ctfy.sdk.admin_resources.instances

 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.sdk._helpers import PagedList, _extract_items, _raise_for_status
 9from ctfy.sdk.base import BaseHttpClient
10from ctfy.server.models import InstanceInfo
11
12
13class AdminInstancesResource:
14    """Every live instance cluster-wide, with forensic pcap / traffic reads.
15    Archived (terminal) records live on :class:`AdminRecordsResource`."""
16
17    def __init__(self, http: BaseHttpClient) -> None:
18        self._http = http
19
20    def stop_all(self) -> builtins.list[dict[str, Any]]:
21        """Cluster-wide tear-down: stop every running instance. Used for
22        maintenance windows. Returns one :class:`StopResponse` row per
23        instance that was stopped."""
24        resp = self._http.request("DELETE", "/instances")
25        _raise_for_status(resp)
26        rows: builtins.list[dict[str, Any]] = resp.json()
27        return rows
28
29    def stop(self, instance_id: str) -> dict[str, Any]:
30        """Stop one running instance, whoever owns it — the surgical
31        counterpart to :meth:`stop_all` for a wedged or abusive box.
32        The owner team's record is archived (logs captured) first."""
33        resp = self._http.request("DELETE", f"/admin/instances/{instance_id}")
34        _raise_for_status(resp)
35        row: dict[str, Any] = resp.json()
36        return row
37
38    def list(
39        self,
40        *,
41        node_id: str = "",
42        status: str = "",
43        offset: int = 0,
44        limit: int = 50,
45    ) -> PagedList[InstanceInfo]:
46        """Every live instance cluster-wide (no team-ownership filter)."""
47        params: dict[str, Any] = {"offset": offset, "limit": limit}
48        if node_id:
49            params["node_id"] = node_id
50        if status:
51            params["status"] = status
52        resp = self._http.request("GET", "/admin/instances", params=params)
53        _raise_for_status(resp)
54        return _extract_items(resp.json(), InstanceInfo)
55
56    def get(self, instance_id: str) -> InstanceInfo:
57        """Single live instance, admin view (any team's)."""
58        resp = self._http.request("GET", f"/admin/instances/{instance_id}")
59        _raise_for_status(resp)
60        return InstanceInfo.model_validate(resp.json())
61
62    def pcap(self, instance_id: str) -> bytes:
63        """Live tcpdump capture for a running instance. Returns empty
64        ``bytes`` when none exists (sidecar off, never ran) so callers
65        can persist conditionally without try/except."""
66        resp = self._http.request("GET", f"/admin/instances/{instance_id}/pcap")
67        if resp.status_code == 404:
68            return b""
69        _raise_for_status(resp)
70        return resp.content
71
72    def traffic(self, instance_id: str, *, cursor: str = "", limit: int = 0) -> dict[str, Any]:
73        """Live mitmproxy traffic for any running instance (admin twin
74        of :meth:`InstancesResource.traffic`, no ownership check).
75
76        Same ``{flows, cursor, truncated, capture}`` envelope and the same
77        polling contract; see the player's method for why an empty
78        cursor is not a value to send on every call.
79        """
80        params: dict[str, Any] = {"cursor": cursor} if cursor else {}
81        if limit > 0:
82            params["limit"] = limit
83        resp = self._http.request("GET", f"/admin/instances/{instance_id}/traffic", params=params)
84        _raise_for_status(resp)
85        data = resp.json()
86        return data if isinstance(data, dict) else {}