vacuum-wall/webui/api/wireguard.py

"""
webui/api/wireguard.py - WireGuard tunnel management API blueprint.

Exposed at /api/wireguard/* and delegates to lib.wireguard.
"""

from flask import Blueprint, jsonify, request

from lib.wireguard import (
    add_peer,
    apply,
    down,
    generate_client_conf,
    get_config,
    get_peer_status,
    get_peers,
    initialize,
    remove_peer,
    save_config,
    status,
)

bp = Blueprint("wireguard", __name__)


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------


def _error(msg, code=400):
    return jsonify({"ok": False, "error": msg}), code


def _ok(data=None):
    return jsonify({"ok": True, "data": data})


# ---------------------------------------------------------------------------
# Config
# ---------------------------------------------------------------------------


@bp.route("/config", methods=["GET"])
def get_config_bp():
    try:
        cfg = get_config()
        safe = dict(cfg)
        if "interface" in safe:
            safe["interface"] = dict(safe["interface"])
            safe["interface"].pop("private_key", None)
        return _ok(safe)
    except RuntimeError as exc:
        return _error(str(exc), 500)


@bp.route("/config", methods=["POST"])
def post_config():
    body = request.get_json(silent=True) or {}
    if not isinstance(body, dict):
        return _error("Request body must be a JSON object", 400)
    try:
        save_config(body)
        safe = dict(body)
        if "interface" in safe:
            safe["interface"] = dict(safe["interface"])
            safe["interface"].pop("private_key", None)
        return _ok(safe)
    except RuntimeError as exc:
        return _error(str(exc), 500)


# ---------------------------------------------------------------------------
# Apply / down
# ---------------------------------------------------------------------------


@bp.route("/apply", methods=["POST"])
def apply_bp():
    try:
        apply()
        return _ok(None)
    except RuntimeError as exc:
        return _error(str(exc), 500)


@bp.route("/down", methods=["POST"])
def down_bp():
    try:
        down()
        return _ok(None)
    except RuntimeError as exc:
        return _error(str(exc), 500)


# ---------------------------------------------------------------------------
# Status
# ---------------------------------------------------------------------------


@bp.route("/status", methods=["GET"])
def status_bp():
    try:
        return _ok(status())
    except RuntimeError as exc:
        return _error(str(exc), 500)


# ---------------------------------------------------------------------------
# Initialize (first-time setup)
# ---------------------------------------------------------------------------


@bp.route("/initialize", methods=["POST"])
def initialize_bp():
    try:
        initialize()
        return _ok(None)
    except RuntimeError as exc:
        return _error(str(exc), 500)


# ---------------------------------------------------------------------------
# Peer management
# ---------------------------------------------------------------------------


@bp.route("/add-peer", methods=["POST"])
def add_peer_bp():
    body = request.get_json(silent=True) or {}
    name = body.get("name", "").strip()
    if not name:
        return _error("'name' is required", 400)
    try:
        peer = add_peer(
            name=name,
            endpoint=body.get("endpoint"),
            allowed_ips=body.get("allowed_ips", []),
            persistent_keepalive=body.get("persistent_keepalive"),
            preshared_key=body.get("preshared_key"),
        )
        safe = dict(peer)
        safe.pop("private_key", None)
        return _ok(safe)
    except RuntimeError as exc:
        return _error(str(exc), 500)


@bp.route("/remove-peer", methods=["DELETE"])
def remove_peer_bp():
    name = request.args.get("name", "").strip()
    if not name:
        return _error("Query parameter 'name' is required", 400)
    try:
        cfg = get_config()
        if name not in cfg.get("peers", {}):
            return _error(f"Peer '{name}' not found", 404)
        remove_peer(name)
        return _ok({"name": name})
    except RuntimeError as exc:
        return _error(str(exc), 500)


@bp.route("/peers", methods=["GET"])
def peers_bp():
    try:
        return _ok(get_peers())
    except RuntimeError as exc:
        return _error(str(exc), 500)


@bp.route("/peer-status", methods=["GET"])
def peer_status_bp():
    try:
        return _ok(get_peer_status())
    except RuntimeError as exc:
        return _error(str(exc), 500)


# ---------------------------------------------------------------------------
# Client config generation
# ---------------------------------------------------------------------------


@bp.route("/generate-client", methods=["POST"])
def generate_client_bp():
    body = request.get_json(silent=True) or {}
    name = body.get("name", "").strip()
    if not name:
        return _error("Field 'name' is required", 400)
    try:
        cfg = get_config()
        if name not in cfg.get("peers", {}):
            return _error(f"Peer '{name}' not found", 404)
        server_endpoint = body.get("server_endpoint", "")
        server_pubkey = cfg["interface"].get("public_key", "")
        if not server_endpoint:
            _ = cfg["interface"].get("listen_port", 51820)
            # Can't auto-derive public IP; ask user to provide it
            return _error(
                "Field 'server_endpoint' is required (e.g., '203.0.113.1:51820')", 400
            )
        conf_text = generate_client_conf(name, server_endpoint, server_pubkey)
        return _ok({"config": conf_text})
    except (KeyError, ValueError, RuntimeError) as exc:
        code = 404 if isinstance(exc, (KeyError, ValueError)) else 500
        return _error(str(exc), code)