ctfy.sdk.admin_resources.instances

client.admin.instances — live cluster-wide instance fleet (admin).

  1"""``client.admin.instances`` — live cluster-wide instance fleet (admin)."""
  2
  3from __future__ import annotations
  4
  5import builtins
  6from typing import Any
  7
  8from ctfy.core.enums import InternetAccess
  9from ctfy.core.state.models import TimingGroup
 10from ctfy.sdk._helpers import PagedList, _extract_items, _raise_for_status
 11from ctfy.sdk.base import BaseHttpClient
 12from ctfy.server.models import (
 13    AdminInstanceTimings,
 14    InstanceInfo,
 15    InstanceInternetRequest,
 16    InstanceInternetResult,
 17    InstanceStorageReport,
 18)
 19
 20
 21class AdminInstancesResource:
 22    """Every live instance cluster-wide, with forensic pcap / traffic reads.
 23    Archived (terminal) records live on :class:`AdminRecordsResource`."""
 24
 25    def __init__(self, http: BaseHttpClient) -> None:
 26        self._http = http
 27
 28    def stop_all(self) -> builtins.list[dict[str, Any]]:
 29        """Cluster-wide tear-down: stop every running instance. Used for
 30        maintenance windows. Returns one :class:`StopResponse` row per
 31        instance that was stopped."""
 32        resp = self._http.request("DELETE", "/instances")
 33        _raise_for_status(resp)
 34        rows: builtins.list[dict[str, Any]] = resp.json()
 35        return rows
 36
 37    def stop(self, instance_id: str) -> dict[str, Any]:
 38        """Stop one running instance, whoever owns it — the surgical
 39        counterpart to :meth:`stop_all` for a wedged or abusive box.
 40        The owner team's record is archived (logs captured) first."""
 41        resp = self._http.request("DELETE", f"/admin/instances/{instance_id}")
 42        _raise_for_status(resp)
 43        row: dict[str, Any] = resp.json()
 44        return row
 45
 46    def set_internet(
 47        self, instance_id: str, internet: InternetAccess, *, reason: str
 48    ) -> InstanceInternetResult:
 49        """Open or cut one running box's internet access (#2067).
 50
 51        The operator's escape hatch over the verdict the box was started
 52        with: the owning node applies it first, then the platform records it
 53        and audits *reason*. ``outcome`` is ``changed`` or ``unchanged`` (it
 54        already had that setting); every failure raises with its code and
 55        leaves the box as it was. Holds until the box is reset.
 56        """
 57        resp = self._http.request(
 58            "POST",
 59            f"/admin/instances/{instance_id}/internet",
 60            json=InstanceInternetRequest(internet=internet, reason=reason).model_dump(mode="json"),
 61        )
 62        _raise_for_status(resp)
 63        return InstanceInternetResult.model_validate(resp.json())
 64
 65    def list(
 66        self,
 67        *,
 68        node_id: str = "",
 69        status: str = "",
 70        offset: int = 0,
 71        limit: int = 50,
 72    ) -> PagedList[InstanceInfo]:
 73        """Every live instance cluster-wide (no team-ownership filter)."""
 74        params: dict[str, Any] = {"offset": offset, "limit": limit}
 75        if node_id:
 76            params["node_id"] = node_id
 77        if status:
 78            params["status"] = status
 79        resp = self._http.request("GET", "/admin/instances", params=params)
 80        _raise_for_status(resp)
 81        return _extract_items(resp.json(), InstanceInfo)
 82
 83    def get(self, instance_id: str) -> InstanceInfo:
 84        """Single live instance, admin view (any team's)."""
 85        resp = self._http.request("GET", f"/admin/instances/{instance_id}")
 86        _raise_for_status(resp)
 87        return InstanceInfo.model_validate(resp.json())
 88
 89    def pcap(self, instance_id: str) -> bytes:
 90        """Live tcpdump capture for a running instance. Returns empty
 91        ``bytes`` when none exists (sidecar off, never ran) so callers
 92        can persist conditionally without try/except."""
 93        resp = self._http.request("GET", f"/admin/instances/{instance_id}/pcap")
 94        if resp.status_code == 404:
 95            return b""
 96        _raise_for_status(resp)
 97        return resp.content
 98
 99    def traffic(self, instance_id: str, *, cursor: str = "", limit: int = 0) -> dict[str, Any]:
