diff --git a/decnet/engine/deployer.py b/decnet/engine/deployer.py
index 4482024e..20a62765 100644
--- a/decnet/engine/deployer.py
+++ b/decnet/engine/deployer.py
@@ -55,6 +55,8 @@ _CANONICAL_SESSREC_DIR = Path(__file__).parent.parent / "templates" / "_shared"
 _SESSREC_SERVICES = {"ssh", "telnet"}
 _CANONICAL_AUTH_HELPER_DIR = Path(__file__).parent.parent / "templates" / "_shared" / "auth-helper"
 _AUTH_HELPER_SERVICES = {"ssh", "telnet"}
+_CANONICAL_NTLMSSP = Path(__file__).parent.parent / "templates" / "_shared" / "ntlmssp.py"
+_NTLMSSP_SERVICES = {"smb", "rdp"}
 
 
 def _sync_logging_helper(config: DecnetConfig) -> None:
@@ -108,6 +110,32 @@ def _sync_auth_helper_sources(config: DecnetConfig) -> None:
                     shutil.copy2(src, dest)
 
 
+def _sync_ntlmssp_sources(config: DecnetConfig) -> None:
+    """Copy _shared/ntlmssp.py into SMB/RDP build contexts.
+
+    Both templates parse NTLMSSP Type 3 messages (SMB Session Setup,
+    RDP NLA CredSSP); the canonical parser lives at
+    ``templates/_shared/ntlmssp.py`` and is mirrored into each active
+    build context here, mirroring the auth-helper / sessrec patterns.
+    """
+    from decnet.services.registry import get_service
+    seen: set[Path] = set()
+    for decky in config.deckies:
+        for svc_name in decky.services:
+            if svc_name not in _NTLMSSP_SERVICES:
+                continue
+            svc = get_service(svc_name)
+            if svc is None:
+                continue
+            ctx = svc.dockerfile_context()
+            if ctx is None or ctx in seen:
+                continue
+            seen.add(ctx)
+            dest = ctx / _CANONICAL_NTLMSSP.name
+            if not dest.exists() or dest.read_bytes() != _CANONICAL_NTLMSSP.read_bytes():
+                shutil.copy2(_CANONICAL_NTLMSSP, dest)
+
+
 def _sync_sessrec_sources(config: DecnetConfig) -> None:
     """Copy sessrec.c + Makefile into SSH/Telnet build contexts as sessrec/."""
     from decnet.services.registry import get_service
