#!/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())