feat(oast): Collaborator-style tokens, DNS/SMTP/HTTP hits

Add durable OAST token minting and multi-protocol hit store with
API (/oast/tokens, /oast/hits), sc5 CLI (token create/hits, --insecure),
DNS dual mode (beacon|oast|both) with A/SOA/NS answers, SMTP log-only
listener, HTTP catch-all correlation, rate limits, and deployment docs
for oast.squidoffense.com delegation.

AGENTS.md

@@ -158,9 +158,15 @@ sc5 tasks get <id>
 sc5 tasks create <session_id> "<command>" [--args-json '{}'] [--hitl]
 
 sc5 listeners list
-sc5 listeners create <name> <port> [--kind http|tcp|reverse_shell] [--host 0.0.0.0]
+sc5 listeners create <name> <port> [--kind http|tcp|reverse_shell|dns|smtp] [--host 0.0.0.0] [--zone ZONE]
 sc5 listeners start|stop|delete <id>
 
+sc5 --insecure ...                  # skip TLS verify (self-signed teamserver)
+sc5 oast token create [--note "..."]
+sc5 oast tokens list
+sc5 oast hits --token T [--protocol dns|http|smtp]
+# aliases: oast mint | oast poll
+
 sc5 payloads templates
 sc5 payloads generate <template> <host> <port> [--interval 5] [--raw]
 # templates: http_beacon_python | http_beacon_bash | reverse_shell_bash | reverse_shell_python

docs/deployment.md

@@ -63,6 +63,69 @@ Disable unused services that might bind ports (ModemManager, etc.) on the host.
 
 Environment variables: see `.env.example` (`SQUIDC5_*`).
 
+## OAST Collaborator (DNS + HTTP + SMTP)
+
+Authorized out-of-band interaction capture (Burp Collaborator / Interactsh style).
+
+### Env
+
+| Var | Example | Purpose |
+|-----|---------|---------|
+| `SQUIDC5_OAST_ZONE` | `oast.squidoffense.com` | Authoritative OAST zone |
+| `SQUIDC5_PUBLIC_HOST` | `oast.squidoffense.com` | Hostnames in payload URLs |
+| `SQUIDC5_PUBLIC_IP` | `159.203.99.184` | A-record answers for DNS OAST |
+| `SQUIDC5_OAST_ENABLED` | `true` | Gate feature |
+| `SQUIDC5_OAST_HTTP_PORT` | `80` | Port shown in HTTP payload URLs |
+
+### DNS delegation (subdomain only — do not change apex NS)
+
+Example for `oast.squidoffense.com` → teamserver `159.203.99.184`:
+
+| Type | Name | Data |
+|------|------|------|
+| A | ns1.oast | 159.203.99.184 (glue) |
+| A | oast | 159.203.99.184 |
+| NS | oast | ns1.oast.squidoffense.com |
+| MX | oast | oast.squidoffense.com (priority 0) |
+
+Leave apex `@` A (website) and apex NS (registrar) alone.
+
+### Firewall ports (teamserver)
+
+Open at minimum: **53/udp, 53/tcp, 25/tcp, 80/tcp, 443/tcp, 8443/tcp**.
+
+Port 53 needs root or `ip_unprivileged_port_start=0`. Port 25 is often blocked by cloud providers — use 2525 for lab SMTP OAST if needed.
+
+### Listeners
+
+```bash
+# DNS OAST + beacon (mode both|oast|beacon)
+sc5 --insecure listeners create oast-dns 53 --kind dns --zone oast.squidoffense.com --dns-mode both
+sc5 --insecure listeners start <id>
+
+# HTTP OAST catch-all
+sc5 --insecure listeners create oast-http 80 --kind http
+sc5 --insecure listeners start <id>
+
+# SMTP OAST (enable feature first; never relays)
+# PUT /api/v1/features {"features":{"smtp_oast":true}}
+sc5 --insecure listeners create oast-smtp 25 --kind smtp
+```
+
+### Operator verify
+
+```bash
+sc5 --insecure login --url https://159.203.99.184:8443 --token sc5_...
+sc5 --insecure oast token create --note "xss"
+# TOKEN=... from response
+dig @8.8.8.8 $TOKEN.oast.squidoffense.com A
+curl -sS "http://oast.squidoffense.com/$TOKEN/"
+# swaks --to $TOKEN@oast.squidoffense.com --server 159.203.99.184
+sc5 --insecure oast hits --token $TOKEN
+```
+
+CLI: global `--insecure` / config `verify_ssl: false` for self-signed API TLS.
+
 ## TLS (HTTPS) — default on new instances
 
 On first start, SquidC5 generates a **unique self-signed certificate** under:

docs/operator-runbook.md