100        """Live mitmproxy traffic for any running instance (admin twin
101        of :meth:`InstancesResource.traffic`, no ownership check).
102
103        Same ``{flows, cursor, truncated, capture}`` envelope and the same
104        polling contract; see the player's method for why an empty
105        cursor is not a value to send on every call.
106        """
107        params: dict[str, Any] = {"cursor": cursor} if cursor else {}
108        if limit > 0:
109            params["limit"] = limit
110        resp = self._http.request("GET", f"/admin/instances/{instance_id}/traffic", params=params)
111        _raise_for_status(resp)
112        data = resp.json()
113        return data if isinstance(data, dict) else {}
114
115    def storage(self) -> InstanceStorageReport:
116        """Disk held by the per-instance archive and each node's captures (#1714).
117
118        The archive is the platform's own last reading; each capture row is
119        that node's, carried on its heartbeat. ``report is None`` on a row is
120        *not measured yet*, never an empty store.
121        """
122        resp = self._http.request("GET", "/admin/instance-storage")
123        _raise_for_status(resp)
124        return InstanceStorageReport.model_validate(resp.json())
125
126    def timings(
127        self,
128        *,
129        group_by: TimingGroup = "challenge",
130        competition_id: str = "",
131        window: int = 0,
132        image_kind: str = "",
133    ) -> AdminInstanceTimings:
134        """Where instance starts spend their time, per challenge or node (#2095).
135
136        Wraps ``GET /admin/instance-timings``: p50 / p95 of the wait and of
137        its phases over the starts the node measured, slowest first.
138        *window* is the last N seconds (``0`` = the whole archive).
139        """
140        params: dict[str, Any] = {"group_by": group_by}
141        if competition_id:
142            params["competition_id"] = competition_id
143        if window:
144            params["window"] = window
145        if image_kind:
146            params["image_kind"] = image_kind
147        resp = self._http.request("GET", "/admin/instance-timings", params=params)
148        _raise_for_status(resp)
149        return AdminInstanceTimings.model_validate(resp.json())
class AdminInstancesResource: 
 22class AdminInstancesResource:
 23    """Every live instance cluster-wide, with forensic pcap / traffic reads.
 24    Archived (terminal) records live on :class:`AdminRecordsResource`."""
 25
 26    def __init__(self, http: BaseHttpClient) -> None:
 27        self._http = http
 28
 29    def stop_all(self) -> builtins.list[dict[str, Any]]:
 30        """Cluster-wide tear-down: stop every running instance. Used for
 31        maintenance windows. Returns one :class:`StopResponse` row per
 32        instance that was stopped."""
 33        resp = self._http.request("DELETE", "/instances")
 34        _raise_for_status(resp)
 35        rows: builtins.list[dict[str, Any]] = resp.json()
 36        return rows
 37
 38    def stop(self, instance_id: str) -> dict[str, Any]:
 39        """Stop one running instance, whoever owns it — the surgical
 40        counterpart to :meth:`stop_all` for a wedged or abusive box.
 41        The owner team's record is archived (logs captured) first."""
 42        resp = self._http.request("DELETE", f"/admin/instances/{instance_id}")
 43        _raise_for_status(resp)
 44        row: dict[str, Any] = resp.json()
 45        return row
 46
 47    def set_internet(
 48        self, instance_id: str, internet: InternetAccess, *, reason: str
 49    ) -> InstanceInternetResult:
 50        """Open or cut one running box's internet access (#2067).
 51
 52        The operator's escape hatch over the verdict the box was started
 53        with: the owning node applies it first, then the platform records it
 54        and audits *reason*. ``outcome`` is ``changed`` or ``unchanged`` (it
 55        already had that setting); every failure raises with its code and
 56        leaves the box as it was. Holds until the box is reset.
 57        """
 58        resp = self._http.request(
 59            "POST",
 60            f"/admin/instances/{instance_id}/internet",
 61            json=InstanceInternetRequest(internet=internet, reason=reason).model_dump(mode="json"),
 62        )
 63        _raise_for_status(resp)
 64        return InstanceInternetResult.model_validate(resp.json())
 65
 66    def list(
 67        self,
 68        *,
 69        node_id: str = "",
 70        status: str = "",
 71        offset: int = 0,
 72        limit: int = 50,
 73    ) -> PagedList[InstanceInfo]:
 74        """Every live instance cluster-wide (no team-ownership filter)."""
 75        params: dict[str, Any] = {"offset": offset, "limit": limit}
 76        if node_id:
 77            params["node_id"] = node_id
 78        if status:
 79            params["status"] = status
 80        resp = self._http.request("GET", "/admin/instances", params=params)
 81        _raise_for_status(resp)
 82        return _extract_items(resp.json(), InstanceInfo)
 83
 84    def get(self, instance_id: str) -> InstanceInfo:
 85        """Single live instance, admin view (any team's)."""
 86        resp = self._http.request("GET", f"/admin/instances/{instance_id}")
 87        _raise_for_status(resp)
 88        return InstanceInfo.model_validate(resp.json())
 89
 90    def pcap(self, instance_id: str) -> bytes:
 91        """Live tcpdump capture for a running instance. Returns empty
 92        ``bytes`` when none exists (sidecar off, never ran) so callers
 93        can persist conditionally without try/except."""
 94        resp = self._http.request("GET", f"/admin/instances/{instance_id}/pcap")
 95        if resp.status_code == 404:
 96            return b""
 97        _raise_for_status(resp)
 98        return resp.content
 99
