Coverage for node / src / stigmem_node / cli / federation.py: 88%
142 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-06-18 05:34 +0000
1"""Federation CLI handlers."""
3from __future__ import annotations
5import argparse
6import sys
7from typing import Any
10def _dnssec_pending_base_url(args: argparse.Namespace) -> str:
11 """Resolve the local node base URL for the dnssec first-trust admin API."""
12 from ..settings import settings
14 return (args.node_url or settings.node_url).rstrip("/")
17def _dnssec_auth_headers(args: argparse.Namespace) -> dict[str, str]:
18 headers = {"Content-Type": "application/json"}
19 if args.api_key: 19 ↛ 21line 19 didn't jump to line 21 because the condition on line 19 was always true
20 headers["Authorization"] = f"Bearer {args.api_key}"
21 return headers
24def _cmd_federation_dnssec_pending(args: argparse.Namespace) -> int:
25 """List quarantined DNSSEC first-trust candidates (operator-confirm queue).
27 Calls ``GET /v1/federation/dnssec/pending`` on the local node (admin-gated).
28 """
29 import json
31 import httpx
33 base = _dnssec_pending_base_url(args)
34 try:
35 resp = httpx.get(
36 f"{base}/v1/federation/dnssec/pending",
37 headers=_dnssec_auth_headers(args),
38 timeout=15.0,
39 )
40 except Exception as exc:
41 print(f"error: cannot reach node at {base}: {exc}", file=sys.stderr)
42 return 1
44 if resp.status_code != 200: 44 ↛ 45line 44 didn't jump to line 45 because the condition on line 44 was never true
45 print(f"error: node returned {resp.status_code}: {resp.text}", file=sys.stderr)
46 return 1
48 pending: list[dict[str, Any]] = resp.json().get("pending", [])
49 print(json.dumps({"pending": pending}, indent=2))
50 if not pending: 50 ↛ 51line 50 didn't jump to line 51 because the condition on line 50 was never true
51 print("no pending first-trust candidates", file=sys.stderr)
52 return 0
55def _cmd_federation_dnssec_confirm(args: argparse.Namespace) -> int:
56 """Confirm a quarantined DNSSEC first-trust candidate (paste-to-confirm).
58 Calls ``POST /v1/federation/dnssec/pending/confirm`` on the local node. The
59 operator-supplied ``--key-fpr`` MUST byte-equal the stored candidate
60 fingerprint (NF-D4-5); a mismatch is rejected by the node (no trust) and this
61 command exits non-zero.
62 """
63 import json
65 import httpx
67 base = _dnssec_pending_base_url(args)
68 payload = {
69 "entity_uri": args.entity_uri,
70 "node_id": args.node_id,
71 "key_fpr": args.key_fpr,
72 }
73 try:
74 resp = httpx.post(
75 f"{base}/v1/federation/dnssec/pending/confirm",
76 json=payload,
77 headers=_dnssec_auth_headers(args),
78 timeout=15.0,
79 )
80 except Exception as exc:
81 print(f"error: cannot reach node at {base}: {exc}", file=sys.stderr)
82 return 1
84 if resp.status_code in (200, 201):
85 print(json.dumps(resp.json(), indent=2))
86 print("first-trust candidate confirmed and pinned", file=sys.stderr)
87 return 0
88 if resp.status_code == 422:
89 print(
90 "error: fingerprint did not match the quarantined candidate — not trusted",
91 file=sys.stderr,
92 )
93 return 1
94 if resp.status_code == 404: 94 ↛ 97line 94 didn't jump to line 97 because the condition on line 94 was always true
95 print("error: no such pending first-trust candidate", file=sys.stderr)
96 return 1
97 print(f"error: node returned {resp.status_code}: {resp.text}", file=sys.stderr)
98 return 1
101def _cmd_federation_dnssec_reject(args: argparse.Namespace) -> int:
102 """Reject a quarantined DNSSEC first-trust candidate WITHOUT trusting it.
104 Calls ``POST /v1/federation/dnssec/pending/reject`` on the local node.
105 """
106 import json
108 import httpx
110 base = _dnssec_pending_base_url(args)
111 payload = {"entity_uri": args.entity_uri, "node_id": args.node_id}
112 try:
113 resp = httpx.post(
114 f"{base}/v1/federation/dnssec/pending/reject",
115 json=payload,
116 headers=_dnssec_auth_headers(args),
117 timeout=15.0,
118 )
119 except Exception as exc:
120 print(f"error: cannot reach node at {base}: {exc}", file=sys.stderr)
121 return 1
123 if resp.status_code in (200, 204):
124 if resp.status_code == 200 and resp.text: 124 ↛ 126line 124 didn't jump to line 126 because the condition on line 124 was always true
125 print(json.dumps(resp.json(), indent=2))
126 print("first-trust candidate rejected", file=sys.stderr)
127 return 0
128 if resp.status_code == 404: 128 ↛ 131line 128 didn't jump to line 131 because the condition on line 128 was always true
129 print("error: no such pending first-trust candidate", file=sys.stderr)
130 return 1
131 print(f"error: node returned {resp.status_code}: {resp.text}", file=sys.stderr)
132 return 1
135def _cmd_federation_register_peer(args: argparse.Namespace) -> int:
136 """Register this node as a peer with a remote node (Spec-05-Federation-Trust)."""
137 import base64
138 import json
139 import ssl
140 from datetime import UTC, datetime
142 import httpx
144 from ..db import apply_migrations
145 from ..settings import settings
147 # Ensure migrations are applied so keypair tables exist.
148 apply_migrations()
150 # Resolve local node URL: explicit flag > settings.
151 local_url = (args.local_url or settings.node_url).rstrip("/")
152 remote_url = args.remote_url.rstrip("/")
153 allowed_scopes: list[str] = [s.strip() for s in args.scopes.split(",") if s.strip()]
154 cert = (args.tls_cert, args.tls_key) if args.tls_cert and args.tls_key else None
155 verify: ssl.SSLContext | str | bool | None = None
156 if cert is not None:
157 ssl_ctx = ssl.create_default_context(cafile=args.ca_bundle or None)
158 ssl_ctx.load_cert_chain(*cert)
159 verify = ssl_ctx
160 elif args.ca_bundle: 160 ↛ 161line 160 didn't jump to line 161 because the condition on line 160 was never true
161 verify = args.ca_bundle
163 # ------------------------------------------------------------------
164 # 1. Fetch local /.well-known/stigmem to get our published metadata.
165 # ------------------------------------------------------------------
166 try:
167 if verify is not None:
168 with httpx.Client(timeout=15.0, trust_env=False, verify=verify) as client:
169 wk = client.get(f"{local_url}/.well-known/stigmem")
170 else:
171 wk = httpx.get(f"{local_url}/.well-known/stigmem", timeout=10.0)
172 wk.raise_for_status()
173 except Exception as exc:
174 print(f"error: cannot reach local node at {local_url}: {exc}", file=sys.stderr)
175 return 1
177 wk_data = wk.json()
178 local_node_id: str = wk_data["node_id"]
179 local_pubkey: str = wk_data.get("federation_pubkey", "")
180 if not local_pubkey:
181 print(
182 "error: local node has no federation_pubkey in /.well-known/stigmem — "
183 "set STIGMEM_FEDERATION_ENABLED=true and restart",
184 file=sys.stderr,
185 )
186 return 1
188 # ------------------------------------------------------------------
189 # 2. Load local private key and sign the PeerDeclaration.
190 # ------------------------------------------------------------------
191 from ..federation.peer_token import init_federation_keys
193 _, priv_b64 = init_federation_keys()
195 def _pad(s: str) -> str:
196 return s + "=" * (-len(s) % 4)
198 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
200 priv_key = Ed25519PrivateKey.from_private_bytes(base64.urlsafe_b64decode(_pad(priv_b64)))
202 signed_at = datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%SZ")
203 signed_fields: dict[str, object] = {
204 "allowed_scopes": sorted(allowed_scopes),
205 "federation_pubkey": local_pubkey,
206 "node_id": local_node_id,
207 "node_url": local_url,
208 "signed_at": signed_at,
209 }
210 canonical = json.dumps(signed_fields, sort_keys=True, separators=(",", ":")).encode("utf-8")
211 sig_bytes = priv_key.sign(canonical)
212 declaration_sig = base64.urlsafe_b64encode(sig_bytes).decode().rstrip("=")
214 # ------------------------------------------------------------------
215 # 3. POST to the remote node.
216 # ------------------------------------------------------------------
217 payload = {
218 "node_id": local_node_id,
219 "node_url": local_url,
220 "federation_pubkey": local_pubkey,
221 "allowed_scopes": sorted(allowed_scopes),
222 "signed_at": signed_at,
223 "declaration_sig": declaration_sig,
224 }
226 headers = {"Content-Type": "application/json"}
227 if args.api_key:
228 headers["Authorization"] = f"Bearer {args.api_key}"
230 try:
231 if verify is not None:
232 with httpx.Client(timeout=15.0, trust_env=False, verify=verify) as client:
233 resp = client.post(
234 f"{remote_url}/v1/federation/peers",
235 json=payload,
236 headers=headers,
237 )
238 else:
239 resp = httpx.post(
240 f"{remote_url}/v1/federation/peers",
241 json=payload,
242 headers=headers,
243 timeout=15.0,
244 )
245 except Exception as exc:
246 print(f"error: cannot reach remote node at {remote_url}: {exc}", file=sys.stderr)
247 return 1
249 if resp.status_code in (200, 201):
250 result = resp.json()
251 peer_status = result.get("status", "unknown")
252 peer_id = result.get("peer_id", "")
253 if peer_status == "active":
254 print(f"peer registered and verified (peer_id={peer_id})")
255 else:
256 print(
257 f"peer registered but not yet active (status={peer_status}, peer_id={peer_id})\n"
258 "Check that the remote node can reach this node's /.well-known/stigmem endpoint.",
259 file=sys.stderr,
260 )
261 return 1
262 elif resp.status_code == 409:
263 print("peer already registered — nothing to do")
264 else:
265 print(
266 f"error: remote node returned {resp.status_code}: {resp.text}",
267 file=sys.stderr,
268 )
269 return 1
271 return 0