ctfy.sdk.admin_resources.nodes

  1"""``client.admin.nodes`` — worker-node fleet management + bootstrap invites.
  2
  3The non-admin ``/nodes`` registration routes are operator-only in practice
  4(no player registers a worker node), so ``register`` / ``remove`` live here
  5alongside the rest of the fleet tooling. The public node *list* is on
  6:meth:`ctfy.sdk.resources.nodes.NodesResource.list`.
  7"""
  8
  9from __future__ import annotations
 10
 11from typing import Any
 12
 13from ctfy.core.node_join import NodeProfile
 14from ctfy.core.state.models import NodeHealthSample
 15from ctfy.sdk._helpers import PagedList, _extract_items, _raise_for_status
 16from ctfy.sdk.base import BaseHttpClient
 17from ctfy.server.models import (
 18    Activity,
 19    CreateInviteResponse,
 20    CreateJoinCommandResponse,
 21    FleetReachAck,
 22    FleetReachAckCleared,
 23    FleetReachAckResult,
 24    FleetReachManifest,
 25    FleetReachToken,
 26    NodeAddress,
 27    NodeFeatures,
 28    NodeInfo,
 29    NodeInviteInfo,
 30)
 31
 32
 33class AdminNodesResource:
 34    """Register / remove / patch worker nodes, read health + events, and
 35    mint the one-time bootstrap invites a fresh node redeems."""
 36
 37    def __init__(self, http: BaseHttpClient) -> None:
 38        self._http = http
 39
 40    def register(
 41        self, url: str, capacity: int = 10, labels: dict[str, str] | None = None
 42    ) -> NodeInfo:
 43        resp = self._http.request(
 44            "POST",
 45            "/nodes/register",
 46            json={
 47                "url": url,
 48                "capacity": capacity,
 49                "labels": labels or {},
 50            },
 51        )
 52        _raise_for_status(resp)
 53        return NodeInfo.model_validate(resp.json())
 54
 55    def remove(self, node_id: str) -> None:
 56        resp = self._http.request("DELETE", f"/nodes/{node_id}")
 57        _raise_for_status(resp)
 58
 59    def get(self, node_id: str) -> NodeInfo:
 60        """Admin-only single-node lookup (full health + capacity)."""
 61        resp = self._http.request("GET", f"/admin/nodes/{node_id}")
 62        _raise_for_status(resp)
 63        return NodeInfo.model_validate(resp.json())
 64
 65    def patch(self, node_id: str, **fields: Any) -> NodeInfo:
 66        """Admin-only: update editable node metadata. Pass any subset of
 67        ``display_name`` / ``labels`` / ``cordoned`` (drain the node, #1900). Omit a
 68        field to leave unchanged. Wraps
 69        ``PATCH /admin/nodes/{id}``."""
 70        resp = self._http.request("PATCH", f"/admin/nodes/{node_id}", json=fields)
 71        _raise_for_status(resp)
 72        return NodeInfo.model_validate(resp.json())
 73
 74    def approve(
 75        self,
 76        node_id: str,
 77        profile: NodeProfile = NodeProfile.NORMAL,
 78        *,
 79        competition_id: str = "",
 80        disabled_features: list[str] | None = None,
 81        url: str = "",
 82    ) -> NodeInfo:
 83        """Admin-only: approve a pending node into *profile*'s pool (#3037),
 84        with *disabled_features* switched off, and its address corrected to
 85        *url* when given (held from then on). Wraps
 86        ``POST /admin/nodes/{id}/approve``."""
 87        resp = self._http.request(
 88            "POST",
 89            f"/admin/nodes/{node_id}/approve",
 90            json={
 91                "profile": str(profile),
 92                "competition_id": competition_id,
 93                "disabled_features": disabled_features or [],
 94                "url": url,
 95            },
 96        )
 97        _raise_for_status(resp)
 98        return NodeInfo.model_validate(resp.json())
 99
