DeepFal
2d76ed6672
- 特征检测（规则/签名）与行为检测（24h 统计）单进程运行 - 融合名单按首次发现时间倒序，展示位置/归属（GeoIP + ASN） - 钉钉融合推送（加签、24h 节流、去重） - 页面状态灯、CSV 导出、Docker 镜像构建文件
128 lines
4.9 KiB
Python
128 lines
4.9 KiB
Python
"""Unit tests for the client-IP field heuristics."""
|
|
|
|
import os
|
|
import sys
|
|
import unittest
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
from graylog_ops import detect # noqa: E402
|
|
|
|
|
|
def message(client_ip, source="10.180.254.17", **extra):
|
|
payload = {
|
|
"timestamp": "2026-10-17T15:00:00.000Z",
|
|
"source": source,
|
|
"client_ip": client_ip,
|
|
"status": 401,
|
|
"request_uri": "/api/login",
|
|
"message": '%s - - "POST /api/login HTTP/1.1" 401 0.01' % client_ip,
|
|
}
|
|
payload.update(extra)
|
|
return payload
|
|
|
|
|
|
class AsIpTests(unittest.TestCase):
|
|
def test_ipv4(self):
|
|
self.assertEqual(str(detect.as_ip("203.0.113.5")), "203.0.113.5")
|
|
|
|
def test_ipv6(self):
|
|
self.assertEqual(str(detect.as_ip("2001:db8::1")), "2001:db8::1")
|
|
|
|
def test_bracketed_ipv6_with_port(self):
|
|
self.assertEqual(str(detect.as_ip("[2001:db8::1]:443")), "2001:db8::1")
|
|
|
|
def test_ipv4_with_port(self):
|
|
self.assertEqual(str(detect.as_ip("192.0.2.44:8080")), "192.0.2.44")
|
|
|
|
def test_rejects_cidr(self):
|
|
self.assertIsNone(detect.as_ip("10.0.0.0/8"))
|
|
|
|
def test_rejects_hostname(self):
|
|
self.assertIsNone(detect.as_ip("edu-master"))
|
|
|
|
def test_rejects_bad_octet(self):
|
|
self.assertIsNone(detect.as_ip("999.1.1.1"))
|
|
|
|
def test_rejects_empty_and_number(self):
|
|
self.assertIsNone(detect.as_ip(""))
|
|
self.assertIsNone(detect.as_ip(12345))
|
|
|
|
|
|
class AnalyseTests(unittest.TestCase):
|
|
def test_empty_sample(self):
|
|
report = detect.analyse_messages([])
|
|
self.assertEqual(report["sampled"], 0)
|
|
self.assertIsNone(report["recommendation"]["field"])
|
|
self.assertEqual(report["recommendation"]["confidence"], "none")
|
|
|
|
def test_recommends_client_ip_over_source(self):
|
|
messages = [message(ip) for ip in (
|
|
"203.0.113.5", "203.0.113.5", "203.0.113.5",
|
|
"198.51.100.7", "198.51.100.7", "192.0.2.44", "192.0.2.44",
|
|
"10.180.2.9",
|
|
)]
|
|
report = detect.analyse_messages(messages)
|
|
self.assertEqual(report["recommendation"]["field"], "client_ip")
|
|
self.assertEqual(report["recommendation"]["confidence"], "high")
|
|
|
|
candidates = {row["field"]: row for row in report["candidates"]}
|
|
self.assertIn("source", candidates)
|
|
self.assertNotIn("message", candidates)
|
|
self.assertLess(candidates["source"]["score"], candidates["client_ip"]["score"])
|
|
# The constant `source` must not outrank the varying client address.
|
|
self.assertEqual(candidates["source"]["distinct_ips"], 1)
|
|
|
|
def test_warns_when_source_is_a_proxy(self):
|
|
messages = [message("203.0.113.%d" % index) for index in range(1, 7)]
|
|
report = detect.analyse_messages(messages)
|
|
joined = " ".join(report["recommendation"]["warnings"])
|
|
self.assertIn("source", joined)
|
|
|
|
def test_private_ratio_note(self):
|
|
messages = [message("10.180.2.%d" % index) for index in range(1, 7)]
|
|
report = detect.analyse_messages(messages)
|
|
best = report["recommendation"]["field"]
|
|
self.assertEqual(best, "client_ip")
|
|
joined = " ".join(report["recommendation"]["warnings"])
|
|
self.assertIn("内网", joined)
|
|
|
|
def test_no_ip_field_at_all(self):
|
|
messages = [{"source": "edu-master", "message": "no address here"}] * 3
|
|
report = detect.analyse_messages(messages)
|
|
self.assertIsNone(report["recommendation"]["field"])
|
|
self.assertEqual(report["candidates"], [])
|
|
|
|
def test_http_fields_detected(self):
|
|
report = detect.analyse_messages([message("203.0.113.5")])
|
|
self.assertIn("status", report["http_fields"])
|
|
self.assertIn("uri", report["http_fields"])
|
|
|
|
|
|
class InternalAddressTests(unittest.TestCase):
|
|
"""`is_internal` must not use ipaddress.is_private: that flag also covers
|
|
documentation ranges such as 203.0.113.0/24."""
|
|
|
|
def test_rfc1918(self):
|
|
self.assertTrue(detect.is_internal(detect.as_ip("10.1.2.3")))
|
|
self.assertTrue(detect.is_internal(detect.as_ip("172.16.5.5")))
|
|
self.assertTrue(detect.is_internal(detect.as_ip("192.168.1.1")))
|
|
|
|
def test_documentation_range_is_not_internal(self):
|
|
self.assertFalse(detect.is_internal(detect.as_ip("203.0.113.5")))
|
|
self.assertFalse(detect.is_internal(detect.as_ip("2001:db8::1")))
|
|
|
|
def test_loopback_link_local_and_cgnat(self):
|
|
self.assertTrue(detect.is_internal(detect.as_ip("127.0.0.1")))
|
|
self.assertTrue(detect.is_internal(detect.as_ip("169.254.1.1")))
|
|
self.assertTrue(detect.is_internal(detect.as_ip("100.64.1.1")))
|
|
self.assertTrue(detect.is_internal(detect.as_ip("::1")))
|
|
|
|
def test_ipv6_unique_local(self):
|
|
self.assertTrue(detect.is_internal(detect.as_ip("fd00::1")))
|
|
self.assertFalse(detect.is_internal(detect.as_ip("2400:cb00::1")))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|