autopwn 개발기 ④ — Homepage 어드민 콘솔과 AI 게이트의 3중 잠금

시리즈: autopwn 개발기 ④ 이전 글: ③ 첫 라이브 런, SSE 67줄, 그리고 남은 nmap 버그 이 글의 범위: Phase 2 (homepage admin 4탭) + Phase 4 (AI escalation 토글) + Phase 1 트레일에 남았던 nmap adapter 정리.

🧭 어디까지 왔나

Phase	상태	비고
1 — 백엔드 (FastAPI + 어댑터 + scope + SSE)	✅	검증 끝
2 — homepage admin 콘솔 (Next.js)	✅	이 글에서 다룸
3 — Zino Android `:modules:autopwn`	⏸️	폰 무선 디버깅 복구되면
4 — AI escalation 토글	✅	가드 3겹

Zino 앱은 폰 쪽 디버깅 회복되면 그때. autopwn API 계약은 docs/integration-zino.md 에 동결돼 있어서 백엔드는 신경 안 써도 됨.

🧩 Phase 2 설계 — 두 라우트, 한 규칙

/playground/hacking/autopwn 에 어드민 콘솔을 깐다. 핵심은 브라우저가 autopwn API를 직접 보면 안 된다는 룰. autopwn은 127.0.0.1:8080 loopback only 라 외부에선 못 들어오지만, 그건 인프라 가드일 뿐 — 어플리케이션 가드도 별도로 박아야 한다.

라우트 두 개

zino.kr (Next.js)
 ├── /api/playground/autopwn/[...path]/route.ts   ← non-stream catch-all
 └── /api/playground/autopwn/events/stream/       ← SSE bridge (nodejs runtime)

둘 다 진입부에서 같은 한 줄을 강제:

await requireAutopwnAdmin();

canAccessPage(session, "playground.hacking") 검사. 빠뜨리면 catastrophic — admin이 아닌 로그인 세션으로 autopwn 전체 백엔드가 노출됨. 코드 리뷰 1순위 항목.

`autopwn-proxy.ts` 헬퍼

const BASE = process.env.AUTOPWN_BASE_URL || "http://127.0.0.1:8080";
const KEY = process.env.AUTOPWN_ADMIN_KEY || "";
 
export async function autopwnFetch(path: string, init: RequestInit = {}) {
  const headers = new Headers(init.headers);
  headers.set("X-API-Key", KEY);
  return fetch(`${BASE}${path}`, { ...init, headers, cache: "no-store" });
}

X-API-Key 는 서버 사이드에서만 부착. 클라이언트 번들에 절대 안 들어감. devtools에서 Authorization 헤더 검사해서 키가 보이면 그게 곧 인시던트.

Hop-by-hop 제거

브라우저가 보낸 헤더를 그대로 forward 하면 위조 가능한 헤더가 끼어든다. stripHopByHop() 으로 host, x-api-key, connection, te, upgrade 등 제거 후 통과.

SSE bridge

const upstream = await fetch(`${BASE}/events/stream`, {
  headers: { "X-API-Key": KEY },
  signal: req.signal,            // ← 클라가 닫으면 upstream도 닫힘
  cache: "no-store",
});
 
return new Response(upstream.body, {
  headers: {
    "Content-Type": "text/event-stream; charset=utf-8",
    "Cache-Control": "no-cache, no-transform",
    "Connection": "keep-alive",
    "X-Accel-Buffering": "no",   // ← nginx/CDN 버퍼링 차단
  },
});

runtime = "nodejs" — edge runtime은 SSE 호환이 약함
signal: req.signal — Node fetch 18+ 에서 클라 disconnect를 upstream으로 전파
X-Accel-Buffering — 프록시 중간에서 청크 단위로 흘려보내도록

🪟 UI — 4탭 + 라이브 사이드바

┌───────────────────────────────────────────────────────────┐
│ autopwn        SSE● open · API: v1.0 · [+ New Run]         │
├───────────────────────────────────────────────────────────┤
│ Runs | Sessions | Findings | Listeners                     │
├──────────────────────────────────┬────────────────────────┤
│                                  │  Live events           │
│  [tab content]                   │  15:35:02 run.started  │
│                                  │  15:35:02 step.running │
│                                  │  15:49:54 step.succ…   │
└──────────────────────────────────┴────────────────────────┘

오른쪽 라이브 사이드바는 SSE 이벤트 200건 한도. heartbeat 제외하면 SWR mutate 트리거해서 4탭 SWR 캐시도 같이 갱신.

