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

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 .common import logger, router 

14 

15 

16class _CidVerifyResponse(BaseModel): 

17 cid_valid: bool 

18 computed_cid: str 

19 stored_cid: str | None 

20 mismatch_reason: str | None = None 

21 

22 

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

24def verify_cid( 

25 fact_id: str, 

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

27) -> _CidVerifyResponse: 

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

29 

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

31 """ 

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

33 raise HTTPException( 

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

35 ) # noqa: E501 

36 with db() as conn: 

37 row = conn.execute( 

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

39 (fact_id, identity.tenant_id), 

40 ).fetchone() 

41 if row is None: 

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

43 computed = compute_cid_from_row(row) 

44 stored = stored_cid_from_row(row) 

45 if stored is None: 

46 return _CidVerifyResponse( 

47 cid_valid=False, 

48 computed_cid=computed, 

49 stored_cid=None, 

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

51 ) 

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

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

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

55 return _CidVerifyResponse( 

56 cid_valid=False, 

57 computed_cid=computed, 

58 stored_cid=stored, 

59 mismatch_reason="stored_cid does not match computed_cid", 

60 )