feat(tarpit): port-selective tc netem tarpit mode with live log events

- GET/POST/DELETE /api/v1/deckies/{name}/tarpit (admin write, viewer GET)
- get_container_veth() + get_container_pid() in network.py via iflink/ip-link
- TarpitRule SQLModel table + TarpitMixin repo (upsert/get/delete/list)
- Background tarpit_watcher_worker: polls /proc/{pid}/net/tcp every 15s,
  emits tarpit_enter/tarpit_exit log events (edge-triggered, with duration)
- tarpit_enabled/tarpit_disabled logs on operator POST/DELETE actions

decnet/network.py

@@ -411,3 +411,47 @@ def ips_to_range(ips: list[str]) -> str:
         strict=False,
     )
     return str(network)
+
+
+# ---------------------------------------------------------------------------
+# Container veth resolution (for tc netem tarpit)
+# ---------------------------------------------------------------------------
+
+def get_container_pid(container_name: str) -> int:
+    """Return the PID of a running container's init process."""
+    client = docker.from_env()
+    try:
+        container = client.containers.get(container_name)
+    except docker.errors.NotFound:
+        raise LookupError(f"container {container_name!r} not found")
+    pid = container.attrs["State"]["Pid"]
+    if not pid:
+        raise LookupError(f"container {container_name!r} is not running (PID=0)")
+    return pid
+
+
+def get_container_veth(container_name: str) -> str:
+    """Return the host veth interface name paired to container_name's eth0.
+
+    Reads /sys/class/net/eth0/iflink from inside the container to get the
+    peer interface index, then matches it against ``ip link show`` on the host.
+    Requires no nsenter and no elevated privileges beyond what Docker exec grants.
+    """
+    result = _run(
+        ["docker", "exec", container_name, "cat", "/sys/class/net/eth0/iflink"],
+        check=False,
+    )
+    if result.returncode != 0:
+        raise LookupError(
+            f"container {container_name!r} not reachable: {result.stderr.strip()}"
+        )
+    peer_index = result.stdout.strip()
+    links = _run(["ip", "link", "show"])
+    for line in links.stdout.splitlines():
+        if line.startswith(f"{peer_index}:"):
+            # Format: "42: veth3a4b5c@if41: <BROADCAST,...>"
+            iface = line.split(":")[1].strip().split("@")[0]
+            return iface
+    raise LookupError(
+        f"no host veth found for container {container_name!r} (peer ifindex {peer_index})"
+    )

decnet/tarpit/__init__.py

@@ -0,0 +1,3 @@
+from .worker import tarpit_watcher_worker
+
+__all__ = ["tarpit_watcher_worker"]

decnet/tarpit/worker.py

