server-up/server-up/core/networks.py

"""Macvlan/ipvlan-netwerken — stacks een eigen IP-adres in het LAN geven.

In plaats van poorten publiceren op de host (`8222:80`) krijgt een container een
eigen adres op het fysieke netwerk. Voordelen: geen poortconflicten meer, apps
draaien op hun eigen standaardpoort, en je kunt per app firewallen op IP.

Belangrijke beperking van macvlan: de host kan zijn eigen macvlan-containers
niet bereiken zonder een extra shim-interface. Zie docs/netwerken.md.
"""
from __future__ import annotations

import ipaddress
import json
import re
import socket
import struct
import time

from core import docker

_NAME_RE = re.compile(r"^[a-zA-Z0-9][a-zA-Z0-9_.-]{0,62}$")
DRIVERS = ("macvlan", "ipvlan")


class NetworkError(Exception):
    pass


# ── Inspectie ────────────────────────────────────────────────────────────────

def list_docker_networks() -> list[dict]:
    """Alle macvlan/ipvlan-netwerken die de Docker-daemon kent."""
    r = docker._run([docker._bin(), "network", "ls", "--format",
                     "{{.Name}}\t{{.Driver}}\t{{.Scope}}"], timeout=15)
    if r.returncode != 0:
        return []
    out = []
    for line in r.stdout.strip().splitlines():
        p = line.split("\t")
        if len(p) >= 2 and p[1] in DRIVERS:
            out.append({"name": p[0], "driver": p[1],
                        "scope": p[2] if len(p) > 2 else ""})
    return out


def inspect(name: str) -> dict:
    r = docker._run([docker._bin(), "network", "inspect", name], timeout=15)
    if r.returncode != 0:
        return {}
    try:
        data = json.loads(r.stdout)
        return data[0] if data else {}
    except Exception:
        return {}


def exists(name: str) -> bool:
    return bool(inspect(name))


def used_ips(name: str) -> set[str]:
    """IP-adressen die binnen dit netwerk al aan een container zijn toegekend."""
    info = inspect(name)
    ips: set[str] = set()
    for c in (info.get("Containers") or {}).values():
        addr = (c.get("IPv4Address") or "").split("/")[0]
        if addr:
            ips.add(addr)
    return ips


# ── Detectie van het hostnetwerk ─────────────────────────────────────────────
#
# Server Up draait in zijn eigen netwerknamespace, dus /sys/class/net en
# /proc/net binnen deze container beschrijven de bridge-interface van de
# container — niet het LAN waar de host op zit. Om het échte netwerk te zien
# starten we kort een container mét `--network host`; die deelt de namespace van
# de host en ziet dus diens routetabel.

_DETECT_CACHE: dict = {}
_DETECT_TS: float = 0.0
_DETECT_TTL = 300


def _hex_naar_ip(h: str) -> str:
    """/proc/net/route schrijft adressen als little-endian hex."""
    return socket.inet_ntoa(struct.pack("<L", int(h, 16)))


def parse_proc_route(tekst: str) -> list[dict]:
    """Zet /proc/net/route om in [{interface, subnet, gateway}].

    Alleen interfaces met een default route zijn interessant: dat is de kant
    waar het LAN zit, en dus de interface waar een macvlan aan moet hangen.
    """
    gateways: dict[str, str] = {}
    subnetten: dict[str, list[str]] = {}
    for regel in tekst.splitlines()[1:]:
        velden = regel.split()
        if len(velden) < 8:
            continue
        iface, dest, gw, _flags, _r, _u, _m, mask = velden[:8]
        try:
            if dest == "00000000":                    # default route
                gateways[iface] = _hex_naar_ip(gw)
            else:
                net = ipaddress.ip_network(
                    f"{_hex_naar_ip(dest)}/{_hex_naar_ip(mask)}", strict=False)
                subnetten.setdefault(iface, []).append(str(net))
        except Exception:
            continue

    uit = []
    for iface, gateway in gateways.items():
        # Het subnet waar de gateway in valt, is het LAN van deze interface.
        gekozen = ""
        for kandidaat in subnetten.get(iface, []):
            try:
                if ipaddress.ip_address(gateway) in ipaddress.ip_network(kandidaat):
                    gekozen = kandidaat
                    break
            except ValueError:
                continue
        if not gekozen and subnetten.get(iface):
            gekozen = subnetten[iface][0]
        if gekozen:
            uit.append({"interface": iface, "subnet": gekozen, "gateway": gateway})
    return sorted(uit, key=lambda d: d["interface"])