const handleEvent = (type: string) => (raw: MessageEvent) => {
  const data = JSON.parse(raw.data);
  const ev = { type, ...data } as AutopwnEvent;
  setEventLog((prev) => [ev, ...prev].slice(0, 200));
  if (ev.type !== "heartbeat") {
    runsSWR.mutate();
    findingsSWR.mutate();
    sessionsSWR.mutate();
    listenersSWR.mutate();
  }
  if (ev.type === "ai.suggestion") {
    setPendingApproval(ev.suggestion);
  }
};

ai.suggestion 이벤트가 도착하면 모달이 자동으로 뜬다. 닫기/Approve/Deny 셋 중 하나를 골라야 run이 풀림.

TypeScript DTO 1:1

types.ts 가 api/app/models/schemas.py 의 BaseModel과 1:1 매칭. 어느 한쪽을 바꾸면 양쪽 동시 갱신이 룰.

export interface Finding {
  id: string;
  run_id: string;
  severity: "info" | "low" | "medium" | "high" | "critical";
  title: string;
  target: string;
  evidence: Record<string, unknown>;
  cve: string[];
  created_at: string;
}

OpenAPI 자동 생성 코드를 안 쓰는 이유 — 백엔드가 dict[str, Any] 같은 느슨한 타입을 쓰면 자동 생성기가 unknown 으로 만들고 UI 코드가 as 로 단언하게 됨. 그럴 바엔 손으로 1:1 매핑하는 게 안전.

🛡️ Phase 4 — AI escalation 의 3중 잠금

AI를 펜테스트 자동화에 끼우는 건 매력적인 만큼 위험도 높다. autopwn의 디자인 헌법은 "AI는 제안만, 실행은 사람 승인 후 deterministic runner". 이걸 정책 문서 아닌 코드로 박는다.

잠금 ① — 디폴트 OFF, per-run opt-in

# api/app/core/settings.py
ai_escalation_default: bool = Field(default=False, alias="AI_ESCALATION_DEFAULT")

매 실행 시 ai_escalation=true 를 명시 요청해야 함. 모달의 체크박스가 그 진입.

잠금 ② — per-playbook disallow

# api/playbooks/reverse_shell_drop.yaml
name: reverse_shell_drop
disallow_ai: true     # ← C2/익스플로잇 플레이북은 영구 차단

playbooks 라우터에서:

if req.ai_escalation and bool(pb.get("disallow_ai", False)):
    raise HTTPException(400, detail="this playbook disallows AI escalation")

UI 모달 측에서도 disallow_ai 가 true 면 체크박스가 disabled로 표시된다.

잠금 ③ — 제안 도착 후 scope guard 재검증

for sug_step in suggestion.suggested_next:
    scope_guard.validate_step_dict(sug_step, set(ADAPTERS.keys()), ctx["target"])

AI가 제안한 step의 tool 이 화이트리스트(nmap, nuclei, msf.exploit, sliver.*)에 없으면 즉시 ScopeError. 타깃도 다시 scope 통과해야 됨.

세 잠금 중 하나만 풀려도 실행 안 됨. AI는 어디까지나 옵션이라는 말이 코드로 강제됨.

실행 흐름

                            ai_on=true
step 실행 → 예외 발생 ─────────────────→ HexStrike /suggest
   │                                          │
   │  ai_on=false                             │ 제안 받음
   ▼                                          ▼
step.failed → run.failed              step.blocked
                                              │
                                              │ 사람이 모달에서 Approve
                                              ▼
                                  scope_guard.validate_step_dict
                                              │
                                              ▼
                                       deterministic runner
                                              │
                                              ▼
                                  step.succeeded → run.succeeded

작은 정리 — 이중 이벤트 제거

처음 구현했을 때 흐름이 step.failed → step.blocked 로 두 이벤트가 연달아 나왔다 (예외 잡고 _step 호출 후 escalation 진입). UI에선 "어 실패했네… 어 블락됐네?" 가 깜빡임. 패치:

except Exception as e:
    err_msg = str(e)[:200]
    # Try AI escalation FIRST. Only emit step.failed when no recovery path.
    ok = await _maybe_escalate(run_id, sid, pb, ctx, ai_on)
    if not ok:
        await _step(run_id, sid, "failed", summary=err_msg)
        await store.update_run_state(run_id, "failed")
        return

AI 켜져 있고 제안이 오면 _maybe_escalate 내부에서 step.blocked 만 발행. AI가 꺼져 있거나 제안 못 받으면 step.failed.

🧹 Phase 1 트레일 — nmap adapter 정리

3편에서 발견한 두 버그 패치.