100    def features(self, node_id: str) -> NodeFeatures:
101        """Admin-only: per feature, what the node supports and whether the admin
102        enabled it (#3037). Wraps ``GET /admin/nodes/{id}/features``; switch one
103        with ``patch(node_id, disabled_features=[...])``."""
104        resp = self._http.request("GET", f"/admin/nodes/{node_id}/features")
105        _raise_for_status(resp)
106        return NodeFeatures.model_validate(resp.json())
107
108    def fleet_reach(self) -> FleetReachManifest:
109        """Admin-only: what an agent on one node must reach on every other,
110        derived from the node rows (#3428). Wraps ``GET /admin/fleet-reach``;
111        the cb host renders its VM firewall accepts from the same body."""
112        resp = self._http.request("GET", "/admin/fleet-reach")
113        _raise_for_status(resp)
114        return FleetReachManifest.model_validate(resp.json())
115
116    def ack_fleet_reach(self, ack: FleetReachAck) -> FleetReachAckResult:
117        """Record that a renderer applied *ack*'s manifest for its agent pools,
118        which placement then derives reach from (#3428). The cb timer calls
119        this after a converged apply. Wraps ``POST /admin/fleet-reach/ack``."""
120        resp = self._http.request(
121            "POST", "/admin/fleet-reach/ack", json=ack.model_dump(mode="json")
122        )
123        _raise_for_status(resp)
124        return FleetReachAckResult.model_validate(resp.json())
125
126    def clear_fleet_reach_ack(self, pool: str = "") -> FleetReachAckCleared:
127        """Forget the stored ack of *pool*, or of every pool when ``""``
128        (#3428 rollback): placement then reads it as unproven (fails open)
129        until the renderer acks again. Admin only. Wraps
130        ``DELETE /admin/fleet-reach/ack``."""
131        params = {"pool": pool} if pool else None
132        resp = self._http.request("DELETE", "/admin/fleet-reach/ack", params=params)
133        _raise_for_status(resp)
134        return FleetReachAckCleared.model_validate(resp.json())
135
136    def mint_fleet_reach_token(self) -> FleetReachToken:
137        """Mint the cb renderer's bearer, scoped to the fleet reach routes
138        (#3428). Shown once; replaces (voids) any previous one. Wraps
139        ``POST /admin/fleet-reach/token``."""
140        resp = self._http.request("POST", "/admin/fleet-reach/token")
141        _raise_for_status(resp)
142        return FleetReachToken.model_validate(resp.json())
143
144    def revoke_fleet_reach_token(self) -> None:
145        """Void the cb renderer's bearer. Idempotent. Wraps
146        ``DELETE /admin/fleet-reach/token``."""
147        resp = self._http.request("DELETE", "/admin/fleet-reach/token")
148        _raise_for_status(resp)
149
150    def address(self, node_id: str, *, check: bool = False) -> NodeAddress:
151        """Admin-only: the node's claimed address, the source address the
152        platform saw and its UDP readings (#3037). *check* probes the UDP
153        surfaces first. Wraps ``GET /admin/nodes/{id}/address`` and
154        ``POST /admin/nodes/{id}/address/check``."""
155        if check:
156            resp = self._http.request("POST", f"/admin/nodes/{node_id}/address/check")
157        else:
158            resp = self._http.request("GET", f"/admin/nodes/{node_id}/address")
159        _raise_for_status(resp)
160        return NodeAddress.model_validate(resp.json())
161
162    def reject(self, node_id: str) -> None:
163        """Admin-only: reject a pending node — its row and token are removed
164        (#3037). Wraps ``POST /admin/nodes/{id}/reject``."""
165        resp = self._http.request("POST", f"/admin/nodes/{node_id}/reject")
166        _raise_for_status(resp)
167
168    def health_history(
169        self, node_id: str, offset: int = 0, limit: int = 50
170    ) -> list[NodeHealthSample]:
171        """Recent heartbeat samples for one node — for the node detail page."""
172        resp = self._http.request(
173            "GET",
174            f"/admin/nodes/{node_id}/health-history",
175            params={"offset": offset, "limit": limit},
176        )
177        _raise_for_status(resp)
178        return [NodeHealthSample.model_validate(s) for s in resp.json()["items"]]
179
180    def events(self, node_id: str, *, offset: int = 0, limit: int = 50) -> PagedList[Activity]:
181        """Activity rows where ``actor`` or subject is this node."""
182        resp = self._http.request(
183            "GET",
184            f"/admin/nodes/{node_id}/events",
185            params={"offset": offset, "limit": limit},
186        )
187        _raise_for_status(resp)
188        return _extract_items(resp.json(), Activity)
189
190    def list_invites(self, offset: int = 0, limit: int = 50) -> PagedList[NodeInviteInfo]:
191        """Outstanding one-time worker-node bootstrap invites."""
192        resp = self._http.request(
193            "GET", "/nodes/invites", params={"offset": offset, "limit": limit}
194        )
195        _raise_for_status(resp)
196        return _extract_items(resp.json(), NodeInviteInfo)
197
198    def mint_invite(self, ttl_seconds: int = 24 * 3600) -> CreateInviteResponse:
199        """Mint a one-time node bootstrap token. A node redeems it once
200        to obtain its long-lived per-node bearer."""
201        resp = self._http.request("POST", "/nodes/invites", json={"ttl_seconds": ttl_seconds})
202        _raise_for_status(resp)
203        return CreateInviteResponse.model_validate(resp.json())
204
205    def mint_join_command(
206        self,
207        profile: NodeProfile = NodeProfile.NORMAL,
208        *,
209        competition_id: str = "",
210        display_name: str = "",
211        capacity: int = 0,
212        ttl_seconds: int = 3600,
213    ) -> CreateJoinCommandResponse:
214        """Mint a one-time invite wrapped as the single command a bare machine
215        runs (#2115). The plaintext rides the returned ``command`` / ``join_code``
216        once and is never listed again. Wraps ``POST /nodes/join-commands``."""
217        resp = self._http.request(
218            "POST",
219            "/nodes/join-commands",
220            json={
221                "profile": str(profile),
222                "competition_id": competition_id,
223                "display_name": display_name,
224                "capacity": capacity,
225                "ttl_seconds": ttl_seconds,
226            },
227        )
228        _raise_for_status(resp)
229        return CreateJoinCommandResponse.model_validate(resp.json())
230
231    def revoke_invite(self, invite_id: str) -> None:
232        """Revoke an unredeemed node invite. Idempotent."""
233        resp = self._http.request("DELETE", f"/nodes/invites/{invite_id}")
234        _raise_for_status(resp)