n0mad1k
c04ec0b9ea
Cover packet_capture (rotation/zstd/AES), credential_db (dedup/hashcat/bulk), ja3_spoofer (profiles/cipher mapping/randomization), responder_mgr (hash capture/dedup/conf gen), ntlm_relay (protocol inference/command building/ coordination), ids_tester (risk assessment/preflight/caplet), and bridge (setup/teardown/ebtables/watchdog). 187 tests total, all passing.
355 lines
12 KiB
Python
355 lines
12 KiB
Python
"""Tests for modules/stealth/ids_tester — risk assessment, rule loading, preflight checks.
|
|
|
|
Validates action risk assessment, IDS rule parsing and matching,
|
|
caplet assessment, preflight go/nogo decisions, and risk escalation
|
|
without requiring Snort/Suricata 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.stealth.ids_tester import (
|
|
IDSTester, RiskAssessment,
|
|
ACTION_RISK_BASELINE, CAPLET_ACTION_RISK,
|
|
RISK_LOW, RISK_MEDIUM, RISK_HIGH, RISK_CRITICAL,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Fixtures
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@pytest.fixture
|
|
def ids(mock_bus, mock_state, tmp_path):
|
|
"""Return a started IDSTester with no rules loaded."""
|
|
config = {"rules_dir": str(tmp_path / "rules")}
|
|
mod = IDSTester(mock_bus, mock_state, config)
|
|
mod.start()
|
|
return mod
|
|
|
|
|
|
@pytest.fixture
|
|
def ids_with_rules(mock_bus, mock_state, tmp_path):
|
|
"""Return a started IDSTester with sample Snort rules."""
|
|
rules_dir = tmp_path / "rules"
|
|
rules_dir.mkdir()
|
|
|
|
(rules_dir / "local.rules").write_text(
|
|
'alert tcp any any -> any any (msg:"LLMNR Spoof Detected"; sid:1000001; classtype:bad-unknown;)\n'
|
|
'alert udp any any -> any any (msg:"NBT-NS Poisoning"; sid:1000002; classtype:bad-unknown;)\n'
|
|
'alert tcp any any -> any 445 (msg:"SMB Relay Detected"; sid:1000003; classtype:attempted-admin;)\n'
|
|
'alert tcp any any -> any 80 (msg:"HTTP MITM Detected"; sid:1000004; classtype:web-application-attack;)\n'
|
|
'# This is a comment line\n'
|
|
'\n'
|
|
)
|
|
config = {"rules_dir": str(rules_dir)}
|
|
mod = IDSTester(mock_bus, mock_state, config)
|
|
mod.start()
|
|
return mod
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Structure
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestIDSTesterStructure:
|
|
|
|
def test_is_base_module(self):
|
|
assert issubclass(IDSTester, BaseModule)
|
|
|
|
def test_module_attributes(self):
|
|
assert IDSTester.name == "ids_tester"
|
|
assert IDSTester.module_type == "stealth"
|
|
assert IDSTester.requires_root is False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Risk constants
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestRiskConstants:
|
|
|
|
def test_action_baselines_exist(self):
|
|
assert len(ACTION_RISK_BASELINE) > 10
|
|
|
|
def test_passive_actions_are_low(self):
|
|
for action in ["passive_sniff", "dns_logging", "pcap_capture"]:
|
|
assert ACTION_RISK_BASELINE[action] == RISK_LOW
|
|
|
|
def test_active_actions_are_high_or_critical(self):
|
|
for action in ["responder", "ntlm_relay", "mitmproxy_ssl_intercept"]:
|
|
assert ACTION_RISK_BASELINE[action] in (RISK_HIGH, RISK_CRITICAL)
|
|
|
|
def test_caplet_actions_defined(self):
|
|
assert "net.sniff" in CAPLET_ACTION_RISK
|
|
assert "arp.spoof" in CAPLET_ACTION_RISK
|
|
assert "https.proxy" in CAPLET_ACTION_RISK
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Rule loading
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestRuleLoading:
|
|
|
|
def test_no_rules_dir(self, ids):
|
|
assert ids._rules == []
|
|
|
|
def test_rules_loaded_from_file(self, ids_with_rules):
|
|
assert len(ids_with_rules._rules) == 4
|
|
|
|
def test_comments_skipped(self, ids_with_rules):
|
|
sids = [r["sid"] for r in ids_with_rules._rules]
|
|
assert all(s.isdigit() for s in sids)
|
|
|
|
def test_parsed_rule_structure(self, ids_with_rules):
|
|
rule = ids_with_rules._rules[0]
|
|
assert "sid" in rule
|
|
assert "msg" in rule
|
|
assert "raw" in rule
|
|
assert rule["sid"] == "1000001"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Rule parsing
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestRuleParsing:
|
|
|
|
def test_parse_valid_rule(self):
|
|
line = 'alert tcp any any -> any any (msg:"Test Rule"; sid:12345; classtype:trojan-activity;)'
|
|
result = IDSTester._parse_rule(line)
|
|
assert result is not None
|
|
assert result["sid"] == "12345"
|
|
assert result["msg"] == "Test Rule"
|
|
assert result["classtype"] == "trojan-activity"
|
|
|
|
def test_parse_rule_without_sid_returns_none(self):
|
|
result = IDSTester._parse_rule("alert tcp any any -> any any (msg:\"No SID\";)")
|
|
assert result is None
|
|
|
|
def test_parse_rule_without_msg(self):
|
|
result = IDSTester._parse_rule("alert tcp any any -> any any (sid:99999;)")
|
|
assert result is not None
|
|
assert result["msg"] == ""
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Risk assessment (baseline only, no rules)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestBaselineRiskAssessment:
|
|
|
|
def test_passive_action_low_risk(self, ids):
|
|
result = ids.assess_risk("passive_sniff")
|
|
assert result["risk_level"] == RISK_LOW
|
|
|
|
def test_active_action_high_risk(self, ids):
|
|
result = ids.assess_risk("responder")
|
|
assert result["risk_level"] == RISK_HIGH
|
|
|
|
def test_critical_action(self, ids):
|
|
result = ids.assess_risk("mitmproxy_ssl_intercept")
|
|
assert result["risk_level"] == RISK_CRITICAL
|
|
|
|
def test_unknown_action_medium_risk(self, ids):
|
|
result = ids.assess_risk("completely_unknown_action")
|
|
assert result["risk_level"] == RISK_MEDIUM
|
|
|
|
def test_assessment_stored(self, ids):
|
|
ids.assess_risk("arp_spoof")
|
|
assert len(ids._assessments) == 1
|
|
assert ids._last_assessment.action == "arp_spoof"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Risk escalation with rules
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestRiskEscalation:
|
|
|
|
def test_responder_escalates_with_matching_rules(self, ids_with_rules):
|
|
result = ids_with_rules.assess_risk("responder")
|
|
assert result["risk_level"] in (RISK_HIGH, RISK_CRITICAL)
|
|
assert len(result["matching_sids"]) > 0
|
|
|
|
def test_matching_sids_capped_at_20(self, mock_bus, mock_state, tmp_path):
|
|
rules_dir = tmp_path / "rules"
|
|
rules_dir.mkdir()
|
|
lines = "\n".join(
|
|
f'alert tcp any any -> any any (msg:"ARP rule {i}"; sid:{2000+i};)'
|
|
for i in range(30)
|
|
)
|
|
(rules_dir / "arp.rules").write_text(lines)
|
|
|
|
config = {"rules_dir": str(rules_dir)}
|
|
mod = IDSTester(mock_bus, mock_state, config)
|
|
mod.start()
|
|
|
|
result = mod.assess_risk("arp_spoof")
|
|
assert len(result["matching_sids"]) <= 20
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Recommendation text
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestRecommendation:
|
|
|
|
def test_critical_recommendation(self, ids):
|
|
result = ids.assess_risk("mitmproxy_ssl_intercept")
|
|
assert "CRITICAL" in result["recommendation"] or result["risk_level"] == RISK_CRITICAL
|
|
|
|
def test_low_recommendation(self, ids):
|
|
result = ids.assess_risk("passive_sniff")
|
|
assert "LOW" in result["recommendation"]
|
|
|
|
def test_recommendation_includes_action_name(self, ids):
|
|
result = ids.assess_risk("arp_spoof")
|
|
assert "arp_spoof" in result["recommendation"]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Keyword mapping
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestKeywordMapping:
|
|
|
|
def test_responder_keywords(self):
|
|
keywords = IDSTester._action_to_keywords("responder")
|
|
assert "llmnr" in keywords
|
|
assert "nbns" in keywords
|
|
|
|
def test_arp_spoof_keywords(self):
|
|
keywords = IDSTester._action_to_keywords("arp_spoof")
|
|
assert "arp" in keywords
|
|
|
|
def test_unknown_action_uses_name(self):
|
|
keywords = IDSTester._action_to_keywords("custom_action")
|
|
assert "custom action" in keywords
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Caplet assessment
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestCapletAssessment:
|
|
|
|
def test_assess_caplet_detects_actions(self, ids):
|
|
caplet = (
|
|
"# Bettercap caplet\n"
|
|
"net.sniff on\n"
|
|
"arp.spoof on\n"
|
|
)
|
|
results = ids.assess_caplet(caplet)
|
|
assert len(results) == 2
|
|
|
|
def test_assess_caplet_skips_comments(self, ids):
|
|
caplet = "# net.sniff on\n"
|
|
results = ids.assess_caplet(caplet)
|
|
assert len(results) == 0
|
|
|
|
def test_assess_caplet_includes_line(self, ids):
|
|
caplet = "arp.spoof on\n"
|
|
results = ids.assess_caplet(caplet)
|
|
assert results[0]["caplet_line"] == "arp.spoof on"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Preflight check
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestPreflightCheck:
|
|
|
|
def test_low_risk_go(self, ids):
|
|
result = ids.preflight_check("packet_capture", ["passive_sniff", "pcap_capture"])
|
|
assert result["go_nogo"] == "GO"
|
|
assert result["overall_risk"] == RISK_LOW
|
|
|
|
def test_critical_risk_nogo(self, ids):
|
|
result = ids.preflight_check("mitmproxy", ["mitmproxy_ssl_intercept"])
|
|
assert result["go_nogo"] == "NO-GO"
|
|
assert result["overall_risk"] == RISK_CRITICAL
|
|
|
|
def test_high_risk_caution(self, ids):
|
|
result = ids.preflight_check("responder_mgr", ["responder"])
|
|
assert result["go_nogo"] == "CAUTION"
|
|
assert result["overall_risk"] == RISK_HIGH
|
|
|
|
def test_preflight_returns_per_action_risks(self, ids):
|
|
result = ids.preflight_check("test_module", ["passive_sniff", "responder"])
|
|
assert len(result["action_risks"]) == 2
|
|
|
|
def test_preflight_overall_is_max(self, ids):
|
|
result = ids.preflight_check("mixed", ["passive_sniff", "mitmproxy_ssl_intercept"])
|
|
assert result["overall_risk"] == RISK_CRITICAL
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# RiskAssessment dataclass
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestRiskAssessmentDataclass:
|
|
|
|
def test_to_dict(self):
|
|
ra = RiskAssessment(
|
|
action="test",
|
|
risk_level=RISK_LOW,
|
|
matching_sids=["100"],
|
|
matching_rules=["Test rule"],
|
|
recommendation="Safe",
|
|
)
|
|
d = ra.to_dict()
|
|
assert d["action"] == "test"
|
|
assert d["risk_level"] == RISK_LOW
|
|
assert d["matching_sids"] == ["100"]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Status
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestStatus:
|
|
|
|
def test_status_running(self, ids):
|
|
s = ids.status()
|
|
assert s["running"] is True
|
|
assert s["rules_loaded"] == 0
|
|
assert s["total_assessments"] == 0
|
|
|
|
def test_status_after_assessment(self, ids):
|
|
ids.assess_risk("arp_spoof")
|
|
s = ids.status()
|
|
assert s["total_assessments"] == 1
|
|
assert s["last_assessment_action"] == "arp_spoof"
|
|
|
|
def test_status_with_rules(self, ids_with_rules):
|
|
s = ids_with_rules.status()
|
|
assert s["rules_loaded"] == 4
|
|
|
|
def test_configure_reloads_rules(self, ids, tmp_path):
|
|
new_dir = tmp_path / "new_rules"
|
|
new_dir.mkdir()
|
|
(new_dir / "test.rules").write_text(
|
|
'alert tcp any any -> any any (msg:"New"; sid:99999;)\n'
|
|
)
|
|
ids.configure({"rules_dir": str(new_dir)})
|
|
assert len(ids._rules) == 1
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Assessment history
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestAssessmentHistory:
|
|
|
|
def test_history_capped_at_100(self, ids):
|
|
for i in range(110):
|
|
ids.assess_risk("arp_spoof")
|
|
assert len(ids._assessments) == 100
|