"""Kea-DHCP: Optionskatalog, Konfigurationserzeugung, Lease-Auswertung.

Backend ist ``kea-dhcp4-server`` (isc-dhcp-server ist seit 2022 abgekuendigt).

Die im Vorgaengerprojekt teuer erarbeiteten Kea-Eigenheiten sind hier von
Anfang an beruecksichtigt und im Code vermerkt:

* Der Schluessel der Lease-Datei heisst ``name``, nicht ``lease-file``.
* ``loggers`` gehoert **unter** ``Dhcp4``, nicht daneben -- sonst Syntaxfehler.
* Standardoptionen bekommen **kein** eigenes ``option-def`` (Kea kennt sie
  nativ); nur eigene Optionen werden definiert.
* Die Lease-CSV wird mit ``\\n`` geschrieben; Pythons csv-Default ``\\r\\n``
  laesst Kea die Datei mit "invalid column" ablehnen.
* Das Kea-Protokoll darf nicht unter dem Protokollverzeichnis der Anwendung
  liegen: das AppArmor-Profil von Kea erlaubt ausschliesslich den fest
  einprogrammierten Pfad ``/var/log/kea/kea-dhcp4.log``.
"""

from __future__ import annotations

import ipaddress
import json
import logging
import re
import sqlite3
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Any, Iterable, Sequence

from tesm_core.db import Database
from tesm_core.sysops import SysOps

logger = logging.getLogger(__name__)

SERVICE_UNIT = "kea-dhcp4-server"
PACKAGE_NAME = "kea-dhcp4-server"
LEASE_FILE = "/var/lib/kea/kea-leases4.csv"
LOG_PATH = "/var/log/kea/kea-dhcp4.log"
DEFAULT_CONFIG_PATH = "/etc/kea/kea-dhcp4.conf"

