ctfy.sdk.admin_resources.machines

 1"""``client.admin.machines`` -- machine logins an organiser hands to teams (#2132).
 2
 3The platform carries the login to each team's captain and does nothing
 4else: it never connects to, initialises, resets or monitors a machine.
 5Organisers only (global admins and the event's own competition admins).
 6Nothing here returns a secret: the organiser holds the CSV already.
 7"""
 8
 9from __future__ import annotations
10
11from ctfy.sdk._helpers import _raise_for_status
12from ctfy.sdk.base import BaseHttpClient
13from ctfy.server.models import (
14    CompetitionMachineRow,
15    CompetitionMachines,
16    MachineAssignOutcome,
17    MachineAssignResult,
18    MachineImportResult,
19    MachineNotifyResult,
20)
21
22
23class AdminMachinesResource:
24    def __init__(self, http: BaseHttpClient) -> None:
25        self._http = http
26
27    def list(self, competition_id: str) -> CompetitionMachines:
28        """Every machine of the competition and who holds it (no secrets)."""
29        resp = self._http.request("GET", f"/admin/competitions/{competition_id}/machines")
30        _raise_for_status(resp)
31        return CompetitionMachines.model_validate(resp.json())
32
33    def import_csv(
34        self, competition_id: str, csv_text: str, *, auto_assign: bool = False
35    ) -> MachineImportResult:
36        """Import ``host,port,username,password,private_key,note,team`` rows.
37
38        Rows that cannot be read are reported by line and code and the rest
39        still import; a machine already on the list is skipped. With
40        ``auto_assign`` the new machines are served to the teams without one.
41        """
42        resp = self._http.request(
43            "POST",
44            f"/admin/competitions/{competition_id}/machines/import",
45            json={"csv": csv_text, "auto_assign": auto_assign},
46        )
47        _raise_for_status(resp)
48        return MachineImportResult.model_validate(resp.json())
49
50    def auto_assign(self, competition_id: str) -> MachineAssignResult:
51        """Serve the unassigned machines to the teams that have none."""
52        resp = self._http.request(
53            "POST", f"/admin/competitions/{competition_id}/machines/auto-assign"
54        )
55        _raise_for_status(resp)
56        return MachineAssignResult.model_validate(resp.json())
57
58    def assign(
59        self, competition_id: str, machine_id: str, team_id: str, *, replace: bool = False
60    ) -> MachineAssignOutcome:
61        """Give an unassigned machine to a team.
62
63        A team that already holds one is refused (``409 team_has_machine``)
64        unless ``replace``: the old machine is then retired at once and the
65        captain is told.
66        """
67        resp = self._http.request(
68            "PUT",
69            f"/admin/competitions/{competition_id}/machines/{machine_id}/team",
70            json={"team_id": team_id, "replace": replace},
71        )
72        _raise_for_status(resp)
73        return MachineAssignOutcome.model_validate(resp.json())
74
75    def revoke(self, competition_id: str, machine_id: str) -> CompetitionMachineRow:
76        """Retire a machine; the team that held it loses it immediately."""
77        resp = self._http.request(
78            "DELETE", f"/admin/competitions/{competition_id}/machines/{machine_id}"
79        )
80        _raise_for_status(resp)
81        return CompetitionMachineRow.model_validate(resp.json())
82
83    def notify(self, competition_id: str, *, mail: bool = True) -> MachineNotifyResult:
84        """Tell every captain whose machine they have not been told about.
85
86        An inbox card with a link to the team page, and with ``mail`` an
87        email carrying the same link -- never the login.
88        """
89        resp = self._http.request(
90            "POST", f"/admin/competitions/{competition_id}/machines/notify", json={"mail": mail}
91        )
92        _raise_for_status(resp)
93        return MachineNotifyResult.model_validate(resp.json())