ctfy.sdk.admin_resources.records

  1"""``client.admin.records`` — archived instance forensics (admin)."""
  2
  3from __future__ import annotations
  4
  5import builtins
  6from typing import Any
  7
  8from ctfy.sdk._helpers import PagedList, _extract_items, _raise_for_status
  9from ctfy.sdk.base import BaseHttpClient
 10from ctfy.server.models import AnswerTraceResponse, InstanceRecordDetail, InstanceRecordInfo
 11
 12
 13class AdminRecordsResource:
 14    """Post-mortem forensics over terminal (archived) instance records."""
 15
 16    def __init__(self, http: BaseHttpClient) -> None:
 17        self._http = http
 18
 19    def list(
 20        self,
 21        *,
 22        team_id: str = "",
 23        challenge_id: str = "",
 24        node_id: str = "",
 25        status: str = "",
 26        since_ts: float | None = None,
 27        offset: int = 0,
 28        limit: int = 50,
 29    ) -> PagedList[InstanceRecordInfo]:
 30        """Archived (terminal) instance records — the post-mortem
 31        forensics index. ``since_ts`` is a UNIX epoch lower bound."""
 32        params: dict[str, Any] = {"offset": offset, "limit": limit}
 33        if team_id:
 34            params["team_id"] = team_id
 35        if challenge_id:
 36            params["challenge_id"] = challenge_id
 37        if node_id:
 38            params["node_id"] = node_id
 39        if status:
 40            params["status"] = status
 41        if since_ts is not None:
 42            params["since_ts"] = since_ts
 43        resp = self._http.request("GET", "/admin/instance-records", params=params)
 44        _raise_for_status(resp)
 45        return _extract_items(resp.json(), InstanceRecordInfo)
 46
 47    def get(self, instance_id: str) -> InstanceRecordDetail:
 48        """Full archived record for one instance (lifecycle, node,
 49        artifact availability flags)."""
 50        resp = self._http.request("GET", f"/admin/instance-records/{instance_id}")
 51        _raise_for_status(resp)
 52        return InstanceRecordDetail.model_validate(resp.json())
 53
 54    def trace_answer(self, answer: str) -> AnswerTraceResponse:
 55        """Attribute a leaked answer to the team it was minted for.
 56
 57        The counterpart to per-team unique answers: when a flag turns up
 58        in a public channel, this says whose it was. Searches live
 59        instances (minted plaintext still in memory) and archived
 60        records (matched on the digest persisted at teardown).
 61
 62        *answer* is accepted wrapped (``FLAG{...}``) or bare — both
 63        normalise to the same digest, as submissions do. It travels in
 64        the request body, never a query string, so a live flag doesn't
 65        land in access logs on the way.
 66
 67        Check ``traceable_records`` on an empty result: instances
 68        archived before answer fingerprinting shipped carry no digest,
 69        so a miss against a mostly-untraceable archive means "unknown",
 70        not "not ours".
 71        """
 72        resp = self._http.request("POST", "/admin/answer-trace", json={"answer": answer})
 73        _raise_for_status(resp)
 74        return AnswerTraceResponse.model_validate(resp.json())
 75
 76    def container_log(self, instance_id: str) -> str:
 77        """Combined container stdout/stderr captured at teardown.
 78        Returns ``""`` when archival was disabled or nothing was kept."""
 79        resp = self._http.request("GET", f"/admin/instance-records/{instance_id}/container-log")
 80        _raise_for_status(resp)
 81        return str(resp.json().get("logs") or "")
 82
 83    def events(
 84        self, instance_id: str, offset: int = 0, limit: int = 50
 85    ) -> builtins.list[dict[str, Any]]:
 86        """Lifecycle event rows for one archived instance."""
 87        resp = self._http.request(
 88            "GET",
 89            f"/admin/instance-records/{instance_id}/events",
 90            params={"offset": offset, "limit": limit},
 91        )
 92        _raise_for_status(resp)
 93        items: builtins.list[dict[str, Any]] = resp.json()["items"]
 94        return items
 95
 96    def submissions(
 97        self, instance_id: str, offset: int = 0, limit: int = 50
 98    ) -> builtins.list[dict[str, Any]]:
 99        """Submission rows recorded against one archived instance."""
100        resp = self._http.request(
101            "GET",
102            f"/admin/instance-records/{instance_id}/submissions",
103            params={"offset": offset, "limit": limit},
104        )
105        _raise_for_status(resp)
106        items: builtins.list[dict[str, Any]] = resp.json()["items"]
107        return items
108
109    def traffic(
110        self, instance_id: str, offset: int = 0, limit: int = 50
111    ) -> builtins.list[dict[str, Any]]:
112        """Persisted mitmproxy flows for one archived instance."""
113        resp = self._http.request(
114            "GET",
115            f"/admin/instance-records/{instance_id}/traffic",
116            params={"offset": offset, "limit": limit},
117        )
118        _raise_for_status(resp)
119        items: builtins.list[dict[str, Any]] = resp.json()["items"]
120        return items
121
122    def pcap(self, instance_id: str) -> bytes:
123        """Archived tcpdump capture for one instance. Returns empty
124        ``bytes`` when no capture is on file."""
125        resp = self._http.request("GET", f"/admin/instance-records/{instance_id}/pcap")
126        if resp.status_code == 404:
127            return b""
128        _raise_for_status(resp)
129        return resp.content