#: ``(code, name, type, description)`` -- kuratierte Standardoptionen.
STANDARD_OPTIONS: tuple[tuple[int, str, str, str], ...] = (
    (4, "time-servers", "ipv4-address", "Zeitserver (RFC 868)"),
    (7, "log-servers", "ipv4-address", "Log-Server (MIT-LCS UDP)"),
    (8, "cookie-servers", "ipv4-address", "Cookie-Server"),
    (9, "lpr-servers", "ipv4-address", "LPR-Druckserver"),
    (10, "impress-servers", "ipv4-address", "Impress-Server"),
    (11, "resource-location-servers", "ipv4-address", "Resource Location Server"),
    (13, "boot-size", "uint16", "Boot-Image-Groesse (Bloecke)"),
    (14, "merit-dump", "string", "Crash-Dump-Datei"),
    (16, "swap-server", "ipv4-address", "Swap-Server"),
    (17, "root-path", "string", "Root-Pfad (NFS)"),
    (18, "extensions-path", "string", "Extensions-Pfad (BOOTP)"),
    (19, "ip-forwarding", "boolean", "IP-Forwarding aktivieren"),
    (20, "non-local-source-routing", "boolean", "Non-Local Source Routing"),
    (21, "policy-filter", "string", "Policy Filter (Ziel,Maske,...)"),
    (22, "max-dgram-reassembly", "uint16", "Maximale Datagramm-Reassemblierungsgroesse"),
    (23, "default-ip-ttl", "uint8", "Standard-TTL fuer IP"),
    (24, "path-mtu-aging-timeout", "uint32", "Path-MTU-Aging-Timeout (Sek.)"),
    (25, "path-mtu-plateau-table", "uint16", "Path-MTU-Plateau-Tabelle"),
    (26, "interface-mtu", "uint16", "MTU des Interfaces"),
    (27, "all-subnets-local", "boolean", "Alle Subnetze lokal erreichbar"),
    (28, "broadcast-address", "ipv4-address", "Broadcast-Adresse"),
    (29, "perform-mask-discovery", "boolean", "Subnetzmasken-Discovery"),
    (30, "mask-supplier", "boolean", "Subnetzmasken-Anbieter"),
    (31, "router-discovery", "boolean", "Router Discovery (RFC 1256)"),
    (32, "router-solicitation-address", "ipv4-address", "Router-Solicitation-Adresse"),
    (33, "static-routes", "string", "Statische Routen (Ziel,Gateway,...)"),
    (34, "trailer-encapsulation", "boolean", "Trailer Encapsulation"),
    (35, "arp-cache-timeout", "uint32", "ARP-Cache-Timeout (Sek.)"),
    (36, "ieee802-3-encapsulation", "boolean", "Ethernet-Encapsulation (IEEE 802.3)"),
    (37, "default-tcp-ttl", "uint8", "Standard-TTL fuer TCP"),
    (38, "tcp-keepalive-interval", "uint32", "TCP-Keepalive-Intervall (Sek.)"),
    (40, "nis-domain", "string", "NIS-Domain"),
    (41, "nis-servers", "ipv4-address", "NIS-Server"),
    (42, "ntp-servers", "ipv4-address", "NTP-Server"),
    (44, "netbios-name-servers", "ipv4-address", "WINS/NBNS-Server"),
    (45, "netbios-dd-server", "ipv4-address", "NetBIOS Datagram Distribution Server"),
    (46, "netbios-node-type", "uint8", "WINS/NBT-Knotentyp"),
    (47, "netbios-scope", "string", "NetBIOS-Scope-ID"),
    (48, "font-servers", "ipv4-address", "X-Window-Font-Server"),
    (49, "x-display-manager", "ipv4-address", "X-Window Display Manager"),
    (60, "vendor-class-identifier", "string", "Vendor Class Identifier"),
    (64, "nisplus-domain-name", "string", "NIS+-Domain"),
    (65, "nisplus-servers", "ipv4-address", "NIS+-Server"),
    (66, "tftp-server-name", "string", "Boot-Server (TFTP-Hostname)"),
    (67, "boot-file-name", "string", "Bootdatei-Name"),
    (68, "mobile-ip-home-agent", "ipv4-address", "Mobile-IP-Home-Agent"),
    (69, "smtp-server", "ipv4-address", "SMTP-Server"),
    (70, "pop-server", "ipv4-address", "POP3-Server"),
    (71, "nntp-server", "ipv4-address", "NNTP-Server"),
    (72, "www-server", "ipv4-address", "WWW-Server"),
    (73, "finger-server", "ipv4-address", "Finger-Server"),
    (74, "irc-server", "ipv4-address", "IRC-Server"),
    (75, "streettalk-server", "ipv4-address", "StreetTalk-Server"),
    (76, "streettalk-directory-assistance-server", "ipv4-address", "StreetTalk Directory Assistance Server"),
    (118, "subnet-selection", "ipv4-address", "Subnet Selection (RFC 3011)"),
    (119, "domain-search", "string", "DNS-Suchliste (Domain Search)"),
)

OPTION_TYPES = (
    "ipv4-address",
    "string",
    "boolean",
    "uint8",
    "uint16",
    "uint32",
    "int8",
    "int16",
    "int32",
)

_HOSTNAME_CLEAN = re.compile(r"[^a-z0-9-]+")


# --------------------------------------------------------------------------- #
# Netzwerkerkennung
# --------------------------------------------------------------------------- #


@dataclass(frozen=True, slots=True)
class InterfaceNetwork:
    interface: str
    cidr: str

    @property
    def network(self) -> ipaddress.IPv4Network:
        return ipaddress.ip_network(self.cidr, strict=False)  # type: ignore[return-value]

    @property
    def address(self) -> str:
        return self.cidr.split("/")[0]