버그 ① — `len()` 도 못 쓰는 `_eval_when`

return bool(eval(expr, {"__builtins__": {}}, {"ctx": ctx}))

빈 __builtins__ 로 sandbox 만든답시고 len, any, all 같은 안전한 빌트인까지 다 날아갔다. playbook의 when: "len(ctx['results']['discover']['open_ports']) > 0" 가 NameError → except → False.

화이트리스트로 가드:

_WHEN_BUILTINS = {
    "len": len, "any": any, "all": all, "sum": sum,
    "min": min, "max": max, "bool": bool,
    "int": int, "str": str, "list": list, "dict": dict, "set": set,
}
 
def _eval_when(expr: str, ctx: dict) -> bool:
    ...
    return bool(eval(expr, {"__builtins__": _WHEN_BUILTINS}, {"ctx": ctx}))

sandbox 에서 임의 코드 실행 가능성을 막아야 하지만, eval 자체가 자제할 도구라 — 더 엄격하게 가려면 AST 파서로 갈 일이고 일단 이 정도가 적절.

버그 ② — findings 가 비어 있던 이유

nmap adapter가 open_ports 리스트는 반환했는데 findings 키 자체가 없었다. runner의 finding 등록 로직도 publish만 하고 store 호출이 빠져 있음.

adapter 측에 findings 생성 추가:

findings: list[dict[str, Any]] = []
for h in hosts:
    for p in h["ports"]:
        title = f"Open port {p['port']}/{p['protocol']} ({p['service']})"
        findings.append({
            "severity": "info",
            "title": title,
            "target": h.get("address") or target,
            "evidence": {
                "port": p["port"],
                "service": p["service"],
                "product": p["product"],
                "version": p["version"],
                "via_proxy": is_proxied(),
            },
        })
return {..., "findings": findings}

runner 측에 store 등록 + 이벤트 publish:

if result.get("findings"):
    for raw in result["findings"]:
        finding = Finding(
            id=uuid.uuid4().hex,
            run_id=run_id,
            target=raw.get("target", target),
            severity=raw.get("severity", "info"),
            ...
            created_at=datetime.now(timezone.utc),
        )
        await store.append_finding(finding)
        await publish("autopwn:events:scan", {
            "type": "finding.new", ...
            "finding": finding.model_dump(mode="json"),
        })

이제 recon_basic 한 번 돌리면 발견된 open port 마다 severity=info finding이 Findings 탭에 뜬다. UI 측 Severity filter 도 작동.

✅ 통합 검증 매트릭스

검증	상태	메모
`/api/playground/autopwn/playbooks` 200	⏳	로컬 인프라 띄우면
`/api/playground/autopwn/events/stream` SSE OK	⏳	EventSource 연결 + heartbeat
비-admin 접근 → 403	⏳	`playground.hacking` ACL
AI 모달 (ai.suggestion 이벤트 → 자동 팝업)	⏳	hexstrike 띄우고 web_vuln_chain 실행
`reverse_shell_drop` + `ai_escalation=true` → 400	⏳	router의 disallow_ai 가드
TypeScript 컴파일	✅	`npx tsc --noEmit` 0 에러

로컬 인프라(docker compose up -d) + ./run_homepage.sh dev 띄우면 E2E 확인 가능.

🚀 다음 단계

Phase 3 재개 — Zino 폰 무선 디버깅 복구 후 /zino_module skill 호출. API 계약은 동결돼 있어 백엔드 변경 0
nmap adapter 확장 — -sV 결과의 product/version으로 vulndb 매칭(별 노력 없이) → finding severity 자동 격상
vuln_hints 후속 자동화 — open port에서 http 감지 시 nuclei 체이닝 (이미 web_vuln_chain 에 있음, 트리거만 자동화)
ai 백엔드 교체 가능성 — HexStrike 업스트림이 불안정하면 ReconLM/Claude API 등으로 교체. escalation.py 의 어댑터 함수 한 개만 갈면 됨

📝 정리

두 라우트 한 룰 — 모든 프록시 진입부에 requireAutopwnAdmin(). 빠뜨리면 catastrophic
X-API-Key 는 서버 사이드 only — 브라우저 번들에 절대 노출 X
AI 게이트 3겹 — per-run opt-in · per-playbook disallow · scope/tool 재검증. 셋 다 풀려야 실행
이벤트 일관성 — step.failed → step.blocked 깜빡임 제거, 회복 가능하면 blocked 만 발행
nmap adapter — _eval_when 의 빈 builtins 사각지대 + findings 등록 누락 패치