- 特征检测(规则/签名)与行为检测(24h 统计)单进程运行 - 融合名单按首次发现时间倒序,展示位置/归属(GeoIP + ASN) - 钉钉融合推送(加签、24h 节流、去重) - 页面状态灯、CSV 导出、Docker 镜像构建文件
49 lines
2.0 KiB
Python
49 lines
2.0 KiB
Python
import os
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
from graylog_ops.geo import GeoIP, build_geo_db, country_name, flag, region_name
|
|
|
|
|
|
class GeoIPTests(unittest.TestCase):
|
|
def test_build_and_lookup(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
csv_path = os.path.join(directory, "city.csv")
|
|
with open(csv_path, "w", encoding="utf-8") as fh:
|
|
fh.write("1.0.0.0,1.0.0.255,OC,AU,Queensland,\"South Brisbane\",0,0\n")
|
|
fh.write("1.0.1.0,1.0.3.255,AS,CN,Fujian,Wenquan,0,0\n")
|
|
fh.write("8.8.8.0,8.8.8.255,NA,US,California,\"Mountain View\",0,0\n")
|
|
fh.write("2001:db8::,2001:db8::ffff,NA,US,California,X,0,0\n")
|
|
db_path = os.path.join(directory, "geo.db")
|
|
count = build_geo_db(csv_path, db_path)
|
|
self.assertEqual(count, 3) # IPv6 range skipped
|
|
|
|
geo = GeoIP(db_path)
|
|
try:
|
|
cn = geo.lookup("1.0.2.1")
|
|
self.assertEqual(cn["country_code"], "CN")
|
|
self.assertEqual(cn["country_name"], "中国")
|
|
self.assertEqual(cn["region"], "福建")
|
|
self.assertEqual(geo.lookup("8.8.8.8")["country_name"], "美国")
|
|
self.assertIsNone(geo.lookup("9.9.9.9"))
|
|
self.assertIsNone(geo.lookup("not-an-ip"))
|
|
finally:
|
|
geo.close()
|
|
|
|
def test_helpers(self):
|
|
self.assertEqual(flag("CN"), "\U0001F1E8\U0001F1F3")
|
|
self.assertEqual(flag("ZZ"), "")
|
|
self.assertEqual(flag(""), "")
|
|
self.assertEqual(country_name("CN"), "中国")
|
|
self.assertEqual(country_name("US"), "美国")
|
|
self.assertEqual(country_name("XX"), "XX")
|
|
self.assertEqual(region_name("CN", "Guangdong"), "广东")
|
|
self.assertEqual(region_name("US", "California"), "California")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|