pure_backends.py

#!/usr/bin/env python3
"""
Pure-Python framing backends for offline differential observation.

These model documented / source-derived policies. They are not the real C
libraries. Use them when a C toolchain is unavailable, and replace with real
CLI backends when you can compile.

Backends:
  reference          — structural observer (existing)
  tinyproxy-1.11.3   — vulnerable CL-wins body policy (CVE-2026-54387 class)
  tinyproxy-fixed    — fixed policy: CL+TE prefers chunked / strips CL
  pico-headers       — header-only parse (like phr_parse_request scope)

Output JSON for framing_diff --backend subprocess mode, or import run().
"""

from __future__ import annotations

import json
import sys
from dataclasses import asdict, dataclass
from typing import Optional


@dataclass
class Result:
    backend: str
    consumed: int
    messages: int
    incomplete: bool
    error: Optional[str]
    observations: dict


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


def _headers(header_blob: bytes) -> list[tuple[str, str]]:
    out = []
    for line in header_blob.split(b"\r\n")[1:]:
        if b":" not in line:
            continue
        k, v = line.split(b":", 1)
        out.append((k.decode("latin1").strip().lower(), v.decode("latin1").strip()))
    return out


def _cl_values(hdrs):
    vals = []
    for k, v in hdrs:
        if k == "content-length":
            try:
                vals.append(int(v))
            except ValueError:
                vals.append(None)
    return vals


def _te_chunked(hdrs):
    for k, v in hdrs:
        if k == "transfer-encoding" and "chunked" in v.lower():
            return True
    return False


def reference(data: bytes) -> Result:
    from framing_diff import reference_parse
    r = reference_parse(data)
    return Result(r.backend, r.consumed, r.messages, r.incomplete, r.error, r.observations)


def tinyproxy_vuln_1113(data: bytes) -> Result:
    """
    Source: tinyproxy 1.11.3 src/reqs.c process_client_headers

      content_length.client = get_content_length(...)
      if content_length.client == -1 && is_chunked:
          content_length.client = -2
      # else if CL present, TE is ignored for body pull
      remaining headers (including both CL and TE) are forwarded

    Body pull:
      if client > 0: pull exactly CL bytes
      elif client == -2: pull chunked
    """
    header_blob, body, hdr_end = _split(data)
    obs = {"policy": "tinyproxy-1.11.3-cl-wins", "cve_class": "CVE-2026-54387"}
    if header_blob is None:
        return Result("tinyproxy-1.11.3", 0, 0, True, "no header terminator", obs)
    hdrs = _headers(header_blob)
    cls = _cl_values(hdrs)
    te = _te_chunked(hdrs)
    obs["content_length_values"] = cls
    obs["transfer_encoding_chunked"] = te
    obs["forwards_both_cl_and_te"] = bool(cls) and te

    body_mode = "none"
    body_need = 0
    if cls:
        # first CL wins in pseudomap_find style (single map entry); use last
        # inserted in real code — map usually keeps one; model first non-None
        cl = next((c for c in cls if c is not None), None)
        if cl is not None and cl > 0:
            body_mode = "content-length"
            body_need = cl
        elif te:
            body_mode = "chunked"
    elif te:
        body_mode = "chunked"

    obs["body_read_mode"] = body_mode
    obs["body_bytes_read"] = body_need if body_mode == "content-length" else None

    if body_mode == "content-length":
        if body is None or len(body) < body_need:
            return Result(
                "tinyproxy-1.11.3",
                len(data),
                0,
                True,
                None,
                {**obs, "note": "waiting for CL bytes"},
            )
        # Proxy consumes exactly CL bytes from the client and forwards them.
        cl_slice = body[:body_need]
        client_leftover = body[body_need:]
        obs["client_leftover_after_cl"] = len(client_leftover)
        obs["backend_would_see_te"] = te
        # Desync vs a TE-honoring backend: early chunk terminator inside the
        # CL-sized slice leaves a second request for the backend to parse.
        early = cl_slice.find(b"0\r\n\r\n")
        obs["early_chunk_end_offset"] = early if early >= 0 else None
        obs["backend_leftover_if_te"] = (
            len(cl_slice) - (early + 5) if early >= 0 else 0
        )
        obs["desync_precondition"] = bool(
            te and early >= 0 and (early + 5) < len(cl_slice)
        )
        obs["impact_requires"] = (
            "real backend that honors TE while proxy consumed CL; lab only"
        )
        return Result("tinyproxy-1.11.3", hdr_end + body_need, 1, False, None, obs)

    if body_mode == "chunked":
        # minimal: need terminal 0\r\n\r\n in body
        if body is None or b"0\r\n\r\n" not in body:
            return Result("tinyproxy-1.11.3", len(data), 0, True, None, obs)
        end = body.find(b"0\r\n\r\n") + 5
        return Result("tinyproxy-1.11.3", hdr_end + end, 1, False, None, obs)

    return Result("tinyproxy-1.11.3", hdr_end, 1, False, None, obs)