def detect_networks(sysops: SysOps) -> list[InterfaceNetwork]:
    """Liest alle IPv4-Adressen inkl. Alias-Adressen aus ``ip -4 addr show``.

    Subnetz und Maske werden nie von Hand eingegeben -- sie kommen immer vom
    laufenden System. Das war schon im Vorgaenger die richtige Entscheidung.
    """
    result = sysops.read_command(["ip", "-4", "-o", "addr", "show", "scope", "global"])
    if not result.ok:
        return []
    networks: list[InterfaceNetwork] = []
    for line in result.stdout.splitlines():
        parts = line.split()
        if len(parts) < 4 or parts[2] != "inet":
            continue
        networks.append(InterfaceNetwork(interface=parts[1], cidr=parts[3]))
    return networks


def default_gateway(sysops: SysOps) -> str:
    result = sysops.read_command(["ip", "route", "show", "default"])
    if not result.ok:
        return ""
    for line in result.stdout.splitlines():
        parts = line.split()
        if "via" in parts:
            return parts[parts.index("via") + 1]
    return ""



def _now() -> datetime:
    return datetime.now(timezone.utc)

def matching_network(
    networks: Iterable[InterfaceNetwork], interface: str, range_start: str, range_end: str
) -> InterfaceNetwork | None:
    """Findet das Netz, in das ein Adressbereich tatsaechlich passt."""
    try:
        start = ipaddress.ip_address(range_start)
        end = ipaddress.ip_address(range_end)
    except ValueError:
        return None
    if int(end) < int(start):
        return None
    for candidate in networks:
        if candidate.interface != interface:
            continue
        network = candidate.network
        if start in network and end in network:
            return candidate
    return None


def validate_subnet(
    networks: Iterable[InterfaceNetwork], interface: str, range_start: str, range_end: str
) -> str:
    """Rueckgabe: leerer String bei Erfolg, sonst die Fehlermeldung."""
    if not interface:
        return "Es ist kein Interface ausgewaehlt."
    try:
        start = ipaddress.ip_address(range_start)
        end = ipaddress.ip_address(range_end)
    except ValueError:
        return "Start- und Endadresse muessen gueltige IPv4-Adressen sein."
    if int(end) < int(start):
        return "Die Endadresse liegt vor der Startadresse."
    if matching_network(networks, interface, range_start, range_end) is None:
        return (
            f"Der Bereich {range_start}-{range_end} passt in kein Netz von {interface}. "
            "Subnetz und Maske werden vom laufenden System gelesen."
        )
    return ""


# --------------------------------------------------------------------------- #
# Reservierungen
# --------------------------------------------------------------------------- #


def normalize_mac(value: str) -> str:
    """Vereinheitlicht MAC-Adressen auf ``aa:bb:cc:dd:ee:ff``."""
    cleaned = re.sub(r"[^0-9a-fA-F]", "", value or "")
    if len(cleaned) != 12:
        return ""
    return ":".join(cleaned[index : index + 2] for index in range(0, 12, 2)).lower()


def hostname_from_name(name: str, taken: set[str]) -> str:
    """Leitet einen DNS-taugliche Hostnamen ab und loest Kollisionen mit -2/-3."""
    base = _HOSTNAME_CLEAN.sub("-", (name or "").lower()).strip("-") or "client"
    base = base[:60]
    candidate = base
    suffix = 2
    while candidate in taken:
        candidate = f"{base}-{suffix}"
        suffix += 1
    taken.add(candidate)
    return candidate


def configured_networks(
    conn: sqlite3.Connection, networks: Iterable[InterfaceNetwork]
) -> list[ipaddress.IPv4Network | ipaddress.IPv6Network]:
    """Die Netze, fuer die tatsaechlich ein Subnetz eingerichtet ist.

    Nur diese landen in der Konfiguration. Eine Reservierung ausserhalb wird von
    Kea nie ausgeliefert -- sie in der Oberflaeche trotzdem zu zeigen, hat
    Reservierungen versprochen, die es nicht gab.
    """
    treffer = []
    for subnet in Database.all(
        conn, "SELECT * FROM dhcp_subnets WHERE enabled=1 ORDER BY sort_order, id"
    ):
        match = matching_network(
            networks,
            str(subnet["interface"]),
            str(subnet["range_start"]),
            str(subnet["range_end"]),
        )
        if match is not None:
            treffer.append(match.network)
    return treffer


