llm-ingestion-pipeline-security/src/llm_ingestion_guard/okf.py

"""OKF adapter — Open Knowledge Format (Google, v0.1) support on top of the core.

Design principle: the format-agnostic core stays ``text -> findings``. This
adapter knows OKF structure (frontmatter, paths, links, ``resource``, bundles)
and feeds scannable text regions into the existing ``sanitize`` / ``scan_output``
/ ``disposition`` machinery. No YAML/format awareness leaks into the core.

T2 — frontmatter parse-safety gate. ``parse_frontmatter`` is a *strict,
reject-by-default* loader for the minimal OKF frontmatter subset: flat
``key: value`` scalars, block ``- item`` lists, and one typed, allowlisted
mapping form (``{ by: x, at: y }`` — see :func:`_parse_flow_mapping`). Every construct the
"block anchor/alias DoS + dangerous type coercion" requirement names is refused
*by construction* — you cannot suffer a billion-laughs alias expansion or a
``!!python/object`` coercion if anchors, aliases and explicit tags are rejected
before any value is interpreted. This is the "reject, don't parse-then-sanitize"
philosophy, the frontmatter analogue of the ``resource`` reject-gate (T3).

Deliberately NOT a general YAML parser. A security tool whose thesis is
minimal-dependency should not pull in a full YAML engine whose own features
(anchors, tags, merges) are the attack surface being defended against. The one
mapping form it does admit is admitted key-by-key against an allowlist, not
parsed generally: the mapping class is expressible, never trusted. Quoted
scalars are kept verbatim (quotes included) rather than unquoted — the value is
still scanned as text downstream, so an injection inside a quoted value is not
lost; richer scalar forms are a future refinement, not a silent parse.
"""

import re
from dataclasses import dataclass
from enum import Enum

from .calibration import MAX_SCAN_CHARS
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",
    "import_bundle",
    "extract_link_targets",
    "resolve_link",
    "link_graph",
    "Origin",
    "Channel",
    "ProvenanceStamp",
    "ConceptResult",
    "BundleResult",
    "LinkGraphResult",
    "OKFError",
    "OKFFrontmatterError",
    "OKFPathError",
    "OKFResourceError",
    "OKFLinkError",
]

_FENCE = "---"
_KEY_RE = re.compile(r"^[A-Za-z0-9_][A-Za-z0-9_-]*$")

# A plain OKF scalar cannot *begin* with a YAML structural indicator. Any value
# starting with one signals an anchor (&), alias (*), explicit tag (!), block
# scalar (|, >), flow collection ([ ] { }), directive (%) or reserved char
# (@ `) — all outside the supported subset and all rejected. `{` and `[` are
# tried as the allowlisted mapping form (G3) and the flow sequence of them (G30)
# FIRST; they reach this predicate only as a leaf inside one, where a nested
# collection is refused before it can be read.
_DANGEROUS_VALUE_STARTS = frozenset("&*!|>[]{}%@`")
# A quoted scalar is a scalar in YAML however many colons it carries, so the
# mapping check steps aside for one. The quotes are retained rather than
# stripped — a pre-existing divergence, pinned in tests/test_okf.py.
_QUOTE_STARTS = frozenset("\"'")

# G3 - the one mapping form T2 can express (operator decision, 2026-08-21).
# Every key inside a mapping must be on this allowlist: the form is safe because
# the allowlist inspects each key, not because mappings became trusted. The keys
# are the ones OKF v0.2 names inside a mapping - `by`/`at` (SPEC.md @ 62432a09
# §5.2 `generated`/`verified`) and `from`/`to` (§5.1 `usage_window`), plus the
# §5.1 `sources`-entry labels.
_MAPPING_KEY_ALLOWLIST = frozenset({
    "by", "at", "from", "to", "id", "title", "author", "usage_count",
    "last_modified",
})

# G30 - the two §5.1 keys admitted inside a `sources` entry and NOWHERE else
# (operator decision, 2026-09-02). `resource` is REQUIRED within a `sources`
# entry, so leaving it off left the whole provenance family unwritable; but the
# same field name in §10 (`executor.resource`, `attester.resource`) names run
# instructions and code - the door-C route closed in 1.1.0. 1.2.0 argued the
# parser could not tell the two apart without parent-key context it did not
# have. That premise was false: the owning key is in scope at every call site
# below, it was simply never threaded through. It is threaded now, so the
# discrimination is structural rather than a judgement about the value.
# `usage_window` is allowlisted here for accuracy of refusal - §5.1 permits it
# per entry, and it is then refused on the depth rule (a mapping inside a
# mapping, which this parser admits at no key) rather than refused as if the
# key were unknown.
_SOURCES_ENTRY_KEYS = frozenset({"resource", "usage_window"})


