178 lines
8.2 KiB
Python
178 lines
8.2 KiB
Python
"""Explicit configuration discovery and metadata-preservation checks."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import contextlib
|
|
import io
|
|
import json
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
from netscaler_triage.cli import main
|
|
from netscaler_triage.config_loading import load_config, load_configs
|
|
from netscaler_triage.matching import Matcher
|
|
|
|
|
|
class ConfigLoadingTests(unittest.TestCase):
|
|
def test_directory_expansion_is_explicit_sorted_and_deduplicated(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
(root / "sub").mkdir()
|
|
(root / ".hidden").mkdir()
|
|
(root / "__pycache__").mkdir()
|
|
(root / "z.txt").write_text("last\n")
|
|
(root / "sub/a.txt").write_text("first\n")
|
|
(root / ".hidden/a.txt").write_text("hidden\n")
|
|
(root / "__pycache__/a.txt").write_text("cache\n")
|
|
(root / "ignored.log").write_text("ignored\n")
|
|
for name in ["README.txt", "LICENSE.txt", "CHANGELOG.txt"]:
|
|
(root / name).write_text("Documentation, not an indicator.\n")
|
|
configs = load_configs([root, root / "z.txt", root])
|
|
self.assertEqual([c["indicators"][0]["value"] for c in configs], ["first", "last"])
|
|
self.assertEqual(load_configs([]), [])
|
|
|
|
def test_txt_infers_unambiguous_types_and_enforces_ip_boundaries(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
path = Path(directory) / "references.txt"
|
|
path.write_text(
|
|
"192.0.2.1\n2001:db8::1\n/example/artifact\n"
|
|
"https://example.invalid/payload\n" + "a" * 64 + "\nwhoami; id\n"
|
|
)
|
|
config = load_config(path)
|
|
self.assertEqual(
|
|
[row["type"] for row in config["indicators"]],
|
|
["ipv4", "ipv6", "path", "url", "sha256", "literal"],
|
|
)
|
|
matcher = Matcher([config], defaults=False)
|
|
self.assertTrue(matcher.search("source=192.0.2.1:443")[1])
|
|
self.assertFalse(matcher.search("source=192.0.2.10")[1])
|
|
self.assertTrue(matcher.search("source=2001:db8::1")[1])
|
|
self.assertFalse(matcher.search("source=2001:db8::10")[1])
|
|
self.assertTrue(matcher.search("whoami; id")[1])
|
|
|
|
def test_csv_preserves_metadata_and_aliases(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
path = Path(directory) / "customer.csv"
|
|
path.write_text(
|
|
"indicator_type,indicator,status,source,description,confidence,evidence\n"
|
|
"ipv4,192.0.2.9,authorized_test,local,Approved test,high,auth.log:3\n"
|
|
)
|
|
row = load_config(path)["indicators"][0]
|
|
self.assertEqual(row["value"], "192.0.2.9")
|
|
self.assertEqual(row["status"], "authorized_test")
|
|
self.assertEqual(row["evidence"], "auth.log:3")
|
|
self.assertEqual(row["source"], "local")
|
|
self.assertEqual(row["confidence"], "high")
|
|
|
|
def test_legacy_regex_csv_preserves_description(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
path = Path(directory) / "rules.csv"
|
|
path.write_text("ID,Muster,description,source\ncustom,test.*value,Example,local\n")
|
|
row = load_config(path)["patterns"][0]
|
|
self.assertEqual(row["description"], "Example")
|
|
self.assertEqual(row["pattern"], "test.*value")
|
|
|
|
def test_symlink_file_and_directory_are_rejected(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
target = root / "target"
|
|
target.mkdir()
|
|
(target / "one.txt").write_text("reference")
|
|
link = root / "linked"
|
|
link.symlink_to(target, target_is_directory=True)
|
|
with self.assertRaisesRegex(ValueError, "symlinks"):
|
|
load_configs([root])
|
|
with self.assertRaisesRegex(ValueError, "symlinks"):
|
|
load_configs([link])
|
|
link.unlink()
|
|
alias = root / "alias.txt"
|
|
alias.symlink_to(target / "one.txt")
|
|
with self.assertRaisesRegex(ValueError, "symlinks"):
|
|
load_config(alias)
|
|
|
|
def test_file_count_and_combined_bytes_are_bounded(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
(root / "a.txt").write_text("first")
|
|
(root / "b.txt").write_text("second")
|
|
with (
|
|
patch("netscaler_triage.config_loading.MAX_CONFIG_FILES", 1),
|
|
self.assertRaisesRegex(ValueError, "exceeds 1 files"),
|
|
):
|
|
load_configs([root])
|
|
with (
|
|
patch("netscaler_triage.config_loading.MAX_TOTAL_BYTES", 8),
|
|
self.assertRaisesRegex(ValueError, "Combined rule packs"),
|
|
):
|
|
load_configs([root])
|
|
with (
|
|
patch("netscaler_triage.config_loading.MAX_CONFIG_BYTES", 4),
|
|
self.assertRaisesRegex(ValueError, "exceeds"),
|
|
):
|
|
load_config(root / "a.txt")
|
|
|
|
def test_invalid_inputs_have_named_errors(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
with self.assertRaisesRegex(ValueError, "No supported rule packs"):
|
|
load_configs([root])
|
|
invalid = root / "broken.json"
|
|
invalid.write_text("{broken")
|
|
with self.assertRaisesRegex(ValueError, "broken.json"):
|
|
load_configs([root])
|
|
invalid.unlink()
|
|
csv = root / "broken.csv"
|
|
csv.write_text("value,status\none,two,three\n")
|
|
with self.assertRaisesRegex(ValueError, "column count mismatch"):
|
|
load_config(csv)
|
|
|
|
def test_sigma_relative_source_distinguishes_nested_names(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
for name in ["a", "b"]:
|
|
(root / name).mkdir()
|
|
(root / name / "rules.yml").write_text("title: example\n")
|
|
configs = load_configs([root])
|
|
self.assertEqual(
|
|
[c["sigma_rules"][0]["source"] for c in configs],
|
|
["a/rules.yml", "b/rules.yml"],
|
|
)
|
|
|
|
|
|
class RulesCommandTests(unittest.TestCase):
|
|
def test_valid_directory_compiles_without_output_artifacts(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
(root / "refs.csv").write_text("type,value,status\nipv4,192.0.2.1,reference_only\n")
|
|
before = sorted(p.name for p in root.iterdir())
|
|
output = io.StringIO()
|
|
with contextlib.redirect_stdout(output):
|
|
code = main(["rules", "--config", str(root), "--json"])
|
|
self.assertEqual(code, 0)
|
|
result = json.loads(output.getvalue())
|
|
self.assertEqual(result["status"], "valid")
|
|
self.assertGreater(result["sigma_rules"], 0)
|
|
self.assertGreaterEqual(result["explicit_active_indicators"], 1)
|
|
self.assertEqual(before, sorted(p.name for p in root.iterdir()))
|
|
|
|
def test_bad_sigma_and_duplicate_ids_fail_validation(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
first = root / "one.yml"
|
|
first.write_text("title: missing required Sigma fields\n")
|
|
with contextlib.redirect_stderr(io.StringIO()):
|
|
self.assertEqual(main(["rules", "--config", str(root)]), 2)
|
|
sigma = (
|
|
"title: Test rule\nid: d78e2aad-98f2-41fa-a552-f55aa0ad786d\n"
|
|
"logsource:\n product: netscaler\ndetection:\n"
|
|
" selection:\n raw|contains: test-marker\n condition: selection\n"
|
|
)
|
|
first.write_text(sigma)
|
|
(root / "two.yml").write_text(sigma)
|
|
stderr = io.StringIO()
|
|
with contextlib.redirect_stderr(stderr):
|
|
self.assertEqual(main(["rules", "--config", str(root)]), 2)
|
|
self.assertIn("Duplicate rule identifier", stderr.getvalue())
|