diff --git a/src/llm_ingestion_guard/okf.py b/src/llm_ingestion_guard/okf.py index 2a7ecea..7837af2 100644 --- a/src/llm_ingestion_guard/okf.py +++ b/src/llm_ingestion_guard/okf.py @@ -23,15 +23,24 @@ lost; richer scalar forms are a future refinement, not a silent parse. """ import re +from dataclasses import dataclass +from enum import Enum from .output import scan_output from .report import Report, Source +from .disposition import Trust, Disposition, Policy, decide __all__ = [ "parse_frontmatter", "scan_concept", "validate_concept_path", "validate_resource_url", + "trust_for", + "stamp_concept", + "format_log_entry", + "Origin", + "Channel", + "ProvenanceStamp", "OKFError", "OKFFrontmatterError", "OKFPathError", @@ -195,6 +204,75 @@ def validate_resource_url(url): return url +class Origin(str, Enum): + """Where the data actually came from (brief §5) — drives trust.""" + + EXTERNAL = "external" + INTERNAL = "internal" + + +class Channel(str, Enum): + """How it was inserted — recorded for the log, but never upgrades trust.""" + + AUTOMATIC = "automatic" + MANUAL = "manual" + + +@dataclass(frozen=True) +class ProvenanceStamp: + """A per-concept provenance record for ``log.md`` (brief §6 T6). + + Composes ``Origin`` x ``Channel`` x ``Trust`` x ``Disposition`` — it adds no + new disposition value (brief §8 naming caveat); the disposition is whatever + :func:`decide` returns for the concept's scan under its origin-derived trust. + """ + + concept_id: str + origin: Origin + channel: Channel + trust: Trust + disposition: Disposition + + +def trust_for(origin, channel=None): + """Map a concept's origin to a :class:`Trust` tier (brief §5). + + Trust follows the *origin*, never the insertion *channel*: a manual paste of + external material is still external. The channel is recorded on the stamp for + the audit log but grants no trust discount. + """ + return Trust.TRUSTED if origin is Origin.INTERNAL else Trust.UNTRUSTED + + +def stamp_concept(concept_id, report, origin, channel): + """Stamp one scanned concept with its provenance and disposition (T6). + + ``report`` is the concept's scan (e.g. from :func:`scan_concept`); the + disposition is decided under a policy at the origin-derived trust tier. + """ + trust = trust_for(origin, channel) + decision = decide(report, Policy(trust=trust)) + return ProvenanceStamp(concept_id, origin, channel, trust, decision.disposition) + + +def format_log_entry(stamp, *, timestamp=None): + """Render a :class:`ProvenanceStamp` as one tab-separated ``log.md`` line. + + ``timestamp`` is caller-supplied (kept out of the stamp so stamping stays + deterministic and wall-clock-free); when given it is prepended. + """ + fields = [ + stamp.concept_id, + stamp.origin.value, + stamp.channel.value, + stamp.trust.value, + stamp.disposition.value, + ] + if timestamp is not None: + fields.insert(0, timestamp) + return "\t".join(fields) + + def _parse_flat(fm_lines): result = {} i = 0 diff --git a/tests/test_okf.py b/tests/test_okf.py index 1d6e0e4..0b6fac1 100644 --- a/tests/test_okf.py +++ b/tests/test_okf.py @@ -22,11 +22,17 @@ from llm_ingestion_guard.okf import ( scan_concept, validate_concept_path, validate_resource_url, + stamp_concept, + trust_for, + format_log_entry, + Origin, + Channel, OKFFrontmatterError, OKFPathError, OKFResourceError, ) from llm_ingestion_guard.report import Report +from llm_ingestion_guard.disposition import Trust, Disposition # --- happy path: split + parse the minimal flat subset ----------------------- @@ -256,3 +262,44 @@ def test_validate_resource_url_rejects_embedded_whitespace(): # a space-split URL can smuggle a second target past a naive consumer parser with pytest.raises(OKFResourceError): validate_resource_url("https://good.example/x javascript:alert(1)") + + +# --- T6: provenance stamping (origin + channel -> trust + disposition) -------- +# brief §5: trust follows the data's ORIGIN, not the insertion channel — a manual +# paste of external material is still external. The channel is recorded but never +# upgrades trust. T6 composes Trust x Disposition; it adds no new disposition. + +def test_trust_follows_origin_not_channel(): + # the load-bearing §5 property: "channel grants no discount" + assert trust_for(Origin.EXTERNAL, Channel.AUTOMATIC) is Trust.UNTRUSTED + assert trust_for(Origin.EXTERNAL, Channel.MANUAL) is Trust.UNTRUSTED + assert trust_for(Origin.INTERNAL, Channel.AUTOMATIC) is Trust.TRUSTED + assert trust_for(Origin.INTERNAL, Channel.MANUAL) is Trust.TRUSTED + + +def test_stamp_concept_records_origin_channel_and_untrusted_external(): + stamp = stamp_concept("tables/users", Report(), Origin.EXTERNAL, Channel.MANUAL) + assert stamp.concept_id == "tables/users" + assert stamp.origin is Origin.EXTERNAL + assert stamp.channel is Channel.MANUAL + assert stamp.trust is Trust.UNTRUSTED + assert isinstance(stamp.disposition, Disposition) + + +def test_stamp_concept_injection_escalates_disposition(): + report = scan_concept("---\ntype: table\n---\n" + _INJECTION + "\n") + stamp = stamp_concept("tables/users", report, Origin.EXTERNAL, Channel.AUTOMATIC) + assert stamp.disposition in (Disposition.QUARANTINE_REVIEW, Disposition.FAIL_SECURE) + + +def test_format_log_entry_contains_all_fields(): + stamp = stamp_concept("tables/users", Report(), Origin.INTERNAL, Channel.AUTOMATIC) + line = format_log_entry(stamp) + for token in ("tables/users", "internal", "automatic", "trusted", stamp.disposition.value): + assert token in line + + +def test_format_log_entry_prepends_timestamp(): + stamp = stamp_concept("a/b", Report(), Origin.INTERNAL, Channel.AUTOMATIC) + line = format_log_entry(stamp, timestamp="2026-07-06T07:00:00Z") + assert line.startswith("2026-07-06T07:00:00Z")