Coverage for node / src / stigmem_node / routes / facts / provenance.py: 82%

1"""Fact provenance route and helpers.""" 

2 

3from __future__ import annotations 

4 

5from typing import Annotated, Any 

6 

7from fastapi import Depends, HTTPException, status 

8 

9from ...auth import Identity, resolve_identity 

10from ...db import db 

11from ...fact_visibility import ReadScope, caller_read_scope 

12from ...garden_acl import require_fact_garden_read 

13from ...models.provenance import ProvenanceEntry, ProvenanceResponse 

14from .common import _get_tombstone_filter, logger, router 

15 

16_REF_SELECT = ( 

17 "SELECT f.*, COALESCE(fgm.garden_id, f.garden_id) AS projected_garden_id FROM facts f" 

18 " LEFT JOIN fact_garden_membership fgm ON fgm.fact_id = f.id" 

19 " WHERE f.id = ? AND f.tenant_id = ?" 

20) 

21 

22 

23def _resolve_provenance_entry( 

24 entry: Any, tenant_id: str, read_scope: ReadScope 

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

26 """Resolve a derived_from entry to (hash_val, ref_row | None); skip non-dict entries. 

27 

28 A ref whose PROJECTED garden the caller cannot see is treated as unresolved 

29 (ref_row=None) so a restricted-garden ancestor's existence / fact_id / entity 

30 URI is not disclosed via lineage (audit F-PROV-REF). It then redacts to the 

31 same {hash, exists:false} shape as tombstoned/missing refs. 

32 """ 

33 if not isinstance(entry, dict): 33 ↛ 34line 33 didn't jump to line 34 because the condition on line 33 was never true

34 return None 

35 hash_val: str = entry.get("hash", "") 

36 entry_fact_id: str | None = entry.get("fact_id") 

37 

38 ref_row = None 

39 with db() as conn: 

40 if entry_fact_id: 40 ↛ 42line 40 didn't jump to line 42 because the condition on line 40 was always true

41 ref_row = conn.execute(_REF_SELECT, (entry_fact_id, tenant_id)).fetchone() 

42 elif hash_val.startswith("sha256:"): 

43 alias = conn.execute( 

44 "SELECT fact_id FROM fact_cid_aliases WHERE cid = ? AND tenant_id = ?", 

45 (hash_val, tenant_id), 

46 ).fetchone() 

47 if alias: 

48 ref_row = conn.execute(_REF_SELECT, (alias["fact_id"], tenant_id)).fetchone() 

49 if ref_row is not None and not read_scope.garden_allows(ref_row["projected_garden_id"]): 

50 ref_row = None # garden-hidden ancestor → redact like tombstoned 

51 return hash_val, ref_row 

52 

53 

54def _format_provenance_entry(hash_val: str, ref_row: Any, excluded: set[str]) -> ProvenanceEntry: 

55 """Render a resolved entry into a ProvenanceEntry, redacting tombstoned/missing rows.""" 

56 if ref_row is None: 

57 return ProvenanceEntry(hash=hash_val, exists=False) 

58 if ref_row["entity"] in excluded: 

59 return ProvenanceEntry(hash=hash_val, exists=False) 

60 return ProvenanceEntry( 

61 hash=hash_val, 

62 fact_id=ref_row["id"], 

63 entity=ref_row["entity"], 

64 exists=True, 

65 ) 

66 

67 

68@router.get("/{fact_id}/provenance", response_model=ProvenanceResponse) 

69def get_provenance( 

70 fact_id: str, 

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

72) -> ProvenanceResponse: 

73 """Provenance walk with tombstone suppression. 

74 

75 Returns the derived_from chain for a fact. Any entry whose referenced entity is 

76 tombstoned — or whose fact is otherwise inaccessible — is redacted to 

77 {"hash": "...", "exists": false}, indistinguishable from unauthorized 

78 cross-scope references to prevent existence leakage. Covered by 

79 Spec-X2-RTBF-Tombstones and Spec-X11-Recall-Graph. 

80 """ 

81 import json as _prov_json 

82 

83 if not identity.can_read(): 83 ↛ 84line 83 didn't jump to line 84 because the condition on line 83 was never true

84 raise HTTPException( 

85 status_code=status.HTTP_403_FORBIDDEN, detail="read permission required" 

86 ) 

87 

88 with db() as conn: 

89 row = conn.execute( 

90 "SELECT * FROM facts WHERE id = ? AND tenant_id = ?", 

91 (fact_id, identity.tenant_id), 

92 ).fetchone() 

93 if row is None: 

94 raise HTTPException(status_code=404, detail="fact not found") 

95 # Garden ACL: don't expose a restricted-garden fact's CID/lineage to a 

96 # non-member (spec §17.3; sibling of the single-get gate). 

97 require_fact_garden_read(conn, fact_id, identity.tenant_id, identity) 

98 

99 derived_from_raw = row["derived_from"] if "derived_from" in row.keys() else None # noqa: SIM118 

100 cid_val = row["cid"] if "cid" in row.keys() else None # noqa: SIM118 

101 root_scope: str = row["scope"] or "local" 

102 

103 if not derived_from_raw: 

104 return ProvenanceResponse(fact_id=fact_id, cid=cid_val, derived_from=[]) 

105 

106 try: 

107 entries_raw: list[Any] = _prov_json.loads(derived_from_raw) 

108 except Exception as exc: 

109 logger.warning("ignoring malformed provenance for fact %s: %s", fact_id, exc) 

110 entries_raw = [] 

111 

112 # Resolve each derived_from entry to its referenced fact row (garden-gated) 

113 read_scope = caller_read_scope(identity) 

114 resolved: list[tuple[str, Any]] = [] # (hash_val, ref_row | None) 

115 for entry in entries_raw: 

116 resolved_entry = _resolve_provenance_entry(entry, identity.tenant_id, read_scope) 

117 if resolved_entry is not None: 117 ↛ 115line 117 didn't jump to line 115 because the condition on line 117 was always true

118 resolved.append(resolved_entry) 

119 

120 # Single tombstone filter call across all resolved entity URIs (§23.3.2 r.4) 

121 accessible_entities = [ref_row["entity"] for _, ref_row in resolved if ref_row is not None] 

122 excluded: set[str] = set() 

123 if accessible_entities: 

124 with db() as _tc_conn: 

125 is_admin = identity.is_admin() 

126 excluded, _ = _get_tombstone_filter( 

127 _tc_conn, accessible_entities, root_scope, is_admin, identity.tenant_id 

128 ) 

129 

130 # Build response — §23.3.2 r.4 tombstone and §20.6.2 unauthorized share identical shape 

131 result: list[ProvenanceEntry] = [ 

132 _format_provenance_entry(hash_val, ref_row, excluded) for hash_val, ref_row in resolved 

133 ] 

134 

135 return ProvenanceResponse(fact_id=fact_id, cid=cid_val, derived_from=result)