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 _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 list(
30        self,
31        *,
32        node_id: str = "",
33        status: str = "",
34        offset: int = 0,
35        limit: int = 50,
36    ) -> builtins.list[InstanceInfo]:
37        """Every live instance cluster-wide (no team-ownership filter)."""
38        params: dict[str, Any] = {"offset": offset, "limit": limit}
39        if node_id:
40            params["node_id"] = node_id
41        if status:
42            params["status"] = status
43        resp = self._http.request("GET", "/admin/instances", params=params)
44        _raise_for_status(resp)
45        return _extract_items(resp.json(), InstanceInfo)
46
47    def get(self, instance_id: str) -> InstanceInfo:
48        """Single live instance, admin view (any team's)."""
49        resp = self._http.request("GET", f"/admin/instances/{instance_id}")
50        _raise_for_status(resp)
51        return InstanceInfo.model_validate(resp.json())
52
53    def pcap(self, instance_id: str) -> bytes:
54        """Live tcpdump capture for a running instance. Returns empty
55        ``bytes`` when none exists (sidecar off, never ran) so callers
56        can persist conditionally without try/except."""
57        resp = self._http.request("GET", f"/admin/instances/{instance_id}/pcap")
58        if resp.status_code == 404:
59            return b""
60        _raise_for_status(resp)
61        return resp.content
62
63    def traffic(self, instance_id: str) -> dict[str, Any]:
64        """Live mitmproxy traffic for any running instance (admin twin
65        of :meth:`InstancesResource.traffic`, no ownership check)."""
66        resp = self._http.request("GET", f"/admin/instances/{instance_id}/traffic")
67        _raise_for_status(resp)
68        data = resp.json()
69        return data if isinstance(data, dict) else {}