def tinyproxy_fixed(data: bytes) -> Result:
    """
    Fixed policy from PR #610 / commit ff45d3b class:
    when both CL and TE present, do not let CL alone drive the body while
    still advertising TE to the backend. Model: prefer chunked body mode and
    strip CL from forwarded headers (no dual advertisement).
    """
    header_blob, body, hdr_end = _split(data)
    obs = {"policy": "tinyproxy-fixed-prefer-chunked-or-reject-dual"}
    if header_blob is None:
        return Result("tinyproxy-fixed", 0, 0, True, "no header terminator", obs)
    hdrs = _headers(header_blob)
    cls = _cl_values(hdrs)
    te = _te_chunked(hdrs)
    obs["content_length_values"] = cls
    obs["transfer_encoding_chunked"] = te

    if cls and te:
        obs["forwards_both_cl_and_te"] = False
        obs["strips_content_length"] = True
        obs["body_read_mode"] = "chunked"
        if body is None or b"0\r\n\r\n" not in body:
            # incomplete chunked
            return Result("tinyproxy-fixed", len(data), 0, True, None, obs)
        end = body.find(b"0\r\n\r\n") + 5
        leftover = body[end:]
        obs["client_leftover_after_chunked"] = len(leftover)
        obs["desync_precondition"] = False
        return Result("tinyproxy-fixed", hdr_end + end, 1, False, None, obs)

    if len(cls) > 1 and len(set([c for c in cls if c is not None])) > 1:
        obs["rejects_duplicate_cl"] = True
        return Result(
            "tinyproxy-fixed",
            0,
            0,
            False,
            "duplicate content-length",
            obs,
        )

    # single CL or single TE path: same as non-dual handling
    base = tinyproxy_vuln_1113(data)
    merged = {**obs, **base.observations}
    merged["policy"] = "tinyproxy-fixed-single-path"
    return Result(
        "tinyproxy-fixed",
        base.consumed,
        base.messages,
        base.incomplete,
        base.error,
        merged,
    )


def pico_headers(data: bytes) -> Result:
    """
    Models picohttpparser phr_parse_request: headers only; body not consumed.
    Return value style: >=0 bytes through end of headers, -2 partial, -1 error.
    """
    header_blob, body, hdr_end = _split(data)
    obs = {"scope": "headers-only", "library_model": "picohttpparser"}
    if header_blob is None:
        return Result("pico-headers", 0, 0, True, "partial", obs)
    # very small request-line check
    first = header_blob.split(b"\r\n", 1)[0]
    parts = first.split(b" ")
    if len(parts) < 3 or not parts[2].startswith(b"HTTP/"):
        return Result("pico-headers", 0, 0, False, "parse error", obs)
    obs["method"] = parts[0].decode("latin1", "replace")
    obs["path"] = parts[1].decode("latin1", "replace")
    hdrs = _headers(header_blob)
    obs["header_count"] = len(hdrs)
    obs["body_not_parsed"] = True
    obs["consumed_through"] = "end of headers"
    return Result("pico-headers", hdr_end, 1, False, None, obs)


BACKENDS = {
    "reference": reference,
    "tinyproxy-1.11.3": tinyproxy_vuln_1113,
    "tinyproxy-fixed": tinyproxy_fixed,
    "pico-headers": pico_headers,
}


def run(name: str, data: bytes) -> Result:
    if name not in BACKENDS:
        raise SystemExit(f"unknown backend {name}; choose from {list(BACKENDS)}")
    return BACKENDS[name](data)


def main():
    # subprocess mode for framing_diff: NAME as argv[1], bytes on stdin
    if len(sys.argv) < 2:
        print("usage: pure_backends.py BACKEND < input.http", file=sys.stderr)
        print("backends:", ", ".join(BACKENDS), file=sys.stderr)
        return 2
    name = sys.argv[1]
    data = sys.stdin.buffer.read()
    r = run(name, data)
    print(json.dumps({
        "consumed": r.consumed,
        "messages": r.messages,
        "incomplete": r.incomplete,
        "error": r.error,
        "observations": r.observations,
    }))
    return 0


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