evgeny
5 days ago
2 changed files with 345 additions and 0 deletions
@ -0,0 +1,206 @@
|
||||
"""Сетевые операции допускаются только в namespace, созданных этим запуском.""" |
||||
|
||||
import json |
||||
import os |
||||
from pathlib import Path |
||||
import subprocess |
||||
import time |
||||
|
||||
from topology import COUNTRIES, NATS, NODES, WANS, wan |
||||
|
||||
|
||||
def nat_rules(gateway): |
||||
clients = [node for node in NODES if node.nat == gateway.name] |
||||
destinations = sorted({n.ip for n in NODES if not n.nat} | {g.ip for g in NATS}) |
||||
lines = ["table ip lab {", |
||||
" set active { type inet_service; flags dynamic,timeout; timeout 60s; }", |
||||
" set contacts { type ipv4_addr . inet_service . inet_service; flags dynamic,timeout; timeout 60s; }", |
||||
" chain refresh { type filter hook postrouting priority mangle; policy accept;"] |
||||
for node in clients: |
||||
match = f' oifname "wan" ip saddr {node.ip} udp sport {node.udp_port}' |
||||
if gateway.mode == "cone": |
||||
lines.append(f"{match} update @active {{ {node.mapped_port} timeout 60s }}") |
||||
elif gateway.mode == "restricted": |
||||
lines.append(f"{match} update @contacts {{ ip daddr . udp dport . {node.mapped_port} timeout 60s }}") |
||||
lines += [" }", " chain prerouting { type nat hook prerouting priority dstnat; policy accept;"] |
||||
for node in clients: |
||||
match = f' iifname "wan" udp dport {node.mapped_port}' |
||||
if gateway.mode == "cone": |
||||
lines.append(f"{match} udp dport @active dnat to {node.ip}:{node.udp_port}") |
||||
elif gateway.mode == "restricted": |
||||
lines.append(f"{match} ip saddr . udp sport . udp dport @contacts dnat to {node.ip}:{node.udp_port}") |
||||
lines += [" }", " chain postrouting { type nat hook postrouting priority srcnat; policy accept;"] |
||||
for node in clients: |
||||
match = f' oifname "wan" ip saddr {node.ip} udp sport {node.udp_port}' |
||||
if gateway.mode == "strict": |
||||
for index, destination in enumerate(destinations): |
||||
port = 40000 + node.number * 100 + index |
||||
lines.append(f"{match} ip daddr {destination} snat to {gateway.ip}:{port}") |
||||
else: |
||||
lines.append(f"{match} snat to {gateway.ip}:{node.mapped_port}") |
||||
lines += [' oifname "wan" masquerade', " }", |
||||
" chain forward { type filter hook forward priority filter; policy drop;", |
||||
" ct state established,related accept", ' iifname "lan" oifname "wan" accept', |
||||
" ct status dnat accept", " counter drop", " }", "}"] |
||||
return "\n".join(lines) + "\n" |
||||
|
||||
|
||||
class Network: |
||||
def __init__(self, directory, seed, host_namespace): |
||||
self.directory = Path(directory) |
||||
self.seed = seed |
||||
self.host_namespace = host_namespace |
||||
self.controller = os.readlink("/proc/self/ns/net") |
||||
if self.controller == host_namespace or not host_namespace: |
||||
raise RuntimeError("Refusing network changes: controller must be in a new network namespace") |
||||
self.namespaces = {} |
||||
self.children = [] |
||||
self.serial = 0 |
||||
self.log = (self.directory / "network.jsonl").open("w", buffering=1) |
||||
|
||||
def command(self, name, arguments, *, input=None, check=True): |
||||
prefix = self.prefix(name) |
||||
started = time.monotonic() |
||||
result = subprocess.run(prefix + list(arguments), input=input, text=True, capture_output=True, timeout=20) |
||||
self.log.write(json.dumps({"time": started, "namespace": name, "argv": list(arguments), |
||||
"rc": result.returncode, "stdout": result.stdout, "stderr": result.stderr}) + "\n") |
||||
if check and result.returncode: |
||||
raise RuntimeError(f"{name}: {' '.join(arguments)}: {result.stderr.strip() or result.stdout.strip()}") |
||||
return result |
||||
|
||||
def prefix(self, name): |
||||
if os.readlink("/proc/self/ns/net") != self.controller: |
||||
raise RuntimeError("Controller changed network namespace") |
||||
if name is None: |
||||
return [] |
||||
child, inode = self.namespaces[name] |
||||
path = f"/proc/{child.pid}/ns/net" |
||||
if child.poll() is not None or os.readlink(path) != inode or inode == self.host_namespace: |
||||
raise RuntimeError(f"Namespace {name} is no longer owned by this run") |
||||
return ["nsenter", f"--net={path}", "--"] |
||||
|
||||
def spawn(self, name, arguments, **kwargs): |
||||
child = subprocess.Popen(self.prefix(name) + list(arguments), **kwargs) |
||||
self.children.append(child) |
||||
return child |
||||
|
||||
def create_namespace(self, name): |
||||
keeper = "import os,sys; print(os.getpid(),flush=True); sys.stdin.buffer.read()" |
||||
child = subprocess.Popen(["unshare", "--net", "--", "python3", "-u", "-c", keeper], |
||||
stdin=subprocess.PIPE, stdout=subprocess.PIPE, text=True) |
||||
self.children.append(child) |
||||
if child.stdout.readline().strip() != str(child.pid): |
||||
raise RuntimeError(f"Cannot create namespace {name}") |
||||
inode = os.readlink(f"/proc/{child.pid}/ns/net") |
||||
if inode in {self.controller, self.host_namespace}: |
||||
raise RuntimeError(f"Namespace isolation failed for {name}") |
||||
self.namespaces[name] = (child, inode) |
||||
self.command(name, ["ip", "link", "set", "lo", "up"]) |
||||
self.command(name, ["sysctl", "-qw", "net.ipv6.conf.all.disable_ipv6=1"]) |
||||
self.command(name, ["sysctl", "-qw", "net.ipv4.conf.all.rp_filter=0"]) |
||||
self.command(name, ["sysctl", "-qw", "net.ipv4.conf.default.rp_filter=0"]) |
||||
|
||||
def link(self, left, left_interface, right, right_interface): |
||||
self.serial += 1 |
||||
a, b = f"v{self.serial}a", f"v{self.serial}b" |
||||
self.command(None, ["ip", "link", "add", a, "type", "veth", "peer", "name", b]) |
||||
for temporary, name, interface in ((a, left, left_interface), (b, right, right_interface)): |
||||
self.command(None, ["ip", "link", "set", temporary, "netns", str(self.namespaces[name][0].pid)]) |
||||
self.command(name, ["ip", "link", "set", temporary, "name", interface]) |
||||
self.command(name, ["ethtool", "-K", interface, "gro", "off", "gso", "off", "tso", "off"]) |
||||
self.command(name, ["ip", "link", "set", interface, "up"]) |
||||
|
||||
def bridge(self, name, address): |
||||
self.command(name, ["ip", "link", "add", "lan", "type", "bridge"]) |
||||
self.command(name, ["ip", "addr", "add", address + "/24", "dev", "lan"]) |
||||
self.command(name, ["ip", "link", "set", "lan", "up"]) |
||||
self.command(name, ["sysctl", "-qw", "net.ipv4.ip_forward=1"]) |
||||
|
||||
def attach(self, node, gateway, port): |
||||
self.link(node, "eth0" if node.startswith("n") else "wan", gateway, port) |
||||
self.command(gateway, ["ip", "link", "set", port, "master", "lan"]) |
||||
|
||||
def netem(self, node, source_country, *, rtt=None, loss=None): |
||||
index = next(i + 1 for i, c in enumerate(COUNTRIES) if c.name == source_country) |
||||
same = source_country == node.country |
||||
link = None if same else wan(source_country, node.country) |
||||
delay = (2 if same else link.rtt_ms) / 2 if rtt is None else rtt / 2 |
||||
jitter = 0.2 if same else link.jitter_ms |
||||
loss = (0 if same else link.loss_percent) if loss is None else loss |
||||
self.command(node.name, ["tc", "qdisc", "replace", "dev", "ifb0", "parent", f"1:{index}", |
||||
"handle", f"{index + 10}:", "netem", "limit", "4096", "delay", f"{delay}ms", |
||||
f"{jitter}ms", "distribution", "normal", "loss", "random", f"{loss}%", |
||||
"seed", str(self.seed + node.number * 10 + index)]) |
||||
|
||||
def receiver(self, node): |
||||
self.command(node.name, ["ip", "link", "add", "ifb0", "type", "ifb"]) |
||||
self.command(node.name, ["ip", "link", "set", "ifb0", "up"]) |
||||
self.command(node.name, ["tc", "qdisc", "add", "dev", "eth0", "clsact"]) |
||||
self.command(node.name, ["tc", "filter", "add", "dev", "eth0", "ingress", "matchall", "action", |
||||
"mirred", "egress", "redirect", "dev", "ifb0"]) |
||||
own = next(i + 1 for i, c in enumerate(COUNTRIES) if c.name == node.country) |
||||
self.command(node.name, ["tc", "qdisc", "add", "dev", "ifb0", "root", "handle", "1:", "htb", "default", str(own)]) |
||||
for index, country in enumerate(COUNTRIES, 1): |
||||
self.command(node.name, ["tc", "class", "add", "dev", "ifb0", "parent", "1:", "classid", f"1:{index}", |
||||
"htb", "rate", "1gbit", "ceil", "1gbit"]) |
||||
self.netem(node, country.name) |
||||
self.command(node.name, ["tc", "filter", "add", "dev", "ifb0", "parent", "1:", "protocol", "ip", |
||||
"prio", str(index), "flower", "src_ip", country.subnet, "classid", f"1:{index}"]) |
||||
|
||||
def bandwidth(self, link, mbps): |
||||
for country in (link.a, link.b): |
||||
self.command("r" + country, ["tc", "qdisc", "replace", "dev", "w_" + link.a + link.b, |
||||
"root", "tbf", "rate", f"{mbps}mbit", "burst", "64kb", "latency", "100ms"]) |
||||
|
||||
def build(self): |
||||
for name in [n.name for n in NODES] + [c.router for c in COUNTRIES] + [g.name for g in NATS]: |
||||
self.create_namespace(name) |
||||
for country in COUNTRIES: |
||||
self.bridge(country.router, country.gateway) |
||||
for gateway in NATS: |
||||
self.bridge(gateway.name, "10.64.1.1") |
||||
self.attach(gateway.name, "r" + gateway.country, "p" + gateway.name) |
||||
self.command(gateway.name, ["ip", "addr", "add", gateway.ip + "/24", "dev", "wan"]) |
||||
country = next(c for c in COUNTRIES if c.name == gateway.country) |
||||
self.command(gateway.name, ["ip", "route", "add", "default", "via", country.gateway]) |
||||
rules = nat_rules(gateway) |
||||
(self.directory / f"{gateway.name}.nft").write_text(rules) |
||||
self.command(gateway.name, ["nft", "-f", "-"], input=rules) |
||||
for node in NODES: |
||||
self.attach(node.name, node.nat or "r" + node.country, "p" + node.name) |
||||
self.command(node.name, ["ip", "addr", "add", node.ip + "/24", "dev", "eth0"]) |
||||
self.command(node.name, ["ip", "route", "add", "default", "via", node.gateway]) |
||||
self.receiver(node) |
||||
for link in WANS: |
||||
interface = "w_" + link.a + link.b |
||||
self.link("r" + link.a, interface, "r" + link.b, interface) |
||||
for country, address, peer, destination in ((link.a, link.ip_a, link.ip_b, link.b), |
||||
(link.b, link.ip_b, link.ip_a, link.a)): |
||||
subnet = next(c.subnet for c in COUNTRIES if c.name == destination) |
||||
self.command("r" + country, ["ip", "addr", "add", address + "/30", "dev", interface]) |
||||
self.command("r" + country, ["ip", "route", "add", subnet, "via", peer]) |
||||
self.bandwidth(link, link.mbps) |
||||
state = {name: {"pid": child.pid, "namespace": inode} for name, (child, inode) in self.namespaces.items()} |
||||
(self.directory / "namespaces.json").write_text(json.dumps(state, indent=2) + "\n") |
||||
|
||||
def snapshot(self, label): |
||||
snapshot = {} |
||||
for name in self.namespaces: |
||||
snapshot[name] = {"links": json.loads(self.command(name, ["ip", "-j", "link"]).stdout), |
||||
"routes": json.loads(self.command(name, ["ip", "-j", "route"]).stdout), |
||||
"qdisc": json.loads(self.command(name, ["tc", "-j", "-s", "qdisc"]).stdout)} |
||||
if name.startswith("g_"): |
||||
snapshot[name]["nft"] = json.loads(self.command(name, ["nft", "-j", "list", "ruleset"]).stdout) |
||||
(self.directory / f"network-{label}.json").write_text(json.dumps(snapshot, indent=2) + "\n") |
||||
|
||||
def close(self): |
||||
for child in reversed(self.children): |
||||
if child.poll() is None: |
||||
child.terminate() |
||||
for child in reversed(self.children): |
||||
try: |
||||
child.wait(timeout=3) |
||||
except subprocess.TimeoutExpired: |
||||
child.kill() |
||||
child.wait(timeout=3) |
||||
self.log.close() |
||||
@ -0,0 +1,139 @@
|
||||
"""UDP-калибровка из настоящих namespace, до запуска ETCP.""" |
||||
|
||||
import argparse |
||||
import json |
||||
from pathlib import Path |
||||
import select |
||||
import socket |
||||
import statistics |
||||
import sys |
||||
import time |
||||
import uuid |
||||
|
||||
from topology import NATS, NODES, WANS |
||||
|
||||
|
||||
def serve(address): |
||||
udp = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) |
||||
udp.bind((address, 32000)) |
||||
udp.setblocking(False) |
||||
print(json.dumps({"ready": address}), flush=True) |
||||
|
||||
def receive(): |
||||
data, peer = udp.recvfrom(4096) |
||||
message = json.loads(data) |
||||
if message.get("request"): |
||||
udp.sendto(json.dumps({"token": message["token"], "observed": peer}).encode(), peer) |
||||
return None |
||||
return message |
||||
|
||||
while True: |
||||
readable, _, _ = select.select([sys.stdin, udp], [], []) |
||||
if udp in readable: |
||||
receive() |
||||
if sys.stdin not in readable: |
||||
continue |
||||
line = sys.stdin.readline() |
||||
if not line: |
||||
break |
||||
command = json.loads(line) |
||||
if command.get("quit"): |
||||
break |
||||
samples, observed, lost = [], None, 0 |
||||
for _ in range(command.get("count", 1)): |
||||
token = uuid.uuid4().hex |
||||
packet = json.dumps({"request": True, "token": token}).encode() |
||||
started = time.monotonic() |
||||
udp.sendto(packet, (command["ip"], command["port"])) |
||||
deadline = started + command.get("timeout", 0.8) |
||||
while time.monotonic() < deadline: |
||||
if not select.select([udp], [], [], max(0, deadline - time.monotonic()))[0]: |
||||
break |
||||
reply = receive() |
||||
if reply and reply.get("token") == token: |
||||
samples.append((time.monotonic() - started) * 1000) |
||||
observed = reply["observed"] |
||||
break |
||||
else: |
||||
pass |
||||
if len(samples) + lost < _ + 1: |
||||
lost += 1 |
||||
print(json.dumps({"samples_ms": samples, "observed": observed, "lost": lost}), flush=True) |
||||
udp.close() |
||||
|
||||
|
||||
class Probes: |
||||
def __init__(self, network): |
||||
self.network = network |
||||
self.processes = {} |
||||
|
||||
def start(self): |
||||
for node in NODES: |
||||
process = self.network.spawn(node.name, [sys.executable, "-u", str(Path(__file__).resolve()), "--serve", node.ip], |
||||
stdin=-1, stdout=-1, text=True) |
||||
if not select.select([process.stdout], [], [], 5)[0]: |
||||
raise RuntimeError(f"Probe {node.name} failed to start") |
||||
ready = json.loads(process.stdout.readline()) |
||||
if ready != {"ready": node.ip}: |
||||
raise RuntimeError(f"Unexpected probe handshake: {ready}") |
||||
self.processes[node.name] = process |
||||
|
||||
def exchange(self, name, ip, port=32000, count=3, timeout=0.8): |
||||
process = self.processes[name] |
||||
process.stdin.write(json.dumps({"ip": ip, "port": port, "count": count, "timeout": timeout}) + "\n") |
||||
process.stdin.flush() |
||||
if not select.select([process.stdout], [], [], count * timeout + 5)[0]: |
||||
raise RuntimeError(f"Probe {name} command timed out") |
||||
line = process.stdout.readline() |
||||
if not line: |
||||
raise RuntimeError(f"Probe {name} exited: {process.poll()}") |
||||
return json.loads(line) |
||||
|
||||
def calibrate(self): |
||||
results = {"wan": [], "nat": []} |
||||
public = {n.country: n for n in NODES if not n.nat and n.client_type == 0} |
||||
for link in WANS: |
||||
result = self.exchange(public[link.a].name, public[link.b].ip, count=40) |
||||
if len(result["samples_ms"]) < 30: |
||||
raise RuntimeError(f"WAN {link.a}-{link.b}: excessive loss: {result}") |
||||
median = statistics.median(result["samples_ms"]) |
||||
if abs(median - link.rtt_ms) > max(12, link.rtt_ms * 0.25): |
||||
raise RuntimeError(f"WAN {link.a}-{link.b}: measured RTT {median:.1f}ms, expected {link.rtt_ms}ms") |
||||
results["wan"].append(dict(result, a=link.a, b=link.b, expected_rtt_ms=link.rtt_ms, median_ms=median)) |
||||
for gateway in NATS: |
||||
for node in (n for n in NODES if n.nat == gateway.name): |
||||
a, b, c = public["a"], public["b"], public["c"] |
||||
before = self.exchange(c.name, gateway.ip, node.mapped_port, count=1, timeout=0.35) |
||||
if before["observed"]: |
||||
raise RuntimeError(f"NAT {node.name}: incoming mapping exists before outbound traffic") |
||||
first = self.exchange(node.name, a.ip) |
||||
second = self.exchange(node.name, b.ip) |
||||
if not first["observed"] or not second["observed"]: |
||||
raise RuntimeError(f"NAT {node.name}: outgoing UDP failed: {first}, {second}") |
||||
stable = first["observed"] == second["observed"] |
||||
third = self.exchange(c.name, gateway.ip, first["observed"][1], count=3, timeout=0.4) |
||||
inbound = bool(third["observed"]) |
||||
expected = (gateway.mode != "strict", gateway.mode == "cone") |
||||
if (stable, inbound) != expected: |
||||
raise RuntimeError(f"NAT {node.name}/{gateway.mode}: stable={stable}, unsolicited={inbound}, expected={expected}") |
||||
results["nat"].append({"node": node.name, "gateway": gateway.name, "mode": gateway.mode, |
||||
"first": first["observed"], "second": second["observed"], |
||||
"stable": stable, "unsolicited": inbound}) |
||||
return results |
||||
|
||||
def close(self): |
||||
for process in self.processes.values(): |
||||
if process.poll() is None: |
||||
process.stdin.close() |
||||
for process in self.processes.values(): |
||||
try: |
||||
process.wait(timeout=3) |
||||
except Exception: |
||||
process.kill() |
||||
process.wait() |
||||
|
||||
|
||||
if __name__ == "__main__": |
||||
parser = argparse.ArgumentParser() |
||||
parser.add_argument("--serve", required=True) |
||||
serve(parser.parse_args().serve) |
||||
Loading…
Reference in new issue