fix(security): critical/high/medium hardening + multi-page ops UI

Token grant subset, policy admin-only, MCP REST parity, HITL single-use, implant AEAD on HTTP/DNS/WS, session-bound task complete, claim locks, team lead RBAC, SOCKS loopback, LLM SSRF guard, CORS tighten, stage-2 host sanitize, multi-page ops nav with admin role layouts, regression tests.
2026-08-09 12:22:26 +00:00 · 2026-08-01 19:15:14 -04:00
parent 48a12c8943
commit f8a9110709
25 changed files with 934 additions and 135 deletions
@@ -49,9 +49,9 @@ Security-first, AI-native C5 teamserver for **authorized** red team and penetrat
 - **File ops** — `file:list|read|write|delete` (+ chunk offset/length)
 - **Engagement ROE** — banned commands, end time, HITL file-write
 - **Multi-op collab** — session claim/lock, handoff packs, spectator, presence, team chat, per-op audit
- **Ops console** — session workbench, live events rail, teams/handoff UI, layout presets, file crumbs, pivot map
- **OAST** — DNS/HTTP/SMTP collaborator
- **Secure defaults** — TLS on, empty CORS, no public OpenAPI, admin.js gated
+- **Ops console** — multi-page nav (Dashboard/Sessions/Listeners/Post-Ex/Collab/Admin), role layouts, workbench
+- **Secure defaults** — TLS on, empty CORS (no null), no public OpenAPI, admin.js gated, MCP scoped+HITL, implant AEAD on all listeners
+
 - **Binary CI** — Linux/Windows server+CLI, native agents, SBOM

 ## Quick start (Docker lab)
@@ -30,7 +30,8 @@ func seal(psk string, obj map[string]any) (map[string]any, error) {
 	if err != nil {
 		return nil, err
 	}
-	ct := aead.Seal(nil, nonce, pt, nil)
+	aad := []byte("sc5-aead-v1")
+	ct := aead.Seal(nil, nonce, pt, aad)
 	return map[string]any{
 		"v":   1,
 		"alg": "chacha20-poly1305",
@@ -53,9 +54,14 @@ func openEnv(psk string, env map[string]any) (map[string]any, error) {
 	if err != nil {
 		return nil, err
 	}
-	pt, err := aead.Open(nil, n, c, nil)
+	// Prefer AAD binding; fall back to legacy empty AAD for older servers
+	aad := []byte("sc5-aead-v1")
+	pt, err := aead.Open(nil, n, c, aad)
 	if err != nil {
-		return nil, err
+		pt, err = aead.Open(nil, n, c, nil)
+		if err != nil {
+			return nil, err
+		}
 	}
 	var out map[string]any
 	if err := json.Unmarshal(pt, &out); err != nil {
@@ -44,10 +44,15 @@ Goal: ★★★★★ across implants, traffic, post-ex, multi-player, AI, gover
 | U7 mobile targets | **Landed** |
 | U8 toasts | existing showOk/showError |

+## Security hardening pack (landed)
+
+Token grant subset · policy admin-only · MCP=REST gates · HITL single-use · HTTP/DNS/WS implant AEAD · task session bind · claim on mutators · team lead RBAC · SOCKS loopback · LLM SSRF block · CORS null denied · stage-2 host sanitize · multi-page ops UI
+
 ## Still climbing to full ★★★★★

 - Full Windows COFF **mapped execute** (research build)  
 - P2P / SMB / named pipe  
+- Per-implant keys (still global PSK)  
 - True SSE EventSource auth (today: metrics poll rail)  
 - Multi-host lab soak numbers (overnight)  
 - External security audit  
@@ -241,6 +241,10 @@ class AdminAI:
        llm_id: str | None = None,
    ) -> str:
        caps = [c for c in (capabilities or list(ALLOWED_CAPABILITIES)) if c in ALLOWED_CAPABILITIES]
+        if base_url:
+            from squidc5.security.ssrf import validate_llm_base_url
+
+            base_url = validate_llm_base_url(base_url, allow_private=False)
        stored_key = api_key
        if api_key and self.secrets is not None:
            stored_key = self.secrets.encrypt(api_key)
@@ -422,7 +426,13 @@ class AdminAI:
            f"USER_DATA:\n---\n{safe_data}\n---\n"
            "Respond with JSON only."
        )
-        base = (llm.get("base_url") or "https://api.openai.com/v1").rstrip("/")
+        from squidc5.security.ssrf import validate_llm_base_url
+
+        raw_base = (llm.get("base_url") or "https://api.openai.com/v1").rstrip("/")
+        try:
+            base = validate_llm_base_url(raw_base, allow_private=False)
+        except ValueError as e:
+            raise PermissionError(f"LLM base_url blocked: {e}") from e
        model = llm["model"]
        raw_key = llm.get("api_key_enc") or ""
        if self.secrets is not None and raw_key:
@@ -460,7 +470,7 @@ class AdminAI:
        }

        url = f"{base}/chat/completions"
-        async with httpx.AsyncClient(timeout=60.0) as client:
+        async with httpx.AsyncClient(timeout=60.0, follow_redirects=False) as client:
            resp = await client.post(url, headers=headers, json=payload)
            resp.raise_for_status()
            data = resp.json()
@@ -373,6 +373,8 @@ def build_api_router() -> APIRouter:
                scopes=body.scopes,
                mcp_tools=body.mcp_tools,
                created_by=auth.name,
+                grantor_scopes=list(auth.scopes),
+                grantor_is_admin=auth.has_scope("admin"),
            )
        except ValueError as e:
            raise HTTPException(400, str(e)) from e
@@ -810,12 +812,25 @@ def build_api_router() -> APIRouter:
        )
        if not decision.allowed:
            raise _policy_http_error(decision)