def _allowed_mapping_keys(parent_key):
    """The mapping-key allowlist for a mapping owned by ``parent_key``."""
    if parent_key == "sources":
        return _MAPPING_KEY_ALLOWLIST | _SOURCES_ENTRY_KEYS
    return _MAPPING_KEY_ALLOWLIST


class OKFError(Exception):
    """Base class for OKF adapter rejections."""


class OKFFrontmatterError(OKFError):
    """Frontmatter violates the strict, reject-by-default OKF subset."""


class OKFPathError(OKFError):
    """A concept path is unsafe (traversal, absolute, or reserved-name shadow)."""


class OKFResourceError(OKFError):
    """A ``resource`` URL is not on the https allowlist."""


class OKFLinkError(OKFError):
    """A cross-link target is unsafe (dangerous scheme or bundle escape)."""


_URL_SCHEME_RE = re.compile(r"^([A-Za-z][A-Za-z0-9+.\-]*):")


# `index.md` (directory listing) and `log.md` (update history) are reserved by
# the OKF spec and MUST NOT name concept documents — at any directory level.
_RESERVED_BASENAMES = frozenset({"index.md", "log.md"})


def parse_frontmatter(document):
    """Split leading OKF frontmatter from the body and parse it strictly.

    Returns ``(frontmatter: dict, body: str)``. A document with no leading
    ``---`` fence has no frontmatter: ``({}, document)`` is returned unchanged.

    Raises ``OKFFrontmatterError`` on an unterminated fence or any construct
    outside the minimal flat subset (anchors, aliases, explicit tags, merge
    keys, block scalars, flow sequences, nested mappings). The single exception
    is the typed, allowlisted flow mapping (:func:`_parse_flow_mapping`), which
    parses into a ``dict`` of allowlisted keys with plain-scalar leaves — every
    other route to a mapping still raises.
    """
    lines = document.split("\n")
    if not lines or lines[0].strip() != _FENCE:
        return {}, document

    close_idx = None
    for i in range(1, len(lines)):
        if lines[i].strip() == _FENCE:
            close_idx = i
            break
    if close_idx is None:
        raise OKFFrontmatterError("unterminated frontmatter: no closing '---' fence")

    frontmatter = _parse_flat(lines[1:close_idx])
    body = "\n".join(lines[close_idx + 1:])
    return frontmatter, body


def scan_concept(document, *, source=Source.OUTPUT):
    """Scan every scannable region of one OKF concept, merged into one Report.

    T1 — whole-concept scan surface. The body is not the only injectable region:
    OKF frontmatter *values* (notably ``description``, which propagates into
    ``index.md`` and is read first under progressive disclosure), ``tags`` items
    and the ``resource`` string are all attacker-controlled and must go through
    the same ``scan_output`` path as the body. Findings from all regions are
    merged so nothing in the frontmatter escapes the gate.

    Frontmatter is parsed with the strict :func:`parse_frontmatter` gate first,
    so a parse-safety violation (T2) raises before any scanning.
    """
    frontmatter, body = parse_frontmatter(document)
    report = Report()
    for region in _scannable_regions(frontmatter, body):
        report.extend(scan_output(region, source=source).findings)
    return report


def _scannable_regions(frontmatter, body):
    """The text regions of a concept that carry attacker-controlled content."""
    regions = [body]
    for value in frontmatter.values():
        regions.extend(_value_regions(value))
    return regions


def _value_regions(value):
    """Every scannable leaf of one frontmatter value.

    A mapping value (G3) is a new *shape* on this surface, not a new exemption:
    its leaves are scanned exactly like a scalar or a list item, so an injection
    parked in ``generated: { by: ... }`` reaches ``scan_output`` like any other
    frontmatter text. The same holds for a *list* of mappings (G30, ``sources``),
    which this function already flattens through its list branch. Mapping *keys*
    are not scanned because they cannot carry attacker text - the allowlist
    admits a fixed, per-parent name set and nothing else.
    """
    if isinstance(value, dict):
        return [leaf for leaf in value.values() if leaf]
    if isinstance(value, list):
        regions = []
        for item in value:
            regions.extend(_value_regions(item))
        return regions
    return [value] if value else []


