DibaGuard open release: licence-clean network models, runtime, docs (no HTTP model)

e02228b verified 4 days ago

16.8 kB

"""Every HTTP input form -> one Request: raw requests (CSIC-style text), access-log lines (nginx/apache combined),
bare URLs. Parameters are extracted from the query, a form, JSON or multipart body and cookies; the model text is
built from the canonical form of each part (canon.py).
The text is what every tier judges (playbook signatures, the A22 SQL-context tier, the HTTP model), and the Go fast
path builds the very same text (go/dibaguard/canon.go, proven by the canon_parity.json fixture of
tests/test_parity.py) - so the Python-side defences hold in a Go deployment and in its sidecar too (V-08).
Nothing is dropped from judgement silently (V-05, V-06, V-07): every occurrence of a duplicate JSON key, every
multipart field, every value of a repeated header, every field and the whole of a value are written into the text
while a judging budget lasts (MAX_TEXT characters); past it, only the parts that can carry a payload are written
(_keep, up to EXTRA_TEXT more), and whatever still does not fit is counted in an `x:oversized=<chars>` part, which the
SQL-context tier answers with a soft alarm (sqlctx.analyse) - a request too large to judge is never simply allowed.
The work stays linear in the request size and bounded: at most MAX_FIELDS fields are canonicalised one by one (the
rest as one blob), a value is decoded up to canon.MAX_LEN characters, the text never exceeds MAX_TEXT + EXTRA_TEXT."""
from __future__ import annotations

import json
import re
from dataclasses import dataclass, field
from urllib.parse import urlsplit

from dibaguard.http.canon import MAX_LEN, canon_views

COMBINED = re.compile(r'^(?P<ip>\S+) \S+ \S+ \[(?P<time>[^\]]*)\] "(?P<req>[^"]*)" (?P<status>\d{3}|-) (?P<size>\d+|-)'
                      r'(?: "(?P<ref>[^"]*)" "(?P<ua>[^"]*)")?')
KEEP_HEADERS = ("host", "user-agent", "referer", "cookie", "content-type", "x-forwarded-for", "accept", "authorization",
                "x-requested-with", "origin")
# judging budget (characters of canonical text; every limit below is mirrored in go/dibaguard/canon.go)
MAX_PART = 8192              # a value longer than this that does not fit the budget whole is cut into overlapping chunks
CHUNK_OVERLAP = 512          # a payload up to this long is whole in one of two neighbouring chunks
MAX_TEXT = 32768             # everything is written while the text is shorter than this
EXTRA_TEXT = 32768           # then only parts that can carry a payload, up to this much more
MAX_FIELDS = 2000            # fields canonicalised one by one; the rest are judged as one blob (part "x:rest")
MAX_DEPTH = 512              # JSON nesting deeper than this is judged as one raw value
MAX_FILE = MAX_PART          # an uploaded file's content is data: its head is judged (a script / web shell shows there)
REST_MAX = 1 << 20           # raw characters past a decoding limit (MAX_LEN per value, MAX_FIELDS) canonicalised per request
_STD_HEADERS = ("user-agent", "referer", "x-forwarded-for", "content-type")
_NEVER_HEADERS = {"cookie", "host", "content-length", "connection", "accept-encoding", "cache-control", "pragma", "upgrade-insecure-requests"}
# word boundaries are written as explicit ASCII classes (no \b): Python's \b is Unicode-aware and RE2's is not, and the
# Go fast path must include exactly the same headers
_HDR_SUSPICIOUS = re.compile(r"['\"<>`]|\$\{|\{\{|--|/\*|\.\./|(?:^|[^a-z0-9_])(?:union|select|sleep|benchmark|jndi|script|javascript|vbscript|document|onerror|eval|exec"
                             r"|cmd|wget|curl|waitfor|extractvalue|updatexml|information_schema|passwd|whoami)(?:$|[^a-z0-9_])")
# what a part needs to be able to carry a payload any tier knows: a character other than letters, digits, space, '.',
# '_' and '-', or "--" / "..", or one of these words (ASCII tokens, the same in Go)
_PLAIN = re.compile(r"[\w .-]*")
_TOKEN = re.compile(r"[a-z0-9_]+")
_KEEP_WORDS = frozenset(("or", "and", "xor", "not", "union", "select", "sleep", "benchmark", "waitfor", "having", "exec", "execute",
                         "declare", "drop", "insert", "update", "delete", "truncate", "shutdown", "grant", "information_schema",
                         "pg_sleep", "extractvalue", "updatexml", "dual", "concat", "char", "chr", "rlike", "regexp", "like", "null",
                         "jndi", "script", "javascript", "passwd", "whoami", "cmd", "eval", "system"))
