Coverage for node / src / stigmem_node / federation / dnssec / quarantine.py: 95%

1"""Operator-confirm quarantine for the DNSSEC first-trust ladder (Rev 6 I9/§9). 

2 

3DB layer over ``pending_first_trust`` (migration 055). Operator-confirm is the 

4SOLE non-DNSSEC first-trust fallback (Rev 6 §2/§15.1): when the ladder cannot 

5root an unknown origin via an operator-pin or a DNSSEC binding, the candidate 

6binding is parked here for an explicit human action (paste/confirm the 

7fingerprint out-of-band, never one-click). Because it is the only fallback for 

8the entire non-DNSSEC long tail, the queue MUST be bounded so an untrusted relay 

9cannot flood it (I9 / NF-R5C-4): 

10 

11 * ``quarantine`` parks/refreshes a row keyed by ``(entity_uri, node_id)``, 

12 enforcing a per-``relay_peer`` insert cap (a NEW row from a peer already 

13 at/over the cap is rejected + audited; a REFRESH of an existing row is always 

14 allowed so a legitimate re-observation is never starved). 

15 * ``evict_expired`` deletes rows older than the TTL by ``seen_at``. 

16 

17``list_pending`` / ``get_pending`` / ``remove_pending`` are the read/delete layer 

18the operator-confirm CLI + admin API (next batch) will call. This module builds 

19NO routes and is not wired into the resolver yet (off-path). It imports 

20``settings`` for the cap/TTL defaults and ``emit_nofail`` function-locally for 

21the flood audit; no DNSSEC / ``dnspython`` import is reachable from it (I11). All 

22helpers take an open connection and participate in its transaction; the caller 

23owns commit/rollback. 

24""" 

25 

26from __future__ import annotations 

27 

28from datetime import datetime, timedelta 

29from typing import Any 

30 

31# The operator-facing columns surfaced by get_pending/list_pending (the API/CLI 

32# render exactly these). Kept in one place so the row-dict shape is consistent. 

33_ROW_COLUMNS = ( 

34 "entity_uri", 

35 "node_id", 

36 "candidate_key_fpr", 

37 "source", 

38 "relay_peer", 

39 "seen_at", 

40) 

41# Pre-built SELECT strings. The column list is the hardcoded tuple above (never 

42# user input), so interpolating it is safe — the nosec annotations mark the two 

43# f-strings bandit flags (B608) as constant-only, not injectable. 

44_ROW_COLUMNS_SQL = ", ".join(_ROW_COLUMNS) 

45_SELECT_ALL_SQL = ( 

46 f"SELECT {_ROW_COLUMNS_SQL} FROM pending_first_trust ORDER BY seen_at DESC" # noqa: S608 # nosec B608 

47) 

48_SELECT_ONE_SQL = ( 

49 f"SELECT {_ROW_COLUMNS_SQL} FROM pending_first_trust WHERE entity_uri=? AND node_id=?" # noqa: S608 # nosec B608 

50) 

51 

52# Sentinel so callers can pass cap=None / ttl=None meaning "use the setting". 

53_UNSET = object() 

54 

55 

56def _row_to_dict(row: Any) -> dict[str, Any]: 

57 return dict(zip(_ROW_COLUMNS, row, strict=True)) 

58 

59 

60def _peer_pending_count(conn: Any, relay_peer: str | None) -> int: 

61 """Number of currently-parked rows attributed to ``relay_peer``.""" 

62 if relay_peer is None: 

63 row = conn.execute( 

64 "SELECT COUNT(*) FROM pending_first_trust WHERE relay_peer IS NULL" 

65 ).fetchone() 

66 else: 

67 row = conn.execute( 

68 "SELECT COUNT(*) FROM pending_first_trust WHERE relay_peer=?", (relay_peer,) 

69 ).fetchone() 

70 return int(row[0]) 

71 

72 

73def _audit_cap_exceeded(entity_uri: str, node_id: str, relay_peer: str | None) -> None: 

74 """Best-effort flood-signal audit (never blocks the caller).""" 

75 from ...observability.audit_event import emit_nofail 

76 

77 emit_nofail( 

78 "federation_dnssec_pending_first_trust_cap_exceeded", 

79 entity_uri=entity_uri, 

80 source="federation_relay", 

81 detail={ 

82 "entity_uri": entity_uri, 

83 "node_id": node_id, 

84 "relay_peer": relay_peer, 

85 }, 

86 ) 

87 

88 

89def quarantine( 

90 conn: Any, 

91 *, 

92 entity_uri: str, 

93 node_id: str, 

94 candidate_key_fpr: str, 

95 source: str, 

96 relay_peer: str | None, 

97 now: datetime, 

98 cap: Any = _UNSET, 

99) -> bool: 

