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

1"""Operator origin-pin store — Phase 2c W4.1. 

2 

3The operator pins an ``(entity_uri, node_id, key_fingerprint)`` triple obtained 

4out-of-band, asserting that a specific key controls a specific origin. This is 

5the same trust primitive as 2a direct-peer approval: the operator confirms a 

6fingerprint they verified independently rather than trusting a network-delivered 

7key at run time. 

8 

9The ``key_fingerprint`` stored here is produced by ``peer_pubkey_fingerprint`` 

10(sha256:<hex>) so it is directly comparable to the fingerprint computed from a 

11manifest's ``public_key`` field — the relay resolver (W4.2) calls 

12``get_origin_pin`` then compares against the manifest key without needing any 

13format conversion. 

14""" 

15 

16from __future__ import annotations 

17 

18from datetime import UTC, datetime 

19from typing import Any 

20 

21from ..routes._federation_impl import peer_pubkey_fingerprint 

22 

23 

24def fingerprint_from_pubkey(pubkey: str) -> str: 

25 """Return the sha256: fingerprint for *pubkey* using the canonical primitive. 

26 

27 Delegates to ``peer_pubkey_fingerprint`` so the stored fingerprint is 

28 identical in form to what ``approve_peer_impl`` computes — the relay 

29 resolver can compare them directly. 

30 """ 

31 return peer_pubkey_fingerprint(pubkey) 

32 

33 

34def put_origin_pin( 

35 conn: Any, 

36 *, 

37 entity_uri: str, 

38 node_id: str, 

39 key_fingerprint: str, 

40 pinned_by: str | None, 

41) -> None: 

42 """Upsert an origin pin. 

43 

44 Idempotent: re-pinning the same key is a no-op update. Pinning a 

45 *different* key for an existing ``(entity_uri, node_id)`` replaces the 

46 old fingerprint — that is an explicit operator action (key rotation). 

47 

48 ``pinned_at`` is set to UTC now on every upsert so the timestamp reflects 

49 the most recent operator confirmation. 

50 """ 

51 now = datetime.now(UTC).isoformat() 

52 conn.execute( 

53 """INSERT INTO origin_pins (entity_uri, node_id, key_fingerprint, pinned_by, pinned_at) 

54 VALUES (?, ?, ?, ?, ?) 

55 ON CONFLICT (entity_uri, node_id) 

56 DO UPDATE SET 

57 key_fingerprint = excluded.key_fingerprint, 

58 pinned_by = excluded.pinned_by, 

59 pinned_at = excluded.pinned_at""", 

60 (entity_uri, node_id, key_fingerprint, pinned_by, now), 

61 ) 

62 

63 

64def get_origin_pin( 

65 conn: Any, 

66 *, 

67 entity_uri: str, 

68 node_id: str, 

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

70 """Return the pin row for ``(entity_uri, node_id)``, or ``None`` if absent.""" 

71 row = conn.execute( 

72 "SELECT entity_uri, node_id, key_fingerprint, pinned_by, pinned_at" 

73 " FROM origin_pins WHERE entity_uri = ? AND node_id = ?", 

74 (entity_uri, node_id), 

75 ).fetchone() 

76 if row is None: 

77 return None 

78 return dict(row) 

79 

80 

81def list_origin_pins(conn: Any) -> list[dict[str, Any]]: 

82 """Return all pinned origins as a list of dicts.""" 

83 rows = conn.execute( 

84 "SELECT entity_uri, node_id, key_fingerprint, pinned_by, pinned_at" 

85 " FROM origin_pins ORDER BY entity_uri, node_id" 

86 ).fetchall() 

87 return [dict(r) for r in rows] 

88 

89 

90def delete_origin_pin( 

91 conn: Any, 

92 *, 

93 entity_uri: str, 

94 node_id: str, 

95) -> bool: 

96 """Delete the pin for ``(entity_uri, node_id)``. 

97 

98 Returns ``True`` if a row was deleted, ``False`` if no such pin existed. 

99 """ 

100 cur = conn.execute( 

101 "DELETE FROM origin_pins WHERE entity_uri = ? AND node_id = ?", 

102 (entity_uri, node_id), 

103 ) 

104 return bool(cur.rowcount > 0)