100    def traffic(self, instance_id: str, *, cursor: str = "", limit: int = 0) -> dict[str, Any]:
101        """Live mitmproxy traffic for any running instance (admin twin
102        of :meth:`InstancesResource.traffic`, no ownership check).
103
104        Same ``{flows, cursor, truncated, capture}`` envelope and the same
105        polling contract; see the player's method for why an empty
106        cursor is not a value to send on every call.
107        """
108        params: dict[str, Any] = {"cursor": cursor} if cursor else {}
109        if limit > 0:
110            params["limit"] = limit
111        resp = self._http.request("GET", f"/admin/instances/{instance_id}/traffic", params=params)
112        _raise_for_status(resp)
113        data = resp.json()
114        return data if isinstance(data, dict) else {}
115
116    def storage(self) -> InstanceStorageReport:
117        """Disk held by the per-instance archive and each node's captures (#1714).
118
119        The archive is the platform's own last reading; each capture row is
120        that node's, carried on its heartbeat. ``report is None`` on a row is
121        *not measured yet*, never an empty store.
122        """
123        resp = self._http.request("GET", "/admin/instance-storage")
124        _raise_for_status(resp)
125        return InstanceStorageReport.model_validate(resp.json())
126
127    def timings(
128        self,
129        *,
130        group_by: TimingGroup = "challenge",
131        competition_id: str = "",
132        window: int = 0,
133        image_kind: str = "",
134    ) -> AdminInstanceTimings:
135        """Where instance starts spend their time, per challenge or node (#2095).
136
137        Wraps ``GET /admin/instance-timings``: p50 / p95 of the wait and of
138        its phases over the starts the node measured, slowest first.
139        *window* is the last N seconds (``0`` = the whole archive).
140        """
141        params: dict[str, Any] = {"group_by": group_by}
142        if competition_id:
143            params["competition_id"] = competition_id
144        if window:
145            params["window"] = window
146        if image_kind:
147            params["image_kind"] = image_kind
148        resp = self._http.request("GET", "/admin/instance-timings", params=params)
149        _raise_for_status(resp)
150        return AdminInstanceTimings.model_validate(resp.json())

