home-network-k3s-impl/platform/vuln-scanner/src/acceptances.py

"""R-VULN-08 acceptances: a strict parser for exactly the spec's flat list, no YAML library.

    - id: CVE-2026-00000
      image: registry/repository     # exact match; omit only for platform-wide
      reason: not reachable - feature disabled
      expires: 2027-02-01            # mandatory, at most 180 days out

Fail-safe: a malformed entry is ignored and reported, so it can never hide a finding. Anything the
parser cannot place (bad indentation, a stray line) invalidates the whole file for the same reason.
"""
import re
import sys
from datetime import date, timedelta

from state import utcnow

MAX_DAYS = 180
MAX_REASON = 2048  # report.schema.json's limit; a longer reason would make the whole report invalid
_KEYS = ("id", "image", "reason", "expires")
_ENTRY = re.compile(r"^- ([a-z]+):(?: (.*))?$")
_FIELD = re.compile(r"^  ([a-z]+):(?: (.*))?$")
_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]*$")
_DATE = re.compile(r"^\d{4}-\d{2}-\d{2}$")
_FORBIDDEN_START = tuple("[{&*!|>%@`")


def _scalar(raw):
    """(value, None) or (None, error) for an unquoted or simply-quoted YAML scalar."""
    value = (raw or "").strip()
    if value[:1] in ("'", '"'):
        quote = value[0]
        end = value.find(quote, 1)
        if end == -1 or "\\" in value[:end]:
            return None, "bad quoting"
        rest = value[end + 1:].strip()
        if rest and not rest.startswith("#"):
            return None, "text after the closing quote"
        return value[1:end], None
    if value.startswith(_FORBIDDEN_START):
        return None, "only plain scalars are allowed"
    comment = re.search(r"\s#", value)
    if comment:
        value = value[:comment.start()].rstrip()
    return value, None


def parse_text(source, today=None):
    """(valid entries, error strings). Entries carry id, reason, expires (date) and optional image."""
    today = today or utcnow().date()
    errors = []
    raw_entries = []
    current = None
    for lineno, line in enumerate(source.splitlines(), 1):
        if not line.strip() or line.lstrip().startswith("#"):
            continue
        entry = _ENTRY.match(line)
        field = _FIELD.match(line)
        if line.strip().split("#", 1)[0].strip() == "[]" and not raw_entries and current is None:
            continue
        if entry:
            current = {"line": lineno, "fields": {}, "errors": []}
            raw_entries.append(current)
            key, raw = entry.group(1), entry.group(2)
        elif field and current is not None:
            key, raw = field.group(1), field.group(2)
        else:
            return [], ["line " + str(lineno) + ": not a flat acceptance entry; the whole file is ignored"]
        value, problem = _scalar(raw)
        if key not in _KEYS:
            current["errors"].append("unknown key " + key)
        elif problem:
            current["errors"].append(key + ": " + problem)
        elif key in current["fields"]:
            current["errors"].append("duplicate key " + key)
        else:
            current["fields"][key] = value
    entries = []
    for raw in raw_entries:
        problems = list(raw["errors"]) + _validate(raw["fields"], today)
        if problems:
            errors.append("entry at line " + str(raw["line"]) + ": " + "; ".join(problems))
            continue
        fields = raw["fields"]
        entries.append({"id": fields["id"], "image": fields.get("image") or None, "reason": fields["reason"],
                        "expires": date.fromisoformat(fields["expires"])})
    return entries, errors


def _validate(fields, today):
    problems = []
    if not _ID.match(fields.get("id") or ""):
        problems.append("id is missing or malformed")
    if not fields.get("reason"):
        problems.append("reason is mandatory")
    elif len(fields["reason"]) > MAX_REASON:
        problems.append("reason is longer than " + str(MAX_REASON) + " characters")
    if "image" in fields and not fields["image"]:
        problems.append("image is empty")
    expires = fields.get("expires") or ""
    if not _DATE.match(expires):
        problems.append("expires is mandatory and must be YYYY-MM-DD")
        return problems
    try:
        when = date.fromisoformat(expires)
    except ValueError:
        problems.append("expires is not a real date")
        return problems
    if when > today + timedelta(days=MAX_DAYS):
        problems.append("expires is more than " + str(MAX_DAYS) + " days out")
    return problems


def load(path, today=None):
    """Entries and errors from a file. An unreadable file is an error, never silently empty."""
    try:
        with open(path, encoding="utf-8") as handle:
            source = handle.read(1024 * 1024 + 1)
    except (OSError, UnicodeDecodeError) as exc:
        return [], ["acceptances file unreadable: " + type(exc).__name__]
    if len(source) > 1024 * 1024:
        return [], ["acceptances file exceeds 1 MiB"]
    return parse_text(source, today)


def match(entries, finding_id, image, today=None):
    """The unexpired acceptance covering a finding, if any. Expiry day itself still counts."""
    today = today or utcnow().date()
    best = None
    for entry in entries:
        if entry["id"] != finding_id or entry["expires"] < today:
            continue
        if entry["image"] is not None and entry["image"] != image:
            continue
        if best is None or entry["expires"] > best["expires"]:
            best = entry
    return best


if __name__ == "__main__":
    found, problems = load(sys.argv[1])
    for problem in problems:
        print("INVALID: " + problem, file=sys.stderr)
    print(str(len(found)) + " valid entries")
    sys.exit(1 if problems else 0)