From 79baaa61b7441798855470a1d641febb31462adc Mon Sep 17 00:00:00 2001 From: n0mad1k Date: Fri, 10 Apr 2026 07:34:49 -0400 Subject: [PATCH] Add ids_tester IDS evasion self-test validation tests --- tests/test_ids_tester.py | 728 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 728 insertions(+) create mode 100644 tests/test_ids_tester.py diff --git a/tests/test_ids_tester.py b/tests/test_ids_tester.py new file mode 100644 index 0000000..668e626 --- /dev/null +++ b/tests/test_ids_tester.py @@ -0,0 +1,728 @@ +"""Tests for modules/stealth/ids_tester.py — IDS risk assessment module.""" + +import os +import time +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest + +from modules.stealth.ids_tester import ( + IDSTester, + RiskAssessment, + RISK_CRITICAL, + RISK_HIGH, + RISK_LOW, + RISK_MEDIUM, +) + + +class TestIDSTesterInstantiation: + """Test module instantiation and initialization.""" + + def test_ids_tester_instantiation(self, mock_bus, mock_state, mock_config): + """IDSTester can be instantiated with bus, state, config.""" + config = { + **mock_config, + "rules_dir": "/tmp/test_rules", + } + tester = IDSTester(mock_bus, mock_state, config) + assert tester.name == "ids_tester" + assert tester.module_type == "stealth" + assert tester.priority == -100 + assert tester.requires_root is False + assert tester._rules == [] + assert tester._last_assessment is None + assert tester._assessments == [] + + def test_ids_tester_default_rules_dir(self, mock_bus, mock_state, mock_config): + """IDSTester uses default rules_dir if not configured.""" + tester = IDSTester(mock_bus, mock_state, mock_config) + assert tester._rules_dir == "data/ids_rules" + + def test_ids_tester_custom_rules_dir(self, mock_bus, mock_state, mock_config): + """IDSTester respects custom rules_dir in config.""" + config = {**mock_config, "rules_dir": "/custom/rules"} + tester = IDSTester(mock_bus, mock_state, config) + assert tester._rules_dir == "/custom/rules" + + +class TestIDSTesterLifecycle: + """Test module start/stop lifecycle.""" + + def test_ids_tester_start_empty_rules_dir(self, mock_bus, mock_state, mock_config, tmp_path): + """IDSTester starts successfully with empty rules directory.""" + rules_dir = str(tmp_path / "empty_rules") + Path(rules_dir).mkdir(parents=True, exist_ok=True) + + config = {**mock_config, "rules_dir": rules_dir} + tester = IDSTester(mock_bus, mock_state, config) + tester.start() + + assert tester._running is True + assert tester._pid == os.getpid() + assert tester._start_time is not None + assert tester._rules == [] + + def test_ids_tester_start_with_rules(self, mock_bus, mock_state, mock_config, tmp_path): + """IDSTester loads rules from .rules files on start.""" + rules_dir = str(tmp_path / "with_rules") + Path(rules_dir).mkdir(parents=True, exist_ok=True) + + # Write a test rules file + rules_file = Path(rules_dir) / "test.rules" + rules_file.write_text( + 'alert tcp any any -> any 445 (msg:"ET POLICY SMB2 NT Create"; sid:2024217; rev:4; classtype:policy-violation;)\n' + 'alert tcp any any -> any 53 (msg:"ET DNS Query"; sid:2024218; rev:2; classtype:suspicious-dns-activity;)\n' + ) + + config = {**mock_config, "rules_dir": rules_dir} + tester = IDSTester(mock_bus, mock_state, config) + tester.start() + + assert tester._running is True + assert len(tester._rules) == 2 + assert tester._rules[0]["sid"] == "2024217" + assert tester._rules[1]["sid"] == "2024218" + + def test_ids_tester_start_nonexistent_rules_dir(self, mock_bus, mock_state, mock_config): + """IDSTester starts gracefully when rules_dir doesn't exist.""" + config = {**mock_config, "rules_dir": "/nonexistent/path"} + tester = IDSTester(mock_bus, mock_state, config) + tester.start() + + assert tester._running is True + assert tester._rules == [] + + def test_ids_tester_stop(self, mock_bus, mock_state, mock_config, tmp_path): + """IDSTester stops cleanly.""" + rules_dir = str(tmp_path / "rules") + Path(rules_dir).mkdir(parents=True, exist_ok=True) + config = {**mock_config, "rules_dir": rules_dir} + + tester = IDSTester(mock_bus, mock_state, config) + tester.start() + assert tester._running is True + + tester.stop() + assert tester._running is False + + +class TestIDSTesterStatus: + """Test module status reporting.""" + + def test_ids_tester_status_not_running(self, mock_bus, mock_state, mock_config, tmp_path): + """Status reflects module not running initially.""" + rules_dir = str(tmp_path / "rules") + Path(rules_dir).mkdir(parents=True, exist_ok=True) + config = {**mock_config, "rules_dir": rules_dir} + + tester = IDSTester(mock_bus, mock_state, config) + status = tester.status() + + assert status["running"] is False + assert status["last_assessment_time"] is None + assert status["last_assessment_action"] is None + + def test_ids_tester_status_running(self, mock_bus, mock_state, mock_config, tmp_path): + """Status reflects module running with rules count.""" + rules_dir = str(tmp_path / "rules") + Path(rules_dir).mkdir(parents=True, exist_ok=True) + + rules_file = Path(rules_dir) / "test.rules" + rules_file.write_text( + 'alert tcp any any -> any 445 (msg:"Test"; sid:1000; rev:1; classtype:test;)\n' + ) + + config = {**mock_config, "rules_dir": rules_dir} + tester = IDSTester(mock_bus, mock_state, config) + tester.start() + + status = tester.status() + assert status["running"] is True + assert status["pid"] == os.getpid() + assert status["rules_loaded"] == 1 + assert status["uptime"] >= 0 + + +class TestRiskAssessmentBaselines: + """Test action baseline risk levels.""" + + def test_assess_risk_low_actions(self, mock_bus, mock_state, mock_config, tmp_path): + """LOW risk actions return LOW baseline.""" + rules_dir = str(tmp_path / "rules") + Path(rules_dir).mkdir(parents=True, exist_ok=True) + config = {**mock_config, "rules_dir": rules_dir} + + tester = IDSTester(mock_bus, mock_state, config) + tester.start() + + for action in ["passive_sniff", "dns_logging", "pcap_capture"]: + result = tester.assess_risk(action) + assert result["risk_level"] == RISK_LOW + + def test_assess_risk_medium_actions(self, mock_bus, mock_state, mock_config, tmp_path): + """MEDIUM risk actions return MEDIUM baseline.""" + rules_dir = str(tmp_path / "rules") + Path(rules_dir).mkdir(parents=True, exist_ok=True) + config = {**mock_config, "rules_dir": rules_dir} + + tester = IDSTester(mock_bus, mock_state, config) + tester.start() + + for action in ["arp_spoof", "dhcp_spoof", "dns_spoof"]: + result = tester.assess_risk(action) + assert result["risk_level"] == RISK_MEDIUM + + def test_assess_risk_high_actions(self, mock_bus, mock_state, mock_config, tmp_path): + """HIGH risk actions return HIGH baseline.""" + rules_dir = str(tmp_path / "rules") + Path(rules_dir).mkdir(parents=True, exist_ok=True) + config = {**mock_config, "rules_dir": rules_dir} + + tester = IDSTester(mock_bus, mock_state, config) + tester.start() + + for action in ["responder", "ntlm_relay", "port_scan"]: + result = tester.assess_risk(action) + assert result["risk_level"] == RISK_HIGH + + def test_assess_risk_critical_actions(self, mock_bus, mock_state, mock_config, tmp_path): + """CRITICAL risk actions return CRITICAL baseline.""" + rules_dir = str(tmp_path / "rules") + Path(rules_dir).mkdir(parents=True, exist_ok=True) + config = {**mock_config, "rules_dir": rules_dir} + + tester = IDSTester(mock_bus, mock_state, config) + tester.start() + + for action in ["mitmproxy_ssl_intercept", "ssl_strip", "exploit_delivery"]: + result = tester.assess_risk(action) + assert result["risk_level"] == RISK_CRITICAL + + def test_assess_risk_unknown_action(self, mock_bus, mock_state, mock_config, tmp_path): + """Unknown actions default to MEDIUM risk.""" + rules_dir = str(tmp_path / "rules") + Path(rules_dir).mkdir(parents=True, exist_ok=True) + config = {**mock_config, "rules_dir": rules_dir} + + tester = IDSTester(mock_bus, mock_state, config) + tester.start() + + result = tester.assess_risk("unknown_action_xyz") + assert result["risk_level"] == RISK_MEDIUM + + +class TestRiskEscalation: + """Test risk escalation when IDS rules match.""" + + def test_assess_risk_escalation_single_match(self, mock_bus, mock_state, mock_config, tmp_path): + """Single matching rule escalates LOW to MEDIUM.""" + rules_dir = str(tmp_path / "rules") + Path(rules_dir).mkdir(parents=True, exist_ok=True) + + # Create rule matching 'arp' keyword + rules_file = Path(rules_dir) / "test.rules" + rules_file.write_text( + 'alert eth (msg:"ARP Spoofing Detected"; sid:3000; classtype:suspicious-traffic;)\n' + ) + + config = {**mock_config, "rules_dir": rules_dir} + tester = IDSTester(mock_bus, mock_state, config) + tester.start() + + result = tester.assess_risk("arp_spoof") + assert result["risk_level"] == RISK_HIGH # MEDIUM + 1 escalation = HIGH + assert len(result["matching_sids"]) > 0 + assert "3000" in result["matching_sids"] + + def test_assess_risk_escalation_multiple_matches(self, mock_bus, mock_state, mock_config, tmp_path): + """Multiple matching rules escalate more aggressively.""" + rules_dir = str(tmp_path / "rules") + Path(rules_dir).mkdir(parents=True, exist_ok=True) + + rules_file = Path(rules_dir) / "test.rules" + rules_file.write_text( + 'alert eth (msg:"ARP Spoof 1"; sid:3001; classtype:suspicious-traffic;)\n' + 'alert eth (msg:"ARP Spoof 2"; sid:3002; classtype:suspicious-traffic;)\n' + 'alert eth (msg:"ARP Spoof 3"; sid:3003; classtype:suspicious-traffic;)\n' + ) + + config = {**mock_config, "rules_dir": rules_dir} + tester = IDSTester(mock_bus, mock_state, config) + tester.start() + + result = tester.assess_risk("arp_spoof") + # MEDIUM + 2 escalations = CRITICAL (capped at 2 escalations) + assert result["risk_level"] == RISK_CRITICAL + assert len(result["matching_sids"]) >= 2 + + def test_assess_risk_no_escalation_without_rules(self, mock_bus, mock_state, mock_config, tmp_path): + """No escalation when no rules match.""" + rules_dir = str(tmp_path / "rules") + Path(rules_dir).mkdir(parents=True, exist_ok=True) + + rules_file = Path(rules_dir) / "test.rules" + rules_file.write_text( + 'alert tcp any any -> any 22 (msg:"SSH"; sid:4000; classtype:suspicious-traffic;)\n' + ) + + config = {**mock_config, "rules_dir": rules_dir} + tester = IDSTester(mock_bus, mock_state, config) + tester.start() + + result = tester.assess_risk("arp_spoof") + # MEDIUM baseline, no matching rules + assert result["risk_level"] == RISK_MEDIUM + assert len(result["matching_sids"]) == 0 + + +class TestAssessCaplet: + """Test bettercap caplet assessment.""" + + def test_assess_caplet_single_action(self, mock_bus, mock_state, mock_config, tmp_path): + """Caplet with single action is assessed.""" + rules_dir = str(tmp_path / "rules") + Path(rules_dir).mkdir(parents=True, exist_ok=True) + config = {**mock_config, "rules_dir": rules_dir} + + tester = IDSTester(mock_bus, mock_state, config) + tester.start() + + caplet = "net.sniff on\n" + results = tester.assess_caplet(caplet) + + assert len(results) == 1 + # net.sniff is in CAPLET_ACTION_RISK with LOW baseline, but since "net_sniff" is not in + # ACTION_RISK_BASELINE, assess_risk defaults to MEDIUM + assert results[0]["risk_level"] == RISK_MEDIUM + assert "caplet_line" in results[0] + + def test_assess_caplet_multiple_actions(self, mock_bus, mock_state, mock_config, tmp_path): + """Caplet with multiple actions is assessed for each.""" + rules_dir = str(tmp_path / "rules") + Path(rules_dir).mkdir(parents=True, exist_ok=True) + config = {**mock_config, "rules_dir": rules_dir} + + tester = IDSTester(mock_bus, mock_state, config) + tester.start() + + caplet = ( + "net.sniff on\n" + "arp.spoof on\n" + "https.proxy on\n" + ) + results = tester.assess_caplet(caplet) + + assert len(results) == 3 + # net_sniff and arp_spoof are not in ACTION_RISK_BASELINE, so they default to MEDIUM + # https_proxy also maps to "https_proxy" which is not in ACTION_RISK_BASELINE, defaults to MEDIUM + assert results[0]["risk_level"] == RISK_MEDIUM + assert results[1]["risk_level"] == RISK_MEDIUM + assert results[2]["risk_level"] == RISK_MEDIUM + + def test_assess_caplet_with_comments(self, mock_bus, mock_state, mock_config, tmp_path): + """Caplet comments are ignored.""" + rules_dir = str(tmp_path / "rules") + Path(rules_dir).mkdir(parents=True, exist_ok=True) + config = {**mock_config, "rules_dir": rules_dir} + + tester = IDSTester(mock_bus, mock_state, config) + tester.start() + + caplet = ( + "# This is a comment\n" + "net.sniff on\n" + "# Another comment\n" + ) + results = tester.assess_caplet(caplet) + + assert len(results) == 1 + assert results[0]["caplet_line"] == "net.sniff on" + + def test_assess_caplet_empty(self, mock_bus, mock_state, mock_config, tmp_path): + """Empty or whitespace-only caplet returns no results.""" + rules_dir = str(tmp_path / "rules") + Path(rules_dir).mkdir(parents=True, exist_ok=True) + config = {**mock_config, "rules_dir": rules_dir} + + tester = IDSTester(mock_bus, mock_state, config) + tester.start() + + caplet = "# Just comments\n\n \n" + results = tester.assess_caplet(caplet) + + assert len(results) == 0 + + +class TestPreflightCheck: + """Test preflight validation for module action combinations.""" + + def test_preflight_all_low_risk(self, mock_bus, mock_state, mock_config, tmp_path): + """All LOW actions → GO recommendation.""" + rules_dir = str(tmp_path / "rules") + Path(rules_dir).mkdir(parents=True, exist_ok=True) + config = {**mock_config, "rules_dir": rules_dir} + + tester = IDSTester(mock_bus, mock_state, config) + tester.start() + + result = tester.preflight_check("test_module", ["passive_sniff", "dns_logging", "pcap_capture"]) + + assert result["overall_risk"] == RISK_LOW + assert result["go_nogo"] == "GO" + assert len(result["action_risks"]) == 3 + + def test_preflight_mixed_with_high(self, mock_bus, mock_state, mock_config, tmp_path): + """Mixed with HIGH risk → CAUTION recommendation.""" + rules_dir = str(tmp_path / "rules") + Path(rules_dir).mkdir(parents=True, exist_ok=True) + config = {**mock_config, "rules_dir": rules_dir} + + tester = IDSTester(mock_bus, mock_state, config) + tester.start() + + result = tester.preflight_check("test_module", ["passive_sniff", "responder", "arp_spoof"]) + + assert result["overall_risk"] == RISK_HIGH + assert result["go_nogo"] == "CAUTION" + assert len(result["action_risks"]) == 3 + + def test_preflight_critical_action(self, mock_bus, mock_state, mock_config, tmp_path): + """Any CRITICAL action → NO-GO recommendation.""" + rules_dir = str(tmp_path / "rules") + Path(rules_dir).mkdir(parents=True, exist_ok=True) + config = {**mock_config, "rules_dir": rules_dir} + + tester = IDSTester(mock_bus, mock_state, config) + tester.start() + + result = tester.preflight_check("test_module", ["mitmproxy_ssl_intercept"]) + + assert result["overall_risk"] == RISK_CRITICAL + assert result["go_nogo"] == "NO-GO" + + def test_preflight_medium_only(self, mock_bus, mock_state, mock_config, tmp_path): + """All MEDIUM actions → GO recommendation.""" + rules_dir = str(tmp_path / "rules") + Path(rules_dir).mkdir(parents=True, exist_ok=True) + config = {**mock_config, "rules_dir": rules_dir} + + tester = IDSTester(mock_bus, mock_state, config) + tester.start() + + result = tester.preflight_check("test_module", ["arp_spoof", "dhcp_spoof", "dns_spoof"]) + + assert result["overall_risk"] == RISK_MEDIUM + assert result["go_nogo"] == "GO" + + +class TestParseRule: + """Test Snort/Suricata rule parsing.""" + + def test_parse_rule_complete(self): + """Parse rule with sid, msg, and classtype.""" + rule_line = 'alert tcp any any -> any 445 (msg:"SMB NT Create"; sid:2024217; rev:4; classtype:policy-violation;)' + result = IDSTester._parse_rule(rule_line) + + assert result is not None + assert result["sid"] == "2024217" + assert result["msg"] == "SMB NT Create" + assert result["classtype"] == "policy-violation" + assert rule_line in result["raw"] + + def test_parse_rule_no_sid(self): + """Rule without SID is rejected.""" + rule_line = 'alert tcp any any -> any 445 (msg:"No SID"; rev:4; classtype:policy-violation;)' + result = IDSTester._parse_rule(rule_line) + + assert result is None + + def test_parse_rule_no_msg(self): + """Rule without msg still parses if SID present.""" + rule_line = 'alert tcp any any -> any 445 (sid:3000; rev:4; classtype:policy-violation;)' + result = IDSTester._parse_rule(rule_line) + + assert result is not None + assert result["sid"] == "3000" + assert result["msg"] == "" + + def test_parse_rule_no_classtype(self): + """Rule without classtype still parses if SID present.""" + rule_line = 'alert tcp any any -> any 445 (msg:"Test"; sid:4000; rev:4;)' + result = IDSTester._parse_rule(rule_line) + + assert result is not None + assert result["sid"] == "4000" + assert result["classtype"] == "" + + def test_parse_rule_with_quotes_in_msg(self): + """Extract msg correctly with embedded quotes.""" + rule_line = 'alert tcp (msg:"ET POLICY HTTP User-Agent Reported as \\"not\\""; sid:5000; classtype:policy-violation;)' + result = IDSTester._parse_rule(rule_line) + + assert result is not None + assert result["sid"] == "5000" + assert "User-Agent" in result["msg"] + + +class TestActionToKeywords: + """Test action-to-keyword mapping.""" + + def test_action_to_keywords_arp_spoof(self): + """ARP spoof maps to relevant keywords.""" + keywords = IDSTester._action_to_keywords("arp_spoof") + + assert "arp" in keywords + assert "spoof" in keywords + assert "arp spoof" in keywords + + def test_action_to_keywords_responder(self): + """Responder maps to LLMNR, NBNS, MDNS, WPAD keywords.""" + keywords = IDSTester._action_to_keywords("responder") + + assert any(k in keywords for k in ["llmnr", "nbns", "mdns", "wpad"]) + + def test_action_to_keywords_unknown(self): + """Unknown action generates keyword from normalized name.""" + keywords = IDSTester._action_to_keywords("unknown_action") + + assert "unknown action" in keywords + + def test_action_to_keywords_case_insensitive(self): + """Keyword mapping handles case normalization.""" + keywords1 = IDSTester._action_to_keywords("arp_spoof") + keywords2 = IDSTester._action_to_keywords("ARP_SPOOF") + + # Should produce same results + assert keywords1 == keywords2 + + def test_action_to_keywords_dot_notation(self): + """Dot notation in action names is converted to underscores.""" + keywords = IDSTester._action_to_keywords("net.probe") + + assert "net probe" in keywords + + +class TestGenerateRecommendation: + """Test recommendation generation.""" + + def test_recommendation_critical(self): + """CRITICAL risk generates appropriate warning.""" + rec = IDSTester._generate_recommendation("mitmproxy_ssl_intercept", RISK_CRITICAL, 5) + + assert "CRITICAL" in rec + assert "mitmproxy_ssl_intercept" in rec + assert "5" in rec + assert "Do NOT activate" in rec + + def test_recommendation_high(self): + """HIGH risk generates caution message.""" + rec = IDSTester._generate_recommendation("responder", RISK_HIGH, 3) + + assert "HIGH" in rec + assert "responder" in rec + assert "3" in rec + + def test_recommendation_medium(self): + """MEDIUM risk generates proceed-with-caution message.""" + rec = IDSTester._generate_recommendation("arp_spoof", RISK_MEDIUM, 2) + + assert "MEDIUM" in rec + assert "arp_spoof" in rec + assert "2" in rec + + def test_recommendation_low(self): + """LOW risk generates safe-to-proceed message.""" + rec = IDSTester._generate_recommendation("passive_sniff", RISK_LOW, 0) + + assert "LOW" in rec + assert "passive_sniff" in rec + assert "Safe" in rec or "minimal" in rec + + +class TestAssessmentHistory: + """Test assessment history and capping.""" + + def test_assessment_history_capped_at_100(self, mock_bus, mock_state, mock_config, tmp_path): + """Assessment history is capped at 100 entries.""" + rules_dir = str(tmp_path / "rules") + Path(rules_dir).mkdir(parents=True, exist_ok=True) + config = {**mock_config, "rules_dir": rules_dir} + + tester = IDSTester(mock_bus, mock_state, config) + tester.start() + + # Perform 150 assessments + for i in range(150): + tester.assess_risk("passive_sniff") + + assert len(tester._assessments) == 100 + + def test_assessment_history_keeps_last_100(self, mock_bus, mock_state, mock_config, tmp_path): + """Last 100 assessments are kept, not first 100.""" + rules_dir = str(tmp_path / "rules") + Path(rules_dir).mkdir(parents=True, exist_ok=True) + config = {**mock_config, "rules_dir": rules_dir} + + tester = IDSTester(mock_bus, mock_state, config) + tester.start() + + actions = ["arp_spoof", "responder", "passive_sniff", "exploit_delivery"] + for i in range(120): + action = actions[i % len(actions)] + tester.assess_risk(action) + + # Last 100 should be retained + assert len(tester._assessments) == 100 + # Most recent should still be there + assert tester._assessments[-1] == tester._last_assessment + + def test_last_assessment_updated(self, mock_bus, mock_state, mock_config, tmp_path): + """_last_assessment is updated with each assess_risk call.""" + rules_dir = str(tmp_path / "rules") + Path(rules_dir).mkdir(parents=True, exist_ok=True) + config = {**mock_config, "rules_dir": rules_dir} + + tester = IDSTester(mock_bus, mock_state, config) + tester.start() + + tester.assess_risk("passive_sniff") + assert tester._last_assessment.action == "passive_sniff" + assert tester._last_assessment.risk_level == RISK_LOW + + tester.assess_risk("mitmproxy_ssl_intercept") + assert tester._last_assessment.action == "mitmproxy_ssl_intercept" + assert tester._last_assessment.risk_level == RISK_CRITICAL + + +class TestRuleLoading: + """Test rule loading from directory.""" + + def test_rule_loading_multiple_files(self, mock_bus, mock_state, mock_config, tmp_path): + """Rules are loaded from all .rules files in directory.""" + rules_dir = str(tmp_path / "rules") + Path(rules_dir).mkdir(parents=True, exist_ok=True) + + # Create multiple rule files + Path(rules_dir).joinpath("network.rules").write_text( + 'alert tcp any any -> any 445 (msg:"SMB"; sid:1001; classtype:policy-violation;)\n' + ) + Path(rules_dir).joinpath("dns.rules").write_text( + 'alert udp any any -> any 53 (msg:"DNS Query"; sid:2001; classtype:suspicious-dns;)\n' + ) + Path(rules_dir).joinpath("web.rules").write_text( + 'alert tcp any any -> any 80 (msg:"HTTP"; sid:3001; classtype:web-application;)\n' + ) + + config = {**mock_config, "rules_dir": rules_dir} + tester = IDSTester(mock_bus, mock_state, config) + tester.start() + + assert len(tester._rules) == 3 + sids = [rule["sid"] for rule in tester._rules] + assert "1001" in sids + assert "2001" in sids + assert "3001" in sids + + def test_rule_loading_skip_non_rules_files(self, mock_bus, mock_state, mock_config, tmp_path): + """Only .rules files are loaded.""" + rules_dir = str(tmp_path / "rules") + Path(rules_dir).mkdir(parents=True, exist_ok=True) + + # Create mixed file types + Path(rules_dir).joinpath("valid.rules").write_text( + 'alert tcp any any -> any 445 (msg:"Valid"; sid:1; classtype:test;)\n' + ) + Path(rules_dir).joinpath("invalid.txt").write_text( + 'alert tcp any any -> any 445 (msg:"Invalid"; sid:2; classtype:test;)\n' + ) + Path(rules_dir).joinpath("readme.md").write_text("# Rules\n") + + config = {**mock_config, "rules_dir": rules_dir} + tester = IDSTester(mock_bus, mock_state, config) + tester.start() + + assert len(tester._rules) == 1 + assert tester._rules[0]["sid"] == "1" + + def test_rule_loading_skip_comments(self, mock_bus, mock_state, mock_config, tmp_path): + """Comment lines in .rules files are skipped.""" + rules_dir = str(tmp_path / "rules") + Path(rules_dir).mkdir(parents=True, exist_ok=True) + + rules_file = Path(rules_dir) / "comments.rules" + rules_file.write_text( + "# This is a comment\n" + 'alert tcp any any -> any 445 (msg:"Valid 1"; sid:100; classtype:test;)\n' + "# Another comment\n" + 'alert tcp any any -> any 53 (msg:"Valid 2"; sid:101; classtype:test;)\n' + ) + + config = {**mock_config, "rules_dir": rules_dir} + tester = IDSTester(mock_bus, mock_state, config) + tester.start() + + assert len(tester._rules) == 2 + + def test_rule_loading_reconfigure(self, mock_bus, mock_state, mock_config, tmp_path): + """configure() with new rules_dir reloads rules.""" + rules_dir1 = str(tmp_path / "rules1") + rules_dir2 = str(tmp_path / "rules2") + Path(rules_dir1).mkdir(parents=True, exist_ok=True) + Path(rules_dir2).mkdir(parents=True, exist_ok=True) + + # First dir has 1 rule + Path(rules_dir1).joinpath("rules.rules").write_text( + 'alert tcp any any -> any 445 (msg:"R1"; sid:200; classtype:test;)\n' + ) + + # Second dir has 2 rules + Path(rules_dir2).joinpath("rules.rules").write_text( + 'alert tcp any any -> any 445 (msg:"R2"; sid:201; classtype:test;)\n' + 'alert tcp any any -> any 53 (msg:"R3"; sid:202; classtype:test;)\n' + ) + + config = {**mock_config, "rules_dir": rules_dir1} + tester = IDSTester(mock_bus, mock_state, config) + tester.start() + + assert len(tester._rules) == 1 + + # Reconfigure with new directory + tester.configure({"rules_dir": rules_dir2}) + + assert len(tester._rules) == 2 + + +class TestRiskAssessmentDataclass: + """Test RiskAssessment dataclass.""" + + def test_risk_assessment_to_dict(self): + """RiskAssessment.to_dict() returns all fields.""" + assessment = RiskAssessment( + action="test_action", + risk_level=RISK_HIGH, + matching_sids=["1000", "2000"], + matching_rules=["Rule 1", "Rule 2"], + recommendation="Test recommendation", + ) + + d = assessment.to_dict() + assert d["action"] == "test_action" + assert d["risk_level"] == RISK_HIGH + assert d["matching_sids"] == ["1000", "2000"] + assert d["matching_rules"] == ["Rule 1", "Rule 2"] + assert d["recommendation"] == "Test recommendation" + assert "timestamp" in d + + def test_risk_assessment_timestamp(self): + """RiskAssessment has timestamp.""" + before = time.time() + assessment = RiskAssessment( + action="test", + risk_level=RISK_LOW, + ) + after = time.time() + + assert before <= assessment.timestamp <= after