def in_configured_network(
    ip: str, networks: Iterable[ipaddress.IPv4Network | ipaddress.IPv6Network]
) -> bool:
    try:
        adresse = ipaddress.ip_address(ip)
    except ValueError:
        return False
    return any(adresse in netz for netz in networks)


def collect_reservations(
    conn: sqlite3.Connection,
    *,
    only_networks: Iterable[ipaddress.IPv4Network | ipaddress.IPv6Network] | None = None,
    skipped: list[dict[str, str]] | None = None,
) -> list[dict[str, Any]]:
    """Automatische Reservierungen aus dem Geraetebestand plus manuelle Eintraege.

    Bei einer MAC-Kollision gewinnt die automatische Reservierung -- ein
    ueberwachtes Geraet ist die verlaesslichere Quelle.

    ``only_networks`` grenzt auf Adressen in eingerichteten Bereichen ein. Ohne
    diese Eingrenzung entstanden Eintraege fuer Adressen, fuer die es kein
    Subnetz gibt: in der Liste sichtbar, in der Kea-Konfiguration nie
    vorhanden. Was dabei wegfaellt, wird in ``skipped`` gesammelt, damit die
    Oberflaeche es benennen kann statt es zu verschweigen.
    """
    grenzen = list(only_networks) if only_networks is not None else None
    taken: set[str] = set()
    reservations: list[dict[str, Any]] = []
    seen_macs: set[str] = set()

    for row in Database.all(
        conn,
        "SELECT id, mac, ip, name FROM devices "
        "WHERE deleted_at IS NULL AND is_active=1 AND ip <> '' ORDER BY name COLLATE NOCASE",
    ):
        mac = normalize_mac(str(row["mac"]))
        if not mac or mac in seen_macs:
            continue
        try:
            ipaddress.ip_address(str(row["ip"]))
        except ValueError:
            continue
        if grenzen is not None and not in_configured_network(str(row["ip"]), grenzen):
            if skipped is not None:
                skipped.append({"name": str(row["name"]), "ip": str(row["ip"]), "source": "auto"})
            continue
        seen_macs.add(mac)
        reservations.append(
            {
                "device_id": int(row["id"]),
                "mac": mac,
                "ip": str(row["ip"]),
                "hostname": hostname_from_name(str(row["name"]), taken),
                "source": "auto",
                "name": str(row["name"]),
            }
        )

    for row in Database.all(
        conn, "SELECT id, mac, ip, hostname, comment FROM dhcp_reservations ORDER BY ip"
    ):
        mac = normalize_mac(str(row["mac"]))
        if not mac or mac in seen_macs:
            continue
        if grenzen is not None and not in_configured_network(str(row["ip"]), grenzen):
            if skipped is not None:
                skipped.append(
                    {
                        "name": str(row["hostname"] or row["comment"] or mac),
                        "ip": str(row["ip"]),
                        "source": "manuell",
                    }
                )
            continue
        seen_macs.add(mac)
        reservations.append(
            {
                "device_id": None,
                "reservation_id": int(row["id"]),
                "mac": mac,
                "ip": str(row["ip"]),
                "hostname": hostname_from_name(
                    str(row["hostname"] or row["comment"] or mac.replace(":", "")), taken
                ),
                "source": "manuell",
                "name": str(row["comment"] or ""),
            }
        )
    return reservations


# --------------------------------------------------------------------------- #
# Konfiguration
# --------------------------------------------------------------------------- #