_BOUNDARY = re.compile(r'boundary=(?:"([^"]+)"|([^";,\s]+))', re.I)
_DISP_NAME = re.compile(r'(?:^|;)\s*name\s*=\s*(?:"([^"]*)"|([^;\s]*))', re.I)
_DISP_FILE = re.compile(r'(?:^|;)\s*filename\s*=\s*(?:"([^"]*)"|([^;\s]*))', re.I)


def _keep(s: str) -> bool:
    """can this canonical text carry a payload? (past the plain budget only such parts are written)"""
    if "--" in s or ".." in s or not _PLAIN.fullmatch(s):
        return True
    return any(t in _KEEP_WORDS for t in _TOKEN.findall(s))


def _join_header(name: str, values) -> str:
    """every value of a repeated header, joined the way HTTP folds them (cookies with '; '): a stack that reads the
    last value or joins them gets what is judged here (V-07)"""
    if isinstance(values, (list, tuple)):
        return ("; " if name == "cookie" else ", ").join(str(v) for v in values)
    return "" if values is None else str(values)


@dataclass
class Request:
    method: str = "GET"
    path: str = "/"
    query: str = ""
    headers: dict = field(default_factory=dict)   # name -> value, or name -> [values] of a repeated header
    body: str = ""
    meta: dict = field(default_factory=dict)   # status, size, ip, time from logs (never model text)

    def __post_init__(self):
        h: dict = {}
        for k, v in (self.headers or {}).items():
            k = str(k).lower()
            v = _join_header(k, v)
            h[k] = (h[k] + ("; " if k == "cookie" else ", ") + v) if k in h else v
        self.headers = h

    def params(self) -> list[tuple[str, str, str]]:
        """[(where, name, value)] from the query, the body (form / JSON / multipart) and cookies - all of them"""
        out = [("q", k, v) for k, v in _pairs(self.query)]
        ct = self.headers.get("content-type", "").lower()
        b = self.body.strip()
        if b:
            if "json" in ct or b[:1] in "[{":
                try:
                    out += [("b", k, v) for k, v in _flatten(json.loads(b, object_pairs_hook=_Obj, parse_constant=_no_constant,
                                                                      parse_int=_int))]
                except (ValueError, RecursionError):
                    out.append(("b", "", b))
            elif "multipart/" in ct:
                mp = _multipart(b, self.headers.get("content-type", ""))     # the boundary keeps its case
                out += [("b", "", b)] if mp is None else [("b", k, v) for k, v in mp]
            elif "=" in b:
                out += [("b", k, v) for k, v in _pairs(b)]
            else:
                out.append(("b", "", b))
        ck = self.headers.get("cookie", "")
        if ck:
            out += [("c", k.strip(), v) for k, v in (p.partition("=")[::2] for p in ck.split(";") if p.strip())]
        return out

    def model_text(self) -> tuple[str, int]:
        """Canonical text the models read, and the deepest encoding layer peeled anywhere in the request."""
        t = _Text()
        p, pa, r = canon_views(self.path)
        t.deepest = r
        t.add(f"{self.method.lower()} ", p)
        if pa is not None:
            t.add("p:=", pa)
        ps = self.params()
        for where, k, v in ps[:MAX_FIELDS]:
            cv, va, r = canon_views(v, plus_is_space=(where != "c"))
            ck, ka, _ = canon_views(k)
            t.deepest = max(t.deepest, r)
            t.add(f"{where}:{ck}=", cv)
            if va is not None:                     # the value with its comments kept (canon.views)
                t.add(f"{where}:{ck}=", va)
            if ka is not None:
                t.add(f"{where}:{ka}=", cv)
            if len(v) > MAX_LEN:                   # decoded up to MAX_LEN: the rest is judged as raw text, never dropped
                t.add_raw(f"{where}:{ck}=", v[MAX_LEN - CHUNK_OVERLAP:])
        if len(ps) > MAX_FIELDS:                   # fields past MAX_FIELDS: one blob, canonicalised once
            t.add_raw("x:rest=", "&".join(f"{k}={v}" for _, k, v in ps[MAX_FIELDS:]))
        for h in _STD_HEADERS:
            if self.headers.get(h):
                ch, ha, r = canon_views(self.headers[h])
                t.deepest = max(t.deepest, r)
                t.add(f"h:{h}=", ch)
                if ha is not None:
                    t.add(f"h:{h}=", ha)
        # any other header joins the text only when its value looks like a payload: a normal Accept-Language or
        # Authorization leaves the model's input exactly as it was trained, a payload hidden in a header is scored.
        # Sorted by name: the Go fast path sees the headers as a map, and both must write the same text
        for h in sorted(self.headers):
            v = self.headers[h]
            if h in _STD_HEADERS or h in _NEVER_HEADERS or not v:
                continue
            ch, ha, r = canon_views(v)
            if _HDR_SUSPICIOUS.search(ch) or (ha is not None and _HDR_SUSPICIOUS.search(ha)):
                t.deepest = max(t.deepest, r)
                t.add(f"h:{h}=", ch)
                if ha is not None:
                    t.add(f"h:{h}=", ha)
        return t.text(), t.deepest


