Public/fgAI
SHA256
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix anon

Browse Source
This commit is contained in:
larssand
2026-09-23 13:32:09 +02:00
parent bf1109be81
commit a3f545f1b0
6 changed files with 159 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

56
tests/test_anomaly.py
View File

@@ -24,6 +24,62 @@ class AnomalyTests(unittest.TestCase):
self.assertGreaterEqual(findings[0].score, 60)
self.assertIn(findings[0].severity, {"high", "critical"})
def test_entity_scope_feedback_suppresses_anomaly_even_without_field_deviations(self):
events = [
parse_log_line(f'type=traffic srcip=10.0.0.{i} dstip=1.1.1.1 service=https action=accept sentbyte=100 rcvdbyte=100')
for i in range(1, 8)
]
events.extend(
parse_log_line(
'type=utm subtype=ips srcip=8.8.8.8 dstip=10.0.0.10 service=https '
'action=blocked severity=critical sentbyte=0 rcvdbyte=0'
)
for _ in range(5)
)
# Confirm it would normally fire (no field_deviations involved at all).
self.assertTrue(detect_source_anomalies(events))
feedback = [{"scope": "entity", "entity": "8.8.8.8", "status": "false_positive"}]
findings = detect_source_anomalies(events, feedback=feedback)
self.assertEqual(findings, [])
def test_entity_scope_feedback_does_not_affect_other_sources(self):
events = [
parse_log_line(
'type=utm subtype=ips srcip=8.8.8.8 dstip=10.0.0.10 service=https '
'action=blocked severity=critical sentbyte=0 rcvdbyte=0'
)
for _ in range(5)
]
events.extend(
parse_log_line(
'type=utm subtype=ips srcip=9.9.9.9 dstip=10.0.0.11 service=https '
'action=blocked severity=critical sentbyte=0 rcvdbyte=0'
)
for _ in range(5)
)
feedback = [{"scope": "entity", "entity": "8.8.8.8", "status": "false_positive"}]
findings = detect_source_anomalies(events, feedback=feedback)
self.assertEqual([finding.subject for finding in findings], ["9.9.9.9"])
def test_missing_feedback_preserves_default_behavior(self):
events = [
parse_log_line(
'type=utm subtype=ips srcip=8.8.8.8 dstip=10.0.0.10 service=https '
'action=blocked severity=critical sentbyte=0 rcvdbyte=0'
)
for _ in range(5)
]
with_none = detect_source_anomalies(events, feedback=None)
without_param = detect_source_anomalies(events)
self.assertEqual([f.subject for f in with_none], [f.subject for f in without_param])
self.assertTrue(with_none)
def test_summary_counts_severities(self):
events = [
parse_log_line(