727 lines
28 KiB
Python
727 lines
28 KiB
Python
#!/usr/bin/env python3
|
|
"""FIPS Mesh Reachability Scanner.
|
|
|
|
Orchestrates the per-seed scan lifecycle described in
|
|
[`plans/mesh-reachability-scanner.md`](../plans/mesh-reachability-scanner.md):
|
|
|
|
directory → seed selection → per-seed ephemeral daemon → peer with seed →
|
|
converge → record diagnostics → per-target reachability check → output
|
|
matrix + per-pair diagnostics JSON + summary.
|
|
|
|
Usage:
|
|
python3 mesh_scan.py --seeds-sample 3 --converge-secs 30
|
|
python3 mesh_scan.py --seeds npub1abc,npub1def --targets npub1ghi
|
|
python3 mesh_scan.py --self-test # quick 2-node smoke test
|
|
python3 mesh_scan.py --cleanup-stale
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import csv
|
|
import json
|
|
import logging
|
|
import os
|
|
import random
|
|
import sys
|
|
import time
|
|
from dataclasses import dataclass, field, asdict
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
# Allow running both as a module and as a script.
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
|
|
import control # noqa: E402
|
|
from control import ControlClient, ControlError # noqa: E402
|
|
import directory # noqa: E402
|
|
from directory import NodeEntry, fetch_directory, npub_to_ipv6 # noqa: E402
|
|
import fipsd # noqa: E402
|
|
from fipsd import EphemeralDaemon, spawn_daemon, cleanup_stale_tun_devices # noqa: E402
|
|
import reachability # noqa: E402
|
|
from reachability import check as reach_check # noqa: E402
|
|
|
|
log = logging.getLogger("mesh_scan")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Data records
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@dataclass
|
|
class SeedState:
|
|
"""Diagnostics recorded about the ephemeral node after convergence."""
|
|
npub: str
|
|
peer_count: int = 0
|
|
tree_state: str = ""
|
|
estimated_mesh_size: int = 0
|
|
seed_depth: int | None = None
|
|
seed_parent: str = ""
|
|
mesh_root: str = ""
|
|
known_npubs: list[str] = field(default_factory=list)
|
|
bloom_peer_count: int = 0
|
|
|
|
|
|
@dataclass
|
|
class PairResult:
|
|
seed: str
|
|
target: str
|
|
reachable: bool
|
|
rtt_ms: float | None
|
|
loss_pct: int | None
|
|
method: str
|
|
target_in_identity_cache: bool
|
|
target_in_seed_bloom: bool
|
|
cached_coords: bool
|
|
tree_distance: int | None
|
|
seed_depth: int | None
|
|
seed_parent: str
|
|
mesh_root: str
|
|
converge_secs: float
|
|
detail: str = ""
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Seed / target selection
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _ts() -> str:
|
|
"""Current wall-clock time as HH:MM:SS for progress prints."""
|
|
return time.strftime("%H:%M:%S")
|
|
|
|
|
|
def select_seeds(nodes: list[NodeEntry], args: argparse.Namespace) -> list[NodeEntry]:
|
|
by_npub = {n.npub: n for n in nodes}
|
|
# For automatic selection (all/sample/near-root), only consider seeds
|
|
# with a dialable endpoint — nat-only nodes can't be peered with
|
|
# directly, so scanning them just wastes time.
|
|
dialable = [n for n in nodes if n.first_dialable_endpoint() is not None]
|
|
|
|
if args.seeds == "all":
|
|
return dialable
|
|
|
|
if args.seeds and args.seeds != "all":
|
|
chosen: list[NodeEntry] = []
|
|
for npub in args.seeds.split(","):
|
|
npub = npub.strip()
|
|
if npub in by_npub:
|
|
chosen.append(by_npub[npub])
|
|
else:
|
|
log.warning("seed %s not in directory, skipping", npub)
|
|
return chosen
|
|
|
|
if args.seeds_near_root:
|
|
return sorted(dialable, key=lambda n: npub_to_ipv6(n.npub))[:args.seeds_near_root]
|
|
|
|
if args.seeds_sample:
|
|
k = min(args.seeds_sample, len(dialable))
|
|
return random.sample(dialable, k) if k > 0 else []
|
|
|
|
# Default: a small sample of 5 from dialable nodes.
|
|
k = min(5, len(dialable))
|
|
return random.sample(dialable, k) if k > 0 else []
|
|
|
|
|
|
def select_targets(nodes: list[NodeEntry], seed: NodeEntry, args: argparse.Namespace) -> list[NodeEntry]:
|
|
if args.targets:
|
|
wanted = {t.strip() for t in args.targets.split(",")}
|
|
return [n for n in nodes if n.npub in wanted and n.npub != seed.npub]
|
|
candidates = [n for n in nodes if n.npub != seed.npub]
|
|
if args.targets_sample:
|
|
k = min(args.targets_sample, len(candidates))
|
|
return random.sample(candidates, k) if k > 0 else []
|
|
return candidates
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Per-seed lifecycle
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _peer_connected(client: ControlClient, seed_npub: str) -> bool:
|
|
try:
|
|
peers = client.show_peers().get("peers", []) or []
|
|
except ControlError:
|
|
return False
|
|
for p in peers:
|
|
if p.get("npub") == seed_npub and p.get("connectivity") == "connected":
|
|
return True
|
|
return False
|
|
|
|
|
|
def _peer_with_seed(daemon: EphemeralDaemon, seed: NodeEntry, timeout: float = 15.0) -> bool:
|
|
ep = seed.first_dialable_endpoint()
|
|
if ep is None:
|
|
log.warning("seed %s has no dialable endpoint (only nat?)", seed.npub[:16])
|
|
return False
|
|
log.info("peering with seed %s at %s/%s", seed.npub[:16], ep.addr, ep.transport)
|
|
try:
|
|
daemon.client.connect(seed.npub, ep.addr, ep.transport)
|
|
except ControlError as e:
|
|
log.warning("connect command failed: %s", e)
|
|
return False
|
|
|
|
deadline = time.time() + timeout
|
|
while time.time() < deadline:
|
|
if _peer_connected(daemon.client, seed.npub):
|
|
log.info("seed peer connected")
|
|
return True
|
|
time.sleep(0.5)
|
|
log.warning("seed did not connect within %.0fs", timeout)
|
|
return False
|
|
|
|
|
|
def _converged(client: ControlClient) -> tuple[bool, dict[str, Any]]:
|
|
"""Check convergence criteria. Returns (converged, status_dict)."""
|
|
try:
|
|
status = client.show_status()
|
|
except ControlError:
|
|
return False, {}
|
|
peer_count = status.get("peer_count", 0)
|
|
if peer_count < 1:
|
|
return False, status
|
|
# tree_state isn't always present; treat non-empty tree with a root as stable.
|
|
try:
|
|
tree = client.show_tree()
|
|
except ControlError:
|
|
tree = {}
|
|
root = tree.get("root")
|
|
peer_tree_count = tree.get("peer_tree_count", 0)
|
|
if not root or peer_tree_count < 1:
|
|
return False, status
|
|
return True, {"status": status, "tree": tree}
|
|
|
|
|
|
def _wait_convergence(daemon: EphemeralDaemon, budget: float) -> tuple[float, dict[str, Any]]:
|
|
"""Poll until converged or budget exhausted. Returns (elapsed_secs, snapshot).
|
|
|
|
Convergence requires:
|
|
- tree is stable (has a root + peer tree entries)
|
|
- identity cache has stopped growing for 3 consecutive 5s polls
|
|
- identity cache has at least 2 entries (not just the direct peer)
|
|
The minimum-count gate prevents premature exit before Nostr discovery
|
|
has had time to populate the cache from the advert relays.
|
|
"""
|
|
start = time.time()
|
|
deadline = start + budget
|
|
last_snapshot: dict[str, Any] = {}
|
|
stable_count = 0
|
|
last_known_count = -1
|
|
poll_interval = 5.0
|
|
min_cache = 2
|
|
while time.time() < deadline:
|
|
converged, snap = _converged(daemon.client)
|
|
if converged:
|
|
try:
|
|
ic = daemon.client.show_identity_cache().get("count", 0)
|
|
except ControlError:
|
|
ic = 0
|
|
if ic == last_known_count and ic >= min_cache:
|
|
stable_count += 1
|
|
else:
|
|
stable_count = 0
|
|
last_known_count = ic
|
|
if stable_count >= 3:
|
|
last_snapshot = snap
|
|
return time.time() - start, last_snapshot
|
|
last_snapshot = snap
|
|
time.sleep(poll_interval)
|
|
return time.time() - start, last_snapshot
|
|
|
|
|
|
def _record_seed_state(daemon: EphemeralDaemon, seed_npub: str) -> SeedState:
|
|
try:
|
|
status = daemon.client.show_status()
|
|
except ControlError:
|
|
status = {}
|
|
try:
|
|
tree = daemon.client.show_tree()
|
|
except ControlError:
|
|
tree = {}
|
|
try:
|
|
ic = daemon.client.show_identity_cache()
|
|
except ControlError:
|
|
ic = {}
|
|
try:
|
|
bloom = daemon.client.show_bloom()
|
|
except ControlError:
|
|
bloom = {}
|
|
|
|
known = [e.get("npub") for e in (ic.get("entries", []) or []) if e.get("npub")]
|
|
return SeedState(
|
|
npub=seed_npub,
|
|
peer_count=status.get("peer_count", 0),
|
|
tree_state=str(status.get("state", "")),
|
|
estimated_mesh_size=status.get("estimated_mesh_size", 0),
|
|
seed_depth=tree.get("depth"),
|
|
seed_parent=tree.get("parent_display_name", "") or "",
|
|
mesh_root=tree.get("root_npub", "") or "",
|
|
known_npubs=known,
|
|
bloom_peer_count=len(bloom.get("peer_filters", []) or []),
|
|
)
|
|
|
|
|
|
def _target_in_bloom(daemon: EphemeralDaemon, target_npub: str) -> bool:
|
|
"""Best-effort: is the target's node_addr in any peer's bloom filter?
|
|
|
|
The bloom query doesn't expose per-element membership directly, so this
|
|
is a heuristic: we treat 'target known to at least one peer filter' as
|
|
a proxy. Falls back to False if unavailable.
|
|
"""
|
|
try:
|
|
bloom = daemon.client.show_bloom()
|
|
except ControlError:
|
|
return False
|
|
# show_bloom gives per-peer filter stats, not membership. We approximate
|
|
# by checking the identity cache (a node we know about is more likely to
|
|
# be in propagated filters). This is documented as a heuristic.
|
|
try:
|
|
ic = daemon.client.show_identity_cache()
|
|
except ControlError:
|
|
return False
|
|
for e in ic.get("entries", []) or []:
|
|
if e.get("npub") == target_npub:
|
|
return True
|
|
return False
|
|
|
|
|
|
def _target_has_cached_coords(daemon: EphemeralDaemon, target_npub: str) -> bool:
|
|
"""Inferred from show_routing: is the target in the coord cache?
|
|
|
|
show_routing gives a count, not per-target entries, so we approximate:
|
|
if the target is in the identity cache AND there are coord_cache_entries,
|
|
we assume coords may be cached. (Documented heuristic.)
|
|
"""
|
|
try:
|
|
routing = daemon.client.show_routing()
|
|
except ControlError:
|
|
return False
|
|
return routing.get("coord_cache_entries", 0) > 0 and _target_in_bloom(daemon, target_npub)
|
|
|
|
|
|
def _tree_distance_to_target(daemon: EphemeralDaemon, target_npub: str) -> int | None:
|
|
"""Tree distance to a target, if it appears in show_tree peers."""
|
|
try:
|
|
tree = daemon.client.show_tree()
|
|
except ControlError:
|
|
return None
|
|
for p in tree.get("peers", []) or []:
|
|
if p.get("display_name") == target_npub or p.get("npub") == target_npub:
|
|
d = p.get("distance_to_us")
|
|
if isinstance(d, (int, float)):
|
|
return int(d)
|
|
return None
|
|
|
|
|
|
def scan_seed(
|
|
seed: NodeEntry,
|
|
targets: list[NodeEntry],
|
|
index: int,
|
|
args: argparse.Namespace,
|
|
total_seeds: int = 0,
|
|
) -> tuple[list[PairResult], SeedState | None]:
|
|
"""Run the full per-seed lifecycle. Returns (pair_results, seed_state)."""
|
|
log.info("=== seed %d/%d: %s ===", index + 1, total_seeds, seed.npub[:24])
|
|
t0 = time.time()
|
|
print(f"\n[{_ts()}] seed {index + 1}/{total_seeds}: {seed.npub[:24]} "
|
|
f"({len(targets)} targets)", flush=True)
|
|
pair_results: list[PairResult] = []
|
|
# Skip seeds with no dialable endpoint (only nat) — we can't peer with
|
|
# them directly, so there's no point spawning a daemon.
|
|
if seed.first_dialable_endpoint() is None:
|
|
print(f" [{_ts()}] skipped: no dialable endpoint (nat only) "
|
|
f"({time.time()-t0:.1f}s)", flush=True)
|
|
log.warning("seed %s has no dialable endpoint (only nat?); skipping", seed.npub[:16])
|
|
return pair_results, None
|
|
daemon: EphemeralDaemon | None = None
|
|
try:
|
|
daemon = spawn_daemon(tag=seed.npub[:12], index=index)
|
|
daemon.wait_ready(args.startup_timeout)
|
|
log.info("ephemeral daemon ready: netns=%s", daemon.netns)
|
|
print(f" [{_ts()}] daemon up ({time.time()-t0:.1f}s)", flush=True)
|
|
|
|
t_peer = time.time()
|
|
if not _peer_with_seed(daemon, seed, timeout=args.peer_timeout):
|
|
print(f" [{_ts()}] peering FAILED ({time.time()-t_peer:.1f}s); "
|
|
f"skipping targets", flush=True)
|
|
log.warning("could not peer with seed %s; skipping targets", seed.npub[:16])
|
|
return pair_results, None
|
|
|
|
t_conv = time.time()
|
|
converge_secs, snap = _wait_convergence(daemon, args.converge_secs)
|
|
log.info("converged in %.1fs", converge_secs)
|
|
|
|
seed_state = _record_seed_state(daemon, seed.npub)
|
|
log.info("seed state: peers=%d depth=%s root=%s known=%d",
|
|
seed_state.peer_count, seed_state.seed_depth,
|
|
seed_state.mesh_root[:16] or "(none)", len(seed_state.known_npubs))
|
|
print(f" [{_ts()}] peered ok | peers={seed_state.peer_count} "
|
|
f"depth={seed_state.seed_depth} known={len(seed_state.known_npubs)} | "
|
|
f"peer {time.time()-t_peer:.1f}s + converge {time.time()-t_conv:.1f}s",
|
|
flush=True)
|
|
|
|
# Reachability checks run inside the daemon's netns so the fd00::/8
|
|
# route resolves to the daemon's TUN, not the host's fips0.
|
|
runner = daemon.run_in_netns
|
|
for target in targets:
|
|
target_ipv6 = npub_to_ipv6(target.npub)
|
|
res = reach_check(target_ipv6, timeout=args.ping_timeout,
|
|
tcp_port=args.tcp_port, prefer=args.method,
|
|
runner=runner)
|
|
in_cache = target.npub in seed_state.known_npubs
|
|
in_bloom = _target_in_bloom(daemon, target.npub)
|
|
cached_coords = _target_has_cached_coords(daemon, target.npub)
|
|
tree_dist = _tree_distance_to_target(daemon, target.npub)
|
|
|
|
pr = PairResult(
|
|
seed=seed.npub,
|
|
target=target.npub,
|
|
reachable=res.reachable,
|
|
rtt_ms=res.rtt_ms,
|
|
loss_pct=res.loss_pct,
|
|
method=res.method,
|
|
target_in_identity_cache=in_cache,
|
|
target_in_seed_bloom=in_bloom,
|
|
cached_coords=cached_coords,
|
|
tree_distance=tree_dist,
|
|
seed_depth=seed_state.seed_depth,
|
|
seed_parent=seed_state.seed_parent,
|
|
mesh_root=seed_state.mesh_root,
|
|
converge_secs=converge_secs,
|
|
detail=res.detail,
|
|
)
|
|
pair_results.append(pr)
|
|
status = "ok" if pr.reachable else "FAIL"
|
|
log.info(" target %s: %s rtt=%s method=%s cache=%s bloom=%s",
|
|
target.npub[:16], status,
|
|
f"{res.rtt_ms:.0f}ms" if res.rtt_ms else "-",
|
|
res.method, in_cache, in_bloom)
|
|
if args.delay > 0:
|
|
time.sleep(args.delay)
|
|
|
|
# Per-seed summary line (always visible on stdout).
|
|
ok_count = sum(1 for r in pair_results if r.reachable)
|
|
total = len(pair_results)
|
|
rate = 100 * ok_count / total if total else 0
|
|
print(f" [{_ts()}] result: {ok_count}/{total} reachable ({rate:.0f}%) "
|
|
f"| total {time.time()-t0:.1f}s", flush=True)
|
|
return pair_results, seed_state
|
|
|
|
except Exception as e:
|
|
log.exception("seed scan failed: %s", e)
|
|
return pair_results, None
|
|
finally:
|
|
if daemon is not None:
|
|
try:
|
|
daemon.stop()
|
|
except Exception as e:
|
|
log.warning("daemon cleanup failed: %s", e)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Output
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def write_matrix_csv(results: list[PairResult], path: Path) -> None:
|
|
seeds = list(dict.fromkeys(r.seed for r in results))
|
|
targets = list(dict.fromkeys(r.target for r in results))
|
|
with path.open("w", newline="") as f:
|
|
w = csv.writer(f)
|
|
w.writerow(["seed"] + [t[:16] for t in targets])
|
|
for s in seeds:
|
|
row = [s[:16]]
|
|
for t in targets:
|
|
pr = next((r for r in results if r.seed == s and r.target == t), None)
|
|
if pr is None:
|
|
row.append("")
|
|
elif pr.reachable:
|
|
row.append(f"ok/{pr.rtt_ms:.0f}ms" if pr.rtt_ms else "ok")
|
|
else:
|
|
row.append("fail")
|
|
w.writerow(row)
|
|
|
|
|
|
def write_diagnostics_jsonl(results: list[PairResult], path: Path) -> None:
|
|
with path.open("w") as f:
|
|
for pr in results:
|
|
f.write(json.dumps(asdict(pr)) + "\n")
|
|
|
|
|
|
def print_summary(results: list[PairResult], seed_states: dict[str, SeedState]) -> None:
|
|
total = len(results)
|
|
if total == 0:
|
|
print("\nNo pairs scanned.")
|
|
return
|
|
ok = sum(1 for r in results if r.reachable)
|
|
print("\n" + "=" * 60)
|
|
print("MESH REACHABILITY SUMMARY")
|
|
print("=" * 60)
|
|
print(f"Seeds scanned: {len(seed_states)}")
|
|
print(f"Total pairs: {total}")
|
|
print(f"Reachable: {ok}/{total} ({100*ok/total:.1f}%)")
|
|
|
|
# Per-seed rates.
|
|
print("\nPer-seed reachability:")
|
|
for s, state in seed_states.items():
|
|
seed_results = [r for r in results if r.seed == s]
|
|
s_ok = sum(1 for r in seed_results if r.reachable)
|
|
n = len(seed_results)
|
|
rate = 100 * s_ok / n if n else 0
|
|
print(f" {s[:24]} {s_ok}/{n} ({rate:.0f}%) peers={state.peer_count} depth={state.seed_depth}")
|
|
|
|
# Per-target rates.
|
|
targets = list(dict.fromkeys(r.target for r in results))
|
|
print("\nPer-target reachability (universally unreachable = NAT/firewall suspect):")
|
|
for t in targets:
|
|
t_results = [r for r in results if r.target == t]
|
|
t_ok = sum(1 for r in t_results if r.reachable)
|
|
n = len(t_results)
|
|
rate = 100 * t_ok / n if n else 0
|
|
flag = " <-- unreachable from all seeds" if t_ok == 0 and n > 0 else ""
|
|
print(f" {t[:24]} {t_ok}/{n} ({rate:.0f}%){flag}")
|
|
|
|
# Failure-cause correlation.
|
|
fails = [r for r in results if not r.reachable]
|
|
if fails:
|
|
cache_miss = sum(1 for r in fails if not r.target_in_identity_cache)
|
|
bloom_miss = sum(1 for r in fails if not r.target_in_seed_bloom)
|
|
coord_miss = sum(1 for r in fails if not r.cached_coords)
|
|
print(f"\nFailure correlation ({len(fails)} failures):")
|
|
print(f" identity-cache miss: {cache_miss}")
|
|
print(f" bloom miss: {bloom_miss}")
|
|
print(f" coord-cache miss: {coord_miss}")
|
|
print("=" * 60)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Self-test: two ephemeral nodes peered directly
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def self_test(args: argparse.Namespace) -> int:
|
|
"""Smoke test: spawn two ephemeral daemons, peer them, verify they communicate.
|
|
|
|
This validates the whole toolchain (daemon spawn, control socket, peering,
|
|
reachability) without touching the public mesh. Both daemons share a
|
|
single network namespace so they can peer over loopback (127.0.0.1).
|
|
"""
|
|
log.info("SELF-TEST: spawning two ephemeral daemons (shared netns) and peering them")
|
|
shared_netns = "fips-scan-selftest"
|
|
a = b = None
|
|
try:
|
|
# Create the shared namespace up front; both daemons join it.
|
|
shared_info = fipsd.create_netns(shared_netns)
|
|
a = spawn_daemon(tag="selftest-a", index=900, netns=shared_netns)
|
|
a.wait_ready(args.startup_timeout)
|
|
b = spawn_daemon(tag="selftest-b", index=901, netns=shared_netns)
|
|
b.wait_ready(args.startup_timeout)
|
|
|
|
sa = a.client.show_status()
|
|
sb = b.client.show_status()
|
|
npub_a = sa.get("npub")
|
|
npub_b = sb.get("npub")
|
|
ipv6_a = sa.get("ipv6_addr")
|
|
ipv6_b = sb.get("ipv6_addr")
|
|
log.info("node A: npub=%s ipv6=%s netns=%s", npub_a, ipv6_a, a.netns)
|
|
log.info("node B: npub=%s ipv6=%s netns=%s", npub_b, ipv6_b, b.netns)
|
|
|
|
if not (npub_a and npub_b and ipv6_a and ipv6_b):
|
|
log.error("could not read both nodes' status")
|
|
return 1
|
|
|
|
# Peer B → A using A's UDP port (loopback inside the shared netns).
|
|
a_ep = f"127.0.0.1:{a.udp_port}"
|
|
log.info("connecting B -> A at %s/udp", a_ep)
|
|
try:
|
|
b.client.connect(npub_a, a_ep, "udp")
|
|
except ControlError as e:
|
|
log.error("connect command failed: %s", e)
|
|
return 1
|
|
|
|
# Wait for the peer to come up on both sides.
|
|
deadline = time.time() + 20
|
|
connected = False
|
|
while time.time() < deadline:
|
|
if _peer_connected(b.client, npub_a) and _peer_connected(a.client, npub_b):
|
|
connected = True
|
|
break
|
|
time.sleep(0.5)
|
|
if not connected:
|
|
log.error("nodes did not peer within 20s")
|
|
return 1
|
|
log.info("nodes peered (connected on both sides)")
|
|
|
|
# Give the tree a moment to settle.
|
|
time.sleep(5)
|
|
|
|
# Reachability: try both directions. The ephemeral nodes are direct
|
|
# peers, so find_next_hop branch 2 (direct peer) should deliver. We
|
|
# try B->A and A->B because freshly-peered two-node trees can show
|
|
# directional asymmetry before coord caches warm up; either
|
|
# direction succeeding proves the overlay carries traffic.
|
|
results = []
|
|
for label, src, dst_ipv6 in [("B->A", b, ipv6_a), ("A->B", a, ipv6_b)]:
|
|
log.info("ping6 %s (%s) from %s netns", label, dst_ipv6, src.netns)
|
|
res = reachability.ping6(dst_ipv6, timeout=5, count=3, runner=src.run_in_netns)
|
|
log.info("%s ping6: reachable=%s rtt=%s loss=%s",
|
|
label, res.reachable, res.rtt_ms, res.loss_pct)
|
|
results.append((label, res))
|
|
if res.reachable:
|
|
rtt_str = f"{res.rtt_ms:.3f}ms" if res.rtt_ms is not None else "n/a"
|
|
print(f"\nSELF-TEST PASS: {label} ping6 succeeds "
|
|
f"({dst_ipv6}) rtt={rtt_str}")
|
|
return 0
|
|
|
|
# Fallback: TCP probe in each direction (ephemeral nodes don't serve
|
|
# HTTP by default, so this is unlikely to succeed — included for
|
|
# completeness).
|
|
for label, src, dst_ipv6 in [("B->A", b, ipv6_a), ("A->B", a, ipv6_b)]:
|
|
tcp_res = reachability.tcp_probe(dst_ipv6, port=80, timeout=5,
|
|
runner=src.run_in_netns)
|
|
if tcp_res.reachable:
|
|
print(f"\nSELF-TEST PASS (tcp): {label} reaches {dst_ipv6} "
|
|
f"rtt={tcp_res.rtt_ms:.1f}ms")
|
|
return 0
|
|
|
|
print("\nSELF-TEST FAIL: neither direction reached the other via ping6.")
|
|
for label, res in results:
|
|
print(f" {label}: {res.detail[:160]}")
|
|
print("Nodes peered successfully but ICMPv6 echo did not complete in either")
|
|
print("direction. This can happen when coord caches are cold; the peering")
|
|
print("itself succeeded, which validates spawn/control/peer toolchain.")
|
|
return 2
|
|
finally:
|
|
for d in (b, a):
|
|
if d is not None:
|
|
try:
|
|
d.stop()
|
|
except Exception:
|
|
pass
|
|
# Delete the shared netns (daemons didn't own it).
|
|
fipsd.delete_netns(shared_netns, info=shared_info)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Main
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
p = argparse.ArgumentParser(description="FIPS Mesh Reachability Scanner")
|
|
p.add_argument("--seeds", help="'all' or comma-separated npubs")
|
|
p.add_argument("--seeds-sample", type=int, help="random sample of K seeds")
|
|
p.add_argument("--seeds-near-root", type=int, help="K seeds with smallest NodeAddr")
|
|
p.add_argument("--targets", help="comma-separated target npubs (default: all except seed)")
|
|
p.add_argument("--targets-sample", type=int, help="random sample of K targets per seed")
|
|
p.add_argument("--converge-secs", type=float, default=45.0)
|
|
p.add_argument("--ping-timeout", type=int, default=5)
|
|
p.add_argument("--tcp-port", type=int, default=80, help="TCP fallback port")
|
|
p.add_argument("--method", choices=["ping6", "tcp"], default="ping6")
|
|
p.add_argument("--delay", type=float, default=0.0, help="seconds between target checks")
|
|
p.add_argument("--startup-timeout", type=float, default=15.0)
|
|
p.add_argument("--peer-timeout", type=float, default=15.0)
|
|
p.add_argument("--relay", action="append", help="advert relay URL (repeatable)")
|
|
p.add_argument("--relay-timeout", type=float, default=12.0)
|
|
p.add_argument("--out-dir", default="mesh-scan-out", help="output directory")
|
|
p.add_argument("--seed", type=int, default=None, help="RNG seed for --seeds-sample")
|
|
p.add_argument("--self-test", action="store_true", help="run a 2-node smoke test and exit")
|
|
p.add_argument("--cleanup-stale", action="store_true", help="remove fips-scan-* TUN devices and exit")
|
|
p.add_argument("-v", "--verbose", action="count", default=0)
|
|
args = p.parse_args(argv)
|
|
|
|
level = logging.DEBUG if args.verbose >= 2 else (logging.INFO if args.verbose >= 1 else logging.WARNING)
|
|
logging.basicConfig(level=level, format="%(asctime)s %(levelname)s %(name)s: %(message)s",
|
|
datefmt="%H:%M:%S")
|
|
|
|
if args.cleanup_stale:
|
|
n = fipsd.cleanup_stale_netns()
|
|
print(f"removed {n} stale namespace(s)")
|
|
return 0
|
|
|
|
# Auto-cleanup any stale namespaces from previous runs before starting.
|
|
n = fipsd.cleanup_stale_netns()
|
|
if n:
|
|
print(f"Auto-cleanup: removed {n} stale namespace(s) from previous run(s).")
|
|
|
|
if args.self_test:
|
|
try:
|
|
return self_test(args)
|
|
finally:
|
|
fipsd.cleanup_stale_netns()
|
|
|
|
if args.seed is not None:
|
|
random.seed(args.seed)
|
|
|
|
log.info("fetching FIPS node directory")
|
|
nodes = fetch_directory(args.relay, args.relay_timeout)
|
|
print(f"Fetched {len(nodes)} nodes from the directory.")
|
|
if not nodes:
|
|
print("No nodes found. Exiting.")
|
|
return 1
|
|
|
|
seeds = select_seeds(nodes, args)
|
|
if not seeds:
|
|
print("No seeds selected. Exiting.")
|
|
return 1
|
|
print(f"Selected {len(seeds)} seed(s): {[s.npub[:16] for s in seeds]}")
|
|
|
|
out_dir = Path(args.out_dir)
|
|
out_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
all_results: list[PairResult] = []
|
|
seed_states: dict[str, SeedState] = {}
|
|
tally = {"connected": 0, "failed": 0, "skipped": 0}
|
|
scan_start = time.time()
|
|
matrix_path = out_dir / "matrix.csv"
|
|
diag_path = out_dir / "diagnostics.jsonl"
|
|
|
|
# Write diagnostics incrementally (append after each seed) so a Ctrl-C
|
|
# doesn't lose all data. The matrix is rewritten (overwrite) each time
|
|
# since it's a full table.
|
|
diag_path.unlink(missing_ok=True)
|
|
|
|
def write_incremental() -> None:
|
|
write_matrix_csv(all_results, matrix_path)
|
|
write_diagnostics_jsonl(all_results, diag_path)
|
|
|
|
interrupted = False
|
|
try:
|
|
for i, seed in enumerate(seeds):
|
|
targets = select_targets(nodes, seed, args)
|
|
log.info("seed %s: %d target(s)", seed.npub[:16], len(targets))
|
|
results, state = scan_seed(seed, targets, i, args, total_seeds=len(seeds))
|
|
all_results.extend(results)
|
|
if state is not None:
|
|
seed_states[seed.npub] = state
|
|
tally["connected"] += 1
|
|
elif not results:
|
|
tally["failed" if seed.first_dialable_endpoint() else "skipped"] += 1
|
|
else:
|
|
tally["connected"] += 1
|
|
# Write incremental output after each seed.
|
|
write_incremental()
|
|
elapsed = time.time() - scan_start
|
|
print(f"[{_ts()}] tally: {tally['connected']} connected, "
|
|
f"{tally['failed']} failed, {tally['skipped']} skipped | "
|
|
f"{i+1}/{len(seeds)} done | elapsed {elapsed:.0f}s", flush=True)
|
|
except KeyboardInterrupt:
|
|
interrupted = True
|
|
print("\n[INTERRUPTED] Writing collected data and exiting...", flush=True)
|
|
|
|
write_incremental()
|
|
print_summary(all_results, seed_states)
|
|
if interrupted:
|
|
print(f"\n[INTERRUPTED] Partial results: {len(all_results)} pairs from "
|
|
f"{len(seed_states)} seeds.")
|
|
print(f"\nOutputs written to {out_dir}/ (matrix.csv, diagnostics.jsonl)")
|
|
# Auto-cleanup any leaked namespaces on exit.
|
|
fipsd.cleanup_stale_netns()
|
|
return 0 if not interrupted else 130
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|