Coverage for node / src / stigmem_node / source_attestation.py: 86%

1"""Source ↔ identity attestation (P-INJ-1), graduated to core. 

2 

3A fact's ``source`` is *attested* when it matches the principal writing it. By 

4default mismatches are flagged (``attested=False``) but allowed; with 

5``settings.source_attestation_enforce`` they are rejected at write. This closes 

6"any write key can forge any source" while making the attributable-memory claim 

7true: forged sources are always detectable, even when not blocked. 

8 

9Delegated source entities (writing as a source other than yourself) are a 

10deferred surface (§18); ``authorized_source_entities`` already reads the 

11``allowed_source_entities`` identity field so the delegation hook lands cleanly. 

12""" 

13 

14from __future__ import annotations 

15 

16from typing import Any 

17 

18from .entity_normalizer import NormalizationError, normalize_entity_uri 

19 

20 

21def _live_settings() -> Any: 

22 import sys 

23 

24 return sys.modules["stigmem_node.settings"].settings 

25 

26 

27def _normalized_or_none(raw: Any) -> str | None: 

28 if not isinstance(raw, str): 28 ↛ 29line 28 didn't jump to line 29 because the condition on line 28 was never true

29 return None 

30 try: 

31 return normalize_entity_uri(raw) 

32 except NormalizationError: 

33 return None 

34 

35 

36def authorized_source_entities(identity: Any) -> set[str]: 

37 """Normalized set of source entities the principal may attest to.""" 

38 raw = {getattr(identity, "entity_uri", None)} 

39 raw.update(getattr(identity, "allowed_source_entities", ()) or ()) # §18 delegation 

40 return {n for r in raw if (n := _normalized_or_none(r)) is not None} 

41 

42 

43def evaluate_source_attested(source: Any, identity: Any) -> bool | None: 

44 """Return whether *source* is attested to the writing *identity*. 

45 

46 ``None`` when there is no authenticated principal (anonymous / auth-disabled 

47 mode) — attestation is meaningless without a real writer. Otherwise True when 

48 the normalized source matches the principal (or a delegated source), else False. 

49 """ 

50 if not _live_settings().auth_required: 

51 return None 

52 normalized = _normalized_or_none(source) 

53 return normalized is not None and normalized in authorized_source_entities(identity) 

54 

55 

56def source_attestation_enforce_enabled() -> bool: 

57 """Return True when an unattested source must be rejected (default off).""" 

58 return bool(_live_settings().source_attestation_enforce)