49db614262
Expand iplib, iptype, and ioc plugins with better caching, throttling, and lookup logic. Update validation script and showcase journal accordingly. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
240 lines
7.6 KiB
Python
240 lines
7.6 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Offline validation for normalization/extraction logic in plugins/iptype.py.
|
|
|
|
This intentionally does not perform live network calls (which may be blocked in CI/sandboxes).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import sys
|
|
|
|
# Ensure repo root is on sys.path when running as a script.
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
from plugins.iplib import ( # noqa: E402
|
|
ASNInfo,
|
|
GeoInfo,
|
|
IPInfo,
|
|
JSONNode,
|
|
VTInfo,
|
|
parse_asn_ipapi,
|
|
parse_asn_ipinfo,
|
|
parse_asn_ipwho,
|
|
parse_geo_ipapi,
|
|
parse_geo_ipinfo,
|
|
parse_geo_ipwho,
|
|
parse_geo_maxmind,
|
|
parse_vt_domain,
|
|
parse_vt_file,
|
|
parse_vt_ip,
|
|
parse_vt_url,
|
|
)
|
|
|
|
|
|
def _assert(cond: bool, msg: str):
|
|
if not cond:
|
|
raise AssertionError(msg)
|
|
|
|
|
|
def main() -> int:
|
|
# JSONNode chaining behavior
|
|
n = JSONNode({"a": {"b": 1}, "x": None})
|
|
_assert(str(n.a.b) == "1", "JSONNode dict nesting")
|
|
_assert(str(n.missing.anything) == "", "JSONNode missing keys should render empty")
|
|
_assert(str(n.x.y) == "", "JSONNode None chaining should render empty")
|
|
|
|
# ipinfo basic
|
|
ipinfo_raw = {
|
|
"country": "US",
|
|
"region": "California",
|
|
"city": "Mountain View",
|
|
"org": "AS15169 Google LLC",
|
|
"loc": "37.4056,-122.0775",
|
|
}
|
|
ipi = IPInfo(ipinfo_raw, source="ipinfo")
|
|
_assert(ipi.country == "US", "ipinfo.country")
|
|
_assert(ipi.asn == "AS15169", "ipinfo.asn derived from org")
|
|
_assert(str(ipi.data.country) == "US", "ipinfo.data.country")
|
|
|
|
# ASNInfo basics
|
|
asn = ASNInfo(
|
|
asn="AS15169",
|
|
name="Google LLC",
|
|
country="US",
|
|
raw={"org": "AS15169 Google LLC"},
|
|
source="ipinfo",
|
|
)
|
|
_assert(asn.asn == "AS15169", "asn.asn")
|
|
_assert(str(asn.data.org) == "AS15169 Google LLC", "asn.data.org")
|
|
|
|
# VTInfo basics
|
|
vt = VTInfo(
|
|
malicious=3,
|
|
suspicious=1,
|
|
harmless=90,
|
|
total=94,
|
|
score=3 / 94,
|
|
categories=("search engine",),
|
|
names=("foo", "bar"),
|
|
name="foo",
|
|
raw={"data": "x"},
|
|
source="virustotal",
|
|
)
|
|
_assert(vt.verdict == "3/94", "vt.verdict")
|
|
_assert(vt.category == "search engine", "vt.category")
|
|
_assert(vt.type == "search engine", "vt.type alias")
|
|
_assert(vt.name == "foo" and vt.names[1] == "bar", "vt names")
|
|
_assert(vt.confidence is not None and vt.confidence > 0, "vt.confidence")
|
|
|
|
# Parse helpers
|
|
_assert(parse_asn_ipinfo(ipinfo_raw).asn == "AS15169", "parse_asn_ipinfo")
|
|
ipapi_raw = {
|
|
"asn": "AS123",
|
|
"org": "Example ISP",
|
|
"country_code": "DE",
|
|
"country_name": "Germany",
|
|
}
|
|
_assert(parse_asn_ipapi(ipapi_raw).asn == "AS123", "parse_asn_ipapi")
|
|
ipwho_raw = {"country_code": "NL", "connection": {"asn": 9009, "isp": "M247"}}
|
|
_assert(parse_asn_ipwho(ipwho_raw).asn == "AS9009", "parse_asn_ipwho")
|
|
|
|
geo1 = parse_geo_ipinfo(ipinfo_raw)
|
|
_assert(geo1.country_code == "US", "parse_geo_ipinfo country_code")
|
|
_assert(geo1.lat is not None and geo1.lon is not None, "parse_geo_ipinfo loc")
|
|
geo2 = parse_geo_ipapi(
|
|
{"country_code": "DE", "country_name": "Germany", "latitude": 1, "longitude": 2}
|
|
)
|
|
_assert(geo2.country_code == "DE", "parse_geo_ipapi country_code")
|
|
geo3 = parse_geo_ipwho(
|
|
{"country_code": "NL", "country": "Netherlands", "latitude": 1, "longitude": 2}
|
|
)
|
|
_assert(geo3.country_code == "NL", "parse_geo_ipwho country_code")
|
|
|
|
mm_raw = {
|
|
"country": {"iso_code": "US", "names": {"en": "United States"}},
|
|
"location": {"latitude": 1, "longitude": 2},
|
|
}
|
|
mm = parse_geo_maxmind(mm_raw)
|
|
_assert(
|
|
mm.country_code == "US" and mm.country == "United States", "parse_geo_maxmind"
|
|
)
|
|
|
|
vt_raw = {
|
|
"data": {
|
|
"attributes": {
|
|
"last_analysis_stats": {
|
|
"malicious": 2,
|
|
"suspicious": 1,
|
|
"harmless": 10,
|
|
},
|
|
"reputation": 5,
|
|
"categories": {"foo": "search engine"},
|
|
"asn": 15169,
|
|
"as_owner": "Google",
|
|
"country": "US",
|
|
"network": "8.8.8.0/24",
|
|
}
|
|
}
|
|
}
|
|
vt2 = parse_vt_ip(vt_raw)
|
|
_assert(vt2.verdict == "2/13", "parse_vt_ip verdict")
|
|
_assert(vt2.score is not None and vt2.score > 0, "parse_vt_ip score")
|
|
_assert(vt2.asn == "AS15169", "parse_vt_ip asn")
|
|
|
|
vd = parse_vt_domain(
|
|
{
|
|
"data": {
|
|
"attributes": {
|
|
"last_analysis_stats": {"malicious": 1, "undetected": 9},
|
|
"last_dns_records": [
|
|
{"type": "A", "value": "1.2.3.4"},
|
|
{"type": "AAAA", "value": "2001:db8::1"},
|
|
],
|
|
"categories": {"x": "phishing"},
|
|
}
|
|
}
|
|
}
|
|
)
|
|
_assert(vd.ip == "1.2.3.4", "parse_vt_domain last ip")
|
|
_assert(vd.ips == ("1.2.3.4", "2001:db8::1"), "parse_vt_domain ips")
|
|
_assert(vd.verdict == "1/10", "parse_vt_domain verdict")
|
|
|
|
vf = parse_vt_file(
|
|
{
|
|
"data": {
|
|
"attributes": {
|
|
"last_analysis_stats": {"malicious": 5, "undetected": 5},
|
|
"popular_threat_classification": {
|
|
"popular_threat_name": [
|
|
{"value": "emotet", "count": 10},
|
|
{"value": "trojan", "count": 7},
|
|
],
|
|
"suggested_threat_label": "trojan.emotet",
|
|
},
|
|
"meaningful_name": "sample.exe",
|
|
}
|
|
}
|
|
}
|
|
)
|
|
_assert(vf.name == "emotet", "parse_vt_file best name")
|
|
_assert("trojan.emotet" in vf.names, "parse_vt_file names")
|
|
_assert(vf.score == 0.5, "parse_vt_file score")
|
|
|
|
vu = parse_vt_url(
|
|
{
|
|
"data": {
|
|
"attributes": {
|
|
"last_analysis_stats": {"malicious": 0, "undetected": 10},
|
|
"threat_names": ["brand-impersonation"],
|
|
}
|
|
}
|
|
}
|
|
)
|
|
_assert(vu.name == "brand-impersonation", "parse_vt_url name")
|
|
|
|
try:
|
|
from plugins.iptype import ip as ipconv
|
|
|
|
n = ipconv("192.168.12.1/24")
|
|
if n is None:
|
|
raise AssertionError("ip converter network parse")
|
|
_assert(n.type == "cidr4", "cidr type")
|
|
_assert(n.mask == "255.255.255.0", "cidr netmask")
|
|
_assert(n.identity == "192.168.12.0", "cidr identity")
|
|
_assert(n.broadcast == "192.168.12.255", "cidr broadcast")
|
|
_assert(n.range == "192.168.12.0-192.168.12.255", "cidr range")
|
|
_assert(n.hostcount == 254, "cidr hostcount")
|
|
_assert(n.rfc_type == "private", "cidr rfc type")
|
|
|
|
a = ipconv("8.8.8.8")
|
|
if a is None:
|
|
raise AssertionError("ip converter ip parse")
|
|
_assert(a.type == "ipv4", "ip type")
|
|
_assert(a.rfc_type == "global", "ip rfc type")
|
|
except ModuleNotFoundError:
|
|
# Script is still useful in limited environments without VisiData runtime.
|
|
pass
|
|
|
|
# GeoInfo basics
|
|
geo = GeoInfo(
|
|
country="United States",
|
|
country_code="US",
|
|
region="California",
|
|
city="Mountain View",
|
|
lat=1.0,
|
|
lon=2.0,
|
|
raw={"country": {"iso_code": "US"}},
|
|
source="maxmind",
|
|
)
|
|
_assert(geo.country_code == "US", "geo.country_code")
|
|
_assert(str(geo.data.country.iso_code) == "US", "geo.data.country.iso_code")
|
|
|
|
print("ok")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|