"""Tests for modules/active/ntlm_relay — target management, protocol inference, command building. Validates relay target file writing, protocol inference from URLs, ntlmrelayx command construction, Responder coordination, output parsing, and status reporting without requiring ntlmrelayx binary, root, or network access. """ import os import time from pathlib import Path from unittest.mock import MagicMock, patch import pytest from modules.base import BaseModule from modules.active.ntlm_relay import NTLMRelay, RELAY_PROTOCOLS # --------------------------------------------------------------------------- # Fixtures # --------------------------------------------------------------------------- @pytest.fixture def relay(mock_bus, mock_state, tmp_path): """Return a started NTLMRelay instance with temp dirs.""" config = { "interface": "eth0", "ntlmrelayx_binary": str(tmp_path / "ntlmrelayx.py"), } mod = NTLMRelay(mock_bus, mock_state, config) mod._target_file = str(tmp_path / "relay_targets.txt") mod._loot_dir = str(tmp_path / "loot") os.makedirs(mod._loot_dir, exist_ok=True) os.makedirs(os.path.dirname(mod._target_file), exist_ok=True) mod.start() return mod # --------------------------------------------------------------------------- # Structure # --------------------------------------------------------------------------- class TestNTLMRelayStructure: def test_is_base_module(self): assert issubclass(NTLMRelay, BaseModule) def test_module_attributes(self): assert NTLMRelay.name == "ntlm_relay" assert NTLMRelay.module_type == "active" assert NTLMRelay.requires_root is True def test_relay_protocols_frozen(self): assert isinstance(RELAY_PROTOCOLS, frozenset) assert "smb" in RELAY_PROTOCOLS assert "ldap" in RELAY_PROTOCOLS assert "adcs" in RELAY_PROTOCOLS # --------------------------------------------------------------------------- # Protocol inference # --------------------------------------------------------------------------- class TestProtocolInference: def test_infer_smb_from_url(self, relay): protos = relay._infer_protocols(["smb://10.0.0.0"]) assert "smb" in protos def test_infer_ldap_from_url(self, relay): protos = relay._infer_protocols(["ldap://10.0.0.0"]) assert "ldap" in protos def test_infer_multiple_protocols(self, relay): protos = relay._infer_protocols([ "smb://10.0.0.0", "ldap://10.0.0.0", "adcs://10.0.0.0", ]) assert protos == {"smb", "ldap", "adcs"} def test_infer_defaults_to_smb(self, relay): protos = relay._infer_protocols(["10.0.0.0"]) assert "smb" in protos def test_unknown_protocol_ignored(self, relay): protos = relay._infer_protocols(["fake://10.0.0.0", "smb://10.0.0.1"]) assert "smb" in protos assert "fake" not in protos def test_empty_targets_defaults_smb(self, relay): protos = relay._infer_protocols([]) assert protos == {"smb"} # --------------------------------------------------------------------------- # Target file # --------------------------------------------------------------------------- class TestTargetFile: def test_write_target_file(self, relay): relay._targets = ["smb://10.0.0.0", "ldap://10.0.0.0"] relay._write_target_file() content = Path(relay._target_file).read_text() assert "smb://10.0.0.0" in content assert "ldap://10.0.0.0" in content def test_write_target_file_one_per_line(self, relay): relay._targets = ["smb://10.0.0.1", "smb://10.0.0.2", "smb://10.0.0.3"] relay._write_target_file() lines = Path(relay._target_file).read_text().strip().split("\n") assert len(lines) == 3 # --------------------------------------------------------------------------- # Command building # --------------------------------------------------------------------------- class TestCommandBuilding: def test_basic_command(self, relay): relay._targets = ["smb://10.0.0.1"] relay._protocols = {"smb"} cmd = relay._build_command() assert "python3" in cmd assert "-tf" in cmd assert "-smb2support" in cmd assert "-socks" in cmd def test_adcs_flags(self, relay): relay._targets = ["adcs://10.0.0.1"] relay._protocols = {"adcs"} relay._adcs_template = "ESC1" cmd = relay._build_command() assert "--adcs" in cmd assert "--template" in cmd assert "ESC1" in cmd def test_ldap_delegate_access(self, relay): relay._targets = ["ldap://10.0.0.1"] relay._protocols = {"ldap"} cmd = relay._build_command() assert "--delegate-access" in cmd def test_no_adcs_without_template(self, relay): relay._targets = ["smb://10.0.0.1"] relay._protocols = {"smb", "adcs"} relay._adcs_template = None cmd = relay._build_command() assert "--adcs" not in cmd # --------------------------------------------------------------------------- # Responder coordination # --------------------------------------------------------------------------- class TestResponderCoordination: def test_update_exclusions_extracts_ips(self, relay): mock_responder = MagicMock() relay._responder_mgr = mock_responder relay._targets = ["smb://10.0.0.0", "ldap://10.0.0.0:389"] relay._update_responder_exclusions() mock_responder.set_relay_targets.assert_called_once() ips = mock_responder.set_relay_targets.call_args[0][0] assert "10.0.0.0" in ips assert "10.0.0.0" in ips def test_no_responder_no_error(self, relay): relay._responder_mgr = None relay._targets = ["smb://10.0.0.1"] relay._update_responder_exclusions() def test_extract_ip_from_plain_target(self, relay): mock_responder = MagicMock() relay._responder_mgr = mock_responder relay._targets = ["10.0.0.5"] relay._update_responder_exclusions() ips = mock_responder.set_relay_targets.call_args[0][0] assert "10.0.0.5" in ips # --------------------------------------------------------------------------- # Add target # --------------------------------------------------------------------------- class TestAddTarget: def test_add_new_target(self, relay): result = relay.add_target("smb://10.0.0.0") assert result is True assert "smb://10.0.0.0" in relay._targets def test_add_duplicate_target(self, relay): relay.add_target("smb://10.0.0.0") result = relay.add_target("smb://10.0.0.0") assert result is False assert relay._targets.count("smb://10.0.0.0") == 1 # --------------------------------------------------------------------------- # Output parsing # --------------------------------------------------------------------------- class TestOutputParsing: def test_parse_auth_success(self, relay): relay._parse_relay_output("[*] Authenticated successfully against smb://10.0.0.1") assert len(relay._successful_relays) == 1 assert relay._successful_relays[0]["type"] == "auth_success" def test_parse_adcs_certificate(self, relay): relay._parse_relay_output("[*] Certificate saved to /tmp/cert.pem") def test_parse_unrelated_line(self, relay): relay._parse_relay_output("[*] Trying to connect to target smb://10.0.0.1") assert len(relay._successful_relays) == 0 # --------------------------------------------------------------------------- # Configure # --------------------------------------------------------------------------- class TestConfigure: def test_configure_binary_path(self, relay): relay.configure({"ntlmrelayx_binary": "/opt/new/ntlmrelayx.py"}) assert relay._ntlmrelayx_binary == "/opt/new/ntlmrelayx.py" def test_configure_adcs_template(self, relay): relay.configure({"adcs_template": "ESC8"}) assert relay._adcs_template == "ESC8" # --------------------------------------------------------------------------- # Status # --------------------------------------------------------------------------- class TestStatus: def test_status_when_idle(self, relay): s = relay.status() assert s["running"] is True assert s["relay_active"] is False assert s["relay_pid"] is None assert s["targets"] == [] assert s["successful_relays"] == 0 def test_status_reflects_targets(self, relay): relay._targets = ["smb://10.0.0.1"] s = relay.status() assert s["targets"] == ["smb://10.0.0.1"] def test_status_reflects_protocols(self, relay): relay._protocols = {"smb", "ldap"} s = relay.status() assert set(s["protocols"]) == {"smb", "ldap"}