Commander1024
00f226dc59
Implement strict origin, cookie, and login JSON parsing with fail-closed validation and output wiping. Add focused host contract tests and document the preparatory 8D.3 parser split.
174 lines
10 KiB
Python
174 lines
10 KiB
Python
#!/usr/bin/env python3
|
|
"""Dependency-free contract tests against the production parser, not a model."""
|
|
import ctypes as C
|
|
import json
|
|
from pathlib import Path
|
|
import subprocess
|
|
import tempfile
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
|
|
|
|
class Credentials(C.Structure):
|
|
_fields_ = [("username_length", C.c_size_t), ("password_length", C.c_size_t),
|
|
("username", C.c_uint8 * 17), ("password", C.c_uint8 * 65)]
|
|
|
|
|
|
def span(value):
|
|
# No implicit NUL terminator; keep the allocation alive throughout the call.
|
|
return None if value is None else (C.c_char * max(1, len(value))).from_buffer_copy(value or b"\0")
|
|
|
|
|
|
def size(value):
|
|
return 0 if value is None else len(value)
|
|
|
|
|
|
def login(username=b"u", password=b"p"):
|
|
return b'{"username":"' + username + b'","password":"' + password + b'"}'
|
|
|
|
|
|
def main():
|
|
with tempfile.TemporaryDirectory(prefix="web-auth-parse-") as temporary:
|
|
library = Path(temporary) / "parser.so"
|
|
subprocess.run(["cc", "-std=c11", "-Wall", "-Wextra", "-Werror",
|
|
"-shared", "-fPIC", "-I", str(ROOT / "src"),
|
|
str(ROOT / "src/web_auth_parse.c"), "-o", str(library)], check=True)
|
|
api = C.CDLL(str(library))
|
|
api.web_auth_parse_origin.argtypes = [C.c_void_p, C.c_size_t, C.c_void_p, C.c_size_t, C.c_void_p]
|
|
api.web_auth_parse_cookie.argtypes = [C.c_void_p, C.c_size_t, C.c_char_p, C.c_void_p]
|
|
api.web_auth_parse_login.argtypes = [C.c_void_p, C.c_size_t, C.POINTER(Credentials)]
|
|
for name in ("origin", "cookie", "login"):
|
|
getattr(api, "web_auth_parse_" + name).restype = C.c_bool
|
|
failures, count = [], 0
|
|
|
|
def check(kind, label, invoke, output, expected, extract):
|
|
nonlocal count
|
|
count += 1
|
|
C.memset(C.addressof(output), 0xA5, C.sizeof(output))
|
|
result = invoke()
|
|
raw = C.string_at(C.addressof(output), C.sizeof(output))
|
|
if result != (expected is not None):
|
|
failures.append(f"{kind}: {label}: unexpected success={result}")
|
|
if not result and raw != bytes(len(raw)):
|
|
failures.append(f"{kind}: {label}: failure did not wipe every output byte")
|
|
if result and expected is not None and extract(output) != expected:
|
|
failures.append(f"{kind}: {label}: incorrect decoded/canonical output")
|
|
|
|
origins = [(b"EXAMPLE.Com", b"https://example.com", b"https://example.com"),
|
|
(b"a-b.local", b"https://A-B.LOCAL", b"https://a-b.local"),
|
|
(b"localhost", b"https://localhost", b"https://localhost"),
|
|
(b"192.168.1.1", b"https://192.168.1.1", b"https://192.168.1.1")]
|
|
for host in (b"EXAMPLE.COM", b"192.168.1.1"):
|
|
for hp in (b"", b":443"):
|
|
for op in (b"", b":443"):
|
|
origins.append((host + hp, b"https://" + host.lower() + op,
|
|
b"https://" + host.lower()))
|
|
bad_hosts = [b"", b" ", b"example.com ", b" example.com", b"a..b", b".a", b"a.",
|
|
b"-a", b"a-", b"a_b", b"a/b", b"a?b", b"a#b", b"u@a", b"a,b",
|
|
b"a\\b", b"a\tb", b"a\r\nb", b"a\0b", b"caf\xc3\xa9", b"[::1]", b"::1",
|
|
b"[::1]:443", b"a:80", b"a:444", b"a:", b"a:0443", b"a:+443",
|
|
b"a:443:443", b"a" * 64 + b".com", b"a" * 130]
|
|
origins += [(h, b"https://" + h, None) for h in bad_hosts]
|
|
origins += [(b"example.com", o, None) for o in
|
|
(None, b"", b"null", b"http://example.com", b"https://other.com",
|
|
b"https://example.com/", b"https://example.com/path", b"https://example.com?x",
|
|
b"https://example.com#x", b"https://u@example.com", b"https://example.com:80",
|
|
b"https://example.com:0443", b" https://example.com", b"https://example.com ",
|
|
b"https://example.com\0", b"https://example.com\r\n", b"https://[::1]",
|
|
b"https://example.com https://example.com", b"https:///example.com")]
|
|
origins += [(None, b"https://example.com", None)]
|
|
# 128 canonical bytes fit with the terminator; 129 do not.
|
|
for n in (56, 57):
|
|
host = b"a" * 63 + b"." + b"b" * n
|
|
for hp in (b"", b":443"):
|
|
for op in (b"", b":443"):
|
|
origins.append((host + hp, b"https://" + host + op,
|
|
b"https://" + host if n == 56 else None))
|
|
for index, (host, origin, expected) in enumerate(origins):
|
|
h, o, out = span(host), span(origin), C.create_string_buffer(129)
|
|
check("origin", str(index), lambda: api.web_auth_parse_origin(h, size(host), o, size(origin), out),
|
|
out, expected, lambda x: x.value)
|
|
|
|
token = b"0123456789abcdef" * 4
|
|
selected = b"sid=" + token
|
|
cookies = [(selected, token), (b"x=1; " + selected + b"; y=two", token),
|
|
(b"sid2=other; " + selected, token), (b"SID=other; " + selected, token),
|
|
(b"empty=; " + selected, token)]
|
|
cookies += [(h, None) for h in
|
|
(None, b"", b"x=1", b"SID=" + token, b"sid2=" + token,
|
|
selected + b"; " + selected, selected + b"; sid=bad", b"sid=bad; " + selected,
|
|
b"sid=" + token[:-1], selected + b"0", b"sid=" + token.upper(),
|
|
b"sid=" + b"g" * 64, b'sid="' + token + b'"', b"sid=", b"sid",
|
|
selected + b"; broken", b"broken; " + selected, selected + b"; =x",
|
|
selected + b"; bad name=x", selected + b"; x=bad,value",
|
|
selected + b"; x=bad\\value", selected + b"\r\n", selected + b"\0",
|
|
selected + b"; x=\x01", selected + b"; x=\x7f", selected + b"; x=\xff")]
|
|
for length in (1024, 1025):
|
|
cookies.append((selected + b"; x=" + b"a" * (length - len(selected) - 4),
|
|
token if length == 1024 else None))
|
|
for index, (header, expected) in enumerate(cookies):
|
|
h, out = span(header), C.create_string_buffer(65)
|
|
check("cookie", str(index), lambda: api.web_auth_parse_cookie(h, size(header), b"sid", out),
|
|
out, expected, lambda x: x.value)
|
|
|
|
cases = [(login(), (b"u", b"p")), (login(b"", b""), (b"", b"")),
|
|
(b' \r\n\t{ "password" : "p", "username" : "u" } \t', (b"u", b"p")),
|
|
(b'{"user\\u006eame":"u","pass\\u0077ord":"p"}', (b"u", b"p")),
|
|
(login(b'\\"\\\\\\/\\b\\f\\n\\r\\t', b"\\u0041"), (b'"\\/\b\f\n\r\t', b"A")),
|
|
(login(b"\\u00e9", b"\\ud83d\\ude00"), ("é".encode(), "😀".encode()))]
|
|
for text in ("é", "€", "😀", "\U0010ffff"):
|
|
for ascii_only in (True, False):
|
|
body = json.dumps({"username": text, "password": text}, ensure_ascii=ascii_only).encode()
|
|
cases.append((body, (text.encode(), text.encode())))
|
|
for n in (15, 16, 17):
|
|
cases.append((login(b"a" * n), (b"a" * n, b"p") if n <= 16 else None))
|
|
cases.append((login(b"\\u0061" * n), (b"a" * n, b"p") if n <= 16 else None))
|
|
for n in (63, 64, 65):
|
|
cases.append((login(password=b"a" * n), (b"u", b"a" * n) if n <= 64 else None))
|
|
cases.append((login(password=b"\\u0061" * n), (b"u", b"a" * n) if n <= 64 else None))
|
|
for field, limit in (("username", 16), ("password", 64)):
|
|
for extra in (b"", b"a"):
|
|
value = "😀".encode() * (limit // 4) + extra
|
|
u, p = (value, b"p") if field == "username" else (b"u", value)
|
|
cases.append((login(u, p), (u, p) if not extra else None))
|
|
for n in (511, 512, 513):
|
|
cases.append((login() + b" " * (n - len(login())), (b"u", b"p") if n <= 512 else None))
|
|
bad_json = [None, b"", b"{}", b"[]", b"null", b'{"username":"u"}', b'{"password":"p"}',
|
|
login() + b"x", login() + login(), b"\xef\xbb\xbf" + login(),
|
|
login()[:-1] + b',}', login()[:-1] + b',"extra":"x"}',
|
|
login()[:-1] + b',"username":"v"}', login()[:-1] + b',"password":"q"}',
|
|
login()[:-1] + b',"user\\u006eame":"v"}',
|
|
b'{"username":1,"password":"p"}', b'{"username":"u","password":null}',
|
|
b'{"username":[],"password":"p"}', b'{"username":"u" "password":"p"}',
|
|
b"{'username':'u','password':'p'}", b'/*x*/' + login()]
|
|
for value in (b"\0", b"\\u0000", b"\\x41", b"\\q", b"\\u123", b"\\uZZZZ",
|
|
b"\\ud800", b"\\udc00", b"\\ud800\\u0041", b"\\udc00\\ud800",
|
|
b"\x80", b"\xc0\xaf", b"\xc1\xbf", b"\xc2", b"\xe2\x82", b"\xf0\x9f\x98",
|
|
b"\xe0\x80\x80", b"\xed\xa0\x80", b"\xf0\x80\x80\x80",
|
|
b"\xf4\x90\x80\x80", b"\xf5\x80\x80\x80", b"\xff", b"\xc2A"):
|
|
bad_json.extend((login(value), login(password=value)))
|
|
bad_json += [login(bytes([n])) for n in range(1, 32)]
|
|
# Every truncated prefix of a valid document must fail, including late failures.
|
|
bad_json += [login()[:n] for n in range(len(login()))]
|
|
bad_json += [login() + b"\0", b'{"user\xffname":"u","password":"p"}']
|
|
cases += [(body, None) for body in bad_json]
|
|
|
|
def decoded(out):
|
|
u, p = out.username_length, out.password_length
|
|
if u > 16 or p > 64 or out.username[u] or out.password[p]:
|
|
return "invalid lengths or missing terminator"
|
|
return bytes(out.username[:u]), bytes(out.password[:p])
|
|
|
|
for index, (body, expected) in enumerate(cases):
|
|
b, out = span(body), Credentials()
|
|
check("login", str(index), lambda: api.web_auth_parse_login(b, size(body), C.byref(out)),
|
|
out, expected, decoded)
|
|
for failure in failures:
|
|
print("FAIL:", failure)
|
|
print(f"{count} cases; {len(failures)} failures")
|
|
return bool(failures)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|