Every live instance cluster-wide, with forensic pcap / traffic reads. Archived (terminal) records live on AdminRecordsResource.

AdminInstancesResource(http: ctfy.sdk.base.BaseHttpClient) 
26    def __init__(self, http: BaseHttpClient) -> None:
27        self._http = http
def stop_all(self) -> list[dict[str, typing.Any]]: 
29    def stop_all(self) -> builtins.list[dict[str, Any]]:
30        """Cluster-wide tear-down: stop every running instance. Used for
31        maintenance windows. Returns one :class:`StopResponse` row per
32        instance that was stopped."""
33        resp = self._http.request("DELETE", "/instances")
34        _raise_for_status(resp)
35        rows: builtins.list[dict[str, Any]] = resp.json()
36        return rows

Cluster-wide tear-down: stop every running instance. Used for maintenance windows. Returns one StopResponse row per instance that was stopped.

def stop(self, instance_id: str) -> dict[str, typing.Any]: 
38    def stop(self, instance_id: str) -> dict[str, Any]:
39        """Stop one running instance, whoever owns it — the surgical
40        counterpart to :meth:`stop_all` for a wedged or abusive box.
41        The owner team's record is archived (logs captured) first."""
42        resp = self._http.request("DELETE", f"/admin/instances/{instance_id}")
43        _raise_for_status(resp)
44        row: dict[str, Any] = resp.json()
45        return row

Stop one running instance, whoever owns it — the surgical counterpart to stop_all() for a wedged or abusive box. The owner team's record is archived (logs captured) first.

def set_internet( self, instance_id: str, internet: ctfy.core.enums.InternetAccess, *, reason: str) -> ctfy.server.models.InstanceInternetResult: 
47    def set_internet(
48        self, instance_id: str, internet: InternetAccess, *, reason: str
49    ) -> InstanceInternetResult:
50        """Open or cut one running box's internet access (#2067).
51
52        The operator's escape hatch over the verdict the box was started
53        with: the owning node applies it first, then the platform records it
54        and audits *reason*. ``outcome`` is ``changed`` or ``unchanged`` (it
55        already had that setting); every failure raises with its code and
56        leaves the box as it was. Holds until the box is reset.
57        """
58        resp = self._http.request(
59            "POST",
60            f"/admin/instances/{instance_id}/internet",
61            json=InstanceInternetRequest(internet=internet, reason=reason).model_dump(mode="json"),
62        )
63        _raise_for_status(resp)
64        return InstanceInternetResult.model_validate(resp.json())

Open or cut one running box's internet access (#2067).

The operator's escape hatch over the verdict the box was started with: the owning node applies it first, then the platform records it and audits reason. outcome is changed or unchanged (it already had that setting); every failure raises with its code and leaves the box as it was. Holds until the box is reset.

def list( self, *, node_id: str = '', status: str = '', offset: int = 0, limit: int = 50) -> ctfy.sdk._helpers.PagedList[ctfy.server.models.InstanceInfo]: 
66    def list(
67        self,
68        *,
69        node_id: str = "",
70        status: str = "",
71        offset: int = 0,
72        limit: int = 50,
73    ) -> PagedList[InstanceInfo]:
74        """Every live instance cluster-wide (no team-ownership filter)."""
75        params: dict[str, Any] = {"offset": offset, "limit": limit}
76        if node_id:
77            params["node_id"] = node_id
78        if status:
79            params["status"] = status
80        resp = self._http.request("GET", "/admin/instances", params=params)
81        _raise_for_status(resp)
82        return _extract_items(resp.json(), InstanceInfo)

Every live instance cluster-wide (no team-ownership filter).

def get(self, instance_id: str) -> ctfy.server.models.InstanceInfo: 
84    def get(self, instance_id: str) -> InstanceInfo:
85        """Single live instance, admin view (any team's)."""
86        resp = self._http.request("GET", f"/admin/instances/{instance_id}")
87        _raise_for_status(resp)
88        return InstanceInfo.model_validate(resp.json())

