framing_diff.py

#!/usr/bin/env python3
"""
HTTP framing differential harness (offline).

Feeds the same complete or incomplete byte stream to one or more parser
backends and prints a structured comparison. This tool never assigns a
smuggling verdict. A security claim requires:

  1. valid or intentionally classified syntax;
  2. complete / fragmented / EOF cases where relevant;
  3. negative controls;
  4. at least two real components that disagree on message boundaries;
  5. an impact oracle outside pure parser state.

Built-in backends:
  - reference: pure-Python structural observer (headers, CL/TE conflict flags)
  - subprocess backends (optional): path to a CLI that reads stdin and prints
    JSON with keys consumed, messages, incomplete, error

Example:
  python framing_diff.py --bytes-file sample.http --label incomplete-chunk
  python framing_diff.py --variant incomplete-declared-chunk --list-classes
"""

from __future__ import annotations

import argparse
import json
import subprocess
import sys
from dataclasses import asdict, dataclass, field
from pathlib import Path
from typing import Optional

# ---------------------------------------------------------------------------
# Classification vocabulary — keep incomplete streams out of "smuggle"
# ---------------------------------------------------------------------------

CLASS_VALID = "valid"
CLASS_INVALID_SYNTAX = "invalid-syntax"
CLASS_AMBIGUOUS_FRAMING = "ambiguous-framing"
CLASS_INCOMPLETE = "incomplete-stream"
CLASS_KNOWN_CVE_SEED = "known-cve-seed"

ALL_CLASSES = (
    CLASS_VALID,
    CLASS_INVALID_SYNTAX,
    CLASS_AMBIGUOUS_FRAMING,
    CLASS_INCOMPLETE,
    CLASS_KNOWN_CVE_SEED,
)


@dataclass
class Case:
    name: str
    classification: str
    data: bytes
    notes: str = ""


@dataclass
class ParseResult:
    backend: str
    consumed: int
    messages: int
    incomplete: bool
    error: Optional[str]
    observations: dict = field(default_factory=dict)
    raw: str = ""


# ---------------------------------------------------------------------------
# Built-in cases (including the invalidated false-positive pattern)
# ---------------------------------------------------------------------------

def incomplete_declared_chunk() -> Case:
    """Declared chunk larger than supplied body — streaming wait, not smuggling."""
    body = (
        b"POST / HTTP/1.1\r\n"
        b"Host: a.example\r\n"
        b"Transfer-Encoding: chunked\r\n"
        b"\r\n"
        b"FFFFF\r\n"
        b"GET /admin HTTP/1.1\r\n"
        b"Host: evil.example\r\n"
        b"\r\n"
    )
    return Case(
        "incomplete-declared-chunk",
        CLASS_INCOMPLETE,
        body,
        "Declares 0xFFFFF chunk bytes but supplies far fewer. A streaming "
        "parser that keeps remaining content_length is expected behavior.",
    )


def space_not_chunk_extension() -> Case:
    """'FFFFF F' is not a valid RFC 9112 chunk extension (needs ';')."""
    body = (
        b"POST / HTTP/1.1\r\n"
        b"Host: a.example\r\n"
        b"Transfer-Encoding: chunked\r\n"
        b"\r\n"
        b"FFFFF F\r\n"
        b"0\r\n\r\n"
    )
    return Case(
        "space-not-chunk-extension",
        CLASS_INVALID_SYNTAX,
        body,
        "Space after hex size is not a legal chunk-extension delimiter.",
    )


def valid_chunked_post() -> Case:
    body = (
        b"POST / HTTP/1.1\r\n"
        b"Host: a.example\r\n"
        b"Transfer-Encoding: chunked\r\n"
        b"\r\n"
        b"5\r\nhello\r\n"
        b"0\r\n\r\n"
    )
    return Case("valid-chunked-post", CLASS_VALID, body, "Normal chunked body.")


def cl_te_conflict() -> Case:
    smuggled = b"GPOST / HTTP/1.1\r\nHost: a.example\r\nContent-Length: 10\r\n\r\nx="
    body = b"0\r\n\r\n" + smuggled
    head = (
        f"POST / HTTP/1.1\r\n"
        f"Host: a.example\r\n"
        f"Content-Length: {len(body)}\r\n"
        f"Transfer-Encoding: chunked\r\n"
        f"\r\n"
    ).encode()
    return Case(
        "cl-te-conflict",
        CLASS_AMBIGUOUS_FRAMING,
        head + body,
        "Conflicting CL and TE. Ambiguous framing seed; not a finding by itself.",
    )


