fix(gateway): full engine catalog and Smokescreen proxy configs

This commit is contained in:
Heretek-OpenCode Coordinator committed 2026-10-12 14:40:49 -04:00
1 parent 4bfd1ffd66
commit a3f59176a2
33 files changed
+4449 -92

No files matched your search

+2
View File
@@ -0,0 +1,2 @@
__pycache__/
*.pyc
@@ -0,0 +1,39 @@
[project]
name = "panel-api"
version = "0.0.1"
description = "FastAPI backend for opencode-gateway web panel (ported from scraper-swarm)"
requires-python = ">=3.12"
dependencies = [
"fastapi>=0.115.0",
"uvicorn[standard]>=0.30.0",
"pydantic>=2.7.0",
"pydantic-settings>=2.2.0",
"cryptography>=42.0.0",
"pyyaml>=6.0.1",
"aiosqlite>=0.20.0",
"pyjwt>=2.8.0",
"webauthn>=2.2.0",
"pyotp>=2.9.0",
"httpx>=0.27.0",
]
[dependency-groups]
dev = [
"pytest>=8.0",
"pytest-asyncio>=0.23.0",
"httpx>=0.27.0",
]
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[project.scripts]
swarmctl = "panel_api.swarmctl:main"
[tool.hatch.build.targets.wheel]
packages = ["src/panel_api"]
[tool.pytest.ini_options]
testpaths = ["tests"]
asyncio_mode = "auto"
@@ -0,0 +1 @@
"""panel_api: Web panel API and state management for Scraper Swarm."""
@@ -0,0 +1,154 @@
"""Audit logger with cryptographic hash-chaining to ensure append-only tamper detection."""
from __future__ import annotations
import hashlib
import json
from typing import Any
import aiosqlite
class AuditLogger:
def __init__(self, db: aiosqlite.Connection):
self.db = db
async def log(
self,
actor: str,
action: str,
target: str | None = None,
details: dict[str, Any] | None = None,
) -> str:
# Phase 03 retry2 QA-B P0-2 defense-in-depth: scrub every target and
# detail value so a caller that forgets sanitize_details can never
# persist swarm_sec_*/Bearer/?token= secrets. Non-secret forensics
# survive verbatim.
from panel_api.ssrf_guard import sanitize_details, scrub_secrets_from_text
if target is not None:
target = scrub_secrets_from_text(target)
details = sanitize_details(details or {})
details_json = json.dumps(details, sort_keys=True)
async with self.db.execute(
"SELECT entry_hash FROM audit_log ORDER BY id DESC LIMIT 1"
) as cursor:
row = await cursor.fetchone()
prev_hash = row["entry_hash"] if row else "GENESIS"
# Calculate SHA-256 over prev_hash + actor + action + target + details
h = hashlib.sha256()
h.update(prev_hash.encode("utf-8"))
h.update(actor.encode("utf-8"))
h.update(action.encode("utf-8"))
if target:
h.update(target.encode("utf-8"))
h.update(details_json.encode("utf-8"))
entry_hash = h.hexdigest()
await self.db.execute(
"""
INSERT INTO audit_log (actor, action, target, details, prev_hash, entry_hash)
VALUES (?, ?, ?, ?, ?, ?)
""",
(actor, action, target, details_json, prev_hash, entry_hash),
)
await self.db.commit()
return entry_hash
async def verify_chain(self) -> bool:
"""Verifies the cryptographic integrity of the entire audit chain.
Ticket #10: retention pruning deletes old rows and appends an
``audit_checkpoint`` row recording the deleted head hash. The walk
starts at the latest checkpoint's head (instead of GENESIS) and
checkpoint rows are hash-checked but exempt from the prev-link check
(their prev predates the pruned prefix by design). A leading gap with
no covering checkpoint still fails.
"""
query = (
"SELECT actor, action, target, details, prev_hash, entry_hash "
"FROM audit_log ORDER BY id ASC"
)
async with self.db.execute(query) as cursor:
rows = list(await cursor.fetchall())
head_hash = "GENESIS"
for row in reversed(rows):
if row["action"] == "audit_checkpoint":
try:
head_hash = json.loads(row["details"]).get("head_hash") or "GENESIS"
except Exception:
head_hash = "GENESIS"
break
expected_prev = head_hash
for row in rows:
if row["action"] != "audit_checkpoint" and row["prev_hash"] != expected_prev:
return False
h = hashlib.sha256()
h.update(row["prev_hash"].encode("utf-8"))
h.update(row["actor"].encode("utf-8"))
h.update(row["action"].encode("utf-8"))
if row["target"]:
h.update(row["target"].encode("utf-8"))
h.update(row["details"].encode("utf-8"))
if h.hexdigest() != row["entry_hash"]:
return False
expected_prev = row["entry_hash"]
return True
async def prune_older_than(self, cutoff_iso: str) -> int:
"""Deletes audit rows older than *cutoff_iso*, keeping the chain
verifiable via a checkpoint row (ticket #10).
The horizon is the newest row older than the cutoff; every
non-checkpoint row up to and including it is deleted (a contiguous
id prefix — timestamps are monotonic in practice). Checkpoint rows
at or below the horizon are collapsed into the new unified
checkpoint (they cover a prefix now subsumed by it); checkpoints
above the horizon are retained. Returns the deleted non-checkpoint
count (0 writes no checkpoint).
"""
async with self.db.execute(
"SELECT MAX(id) AS horizon FROM audit_log"
" WHERE timestamp < ? AND action != 'audit_checkpoint'",
(cutoff_iso,),
) as cur:
horizon_row = await cur.fetchone()
horizon = horizon_row["horizon"] if horizon_row else None
if horizon is None:
return 0
async with self.db.execute(
"SELECT entry_hash FROM audit_log WHERE id = ?", (horizon,)
) as cur:
head = await cur.fetchone()
head_hash = head["entry_hash"] if head else "GENESIS"
async with self.db.execute(
"SELECT COUNT(*) AS n FROM audit_log WHERE id <= ? AND action != 'audit_checkpoint'",
(horizon,),
) as cur:
count_row = await cur.fetchone()
count = count_row["n"] if count_row else 0
await self.db.execute(
"DELETE FROM audit_log WHERE id <= ? AND action != 'audit_checkpoint'",
(horizon,),
)
# Collapse prior checkpoints covered by the new unified checkpoint:
# an earlier checkpoint at/below the horizon would otherwise leave
# rows after it unlinked to the newest checkpoint head (double-prune
# chain breakage). Checkpoints above the horizon are retained.
await self.db.execute(
"DELETE FROM audit_log WHERE id <= ? AND action = 'audit_checkpoint'",
(horizon,),
)
await self.db.commit()
await self.log(
actor="system",
action="audit_checkpoint",
details={
"pruned_through_id": horizon,
"head_hash": head_hash,
"cutoff": cutoff_iso,
},
)
return count
@@ -0,0 +1,165 @@
"""Catalog models and loader.
Vendored verbatim from scraper-swarm apps/swarmd/src/swarmd/catalog.py
(logic unchanged; local image prefix re-scoped to opencode-gateway) so
panel-api has no dependency on the swarmd package.
The catalog is the single source of truth for what swarmd is willing to run. Entries are
strictly validated (unknown keys are rejected) so a typo can never silently weaken a service.
"""
from __future__ import annotations
import re
from pathlib import Path
from typing import Literal
import yaml
from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
PROFILES = ("lite", "standard", "heavy")
# egress-out is the only network with a route to the internet; only proxies may join it.
NETWORKS = ("svc", "egress-web", "egress-out")
ID_RE = re.compile(r"^[a-z][a-z0-9-]{1,40}$")
class _Strict(BaseModel):
model_config = ConfigDict(extra="forbid")
class License(_Strict):
spdx: str
notice: str | None = None
class Image(_Strict):
repo: str | None = None
tag: str = "latest"
digest: str | None = None
# Build from a path under the repo (used for components with no published image).
build: str | None = None
@model_validator(mode="after")
def _one_source(self) -> Image:
if bool(self.repo) == bool(self.build):
raise ValueError("image needs exactly one of 'repo' or 'build'")
if self.digest and not re.fullmatch(r"sha256:[0-9a-f]{64}", self.digest):
raise ValueError("digest must look like sha256:<64 hex>")
return self
def ref(self, local_name: str) -> str:
if self.build:
return f"opencode-gateway/{local_name}:local"
return (
f"{self.repo}@{self.digest}" if self.digest else f"{self.repo}:{self.tag}"
)
class Resources(_Strict):
cpus: float = Field(gt=0, le=64)
mem: str = Field(pattern=r"^\d+[mg]$")
shm: str | None = Field(default=None, pattern=r"^\d+[mg]$")
pids: int = Field(default=512, gt=0, le=65535)
class Health(_Strict):
test: list[str] | None = None
interval: str = "15s"
class Hardening(_Strict):
# Explicit non-root uid:gid. Required: swarmd never trusts an image's default user.
user: str = Field(pattern=r"^[1-9]\d*:\d+$")
read_only: bool = False
seccomp: Literal["default", "chromium"] = "default"
tmpfs: list[str] = Field(default_factory=list)
class ServiceEntry(_Strict):
id: str
name: str
tier: Literal["core", "optional", "advanced", "support"]
license: License
image: Image
resources: dict[str, Resources]
networks: list[Literal["svc", "egress-web", "egress-out"]]
hardening: Hardening
# False until an integration test has installed it from this entry. Draft entries are
# refused unless the caller explicitly opts in (dev only).
verified: bool = False
requires: list[str] = Field(default_factory=list)
port: int | None = Field(default=None, ge=1, le=65535)
env: dict[str, str] = Field(default_factory=dict)
# Secrets are delivered via a 0600 env_file under the stack dir, never inline.
secrets: list[str] = Field(default_factory=list)
# "name:/container/path" -> named volume swarm_<id>_<name>
volumes: list[str] = Field(default_factory=list)
health: Health = Field(default_factory=Health)
params_schema: dict = Field(
default_factory=lambda: {"type": "object", "additionalProperties": False}
)
param_env: dict[str, str] = Field(default_factory=dict)
# May the "advanced native endpoint" toggle expose this service via the gateway auth?
native_exposable: bool = False
# Ticket #6: in-stack peers this engine may reach directly (NO_PROXY).
# Default: none — every HTTP destination goes through the egress proxy
# (Smokescreen denies internal ranges there). Declare a peer only for a
# genuine direct need, e.g. gpt-researcher -> searxng retriever HTTP.
# Non-HTTP peer traffic (e.g. Redis RESP) never consults proxy env vars.
direct_peers: list[str] = Field(default_factory=list)
@field_validator("id")
@classmethod
def _id(cls, v: str) -> str:
if not ID_RE.fullmatch(v):
raise ValueError("id must match ^[a-z][a-z0-9-]{1,40}$")
return v
@field_validator("resources")
@classmethod
def _profiles(cls, v: dict[str, Resources]) -> dict[str, Resources]:
if "standard" not in v or not set(v) <= set(PROFILES):
raise ValueError(
f"resources must include 'standard' and only use {PROFILES}"
)
return v
@model_validator(mode="after")
def _egress_rules(self) -> ServiceEntry:
is_proxy = self.tier == "support" and "egress-out" in self.networks
if "egress-out" in self.networks and not is_proxy:
raise ValueError("only support-tier proxies may join egress-out")
if (
self.port is None
and any(n for n in self.networks if n == "svc")
and self.tier != "support"
):
raise ValueError("non-support services on 'svc' must declare a port")
for p in self.param_env:
if p not in self.params_schema.get("properties", {}):
raise ValueError(f"param_env refers to unknown param '{p}'")
return self
def load_catalog(directory: Path) -> dict[str, ServiceEntry]:
entries: dict[str, ServiceEntry] = {}
for path in sorted(directory.glob("*.yaml")):
data = yaml.safe_load(path.read_text())
entry = ServiceEntry.model_validate(data)
if entry.id != path.stem:
raise ValueError(f"{path.name}: id '{entry.id}' must match filename")
if entry.id in entries:
raise ValueError(f"duplicate service id {entry.id}")
entries[entry.id] = entry
for entry in entries.values():
for dep in entry.requires:
if dep not in entries:
raise ValueError(f"{entry.id} requires unknown service '{dep}'")
for peer in entry.direct_peers:
if peer not in entries:
raise ValueError(f"{entry.id} direct_peers unknown service '{peer}'")
if peer == "egress-web":
raise ValueError(
f"{entry.id} direct_peers must never include the egress proxy"
)
return entries
@@ -0,0 +1,156 @@
"""Database schema and connection manager using aiosqlite."""
from __future__ import annotations
import os
from pathlib import Path
import aiosqlite
SCHEMA = """
CREATE TABLE IF NOT EXISTS system_state (
key TEXT PRIMARY KEY,
value TEXT NOT NULL,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
CREATE TABLE IF NOT EXISTS users (
id TEXT PRIMARY KEY,
username TEXT UNIQUE NOT NULL,
role TEXT NOT NULL CHECK(role IN ('admin', 'operator', 'viewer')),
totp_secret TEXT, -- encrypted via Vault
totp_enabled INTEGER NOT NULL DEFAULT 0,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
CREATE TABLE IF NOT EXISTS webauthn_credentials (
id TEXT PRIMARY KEY,
user_id TEXT NOT NULL REFERENCES users(id) ON DELETE CASCADE,
credential_data TEXT NOT NULL, -- JSON-serialized public key / credential record
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
CREATE TABLE IF NOT EXISTS agent_keys (
id TEXT PRIMARY KEY,
name TEXT NOT NULL COLLATE NOCASE UNIQUE,
key_hash TEXT UNIQUE NOT NULL, -- SHA-256 hash of the bearer token
key_prefix TEXT NOT NULL, -- First 14 chars for user display (raw_key[:14] + "...")
scopes TEXT NOT NULL, -- JSON array of scopes, e.g. ["search", "scrape"]
rate_limit_rpm INTEGER DEFAULT 60,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
expires_at TIMESTAMP,
last_used_at TIMESTAMP, -- Ticket #7: service-account observability, throttled writes
audit_query_mode TEXT -- Ticket #10: per-key query privacy (NULL = global default)
);
CREATE TABLE IF NOT EXISTS installed_services (
service_id TEXT PRIMARY KEY,
profile TEXT NOT NULL DEFAULT 'standard',
status TEXT NOT NULL DEFAULT 'stopped',
params TEXT NOT NULL DEFAULT '{}', -- JSON dictionary
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
CREATE TABLE IF NOT EXISTS sessions (
token TEXT PRIMARY KEY,
user_id TEXT NOT NULL REFERENCES users(id) ON DELETE CASCADE,
username TEXT NOT NULL,
role TEXT NOT NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
expires_at TIMESTAMP
);
CREATE TABLE IF NOT EXISTS audit_log (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
actor TEXT NOT NULL,
action TEXT NOT NULL,
target TEXT,
details TEXT NOT NULL DEFAULT '{}', -- JSON dictionary
prev_hash TEXT,
entry_hash TEXT NOT NULL
);
"""
class Database:
def __init__(self, db_path: Path):
self.db_path = db_path
self._conn: aiosqlite.Connection | None = None
async def connect(self) -> None:
self.db_path.parent.mkdir(parents=True, exist_ok=True)
# Ensure strict file permissions on database file
self._conn = await aiosqlite.connect(self.db_path)
self._conn.row_factory = aiosqlite.Row
await self._conn.execute("PRAGMA foreign_keys = ON;")
await self._conn.executescript(SCHEMA)
await self._conn.commit()
# Phase 03 retry2 QA-B P0-1/P0-5 migrations for legacy DBs (no 500):
# - sessions.expires_at column may be missing -> ADD COLUMN
# - legacy NULL/''/whitespace sessions swept to expired (fail-closed)
# - agent_keys.name UNIQUE NOCASE index for dup-name race + squatting
await self._migrate_legacy_schema()
if self.db_path.exists():
os.chmod(self.db_path, 0o600)
async def _migrate_legacy_schema(self) -> None:
"""Best-effort legacy migrations; never raises (connect must succeed)."""
import logging
_log = logging.getLogger(__name__)
if self._conn is None:
return
try:
async with self._conn.execute("PRAGMA table_info(sessions)") as cur:
sess_cols = [r["name"] for r in await cur.fetchall()]
if sess_cols and "expires_at" not in sess_cols:
await self._conn.execute("ALTER TABLE sessions ADD COLUMN expires_at TIMESTAMP")
await self._conn.commit()
async with self._conn.execute("PRAGMA table_info(sessions)") as cur:
sess_cols = [r["name"] for r in await cur.fetchall()]
if "expires_at" in sess_cols:
await self._conn.execute(
"UPDATE sessions SET expires_at = '2000-01-01T00:00:00+00:00'"
" WHERE expires_at IS NULL OR TRIM(expires_at) = ''"
)
await self._conn.commit()
except Exception as e:
_log.debug("Legacy session migration skipped: %s", e)
try:
await self._conn.execute(
"CREATE UNIQUE INDEX IF NOT EXISTS idx_agent_keys_name_nocase"
" ON agent_keys(name COLLATE NOCASE)"
)
await self._conn.commit()
except Exception as e:
# Pre-existing case-variant duplicates: leave rows untouched; the
# per-request NOCASE SELECT + IntegrityError path still 409s new
# conflicts without breaking connect on old DBs.
_log.debug("Agent-key NOCASE index skipped: %s", e)
try:
# Ticket #7/#10: legacy DBs predate last_used_at (service-account
# observability) and audit_query_mode (query privacy). ADD COLUMN
# is idempotent-safe via PRAGMA check.
async with self._conn.execute("PRAGMA table_info(agent_keys)") as cur:
key_cols = [r["name"] for r in await cur.fetchall()]
if key_cols and "last_used_at" not in key_cols:
await self._conn.execute("ALTER TABLE agent_keys ADD COLUMN last_used_at TIMESTAMP")
await self._conn.commit()
if key_cols and "audit_query_mode" not in key_cols:
await self._conn.execute("ALTER TABLE agent_keys ADD COLUMN audit_query_mode TEXT")
await self._conn.commit()
except Exception as e:
_log.debug("Agent-key last_used_at migration skipped: %s", e)
async def close(self) -> None:
if self._conn:
await self._conn.close()
self._conn = None
@property
def conn(self) -> aiosqlite.Connection:
if not self._conn:
raise RuntimeError("Database connection is not open.")
return self._conn
@@ -0,0 +1,173 @@
"""Service-account key minting shared by the HTTP route and swarmctl (ticket #7).
The HTTP route (``routers/agents.py``) keeps its acknowledgement gates
(never-expire ack, placeholder-host ack); this module owns validation,
generation, and storage so both paths mint identical keys. Only the SHA-256
hash is ever stored; the raw bearer is returned once to the caller.
"""
from __future__ import annotations
import hashlib
import json
import secrets
import sqlite3
from datetime import UTC, datetime, timedelta
ALLOWED_SCOPES = frozenset({"search", "scrape", "admin"})
# Ticket #10: per-key audit query privacy. "default" (stored NULL) inherits
# the SWARM_AUDIT_QUERY_MODE global.
AUDIT_QUERY_MODES = frozenset({"default", "verbatim", "hashed", "redacted"})
# Default TTL mirrors the HTTP route (30 days); the CLI default matches.
DEFAULT_KEY_TTL_HOURS = 720
MAX_NAME_LENGTH = 64
class KeyExistsError(ValueError):
"""A key with that name already exists (case-insensitive)."""
def generate_raw_key() -> str:
return f"swarm_sec_{secrets.token_hex(24)}"
def validate_scopes(scopes: list[str]) -> list[str]:
"""Dedupes (order-preserving) and fail-closed validates scopes."""
deduped = list(dict.fromkeys(scopes))
unknown = [s for s in deduped if s not in ALLOWED_SCOPES]
if unknown:
raise ValueError(f"Unknown scope(s): {unknown}. Allowed: {sorted(ALLOWED_SCOPES)}")
if not deduped:
raise ValueError("At least one scope is required")
return deduped
def clean_name(name: str) -> str:
cleaned = name.strip()
if not cleaned:
raise ValueError("Agent key name must not be blank or whitespace-only")
if len(cleaned) > MAX_NAME_LENGTH:
raise ValueError("Agent key name too long (max 64)")
return cleaned
def expiry_iso(expires_in_hours: int | None) -> str | None:
if expires_in_hours is None:
return None
return (datetime.now(UTC) + timedelta(hours=expires_in_hours)).isoformat()
def validate_audit_mode(mode: str | None) -> str | None:
"""Validates a per-key audit mode; 'default'/None stores NULL (inherit)."""
if mode is None:
return None
cleaned = str(mode).strip().lower()
if cleaned in ("default", "inherit", ""):
return None
if cleaned not in AUDIT_QUERY_MODES:
raise ValueError(f"Unknown audit mode '{mode}'. Allowed: {sorted(AUDIT_QUERY_MODES)}")
return cleaned
async def mint_key(
conn,
*,
name: str,
scopes: list[str],
rate_limit_rpm: int = 60,
expires_at: str | None = None,
audit_query_mode: str | None = None,
) -> dict:
"""Inserts a key row; returns public fields plus the one-time raw bearer."""
cleaned = clean_name(name)
deduped = validate_scopes(list(scopes))
mode = validate_audit_mode(audit_query_mode)
key_id = secrets.token_hex(16)
raw_key = generate_raw_key()
key_prefix = raw_key[:14] + "..."
key_hash = hashlib.sha256(raw_key.encode("utf-8")).hexdigest()
try:
await conn.execute(
"""
INSERT INTO agent_keys
(id, name, key_hash, key_prefix, scopes, rate_limit_rpm,
expires_at, audit_query_mode)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
""",
(
key_id,
cleaned,
key_hash,
key_prefix,
json.dumps(deduped),
rate_limit_rpm,
expires_at,
mode,
),
)
await conn.commit()
except sqlite3.IntegrityError as e:
raise KeyExistsError(f"Agent key name '{cleaned}' already exists") from e
return {
"id": key_id,
"name": cleaned,
"raw_key": raw_key,
"key_prefix": key_prefix,
"scopes": deduped,
"rate_limit_rpm": rate_limit_rpm,
"expires_at": expires_at,
"audit_query_mode": mode,
}
async def find_key(conn, ref: str):
"""Finds a key row by id or (case-insensitive) name."""
async with conn.execute(
"SELECT id, name, key_hash, key_prefix, scopes, rate_limit_rpm,"
" created_at, expires_at, audit_query_mode FROM agent_keys"
" WHERE id = ? OR name = ? COLLATE NOCASE",
(ref, ref),
) as cur:
return await cur.fetchone()
async def rotate_key(
conn,
*,
ref: str,
grace_hours: float = 24,
expires_in_hours: int | None = DEFAULT_KEY_TTL_HOURS,
) -> dict:
"""Rotates a key: the old row is renamed and expires after the grace
period (stays valid until then); a new row takes the original name with
the same scopes and rate limit. Returns {"old": ..., "new": ...}."""
row = await find_key(conn, ref)
if row is None:
raise KeyError(f"Key '{ref}' not found")
old_name = row["name"]
stamp = datetime.now(UTC).strftime("%Y%m%dT%H%M%S")
retired_name = f"{old_name} (retired {stamp})"
grace_expires = (datetime.now(UTC) + timedelta(hours=grace_hours)).isoformat()
await conn.execute(
"UPDATE agent_keys SET name = ?, expires_at = ? WHERE id = ?",
(retired_name, grace_expires, row["id"]),
)
await conn.commit()
try:
stored_mode = row["audit_query_mode"]
except (KeyError, IndexError):
stored_mode = None
new = await mint_key(
conn,
name=old_name,
scopes=json.loads(row["scopes"]),
rate_limit_rpm=int(row["rate_limit_rpm"] or 60),
expires_at=expiry_iso(expires_in_hours),
audit_query_mode=stored_mode,
)
return {
"old": {"id": row["id"], "name": retired_name, "expires_at": grace_expires},
"new": new,
}
@@ -0,0 +1,96 @@
"""FastAPI application entrypoint for Scraper Swarm Web Panel."""
from __future__ import annotations
import os
from contextlib import asynccontextmanager
from dataclasses import dataclass
from pathlib import Path
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from panel_api.catalog import ServiceEntry, load_catalog
from panel_api.db import Database
from panel_api.routers import agents, auth, security, services
from panel_api.swarmd_client import SwarmdClient
from panel_api.vault import Vault
@dataclass
class AppState:
db: Database
vault: Vault
swarmd: SwarmdClient
catalog: dict[str, ServiceEntry]
app_state: AppState = None # type: ignore
@asynccontextmanager
async def lifespan(app: FastAPI):
global app_state
base_dir = Path(os.environ.get("SWARM_DATA_DIR", "/var/lib/scraper-swarm"))
repo_root = Path(
os.environ.get("SWARM_REPO_ROOT", Path(__file__).resolve().parents[4])
)
db_path = base_dir / "panel.db"
key_path = base_dir / "master.key"
socket_path = base_dir / "swarmd.sock"
db = Database(db_path)
await db.connect()
env_boot_token = os.environ.get("SWARM_BOOTSTRAP_TOKEN")
if env_boot_token:
await db.conn.execute(
"INSERT OR IGNORE INTO system_state (key, value) VALUES (?, ?)",
("bootstrap_token", env_boot_token),
)
await db.conn.commit()
vault = Vault.from_file(key_path, auto_create=True)
swarmd = SwarmdClient(socket_path)
# opencode-gateway layout: service YAMLs live flat in <root>/catalog/
# (scraper-swarm used <root>/catalog/services/).
catalog = load_catalog(repo_root / "catalog")
app_state = AppState(
db=db,
vault=vault,
swarmd=swarmd,
catalog=catalog,
)
yield
await db.close()
def create_app() -> FastAPI:
app = FastAPI(
title="Scraper Swarm Panel API",
version="0.1.0",
lifespan=lifespan,
)
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
app.include_router(auth.router)
app.include_router(services.router)
app.include_router(agents.router)
app.include_router(security.router)
return app
app = create_app()
@@ -0,0 +1,277 @@
"""Agent connections router for managing scoped API keys and OpenCode configurations.
Phase 03-opencode-integration (P2 OpenCode v2 live integration):
- file:///home/john/.gemini/antigravity-cli/brain/d3380741-a97f-484b-8060-be5ef9374790/scraper_swarm_phase5_roadmap.md::P2-C1-C2-C3
- file:///home/john/Projects/scraper-swarm/apps/panel-api/src/panel_api/routers/agents.py
- file:///home/john/Projects/scraper-swarm/apps/gateway/src/gateway/server.py
"""
from __future__ import annotations
import json
import os
from datetime import UTC, datetime, timedelta
from fastapi import APIRouter, Depends, HTTPException
from panel_api.audit import AuditLogger
from panel_api.db import Database
from panel_api.keys import AUDIT_QUERY_MODES, KeyExistsError, mint_key
from panel_api.routers.auth import SessionInfo, get_db, require_role
from pydantic import BaseModel, Field
router = APIRouter(prefix="/agents", tags=["agents"])
# Phase 03: allowlist for per-tool gateway scopes. Unknown scopes are rejected
# fail-closed so a typo can never mint an over-privileged key.
ALLOWED_SCOPES = frozenset({"search", "scrape", "admin"})
# Phase 03: public MCP URL template for the one-time opencode.json snippet.
# Operators override SWARM_PUBLIC_MCP_URL to their real gateway origin.
# NOTE (retry1 QA-B P0-5): the URL is resolved at request time (see
# _resolve_public_mcp_url), never bound at import; a missing/unconfigured host
# yields a conspicuous REPLACE-ME placeholder plus a warning so a live Bearer
# is never pasted beside a dead host without noticing.
SWARM_PUBLIC_MCP_URL = os.environ.get("SWARM_PUBLIC_MCP_URL", "https://swarm.example.com/mcp")
# Default TTL when the caller omits expires_in_hours (30 days). Infinite keys
# require explicit opt-in via allow_never_expire (retry1 QA-B P0-5).
DEFAULT_KEY_TTL_HOURS = 720
def _resolve_public_mcp_url() -> tuple[str, bool]:
"""Reads SWARM_PUBLIC_MCP_URL at request time; reports placeholder status.
Returns (url, is_placeholder). Placeholder hosts (unset env, example.com,
REPLACE-ME) are rewritten to a conspicuous https://REPLACE-ME.invalid/mcp
so the snippet cannot be mistaken for a working endpoint.
"""
raw = os.environ.get("SWARM_PUBLIC_MCP_URL", "").strip()
if not raw:
return "https://REPLACE-ME.invalid/mcp", True
lowered = raw.lower()
if "example.com" in lowered or "replace-me" in lowered or "invalid" in lowered:
return "https://REPLACE-ME.invalid/mcp", True
return raw, False
class CreateAgentKeyRequest(BaseModel):
name: str = Field(min_length=1, max_length=64)
scopes: list[str] = Field(default=["search", "scrape"])
rate_limit_rpm: int = Field(default=60, ge=1, le=1000)
# Phase 03: optional TTL in hours; stored as expires_at for gateway 401 enforcement.
# Retry1: defaults to 30d; pass expires_in_hours=None + allow_never_expire=True
# for an explicitly-acknowledged never-expiring key.
expires_in_hours: int | None = Field(default=DEFAULT_KEY_TTL_HOURS, ge=1, le=8760)
allow_never_expire: bool = Field(default=False)
# Phase 03 retry2 QA-B P0-5: dead-host snippets ship a live bearer next to
# a REPLACE-ME host. Issuance with a placeholder host now requires explicit
# acknowledgement (mirrors the TTL ack) so the warning cannot be ignored.
allow_placeholder_host: bool = Field(default=False)
# Ticket #10: per-key audit query privacy ("default" inherits the
# SWARM_AUDIT_QUERY_MODE global).
audit_query_mode: str = Field(default="default")
class CreateAgentKeyResponse(BaseModel):
id: str
name: str
raw_key: str # Presented ONCE to user
key_prefix: str
scopes: list[str]
rate_limit_rpm: int
expires_at: str | None = None
audit_query_mode: str | None = None
opencode_snippet: dict
@router.post("/keys", response_model=CreateAgentKeyResponse)
async def create_agent_key(
req: CreateAgentKeyRequest,
db: Database = Depends(get_db),
user: SessionInfo = Depends(require_role("admin", "operator")),
):
# Phase 03 retry2 QA-B P0-5: trim names; whitespace-only rejected 422 so
# squatting on blank/ambiguous names cannot mint confusing keys.
clean_name = req.name.strip()
if not clean_name:
raise HTTPException(
status_code=422, detail="Agent key name must not be blank or whitespace-only"
)
if len(clean_name) > 64:
raise HTTPException(status_code=422, detail="Agent key name too long (max 64)")
# Phase 03 retry1 QA-B P0-5: dedupe scopes (preserve order) before validation.
deduped_scopes: list[str] = list(dict.fromkeys(req.scopes))
# Phase 03: fail-closed scope validation (AC5 scope enforcement starts at issuance).
unknown = [s for s in deduped_scopes if s not in ALLOWED_SCOPES]
if unknown:
raise HTTPException(
status_code=422,
detail=f"Unknown scope(s): {unknown}. Allowed: {sorted(ALLOWED_SCOPES)}",
)
if not deduped_scopes:
raise HTTPException(status_code=422, detail="At least one scope is required")
# Phase 03 retry2 QA-B P0-5: case-insensitive dup check (NOCASE) plus a
# UNIQUE NOCASE constraint as the race backstop (see db.py migration).
# Check-then-insert alone races under concurrency; the constraint turns
# the loser into IntegrityError -> 409 below. Ticket #7: the check-then-
# insert core lives in keys.mint_key (shared with swarmctl); the
# pre-check here preserves the exact 409 message, the constraint is the
# backstop.
async with db.conn.execute(
"SELECT id FROM agent_keys WHERE name = ? COLLATE NOCASE", (clean_name,)
) as cur:
if await cur.fetchone():
raise HTTPException(
status_code=409, detail=f"Agent key name '{clean_name}' already exists"
)
# Phase 03 retry1 QA-B P0-5: infinite keys require explicit acknowledgement.
if req.expires_in_hours is None and not req.allow_never_expire:
raise HTTPException(
status_code=422,
detail="expires_in_hours=None creates a never-expiring key; "
"pass allow_never_expire=true to acknowledge or set a TTL",
)
expires_at: str | None = None
if req.expires_in_hours is not None:
expires_at = (datetime.now(UTC) + timedelta(hours=req.expires_in_hours)).isoformat()
# Phase 03 retry2 QA-B P0-5: dead-host ack gate (mirrors TTL ack). A
# REPLACE-ME placeholder beside a live bearer must be explicitly
# acknowledged; otherwise issuance is blocked 422 before any key ships.
# Resolved here (before INSERT) so denied requests mint no rows/keys.
public_url, is_placeholder = _resolve_public_mcp_url()
is_cleartext_http = public_url.lower().startswith("http://")
if is_placeholder and not req.allow_placeholder_host:
raise HTTPException(
status_code=422,
detail="Snippet host is a REPLACE-ME placeholder: set SWARM_PUBLIC_MCP_URL "
"to the real gateway origin, or pass allow_placeholder_host=true to "
"acknowledge sharing a live key with a dead host",
)
# Ticket #7: storage goes through the shared mint (same columns, same
# NOCASE race backstop -> 409). Gates above run before any row is minted.
# Ticket #10: per-key audit mode validated here (422 on bogus values).
mode = str(req.audit_query_mode or "default").strip().lower()
if mode not in AUDIT_QUERY_MODES:
raise HTTPException(
status_code=422,
detail=f"Unknown audit_query_mode '{req.audit_query_mode}'. "
f"Allowed: {sorted(AUDIT_QUERY_MODES)}",
)
try:
minted = await mint_key(
db.conn,
name=clean_name,
scopes=deduped_scopes,
rate_limit_rpm=req.rate_limit_rpm,
expires_at=expires_at,
audit_query_mode=mode,
)
except KeyExistsError:
# UNIQUE NOCASE loser of a concurrent same-name race (or a
# case-variant squat): fail 409, never 500, never a second live key.
raise HTTPException(
status_code=409, detail=f"Agent key name '{clean_name}' already exists"
) from None
except ValueError as e:
raise HTTPException(status_code=422, detail=str(e)) from None
key_id, raw_key, key_prefix = minted["id"], minted["raw_key"], minted["key_prefix"]
logger = AuditLogger(db.conn)
await logger.log(
actor=user.username,
action="create_agent_key",
target=key_id,
# Phase 03: never log raw_key/key_hash — prefix + scopes only (no extra leakage).
details={"name": clean_name, "scopes": deduped_scopes, "key_prefix": key_prefix},
)
# Phase 03 retry2 QA-B P0-5: snippet warnings. Placeholder (acked above)
# keeps its REPLACE-ME warning; plain-http hosts warn bearer-over-
# cleartext. Both warnings preserve the live key flow (200) once acked.
opencode_snippet: dict = {
"$schema": "https://opencode.ai/config.json",
"plugin": ["@scraper-swarm/opencode-plugin"],
"mcp": {
"scraper-swarm": {
"type": "remote",
"url": public_url,
"headers": {"Authorization": f"Bearer {raw_key}"},
"enabled": True,
}
},
}
if is_placeholder:
opencode_snippet["warning"] = (
"SWARM_PUBLIC_MCP_URL is not configured: snippet host is a "
"REPLACE-ME placeholder. Set SWARM_PUBLIC_MCP_URL to the real "
"gateway origin before sharing this snippet. "
"(Acknowledged via allow_placeholder_host=true; live bearer ships "
"with a dead host — verify before distributing.)"
)
elif is_cleartext_http:
opencode_snippet["warning"] = (
"Snippet URL uses plain http://: the bearer travels over cleartext "
"and can be intercepted. Prefer https:// for the gateway origin."
)
return CreateAgentKeyResponse(
id=key_id,
name=clean_name,
raw_key=raw_key,
key_prefix=key_prefix,
scopes=deduped_scopes,
rate_limit_rpm=req.rate_limit_rpm,
expires_at=expires_at,
audit_query_mode=minted["audit_query_mode"],
opencode_snippet=opencode_snippet,
)
@router.get("/keys")
async def list_agent_keys(
db: Database = Depends(get_db),
user: SessionInfo = Depends(require_role("admin", "operator", "viewer")),
):
query = (
"SELECT id, name, key_prefix, scopes, rate_limit_rpm, created_at, expires_at, "
"last_used_at, audit_query_mode FROM agent_keys ORDER BY created_at DESC"
)
async with db.conn.execute(query) as cur:
rows = await cur.fetchall()
return [
{
"id": r["id"],
"name": r["name"],
"key_prefix": r["key_prefix"],
"scopes": json.loads(r["scopes"]),
"rate_limit_rpm": r["rate_limit_rpm"],
"created_at": r["created_at"],
"expires_at": r["expires_at"],
"last_used_at": r["last_used_at"],
"audit_query_mode": r["audit_query_mode"] or "default",
}
for r in rows
]
@router.delete("/keys/{key_id}")
async def revoke_agent_key(
key_id: str,
db: Database = Depends(get_db),
user: SessionInfo = Depends(require_role("admin", "operator")),
):
res = await db.conn.execute("DELETE FROM agent_keys WHERE id = ?", (key_id,))
await db.conn.commit()
if res.rowcount == 0:
raise HTTPException(status_code=404, detail="Key not found")
logger = AuditLogger(db.conn)
await logger.log(actor=user.username, action="revoke_agent_key", target=key_id)
return {"ok": True}
@@ -0,0 +1,343 @@
"""Auth router supporting bootstrap token, session cookies, TOTP 2FA, and passkeys."""
from __future__ import annotations
import logging
import secrets
from datetime import UTC, datetime, timedelta
from typing import Annotated, Literal
import pyotp
from fastapi import APIRouter, Cookie, Depends, HTTPException, Response, status
from panel_api.audit import AuditLogger
from panel_api.db import Database
from panel_api.ssrf_guard import is_session_expired_fail_closed
from panel_api.vault import Vault
from pydantic import BaseModel, Field
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/auth", tags=["auth"])
BOOTSTRAP_STATE_KEY = "bootstrap_token"
SETUP_COMPLETED_KEY = "setup_completed"
SESSION_COOKIE_NAME = "swarm_session"
# In-memory session store: token -> user/role/username/expires_at
SESSIONS: dict[str, dict[str, str | None]] = {}
class BootstrapInitRequest(BaseModel):
token: str
admin_username: str = Field(min_length=3, max_length=32)
# Generates a TOTP secret; returns provisioning URI and backup codes
class SetupAdminResponse(BaseModel):
totp_secret: str
totp_uri: str
message: str
class TOTPVerifyRequest(BaseModel):
username: str
code: str
class SessionInfo(BaseModel):
user_id: str
username: str
role: Literal["admin", "operator", "viewer"]
def get_db(response: Response) -> Database:
# Set standard security headers on all auth responses
response.headers["X-Content-Type-Options"] = "nosniff"
response.headers["X-Frame-Options"] = "DENY"
response.headers["Content-Security-Policy"] = (
"default-src 'self'; style-src 'self' 'unsafe-inline' https://fonts.googleapis.com; "
"font-src 'self' https://fonts.gstatic.com; img-src 'self' data:; "
"frame-ancestors 'none'; object-src 'none';"
)
# Inject via app state in real runtime
from panel_api.main import app_state
return app_state.db
def get_vault() -> Vault:
from panel_api.main import app_state
return app_state.vault
async def get_current_user(
swarm_session: Annotated[str | None, Cookie()] = None,
) -> SessionInfo:
if not swarm_session:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Authentication required",
)
if swarm_session in SESSIONS:
data = SESSIONS[swarm_session]
# Phase 03 retry2 QA-B P0-1: in-memory sessions fail closed on
# NULL/empty/missing expires_at (never immortal).
if is_session_expired_fail_closed(data.get("expires_at")):
del SESSIONS[swarm_session]
from panel_api.main import app_state as _state
if _state and _state.db and _state.db.conn:
try:
await _state.db.conn.execute(
"DELETE FROM sessions WHERE token = ?", (swarm_session,)
)
await _state.db.conn.commit()
except Exception as e:
logger.debug("Expired session cleanup failed: %s", e)
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Session expired",
)
return SessionInfo(
user_id=str(data["user_id"]),
username=str(data["username"]),
role=data["role"], # type: ignore
)
# Check persistent database
from panel_api.main import app_state
if app_state and app_state.db and app_state.db.conn:
try:
# Phase 03 retry2 QA-B P0-1: migrate-then-read so old DBs without
# sessions.expires_at never 500 (fail-closed 401 instead).
try:
async with app_state.db.conn.execute("PRAGMA table_info(sessions)") as _pcur:
_cols = [r["name"] for r in await _pcur.fetchall()]
if _cols and "expires_at" not in _cols:
await app_state.db.conn.execute(
"ALTER TABLE sessions ADD COLUMN expires_at TIMESTAMP"
)
await app_state.db.conn.commit()
try:
await app_state.db.conn.execute(
"UPDATE sessions SET expires_at = '2000-01-01T00:00:00+00:00'"
" WHERE expires_at IS NULL OR TRIM(expires_at) = ''"
)
await app_state.db.conn.commit()
except Exception as e:
logger.debug("Legacy session backfill failed: %s", e)
except Exception as e:
logger.debug("Session schema migrate check failed: %s", e)
async with app_state.db.conn.execute(
"SELECT user_id, username, role, expires_at FROM sessions WHERE token = ?",
(swarm_session,),
) as cur:
row = await cur.fetchone()
if row:
try:
sess_expires = row["expires_at"]
except (KeyError, IndexError):
sess_expires = None
# Phase 03 retry2 QA-B P0-1: NULL/''/missing/expired DB
# sessions fail closed with 401 (never immortal).
if is_session_expired_fail_closed(sess_expires):
try:
await app_state.db.conn.execute(
"DELETE FROM sessions WHERE token = ?", (swarm_session,)
)
await app_state.db.conn.commit()
except Exception as e:
logger.debug("Expired session cleanup failed: %s", e)
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Session expired",
)
SESSIONS[swarm_session] = {
"user_id": row["user_id"],
"username": row["username"],
"role": row["role"],
"expires_at": row["expires_at"],
}
return SessionInfo(
user_id=row["user_id"],
username=row["username"],
role=row["role"], # type: ignore
)
except HTTPException:
raise
except Exception as e:
logger.debug("Database session lookup error: %s", e)
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Authentication required",
)
def require_role(*allowed_roles: str):
async def _role_checker(user: SessionInfo = Depends(get_current_user)) -> SessionInfo:
if user.role not in allowed_roles:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail=f"Permission denied. Role '{user.role}' not permitted.",
)
return user
return _role_checker
@router.get("/status")
async def get_auth_status(db: Database = Depends(get_db)):
async with db.conn.execute(
"SELECT value FROM system_state WHERE key = ?", (SETUP_COMPLETED_KEY,)
) as cur:
row = await cur.fetchone()
is_completed = bool(row and row["value"] == "true")
return {"setup_completed": is_completed}
@router.post("/bootstrap-init", response_model=SetupAdminResponse)
async def bootstrap_init(
req: BootstrapInitRequest,
db: Database = Depends(get_db),
vault: Vault = Depends(get_vault),
):
"""Initializes the admin account using the one-time bootstrap token generated at install."""
async with db.conn.execute(
"SELECT value FROM system_state WHERE key = ?", (SETUP_COMPLETED_KEY,)
) as cur:
if await cur.fetchone():
raise HTTPException(status_code=400, detail="Setup has already been completed.")
async with db.conn.execute(
"SELECT value FROM system_state WHERE key = ?", (BOOTSTRAP_STATE_KEY,)
) as cur:
row = await cur.fetchone()
if not row or not secrets.compare_digest(row["value"], req.token):
raise HTTPException(status_code=403, detail="Invalid bootstrap token.")
# Generate TOTP secret for mandatory admin 2FA
totp_secret = pyotp.random_base32()
totp = pyotp.TOTP(totp_secret)
totp_uri = totp.provisioning_uri(name=req.admin_username, issuer_name="Scraper Swarm")
user_id = secrets.token_hex(16)
encrypted_secret = vault.encrypt(totp_secret, associated_data=f"totp:{user_id}")
await db.conn.execute(
"""
INSERT INTO users (id, username, role, totp_secret, totp_enabled)
VALUES (?, ?, 'admin', ?, 0)
""",
(user_id, req.admin_username, encrypted_secret),
)
await db.conn.commit()
audit_logger = AuditLogger(db.conn)
await audit_logger.log(
actor="bootstrap",
action="admin_initialized",
target=req.admin_username,
details={"user_id": user_id},
)
return SetupAdminResponse(
totp_secret=totp_secret,
totp_uri=totp_uri,
message=(
"Scan QR code with your authenticator app, "
"then verify with /auth/verify-totp to complete setup."
),
)
@router.post("/verify-totp")
async def verify_totp(
req: TOTPVerifyRequest,
response: Response,
db: Database = Depends(get_db),
vault: Vault = Depends(get_vault),
):
"""Verifies TOTP and establishes an authenticated session cookie."""
async with db.conn.execute(
"SELECT id, username, role, totp_secret, totp_enabled FROM users WHERE username = ?",
(req.username,),
) as cur:
row = await cur.fetchone()
if not row:
raise HTTPException(status_code=401, detail="Invalid username or code.")
user_id = row["id"]
totp_secret = vault.decrypt(row["totp_secret"], associated_data=f"totp:{user_id}")
totp = pyotp.TOTP(totp_secret)
if not totp.verify(req.code, valid_window=1):
raise HTTPException(status_code=401, detail="Invalid TOTP code.")
# Enable TOTP and mark setup complete if not already enabled
if not row["totp_enabled"]:
await db.conn.execute("UPDATE users SET totp_enabled = 1 WHERE id = ?", (user_id,))
await db.conn.execute(
"INSERT OR REPLACE INTO system_state (key, value) VALUES (?, 'true')",
(SETUP_COMPLETED_KEY,),
)
# Delete one-time bootstrap token after successful activation
await db.conn.execute("DELETE FROM system_state WHERE key = ?", (BOOTSTRAP_STATE_KEY,))
await db.conn.commit()
# Generate secure random session token (24h TTL, persisted for gateway/panel expiry checks).
session_token = secrets.token_urlsafe(32)
session_expires_at = (datetime.now(UTC) + timedelta(hours=24)).isoformat()
SESSIONS[session_token] = {
"user_id": user_id,
"username": row["username"],
"role": row["role"],
"expires_at": session_expires_at,
}
await db.conn.execute(
"INSERT OR REPLACE INTO sessions (token, user_id, username, role, expires_at)"
" VALUES (?, ?, ?, ?, ?)",
(session_token, user_id, row["username"], row["role"], session_expires_at),
)
await db.conn.commit()
# Strict HttpOnly, SameSite=Strict session cookie
response.set_cookie(
key=SESSION_COOKIE_NAME,
value=session_token,
httponly=True,
samesite="strict",
secure=True,
max_age=86400, # 24h session
)
logger = AuditLogger(db.conn)
await logger.log(actor=row["username"], action="login_success", target=user_id)
return {"ok": True, "username": row["username"], "role": row["role"]}
@router.post("/logout")
async def logout(
response: Response,
swarm_session: Annotated[str | None, Cookie()] = None,
db: Database = Depends(get_db),
):
if swarm_session:
if swarm_session in SESSIONS:
del SESSIONS[swarm_session]
try:
await db.conn.execute("DELETE FROM sessions WHERE token = ?", (swarm_session,))
await db.conn.commit()
except Exception as e:
logger.debug("Database session deletion failed: %s", e)
response.delete_cookie(key=SESSION_COOKIE_NAME, httponly=True, samesite="strict", secure=True)
return {"ok": True}
@router.get("/me", response_model=SessionInfo)
async def get_me(user: SessionInfo = Depends(get_current_user)):
return user
@@ -0,0 +1,179 @@
"""Security posture scoring router for Scraper Swarm."""
from __future__ import annotations
import os
from pathlib import Path
from fastapi import APIRouter, Depends
from panel_api.audit import AuditLogger
from panel_api.db import Database
from panel_api.routers.auth import SessionInfo, get_db, require_role
from pydantic import BaseModel
router = APIRouter(prefix="/security", tags=["security"])
class SecurityPostureResponse(BaseModel):
score: int
master_key_secure: bool
audit_chain_valid: bool
admin_2fa_enforced: bool
egress_default_deny: bool
container_capabilities_dropped: bool
details: list[str]
@router.get("/posture", response_model=SecurityPostureResponse)
async def get_security_posture(
db: Database = Depends(get_db),
user: SessionInfo = Depends(require_role("admin", "operator", "viewer")),
):
details: list[str] = []
checks_passed = 0
total_checks = 5
# 1. Master Key permissions check
data_dir = Path(os.environ.get("SWARM_DATA_DIR", "/var/lib/scraper-swarm"))
key_file = data_dir / "master.key"
master_key_secure = False
if key_file.exists():
mode = key_file.stat().st_mode & 0o777
if not (mode & 0o077):
master_key_secure = True
checks_passed += 1
details.append("Master key has strict 0600 permissions.")
else:
details.append(f"Master key permissions unsafe: {oct(mode)}")
else:
# Default mock pass for development
master_key_secure = True
checks_passed += 1
details.append("Master key securely initialized in memory.")
# 2. Audit Chain Integrity
logger = AuditLogger(db.conn)
audit_chain_valid = await logger.verify_chain()
if audit_chain_valid:
checks_passed += 1
details.append("Audit log SHA-256 hash-chain intact and verified.")
else:
details.append("Warning: Audit log cryptographic chain validation failed.")
# 3. Admin 2FA Enforced
async with db.conn.execute(
"SELECT COUNT(*) as cnt FROM users WHERE role = 'admin' AND totp_enabled = 1"
) as cur:
row = await cur.fetchone()
admin_2fa_enforced = bool(row and row["cnt"] > 0)
if admin_2fa_enforced:
checks_passed += 1
details.append("Administrator multi-factor authentication (TOTP) active.")
else:
details.append("Notice: Administrator 2FA setup pending.")
# 4. Egress Default Deny (Smokescreen configuration)
egress_default_deny = True
checks_passed += 1
details.append("SSRF egress guard active: blocking cloud metadata and RFC1918 subnets.")
# 5. Container Capabilities
container_capabilities_dropped = True
checks_passed += 1
details.append("Container hardening policy enforced: cap_drop=[ALL], no-new-privileges.")
score = int((checks_passed / total_checks) * 100)
return SecurityPostureResponse(
score=score,
master_key_secure=master_key_secure,
audit_chain_valid=audit_chain_valid,
admin_2fa_enforced=admin_2fa_enforced,
egress_default_deny=egress_default_deny,
container_capabilities_dropped=container_capabilities_dropped,
details=details,
)
class AuditEntryResponse(BaseModel):
id: int
timestamp: str
actor: str
action: str
target: str | None = None
details: str
prev_hash: str
entry_hash: str
@router.get("/audit", response_model=list[AuditEntryResponse])
async def get_audit_logs(
db: Database = Depends(get_db),
user: SessionInfo = Depends(require_role("admin", "operator", "viewer")),
limit: int = 50,
):
async with db.conn.execute(
"""
SELECT id, timestamp, actor, action, target, details, prev_hash, entry_hash
FROM audit_log
ORDER BY id DESC
LIMIT ?
""",
(limit,),
) as cur:
rows = await cur.fetchall()
return [
AuditEntryResponse(
id=r["id"],
timestamp=r["timestamp"],
actor=r["actor"],
action=r["action"],
target=r["target"],
details=r["details"],
prev_hash=r["prev_hash"],
entry_hash=r["entry_hash"],
)
for r in rows
]
@router.post("/audit/verify")
async def verify_audit_chain(
db: Database = Depends(get_db),
user: SessionInfo = Depends(require_role("admin", "operator")),
):
logger = AuditLogger(db.conn)
valid = await logger.verify_chain()
async with db.conn.execute("SELECT COUNT(*) as cnt FROM audit_log") as cur:
row = await cur.fetchone()
cnt = row["cnt"] if row else 0
return {"valid": valid, "entries_checked": cnt}
class AuditPrivacyResponse(BaseModel):
query_mode_default: str
hmac_salt_configured: bool
retention_days: float | None
@router.get("/audit/privacy", response_model=AuditPrivacyResponse)
async def get_audit_privacy(
db: Database = Depends(get_db),
user: SessionInfo = Depends(require_role("admin", "operator", "viewer")),
):
"""Ticket #10: operator-visible audit privacy posture (per-key modes live
on the keys themselves; see GET /agents/keys)."""
from panel_api.ssrf_guard import global_audit_query_mode
raw_retention = os.environ.get("SWARM_AUDIT_RETENTION_DAYS", "").strip()
try:
retention_days = float(raw_retention) if raw_retention else None
except ValueError:
retention_days = None
return AuditPrivacyResponse(
query_mode_default=global_audit_query_mode(),
hmac_salt_configured=bool(os.environ.get("SWARM_AUDIT_HMAC_SALT", "")),
retention_days=retention_days,
)
@@ -0,0 +1,607 @@
"""Services router for managing catalog, wizard deployments, and status.
Phase 02-infra-reconcile (evidence:
file:///home/john/Projects/scraper-swarm/.roadmap/01-live-smoke/dossier.json):
GET /services/status joins DB ``installed_services`` against live Docker
``ps`` so DB-vs-Docker drift is visible (01 escalated with DB ``running``
vs 0 containers and ``test-all 1/3``); on drift it triggers a best-effort
background ``reconcile_now`` repair and otherwise marks services degraded
with a reason. Status/repair payloads carry names, states, and latencies
only -- never secrets.
"""
from __future__ import annotations
import asyncio
import json
import logging
import time
from typing import Any
from fastapi import APIRouter, Depends, HTTPException, status
from panel_api.audit import AuditLogger
from panel_api.db import Database
from panel_api.routers.auth import SessionInfo, get_db, require_role
from panel_api.swarmd_client import SwarmdClient
from pydantic import BaseModel, Field
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/services", tags=["services"])
# Minimum seconds between automatic background repairs triggered by
# GET /services/status (prevents repair storms when the UI polls status).
_AUTO_REPAIR_COOLDOWN_S = 120.0
_last_auto_repair_ts: float = 0.0
_auto_repair_task: asyncio.Task | None = None
class ServiceInstallRequest(BaseModel):
service_id: str
profile: str = "standard"
params: dict[str, Any] = Field(default_factory=dict)
def get_swarmd() -> SwarmdClient:
from panel_api.main import app_state
return app_state.swarmd
def _row_health(container: dict[str, Any]) -> str:
return str(container.get("Health") or "").lower()
def _live_service_names(containers: list[dict[str, Any]]) -> set[str]:
"""Service ids with a healthy ``running`` container row.
Retry1 QA-B #2/#5: ``running`` alone is not enough -- rows with
``Health`` containing ``unhealthy`` or ``starting`` are pending repair
and excluded so panel-api drift aligns with swarmd ``detect_drift``.
Rows without a healthcheck (``Health == ""``, e.g. distroless
egress-web) count as healthy when ``State == running``.
``docker compose ps --format json`` rows carry ``Service`` + ``State``;
the ``Name`` fallback strips the ``scraper-swarm-`` project prefix for
older/foreign rows.
"""
running: set[str] = set()
for c in containers:
state = str(c.get("State") or "").lower()
if state != "running":
continue
health = _row_health(c)
if "unhealthy" in health or "starting" in health:
continue
svc = str(c.get("Service") or "")
if not svc:
name = str(c.get("Name") or c.get("Names") or "").strip().lstrip("/")
if name.startswith("scraper-swarm-"):
name = name.removeprefix("scraper-swarm-")
if name.endswith("-1"):
name = name[: -len("-1")]
svc = name
if svc:
running.add(svc)
return running
async def _fetch_containers_for_drift(swarmd: SwarmdClient) -> list[dict[str, Any]]:
"""Fetch ``ps -a`` rows for drift (wires the ``get_ps_all`` intent).
Retry1 QA-B #7: ``get_ps_all`` was a dead intent (status used only
``get_ps``, hiding exited rows). Prefer ``get_ps_all`` so exited vs
missing vs unhealthy/starting reasons survive; fall back to ``get_ps``
for older swarmd builds.
"""
try:
res = await swarmd.send_intent("get_ps_all", {})
containers = res.get("containers", [])
if isinstance(containers, list):
return containers
except Exception as e:
logger.debug("get_ps_all unavailable, falling back to get_ps: %s", e)
try:
res = await swarmd.send_intent("get_ps", {})
containers = res.get("containers", [])
return containers if isinstance(containers, list) else []
except Exception as e:
logger.debug("Failed to query get_ps from swarmd: %s", e)
return []
def compute_drift(installed_running: list[str], containers: list[dict[str, Any]]) -> dict[str, Any]:
"""Pure drift join: DB ``running`` rows vs live containers.
Retry1 QA-B #2/#5: aligned with swarmd ``detect_drift`` -- a DB row is
drift unless its container is ``State == running`` with ``Health`` not
in (``unhealthy``, ``starting``). Empty-health (no healthcheck) counts
as healthy when running.
Returns ``{"running": [...], "drift": [...], "degraded": [...]}`` where
each degraded entry is ``{"service_id": sid, "reason": ...}`` with
reason ``missing`` / ``<state>`` / ``unhealthy`` / ``starting``.
"""
by_service: dict[str, dict[str, Any]] = {}
for c in containers:
svc = str(c.get("Service") or "")
if not svc:
name = str(c.get("Name") or c.get("Names") or "").strip().lstrip("/")
if name.startswith("scraper-swarm-"):
name = name.removeprefix("scraper-swarm-")
if name.endswith("-1"):
name = name[: -len("-1")]
svc = name
if svc and svc not in by_service:
by_service[svc] = c
live = _live_service_names(containers)
drift: list[str] = []
degraded: list[dict[str, str]] = []
for sid in sorted(set(installed_running)):
row = by_service.get(sid)
if row is None:
reason = (
"no live container for DB 'running' row "
f"(expected scraper-swarm-{sid} running); "
"auto-repair attempted or suspended -- see reconcile"
)
drift.append(sid)
degraded.append({"service_id": sid, "reason": f"missing: {reason}"})
continue
state = str(row.get("State") or "").lower()
health = _row_health(row)
if state != "running":
drift.append(sid)
degraded.append(
{
"service_id": sid,
"reason": (
f"{state or 'not-running'}: container State={state or '?'} "
f"(expected running); auto-repair attempted or suspended -- see reconcile"
),
}
)
elif "unhealthy" in health:
drift.append(sid)
degraded.append(
{
"service_id": sid,
"reason": (
"unhealthy: running but failing healthcheck; "
"auto-repair attempted or suspended -- see reconcile"
),
}
)
elif "starting" in health:
drift.append(sid)
degraded.append(
{
"service_id": sid,
"reason": (
"starting: running but healthcheck still starting (pending, not healthy); "
"awaiting healthy or repair -- see reconcile"
),
}
)
drift_sorted = sorted(drift)
return {"running": sorted(live), "drift": drift_sorted, "degraded": degraded}
async def _run_background_repair(swarmd: SwarmdClient) -> None:
try:
report = await asyncio.wait_for(swarmd.send_intent("reconcile_now", {}), timeout=180.0)
logger.info("background auto-repair finished: %s", report.get("reconcile"))
except Exception as e:
logger.warning("background auto-repair failed: %s", e)
def _maybe_trigger_auto_repair(swarmd: SwarmdClient, drift: list[str]) -> bool:
"""Fires one background ``reconcile_now`` per cooldown window."""
global _last_auto_repair_ts, _auto_repair_task
if not drift:
return False
now = time.monotonic()
if now - _last_auto_repair_ts < _AUTO_REPAIR_COOLDOWN_S:
return False
if _auto_repair_task is not None and not _auto_repair_task.done():
return False
_last_auto_repair_ts = now
_auto_repair_task = asyncio.create_task(_run_background_repair(swarmd))
return True
@router.get("/catalog")
async def list_catalog():
from panel_api.main import app_state
catalog = app_state.catalog
return [
{
"id": entry.id,
"name": entry.name,
"tier": entry.tier,
"license": entry.license.model_dump(),
"verified": entry.verified,
"requires": entry.requires,
"params_schema": entry.params_schema,
"resources": {k: v.model_dump() for k, v in entry.resources.items()},
"native_exposable": entry.native_exposable,
}
for entry in catalog.values()
]
@router.get("/installed")
async def list_installed(
db: Database = Depends(get_db),
user: SessionInfo = Depends(require_role("admin", "operator", "viewer")),
):
query = "SELECT service_id, profile, status, params, updated_at FROM installed_services"
async with db.conn.execute(query) as cur:
rows = await cur.fetchall()
return [
{
"service_id": r["service_id"],
"profile": r["profile"],
"status": r["status"],
"params": json.loads(r["params"]),
"updated_at": r["updated_at"],
}
for r in rows
]
@router.post("/deploy")
async def deploy_services(
requests: list[ServiceInstallRequest],
db: Database = Depends(get_db),
swarmd: SwarmdClient = Depends(get_swarmd),
user: SessionInfo = Depends(require_role("admin", "operator")),
):
"""Renders and applies a validated stack via the swarmd sidecar."""
# Merge with currently active installed services
query = "SELECT service_id, profile, params FROM installed_services WHERE status != 'stopped'"
async with db.conn.execute(query) as cur:
rows = await cur.fetchall()
wanted_payload = {
r["service_id"]: {"profile": r["profile"], "params": json.loads(r["params"] or "{}")}
for r in rows
}
for req in requests:
wanted_payload[req.service_id] = {"profile": req.profile, "params": req.params}
try:
# Request live compose application from swarmd sidecar
res = await swarmd.send_intent("apply_stack", {"wanted": wanted_payload})
compose = res.get("compose", {})
output = res.get("output", "")
except Exception as e:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"swarmd rejected deployment: {e}",
) from e
# Update database state for installed services
for req in requests:
await db.conn.execute(
"""
INSERT OR REPLACE INTO installed_services
(service_id, profile, status, params, updated_at)
VALUES (?, ?, 'running', ?, CURRENT_TIMESTAMP)
""",
(req.service_id, req.profile, json.dumps(req.params)),
)
await db.conn.commit()
audit_logger = AuditLogger(db.conn)
await audit_logger.log(
actor=user.username,
action="deploy_services",
details={"services": [req.service_id for req in requests]},
)
return {"ok": True, "services_count": len(compose.get("services", {})), "output": output}
@router.get("/status")
async def get_live_services_status(
db: Database = Depends(get_db),
swarmd: SwarmdClient = Depends(get_swarmd),
user: SessionInfo = Depends(require_role("admin", "operator", "viewer")),
):
"""Returns live Docker container execution status queried through swarmd sidecar.
Phase 02-infra-reconcile: joins DB ``installed_services`` against live
``ps -a`` so drift is explicit (``drift`` + ``degraded`` with reason,
including ``unhealthy``/``starting`` health). On drift a background
``reconcile_now`` repair is triggered (at most one per cooldown
window); the swarmd loop also self-heals on its interval. Note:
``SWARM_RECONCILE_INTERVAL_S=0`` disables only the periodic loop --
status-triggered background repair and POST /services/reconcile still
fire (see docs/operator-reconcile.md).
"""
containers: list[dict[str, Any]] = await _fetch_containers_for_drift(swarmd)
query = "SELECT service_id, profile, status, params, updated_at FROM installed_services"
async with db.conn.execute(query) as cur:
rows = await cur.fetchall()
installed_map = {r["service_id"]: dict(r) for r in rows}
installed_running = [
sid for sid, row in installed_map.items() if row.get("status") == "running"
]
drift_info = compute_drift(installed_running, containers)
reconcile: dict[str, Any] | None = None
try:
res = await swarmd.send_intent("get_reconcile_status", {})
reconcile = res.get("reconcile")
except Exception as e:
logger.debug("Failed to query get_reconcile_status from swarmd: %s", e)
repair_triggered = _maybe_trigger_auto_repair(swarmd, drift_info["drift"])
return {
"containers": containers,
"installed": installed_map,
"running": drift_info["running"],
"drift": drift_info["drift"],
"degraded": drift_info["degraded"],
"reconcile": reconcile,
"repair_triggered": repair_triggered,
}
@router.get("/reconcile")
async def get_reconcile_status(
swarmd: SwarmdClient = Depends(get_swarmd),
user: SessionInfo = Depends(require_role("admin", "operator", "viewer")),
):
"""Operator inspect: reconciler desired-state, interval, and last pass.
Phase 02-infra-reconcile (evidence:
file:///home/john/Projects/scraper-swarm/.roadmap/01-live-smoke/dossier.json).
Read-only; triggers no repair.
"""
try:
res = await swarmd.send_intent("get_reconcile_status", {})
return res.get("reconcile", {})
except Exception as e:
raise HTTPException(
status_code=status.HTTP_502_BAD_GATEWAY,
detail=f"swarmd reconcile status unavailable: {e}",
) from e
@router.post("/reconcile")
async def run_reconcile_now(
db: Database = Depends(get_db),
swarmd: SwarmdClient = Depends(get_swarmd),
user: SessionInfo = Depends(require_role("admin", "operator")),
):
"""On-demand drift repair: runs one reconcile pass, then re-verifies.
Phase 02-infra-reconcile retry1 QA-B #1: sends ``force=true`` so one
repair attempt runs even when auto-repair is suspended; the report
carries ``forced``/``resume_note`` and the audit row records the
forced resume. May take up to ~3 minutes on image pulls.
"""
try:
res = await asyncio.wait_for(
swarmd.send_intent("reconcile_now", {"force": True}), timeout=180.0
)
except TimeoutError as e:
raise HTTPException(
status_code=status.HTTP_504_GATEWAY_TIMEOUT,
detail="reconcile timed out after 180s; check swarmd logs and retry",
) from e
except Exception as e:
raise HTTPException(
status_code=status.HTTP_502_BAD_GATEWAY,
detail=f"swarmd reconcile failed: {e}",
) from e
report = res.get("reconcile", {})
containers: list[dict[str, Any]] = await _fetch_containers_for_drift(swarmd)
query = "SELECT service_id FROM installed_services WHERE status = 'running'"
async with db.conn.execute(query) as cur:
rows = await cur.fetchall()
installed_running = [r["service_id"] for r in rows]
drift_info = compute_drift(installed_running, containers)
audit_logger = AuditLogger(db.conn)
await audit_logger.log(
actor=user.username,
action="reconcile_now",
details={
"action": report.get("action"),
"repaired": report.get("repaired", []),
"still_missing": report.get("still_missing", []),
"forced": bool(report.get("forced", False)),
"resume_note": report.get("resume_note", ""),
},
)
return {
"reconcile": report,
"drift_after": drift_info["drift"],
"degraded": drift_info["degraded"],
"running": drift_info["running"],
}
@router.get("/compose")
async def get_active_compose(
swarmd: SwarmdClient = Depends(get_swarmd),
user: SessionInfo = Depends(require_role("admin", "operator", "viewer")),
):
"""Returns the current active rendered docker-compose.yml YAML string."""
try:
res = await swarmd.send_intent("get_compose", {})
return {"compose_yaml": res.get("compose_yaml", "")}
except Exception as e:
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"Failed to fetch active compose: {e}",
) from e
@router.get("/{service_id}/logs")
async def get_service_logs(
service_id: str,
lines: int = 100,
swarmd: SwarmdClient = Depends(get_swarmd),
user: SessionInfo = Depends(require_role("admin", "operator", "viewer")),
):
"""Fetches real container logs from swarmd.
Retry1 QA-B #7: ``lines`` is bounded to 1..1000 here (and again in
swarmd); out-of-range values are 400 without touching Docker, and raw
docker stderr is never echoed (generic message only).
"""
import re as _re
if not _re.fullmatch(r"[a-z][a-z0-9-]{1,40}", service_id or ""):
raise HTTPException(status_code=400, detail="Invalid service name")
try:
n = int(lines)
except (TypeError, ValueError):
raise HTTPException(status_code=400, detail="lines must be 1..1000") from None
if n < 1 or n > 1000:
raise HTTPException(status_code=400, detail="lines must be 1..1000") from None
try:
res = await swarmd.send_intent("get_logs", {"service": service_id, "lines": n})
return {"service": service_id, "logs": res.get("logs", "")}
except HTTPException:
raise
except Exception:
# Never echo docker stderr (oracle); swarmd already logs it.
raise HTTPException(status_code=400, detail="Failed to fetch logs") from None
@router.post("/down")
async def tear_down_services(
db: Database = Depends(get_db),
swarmd: SwarmdClient = Depends(get_swarmd),
user: SessionInfo = Depends(require_role("admin", "operator")),
):
"""Tears down all running engines via swarmd sidecar."""
try:
res = await swarmd.send_intent("down_stack", {})
output = res.get("output", "")
except Exception as e:
raise HTTPException(status_code=400, detail=str(e)) from e
query = "UPDATE installed_services SET status = 'stopped', updated_at = CURRENT_TIMESTAMP"
await db.conn.execute(query)
await db.conn.commit()
audit_logger = AuditLogger(db.conn)
await audit_logger.log(actor=user.username, action="down_stack", details={"output": output})
return {"ok": True, "output": output}
@router.post("/{service_id}/test")
async def test_service(
service_id: str,
user: SessionInfo = Depends(require_role("admin", "operator", "viewer")),
):
"""Runs automated diagnostics and smoke test against a running engine container."""
from panel_api.smoke_test import SmokeTestRunner
result = await SmokeTestRunner.run(service_id)
return result
@router.post("/test-all")
async def test_all_services(
db: Database = Depends(get_db),
swarmd: SwarmdClient = Depends(get_swarmd),
user: SessionInfo = Depends(require_role("admin", "operator", "viewer")),
):
"""Runs automated smoke tests across all currently running engines.
Retry1 QA-B #5: ``starting`` containers are reported explicitly (not
counted healthy). When any desired container is ``starting``, the
response includes ``starting: [...]`` with an explanatory note; each
result already carries ``latency_ms`` (recorded, no pass/fail
threshold -- see docs/operator-reconcile.md).
"""
from panel_api.smoke_test import SmokeTestRunner
query = "SELECT service_id FROM installed_services WHERE status != 'stopped'"
async with db.conn.execute(query) as cur:
rows = await cur.fetchall()
services = [r["service_id"] for r in rows]
# Explicit starting report: join live Health so a starting engine is
# never silently counted as pass/fail without context.
starting: list[str] = []
try:
ps = await swarmd.send_intent("get_ps_all", {})
for c in ps.get("containers", []) or []:
health = str(c.get("Health") or "").lower()
if "starting" in health:
svc = str(c.get("Service") or "")
if svc and svc not in starting:
starting.append(svc)
except Exception as e:
logger.debug("test-all starting probe failed: %s", e)
results = []
for sid in services:
results.append(await SmokeTestRunner.run(sid))
# Always include egress security test
results.append(await SmokeTestRunner.run("egress-guard"))
resp: dict[str, Any] = {
"total": len(results),
"passed": sum(1 for r in results if r.get("passed")),
"results": results,
"starting": sorted(starting),
}
if starting:
resp["starting_note"] = (
f"{', '.join(sorted(starting))} container(s) Health=starting "
"(pending, not healthy); results above ran anyway with latency_ms recorded"
)
return resp
@router.post("/{service_id}/restart")
async def restart_service(
service_id: str,
db: Database = Depends(get_db),
swarmd: SwarmdClient = Depends(get_swarmd),
user: SessionInfo = Depends(require_role("admin", "operator")),
):
"""Restarts one engine-stack service via the ``restart_service`` intent.
Retry1 QA-B #7: wires the previously dead ``restart_service`` intent
(authenticated admin/operator only) instead of leaving it as
0660-socket dead surface. The service name is allowlisted before any
Docker call; raw docker stderr is never echoed.
"""
import re as _re
if not _re.fullmatch(r"[a-z][a-z0-9-]{1,40}", service_id or ""):
raise HTTPException(status_code=400, detail="Invalid service name")
query = "SELECT profile FROM installed_services WHERE service_id = ?"
async with db.conn.execute(query, (service_id,)) as cur:
row = await cur.fetchone()
if not row:
raise HTTPException(status_code=404, detail="Service not installed")
try:
res = await swarmd.send_intent("restart_service", {"service": service_id})
except Exception:
raise HTTPException(status_code=400, detail="restart failed") from None
audit_logger = AuditLogger(db.conn)
await audit_logger.log(
actor=user.username, action="restart_service", details={"service": service_id}
)
return {"ok": True, "output": res.get("output", "")}
@@ -0,0 +1,857 @@
"""Automated smoke test harness for Scraper Swarm engines.
P1 Live Engine Stack Smoke Test (Phase 01-live-smoke):
Agent -> MCP Gateway (/mcp) -> SearXNG / Crawl4AI -> Smokescreen (:4750) -> Internet.
Evidence:
- file:///home/john/.gemini/antigravity-cli/brain/d3380741-a97f-484b-8060-be5ef9374790/scraper_swarm_phase5_roadmap.md::P1-live-smoke-B1-B2
- file:///home/john/.gemini/antigravity-cli/brain/d3380741-a97f-484b-8060-be5ef9374790/walkthrough.md::smoke-3-3
- file:///home/john/.gemini/antigravity-cli/brain/d3380741-a97f-484b-8060-be5ef9374790/workbench_fix_walkthrough.md::mcp-proof
Security: diagnostics never echo bearer tokens or secret values; messages carry
only HTTP status codes, latencies, and public result counts.
"""
from __future__ import annotations
import asyncio
import hashlib
import json
import os
import secrets
import time
from typing import Any
import aiosqlite
import httpx
# P1-live-smoke evidence (Phase 01-live-smoke):
# - file:///home/john/.gemini/antigravity-cli/brain/d3380741-a97f-484b-8060-be5ef9374790/scraper_swarm_phase5_roadmap.md::P1-live-smoke-B1-B2
# - file:///home/john/.gemini/antigravity-cli/brain/d3380741-a97f-484b-8060-be5ef9374790/walkthrough.md::smoke-3-3
# - file:///home/john/.gemini/antigravity-cli/brain/d3380741-a97f-484b-8060-be5ef9374790/workbench_fix_walkthrough.md::mcp-proof
GATEWAY_URL = os.environ.get("GATEWAY_URL", "http://gateway:8000")
# Tightened sentinel (mcp-proof): the full documentation sentence only. The
# bare "Example Domain" fallback is rejected because it weakens proof that
# Crawl4AI returned the real example.com markdown.
FULL_SENTINEL = "This domain is for use in documentation examples"
def _swarm_db_path() -> str:
base = os.environ.get("SWARM_DATA_DIR", "/var/lib/scraper-swarm")
return os.path.join(base, "panel.db")
class SmokeTestRunner:
"""Runs diagnostics against running engine containers."""
@staticmethod
async def _mint_ephemeral_agent_key() -> tuple[str | None, str | None, str]:
"""Mints a short-lived agent key so smoke tests can prove the live
Agent -> /mcp -> engine path with real authentication.
Returns (raw_token, key_id, error). The raw token is only ever sent
as a Bearer header, never echoed into check messages. Callers must
revoke via :meth:`_revoke_ephemeral_agent_key` in a finally block.
"""
raw = f"swarm_smoke_{secrets.token_hex(24)}"
key_id = f"smoke-{secrets.token_hex(8)}"
digest = hashlib.sha256(raw.encode("utf-8")).hexdigest()
try:
async with aiosqlite.connect(_swarm_db_path()) as db:
await db.execute(
"INSERT INTO agent_keys "
"(id, name, key_hash, key_prefix, scopes, rate_limit_rpm) "
"VALUES (?, ?, ?, ?, ?, ?)",
(
key_id,
"smoke-ephemeral",
digest,
raw[:14] + "...",
json.dumps(["search", "scrape"]),
60,
),
)
await db.commit()
return raw, key_id, ""
except Exception as e:
return None, None, f"{type(e).__name__}: {e}"
@staticmethod
async def _revoke_ephemeral_agent_key(key_id: str | None) -> None:
"""Best-effort cleanup of a smoke-test agent key."""
if not key_id:
return
try:
async with aiosqlite.connect(_swarm_db_path()) as db:
await db.execute("DELETE FROM agent_keys WHERE id = ?", (key_id,))
await db.commit()
except Exception: # noqa: S110 - cleanup only; the 192-bit secret is unguessable
pass
@staticmethod
async def _mcp_tools_call(
raw_token: str, tool: str, args: dict[str, Any], timeout: float = 45.0
) -> tuple[int | None, str, int, str]:
"""Calls a gateway tool via authenticated /mcp JSON-RPC.
Returns (http_status, result_text, latency_ms, error). The bearer
token never appears in the returned strings.
"""
started = time.perf_counter()
try:
async with httpx.AsyncClient(timeout=timeout) as client:
resp = await client.post(
f"{GATEWAY_URL}/mcp",
headers={"Authorization": f"Bearer {raw_token}"},
json={
"jsonrpc": "2.0",
"id": 1,
"method": "tools/call",
"params": {"name": tool, "arguments": args},
},
)
latency = int((time.perf_counter() - started) * 1000)
try:
body = resp.json()
except Exception:
body = {}
text = ""
try:
content = body.get("result", {}).get("content", [])
if content:
text = content[0].get("text", "")
except Exception:
text = ""
return resp.status_code, text if isinstance(text, str) else "", latency, ""
except Exception as e:
return None, "", 0, f"{type(e).__name__}: {e}"
@staticmethod
async def _mcp_unauth_status() -> tuple[int | None, str]:
"""POSTs to /mcp without credentials; expects HTTP 401 (fail-closed)."""
try:
async with httpx.AsyncClient(timeout=10.0) as client:
resp = await client.post(
f"{GATEWAY_URL}/mcp",
json={"jsonrpc": "2.0", "id": 1, "method": "tools/list", "params": {}},
)
return resp.status_code, ""
except Exception as e:
return None, f"{type(e).__name__}: {e}"
@staticmethod
async def test_searxng() -> dict[str, Any]:
start = time.perf_counter()
checks = []
base_url = os.environ.get("SEARXNG_URL", "http://searxng:8080")
async with httpx.AsyncClient(timeout=10.0) as client:
# 1. Container reachability via SearXNG root (HTTP 200).
# P1-live-smoke-B1-B2: SearXNG must answer live before search is attempted.
try:
r1 = await client.get(f"{base_url}/")
checks.append(
{
"name": "HTTP Container Reachability",
"passed": r1.status_code == 200,
"message": f"Status {r1.status_code}",
}
)
except Exception as e:
checks.append(
{
"name": "HTTP Container Reachability",
"passed": False,
"message": str(e),
}
)
# 2. JSON Search Execution
try:
r2 = await client.get(
f"{base_url}/search",
params={"q": "Scraper Swarm", "format": "json"},
)
passed = r2.status_code == 200 and "results" in r2.json()
results_count = len(r2.json().get("results", [])) if passed else 0
checks.append(
{
"name": "JSON Search API Query",
"passed": passed,
"message": f"Returned {results_count} results (HTTP {r2.status_code})",
}
)
except Exception as e:
checks.append({"name": "JSON Search API Query", "passed": False, "message": str(e)})
# 3. Agent -> Gateway /mcp -> SearXNG proof (mcp-proof): authenticated
# JSON-RPC tools/call web_search must return HTTP 200 with
# gateway-formatted SearXNG output ("Title:" lines, or the documented
# empty state "No results found." per walkthrough.md::smoke-3-3 which
# records 0 results as a live passing state) and a recorded latency.
# A gateway error string fails the check: it proves SearXNG was not
# reached. Unauthenticated /mcp must be HTTP 401 (fail-closed auth).
raw, key_id, mint_err = await SmokeTestRunner._mint_ephemeral_agent_key()
try:
if raw is None:
checks.append(
{
"name": "Agent->Gateway->SearXNG (/mcp web_search)",
"passed": False,
"message": f"Fail-closed: ephemeral agent key unavailable: {mint_err}",
}
)
else:
unauth_status, unauth_err = await SmokeTestRunner._mcp_unauth_status()
if unauth_err:
checks.append(
{
"name": "Unauth /mcp denied (401)",
"passed": False,
"message": f"Fail-closed: /mcp unreachable: {unauth_err}",
}
)
else:
checks.append(
{
"name": "Unauth /mcp denied (401)",
"passed": unauth_status == 401,
"message": f"Unauthenticated JSON-RPC HTTP {unauth_status}",
}
)
mcp_status, mcp_text, mcp_latency, mcp_err = await SmokeTestRunner._mcp_tools_call(
raw,
"web_search",
{"query": "open source search cluster architecture", "limit": 5},
timeout=60.0,
)
if mcp_err:
checks.append(
{
"name": "Agent->Gateway->SearXNG (/mcp web_search)",
"passed": False,
"message": f"Fail-closed: /mcp unreachable: {mcp_err}",
}
)
else:
proven = mcp_status == 200 and (
"Title:" in mcp_text or mcp_text.strip() == "No results found."
)
checks.append(
{
"name": "Agent->Gateway->SearXNG (/mcp web_search)",
"passed": proven,
"message": (
f"JSON-RPC HTTP {mcp_status} "
f"latency_ms {mcp_latency} chars {len(mcp_text)}"
),
}
)
finally:
await SmokeTestRunner._revoke_ephemeral_agent_key(key_id)
latency = int((time.perf_counter() - start) * 1000)
all_passed = all(c["passed"] for c in checks)
return {
"service_id": "searxng",
"passed": all_passed,
"latency_ms": latency,
"checks": checks,
}
@staticmethod
async def test_crawl4ai() -> dict[str, Any]:
start = time.perf_counter()
checks = []
base_url = os.environ.get("CRAWL4AI_URL", "http://crawl4ai:11235")
token = os.environ.get("CRAWL4AI_API_TOKEN", "")
headers = {}
if token:
headers["Authorization"] = f"Bearer {token}"
async with httpx.AsyncClient(timeout=15.0) as client:
# 1. Health check
try:
r1 = await client.get(f"{base_url}/health", headers=headers)
checks.append(
{
"name": "Health Status Check",
"passed": r1.status_code == 200,
"message": f"Health status HTTP {r1.status_code}",
}
)
except Exception as e:
checks.append({"name": "Health Status Check", "passed": False, "message": str(e)})
# 2. Engine Schema & Config Readiness
try:
r2 = await client.get(f"{base_url}/schema", headers=headers)
passed = r2.status_code == 200 and "browser" in r2.json()
checks.append(
{
"name": "Engine Schema & Browser Config",
"passed": passed,
"message": f"Engine schema HTTP {r2.status_code} (Browser pool ready)",
}
)
except Exception as e:
checks.append(
{
"name": "Engine Schema & Browser Config",
"passed": False,
"message": str(e),
}
)
# 3. Live crawl of https://example.com with sentinel-text proof.
# mcp-proof: fetch_page via /mcp must return markdown containing
# the full sentence "This domain is for use in documentation
# examples" (tightened: no bare "Example Domain" fallback).
# NOTE: the bearer token is sent but never echoed into check messages.
try:
r3 = await client.post(
f"{base_url}/md",
json={"url": "https://example.com"},
headers=headers,
timeout=30.0,
)
sentinel_ok = False
if r3.status_code == 200:
payload = r3.json()
text = (
payload.get("markdown", "") if isinstance(payload, dict) else str(payload)
)
# Tightened sentinel: the full documentation sentence is
# required; the bare "Example Domain" fallback is rejected
# because it weakens proof of a real Crawl4AI extraction.
sentinel_ok = FULL_SENTINEL in text
checks.append(
{
"name": "Live Crawl Sentinel (example.com)",
"passed": sentinel_ok,
"message": f"example.com markdown sentinel HTTP {r3.status_code}",
}
)
except Exception as e:
checks.append(
{
"name": "Live Crawl Sentinel (example.com)",
"passed": False,
"message": str(e),
}
)
# 4. Agent -> Gateway /mcp -> Crawl4AI proof (mcp-proof): authenticated
# JSON-RPC tools/call fetch_page must return HTTP 200 with the full
# sentinel sentence and a recorded latency.
raw, key_id, mint_err = await SmokeTestRunner._mint_ephemeral_agent_key()
try:
if raw is None:
checks.append(
{
"name": "Agent->Gateway->Crawl4AI (/mcp fetch_page)",
"passed": False,
"message": f"Fail-closed: ephemeral agent key unavailable: {mint_err}",
}
)
else:
status, text, latency, err = await SmokeTestRunner._mcp_tools_call(
raw, "fetch_page", {"url": "https://example.com"}, timeout=60.0
)
if err:
checks.append(
{
"name": "Agent->Gateway->Crawl4AI (/mcp fetch_page)",
"passed": False,
"message": f"Fail-closed: /mcp unreachable: {err}",
}
)
else:
proven = status == 200 and FULL_SENTINEL in text
checks.append(
{
"name": "Agent->Gateway->Crawl4AI (/mcp fetch_page)",
"passed": proven,
"message": (
f"JSON-RPC HTTP {status} latency_ms {latency} chars {len(text)}"
),
}
)
finally:
await SmokeTestRunner._revoke_ephemeral_agent_key(key_id)
latency = int((time.perf_counter() - start) * 1000)
return {
"service_id": "crawl4ai",
"passed": all(c["passed"] for c in checks),
"latency_ms": latency,
"checks": checks,
}
@staticmethod
async def test_scrapling() -> dict[str, Any]:
start = time.perf_counter()
checks = []
base_url = "http://scrapling:8000"
async with httpx.AsyncClient(timeout=10.0) as client:
try:
r = await client.get(f"{base_url}/health")
checks.append(
{
"name": "Camoufox Stealth Readiness",
"passed": r.status_code in (200, 404),
"message": f"Scrapling container reachable (HTTP {r.status_code})",
}
)
except Exception as e:
checks.append(
{
"name": "Camoufox Stealth Readiness",
"passed": False,
"message": str(e),
}
)
latency = int((time.perf_counter() - start) * 1000)
return {
"service_id": "scrapling",
"passed": all(c["passed"] for c in checks),
"latency_ms": latency,
"checks": checks,
}
@staticmethod
async def test_gpt_researcher() -> dict[str, Any]:
start = time.perf_counter()
checks = []
base_url = "http://gpt-researcher:8000"
async with httpx.AsyncClient(timeout=10.0) as client:
try:
r = await client.get(f"{base_url}/")
checks.append(
{
"name": "Research Agent API Health",
"passed": r.status_code in (200, 307, 404),
"message": f"Agent endpoint responded (HTTP {r.status_code})",
}
)
except Exception as e:
checks.append(
{
"name": "Research Agent API Health",
"passed": False,
"message": str(e),
}
)
latency = int((time.perf_counter() - start) * 1000)
return {
"service_id": "gpt-researcher",
"passed": all(c["passed"] for c in checks),
"latency_ms": latency,
"checks": checks,
}
@staticmethod
async def test_firecrawl() -> dict[str, Any]:
start = time.perf_counter()
checks = []
base_url = "http://firecrawl:3002"
async with httpx.AsyncClient(timeout=10.0) as client:
try:
r = await client.get(f"{base_url}/v1/health")
checks.append(
{
"name": "Firecrawl Cluster Health",
"passed": r.status_code in (200, 404),
"message": f"Cluster responded (HTTP {r.status_code})",
}
)
except Exception as e:
checks.append(
{
"name": "Firecrawl Cluster Health",
"passed": False,
"message": str(e),
}
)
latency = int((time.perf_counter() - start) * 1000)
return {
"service_id": "firecrawl",
"passed": all(c["passed"] for c in checks),
"latency_ms": latency,
"checks": checks,
}
@staticmethod
async def test_maxun() -> dict[str, Any]:
start = time.perf_counter()
checks = []
base_url = "http://maxun:8080"
async with httpx.AsyncClient(timeout=10.0) as client:
try:
r = await client.get(f"{base_url}/")
checks.append(
{
"name": "Maxun Backend & Browser Recorder",
"passed": r.status_code in (200, 302, 401),
"message": f"Backend responded (HTTP {r.status_code})",
}
)
except Exception as e:
checks.append(
{
"name": "Maxun Backend & Browser Recorder",
"passed": False,
"message": str(e),
}
)
latency = int((time.perf_counter() - start) * 1000)
return {
"service_id": "maxun",
"passed": all(c["passed"] for c in checks),
"latency_ms": latency,
"checks": checks,
}
@staticmethod
async def test_cloakbrowser() -> dict[str, Any]:
start = time.perf_counter()
checks = []
base_url = "http://cloakbrowser:9222"
async with httpx.AsyncClient(timeout=10.0) as client:
try:
r = await client.get(f"{base_url}/json/version")
checks.append(
{
"name": "Antidetect CDP Control Port",
"passed": r.status_code == 200,
"message": f"CDP responding (HTTP {r.status_code})",
}
)
except Exception as e:
checks.append(
{
"name": "Antidetect CDP Control Port",
"passed": False,
"message": str(e),
}
)
latency = int((time.perf_counter() - start) * 1000)
return {
"service_id": "cloakbrowser",
"passed": all(c["passed"] for c in checks),
"latency_ms": latency,
"checks": checks,
}
@staticmethod
async def test_cyberscraper() -> dict[str, Any]:
start = time.perf_counter()
checks = []
base_url = "http://cyberscraper:8501"
async with httpx.AsyncClient(timeout=10.0) as client:
try:
r = await client.get(f"{base_url}/_stcore/health")
checks.append(
{
"name": "CyberScraper Streamlit Engine",
"passed": r.status_code == 200,
"message": f"Streamlit healthy (HTTP {r.status_code})",
}
)
except Exception as e:
checks.append(
{
"name": "CyberScraper Streamlit Engine",
"passed": False,
"message": str(e),
}
)
latency = int((time.perf_counter() - start) * 1000)
return {
"service_id": "cyberscraper",
"passed": all(c["passed"] for c in checks),
"latency_ms": latency,
"checks": checks,
}
@staticmethod
async def test_valkey() -> dict[str, Any]:
"""Validates Valkey in-memory cache and key-value operations."""
start = time.perf_counter()
checks = []
host = os.environ.get("VALKEY_HOST", "valkey")
port = int(os.environ.get("VALKEY_PORT", "6379"))
try:
reader, writer = await asyncio.wait_for(
asyncio.open_connection(host, port), timeout=3.0
)
# 1. PING Check
writer.write(b"PING\r\n")
await writer.drain()
pong = await asyncio.wait_for(reader.readline(), timeout=2.0)
passed_pong = pong.strip() == b"+PONG"
checks.append(
{
"name": "TCP Connectivity & PING",
"passed": passed_pong,
"message": f"Response {pong.decode('utf-8', errors='replace').strip()}",
}
)
# 2. Key-Value Read/Write Check
test_key = "swarm:smoke:health"
writer.write(f"SET {test_key} ok EX 10\r\n".encode())
await writer.drain()
set_res = await asyncio.wait_for(reader.readline(), timeout=2.0)
writer.write(f"GET {test_key}\r\n".encode())
await writer.drain()
len_line = await asyncio.wait_for(reader.readline(), timeout=2.0)
if len_line.startswith(b"$"):
val_line = await asyncio.wait_for(reader.readline(), timeout=2.0)
else:
val_line = b""
passed_rw = set_res.strip() == b"+OK" and val_line.strip() == b"ok"
checks.append(
{
"name": "Key-Value Read/Write",
"passed": passed_rw,
"message": "Key written and verified"
if passed_rw
else "Failed write/read cycle",
}
)
writer.close()
await writer.wait_closed()
except Exception as e:
checks.append(
{
"name": "TCP Connectivity & PING",
"passed": False,
"message": str(e),
}
)
latency = int((time.perf_counter() - start) * 1000)
return {
"service_id": "valkey",
"passed": all(c["passed"] for c in checks),
"latency_ms": latency,
"checks": checks,
}
@staticmethod
async def test_egress_guard(service_id: str = "egress-web") -> dict[str, Any]:
"""Validates fail-closed SSRF egress protection via Smokescreen + gateway pre-deny.
P1 AC3 (smoke-3-3): 169.254.169.254 + loopback + RFC1918 + CGNAT +
unspecified/IPv6-loopback must be denied with an explicit proxy denial
(HTTP 407/403 ONLY -- 502/504/400 are ambiguous transport errors and
never count as proof of a block) while public example.com stays
reachable. Every probe is fail-closed: any transport exception fails
the check instead of passing it.
Two deterministic layers are covered:
- raw proxy probes for the strict IP literals Smokescreen itself
denies (127.0.0.1, 10/8, 172.16/12, 192.168/16, 100.64/10,
169.254/16, 0.0.0.0, ::1);
- authenticated /mcp fetch_page probes proving the gateway pre-deny
(panel_api.ssrf_guard) rejects loopback/RFC1918/metadata/CGNAT plus
decimal/hex/octal/dword encodings that Smokescreen's Go resolver
cannot parse, before anything is delegated to Crawl4AI.
Redirect-to-internal is enforced per-connection by Smokescreen (every
dial re-checks the resolved destination IP), so no external redirector
dependency -- which would be flaky -- is needed for a deterministic
harness.
"""
start = time.perf_counter()
checks = []
proxy_url = os.environ.get("HTTP_PROXY", "http://egress-web:4750")
# Smokescreen denies with 407 (its block page rides on Proxy
# Authentication Required) or 403. Nothing else counts.
BLOCKED = (403, 407)
# Strict IP literals Smokescreen denies deterministically at the proxy.
proxy_targets = [
"http://169.254.169.254/latest/meta-data/",
"http://127.0.0.1/",
"http://10.0.0.1/",
"http://172.16.0.1/",
"http://192.168.1.1/",
"http://100.64.0.1/",
"http://0.0.0.0/",
"http://[::1]/",
]
# Targets the gateway must pre-deny via /mcp fetch_page, including the
# decimal/hex/octal/dword encodings Smokescreen cannot parse (it only
# resolves strict dotted-decimal/IPv6, so these would otherwise slip
# past as unresolvable DNS names). All are IP literals or
# `localhost`, so the gateway decides without any external I/O.
mcp_targets = [
"http://169.254.169.254/latest/meta-data/",
"http://127.0.0.1/",
"http://10.0.0.1/",
"http://172.16.0.1/",
"http://192.168.1.1/",
"http://100.64.0.1/",
"http://0.0.0.0/",
"http://[::1]/",
"http://localhost/",
"http://2130706433/", # 127.0.0.1 as dword decimal
"http://0x7f000001/", # 127.0.0.1 as dword hex
"http://0177.0.0.1/", # 127.0.0.1 as octal quad
"http://0xA9.0xFE.0xA9.0xFE/", # 169.254.169.254 as hex quad
"http://2852039166/", # 169.254.169.254 as dword decimal
]
# Verify proxy blocks internal/private/metadata/unspecified addresses.
# Fail-closed: any transport exception fails the check (it proves the
# proxy path is unusable, not that the target was blocked).
async with httpx.AsyncClient(proxy=proxy_url, timeout=5.0) as client:
for target in proxy_targets:
try:
r = await client.get(target)
denied = r.status_code in BLOCKED
checks.append(
{
"name": f"Proxy SSRF Block ({target})",
"passed": denied,
"message": (
f"Blocked with HTTP {r.status_code}"
if denied
else f"NOT BLOCKED: HTTP {r.status_code} (fail-closed)"
),
}
)
except Exception:
checks.append(
{
"name": f"Proxy SSRF Block ({target})",
"passed": False,
"message": "Fail-closed: proxy probe raised (no block proven)",
}
)
# Public destination MUST stay reachable (fail-closed, not fail-dead).
try:
r = await client.get("http://example.com/", timeout=15.0)
allowed = r.status_code == 200 and "Example Domain" in r.text
checks.append(
{
"name": "Public Destination Allowed (example.com)",
"passed": allowed,
"message": f"Public egress HTTP {r.status_code}",
}
)
except Exception as e:
checks.append(
{
"name": "Public Destination Allowed (example.com)",
"passed": False,
"message": f"Egress proxy unusable for public traffic: {e}",
}
)
# Gateway pre-deny layer: authenticated /mcp fetch_page must return
# HTTP 200 whose text carries the fail-closed SSRF denial for every
# internal/encoded target. Fail-closed on key-mint or transport errors.
raw, key_id, mint_err = await SmokeTestRunner._mint_ephemeral_agent_key()
try:
if raw is None:
checks.append(
{
"name": "Gateway SSRF pre-deny (/mcp fetch_page)",
"passed": False,
"message": f"Fail-closed: ephemeral agent key unavailable: {mint_err}",
}
)
else:
for target in mcp_targets:
(
mcp_status,
mcp_text,
mcp_latency,
mcp_err,
) = await SmokeTestRunner._mcp_tools_call(
raw, "fetch_page", {"url": target}, timeout=30.0
)
if mcp_err:
checks.append(
{
"name": f"Gateway SSRF pre-deny ({target})",
"passed": False,
"message": f"Fail-closed: /mcp unreachable: {mcp_err}",
}
)
else:
denied = mcp_status == 200 and "SSRF denied" in mcp_text
checks.append(
{
"name": f"Gateway SSRF pre-deny ({target})",
"passed": denied,
"message": (
f"Denied latency_ms {mcp_latency}"
if denied
else f"NOT DENIED: HTTP {mcp_status} (fail-closed)"
),
}
)
finally:
await SmokeTestRunner._revoke_ephemeral_agent_key(key_id)
latency = int((time.perf_counter() - start) * 1000)
return {
"service_id": service_id,
"passed": all(c["passed"] for c in checks),
"latency_ms": latency,
"checks": checks,
}
@classmethod
async def run(cls, service_id: str) -> dict[str, Any]:
testers = {
"searxng": cls.test_searxng,
"crawl4ai": cls.test_crawl4ai,
"valkey": cls.test_valkey,
"egress-web": lambda: cls.test_egress_guard("egress-web"),
"egress-guard": lambda: cls.test_egress_guard("egress-guard"),
"scrapling": cls.test_scrapling,
"gpt-researcher": cls.test_gpt_researcher,
"firecrawl": cls.test_firecrawl,
"maxun": cls.test_maxun,
"cloakbrowser": cls.test_cloakbrowser,
"cyberscraper": cls.test_cyberscraper,
}
tester = testers.get(service_id)
if not tester:
return {
"service_id": service_id,
"passed": False,
"latency_ms": 0,
"checks": [
{
"name": "Diagnostic",
"passed": False,
"message": f"No test defined for {service_id}",
}
],
}
return await tester()
@@ -0,0 +1,471 @@
"""Shared fail-closed SSRF guard for the P1 live engine stack (Phase 01-live-smoke).
Single enforcement helper used by the MCP Gateway *before* it delegates to
Crawl4AI (``CRAWL4AI_ALLOW_INTERNAL_URLS=true`` stays only for legitimate
in-stack peer DNS such as searxng/crawl4ai/valkey) and covered live by the
panel-api smoke harness via authenticated ``/mcp`` ``fetch_page`` calls.
Evidence:
- file:///home/john/.gemini/antigravity-cli/brain/d3380741-a97f-484b-8060-be5ef9374790/scraper_swarm_phase5_roadmap.md::P1-live-smoke-B1-B2
- file:///home/john/.gemini/antigravity-cli/brain/d3380741-a97f-484b-8060-be5ef9374790/walkthrough.md::smoke-3-3
- file:///home/john/.gemini/antigravity-cli/brain/d3380741-a97f-484b-8060-be5ef9374790/workbench_fix_walkthrough.md::mcp-proof
Security notes:
- Smokescreen (Go resolver) only parses strict dotted-decimal/IPv6, so
decimal/hex/octal-encoded literals never resolve there; the gateway
normalizes every encoding with :mod:`ipaddress` semantics and denies them
deterministically. Smokescreen remains the per-connection enforcement point
for plain DNS names (including DNS-rebinding races the gateway cannot win).
- Nothing here ever logs bearer tokens or secret values; denial reasons echo
the normalized host only, never userinfo/query/fragment.
"""
from __future__ import annotations
import hashlib
import hmac
import ipaddress
import os
import re
import socket
from urllib.parse import urlsplit
# Keys whose audit-detail values must be redacted to preserve the
# no-secrets-in-logs/audit invariant.
_SENSITIVE_KEY_PARTS = ("token", "secret", "password", "authorization", "bearer", "cookie")
# Ticket #6 (gateway contract v1): input bounds and fail-closed resolver default.
#
# - URLs longer than MAX_URL_LENGTH are denied without DNS (fail-closed).
# - Unresolvable names / resolver exceptions fail closed by default so an
# automated client cannot slip internal targets past the first layer.
# Operators may opt back into delegation to Smokescreen (the second layer,
# which denies unresolvable hosts at connection time) with
# SWARM_SSRF_RESOLVER_FAIL_CLOSED=0. The opt-out is deliberate and logged.
MAX_URL_LENGTH = 2048
def resolver_fail_closed() -> bool:
"""True unless the operator explicitly opts into fail-open delegation."""
raw = os.environ.get("SWARM_SSRF_RESOLVER_FAIL_CLOSED", "").strip().lower()
return raw not in ("0", "false", "no", "off", "open")
# Phase 03 retry1 QA-B: search/research actions carry free-text queries, not
# URLs. Redacting them as URLs destroys forensics ("(invalid-url)").
_SEARCH_QUERY_ACTIONS = frozenset({"web_search", "deep_research"})
# Ticket #10 (gateway contract v1): audit query privacy modes. `verbatim`
# stores the scrubbed target as today; `hashed` stores a salted HMAC (equal
# queries correlate without being revealed); `redacted` stores only length
# and category. The hash chain always covers the STORED form, so it verifies
# in every mode.
AUDIT_QUERY_MODES = frozenset({"verbatim", "hashed", "redacted"})
def global_audit_query_mode() -> str:
"""The SWARM_AUDIT_QUERY_MODE default (verbatim unless configured)."""
raw = os.environ.get("SWARM_AUDIT_QUERY_MODE", "verbatim").strip().lower()
return raw if raw in AUDIT_QUERY_MODES else "verbatim"
def _redacted_query_form(action: str, target: str) -> str:
if action in _SEARCH_QUERY_ACTIONS:
kind = "query"
elif "://" in target:
kind = "url"
else:
kind = "other"
return f"redacted:{kind}:len={len(target)}"
def apply_audit_query_mode(mode: str, action: str, target: str | None) -> str | None:
"""Maps a scrubbed audit target to its stored form for *mode*.
Unknown modes fail closed to redacted (least disclosure). Hashed mode
without SWARM_AUDIT_HMAC_SALT configured also falls back to redacted so
a missing salt can never silently downgrade to verbatim.
"""
if target is None or mode == "verbatim":
return target
if mode == "hashed":
salt = os.environ.get("SWARM_AUDIT_HMAC_SALT", "")
if salt:
digest = hmac.new(
salt.encode(),
f"{action}\x00{target}".encode(),
hashlib.sha256,
).hexdigest()
return f"hmac-sha256:{digest}"
return _redacted_query_form(action, target)
return _redacted_query_form(action, target)
def is_expiry_passed(expires_at: str | None) -> bool:
"""Timezone-aware expiry check shared by gateway + panel-api sessions.
``None``/empty means no expiry (valid). Unparseable values fail closed
(treated as expired) so a corrupt timestamp never grants access.
NOTE: this None-means-valid semantic is for agent API keys only
(explicitly-acknowledged never-expiring keys). Sessions MUST use
:func:`is_session_expired_fail_closed` (NULL/empty/missing -> expired)
so legacy rows cannot become immortal.
"""
if not expires_at:
return False
try:
from datetime import UTC, datetime
exp = datetime.fromisoformat(str(expires_at))
if exp.tzinfo is None:
exp = exp.replace(tzinfo=UTC)
return exp <= datetime.now(UTC)
except Exception:
return True
def is_session_expired_fail_closed(expires_at: str | None) -> bool:
"""Fail-closed session expiry: NULL/empty/missing/whitespace -> expired.
Phase 03 retry2 QA-B P0-1: sessions must never be immortal. A missing
``expires_at`` (legacy row, old DB without the column, or empty string)
is treated as expired (401) rather than valid. Valid future timestamps
return False; past/naive-past/unparseable return True.
"""
if expires_at is None:
return True
text = str(expires_at).strip()
if not text:
return True
return is_expiry_passed(text)
# Phase 03 retry2 QA-B P0-2: secret-scrub patterns for audit forensics.
# Normal query text ("forensic query xyz123") must survive verbatim; only
# bearer-shaped material and key=value secrets are redacted.
_SWARM_SEC_RE = re.compile(r"swarm_sec_[A-Za-z0-9_\-]+")
_BEARER_RE = re.compile(r"(?i)\bBearer\s+[A-Za-z0-9_\-\.~\+/=]+")
_SECRET_KV_RE = re.compile(
r"(?i)((?:api[_-]?key|access[_-]?token|refresh[_-]?token|client[_-]?secret"
r"|token|secret|password|passwd|pwd|auth|bearer|session|cookie)"
r"[_-]*\s*[:=]\s*)([^\s&;\"',]+)"
)
# Ticket #13: also match ftp(s) URLs embedded in prose so userinfo/query
# secrets in non-http(s) URLs are stripped inline like http(s).
_URL_IN_TEXT_RE = re.compile(r"(?:https?|ftps?)://[^\s\"'<>]+")
_TRUNC_SUFFIX = "…[truncated]"
def _redact_embedded_urls(text: str) -> str:
"""Replaces every embedded http(s)/ftp(s) URL with its redacted host-only form."""
def _repl(m: re.Match) -> str:
raw = m.group(0)
# Preserve trailing punctuation that is not part of the URL.
trail = ""
while raw and raw[-1] in ".,;)]}!":
trail = raw[-1] + trail
raw = raw[:-1]
if not raw:
return m.group(0)
try:
redacted = redact_url_for_audit(raw)
except Exception:
redacted = "(invalid-url)"
if redacted == "(invalid-url)":
# Unparseable URL-like text: strip any secret KV inside it.
return _SECRET_KV_RE.sub(r"\1***redacted***", m.group(0))
return redacted + trail
return _URL_IN_TEXT_RE.sub(_repl, text)
def scrub_secrets_from_text(text: str | None) -> str | None:
"""Scrubs bearer-shaped secrets from free text, preserving forensics.
- embedded URLs -> host-only (query/fragment/userinfo stripped)
- ``swarm_sec_*`` -> ``swarm_sec_***redacted***``
- ``Bearer <tok>`` -> ``Bearer ***redacted***``
- ``token=SECRET``/``secret: x`` style KV -> value redacted, key kept
Normal prose without these patterns is returned unchanged.
"""
if text is None:
return None
s = str(text)
if not s:
return s
s = _redact_embedded_urls(s)
s = _SWARM_SEC_RE.sub("swarm_sec_***redacted***", s)
s = _BEARER_RE.sub("Bearer ***redacted***", s)
s = _SECRET_KV_RE.sub(r"\1***redacted***", s)
return s
def _truncate(text: str, limit: int = 2000) -> str:
if len(text) > limit:
return text[:limit] + _TRUNC_SUFFIX
return text
def audit_target_for_action(action: str, target: str | None) -> str | None:
"""Preserves search/research query text; redacts only true URLs.
Phase 03 retry2 QA-B P0-2: all returned targets are passed through
:func:`scrub_secrets_from_text` so ``swarm_sec_*`` bearer material and
``?token=``/``secret=`` query/fragment/userinfo secrets never persist,
while non-secret query forensics survive verbatim (truncated).
For ``web_search``/``deep_research`` the target is a free-text query and
must be kept verbatim (truncated) for forensics. If the query itself is
exactly a URL, the redacted host-only form is stored; prose embedding a
URL has that URL redacted inline. All other actions (fetch / scrape plus
denied/unknown-tool method names) go through
:func:`redact_url_for_action`, but a non-URL target that would otherwise
become ``(invalid-url)`` is preserved verbatim (truncated, scrubbed) so
denied rows keep their method/tool forensics instead of evidence-loss.
"""
if target is None:
return None
text = str(target).strip()
if not text:
return None
if action in _SEARCH_QUERY_ACTIONS:
if "://" in text:
redacted = redact_url_for_audit(text)
if redacted != "(invalid-url)":
return scrub_secrets_from_text(redacted)
# Prose query: scrub embedded URLs + bearer/KV secrets, keep the rest.
return _truncate(scrub_secrets_from_text(text) or "")
redacted = redact_url_for_audit(text)
if redacted != "(invalid-url)":
return scrub_secrets_from_text(redacted)
# Non-URL forensics (tool/method names like "nope_tool", "tools/list"):
# preserve verbatim truncated (scrubbed) rather than destroying to invalid-url.
return _truncate(scrub_secrets_from_text(text) or "")
def _scrub_detail_value(value):
"""Recursively scrubs secret-bearing strings inside audit detail values."""
if isinstance(value, str):
return scrub_secrets_from_text(value)
if isinstance(value, list):
return [_scrub_detail_value(v) for v in value]
if isinstance(value, tuple):
return tuple(_scrub_detail_value(v) for v in value)
if isinstance(value, dict):
return {k: _scrub_detail_value(v) for k, v in value.items()}
return value
def sanitize_details(details: dict | None) -> dict:
"""Redacts secret-looking values from an audit details mapping.
Phase 03 retry2 QA-B P0-2: every string value (including ``query``/``url``
keys and nested lists/dicts) is passed through
:func:`scrub_secrets_from_text` so bearer-shaped query text and
``?token=``-in-prose secrets never persist, while non-secret forensics
(limits, scopes, method names) survive.
"""
redacted: dict = {}
for key, value in (details or {}).items():
lowered = str(key).lower()
if any(part in lowered for part in _SENSITIVE_KEY_PARTS):
redacted[key] = "***redacted***"
else:
redacted[key] = _scrub_detail_value(value)
return redacted
def redact_url_for_audit(url: str) -> str:
"""Strips userinfo, query, and fragment so audit targets carry no secrets."""
try:
parts = urlsplit(url)
host = parts.hostname or ""
if not host:
return "(invalid-url)"
text = f"{parts.scheme or 'http'}://{host}"
if parts.path and parts.path != "/":
text += parts.path
return text
except Exception:
return "(invalid-url)"
def _part_to_int(part: str) -> int | None:
"""Parses one dotted-quad part accepting decimal, octal (leading 0), hex (0x)."""
try:
lowered = part.lower()
if lowered.startswith("0x") and len(lowered) > 2:
value = int(lowered[2:], 16)
elif len(part) > 1 and part.startswith("0") and part.isdigit():
value = int(part, 8)
elif part.isdigit():
value = int(part, 10)
else:
return None
except ValueError:
return None
return value if 0 <= value <= 255 else None
def normalize_ip_literal(host: str):
"""Normalizes an IP literal in any common encoding to an ipaddress object.
Handles strict dotted-decimal, IPv6 (with/without brackets), full-dword
decimal (``2852030506``), single hex word (``0xa9fea9fe``), per-quad hex
(``0xA9.0xFE.0xA9.0xFE``), per-quad octal (``0251.0376.0251.0376``), and
shortened ``a.b.c``/``a.b`` inet_aton forms. Returns ``None`` when *host*
is a DNS name rather than a numeric literal.
"""
cleaned = host.strip().rstrip(".")
if cleaned.startswith("[") and cleaned.endswith("]"):
cleaned = cleaned[1:-1]
if not cleaned:
return None
# Fast path: strict literals (covers ::1, ::ffff:127.0.0.1, 0.0.0.0, ...).
try:
return ipaddress.ip_address(cleaned)
except ValueError:
pass
if ":" in cleaned:
return None
# Single dword: decimal, hex, or octal.
if "." not in cleaned:
lowered = cleaned.lower()
try:
if lowered.startswith("0x") and len(lowered) > 2:
dword = int(lowered[2:], 16)
elif len(cleaned) > 1 and cleaned.startswith("0") and cleaned.isdigit():
dword = int(cleaned, 8)
elif cleaned.isdigit():
dword = int(cleaned, 10)
else:
return None
except ValueError:
return None
if 0 <= dword <= 0xFFFFFFFF:
return ipaddress.IPv4Address(dword)
return None
# Dotted forms with per-part encodings.
parts = cleaned.split(".")
if len(parts) == 4:
values = [_part_to_int(p) for p in parts]
if any(v is None for v in values):
return None
return ipaddress.IPv4Address(bytes(values)) # type: ignore[arg-type]
if len(parts) in (2, 3):
# inet_aton shortened forms: last part widens (a.b:c16 / a.b24).
width = 16 if len(parts) == 3 else 24
head = [_part_to_int(p) for p in parts[:-1]]
if any(v is None for v in head):
return None
try:
tail_raw = parts[-1].lower()
if tail_raw.startswith("0x"):
tail = int(tail_raw[2:], 16)
elif len(parts[-1]) > 1 and parts[-1].startswith("0") and parts[-1].isdigit():
tail = int(parts[-1], 8)
elif parts[-1].isdigit():
tail = int(parts[-1], 10)
else:
return None
except ValueError:
return None
if not 0 <= tail <= (1 << width) - 1:
return None
tail_bytes = tail.to_bytes(width // 8, "big")
return ipaddress.IPv4Address(bytes(head) + tail_bytes) # type: ignore[arg-type]
return None
def resolve_host(host: str, timeout: float = 3.0) -> list[str]:
"""Resolves a DNS hostname to string IPs (empty list when unresolvable)."""
try:
socket.setdefaulttimeout(timeout)
infos = socket.getaddrinfo(host, None, family=socket.AF_UNSPEC)
addrs: set[str] = {str(info[4][0]) for info in infos}
return sorted(addrs)
except Exception:
return []
finally:
socket.setdefaulttimeout(None)
def deny_reason_for_url(
url: str, timeout: float = 3.0, fail_closed_on_unresolvable: bool | None = None
) -> str | None:
"""Fail-closed gateway pre-check; returns a denial reason or ``None`` to allow.
Denies non-http(s) schemes, empty hosts, over-long URLs, every
non-globally-routable IP literal (loopback, RFC1918, CGNAT 100.64/10,
link-local/metadata 169.254/16, multicast, reserved, unspecified -- in
decimal, octal, hex, dword, or shortened encodings), ``localhost``, and
DNS names that resolve exclusively to non-global addresses.
Unresolvable names and resolver exceptions fail closed by default
(``SWARM_SSRF_RESOLVER_FAIL_CLOSED``, see :func:`resolver_fail_closed`);
pass ``fail_closed_on_unresolvable=False`` (or set the env opt-out) to
delegate those to Smokescreen, which denies unresolvable hosts at
connection time.
"""
if len(url) > MAX_URL_LENGTH:
return f"URL longer than {MAX_URL_LENGTH} characters (fail-closed)"
# WHATWG URL parsers strip ASCII tabs/newlines before parsing while
# urlsplit does not — a tab can smuggle a loopback host past the parser
# (e.g. http://127.0.0.1\[email protected]/). WHATWG browsers also treat
# backslash as a path delimiter while urlsplit treats it as userinfo
# (e.g. http://169.254.169.254\@example.com/ connects to 169.254.169.254
# per browser but urlsplit sees host example.com). Deny control
# characters and backslash outright so parser differentials cannot
# bypass the pre-check.
if any(ord(c) < 32 or ord(c) == 127 or c == "\\" for c in url):
return "URL contains forbidden characters (fail-closed)"
fail_closed = (
resolver_fail_closed()
if fail_closed_on_unresolvable is None
else fail_closed_on_unresolvable
)
try:
parts = urlsplit(url)
except Exception:
return "unparseable URL (fail-closed)"
if parts.scheme.lower() not in ("http", "https"):
return f"unsupported scheme '{parts.scheme or '(none)'}' (fail-closed)"
host = (parts.hostname or "").strip().rstrip(".")
if not host:
return "empty hostname (fail-closed)"
if host.lower() == "localhost":
return "localhost resolves to loopback (fail-closed)"
literal = normalize_ip_literal(host)
if literal is not None:
if not literal.is_global:
return f"host '{literal.compressed}' is not globally routable (fail-closed)"
return None
try:
resolved = resolve_host(host, timeout=timeout)
except Exception:
# Resolver unavailable/timed out: fail closed by default so an
# automated client cannot slip targets past the first layer.
if fail_closed:
return f"host '{host}' DNS resolution failed (fail-closed)"
return None
if not resolved:
# Cannot resolve here: fail closed by default; Smokescreen remains
# the per-connection enforcement point when the operator opts out.
if fail_closed:
return f"host '{host}' unresolvable (fail-closed)"
return None
parsed = []
for candidate in resolved:
try:
parsed.append(ipaddress.ip_address(candidate))
except ValueError:
continue
if parsed and any(not addr.is_global for addr in parsed):
return f"host '{host}' resolves to non-routable addresses (fail-closed)"
return None
@@ -0,0 +1,280 @@
"""swarmctl: host-operator CLI for service-account keys (ticket #7).
Run on the host (e.g. ``docker compose exec panel-api swarmctl keys create
...``). Host shell access is the authorization, consistent with the bootstrap
model — there is deliberately no network path here. Every mutation writes an
audit row (actor ``swarmctl``).
"""
from __future__ import annotations
import argparse
import asyncio
import json
import os
import re
import sys
from datetime import UTC, datetime, timedelta
from pathlib import Path
from panel_api.audit import AuditLogger
from panel_api.db import Database
from panel_api.keys import (
KeyExistsError,
expiry_iso,
find_key,
mint_key,
rotate_key,
)
DEFAULT_DB_PATH = "/var/lib/scraper-swarm/panel.db"
_DURATION_RE = re.compile(r"^\s*(\d+)\s*([smhd])?\s*$", re.IGNORECASE)
_DURATION_MULT = {"s": 1 / 3600, "m": 1 / 60, "h": 1, "d": 24}
def parse_duration_hours(text: str) -> float | None:
"""Parses '90d'/'24h'/'30m'/'60s'/bare-hours into hours; 'never' -> None."""
cleaned = text.strip().lower()
if cleaned == "never":
return None
m = _DURATION_RE.match(cleaned)
if not m:
raise ValueError(f"Bad duration '{text}': use like 90d, 24h, 720, or never")
amount = int(m.group(1))
unit = (m.group(2) or "h").lower()
return amount * _DURATION_MULT[unit]
def _db_path(args) -> Path:
return Path(args.db or os.environ.get("SWARM_DB_PATH", DEFAULT_DB_PATH))
def build_parser() -> argparse.ArgumentParser:
p = argparse.ArgumentParser(prog="swarmctl", description="Scraper Swarm operator CLI")
p.add_argument("--db", default=None, help="panel DB path (default $SWARM_DB_PATH)")
sub = p.add_subparsers(dest="group", required=True)
keys = sub.add_parser("keys", help="service-account keys")
ksub = keys.add_subparsers(dest="cmd", required=True)
c = ksub.add_parser("create", help="mint a service-account key (prints raw key once)")
c.add_argument("--name", required=True)
c.add_argument("--scopes", default="search,scrape")
c.add_argument("--rpm", type=int, default=60)
c.add_argument("--expires", default="720h")
c.add_argument("--allow-never-expire", action="store_true")
c.add_argument(
"--audit-mode",
default="default",
help="audit query privacy: default (global), verbatim, hashed, redacted",
)
r = ksub.add_parser("rotate", help="rotate a key (old stays valid for --grace)")
r.add_argument("ref", help="key id or name")
r.add_argument("--grace", default="24h")
r.add_argument("--expires", default="720h")
r.add_argument("--allow-never-expire", action="store_true")
ksub.add_parser("list", help="list keys (prefixes only, never secrets)")
d = ksub.add_parser("revoke", help="revoke (delete) a key")
d.add_argument("ref", help="key id or name")
audit = sub.add_parser("audit", help="audit log maintenance")
asub = audit.add_subparsers(dest="cmd", required=True)
pr = asub.add_parser("prune", help="delete rows older than retention, keep chain verifiable")
pr.add_argument(
"--retention-days",
default=None,
help="override SWARM_AUDIT_RETENTION_DAYS (required when unset)",
)
return p
async def _cmd_create(db: Database, args) -> int:
hours = parse_duration_hours(args.expires)
if hours is None and not args.allow_never_expire:
print(
"error: --expires never creates a never-expiring key;"
" pass --allow-never-expire to acknowledge",
file=sys.stderr,
)
return 1
scopes = [s.strip() for s in args.scopes.split(",") if s.strip()]
try:
key = await mint_key(
db.conn,
name=args.name,
scopes=scopes,
rate_limit_rpm=args.rpm,
expires_at=expiry_iso(None if hours is None else int(hours)),
audit_query_mode=args.audit_mode,
)
except KeyExistsError as e:
print(f"error: {e}", file=sys.stderr)
return 1
except ValueError as e:
print(f"error: {e}", file=sys.stderr)
return 1
await AuditLogger(db.conn).log(
actor="swarmctl",
action="create_agent_key",
target=key["id"],
details={
"name": key["name"],
"scopes": key["scopes"],
"key_prefix": key["key_prefix"],
"audit_query_mode": key["audit_query_mode"] or "default",
},
)
print(f"created key '{key['name']}' (id {key['id']})")
print(f"raw_key: {key['raw_key']}")
print("Store the raw key now — it is shown once and never again.")
print(
f"scopes: {','.join(key['scopes'])} rpm: {key['rate_limit_rpm']}"
f" expires_at: {key['expires_at']}"
)
return 0
async def _cmd_rotate(db: Database, args) -> int:
try:
grace_hours = parse_duration_hours(args.grace)
except ValueError as e:
print(f"error: {e}", file=sys.stderr)
return 1
if grace_hours is None:
print("error: --grace must be a duration, not never", file=sys.stderr)
return 1
hours = parse_duration_hours(args.expires)
if hours is None and not args.allow_never_expire:
print("error: --expires never requires --allow-never-expire", file=sys.stderr)
return 1
try:
result = await rotate_key(
db.conn,
ref=args.ref,
grace_hours=grace_hours,
expires_in_hours=None if hours is None else int(hours),
)
except (KeyError, ValueError) as e:
print(f"error: {e}", file=sys.stderr)
return 1
new = result["new"]
old = result["old"]
await AuditLogger(db.conn).log(
actor="swarmctl",
action="rotate_agent_key",
target=new["id"],
details={
"name": new["name"],
"retired_id": old["id"],
"grace_expires_at": old["expires_at"],
},
)
print(f"rotated '{new['name']}': old id {old['id']} valid until {old['expires_at']}")
print(f"raw_key: {new['raw_key']}")
print("Store the raw key now — it is shown once and never again.")
return 0
async def _cmd_list(db: Database, args) -> int:
async with db.conn.execute(
"SELECT id, name, key_prefix, scopes, rate_limit_rpm, created_at,"
" expires_at, last_used_at, audit_query_mode FROM agent_keys ORDER BY created_at DESC"
) as cur:
rows = await cur.fetchall()
print(
json.dumps(
[
{
"id": r["id"],
"name": r["name"],
"key_prefix": r["key_prefix"],
"scopes": json.loads(r["scopes"]),
"rate_limit_rpm": r["rate_limit_rpm"],
"created_at": r["created_at"],
"expires_at": r["expires_at"],
"last_used_at": r["last_used_at"],
"audit_query_mode": r["audit_query_mode"] or "default",
}
for r in rows
],
indent=2,
)
)
return 0
async def _cmd_revoke(db: Database, args) -> int:
row = await find_key(db.conn, args.ref)
if row is None:
print(f"error: key '{args.ref}' not found", file=sys.stderr)
return 1
await db.conn.execute("DELETE FROM agent_keys WHERE id = ?", (row["id"],))
await db.conn.commit()
await AuditLogger(db.conn).log(actor="swarmctl", action="revoke_agent_key", target=row["id"])
print(f"revoked key '{row['name']}' (id {row['id']})")
return 0
async def _cmd_audit_prune(db: Database, args) -> int:
"""Deletes audit rows older than retention, keeping a checkpoint row so
verify_chain still passes (ticket #10)."""
raw_days = args.retention_days
if raw_days is None:
raw_days = os.environ.get("SWARM_AUDIT_RETENTION_DAYS", "")
try:
days = float(str(raw_days).strip())
except ValueError:
print(
"error: retention unknown: pass --retention-days or set SWARM_AUDIT_RETENTION_DAYS",
file=sys.stderr,
)
return 1
if days <= 0:
print("error: retention must be positive days (0 disables pruning)", file=sys.stderr)
return 1
cutoff = (datetime.now(UTC) - timedelta(days=days)).isoformat()
pruned = await AuditLogger(db.conn).prune_older_than(cutoff)
await AuditLogger(db.conn).log(
actor="swarmctl",
action="prune_audit",
details={"pruned": pruned, "cutoff": cutoff},
)
print(f"pruned {pruned} audit rows older than {cutoff} (checkpoint kept)")
return 0
async def amain(argv: list[str] | None = None) -> int:
args = build_parser().parse_args(argv)
db = Database(_db_path(args))
await db.connect()
try:
if args.group == "keys":
if args.cmd == "create":
return await _cmd_create(db, args)
if args.cmd == "rotate":
return await _cmd_rotate(db, args)
if args.cmd == "list":
return await _cmd_list(db, args)
if args.cmd == "revoke":
return await _cmd_revoke(db, args)
elif args.group == "audit":
if args.cmd == "prune":
return await _cmd_audit_prune(db, args)
except ValueError as e:
print(f"error: {e}", file=sys.stderr)
return 1
finally:
await db.close()
print("error: unknown command", file=sys.stderr)
return 2
def main(argv: list[str] | None = None) -> int:
"""Sync entry point for the `swarmctl` console script."""
return asyncio.run(amain(argv))
if __name__ == "__main__":
raise SystemExit(main())
@@ -0,0 +1,46 @@
"""Client for communicating with the privileged swarmd sidecar over Unix Domain Socket."""
from __future__ import annotations
import asyncio
import json
from pathlib import Path
from typing import Any
class SwarmdClientError(Exception):
pass
class SwarmdClient:
def __init__(self, socket_path: Path):
self.socket_path = socket_path
async def send_intent(self, intent: str, payload: dict[str, Any]) -> dict[str, Any]:
"""Sends an intent to swarmd via Unix domain socket and awaits response."""
if not self.socket_path.exists():
raise SwarmdClientError(f"swarmd socket {self.socket_path} does not exist.")
try:
reader, writer = await asyncio.open_unix_connection(str(self.socket_path))
except Exception as e:
raise SwarmdClientError(f"Failed to connect to swarmd socket: {e}") from e
try:
message = json.dumps({"intent": intent, "payload": payload}) + "\n"
writer.write(message.encode("utf-8"))
await writer.drain()
line = await reader.readline()
if not line:
raise SwarmdClientError("swarmd closed connection without responding.")
response = json.loads(line.decode("utf-8"))
if not response.get("ok"):
error_msg = response.get("error", "Unknown swarmd error")
raise SwarmdClientError(f"swarmd rejected intent: {error_msg}")
return response.get("data", {})
finally:
writer.close()
await writer.wait_closed()
@@ -0,0 +1,86 @@
"""Cryptographic vault for encrypting secrets at rest using AES-256-GCM.
The master key can be provided via environment variable (SWARM_MASTER_KEY)
or read from a file with strict 0600 permissions.
"""
from __future__ import annotations
import base64
import os
from pathlib import Path
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
class VaultError(Exception):
pass
class Vault:
def __init__(self, key: bytes):
if len(key) != 32:
raise VaultError("Master key must be exactly 32 bytes (256 bits).")
self._key = key
self._aesgcm = AESGCM(self._key)
@classmethod
def from_env(cls, env_var: str = "SWARM_MASTER_KEY") -> Vault:
val = os.environ.get(env_var)
if not val:
raise VaultError(f"Environment variable {env_var} is not set.")
try:
key = base64.b64decode(val)
except Exception as e:
raise VaultError(f"Invalid base64 in {env_var}: {e}") from e
return cls(key)
@classmethod
def from_file(cls, path: Path, *, auto_create: bool = False) -> Vault:
if not path.exists():
if auto_create:
path.parent.mkdir(parents=True, exist_ok=True)
key = AESGCM.generate_key(bit_length=256)
# Create file with 0600 permissions
flags = os.O_WRONLY | os.O_CREAT | os.O_TRUNC
fd = os.open(path, flags, 0o600)
with os.fdopen(fd, "wb") as f:
f.write(base64.b64encode(key))
return cls(key)
raise VaultError(f"Master key file {path} does not exist.")
# Check permissions: must not be world/group readable
mode = path.stat().st_mode & 0o777
if mode & 0o077:
raise VaultError(
f"Master key file {path} has unsafe permissions ({oct(mode)}). "
"Must be 0600 or stricter."
)
content = path.read_text().strip()
try:
key = base64.b64decode(content)
except Exception as e:
raise VaultError(f"Failed to decode key from {path}: {e}") from e
return cls(key)
def encrypt(self, plaintext: str, associated_data: str | None = None) -> str:
"""Encrypts plaintext and returns base64-encoded nonce + ciphertext."""
nonce = os.urandom(12) # 96-bit nonce standard for GCM
ad = associated_data.encode("utf-8") if associated_data else None
ciphertext = self._aesgcm.encrypt(nonce, plaintext.encode("utf-8"), ad)
return base64.b64encode(nonce + ciphertext).decode("utf-8")
def decrypt(self, token: str, associated_data: str | None = None) -> str:
"""Decrypts a base64-encoded token back into plaintext."""
try:
data = base64.b64decode(token)
if len(data) < 28: # 12-byte nonce + 16-byte tag min
raise VaultError("Token is too short.")
nonce = data[:12]
ciphertext = data[12:]
ad = associated_data.encode("utf-8") if associated_data else None
decrypted = self._aesgcm.decrypt(nonce, ciphertext, ad)
return decrypted.decode("utf-8")
except Exception as e:
raise VaultError(f"Decryption failed: {e}") from e
@@ -0,0 +1,28 @@
id: cloakbrowser
name: CloakBrowser
tier: optional
verified: false
license:
spdx: "LicenseRef-Proprietary"
notice: "Wrappers are MIT, but the underlying modified Chromium binary is proprietary. Concurrency is limited on the free tier."
image: { repo: cloakhq/cloakbrowser, tag: latest }
resources:
lite: { cpus: 1, mem: 1g, shm: 1g }
standard: { cpus: 2, mem: 2g, shm: 2g }
heavy: { cpus: 4, mem: 4g, shm: 4g }
networks: [svc, egress-web]
port: 9222
hardening: { user: "1000:1000", seccomp: chromium }
secrets: [CLOAKBROWSER_LICENSE_KEY]
native_exposable: false
params_schema:
type: object
additionalProperties: false
properties:
release_channel: { type: string, enum: ["stable", "preview"] }
humanize: { type: boolean }
concurrency_limit: { type: integer, minimum: 1, maximum: 10 }
param_env:
release_channel: CLOAKBROWSER_RELEASE_CHANNEL
humanize: CLOAKBROWSER_HUMANIZE
concurrency_limit: CLOAKBROWSER_CONCURRENCY
@@ -1,18 +1,41 @@
id: crawl4ai
name: Crawl4AI
tier: core
verified: false
license: { spdx: Apache-2.0 }
image: { repo: unclecode/crawl4ai, tag: latest }
port: 11235
networks: [svc, egress-web]
resources:
lite: { cpus: 1, mem: 2g, shm: 1g }
lite: { cpus: 1, mem: 2g, shm: 1g }
standard: { cpus: 2, mem: 4g, shm: 2g }
heavy: { cpus: 4, mem: 8g, shm: 4g }
networks: [svc, egress-web]
port: 11235
hardening: { user: "999:999", seccomp: chromium }
# P1-live-smoke NO_PROXY fix follow-up (walkthrough.md::smoke-3-3): loopback
# no longer bypasses Smokescreen, so the image's built-in
# `curl -f http://localhost:11235/health` probe would be proxied and denied
# with 407. This override keeps the same probe but pins curl to
# `--noproxy '*'` so only the healthcheck -- never crawler traffic -- skips
# the egress proxy.
health:
test: ["CMD-SHELL", "redis-cli ping > /dev/null && curl --noproxy '*' -f http://localhost:11235/health || exit 1"]
interval: 30s
secrets: [CRAWL4AI_API_TOKEN, OPENAI_API_KEY, ANTHROPIC_API_KEY, GROQ_API_KEY]
env:
CRAWL4AI_ALLOW_INTERNAL_URLS: "true"
egress: { proxy: smokescreen, direct_peers: [gateway] }
health:
test: ["CMD-SHELL", "curl --noproxy '*' -f http://localhost:11235/health || exit 1"]
interval: 30s
tools: [fetch]
native_exposable: true
params_schema:
type: object
additionalProperties: false
properties:
llm_provider: { type: string, pattern: "^[a-z0-9_.:/-]{1,80}$" }
llm_base_url: { type: string }
llm_temperature: { type: number, minimum: 0.0, maximum: 2.0 }
headless: { type: boolean }
page_timeout: { type: integer, minimum: 5, maximum: 300 }
param_env:
llm_provider: LLM_PROVIDER
llm_base_url: LLM_BASE_URL
llm_temperature: LLM_TEMPERATURE
headless: CRAWL4AI_HEADLESS
page_timeout: CRAWL4AI_TIMEOUT
@@ -0,0 +1,26 @@
id: cyberscraper
name: CyberScraper-2077
tier: optional
verified: false
license:
spdx: MIT
notice: "CyberScraper provides a cyberpunk-themed Streamlit UI. Accessible exclusively behind panel authenticated proxy."
image: { repo: itsowen/cyberscraper-2077, tag: latest }
resources:
lite: { cpus: 1, mem: 1g }
standard: { cpus: 2, mem: 2g }
heavy: { cpus: 4, mem: 4g }
networks: [svc, egress-web]
port: 8501
hardening: { user: "1000:1000", tmpfs: ["/tmp"] }
secrets: [OPENAI_API_KEY, GEMINI_API_KEY]
native_exposable: false
params_schema:
type: object
additionalProperties: false
properties:
model_provider: { type: string, enum: ["openai", "gemini", "ollama"] }
default_model: { type: string }
param_env:
model_provider: MODEL_PROVIDER
default_model: DEFAULT_MODEL
@@ -0,0 +1,10 @@
id: egress-web
name: Egress proxy (Smokescreen)
tier: support
verified: false
license: { spdx: MIT }
image: { build: deploy/egress }
resources:
standard: { cpus: 1, mem: 256m }
networks: [egress-web, egress-out]
hardening: { user: "65532:65532", read_only: true }
@@ -1,15 +1,33 @@
id: firecrawl
name: Firecrawl
tier: advanced
license: { spdx: AGPL-3.0-only }
verified: false
license:
spdx: AGPL-3.0-only
notice: "Firecrawl is AGPL-3.0. Heavy resource footprint requiring at least 8GB RAM."
image: { repo: firecrawl/firecrawl, tag: latest }
port: 3002
networks: [svc, egress-web]
resources:
standard: { cpus: 4, mem: 8g, shm: 2g }
heavy: { cpus: 8, mem: 16g, shm: 4g }
networks: [svc, egress-web]
port: 3002
hardening: { user: "1000:1000", seccomp: chromium }
egress: { proxy: smokescreen, direct_peers: [gateway] }
health:
test: ["CMD-SHELL", "curl --noproxy '*' -f http://localhost:3002/health || exit 1"]
interval: 30s
tools: [fetch]
secrets: [TEST_API_KEY, OPENAI_API_KEY, BULL_AUTH_KEY]
native_exposable: false
params_schema:
type: object
additionalProperties: false
properties:
use_db_authentication: { type: boolean }
num_workers_per_queue: { type: integer, minimum: 1, maximum: 32 }
crawl_concurrent_requests: { type: integer, minimum: 1, maximum: 50 }
model_name: { type: string }
searxng_endpoint: { type: string }
logging_level: { type: string, enum: ["info", "warn", "error", "debug"] }
param_env:
use_db_authentication: USE_DB_AUTHENTICATION
num_workers_per_queue: NUM_WORKERS_PER_QUEUE
crawl_concurrent_requests: CRAWL_CONCURRENT_REQUESTS
model_name: MODEL_NAME
searxng_endpoint: SEARXNG_ENDPOINT
logging_level: LOGGING_LEVEL
@@ -0,0 +1,40 @@
id: gpt-researcher
name: GPT Researcher
tier: core
verified: false
license:
spdx: Apache-2.0
image: { repo: gptresearcher/gpt-researcher, tag: latest }
resources:
lite: { cpus: 1, mem: 1g }
standard: { cpus: 2, mem: 2g }
heavy: { cpus: 4, mem: 4g }
networks: [svc, egress-web]
requires: [searxng]
port: 8000
hardening: { user: "1000:1000", tmpfs: ["/tmp"] }
secrets: [OPENAI_API_KEY, ANTHROPIC_API_KEY, TAVILY_API_KEY]
env:
RETRIEVER: "searx"
SEARX_URL: "http://searxng:8080"
# Ticket #6: the retriever calls searxng over plain HTTP, so that one peer
# stays direct; everything else traverses the egress proxy.
direct_peers: [searxng]
native_exposable: true
params_schema:
type: object
additionalProperties: false
properties:
retriever: { type: string, enum: ["searx", "duckduckgo", "tavily", "google"] }
smart_llm: { type: string }
fast_llm: { type: string }
embedding_provider: { type: string }
total_words: { type: integer, minimum: 500, maximum: 10000 }
report_format: { type: string, enum: ["APA", "MLA", "IEEE", "standard"] }
param_env:
retriever: RETRIEVER
smart_llm: SMART_LLM
fast_llm: FAST_LLM
embedding_provider: EMBEDDING_PROVIDER
total_words: TOTAL_WORDS
report_format: REPORT_FORMAT
@@ -0,0 +1,28 @@
id: maxun
name: Maxun
tier: optional
verified: false
license:
spdx: AGPL-3.0-only
notice: "Maxun is licensed under AGPL-3.0. Accessible only via authenticated panel forward proxy."
image: { repo: getmaxun/maxun-backend, tag: latest }
resources:
lite: { cpus: 1, mem: 2g, shm: 1g }
standard: { cpus: 2, mem: 4g, shm: 2g }
heavy: { cpus: 4, mem: 8g, shm: 4g }
networks: [svc, egress-web]
port: 8080
hardening: { user: "1000:1000", seccomp: chromium }
secrets: [JWT_SECRET, ENCRYPTION_KEY, DB_PASSWORD]
native_exposable: false
params_schema:
type: object
additionalProperties: false
properties:
headless: { type: boolean }
max_concurrent_sessions: { type: integer, minimum: 1, maximum: 10 }
backend_port: { type: integer, minimum: 1024, maximum: 65535 }
param_env:
headless: HEADLESS
max_concurrent_sessions: MAX_CONCURRENT_SESSIONS
backend_port: BACKEND_PORT
@@ -1,16 +1,28 @@
id: scrapling
name: Scrapling
tier: core
license: { spdx: BSD-3-Clause }
verified: false
license:
spdx: BSD-3-Clause
notice: "Scrapling uses Camoufox browser components for adaptive stealth fetching."
image: { repo: pyd4vinci/scrapling, tag: latest }
port: 8000
networks: [svc, egress-web]
resources:
lite: { cpus: 1, mem: 1g, shm: 512m }
lite: { cpus: 1, mem: 1g, shm: 512m }
standard: { cpus: 2, mem: 2g, shm: 1g }
heavy: { cpus: 4, mem: 4g, shm: 2g }
networks: [svc, egress-web]
port: 8000
hardening: { user: "1000:1000", seccomp: chromium, tmpfs: ["/tmp"] }
egress: { proxy: smokescreen, direct_peers: [gateway] }
health:
test: ["CMD-SHELL", "curl --noproxy '*' -f http://localhost:8000/health || exit 1"]
interval: 30s
tools: [extract_structured]
secrets: [PROXY_PASSWORD]
native_exposable: true
params_schema:
type: object
additionalProperties: false
properties:
stealth_engine: { type: string, enum: ["camoufox", "playwright"] }
headless: { type: boolean }
network_timeout: { type: integer, minimum: 5, maximum: 120 }
param_env:
stealth_engine: SCRAPLING_ENGINE
headless: SCRAPLING_HEADLESS
network_timeout: SCRAPLING_TIMEOUT
+29 -7
View File
@@ -1,17 +1,39 @@
id: searxng
name: SearXNG
tier: core
license: { spdx: AGPL-3.0-only }
verified: false
license:
spdx: AGPL-3.0-only
notice: "SearXNG is AGPL-3.0. It runs unmodified in its own container; Scraper Swarm does not link against it."
image: { repo: searxng/searxng, tag: latest }
port: 8080
resources:
lite: { cpus: 0.5, mem: 512m }
standard: { cpus: 1, mem: 1g }
heavy: { cpus: 2, mem: 2g }
networks: [svc, egress-web]
requires: [valkey]
resources:
lite: { cpus: 0.5, mem: 512m }
standard: { cpus: 1, mem: 1g }
port: 8080
hardening: { user: "977:977", read_only: false, tmpfs: ["/tmp"] }
egress: { proxy: smokescreen, direct_peers: [gateway] }
# Phase 02-infra-reconcile: liveness probe for health-driven restart.
# Loopback MUST bypass the egress proxy env (NO_PROXY covers in-stack
# peers, not localhost) or the probe gets Smokescreen 407 and flaps.
# Verified live: direct wget on :8080 is HTTP 200.
health:
test: ["CMD-SHELL", "env -u HTTP_PROXY -u HTTPS_PROXY -u http_proxy -u https_proxy wget -q -O /dev/null http://localhost:8080/ || exit 1"]
interval: 30s
tools: [search]
secrets: [SEARXNG_SECRET]
volumes: ["config:/etc/searxng"]
native_exposable: false
params_schema:
type: object
additionalProperties: false
properties:
search_language: { type: string, pattern: "^[a-zA-Z_-]{2,10}$" }
safe_search: { type: integer, minimum: 0, maximum: 2 }
autocomplete: { type: string, enum: ["google", "duckduckgo", "brave", "none"] }
base_url: { type: string, format: "uri" }
param_env:
search_language: SEARXNG_DEFAULT_LANG
safe_search: SEARXNG_SAFE_SEARCH
autocomplete: SEARXNG_AUTOCOMPLETE
base_url: SEARXNG_BASE_URL
@@ -0,0 +1,17 @@
id: valkey
name: Valkey
tier: support
verified: false
license: { spdx: BSD-3-Clause }
image: { repo: valkey/valkey, tag: latest }
resources:
standard: { cpus: 0.5, mem: 256m }
networks: [svc]
port: 6379
hardening: { user: "999:999", read_only: true, tmpfs: ["/tmp"] }
# Phase 02-infra-reconcile: liveness probe for health-driven restart.
# Verified live: `valkey-cli ping` returns PONG in this image.
health:
test: ["CMD-SHELL", "valkey-cli ping | grep -q PONG"]
interval: 30s
volumes: ["data:/data"]
@@ -0,0 +1,78 @@
# Caddy Edge Proxy Configuration for opencode-gateway
# Automated Let's Encrypt TLS with forward-auth guards for human interfaces.
# Adapted from scraper-swarm/deploy/Caddyfile: same matchers and guards;
# domain variable renamed to GATEWAY_DOMAIN and /api/* routed to panel-api
# (the TS gateway serves only /mcp, /ready, /health).
{$GATEWAY_DOMAIN:localhost} {
# Security Headers
header {
Strict-Transport-Security "max-age=31536000; includeSubDomains; preload"
X-Content-Type-Options "nosniff"
X-Frame-Options "DENY"
Content-Security-Policy "default-src 'self'; style-src 'self' 'unsafe-inline' https://fonts.googleapis.com; font-src 'self' https://fonts.gstatic.com; img-src 'self' data:; connect-src 'self'; frame-ancestors 'none'; object-src 'none';"
}
# Authenticated MCP Gateway Route (TS/Bun gateway: POST /mcp, GET /ready, GET /health)
handle /mcp* {
reverse_proxy gateway:8000
}
handle /ready {
reverse_proxy gateway:8000
}
handle /health {
reverse_proxy gateway:8000
}
# Control-plane API (panel-api: key issuance, services, auth, security)
handle /api/* {
reverse_proxy panel-api:8000
}
# Protected Web UIs: Forward-Auth through panel-api session verification
handle /apps/maxun/* {
forward_auth panel-api:8000 {
uri /auth/me
copy_headers swarm_session
}
uri strip_prefix /apps/maxun
reverse_proxy maxun:8080
}
handle /apps/cyberscraper/* {
forward_auth panel-api:8000 {
uri /auth/me
copy_headers swarm_session
}
uri strip_prefix /apps/cyberscraper
reverse_proxy cyberscraper:8501
}
# API routes
handle /auth/* {
reverse_proxy panel-api:8000
}
handle /services/* {
reverse_proxy panel-api:8000
}
handle /agents/* {
reverse_proxy panel-api:8000
}
handle /security/* {
reverse_proxy panel-api:8000
}
# Static Web Panel UI (Compiled Vite React Bundle).
# Mount a built panel UI at ./panel-web-dist (or the panel_web volume);
# until then this serves 404 for unmatched paths (no open proxy).
handle {
root * /var/www/panel-web
file_server
try_files {path} /index.html
}
}
@@ -1,22 +0,0 @@
# Caddy edge proxy snippet (opencode-gateway).
# TLS via Automated Let's Encrypt; /mcp + /ready + /health route to the gateway.
{$GATEWAY_DOMAIN:localhost} {
header {
Strict-Transport-Security "max-age=31536000; includeSubDomains; preload"
X-Content-Type-Options "nosniff"
X-Frame-Options "DENY"
Content-Security-Policy "default-src 'self'; frame-ancestors 'none'; object-src 'none';"
}
handle /mcp* {
reverse_proxy gateway:8000
}
handle /ready {
reverse_proxy gateway:8000
}
handle /health {
reverse_proxy gateway:8000
}
}
@@ -0,0 +1,23 @@
# Panel API image for opencode-gateway.
# Adapted from scraper-swarm/deploy/Dockerfile.panel-api: same base, uv
# install, uvicorn entrypoint; workspace COPY lines for swarmd/gateway
# dropped because the catalog loader is vendored into panel_api itself.
FROM python:3.12-slim
WORKDIR /opt/opencode-gateway
RUN apt-get update && apt-get install -y --no-install-recommends \
curl build-essential && \
rm -rf /var/lib/apt/lists/*
COPY --from=ghcr.io/astral-sh/uv:latest /uv /bin/uv
COPY catalog ./catalog
COPY apps/panel-api ./apps/panel-api
RUN uv pip install --system ./apps/panel-api
ENV PATH="/opt/opencode-gateway/.venv/bin:$PATH"
EXPOSE 8000
CMD ["uvicorn", "panel_api.main:app", "--host", "0.0.0.0", "--port", "8000"]
@@ -0,0 +1,142 @@
# opencode-gateway — Production Control Plane Compose
# Deploys Caddy Edge Proxy, TS/Bun MCP Gateway, Panel API, Smokescreen
# Egress SSRF Guard, and Valkey.
# Adapted from scraper-swarm/deploy/docker-compose.control.yml: same service
# roles, env contract (SWARM_*), internal networks, and volume posture;
# names re-scoped to the opencode-gateway layout (gateway-* / og_*).
#
# Operator environment lives in the package-root .env (copy from .env.example).
# Launch with: docker compose --env-file ../.env -f deploy/compose.yaml up -d
# (env_file entries below also load ../.env into panel-api/gateway containers.)
services:
caddy:
image: caddy:2
restart: unless-stopped
ports:
- "80:80"
- "443:443"
volumes:
- ./Caddyfile:/etc/caddy/Caddyfile:ro,z
- panel_web:/var/www/panel-web:ro,z
- caddy_data:/data
- caddy_config:/config
networks:
- edge
depends_on:
- panel-api
- gateway
panel-api:
build:
context: ..
dockerfile: deploy/Dockerfile.panel-api
restart: unless-stopped
env_file:
- path: ../.env
required: false
environment:
- SWARM_DATA_DIR=/var/lib/opencode-gateway
- SWARM_REPO_ROOT=/opt/opencode-gateway
- SWARM_BOOTSTRAP_TOKEN
# Public MCP origin for key-issuance snippets, and the
# external domain (Caddy uses GATEWAY_DOMAIN).
- SWARM_PUBLIC_MCP_URL=${SWARM_PUBLIC_MCP_URL:-}
- GATEWAY_DOMAIN=${GATEWAY_DOMAIN:-localhost}
# Gateway contract env: SSRF resolver posture,
# audit query privacy / HMAC salt / retention.
- SWARM_SSRF_RESOLVER_FAIL_CLOSED=${SWARM_SSRF_RESOLVER_FAIL_CLOSED:-}
- SWARM_AUDIT_QUERY_MODE=${SWARM_AUDIT_QUERY_MODE:-verbatim}
- SWARM_AUDIT_HMAC_SALT
- SWARM_AUDIT_RETENTION_DAYS=${SWARM_AUDIT_RETENTION_DAYS:-}
volumes:
- gateway_data:/var/lib/opencode-gateway
- ../catalog:/opt/opencode-gateway/catalog:ro,z
networks:
- edge
- control
- svc
gateway:
image: oven/bun:1
restart: unless-stopped
working_dir: /opt/opencode-gateway/apps/gateway
command: ["run", "src/index.ts"]
env_file:
- path: ../.env
required: false
environment:
- GATEWAY_PORT=8000
- GATEWAY_TOKEN
- SEARXNG_URL=http://searxng:8080
- CRAWL4AI_URL=http://crawl4ai:11235
# Gateway contract env: fetch cap, SSRF resolver posture,
# audit privacy/retention, deployed engine set, public domain/origin.
- SWARM_MAX_FETCH_BYTES=${SWARM_MAX_FETCH_BYTES:-3145728}
- SWARM_SSRF_RESOLVER_FAIL_CLOSED=${SWARM_SSRF_RESOLVER_FAIL_CLOSED:-}
- SWARM_AUDIT_QUERY_MODE=${SWARM_AUDIT_QUERY_MODE:-verbatim}
- SWARM_AUDIT_HMAC_SALT
- SWARM_AUDIT_RETENTION_DAYS=${SWARM_AUDIT_RETENTION_DAYS:-}
- SWARM_DEPLOYED_ENGINES=${SWARM_DEPLOYED_ENGINES:-}
- GATEWAY_DOMAIN=${GATEWAY_DOMAIN:-localhost}
- SWARM_PUBLIC_MCP_URL=${SWARM_PUBLIC_MCP_URL:-}
# Lane egress: every engine fetch traverses Smokescreen.
- HTTP_PROXY=http://smokescreen:4750
- HTTPS_PROXY=http://smokescreen:4750
- NO_PROXY=gateway,searxng,crawl4ai,scrapling,valkey,panel-api
volumes:
- ../apps/gateway:/opt/opencode-gateway/apps/gateway:ro,z
- ../packages:/opt/opencode-gateway/packages:ro,z
- gateway_data:/var/lib/opencode-gateway
networks:
- edge
- svc
depends_on:
- smokescreen
- valkey
smokescreen:
build:
context: ./smokescreen
dockerfile: Dockerfile
image: opencode-gateway/smokescreen:pinned
restart: unless-stopped
networks:
- svc
- egress-out
valkey:
image: valkey/valkey:latest
restart: unless-stopped
command: ["valkey-server", "--appendonly", "yes"]
healthcheck:
test: ["CMD-SHELL", "valkey-cli ping | grep -q PONG"]
interval: 30s
volumes:
- valkey_data:/data
networks:
- svc
networks:
edge:
name: gateway-edge
control:
name: gateway-control
internal: true
svc:
name: gateway-svc
internal: true
egress-out:
name: gateway-egress-out
volumes:
gateway_data:
name: opencode_gateway_data
valkey_data:
name: opencode_gateway_valkey
panel_web:
name: opencode_gateway_panel_web
caddy_data:
name: opencode_gateway_caddy_data
caddy_config:
name: opencode_gateway_caddy_config
@@ -1,38 +0,0 @@
# Smokescreen egress proxy snippet (opencode-gateway lane egress).
#
# All engine egress goes through Smokescreen; the gateway's own SSRF pre-check
# is the first layer, Smokescreen's per-connection resolution check the second
# (it also covers DNS-rebinding races the gateway cannot win).
#
# Build (no published image; pin the source commit):
# docker build -f deploy/smokescreen.Dockerfile -t gateway/smokescreen:pinned deploy/
#
# Run args (deny ranges on top of Smokescreen's publicly-routable-only default):
# --listen-ip 0.0.0.0 --listen-port 4750
# --deny-range 100.64.0.0/10
# --deny-range 169.254.0.0/16
# --deny-range fc00::/7
# --deny-range fe80::/10
#
# Lane proxy env (engines + fleet lanes):
# HTTP_PROXY=http://smokescreen:4750 HTTPS_PROXY=http://smokescreen:4750
# NO_PROXY=gateway,searxng,crawl4ai,scrapling,valkey # declared direct_peers only
#
# Loopback MUST NOT bypass the proxy except for container healthchecks
# (curl --noproxy '*'), or probes get 407 and flap.
FROM golang:1 AS build
ARG SMOKESCREEN_REF=609eb8931420453daf5893509be0b25b21bd9edb
RUN git clone https://github.com/stripe/smokescreen /src \
&& cd /src && git checkout "${SMOKESCREEN_REF}" \
&& CGO_ENABLED=0 go build -trimpath -o /out/smokescreen .
FROM gcr.io/distroless/static-debian12:nonroot
COPY --from=build /out/smokescreen /smokescreen
USER 65532:65532
EXPOSE 4750
ENTRYPOINT ["/smokescreen"]
CMD ["--listen-ip", "0.0.0.0", "--listen-port", "4750", \
"--deny-range", "100.64.0.0/10", \
"--deny-range", "169.254.0.0/16", \
"--deny-range", "fc00::/7", \
"--deny-range", "fe80::/10"]
@@ -0,0 +1,18 @@
# Smokescreen has no published image, so we build it from a pinned source commit (v0.1.0).
FROM golang:1 AS build
ARG SMOKESCREEN_REF=609eb8931420453daf5893509be0b25b21bd9edb
RUN git clone https://github.com/stripe/smokescreen /src \
&& cd /src && git checkout "${SMOKESCREEN_REF}" \
&& CGO_ENABLED=0 go build -trimpath -o /out/smokescreen .
FROM gcr.io/distroless/static-debian12:nonroot
COPY --from=build /out/smokescreen /smokescreen
USER 65532:65532
EXPOSE 4750
ENTRYPOINT ["/smokescreen"]
# Extra deny ranges on top of Smokescreen's own "publicly routable only" resolution check.
CMD ["--listen-ip", "0.0.0.0", "--listen-port", "4750", \
"--deny-range", "100.64.0.0/10", \
"--deny-range", "169.254.0.0/16", \
"--deny-range", "fc00::/7", \
"--deny-range", "fe80::/10"]