@@ -0,0 +1,191 @@
+"""Tarpit connection watcher — edge-triggered enter/exit log events.
+
+Polls active tarpit rules every ``DECNET_TARPIT_POLL_INTERVAL`` seconds
+(default 15).  For each rule, reads ``/proc/{pid}/net/tcp`` on the host
+(no docker exec, no ss needed inside the container) to find ESTABLISHED
+connections on the tarpitted ports.  Emits structured log events:
+
+* ``tarpit_enter`` — new connection seen on a tarpitted port
+* ``tarpit_exit``  — connection gone; includes elapsed time in seconds
+
+Runs embedded in the API process (always-on, near-zero cost when no
+rules exist).
+"""
+from __future__ import annotations
+
+import asyncio
+import json
+import socket
+from datetime import datetime, timezone
+from typing import Any, Optional
+
+from decnet.logging import get_logger
+from decnet.network import get_container_pid
+from decnet.web.db.repository import BaseRepository
+
+log = get_logger("tarpit.watcher")
+
+_POLL_INTERVAL_ENV = "DECNET_TARPIT_POLL_INTERVAL"
+_DEFAULT_POLL_S = 15
+
+_TCP_ESTABLISHED = "01"
+
+
+def _read_proc_net_tcp(pid: int) -> str:
+    """Read /proc/{pid}/net/tcp from the host (namespace-aware symlink)."""
+    path = f"/proc/{pid}/net/tcp"
+    try:
+        with open(path) as f:
+            return f.read()
+    except OSError:
+        return ""
+
+
+def _parse_connections(content: str, target_port: int) -> list[str]:
+    """Return list of remote IPs in ESTABLISHED state on target_port."""
+    ips: list[str] = []
+    for line in content.strip().splitlines()[1:]:
+        parts = line.split()
+        if len(parts) < 4:
+            continue
+        local_hex, rem_hex, state = parts[1], parts[2], parts[3]
+        if state != _TCP_ESTABLISHED:
+            continue
+        local_port = int(local_hex.split(":")[1], 16)
+        if local_port != target_port:
+            continue
+        rem_ip_hex = rem_hex.split(":")[0]
+        try:
+            ip_bytes = bytes.fromhex(rem_ip_hex)[::-1]
+            ip = socket.inet_ntoa(ip_bytes)
+        except (ValueError, OSError):
+            continue
+        if ip != "0.0.0.0":  # nosec B104
+            ips.append(ip)
+    return ips
+
+
+def _get_poll_interval() -> int:
+    import os
+    try:
+        return int(os.environ.get(_POLL_INTERVAL_ENV, _DEFAULT_POLL_S))
+    except (TypeError, ValueError):
+        return _DEFAULT_POLL_S
+
+
+async def _get_attacker_uuid(repo: BaseRepository, ip: str) -> Optional[str]:
+    try:
+        from decnet.web.db.models import Attacker
+        from sqlalchemy import select
+        async with repo._session() as session:  # type: ignore[attr-defined]
+            result = await session.execute(
+                select(Attacker).where(Attacker.ip == ip)
+            )
+            row = result.scalar_one_or_none()
+            return row.uuid if row else None
+    except Exception:
+        return None
+
+
+async def _emit_log(
+    repo: BaseRepository,
+    *,
+    event_type: str,
+    decky_name: str,
+    src_ip: str,
+    port: int,
+    extra: dict[str, Any] | None = None,
+) -> None:
+    attacker_uuid = await _get_attacker_uuid(repo, src_ip)
+    fields: dict[str, Any] = {"port": port, "attacker_uuid": attacker_uuid}
+    if extra:
+        fields.update(extra)
+    try:
+        await repo.add_log({
+            "decky": decky_name,
+            "service": "tarpit",
+            "event_type": event_type,
+            "attacker_ip": src_ip,
+            "raw_line": f"tarpit {event_type} src={src_ip} decky={decky_name} port={port}",
+            "fields": json.dumps(fields),
+        })
+    except Exception as exc:
+        log.warning("tarpit log emit failed: %s", exc)
+
+
+async def tarpit_watcher_worker(repo: BaseRepository) -> None:
+    """Main loop — runs forever, wakes every DECNET_TARPIT_POLL_INTERVAL seconds."""
+    poll_interval = _get_poll_interval()
+    log.info("tarpit watcher started poll_interval=%ds", poll_interval)
+
+    # (decky_name, src_ip, port) → first_seen timestamp
+    seen: dict[tuple[str, str, int], datetime] = {}
+
+    while True:
+        try:
+            await _tick(repo, seen)
+        except asyncio.CancelledError:
+            raise
+        except Exception as exc:
+            log.warning("tarpit watcher tick error: %s", exc)
+        await asyncio.sleep(poll_interval)
+
+
+async def _tick(
+    repo: BaseRepository,
+    seen: dict[tuple[str, str, int], datetime],
+) -> None:
+    rules = await repo.list_tarpit_rules()
+    if not rules:
+        # No active tarpit rules — clear stale seen state and bail early.
+        seen.clear()
+        return
+
+    current: set[tuple[str, str, int]] = set()
+
+    for rule in rules:
+        decky_name: str = rule["decky_name"]
+        ports: list[int] = rule["ports"]
+        try:
+            pid = await asyncio.to_thread(get_container_pid, decky_name)
+        except LookupError as exc:
+            log.debug("tarpit watcher: %s", exc)
+            continue
+
+        tcp_content = await asyncio.to_thread(_read_proc_net_tcp, pid)
+
+        for port in ports:
+            for src_ip in _parse_connections(tcp_content, port):
+                key = (decky_name, src_ip, port)
+                current.add(key)
+                if key not in seen:
+                    seen[key] = datetime.now(timezone.utc)
+                    log.info(
+                        "tarpit enter decky=%s src=%s port=%d",
+                        decky_name, src_ip, port,
+                    )
+                    await _emit_log(
+                        repo,
+                        event_type="tarpit_enter",
+                        decky_name=decky_name,
+                        src_ip=src_ip,
+                        port=port,
+                    )
+
+    for key in list(seen):
+        if key not in current:
+            first_seen = seen.pop(key)
+            elapsed = int((datetime.now(timezone.utc) - first_seen).total_seconds())
+            decky_name, src_ip, port = key
+            log.info(
+                "tarpit exit decky=%s src=%s port=%d elapsed=%ds",
+                decky_name, src_ip, port, elapsed,
+            )
+            await _emit_log(
+                repo,
+                event_type="tarpit_exit",
+                decky_name=decky_name,
+                src_ip=src_ip,
+                port=port,
+                extra={"duration_s": elapsed},
+            )