class _Text:
    """the parts of a model text under the judging budget (see the module docstring)"""

    def __init__(self):
        self.parts: list[str] = []
        self.n = 0                 # characters written within MAX_TEXT (separators included)
        self.extra = 0             # characters written past it (parts that can carry a payload)
        self.over = 0              # characters of such parts that fit neither, and raw past REST_MAX: x:oversized
        self.raw_left = REST_MAX
        self.deepest = 0

    def add(self, prefix: str, value: str) -> None:
        if self.n + len(prefix) + len(value) + 3 <= MAX_TEXT:
            self.parts.append(prefix + value)
            self.n += len(prefix) + len(value) + 3
            return
        for c in _chunks(value):
            k = len(prefix) + len(c) + 3
            if self.n + k <= MAX_TEXT:
                self.parts.append(prefix + c)
                self.n += k
            elif _keep(c):
                if self.extra + k <= EXTRA_TEXT:
                    self.parts.append(prefix + c)
                    self.extra += k
                else:
                    self.over += len(c)

    def add_raw(self, prefix: str, raw: str) -> None:
        """raw text past a decoding limit: canonicalised in MAX_LEN pieces (overlapping) and written like a value; past
        REST_MAX raw characters per request it is only counted (x:oversized)"""
        if len(raw) > self.raw_left:
            self.over += len(raw) - self.raw_left
            raw = raw[:self.raw_left]
        self.raw_left -= len(raw)
        if not raw:
            return
        step = MAX_LEN - CHUNK_OVERLAP
        for i in range(0, max(1, len(raw) - CHUNK_OVERLAP), step):
            a, b, r = canon_views(raw[i:i + MAX_LEN])
            self.deepest = max(self.deepest, r)
            self.add(prefix, a)
            if b is not None:
                self.add(prefix, b)

    def text(self) -> str:
        if self.over:
            self.parts.append(f"x:oversized={self.over}")
        return " | ".join(self.parts)


def _chunks(v: str) -> list[str]:
    """overlapping MAX_PART pieces of a long value (a short one is its own only piece)"""
    if len(v) <= MAX_PART:
        return [v]
    step = MAX_PART - CHUNK_OVERLAP
    return [v[i:i + MAX_PART] for i in range(0, len(v) - CHUNK_OVERLAP, step)]


def _pairs(s: str) -> list[tuple[str, str]]:
    return [(k, v) for k, _, v in (p.partition("=") for p in s.split("&") if p)]


class _Obj(list):
    """a JSON object as the list of its (key, value) pairs in order: a duplicate key keeps every value (V-05) - parsers
    that give the application the first one (several streaming / Go parsers) or the last one (json.loads) both get a
    value that was judged"""


def _no_constant(name: str):
    raise ValueError(f"not JSON: {name}")      # NaN / Infinity are not JSON (and the Go decoder refuses them too)


def _int(s: str) -> int:
    if len(s) > 4300:                          # Python's own int-from-string limit, explicit so Go can refuse the same
        raise ValueError("integer too long")
    return int(s)


