336 lines
12 KiB
Python
336 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""IPVarch test runner.
|
|
|
|
Runs every case in cases.py against the IPVarch binary and checks
|
|
validity (exit codes, output contracts, structural table rules) and,
|
|
with --sanitize, memory safety by rebuilding ipvarch.c with
|
|
AddressSanitizer + UndefinedBehaviorSanitizer via gcc/clang.
|
|
|
|
Usage:
|
|
python3 tests/runner.py [--bin ./ipvarch]
|
|
[--sanitize] [--filter substring-of-group-or-case]
|
|
"""
|
|
|
|
import argparse
|
|
import os
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
from cases import GROUPS # noqa: E402
|
|
|
|
REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
SAN_BIN = os.path.join("/tmp/opencode", "ipvarch-asan")
|
|
_usage_bin = None
|
|
|
|
# ---------------------------------------------------------------- table
|
|
|
|
|
|
def parse_table(out):
|
|
"""Parse a network-plan table into (rows dict, prefix or None)."""
|
|
rows = {}
|
|
prefix = None
|
|
for line in out.splitlines():
|
|
if not line.startswith("│"):
|
|
continue
|
|
if "IPVarch Network Plan" in line: # title row spans both cells
|
|
label = line.strip("│").strip()
|
|
m = re.search(r"/(\d+)$", label)
|
|
prefix = int(m.group(1)) if m else None
|
|
rows["__title__"] = label
|
|
continue
|
|
cells = [c.strip() for c in line.split("│")]
|
|
if len(cells) < 4:
|
|
continue
|
|
label, value = cells[1], cells[2]
|
|
if label:
|
|
rows[label] = value
|
|
return rows, prefix
|
|
|
|
|
|
def v_table_layout(out):
|
|
"""Every non-title table row must have exactly 3 vertical bars."""
|
|
rows, prefix = parse_table(out)
|
|
if not rows.get("__title__"):
|
|
return False, "no table title"
|
|
if prefix is None:
|
|
return False, "title missing prefix"
|
|
for line in out.splitlines():
|
|
if line.startswith("│") and "IPVarch Network Plan" not in line:
|
|
if line.count("│") != 3:
|
|
return False, "malformed row: %r" % line
|
|
return True, ""
|
|
|
|
|
|
def _ip_u32(s):
|
|
parts = [int(p) for p in s.split(".")]
|
|
if len(parts) != 4 or any(p < 0 or p > 255 for p in parts):
|
|
return None
|
|
return (parts[0] << 24) | (parts[1] << 16) | (parts[2] << 8) | parts[3]
|
|
|
|
|
|
def _u32_ip(u):
|
|
return "%u.%u.%u.%u" % ((u >> 24) & 255, (u >> 16) & 255,
|
|
(u >> 8) & 255, u & 255)
|
|
|
|
|
|
EXPECTED_HOSTS = {8: "16777214", 16: "65534", 24: "254", 32: "1"}
|
|
PREFIX_MASK = {8: 0xFF000000, 16: 0xFFFF0000, 24: 0xFFFFFF00, 32: 0xFFFFFFFF}
|
|
|
|
|
|
def v_plan(out):
|
|
ok, msg = v_table_layout(out)
|
|
if not ok:
|
|
return ok, msg
|
|
rows, prefix = parse_table(out)
|
|
labels = ["Network address", "Netmask", "Broadcast", "Usable host range",
|
|
"Usable hosts", "Class", "Scope", "Gateway IP",
|
|
"DNS/DHCP server", "DHCP pool"]
|
|
for label in labels:
|
|
if label not in rows:
|
|
return False, "missing row %r" % label
|
|
if rows["Usable hosts"] != EXPECTED_HOSTS.get(prefix):
|
|
return False, ("usable hosts %r != %r" %
|
|
(rows["Usable hosts"], EXPECTED_HOSTS.get(prefix)))
|
|
net = _ip_u32(rows["Network address"])
|
|
if net is None:
|
|
return False, "bad network address %r" % rows["Network address"]
|
|
if prefix == 32:
|
|
if not rows["Gateway IP"].startswith("n/a"):
|
|
return False, "/32 gateway should be n/a: %r" % rows["Gateway IP"]
|
|
if rows["DHCP pool"] != "disabled (single-host network)":
|
|
return False, "/32 DHCP pool wrong: %r" % rows["DHCP pool"]
|
|
return True, ""
|
|
first = (net & PREFIX_MASK[prefix]) + 1
|
|
if rows["Gateway IP"] != _u32_ip(first):
|
|
return False, "gateway %r != first usable %r" % (rows["Gateway IP"],
|
|
_u32_ip(first))
|
|
if rows["DNS/DHCP server"] != _u32_ip(first + 1):
|
|
return False, "DNS/DHCP %r != second usable %r" % (
|
|
rows["DNS/DHCP server"], _u32_ip(first + 1))
|
|
base = rows["Network address"].rsplit(".", 1)[0]
|
|
if rows["DHCP pool"] != base + ".100 - " + base + ".254":
|
|
return False, "DHCP pool %r != %r" % (rows["DHCP pool"],
|
|
base + ".100 - " + base + ".254")
|
|
return True, ""
|
|
|
|
|
|
def _is_private_net(net, prefix):
|
|
a = [(net >> 24) & 255, (net >> 16) & 255, (net >> 8) & 255, net & 255]
|
|
if a[0] == 10 or a[0] == 192 and a[1] == 168:
|
|
return True
|
|
if a[0] == 172 and 16 <= a[1] <= 31:
|
|
return True
|
|
return False
|
|
|
|
|
|
def _is_public_net(net):
|
|
a = [(net >> 24) & 255, (net >> 16) & 255, (net >> 8) & 255, net & 255]
|
|
if a[0] in (0, 10, 127):
|
|
return False
|
|
if a[0] == 172 and 16 <= a[1] <= 31:
|
|
return False
|
|
if a[0] == 192 and a[1] == 168:
|
|
return False
|
|
if a[0] == 169 and a[1] == 254:
|
|
return False
|
|
if a[0] == 100 and 64 <= a[1] <= 127:
|
|
return False
|
|
if a[0] == 198 and a[1] in (18, 19):
|
|
return False
|
|
if a[0] == 192 and a[1] == 0 and a[2] == 2:
|
|
return False
|
|
if a[0] == 198 and a[1] == 51 and a[2] == 100:
|
|
return False
|
|
if a[0] == 203 and a[1] == 0 and a[2] == 113:
|
|
return False
|
|
if a[0] >= 224:
|
|
return False
|
|
return True
|
|
|
|
|
|
def _random_check(out, want_private):
|
|
ok, msg = v_plan(out)
|
|
if not ok:
|
|
return ok, msg
|
|
rows, prefix = parse_table(out)
|
|
net = _ip_u32(rows["Network address"])
|
|
shape = "zeroed" if prefix != 32 else "any"
|
|
if want_private:
|
|
if not rows["Scope"].startswith("Private"):
|
|
return False, "scope should be Private: %r" % rows["Scope"]
|
|
if not _is_private_net(net, prefix):
|
|
return False, "network %r not private" % rows["Network address"]
|
|
else:
|
|
if not rows["Scope"].startswith("Public"):
|
|
return False, "scope should be Public: %r" % rows["Scope"]
|
|
if not _is_public_net(net):
|
|
return False, ("network %r is in a reserved/special range" %
|
|
rows["Network address"])
|
|
if prefix != 32:
|
|
expect_net = net & PREFIX_MASK[prefix]
|
|
if net != expect_net:
|
|
return False, "host bits set in random network %r" % _u32_ip(net)
|
|
expect_bc = net | (~PREFIX_MASK[prefix] & 0xFFFFFFFF)
|
|
if rows["Broadcast"] != _u32_ip(expect_bc):
|
|
return False, "broadcast %r wrong" % rows["Broadcast"]
|
|
return True, ""
|
|
|
|
|
|
def v_usage_examples(out):
|
|
"""Every <addr>/<prefix> example in usage output must be planned OK
|
|
when fed back to the binary."""
|
|
bin_path = _usage_bin
|
|
rows = []
|
|
for m in re.finditer(r"(\d{1,3}(?:\.\d{1,3}){3})/(\d+)", out):
|
|
rows.append((m.group(1), int(m.group(2))))
|
|
if not rows:
|
|
return False, "usage shows no <addr>/<prefix> examples"
|
|
for net, prefix in rows:
|
|
rc, out2, err2 = run_bin(bin_path, ["--plan-network",
|
|
"%s/%d" % (net, prefix)])
|
|
if rc != 0 or "IPVarch Network Plan" not in out2:
|
|
return False, ("usage example %s/%d is not a plannable network "
|
|
"(exit=%d, stderr=%r)" %
|
|
(net, prefix, rc, err2.strip()))
|
|
return True, ""
|
|
|
|
|
|
VALIDATORS = {
|
|
"plan": v_plan,
|
|
"plan-gateway": v_plan,
|
|
"table": v_table_layout,
|
|
"random-private": lambda out: _random_check(out, True),
|
|
"random-public": lambda out: _random_check(out, False),
|
|
"usage-examples": v_usage_examples,
|
|
}
|
|
|
|
# ---------------------------------------------------------------- run
|
|
|
|
|
|
def run_bin(bin_path, args, timeout=15):
|
|
try:
|
|
proc = subprocess.run([bin_path] + args, capture_output=True,
|
|
text=True, timeout=timeout)
|
|
except subprocess.TimeoutExpired:
|
|
return None, "TIMEOUT (%ss) on args=%r" % (timeout, args), ""
|
|
except OSError as exc:
|
|
return None, "cannot run binary: %s" % exc, ""
|
|
return proc.returncode, proc.stdout, proc.stderr
|
|
|
|
|
|
def check_once(bin_path, case):
|
|
rc, out, err = run_bin(bin_path, case["args"])
|
|
if rc is None:
|
|
return False, err
|
|
if rc != case["exit"]:
|
|
head = err.strip().splitlines()[0] if err.strip() else ""
|
|
if any(s in err for s in ("AddressSanitizer", "LeakSanitizer",
|
|
"runtime error:")):
|
|
return False, "sanitizer abort (rc=%d): %s" % (rc, head)
|
|
return False, "exit %d, expected %d; stderr=%r" % (rc, case["exit"],
|
|
head)
|
|
for sub in case["must"]:
|
|
hay = err if sub.startswith("stderr:") else out
|
|
clean = sub[len("stderr:"):] if sub.startswith("stderr:") else sub
|
|
if clean not in hay:
|
|
return False, "output missing %r" % clean
|
|
for sub in case["must_not"]:
|
|
if sub in out:
|
|
return False, "output unexpectedly contains %r" % sub
|
|
if case["validator"]:
|
|
ok, msg = VALIDATORS[case["validator"]](out)
|
|
if not ok:
|
|
return False, "validator %s: %s" % (case["validator"], msg)
|
|
return True, ""
|
|
|
|
|
|
def run_case(bin_path, case):
|
|
reps = case.get("repeat", 1)
|
|
for i in range(reps):
|
|
ok, msg = check_once(bin_path, case)
|
|
if not ok:
|
|
suffix = " (iteration %d/%d)" % (i + 1, reps) if reps > 1 else ""
|
|
return False, msg + suffix
|
|
return True, ""
|
|
|
|
|
|
def build_sanitize(repo):
|
|
cc = shutil.which("gcc") or shutil.which("clang")
|
|
if cc is None:
|
|
return None, "no ASan-capable compiler (gcc/clang) found"
|
|
cmd = [cc, "-std=c89", "-O1", "-g",
|
|
"-fsanitize=address,undefined", "-fno-omit-frame-pointer",
|
|
os.path.join(repo, "ipvarch.c"), "-o", SAN_BIN]
|
|
proc = subprocess.run(cmd, capture_output=True, text=True)
|
|
if proc.returncode != 0:
|
|
return None, "sanitized build failed:\n%s" % proc.stderr.strip()
|
|
return SAN_BIN, None
|
|
|
|
|
|
def main():
|
|
global _usage_bin
|
|
ap = argparse.ArgumentParser(description="IPVarch test runner")
|
|
ap.add_argument("--bin", default=os.path.join(REPO, "ipvarch"))
|
|
ap.add_argument("--sanitize", action="store_true",
|
|
help="rebuild with ASan+UBSan (gcc/clang) and rerun matrix")
|
|
ap.add_argument("--filter", default=None,
|
|
help="only run groups/cases whose name contains this")
|
|
args = ap.parse_args()
|
|
|
|
os.makedirs(os.path.dirname(SAN_BIN), exist_ok=True)
|
|
memory = "not instrumented (run 'make test-sanitize' to enable)"
|
|
|
|
if args.sanitize:
|
|
bin_path, err = build_sanitize(REPO)
|
|
if bin_path is None:
|
|
print("sanitize: SKIPPED (%s)" % err)
|
|
return 0
|
|
memory = "ASan+UBSan instrumentation active"
|
|
else:
|
|
bin_path = args.bin
|
|
|
|
_usage_bin = bin_path
|
|
|
|
if not os.path.exists(bin_path):
|
|
print("error: binary not found: %s (run 'make' first)" % bin_path,
|
|
file=sys.stderr)
|
|
return 1
|
|
|
|
passed = failed = 0
|
|
failures = []
|
|
ran_any = False
|
|
|
|
for gname, cases in GROUPS:
|
|
if args.filter and args.filter not in gname and \
|
|
not any(args.filter in case["name"] for case in cases):
|
|
continue
|
|
print("== %s ==" % gname)
|
|
for case in cases:
|
|
if args.filter and args.filter not in case["name"]:
|
|
continue
|
|
ran_any = True
|
|
ok, msg = run_case(bin_path, case)
|
|
if ok:
|
|
passed += 1
|
|
print(" PASS %s" % case["name"])
|
|
else:
|
|
failed += 1
|
|
failures.append((gname, case["name"], msg))
|
|
print(" FAIL %s\n %s" % (case["name"], msg))
|
|
print()
|
|
|
|
if not ran_any:
|
|
print("error: filter %r matched nothing" % args.filter, file=sys.stderr)
|
|
return 1
|
|
|
|
print("Summary: %d passed, %d failed [binary: %s]" %
|
|
(passed, failed, bin_path))
|
|
print(" memory safety: %s" % memory)
|
|
return 1 if failed else 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|