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

await requireAutopwnAdmin();

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" });
}

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 버퍼링 차단
  },
});

┌───────────────────────────────────────────────────────────┐
│ 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…   │
└──────────────────────────────────┴────────────────────────┘

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);
  }
};

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;
}

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

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

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

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

                            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

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

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

_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}))

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}

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"),
        })