def render_config(
    conn: sqlite3.Connection,
    *,
    networks: list[InterfaceNetwork],
    settings: dict[str, Any],
) -> tuple[dict[str, Any], list[str]]:
    """Erzeugt die Kea-Konfiguration. Rueckgabe: ``(config, warnungen)``."""
    warnings: list[str] = []

    subnets = Database.all(
        conn,
        "SELECT * FROM dhcp_subnets WHERE enabled=1 ORDER BY sort_order, id",
    )
    reservations = collect_reservations(conn)
    option_rows = Database.all(conn, "SELECT * FROM dhcp_options ORDER BY code")
    options_by_id = {int(row["id"]): row for row in option_rows}

    global_values: dict[int, str] = {}
    device_values: dict[int, dict[int, str]] = {}
    for row in Database.all(conn, "SELECT * FROM dhcp_option_values"):
        option_id = int(row["option_id"])
        value = str(row["value"] or "").strip()
        if not value:
            continue
        if row["device_id"] is None:
            global_values[option_id] = value
        else:
            device_values.setdefault(int(row["device_id"]), {})[option_id] = value

    def option_data(values: dict[int, str]) -> list[dict[str, Any]]:
        entries: list[dict[str, Any]] = []
        for option_id, value in sorted(values.items()):
            option = options_by_id.get(option_id)
            if option is None:
                continue
            entries.append({"name": str(option["name"]), "data": value})
        return entries

    subnet_entries: list[dict[str, Any]] = []
    interfaces: set[str] = set()
    for index, subnet in enumerate(subnets, start=1):
        interface = str(subnet["interface"])
        match = matching_network(
            networks, interface, str(subnet["range_start"]), str(subnet["range_end"])
        )
        if match is None:
            # Nicht passende Subnetze werden uebersprungen, nicht die ganze
            # Aktion abgebrochen -- eine geaenderte Host-IP soll nicht dazu
            # fuehren, dass gar keine Konfiguration mehr geschrieben wird.
            warnings.append(
                f"Subnetz {subnet['range_start']}-{subnet['range_end']} auf {interface} "
                "passt in kein aktuelles Netz und wurde uebersprungen."
            )
            continue
        interfaces.add(interface)
        network = match.network

        subnet_options: list[dict[str, Any]] = []
        if str(subnet["gateway"] or "").strip():
            subnet_options.append({"name": "routers", "data": str(subnet["gateway"]).strip()})
        if str(subnet["dns"] or "").strip():
            subnet_options.append(
                {"name": "domain-name-servers", "data": str(subnet["dns"]).strip()}
            )

        subnet_reservations = []
        for reservation in reservations:
            if ipaddress.ip_address(reservation["ip"]) not in network:
                continue
            entry: dict[str, Any] = {
                "hw-address": reservation["mac"],
                "ip-address": reservation["ip"],
                "hostname": reservation["hostname"],
            }
            overrides = device_values.get(reservation.get("device_id") or -1, {})
            if overrides:
                entry["option-data"] = option_data(overrides)
            subnet_reservations.append(entry)

        subnet_entries.append(
            {
                "id": index,
                "subnet": str(network),
                "pools": [{"pool": f"{subnet['range_start']} - {subnet['range_end']}"}],
                **({"option-data": subnet_options} if subnet_options else {}),
                "reservations": subnet_reservations,
            }
        )

    if not subnet_entries:
        warnings.append("Es ist kein gueltiges Subnetz konfiguriert -- Kea wuerde nicht starten.")

    global_option_data = option_data(global_values)
    domain = str(settings.get("dhcp_domain") or "").strip()
    if domain:
        global_option_data.insert(0, {"name": "domain-name", "data": domain})

    # Eigene (nicht-Standard-) Optionen brauchen eine Definition, sonst
    # verweigert Kea den Start: "definition for the option 'dhcp4.<name>' does
    # not exist". Deklariert werden **alle** eigenen Optionen, nicht nur die
    # mit globalem Wert -- ein Wert kann auch nur in einer einzelnen
    # Reservierung stehen, und genau das hat den Dienst umgebracht: die
    # Definition fehlte, weil hier nur die globalen Werte geprueft wurden.
    # Eine ungenutzte Definition ist fuer Kea harmlos; eine fehlende nicht.
    custom_defs = [
        {
            "name": str(row["name"]),
            "code": int(row["code"]),
            "type": str(row["type"]),
            "space": "dhcp4",
        }
        for row in option_rows
        if not int(row["is_standard"] or 0)
    ]

    config = {
        "Dhcp4": {
            "interfaces-config": {"interfaces": sorted(interfaces)},
            "lease-database": {
                "type": "memfile",
                "lfc-interval": 3600,
                # Schluessel heisst "name" -- nicht "lease-file".
                "name": LEASE_FILE,
            },
            "valid-lifetime": int(settings.get("dhcp_lease_default") or 600),
            "max-valid-lifetime": int(settings.get("dhcp_lease_max") or 7200),
            **({"option-def": custom_defs} if custom_defs else {}),
            **({"option-data": global_option_data} if global_option_data else {}),
            "subnet4": subnet_entries,
            # loggers gehoert UNTER Dhcp4 -- daneben ist es ein Syntaxfehler.
            "loggers": [
                {
                    "name": "kea-dhcp4",
                    "output_options": [{"output": LOG_PATH, "pattern": "%D{%Y-%m-%d %H:%M:%S} %-5p %m\n"}],
                    "severity": "INFO",
                    "debuglevel": 0,
                }
            ],
        }
    }
    return config, warnings


