Coverage for node / src / stigmem_node / routes / federation / common.py: 77%

1"""Shared helpers and compatibility exports for federation route modules.""" 

2 

3from __future__ import annotations 

4 

5import json 

6import logging 

7import sys 

8from typing import Annotated, Any 

9 

10from fastapi import APIRouter, Depends, Header, HTTPException, Request 

11 

12from ...db import db 

13from ...federation.peer_token import TokenError, verify_peer_token 

14from ...federation.tls import check_peer_san 

15 

16logger = logging.getLogger("stigmem.federation") 

17 

18router = APIRouter(tags=["federation"]) 

19 

20 

21def _public_module() -> Any: 

22 """Return the public federation module so test monkey-patches stay visible.""" 

23 return sys.modules["stigmem_node.routes.federation"] 

24 

25def _allowed_output_scopes( 

26 peer: dict[str, Any], 

27 token_payload: dict[str, Any], 

28 origin_allowed_scopes: list[str] | None = None, 

29) -> set[str]: 

30 """Scopes this peer may receive: peer.allowed_scopes ∩ token.scopes (§5.8). 

31 

32 For a RELAYED fact (F-FED-2c W2.3) the origin's signed propagation grant 

33 further constrains the set: a relayed fact may only egress for scopes the 

34 ORIGIN authorised. Pass ``origin_allowed_scopes`` (the per-fact grant) to 

35 additionally intersect with it. For self-originated facts leave it ``None`` 

36 — the result is then exactly the 2b behaviour (peer ∩ token only). 

37 

38 ``local`` is always stripped; ``team`` is stripped unless 

39 ``federation_allow_team`` is set. These node-policy filters apply to the 

40 origin grant too — a relay never widens scope beyond local node policy. 

41 """ 

42 peer_scopes = set(json.loads(peer["allowed_scopes"])) 

43 token_scopes = set(token_payload.get("scopes", [])) 

44 combined = peer_scopes & token_scopes 

45 if origin_allowed_scopes is not None: 45 ↛ 46line 45 didn't jump to line 46 because the condition on line 45 was never true

46 combined &= set(origin_allowed_scopes) 

47 combined.discard("local") 

48 if not _public_module().settings.federation_allow_team: 48 ↛ 50line 48 didn't jump to line 50 because the condition on line 48 was always true

49 combined.discard("team") 

50 return combined 

51 

52 

53def _allowed_output_tenants( 

54 peer: dict[str, Any], 

55 origin_allowed_tenants: list[str] | None = None, 

56) -> set[str]: 

57 """Tenants this peer is authorised to receive facts for (F-FED-2c W2.3). 

58 

59 Parallels ``_allowed_output_scopes`` on the tenant axis. The base set is the 

60 peer's declared ``allowed_tenants`` (migration 041, JSON array); when that 

61 column is NULL/empty the peer's resolved ``pull_tenant`` (or ``"default"``) 

62 is the sole authorised tenant. For a RELAYED fact the origin's signed 

63 ``origin_allowed_tenants`` grant additionally intersects: a relayed fact may 

64 only egress when the origin authorised at least one tenant the peer is 

65 allowed to receive. Pass ``None`` (self-originated) to skip the origin 

66 constraint. 

67 """ 

68 raw = peer.get("allowed_tenants") 

69 peer_tenants: set[str] = set(json.loads(raw)) if raw else set() 

70 if not peer_tenants: 70 ↛ 71line 70 didn't jump to line 71 because the condition on line 70 was never true

71 peer_tenants = {peer.get("pull_tenant") or "default"} 

72 if origin_allowed_tenants is not None: 72 ↛ 73line 72 didn't jump to line 73 because the condition on line 72 was never true

73 peer_tenants &= set(origin_allowed_tenants) 

74 return peer_tenants 

75 

76 

77# --------------------------------------------------------------------------- 

78# Peer-token dependency 

79# --------------------------------------------------------------------------- 

80 

81 

82def _get_mtls_peer_cert(request: Request) -> dict[str, Any]: 

83 """Extract the TLS peer certificate dict from the ASGI transport (uvicorn). 

84 

85 Returns an empty dict when not running under TLS (tests, plaintext mode). 

86 """ 

87 transport = request.scope.get("transport") 

88 if transport is None: 

89 return {} 

90 ssl_obj = transport.get_extra_info("ssl_object") 

91 if ssl_obj is None: 91 ↛ 92line 91 didn't jump to line 92 because the condition on line 91 was never true

92 return {} 

93 return ssl_obj.getpeercert() or {} 

94 

95 

96def _require_peer_token( 

97 request: Request, 

98 authorization: Annotated[str | None, Header(alias="authorization")] = None, 

99) -> tuple[dict[str, Any], dict[str, Any]]: 

100 """Verify incoming peer token. Returns (peer_dict, token_payload) or raises 401.""" 

101 if authorization is None or not authorization.lower().startswith("bearer "): 

102 raise HTTPException(status_code=401, detail="peer token required") 

103 

104 raw_token = authorization[7:] 

105 

106 # Decode header without sig verification to extract iss 

107 import jwt as _jwt 

