diff --git a/tests/test_responder_mgr.py b/tests/test_responder_mgr.py new file mode 100644 index 0000000..43c9368 --- /dev/null +++ b/tests/test_responder_mgr.py @@ -0,0 +1,247 @@ +"""Tests for modules/active/responder_mgr — hash capture, dedup, config generation. + +Validates hash parsing, deduplication, Responder.conf generation, +relay target exclusion, and status reporting without requiring +Responder 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.responder_mgr import ( + ResponderManager, HASH_FILE_PATTERN, NTLMV2_REGEX, NTLMV1_REGEX, +) + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + +@pytest.fixture +def responder(mock_bus, mock_state, tmp_path): + """Return a ResponderManager with temp dirs and mocked externals.""" + config = { + "responder_path": str(tmp_path / "Responder"), + "interface": "eth0", + "template_dir": str(tmp_path / "templates"), + } + resp_dir = tmp_path / "Responder" + resp_dir.mkdir() + (resp_dir / "logs").mkdir() + (resp_dir / "Responder.py").touch() + + mod = ResponderManager(mock_bus, mock_state, config) + mod.start() + return mod + + +@pytest.fixture +def hash_log_dir(responder, tmp_path): + """Return the Responder logs directory.""" + return tmp_path / "Responder" / "logs" + + +# --------------------------------------------------------------------------- +# Structure +# --------------------------------------------------------------------------- + +class TestResponderStructure: + + def test_is_base_module(self): + assert issubclass(ResponderManager, BaseModule) + + def test_module_attributes(self): + assert ResponderManager.name == "responder_mgr" + assert ResponderManager.module_type == "active" + assert ResponderManager.requires_root is True + + +# --------------------------------------------------------------------------- +# Regex patterns +# --------------------------------------------------------------------------- + +class TestHashRegex: + + def test_hash_file_pattern_matches_ntlmv2(self): + assert HASH_FILE_PATTERN.search("SMB-NTLMv2-10.0.0.5.txt") + + def test_hash_file_pattern_matches_ntlmv1(self): + assert HASH_FILE_PATTERN.search("HTTP-NTLMv1-10.0.0.5.txt") + + def test_hash_file_pattern_rejects_other(self): + assert not HASH_FILE_PATTERN.search("Responder-Session.log") + + def test_ntlmv2_regex_parses_valid_hash(self): + line = "admin::CORP:aabbccdd11223344:eeff00112233445566778899aabbccdd:0011223344556677" + m = NTLMV2_REGEX.match(line) + assert m is not None + assert m.group("username") == "admin" + assert m.group("domain") == "CORP" + + def test_ntlmv1_regex_parses_valid_hash(self): + line = "user1::DOMAIN:aabbccdd11223344:eeff00112233445566778899aabbccdd:0011223344556677" + m = NTLMV1_REGEX.match(line) + assert m is not None + assert m.group("username") == "user1" + assert m.group("domain") == "DOMAIN" + + def test_ntlmv2_regex_rejects_garbage(self): + assert NTLMV2_REGEX.match("not a hash line") is None + + +# --------------------------------------------------------------------------- +# Hash line processing +# --------------------------------------------------------------------------- + +class TestHashProcessing: + + def test_process_ntlmv2_hash_line(self, responder): + line = "admin::CORP:aabbccdd11223344:eeff00112233445566778899aabbccdd:0011223344556677" + responder._process_hash_line(line, "SMB-NTLMv2-10.0.0.5.txt") + + hashes = responder.get_captured_hashes() + assert len(hashes) == 1 + assert hashes[0]["username"] == "admin" + assert hashes[0]["domain"] == "CORP" + assert hashes[0]["hash_type"] == "ntlmv2" + assert hashes[0]["hashcat_mode"] == 5600 + + def test_process_ntlmv1_hash_line(self, responder): + line = "user1::DOMAIN:aabb:ccdd:eeff" + responder._process_hash_line(line, "HTTP-NTLMv1-10.0.0.5.txt") + + hashes = responder.get_captured_hashes() + assert len(hashes) == 1 + assert hashes[0]["hash_type"] == "ntlmv1" + assert hashes[0]["hashcat_mode"] == 5500 + + def test_process_unparseable_line(self, responder): + """Lines without :: still get processed with username='unknown'.""" + responder._process_hash_line("garbage", "SMB-NTLMv2-test.txt") + hashes = responder.get_captured_hashes() + assert len(hashes) == 1 + assert hashes[0]["username"] == "unknown" + + +# --------------------------------------------------------------------------- +# Hash file scanning +# --------------------------------------------------------------------------- + +class TestHashFileScanning: + + def test_scan_hash_files_discovers_new_hashes(self, responder, hash_log_dir): + hash_file = hash_log_dir / "SMB-NTLMv2-10.0.0.5.txt" + hash_file.write_text( + "admin::CORP:aa:bb:cc\n" + "user2::CORP:dd:ee:ff\n" + ) + responder._scan_hash_files() + assert len(responder.get_captured_hashes()) == 2 + + def test_scan_deduplicates_seen_hashes(self, responder, hash_log_dir): + hash_file = hash_log_dir / "SMB-NTLMv2-10.0.0.5.txt" + hash_file.write_text("admin::CORP:aa:bb:cc\n") + + responder._scan_hash_files() + responder._scan_hash_files() + + assert len(responder.get_captured_hashes()) == 1 + + def test_scan_ignores_empty_lines(self, responder, hash_log_dir): + hash_file = hash_log_dir / "SMB-NTLMv2-10.0.0.5.txt" + hash_file.write_text("\n\nadmin::CORP:aa:bb:cc\n\n") + + responder._scan_hash_files() + assert len(responder.get_captured_hashes()) == 1 + + +# --------------------------------------------------------------------------- +# Responder.conf generation +# --------------------------------------------------------------------------- + +class TestConfGeneration: + + def test_fallback_conf_written(self, responder, tmp_path): + responder._write_responder_conf() + conf_path = tmp_path / "Responder" / "Responder.conf" + assert conf_path.exists() + content = conf_path.read_text() + assert "SMB = On" in content + assert "HTTP = On" in content + + def test_protocol_toggles_reflected(self, responder, tmp_path): + responder._protocols["SMB"] = False + responder._protocols["HTTP"] = False + responder._write_responder_conf() + + conf_path = tmp_path / "Responder" / "Responder.conf" + content = conf_path.read_text() + assert "SMB = Off" in content + assert "HTTP = Off" in content + + +# --------------------------------------------------------------------------- +# Relay target exclusion +# --------------------------------------------------------------------------- + +class TestRelayExclusion: + + def test_set_relay_targets(self, responder): + responder.set_relay_targets({"10.0.0.1", "10.0.0.2"}) + assert "10.0.0.1" in responder._relay_targets + assert "10.0.0.2" in responder._relay_targets + + def test_set_relay_targets_replaces_previous(self, responder): + responder.set_relay_targets({"10.0.0.1"}) + responder.set_relay_targets({"10.0.0.2"}) + assert "10.0.0.1" not in responder._relay_targets + assert "10.0.0.2" in responder._relay_targets + + +# --------------------------------------------------------------------------- +# Configure +# --------------------------------------------------------------------------- + +class TestConfigure: + + def test_configure_updates_interface(self, responder): + responder.configure({"interface": "wlan0"}) + assert responder._iface == "wlan0" + + def test_configure_updates_relay_targets(self, responder): + responder.configure({"relay_targets": ["10.0.0.5"]}) + assert "10.0.0.5" in responder._relay_targets + + def test_configure_updates_protocols(self, responder): + responder.configure({"protocols": {"FTP": True, "SMB": False}}) + assert responder._protocols["FTP"] is True + assert responder._protocols["SMB"] is False + + +# --------------------------------------------------------------------------- +# Status +# --------------------------------------------------------------------------- + +class TestStatus: + + def test_status_when_running(self, responder): + s = responder.status() + assert s["running"] is True + assert s["responder_running"] is False + assert s["captured_hash_count"] == 0 + assert s["interface"] == "eth0" + + def test_status_reflects_captured_hashes(self, responder): + responder._process_hash_line("admin::CORP:aa:bb:cc", "SMB-NTLMv2-test.txt") + s = responder.status() + assert s["captured_hash_count"] == 1 + + def test_status_shows_relay_targets(self, responder): + responder.set_relay_targets({"10.0.0.1"}) + s = responder.status() + assert "10.0.0.1" in s["relay_targets"]