"""Synthetic protocol fixtures only: these tests never open a network socket."""
import socket
import struct
import threading
import unittest
from unittest.mock import patch

from rpcscope import longlogon


ADDRESS = "192.0.2.10"
GUID = bytes(range(16))


def tlv(tag, value):
    size = len(value)
    length = bytes([size]) if size < 128 else b"\x81" + bytes([size]) if size < 256 else b"\x82" + size.to_bytes(2, "big")
    return bytes([tag]) + length + value


def dns_name(name):
    return b"".join(bytes([len(label.encode())]) + label.encode() for label in name.split(".") if label) + b"\0"


def identity(domain="lab.example.test", forest="example.test", host=None, nb_domain="LAB", nb_host="DC1", guid=GUID):
    return longlogon._Identity(forest, domain, host or "dc1." + domain, nb_domain.casefold(), nb_host.casefold(), guid)


def ex_blob(value=None):
    value = value or identity()
    names = (value.forest, value.domain, value.host, value.netbios_domain,
             value.netbios_host, "x", "Default-First-Site-Name", "")
    return (struct.pack("<HHI", 25, 0, 0x18) + value.guid
            + b"".join(dns_name(name) for name in names)
            + struct.pack("<IHH", 5, 0xFFFF, 0xFFFF))


def legacy_blob(value=None):
    value = value or identity()
    strings = (value.netbios_host, "x", value.netbios_domain)
    return (struct.pack("<H", 21)
            + b"".join(name.encode("utf-16-le") + b"\0\0" for name in strings)
            + value.guid + bytes(16)
            + b"".join(dns_name(name) for name in (value.forest, value.domain, value.host))
            + socket.inet_aton(ADDRESS) + struct.pack("<IIHH", 0x10, 3, 0xFFFF, 0xFFFF))


def ldap(blob, message_id, *, done=True):
    attribute = tlv(0x30, tlv(4, b"Netlogon") + tlv(0x31, tlv(4, blob)))
    entry = tlv(0x30, tlv(2, bytes([message_id])) + tlv(0x64, tlv(4, b"") + tlv(0x30, attribute)))
    if done:
        entry += tlv(0x30, tlv(2, bytes([message_id])) + tlv(0x65, tlv(10, b"\0") + tlv(4, b"") + tlv(4, b"")))
    return entry


def request_id(request):
    outer = longlogon._Reader(request)
    inner = longlogon._Reader(outer.take(0x30))
    return int.from_bytes(inner.take(2), "big")


class FakeSocket:
    def __init__(self, responses):
        self.responses = list(responses)
        self.sent = []
        self.peer = None
        self.closed = False

    def __enter__(self):
        return self

    def __exit__(self, *args):
        self.closed = True

    def connect(self, peer):
        self.peer = peer

    def settimeout(self, timeout):
        self.timeout = timeout

    def send(self, request):
        self.sent.append(request)
        return len(request)

    def recv(self, maximum):
        if not self.responses:
            raise OSError("no UDP reply")
        response = self.responses.pop(0)
        if isinstance(response, Exception):
            raise response
        if callable(response):
            return response(self.sent[-1])
        return response


def response(blob):
    return lambda request: ldap(blob, request_id(request))