100 """Park (or refresh) a candidate binding in ``pending_first_trust`` (I9). 

101 

102 Returns True if the row is now parked (inserted or refreshed), False if a NEW 

103 insert was rejected because ``relay_peer`` is already at/over ``cap`` (an 

104 audit event is emitted naming the peer). 

105 

106 * If a row already exists for ``(entity_uri, node_id)`` it is REFRESHED in 

107 place (candidate_key_fpr / source / relay_peer / seen_at) regardless of the 

108 cap — refreshing adds no row, so it cannot contribute to a flood. 

109 * Otherwise a NEW row is inserted only if ``relay_peer`` has fewer than 

110 ``cap`` parked rows; at/over the cap the insert is rejected + audited. 

111 

112 ``cap`` defaults to ``settings.federation_dnssec_pending_confirm_cap`` when 

113 left unset. The caller owns the transaction. 

114 """ 

115 if cap is _UNSET: 

116 from ...settings import settings 

117 

118 cap = settings.federation_dnssec_pending_confirm_cap 

119 

120 seen_at = now.isoformat() 

121 

122 existing = conn.execute( 

123 "SELECT entity_uri FROM pending_first_trust WHERE entity_uri=? AND node_id=?", 

124 (entity_uri, node_id), 

125 ).fetchone() 

126 

127 if existing is not None: 

128 # Refresh in place — never counts against the cap. 

129 conn.execute( 

130 "UPDATE pending_first_trust " 

131 "SET candidate_key_fpr=?, source=?, relay_peer=?, seen_at=? " 

132 "WHERE entity_uri=? AND node_id=?", 

133 (candidate_key_fpr, source, relay_peer, seen_at, entity_uri, node_id), 

134 ) 

135 return True 

136 

137 if _peer_pending_count(conn, relay_peer) >= cap: 

138 # Flood bound hit: refuse the NEW row and surface the signal to the 

139 # operator. Distinct event so it is not confused with an ordinary 

140 # unknown-origin confirm. 

141 _audit_cap_exceeded(entity_uri, node_id, relay_peer) 

142 return False 

143 

144 conn.execute( 

145 "INSERT INTO pending_first_trust " 

146 "(entity_uri, node_id, candidate_key_fpr, source, relay_peer, seen_at) " 

147 "VALUES (?, ?, ?, ?, ?, ?)", 

148 (entity_uri, node_id, candidate_key_fpr, source, relay_peer, seen_at), 

149 ) 

150 return True 

151 

152 

153def evict_expired(conn: Any, *, now: datetime, ttl: Any = _UNSET) -> int: 

154 """Delete unconfirmed rows older than ``ttl`` seconds by ``seen_at`` (I9). 

155 

156 A row is expired when ``seen_at < now - ttl`` (strict, so a row exactly at 

157 the TTL boundary is kept). Returns the number of rows removed. ``ttl`` 

158 defaults to ``settings.federation_dnssec_pending_confirm_ttl`` when unset. The 

159 caller owns the transaction. 

160 """ 

161 if ttl is _UNSET: 161 ↛ 162line 161 didn't jump to line 162 because the condition on line 161 was never true

162 from ...settings import settings 

163 

164 ttl = settings.federation_dnssec_pending_confirm_ttl 

165 

166 cutoff = (now - timedelta(seconds=ttl)).isoformat() 

167 cur = conn.execute( 

168 "DELETE FROM pending_first_trust WHERE seen_at < ?", (cutoff,) 

169 ) 

170 return int(cur.rowcount) 

171 

172 

173def list_pending(conn: Any) -> list[dict[str, Any]]: 

174 """Return every parked binding (the operator-confirm queue), newest first.""" 

175 rows = conn.execute(_SELECT_ALL_SQL).fetchall() 

176 return [_row_to_dict(r) for r in rows] 

177 

178 

179def get_pending(conn: Any, entity_uri: str, node_id: str) -> dict[str, Any] | None: 

180 """Return the parked binding for ``(entity_uri, node_id)``, or None.""" 

181 row = conn.execute(_SELECT_ONE_SQL, (entity_uri, node_id)).fetchone() 

182 return _row_to_dict(row) if row is not None else None 

183 

184 

185def remove_pending(conn: Any, entity_uri: str, node_id: str) -> bool: 

186 """Delete the parked binding for ``(entity_uri, node_id)``. 

187 

188 Returns True if a row was removed (the operator confirmed or rejected it), 

189 False if there was nothing to remove. The caller owns the transaction. 

190 """ 

191 cur = conn.execute( 

192 "DELETE FROM pending_first_trust WHERE entity_uri=? AND node_id=?", 

193 (entity_uri, node_id), 

194 ) 

195 return int(cur.rowcount) > 0