n0mad1k
248 lines
8.7 KiB
Python
248 lines
8.7 KiB
Python
"""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"]
|