368 lines
17 KiB
Python
368 lines
17 KiB
Python
"""Local, single-interface WireGuard controller. Never executes wg-quick/hooks."""
|
|
import base64
|
|
import binascii
|
|
import http.client
|
|
import http.server
|
|
import ipaddress
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import socket
|
|
import socketserver
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
|
|
IFACE = "wg-flow"
|
|
CONFIG = Path(os.environ.get("FLOW_AGENT_CONFIG_DIR", "/etc/wireguard"))
|
|
SOCKET = os.environ.get("FLOW_AGENT_SOCKET", "/run/mm-netflow-host/agent.sock")
|
|
LOCK = threading.RLock()
|
|
MAX_BODY = 262144
|
|
|
|
|
|
class AgentError(Exception):
|
|
pass
|
|
|
|
|
|
def run(*args, input=None):
|
|
try:
|
|
result = subprocess.run(args, input=input, text=True, capture_output=True, timeout=10)
|
|
except (OSError, subprocess.TimeoutExpired):
|
|
raise AgentError("Команда WireGuard/IP недоступна или превысила время ожидания") from None
|
|
if result.returncode:
|
|
# stderr/stdout can contain keys and must never reach an HTTP response/log.
|
|
raise AgentError("Операция WireGuard/IP не выполнена")
|
|
return result.stdout.strip()
|
|
|
|
|
|
def key(value):
|
|
try:
|
|
if not isinstance(value, str) or len(value) != 44 or len(base64.b64decode(value, validate=True)) != 32:
|
|
raise ValueError()
|
|
except (ValueError, binascii.Error):
|
|
raise AgentError("Некорректный ключ WireGuard") from None
|
|
return value
|
|
|
|
|
|
def port(value):
|
|
if type(value) is not int or not 1 <= value <= 65535:
|
|
raise AgentError("Некорректный UDP-порт")
|
|
return value
|
|
|
|
|
|
def endpoint(value):
|
|
if not isinstance(value, str) or not re.fullmatch(r"(?:\[[0-9a-fA-F:]+\]|[a-zA-Z0-9._-]+):[0-9]{1,5}", value):
|
|
raise AgentError("Некорректный endpoint")
|
|
port(int(value.rsplit(":", 1)[1]))
|
|
return value
|
|
|
|
|
|
def parse_conf(text):
|
|
interface, peers, section = {}, [], None
|
|
allowed = {"interface": {"PrivateKey", "Address", "ListenPort", "MTU"},
|
|
"peer": {"PublicKey", "PresharedKey", "AllowedIPs", "Endpoint", "PersistentKeepalive"}}
|
|
for raw in text.splitlines():
|
|
line = raw.split("#", 1)[0].strip()
|
|
if not line:
|
|
continue
|
|
if line == "[Interface]" and section is None:
|
|
section, current = "interface", interface
|
|
elif line == "[Peer]" and section is not None:
|
|
section, current = "peer", {}
|
|
peers.append(current)
|
|
else:
|
|
name, sep, value = line.partition("=")
|
|
name, value = name.strip(), value.strip()
|
|
if not sep or section is None or name not in allowed[section] or name in current or not value:
|
|
raise AgentError("Конфигурация wg-flow содержит неподдерживаемые или повторные параметры; требуется ручная проверка")
|
|
current[name] = value
|
|
if not interface or not {"PrivateKey", "Address", "ListenPort"} <= interface.keys():
|
|
raise AgentError("Неполная конфигурация wg-flow")
|
|
key(interface["PrivateKey"])
|
|
try:
|
|
ipaddress.IPv4Interface(interface["Address"])
|
|
port(int(interface["ListenPort"]))
|
|
if "MTU" in interface and not 576 <= int(interface["MTU"]) <= 9000:
|
|
raise ValueError()
|
|
for peer in peers:
|
|
key(peer.get("PublicKey"))
|
|
if "PresharedKey" in peer:
|
|
key(peer["PresharedKey"])
|
|
if "Endpoint" in peer:
|
|
endpoint(peer["Endpoint"])
|
|
if peer.get("AllowedIPs"):
|
|
networks(peer["AllowedIPs"].split(","))
|
|
if "PersistentKeepalive" in peer and not 0 <= int(peer["PersistentKeepalive"]) <= 65535:
|
|
raise ValueError()
|
|
except (ValueError, TypeError):
|
|
raise AgentError("Некорректные параметры wg-flow") from None
|
|
if len({p["PublicKey"] for p in peers}) != len(peers):
|
|
raise AgentError("Повторный публичный ключ пира")
|
|
return interface, peers
|
|
|
|
|
|
def networks(values):
|
|
if not isinstance(values, list) or not values:
|
|
raise AgentError("AllowedIPs должны содержать адреса")
|
|
try:
|
|
return [ipaddress.ip_network(v.strip(), strict=False) for v in values]
|
|
except (ValueError, AttributeError):
|
|
raise AgentError("Некорректные AllowedIPs") from None
|
|
|
|
|
|
def validate(raw):
|
|
fields = {"collectorIp", "prefix", "flowListenPort", "wgListenPort", "hostPublicKey", "hostPrivateKey", "peers"}
|
|
if not isinstance(raw, dict) or set(raw) != fields:
|
|
raise AgentError("Некорректный снимок NetFlow")
|
|
try:
|
|
collector = ipaddress.IPv4Address(raw["collectorIp"])
|
|
prefix = ipaddress.IPv4Network(raw["prefix"], strict=False)
|
|
if prefix.prefixlen != 24 or collector not in prefix or collector in (prefix.network_address, prefix.broadcast_address):
|
|
raise ValueError()
|
|
except (ValueError, TypeError):
|
|
raise AgentError("Требуется IPv4 overlay /24 и адрес коллектора внутри него") from None
|
|
port(raw["flowListenPort"])
|
|
port(raw["wgListenPort"])
|
|
key(raw["hostPublicKey"])
|
|
key(raw["hostPrivateKey"])
|
|
if run("wg", "pubkey", input=raw["hostPrivateKey"] + "\n") != raw["hostPublicKey"]:
|
|
raise AgentError("Публичный и приватный ключи MM не соответствуют друг другу")
|
|
binding = json.loads((CONFIG / "binding.json").read_text())
|
|
for name in ("collectorIp", "prefix", "flowListenPort", "wgListenPort"):
|
|
if binding.get(name) != raw[name]:
|
|
raise AgentError("Адрес, префикс или порт отличаются от установки агента; повторите установщик")
|
|
if not isinstance(raw["peers"], list) or len(raw["peers"]) > 253:
|
|
raise AgentError("Некорректный список пиров")
|
|
peers, ids, keys = [], set(), set()
|
|
for peer in raw["peers"]:
|
|
if not isinstance(peer, dict) or not {"serverId", "publicKey", "allowedIps"} <= peer.keys() or set(peer) - {"serverId", "publicKey", "allowedIps", "endpoint"}:
|
|
raise AgentError("Некорректный пир")
|
|
peer_key = key(peer["publicKey"])
|
|
if type(peer["serverId"]) is not int or peer["serverId"] <= 0 or peer["serverId"] in ids or peer_key in keys or peer_key == raw["hostPublicKey"]:
|
|
raise AgentError("Повторный или некорректный пир")
|
|
ids.add(peer["serverId"])
|
|
keys.add(peer_key)
|
|
nets = networks(peer["allowedIps"])
|
|
if len(nets) != 1 or nets[0].version != 4 or nets[0].prefixlen != 32 or nets[0].network_address not in prefix or nets[0].network_address in (collector, prefix.network_address, prefix.broadcast_address):
|
|
raise AgentError("Пиру разрешён только собственный IPv4 /32 внутри overlay")
|
|
data = {"PublicKey": peer_key, "AllowedIPs": str(nets[0])}
|
|
if peer.get("endpoint"):
|
|
data.update(Endpoint=endpoint(peer["endpoint"]), PersistentKeepalive="25")
|
|
peers.append(data)
|
|
return peers
|
|
|
|
|
|
def merge_peers(existing, desired):
|
|
merged = {peer["PublicKey"]: dict(peer) for peer in existing}
|
|
for peer in desired:
|
|
public = peer["PublicKey"]
|
|
# Replacing a PSK would silently break a previously working tunnel.
|
|
if merged.get(public, {}).get("PresharedKey", "A" * 43 + "=") != "A" * 43 + "=":
|
|
raise AgentError("Управляемый пир использует preshared key; требуется ручная проверка")
|
|
merged[public] = dict(peer)
|
|
rows = list(merged.values())
|
|
wanted = {p["PublicKey"] for p in desired}
|
|
for i, left in enumerate(rows):
|
|
for right in rows[i + 1:]:
|
|
if left["PublicKey"] not in wanted and right["PublicKey"] not in wanted:
|
|
continue
|
|
if not left.get("AllowedIPs") or not right.get("AllowedIPs"):
|
|
continue
|
|
if any(a.version == b.version and a.overlaps(b) for a in networks(left["AllowedIPs"].split(",")) for b in networks(right["AllowedIPs"].split(","))):
|
|
raise AgentError("AllowedIPs управляемого пира пересекаются с другим пиром")
|
|
return rows
|
|
|
|
|
|
def render(interface, peers, native=False):
|
|
rows = ["[Interface]"]
|
|
rows += [f"{k} = {v}" for k, v in interface.items() if not native or k not in ("Address", "MTU")]
|
|
for peer in peers:
|
|
rows += ["", "[Peer]"] + [f"{k} = {v}" for k, v in peer.items() if k != "AllowedIPs" or v]
|
|
return "\n".join(rows) + "\n"
|
|
|
|
|
|
def atomic_write(path, text):
|
|
fd, name = tempfile.mkstemp(prefix=".flow-", dir=path.parent)
|
|
try:
|
|
with os.fdopen(fd, "w") as stream:
|
|
stream.write(text)
|
|
stream.flush()
|
|
os.fsync(stream.fileno())
|
|
os.replace(name, path)
|
|
if os.name == "posix":
|
|
directory = os.open(path.parent, os.O_RDONLY)
|
|
try:
|
|
os.fsync(directory)
|
|
finally:
|
|
os.close(directory)
|
|
finally:
|
|
if os.path.exists(name):
|
|
os.unlink(name)
|
|
|
|
|
|
def sync(text):
|
|
# wg requires a file. mkstemp creates it with mode 0600.
|
|
fd, name = tempfile.mkstemp(prefix="flow-sync-")
|
|
try:
|
|
with os.fdopen(fd, "w") as stream:
|
|
stream.write(text)
|
|
run("wg", "syncconf", IFACE, name)
|
|
finally:
|
|
os.unlink(name)
|
|
|
|
|
|
def live():
|
|
dump = run("wg", "show", IFACE, "dump").splitlines()
|
|
if not dump or len(dump[0].split("\t")) != 4:
|
|
raise AgentError("Интерфейс wg-flow недоступен")
|
|
interface = dump[0].split("\t")
|
|
peers, raw_peers = [], []
|
|
for line in dump[1:]:
|
|
cols = line.split("\t")
|
|
if len(cols) != 8:
|
|
raise AgentError("Некорректный ответ WireGuard")
|
|
public, psk, ep, ips, handshake, rx, tx, keepalive = cols
|
|
peers.append({"publicKey": public, "allowedIps": ips.split(",") if ips != "(none)" else [],
|
|
"lastHandshakeAt": int(handshake) or None, "receivedBytes": int(rx), "sentBytes": int(tx)})
|
|
item = {"PublicKey": public, "AllowedIPs": ips if ips != "(none)" else ""}
|
|
if psk != "(none)":
|
|
item["PresharedKey"] = psk
|
|
if ep != "(none)":
|
|
item["Endpoint"] = ep
|
|
item["PersistentKeepalive"] = keepalive
|
|
raw_peers.append(item)
|
|
addresses = json.loads(run("ip", "-j", "-4", "address", "show", "dev", IFACE))
|
|
cidrs = [f"{a['local']}/{a['prefixlen']}" for row in addresses for a in row.get("addr_info", []) if a.get("family") == "inet"]
|
|
return {"interfaceUp": any("UP" in row.get("flags", []) for row in addresses), "publicKey": interface[1], "listenPort": int(interface[2]), "addresses": cidrs, "peers": peers}, raw_peers
|
|
|
|
|
|
def check(raw, apply=False):
|
|
desired = validate(raw)
|
|
path = CONFIG / "wg-flow.conf"
|
|
previous = path.read_text()
|
|
interface, stored = parse_conf(previous)
|
|
expected_address = f"{raw['collectorIp']}/{raw['prefix'].split('/')[1]}"
|
|
if run("wg", "pubkey", input=interface["PrivateKey"] + "\n") != raw["hostPublicKey"] or interface["Address"] != expected_address or int(interface["ListenPort"]) != raw["wgListenPort"]:
|
|
raise AgentError("Ключ, адрес или порт существующего wg-flow отличаются от MM; требуется ручная проверка")
|
|
status, runtime = live()
|
|
if not status["interfaceUp"]:
|
|
raise AgentError("Интерфейс wg-flow выключен; проверьте wg-quick на хосте")
|
|
if status["publicKey"] != raw["hostPublicKey"] or expected_address not in status["addresses"] or status["listenPort"] != raw["wgListenPort"]:
|
|
raise AgentError("Работающий wg-flow отличается от конфигурации; требуется ручная проверка")
|
|
# Preserve both persistent and runtime-only third-party peers. Detect conflicts before writing.
|
|
combined = {p["PublicKey"]: p for p in runtime}
|
|
combined.update({p["PublicKey"]: p for p in stored})
|
|
merged = merge_peers(list(combined.values()), desired)
|
|
runtime_merged = merge_peers(runtime, desired)
|
|
# Do not reset a third-party peer's runtime settings to an older persistent file.
|
|
# Preserve roaming endpoints of managed peers unless the persisted endpoint was edited.
|
|
stored_by_key = {p["PublicKey"]: p for p in stored}
|
|
runtime_by_key = {p["PublicKey"]: p for p in runtime}
|
|
desired_by_key = {p["PublicKey"]: p for p in desired}
|
|
for peer in runtime_merged:
|
|
public = peer["PublicKey"]
|
|
if public in desired_by_key and stored_by_key.get(public, {}).get("Endpoint") == desired_by_key[public].get("Endpoint"):
|
|
if runtime_by_key.get(public, {}).get("Endpoint"):
|
|
peer["Endpoint"] = runtime_by_key[public]["Endpoint"]
|
|
def matches(actual, persistent=False):
|
|
by_key = {p["PublicKey"]: p for p in actual}
|
|
return all(p["PublicKey"] in by_key and all(by_key[p["PublicKey"]].get(k) == v for k, v in p.items() if persistent or k != "Endpoint") for p in desired)
|
|
matched = matches(stored, persistent=True) and matches(runtime)
|
|
if apply:
|
|
rollback = run("wg", "showconf", IFACE)
|
|
try:
|
|
atomic_write(path, render(interface, merged))
|
|
# Keep the running interface header verbatim (including a possible fwmark).
|
|
header = rollback.split("[Peer]", 1)[0].rstrip()
|
|
peer_config = render({}, runtime_merged, native=True).removeprefix("[Interface]\n")
|
|
sync(header + "\n" + peer_config)
|
|
status, after = live()
|
|
if not matches(after):
|
|
raise AgentError("Применение пиров не подтверждено")
|
|
except Exception:
|
|
failures = []
|
|
for action in (lambda: atomic_write(path, previous), lambda: sync(rollback)):
|
|
try:
|
|
action()
|
|
except Exception:
|
|
failures.append(True)
|
|
raise AgentError("Применение не удалось; " + ("восстановление требует ручной проверки" if failures else "предыдущая конфигурация восстановлена")) from None
|
|
matched = True
|
|
return {**status, "configMatches": matched}
|
|
|
|
|
|
class Handler(http.server.BaseHTTPRequestHandler):
|
|
def log_message(self, *_):
|
|
pass
|
|
|
|
def handle_request(self):
|
|
try:
|
|
with LOCK:
|
|
if self.command == "GET" and self.path == "/status":
|
|
result, _ = live()
|
|
result["configMatches"] = None
|
|
elif self.command == "POST" and self.path in ("/check", "/apply"):
|
|
size = int(self.headers.get("Content-Length", "0"))
|
|
if not 0 < size <= MAX_BODY or self.headers.get("Transfer-Encoding"):
|
|
raise AgentError("Некорректный размер запроса")
|
|
raw = json.loads(self.rfile.read(size))
|
|
result = check(raw, apply=self.path == "/apply")
|
|
else:
|
|
self.respond(404, {"error": "Операция не поддерживается"})
|
|
return
|
|
self.respond(200, result)
|
|
except AgentError as error:
|
|
self.respond(409, {"error": str(error)})
|
|
except Exception:
|
|
self.respond(503, {"error": "Агент NetFlow не готов; проверьте конфигурацию и интерфейс хоста"})
|
|
|
|
def setup(self):
|
|
super().setup()
|
|
self.connection.settimeout(15)
|
|
|
|
def respond(self, code, body):
|
|
data = json.dumps(body, ensure_ascii=False).encode()
|
|
self.send_response(code)
|
|
self.send_header("Content-Type", "application/json; charset=utf-8")
|
|
self.send_header("Content-Length", str(len(data)))
|
|
self.end_headers()
|
|
self.wfile.write(data)
|
|
|
|
do_GET = handle_request
|
|
do_POST = handle_request
|
|
|
|
|
|
class Server(socketserver.ThreadingMixIn, getattr(socketserver, "UnixStreamServer", socketserver.TCPServer)):
|
|
daemon_threads = True
|
|
|
|
|
|
class UnixConnection(http.client.HTTPConnection):
|
|
def connect(self):
|
|
self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
|
|
self.sock.settimeout(3)
|
|
self.sock.connect(SOCKET)
|
|
|
|
|
|
def main():
|
|
if not hasattr(socketserver, "UnixStreamServer"):
|
|
raise SystemExit("NetFlow agent requires Linux Unix sockets")
|
|
if "--healthcheck" in sys.argv:
|
|
client = UnixConnection("localhost")
|
|
client.request("GET", "/status")
|
|
response = client.getresponse()
|
|
sys.exit(0 if response.status == 200 and json.loads(response.read()).get("interfaceUp") else 1)
|
|
os.umask(0o077)
|
|
Path(SOCKET).parent.mkdir(parents=True, exist_ok=True)
|
|
if os.path.lexists(SOCKET):
|
|
Path(SOCKET).unlink()
|
|
with Server(SOCKET, Handler) as server:
|
|
os.chmod(SOCKET, 0o660)
|
|
print("NetFlow agent: Unix API ready", flush=True)
|
|
server.serve_forever()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|