53 lines
1.8 KiB
Python
53 lines
1.8 KiB
Python
import unittest
|
|
|
|
from app.policy import PolicyEngine
|
|
|
|
|
|
def event(src, dst, severity=1):
|
|
return {
|
|
"src_ip": src,
|
|
"dest_ip": dst,
|
|
"alert": {"severity": severity, "signature_id": 1234, "signature": "test"},
|
|
}
|
|
|
|
|
|
class PolicyTests(unittest.TestCase):
|
|
def test_observation_mode_selects_candidate_but_does_not_block(self):
|
|
p = PolicyEngine(False, 1, "192.168.100.0/24", "")
|
|
d = p.evaluate(event("192.168.100.10", "9.9.9.9"))
|
|
self.assertFalse(d.should_block)
|
|
self.assertEqual(d.target, "9.9.9.9")
|
|
self.assertIn("observation", d.reason)
|
|
|
|
def test_blocks_public_remote_when_enabled(self):
|
|
p = PolicyEngine(True, 1, "192.168.100.0/24", "")
|
|
d = p.evaluate(event("192.168.100.10", "9.9.9.9"))
|
|
self.assertTrue(d.should_block)
|
|
self.assertEqual(d.target, "9.9.9.9")
|
|
|
|
def test_inbound_selects_source(self):
|
|
p = PolicyEngine(True, 1, "192.168.100.0/24", "")
|
|
d = p.evaluate(event("9.9.9.9", "192.168.100.10"))
|
|
self.assertTrue(d.should_block)
|
|
self.assertEqual(d.target, "9.9.9.9")
|
|
|
|
def test_private_remote_is_never_blocked(self):
|
|
p = PolicyEngine(True, 1, "192.168.100.0/24", "")
|
|
d = p.evaluate(event("192.168.100.10", "10.10.10.10"))
|
|
self.assertFalse(d.should_block)
|
|
|
|
def test_never_block_list_wins(self):
|
|
p = PolicyEngine(True, 1, "192.168.100.0/24", "9.9.9.9/32")
|
|
d = p.evaluate(event("192.168.100.10", "9.9.9.9"))
|
|
self.assertFalse(d.should_block)
|
|
self.assertIn("NEVER_BLOCK", d.reason)
|
|
|
|
def test_severity_threshold(self):
|
|
p = PolicyEngine(True, 1, "192.168.100.0/24", "")
|
|
d = p.evaluate(event("192.168.100.10", "9.9.9.9", severity=2))
|
|
self.assertFalse(d.should_block)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|