feat(okf): strict reject-by-default frontmatter parser (T2, TDD, +12)

src/llm_ingestion_guard/okf.py

@@ -0,0 +1,149 @@
+"""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 plus block ``- item`` lists. 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. 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
+
+__all__ = ["parse_frontmatter", "OKFError", "OKFFrontmatterError"]
+
+_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.
+_DANGEROUS_VALUE_STARTS = frozenset("&*!|>[]{}%@`")
+
+
+class OKFError(Exception):
+    """Base class for OKF adapter rejections."""
+
+
+class OKFFrontmatterError(OKFError):
+    """Frontmatter violates the strict, reject-by-default OKF subset."""
+
+
+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 collections, nested mappings).
+    """
+    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 _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)
+            result[key] = items if items is not None else ""
+            continue
+
+        _reject_dangerous_value(value)
+        result[key] = value
+        i += 1
+
+    return result
+
+
+def _consume_block_list(fm_lines, start):
+    """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.
+    """
+    items = []
+    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 raw[:1] in (" ", "\t") and stripped.startswith("- "):
+            item = stripped[2:].strip()
+            _reject_dangerous_value(item)
+            items.append(item)
+            i += 1
+            continue
+        break
+    if not items:
+        return None, start
+    return items, i
+
+
+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)
+        )