108 

109 try: 

110 # exp/iat are epoch_ms per spec §3.5; disable all claim validation for header-only peek 

111 unverified: dict[str, Any] = _jwt.decode( 

112 raw_token, 

113 options={ 

114 "verify_signature": False, 

115 "verify_exp": False, 

116 "verify_iat": False, 

117 "verify_nbf": False, 

118 "verify_aud": False, 

119 }, 

120 algorithms=["EdDSA"], 

121 ) 

122 except Exception as exc: 

123 raise HTTPException(status_code=401, detail="malformed token") from exc 

124 

125 iss = unverified.get("iss", "") 

126 

127 with db() as conn: 

128 peer_row = conn.execute( 

129 "SELECT * FROM peers WHERE node_id = ?", 

130 (iss,), 

131 ).fetchone() 

132 

133 if peer_row is None: 133 ↛ 134line 133 didn't jump to line 134 because the condition on line 133 was never true

134 _public_module().write_audit_log( 

135 iss, "rejected_token", {"reason": "peer_not_found", "iss": iss} 

136 ) 

137 raise HTTPException(status_code=401, detail="peer not registered") 

138 

139 if peer_row["status"] != "active": 

140 _public_module().write_audit_log( 

141 peer_row["id"], 

142 "rejected_token", 

143 {"reason": "peer_not_approved", "iss": iss, "status": peer_row["status"]}, 

144 ) 

145 raise HTTPException(status_code=401, detail="peer_not_approved") 

146 

147 peer = dict(peer_row) 

148 

149 try: 

150 payload = verify_peer_token(raw_token, peer["federation_pubkey"], peer["id"]) 

151 except TokenError as exc: 

152 event = "replay_attempt" if exc.kind == "nonce_already_seen" else "rejected_token" 

153 _public_module().write_audit_log(peer["id"], event, {"reason": exc.kind}) 

154 raise HTTPException(status_code=401, detail=exc.kind) from exc 

155 

156 # §22.1.2.4 — bind TLS cert identity to JWT iss; rejects cert-swapping attacks. 

157 if _public_module().settings.mtls_enabled: 157 ↛ 158line 157 didn't jump to line 158 because the condition on line 157 was never true

158 peer_cert = _get_mtls_peer_cert(request) 

159 if peer_cert and not check_peer_san(peer_cert, peer["node_id"]): 

160 _public_module().write_audit_log( 

161 peer["id"], "san_mismatch", {"node_id": peer["node_id"]} 

162 ) 

163 raise HTTPException( 

164 status_code=401, 

165 detail="peer certificate URI SAN does not match node_id", 

166 ) 

167 if not peer_cert: 

168 logger.warning( 

169 "mTLS peer certificate was not exposed by the ASGI server; " 

170 "falling back to TLS-layer client certificate verification for %s", 

171 peer["node_id"], 

172 ) 

173 

174 return peer, payload 

175 

176 

177PeerTokenDep = Annotated[tuple[dict[str, Any], dict[str, Any]], Depends(_require_peer_token)] 

178 

179 

180def _try_peer_token_auth( 

181 authorization: str | None, 

182) -> tuple[dict[str, Any], dict[str, Any]] | None: 

183 """Soft peer-JWT auth: returns (peer, payload) on success, None on failure. 

184 

185 Unlike _require_peer_token, never raises — used so push_facts can fall 

186 through to the capability-token path when peer JWT is absent or invalid. 

187 """ 

188 if authorization is None or not authorization.lower().startswith("bearer "): 

189 return None 

190 

191 raw_token = authorization[7:] 

192 

193 import jwt as _jwt 

194 

195 try: 

196 unverified: dict[str, Any] = _jwt.decode( 

197 raw_token, 

198 options={ 

199 "verify_signature": False, 

200 "verify_exp": False, 

201 "verify_iat": False, 

202 "verify_nbf": False, 

203 "verify_aud": False, 

204 }, 

205 algorithms=["EdDSA"], 

206 ) 

207 except Exception: 

208 return None 

209 

210 iss = unverified.get("iss", "") 

211 with db() as conn: 

212 peer_row = conn.execute( 

213 "SELECT * FROM peers WHERE node_id = ?", 

214 (iss,), 

215 ).fetchone() 

216 

217 if peer_row is None or peer_row["status"] != "active": 217 ↛ 218line 217 didn't jump to line 218 because the condition on line 217 was never true

218 return None 

219 

220 peer = dict(peer_row) 

221 try: 

222 payload = verify_peer_token(raw_token, peer["federation_pubkey"], peer["id"]) 

223 except TokenError: 

224 return None 

225 

226 return peer, payload 

227 

228 

229def _cap_token_covers_scope(token_object: str, scope: str) -> bool: 

230 """Return True if the capability token's object covers the given fact scope (H-SEC-2).""" 

231 # "stigmem://facts" is a wildcard covering all scopes 

232 if token_object == "stigmem://facts": # nosec B105 — URI scheme constant, not a password 

233 return True 

234 # "stigmem://facts/scope:X" covers exactly scope X 

235 return token_object == f"stigmem://facts/scope:{scope}"