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 (
 11    AnswerTraceResponse,
 12    InstanceAnswersResponse,
 13    InstanceRecordDetail,
 14    InstanceRecordInfo,
 15)
 16
 17
 18class AdminRecordsResource:
 19    """Post-mortem forensics over terminal (archived) instance records."""
 20
 21    def __init__(self, http: BaseHttpClient) -> None:
 22        self._http = http
 23
 24    def list(
 25        self,
 26        *,
 27        team_id: str = "",
 28        challenge_id: str = "",
 29        node_id: str = "",
 30        status: str = "",
 31        since_ts: float | None = None,
 32        offset: int = 0,
 33        limit: int = 50,
 34    ) -> PagedList[InstanceRecordInfo]:
 35        """Archived (terminal) instance records — the post-mortem
 36        forensics index. ``since_ts`` is a UNIX epoch lower bound."""
 37        params: dict[str, Any] = {"offset": offset, "limit": limit}
 38        if team_id:
 39            params["team_id"] = team_id
 40        if challenge_id:
 41            params["challenge_id"] = challenge_id
 42        if node_id:
 43            params["node_id"] = node_id
 44        if status:
 45            params["status"] = status
 46        if since_ts is not None:
 47            params["since_ts"] = since_ts
 48        resp = self._http.request("GET", "/admin/instance-records", params=params)
 49        _raise_for_status(resp)
 50        return _extract_items(resp.json(), InstanceRecordInfo)
 51
 52    def get(self, instance_id: str) -> InstanceRecordDetail:
 53        """Full archived record for one instance (lifecycle, node,
 54        artifact availability flags)."""
 55        resp = self._http.request("GET", f"/admin/instance-records/{instance_id}")
 56        _raise_for_status(resp)
 57        return InstanceRecordDetail.model_validate(resp.json())
 58
 59    def trace_answer(self, answer: str) -> AnswerTraceResponse:
 60        """Attribute a leaked answer to the team it was minted for.
 61
 62        The counterpart to per-team unique answers: when a flag turns up
 63        in a public channel, this says whose it was. Searches live
 64        instances (minted plaintext still in memory) and archived
 65        records (matched on the digest persisted at teardown).
 66
 67        *answer* is accepted wrapped (``FLAG{...}``) or bare — both
 68        normalise to the same digest, as submissions do. It travels in
 69        the request body, never a query string, so a live flag doesn't
 70        land in access logs on the way.
 71
 72        Check ``traceable_records`` on an empty result: instances
 73        archived before answer fingerprinting shipped carry no digest,
 74        so a miss against a mostly-untraceable archive means "unknown",
 75        not "not ours".
 76        """
 77        resp = self._http.request("POST", "/admin/answer-trace", json={"answer": answer})
 78        _raise_for_status(resp)
 79        return AnswerTraceResponse.model_validate(resp.json())
 80
 81    def instance_answers(self, instance_id: str) -> InstanceAnswersResponse:
 82        """The answers minted for one **live** instance.
 83
 84        ⚠️ The one call on this SDK that returns groundtruth. Everything
 85        else goes the other way: the catalog redacts gated prompts,
 86        :meth:`ChallengesResource.questions` widens *who may read the
 87        question* and not *what is said*, and :meth:`answer_trace` takes
 88        a value rather than handing one out. It exists because a dispute
 89        about a submission otherwise has nothing to compare against.
 90
 91        Two consequences of that, both deliberate:
 92
 93        * **Every call is recorded.** The read emits an
 94          ``instance_answers_revealed`` activity row naming the caller
 95          and carrying the count — never the values. Do not put this in
 96          a polling loop; each pass is a disclosure in the audit log.
 97        * **Live instances only.** An archived record keeps a one-way
 98          fingerprint of each answer and not the answer, so a stopped
 99          instance raises rather than returning an empty list —
100          ``no answers here`` and ``this instance is gone`` are opposite
101          facts. Use :meth:`answer_trace` for an archived one.
102
103        Wraps ``GET /admin/instances/{id}/answers``."""
104        resp = self._http.request("GET", f"/admin/instances/{instance_id}/answers")
105        _raise_for_status(resp)
106        return InstanceAnswersResponse.model_validate(resp.json())
107
108    def container_log(self, instance_id: str) -> str:
109        """Combined container stdout/stderr captured at teardown.
110        Returns ``""`` when archival was disabled or nothing was kept."""
111        resp = self._http.request("GET", f"/admin/instance-records/{instance_id}/container-log")
112        _raise_for_status(resp)
113        return str(resp.json().get("logs") or "")
114
115    def events(
116        self, instance_id: str, offset: int = 0, limit: int = 50
117    ) -> builtins.list[dict[str, Any]]:
118        """Lifecycle event rows for one archived instance."""
119        resp = self._http.request(
120            "GET",
121            f"/admin/instance-records/{instance_id}/events",
122            params={"offset": offset, "limit": limit},
123        )
124        _raise_for_status(resp)
125        items: builtins.list[dict[str, Any]] = resp.json()["items"]
126        return items
127
128    def submissions(
129        self, instance_id: str, offset: int = 0, limit: int = 50
130    ) -> builtins.list[dict[str, Any]]:
131        """Submission rows recorded against one archived instance."""
132        resp = self._http.request(
133            "GET",
134            f"/admin/instance-records/{instance_id}/submissions",
135            params={"offset": offset, "limit": limit},
136        )
137        _raise_for_status(resp)
138        items: builtins.list[dict[str, Any]] = resp.json()["items"]
139        return items
140
141    def traffic(
142        self, instance_id: str, offset: int = 0, limit: int = 50
143    ) -> builtins.list[dict[str, Any]]:
144        """Persisted mitmproxy flows for one archived instance."""
145        resp = self._http.request(
146            "GET",
147            f"/admin/instance-records/{instance_id}/traffic",
148            params={"offset": offset, "limit": limit},
149        )
150        _raise_for_status(resp)
151        items: builtins.list[dict[str, Any]] = resp.json()["items"]
152        return items
153
154    def pcap(self, instance_id: str) -> bytes:
155        """Archived tcpdump capture for one instance. Returns empty
156        ``bytes`` when no capture is on file."""
157        resp = self._http.request("GET", f"/admin/instance-records/{instance_id}/pcap")
158        if resp.status_code == 404:
159            return b""
160        _raise_for_status(resp)
161        return resp.content