class LongLogonTests(unittest.TestCase):
    def probe(self, responses, **kwargs):
        fake = FakeSocket(responses)
        with patch.object(longlogon.socket, "socket", return_value=fake) as factory:
            result = longlogon.probe_longlogon(ADDRESS, **kwargs)
        self.assertTrue(fake.closed)
        self.assertEqual(fake.peer, (ADDRESS, 389))
        self.assertEqual(factory.call_args.args, (socket.AF_INET, socket.SOCK_DGRAM))
        return result, fake

    def test_valid_pair_is_precondition_only_and_requests_are_fixed_small(self):
        result, fake = self.probe([response(ex_blob()), response(legacy_blob())])
        self.assertEqual(result.status, "precondition-not-met")
        self.assertIn("patch state is not determined", result.evidence)
        self.assertIn("projected full-user length=", result.evidence)
        self.assertEqual(len(fake.sent), 2)
        self.assertNotEqual(request_id(fake.sent[0]), request_id(fake.sent[1]))
        self.assertLess(max(map(len, fake.sent)), 256)
        self.assertNotIn(b"DnsDomain", fake.sent[0])
        self.assertIn(b"DnsDomain", fake.sent[1])
        self.assertIn(b"lab.example.test", fake.sent[1])
        for request, version in zip(fake.sent, (6, 2)):
            self.assertIn(tlv(4, b"User") + tlv(4, b"x"), request)
            self.assertIn(tlv(4, b"NtVer") + tlv(4, struct.pack("<I", version)), request)
            self.assertNotIn(b"x" * 2, request)

    def test_long_identity_can_produce_positive_precondition_without_long_user(self):
        value = identity(domain="a" * 60 + ".example.test")
        result, fake = self.probe([response(ex_blob(value)), response(legacy_blob(value))])
        self.assertEqual(result.status, "precondition-met")
        self.assertIn("empirical", result.evidence)
        self.assertTrue(all(b"x" * 2 not in request for request in fake.sent))

    def test_exact_empirical_thresholds(self):
        for estimate, expected in ((527, "precondition-not-met"), (528, "precondition-borderline"),
                                   (535, "precondition-borderline"), (536, "precondition-met")):
            with self.subTest(estimate=estimate):
                result = longlogon._result("lab.test", 180, estimate - 258, 400)
                self.assertEqual(result.status, expected)
                self.assertIn(f"length={estimate} B", result.evidence)

    def test_known_domain_is_normalized_and_must_match_response(self):
        result, fake = self.probe([response(ex_blob()), response(legacy_blob())], domain="LAB.EXAMPLE.TEST.")
        self.assertEqual(result.status, "precondition-not-met")
        self.assertIn(b"lab.example.test", fake.sent[0])
        result, fake = self.probe([response(ex_blob())], domain="other.example.test")
        self.assertEqual(result.status, "inconclusive")
        self.assertIn("does not match", result.evidence)
        self.assertEqual(len(fake.sent), 1)

    def test_unsafe_baseline_withholds_legacy_query(self):
        domain = ".".join(("a" * 60, "b" * 60, "c" * 60, "test"))
        value = identity(domain=domain, forest=domain)
        result, fake = self.probe([response(ex_blob(value))])
        self.assertEqual(result.status, "inconclusive")
        self.assertIn("legacy query withheld", result.evidence)
        self.assertIn("upper bound=", result.evidence)
        self.assertEqual(len(fake.sent), 1)

    def test_no_control_response_and_ex_only_are_unknown_not_negative(self):
        result, fake = self.probe([OSError("connection refused")])
        self.assertEqual(result.status, "inconclusive")
        self.assertEqual(len(fake.sent), 1)
        result, fake = self.probe([response(ex_blob()), OSError("no UDP reply")])
        self.assertEqual(result.status, "inconclusive")
        self.assertIn("EX=", result.evidence)
        self.assertIn("legacy=no valid response", result.evidence)
        self.assertEqual(len(fake.sent), 2)

    def test_socket_timeouts_are_bounded_and_unknown(self):
        fake = FakeSocket([socket.timeout()])
        with patch.object(longlogon.socket, "socket", return_value=fake), \
                patch.object(longlogon.time, "monotonic", side_effect=(0.0, 0.01, 1.0)):
            result = longlogon.probe_longlogon(ADDRESS, timeout=0.5)
        self.assertEqual(result.status, "inconclusive")
        self.assertIn("timed out", result.evidence)
        self.assertEqual(len(fake.sent), 1)

    def test_invalid_control_packets_never_send_legacy(self):
        cases = [b"", b"x" * 8193, b"\x30\x80", b"\x30\x82\xff\xff", b"\x30\x04\x02\x01\x01\x64"]
        for raw in cases:
            with self.subTest(packet=raw[:8]):
                result, fake = self.probe([raw])
                self.assertEqual(result.status, "inconclusive")
                self.assertEqual(len(fake.sent), 1)

    def test_wrong_transaction_id_and_error_only_response_are_unknown(self):
        bad_id = lambda request: ldap(ex_blob(), request_id(request) % 127 + 1)
        error = lambda request: tlv(0x30, tlv(2, bytes([request_id(request)])) + tlv(0x65, tlv(10, b"\x20") + tlv(4, b"") + tlv(4, b"no object")))
        for response_value in (bad_id, error):
            result, fake = self.probe([response_value])
            self.assertEqual(result.status, "inconclusive")
            self.assertEqual(len(fake.sent), 1)

    def test_unsupported_control_version_and_opcode_are_unknown(self):
        original = ex_blob()
        cases = [struct.pack("<H", 24) + original[2:], original[:-8] + struct.pack("<IHH", 3, 0xFFFF, 0xFFFF),
                 original[:2] + b"\x01\0" + original[4:], original[:-1] + b"\0"]
        for blob in cases:
            result, fake = self.probe([response(blob)])
            self.assertEqual(result.status, "inconclusive")
            self.assertEqual(len(fake.sent), 1)

    def test_legacy_identity_and_variant_must_match_control(self):
        other = identity(guid=b"1" * 16)
        old = legacy_blob()
        for blob in (legacy_blob(other), old[:-8] + struct.pack("<IHH", 1, 0xFFFF, 0xFFFF), ex_blob()):
            result, fake = self.probe([response(ex_blob()), response(blob)])
            self.assertEqual(result.status, "inconclusive")
            self.assertEqual(len(fake.sent), 2)

    def test_ldap_requires_single_netlogon_value_and_success_when_done_present(self):
        original = ldap(ex_blob(), 1)
        for data in (original.replace(b"Netlogon", b"Anything"), original + original,
                     original[:-7] + b"\x01" + original[-6:]):
            with self.assertRaises(longlogon._Invalid):
                longlogon._ldap_netlogon(data, 1)
        self.assertEqual(longlogon._ldap_netlogon(ldap(ex_blob(), 1, done=False), 1), ex_blob())

    def test_compressed_names_and_cycle_detection(self):
        data = b"\x03lab\x04test\0\x02dc\xc0\0"
        self.assertEqual(longlogon._name(data, 10, len(data)), ("dc.lab.test", len(data)))
        for data, offset in ((b"\xc0\0", 0), (b"\xc0\x02\0", 0), (b"\x40x\0", 0), (b"\x03ab", 0)):
            with self.assertRaises(longlogon._Invalid):
                longlogon._name(data, offset, len(data))

    def test_invalid_target_domain_or_timeout_opens_no_socket(self):
        for address, kwargs in (("dc.example.test", {}), (ADDRESS, {"domain": "bad)(*"}),
                                (ADDRESS, {"timeout": float("inf")}), (ADDRESS, {"timeout": 0})):
            with patch.object(longlogon.socket, "socket") as factory:
                self.assertEqual(longlogon.probe_longlogon(address, **kwargs).status, "inconclusive")
            factory.assert_not_called()

    def test_cancel_before_start_sends_nothing(self):
        cancel = threading.Event()
        cancel.set()
        with patch.object(longlogon.socket, "socket") as factory:
            result = longlogon.probe_longlogon(ADDRESS, cancel=cancel)
        self.assertEqual(result.status, "skipped-cancelled")
        factory.assert_not_called()

    def test_cancel_after_control_withholds_legacy(self):
        cancel = threading.Event()

        def control(request):
            cancel.set()
            return ldap(ex_blob(), request_id(request))

        result, fake = self.probe([control], cancel=cancel)
        self.assertEqual(result.status, "skipped-cancelled")
        self.assertEqual(len(fake.sent), 1)

    def test_ipv6_connects_only_to_given_literal_target(self):
        fake = FakeSocket([response(ex_blob()), response(legacy_blob())])
        with patch.object(longlogon.socket, "socket", return_value=fake) as factory:
            result = longlogon.probe_longlogon("2001:db8::10")
        self.assertEqual(result.status, "precondition-not-met")
        self.assertEqual(factory.call_args.args[0], socket.AF_INET6)
        self.assertEqual(fake.peer, ("2001:db8::10", 389, 0, 0))


if __name__ == "__main__":
    unittest.main()