gatehouse-api/gatehouse_app/services/zerotier_api_service.py

"""ZeroTier API service — thin Flask adapter around the ZeroTierClient SDK.

Reads configuration from app config and translates SDK exceptions to
Secuird typed exceptions.
"""

import logging
from typing import Optional

from gatehouse_app.exceptions import ZeroTierAPIError
from gatehouse_app.utils.zerotier_client import (
    APIMode,
    ZeroTierAPIError as SDKZeroTierAPIError,
    ZeroTierAuthError,
    ZeroTierClient,
    ZeroTierNotFoundError,
)

logger = logging.getLogger(__name__)


def _get_client(app=None) -> ZeroTierClient:
    """Build a ZeroTierClient from current app config."""
    from flask import current_app

    app = app or current_app

    mode_str = app.config.get("ZEROTIER_API_MODE", "controller")
    mode = APIMode.CENTRAL if mode_str == "central" else APIMode.CONTROLLER

    return ZeroTierClient(
        api_token=app.config.get("ZEROTIER_API_TOKEN", ""),
        base_url=app.config.get("ZEROTIER_API_URL", "http://localhost:9993"),
        mode=mode,
    )


def get_status() -> dict:
    """Verify connectivity to the ZeroTier controller."""
    client = _get_client()
    try:
        return client.get_status()
    except SDKZeroTierAPIError as exc:
        raise ZeroTierAPIError(str(exc), status_code=exc.status_code) from exc


def list_networks():
    """List all networks accessible to the configured token."""
    client = _get_client()
    try:
        return client.list_networks()
    except SDKZeroTierAPIError as exc:
        raise ZeroTierAPIError(str(exc), status_code=exc.status_code) from exc


def get_network(network_id: str):
    """Fetch a single network by ID."""
    client = _get_client()
    try:
        return client.get_network(network_id)
    except SDKZeroTierAPIError as exc:
        raise ZeroTierAPIError(str(exc), status_code=exc.status_code) from exc


def list_members(network_id: str):
    """List all members on a network."""
    client = _get_client()
    try:
        return client.list_members(network_id)
    except SDKZeroTierAPIError as exc:
        raise ZeroTierAPIError(str(exc), status_code=exc.status_code) from exc


def get_member(network_id: str, node_id: str):
    """Fetch a single member on a network."""
    client = _get_client()
    try:
        return client.get_member(network_id, node_id)
    except SDKZeroTierAPIError as exc:
        raise ZeroTierAPIError(str(exc), status_code=exc.status_code) from exc


def authorize_member(network_id: str, node_id: str):
    """Authorize a member on a network. Returns updated member."""
    client = _get_client()
    try:
        return client.authorize_member(network_id, node_id)
    except SDKZeroTierAPIError as exc:
        raise ZeroTierAPIError(str(exc), status_code=exc.status_code) from exc


def deauthorize_member(network_id: str, node_id: str):
    """De-authorize a member on a network. Returns updated member."""
    client = _get_client()
    try:
        return client.deauthorize_member(network_id, node_id)
    except SDKZeroTierAPIError as exc:
        raise ZeroTierAPIError(str(exc), status_code=exc.status_code) from exc


def add_member(network_id: str, node_id: str, authorized: bool = False):
    """Manually add/pre-provision a member on a network."""
    client = _get_client()
    try:
        return client.add_member(network_id, node_id, authorized=authorized)
    except SDKZeroTierAPIError as exc:
        raise ZeroTierAPIError(str(exc), status_code=exc.status_code) from exc


def delete_network_member(network_id: str, node_id: str):
    """Remove a member entirely from a ZeroTier network."""
    client = _get_client()
    try:
        return client.delete_member(network_id, node_id)
    except SDKZeroTierAPIError as exc:
        raise ZeroTierAPIError(str(exc), status_code=exc.status_code) from exc