@@ -437,6 +465,7 @@ def deploy(config: DecnetConfig, dry_run: bool = False, no_cache: bool = False,
     _sync_logging_helper(config)
     _sync_sessrec_sources(config)
     _sync_auth_helper_sources(config)
+    _sync_ntlmssp_sources(config)
 
     compose_path = write_compose(config, COMPOSE_FILE)
     console.print(f"[bold cyan]Compose file written[/] → {compose_path}")
diff --git a/decnet/templates/smb/Dockerfile b/decnet/templates/smb/Dockerfile
index 64120be7..d676620e 100644
--- a/decnet/templates/smb/Dockerfile
+++ b/decnet/templates/smb/Dockerfile
@@ -2,13 +2,11 @@ ARG BASE_IMAGE=debian:bookworm-slim
 FROM ${BASE_IMAGE}
 
 RUN apt-get update && apt-get install -y --no-install-recommends \
-    python3 python3-pip \
+    python3 \
     && rm -rf /var/lib/apt/lists/*
 
-ENV PIP_BREAK_SYSTEM_PACKAGES=1
-RUN pip3 install --no-cache-dir impacket jinja2
-
 COPY syslog_bridge.py /opt/syslog_bridge.py
+COPY ntlmssp.py /opt/ntlmssp.py
 COPY server.py /opt/server.py
 COPY entrypoint.sh /entrypoint.sh
 RUN chmod +x /entrypoint.sh
diff --git a/decnet/templates/smb/entrypoint.sh b/decnet/templates/smb/entrypoint.sh
index 89e48290..c830b733 100644
--- a/decnet/templates/smb/entrypoint.sh
+++ b/decnet/templates/smb/entrypoint.sh
@@ -1,4 +1,3 @@
 #!/bin/bash
 set -e
-mkdir -p /tmp/smb_share
 exec python3 /opt/server.py
diff --git a/decnet/templates/smb/server.py b/decnet/templates/smb/server.py
index 24356a85..22c8ec9f 100644
--- a/decnet/templates/smb/server.py
+++ b/decnet/templates/smb/server.py
@@ -1,19 +1,57 @@
 #!/usr/bin/env python3
-"""
-Minimal SMB server using Impacket's SimpleSMBServer.
-Logs all connection attempts, optionally forwarding them as JSON to LOG_TARGET.
+"""Minimal honeypot SMB2 server.
+
+Hand-rolled asyncio framer that does just enough of MS-SMB2 to lure a
+client through Negotiate → Session Setup (Type1) → Session Setup
+(Type3), at which point we extract the inner NTLMSSP Type 3 with the
+shared :func:`ntlmssp.parse_type3` parser and emit a credential SD
+block. Authentication always fails with STATUS_LOGON_FAILURE — the
+attacker's hash lands in the Credential table; the attacker does not
+land on the host.
+
+References:
+- MS-SMB2 §2.2.3 NEGOTIATE Request, §2.2.4 NEGOTIATE Response
+- MS-SMB2 §2.2.5 SESSION_SETUP Request, §2.2.6 SESSION_SETUP Response
+- MS-NLMP §2.2.1 NTLMSSP messages (CHALLENGE_MESSAGE Type 2)
+- RFC 1002 §4.3 NetBIOS Session Service framing
 """
 
+from __future__ import annotations
+
+import asyncio
 import os
+import struct
 
-from impacket import smbserver
+from ntlmssp import find_ntlmssp, parse_type3
 from syslog_bridge import syslog_line, write_syslog_file, forward_syslog
 
 NODE_NAME = os.environ.get("NODE_NAME", "WORKSTATION")
-SERVICE_NAME   = "smb"
+SERVICE_NAME = "smb"
 LOG_TARGET = os.environ.get("LOG_TARGET", "")
 
+LISTEN_HOST = "0.0.0.0"  # nosec B104 — honeypot binds all interfaces by design
+LISTEN_PORT = 445
 
+# SMB2 status codes
+STATUS_SUCCESS = 0x00000000
+STATUS_MORE_PROCESSING_REQUIRED = 0xC0000016
+STATUS_LOGON_FAILURE = 0xC000006D
+
+# SMB2 commands
+SMB2_NEGOTIATE = 0x0000
+SMB2_SESSION_SETUP = 0x0001
+
+SMB2_MAGIC = b"\xfeSMB"
+NBSS_SESSION_MESSAGE = 0x00
+
+# Server's fixed 8-byte NTLM challenge (random-looking; honeypot, not crypto)
+SERVER_CHALLENGE = b"\x11\x22\x33\x44\x55\x66\x77\x88"
+# Stable server GUID — 16 zero bytes is fine for a honeypot.
+SERVER_GUID = b"\x00" * 16
+
+# Read caps; an attacker shouldn't be able to make us allocate
+# unbounded memory just by lying about NetBIOS frame length.
+MAX_NBSS_LEN = 1 * 1024 * 1024  # 1 MiB is plenty for SessionSetup blobs
 
 
 def _log(event_type: str, severity: int = 6, **kwargs) -> None:
@@ -22,15 +60,233 @@ def _log(event_type: str, severity: int = 6, **kwargs) -> None:
     forward_syslog(line, LOG_TARGET)
 
 
-if __name__ == "__main__":
-    _log("startup", msg=f"SMB server starting as {NODE_NAME}")
-    os.makedirs("/tmp/smb_share", exist_ok=True)  # nosec B108
+# ── SPNEGO / NTLMSSP Type 2 builder ──────────────────────────────────────────
 
-    server = smbserver.SimpleSMBServer(listenAddress="0.0.0.0", listenPort=445)  # nosec B104
-    server.setSMB2Support(True)
-    server.setSMBChallenge("")
-    server.addShare("SHARE", "/tmp/smb_share", "Shared Documents")  # nosec B108
+
+def _build_ntlmssp_type2(challenge: bytes) -> bytes:
+    """Build a minimal NTLMSSP CHALLENGE_MESSAGE (MS-NLMP §2.2.1.2).
+
+    Layout (all little-endian):
+      0   "NTLMSSP\\0"          8 bytes
+      8   MessageType=2         u32
+     12   TargetNameFields      8 bytes (Len, MaxLen, Offset)
+     20   NegotiateFlags        u32
+     24   ServerChallenge       8 bytes
+     32   Reserved              8 bytes
+     40   TargetInfoFields      8 bytes
+     48   Version               8 bytes
+     56   Payload               TargetName + TargetInfo
+
+    We advertise NEGOTIATE_UNICODE | NEGOTIATE_NTLM | NEGOTIATE_TARGET_INFO
+    (0x00828201) which is what real Windows servers send in practice; the
+    attacker's client uses these flags to decide whether to send Unicode
+    field strings in its Type 3 — the parser handles either.
+    """
+    target = "WORKGROUP".encode("utf-16-le")
+    # AV pair list: NetBIOS computer name + EOL terminator
+    av_name = "WORKGROUP".encode("utf-16-le")
+    target_info = struct.pack("<HH", 1, len(av_name)) + av_name + struct.pack("<HH", 0, 0)
+
+    flags = 0x00828201  # UNICODE | NTLM | TARGET_INFO | always_sign
+    payload = target + target_info
+    target_off = 56
+    info_off = target_off + len(target)
+
+    return (
+        b"NTLMSSP\x00"
+        + struct.pack("<I", 2)  # Type 2
+        + struct.pack("<HHI", len(target), len(target), target_off)
+        + struct.pack("<I", flags)
+        + challenge
+        + b"\x00" * 8  # reserved
+        + struct.pack("<HHI", len(target_info), len(target_info), info_off)
+        + b"\x00" * 8  # version
+        + payload
+    )
+
+
+def _wrap_spnego_type2(ntlm_type2: bytes) -> bytes:
+    """SPNEGO NegTokenResp DER carrying the NTLMSSP Type 2 blob.
+
+    Real Windows wraps Type 2 in an SPNEGO NegTokenResp (RFC 4178). A
+    well-formed wrapping is rarely required by attacker tools (Hydra,
+    Metasploit's smb_login, Impacket scanners all accept a raw Type 2
+    too) — but we ship the SPNEGO envelope so that finicky clients
+    don't bail out before sending Type 3, which is what we actually
+    want on the wire. The DER below hand-encodes a single
+    ``NegTokenResp`` with negState=accept-incomplete, supportedMech =
+    NTLMSSP OID, and responseToken = ntlm_type2.
+    """
+    # NTLMSSP OID = 1.3.6.1.4.1.311.2.2.10 → DER bytes
+    ntlmssp_oid = bytes.fromhex("06 0a 2b 06 01 04 01 82 37 02 02 0a".replace(" ", ""))
+    # negState [0] enum 1 (accept-incomplete)
+    neg_state = bytes.fromhex("a0 03 0a 01 01".replace(" ", ""))
+    # supportedMech [1] OID
+    supported = b"\xa1" + _der_len(len(ntlmssp_oid)) + ntlmssp_oid
+    # responseToken [2] OCTET STRING
+    rt_inner = b"\x04" + _der_len(len(ntlm_type2)) + ntlm_type2
+    response_token = b"\xa2" + _der_len(len(rt_inner)) + rt_inner
+    inner = neg_state + supported + response_token
+    neg_token_resp = b"\x30" + _der_len(len(inner)) + inner  # SEQUENCE
+    # NegTokenResp is itself tagged [1] in the outer choice
+    return b"\xa1" + _der_len(len(neg_token_resp)) + neg_token_resp
+
+
+def _der_len(n: int) -> bytes:
+    if n < 0x80:
+        return bytes([n])
+    body = n.to_bytes((n.bit_length() + 7) // 8, "big")
+    return bytes([0x80 | len(body)]) + body
+
+
+# ── SMB2 PDU helpers ─────────────────────────────────────────────────────────
+
+
+def _smb2_header(command: int, status: int, message_id: int, session_id: int = 0) -> bytes:
+    """SMB2 sync header (64 bytes), MS-SMB2 §2.2.1."""
+    return (
+        SMB2_MAGIC                          # ProtocolId
+        + struct.pack("<H", 64)             # StructureSize
+        + struct.pack("<H", 0)              # CreditCharge
+        + struct.pack("<I", status)         # Status
+        + struct.pack("<H", command)        # Command
+        + struct.pack("<H", 1)              # CreditResponse
+        + struct.pack("<I", 0x00000001)     # Flags = SERVER_TO_REDIR
+        + struct.pack("<I", 0)              # NextCommand
+        + struct.pack("<Q", message_id)     # MessageId
+        + struct.pack("<I", 0)              # Reserved (sync)
+        + struct.pack("<I", 0)              # TreeId
+        + struct.pack("<Q", session_id)     # SessionId
+        + b"\x00" * 16                      # Signature
+    )
+
+
+def _negotiate_response(message_id: int) -> bytes:
+    """SMB2 NEGOTIATE response (MS-SMB2 §2.2.4) — dialect 0x0210 (SMB 2.1)."""
+    body = (
+        struct.pack("<H", 65)               # StructureSize
+        + struct.pack("<H", 0)              # SecurityMode
+        + struct.pack("<H", 0x0210)         # DialectRevision = SMB 2.1
+        + struct.pack("<H", 0)              # Reserved
+        + SERVER_GUID
+        + struct.pack("<I", 0)              # Capabilities
+        + struct.pack("<I", 0x00010000)     # MaxTransactSize
+        + struct.pack("<I", 0x00010000)     # MaxReadSize
+        + struct.pack("<I", 0x00010000)     # MaxWriteSize
+        + struct.pack("<Q", 0)              # SystemTime
+        + struct.pack("<Q", 0)              # ServerStartTime
+        + struct.pack("<H", 128)            # SecurityBufferOffset (header64+body64)
+        + struct.pack("<H", 0)              # SecurityBufferLength
+        + struct.pack("<I", 0)              # Reserved2
+    )
+    return _smb2_header(SMB2_NEGOTIATE, STATUS_SUCCESS, message_id) + body
+
+
+def _session_setup_response(message_id: int, session_id: int, sec_blob: bytes, status: int) -> bytes:
+    """SMB2 SESSION_SETUP response (MS-SMB2 §2.2.6) carrying SPNEGO blob."""
+    body = (
+        struct.pack("<H", 9)                # StructureSize
+        + struct.pack("<H", 0)              # SessionFlags
+        + struct.pack("<H", 64 + 8)         # SecurityBufferOffset
+        + struct.pack("<H", len(sec_blob))  # SecurityBufferLength
+    )
+    return _smb2_header(SMB2_SESSION_SETUP, status, message_id, session_id) + body + sec_blob
+
+
+# ── Connection handler ───────────────────────────────────────────────────────
+
+
+async def _handle_client(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
+    peer = writer.get_extra_info("peername") or ("?", 0)
+    src_ip, src_port = peer[0], peer[1]
+    _log("connection", src_ip=src_ip, src_port=src_port)
+    session_id = 0x1000_0000_0000_0001
+    setup_round = 0
     try:
-        server.start()
+        while True:
+            # NetBIOS Session Service framing: 1 type byte + 3 length bytes
+            hdr = await reader.readexactly(4)
+            if hdr[0] != NBSS_SESSION_MESSAGE:
+                # Session Request / Keepalive / etc — quietly drop.
+                break
+            nb_len = int.from_bytes(hdr[1:4], "big")
+            if nb_len < 64 or nb_len > MAX_NBSS_LEN:
+                break
+            pdu = await reader.readexactly(nb_len)
+            if not pdu.startswith(SMB2_MAGIC):
+                # SMB1 Negotiate or other — not implemented; drop.
+                break
+            command = struct.unpack_from("<H", pdu, 12)[0]
+            message_id = struct.unpack_from("<Q", pdu, 24)[0]
+            if command == SMB2_NEGOTIATE:
+                resp = _negotiate_response(message_id)
+                _send_nbss(writer, resp)
+            elif command == SMB2_SESSION_SETUP:
+                setup_round += 1
+                # Body starts after 64-byte header; parse SecurityBufferOffset/Length
+                if len(pdu) < 64 + 24:
+                    break
+                sec_off = struct.unpack_from("<H", pdu, 64 + 12)[0]
+                sec_len = struct.unpack_from("<H", pdu, 64 + 14)[0]
+                blob = pdu[sec_off:sec_off + sec_len] if sec_len else b""
+                if setup_round == 1:
+                    # First Session Setup → respond with NTLMSSP Type 2
+                    type2 = _build_ntlmssp_type2(SERVER_CHALLENGE)
+                    spnego = _wrap_spnego_type2(type2)
+                    resp = _session_setup_response(
+                        message_id, session_id, spnego, STATUS_MORE_PROCESSING_REQUIRED
+                    )
+                    _send_nbss(writer, resp)
+                else:
+                    # Second Session Setup → contains NTLMSSP Type 3
+                    off = find_ntlmssp(blob)
+                    if off >= 0:
+                        cred = parse_type3(blob[off:])
+                        if cred:
+                            _log(
+                                "auth_attempt",
+                                src_ip=src_ip,
+                                src_port=src_port,
+                                **cred,
+                            )
+                    # Always fail authentication
+                    resp = _session_setup_response(
+                        message_id, session_id, b"", STATUS_LOGON_FAILURE
+                    )
+                    _send_nbss(writer, resp)
+                    break
+            else:
+                # We only implement Negotiate + SessionSetup; other commands
+                # could keep an attacker engaged longer but require state we
+                # don't carry. Disconnect.
+                break
+    except (asyncio.IncompleteReadError, ConnectionError):
+        pass
+    except Exception as exc:  # noqa: BLE001 — honeypot must never crash the worker
+        _log("error", severity=4, src_ip=src_ip, msg=str(exc))
+    finally:
+        try:
+            writer.close()
+            await writer.wait_closed()
+        except Exception:
+            pass
+        _log("disconnect", src_ip=src_ip, src_port=src_port)
+
+
+def _send_nbss(writer: asyncio.StreamWriter, smb_pdu: bytes) -> None:
+    nbss = bytes([NBSS_SESSION_MESSAGE]) + len(smb_pdu).to_bytes(3, "big")
+    writer.write(nbss + smb_pdu)
+
+
+async def _main() -> None:
+    _log("startup", msg=f"SMB server starting as {NODE_NAME}")
+    server = await asyncio.start_server(_handle_client, LISTEN_HOST, LISTEN_PORT)
+    async with server:
+        await server.serve_forever()
+
+
+if __name__ == "__main__":
+    try:
+        asyncio.run(_main())
     except KeyboardInterrupt:
         _log("shutdown")
diff --git a/tests/service_testing/test_smb_server.py b/tests/service_testing/test_smb_server.py
new file mode 100644
index 00000000..458b27c6
--- /dev/null
+++ b/tests/service_testing/test_smb_server.py
@@ -0,0 +1,268 @@
+"""Tests for decnet/templates/smb/server.py — hand-rolled SMB2 framer.
+
+Drives the asyncio handler with an in-memory StreamReader and a mocked
+StreamWriter. Exercises the full Negotiate → SessionSetup(Type1) →
+SessionSetup(Type3) flow and asserts that an NTLMSSP Type 3 lands in
+the universal credential SD shape.
+"""
+from __future__ import annotations
+
+import asyncio
+import importlib.util
+import struct
+import sys
+from unittest.mock import MagicMock
+
+import pytest
+
+from .conftest import load_real_instance_seed, make_fake_syslog_bridge
+
+
+# ── Module loader ─────────────────────────────────────────────────────────────
+
+
+def _load_real_ntlmssp():
+    spec = importlib.util.spec_from_file_location(
+        "ntlmssp", "decnet/templates/_shared/ntlmssp.py"
+    )
+    mod = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(mod)
+    return mod
+
+
+def _load_smb():
+    for key in ("smb_server", "syslog_bridge", "instance_seed", "ntlmssp"):
+        sys.modules.pop(key, None)
+    sys.modules["syslog_bridge"] = make_fake_syslog_bridge()
+    sys.modules["instance_seed"] = load_real_instance_seed()
+    sys.modules["ntlmssp"] = _load_real_ntlmssp()
+    spec = importlib.util.spec_from_file_location(
+        "smb_server", "decnet/templates/smb/server.py"
+    )
+    mod = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(mod)
+    return mod
+
+
+# ── Fixtures ──────────────────────────────────────────────────────────────────
+
+
+@pytest.fixture
+def smb_mod():
+    return _load_smb()
+
+
+def _make_streams():
+    """Return (reader, writer, written) — writer.write() collects bytes.
+
+    Must be called from inside a running event loop because
+    asyncio.StreamReader's __init__ needs one in Python 3.11.
+    """
+    reader = asyncio.StreamReader()
+    writer = MagicMock()
+    written: list[bytes] = []
+    writer.write.side_effect = written.append
+    writer.get_extra_info.return_value = ("198.51.100.7", 51234)
+
+    async def _wait_closed():
+        return None
+
+    writer.wait_closed = _wait_closed
+    return reader, writer, written
+
+
+# ── PDU builders ──────────────────────────────────────────────────────────────
+
+
+def _nbss(payload: bytes) -> bytes:
+    return bytes([0x00]) + len(payload).to_bytes(3, "big") + payload
+
+
+def _smb2_header(command: int, message_id: int, session_id: int = 0) -> bytes:
+    return (
+        b"\xfeSMB"
+        + struct.pack("<H", 64)
+        + struct.pack("<H", 0)
+        + struct.pack("<I", 0)
+        + struct.pack("<H", command)
+        + struct.pack("<H", 1)
+        + struct.pack("<I", 0)
+        + struct.pack("<I", 0)
+        + struct.pack("<Q", message_id)
+        + struct.pack("<I", 0)
+        + struct.pack("<I", 0)
+        + struct.pack("<Q", session_id)
+        + b"\x00" * 16
+    )
+
+
+def _negotiate_request() -> bytes:
+    # SMB2 NEGOTIATE Request (MS-SMB2 §2.2.3) — minimal, 1 dialect
+    body = (
+        struct.pack("<H", 36)             # StructureSize
+        + struct.pack("<H", 1)            # DialectCount
+        + struct.pack("<H", 0)            # SecurityMode
+        + struct.pack("<H", 0)            # Reserved
+        + struct.pack("<I", 0)            # Capabilities
+        + b"\x00" * 16                    # ClientGuid
+        + struct.pack("<Q", 0)            # ClientStartTime
+        + struct.pack("<H", 0x0210)       # Dialect = SMB 2.1
+        + struct.pack("<H", 0)            # padding
+    )
+    return _smb2_header(0x0000, 0) + body
+
+
+def _session_setup_request(message_id: int, sec_blob: bytes) -> bytes:
+    body = (
+        struct.pack("<H", 25)             # StructureSize
+        + struct.pack("<B", 0)            # Flags
+        + struct.pack("<B", 0)            # SecurityMode
+        + struct.pack("<I", 0)            # Capabilities
+        + struct.pack("<I", 0)            # Channel
+        + struct.pack("<H", 64 + 24)      # SecurityBufferOffset
+        + struct.pack("<H", len(sec_blob))
+        + struct.pack("<Q", 0)            # PreviousSessionId
+    )
+    return _smb2_header(0x0001, message_id) + body + sec_blob
+
+
+def _ntlmssp_type1() -> bytes:
+    return b"NTLMSSP\x00" + struct.pack("<I", 1) + struct.pack("<I", 0xE2088297) + b"\x00" * 24
+
+
+def _ntlmssp_type3(username: str, domain: str, nt_response: bytes) -> bytes:
+    """Build a minimal valid NTLMSSP Type 3 with NEGOTIATE_UNICODE."""
+    user_b = username.encode("utf-16-le")
+    dom_b = domain.encode("utf-16-le")
+    workstation = b""
+    payload = nt_response + dom_b + user_b + workstation
+
+    # 64-byte header + 8-byte version
+    nt_off = 72
+    dom_off = nt_off + len(nt_response)
+    user_off = dom_off + len(dom_b)
+    ws_off = user_off + len(user_b)
+    flags = 0x00000001  # NEGOTIATE_UNICODE
+
+    return (
+        b"NTLMSSP\x00"
+        + struct.pack("<I", 3)
+        + struct.pack("<HHI", 0, 0, ws_off)              # LmChallengeResponseFields (empty)
+        + struct.pack("<HHI", len(nt_response), len(nt_response), nt_off)
+        + struct.pack("<HHI", len(dom_b), len(dom_b), dom_off)
+        + struct.pack("<HHI", len(user_b), len(user_b), user_off)
+        + struct.pack("<HHI", 0, 0, ws_off)              # WorkstationFields (empty)
+        + struct.pack("<HHI", 0, 0, ws_off)              # EncryptedRandomSessionKey (empty)
+        + struct.pack("<I", flags)
+        + b"\x00" * 8                                    # Version
+        + payload
+    )
+
+
+# ── Helpers ───────────────────────────────────────────────────────────────────
+
+
+def _drive(smb_mod, request_bytes: bytes):
+    async def _run():
+        reader, writer, written = _make_streams()
+        reader.feed_data(request_bytes)
+        reader.feed_eof()
+        await asyncio.wait_for(smb_mod._handle_client(reader, writer), timeout=2.0)
+        return writer, written
+
+    return asyncio.run(_run())
+
+
+# ── Tests ─────────────────────────────────────────────────────────────────────
+
+
+def test_negotiate_response_is_smb2_dialect_0x0210(smb_mod):
+    _, written = _drive(smb_mod, _nbss(_negotiate_request()))
+    blob = b"".join(written)
+    # Skip NBSS header (4 bytes), then SMB2 header (64), body StructureSize, body[2:4]=DialectRevision
+    assert blob[:4] == b"\x00\x00\x00\x83" or blob[0] == 0x00
+    smb = blob[4:]
+    assert smb.startswith(b"\xfeSMB")
+    body = smb[64:]
+    dialect = struct.unpack_from("<H", body, 4)[0]
+    assert dialect == 0x0210
+
+
+def test_first_session_setup_returns_more_processing_required(smb_mod):
+    pkt1 = _nbss(_negotiate_request())
+    pkt2 = _nbss(_session_setup_request(1, _ntlmssp_type1()))
+    _, written = _drive(smb_mod, pkt1 + pkt2)
+    # second response
+    assert len(written) >= 2
+    smb = written[1][4:]
+    status = struct.unpack_from("<I", smb, 8)[0]
+    assert status == 0xC0000016  # STATUS_MORE_PROCESSING_REQUIRED
+    # SecurityBuffer should carry an NTLMSSP Type 2
+    body = smb[64:]
+    sec_off = struct.unpack_from("<H", body, 4)[0]
+    sec_len = struct.unpack_from("<H", body, 6)[0]
+    sec = smb[sec_off:sec_off + sec_len]
+    assert b"NTLMSSP\x00" in sec
+    type_byte = sec[sec.index(b"NTLMSSP\x00") + 8]
+    assert type_byte == 0x02
+
+
+def test_type3_credential_lands_in_log():
+    mod = _load_smb()
+    log_mock = sys.modules["syslog_bridge"]
+    nt_response = b"\xaa" * 32  # 32-byte NTLMv2 response
+    type3 = _ntlmssp_type3("alice", "ACME", nt_response)
+    pkts = (
+        _nbss(_negotiate_request())
+        + _nbss(_session_setup_request(1, _ntlmssp_type1()))
+        + _nbss(_session_setup_request(2, type3))
+    )
+    _drive(mod, pkts)
+
+    # Find the auth_attempt call
+    auth_calls = [
+        c for c in log_mock.syslog_line.call_args_list
+        if len(c.args) >= 3 and c.args[2] == "auth_attempt"
+    ]
+    assert auth_calls, f"no auth_attempt logged; got: {log_mock.syslog_line.call_args_list}"
+    kwargs = auth_calls[0].kwargs
+    assert kwargs["principal"] == "ACME\\alice"
+    assert kwargs["secret_kind"] == "ntlmssp_v2"
+    assert kwargs["username"] == "alice"
+    assert kwargs["domain"] == "ACME"
+    assert "secret_b64" in kwargs and kwargs["secret_b64"]
+
+
+def test_second_session_setup_returns_logon_failure(smb_mod):
+    nt_response = b"\xbb" * 32
+    type3 = _ntlmssp_type3("bob", "", nt_response)
+    pkts = (
+        _nbss(_negotiate_request())
+        + _nbss(_session_setup_request(1, _ntlmssp_type1()))
+        + _nbss(_session_setup_request(2, type3))
+    )
+    _, written = _drive(smb_mod, pkts)
+    smb = written[-1][4:]
+    status = struct.unpack_from("<I", smb, 8)[0]
+    assert status == 0xC000006D  # STATUS_LOGON_FAILURE
+
+
+def test_oversized_nbss_length_drops_connection(smb_mod):
+    # nb_len = 8 MiB > MAX_NBSS_LEN; framer should bail before allocating
+    bad = bytes([0x00]) + (8 * 1024 * 1024).to_bytes(3, "big")
+    _, written = _drive(smb_mod, bad)
+    assert written == []
+
+
+def test_smb1_negotiate_drops_connection(smb_mod):
+    # 0xff 'SMB' is the SMB1 magic — our framer doesn't speak it
+    pdu = b"\xffSMB" + b"\x00" * 60
+    _, written = _drive(smb_mod, _nbss(pdu))
+    assert written == []
+
+
+def test_short_pdu_below_64_drops(smb_mod):
+    # NBSS length < 64 should be rejected
+    bad = bytes([0x00]) + (32).to_bytes(3, "big") + b"\x00" * 32
+    _, written = _drive(smb_mod, bad)
+    assert written == []