def suggest_range(subnet: str, gateway: str = "") -> str:
    """Stel een blok adressen voor aan het eind van het subnet.

    Het einde van een subnet is meestal vrij: DHCP-servers delen doorgaans vanaf
    onderen uit, en de gateway zit vrijwel altijd op .1. Zestien adressen is
    genoeg voor een flinke stapel stacks en klein genoeg om in je router als
    uitzondering vrij te houden.
    """
    try:
        net = ipaddress.ip_network(subnet, strict=False)
    except ValueError:
        return ""
    # Een /28 of kleiner valt niet zinnig op te delen: dan is het hele subnet
    # het bereik en heeft een aparte range geen toegevoegde waarde.
    if net.version != 4 or net.prefixlen >= 28:
        return ""
    blokken = list(net.subnets(new_prefix=28))
    for blok in reversed(blokken):
        if gateway:
            try:
                if ipaddress.ip_address(gateway) in blok:
                    continue          # nooit een blok met de gateway erin
            except ValueError:
                pass
        return str(blok)
    return ""


def _helper_image() -> str:
    """Image voor de hulpcontainer: dat van Server Up zelf, dus altijd aanwezig."""
    try:
        from core import selfupdate
        return selfupdate.inspect_self().get("image", "")
    except Exception:
        return ""


def detect_host_networks(force: bool = False) -> dict:
    """Detecteer de netwerken van de host.

    Retourneert {"networks": [...], "error": "..."} — bij een fout een lege
    lijst plus uitleg, zodat de interface kan terugvallen op handmatig invullen.
    """
    global _DETECT_CACHE, _DETECT_TS
    if not force and _DETECT_CACHE and (time.time() - _DETECT_TS) < _DETECT_TTL:
        return dict(_DETECT_CACHE)

    image = _helper_image()
    if not image:
        return {"networks": [], "error":
                "Kan het eigen image niet bepalen; vul de gegevens handmatig in."}

    r = docker._run([docker._bin(), "run", "--rm", "--network", "host",
                     "--entrypoint", "cat", image, "/proc/net/route"], timeout=30)
    if r.returncode != 0:
        return {"networks": [], "error":
                (r.stderr or r.stdout or "").strip()[:200] or "Detectie mislukt."}

    netwerken = parse_proc_route(r.stdout)
    for n in netwerken:
        n["suggested_range"] = suggest_range(n["subnet"], n["gateway"])
        n["driver"] = "macvlan"
        n["name"] = "lan" if len(netwerken) == 1 else f"lan-{n['interface']}"
    resultaat = {"networks": netwerken, "error": ""}

    _DETECT_CACHE, _DETECT_TS = dict(resultaat), time.time()
    return resultaat


def host_interfaces() -> list[str]:
    """Interfacenamen van de host, voor de keuzelijst bij `parent`."""
    return [n["interface"] for n in detect_host_networks().get("networks", [])]


# ── IP-toewijzing ────────────────────────────────────────────────────────────

def next_free_ip(net: dict, extra_used: set[str] | None = None) -> str:
    """Eerste vrije adres binnen ip_range (of subnet), gateway uitgezonderd.

    Zelfde patroon als docker.next_free_port(): een voorstel, geen reservering.
    """
    subnet = net.get("ip_range") or net.get("subnet") or ""
    if not subnet:
        return ""
    try:
        network = ipaddress.ip_network(subnet, strict=False)
    except ValueError:
        return ""
    bezet = set(used_ips(net.get("name", "")) if net.get("name") else set())
    bezet |= set(extra_used or set())
    gateway = (net.get("gateway") or "").strip()
    if gateway:
        bezet.add(gateway)
    for host in network.hosts():
        if str(host) not in bezet:
            return str(host)
    return ""


def ip_in_network(ip: str, net: dict) -> bool:
    subnet = net.get("subnet") or ""
    if not subnet:
        return False
    try:
        return ipaddress.ip_address(ip) in ipaddress.ip_network(subnet, strict=False)
    except ValueError:
        return False


# ── Aanmaken / verwijderen ───────────────────────────────────────────────────