def validate_concept_path(path, *, allow_reserved=False):
    """Validate a bundle-relative concept path and return its concept-ID.

    T4 — path / reserved-name gate. The concept-ID is the path with the ``.md``
    suffix removed (OKF spec). Rejects, before the path is ever used to write:

    - ``..`` traversal at any segment (escape the bundle);
    - absolute paths (``/...``) and backslashes (platform-separator ambiguity);
    - the reserved basenames ``index.md`` / ``log.md`` (shadow the directory
      listing / update log), case-insensitively — a case-insensitive filesystem
      lets ``Index.md`` shadow ``index.md``;
    - non-``.md`` files (not a concept document).

    ``allow_reserved`` (default ``False``) keeps this a strict concept-path
    validator: a reserved basename is not a concept and is rejected. A mode-b
    bundle import passes ``allow_reserved=True`` because a *received* bundle MAY
    legitimately carry ``index.md`` / ``log.md`` as structural files — the caller
    then scans their body rather than persisting them as concepts. The path-safety
    checks (traversal / absolute / backslash / ``.md``) still apply either way.

    Raises :class:`OKFPathError` on any of these; returns the concept-ID string.
    """
    if not path or not isinstance(path, str):
        raise OKFPathError("empty or non-string concept path: %r" % (path,))
    if path.startswith("/"):
        raise OKFPathError("concept path must be bundle-relative, not absolute: %r" % path)
    if "\\" in path:
        raise OKFPathError("backslashes are not permitted in a concept path: %r" % path)

    segments = path.split("/")
    for seg in segments:
        if seg == "..":
            raise OKFPathError("path traversal ('..') is not permitted: %r" % path)
        if seg == "" or seg == ".":
            raise OKFPathError("malformed path segment in %r" % path)

    basename = segments[-1]
    if not allow_reserved and basename.lower() in _RESERVED_BASENAMES:
        raise OKFPathError("reserved filename may not name a concept: %r" % basename)
    if not basename.lower().endswith(".md"):
        raise OKFPathError("a concept document must be a .md file: %r" % path)

    return path[: -len(".md")]