def _flatten(o, prefix: str = "", depth: int = 0) -> list[tuple[str, str]]:
    if depth > MAX_DEPTH:
        raise ValueError("JSON nested too deeply")
    if isinstance(o, _Obj):
        return [x for k, v in o for x in _flatten(v, f"{prefix}.{k}" if prefix else str(k), depth + 1)]
    if isinstance(o, dict):
        return [x for k, v in o.items() for x in _flatten(v, f"{prefix}.{k}" if prefix else str(k), depth + 1)]
    if isinstance(o, list):
        return [x for i, v in enumerate(o) for x in _flatten(v, f"{prefix}[{i}]", depth + 1)]
    return [(prefix, "" if o is None else str(o))]


def _multipart(body: str, ct: str) -> list[tuple[str, str]] | None:
    """multipart/form-data -> [(name, value)]: every field is a value of its own, a file part gives its filename and
    the head of its content (MAX_FILE characters: a file is data, stored rather than interpreted). None when the body is
    not multipart after all (judged as one raw value then). The same simple reading as the Go fast path."""
    m = _BOUNDARY.search(ct)
    if not m:
        return None
    delim = "--" + (m.group(1) or m.group(2))
    pieces = body.split(delim)
    if len(pieces) < 2:
        return None
    out = []
    for piece in pieces[1:]:
        if piece.startswith("--"):               # the closing delimiter: the epilogue is not delivered
            break
        if piece.startswith("\r\n"):
            piece = piece[2:]
        elif piece.startswith("\n"):
            piece = piece[1:]
        i, j = piece.find("\r\n\r\n"), piece.find("\n\n")
        if i < 0 and j < 0:
            out.append(("", piece))              # no header block: judged as it is
            continue
        cut, skip = (i, 4) if j < 0 or (0 <= i < j) else (j, 2)
        head, content = piece[:cut], piece[cut + skip:]
        if content.endswith("\r\n"):
            content = content[:-2]
        elif content.endswith("\n"):
            content = content[:-1]
        disp = ""
        for ln in head.split("\n"):
            hk, sep, hv = ln.partition(":")
            if sep and hk.strip().lower() == "content-disposition":
                disp = hv.strip()
        nm, fn = _DISP_NAME.search(disp), _DISP_FILE.search(disp)
        name = (nm.group(1) if nm.group(1) is not None else nm.group(2)) if nm else ""
        if fn:
            out.append((name, fn.group(1) if fn.group(1) is not None else fn.group(2)))
            out.append((name, content[:MAX_FILE]))
        else:
            out.append((name, content))
    return out


def from_url(url: str, method: str = "GET", headers: dict | None = None, body: str = "") -> Request:
    u = urlsplit(url if "://" in url or url.startswith("/") else "/" + url)
    h = {k.lower(): v for k, v in (headers or {}).items()}
    if u.netloc and "host" not in h:
        h["host"] = u.netloc
    return Request(method.upper(), u.path or "/", u.query, h, body or "")


def from_raw(text: str) -> Request:
    """'GET /a?b=1 HTTP/1.1\\nHeader: v\\n\\nbody' (CRLF or LF). A repeated header keeps every value."""
    text = text.replace("\r\n", "\n")
    head, _, body = text.partition("\n\n")
    lines = head.split("\n")
    first = lines[0].split(" ")
    method = first[0] if first else "GET"
    url = first[1] if len(first) > 1 else "/"
    headers: dict = {}
    for ln in lines[1:]:
        k, sep, v = ln.partition(":")
        if sep and k.strip().lower() in KEEP_HEADERS:
            headers.setdefault(k.strip().lower(), []).append(v.strip())
    return from_url(url, method, headers, body.strip("\n"))


