nexgen_mirrors
1789590167
Add comprehensive ZeroTier integration for managing network access: - Portal networks: manager-created ZeroTier network bindings - Device registration: user-owned ZeroTier node endpoints - Approval workflows: request/approve/revoke network access - Activation sessions: time-limited network authorization - Kill switch: emergency access revocation - Reconciliation job: sync portal state with ZeroTier controller Includes ZeroTier client SDK supporting both Central and self-hosted controller APIs, with full CRUD operations for networks and members.
117 lines
3.7 KiB
Python
117 lines
3.7 KiB
Python
"""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
|