diff --git a/tests/test_ntlm_relay.py b/tests/test_ntlm_relay.py new file mode 100644 index 0000000..92edc5a --- /dev/null +++ b/tests/test_ntlm_relay.py @@ -0,0 +1,808 @@ +#!/usr/bin/env python3 +"""Tests for modules/active/ntlm_relay.py — NTLM relay subprocess wrapper.""" + +import os +import pytest +import subprocess +import tempfile +import threading +import time +from unittest.mock import MagicMock, Mock, patch, call, mock_open + +from modules.active.ntlm_relay import NTLMRelay, RELAY_PROTOCOLS + + +class TestNTLMRelayInstantiation: + """Test module instantiation and configuration.""" + + def test_ntlm_relay_instantiate(self, mock_bus, mock_state, mock_config): + """NTLMRelay can be instantiated with mock bus/state/config.""" + instance = NTLMRelay(bus=mock_bus, state=mock_state, config=mock_config) + + assert instance is not None + assert instance.bus is mock_bus + assert instance.state is mock_state + assert instance.config is mock_config + assert instance._running is False + assert instance.name == "ntlm_relay" + assert instance.module_type == "active" + assert instance.priority == 150 + assert instance.requires_root is True + + def test_ntlm_relay_default_binary_path(self, mock_bus, mock_state, mock_config): + """Default ntlmrelayx binary path is set correctly.""" + instance = NTLMRelay(bus=mock_bus, state=mock_state, config=mock_config) + assert instance._ntlmrelayx_binary == "/opt/tools/impacket/examples/ntlmrelayx.py" + + def test_ntlm_relay_custom_binary_path(self, mock_bus, mock_state, mock_config): + """Custom ntlmrelayx binary path is respected.""" + custom_path = "/custom/path/ntlmrelayx.py" + mock_config["ntlmrelayx_binary"] = custom_path + instance = NTLMRelay(bus=mock_bus, state=mock_state, config=mock_config) + assert instance._ntlmrelayx_binary == custom_path + + def test_ntlm_relay_custom_interface(self, mock_bus, mock_state, mock_config): + """Custom interface is set correctly.""" + mock_config["interface"] = "wlan0" + instance = NTLMRelay(bus=mock_bus, state=mock_state, config=mock_config) + assert instance._iface == "wlan0" + + def test_ntlm_relay_attributes_initialized(self, mock_bus, mock_state, mock_config): + """All internal attributes are initialized.""" + instance = NTLMRelay(bus=mock_bus, state=mock_state, config=mock_config) + + assert instance._proc is None + assert instance._output_thread is None + assert instance._targets == [] + assert instance._protocols == set() + assert instance._adcs_template is None + assert instance._relay_active is False + assert instance._successful_relays == [] + assert isinstance(instance._relays_lock, type(threading.Lock())) + assert instance._responder_mgr is None + + +class TestNTLMRelayStartStop: + """Test start/stop lifecycle.""" + + def test_start_creates_directories(self, mock_bus, mock_state, mock_config, tmp_path): + """start() creates relay_targets and loot directories.""" + instance = NTLMRelay(bus=mock_bus, state=mock_state, config=mock_config) + + # Override paths to temp directory + instance._target_file = os.path.join(str(tmp_path), ".implant", "relay_targets.txt") + instance._loot_dir = os.path.join(str(tmp_path), ".implant", "ntlmrelay_loot") + + with patch.object(instance.state, 'set_module_status') as mock_set_status: + instance.start() + + assert instance._running is True + assert os.path.isdir(os.path.dirname(instance._target_file)) + assert os.path.isdir(instance._loot_dir) + mock_set_status.assert_called_once_with("ntlm_relay", "running", pid=instance._pid) + + def test_start_idempotent(self, mock_bus, mock_state, mock_config): + """Calling start() twice doesn't re-initialize.""" + instance = NTLMRelay(bus=mock_bus, state=mock_state, config=mock_config) + + with patch.object(instance.state, 'set_module_status') as mock_set_status: + instance.start() + call_count_1 = mock_set_status.call_count + + instance.start() + call_count_2 = mock_set_status.call_count + + # Second start() should return early, not call set_module_status again + assert call_count_2 == call_count_1 + + def test_stop_when_not_running(self, mock_bus, mock_state, mock_config): + """stop() returns early if not running.""" + instance = NTLMRelay(bus=mock_bus, state=mock_state, config=mock_config) + + with patch.object(instance.state, 'set_module_status') as mock_set_status: + instance.stop() + mock_set_status.assert_not_called() + + def test_stop_stops_relay(self, mock_bus, mock_state, mock_config): + """stop() calls stop_relay if relay is active.""" + instance = NTLMRelay(bus=mock_bus, state=mock_state, config=mock_config) + instance._running = True + instance._relay_active = True + + with patch.object(instance, 'stop_relay') as mock_stop_relay: + with patch.object(instance.state, 'set_module_status'): + instance.stop() + mock_stop_relay.assert_called_once() + + def test_stop_sets_status(self, mock_bus, mock_state, mock_config): + """stop() updates module status.""" + instance = NTLMRelay(bus=mock_bus, state=mock_state, config=mock_config) + instance._running = True + + with patch.object(instance.state, 'set_module_status') as mock_set_status: + instance.stop() + mock_set_status.assert_called_once_with("ntlm_relay", "stopped") + + +class TestBuildCommand: + """Test ntlmrelayx command building.""" + + def test_build_command_basic(self, mock_bus, mock_state, mock_config, tmp_path): + """_build_command builds basic command without ADCS.""" + instance = NTLMRelay(bus=mock_bus, state=mock_state, config=mock_config) + instance._target_file = str(tmp_path / "targets.txt") + instance._loot_dir = str(tmp_path) + instance._protocols = {"smb"} + + cmd = instance._build_command() + + assert "python3" in cmd + assert instance._ntlmrelayx_binary in cmd + assert "-tf" in cmd + assert instance._target_file in cmd + assert "-of" in cmd + assert "-smb2support" in cmd + assert "-socks" in cmd + + def test_build_command_with_adcs(self, mock_bus, mock_state, mock_config, tmp_path): + """_build_command includes ADCS options when template is set.""" + instance = NTLMRelay(bus=mock_bus, state=mock_state, config=mock_config) + instance._target_file = str(tmp_path / "targets.txt") + instance._loot_dir = str(tmp_path) + instance._protocols = {"adcs"} + instance._adcs_template = "User" + + cmd = instance._build_command() + + assert "--adcs" in cmd + assert "--template" in cmd + assert "User" in cmd + + def test_build_command_without_adcs_if_not_in_protocols(self, mock_bus, mock_state, mock_config, tmp_path): + """_build_command does not include ADCS options if adcs not in protocols.""" + instance = NTLMRelay(bus=mock_bus, state=mock_state, config=mock_config) + instance._target_file = str(tmp_path / "targets.txt") + instance._loot_dir = str(tmp_path) + instance._protocols = {"smb"} + instance._adcs_template = "User" + + cmd = instance._build_command() + + assert "--adcs" not in cmd + assert "--template" not in cmd + + def test_build_command_with_ldap(self, mock_bus, mock_state, mock_config, tmp_path): + """_build_command includes delegate-access for LDAP.""" + instance = NTLMRelay(bus=mock_bus, state=mock_state, config=mock_config) + instance._target_file = str(tmp_path / "targets.txt") + instance._loot_dir = str(tmp_path) + instance._protocols = {"ldap"} + + cmd = instance._build_command() + + assert "--delegate-access" in cmd + + def test_build_command_with_ldaps(self, mock_bus, mock_state, mock_config, tmp_path): + """_build_command includes delegate-access for LDAPS.""" + instance = NTLMRelay(bus=mock_bus, state=mock_state, config=mock_config) + instance._target_file = str(tmp_path / "targets.txt") + instance._loot_dir = str(tmp_path) + instance._protocols = {"ldaps"} + + cmd = instance._build_command() + + assert "--delegate-access" in cmd + + def test_build_command_hashes_output_path(self, mock_bus, mock_state, mock_config, tmp_path): + """_build_command sets correct hash output path.""" + instance = NTLMRelay(bus=mock_bus, state=mock_state, config=mock_config) + instance._target_file = str(tmp_path / "targets.txt") + instance._loot_dir = str(tmp_path) + + cmd = instance._build_command() + hash_idx = cmd.index("-of") + hash_path = cmd[hash_idx + 1] + + assert hash_path == os.path.join(instance._loot_dir, "hashes") + + +class TestInferProtocols: + """Test protocol inference from target URLs.""" + + def test_infer_protocols_from_smb_url(self, mock_bus, mock_state, mock_config): + """_infer_protocols extracts SMB from smb:// URL.""" + instance = NTLMRelay(bus=mock_bus, state=mock_state, config=mock_config) + protocols = instance._infer_protocols(["smb://10.0.0.0"]) + assert "smb" in protocols + + def test_infer_protocols_from_ldap_url(self, mock_bus, mock_state, mock_config): + """_infer_protocols extracts LDAP from ldap:// URL.""" + instance = NTLMRelay(bus=mock_bus, state=mock_state, config=mock_config) + protocols = instance._infer_protocols(["ldap://10.0.0.0"]) + assert "ldap" in protocols + + def test_infer_protocols_from_adcs_url(self, mock_bus, mock_state, mock_config): + """_infer_protocols extracts ADCS from adcs:// URL.""" + instance = NTLMRelay(bus=mock_bus, state=mock_state, config=mock_config) + protocols = instance._infer_protocols(["adcs://10.0.0.0"]) + assert "adcs" in protocols + + def test_infer_protocols_plain_ip_defaults_to_smb(self, mock_bus, mock_state, mock_config): + """_infer_protocols defaults to SMB for plain IP addresses.""" + instance = NTLMRelay(bus=mock_bus, state=mock_state, config=mock_config) + protocols = instance._infer_protocols(["10.0.0.0"]) + assert "smb" in protocols + + def test_infer_protocols_multiple_targets(self, mock_bus, mock_state, mock_config): + """_infer_protocols handles multiple targets with different protocols.""" + instance = NTLMRelay(bus=mock_bus, state=mock_state, config=mock_config) + protocols = instance._infer_protocols([ + "smb://10.0.0.0", + "ldap://10.0.0.0", + "adcs://10.0.0.0", + ]) + assert "smb" in protocols + assert "ldap" in protocols + assert "adcs" in protocols + + def test_infer_protocols_case_insensitive(self, mock_bus, mock_state, mock_config): + """_infer_protocols is case insensitive.""" + instance = NTLMRelay(bus=mock_bus, state=mock_state, config=mock_config) + protocols = instance._infer_protocols(["SMB://10.0.0.0", "LDAP://10.0.0.0"]) + assert "smb" in protocols + assert "ldap" in protocols + + def test_infer_protocols_empty_defaults_to_smb(self, mock_bus, mock_state, mock_config): + """_infer_protocols defaults to SMB if protocols are empty.""" + instance = NTLMRelay(bus=mock_bus, state=mock_state, config=mock_config) + protocols = instance._infer_protocols([]) + assert protocols == {"smb"} + + +class TestParseRelayOutput: + """Test relay output parsing for events.""" + + def test_parse_authenticated_successfully(self, mock_bus, mock_state, mock_config): + """_parse_relay_output detects 'authenticated successfully' and emits CREDENTIAL_FOUND.""" + instance = NTLMRelay(bus=mock_bus, state=mock_state, config=mock_config) + instance._running = True + + with patch.object(instance.bus, 'emit') as mock_emit: + instance._parse_relay_output("SMB session opened. User authenticated successfully") + + mock_emit.assert_called_once() + call_args = mock_emit.call_args + assert call_args[0][0] == "CREDENTIAL_FOUND" + assert call_args[0][1]["source_module"] == "ntlm_relay" + assert call_args[0][1]["credential_type"] == "relay_success" + + def test_parse_authenticated_successfully_adds_to_list(self, mock_bus, mock_state, mock_config): + """_parse_relay_output adds authenticated entry to _successful_relays.""" + instance = NTLMRelay(bus=mock_bus, state=mock_state, config=mock_config) + instance._running = True + + with patch.object(instance.bus, 'emit'): + instance._parse_relay_output("User authenticated successfully to DC") + + assert len(instance._successful_relays) == 1 + assert instance._successful_relays[0]["type"] == "auth_success" + + def test_parse_sam_dump_detection(self, mock_bus, mock_state, mock_config): + """_parse_relay_output detects SAM dump.""" + instance = NTLMRelay(bus=mock_bus, state=mock_state, config=mock_config) + instance._running = True + + with patch.object(instance.bus, 'emit'): + instance._parse_relay_output("Dumping SAM hashes from target") + + # Should not crash, may log + + def test_parse_adcs_certificate_obtained(self, mock_bus, mock_state, mock_config): + """_parse_relay_output detects ADCS certificate and emits event.""" + instance = NTLMRelay(bus=mock_bus, state=mock_state, config=mock_config) + instance._running = True + + with patch.object(instance.bus, 'emit') as mock_emit: + instance._parse_relay_output("Certificate saved to file: cert.pem") + + mock_emit.assert_called_once() + call_args = mock_emit.call_args + assert call_args[0][0] == "CREDENTIAL_FOUND" + assert call_args[0][1]["credential_type"] == "certificate" + assert call_args[0][1]["target_service"] == "adcs_relay" + + def test_parse_adcs_certificate_base64(self, mock_bus, mock_state, mock_config): + """_parse_relay_output detects ADCS certificate in base64 format.""" + instance = NTLMRelay(bus=mock_bus, state=mock_state, config=mock_config) + instance._running = True + + with patch.object(instance.bus, 'emit') as mock_emit: + instance._parse_relay_output("Certificate in base64 format:") + + mock_emit.assert_called_once() + assert mock_emit.call_args[0][1]["credential_type"] == "certificate" + + def test_parse_socks_connection(self, mock_bus, mock_state, mock_config): + """_parse_relay_output detects SOCKS proxy connection.""" + instance = NTLMRelay(bus=mock_bus, state=mock_state, config=mock_config) + instance._running = True + + with patch.object(instance.bus, 'emit'): + instance._parse_relay_output("SOCKS proxy connection established") + + # Should not crash, may log + + def test_parse_empty_line_ignored(self, mock_bus, mock_state, mock_config): + """_parse_relay_output ignores empty lines.""" + instance = NTLMRelay(bus=mock_bus, state=mock_state, config=mock_config) + instance._running = True + + with patch.object(instance.bus, 'emit'): + instance._parse_relay_output("") + + # Should not crash or emit + + def test_parse_case_insensitive(self, mock_bus, mock_state, mock_config): + """_parse_relay_output matching is case insensitive.""" + instance = NTLMRelay(bus=mock_bus, state=mock_state, config=mock_config) + instance._running = True + + with patch.object(instance.bus, 'emit') as mock_emit: + instance._parse_relay_output("USER AUTHENTICATED SUCCESSFULLY to target") + + mock_emit.assert_called_once() + + +class TestWriteTargetFile: + """Test target file writing.""" + + def test_write_target_file_creates_file(self, mock_bus, mock_state, mock_config, tmp_path): + """_write_target_file creates targets file with all targets.""" + instance = NTLMRelay(bus=mock_bus, state=mock_state, config=mock_config) + instance._target_file = os.path.join(str(tmp_path), "targets.txt") + instance._targets = ["smb://10.0.0.0", "ldap://10.0.0.0"] + + instance._write_target_file() + + assert os.path.exists(instance._target_file) + with open(instance._target_file) as f: + content = f.read() + 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, mock_bus, mock_state, mock_config, tmp_path): + """_write_target_file writes one target per line.""" + instance = NTLMRelay(bus=mock_bus, state=mock_state, config=mock_config) + instance._target_file = os.path.join(str(tmp_path), "targets.txt") + instance._targets = ["10.0.0.0", "10.0.0.0", "10.0.0.0"] + + instance._write_target_file() + + with open(instance._target_file) as f: + lines = f.read().strip().split('\n') + assert len(lines) == 3 + + def test_write_target_file_empty_targets(self, mock_bus, mock_state, mock_config, tmp_path): + """_write_target_file handles empty targets list.""" + instance = NTLMRelay(bus=mock_bus, state=mock_state, config=mock_config) + instance._target_file = os.path.join(str(tmp_path), "targets.txt") + instance._targets = [] + + instance._write_target_file() + + assert os.path.exists(instance._target_file) + with open(instance._target_file) as f: + content = f.read() + assert content == "" + + +class TestResponderExclusions: + """Test Responder exclusion coordination.""" + + def test_update_responder_exclusions_no_responder_mgr(self, mock_bus, mock_state, mock_config): + """_update_responder_exclusions returns early if no responder_mgr.""" + instance = NTLMRelay(bus=mock_bus, state=mock_state, config=mock_config) + instance._responder_mgr = None + + # Should not crash + instance._update_responder_exclusions() + + def test_update_responder_exclusions_calls_set_relay_targets(self, mock_bus, mock_state, mock_config): + """_update_responder_exclusions calls responder_mgr.set_relay_targets with extracted IPs.""" + instance = NTLMRelay(bus=mock_bus, state=mock_state, config=mock_config) + mock_responder_mgr = MagicMock() + instance._responder_mgr = mock_responder_mgr + instance._targets = ["smb://10.0.0.0", "ldap://10.0.0.0"] + + instance._update_responder_exclusions() + + mock_responder_mgr.set_relay_targets.assert_called_once() + called_ips = mock_responder_mgr.set_relay_targets.call_args[0][0] + assert "10.0.0.0" in called_ips + assert "10.0.0.0" in called_ips + + def test_update_responder_exclusions_extracts_ip_from_url(self, mock_bus, mock_state, mock_config): + """_update_responder_exclusions correctly extracts IP from full URLs.""" + instance = NTLMRelay(bus=mock_bus, state=mock_state, config=mock_config) + mock_responder_mgr = MagicMock() + instance._responder_mgr = mock_responder_mgr + instance._targets = ["ldap://10.0.0.0:389/CN=Users"] + + instance._update_responder_exclusions() + + called_ips = mock_responder_mgr.set_relay_targets.call_args[0][0] + assert "10.0.0.0" in called_ips + + def test_update_responder_exclusions_handles_plain_ip(self, mock_bus, mock_state, mock_config): + """_update_responder_exclusions handles plain IP addresses.""" + instance = NTLMRelay(bus=mock_bus, state=mock_state, config=mock_config) + mock_responder_mgr = MagicMock() + instance._responder_mgr = mock_responder_mgr + instance._targets = ["10.0.0.0"] + + instance._update_responder_exclusions() + + called_ips = mock_responder_mgr.set_relay_targets.call_args[0][0] + assert "10.0.0.0" in called_ips + + def test_update_responder_exclusions_handles_exception(self, mock_bus, mock_state, mock_config): + """_update_responder_exclusions catches and logs exceptions.""" + instance = NTLMRelay(bus=mock_bus, state=mock_state, config=mock_config) + mock_responder_mgr = MagicMock() + mock_responder_mgr.set_relay_targets.side_effect = Exception("Test error") + instance._responder_mgr = mock_responder_mgr + + # Should not crash + instance._update_responder_exclusions() + + +class TestAddTarget: + """Test add_target functionality.""" + + def test_add_target_adds_new_target(self, mock_bus, mock_state, mock_config): + """add_target adds new target to list.""" + instance = NTLMRelay(bus=mock_bus, state=mock_state, config=mock_config) + + result = instance.add_target("smb://10.0.0.0") + + assert result is True + assert "smb://10.0.0.0" in instance._targets + + def test_add_target_ignores_duplicate(self, mock_bus, mock_state, mock_config): + """add_target returns False for duplicate target.""" + instance = NTLMRelay(bus=mock_bus, state=mock_state, config=mock_config) + instance._targets = ["smb://10.0.0.0"] + + result = instance.add_target("smb://10.0.0.0") + + assert result is False + + def test_add_target_writes_target_file(self, mock_bus, mock_state, mock_config, tmp_path): + """add_target writes updated target file.""" + instance = NTLMRelay(bus=mock_bus, state=mock_state, config=mock_config) + instance._target_file = os.path.join(str(tmp_path), "targets.txt") + + with patch.object(instance, '_update_responder_exclusions'): + instance.add_target("smb://10.0.0.0") + + assert os.path.exists(instance._target_file) + + def test_add_target_updates_responder_exclusions(self, mock_bus, mock_state, mock_config): + """add_target calls _update_responder_exclusions.""" + instance = NTLMRelay(bus=mock_bus, state=mock_state, config=mock_config) + + with patch.object(instance, '_update_responder_exclusions') as mock_update: + with patch.object(instance, '_write_target_file'): + instance.add_target("smb://10.0.0.0") + mock_update.assert_called_once() + + +class TestStartRelay: + """Test start_relay functionality.""" + + def test_start_relay_not_running(self, mock_bus, mock_state, mock_config): + """start_relay returns False if module not running.""" + instance = NTLMRelay(bus=mock_bus, state=mock_state, config=mock_config) + instance._running = False + + result = instance.start_relay(["10.0.0.0"]) + + assert result is False + + def test_start_relay_already_running(self, mock_bus, mock_state, mock_config): + """start_relay returns True if relay already active.""" + instance = NTLMRelay(bus=mock_bus, state=mock_state, config=mock_config) + instance._running = True + instance._relay_active = True + mock_proc = MagicMock() + mock_proc.poll.return_value = None + instance._proc = mock_proc + + result = instance.start_relay(["10.0.0.0"]) + + assert result is True + + def test_start_relay_infers_protocols(self, mock_bus, mock_state, mock_config, tmp_path): + """start_relay infers protocols from targets.""" + instance = NTLMRelay(bus=mock_bus, state=mock_state, config=mock_config) + instance._running = True + instance._target_file = os.path.join(str(tmp_path), "targets.txt") + instance._loot_dir = str(tmp_path) + + with patch('subprocess.Popen') as mock_popen: + mock_popen.return_value = MagicMock(poll=MagicMock(return_value=None), pid=1234) + with patch.object(instance, '_monitor_output'): + instance.start_relay(["smb://10.0.0.0", "ldap://10.0.0.0"]) + + assert "smb" in instance._protocols + assert "ldap" in instance._protocols + + def test_start_relay_respects_provided_protocols(self, mock_bus, mock_state, mock_config, tmp_path): + """start_relay uses provided protocols if given.""" + instance = NTLMRelay(bus=mock_bus, state=mock_state, config=mock_config) + instance._running = True + instance._target_file = os.path.join(str(tmp_path), "targets.txt") + instance._loot_dir = str(tmp_path) + + with patch('subprocess.Popen') as mock_popen: + mock_popen.return_value = MagicMock(poll=MagicMock(return_value=None), pid=1234) + with patch.object(instance, '_monitor_output'): + instance.start_relay(["10.0.0.0"], protocols={"ldap"}) + + assert instance._protocols == {"ldap"} + + def test_start_relay_writes_target_file(self, mock_bus, mock_state, mock_config, tmp_path): + """start_relay writes targets to file.""" + instance = NTLMRelay(bus=mock_bus, state=mock_state, config=mock_config) + instance._running = True + instance._target_file = os.path.join(str(tmp_path), "targets.txt") + instance._loot_dir = str(tmp_path) + + with patch('subprocess.Popen') as mock_popen: + mock_popen.return_value = MagicMock(poll=MagicMock(return_value=None), pid=1234) + with patch.object(instance, '_monitor_output'): + instance.start_relay(["10.0.0.0"]) + + assert os.path.exists(instance._target_file) + + def test_start_relay_spawns_subprocess(self, mock_bus, mock_state, mock_config, tmp_path): + """start_relay spawns ntlmrelayx subprocess.""" + instance = NTLMRelay(bus=mock_bus, state=mock_state, config=mock_config) + instance._running = True + instance._target_file = os.path.join(str(tmp_path), "targets.txt") + instance._loot_dir = str(tmp_path) + + with patch('subprocess.Popen') as mock_popen: + mock_popen.return_value = MagicMock(poll=MagicMock(return_value=None), pid=1234) + with patch.object(instance, '_monitor_output'): + instance.start_relay(["10.0.0.0"]) + + mock_popen.assert_called_once() + call_args = mock_popen.call_args + cmd = call_args[0][0] + assert "python3" in cmd + + def test_start_relay_returns_false_if_process_dies(self, mock_bus, mock_state, mock_config, tmp_path): + """start_relay returns False if subprocess exits immediately.""" + instance = NTLMRelay(bus=mock_bus, state=mock_state, config=mock_config) + instance._running = True + instance._target_file = os.path.join(str(tmp_path), "targets.txt") + instance._loot_dir = str(tmp_path) + + with patch('subprocess.Popen') as mock_popen: + mock_proc = MagicMock() + mock_proc.poll.return_value = 1 # Process exited + mock_proc.stderr.read.return_value = b"Error message" + mock_popen.return_value = mock_proc + + result = instance.start_relay(["10.0.0.0"]) + + assert result is False + + def test_start_relay_handles_file_not_found(self, mock_bus, mock_state, mock_config, tmp_path): + """start_relay returns False if binary not found.""" + instance = NTLMRelay(bus=mock_bus, state=mock_state, config=mock_config) + instance._running = True + instance._ntlmrelayx_binary = "/nonexistent/path" + instance._target_file = os.path.join(str(tmp_path), "targets.txt") + instance._loot_dir = str(tmp_path) + + with patch('subprocess.Popen', side_effect=FileNotFoundError): + result = instance.start_relay(["10.0.0.0"]) + assert result is False + + def test_start_relay_sets_relay_active(self, mock_bus, mock_state, mock_config, tmp_path): + """start_relay sets _relay_active to True on success.""" + instance = NTLMRelay(bus=mock_bus, state=mock_state, config=mock_config) + instance._running = True + instance._target_file = os.path.join(str(tmp_path), "targets.txt") + instance._loot_dir = str(tmp_path) + + with patch('subprocess.Popen') as mock_popen: + mock_popen.return_value = MagicMock(poll=MagicMock(return_value=None), pid=1234) + with patch.object(instance, '_monitor_output'): + instance.start_relay(["10.0.0.0"]) + + assert instance._relay_active is True + + def test_start_relay_starts_monitor_thread(self, mock_bus, mock_state, mock_config, tmp_path): + """start_relay starts output monitoring thread.""" + instance = NTLMRelay(bus=mock_bus, state=mock_state, config=mock_config) + instance._running = True + instance._target_file = os.path.join(str(tmp_path), "targets.txt") + instance._loot_dir = str(tmp_path) + + with patch('subprocess.Popen') as mock_popen: + mock_popen.return_value = MagicMock(poll=MagicMock(return_value=None), pid=1234) + with patch('threading.Thread') as mock_thread: + instance.start_relay(["10.0.0.0"]) + + mock_thread.assert_called_once() + thread_kwargs = mock_thread.call_args[1] + assert thread_kwargs['daemon'] is True + + +class TestStopRelay: + """Test stop_relay functionality.""" + + def test_stop_relay_no_process(self, mock_bus, mock_state, mock_config): + """stop_relay returns True if no process running.""" + instance = NTLMRelay(bus=mock_bus, state=mock_state, config=mock_config) + instance._proc = None + + result = instance.stop_relay() + + assert result is True + assert instance._relay_active is False + + def test_stop_relay_terminates_process(self, mock_bus, mock_state, mock_config): + """stop_relay terminates running process.""" + instance = NTLMRelay(bus=mock_bus, state=mock_state, config=mock_config) + mock_proc = MagicMock() + mock_proc.poll.return_value = None # Still running + instance._proc = mock_proc + + instance.stop_relay() + + mock_proc.terminate.assert_called_once() + mock_proc.wait.assert_called_once() + + def test_stop_relay_kills_if_timeout(self, mock_bus, mock_state, mock_config): + """stop_relay kills process if terminate times out.""" + instance = NTLMRelay(bus=mock_bus, state=mock_state, config=mock_config) + mock_proc = MagicMock() + mock_proc.poll.return_value = None + mock_proc.terminate.side_effect = lambda: None + mock_proc.wait.side_effect = [ + subprocess.TimeoutExpired("cmd", 5), # First wait times out (after terminate) + ] + instance._proc = mock_proc + + instance.stop_relay() + + mock_proc.kill.assert_called_once() + + def test_stop_relay_sets_proc_to_none(self, mock_bus, mock_state, mock_config): + """stop_relay sets _proc to None.""" + instance = NTLMRelay(bus=mock_bus, state=mock_state, config=mock_config) + mock_proc = MagicMock() + mock_proc.poll.return_value = None + instance._proc = mock_proc + + instance.stop_relay() + + assert instance._proc is None + + def test_stop_relay_sets_relay_active_false(self, mock_bus, mock_state, mock_config): + """stop_relay sets _relay_active to False.""" + instance = NTLMRelay(bus=mock_bus, state=mock_state, config=mock_config) + instance._relay_active = True + mock_proc = MagicMock() + mock_proc.poll.return_value = None + instance._proc = mock_proc + + instance.stop_relay() + + assert instance._relay_active is False + + +class TestStatus: + """Test status reporting.""" + + def test_status_when_not_running(self, mock_bus, mock_state, mock_config): + """status() returns running=False when not started.""" + instance = NTLMRelay(bus=mock_bus, state=mock_state, config=mock_config) + + status = instance.status() + + assert status["running"] is False + + def test_status_when_running(self, mock_bus, mock_state, mock_config): + """status() returns running=True when started.""" + instance = NTLMRelay(bus=mock_bus, state=mock_state, config=mock_config) + instance._running = True + + status = instance.status() + + assert status["running"] is True + + def test_status_includes_relay_info(self, mock_bus, mock_state, mock_config): + """status() includes relay targets and protocols.""" + instance = NTLMRelay(bus=mock_bus, state=mock_state, config=mock_config) + instance._running = True + instance._targets = ["smb://10.0.0.0"] + instance._protocols = {"smb", "ldap"} + instance._successful_relays = [{"type": "auth_success"}] + + status = instance.status() + + assert status["targets"] == ["smb://10.0.0.0"] + assert "smb" in status["protocols"] + assert "ldap" in status["protocols"] + assert status["successful_relays"] == 1 + + def test_status_includes_adcs_template(self, mock_bus, mock_state, mock_config): + """status() includes ADCS template if set.""" + instance = NTLMRelay(bus=mock_bus, state=mock_state, config=mock_config) + instance._adcs_template = "User" + + status = instance.status() + + assert status["adcs_template"] == "User" + + def test_status_relay_active_checks_process(self, mock_bus, mock_state, mock_config): + """status() checks if process is still alive for relay_active.""" + instance = NTLMRelay(bus=mock_bus, state=mock_state, config=mock_config) + instance._running = True + instance._relay_active = True + mock_proc = MagicMock() + mock_proc.poll.return_value = None # Still running + mock_proc.pid = 1234 + instance._proc = mock_proc + + status = instance.status() + + assert status["relay_active"] is True + assert status["relay_pid"] == 1234 + + def test_status_relay_inactive_if_process_dead(self, mock_bus, mock_state, mock_config): + """status() reports relay_active=False if process exited.""" + instance = NTLMRelay(bus=mock_bus, state=mock_state, config=mock_config) + instance._running = True + instance._relay_active = True + mock_proc = MagicMock() + mock_proc.poll.return_value = 1 # Process exited + instance._proc = mock_proc + + status = instance.status() + + assert status["relay_active"] is False + + +class TestConfigure: + """Test module configuration.""" + + def test_configure_updates_binary_path(self, mock_bus, mock_state, mock_config): + """configure() updates ntlmrelayx_binary.""" + instance = NTLMRelay(bus=mock_bus, state=mock_state, config=mock_config) + new_binary = "/custom/ntlmrelayx.py" + + instance.configure({"ntlmrelayx_binary": new_binary}) + + assert instance._ntlmrelayx_binary == new_binary + + def test_configure_updates_adcs_template(self, mock_bus, mock_state, mock_config): + """configure() updates ADCS template.""" + instance = NTLMRelay(bus=mock_bus, state=mock_state, config=mock_config) + + instance.configure({"adcs_template": "Computer"}) + + assert instance._adcs_template == "Computer" + + def test_configure_updates_config_dict(self, mock_bus, mock_state, mock_config): + """configure() updates the config dict.""" + instance = NTLMRelay(bus=mock_bus, state=mock_state, config=mock_config) + + instance.configure({"custom_key": "custom_value"}) + + assert instance.config["custom_key"] == "custom_value"