Coverage for node / src / stigmem_node / federation / dnssec / pin.py: 100%

1"""Rotation-aware DNSSEC origin-pin store (Rev 6 I1/I6 / build-phase 3b). 

2 

3DB layer over ``dnssec_origin_pins`` (migration 053), keyed by 

4``(entity_uri, node_id)`` — the pinned DNSSEC binding per identity (I1). When 

5the first-trust ladder accepts a binding it pins the validated fingerprint + 

6rotation epoch + the canonical query host (I3) here; a later binding that 

7disagrees with this stored anchor is an attack and is rejected (I1/I8). 

8 

9Two pieces: 

10 

11 * ``upsert_pin`` / ``get_pin`` — write/read the trusted pin. ``upsert_pin`` 

12 stamps ``last_validated_at=now`` on every accepted (re)validation; the row is 

13 keyed by the identity PK so a re-validation updates in place, never adds a 

14 second row. 

15 * ``pin_matches`` — the I6 rotation-grace predicate. A candidate fingerprint 

16 matches the *current* ``key_fpr`` always; it also matches the committed 

17 ``prev_fpr`` but **only** while ``now <= prev_until`` (a live grace window). 

18 A missing/empty ``prev_until`` means there is no live grace window, so 

19 ``prev_fpr`` never matches (fail-closed — a missing deadline is not 

20 "forever"). Carried bytes are never a key source (I7); this module only 

21 compares an independently-resolved candidate against the stored anchor. 

22 

23Off-path DB primitives: nothing here is wired into the resolver yet (the ladder 

24in the sibling commit consumes it). All helpers take an open ``sqlite3``-style 

25connection and participate in its transaction; the caller owns commit/rollback. 

26No DNSSEC / ``dnspython`` import is reachable from this module (Rev 6 I11). 

27""" 

28 

29from __future__ import annotations 

30 

31from dataclasses import dataclass 

32from datetime import UTC, datetime 

33from typing import Any 

34 

35_COLUMNS = ( 

36 "entity_uri", 

37 "node_id", 

38 "key_fpr", 

39 "epoch", 

40 "prev_fpr", 

41 "prev_until", 

42 "host", 

43 "last_validated_at", 

44) 

45_COLUMNS_SQL = ", ".join(_COLUMNS) 

46_SELECT_ONE_SQL = ( 

47 f"SELECT {_COLUMNS_SQL} FROM dnssec_origin_pins " # noqa: S608 # nosec B608 

48 "WHERE entity_uri=? AND node_id=?" 

49) 

50 

51 

52@dataclass(frozen=True) 

53class Pin: 

54 """A pinned, DNSSEC-validated origin binding (one ``dnssec_origin_pins`` row).""" 

55 

56 entity_uri: str 

57 node_id: str 

58 key_fpr: str 

59 epoch: int 

60 prev_fpr: str | None 

61 prev_until: str | None 

62 host: str 

63 last_validated_at: str 

64 

65 

66def _row_to_pin(row: Any) -> Pin: 

67 return Pin(*row) 

68 

69 

70def get_pin(conn: Any, entity_uri: str, node_id: str) -> Pin | None: 

71 """Return the pinned binding for ``(entity_uri, node_id)``, or ``None``.""" 

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

73 return _row_to_pin(row) if row is not None else None 

74 

75 

76def upsert_pin( 

77 conn: Any, 

78 *, 

79 entity_uri: str, 

80 node_id: str, 

81 key_fpr: str, 

82 epoch: int, 

83 host: str, 

84 prev_fpr: str | None = None, 

85 prev_until: str | None = None, 

86 now: datetime, 

87) -> None: 

88 """Insert or update the trusted pin for ``(entity_uri, node_id)`` (I1/I6). 

89 

90 Stamps ``last_validated_at=now`` (ISO-8601). Keyed by the identity PK, so a 

91 re-validation of an existing identity updates the row in place (fingerprint / 

92 epoch / grace fields / host / validation time) rather than inserting a 

93 duplicate. The caller owns the transaction. 

94 """ 

95 validated_at = now.isoformat() 

96 existing = conn.execute( 

97 "SELECT entity_uri FROM dnssec_origin_pins WHERE entity_uri=? AND node_id=?", 

98 (entity_uri, node_id), 

99 ).fetchone() 

100 

101 if existing is not None: 

102 conn.execute( 

103 "UPDATE dnssec_origin_pins " 

104 "SET key_fpr=?, epoch=?, prev_fpr=?, prev_until=?, host=?, last_validated_at=? " 

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

106 (key_fpr, epoch, prev_fpr, prev_until, host, validated_at, entity_uri, node_id), 

107 ) 

108 return 

109 

110 conn.execute( 

111 "INSERT INTO dnssec_origin_pins " 

112 "(entity_uri, node_id, key_fpr, epoch, prev_fpr, prev_until, host, last_validated_at) " 

113 "VALUES (?, ?, ?, ?, ?, ?, ?, ?)", 

114 (entity_uri, node_id, key_fpr, epoch, prev_fpr, prev_until, host, validated_at), 

115 ) 

116 

117 

118def pin_matches(pin: Pin, candidate_fpr: str, *, now: datetime) -> bool: 

119 """Whether ``candidate_fpr`` is honored against the stored pin (Rev 6 I6). 

120 

121 * Matches the current ``key_fpr`` -> True (always). 

122 * Matches the committed ``prev_fpr`` -> True only while ``now <= prev_until`` 

123 (a live rotation-grace window). A missing/empty ``prev_until``, or a ``now`` 

124 past it, means the prior key is no longer honored (fail-closed). 

125 * Otherwise -> False. 

126 

127 An empty ``candidate_fpr`` never matches (an empty fingerprint is not a key). 

128 """ 

129 if not candidate_fpr: 

130 return False 

131 

132 if candidate_fpr == pin.key_fpr: 

133 return True 

134 

135 if pin.prev_fpr and candidate_fpr == pin.prev_fpr and pin.prev_until: 

136 try: 

137 deadline = datetime.fromisoformat(pin.prev_until) 

138 except (ValueError, TypeError): 

139 # An unparseable/odd deadline is fail-closed: no live grace window. 

140 # ``prev_until`` is record input the grammar does not constrain, so 

141 # the comparison below must never raise out of the ladder. 

142 return False 

143 # A legitimate NAIVE deadline (no tzinfo) is normalized to UTC so a 

144 # bare ISO timestamp still honors rotation grace rather than raising a 

145 # TypeError against the tz-aware `now`. 

146 if deadline.tzinfo is None: 

147 deadline = deadline.replace(tzinfo=UTC) 

148 return now <= deadline 

149 

150 return False