"""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"]