n0mad1k
c20f189d49
masscan port discovery → nmap fingerprinting → probe engine → pattern/version matching Dynamic YAML inputs: countries.yaml (country codes + CIDR resolution) and targets.yaml (fingerprint probes) RIR delegated stats for authoritative country CIDR data (no external API) Supports tcp_banner, http, https, rtsp, udp probe types Version extraction and comparison operators (<=, >=, ==, etc.) SQLite results per engagement, CSV/JSON export format-spec/ docs for generating input files externally
105 lines
3.3 KiB
Python
105 lines
3.3 KiB
Python
import ipaddress
|
|
import json
|
|
import os
|
|
import time
|
|
import urllib.request
|
|
from pathlib import Path
|
|
|
|
CACHE_DIR = Path(__file__).parent.parent / "data" / "cidr-cache"
|
|
CACHE_TTL = 86400 * 7 # 7 days
|
|
|
|
RIR_URLS = [
|
|
"https://ftp.ripe.net/pub/stats/ripencc/delegated-ripencc-latest",
|
|
"https://ftp.arin.net/pub/stats/arin/delegated-arin-extended-latest",
|
|
"https://ftp.apnic.net/pub/stats/apnic/delegated-apnic-latest",
|
|
"https://ftp.lacnic.net/pub/stats/lacnic/delegated-lacnic-latest",
|
|
"https://ftp.afrinic.net/pub/stats/afrinic/delegated-afrinic-latest",
|
|
]
|
|
|
|
RIR_NAMES = ["ripe", "arin", "apnic", "lacnic", "afrinic"]
|
|
|
|
|
|
def _cache_path(rir: str) -> Path:
|
|
return CACHE_DIR / f"{rir}.txt"
|
|
|
|
|
|
def _index_path() -> Path:
|
|
return CACHE_DIR / "index.json"
|
|
|
|
|
|
def _is_stale(path: Path) -> bool:
|
|
if not path.exists():
|
|
return True
|
|
return time.time() - path.stat().st_mtime > CACHE_TTL
|
|
|
|
|
|
def fetch_rir_data(force: bool = False) -> None:
|
|
CACHE_DIR.mkdir(parents=True, exist_ok=True)
|
|
for name, url in zip(RIR_NAMES, RIR_URLS):
|
|
path = _cache_path(name)
|
|
if not force and not _is_stale(path):
|
|
continue
|
|
print(f" Fetching {name}...", end=" ", flush=True)
|
|
try:
|
|
req = urllib.request.Request(url, headers={"User-Agent": "geo-scout/1.0"})
|
|
with urllib.request.urlopen(req, timeout=30) as resp:
|
|
path.write_bytes(resp.read())
|
|
print("done")
|
|
except Exception as e:
|
|
print(f"failed ({e})")
|
|
|
|
|
|
def _build_index() -> dict[str, list[str]]:
|
|
index: dict[str, list[str]] = {}
|
|
for name in RIR_NAMES:
|
|
path = _cache_path(name)
|
|
if not path.exists():
|
|
continue
|
|
for line in path.read_text(errors="replace").splitlines():
|
|
if line.startswith("#") or not line.strip():
|
|
continue
|
|
parts = line.split("|")
|
|
if len(parts) < 7:
|
|
continue
|
|
_, cc, record_type, start, value, _, status = parts[:7]
|
|
if record_type != "ipv4" or status not in ("allocated", "assigned"):
|
|
continue
|
|
if not cc or cc == "*":
|
|
continue
|
|
cc = cc.upper()
|
|
try:
|
|
count = int(value)
|
|
prefix_len = 32 - (count - 1).bit_length()
|
|
cidr = f"{start}/{prefix_len}"
|
|
ipaddress.ip_network(cidr, strict=False)
|
|
except (ValueError, TypeError):
|
|
continue
|
|
index.setdefault(cc, []).append(cidr)
|
|
return index
|
|
|
|
|
|
def load_index(force_rebuild: bool = False) -> dict[str, list[str]]:
|
|
idx_path = _index_path()
|
|
if not force_rebuild and idx_path.exists() and not _is_stale(idx_path):
|
|
return json.loads(idx_path.read_text())
|
|
index = _build_index()
|
|
idx_path.write_text(json.dumps(index))
|
|
return index
|
|
|
|
|
|
def get_cidrs(country_code: str, index: dict[str, list[str]]) -> list[str]:
|
|
return index.get(country_code.upper(), [])
|
|
|
|
|
|
def merge_cidrs(cidrs: list[str], exclude: list[str]) -> list[str]:
|
|
if not exclude:
|
|
return cidrs
|
|
exclude_nets = [ipaddress.ip_network(c, strict=False) for c in exclude]
|
|
result = []
|
|
for cidr in cidrs:
|
|
net = ipaddress.ip_network(cidr, strict=False)
|
|
excluded = any(net.overlaps(ex) for ex in exclude_nets)
|
|
if not excluded:
|
|
result.append(cidr)
|
|
return result
|