def config_to_text(config: dict[str, Any]) -> str:
    return json.dumps(config, indent=2, ensure_ascii=False) + "\n"


# --------------------------------------------------------------------------- #
# Leases
# --------------------------------------------------------------------------- #


@dataclass(frozen=True, slots=True)
class Lease:
    address: str
    mac: str
    hostname: str
    expires_at: int
    state: str

    @property
    def is_active(self) -> bool:
        return self.state == "0"

    @property
    def is_expired(self) -> bool:
        return self.expires_at > 0 and self.expires_at <= int(_now().timestamp())

    @property
    def expires_text(self) -> str:
        """Ablauf als Datum und Uhrzeit in der Zeitzone des Servers.

        Kea schreibt einen Unix-Zeitstempel in die Lease-Datei. Der stand hier
        vorher unveraendert in der Tabelle -- eine zehnstellige Zahl, aus der
        niemand ablesen kann, ob eine Lease in zehn Minuten oder erst morgen
        ablaeuft.
        """
        if self.expires_at <= 0:
            return "unbegrenzt"
        return datetime.fromtimestamp(self.expires_at).strftime("%d.%m.%Y %H:%M")

    @property
    def remaining_text(self) -> str:
        """Restlaufzeit in Worten -- die eigentlich interessierende Angabe."""
        if self.expires_at <= 0:
            return ""
        sekunden = self.expires_at - int(_now().timestamp())
        if sekunden <= 0:
            return "abgelaufen"
        if sekunden < 60:
            return "unter einer Minute"
        if sekunden < 3600:
            return f"in {sekunden // 60} min"
        if sekunden < 86400:
            stunden, rest = divmod(sekunden, 3600)
            return f"in {stunden} h {rest // 60} min" if rest >= 60 else f"in {stunden} h"
        tage, rest = divmod(sekunden, 86400)
        return f"in {tage} Tg {rest // 3600} h" if rest >= 3600 else f"in {tage} Tg"