def validate_resource_url(url):
    """Validate a concept's ``resource`` URL against the https allowlist (T3).

    The OKF format places no constraint on the ``resource`` scheme (verified
    against SPEC.md), so this default-deny allowlist is the only gate: it
    **rejects** anything that is not ``https`` — ``http``, ``data:``,
    ``javascript:``, ``file:``, ``blob:``, ``ftp:`` and schemeless/relative
    strings — *before commit*. This is reject, not defang: ``neutralize`` renders
    dangerous schemes inert for human audit; this refuses to persist them at all.

    Returns ``url`` unchanged on success; raises :class:`OKFResourceError`
    otherwise.
    """
    if not url or not isinstance(url, str):
        raise OKFResourceError("empty or non-string resource URL: %r" % (url,))
    stripped = url.strip()
    if " " in stripped or any(ord(c) < 0x20 for c in stripped):
        raise OKFResourceError("resource URL contains whitespace/control chars: %r" % url)
    match = _URL_SCHEME_RE.match(stripped)
    scheme = match.group(1).lower() if match else None
    if scheme != "https":
        raise OKFResourceError(
            "resource URL must use the https scheme (got %r): %r" % (scheme, 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)


# --- T7: bundle-import iterator (mode b) -------------------------------------

# WARN < QUARANTINE_REVIEW < FAIL_SECURE — the aggregate is the most severe.
_DISPOSITION_ORDER = (
    Disposition.WARN,
    Disposition.QUARANTINE_REVIEW,
    Disposition.FAIL_SECURE,
)


@dataclass(frozen=True)
class ConceptResult:
    """The outcome of validating one concept in a bundle.

    ``error`` is ``None`` on a concept that passed the gates (and then carries a
    ``stamp``); a non-``None`` ``error`` means a hard reject (bad path, unsafe
    frontmatter, or a non-https ``resource``) — ``disposition`` is FAIL_SECURE
    and no stamp is produced, so the concept must not be merged.
    """

    path: str
    concept_id: str | None
    disposition: Disposition
    stamp: ProvenanceStamp | None
    report: Report
    error: str | None


@dataclass(frozen=True)
class BundleResult:
    """Per-concept results, the aggregate disposition, and the cross-link graph.

    ``links`` is the in-import :class:`LinkGraphResult` for the whole bundle
    (dangling / rejected / resolved edges), so a mode-b import returns both halves
    of the gate together. Whether a dangling or rejected link should block is the
    caller's disposition call (design principle 4).
    """

    concepts: tuple
    disposition: Disposition
    links: "LinkGraphResult"

    def log(self):
        """The ``log.md`` body — one line per concept, rejected ones marked."""
        lines = []
        for c in self.concepts:
            if c.stamp is not None:
                lines.append(format_log_entry(c.stamp))
            else:
                lines.append("\t".join([c.path, "REJECTED", c.disposition.value, c.error or ""]))
        return "\n".join(lines)


def import_bundle(bundle, *, origin=Origin.EXTERNAL, channel=Channel.AUTOMATIC, allow_reserved=True):
    """Validate a received OKF bundle concept-by-concept before merge (mode b).

    ``bundle`` maps concept path (e.g. ``tables/users.md``) to its raw document
    text. Each concept runs the full per-concept gate — path/reserved-name (T4),
    frontmatter parse-safety (T2), ``resource`` allowlist (T3), whole-concept
    scan (T1) and provenance stamping (T6). A concept that fails a hard gate is
    rejected (FAIL_SECURE) and recorded, but iteration continues, so the caller
    sees every issue in the bundle, not only the first. The bundle disposition is
    the most severe across its concepts.

    ``allow_reserved`` (default ``True``) reflects that this is the mode-b
    *received-bundle* path: ``index.md`` / ``log.md`` are legitimate structural
    files (OKF spec §3.1/§6/§7) that MAY appear at any level, so they are scanned
    (their body is the highest-priority injection surface) rather than
    path-rejected — over-blocking a conformant third-party bundle is itself a
    failure mode (brief principle 5). A front-end materialising individual
    *uploads* passes ``allow_reserved=False``: there a reserved basename is a
    shadow of the directory listing and must be refused.
    """
    results = tuple(
        _validate_concept(path, bundle[path], origin, channel, allow_reserved=allow_reserved)
        for path in sorted(bundle)
    )
    aggregate = _most_severe(r.disposition for r in results)
    return BundleResult(results, aggregate, link_graph(bundle))


def _validate_concept(path, doc, origin, channel, *, allow_reserved=True):
    try:
        concept_id = validate_concept_path(path, allow_reserved=allow_reserved)
    except OKFPathError as exc:
        return ConceptResult(path, None, Disposition.FAIL_SECURE, None, Report(), str(exc))
    try:
        frontmatter, _body = parse_frontmatter(doc)
    except OKFFrontmatterError as exc:
        return ConceptResult(path, concept_id, Disposition.FAIL_SECURE, None, Report(), str(exc))
    resource = frontmatter.get("resource")
    if isinstance(resource, str):
        try:
            validate_resource_url(resource)
        except OKFResourceError as exc:
            return ConceptResult(path, concept_id, Disposition.FAIL_SECURE, None, Report(), str(exc))
    report = scan_concept(doc)
    stamp = stamp_concept(concept_id, report, origin, channel)
    return ConceptResult(path, concept_id, stamp.disposition, stamp, report, None)


def _most_severe(dispositions):
    worst = Disposition.WARN
    for disposition in dispositions:
        if _DISPOSITION_ORDER.index(disposition) > _DISPOSITION_ORDER.index(worst):
            worst = disposition
    return worst


# --- T5a / A: cross-link graph (in-import) -----------------------------------
# The persisted cross-run graph (B) that would catch "plant a link now, write
# the poisoned target in a LATER run" (§7.2) is deferred to stream 2, where the
# consumer that owns the corpus decides where the durable graph state lives.
# This in-import graph resolves links within a single bundle merge.

# ReDoS note (OWASP LLM10): the label run excludes `[`, the character that opens
# this pattern's own anchor. Without it, a bundle body repeating `[` and never
# closing it makes every start position rescan the tail — 7.1s at 100_000 chars,
# exponent ~2.0, over attacker-supplied bodies this adapter reads with no input
# cap. Same defect and same fix as `active_content.MD_LINK_RE`, including the
# trade it names: a label containing a nested `[...]` is given up on, which costs
# no exfil coverage because the inner link is matched on its own.
_MD_LINK_RE = re.compile(r"\[[^\]\[]*\]\(\s*([^)\s]+)")
# Active-content schemes are refused in a link, mirroring the resource gate (T3).
_DANGEROUS_LINK_SCHEMES = frozenset({"javascript", "data", "vbscript", "file", "blob"})


@dataclass(frozen=True)
class LinkGraphResult:
    """In-import cross-link resolution over one bundle.

    ``dangling`` — ``(from_id, target_concept_id)`` for in-bundle ``.md`` links
    whose target concept is **not present** in the bundle: the dormant-injection
    signal of §7.2 (a link planted to a not-yet-written concept). ``rejected`` —
    ``(from_id, target, reason)`` for links refused outright (dangerous scheme or
    bundle escape). ``resolved`` — ``(from_id, target_concept_id)`` for links to
    concepts present in the bundle. ``truncated`` — ``(from_id, body_length)`` for
    bodies read only as far as the scan cap, so a caller can tell "no links past
    here" apart from "no links *read* past here" (OWASP LLM10).
    """

    dangling: tuple
    rejected: tuple
    resolved: tuple
    truncated: tuple = ()


def extract_link_targets(body):
    """Return the destinations of markdown ``[text](target)`` links in ``body``."""
    return _MD_LINK_RE.findall(body)


def resolve_link(target, from_concept_id):
    """Resolve one link target to an in-bundle concept-ID, or reject it (T5a).

    Returns the target concept-ID for an in-bundle ``.md`` link (bundle-absolute
    ``/x.md`` or relative ``./x.md`` / ``../y.md``, resolved against the linking
    concept's directory). Returns ``None`` for an external ``http(s)``/other
    non-active link (not a concept edge) and for non-``.md`` targets. Raises
    :class:`OKFLinkError` for an active-content scheme or a ``..`` escape past the
    bundle root.
    """
    candidate = target.strip().split("#", 1)[0].split("?", 1)[0]
    if not candidate:
        return None

    scheme_match = _URL_SCHEME_RE.match(candidate)
    if scheme_match:
        scheme = scheme_match.group(1).lower()
        if scheme in _DANGEROUS_LINK_SCHEMES:
            raise OKFLinkError("link uses a dangerous scheme %r: %r" % (scheme, target))
        return None  # external (http/https/mailto/…): not an in-bundle concept edge

    if not candidate.endswith(".md"):
        return None  # not a concept-document link (asset, anchor, …)

    if candidate.startswith("/"):
        normalized = _normalize_bundle_path(candidate[1:])
    else:
        from_dir = from_concept_id.rsplit("/", 1)[0] if "/" in from_concept_id else ""
        joined = from_dir + "/" + candidate if from_dir else candidate
        normalized = _normalize_bundle_path(joined)

    return normalized[: -len(".md")]


def link_graph(bundle, max_scan_chars=MAX_SCAN_CHARS):
    """Resolve every cross-link in ``bundle`` against the concepts it contains.

    ``bundle`` maps concept path to document text (as :func:`import_bundle`). Only
    the body is scanned for links. See :class:`LinkGraphResult` for the outcome.

    Self-safety (OWASP LLM10): every body is attacker-supplied and each is walked
    by a `findall`, so each body is capped at ``max_scan_chars`` and recorded in
    ``truncated``. It truncates rather than raising, the way the scanners do: the
    graph reports on documents, it does not hand them back, so a shortened scan
    costs edges — not the caller's content.
    """
    present = {p[: -len(".md")] for p in bundle if p.endswith(".md")}
    dangling, rejected, resolved, truncated = [], [], [], []

    for path in sorted(bundle):
        if not path.endswith(".md"):
            continue
        from_id = path[: -len(".md")]
        try:
            _frontmatter, body = parse_frontmatter(bundle[path])
        except OKFFrontmatterError:
            body = bundle[path]  # unparseable frontmatter is T2's reject, not ours

        if len(body) > max_scan_chars:
            truncated.append((from_id, len(body)))
            body = body[:max_scan_chars]

        for target in extract_link_targets(body):
            try:
                concept_id = resolve_link(target, from_id)
            except OKFLinkError as exc:
                rejected.append((from_id, target, str(exc)))
                continue
            if concept_id is None:
                continue
            if concept_id in present:
                resolved.append((from_id, concept_id))
            else:
                dangling.append((from_id, concept_id))

    return LinkGraphResult(
        tuple(dangling), tuple(rejected), tuple(resolved), tuple(truncated)
    )


def _normalize_bundle_path(path):
    """Normalize a ``/``-separated bundle path; raise if it escapes the root."""
    parts = []
    for segment in path.split("/"):
        if segment in ("", "."):
            continue
        if segment == "..":
            if not parts:
                raise OKFLinkError("link target escapes the bundle root: %r" % path)
            parts.pop()
        else:
            parts.append(segment)
    return "/".join(parts)


def _parse_flat(fm_lines):
    result = {}
    i = 0
    n = len(fm_lines)
    while i < n:
        raw = fm_lines[i]
        stripped = raw.strip()

        if stripped == "" or stripped.startswith("#"):
            i += 1
            continue

        # An indented line with no active list key is a nested structure.
        if raw[:1] in (" ", "\t"):
            raise OKFFrontmatterError(
                "nested mappings are not supported in OKF frontmatter: %r" % raw
            )

        if stripped.startswith("<<"):
            raise OKFFrontmatterError("YAML merge keys are not permitted")

        if ":" not in stripped:
            raise OKFFrontmatterError("malformed frontmatter line: %r" % raw)

        key, _, value = stripped.partition(":")
        key = key.strip()
        value = value.strip()
        if not _KEY_RE.match(key):
            raise OKFFrontmatterError("invalid frontmatter key: %r" % key)

        if value == "":
            items, i = _consume_block_list(fm_lines, i + 1, key)
            result[key] = items if items is not None else ""
            continue

        mapping = _parse_flow_mapping(value, key)
        if mapping is not None:
            result[key] = mapping
            i += 1
            continue

        sequence = _parse_flow_sequence(value, key)
        if sequence is not None:
            result[key] = sequence
            i += 1
            continue

        _reject_dangerous_value(value)
        _reject_mapping_construct(value)
        result[key] = value
        i += 1

    return result


def _consume_block_list(fm_lines, start, parent_key=None):
    """Consume ``  - item`` lines following a bare ``key:``.

    Returns ``(items, next_index)`` - ``items`` is ``None`` (and ``next_index``
    unchanged) when no list item follows, so the caller can treat the key as an
    empty scalar and let the next line trip the nested-structure guard.

    An item is one of three shapes, decided by the item text alone: a flow
    mapping (G3), a block mapping (G30 - an unquoted ``key: value`` opening a
    run of more-indented sibling entries, which is SPEC.md §5.1's own carrier
    for ``sources``), or a plain scalar. ``parent_key`` is the key that owns the
    list; it decides the mapping-key allowlist, which is how ``sources[].resource``
    is admitted while ``executor``/``attester`` ``resource`` stays refused.

    A list may not mix scalars and mappings. YAML permits it, but a consumer
    iterating ``sources`` and reading ``entry.get("id")`` gets an
    ``AttributeError`` off the first ``str`` - refusing is the cheaper failure.
    """
    items = []
    kinds = set()
    i = start
    n = len(fm_lines)
    while i < n:
        raw = fm_lines[i]
        stripped = raw.strip()
        if stripped == "" or stripped.startswith("#"):
            i += 1
            continue
        if not (raw[:1] in (" ", "\t") and stripped.startswith("- ")):
            break

        item = stripped[2:].strip()

        mapping = _parse_flow_mapping(item, parent_key)
        if mapping is not None:
            items.append(mapping)
            kinds.add("mapping")
            i += 1
            continue

        entry = _block_mapping_entry(item)
        if entry is not None:
            mapping, i = _consume_block_mapping(fm_lines, i + 1, entry, parent_key)
            items.append(mapping)
            kinds.add("mapping")
            continue

        _reject_dangerous_value(item)
        _reject_mapping_construct(item)
        items.append(item)
        kinds.add("scalar")
        i += 1

    if len(kinds) > 1:
        raise OKFFrontmatterError(
            "a block list may not mix scalar items and mappings: %r" % (parent_key,)
        )
    if not items:
        return None, start
    return items, i


def _block_mapping_entry(text):
    """Read ``text`` as one ``key: value`` block-mapping entry, or return ``None``.

    The trigger is deliberately the same shape ``_reject_mapping_construct``
    uses to *refuse* a scalar: an unquoted ``": "``. What changes in 1.3.0 is
    only what happens next - the entry is admitted key-by-key against the
    allowlist instead of refused wholesale. Every shape that is a scalar to
    PyYAML stays one here: a quoted item, a colon with no space
    (``domain:security``, ``https://e.com:8443/a``) and a trailing colon all
    return ``None`` and fall through to the unchanged scalar rules.
    """
    if not text or text[0] in _QUOTE_STARTS:
        return None
    key, sep, leaf = text.partition(": ")
    if not sep:
        return None
    key = key.strip()
    if not _KEY_RE.match(key):
        return None
    return key, leaf.strip()


def _consume_block_mapping(fm_lines, start, first_entry, parent_key):
    """Consume the sibling entries of a block mapping opened by a ``- `` item.

    Returns ``(mapping, next_index)``. A sibling is an indented line that does
    not open a new list item; the run ends at a blank line, a comment, a new
    ``- `` item, or a line at column zero. Depth is capped at one by giving the
    leaves the *unchanged* scalar predicates: a nested collection opens with
    ``{`` or ``[`` and is refused by ``_reject_dangerous_value``, and a further
    block level is refused by ``_reject_mapping_construct``.
    """
    allowed = _allowed_mapping_keys(parent_key)
    mapping = {}
    _admit_mapping_entry(mapping, first_entry[0], first_entry[1], allowed, parent_key)

    i = start
    n = len(fm_lines)
    while i < n:
        raw = fm_lines[i]
        stripped = raw.strip()
        if stripped == "" or stripped.startswith("#"):
            break
        if raw[:1] not in (" ", "\t") or stripped.startswith("- "):
            break
        entry = _block_mapping_entry(stripped)
        if entry is None:
            _reject_dangerous_value(stripped)
            _reject_mapping_construct(stripped)
            raise OKFFrontmatterError(
                "a block-mapping entry must be 'key: value': %r" % (raw,)
            )
        _admit_mapping_entry(mapping, entry[0], entry[1], allowed, parent_key)
        i += 1
    return mapping, i


def _admit_mapping_entry(mapping, key, leaf, allowed, parent_key):
    """Admit one mapping entry, or raise. The single gate both carriers pass."""
    if not _KEY_RE.match(key):
        raise OKFFrontmatterError("invalid mapping key: %r" % (key,))
    if key not in allowed:
        raise OKFFrontmatterError(
            "mapping key %r is not on the OKF mapping allowlist under %r"
            % (key, parent_key)
        )
    if key in mapping:
        raise OKFFrontmatterError("duplicate mapping key %r" % (key,))
    _reject_dangerous_value(leaf)
    _reject_mapping_construct(leaf)
    mapping[key] = leaf


def _reject_dangerous_value(value):
    if value and value[0] in _DANGEROUS_VALUE_STARTS:
        raise OKFFrontmatterError(
            "value begins with a disallowed YAML indicator %r: %r"
            % (value[0], value)
        )


def _reject_mapping_construct(value):
    """Reject a scalar that YAML reads as a mapping rather than as a string.

    T2 gives the mapping *class* exactly one expressible form, the typed
    allowlisted flow mapping (G3); the nested-block and dotted-key routes still
    raise, and this predicate is what keeps them raising — both at the top level
    and on a leaf *inside* an admitted mapping. Two routes used to escape by degrading
    into a string instead: a block-sequence item carrying exactly one key
    (``- uri: x``), and an inline second colon (``attester: resource: x``).
    Both parsed "successfully" into the wrong *type*, and a pointer parked in
    one rode through in a key the ``resource`` allowlist never inspects.

    ``": "`` and a trailing ``":"`` are exactly the two shapes where a plain
    scalar stops being one — ground-truthed against PyYAML 6.0.3, which reads
    ``- uri: x`` as ``[{'uri': 'x'}]``, ``- uri:`` as ``[{'uri': None}]``, and
    refuses ``k: sub: v`` outright. A colon carrying neither a space nor a line
    end opens no mapping (``domain:security``, ``https://e.com:8443/a``) and is
    left alone, as is a quoted scalar — over-blocking a conformant bundle is
    itself a failure mode.
    """
    if not value or value[0] in _QUOTE_STARTS:
        return
    if ": " in value or value.endswith(":"):
        raise OKFFrontmatterError(
            "a mapping is not expressible in OKF frontmatter: %r" % (value,)
        )


def _parse_flow_mapping(value, parent_key=None):
    """Parse ``{ key: value, ... }`` into a typed dict, or refuse it (G3).

    Returns ``None`` when ``value`` does not open a flow mapping, so the caller
    falls through to the unchanged scalar rules. Otherwise the value either
    parses into a ``dict`` of allowlisted keys with plain-scalar leaves, or
    raises - it never degrades into a string, which is the defect closed in
    1.1.0 and not reopened here.

    Why the mapping class needed *a* form at all: OKF v0.2 writes its whole
    trust and provenance layer as mappings, and SPEC.md @ ``62432a09`` uses flow
    form in its own examples (§5.1 ``usage_window``, §5.2 ``generated`` /
    ``verified``). §11 goes further than "should": a consumer *MUST* treat a
    bare ``verified`` mapping as a one-element list - a rule that presupposes
    the mapping parses. With no form, 0 of 53 upstream concepts reached the
    gate, and no threshold would have changed that.

    Why this form is safe: the allowlist inspects **every key**, which is the
    property that actually carried the security in T2 - the blanket refusal was
    the enforcement, not the point. Admitted, ground-truthed against PyYAML
    6.0.3:

    - one flow mapping per value, closed on the same line (``{ a: b }``);
    - keys on :data:`_MAPPING_KEY_ALLOWLIST` and matching ``_KEY_RE``, no
      duplicates - PyYAML resolves a duplicate last-wins, which is a way to
      show one claim and mean another;
    - plain-scalar leaves only, each run through the *unchanged*
      ``_reject_dangerous_value`` / ``_reject_mapping_construct`` predicates, so
      a leaf can no more open an anchor, a tag or a nested mapping than a
      top-level scalar can.

    Refused, each on its own rule: nested collections (``{ a: { b: c } }``,
    ``{ a: [1] }``), quoted leaves, an empty mapping, an unclosed or
    trailing-junk value (``{ a: b } x``, which PyYAML also refuses), a key
    outside the allowlist, and ``{a:b}`` - which PyYAML reads as the *key*
    ``a:b``, not as a scalar, and which the required ``": "`` separator catches.

    Two deliberate divergences from PyYAML, both toward refusal: a quoted leaf
    (``{ title: 'a, b' }``) and a trailing comment (``{ a: b } # note``) are
    conformant YAML that this rejects. Splitting quoted commas correctly needs a
    quote state machine whose failure mode is *accepting* something YAML would
    refuse; refusing is the cheaper side to be wrong on, and the keys that
    plausibly need a comma (``title``, ``author``) only occur inside ``sources``
    entries, whose block-sequence carrier is refused anyway.
    """
    if not value or value[0] != "{":
        return None
    if not value.endswith("}"):
        raise OKFFrontmatterError(
            "a flow mapping must be closed by '}' on the same line: %r" % (value,)
        )

    inner = value[1:-1].strip()
    if inner.endswith(","):  # a trailing comma is legal YAML; one, and only one
        inner = inner[:-1].strip()
    if not inner:
        raise OKFFrontmatterError("an empty flow mapping carries nothing: %r" % (value,))
    for char in "{}[]":
        if char in inner:
            raise OKFFrontmatterError(
                "a flow mapping admits scalar leaves only, not %r: %r" % (char, value)
            )
    for quote in _QUOTE_STARTS:
        if quote in inner:
            raise OKFFrontmatterError(
                "a quoted scalar inside a flow mapping is not a supported form: %r"
                % (value,)
            )

    allowed = _allowed_mapping_keys(parent_key)
    mapping = {}
    for entry in inner.split(","):
        entry = entry.strip()
        key, sep, leaf = entry.partition(": ")
        if not sep:
            raise OKFFrontmatterError(
                "a flow-mapping entry must be 'key: value': %r" % (entry,)
            )
        _admit_mapping_entry(mapping, key.strip(), leaf.strip(), allowed, parent_key)
    return mapping


def _parse_flow_sequence(value, parent_key=None):
    """Parse ``[{ ... }, { ... }]`` into a list of typed dicts, or refuse it (G30).

    Returns ``None`` when ``value`` does not open a flow sequence, so the caller
    falls through to the unchanged scalar rules - where ``[`` is still a
    disallowed indicator. This carrier is opened for the flow-mapping element
    and nothing else: it is the form the OKF producers emit for ``sources``
    (measured 02.09 against llm-ingestion-okf's golden bundle, where a
    one-element sequence raised on the ``[`` just as a two-element one did).

    A flow sequence of plain *scalars* (``tags: [a, b, c]``) stays refused. It
    is a different shape with its own quoting and comma-splitting problem, whose
    failure mode would be accepting something YAML reads differently - and the
    block-sequence carrier already covers it for every consumer measured so far.

    Elements are split on ``}`` rather than on commas, which is sound precisely
    because ``_parse_flow_mapping`` admits no nested collection: a ``}`` inside
    an element cannot occur, so the first ``}`` after ``{`` always closes it.
    Anything between elements that is not a separating comma is refused, which
    is what makes trailing junk and a mixed sequence fail rather than parse.
    """
    if not value or value[0] != "[":
        return None
    if not value.endswith("]"):
        raise OKFFrontmatterError(
            "a flow sequence must be closed by ']' on the same line: %r" % (value,)
        )

    inner = value[1:-1].strip()
    if not inner:
        raise OKFFrontmatterError("an empty flow sequence carries nothing: %r" % (value,))

    items = []
    i = 0
    n = len(inner)
    while True:
        while i < n and inner[i] in " \t":
            i += 1
        if i >= n:
            break
        if inner[i] != "{":
            raise OKFFrontmatterError(
                "a flow sequence admits flow mappings only: %r" % (value,)
            )
        close = inner.find("}", i)
        if close == -1:
            raise OKFFrontmatterError(
                "an unclosed flow mapping inside a flow sequence: %r" % (value,)
            )
        items.append(_parse_flow_mapping(inner[i:close + 1], parent_key))
        i = close + 1
        while i < n and inner[i] in " \t":
            i += 1
        if i >= n:
            break
        if inner[i] != ",":
            raise OKFFrontmatterError(
                "trailing junk after a flow-sequence element: %r" % (value,)
            )
        i += 1
    return items