def from_log_line(line: str) -> Request | None:
    """nginx/apache 'combined' (or 'common') format; None when the line is not one."""
    m = COMBINED.match(line.strip())
    if not m:
        return None
    req = m.group("req").split(" ")
    if len(req) < 2:
        return None
    h = {}
    if m.group("ua") and m.group("ua") != "-":
        h["user-agent"] = m.group("ua")
    if m.group("ref") and m.group("ref") != "-":
        h["referer"] = m.group("ref")
    r = from_url(req[1], req[0], h)
    r.meta = {"ip": m.group("ip"), "time": m.group("time"), "status": m.group("status"), "size": m.group("size")}
    return r

	"""Every HTTP input form -> one Request: raw requests (CSIC-style text), access-log lines (nginx/apache combined),
	bare URLs. Parameters are extracted from the query, a form, JSON or multipart body and cookies; the model text is
	built from the canonical form of each part (canon.py).

	The text is what every tier judges (playbook signatures, the A22 SQL-context tier, the HTTP model), and the Go fast
	path builds the very same text (go/dibaguard/canon.go, proven by the canon_parity.json fixture of
	tests/test_parity.py) - so the Python-side defences hold in a Go deployment and in its sidecar too (V-08).

	Nothing is dropped from judgement silently (V-05, V-06, V-07): every occurrence of a duplicate JSON key, every
	multipart field, every value of a repeated header, every field and the whole of a value are written into the text
	while a judging budget lasts (MAX_TEXT characters); past it, only the parts that can carry a payload are written
	(_keep, up to EXTRA_TEXT more), and whatever still does not fit is counted in an `x:oversized=<chars>` part, which the
	SQL-context tier answers with a soft alarm (sqlctx.analyse) - a request too large to judge is never simply allowed.
	The work stays linear in the request size and bounded: at most MAX_FIELDS fields are canonicalised one by one (the
	rest as one blob), a value is decoded up to canon.MAX_LEN characters, the text never exceeds MAX_TEXT + EXTRA_TEXT."""
	from __future__ import annotations

	import json
	import re
	from dataclasses import dataclass, field
	from urllib.parse import urlsplit

	from dibaguard.http.canon import MAX_LEN, canon_views

	COMBINED = re.compile(r'^(?P<ip>\S+) \S+ \S+ \[(?P<time>[^\]])\] "(?P<req>[^"])" (?P<status>\d{3}\|-) (?P<size>\d+\|-)'
	r'(?: "(?P<ref>[^"])" "(?P<ua>[^"])")?')
	KEEP_HEADERS = ("host", "user-agent", "referer", "cookie", "content-type", "x-forwarded-for", "accept", "authorization",
	"x-requested-with", "origin")
	# judging budget (characters of canonical text; every limit below is mirrored in go/dibaguard/canon.go)
	MAX_PART = 8192 # a value longer than this that does not fit the budget whole is cut into overlapping chunks
	CHUNK_OVERLAP = 512 # a payload up to this long is whole in one of two neighbouring chunks
	MAX_TEXT = 32768 # everything is written while the text is shorter than this
	EXTRA_TEXT = 32768 # then only parts that can carry a payload, up to this much more
	MAX_FIELDS = 2000 # fields canonicalised one by one; the rest are judged as one blob (part "x:rest")
	MAX_DEPTH = 512 # JSON nesting deeper than this is judged as one raw value
	MAX_FILE = MAX_PART # an uploaded file's content is data: its head is judged (a script / web shell shows there)
	REST_MAX = 1 << 20 # raw characters past a decoding limit (MAX_LEN per value, MAX_FIELDS) canonicalised per request
	_STD_HEADERS = ("user-agent", "referer", "x-forwarded-for", "content-type")
	_NEVER_HEADERS = {"cookie", "host", "content-length", "connection", "accept-encoding", "cache-control", "pragma", "upgrade-insecure-requests"}
	# word boundaries are written as explicit ASCII classes (no \b): Python's \b is Unicode-aware and RE2's is not, and the
	# Go fast path must include exactly the same headers
	_HDR_SUSPICIOUS = re.compile(r"['\"<>`]\|\$\{\|\{\{\|--\|/\*\|\.\./\|(?:^\|[^a-z0-9_])(?:union\|select\|sleep\|benchmark\|jndi\|script\|javascript\|vbscript\|document\|onerror\|eval\|exec"
	r"\|cmd\|wget\|curl\|waitfor\|extractvalue\|updatexml\|information_schema\|passwd\|whoami)(?:$\|[^a-z0-9_])")
	# what a part needs to be able to carry a payload any tier knows: a character other than letters, digits, space, '.',
	# '_' and '-', or "--" / "..", or one of these words (ASCII tokens, the same in Go)
	_PLAIN = re.compile(r"[\w .-]*")
	_TOKEN = re.compile(r"[a-z0-9_]+")
	_KEEP_WORDS = frozenset(("or", "and", "xor", "not", "union", "select", "sleep", "benchmark", "waitfor", "having", "exec", "execute",
	"declare", "drop", "insert", "update", "delete", "truncate", "shutdown", "grant", "information_schema",
	"pg_sleep", "extractvalue", "updatexml", "dual", "concat", "char", "chr", "rlike", "regexp", "like", "null",
	"jndi", "script", "javascript", "passwd", "whoami", "cmd", "eval", "system"))
	_BOUNDARY = re.compile(r'boundary=(?:"([^"]+)"\|([^";,\s]+))', re.I)
	_DISP_NAME = re.compile(r'(?:^\|;)\sname\s=\s(?:"([^"])"\|([^;\s]*))', re.I)
	_DISP_FILE = re.compile(r'(?:^\|;)\sfilename\s=\s(?:"([^"])"\|([^;\s]*))', re.I)


	def _keep(s: str) -> bool:
	"""can this canonical text carry a payload? (past the plain budget only such parts are written)"""
	if "--" in s or ".." in s or not _PLAIN.fullmatch(s):
	return True
	return any(t in _KEEP_WORDS for t in _TOKEN.findall(s))


	def _join_header(name: str, values) -> str:
	"""every value of a repeated header, joined the way HTTP folds them (cookies with '; '): a stack that reads the
	last value or joins them gets what is judged here (V-07)"""
	if isinstance(values, (list, tuple)):
	return ("; " if name == "cookie" else ", ").join(str(v) for v in values)
	return "" if values is None else str(values)


	@dataclass
	class Request:
	method: str = "GET"
	path: str = "/"
	query: str = ""
	headers: dict = field(default_factory=dict) # name -> value, or name -> [values] of a repeated header
	body: str = ""
	meta: dict = field(default_factory=dict) # status, size, ip, time from logs (never model text)

	def __post_init__(self):
	h: dict = {}
	for k, v in (self.headers or {}).items():
	k = str(k).lower()
	v = _join_header(k, v)
	h[k] = (h[k] + ("; " if k == "cookie" else ", ") + v) if k in h else v
	self.headers = h

	def params(self) -> list[tuple[str, str, str]]:
	"""[(where, name, value)] from the query, the body (form / JSON / multipart) and cookies - all of them"""
	out = [("q", k, v) for k, v in _pairs(self.query)]
	ct = self.headers.get("content-type", "").lower()
	b = self.body.strip()
	if b:
	if "json" in ct or b[:1] in "[{":
	try:
	out += [("b", k, v) for k, v in _flatten(json.loads(b, object_pairs_hook=_Obj, parse_constant=_no_constant,
	parse_int=_int))]
	except (ValueError, RecursionError):
	out.append(("b", "", b))
	elif "multipart/" in ct:
	mp = _multipart(b, self.headers.get("content-type", "")) # the boundary keeps its case
	out += [("b", "", b)] if mp is None else [("b", k, v) for k, v in mp]
	elif "=" in b:
	out += [("b", k, v) for k, v in _pairs(b)]
	else:
	out.append(("b", "", b))
	ck = self.headers.get("cookie", "")
	if ck:
	out += [("c", k.strip(), v) for k, v in (p.partition("=")[::2] for p in ck.split(";") if p.strip())]
	return out

	def model_text(self) -> tuple[str, int]:
	"""Canonical text the models read, and the deepest encoding layer peeled anywhere in the request."""
	t = _Text()
	p, pa, r = canon_views(self.path)
	t.deepest = r
	t.add(f"{self.method.lower()} ", p)
	if pa is not None:
	t.add("p:=", pa)
	ps = self.params()
	for where, k, v in ps[:MAX_FIELDS]:
	cv, va, r = canon_views(v, plus_is_space=(where != "c"))
	ck, ka, _ = canon_views(k)
	t.deepest = max(t.deepest, r)
	t.add(f"{where}:{ck}=", cv)
	if va is not None: # the value with its comments kept (canon.views)
	t.add(f"{where}:{ck}=", va)
	if ka is not None:
	t.add(f"{where}:{ka}=", cv)
	if len(v) > MAX_LEN: # decoded up to MAX_LEN: the rest is judged as raw text, never dropped
	t.add_raw(f"{where}:{ck}=", v[MAX_LEN - CHUNK_OVERLAP:])
	if len(ps) > MAX_FIELDS: # fields past MAX_FIELDS: one blob, canonicalised once
	t.add_raw("x:rest=", "&".join(f"{k}={v}" for _, k, v in ps[MAX_FIELDS:]))
	for h in _STD_HEADERS:
	if self.headers.get(h):
	ch, ha, r = canon_views(self.headers[h])
	t.deepest = max(t.deepest, r)
	t.add(f"h:{h}=", ch)
	if ha is not None:
	t.add(f"h:{h}=", ha)
	# any other header joins the text only when its value looks like a payload: a normal Accept-Language or
	# Authorization leaves the model's input exactly as it was trained, a payload hidden in a header is scored.
	# Sorted by name: the Go fast path sees the headers as a map, and both must write the same text
	for h in sorted(self.headers):
	v = self.headers[h]
	if h in _STD_HEADERS or h in _NEVER_HEADERS or not v:
	continue
	ch, ha, r = canon_views(v)
	if _HDR_SUSPICIOUS.search(ch) or (ha is not None and _HDR_SUSPICIOUS.search(ha)):
	t.deepest = max(t.deepest, r)
	t.add(f"h:{h}=", ch)
	if ha is not None:
	t.add(f"h:{h}=", ha)
	return t.text(), t.deepest


	class _Text:
	"""the parts of a model text under the judging budget (see the module docstring)"""

	def __init__(self):
	self.parts: list[str] = []
	self.n = 0 # characters written within MAX_TEXT (separators included)
	self.extra = 0 # characters written past it (parts that can carry a payload)
	self.over = 0 # characters of such parts that fit neither, and raw past REST_MAX: x:oversized
	self.raw_left = REST_MAX
	self.deepest = 0

	def add(self, prefix: str, value: str) -> None:
	if self.n + len(prefix) + len(value) + 3 <= MAX_TEXT:
	self.parts.append(prefix + value)
	self.n += len(prefix) + len(value) + 3
	return
	for c in _chunks(value):
	k = len(prefix) + len(c) + 3
	if self.n + k <= MAX_TEXT:
	self.parts.append(prefix + c)
	self.n += k
	elif _keep(c):
	if self.extra + k <= EXTRA_TEXT:
	self.parts.append(prefix + c)
	self.extra += k
	else:
	self.over += len(c)

	def add_raw(self, prefix: str, raw: str) -> None:
	"""raw text past a decoding limit: canonicalised in MAX_LEN pieces (overlapping) and written like a value; past
	REST_MAX raw characters per request it is only counted (x:oversized)"""
	if len(raw) > self.raw_left:
	self.over += len(raw) - self.raw_left
	raw = raw[:self.raw_left]
	self.raw_left -= len(raw)
	if not raw:
	return
	step = MAX_LEN - CHUNK_OVERLAP
	for i in range(0, max(1, len(raw) - CHUNK_OVERLAP), step):
	a, b, r = canon_views(raw[i:i + MAX_LEN])
	self.deepest = max(self.deepest, r)
	self.add(prefix, a)
	if b is not None:
	self.add(prefix, b)

	def text(self) -> str:
	if self.over:
	self.parts.append(f"x:oversized={self.over}")
	return " \| ".join(self.parts)


	def _chunks(v: str) -> list[str]:
	"""overlapping MAX_PART pieces of a long value (a short one is its own only piece)"""
	if len(v) <= MAX_PART:
	return [v]
	step = MAX_PART - CHUNK_OVERLAP
	return [v[i:i + MAX_PART] for i in range(0, len(v) - CHUNK_OVERLAP, step)]


	def _pairs(s: str) -> list[tuple[str, str]]:
	return [(k, v) for k, _, v in (p.partition("=") for p in s.split("&") if p)]


	class _Obj(list):
	"""a JSON object as the list of its (key, value) pairs in order: a duplicate key keeps every value (V-05) - parsers
	that give the application the first one (several streaming / Go parsers) or the last one (json.loads) both get a
	value that was judged"""


	def _no_constant(name: str):
	raise ValueError(f"not JSON: {name}") # NaN / Infinity are not JSON (and the Go decoder refuses them too)


	def _int(s: str) -> int:
	if len(s) > 4300: # Python's own int-from-string limit, explicit so Go can refuse the same
	raise ValueError("integer too long")
	return int(s)


	def _flatten(o, prefix: str = "", depth: int = 0) -> list[tuple[str, str]]:
	if depth > MAX_DEPTH:
	raise ValueError("JSON nested too deeply")
	if isinstance(o, _Obj):
	return [x for k, v in o for x in _flatten(v, f"{prefix}.{k}" if prefix else str(k), depth + 1)]
	if isinstance(o, dict):
	return [x for k, v in o.items() for x in _flatten(v, f"{prefix}.{k}" if prefix else str(k), depth + 1)]
	if isinstance(o, list):
	return [x for i, v in enumerate(o) for x in _flatten(v, f"{prefix}[{i}]", depth + 1)]
	return [(prefix, "" if o is None else str(o))]


	def _multipart(body: str, ct: str) -> list[tuple[str, str]] \| None:
	"""multipart/form-data -> [(name, value)]: every field is a value of its own, a file part gives its filename and
	the head of its content (MAX_FILE characters: a file is data, stored rather than interpreted). None when the body is
	not multipart after all (judged as one raw value then). The same simple reading as the Go fast path."""
	m = _BOUNDARY.search(ct)
	if not m:
	return None
	delim = "--" + (m.group(1) or m.group(2))
	pieces = body.split(delim)
	if len(pieces) < 2:
	return None
	out = []
	for piece in pieces[1:]:
	if piece.startswith("--"): # the closing delimiter: the epilogue is not delivered
	break
	if piece.startswith("\r\n"):
	piece = piece[2:]
	elif piece.startswith("\n"):
	piece = piece[1:]
	i, j = piece.find("\r\n\r\n"), piece.find("\n\n")
	if i < 0 and j < 0:
	out.append(("", piece)) # no header block: judged as it is
	continue
	cut, skip = (i, 4) if j < 0 or (0 <= i < j) else (j, 2)
	head, content = piece[:cut], piece[cut + skip:]
	if content.endswith("\r\n"):
	content = content[:-2]
	elif content.endswith("\n"):
	content = content[:-1]
	disp = ""
	for ln in head.split("\n"):
	hk, sep, hv = ln.partition(":")
	if sep and hk.strip().lower() == "content-disposition":
	disp = hv.strip()
	nm, fn = _DISP_NAME.search(disp), _DISP_FILE.search(disp)
	name = (nm.group(1) if nm.group(1) is not None else nm.group(2)) if nm else ""
	if fn:
	out.append((name, fn.group(1) if fn.group(1) is not None else fn.group(2)))
	out.append((name, content[:MAX_FILE]))
	else:
	out.append((name, content))
	return out


	def from_url(url: str, method: str = "GET", headers: dict \| None = None, body: str = "") -> Request:
	u = urlsplit(url if "://" in url or url.startswith("/") else "/" + url)
	h = {k.lower(): v for k, v in (headers or {}).items()}
	if u.netloc and "host" not in h:
	h["host"] = u.netloc
	return Request(method.upper(), u.path or "/", u.query, h, body or "")


	def from_raw(text: str) -> Request:
	"""'GET /a?b=1 HTTP/1.1\\nHeader: v\\n\\nbody' (CRLF or LF). A repeated header keeps every value."""
	text = text.replace("\r\n", "\n")
	head, _, body = text.partition("\n\n")
	lines = head.split("\n")
	first = lines[0].split(" ")
	method = first[0] if first else "GET"
	url = first[1] if len(first) > 1 else "/"
	headers: dict = {}
	for ln in lines[1:]:
	k, sep, v = ln.partition(":")
	if sep and k.strip().lower() in KEEP_HEADERS:
	headers.setdefault(k.strip().lower(), []).append(v.strip())
	return from_url(url, method, headers, body.strip("\n"))


	def from_log_line(line: str) -> Request \| None:
	"""nginx/apache 'combined' (or 'common') format; None when the line is not one."""
	m = COMBINED.match(line.strip())
	if not m:
	return None
	req = m.group("req").split(" ")
	if len(req) < 2:
	return None
	h = {}
	if m.group("ua") and m.group("ua") != "-":
	h["user-agent"] = m.group("ua")
	if m.group("ref") and m.group("ref") != "-":
	h["referer"] = m.group("ref")
	r = from_url(req[1], req[0], h)
	r.meta = {"ip": m.group("ip"), "time": m.group("time"), "status": m.group("status"), "size": m.group("size")}
	return r