def validate(net: dict) -> str:
    """Retourneer een foutmelding, of "" als de definitie klopt."""
    if not _NAME_RE.match(net.get("name") or ""):
        return "Ongeldige netwerknaam."
    if net.get("driver") not in DRIVERS:
        return f"Driver moet {' of '.join(DRIVERS)} zijn."
    if not _NAME_RE.match(net.get("parent") or ""):
        return "Geef een geldige host-interface op (bv. eth0)."
    try:
        subnet = ipaddress.ip_network(net.get("subnet") or "", strict=False)
    except ValueError:
        return "Ongeldig subnet (bv. 192.168.1.0/24)."
    gateway = (net.get("gateway") or "").strip()
    if gateway:
        try:
            if ipaddress.ip_address(gateway) not in subnet:
                return "De gateway ligt buiten het subnet."
        except ValueError:
            return "Ongeldig gateway-adres."
    rng = (net.get("ip_range") or "").strip()
    if rng:
        try:
            if not ipaddress.ip_network(rng, strict=False).subnet_of(subnet):
                return "De IP-range ligt buiten het subnet."
        except ValueError:
            return "Ongeldige IP-range (bv. 192.168.1.240/28)."
    return ""


def review(net: dict) -> dict:
    """Beoordeel een netwerkdefinitie en leg in gewone taal uit wat er mis is.

    Gescheiden van validate(): dat geeft één blokkerende fout, dit geeft alle
    opmerkingen — ook waarschuwingen die je mág negeren — plus wat het netwerk
    concreet oplevert. Bedoeld om live mee te typen in de interface.
    """
    fouten: list[str] = []
    waarschuwingen: list[str] = []
    info: dict = {}

    blokkerend = validate(net)
    if blokkerend:
        fouten.append(blokkerend)
        return {"ok": False, "errors": fouten, "warnings": waarschuwingen, "info": info}

    subnet = ipaddress.ip_network(net["subnet"], strict=False)
    gateway = (net.get("gateway") or "").strip()
    rng = (net.get("ip_range") or "").strip()

    if not gateway:
        waarschuwingen.append(
            "Zonder gateway kunnen de containers alleen het eigen subnet bereiken, "
            "geen internet. Meestal is dat het adres van je router.")

    bereik = ipaddress.ip_network(rng, strict=False) if rng else subnet
    bruikbaar = [h for h in bereik.hosts()]
    if gateway:
        try:
            gw = ipaddress.ip_address(gateway)
            # Alleen relevant bij een expliciete range: dat de gateway ergens in
            # het subnet zit is normaal. Docker houdt het gateway-adres apart,
            # maar een range eromheen kiezen is duidelijker en scheelt verwarring.
            if rng and gw in bereik:
                waarschuwingen.append(
                    f"De gateway ({gateway}) valt binnen je IP-range. Docker houdt "
                    "dat adres vrij, maar het is overzichtelijker om een range te "
                    "kiezen die de gateway niet bevat.")
            bruikbaar = [h for h in bruikbaar if h != gw]
        except ValueError:
            pass

    if not rng:
        waarschuwingen.append(
            "Zonder IP-range mag Docker het hele subnet gebruiken, inclusief "
            "adressen die je DHCP-server aan telefoons en laptops uitdeelt. Dat "
            "geeft vroeg of laat twee apparaten op hetzelfde adres.")

    if len(bruikbaar) == 0:
        fouten.append("Er blijft geen enkel bruikbaar adres over in dit bereik.")
    elif len(bruikbaar) < 4:
        waarschuwingen.append(
            f"Er passen maar {len(bruikbaar)} adressen in dit bereik; je kunt dus "
            f"{len(bruikbaar)} stacks een eigen IP geven.")

    if bruikbaar:
        info = {"first": str(bruikbaar[0]), "last": str(bruikbaar[-1]),
                "count": len(bruikbaar), "subnet": str(subnet),
                "range": str(bereik)}

    return {"ok": not fouten, "errors": fouten,
            "warnings": waarschuwingen, "info": info}


def create(net: dict) -> tuple[bool, str]:
    """Maak het netwerk aan bij de Docker-daemon (idempotent)."""
    probleem = validate(net)
    if probleem:
        return False, probleem
    if exists(net["name"]):
        return True, "Netwerk bestaat al."
    cmd = [docker._bin(), "network", "create",
           "-d", net["driver"],
           "--subnet", net["subnet"],
           "-o", f"parent={net['parent']}"]
    if net.get("gateway"):
        cmd += ["--gateway", net["gateway"]]
    if net.get("ip_range"):
        cmd += ["--ip-range", net["ip_range"]]
    cmd.append(net["name"])
    r = docker._run(cmd, timeout=30)
    if r.returncode != 0:
        return False, (r.stderr or r.stdout).strip()[:400]
    return True, "Netwerk aangemaakt."


def remove(name: str) -> tuple[bool, str]:
    if not _NAME_RE.match(name or ""):
        return False, "Ongeldige netwerknaam."
    r = docker._run([docker._bin(), "network", "rm", name], timeout=30)
    if r.returncode != 0:
        return False, (r.stderr or r.stdout).strip()[:400]
    return True, "Netwerk verwijderd."