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

1"""Admin routes for the operator origin-pin store — Phase 2c W4.1. 

2 

3All three endpoints are gated on ``admin:federation`` (403 otherwise) and each 

4mutating call writes a federation audit event, mirroring ``patch_peer_policy`` 

5in ``peers.py``. 

6""" 

7 

8from __future__ import annotations 

9 

10from typing import Annotated, Any 

11 

12from fastapi import Depends, HTTPException, status 

13from pydantic import BaseModel 

14 

15from ...auth import Identity, resolve_identity 

16from ...db import db 

17from ...federation.origin_pins import ( 

18 delete_origin_pin, 

19 get_origin_pin, 

20 list_origin_pins, 

21 put_origin_pin, 

22) 

23from .common import _public_module, router 

24 

25# --------------------------------------------------------------------------- 

26# Request model 

27# --------------------------------------------------------------------------- 

28 

29 

30class OriginPinRequest(BaseModel): 

31 """Body for POST /v1/federation/origin-pins.""" 

32 

33 entity_uri: str 

34 node_id: str 

35 key_fingerprint: str 

36 

37 

38# --------------------------------------------------------------------------- 

39# POST /v1/federation/origin-pins — create/replace a pin 

40# --------------------------------------------------------------------------- 

41 

42 

43@router.post( 

44 "/v1/federation/origin-pins", 

45 status_code=status.HTTP_200_OK, 

46) 

47def create_origin_pin( 

48 req: OriginPinRequest, 

49 identity: Annotated[Identity, Depends(resolve_identity)], 

50) -> dict[str, Any]: 

51 """Pin an origin's (entity_uri, node_id, key_fingerprint) out-of-band. 

52 

53 Idempotent: re-pinning the same key is a no-op update; re-pinning a 

54 different key replaces the previous fingerprint (explicit operator action). 

55 Requires admin:federation. 

56 """ 

57 if not identity.can_admin_federation(): 

58 raise HTTPException(status_code=403, detail="admin:federation required") 

59 with db() as conn: 

60 put_origin_pin( 

61 conn, 

62 entity_uri=req.entity_uri, 

63 node_id=req.node_id, 

64 key_fingerprint=req.key_fingerprint, 

65 pinned_by=identity.entity_uri, 

66 ) 

67 conn.commit() 

68 row = get_origin_pin(conn, entity_uri=req.entity_uri, node_id=req.node_id) 

69 _public_module().write_audit_log( 

70 req.entity_uri, 

71 "origin_pin_set", 

72 { 

73 "entity_uri": req.entity_uri, 

74 "node_id": req.node_id, 

75 "key_fingerprint": req.key_fingerprint, 

76 "pinned_by": identity.entity_uri, 

77 }, 

78 ) 

79 return row # type: ignore[return-value] 

80 

81 

82# --------------------------------------------------------------------------- 

83# GET /v1/federation/origin-pins — list all pins 

84# --------------------------------------------------------------------------- 

85 

86 

87@router.get("/v1/federation/origin-pins") 

88def list_pins( 

89 identity: Annotated[Identity, Depends(resolve_identity)], 

90) -> dict[str, Any]: 

91 """List all operator-pinned origins. Requires admin:federation.""" 

92 if not identity.can_admin_federation(): 

93 raise HTTPException(status_code=403, detail="admin:federation required") 

94 with db() as conn: 

95 pins = list_origin_pins(conn) 

96 return {"pins": pins} 

97 

98 

99# --------------------------------------------------------------------------- 

100# DELETE /v1/federation/origin-pins/{entity_uri}/{node_id} — remove a pin 

101# --------------------------------------------------------------------------- 

102 

103 

104@router.delete("/v1/federation/origin-pins") 

105def delete_pin( 

106 entity_uri: str, 

107 node_id: str, 

108 identity: Annotated[Identity, Depends(resolve_identity)], 

109) -> dict[str, Any]: 

110 """Remove an operator pin for (entity_uri, node_id). Requires admin:federation. 

111 

112 ``entity_uri`` and ``node_id`` are passed as query parameters to avoid 

113 path-encoding complications with ``://`` URIs. Returns 404 when no such 

114 pin exists. 

115 """ 

116 if not identity.can_admin_federation(): 

117 raise HTTPException(status_code=403, detail="admin:federation required") 

118 with db() as conn: 

119 deleted = delete_origin_pin(conn, entity_uri=entity_uri, node_id=node_id) 

120 if deleted: 

121 conn.commit() 

122 if not deleted: 

123 raise HTTPException(status_code=404, detail="origin pin not found") 

124 _public_module().write_audit_log( 

125 entity_uri, 

126 "origin_pin_deleted", 

127 { 

128 "entity_uri": entity_uri, 

129 "node_id": node_id, 

130 "deleted_by": identity.entity_uri, 

131 }, 

132 ) 

133 return {"entity_uri": entity_uri, "node_id": node_id, "deleted": True}