Add responder_mgr NTLM hash capture validation tests

2026-04-10 07:37:22 -04:00
parent 7aab56cbe0
commit fded0ff4e7
1 changed files with 247 additions and 0 deletions
@@ -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"]