def dual_content_length() -> Case:
    body = b"0\r\n\r\nGPOST / HTTP/1.1\r\nHost: a.example\r\n\r\n"
    head = (
        f"POST / HTTP/1.1\r\n"
        f"Host: a.example\r\n"
        f"Content-Length: 5\r\n"
        f"Content-Length: {len(body)}\r\n"
        f"Transfer-Encoding: chunked\r\n"
        f"\r\n"
    ).encode()
    return Case(
        "dual-content-length",
        CLASS_AMBIGUOUS_FRAMING,
        head + body,
        "Two different Content-Length values plus TE.",
    )


def malformed_chunk_size() -> Case:
    body = (
        b"POST / HTTP/1.1\r\n"
        b"Host: a.example\r\n"
        b"Transfer-Encoding: chunked\r\n"
        b"\r\n"
        b"3x\r\nabc\r\n0\r\n\r\n"
    )
    return Case(
        "malformed-chunk-size",
        CLASS_INVALID_SYNTAX,
        body,
        "Non-hex suffix without ';' — not a valid extension.",
    )


def builtin_cases() -> dict[str, Case]:
    cases = [
        incomplete_declared_chunk(),
        space_not_chunk_extension(),
        valid_chunked_post(),
        cl_te_conflict(),
        dual_content_length(),
        malformed_chunk_size(),
    ]
    return {c.name: c for c in cases}


# ---------------------------------------------------------------------------
# Reference backend: structural observation only (no smuggling verdict)
# ---------------------------------------------------------------------------

def _split_headers(data: bytes) -> tuple[Optional[bytes], Optional[bytes]]:
    sep = data.find(b"\r\n\r\n")
    if sep < 0:
        return None, None
    return data[:sep], data[sep + 4:]


def reference_parse(data: bytes) -> ParseResult:
    header_blob, body = _split_headers(data)
    obs: dict = {
        "has_header_terminator": header_blob is not None,
        "content_length_count": 0,
        "transfer_encoding_chunked": False,
        "cl_values": [],
        "looks_incomplete_chunk": False,
        "classification_hint": None,
    }
    if header_blob is None:
        return ParseResult(
            backend="reference",
            consumed=0,
            messages=0,
            incomplete=True,
            error="no header terminator",
            observations=obs,
        )

    lines = header_blob.split(b"\r\n")
    cl_values = []
    te_chunked = False
    for line in lines[1:]:
        lower = line.lower()
        if lower.startswith(b"content-length:"):
            raw = line.split(b":", 1)[1].strip()
            try:
                cl_values.append(int(raw))
            except ValueError:
                cl_values.append(None)
        if lower.startswith(b"transfer-encoding:") and b"chunked" in lower:
            te_chunked = True

    obs["content_length_count"] = len(cl_values)
    obs["cl_values"] = cl_values
    obs["transfer_encoding_chunked"] = te_chunked
    obs["body_len"] = len(body or b"")

    incomplete = False
    error = None
    messages = 1

    if te_chunked and body is not None:
        # Extremely small chunk-size observer: first line of body is size
        first_line, _, rest = body.partition(b"\r\n")
        size_token = first_line.split(b";")[0].strip()
        # space after hex is invalid extension syntax
        if b" " in first_line and b";" not in first_line.split(b"\r\n")[0]:
            obs["invalid_chunk_size_line"] = first_line.decode("latin1", "replace")
        try:
            declared = int(size_token, 16)
            if declared > 0 and len(rest) < declared:
                incomplete = True
                obs["looks_incomplete_chunk"] = True
                obs["declared_chunk"] = declared
                obs["received_after_size_line"] = len(rest)
                obs["classification_hint"] = CLASS_INCOMPLETE
        except ValueError:
            error = f"non-hex chunk size token: {size_token!r}"
            obs["classification_hint"] = CLASS_INVALID_SYNTAX

    if len(cl_values) > 1 and len(set(cl_values)) > 1:
        obs["classification_hint"] = CLASS_AMBIGUOUS_FRAMING
    if te_chunked and cl_values:
        obs["classification_hint"] = obs.get("classification_hint") or CLASS_AMBIGUOUS_FRAMING

    return ParseResult(
        backend="reference",
        consumed=len(data),
        messages=messages,
        incomplete=incomplete,
        error=error,
        observations=obs,
    )


def subprocess_parse(name: str, cmd: list[str], data: bytes, timeout: float = 5.0) -> ParseResult:
    try:
        proc = subprocess.run(
            cmd,
            input=data,
            capture_output=True,
            timeout=timeout,
        )
    except FileNotFoundError as e:
        return ParseResult(name, 0, 0, False, f"backend not found: {e}", {})
    except subprocess.TimeoutExpired:
        return ParseResult(name, 0, 0, False, "timeout", {})

    raw = (proc.stdout or b"") + (proc.stderr or b"")
    text = raw.decode("utf-8", "replace")
    try:
        payload = json.loads(proc.stdout.decode("utf-8"))
        return ParseResult(
            backend=name,
            consumed=int(payload.get("consumed", 0)),
            messages=int(payload.get("messages", 0)),
            incomplete=bool(payload.get("incomplete", False)),
            error=payload.get("error"),
            observations=payload.get("observations") or {},
            raw=text,
        )
    except (json.JSONDecodeError, UnicodeDecodeError, TypeError, ValueError):
        return ParseResult(
            backend=name,
            consumed=len(data),
            messages=0,
            incomplete=False,
            error=f"non-json exit={proc.returncode}",
            observations={},
            raw=text[:500],
        )