@@ -127,6 +127,23 @@ sc5 listeners start <id>
 sc5 payloads generate dns_beacon_python <DNS_HOST> 5353 --zone c2.lab.invalid --raw
 ```
 
+## OAST Collaborator (HTTP / DNS / SMTP)
+
+```bash
+# self-signed teamserver
+sc5 --insecure login --url https://TEAM:8443 --token sc5_...
+
+sc5 --insecure oast token create --note "sqli-oob"
+# → dns_name, http_url, smtp_to
+
+sc5 --insecure oast tokens list
+sc5 --insecure oast hits --token <TOKEN>
+sc5 --insecure oast hits --token <TOKEN> --protocol dns
+```
+
+DNS listener config: `--zone oast.example.com --dns-mode both` (or `oast` / `beacon`).  
+SMTP is log-only (feature `smtp_oast`, default off). See `docs/deployment.md` § OAST.
+
 Point lab zone NS/records at the C2 host (authorized lab only).
 
 ## WebSocket C2

src/squidc5/api/routes.py

@@ -26,12 +26,22 @@ class TokenCreate(BaseModel):
 
 class ListenerCreate(BaseModel):
     name: str
-    kind: str = "http"  # http | tcp | reverse_shell | dns
+    kind: str = "http"  # http | tcp | reverse_shell | dns | smtp
     host: str = "0.0.0.0"
     port: int
     config: dict[str, Any] = Field(default_factory=dict)
 
 
+class OastTokenCreate(BaseModel):
+    note: str = ""
+    label: str = ""  # alias for note
+    meta: dict[str, Any] = Field(default_factory=dict)
+
+
+# backward-compatible alias
+OastClientCreate = OastTokenCreate
+
+
 class TaskCreate(BaseModel):
     session_id: str
     command: str
@@ -407,6 +417,10 @@ def build_api_router() -> APIRouter:
         state = get_state(request)
         if body.kind == "http" and not await state.features.enabled("http_listeners"):
             raise HTTPException(403, "HTTP listeners disabled by feature flag")
+        if body.kind == "dns" and not await state.features.enabled("dns_listeners"):
+            raise HTTPException(403, "DNS listeners disabled by feature flag")
+        if body.kind == "smtp" and not await state.features.enabled("smtp_oast"):
+            raise HTTPException(403, "SMTP OAST disabled by feature flag")
         if body.kind in ("tcp", "reverse_shell") and not await state.features.enabled(
             "reverse_shell_listeners"
         ):
@@ -1389,6 +1403,120 @@ def build_api_router() -> APIRouter:
 
         return cert_rotation_plan(body.domains, body.days)
 
+    @api.post("/deploy/wildcard-cert-plan")
+    async def deploy_wildcard_cert_plan(
+        body: CertPlanRequest,
+        auth: AuthContext = Depends(require_scope("admin", "listeners:write")),
+    ) -> dict[str, Any]:
+        from squidc5.deploy.helpers import wildcard_cert_plan
+
+        return wildcard_cert_plan(body.domains, body.days)
+
+    # ----- OAST Collaborator -----
+
+    async def _oast_or_403(request: Request):
+        state = get_state(request)
+        if not state.settings.oast_enabled or not await state.features.enabled("oast_enabled"):
+            raise HTTPException(403, "OAST disabled")
+        if state.oast is None:
+            raise HTTPException(500, "OAST not initialized")
+        return state
+
+    @api.post("/oast/tokens")
+    async def oast_create_token(
+        body: OastTokenCreate,
+        request: Request,
+        auth: AuthContext = Depends(require_scope("oast:write", "admin")),
+    ) -> dict[str, Any]:
+        state = await _oast_or_403(request)
+        note = body.note or body.label or ""
+        return await state.oast.create_token(note=note, created_by=auth.name, meta=body.meta)
+
+    @api.get("/oast/tokens")
+    async def oast_list_tokens(
+        request: Request,
+        auth: AuthContext = Depends(require_scope("oast:read", "admin")),
+        limit: int = 100,
+    ) -> list[dict[str, Any]]:
+        state = await _oast_or_403(request)
+        return await state.oast.list_tokens(limit=limit)
+
+    @api.get("/oast/tokens/{token_id}")
+    async def oast_get_token(
+        token_id: str,
+        request: Request,
+        auth: AuthContext = Depends(require_scope("oast:read", "admin")),
+    ) -> dict[str, Any]:
+        state = await _oast_or_403(request)
+        c = await state.oast.get_token(token_id)
+        if not c:
+            raise HTTPException(404, "token not found")
+        return c
+
+    @api.get("/oast/hits")
+    async def oast_list_hits(
+        request: Request,
+        auth: AuthContext = Depends(require_scope("oast:read", "admin")),
+        token: str | None = None,
+        protocol: str | None = None,
+        client_id: str | None = None,
+        since: float | None = None,
+        limit: int = 100,
+    ) -> dict[str, Any]:
+        state = await _oast_or_403(request)
+        items = await state.oast.list_hits(
+            client_id=client_id,
+            token=token,
+            protocol=protocol,
+            since=since,
+            limit=limit,
+        )
+        return {"hits": items, "count": len(items)}
+
+    # aliases
+    @api.post("/oast/clients")
+    async def oast_create_client_alias(
+        body: OastTokenCreate,
+        request: Request,
+        auth: AuthContext = Depends(require_scope("oast:write", "admin")),
+    ) -> dict[str, Any]:
+        return await oast_create_token(body, request, auth)
+
+    @api.get("/oast/clients")
+    async def oast_list_clients_alias(
+        request: Request,
+        auth: AuthContext = Depends(require_scope("oast:read", "admin")),
+        limit: int = 100,
+    ) -> list[dict[str, Any]]:
+        return await oast_list_tokens(request, auth, limit)
+
+    @api.get("/oast/interactions")
+    async def oast_interactions_alias(
+        request: Request,
+        auth: AuthContext = Depends(require_scope("oast:read", "admin")),
+        client_id: str | None = None,
+        token: str | None = None,
+        protocol: str | None = None,
+        since: float | None = None,
+        limit: int = 100,
+    ) -> dict[str, Any]:
+        r = await oast_list_hits(
+            request, auth, token=token, protocol=protocol, client_id=client_id, since=since, limit=limit
+        )
+        return {"interactions": r["hits"], "count": r["count"]}
+
+    @api.delete("/oast/tokens/{token_id}")
+    async def oast_delete_token(
+        token_id: str,
+        request: Request,
+        auth: AuthContext = Depends(require_scope("oast:write", "admin")),
+    ) -> dict[str, str]:
+        state = await _oast_or_403(request)
+        ok = await state.oast.delete_client(token_id)
+        if not ok:
+            raise HTTPException(404, "token not found")
+        return {"status": "deleted", "id": token_id}
+
     # ----- Implant (no auth — session-bound beacon) -----
 
     implant = APIRouter(prefix="/implant", tags=["implant"])

src/squidc5/auth/tokens.py

@@ -36,6 +36,8 @@ SCOPES = frozenset(
         "profiles:write",
         "plugins:manage",
         "collab:use",
+        "oast:read",
+        "oast:write",
     }
 )
 

src/squidc5/cli.py

@@ -61,15 +61,40 @@ def pp(data: Any) -> None:
         print(data)
 
 
+def resolve_verify(args: argparse.Namespace) -> bool:
+    """TLS verify; False for self-signed teamservers (--insecure / verify_ssl:false)."""
+    if getattr(args, "insecure", False):
+        return False
+    env = os.environ.get("SQUIDC5_VERIFY_SSL") or os.environ.get("SC5_VERIFY_SSL")
+    if env is not None:
+        return env.strip().lower() not in ("0", "false", "no", "off")
+    cfg = load_config()
+    if "verify_ssl" in cfg:
+        return bool(cfg["verify_ssl"])
+    return True
+
+
 class Client:
-    def __init__(self, base: str, token: str | None, timeout: float = 30.0) -> None:
+    def __init__(
+        self,
+        base: str,
+        token: str | None,
+        timeout: float = 30.0,
+        *,
+        verify: bool = True,
+    ) -> None:
         self.base = base.rstrip("/")
         self.token = token
         headers: dict[str, str] = {"Accept": "application/json"}
         if token:
             headers["Authorization"] = f"Bearer {token}"
         # Default 30s; long reaps/broadcasts can override per-call
-        self._client = httpx.Client(base_url=self.base, headers=headers, timeout=timeout)
+        self._client = httpx.Client(
+            base_url=self.base,
+            headers=headers,
+            timeout=timeout,
+            verify=verify,
+        )
 
     def close(self) -> None:
         self._client.close()
@@ -108,16 +133,27 @@ def cmd_login(args: argparse.Namespace) -> None:
         cfg["url"] = args.url.rstrip("/")
     if args.token:
         cfg["token"] = args.token
+    if getattr(args, "insecure", False):
+        cfg["verify_ssl"] = False
     if not cfg.get("url"):
         cfg["url"] = DEFAULT_BASE
     if not cfg.get("token"):
-        raise SystemExit("Token required: sc5 login --token <token> [--url http://host:8443]")
+        raise SystemExit("Token required: sc5 login --token <token> [--url https://host:8443] [--insecure]")
     save_config(cfg)
-    client = Client(cfg["url"], cfg["token"])
+    verify = resolve_verify(args)
+    client = Client(cfg["url"], cfg["token"], verify=verify)
     try:
         health = client.get("/api/v1/health")
         meta = client.get("/api/v1/meta")
-        pp({"saved": str(CONFIG_FILE), "url": cfg["url"], "health": health, "actor": meta.get("actor")})
+        pp(
+            {
+                "saved": str(CONFIG_FILE),
+                "url": cfg["url"],
+                "verify_ssl": verify,
+                "health": health,
+                "actor": meta.get("actor"),
+            }
+        )
     finally:
         client.close()
 
@@ -224,7 +260,10 @@ def cmd_listeners_create(args: argparse.Namespace, client: Client) -> None:
         "port": args.port,
     }
     if args.kind == "dns":
-        body["config"] = {"zone": getattr(args, "zone", None) or "c2.lab.invalid"}
+        body["config"] = {
+            "zone": getattr(args, "zone", None) or "c2.lab.invalid",
+            "mode": getattr(args, "dns_mode", None) or "both",
+        }
     pp(client.post("/api/v1/listeners", json=body))
 
 
@@ -484,6 +523,34 @@ def cmd_events(args: argparse.Namespace, client: Client) -> None:
                 print(line)
 
 
+def cmd_oast_token_create(args: argparse.Namespace, client: Client) -> None:
+    note = getattr(args, "note", None) or getattr(args, "label", None) or ""
+    pp(client.post("/api/v1/oast/tokens", json={"note": note}))
+
+
+def cmd_oast_tokens_list(args: argparse.Namespace, client: Client) -> None:
+    pp(client.get("/api/v1/oast/tokens"))
+
+
+def cmd_oast_hits(args: argparse.Namespace, client: Client) -> None:
+    q: dict[str, Any] = {"limit": getattr(args, "limit", 100)}
+    if getattr(args, "token", None):
+        q["token"] = args.token
+    if getattr(args, "protocol", None):
+        q["protocol"] = args.protocol
+    if getattr(args, "client_id", None):
+        q["client_id"] = args.client_id
+    if getattr(args, "since", None) is not None:
+        q["since"] = args.since
+    pp(client.get("/api/v1/oast/hits", params=q))
+
+
+# aliases
+cmd_oast_mint = cmd_oast_token_create
+cmd_oast_list = cmd_oast_tokens_list
+cmd_oast_poll = cmd_oast_hits
+
+
 def cmd_tokens_list(args: argparse.Namespace, client: Client) -> None:
     pp(client.get("/api/v1/tokens"))
 
@@ -663,6 +730,12 @@ def build_parser() -> argparse.ArgumentParser:
     p.add_argument("--url", help="C2 base URL (or SQUIDC5_URL / config)")
     p.add_argument("--token", help="API token (or SQUIDC5_TOKEN / config)")
     p.add_argument("--timeout", type=float, default=30.0)
+    p.add_argument(
+        "--insecure",
+        "-k",
+        action="store_true",
+        help="Skip TLS certificate verify (self-signed teamservers)",
+    )
 
     sub = p.add_subparsers(dest="command", required=True)
 
@@ -765,9 +838,16 @@ def build_parser() -> argparse.ArgumentParser:
     l_create.add_argument(
         "--kind",
         default="http",
-        choices=["http", "tcp", "reverse_shell", "dns"],
+        choices=["http", "tcp", "reverse_shell", "dns", "smtp"],
     )
     l_create.add_argument("--zone", default=None, help="DNS zone when kind=dns")
+    l_create.add_argument(
+        "--dns-mode",
+        dest="dns_mode",
+        default="both",
+        choices=["beacon", "oast", "both"],
+        help="DNS listener mode (default both)",
+    )
     l_create.add_argument("--host", default="0.0.0.0")
     l_create.set_defaults(func=cmd_listeners_create, needs_client=True)
     l_start = lis_sub.add_parser("start")
@@ -964,6 +1044,39 @@ def build_parser() -> argparse.ArgumentParser:
     pol_set.add_argument("--file")
     pol_set.set_defaults(func=cmd_policy_set, needs_client=True)
 
+    # oast collaborator
+    oast = sub.add_parser("oast", help="OAST Collaborator (tokens + hits)")
+    oast_sub = oast.add_subparsers(dest="oast_cmd", required=True)
+    o_tok = oast_sub.add_parser("token", help="Token operations")
+    o_tok_sub = o_tok.add_subparsers(dest="oast_token_cmd", required=True)
+    o_tok_c = o_tok_sub.add_parser("create", help="Mint unique OAST payloads")
+    o_tok_c.add_argument("--note", default="", help="Operator note")
+    o_tok_c.set_defaults(func=cmd_oast_token_create, needs_client=True)
+    o_tokens = oast_sub.add_parser("tokens", help="List OAST tokens")
+    o_tokens_sub = o_tokens.add_subparsers(dest="oast_tokens_cmd", required=False)
+    o_tokens_list = o_tokens_sub.add_parser("list", help="List tokens")
+    o_tokens_list.set_defaults(func=cmd_oast_tokens_list, needs_client=True)
+    o_tokens.set_defaults(func=cmd_oast_tokens_list, needs_client=True)
+    o_hits = oast_sub.add_parser("hits", help="Poll OAST hits (Collaborator-style)")
+    o_hits.add_argument("--token")
+    o_hits.add_argument("--protocol", choices=["http", "dns", "smtp"])
+    o_hits.add_argument("--client-id", dest="client_id")
+    o_hits.add_argument("--since", type=float, default=None)
+    o_hits.add_argument("--limit", type=int, default=100)
+    o_hits.set_defaults(func=cmd_oast_hits, needs_client=True)
+    # aliases
+    o_mint = oast_sub.add_parser("mint", help="Alias: token create")
+    o_mint.add_argument("--note", default="")
+    o_mint.add_argument("--label", default="")
+    o_mint.set_defaults(func=cmd_oast_token_create, needs_client=True)
+    o_poll = oast_sub.add_parser("poll", help="Alias: hits")
+    o_poll.add_argument("--token")
+    o_poll.add_argument("--protocol", choices=["http", "dns", "smtp"])
+    o_poll.add_argument("--client-id", dest="client_id")
+    o_poll.add_argument("--since", type=float, default=None)
+    o_poll.add_argument("--limit", type=int, default=100)
+    o_poll.set_defaults(func=cmd_oast_hits, needs_client=True)
+
     return p
 
 
@@ -990,7 +1103,8 @@ def main(argv: list[str] | None = None) -> None:
             if args.command != "health":
                 raise SystemExit(1)
 
-    client = Client(base, token, timeout=args.timeout)
+    verify = resolve_verify(args)
+    client = Client(base, token, timeout=args.timeout, verify=verify)
     try:
         args.func(args, client)
     finally:

src/squidc5/config.py

@@ -39,8 +39,14 @@ class Settings(BaseSettings):
     shell_auto_stabilize: bool = True
     # Host/IP implants should call back to (defaults to request/local bind if empty)
     public_host: str = ""
+    public_ip: str = ""  # A-record for OAST DNS answers (SQUIDC5_PUBLIC_IP)
     shell_stabilize_delay_sec: float = 0.8
     shell_probe_wait_sec: float = 1.5
+    # OAST Collaborator (SQUIDC5_OAST_*)
+    oast_enabled: bool = True
+    oast_zone: str = "oast.lab.invalid"
+    oast_http_port: int = 80
+    oast_rate_limit_per_minute: int = 120
     # Hardened defaults
     expose_health_details: bool = False
     security_headers: bool = True

src/squidc5/core/state.py

@@ -16,6 +16,7 @@ if TYPE_CHECKING:
     from squidc5.implants.registry import ImplantRegistry
     from squidc5.listeners.manager import ListenerManager
     from squidc5.metrics.collector import MetricsCollector
+    from squidc5.oast.store import OastService
     from squidc5.observability.timeline import TimelineService
     from squidc5.payloads.generator import PayloadGenerator
     from squidc5.plugins.registry import PluginRegistry
@@ -44,6 +45,7 @@ class AppState:
     teams: TeamService
     plugins: PluginRegistry
     timeline: TimelineService
+    oast: OastService | None = None
     ai_chain: Any = None
     admin_token_once: str = ""
     shell_buffers: dict[str, list[str]] = field(default_factory=dict)

src/squidc5/db/store.py

@@ -165,6 +165,32 @@ CREATE TABLE IF NOT EXISTS team_members (
     added_at REAL NOT NULL,
     PRIMARY KEY (team_id, actor)
 );
+
+CREATE TABLE IF NOT EXISTS oast_clients (
+    id TEXT PRIMARY KEY,
+    token TEXT NOT NULL UNIQUE,
+    label TEXT NOT NULL DEFAULT '',
+    created_by TEXT,
+    meta TEXT NOT NULL DEFAULT '{}',
+    created_at REAL NOT NULL
+);
+
+CREATE TABLE IF NOT EXISTS oast_interactions (
+    id TEXT PRIMARY KEY,
+    client_id TEXT,
+    token TEXT,
+    protocol TEXT NOT NULL,
+    listener_id TEXT,
+    remote TEXT,
+    raw TEXT NOT NULL DEFAULT '{}',
+    correlation_key TEXT,
+    ts REAL NOT NULL
+);
+
+CREATE INDEX IF NOT EXISTS idx_oast_clients_token ON oast_clients(token);
+CREATE INDEX IF NOT EXISTS idx_oast_interactions_ts ON oast_interactions(ts);
+CREATE INDEX IF NOT EXISTS idx_oast_interactions_client ON oast_interactions(client_id);
+CREATE INDEX IF NOT EXISTS idx_oast_interactions_token ON oast_interactions(token);
 """
 
 def _now() -> float:
@@ -710,3 +736,103 @@ class Database:
             (team_id, actor),
         )
         return cur.rowcount > 0
+
+    # --- OAST (Collaborator-style) ---
+
+    async def create_oast_client(
+        self,
+        token: str,
+        label: str = "",
+        created_by: str | None = None,
+        meta: dict[str, Any] | None = None,
+    ) -> str:
+        cid = _uid("oast_")
+        await self.execute(
+            "INSERT INTO oast_clients (id, token, label, created_by, meta, created_at) "
+            "VALUES (?, ?, ?, ?, ?, ?)",
+            (cid, token, label or token, created_by, json.dumps(meta or {}), _now()),
+        )
+        return cid
+
+    async def get_oast_client(self, client_id: str) -> dict[str, Any] | None:
+        return await self.fetchone("SELECT * FROM oast_clients WHERE id = ?", (client_id,))
+
+    async def get_oast_client_by_token(self, token: str) -> dict[str, Any] | None:
+        return await self.fetchone("SELECT * FROM oast_clients WHERE token = ?", (token,))
+
+    async def list_oast_clients(self, limit: int = 100) -> list[dict[str, Any]]:
+        return await self.fetchall(
+            "SELECT * FROM oast_clients ORDER BY created_at DESC LIMIT ?",
+            (limit,),
+        )
+
+    async def delete_oast_client(self, client_id: str) -> bool:
+        await self.execute("DELETE FROM oast_interactions WHERE client_id = ?", (client_id,))
+        cur = await self.execute("DELETE FROM oast_clients WHERE id = ?", (client_id,))
+        return cur.rowcount > 0
+
+    async def create_oast_interaction(
+        self,
+        *,
+        client_id: str | None,
+        protocol: str,
+        listener_id: str | None,
+        remote: str | None,
+        raw: dict[str, Any],
+        correlation_key: str | None = None,
+        token: str | None = None,
+    ) -> str:
+        iid = _uid("hit_")
+        await self.execute(
+            "INSERT INTO oast_interactions "
+            "(id, client_id, token, protocol, listener_id, remote, raw, correlation_key, ts) "
+            "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
+            (
+                iid,
+                client_id,
+                token,
+                protocol,
+                listener_id,
+                remote,
+                json.dumps(raw)[:200_000],
+                correlation_key,
+                _now(),
+            ),
+        )
+        return iid
+
+    async def get_oast_interaction(self, interaction_id: str) -> dict[str, Any] | None:
+        return await self.fetchone(
+            "SELECT * FROM oast_interactions WHERE id = ?",
+            (interaction_id,),
+        )
+
+    async def list_oast_interactions(
+        self,
+        *,
+        client_id: str | None = None,
+        protocol: str | None = None,
+        since: float | None = None,
+        limit: int = 100,
+        token: str | None = None,
+    ) -> list[dict[str, Any]]:
+        clauses: list[str] = []
+        args: list[Any] = []
+        if client_id:
+            clauses.append("client_id = ?")
+            args.append(client_id)
+        if token:
+            clauses.append("token = ?")
+            args.append(token.lower())
+        if protocol:
+            clauses.append("protocol = ?")
+            args.append(protocol)
+        if since is not None:
+            clauses.append("ts > ?")
+            args.append(since)
+        where = (" WHERE " + " AND ".join(clauses)) if clauses else ""
+        args.append(min(max(limit, 1), 1000))
+        return await self.fetchall(
+            f"SELECT * FROM oast_interactions{where} ORDER BY ts DESC LIMIT ?",
+            tuple(args),
+        )

src/squidc5/deploy/helpers.py

@@ -61,3 +61,42 @@ def cert_rotation_plan(domains: list[str], days: int = 60) -> dict[str, Any]:
         ],
         "notes": "Never commit private keys. Store certs outside the git tree.",
     }
+
+
+def wildcard_cert_plan(domains: list[str], days: int = 60) -> dict[str, Any]:
+    """ACME DNS-01 wildcard plan for multi-protocol OAST (*.zone) + apex."""
+    names: list[str] = []
+    for d in domains:
+        d = d.strip().lstrip("*.")
+        if not d:
+            continue
+        names.append(d)
+        names.append(f"*.{d}")
+    # de-dupe preserve order
+    seen: set[str] = set()
+    ordered = []
+    for n in names:
+        if n not in seen:
+            seen.add(n)
+            ordered.append(n)
+    certbot_flags = " ".join(f"-d {n}" for n in ordered)
+    return {
+        "interval_days": days,
+        "domains": ordered,
+        "challenge": "dns-01",
+        "steps": [
+            "Delegate NS for OAST/C2 zone to this host (or provider API for DNS-01)",
+            f"Issue wildcard via ACME DNS-01: certbot certonly --manual --preferred-challenges dns {certbot_flags}",
+            "Or use DNS provider plugin (cloudflare, route53, etc.) for automation",
+            "Install fullchain.pem + privkey.pem on redirector / SQUIDC5_TLS_* paths",
+            "Point HTTP OAST + beacon Hostnames at apex; DNS OAST uses *.zone tokens",
+            "Correlate hits by oast token across http/dns/smtp in GET /api/v1/oast/interactions",
+            "Rotate before expiry; never commit private keys",
+        ],
+        "correlation": {
+            "http": "Host or path /o/{token}",
+            "dns": "{token}.zone query (any RR type logged)",
+            "smtp": "{token}@zone RCPT",
+        },
+        "notes": "Wildcard requires DNS-01. Lab script: scripts/acme_lab_renew.sh with -d '*.zone' -d zone.",
+    }

src/squidc5/features.py

@@ -25,6 +25,9 @@ DEFAULT_FEATURES: dict[str, bool] = {
     "plugins_enabled": False,  # deny by default until allow-listed
     "collab_teams": True,
     "observability_timeline": True,
+    "oast_enabled": True,
+    "dns_listeners": True,
+    "smtp_oast": False,  # deny by default (port 25 often restricted)
 }
 
 FEATURE_LABELS: dict[str, str] = {
@@ -44,6 +47,9 @@ FEATURE_LABELS: dict[str, str] = {
     "plugins_enabled": "Plugin registry (allow-list)",
     "collab_teams": "Multi-operator teams / handoff",
     "observability_timeline": "Timeline + ATT&CK mapping",
+    "oast_enabled": "OAST Collaborator (unique IDs + poll)",
+    "dns_listeners": "DNS C2 + DNS OAST proof logging",
+    "smtp_oast": "SMTP OAST listener",
 }
 
 

src/squidc5/listeners/dns_listener.py

@@ -1,4 +1,4 @@
-"""UDP DNS C2 listener — TXT/A check-ins for authorized lab use."""
+"""UDP DNS C2 + OAST — authoritative zone answers for authorized lab use."""
 
 from __future__ import annotations
 
@@ -6,7 +6,9 @@ import asyncio
 import base64
 import json
 import logging
+import re
 import struct
+import time
 from typing import TYPE_CHECKING, Any
 
 if TYPE_CHECKING:
@@ -14,6 +16,14 @@ if TYPE_CHECKING:
 
 log = logging.getLogger("squidc5.listeners.dns")
 
+_TOKEN_RE = re.compile(r"^[a-z0-9]{8,32}$")
+
+# DNS types
+T_A = 1
+T_NS = 2
+T_SOA = 6
+T_TXT = 16
+
 
 def _b32pad(s: str) -> str:
     s = s.replace("-", "").replace("_", "").upper()
@@ -25,11 +35,9 @@ def decode_label_payload(labels: list[str], zone_labels: list[str]) -> dict[str,
     """Extract beacon JSON from labels before zone: b.<b32chunks...>.zone."""
     if len(labels) <= len(zone_labels):
         return None
-    # strip zone suffix
     body = labels[: len(labels) - len(zone_labels)]
     if not body:
         return None
-    # first label is mode: b=beacon, r=result
     mode = body[0].lower() if body else ""
     chunks = body[1:] if mode in ("b", "r", "c") else body
     raw = "".join(chunks)
@@ -56,7 +64,6 @@ def parse_dns_query(data: bytes) -> tuple[int, list[str], int] | None:
     if len(data) < 12:
         return None
     txid = struct.unpack("!H", data[0:2])[0]
-    # skip flags, counts
     qdcount = struct.unpack("!H", data[4:6])[0]
     if qdcount < 1:
         return None
@@ -69,7 +76,7 @@ def parse_dns_query(data: bytes) -> tuple[int, list[str], int] | None:
             if ln == 0:
                 break
             if ln & 0xC0:
-                return None  # compression in QNAME unexpected
+                return None
             labels.append(data[pos : pos + ln].decode("ascii", errors="ignore"))
             pos += ln
         if pos + 4 > len(data):
@@ -80,12 +87,9 @@ def parse_dns_query(data: bytes) -> tuple[int, list[str], int] | None:
         return None
 
 
-def build_dns_response(txid: int, query: bytes, txt: str | None, *, nxdomain: bool = False) -> bytes:
-    """Build a simple DNS response echoing question; TXT answer or NXDOMAIN."""
-    # copy question from query
+def _question_slice(query: bytes) -> bytes:
     if len(query) < 12:
         return b""
-    # find end of question
     pos = 12
     while pos < len(query):
         ln = query[pos]
@@ -96,27 +100,93 @@ def build_dns_response(txid: int, query: bytes, txt: str | None, *, nxdomain: bo
             pos += 1
             break
         pos += ln
-    pos += 4  # qtype+qclass
-    question = query[12:pos]
-    flags = 0x8183 if nxdomain else 0x8180  # response, recursion available
-    ancount = 0 if nxdomain or not txt else 1
-    header = struct.pack("!HHHHHH", txid, flags, 1, ancount, 0, 0)
+    pos += 4
+    return query[12:pos]
+
+
+def _encode_name(labels: list[str]) -> bytes:
+    out = bytearray()
+    for lab in labels:
+        b = lab.encode("ascii", errors="ignore")[:63]
+        out.append(len(b))
+        out.extend(b)
+    out.append(0)
+    return bytes(out)
+
+
+def _ipv4_bytes(ip: str) -> bytes:
+    parts = ip.split(".")
+    if len(parts) != 4:
+        return bytes([127, 0, 0, 1])
+    try:
+        return bytes(int(p) & 0xFF for p in parts)
+    except ValueError:
+        return bytes([127, 0, 0, 1])
+
+
+def build_dns_response(
+    txid: int,
+    query: bytes,
+    txt: str | None,
+    *,
+    nxdomain: bool = False,
+) -> bytes:
+    """Legacy helper: TXT answer or NXDOMAIN."""
+    question = _question_slice(query)
+    flags = 0x8403 if nxdomain else 0x8400  # AA + response
     if nxdomain or not txt:
-        return header + question
-    # TXT answer: name pointer to question (0xC00C), type TXT, class IN, TTL, rdlength, txt
+        return struct.pack("!HHHHHH", txid, flags if nxdomain else 0x8400, 1, 0, 0, 0) + question
     name_ptr = b"\xc0\x0c"
     txt_bytes = txt.encode("ascii", errors="ignore")[:200]
     rdata = bytes([len(txt_bytes)]) + txt_bytes
-    answer = name_ptr + struct.pack("!HHIH", 16, 1, 60, len(rdata)) + rdata
+    answer = name_ptr + struct.pack("!HHIH", T_TXT, 1, 60, len(rdata)) + rdata
+    header = struct.pack("!HHHHHH", txid, 0x8400, 1, 1, 0, 0)
     return header + question + answer
 
 
+def build_dns_answers(
+    txid: int,
+    query: bytes,
+    answers: list[tuple[int, bytes]],
+    *,
+    nxdomain: bool = False,
+    aa: bool = True,
+) -> bytes:
+    """answers: list of (rtype, rdata)."""
+    question = _question_slice(query)
+    flags = 0x8000 | (0x0400 if aa else 0) | (0x0003 if nxdomain else 0)
+    ancount = 0 if nxdomain else len(answers)
+    header = struct.pack("!HHHHHH", txid, flags, 1, ancount, 0, 0)
+    if nxdomain or not answers:
+        return header + question
+    body = bytearray()
+    name_ptr = b"\xc0\x0c"
+    for rtype, rdata in answers:
+        body.extend(name_ptr)
+        body.extend(struct.pack("!HHIH", rtype, 1, 60, len(rdata)))
+        body.extend(rdata)
+    return header + question + bytes(body)
+
+
 class DnsProtocol(asyncio.DatagramProtocol):
-    def __init__(self, manager: ListenerManager, listener_id: str, zone: str) -> None:
+    def __init__(
+        self,
+        manager: ListenerManager,
+        listener_id: str,
+        zone: str,
+        *,
+        mode: str = "both",
+        public_ip: str = "127.0.0.1",
+        ns_name: str = "",
+    ) -> None:
         self.manager = manager
         self.listener_id = listener_id
         self.zone = zone.lower().strip(".")
         self.zone_labels = [p for p in self.zone.split(".") if p]
+        self.mode = (mode or "both").lower()  # beacon | oast | both
+        self.public_ip = public_ip or "127.0.0.1"
+        # ns1.<zone> by default for glue-friendly answers
+        self.ns_name = (ns_name or f"ns1.{self.zone}").lower().strip(".")
         self.transport: asyncio.DatagramTransport | None = None
 
     def connection_made(self, transport: asyncio.BaseTransport) -> None:
@@ -132,6 +202,32 @@ class DnsProtocol(asyncio.DatagramProtocol):
         except Exception:
             pass
 
+    def _is_c2_body(self, body: list[str]) -> bool:
+        if not body:
+            return False
+        return body[0] in ("b", "r", "c")
+
+    def _oast_answers(self, qtype: int) -> list[tuple[int, bytes]]:
+        answers: list[tuple[int, bytes]] = []
+        if qtype in (T_A, 255):  # A or ANY
+            answers.append((T_A, _ipv4_bytes(self.public_ip)))
+        if qtype in (T_NS, 255):
+            answers.append((T_NS, _encode_name(self.ns_name.split("."))))
+        if qtype in (T_SOA, 255):
+            # mname rname serial refresh retry expire minimum
+            mname = _encode_name(self.ns_name.split("."))
+            rname = _encode_name(["hostmaster"] + self.zone_labels)
+            soa = mname + rname + struct.pack("!IIIII", 1, 3600, 600, 86400, 60)
+            answers.append((T_SOA, soa))
+        if qtype == T_TXT:
+            answers.append((T_TXT, bytes([2]) + b"ok"))
+        if not answers and qtype == T_A:
+            answers.append((T_A, _ipv4_bytes(self.public_ip)))
+        if not answers:
+            # default A so scanners get something
+            answers.append((T_A, _ipv4_bytes(self.public_ip)))
+        return answers
+
     async def _handle(self, data: bytes, addr: tuple[str | Any, int]) -> None:
         if not self.transport:
             return
@@ -140,36 +236,98 @@ class DnsProtocol(asyncio.DatagramProtocol):
             return
         txid, labels, qtype = parsed
         labels_l = [x.lower() for x in labels]
-        # must end with zone
         if len(labels_l) < len(self.zone_labels) or labels_l[-len(self.zone_labels) :] != self.zone_labels:
-            self._send(build_dns_response(txid, data, None, nxdomain=True), addr)
+            self._send(build_dns_answers(txid, data, [], nxdomain=True), addr)
             return
-        payload = decode_label_payload(labels_l, self.zone_labels) or {}
-        mode = payload.pop("_mode", "b")
-        remote = addr[0] if addr else None
-        try:
-            if mode == "r":
-                result = await self.manager.handle_beacon_result(payload)
-            else:
-                result = await self.manager.handle_beacon(
-                    listener_id=self.listener_id,
-                    remote_addr=str(remote) if remote else None,
-                    payload=payload,
-                    user_agent="dns-c2",
-                )
-            txt = encode_txt_response(result)
-            if qtype == 16:  # TXT
-                resp = build_dns_response(txid, data, txt)
-            else:
-                resp = build_dns_response(txid, data, "ok")
-            self._send(resp, addr)
+
+        body = labels_l[: len(labels_l) - len(self.zone_labels)]
+        is_c2 = self._is_c2_body(body)
+        remote = str(addr[0]) if addr else None
+        qname = ".".join(labels_l)
+        oast = getattr(self.manager, "oast", None)
+
+        # rate limit OAST flood
+        if oast is not None and not oast.allow_remote(remote):
+            self._send(build_dns_answers(txid, data, [], nxdomain=True), addr)
+            return
+
+        # Beacon C2 path
+        if is_c2 and self.mode in ("beacon", "both"):
+            payload = decode_label_payload(labels_l, self.zone_labels) or {}
+            mode = payload.pop("_mode", "b")
             try:
+                if mode == "r":
+                    result = await self.manager.handle_beacon_result(payload)
+                else:
+                    result = await self.manager.handle_beacon(
+                        listener_id=self.listener_id,
+                        remote_addr=remote,
+                        payload=payload,
+                        user_agent="dns-c2",
+                    )
+                if oast is not None:
+                    ctok = None
+                    meta = payload.get("metadata") if isinstance(payload.get("metadata"), dict) else {}
+                    ctok = meta.get("oast_token") or payload.get("oast_token")
+                    if ctok:
+                        await oast.record(
+                            protocol="dns",
+                            listener_id=self.listener_id,
+                            remote=remote,
+                            token=str(ctok).lower(),
+                            raw={"qname": qname, "qtype": qtype, "c2": True, "mode": mode},
+                            correlation_key=str(ctok).lower(),
+                        )
+                txt = encode_txt_response(result)
+                if qtype == T_TXT:
+                    self._send(build_dns_response(txid, data, txt), addr)
+                else:
+                    self._send(build_dns_answers(txid, data, [(T_A, _ipv4_bytes(self.public_ip))]), addr)
+                await self.manager.metrics.incr("dns.queries")
+            except Exception:
+                log.exception("DNS C2 handler error from %s", addr)
+                self._send(build_dns_answers(txid, data, [], nxdomain=True), addr)
+            return
+
+        if is_c2 and self.mode == "oast":
+            # C2 labels ignored in oast-only mode — still log as OAST
+            pass
+
+        # OAST: any query under zone
+        if self.mode in ("oast", "both") or not is_c2:
+            token = None
+            if body and _TOKEN_RE.match(body[0]):
+                token = body[0]
+            elif body and not is_c2 and len(body[0]) >= 8:
+                token = body[0]
+            if oast is not None:
+                try:
+                    await oast.record(
+                        protocol="dns",
+                        listener_id=self.listener_id,
+                        remote=remote,
+                        token=token,
+                        raw={
+                            "qname": qname,
+                            "qtype": qtype,
+                            "labels": labels_l,
+                            "zone": self.zone,
+                            "ts": time.time(),
+                        },
+                        correlation_key=token,
+                    )
+                except Exception:
+                    log.exception("DNS OAST record failed")
+            answers = self._oast_answers(qtype)
+            self._send(build_dns_answers(txid, data, answers), addr)
+            try:
+                await self.manager.metrics.incr("dns.oast")
                 await self.manager.metrics.incr("dns.queries")
             except Exception:
                 pass
-        except Exception:
-            log.exception("DNS C2 handler error from %s", addr)
-            self._send(build_dns_response(txid, data, None, nxdomain=True), addr)
+            return
+
+        self._send(build_dns_answers(txid, data, [], nxdomain=True), addr)
 
 
 async def start_dns_server(
@@ -178,10 +336,21 @@ async def start_dns_server(
     host: str,
     port: int,
     zone: str,
+    *,
+    mode: str = "both",
+    public_ip: str = "127.0.0.1",
+    ns_name: str = "",
 ) -> tuple[asyncio.DatagramTransport, asyncio.BaseProtocol]:
     loop = asyncio.get_running_loop()
     transport, protocol = await loop.create_datagram_endpoint(
-        lambda: DnsProtocol(manager, listener_id, zone),
+        lambda: DnsProtocol(
+            manager,
+            listener_id,
+            zone,
+            mode=mode,
+            public_ip=public_ip,
+            ns_name=ns_name,
+        ),
         local_addr=(host, port),
     )
     return transport, protocol

src/squidc5/listeners/manager.py

@@ -58,6 +58,9 @@ class ListenerManager:
         self._verified: set[str] = set()
         # Optional: async (key) -> bool feature flag checker
         self.feature_check = None
+        # OAST Collaborator store (set from main)
+        self.oast = None
+        self.profile_engine = None
 
     async def create(
         self,
@@ -67,7 +70,7 @@ class ListenerManager:
         host: str = "0.0.0.0",
         config: dict[str, Any] | None = None,
     ) -> dict[str, Any]:
-        if kind not in ("http", "tcp", "reverse_shell", "dns"):
+        if kind not in ("http", "tcp", "reverse_shell", "dns", "smtp"):
             raise ValueError(f"Unsupported listener kind: {kind}")
         if port < 1 or port > 65535:
             raise ValueError("Port must be 1-65535")
@@ -112,11 +115,49 @@ class ListenerManager:
                 cfg = row.get("config") or {}
                 if isinstance(cfg, str):
                     cfg = json.loads(cfg)
-                zone = str((cfg or {}).get("zone") or "c2.lab.invalid")
-                transport, _proto = await start_dns_server(self, listener_id, host, port, zone)
+                cfg = cfg or {}
+                zone = str(cfg.get("zone") or getattr(self, "oast_zone", None) or "c2.lab.invalid")
+                mode = str(cfg.get("mode") or "both")  # beacon | oast | both
+                public_ip = str(
+                    cfg.get("public_ip")
+                    or getattr(self, "public_ip", None)
+                    or self.public_host
+                    or "127.0.0.1"
+                )
+                ns_name = str(cfg.get("ns_name") or f"ns1.{zone}")
+                transport, _proto = await start_dns_server(
+                    self,
+                    listener_id,
+                    host,
+                    port,
+                    zone,
+                    mode=mode,
+                    public_ip=public_ip,
+                    ns_name=ns_name,
+                )
                 self._udp[listener_id] = transport
                 await self.db.set_listener_status(listener_id, "running")
-                log.info("Started dns listener %s on %s:%s zone=%s", listener_id, host, port, zone)
+                log.info(
+                    "Started dns listener %s on %s:%s zone=%s mode=%s",
+                    listener_id,
+                    host,
+                    port,
+                    zone,
+                    mode,
+                )
+            elif kind == "smtp":
+                from squidc5.listeners.smtp_listener import handle_smtp_client
+
+                server = await asyncio.start_server(
+                    lambda r, w: handle_smtp_client(self, listener_id, r, w),
+                    host=host,
+                    port=port,
+                )
+                self._servers[listener_id] = server
+                task = asyncio.create_task(server.serve_forever(), name=f"listener-smtp-{listener_id}")
+                self._tasks[listener_id] = task
+                await self.db.set_listener_status(listener_id, "running")
+                log.info("Started smtp listener %s on %s:%s", listener_id, host, port)
             elif kind in ("tcp", "reverse_shell"):
                 server = await asyncio.start_server(
                     lambda r, w: self._handle_tcp(listener_id, kind, r, w),
@@ -892,6 +933,31 @@ class ListenerManager:
 
     async def record_http_hit(self, listener_id: str, hit: dict[str, Any]) -> None:
         """OAST-style catch-all for non-beacon HTTP requests."""
+        from squidc5.oast.store import (
+            extract_token_from_host,
+            extract_token_from_path,
+            extract_token_from_query,
+        )
+
+        path = str(hit.get("path") or "")
+        query = hit.get("query") if isinstance(hit.get("query"), dict) else {}
+        headers = hit.get("headers") if isinstance(hit.get("headers"), dict) else {}
+        zone = getattr(self, "oast_zone", "") or ""
+        token = (
+            extract_token_from_path(path)
+            or extract_token_from_query(query)
+            or extract_token_from_host(str(headers.get("host") or ""), zone=zone)
+        )
+        hit = {**hit, "token": token}
+        if self.oast is not None:
+            await self.oast.record(
+                protocol="http",
+                listener_id=listener_id,
+                remote=str(hit.get("remote") or ""),
+                token=token,
+                raw=hit,
+                correlation_key=token,
+            )
         await self.metrics.incr("http.hits")
         await self.metrics.emit("http.hit", hit)
         await self.db.audit(
@@ -902,7 +968,8 @@ class ListenerManager:
             details={
                 "remote": hit.get("remote"),
                 "method": hit.get("method"),
-                "path": str(hit.get("path", ""))[:200],
+                "path": path[:200],
+                "token": token,
             },
             risk_score=1,
         )

src/squidc5/listeners/smtp_listener.py

@@ -0,0 +1,162 @@
+"""Minimal SMTP OAST listener — log MAIL/RCPT/DATA only; never relay."""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+import re
+import time
+from typing import TYPE_CHECKING, Any
+
+if TYPE_CHECKING:
+    from squidc5.listeners.manager import ListenerManager
+
+log = logging.getLogger("squidc5.listeners.smtp")
+
+_TOKEN_RE = re.compile(r"([a-z0-9]{8,32})@", re.I)
+_LOCAL_RE = re.compile(r"^([a-z0-9]{8,32})$", re.I)
+
+
+def extract_smtp_token(rcpt: str, mail_from: str = "", data: str = "") -> str | None:
+    for src in (rcpt, mail_from):
+        m = _TOKEN_RE.search(src or "")
+        if m:
+            return m.group(1).lower()
+        local = (src or "").split("@", 1)[0].strip("<> ").lower()
+        if _LOCAL_RE.match(local):
+            return local
+    m = re.search(r"(?:id|oast|c)=([a-z0-9]{8,32})", data or "", re.I)
+    if m:
+        return m.group(1).lower()
+    return None
+
+
+async def handle_smtp_client(
+    manager: ListenerManager,
+    listener_id: str,
+    reader: asyncio.StreamReader,
+    writer: asyncio.StreamWriter,
+) -> None:
+    peer = writer.get_extra_info("peername")
+    remote = f"{peer[0]}:{peer[1]}" if peer else "unknown"
+    oast = getattr(manager, "oast", None)
+    if oast is not None and not oast.allow_remote(remote):
+        try:
+            await _send(writer, "421 rate limited")
+        except Exception:
+            pass
+        try:
+            writer.close()
+            await writer.wait_closed()
+        except Exception:
+            pass
+        return
+
+    mail_from = ""
+    rcpts: list[str] = []
+    try:
+        await _send(writer, "220 squidc5-oast ESMTP ready")
+        while True:
+            try:
+                line = await asyncio.wait_for(reader.readline(), timeout=60.0)
+            except TimeoutError:
+                break
+            if not line:
+                break
+            text = line.decode("utf-8", errors="replace").rstrip("\r\n")
+            upper = text.upper()
+            if upper.startswith("EHLO") or upper.startswith("HELO"):
+                await _send(writer, "250-squidc5-oast")
+                await _send(writer, "250 OK")
+            elif upper.startswith("MAIL FROM:"):
+                mail_from = text[10:].strip()
+                await _send(writer, "250 OK")
+            elif upper.startswith("RCPT TO:"):
+                rcpts.append(text[8:].strip())
+                await _send(writer, "250 OK")
+            elif upper == "DATA":
+                await _send(writer, "354 End data with <CR><LF>.<CR><LF>")
+                chunks: list[str] = []
+                while True:
+                    dline = await asyncio.wait_for(reader.readline(), timeout=120.0)
+                    if not dline:
+                        break
+                    s = dline.decode("utf-8", errors="replace")
+                    if s.rstrip("\r\n") == ".":
+                        break
+                    chunks.append(s)
+                data_buf = "".join(chunks)[:8192]
+                await _record(manager, listener_id, remote, mail_from, rcpts, data_buf)
+                await _send(writer, "250 OK queued")  # accepted + discarded (no relay)
+                mail_from, rcpts = "", []
+            elif upper in ("QUIT", "RSET"):
+                if upper == "RSET":
+                    mail_from, rcpts = "", []
+                    await _send(writer, "250 OK")
+                else:
+                    await _send(writer, "221 bye")
+                    break
+            elif upper.startswith("NOOP") or upper.startswith("VRFY") or upper.startswith("HELP"):
+                await _send(writer, "250 OK")
+            else:
+                await _send(writer, "250 OK")
+    except Exception:
+        log.exception("SMTP handler error from %s", remote)
+    finally:
+        try:
+            writer.close()
+            await writer.wait_closed()
+        except Exception:
+            pass
+
+
+async def _record(
+    manager: ListenerManager,
+    listener_id: str,
+    remote: str,
+    mail_from: str,
+    rcpts: list[str],
+    data: str,
+) -> None:
+    token = None
+    for r in rcpts:
+        token = extract_smtp_token(r, mail_from, data)
+        if token:
+            break
+    if not token:
+        token = extract_smtp_token("", mail_from, data)
+    hit: dict[str, Any] = {
+        "listener_id": listener_id,
+        "remote": remote,
+        "mail_from": mail_from[:500],
+        "rcpt_to": [r[:500] for r in rcpts[:20]],
+        "data_preview": data[:2000],
+        "token": token,
+        "ts": time.time(),
+        "relay": False,
+    }
+    oast = getattr(manager, "oast", None)
+    if oast is not None:
+        await oast.record(
+            protocol="smtp",
+            listener_id=listener_id,
+            remote=remote,
+            token=token,
+            raw=hit,
+            correlation_key=token,
+        )
+    await manager.metrics.incr("smtp.hits")
+    await manager.metrics.emit("smtp.hit", hit)
+    await manager.db.audit(
+        actor="implant",
+        actor_type="smtp_listener",
+        action="smtp.hit",
+        resource=listener_id,
+        details={"remote": remote, "token": token, "rcpt": (rcpts[0] if rcpts else "")[:200]},
+        risk_score=1,
+    )
+
+
+async def _send(writer: asyncio.StreamWriter, line: str) -> None:
+    writer.write((line + "\r\n").encode("utf-8"))
+    await writer.drain()

src/squidc5/main.py

@@ -26,6 +26,7 @@ from squidc5.implants.registry import ImplantRegistry
 from squidc5.listeners.manager import ListenerManager
 from squidc5.mcp.server import build_mcp_router
 from squidc5.metrics.collector import MetricsCollector
+from squidc5.oast.store import OastService
 from squidc5.observability.timeline import TimelineService
 from squidc5.paths import web_dir
 from squidc5.payloads.generator import PayloadGenerator
@@ -65,6 +66,15 @@ async def build_state(settings: Settings) -> AppState:
     plugins = PluginRegistry(db=db)
     await plugins.load_from_db()
     timeline = TimelineService(db)
+    oast = OastService(
+        db,
+        metrics,
+        zone=settings.oast_zone,
+        public_host=settings.public_host or settings.oast_zone,
+        public_ip=settings.public_ip or settings.public_host or "127.0.0.1",
+        http_port=settings.oast_http_port,
+        rate_limit=settings.oast_rate_limit_per_minute,
+    )
     from squidc5.ai.chain import AIChainRunner
 
     ai_chain = AIChainRunner(admin_ai, max_steps=3)
@@ -81,6 +91,9 @@ async def build_state(settings: Settings) -> AppState:
     )
     listeners.task_poll = tasks.poll
     listeners.task_complete = tasks.complete
+    listeners.oast = oast if settings.oast_enabled else None
+    listeners.oast_zone = settings.oast_zone
+    listeners.public_ip = settings.public_ip or settings.public_host
     sessions.interactive_check = listeners.is_live
     sessions.verified_check = listeners.is_verified
     sessions.exec_probe = listeners.probe_exec
@@ -116,6 +129,7 @@ async def build_state(settings: Settings) -> AppState:
         teams=teams,
         plugins=plugins,
         timeline=timeline,
+        oast=oast,
         ai_chain=ai_chain,
         admin_token_once=admin_once,
     )

src/squidc5/oast/__init__.py

@@ -0,0 +1,5 @@
+"""Out-of-band application security testing (Collaborator-style) interactions."""
+
+from squidc5.oast.store import OastService, extract_token_from_host, extract_token_from_path
+
+__all__ = ["OastService", "extract_token_from_host", "extract_token_from_path"]

src/squidc5/oast/store.py

@@ -0,0 +1,341 @@
+"""OAST tokens (unique payload IDs) and multi-protocol hit store."""
+
+from __future__ import annotations
+
+import json
+import re
+import secrets
+import time
+from collections import defaultdict, deque
+from typing import Any
+
+from squidc5.db.store import Database
+from squidc5.metrics.collector import MetricsCollector
+
+_TOKEN_RE = re.compile(r"^[a-z0-9]{8,32}$")
+_PATH_TOKEN_RE = re.compile(
+    r"(?:^|/)(?:o|oast|c|p|id)/([a-z0-9]{8,32})(?:/|$)",
+    re.IGNORECASE,
+)
+
+
+def mint_token(nbytes: int = 6) -> str:
+    """8–12 hex chars, dns/url-safe."""
+    return secrets.token_hex(nbytes)
+
+
+def extract_token_from_path(path: str) -> str | None:
+    if not path:
+        return None
+    m = _PATH_TOKEN_RE.search(path.split("?", 1)[0])
+    if m:
+        return m.group(1).lower()
+    segs = [s for s in path.split("?", 1)[0].strip("/").split("/") if s]
+    if segs and _TOKEN_RE.match(segs[0].lower()):
+        return segs[0].lower()
+    return None
+
+
+def extract_token_from_host(host: str, zone: str = "") -> str | None:
+    if not host:
+        return None
+    h = host.split(":")[0].lower().strip(".")
+    if zone:
+        z = zone.lower().strip(".")
+        if h == z or h.endswith("." + z):
+            left = h[: -(len(z) + 1)] if h != z else ""
+            if left:
+                first = left.split(".")[0]
+                if _TOKEN_RE.match(first):
+                    return first
+    first = h.split(".")[0]
+    if _TOKEN_RE.match(first):
+        return first
+    return None
+
+
+def extract_token_from_query(query: dict[str, Any]) -> str | None:
+    for key in ("c", "id", "oast", "token", "cid"):
+        v = query.get(key)
+        if isinstance(v, list):
+            v = v[0] if v else None
+        if isinstance(v, str) and _TOKEN_RE.match(v.lower()):
+            return v.lower()
+    return None
+
+
+def strip_secrets(headers: dict[str, Any]) -> dict[str, Any]:
+    out = {}
+    for k, v in headers.items():
+        if k.lower() in ("authorization", "cookie", "set-cookie", "proxy-authorization"):
+            continue
+        out[k] = v
+    return out
+
+
+class RateLimiter:
+    """Simple per-IP sliding window (in-memory)."""
+
+    def __init__(self, limit: int = 120, window_sec: float = 60.0) -> None:
+        self.limit = max(1, limit)
+        self.window = window_sec
+        self._hits: dict[str, deque[float]] = defaultdict(deque)
+
+    def allow(self, key: str) -> bool:
+        now = time.time()
+        q = self._hits[key]
+        while q and now - q[0] > self.window:
+            q.popleft()
+        if len(q) >= self.limit:
+            return False
+        q.append(now)
+        return True
+
+
+class OastService:
+    def __init__(
+        self,
+        db: Database,
+        metrics: MetricsCollector | None = None,
+        *,
+        zone: str = "oast.lab.invalid",
+        public_host: str = "",
+        public_ip: str = "",
+        http_port: int = 80,
+        scheme: str = "http",
+        rate_limit: int = 120,
+    ) -> None:
+        self.db = db
+        self.metrics = metrics
+        self.zone = (zone or "oast.lab.invalid").lower().strip(".")
+        self.public_host = public_host or self.zone
+        self.public_ip = public_ip or "127.0.0.1"
+        self.http_port = http_port
+        self.scheme = scheme
+        self.rate = RateLimiter(limit=rate_limit)
+
+    def allow_remote(self, remote: str | None) -> bool:
+        ip = (remote or "unknown").split(":")[0]
+        return self.rate.allow(ip)
+
+    async def create_token(
+        self,
+        *,
+        note: str = "",
+        created_by: str | None = None,
+        meta: dict[str, Any] | None = None,
+    ) -> dict[str, Any]:
+        token = mint_token()
+        for _ in range(8):
+            if not await self.db.get_oast_client_by_token(token):
+                break
+            token = mint_token()
+        m = dict(meta or {})
+        if note:
+            m["note"] = note
+        cid = await self.db.create_oast_client(
+            token=token,
+            label=note or token,
+            created_by=created_by,
+            meta=m,
+        )
+        await self.db.audit(
+            actor=created_by or "operator",
+            actor_type="operator",
+            action="oast.token.created",
+            resource=cid,
+            details={"token": token, "note": note[:200]},
+            risk_score=0,
+        )
+        if self.metrics:
+            await self.metrics.emit("oast.token.created", {"id": cid, "token": token})
+        return self.format_token_response(cid, token, note=note)
+
+    # aliases used by older routes
+    async def create_client(self, **kwargs: Any) -> dict[str, Any]:
+        note = kwargs.get("label") or kwargs.get("note") or ""
+        return await self.create_token(
+            note=str(note),
+            created_by=kwargs.get("created_by"),
+            meta=kwargs.get("meta"),
+        )
+
+    def format_token_response(
+        self, client_id: str, token: str, *, note: str = ""
+    ) -> dict[str, Any]:
+        zone = self.zone
+        host = self.public_host or zone
+        port_s = f":{self.http_port}" if self.http_port not in (80, 443) else ""
+        dns_name = f"{token}.{zone}"
+        http_url = f"{self.scheme}://{token}.{zone}{port_s}/"
+        http_path = f"{self.scheme}://{host}{port_s}/{token}/"
+        smtp_to = f"{token}@{zone}"
+        return {
+            "id": client_id,
+            "token": token,
+            "note": note,
+            "dns_name": dns_name,
+            "http_url": http_url,
+            "http_url_path": http_path,
+            "smtp_to": smtp_to,
+            "payloads": {
+                "dns": dns_name,
+                "http": http_url,
+                "http_path": http_path,
+                "smtp": smtp_to,
+                "token": token,
+            },
+            "zone": zone,
+            "public_ip": self.public_ip,
+        }
+
+    async def list_tokens(self, limit: int = 100) -> list[dict[str, Any]]:
+        rows = await self.db.list_oast_clients(limit=limit)
+        out = []
+        for r in rows:
+            n = self._norm_client(r)
+            note = str((n.get("meta") or {}).get("note") or n.get("label") or "")
+            out.append(self.format_token_response(n["id"], n["token"], note=note))
+        return out
+
+    async def get_token(self, token_id: str) -> dict[str, Any] | None:
+        row = await self.db.get_oast_client(token_id)
+        if not row:
+            return None
+        n = self._norm_client(row)
+        note = str((n.get("meta") or {}).get("note") or n.get("label") or "")
+        return self.format_token_response(n["id"], n["token"], note=note)
+
+    async def list_clients(self, limit: int = 100) -> list[dict[str, Any]]:
+        return await self.list_tokens(limit=limit)
+
+    async def get_client(self, client_id: str) -> dict[str, Any] | None:
+        return await self.get_token(client_id)
+
+    async def get_by_token(self, token: str) -> dict[str, Any] | None:
+        row = await self.db.get_oast_client_by_token(token.lower())
+        if not row:
+            return None
+        n = self._norm_client(row)
+        note = str((n.get("meta") or {}).get("note") or n.get("label") or "")
+        return self.format_token_response(n["id"], n["token"], note=note)
+
+    async def resolve_token(self, token: str | None) -> str | None:
+        if not token:
+            return None
+        row = await self.db.get_oast_client_by_token(token.lower())
+        return str(row["id"]) if row else None
+
+    async def record(
+        self,
+        *,
+        protocol: str,
+        listener_id: str | None = None,
+        remote: str | None = None,
+        token: str | None = None,
+        client_id: str | None = None,
+        raw: dict[str, Any] | None = None,
+        correlation_key: str | None = None,
+    ) -> dict[str, Any]:
+        if not client_id and token:
+            client_id = await self.resolve_token(token)
+        summary = raw or {}
+        if isinstance(summary.get("headers"), dict):
+            summary = {**summary, "headers": strip_secrets(summary["headers"])}
+        iid = await self.db.create_oast_interaction(
+            client_id=client_id,
+            protocol=protocol,
+            listener_id=listener_id,
+            remote=remote,
+            raw=summary,
+            correlation_key=correlation_key or token,
+            token=token,
+        )
+        await self.db.audit(
+            actor="oast",
+            actor_type="listener",
+            action="oast.hit",
+            resource=iid,
+            details={
+                "protocol": protocol,
+                "token": token,
+                "remote": (remote or "")[:80],
+                "client_id": client_id,
+            },
+            risk_score=1,
+        )
+        if self.metrics:
+            await self.metrics.incr("oast.hits")
+            await self.metrics.incr("oast.interactions")
+            await self.metrics.emit(
+                "oast.hit",
+                {
+                    "id": iid,
+                    "protocol": protocol,
+                    "client_id": client_id,
+                    "token": token,
+                    "remote": remote,
+                    "listener_id": listener_id,
+                },
+            )
+        row = await self.db.get_oast_interaction(iid)
+        return self._norm_hit(row)  # type: ignore[arg-type]
+
+    async def list_hits(
+        self,
+        *,
+        client_id: str | None = None,
+        token: str | None = None,
+        protocol: str | None = None,
+        since: float | None = None,
+        limit: int = 100,
+    ) -> list[dict[str, Any]]:
+        if token and not client_id:
+            c = await self.db.get_oast_client_by_token(token.lower())
+            if c:
+                client_id = str(c["id"])
+            # also match by token column even if unregistered
+        rows = await self.db.list_oast_interactions(
+            client_id=client_id,
+            protocol=protocol,
+            since=since,
+            limit=limit,
+            token=token if not client_id else None,
+        )
+        return [self._norm_hit(r) for r in rows]
+
+    async def poll(self, **kwargs: Any) -> list[dict[str, Any]]:
+        return await self.list_hits(**kwargs)
+
+    async def delete_client(self, client_id: str) -> bool:
+        return await self.db.delete_oast_client(client_id)
+
+    def payload_urls(self, token: str, **kwargs: Any) -> dict[str, str]:
+        r = self.format_token_response("x", token)
+        return r["payloads"]
+
+    def _norm_client(self, row: dict[str, Any]) -> dict[str, Any]:
+        out = dict(row)
+        meta = out.get("meta")
+        if isinstance(meta, str):
+            try:
+                out["meta"] = json.loads(meta)
+            except Exception:
+                out["meta"] = {}
+        return out
+
+    def _norm_hit(self, row: dict[str, Any]) -> dict[str, Any]:
+        out = dict(row)
+        raw = out.get("raw")
+        if isinstance(raw, str):
+            try:
+                out["raw"] = json.loads(raw)
+            except Exception:
+                out["raw"] = {"raw": raw}
+        # Collaborator-style aliases
+        out["summary"] = out.get("raw")
+        out["remote_ip"] = out.get("remote")
+        return out
+
+    def _norm_interaction(self, row: dict[str, Any]) -> dict[str, Any]:
+        return self._norm_hit(row)

tests/test_oast.py

@@ -0,0 +1,216 @@
+"""OAST Collaborator: tokens, HTTP/DNS/SMTP hits, correlation."""
+
+from __future__ import annotations
+
+import asyncio
+import struct
+
+import pytest
+
+from squidc5.listeners.dns_listener import build_dns_answers, build_dns_response, parse_dns_query
+from squidc5.listeners.smtp_listener import extract_smtp_token
+from squidc5.oast.store import (
+    extract_token_from_host,
+    extract_token_from_path,
+    extract_token_from_query,
+    mint_token,
+)
+
+
+def test_mint_and_extract_token():
+    t = mint_token()
+    assert 8 <= len(t) <= 32
+    assert extract_token_from_path(f"/o/{t}") == t
+    assert extract_token_from_path(f"/{t}/x") == t
+    assert extract_token_from_query({"c": t}) == t
+    assert extract_token_from_host(f"{t}.oast.squidoffense.com", "oast.squidoffense.com") == t
+
+
+def test_extract_smtp_token():
+    assert extract_smtp_token("abcd1234ef@oast.squidoffense.com") == "abcd1234ef"
+    assert extract_smtp_token("<deadbeefcafe@x.invalid>") == "deadbeefcafe"
+
+
+@pytest.mark.asyncio
+async def test_oast_token_api_shape(client, admin_headers):
+    r = await client.post(
+        "/api/v1/oast/tokens",
+        headers=admin_headers,
+        json={"note": "xss-lab"},
+    )
+    assert r.status_code == 200
+    body = r.json()
+    assert body["token"]
+    assert body["dns_name"].endswith(body["token"] + "." + body["dns_name"].split(".", 1)[-1]) or body[
+        "dns_name"
+    ].startswith(body["token"])
+    assert body["token"] in body["dns_name"]
+    assert body["token"] in body["http_url"]
+    assert body["smtp_to"].startswith(body["token"] + "@")
+    lst = await client.get("/api/v1/oast/tokens", headers=admin_headers)
+    assert lst.status_code == 200
+    assert any(x["id"] == body["id"] for x in lst.json())
+
+
+@pytest.mark.asyncio
+async def test_oast_http_hit(client, admin_headers):
+    r = await client.post(
+        "/api/v1/oast/tokens",
+        headers=admin_headers,
+        json={"note": "http"},
+    )
+    token = r.json()["token"]
+
+    lr = await client.post(
+        "/api/v1/listeners",
+        headers=admin_headers,
+        json={"name": "oast-http", "kind": "http", "port": 19050},
+    )
+    lid = lr.json()["id"]
+    await client.post(f"/api/v1/listeners/{lid}/start", headers=admin_headers)
+
+    reader, writer = await asyncio.open_connection("127.0.0.1", 19050)
+    writer.write(f"GET /{token}/ HTTP/1.1\r\nHost: 127.0.0.1\r\nConnection: close\r\n\r\n".encode())
+    await writer.drain()
+    data = await reader.read(4096)
+    writer.close()
+    try:
+        await writer.wait_closed()
+    except Exception:
+        pass
+    assert b"200" in data
+
+    poll = await client.get(f"/api/v1/oast/hits?token={token}", headers=admin_headers)
+    assert poll.status_code == 200
+    hits = poll.json()["hits"]
+    assert any(h.get("protocol") == "http" and h.get("token") == token for h in hits)
+
+    await client.post(f"/api/v1/listeners/{lid}/stop", headers=admin_headers)
+
+
+@pytest.mark.asyncio
+async def test_dns_oast_subdomain(client, admin_headers):
+    r = await client.post("/api/v1/oast/tokens", headers=admin_headers, json={"note": "dns"})
+    token = r.json()["token"]
+
+    lr = await client.post(
+        "/api/v1/listeners",
+        headers=admin_headers,
+        json={
+            "name": "oast-dns",
+            "kind": "dns",
+            "port": 19553,
+            "config": {"zone": "oast.lab.invalid", "mode": "oast", "public_ip": "159.203.99.184"},
+        },
+    )
+    lid = lr.json()["id"]
+    await client.post(f"/api/v1/listeners/{lid}/start", headers=admin_headers)
+
+    labels = [token, "oast", "lab", "invalid"]
+    q = bytearray(struct.pack("!HHHHHH", 0x1234, 0x0100, 1, 0, 0, 0))
+    for lab in labels:
+        b = lab.encode("ascii")
+        q.append(len(b))
+        q.extend(b)
+    q.append(0)
+    q += struct.pack("!HH", 1, 1)  # A IN
+
+    loop = asyncio.get_running_loop()
+
+    class _P(asyncio.DatagramProtocol):
+        def __init__(self):
+            self.got = asyncio.Event()
+            self.data = b""
+
+        def datagram_received(self, data, addr):
+            self.data = data
+            self.got.set()
+
+    transport, proto = await loop.create_datagram_endpoint(_P, local_addr=("127.0.0.1", 0))
+    transport.sendto(bytes(q), ("127.0.0.1", 19553))
+    try:
+        await asyncio.wait_for(proto.got.wait(), timeout=2.0)
+        # A answer should include public IP bytes
+        assert b"\x9f\xcb\x63\xb8" in proto.data  # 159.203.99.184
+    finally:
+        transport.close()
+
+    poll = await client.get(
+        f"/api/v1/oast/hits?token={token}&protocol=dns",
+        headers=admin_headers,
+    )
+    assert poll.json()["count"] >= 1
+
+    await client.post(f"/api/v1/listeners/{lid}/stop", headers=admin_headers)
+
+
+@pytest.mark.asyncio
+async def test_smtp_oast(client, admin_headers):
+    await client.put(
+        "/api/v1/features",
+        headers=admin_headers,
+        json={"features": {"smtp_oast": True}},
+    )
+    r = await client.post("/api/v1/oast/tokens", headers=admin_headers, json={"note": "smtp"})
+    token = r.json()["token"]
+
+    lr = await client.post(
+        "/api/v1/listeners",
+        headers=admin_headers,
+        json={"name": "oast-smtp", "kind": "smtp", "port": 19025},
+    )
+    assert lr.status_code == 200, lr.text
+    lid = lr.json()["id"]
+    st = await client.post(f"/api/v1/listeners/{lid}/start", headers=admin_headers)
+    assert st.status_code == 200
+
+    reader, writer = await asyncio.open_connection("127.0.0.1", 19025)
+    assert (await reader.readline()).startswith(b"220")
+    writer.write(b"EHLO test\r\n")
+    await writer.drain()
+    while True:
+        line = await reader.readline()
+        if line.startswith(b"250 ") and not line.startswith(b"250-"):
+            break
+    writer.write(b"MAIL FROM:<a@b.invalid>\r\n")
+    await writer.drain()
+    await reader.readline()
+    writer.write(f"RCPT TO:<{token}@oast.lab.invalid>\r\n".encode())
+    await writer.drain()
+    await reader.readline()
+    writer.write(b"DATA\r\n")
+    await writer.drain()
+    await reader.readline()
+    writer.write(b"Subject: x\r\n\r\nhello\r\n.\r\n")
+    await writer.drain()
+    assert b"250" in await reader.readline()
+    writer.write(b"QUIT\r\n")
+    await writer.drain()
+    writer.close()
+    try:
+        await writer.wait_closed()
+    except Exception:
+        pass
+
+    poll = await client.get(
+        f"/api/v1/oast/hits?token={token}&protocol=smtp",
+        headers=admin_headers,
+    )
+    assert poll.json()["count"] >= 1
+    await client.post(f"/api/v1/listeners/{lid}/stop", headers=admin_headers)
+
+
+def test_parse_dns_roundtrip():
+    assert parse_dns_query(b"") is None
+    q = struct.pack("!HHHHHH", 1, 0x0100, 1, 0, 0, 0)
+    for lab in (b"a", b"b"):
+        q += bytes([len(lab)]) + lab
+    q += b"\x00" + struct.pack("!HH", 1, 1)
+    parsed = parse_dns_query(q)
+    assert parsed is not None
+    txid, labels, qtype = parsed
+    assert labels == ["a", "b"]
+    resp = build_dns_response(txid, q, "ok")
+    assert len(resp) > 12
+    resp2 = build_dns_answers(txid, q, [(1, bytes([1, 2, 3, 4]))])
+    assert b"\x01\x02\x03\x04" in resp2