83 lines
2.7 KiB
Python
83 lines
2.7 KiB
Python
import subprocess
|
|
import re
|
|
import time
|
|
import itertools
|
|
from db import init_ping_db, insert_ping_result
|
|
from config import PING_INTERVAL
|
|
|
|
# Curated pool of well-known, stable IPs across DNS resolvers, CDNs, and captive-portal check targets.
|
|
# Cycling 3 at a time means each IP is hit at most once every ~(len(pool)/3 * PING_INTERVAL) seconds.
|
|
IP_POOL = [
|
|
# DNS resolvers
|
|
"1.1.1.1", # Cloudflare
|
|
"1.0.0.1", # Cloudflare
|
|
"8.8.8.8", # Google
|
|
"8.8.4.4", # Google
|
|
"9.9.9.9", # Quad9
|
|
"149.112.112.112", # Quad9
|
|
"208.67.222.222", # OpenDNS
|
|
"208.67.220.220", # OpenDNS
|
|
"64.6.64.6", # Verisign
|
|
"84.200.69.80", # DNS.WATCH
|
|
]
|
|
|
|
TARGETS_PER_ITERATION = 3
|
|
|
|
_LATENCY_RE = re.compile(r"time[=<](\d+(?:\.\d+)?)\s*ms")
|
|
|
|
|
|
def detect_gateway() -> str | None:
|
|
try:
|
|
with open("/proc/net/route") as f:
|
|
for line in f.readlines()[1:]: # skip header row
|
|
fields = line.strip().split()
|
|
# Destination == "00000000" means 0.0.0.0 — the default/catch-all route
|
|
if fields[1] == "00000000":
|
|
addr = int(fields[2], 16) # little-endian hex
|
|
return ".".join(str((addr >> (8 * i)) & 0xFF) for i in range(4))
|
|
except Exception as e:
|
|
print(f"WARNING: gateway detection failed: {e}", flush=True)
|
|
return None
|
|
|
|
|
|
def ping_once(ip: str) -> tuple[bool, float | None]:
|
|
result = subprocess.run(
|
|
["ping", "-c", "1", "-W", "2", ip],
|
|
capture_output=True, text=True
|
|
)
|
|
if result.returncode != 0:
|
|
return False, None
|
|
m = _LATENCY_RE.search(result.stdout)
|
|
latency = float(m.group(1)) if m else None
|
|
return True, latency
|
|
|
|
|
|
def run():
|
|
init_ping_db()
|
|
gateway_ip = detect_gateway()
|
|
if gateway_ip:
|
|
print(f"Gateway detected: {gateway_ip}", flush=True)
|
|
else:
|
|
print("WARNING: No gateway detected, skipping gateway checks", flush=True)
|
|
|
|
pool = itertools.cycle(IP_POOL)
|
|
|
|
while True:
|
|
if gateway_ip:
|
|
success, latency_ms = ping_once(gateway_ip)
|
|
insert_ping_result(gateway_ip, success, latency_ms, is_gateway=True)
|
|
status = f"{latency_ms:.1f}ms" if latency_ms is not None else "FAIL"
|
|
print(f"[GW] {gateway_ip}: {status}", flush=True)
|
|
|
|
targets = [next(pool) for _ in range(TARGETS_PER_ITERATION)]
|
|
for ip in targets:
|
|
success, latency_ms = ping_once(ip)
|
|
insert_ping_result(ip, success, latency_ms)
|
|
status = f"{latency_ms:.1f}ms" if latency_ms is not None else "FAIL"
|
|
print(f"{ip}: {status}", flush=True)
|
|
time.sleep(PING_INTERVAL)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
run()
|