ctfy.sdk.admin_resources.records

client.admin.records — archived instance forensics (admin).

  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 _extract_items, _raise_for_status
  9from ctfy.sdk.base import BaseHttpClient
 10from ctfy.server.models import 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    ) -> builtins.list[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 container_log(self, instance_id: str) -> str:
 55        """Combined container stdout/stderr captured at teardown.
 56        Returns ``""`` when archival was disabled or nothing was kept."""
 57        resp = self._http.request("GET", f"/admin/instance-records/{instance_id}/container-log")
 58        _raise_for_status(resp)
 59        return str(resp.json().get("logs") or "")
 60
 61    def events(
 62        self, instance_id: str, offset: int = 0, limit: int = 50
 63    ) -> builtins.list[dict[str, Any]]:
 64        """Lifecycle event rows for one archived instance."""
 65        resp = self._http.request(
 66            "GET",
 67            f"/admin/instance-records/{instance_id}/events",
 68            params={"offset": offset, "limit": limit},
 69        )
 70        _raise_for_status(resp)
 71        items: builtins.list[dict[str, Any]] = resp.json()["items"]
 72        return items
 73
 74    def submissions(
 75        self, instance_id: str, offset: int = 0, limit: int = 50
 76    ) -> builtins.list[dict[str, Any]]:
 77        """Submission rows recorded against one archived instance."""
 78        resp = self._http.request(
 79            "GET",
 80            f"/admin/instance-records/{instance_id}/submissions",
 81            params={"offset": offset, "limit": limit},
 82        )
 83        _raise_for_status(resp)
 84        items: builtins.list[dict[str, Any]] = resp.json()["items"]
 85        return items
 86
 87    def traffic(
 88        self, instance_id: str, offset: int = 0, limit: int = 50
 89    ) -> builtins.list[dict[str, Any]]:
 90        """Persisted mitmproxy flows for one archived instance."""
 91        resp = self._http.request(
 92            "GET",
 93            f"/admin/instance-records/{instance_id}/traffic",
 94            params={"offset": offset, "limit": limit},
 95        )
 96        _raise_for_status(resp)
 97        items: builtins.list[dict[str, Any]] = resp.json()["items"]
 98        return items
 99
100    def pcap(self, instance_id: str) -> bytes:
101        """Archived tcpdump capture for one instance. Returns empty
102        ``bytes`` when no capture is on file."""
103        resp = self._http.request("GET", f"/admin/instance-records/{instance_id}/pcap")
104        if resp.status_code == 404:
105            return b""
106        _raise_for_status(resp)
107        return resp.content
class AdminRecordsResource: 
 14class AdminRecordsResource:
 15    """Post-mortem forensics over terminal (archived) instance records."""
 16
 17    def __init__(self, http: BaseHttpClient) -> None:
 18        self._http = http
 19
 20    def list(
 21        self,
 22        *,
 23        team_id: str = "",
 24        challenge_id: str = "",
 25        node_id: str = "",
 26        status: str = "",
 27        since_ts: float | None = None,
 28        offset: int = 0,
 29        limit: int = 50,
 30    ) -> builtins.list[InstanceRecordInfo]:
 31        """Archived (terminal) instance records — the post-mortem
 32        forensics index. ``since_ts`` is a UNIX epoch lower bound."""
 33        params: dict[str, Any] = {"offset": offset, "limit": limit}
 34        if team_id:
 35            params["team_id"] = team_id
 36        if challenge_id:
 37            params["challenge_id"] = challenge_id
 38        if node_id:
 39            params["node_id"] = node_id
 40        if status:
 41            params["status"] = status
 42        if since_ts is not None:
 43            params["since_ts"] = since_ts
 44        resp = self._http.request("GET", "/admin/instance-records", params=params)
 45        _raise_for_status(resp)
 46        return _extract_items(resp.json(), InstanceRecordInfo)
 47
 48    def get(self, instance_id: str) -> InstanceRecordDetail:
 49        """Full archived record for one instance (lifecycle, node,
 50        artifact availability flags)."""
 51        resp = self._http.request("GET", f"/admin/instance-records/{instance_id}")
 52        _raise_for_status(resp)
 53        return InstanceRecordDetail.model_validate(resp.json())
 54
 55    def container_log(self, instance_id: str) -> str:
 56        """Combined container stdout/stderr captured at teardown.
 57        Returns ``""`` when archival was disabled or nothing was kept."""
 58        resp = self._http.request("GET", f"/admin/instance-records/{instance_id}/container-log")
 59        _raise_for_status(resp)
 60        return str(resp.json().get("logs") or "")
 61
 62    def events(
 63        self, instance_id: str, offset: int = 0, limit: int = 50
 64    ) -> builtins.list[dict[str, Any]]:
 65        """Lifecycle event rows for one archived instance."""
 66        resp = self._http.request(
 67            "GET",
 68            f"/admin/instance-records/{instance_id}/events",
 69            params={"offset": offset, "limit": limit},
 70        )
 71        _raise_for_status(resp)
 72        items: builtins.list[dict[str, Any]] = resp.json()["items"]
 73        return items
 74
 75    def submissions(
 76        self, instance_id: str, offset: int = 0, limit: int = 50
 77    ) -> builtins.list[dict[str, Any]]:
 78        """Submission rows recorded against one archived instance."""
 79        resp = self._http.request(
 80            "GET",
 81            f"/admin/instance-records/{instance_id}/submissions",
 82            params={"offset": offset, "limit": limit},
 83        )
 84        _raise_for_status(resp)
 85        items: builtins.list[dict[str, Any]] = resp.json()["items"]
 86        return items
 87
 88    def traffic(
 89        self, instance_id: str, offset: int = 0, limit: int = 50
 90    ) -> builtins.list[dict[str, Any]]:
 91        """Persisted mitmproxy flows for one archived instance."""
 92        resp = self._http.request(
 93            "GET",
 94            f"/admin/instance-records/{instance_id}/traffic",
 95            params={"offset": offset, "limit": limit},
 96        )
 97        _raise_for_status(resp)
 98        items: builtins.list[dict[str, Any]] = resp.json()["items"]
 99        return items
100
101    def pcap(self, instance_id: str) -> bytes:
102        """Archived tcpdump capture for one instance. Returns empty
103        ``bytes`` when no capture is on file."""
104        resp = self._http.request("GET", f"/admin/instance-records/{instance_id}/pcap")
105        if resp.status_code == 404:
106            return b""
107        _raise_for_status(resp)
108        return resp.content

Post-mortem forensics over terminal (archived) instance records.

AdminRecordsResource(http: ctfy.sdk.base.BaseHttpClient) 
17    def __init__(self, http: BaseHttpClient) -> None:
18        self._http = http
def list( self, *, team_id: str = '', challenge_id: str = '', node_id: str = '', status: str = '', since_ts: float | None = None, offset: int = 0, limit: int = 50) -> list[ctfy.server.models.InstanceRecordInfo]: 
20    def list(
21        self,
22        *,
23        team_id: str = "",
24        challenge_id: str = "",
25        node_id: str = "",
26        status: str = "",
27        since_ts: float | None = None,
28        offset: int = 0,
29        limit: int = 50,
30    ) -> builtins.list[InstanceRecordInfo]:
31        """Archived (terminal) instance records — the post-mortem
32        forensics index. ``since_ts`` is a UNIX epoch lower bound."""
33        params: dict[str, Any] = {"offset": offset, "limit": limit}
34        if team_id:
35            params["team_id"] = team_id
36        if challenge_id:
37            params["challenge_id"] = challenge_id
38        if node_id:
39            params["node_id"] = node_id
40        if status:
41            params["status"] = status
42        if since_ts is not None:
43            params["since_ts"] = since_ts
44        resp = self._http.request("GET", "/admin/instance-records", params=params)
45        _raise_for_status(resp)
46        return _extract_items(resp.json(), InstanceRecordInfo)

Archived (terminal) instance records — the post-mortem forensics index. since_ts is a UNIX epoch lower bound.

def get( self, instance_id: str) -> ctfy.server.models.InstanceRecordDetail: 
48    def get(self, instance_id: str) -> InstanceRecordDetail:
49        """Full archived record for one instance (lifecycle, node,
50        artifact availability flags)."""
51        resp = self._http.request("GET", f"/admin/instance-records/{instance_id}")
52        _raise_for_status(resp)
53        return InstanceRecordDetail.model_validate(resp.json())

Full archived record for one instance (lifecycle, node, artifact availability flags).

def container_log(self, instance_id: str) -> str: 
55    def container_log(self, instance_id: str) -> str:
56        """Combined container stdout/stderr captured at teardown.
57        Returns ``""`` when archival was disabled or nothing was kept."""
58        resp = self._http.request("GET", f"/admin/instance-records/{instance_id}/container-log")
59        _raise_for_status(resp)
60        return str(resp.json().get("logs") or "")

Combined container stdout/stderr captured at teardown. Returns "" when archival was disabled or nothing was kept.

def events( self, instance_id: str, offset: int = 0, limit: int = 50) -> list[dict[str, typing.Any]]: 
62    def events(
63        self, instance_id: str, offset: int = 0, limit: int = 50
64    ) -> builtins.list[dict[str, Any]]:
65        """Lifecycle event rows for one archived instance."""
66        resp = self._http.request(
67            "GET",
68            f"/admin/instance-records/{instance_id}/events",
69            params={"offset": offset, "limit": limit},
70        )
71        _raise_for_status(resp)
72        items: builtins.list[dict[str, Any]] = resp.json()["items"]
73        return items

Lifecycle event rows for one archived instance.

def submissions( self, instance_id: str, offset: int = 0, limit: int = 50) -> list[dict[str, typing.Any]]: 
75    def submissions(
76        self, instance_id: str, offset: int = 0, limit: int = 50
77    ) -> builtins.list[dict[str, Any]]:
78        """Submission rows recorded against one archived instance."""
79        resp = self._http.request(
80            "GET",
81            f"/admin/instance-records/{instance_id}/submissions",
82            params={"offset": offset, "limit": limit},
83        )
84        _raise_for_status(resp)
85        items: builtins.list[dict[str, Any]] = resp.json()["items"]
86        return items

Submission rows recorded against one archived instance.

def traffic( self, instance_id: str, offset: int = 0, limit: int = 50) -> list[dict[str, typing.Any]]: 
88    def traffic(
89        self, instance_id: str, offset: int = 0, limit: int = 50
90    ) -> builtins.list[dict[str, Any]]:
91        """Persisted mitmproxy flows for one archived instance."""
92        resp = self._http.request(
93            "GET",
94            f"/admin/instance-records/{instance_id}/traffic",
95            params={"offset": offset, "limit": limit},
96        )
97        _raise_for_status(resp)
98        items: builtins.list[dict[str, Any]] = resp.json()["items"]
99        return items

Persisted mitmproxy flows for one archived instance.

def pcap(self, instance_id: str) -> bytes: 
101    def pcap(self, instance_id: str) -> bytes:
102        """Archived tcpdump capture for one instance. Returns empty
103        ``bytes`` when no capture is on file."""
104        resp = self._http.request("GET", f"/admin/instance-records/{instance_id}/pcap")
105        if resp.status_code == 404:
106            return b""
107        _raise_for_status(resp)
108        return resp.content

Archived tcpdump capture for one instance. Returns empty bytes when no capture is on file.