Single live instance, admin view (any team's).

def pcap(self, instance_id: str) -> bytes: 
90    def pcap(self, instance_id: str) -> bytes:
91        """Live tcpdump capture for a running instance. Returns empty
92        ``bytes`` when none exists (sidecar off, never ran) so callers
93        can persist conditionally without try/except."""
94        resp = self._http.request("GET", f"/admin/instances/{instance_id}/pcap")
95        if resp.status_code == 404:
96            return b""
97        _raise_for_status(resp)
98        return resp.content

Live tcpdump capture for a running instance. Returns empty bytes when none exists (sidecar off, never ran) so callers can persist conditionally without try/except.

def traffic( self, instance_id: str, *, cursor: str = '', limit: int = 0) -> dict[str, typing.Any]: 
100    def traffic(self, instance_id: str, *, cursor: str = "", limit: int = 0) -> dict[str, Any]:
101        """Live mitmproxy traffic for any running instance (admin twin
102        of :meth:`InstancesResource.traffic`, no ownership check).
103
104        Same ``{flows, cursor, truncated, capture}`` envelope and the same
105        polling contract; see the player's method for why an empty
106        cursor is not a value to send on every call.
107        """
108        params: dict[str, Any] = {"cursor": cursor} if cursor else {}
109        if limit > 0:
110            params["limit"] = limit
111        resp = self._http.request("GET", f"/admin/instances/{instance_id}/traffic", params=params)
112        _raise_for_status(resp)
113        data = resp.json()
114        return data if isinstance(data, dict) else {}

Live mitmproxy traffic for any running instance (admin twin of InstancesResource.traffic(), no ownership check).

Same {flows, cursor, truncated, capture} envelope and the same polling contract; see the player's method for why an empty cursor is not a value to send on every call.

def storage(self) -> ctfy.server.models.InstanceStorageReport: 
116    def storage(self) -> InstanceStorageReport:
117        """Disk held by the per-instance archive and each node's captures (#1714).
118
119        The archive is the platform's own last reading; each capture row is
120        that node's, carried on its heartbeat. ``report is None`` on a row is
121        *not measured yet*, never an empty store.
122        """
123        resp = self._http.request("GET", "/admin/instance-storage")
124        _raise_for_status(resp)
125        return InstanceStorageReport.model_validate(resp.json())

Disk held by the per-instance archive and each node's captures (#1714).

The archive is the platform's own last reading; each capture row is that node's, carried on its heartbeat. report is None on a row is not measured yet, never an empty store.

def timings( self, *, group_by: Literal['challenge', 'node'] = 'challenge', competition_id: str = '', window: int = 0, image_kind: str = '') -> ctfy.server.models.AdminInstanceTimings: 
127    def timings(
128        self,
129        *,
130        group_by: TimingGroup = "challenge",
131        competition_id: str = "",
132        window: int = 0,
133        image_kind: str = "",
134    ) -> AdminInstanceTimings:
135        """Where instance starts spend their time, per challenge or node (#2095).
136
137        Wraps ``GET /admin/instance-timings``: p50 / p95 of the wait and of
138        its phases over the starts the node measured, slowest first.
139        *window* is the last N seconds (``0`` = the whole archive).
140        """
141        params: dict[str, Any] = {"group_by": group_by}
142        if competition_id:
143            params["competition_id"] = competition_id
144        if window:
145            params["window"] = window
146        if image_kind:
147            params["image_kind"] = image_kind
148        resp = self._http.request("GET", "/admin/instance-timings", params=params)
149        _raise_for_status(resp)
150        return AdminInstanceTimings.model_validate(resp.json())

Where instance starts spend their time, per challenge or node (#2095).

Wraps GET /admin/instance-timings: p50 / p95 of the wait and of its phases over the starts the node measured, slowest first. window is the last N seconds (0 = the whole archive).