decnet/web/api.py

@@ -31,6 +31,7 @@ from decnet.web.dependencies import repo
 from decnet.collector import log_collector_worker
 from decnet.web.ingester import log_ingestion_worker
 from decnet.profiler import attacker_profile_worker
+from decnet.tarpit import tarpit_watcher_worker
 from decnet.web.limiter import limiter
 from decnet.web.router import api_router
 from slowapi import _rate_limit_exceeded_handler
@@ -43,6 +44,7 @@ collector_task: Optional[asyncio.Task[Any]] = None
 attacker_task: Optional[asyncio.Task[Any]] = None
 sniffer_task: Optional[asyncio.Task[Any]] = None
 heartbeat_task: Optional[asyncio.Task[Any]] = None
+tarpit_task: Optional[asyncio.Task[Any]] = None
 
 
 def get_background_tasks() -> dict[str, Optional[asyncio.Task[Any]]]:
@@ -52,13 +54,14 @@ def get_background_tasks() -> dict[str, Optional[asyncio.Task[Any]]]:
         "collector_worker": collector_task,
         "attacker_worker": attacker_task,
         "sniffer_worker": sniffer_task,
+        "tarpit_watcher": tarpit_task,
     }
 
 
 @asynccontextmanager
 async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
     global ingestion_task, collector_task, attacker_task, sniffer_task
-    global heartbeat_task
+    global heartbeat_task, tarpit_task
 
     import resource
     soft, hard = resource.getrlimit(resource.RLIMIT_NOFILE)
@@ -162,6 +165,11 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
                 log.warning("Sniffer worker failed to start — API continues without sniffing: %s", exc)
         else:
             log.debug("API startup: sniffer not embedded — expecting standalone daemon")
+
+        # Tarpit watcher — always-on, near-zero cost when no rules exist.
+        if tarpit_task is None or tarpit_task.done():
+            tarpit_task = asyncio.create_task(tarpit_watcher_worker(repo))
+            log.debug("API startup: tarpit watcher started")
     else:
         log.info("Contract Test Mode: skipping background worker startup")
 
@@ -191,7 +199,7 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
         await get_registry().stop()
     except Exception as exc:  # noqa: BLE001
         log.warning("worker registry stop raised: %s", exc)
-    for task in (ingestion_task, collector_task, attacker_task, sniffer_task, heartbeat_task):
+    for task in (ingestion_task, collector_task, attacker_task, sniffer_task, heartbeat_task, tarpit_task):
         if task and not task.done():
             task.cancel()
             try:

decnet/web/db/models/__init__.py

@@ -179,6 +179,12 @@ from .workers import (
     WorkersResponse,
     WorkerStatus,
 )