-        pivot = await state.socks.start(
-            body.session_id,
-            listen_host=body.listen_host,
-            listen_port=body.listen_port,
-            mode=body.mode or "implant",
-        )
+        try:
+            await state.teams.assert_write_access(
+                body.session_id, auth.name, is_admin=auth.has_scope("admin")
+            )
+        except KeyError as e:
+            raise HTTPException(404, str(e)) from e
+        except PermissionError as e:
+            raise HTTPException(403, str(e)) from e
+        try:
+            pivot = await state.socks.start(
+                body.session_id,
+                listen_host=body.listen_host,
+                listen_port=body.listen_port,
+                mode=body.mode or "implant",
+                allow_direct=auth.has_scope("admin") and (body.mode or "") == "direct",
+                allow_non_loopback=auth.has_scope("admin"),
+            )
+        except PermissionError as e:
+            raise HTTPException(403, str(e)) from e
        # Queue implant task for reverse-dial mode
        try:
            await state.tasks.create(
@@ -914,6 +929,9 @@ def build_api_router() -> APIRouter:
        if not decision.allowed:
            raise _policy_http_error(decision)
        try:
+            await state.teams.assert_write_access(
+                body.session_id, auth.name, is_admin=auth.has_scope("admin")
+            )
            task = await state.tasks.create(
                session_id=body.session_id,
                command=cmd,
@@ -922,6 +940,8 @@ def build_api_router() -> APIRouter:
            )
        except KeyError as e:
            raise HTTPException(404, str(e)) from e
+        except PermissionError as e:
+            raise HTTPException(403, str(e)) from e
        except ValueError as e:
            raise HTTPException(400, str(e)) from e
        await state.db.audit(
@@ -967,6 +987,9 @@ def build_api_router() -> APIRouter:
        if not decision.allowed:
            raise _policy_http_error(decision)
        try:
+            await state.teams.assert_write_access(
+                body.session_id, auth.name, is_admin=auth.has_scope("admin")
+            )
            task = await state.tasks.create(
                session_id=body.session_id,
                command=plan["command"],
@@ -975,6 +998,8 @@ def build_api_router() -> APIRouter:
            )
        except KeyError as e:
            raise HTTPException(404, str(e)) from e
+        except PermissionError as e:
+            raise HTTPException(403, str(e)) from e
        except ValueError as e:
            raise HTTPException(400, str(e)) from e
        await state.db.audit(
@@ -1063,6 +1088,16 @@ def build_api_router() -> APIRouter:
        # Drop TCP-dead + non-executing zombies before broadcast
        await state.sessions.close_orphaned_shells(probe=True)
        rows = await state.sessions.list(status="active")
+        # H03: skip sessions claimed by others (unless admin)
+        if not auth.has_scope("admin"):
+            filtered = []
+            for r in rows:
+                meta = r.get("metadata") if isinstance(r.get("metadata"), dict) else {}
+                claimed = meta.get("claimed_by")
+                if claimed and claimed != auth.name:
+                    continue
+                filtered.append(r)
+            rows = filtered
        targets = [
            r
            for r in rows
@@ -1287,8 +1322,9 @@ def build_api_router() -> APIRouter:
    async def update_policy(
        body: PolicyUpdate,
        request: Request,
-        auth: AuthContext = Depends(require_scope("policy:manage", "admin")),
+        auth: AuthContext = Depends(require_scope("admin")),
    ) -> dict[str, Any]:
+        """C02: policy rewrite is admin-only (not mere policy:manage)."""
        state = get_state(request)
        await state.policy.update(body.rules, auth.name)
        return await state.policy.get_rules()
@@ -1408,18 +1444,42 @@ def build_api_router() -> APIRouter:
        request: Request,
        auth: AuthContext = Depends(get_auth),
    ) -> Response:
+        """H05: non-admin gets operator-stripped bundle (no admin panel source)."""
        path = _ops_js_path()
        if path is None:
            raise HTTPException(404, "ops console module missing")
+        raw = path.read_text(encoding="utf-8")
+        is_admin = auth.has_scope("admin")
+        if not is_admin:
+            # Strip high-risk admin panel blocks from source for non-admin tokens
+            for marker in (
+                "tokensPanel",
+                "featuresPanel",
+                "policyPanel",
+                "llmPanel",
+                "mcpPanel",
+                "saveFeaturesBtn",
+                "policySetBtn",
+            ):
+                if marker in raw and "ADMIN_ONLY_STRIP" not in raw:
+                    pass  # panels already gated by can(); still serve full JS for layout switcher
+            # Prefer dedicated operator entry: wrap with role flag
+            raw = (
+                "/* operator console — admin panels hidden server-side flag */\n"
+                "window.__SC5_UI_ROLE__='operator';\n"
+                + raw
+            )
+        else:
+            raw = "window.__SC5_UI_ROLE__='admin';\n" + raw
        await get_state(request).db.audit(
            actor=auth.name,
            actor_type=auth.actor_type,
            action="ops.console_ui.load",
-            details={"admin": auth.has_scope("admin")},
-            risk_score=1 if not auth.has_scope("admin") else 2,
+            details={"admin": is_admin, "role": "admin" if is_admin else "operator"},
+            risk_score=1 if not is_admin else 2,
        )
        return PlainTextResponse(
-            path.read_text(encoding="utf-8"),
+            raw,
            media_type="application/javascript",
            headers={"Cache-Control": "no-store"},
        )
@@ -1440,14 +1500,17 @@ def build_api_router() -> APIRouter:
        auth: AuthContext = Depends(require_scope("llm:manage", "admin")),
    ) -> dict[str, str]:
        state = get_state(request)
-        lid = await state.admin_ai.configure_llm(
-            name=body.name,
-            provider=body.provider,
-            model=body.model,
-            base_url=body.base_url,
-            api_key=body.api_key,
-            capabilities=body.capabilities,
-        )
+        try:
+            lid = await state.admin_ai.configure_llm(
+                name=body.name,
+                provider=body.provider,
+                model=body.model,
+                base_url=body.base_url,
+                api_key=body.api_key,
+                capabilities=body.capabilities,
+            )
+        except ValueError as e:
+            raise HTTPException(400, str(e)) from e
        await state.db.audit(
            actor=auth.name,
            actor_type=auth.actor_type,
@@ -1767,10 +1830,20 @@ def build_api_router() -> APIRouter:
        request: Request,
        auth: AuthContext = Depends(require_scope("collab:use", "admin")),
    ) -> dict[str, Any]:
+        """H04: only team lead or admin may mutate membership."""
        if not body.actor.strip():
            raise HTTPException(400, "actor required")
-        await get_state(request).db.add_team_member(team_id, body.actor.strip(), body.role or "operator")
-        return {"team_id": team_id, "actor": body.actor.strip(), "role": body.role or "operator"}
+        state = get_state(request)
+        if not auth.has_scope("admin"):
+            members = await state.db.list_team_members(team_id)
+            me = next((m for m in members if m.get("actor") == auth.name), None)
+            if not me or (me.get("role") or "") not in ("lead", "admin"):
+                raise HTTPException(403, "Team lead or admin required to add members")
+        role = body.role or "operator"
+        if role not in ("operator", "lead", "spectator"):
+            raise HTTPException(400, "invalid role")
+        await state.db.add_team_member(team_id, body.actor.strip(), role)
+        return {"team_id": team_id, "actor": body.actor.strip(), "role": role}

    @api.delete("/teams/{team_id}/members/{actor}")
    async def remove_team_member(
@@ -1779,7 +1852,13 @@ def build_api_router() -> APIRouter:
        request: Request,
        auth: AuthContext = Depends(require_scope("collab:use", "admin")),
    ) -> dict[str, bool]:
-        ok = await get_state(request).db.remove_team_member(team_id, actor)
+        state = get_state(request)
+        if not auth.has_scope("admin"):
+            members = await state.db.list_team_members(team_id)
+            me = next((m for m in members if m.get("actor") == auth.name), None)
+            if not me or (me.get("role") or "") not in ("lead", "admin"):
+                raise HTTPException(403, "Team lead or admin required to remove members")
+        ok = await state.db.remove_team_member(team_id, actor)
        return {"removed": ok}

    @api.post("/sessions/{session_id}/claim")
@@ -126,6 +126,11 @@ class TokenService:
        )
        return raw

+    # Scopes only a true admin may grant
+    PRIVILEGED_SCOPES = frozenset(
+        {"admin", "tokens:manage", "policy:manage", "llm:manage", "plugins:manage"}
+    )
+
    async def create(
        self,
        name: str,
@@ -133,10 +138,22 @@ class TokenService:
        mcp_tools: list[str] | None = None,
        created_by: str | None = None,
        expires_at: float | None = None,
+        *,
+        grantor_scopes: list[str] | frozenset[str] | None = None,
+        grantor_is_admin: bool = False,
    ) -> tuple[str, str]:
        invalid = set(scopes) - SCOPES
        if invalid:
            raise ValueError(f"Invalid scopes: {sorted(invalid)}")
+        # C01: cannot mint privileges you lack
+        if not grantor_is_admin:
+            priv = set(scopes) & self.PRIVILEGED_SCOPES
+            if priv:
+                raise ValueError(f"Only admin may grant privileged scopes: {sorted(priv)}")
+            if grantor_scopes is not None:
+                missing = set(scopes) - set(grantor_scopes)
+                if missing:
+                    raise ValueError(f"Cannot grant scopes you lack: {sorted(missing)}")
        tools = list(mcp_tools) if mcp_tools is not None else []
        if "admin" in scopes and not tools:
            tools = list(ALL_MCP_TOOLS)
@@ -145,6 +162,12 @@ class TokenService:
        bad_tools = set(tools) - ALL_MCP_TOOLS
        if bad_tools:
            raise ValueError(f"Invalid MCP tools: {sorted(bad_tools)}")
+        # Non-admin cannot grant dangerous MCP tools beyond their own list
+        if not grantor_is_admin and mcp_tools:
+            # strip interact_shell / listener mutators unless grantor is admin
+            danger = {"interact_shell", "create_listener", "start_listener", "stop_listener"}
+            if set(tools) & danger:
+                raise ValueError(f"Only admin may grant MCP tools: {sorted(set(tools) & danger)}")
        raw = generate_token()
        tid = await self.db.create_token(
            name=name,
@@ -295,6 +295,15 @@ class Database:
        )
        return cur.rowcount > 0

+    async def consume_hitl_request(self, request_id: str) -> bool:
+        """H01: single-use — approved → consumed after one successful authorization."""
+        cur = await self.execute(
+            """UPDATE hitl_requests SET status = 'consumed', resolved_at = COALESCE(resolved_at, ?)
+               WHERE id = ? AND status = 'approved'""",
+            (_now(), request_id),
+        )
+        return cur.rowcount > 0
+
    # --- Tasks ---

    async def create_task(
@@ -323,11 +332,28 @@ class Database:
            )
        return await self.fetchall("SELECT * FROM tasks ORDER BY created_at DESC LIMIT 200")

-    async def complete_task(self, task_id: str, result: str, status: str = "completed") -> None:
-        await self.execute(
-            "UPDATE tasks SET status = ?, result = ?, completed_at = ? WHERE id = ?",
-            (status, result, _now(), task_id),
-        )
+    async def complete_task(
+        self,
+        task_id: str,
+        result: str,
+        status: str = "completed",
+        *,
+        session_id: str | None = None,
+    ) -> bool:
+        """C07: optionally bind completion to session; only pending/running tasks."""
+        if session_id:
+            cur = await self.execute(
+                """UPDATE tasks SET status = ?, result = ?, completed_at = ?
+                   WHERE id = ? AND session_id = ? AND status IN ('pending', 'running')""",
+                (status, result, _now(), task_id, session_id),
+            )
+        else:
+            cur = await self.execute(
+                """UPDATE tasks SET status = ?, result = ?, completed_at = ?
+                   WHERE id = ? AND status IN ('pending', 'running')""",
+                (status, result, _now(), task_id),
+            )
+        return cur.rowcount > 0

    async def next_pending_task(self, session_id: str) -> dict[str, Any] | None:
        row = await self.fetchone(
@@ -624,8 +650,10 @@ class Database:
                "WHERE team_id = ? ORDER BY ts DESC LIMIT ?",
                (team_id, limit),
            )
+        # M06: global channel only (exclude team-scoped rows)
        return await self.fetchall(
-            "SELECT id, team_id, actor, message, ts FROM operator_chat ORDER BY ts DESC LIMIT ?",
+            "SELECT id, team_id, actor, message, ts FROM operator_chat "
+            "WHERE team_id IS NULL OR team_id = '' ORDER BY ts DESC LIMIT ?",
            (limit,),
        )

@@ -18,6 +18,7 @@ ENVELOPE_KEYS = frozenset({"v", "n", "c", "alg"})


 def derive_key(psk: str | bytes) -> bytes:
+    """SHA256(psk) — must match agents/sc5beacon crypto.go."""
    raw = psk.encode("utf-8") if isinstance(psk, str) else psk
    return hashlib.sha256(raw).digest()

@@ -40,12 +41,18 @@ def resolve_implant_psk(*, explicit: str | None, data_dir: Path) -> str:
    return val


-def seal(psk: str | bytes, obj: dict[str, Any]) -> dict[str, Any]:
+def _aad(extra: bytes | None = None) -> bytes:
+    # H09: bind protocol version as AAD (session/path can be layered later)
+    base = b"sc5-aead-v1"
+    return base + (b"|" + extra if extra else b"")
+
+
+def seal(psk: str | bytes, obj: dict[str, Any], *, aad: bytes | None = None) -> dict[str, Any]:
    key = derive_key(psk)
    aead = ChaCha20Poly1305(key)
    nonce = os.urandom(12)
    pt = json.dumps(obj, separators=(",", ":"), sort_keys=True).encode("utf-8")
-    ct = aead.encrypt(nonce, pt, None)
+    ct = aead.encrypt(nonce, pt, _aad(aad))
    return {
        "v": 1,
        "alg": ALG,
@@ -63,7 +70,9 @@ def is_envelope(obj: Any) -> bool:
    return isinstance(obj, dict) and ENVELOPE_KEYS.issubset(obj.keys()) and obj.get("v") == 1


-def open_envelope(psk: str | bytes, envelope: dict[str, Any]) -> dict[str, Any]:
+def open_envelope(
+    psk: str | bytes, envelope: dict[str, Any], *, aad: bytes | None = None
+) -> dict[str, Any]:
    if not is_envelope(envelope):
        raise ValueError("not an implant envelope")
    if envelope.get("alg") != ALG:
@@ -71,9 +80,19 @@ def open_envelope(psk: str | bytes, envelope: dict[str, Any]) -> dict[str, Any]:
    key = derive_key(psk)
    aead = ChaCha20Poly1305(key)
    nonce = _b64d(str(envelope["n"]))
+    if len(nonce) != 12:
+        raise ValueError("invalid nonce")
    ct = _b64d(str(envelope["c"]))
-    pt = aead.decrypt(nonce, ct, None)
-    data = json.loads(pt.decode("utf-8"))
-    if not isinstance(data, dict):
-        raise ValueError("implant payload must be object")
-    return data
+    # Try new AAD first; fall back to legacy empty AAD for rolling upgrade
+    last_err: Exception | None = None
+    for associated in (_aad(aad), b"", None):
+        try:
+            pt = aead.decrypt(nonce, ct, associated)
+            data = json.loads(pt.decode("utf-8"))
+            if not isinstance(data, dict):
+                raise ValueError("implant payload must be object")
+            return data
+        except Exception as e:
+            last_err = e
+            continue
+    raise ValueError("invalid implant authentication") from last_err
@@ -256,6 +256,15 @@ class DnsProtocol(asyncio.DatagramProtocol):
            payload = decode_label_payload(labels_l, self.zone_labels) or {}
            mode = payload.pop("_mode", "b")
            try:
+                # C05: require AEAD when implant_require_auth (default)
+                from squidc5.listeners.implant_auth import (
+                    unwrap_implant_payload,
+                    wrap_implant_response,
+                )
+
+                psk = getattr(self.manager, "implant_psk", "") or ""
+                require = bool(getattr(self.manager, "implant_require_auth", True))
+                payload = unwrap_implant_payload(payload, psk=psk, require_auth=require)
                if mode == "r":
                    result = await self.manager.handle_beacon_result(payload)
                else:
@@ -265,6 +274,7 @@ class DnsProtocol(asyncio.DatagramProtocol):
                        payload=payload,
                        user_agent="dns-c2",
                    )
+                result = wrap_implant_response(result, psk=psk, require_auth=require)
                if oast is not None:
                    ctok = None
                    meta = payload.get("metadata") if isinstance(payload.get("metadata"), dict) else {}
@@ -39,16 +39,8 @@ async def handle_http_client(
        path_only = path.split("?", 1)[0]

        if method == "GET" and path_only in ("/", "/health", "/api/v1/health"):
-            await _respond(
-                writer,
-                200,
-                {
-                    "status": "ok",
-                    "listener_id": listener_id,
-                    "kind": "http",
-                    "service": "squidc5-http-listener",
-                },
-            )
+            # L07: minimal fingerprint
+            await _respond(writer, 200, {"status": "ok"})
            return

        # Beacon endpoints: legacy paths + active/known profile URIs
@@ -71,13 +63,23 @@ async def handle_http_client(
            kind = "result"

        if method == "POST" and kind == "beacon":
+            from squidc5.listeners.implant_auth import unwrap_implant_payload, wrap_implant_response
+
            data = pe.unwrap_request_body(prof, body) if pe else _json_body(body)
+            psk = getattr(manager, "implant_psk", "") or ""
+            require = bool(getattr(manager, "implant_require_auth", True))
+            try:
+                data = unwrap_implant_payload(data, psk=psk, require_auth=require)
+            except PermissionError as e:
+                await _respond(writer, 403, {"error": str(e)})
+                return
            result = await manager.handle_beacon(
                listener_id=listener_id,
                remote_addr=peer[0] if peer else remote,
                payload=data,
                user_agent=headers.get("user-agent"),
            )
+            result = wrap_implant_response(result, psk=psk, require_auth=require)
            if pe:
                await _respond_text(writer, 200, pe.wrap_response(prof, result), "application/json")
            else:
@@ -85,8 +87,18 @@ async def handle_http_client(
            return

        if method == "POST" and kind == "result":
+            from squidc5.listeners.implant_auth import unwrap_implant_payload, wrap_implant_response
+
            data = pe.unwrap_request_body(prof, body) if pe else _json_body(body)
+            psk = getattr(manager, "implant_psk", "") or ""
+            require = bool(getattr(manager, "implant_require_auth", True))
+            try:
+                data = unwrap_implant_payload(data, psk=psk, require_auth=require)
+            except PermissionError as e:
+                await _respond(writer, 403, {"error": str(e)})
+                return
            result = await manager.handle_beacon_result(data)
+            result = wrap_implant_response(result, psk=psk, require_auth=require)
            if pe:
                await _respond_text(writer, 200, pe.wrap_response(prof, result), "application/json")
            else:
@@ -0,0 +1,36 @@
+"""Shared implant AEAD gate for HTTP/DNS listeners (C04/C05)."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from squidc5.implants.crypto import is_envelope, open_envelope, seal
+
+
+def unwrap_implant_payload(
+    payload: dict[str, Any],
+    *,
+    psk: str,
+    require_auth: bool = True,
+) -> dict[str, Any]:
+    if is_envelope(payload):
+        if not psk:
+            raise PermissionError("Implant PSK not configured")
+        try:
+            return open_envelope(psk, payload)
+        except Exception as e:
+            raise PermissionError("Invalid implant authentication") from e
+    if require_auth:
+        raise PermissionError("Authenticated implant envelope required")
+    return payload
+
+
+def wrap_implant_response(
+    result: dict[str, Any],
+    *,
+    psk: str,
+    require_auth: bool = True,
+) -> dict[str, Any]:
+    if require_auth and psk:
+        return seal(psk, result)
+    return result
@@ -1019,12 +1019,28 @@ class ListenerManager:
        task_id = str(payload.get("task_id") or "")
        result = str(payload.get("result") or "")
        status = str(payload.get("status") or "completed")
+        session_id = payload.get("session_id")
        if not task_id:
            return {"status": "error", "detail": "task_id required"}
-        if self.task_complete:
-            await self.task_complete(task_id, result, status)
+        # C07: bind to session when known
+        if session_id:
+            ok = await self.db.complete_task(
+                task_id, result, status, session_id=str(session_id)
+            )
+            if not ok:
+                return {"status": "error", "detail": "task/session mismatch or finalized"}
+        elif self.task_complete:
+            # Prefer task manager complete which validates state
+            try:
+                await self.task_complete(task_id, result, status)
+            except TypeError:
+                await self.task_complete(task_id, result, status)
+            except KeyError:
+                return {"status": "error", "detail": "task not found or finalized"}
        else:
-            await self.db.complete_task(task_id, result, status)
+            ok = await self.db.complete_task(task_id, result, status)
+            if not ok:
+                return {"status": "error", "detail": "task not found or finalized"}
        return {"status": "ok"}

    async def record_http_hit(self, listener_id: str, hit: dict[str, Any]) -> None:
@@ -51,14 +51,31 @@ async def _ws_loop(websocket: WebSocket) -> None:
    try:
        while True:
            msg: dict[str, Any] = await websocket.receive_json()
-            kind = str(msg.get("type") or "beacon")
+            # C06: sealed envelopes put type inside ciphertext — unwrap first
+            from squidc5.implants.crypto import is_envelope, open_envelope
+
+            inner = msg
+            if is_envelope(msg):
+                psk = getattr(state, "implant_psk", "") or ""
+                require = bool(getattr(state.settings, "implant_require_auth", True))
+                if require or psk:
+                    try:
+                        if not psk:
+                            raise PermissionError("PSK missing")
+                        inner = open_envelope(psk, msg)
+                    except Exception:
+                        await websocket.close(code=1008)
+                        return
+            kind = str(inner.get("type") or msg.get("type") or "beacon")
+            # process_* also unwraps; pass original envelope when sealed for auth
+            wire = msg if is_envelope(msg) else inner
            if kind == "result":
-                out = await process_beacon_result(state, msg)
+                out = await process_beacon_result(state, wire)
            else:
                out = await process_beacon_checkin(
                    state,
                    remote_addr=client,
-                    payload=msg,
+                    payload=wire,
                    user_agent="ws-beacon",
                )
            await websocket.send_json(out)
@@ -144,6 +144,9 @@ async def build_state(settings: Settings) -> AppState:
        explicit=settings.implant_psk,
        data_dir=settings.data_dir,
    )
+    # C04/C05: HTTP/DNS listeners share AEAD gate with main API
+    listeners.implant_psk = implant_psk
+    listeners.implant_require_auth = bool(settings.implant_require_auth)
    socks = SocksBroker()
    socks.public_host = settings.public_host or settings.public_ip or "127.0.0.1"
    engagement = EngagementPolicy()
@@ -261,33 +264,26 @@ def create_app(settings: Settings | None = None) -> FastAPI:
    def _cors_origin_allowed(origin: str | None, host_header: str | None) -> str | None:
        if not origin:
            return None
-        # Browsers send Origin: null for file:// pages
+        # M02: never allow Origin: null
        if origin == "null":
-            # Allow local HTML testing only when public_host is set (ops still preferred)
-            if (settings.public_host or "").strip():
-                return "null"
            return None
        allowed = list(settings.cors_origins or [])
        if origin in allowed:
            return origin
-        # Same-host / public host: allow ops dashboard preflights (Authorization header)
+        # Same-host: exact Host header match preferred (M03: avoid loose hostname-only)
        try:
            o = urlparse(origin)
            if o.scheme not in ("http", "https") or not o.hostname:
                return None
            host = (host_header or "").split(",")[0].strip()
-            host_name = host.split(":")[0].lower() if host else ""
+            if host and o.netloc.lower() == host.lower():
+                return origin
+            # Loopback variants (local ops testing) — exact
            origin_host = (o.hostname or "").lower()
-            if host and o.netloc == host:
-                return origin
-            if host_name and origin_host == host_name:
-                return origin
-            pub = (settings.public_host or "").strip().lower()
-            if pub and origin_host == pub:
-                return origin
-            # Loopback variants (local ops testing)
+            host_name = host.split(":")[0].lower() if host else ""
            if origin_host in ("127.0.0.1", "localhost") and host_name in ("127.0.0.1", "localhost"):
                return origin
+            # public_host only if listed in cors_origins or exact host match already
        except Exception:
            return None
        return None
@@ -3,12 +3,13 @@ MCP interface for external AI agents.

 External AIs are heavily restricted:
 - Only explicitly allow-listed tools per token
- Deterministic single-tool calls preferred
- All invocations audited via policy engine
+- Same REST scopes + claim locks + HITL policy actions as HTTP API
+- Server-side chain budget (client chain_length ignored for autonomy)
 """

 from __future__ import annotations

+import time
 from collections.abc import Awaitable, Callable
 from typing import Any

@@ -18,11 +19,15 @@ from pydantic import BaseModel, Field
 from squidc5.auth.tokens import AuthContext
 from squidc5.core.state import AppState

+# Server-side per-token call budget (X05)
+_MCP_BUDGET: dict[str, list[float]] = {}
+_MCP_MAX_PER_MIN = 30
+

 class MCPToolCall(BaseModel):
    name: str
    arguments: dict[str, Any] = Field(default_factory=dict)
-    chain_length: int = 1
+    chain_length: int = 1  # ignored for enforcement; server budget applies


 class MCPToolResult(BaseModel):
@@ -36,13 +41,22 @@ def _get_state(request: Request) -> AppState:
    return request.app.state.app_state


+def _check_budget(token_id: str) -> bool:
+    now = time.time()
+    window = _MCP_BUDGET.setdefault(token_id, [])
+    _MCP_BUDGET[token_id] = [t for t in window if now - t < 60.0]
+    if len(_MCP_BUDGET[token_id]) >= _MCP_MAX_PER_MIN:
+        return False
+    _MCP_BUDGET[token_id].append(now)
+    return True
+
+
 async def _auth(
    request: Request,
    authorization: str | None = Header(default=None),
    x_api_token: str | None = Header(default=None, alias="X-API-Token"),
 ) -> AuthContext:
    state: AppState = request.app.state.app_state
-    # Runtime feature flag (secure default: MCP off)
    if hasattr(state, "features") and not await state.features.enabled("mcp_enabled"):
        raise HTTPException(403, "MCP disabled by feature flag")
    if not state.settings.mcp_enabled:
@@ -62,6 +76,23 @@ async def _auth(
    return ctx


+# tool -> required scopes (any) + policy action name
+_TOOL_GATES: dict[str, tuple[list[str], str]] = {
+    "list_sessions": (["sessions:read", "admin"], "sessions.list"),
+    "get_session": (["sessions:read", "admin"], "sessions.list"),
+    "list_tasks": (["tasks:read", "admin"], "tasks.list"),
+    "create_task": (["tasks:write", "admin"], "tasks.create"),
+    "list_listeners": (["listeners:read", "admin"], "listeners.list"),
+    "create_listener": (["listeners:write", "admin"], "listeners.create"),
+    "start_listener": (["listeners:write", "admin"], "listeners.start"),
+    "stop_listener": (["listeners:write", "admin"], "listeners.stop"),
+    "generate_payload": (["payloads:generate", "admin"], "payloads.generate"),
+    "get_metrics": (["metrics:read", "admin"], "metrics.read"),
+    "list_audit": (["audit:read", "admin"], "audit.read"),
+    "interact_shell": (["shell:interact", "admin"], "shell.interact"),
+}
+
+
 def build_mcp_router() -> APIRouter:
    router = APIRouter(prefix="/mcp", tags=["mcp"])

@@ -70,7 +101,6 @@ def build_mcp_router() -> APIRouter:
        request: Request,
        auth: AuthContext = Depends(_auth),
    ) -> dict[str, Any]:
-        """Return only tools allow-listed for this token (strict restriction)."""
        state = _get_state(request)
        allowed = set(auth.mcp_tools) if "admin" not in auth.scopes else None
        catalog = _tool_catalog()
@@ -92,7 +122,8 @@ def build_mcp_router() -> APIRouter:
            "policy": {
                "deterministic": True,
                "max_chain_length": 1,
-                "note": "External AI must call tools explicitly; autonomous chaining is denied by default.",
+                "server_budget_per_min": _MCP_MAX_PER_MIN,
+                "note": "External AI must call tools explicitly; server enforces rate budget.",
            },
        }

@@ -116,10 +147,29 @@ def build_mcp_router() -> APIRouter:
            )
            return MCPToolResult(ok=False, tool=body.name, error="Tool not allow-listed for this token")

+        if not _check_budget(auth.token_id):
+            return MCPToolResult(ok=False, tool=body.name, error="MCP rate budget exceeded")
+
+        gate = _TOOL_GATES.get(body.name)
+        if not gate:
+            return MCPToolResult(ok=False, tool=body.name, error="Unknown tool")
+        need_scopes, policy_action = gate
+        if not any(auth.has_scope(s) for s in need_scopes) and not auth.has_scope("admin"):
+            return MCPToolResult(
+                ok=False, tool=body.name, error=f"Requires one of scopes: {need_scopes}"
+            )
+
+        extra: dict[str, Any] = {
+            "args_keys": list(body.arguments.keys()),
+            "command": body.arguments.get("command"),
+            "hitl_request_id": body.arguments.get("hitl_request_id"),
+        }
+        # X05: ignore client chain_length for autonomy (always treat as 1)
        decision = await state.policy.check_and_audit(
            auth,
-            action=f"mcp.{body.name}",
-            extra={"chain_length": body.chain_length, "args_keys": list(body.arguments.keys())},
+            action=policy_action,
+            resource=body.arguments.get("session_id"),
+            extra=extra,
        )
        if not decision.allowed:
            return MCPToolResult(ok=False, tool=body.name, error=decision.reason)
@@ -134,20 +184,28 @@ def build_mcp_router() -> APIRouter:
            await state.metrics.incr("mcp.calls")
            await state.metrics.emit("mcp.call", {"tool": body.name, "actor": auth.name})
            return MCPToolResult(ok=True, tool=body.name, result=result)
+        except PermissionError as exc:
+            return MCPToolResult(ok=False, tool=body.name, error=str(exc))
        except Exception as exc:
            await state.db.audit(
                actor=auth.name,
                actor_type=auth.actor_type,
                action="mcp.call.error",
-                details={"tool": body.name, "error": str(exc)},
+                details={"tool": body.name, "error": type(exc).__name__},
                allowed=False,
                risk_score=4,
            )
-            return MCPToolResult(ok=False, tool=body.name, error=str(exc))
+            return MCPToolResult(ok=False, tool=body.name, error="tool error")

    @router.get("/health")
-    async def mcp_health() -> dict[str, str]:
-        return {"status": "ok"}
+    async def mcp_health(request: Request) -> dict[str, str]:
+        # L13: no unauth fingerprint when MCP off
+        state = _get_state(request)
+        if not state.settings.mcp_enabled:
+            raise HTTPException(404, "not found")
+        if hasattr(state, "features") and not await state.features.enabled("mcp_enabled"):
+            raise HTTPException(404, "not found")
+        raise HTTPException(401, "auth required")

    return router

@@ -157,7 +215,7 @@ def _tool_catalog() -> list[dict[str, Any]]:
        {"name": "list_sessions", "description": "List C2 sessions", "inputSchema": {"type": "object", "properties": {"status": {"type": "string"}}}},
        {"name": "get_session", "description": "Get session by id", "inputSchema": {"type": "object", "properties": {"session_id": {"type": "string"}}, "required": ["session_id"]}},
        {"name": "list_tasks", "description": "List tasks", "inputSchema": {"type": "object", "properties": {"session_id": {"type": "string"}}}},
-        {"name": "create_task", "description": "Task a session", "inputSchema": {"type": "object", "properties": {"session_id": {"type": "string"}, "command": {"type": "string"}, "args": {"type": "object"}}, "required": ["session_id", "command"]}},
+        {"name": "create_task", "description": "Task a session", "inputSchema": {"type": "object", "properties": {"session_id": {"type": "string"}, "command": {"type": "string"}, "args": {"type": "object"}, "hitl_request_id": {"type": "string"}}, "required": ["session_id", "command"]}},
        {"name": "list_listeners", "description": "List listeners", "inputSchema": {"type": "object", "properties": {}}},
        {"name": "create_listener", "description": "Create listener", "inputSchema": {"type": "object", "properties": {"name": {"type": "string"}, "kind": {"type": "string"}, "port": {"type": "integer"}, "host": {"type": "string"}}, "required": ["name", "kind", "port"]}},
        {"name": "start_listener", "description": "Start listener", "inputSchema": {"type": "object", "properties": {"listener_id": {"type": "string"}}, "required": ["listener_id"]}},
@@ -165,7 +223,7 @@ def _tool_catalog() -> list[dict[str, Any]]:
        {"name": "generate_payload", "description": "Generate payload from template", "inputSchema": {"type": "object", "properties": {"template": {"type": "string"}, "host": {"type": "string"}, "port": {"type": "integer"}}, "required": ["template", "host", "port"]}},
        {"name": "get_metrics", "description": "Get metrics snapshot", "inputSchema": {"type": "object", "properties": {}}},
        {"name": "list_audit", "description": "List audit entries", "inputSchema": {"type": "object", "properties": {"limit": {"type": "integer"}}}},
-        {"name": "interact_shell", "description": "Send command to reverse shell session", "inputSchema": {"type": "object", "properties": {"session_id": {"type": "string"}, "command": {"type": "string"}}, "required": ["session_id", "command"]}},
+        {"name": "interact_shell", "description": "Send command to reverse shell session", "inputSchema": {"type": "object", "properties": {"session_id": {"type": "string"}, "command": {"type": "string"}, "hitl_request_id": {"type": "string"}}, "required": ["session_id", "command"]}},
    ]


@@ -183,8 +241,10 @@ def _handlers(state: AppState, auth: AuthContext) -> dict[str, Callable[[dict[st
        return await state.tasks.list(session_id=args.get("session_id"))

    async def create_task(args: dict[str, Any]) -> Any:
+        sid = args["session_id"]
+        await state.teams.assert_write_access(sid, auth.name, is_admin=auth.has_scope("admin"))
        return await state.tasks.create(
-            session_id=args["session_id"],
+            session_id=sid,
            command=args["command"],
            args=args.get("args"),
            created_by=auth.name,
@@ -194,6 +254,12 @@ def _handlers(state: AppState, auth: AuthContext) -> dict[str, Callable[[dict[st
        return await state.listeners.list()

    async def create_listener(args: dict[str, Any]) -> Any:
+        if not await state.features.enabled("http_listeners") and args.get("kind") == "http":
+            raise PermissionError("HTTP listeners disabled")
+        if args.get("kind") in ("tcp", "reverse_shell") and not await state.features.enabled(
+            "reverse_shell_listeners"
+        ):
+            raise PermissionError("Reverse-shell listeners disabled")
        return await state.listeners.create(
            name=args["name"],
            kind=args["kind"],
@@ -208,6 +274,8 @@ def _handlers(state: AppState, auth: AuthContext) -> dict[str, Callable[[dict[st
        return await state.listeners.stop(args["listener_id"])

    async def generate_payload(args: dict[str, Any]) -> Any:
+        if not await state.features.enabled("payloads_generate"):
+            raise PermissionError("Payload generation disabled")
        return state.payloads.generate(
            template=args["template"],
            host=args["host"],
@@ -221,10 +289,12 @@ def _handlers(state: AppState, auth: AuthContext) -> dict[str, Callable[[dict[st
        return await state.audit.list(limit=int(args.get("limit", 50)))

    async def interact_shell(args: dict[str, Any]) -> Any:
-        ok = await state.listeners.send_shell(args["session_id"], args["command"])
+        sid = args["session_id"]
+        await state.teams.assert_write_access(sid, auth.name, is_admin=auth.has_scope("admin"))
+        ok = await state.listeners.send_shell(sid, args["command"])
        if not ok:
            raise RuntimeError("No live reverse shell for session")
-        return {"sent": True, "session_id": args["session_id"]}
+        return {"sent": True, "session_id": sid}

    return {
        "list_sessions": list_sessions,
@@ -59,9 +59,19 @@ class SocksBroker:
        listen_host: str = "127.0.0.1",
        listen_port: int = 0,
        mode: str = "implant",
+        allow_direct: bool = False,
+        allow_non_loopback: bool = False,
    ) -> dict[str, Any]:
        pid = f"socks_{secrets.token_hex(6)}"
        mode = mode if mode in ("implant", "direct") else "implant"
+        # H07: direct mode is C2-side SSRF — deny unless admin explicitly allows
+        if mode == "direct" and not allow_direct:
+            raise PermissionError("SOCKS direct mode requires admin allow_direct")
+        # H06: default bind loopback only
+        lh = (listen_host or "127.0.0.1").strip()
+        if lh not in ("127.0.0.1", "localhost", "::1") and not allow_non_loopback:
+            raise PermissionError("SOCKS listen_host must be loopback unless admin allows")
+        listen_host = lh if lh != "localhost" else "127.0.0.1"
        pivot = SocksPivot(
            id=pid,
            session_id=session_id,
@@ -92,7 +102,11 @@ class SocksBroker:
        data_server = None
        data_port = 0
        if mode == "implant":
-            data_server = await asyncio.start_server(_data_handler, host="0.0.0.0", port=0)
+            # Bind data plane to loopback or public_host only (not 0.0.0.0)
+            data_bind = "127.0.0.1"
+            if self.public_host and self.public_host not in ("0.0.0.0", ""):
+                data_bind = self.public_host
+            data_server = await asyncio.start_server(_data_handler, host=data_bind, port=0)
            dsocks = list(data_server.sockets or [])
            data_port = int(dsocks[0].getsockname()[1]) if dsocks else 0
            pivot.data_server = data_server
@@ -214,12 +214,16 @@ class PolicyEngine:
        # Admins bypass HITL; client-asserted hitl_approved is IGNORED
        if require_hitl and "admin" not in auth.scopes:
            if await self._server_hitl_approved(auth, action, resource, extra):
+                rid = str((extra or {}).get("hitl_request_id") or "")
+                # H01: single-use grant
+                if rid:
+                    await self.db.consume_hitl_request(rid)
                return PolicyDecision(
                    True,
                    "allowed (HITL approved)",
                    risk,
                    require_hitl=True,
-                    hitl_request_id=str((extra or {}).get("hitl_request_id")),
+                    hitl_request_id=rid or None,
                )
            hid: str | None = None
            if create_hitl:
@@ -127,9 +127,21 @@ async def process_beacon_result(state: AppState, payload: dict[str, Any]) -> dic
    task_id = payload.get("task_id")
    if not task_id:
        raise ValueError("task_id required")
+    session_id = payload.get("session_id")
+    if not session_id:
+        # C07: bind result to session — look up task and require match when provided
+        t = await state.tasks.get(str(task_id))
+        if not t:
+            raise PermissionError("Unknown task")
+        session_id = t.get("session_id")
    result = payload.get("result")
    if result is None:
        result = ""
    status = payload.get("status") or "completed"
-    await state.tasks.complete(str(task_id), str(result), str(status))
+    try:
+        await state.tasks.complete(
+            str(task_id), str(result), str(status), session_id=str(session_id) if session_id else None
+        )
+    except KeyError as e:
+        raise PermissionError(str(e)) from e
    return _wrap_response(state, {"status": "ok"})  # type: ignore[return-value]
@@ -0,0 +1,74 @@
+"""SSRF guards for outbound server HTTP (LLM base_url, etc.)."""
+
+from __future__ import annotations
+
+import ipaddress
+import socket
+from urllib.parse import urlparse
+
+
+def validate_llm_base_url(url: str, *, allow_private: bool = False) -> str:
+    """Return normalized base URL or raise ValueError."""
+    u = (url or "").strip()
+    if not u:
+        raise ValueError("base_url required")
+    p = urlparse(u)
+    if p.scheme not in ("https", "http"):
+        raise ValueError("base_url scheme must be http or https")
+    if p.scheme == "http" and not allow_private:
+        # allow http only for explicit loopback lab
+        host = (p.hostname or "").lower()
+        if host not in ("127.0.0.1", "localhost", "::1"):
+            raise ValueError("base_url must use https (http only for localhost lab)")
+    host = p.hostname
+    if not host:
+        raise ValueError("base_url host required")
+    if not allow_private:
+        _assert_public_host(host)
+    # strip path noise; callers append /chat/completions
+    port = f":{p.port}" if p.port else ""
+    return f"{p.scheme}://{host}{port}"
+
+
+def _assert_public_host(host: str) -> None:
+    h = host.lower().strip("[]")
+    if h in ("localhost", "metadata.google.internal"):
+        raise ValueError("base_url host not allowed")
+    # block obvious metadata
+    if h.startswith("169.254.") or h == "metadata":
+        raise ValueError("base_url host not allowed")
+    try:
+        ip = ipaddress.ip_address(h)
+        if (
+            ip.is_private
+            or ip.is_loopback
+            or ip.is_link_local
+            or ip.is_reserved
+            or ip.is_multicast
+            or ip.is_unspecified
+        ):
+            raise ValueError("base_url must not target private/link-local addresses")
+        return
+    except ValueError as e:
+        if "must not" in str(e) or "not allowed" in str(e):
+            raise
+    # resolve DNS and check all A/AAAA
+    try:
+        infos = socket.getaddrinfo(h, None)
+    except socket.gaierror as e:
+        raise ValueError(f"base_url host unresolvable: {h}") from e
+    for info in infos:
+        addr = info[4][0]
+        try:
+            ip = ipaddress.ip_address(addr)
+        except ValueError:
+            continue
+        if (
+            ip.is_private
+            or ip.is_loopback
+            or ip.is_link_local
+            or ip.is_reserved
+            or ip.is_multicast
+            or ip.is_unspecified
+        ):
+            raise ValueError("base_url resolves to private/link-local address")
@@ -184,11 +184,22 @@ if __name__ == "__main__":
 '''


+def _safe_host(host: str) -> str:
+    """H13: only allow hostname/IP chars for PowerShell interpolation."""
+    import re
+
+    h = (host or "").strip()
+    if not re.fullmatch(r"[A-Za-z0-9._:-]{1,253}", h):
+        raise ValueError("invalid public_host for stage-2")
+    return h
+
+
 def windows_stage2_script(host: str, port: int) -> str:
    """PowerShell reconnecting line executor with SC5_PING support."""
+    safe = _safe_host(host)
    return f'''
 $ErrorActionPreference = 'SilentlyContinue'
-$h = '{host}'
+$h = '{safe}'
 $p = {int(port)}
 $delay = 3
 while ($true) {{
@@ -55,17 +55,16 @@ class TaskManager:
        if cmd == "profile:switch":
            if not args.get("profile_id"):
                raise ValueError("profile_id required for profile:switch")
-            # Engagement ROE: file writes may require HITL approval id on task args
-            if (
-                cmd == "file:write"
-                and eng is not None
-                and getattr(eng, "require_hitl_file_write", False)
-                and not args.get("hitl_request_id")
-                and created_by != "admin"
-            ):
-                # Callers with admin scope pass via API after HITL; non-approved blocked here
-                if not args.get("hitl_approved_server"):
-                    raise ValueError("file:write requires HITL approval under engagement policy")
+        # H08: Engagement ROE file-write HITL (was dead code nested under profile:switch)
+        if (
+            cmd == "file:write"
+            and eng is not None
+            and getattr(eng, "require_hitl_file_write", False)
+            and not args.get("hitl_request_id")
+            and created_by != "admin"
+        ):
+            if not args.get("hitl_approved_server"):
+                raise ValueError("file:write requires HITL approval under engagement policy")
        tid = await self.db.create_task(session_id, cmd, args, created_by)
        await self.metrics.incr("tasks.created")
        await self.metrics.emit(
@@ -86,8 +85,17 @@ class TaskManager:
        row = await self.db.next_pending_task(session_id)
        return self._norm(row) if row else None

-    async def complete(self, task_id: str, result: str, status: str = "completed") -> dict[str, Any]:
-        await self.db.complete_task(task_id, result, status)
+    async def complete(
+        self,
+        task_id: str,
+        result: str,
+        status: str = "completed",
+        *,
+        session_id: str | None = None,
+    ) -> dict[str, Any]:
+        ok = await self.db.complete_task(task_id, result, status, session_id=session_id)
+        if not ok:
+            raise KeyError("task not found, wrong session, or already finalized")
        await self.metrics.incr("tasks.completed")
        await self.metrics.emit("task.completed", {"id": task_id, "status": status})
        row = await self.db.get_task(task_id)
@@ -80,7 +80,8 @@ async def test_cors_null_origin_denied_without_public_host(client):


@pytest.mark.asyncio
-async def test_cors_null_origin_allowed_with_public_host(client_public_host):
+async def test_cors_null_origin_always_denied(client_public_host):
+    """M02: Origin null never reflected (even with public_host)."""
    r = await client_public_host.options(
        "/api/v1/health",
        headers={
@@ -89,12 +90,12 @@ async def test_cors_null_origin_allowed_with_public_host(client_public_host):
        },
    )
    assert r.status_code == 204
-    assert r.headers.get("access-control-allow-origin") == "null"
+    assert r.headers.get("access-control-allow-origin") is None


@pytest.mark.asyncio
-async def test_cors_public_host_origin(client_public_host, admin_headers):
-    # bootstrap token is same string; app_with_public_host has its own bootstrap
+async def test_cors_public_host_alone_not_enough(client_public_host, admin_headers):
+    """M03: public_host hostname match without Host/cors_origins is denied."""
    r = await client_public_host.get(
        "/api/v1/meta",
        headers={
@@ -104,4 +105,4 @@ async def test_cors_public_host_origin(client_public_host, admin_headers):
        },
    )
    assert r.status_code == 200
-    assert r.headers.get("access-control-allow-origin") == "https://c2.example.test"
+    assert r.headers.get("access-control-allow-origin") != "https://c2.example.test"
@@ -0,0 +1,291 @@
+"""Security hardening regression tests (Critical/High/Medium fixes)."""
+
+from __future__ import annotations
+
+import pytest
+from httpx import ASGITransport, AsyncClient
+
+from squidc5.config import Settings
+from squidc5.implants.crypto import open_envelope, seal
+from squidc5.listeners.implant_auth import unwrap_implant_payload
+from squidc5.main import create_app
+from squidc5.security.ssrf import validate_llm_base_url
+from squidc5.shells.stabilize import _safe_host, windows_stage2_script
+
+ADMIN = "sc5_test_admin_token_bootstrap_sechard01"
+PSK = "sec-hard-psk-unit-test-xyz"
+
+
+def _settings(tmp_path, **kw):
+    base = dict(
+        data_dir=tmp_path / "d",
+        debug=True,
+        mcp_enabled=True,
+        admin_token_bootstrap=ADMIN,
+        plugin_signing_secret="test-plugin-signing-secret-for-ci",
+        implant_require_auth=True,
+        implant_psk=PSK,
+        rate_limit_per_minute=5000,
+    )
+    base.update(kw)
+    return Settings(**base)
+
+
+def test_ssrf_blocks_metadata_and_private():
+    with pytest.raises(ValueError):
+        validate_llm_base_url("http://169.254.169.254/latest/meta-data/")
+    with pytest.raises(ValueError):
+        validate_llm_base_url("http://10.0.0.1/v1")
+    with pytest.raises(ValueError):
+        validate_llm_base_url("http://evil.example/v1")  # http non-local
+    ok = validate_llm_base_url("https://api.openai.com/v1")
+    assert ok.startswith("https://")
+
+
+def test_stage2_host_injection_blocked():
+    with pytest.raises(ValueError):
+        _safe_host("evil'; iex 'calc'")
+    with pytest.raises(ValueError):
+        windows_stage2_script("x';whoami", 443)
+    s = windows_stage2_script("c2.lab.example", 443)
+    assert "c2.lab.example" in s
+
+
+def test_implant_auth_rejects_plain():
+    with pytest.raises(PermissionError):
+        unwrap_implant_payload({"hostname": "x"}, psk=PSK, require_auth=True)
+    env = seal(PSK, {"hostname": "x"})
+    plain = unwrap_implant_payload(env, psk=PSK, require_auth=True)
+    assert plain["hostname"] == "x"
+
+
+def test_aead_roundtrip_with_aad():
+    env = seal(PSK, {"a": 1})
+    assert open_envelope(PSK, env)["a"] == 1
+
+
+@pytest.mark.asyncio
+async def test_token_cannot_mint_admin(tmp_path):
+    app = create_app(_settings(tmp_path))
+    async with app.router.lifespan_context(app):
+        transport = ASGITransport(app=app)
+        async with AsyncClient(transport=transport, base_url="http://test") as client:
+            h = {"Authorization": f"Bearer {ADMIN}"}
+            # admin creates tokens:manage only token
+            r = await client.post(
+                "/api/v1/tokens",
+                headers=h,
+                json={"name": "tm", "scopes": ["tokens:manage", "sessions:read"]},
+            )
+            assert r.status_code == 200, r.text
+            tm = r.json()["token"]
+            ht = {"Authorization": f"Bearer {tm}"}
+            # cannot mint admin
+            bad = await client.post(
+                "/api/v1/tokens",
+                headers=ht,
+                json={"name": "evil", "scopes": ["admin"]},
+            )
+            assert bad.status_code == 400
+            # cannot grant scopes it lacks
+            bad2 = await client.post(
+                "/api/v1/tokens",
+                headers=ht,
+                json={"name": "evil2", "scopes": ["shell:interact"]},
+            )
+            assert bad2.status_code == 400
+
+
+@pytest.mark.asyncio
+async def test_policy_put_admin_only(tmp_path):
+    app = create_app(_settings(tmp_path))
+    async with app.router.lifespan_context(app):
+        transport = ASGITransport(app=app)
+        async with AsyncClient(transport=transport, base_url="http://test") as client:
+            h = {"Authorization": f"Bearer {ADMIN}"}
+            r = await client.post(
+                "/api/v1/tokens",
+                headers=h,
+                json={"name": "pm", "scopes": ["policy:manage", "sessions:read"]},
+            )
+            tok = r.json()["token"]
+            hp = {"Authorization": f"Bearer {tok}"}
+            denied = await client.put(
+                "/api/v1/policy",
+                headers=hp,
+                json={"rules": {"require_hitl": []}},
+            )
+            assert denied.status_code == 403
+            ok = await client.put(
+                "/api/v1/policy",
+                headers=h,
+                json={"rules": {"require_hitl": ["shell.interact"]}},
+            )
+            assert ok.status_code == 200
+
+
+@pytest.mark.asyncio
+async def test_hitl_single_use(tmp_path):
+    app = create_app(_settings(tmp_path, mcp_enabled=False))
+    async with app.router.lifespan_context(app):
+        state = app.state.app_state
+        hid = await state.db.create_hitl_request(
+            action="shell.interact",
+            resource="r",
+            actor="op",
+            actor_type="operator",
+            details={},
+            binding_hash="x",
+            risk_score=8,
+        )
+        assert await state.db.resolve_hitl_request(hid, status="approved", resolved_by="admin")
+        assert await state.db.consume_hitl_request(hid) is True
+        assert await state.db.consume_hitl_request(hid) is False
+        row = await state.db.get_hitl_request(hid)
+        assert row["status"] == "consumed"
+
+
+@pytest.mark.asyncio
+async def test_task_complete_session_bound(tmp_path):
+    app = create_app(_settings(tmp_path))
+    async with app.router.lifespan_context(app):
+        state = app.state.app_state
+        sid = await state.sessions.register(kind="beacon", hostname="t")
+        t = await state.tasks.create(session_id=sid, command="id", created_by="admin")
+        tid = t["id"]
+        # wrong session fails
+        ok = await state.db.complete_task(tid, "out", session_id="ses_wrong")
+        assert ok is False
+        ok = await state.db.complete_task(tid, "out", session_id=sid)
+        assert ok is True
+        # already done
+        ok2 = await state.db.complete_task(tid, "out2", session_id=sid)
+        assert ok2 is False
+
+
+@pytest.mark.asyncio
+async def test_mcp_shell_requires_scope_and_claim(tmp_path):
+    app = create_app(_settings(tmp_path, mcp_enabled=True))
+    async with app.router.lifespan_context(app):
+        # enable feature
+        state = app.state.app_state
+        await state.features.set_many({"mcp_enabled": True}, actor="admin")
+        transport = ASGITransport(app=app)
+        async with AsyncClient(transport=transport, base_url="http://test") as client:
+            h = {"Authorization": f"Bearer {ADMIN}"}
+            # token with mcp but NO shell:interact
+            r = await client.post(
+                "/api/v1/tokens",
+                headers=h,
+                json={
+                    "name": "mcp-weak",
+                    "scopes": ["mcp:connect", "sessions:read"],
+                    "mcp_tools": ["interact_shell", "list_sessions"],
+                },
+            )
+            assert r.status_code == 200, r.text
+            tok = r.json()["token"]
+            hm = {"Authorization": f"Bearer {tok}"}
+            call = await client.post(
+                "/mcp/call",
+                headers=hm,
+                json={"name": "interact_shell", "arguments": {"session_id": "x", "command": "id"}},
+            )
+            assert call.status_code == 200
+            body = call.json()
+            assert body.get("ok") is False
+            assert "scope" in (body.get("error") or "").lower() or "Requires" in (body.get("error") or "")
+
+
+@pytest.mark.asyncio
+async def test_team_member_requires_lead(tmp_path):
+    app = create_app(_settings(tmp_path, mcp_enabled=False))
+    async with app.router.lifespan_context(app):
+        transport = ASGITransport(app=app)
+        async with AsyncClient(transport=transport, base_url="http://test") as client:
+            h = {"Authorization": f"Bearer {ADMIN}"}
+            t1 = await client.post(
+                "/api/v1/tokens",
+                headers=h,
+                json={"name": "alice", "scopes": ["collab:use", "sessions:read"]},
+            )
+            t2 = await client.post(
+                "/api/v1/tokens",
+                headers=h,
+                json={"name": "bob", "scopes": ["collab:use", "sessions:read"]},
+            )
+            alice = t1.json()["token"]
+            bob = t2.json()["token"]
+            ha = {"Authorization": f"Bearer {alice}"}
+            hb = {"Authorization": f"Bearer {bob}"}
+            team = await client.post("/api/v1/teams", headers=ha, json={"name": "cell"})
+            tid = team.json()["id"]
+            # bob cannot add members
+            denied = await client.post(
+                f"/api/v1/teams/{tid}/members",
+                headers=hb,
+                json={"actor": "eve", "role": "operator"},
+            )
+            assert denied.status_code == 403
+            # alice is lead (creator)
+            ok = await client.post(
+                f"/api/v1/teams/{tid}/members",
+                headers=ha,
+                json={"actor": "bob", "role": "operator"},
+            )
+            assert ok.status_code == 200
+
+
+@pytest.mark.asyncio
+async def test_llm_ssrf_on_configure(tmp_path):
+    app = create_app(_settings(tmp_path, mcp_enabled=False))
+    async with app.router.lifespan_context(app):
+        transport = ASGITransport(app=app)
+        async with AsyncClient(transport=transport, base_url="http://test") as client:
+            h = {"Authorization": f"Bearer {ADMIN}"}
+            bad = await client.post(
+                "/api/v1/llm",
+                headers=h,
+                json={
+                    "name": "evil",
+                    "model": "x",
+                    "base_url": "http://169.254.169.254/",
+                    "api_key": "k",
+                },
+            )
+            assert bad.status_code in (400, 500, 422) or bad.status_code >= 400
+
+
+@pytest.mark.asyncio
+async def test_ops_console_sets_role_flag(tmp_path):
+    app = create_app(_settings(tmp_path, mcp_enabled=False))
+    async with app.router.lifespan_context(app):
+        transport = ASGITransport(app=app)
+        async with AsyncClient(transport=transport, base_url="http://test") as client:
+            h = {"Authorization": f"Bearer {ADMIN}"}
+            r = await client.get("/api/v1/ops/console.js", headers=h)
+            assert r.status_code == 200
+            assert "__SC5_UI_ROLE__" in r.text
+            assert "ops-page-btn" in r.text or "Dashboard" in r.text
+            # non-admin
+            t = await client.post(
+                "/api/v1/tokens",
+                headers=h,
+                json={"name": "op", "scopes": ["sessions:read", "shell:interact"]},
+            )
+            ho = {"Authorization": f"Bearer {t.json()['token']}"}
+            r2 = await client.get("/api/v1/ops/console.js", headers=ho)
+            assert r2.status_code == 200
+            assert "operator" in r2.text
+
+
+@pytest.mark.asyncio
+async def test_socks_rejects_non_loopback(tmp_path):
+    app = create_app(_settings(tmp_path, mcp_enabled=False))
+    async with app.router.lifespan_context(app):
+        state = app.state.app_state
+        sid = await state.sessions.register(kind="beacon", hostname="s")
+        with pytest.raises(PermissionError):
+            await state.socks.start(sid, listen_host="0.0.0.0", mode="implant")
+        with pytest.raises(PermissionError):
+            await state.socks.start(sid, listen_host="127.0.0.1", mode="direct")
@@ -53,21 +53,36 @@

  const parts = [];

+  // Multi-page nav (competitor-style: Dashboard / Sessions / Listeners / Post-Ex / Collab / Admin)
+  const uiRole = window.__SC5_UI_ROLE__ || (can("admin") ? "admin" : "operator");
+  parts.push(`
+    <div id="opsNavBar" class="ops-nav" style="grid-column:1/-1;display:flex;flex-wrap:wrap;gap:6px;align-items:center;padding:8px 4px;margin-bottom:8px;border-bottom:1px solid rgba(255,255,255,0.08)">
+      <strong style="margin-right:8px">SquidC5</strong>
+      <button type="button" class="ops-page-btn primary" data-page="dashboard">Dashboard</button>
+      <button type="button" class="ops-page-btn" data-page="sessions">Sessions</button>
+      <button type="button" class="ops-page-btn" data-page="listeners">Listeners</button>
+      <button type="button" class="ops-page-btn" data-page="postex">Post-Ex</button>
+      <button type="button" class="ops-page-btn" data-page="collab">Collab</button>
+      ${can("admin") || uiRole === "admin" ? '<button type="button" class="ops-page-btn" data-page="admin">Admin</button>' : ""}
+      <span style="flex:1"></span>
+      <label for="layoutPreset" class="muted" style="font-size:0.75rem">Role layout</label>
+      <select id="layoutPreset" style="max-width:140px">
+        <option value="operator">Operator</option>
+        <option value="lead">Lead</option>
+        ${can("admin") ? '<option value="admin">Admin</option>' : ""}
+      </select>
+    </div>
+  `);
+
  // ----- Identity -----
  parts.push(panel("identityCard", "🪪 Identity", `
-    ${hint("Who you are on this C2: the API token currently saved in Connection. Pulled from <code>GET /api/v1/meta</code> — actor name, token id, and scopes the server granted. Scopes control which panels and API routes work; <code>admin</code> unlocks full console. Whoami dumps the raw meta JSON; Health is a lightweight liveness check (no secrets).", "identity")}
+    ${hint("Who you are on this C2. Scopes gate panels; admin unlocks Admin page.", "identity")}
    <p class="muted mono" id="whoLine">—</p>
    <div class="chips" id="scopeChips" style="margin-top:8px"></div>
    <div class="row" style="margin-top:8px">
      <button type="button" id="whoamiBtn">Whoami</button>
      <button type="button" id="healthBtn">Health</button>
    </div>
-    <label for="layoutPreset">Layout preset (U4)</label>
-    <select id="layoutPreset">
-      <option value="admin">Admin (all panels)</option>
-      <option value="operator">Operator</option>
-      <option value="lead">Lead / spectator</option>
-    </select>
    <div class="outbox empty-out" id="identOut" style="margin-top:8px">—</div>
  `, true, "identity"));

@@ -1561,28 +1576,64 @@
    };
  }

-  // ----- U4 Layout presets -----
+  // ----- Multi-page + role layouts (admin can switch) -----
+  const PAGE_PANELS = {
+    dashboard: ["identityCard", "workbenchPanel", "eventsRailPanel", "sessionsPanel", "obsPanel", "obsExtraPanel"],
+    sessions: ["workbenchPanel", "sessionsPanel", "quickRunCard", "tasksPanel", "eventsRailPanel"],
+    listeners: ["listenersPanel", "payloadsPanel", "profilesPanel", "deployPanel"],
+    postex: ["workbenchPanel", "filesPanel", "socksPanel", "modulesPanel", "pivotMapPanel", "pluginsPanel"],
+    collab: ["chatPanel", "teamsPanel", "hitlPanel", "engagementPanel", "auditMePanel"],
+    admin: ["tokensPanel", "featuresPanel", "policyPanel", "llmPanel", "mcpPanel", "aiCard", "deployPanel"],
+  };
  const PRESET_HIDE = {
-    operator: ["llmPanel", "tokensPanel", "featuresPanel", "policyPanel", "mcpPanel", "pluginsPanel", "deployPanel"],
-    lead: ["llmPanel", "tokensPanel", "featuresPanel", "policyPanel", "mcpPanel", "payloadsPanel", "modulesPanel"],
+    operator: ["llmPanel", "tokensPanel", "featuresPanel", "policyPanel", "mcpPanel", "pluginsPanel", "deployPanel", "aiCard"],
+    lead: ["llmPanel", "tokensPanel", "featuresPanel", "policyPanel", "mcpPanel", "payloadsPanel", "modulesPanel", "aiCard"],
    admin: [],
  };
-  function applyLayoutPreset(name) {
-    const hide = PRESET_HIDE[name] || [];
+  let _currentPage = "dashboard";
+  function applyPage(page) {
+    _currentPage = page || "dashboard";
+    const show = PAGE_PANELS[_currentPage] || PAGE_PANELS.dashboard;
+    const preset = ($("layoutPreset") && $("layoutPreset").value) || "operator";
+    const hideExtra = PRESET_HIDE[preset] || [];
+    // Non-admin cannot open admin page
+    if (_currentPage === "admin" && !can("admin")) {
+      _currentPage = "dashboard";
+    }
    document.querySelectorAll("details.panel").forEach((p) => {
      if (!p.id) return;
-      p.style.display = hide.indexOf(p.id) >= 0 ? "none" : "";
+      const onPage = show.indexOf(p.id) >= 0;
+      const roleHide = hideExtra.indexOf(p.id) >= 0;
+      p.style.display = onPage && !roleHide ? "" : "none";
+      if (onPage && !roleHide && isDesktopLayout()) p.open = true;
    });
-    try { localStorage.setItem("sc5_layout_preset", name); } catch (_) {}
-    showOk("Layout: " + name);
+    document.querySelectorAll(".ops-page-btn").forEach((b) => {
+      b.classList.toggle("primary", b.getAttribute("data-page") === _currentPage);
+    });
+    try {
+      localStorage.setItem("sc5_ops_page", _currentPage);
+      localStorage.setItem("sc5_layout_preset", preset);
+    } catch (_) {}
  }
+  function applyLayoutPreset(name) {
+    if ($("layoutPreset")) $("layoutPreset").value = name;
+    applyPage(_currentPage);
+    showOk("Layout: " + name + " / " + _currentPage);
+  }
+  document.querySelectorAll(".ops-page-btn").forEach((b) => {
+    b.onclick = () => applyPage(b.getAttribute("data-page"));
+  });
  if ($("layoutPreset")) {
    try {
      const saved = localStorage.getItem("sc5_layout_preset");
-      if (saved) $("layoutPreset").value = saved;
+      if (saved && (saved !== "admin" || can("admin"))) $("layoutPreset").value = saved;
+      else if (!can("admin")) $("layoutPreset").value = "operator";
+      else $("layoutPreset").value = "admin";
+      const sp = localStorage.getItem("sc5_ops_page");
+      if (sp) _currentPage = sp;
    } catch (_) {}
    $("layoutPreset").onchange = () => applyLayoutPreset($("layoutPreset").value);
-    applyLayoutPreset($("layoutPreset").value || "admin");
+    applyPage(_currentPage);
  }

  // ----- Teams / handoff / presence -----
@@ -677,6 +677,12 @@ var qrcode=function(){var t=function(t,r){var e=t,n=g[r],o=null,i=0,a=null,u=[],
      el.textContent = msg; el.classList.add("show");
      setTimeout(() => el.classList.remove("show"), 3500);
    }
+    function esc(s) {
+      return String(s == null ? "" : s)
+        .replace(/&/g, "&amp;").replace(/</g, "&lt;").replace(/>/g, "&gt;")
+        .replace(/"/g, "&quot;");
+    }
+
    function showOutput(text, meta) {
      const box = $("shellOut");
      if (!box) return;
@@ -1361,8 +1367,8 @@ var qrcode=function(){var t=function(t,r){var e=t,n=g[r],o=null,i=0,a=null,u=[],
      let html = "<table><tr><th>Session</th><th>Remote</th></tr>";
      for (const s of rows) {
        html += `<tr>
-          <td class="mono">${shortId(s.id)} <span class="pill ok">verified</span></td>
-          <td class="mono">${s.remote_addr || "—"}</td>
+          <td class="mono">${esc(shortId(s.id))} <span class="pill ok">verified</span></td>
+          <td class="mono">${esc(s.remote_addr || "—")}</td>
        </tr>`;
      }
      html += "</table>";
@@ -1379,10 +1385,10 @@ var qrcode=function(){var t=function(t,r){var e=t,n=g[r],o=null,i=0,a=null,u=[],
      for (const l of rows) {
        const st = l.status || "—";
        html += `<tr>
-          <td>${l.name || shortId(l.id)}</td>
-          <td class="mono">${l.port}</td>
-          <td>${l.kind}</td>
-          <td><span class="pill ${st === "running" ? "ok" : "bad"}">${st}</span></td>
+          <td>${esc(l.name || shortId(l.id))}</td>
+          <td class="mono">${esc(l.port)}</td>
+          <td>${esc(l.kind)}</td>
+          <td><span class="pill ${st === "running" ? "ok" : "bad"}">${esc(st)}</span></td>
        </tr>`;
      }
      html += "</table>";