vacuum-wall/webui/api/dhcp.py

"""DHCP/DNS (dnsmasq) management API blueprint.

Exposed at /api/dhcp/* and delegates all operations to vacuum-walld.
"""

from typing import Any

from flask import Blueprint

from daemon.client import (  # noqa: F401 (resolved via module globals)
    delete,
    get,
    patch,
    post,
)
from daemon.iface import (
    DELETE_DNSMASQ_DNS_RECORD_REMOVE,
    DELETE_DNSMASQ_RANGES_REMOVE,
    DELETE_DNSMASQ_STATIC_LEASE_REMOVE,
    GET_DNSMASQ_CONFIG,
    GET_DNSMASQ_LEASES,
    GET_DNSMASQ_STATUS,
    PATCH_DNSMASQ_CONFIG,
    POST_DNSMASQ_APPLY,
    POST_DNSMASQ_CONFIG,
    POST_DNSMASQ_DNS_RECORD_ADD,
    POST_DNSMASQ_DOMAIN,
    POST_DNSMASQ_RANGES_ADD,
    POST_DNSMASQ_STATIC_LEASE_ADD,
)
from webui.api.common import NO_BODY, daemon_route, require_dict_body, void_transform

bp = Blueprint("dhcp", __name__)


# ---------------------------------------------------------------------------
# Config / status
# ---------------------------------------------------------------------------


@daemon_route(GET_DNSMASQ_CONFIG, bp)
def get_config_bp():
    """GET /api/dhcp/config — Retrieve the current dnsmasq configuration."""


@daemon_route(
    POST_DNSMASQ_CONFIG, bp, precheck=require_dict_body, transform=void_transform
)
def post_config():
    """POST /api/dhcp/config — Save a full replacement dnsmasq configuration."""


@daemon_route(
    PATCH_DNSMASQ_CONFIG, bp, precheck=require_dict_body, transform=void_transform
)
def patch_config():
    """PATCH /api/dhcp/config — Partially update the dnsmasq configuration."""


@daemon_route(POST_DNSMASQ_APPLY, bp, body=NO_BODY, transform=void_transform)
def apply_bp():
    """POST /api/dhcp/apply — Apply the current dnsmasq configuration."""


@daemon_route(GET_DNSMASQ_STATUS, bp)
def status_bp():
    """GET /api/dhcp/status — Retrieve dnsmasq service status."""


# ---------------------------------------------------------------------------
# DHCP ranges
# ---------------------------------------------------------------------------


def _add_range_body(request: Any, _va: Any) -> dict[str, Any]:
    body = request.get_json(silent=True) or {}
    iface = (body.get("interface") or "").strip() or None
    start = (body.get("start") or "").strip()
    end = (body.get("end") or "").strip()
    if not start or not end:
        raise ValueError("'start' and 'end' are required")
    return {
        "interface": iface or "",
        "start": start,
        "end": end,
        "lease_time": body.get("lease_time", "12h"),
    }


def _remove_range_body(request: Any, _va: Any) -> dict[str, Any]:
    body = request.get_json(silent=True) or {}
    iface = (body.get("interface") or "").strip() or ""
    start = (body.get("start") or "").strip()
    end = (body.get("end") or "").strip()
    if not start or not end:
        raise ValueError("'start' and 'end' are required")
    return {"interface": iface, "start": start, "end": end}


@daemon_route(
    POST_DNSMASQ_RANGES_ADD,
    bp,
    rule="/ranges",
    body=_add_range_body,
    transform=void_transform,
)
def add_range_bp():
    """POST /api/dhcp/ranges — Add a DHCP address range for an interface."""


@daemon_route(
    DELETE_DNSMASQ_RANGES_REMOVE,
    bp,
    rule="/ranges",
    body=_remove_range_body,
    transform=void_transform,
)
def remove_range_bp():
    """DELETE /api/dhcp/ranges — Remove a DHCP address range."""


@daemon_route(GET_DNSMASQ_LEASES, bp)
def leases_bp():
    """GET /api/dhcp/leases — Retrieve the current DHCP lease table."""


# ---------------------------------------------------------------------------
# Static leases
# ---------------------------------------------------------------------------


def _add_static_lease_body(request: Any, _va: Any) -> dict[str, Any]:
    body = request.get_json(silent=True) or {}
    mac = (body.get("mac") or "").strip()
    ip = (body.get("ip") or "").strip()
    if not mac or not ip:
        raise ValueError("'mac' and 'ip' are required")
    return {"mac": mac, "ip": ip, "hostname": body.get("hostname")}


def _static_lease_echo(_data: Any, _va: Any, sent: Any) -> Any:
    return {"mac": sent["mac"], "ip": sent["ip"], "hostname": sent["hostname"]}


@daemon_route(
    POST_DNSMASQ_STATIC_LEASE_ADD,
    bp,
    rule="/static-lease",
    body=_add_static_lease_body,
    transform=_static_lease_echo,
)
def add_static_lease_bp():
    """POST /api/dhcp/static-lease — Add a static DHCP lease by MAC address."""


@daemon_route(
    DELETE_DNSMASQ_STATIC_LEASE_REMOVE,
    bp,
    rule="/static-lease/<mac>",
    transform=void_transform,
)
def remove_static_lease_bp():
    """DELETE /api/dhcp/static-lease/<mac> — Remove a static DHCP lease by MAC."""


# ---------------------------------------------------------------------------
# DNS records
# ---------------------------------------------------------------------------


def _add_dns_record_body(request: Any, _va: Any) -> dict[str, Any]:
    body = request.get_json(silent=True) or {}
    name = (body.get("name") or "").strip()
    address = (body.get("address") or "").strip()
    if not name or not address:
        raise ValueError("'name' and 'address' are required")
    return {"name": name, "address": address, "hostname": body.get("hostname")}


def _dns_record_echo(_data: Any, _va: Any, sent: Any) -> Any:
    return {
        "name": sent["name"],
        "address": sent["address"],
        "hostname": sent["hostname"],
    }


@daemon_route(
    POST_DNSMASQ_DNS_RECORD_ADD,
    bp,
    rule="/dns-record",
    body=_add_dns_record_body,
    transform=_dns_record_echo,
)
def add_dns_record_bp():
    """POST /api/dhcp/dns-record — Add a DNS record."""


@daemon_route(
    DELETE_DNSMASQ_DNS_RECORD_REMOVE,
    bp,
    rule="/dns-record/<name>",
    transform=void_transform,
)
def remove_dns_record_bp():
    """DELETE /api/dhcp/dns-record/<name> — Remove a DNS record by name."""


# ---------------------------------------------------------------------------
# DNS domain
# ---------------------------------------------------------------------------


def _set_domain_body(request: Any, _va: Any) -> dict[str, Any]:
    return {"domain": (request.get_json(silent=True) or {}).get("domain")}


@daemon_route(
    POST_DNSMASQ_DOMAIN,
    bp,
    precheck=require_dict_body,
    body=_set_domain_body,
    transform=void_transform,
)
def set_domain_bp():
    """POST /api/dhcp/domain — Set or clear the DNS search domain."""