def compare(results: list[ParseResult]) -> dict:
    """Structural comparison only — never emits SMUGGLE/VULNERABLE labels."""
    incomplete_flags = {r.backend: r.incomplete for r in results}
    message_counts = {r.backend: r.messages for r in results}
    errors = {r.backend: r.error for r in results}
    disagree_messages = len(set(message_counts.values())) > 1
    disagree_incomplete = len(set(incomplete_flags.values())) > 1
    return {
        "message_counts": message_counts,
        "incomplete_flags": incomplete_flags,
        "errors": errors,
        "disagree_on_message_count": disagree_messages,
        "disagree_on_incomplete": disagree_incomplete,
        "verdict": "OBSERVATION_ONLY",
        "note": (
            "Disagreement is not a vulnerability. Upgrade only with a "
            "controlled multi-component impact oracle and complete-stream tests."
        ),
    }


def run_case(case: Case, extra_backends: list[tuple[str, list[str]]]) -> dict:
    results = [reference_parse(case.data)]
    for name, cmd in extra_backends:
        results.append(subprocess_parse(name, cmd, case.data))
    return {
        "case": case.name,
        "classification": case.classification,
        "notes": case.notes,
        "bytes": len(case.data),
        "results": [asdict(r) for r in results],
        "comparison": compare(results),
    }


def main(argv=None):
    ap = argparse.ArgumentParser(description=__doc__,
                                 formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument("--list-cases", action="store_true", help="list built-in cases")
    ap.add_argument("--list-classes", action="store_true", help="list classification labels")
    ap.add_argument("--variant", help="run a built-in case by name")
    ap.add_argument("--bytes-file", type=Path, help="raw HTTP bytes from a file")
    ap.add_argument("--label", default="unlabeled", help="classification label for --bytes-file")
    ap.add_argument("--backend", action="append", default=[],
                    help="NAME=cmd with {stdin} fed; may be repeated. "
                         "Example: llhttp=./parse_llhttp")
    ap.add_argument("--json", action="store_true", help="emit JSON")
    args = ap.parse_args(argv)

    if args.list_classes:
        for c in ALL_CLASSES:
            print(c)
        return 0
    if args.list_cases:
        for c in builtin_cases().values():
            print(f"{c.name:32s} [{c.classification}] {c.notes[:60]}")
        return 0

    extra = []
    for item in args.backend:
        if "=" not in item:
            print(f"bad --backend {item!r}, expected NAME=cmd", file=sys.stderr)
            return 2
        name, cmd = item.split("=", 1)
        extra.append((name, cmd.split()))

    if args.variant:
        cases = builtin_cases()
        if args.variant not in cases:
            print(f"unknown variant {args.variant!r}", file=sys.stderr)
            return 2
        report = run_case(cases[args.variant], extra)
    elif args.bytes_file:
        data = args.bytes_file.read_bytes()
        case = Case(args.bytes_file.name, args.label, data)
        report = run_case(case, extra)
    else:
        # default: run all built-ins
        reports = [run_case(c, extra) for c in builtin_cases().values()]
        if args.json:
            print(json.dumps(reports, indent=2))
        else:
            for report in reports:
                _print_report(report)
        return 0

    if args.json:
        print(json.dumps(report, indent=2))
    else:
        _print_report(report)
    return 0


def _print_report(report: dict):
    print("=" * 60)
    print(f"case           : {report['case']}")
    print(f"classification : {report['classification']}")
    print(f"bytes          : {report['bytes']}")
    if report.get("notes"):
        print(f"notes          : {report['notes']}")
    print("-" * 60)
    for r in report["results"]:
        print(f"[{r['backend']}] consumed={r['consumed']} messages={r['messages']} "
              f"incomplete={r['incomplete']} error={r['error']}")
        if r.get("observations"):
            for k, v in r["observations"].items():
                print(f"    {k}: {v}")
    cmp_ = report["comparison"]
    print("-" * 60)
    print(f"verdict        : {cmp_['verdict']}")
    print(f"msg counts     : {cmp_['message_counts']}")
    print(f"incomplete     : {cmp_['incomplete_flags']}")
    print(f"note           : {cmp_['note']}")
    print()


if __name__ == "__main__":
    sys.exit(main())