#: Zustand einer Reservierung -- wie im Windows-DHCP, wo eine Reservierung
#: "aktiv" oder "inaktiv" ist, je nachdem ob der Client sie gerade benutzt.
#: Ohne diese Angabe sieht eine nie abgeholte Reservierung genauso aus wie eine
#: benutzte, und ein Geraet, das noch auf einer alten Adresse haengt, faellt
#: ueberhaupt nicht auf.
RESERVATION_STATES: dict[str, tuple[str, str]] = {
    "aktiv": ("online", "Das Geraet benutzt diese Adresse gerade."),
    "abgelaufen": ("", "Lease auf dieser Adresse ist abgelaufen -- Geraet derzeit nicht da."),
    "andere_adresse": (
        "warning",
        "Das Geraet haelt noch eine Lease auf einer anderen Adresse. Es uebernimmt die "
        "Reservierung erst nach Ablauf oder Neuanforderung.",
    ),
    "unbenutzt": ("", "Fuer dieses Geraet liegt keine Lease vor."),
}


def annotate_reservations(
    reservations: Sequence[dict[str, Any]], leases: Sequence[Lease]
) -> list[dict[str, Any]]:
    """Ergaenzt jede Reservierung um ihren Zustand aus den laufenden Leases.

    Reine Funktion ueber zwei Listen -- damit pruefbar, ohne Kea zu starten.
    """
    nach_mac: dict[str, Lease] = {}
    for lease in leases:
        mac = normalize_mac(lease.mac)
        if not mac:
            continue
        vorhanden = nach_mac.get(mac)
        # Bei mehreren Leases je MAC gewinnt die, die spaeter ablaeuft.
        if vorhanden is None or lease.expires_at > vorhanden.expires_at:
            nach_mac[mac] = lease

    ergebnis: list[dict[str, Any]] = []
    for reservation in reservations:
        eintrag = dict(reservation)
        lease = nach_mac.get(normalize_mac(str(reservation["mac"])))
        if lease is None:
            zustand = "unbenutzt"
        elif lease.address != str(reservation["ip"]):
            zustand = "andere_adresse"
        elif lease.is_expired or not lease.is_active:
            zustand = "abgelaufen"
        else:
            zustand = "aktiv"
        eintrag["state"] = zustand
        eintrag["state_hint"] = RESERVATION_STATES[zustand][1]
        eintrag["state_style"] = RESERVATION_STATES[zustand][0]
        eintrag["lease"] = lease
        ergebnis.append(eintrag)
    return ergebnis


def read_leases(sysops: SysOps, *, path: str = LEASE_FILE) -> list[Lease]:
    """Liest die memfile-Lease-Datei.

    Die Datei haengt bei jeder Verlaengerung neue Zeilen an; es gewinnt die
    letzte Zeile je Adresse. Gelesen wird ueber den privilegierten Helfer, weil
    ``/var/lib/kea`` dem Kea-Dienstbenutzer gehoert.
    """
    result = sysops.read_privileged_file(path)
    if not result.ok:
        return []
    lines = result.stdout.splitlines()
    if not lines:
        return []
    header = [name.strip().strip('"') for name in lines[0].split(",")]
    try:
        index = {
            "address": header.index("address"),
            "hwaddr": header.index("hwaddr"),
            "expire": header.index("expire"),
            "state": header.index("state"),
            "hostname": header.index("hostname"),
        }
    except ValueError:
        logger.warning("Kea-Lease-Datei hat unerwartete Spalten: %s", header)
        return []

    latest: dict[str, Lease] = {}
    for line in lines[1:]:
        fields = [field.strip().strip('"') for field in line.split(",")]
        if len(fields) < len(header):
            continue
        try:
            lease = Lease(
                address=fields[index["address"]],
                mac=normalize_mac(fields[index["hwaddr"]]),
                hostname=fields[index["hostname"]],
                expires_at=int(fields[index["expire"]] or 0),
                state=fields[index["state"]] or "0",
            )
        except (ValueError, IndexError):
            continue
        latest[lease.address] = lease
    return sorted(latest.values(), key=lambda entry: tuple(int(p) for p in entry.address.split(".") if p.isdigit()))


def service_state(sysops: SysOps) -> dict[str, str]:
    state = sysops.service_state(SERVICE_UNIT)
    import shutil

    state["binary"] = "vorhanden" if shutil.which("kea-dhcp4") else "fehlt"
    return state
