Coverage for node / src / stigmem_node / routes / facts / cid.py: 87%

1"""CID verification route for facts.""" 

2 

3from __future__ import annotations 

4 

5from typing import Annotated 

6 

7from fastapi import Depends, HTTPException, status 

8from pydantic import BaseModel 

9 

10from ...auth import Identity, resolve_identity 

11from ...cid import compute_cid_from_row, stored_cid_from_row 

12from ...db import db 

13from ...garden_acl import require_fact_garden_read 

14from .common import logger, router 

15 

16 

17class _CidVerifyResponse(BaseModel): 

18 cid_valid: bool 

19 computed_cid: str 

20 stored_cid: str | None 

21 mismatch_reason: str | None = None 

22 

23 

24@router.post("/{fact_id}/verify-cid", response_model=_CidVerifyResponse, tags=["facts"]) 

25def verify_cid( 

26 fact_id: str, 

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

28) -> _CidVerifyResponse: 

29 """Verify a fact's stored CID against a freshly computed one. 

30 

31 Covered by Spec-21-Content-Addressed-IDs. 

32 """ 

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

34 raise HTTPException( 

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

36 ) # noqa: E501 

37 with db() as conn: 

38 row = conn.execute( 

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

40 (fact_id, identity.tenant_id), 

41 ).fetchone() 

42 if row is None: 

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

44 # Garden ACL: a restricted-garden fact's CID is an existence/content 

45 # oracle — hide it from non-members (spec §17.3; sibling of single-get). 

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

47 computed = compute_cid_from_row(row) 

48 stored = stored_cid_from_row(row) 

49 if stored is None: 

50 return _CidVerifyResponse( 

51 cid_valid=False, 

52 computed_cid=computed, 

53 stored_cid=None, 

54 mismatch_reason="stored_cid is null (pre-Phase-13 record pending backfill)", 

55 ) 

56 if computed == stored: 56 ↛ 58line 56 didn't jump to line 58 because the condition on line 56 was always true

57 return _CidVerifyResponse(cid_valid=True, computed_cid=computed, stored_cid=stored) 

58 logger.warning("CID mismatch for fact %s: computed=%s stored=%s", fact_id, computed, stored) 

59 return _CidVerifyResponse( 

60 cid_valid=False, 

61 computed_cid=computed, 

62 stored_cid=stored, 

63 mismatch_reason="stored_cid does not match computed_cid", 

64 )