diff --git a/decnet/templates/mssql/server.py b/decnet/templates/mssql/server.py index f9eb5558..d647f513 100644 --- a/decnet/templates/mssql/server.py +++ b/decnet/templates/mssql/server.py @@ -6,6 +6,7 @@ a login failed error. Logs auth attempts as JSON. """ import asyncio +import base64 import os import struct @@ -142,26 +143,63 @@ class MSSQLProtocol(asyncio.Protocol): self._transport.write(_PRELOGIN_RESP) self._prelogin_done = True elif pkt_type == 0x10: # Login7 - username = self._parse_login7_username(payload) - _log("auth", src=self._peer[0], username=username) + username, password = self._parse_login7_creds(payload) + extra: dict = {} + if password: + pw_bytes = password.encode("utf-8") + extra = { + "principal": username, + "secret_kind": "plaintext", + "secret_printable": password, + "secret_b64": base64.b64encode(pw_bytes).decode("ascii"), + } + _log("auth", src=self._peer[0], username=username, **extra) self._transport.write(_tds_error_packet("Login failed for user.")) self._transport.close() else: _log("unknown_packet", src=self._peer[0], pkt_type=hex(pkt_type)) self._transport.close() - def _parse_login7_username(self, payload: bytes) -> str: + @staticmethod + def _deobfuscate_login7_password(blob: bytes) -> str: + """MS-TDS Login7 password obfuscation: each byte was rotated-right + 4 bits then XOR'd with 0xa5. Inverse is XOR 0xa5 then rotate-left + 4 bits (== nibble swap). Plaintext-recoverable. + + After deobfuscation the bytes are UTF-16-LE encoded characters.""" + out = bytearray(len(blob)) + for i, b in enumerate(blob): + x = b ^ 0xa5 + out[i] = ((x & 0x0f) << 4) | ((x & 0xf0) >> 4) + return bytes(out).decode("utf-16-le", errors="replace") + + def _parse_login7_creds(self, payload: bytes) -> tuple[str, str]: + """Login7 offset table starts at payload offset 36: + + 36-37 ibHostName 38-39 cchHostName + 40-41 ibUserName 42-43 cchUserName + 44-45 ibPassword 46-47 cchPassword + + Both lengths are CHARACTER counts; multiply by 2 for byte length. + The password field is XOR/swap-obfuscated — see + :meth:`_deobfuscate_login7_password`. Plaintext-recoverable. + """ try: - # Login7 layout: fixed header 36 bytes, then offsets - # Username offset at bytes 36-37, length at 38-39 - if len(payload) < 40: - return "" - offset = struct.unpack("", "" + user_off, user_len = struct.unpack("" + return "", "" def connection_lost(self, exc): _log("disconnect", src=self._peer[0] if self._peer else "?") diff --git a/decnet/templates/mysql/server.py b/decnet/templates/mysql/server.py index 13d8f323..17de3f33 100644 --- a/decnet/templates/mysql/server.py +++ b/decnet/templates/mysql/server.py @@ -7,6 +7,7 @@ attempts as JSON. """ import asyncio +import base64 import itertools import os import struct @@ -109,17 +110,38 @@ class MySQLProtocol(asyncio.Protocol): def _handle_packet(self, payload: bytes): if not payload: return - # Login packet: capability flags (4), max_packet (4), charset (1), reserved (23), username (NUL-terminated) + # Login packet: capability flags (4), max_packet (4), charset (1), + # reserved (23), username (NUL-terminated), auth-response. + # mysql_native_password puts a 1-byte length followed by exactly + # 20 bytes: SHA1(password) XOR SHA1(salt + SHA1(SHA1(password))) — + # plaintext is unrecoverable but the 20 bytes ARE a credential the + # attacker knew, so they land as secret_kind="mysql_native_password". username = "" + auth_response = b"" if len(payload) > 32: try: username_start = 32 nul = payload.index(b"\x00", username_start) username = payload[username_start:nul].decode(errors="replace") + # auth-response length byte + bytes + if len(payload) > nul + 1: + resp_len = payload[nul + 1] + if resp_len and len(payload) >= nul + 2 + resp_len: + auth_response = payload[nul + 2:nul + 2 + resp_len] except (ValueError, IndexError): username = "" + + extra: dict = {} + if auth_response: + _b64 = base64.b64encode(auth_response).decode("ascii") + extra = { + "principal": username, + "secret_kind": "mysql_native_password", + "secret_printable": auth_response.hex(), + "secret_b64": _b64, + } _log("auth", src=self._peer[0], username=username, - connection_id=self._conn_id) + connection_id=self._conn_id, **extra) # Real mysqld includes client IP in the error text. src_ip = self._peer[0] if self._peer else "?" msg = f"Access denied for user '{username}'@'{src_ip}' (using password: YES)" diff --git a/tests/services/test_cred_emitters.py b/tests/services/test_cred_emitters.py index e81939a1..bc69ae51 100644 --- a/tests/services/test_cred_emitters.py +++ b/tests/services/test_cred_emitters.py @@ -325,6 +325,104 @@ async def test_sip_digest_native_shape(): assert cred["principal"] == "alice" +@pytest.mark.asyncio +async def test_mysql_native_password_hash(): + """MySQL handshake auth-response: 20-byte sha1 chain hash. Plaintext + irrecoverable; lands as secret_kind=mysql_native_password.""" + from decnet.web.ingester import _extract_bounty + repo = MagicMock(); repo.upsert_credential = AsyncMock() + raw = bytes(range(20)) # arbitrary 20-byte "hash" + log_data = { + "decky": "decky-01", + "service": "mysql", + "attacker_ip": "10.0.0.5", + "fields": { + "username": "root", + "principal": "root", + "secret_kind": "mysql_native_password", + "secret_printable": raw.hex(), + "secret_b64": base64.b64encode(raw).decode("ascii"), + }, + } + await _extract_bounty(repo, log_data) + cred = repo.upsert_credential.call_args[0][0] + assert cred["service"] == "mysql" + assert cred["secret_kind"] == "mysql_native_password" + assert cred["principal"] == "root" + assert cred["secret_sha256"] == hashlib.sha256(raw).hexdigest() + + +@pytest.mark.asyncio +async def test_mssql_login7_plaintext(): + """MSSQL Login7 password is XOR/nibble-obfuscated but plaintext- + recoverable. Lands as secret_kind=plaintext after deobfuscation.""" + from decnet.web.ingester import _extract_bounty + repo = MagicMock(); repo.upsert_credential = AsyncMock() + log_data = { + "decky": "decky-01", + "service": "mssql", + "attacker_ip": "10.0.0.5", + "fields": { + "username": "sa", + "principal": "sa", + "secret_kind": "plaintext", + "secret_printable": "hunter2", + "secret_b64": base64.b64encode(b"hunter2").decode("ascii"), + }, + } + await _extract_bounty(repo, log_data) + cred = repo.upsert_credential.call_args[0][0] + assert cred["service"] == "mssql" + assert cred["principal"] == "sa" + assert cred["secret_printable"] == "hunter2" + + +def test_mssql_deobfuscate_roundtrip(): + """Direct unit test of the MSSQL Login7 deobfuscation against a + handcrafted obfuscated buffer. Exercises the algorithm itself.""" + import importlib.util + import sys + from pathlib import Path + from types import ModuleType + from unittest.mock import MagicMock + # Stand up a fake syslog_bridge so the template imports cleanly, + # then load the mssql module and test the static helper. + fake = ModuleType("syslog_bridge") + fake.syslog_line = MagicMock(return_value="") + fake.write_syslog_file = MagicMock() + fake.forward_syslog = MagicMock() + fake.SEVERITY_INFO = 6 + fake.SEVERITY_WARNING = 4 + fake.encode_secret = MagicMock(return_value={"secret_printable": "", "secret_b64": ""}) + fake.classify_authorization = MagicMock(return_value=None) + sys.modules["syslog_bridge"] = fake + # Load the real instance_seed so the mssql module's top-level + # _seed.pick(...) tuple-unpack works. MagicMock returns sentinels + # that don't satisfy iterable unpacking. + repo_root = Path(__file__).resolve().parents[2] + if "instance_seed" not in sys.modules: + seed_spec = importlib.util.spec_from_file_location( + "instance_seed", repo_root / "decnet" / "templates" / "instance_seed.py" + ) + seed_mod = importlib.util.module_from_spec(seed_spec) + seed_spec.loader.exec_module(seed_mod) + sys.modules["instance_seed"] = seed_mod + spec = importlib.util.spec_from_file_location( + "_mssql_under_test", repo_root / "decnet" / "templates" / "mssql" / "server.py" + ) + mod = importlib.util.module_from_spec(spec) + spec.loader.exec_module(mod) + + # Build the obfuscated form of "abc": each byte → swap nibbles, XOR 0xa5. + plain = "abc".encode("utf-16-le") # 6 bytes + obfuscated = bytes( + (((b & 0x0f) << 4) | ((b & 0xf0) >> 4)) ^ 0xa5 + for b in plain + ) + decoded = mod.MSSQLProtocol._deobfuscate_login7_password(obfuscated) + assert decoded == "abc" + + @pytest.mark.asyncio async def test_lossless_b64_survives_nonprintable_password(): """Even when secret_printable is sanitized, secret_b64 still decodes