+from .tarpit import (
+    TarpitEnableRequest,
+    TarpitRule,
+    TarpitRuleResponse,
+    TarpitStatusResponse,
+)
 
 __all__ = [
     # _base
@@ -334,4 +340,9 @@ __all__ = [
     "WorkerControlResponse",
     "WorkersResponse",
     "WorkerStatus",
+    # tarpit
+    "TarpitEnableRequest",
+    "TarpitRule",
+    "TarpitRuleResponse",
+    "TarpitStatusResponse",
 ]

decnet/web/db/models/tarpit.py

@@ -0,0 +1,44 @@
+"""Tarpit rule table + HTTP request/response shapes."""
+from datetime import datetime, timezone
+from typing import Any
+
+from pydantic import BaseModel, Field as PydanticField
+from sqlmodel import Field, SQLModel
+
+
+class TarpitRule(SQLModel, table=True):
+    """One active tarpit rule — one per decky at a time.
+
+    ``ports`` is JSON-encoded (e.g. ``"[22, 80]"``).  One row per decky;
+    ``set_tarpit_rule`` upserts on ``decky_name`` so re-enabling with
+    different parameters replaces the old rule.
+    """
+    __tablename__ = "tarpit_rules"
+
+    id: str = Field(primary_key=True)
+    decky_name: str = Field(index=True, unique=True)
+    ports: str          # JSON list[int]
+    delay_ms: int
+    created_at: datetime = Field(
+        default_factory=lambda: datetime.now(timezone.utc)
+    )
+    created_by: str     # operator UUID from JWT
+
+
+class TarpitEnableRequest(BaseModel):
+    ports: list[int] = PydanticField(..., min_length=1)
+    delay_ms: int = PydanticField(..., ge=100, le=300_000)
+
+
+class TarpitRuleResponse(BaseModel):
+    id: str
+    decky_name: str
+    ports: list[int]
+    delay_ms: int
+    created_at: datetime
+    created_by: str
+
+
+class TarpitStatusResponse(BaseModel):
+    rule: TarpitRuleResponse
+    active_connections: list[dict[str, Any]]

decnet/web/db/sqlmodel_repo/__init__.py

@@ -48,6 +48,7 @@ from decnet.web.db.sqlmodel_repo.orchestrator import OrchestratorMixin
 from decnet.web.db.sqlmodel_repo.realism import RealismMixin
 from decnet.web.db.sqlmodel_repo.swarm import SwarmMixin
 from decnet.web.db.sqlmodel_repo.topology import TopologyMixin
+from decnet.web.db.sqlmodel_repo.tarpit import TarpitMixin
 from decnet.web.db.sqlmodel_repo.webhooks import WebhooksMixin
 
 
@@ -66,6 +67,7 @@ class SQLModelRepository(
     OrchestratorMixin,
     RealismMixin,
     SwarmMixin,
+    TarpitMixin,
     TopologyMixin,
     WebhooksMixin,
     BaseRepository,

decnet/web/db/sqlmodel_repo/tarpit.py

@@ -0,0 +1,70 @@
+"""Tarpit rule CRUD."""
+from __future__ import annotations
+
+import json
+import uuid
+from datetime import datetime, timezone
+from typing import Any, Optional
+
+from sqlalchemy import select
+
+from decnet.web.db.models import TarpitRule
+
+
+class TarpitMixin:
+    """Mixin: composed onto ``SQLModelRepository``."""
+
+    async def set_tarpit_rule(self, data: dict[str, Any]) -> None:
+        """Upsert a tarpit rule keyed on ``decky_name`` (one rule per decky)."""
+        async with self._session() as session:
+            result = await session.execute(
+                select(TarpitRule).where(TarpitRule.decky_name == data["decky_name"])
+            )
+            existing = result.scalar_one_or_none()
+            if existing:
+                for k, v in data.items():
+                    setattr(existing, k, v)
+                session.add(existing)
+            else:
+                payload = {
+                    "id": str(uuid.uuid4()),
+                    "created_at": datetime.now(timezone.utc),
+                    **data,
+                }
+                session.add(TarpitRule(**payload))
+            await session.commit()
+
+    async def get_tarpit_rule(self, decky_name: str) -> Optional[dict[str, Any]]:
+        async with self._session() as session:
+            result = await session.execute(
+                select(TarpitRule).where(TarpitRule.decky_name == decky_name)
+            )
+            row = result.scalar_one_or_none()
+            if row is None:
+                return None
+            d = row.model_dump(mode="json")
+            d["ports"] = json.loads(d["ports"])
+            return d
+
+    async def delete_tarpit_rule(self, decky_name: str) -> bool:
+        async with self._session() as session:
+            result = await session.execute(
+                select(TarpitRule).where(TarpitRule.decky_name == decky_name)
+            )
+            row = result.scalar_one_or_none()
+            if row is None:
+                return False
+            await session.delete(row)
+            await session.commit()
+            return True
+
+    async def list_tarpit_rules(self) -> list[dict[str, Any]]:
+        async with self._session() as session:
+            result = await session.execute(select(TarpitRule))
+            rows = result.scalars().all()
+            out = []
+            for row in rows:
+                d = row.model_dump(mode="json")
+                d["ports"] = json.loads(d["ports"])
+                out.append(d)
+            return out

decnet/web/router/deckies/__init__.py

@@ -18,6 +18,7 @@ from .api_services import (
     fleet_services_router,
     topology_services_router,
 )
+from .api_tarpit import router as tarpit_router
 
 deckies_router = APIRouter()
 deckies_router.include_router(file_drop_router)
@@ -27,5 +28,6 @@ deckies_router.include_router(fleet_services_router)
 # umbrella because the *operation* (add/remove a service on a deployed
 # decky) is identical; only the addressing scheme differs.
 deckies_router.include_router(topology_services_router)
+deckies_router.include_router(tarpit_router)
 
 __all__ = ["deckies_router"]

decnet/web/router/deckies/api_tarpit.py

@@ -0,0 +1,229 @@
+"""POST/GET/DELETE /api/v1/deckies/{decky_name}/tarpit — per-decky tc netem tarpit.
+
+Applies port-selective traffic delay on the host veth paired to the target
+decky container using tc qdisc (HTB + netem).  Requires CAP_NET_ADMIN on
+the API process (provided by decnet-api.service AmbientCapabilities).
+
+Auth: ``require_admin`` for write operations, ``require_viewer`` for GET.
+"""
+from __future__ import annotations
+
+import asyncio
+import json
+import socket
+import subprocess  # nosec B404
+
+from fastapi import APIRouter, Depends, HTTPException, Path
+
+from decnet.logging import get_logger
+from decnet.network import get_container_pid, get_container_veth
+from decnet.web.db.models import (
+    MessageResponse,
+    TarpitEnableRequest,
+    TarpitRuleResponse,
+    TarpitStatusResponse,
+)
+from decnet.web.dependencies import repo, require_admin, require_viewer
+
+log = get_logger("api.deckies.tarpit")
+
+router = APIRouter(prefix="/deckies/{decky_name}/tarpit", tags=["Deckies"])
+
+_DECKY_RE = r"^[a-z0-9\-]{1,64}$"
+
+
+def _tc(*args: str) -> subprocess.CompletedProcess:  # type: ignore[type-arg]
+    cmd = ["tc", *args]
+    return subprocess.run(cmd, capture_output=True, text=True)  # nosec B603 B404
+
+
+def _apply_tarpit(veth: str, ports: list[int], delay_ms: int) -> None:
+    """Build tc qdisc + class + netem + per-port filters on veth."""
+    steps = [
+        ["qdisc", "add", "dev", veth, "root", "handle", "1:", "htb"],
+        ["class", "add", "dev", veth, "parent", "1:", "classid", "1:1",
+         "htb", "rate", "1gbit"],
+        ["qdisc", "add", "dev", veth, "parent", "1:1", "handle", "10:",
+         "netem", "delay", f"{delay_ms}ms"],
+    ]
+    for args in steps:
+        r = _tc(*args)
+        if r.returncode != 0:
+            raise RuntimeError(r.stderr.strip())
+
+    for port in ports:
+        r = _tc(
+            "filter", "add", "dev", veth,
+            "protocol", "ip", "parent", "1:", "prio", "1",
+            "u32", "match", "ip", "dport", str(port), "0xffff",
+            "flowid", "1:1",
+        )
+        if r.returncode != 0:
+            raise RuntimeError(r.stderr.strip())
+
+
+def _remove_tarpit(veth: str) -> bool:
+    """Tear down the qdisc tree.  Returns False if nothing was there."""
+    r = _tc("qdisc", "del", "dev", veth, "root")
+    if r.returncode != 0:
+        if "Cannot find" in r.stderr or "No such" in r.stderr:
+            return False
+        raise RuntimeError(r.stderr.strip())
+    return True
+
+
+def _get_active_connections(pid: int, ports: list[int]) -> list[dict]:
+    """Read /proc/{pid}/net/tcp and return active connections on tarpitted ports."""
+    try:
+        with open(f"/proc/{pid}/net/tcp") as f:
+            content = f.read()
+    except OSError:
+        return []
+
+    conns: list[dict] = []
+    for line in content.strip().splitlines()[1:]:
+        parts = line.split()
+        if len(parts) < 4:
+            continue
+        local_hex, rem_hex, state = parts[1], parts[2], parts[3]
+        if state != "01":
+            continue
+        local_port = int(local_hex.split(":")[1], 16)
+        if local_port not in ports:
+            continue
+        rem_ip_hex = rem_hex.split(":")[0]
+        try:
+            ip = socket.inet_ntoa(bytes.fromhex(rem_ip_hex)[::-1])
+        except (ValueError, OSError):
+            continue
+        if ip != "0.0.0.0":  # nosec B104
+            conns.append({"ip": ip, "port": local_port})
+    return conns
+
+
+@router.post(
+    "",
+    response_model=MessageResponse,
+    status_code=201,
+    responses={
+        401: {"description": "Could not validate credentials"},
+        403: {"description": "Insufficient permissions"},
+        404: {"description": "Decky not found in active deployment"},
+        409: {"description": "tc command failed (qdisc already exists or veth unreachable)"},
+    },
+)
+async def api_enable_tarpit(
+    decky_name: str = Path(..., pattern=_DECKY_RE),
+    req: TarpitEnableRequest = ...,
+    admin: dict = Depends(require_admin),
+) -> MessageResponse:
+    try:
+        veth = await asyncio.to_thread(get_container_veth, decky_name)
+    except LookupError as exc:
+        raise HTTPException(status_code=404, detail=str(exc)) from exc
+
+    try:
+        await asyncio.to_thread(_apply_tarpit, veth, req.ports, req.delay_ms)
+    except RuntimeError as exc:
+        raise HTTPException(status_code=409, detail=str(exc)) from exc
+
+    ports_json = json.dumps(req.ports)
+    await repo.set_tarpit_rule({
+        "decky_name": decky_name,
+        "ports": ports_json,
+        "delay_ms": req.delay_ms,
+        "created_by": admin.get("uuid", "unknown"),
+    })
+    await repo.add_log({
+        "decky": decky_name,
+        "service": "tarpit",
+        "event_type": "tarpit_enabled",
+        "attacker_ip": "0.0.0.0",  # nosec B104
+        "raw_line": (
+            f"tarpit enabled decky={decky_name} ports={req.ports} delay={req.delay_ms}ms"
+            f" by={admin.get('uuid', 'unknown')}"
+        ),
+        "fields": json.dumps({
+            "ports": req.ports,
+            "delay_ms": req.delay_ms,
+            "veth": veth,
+            "operator": admin.get("uuid"),
+        }),
+    })
+    log.info(
+        "tarpit enabled decky=%s ports=%s delay_ms=%d veth=%s by=%s",
+        decky_name, req.ports, req.delay_ms, veth, admin.get("uuid"),
+    )
+    return MessageResponse(message="tarpit active")
+
+
+@router.get(
+    "",
+    response_model=TarpitStatusResponse,
+    responses={
+        401: {"description": "Could not validate credentials"},
+        403: {"description": "Insufficient permissions"},
+        404: {"description": "No active tarpit rule for this decky"},
+    },
+)
+async def api_get_tarpit(
+    decky_name: str = Path(..., pattern=_DECKY_RE),
+    viewer: dict = Depends(require_viewer),
+) -> TarpitStatusResponse:
+    rule = await repo.get_tarpit_rule(decky_name)
+    if rule is None:
+        raise HTTPException(status_code=404, detail="No active tarpit rule for this decky")
+
+    conns: list[dict] = []
+    try:
+        pid = await asyncio.to_thread(get_container_pid, decky_name)
+        raw_conns = await asyncio.to_thread(_get_active_connections, pid, rule["ports"])
+        for c in raw_conns:
+            conns.append({"ip": c["ip"], "port": c["port"]})
+    except LookupError:
+        pass
+
+    return TarpitStatusResponse(
+        rule=TarpitRuleResponse(**rule),
+        active_connections=conns,
+    )
+
+
+@router.delete(
+    "",
+    response_model=MessageResponse,
+    responses={
+        401: {"description": "Could not validate credentials"},
+        403: {"description": "Insufficient permissions"},
+        404: {"description": "Decky container not found"},
+        409: {"description": "tc teardown failed"},
+    },
+)
+async def api_disable_tarpit(
+    decky_name: str = Path(..., pattern=_DECKY_RE),
+    admin: dict = Depends(require_admin),
+) -> MessageResponse:
+    try:
+        veth = await asyncio.to_thread(get_container_veth, decky_name)
+    except LookupError as exc:
+        raise HTTPException(status_code=404, detail=str(exc)) from exc
+
+    try:
+        await asyncio.to_thread(_remove_tarpit, veth)
+    except RuntimeError as exc:
+        raise HTTPException(status_code=409, detail=str(exc)) from exc
+
+    await repo.delete_tarpit_rule(decky_name)
+    await repo.add_log({
+        "decky": decky_name,
+        "service": "tarpit",
+        "event_type": "tarpit_disabled",
+        "attacker_ip": "0.0.0.0",  # nosec B104
+        "raw_line": (
+            f"tarpit disabled decky={decky_name}"
+            f" by={admin.get('uuid', 'unknown')}"
+        ),
+        "fields": json.dumps({"veth": veth, "operator": admin.get("uuid")}),
+    })
+    log.info("tarpit disabled decky=%s veth=%s by=%s", decky_name, veth, admin.get("uuid"))
+    return MessageResponse(message="tarpit removed")