coryHawkvelt/gatehouse-api
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Feat(Fix): Multi-Tenant Zerotier Org Setups

Browse Source 
Imports Network From Zerotier
Async Emails
Migration guardrails
Admin to see all approvals states
This commit is contained in:
James Bhattarai
2026-03-29 23:14:20 +05:45
parent 05eb092228
commit 2b6f7e15af
21 changed files with 974 additions and 239 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gatehouse_app/services/network_access_service.py
+76 -17
View File
@@ -33,6 +33,7 @@ from gatehouse_app.exceptions import (
DeviceNotFoundError,
ApprovalAlreadyExistsError,
ValidationError,
ZeroTierAPIError,
)
logger = logging.getLogger(__name__)
@@ -74,9 +75,30 @@ def request_access(
raise ApprovalAlreadyExistsError(
"An access request or approval already exists for this user and network."
)
existing.state = ApprovalState.PENDING
is_open = network.request_mode.value == "open"
existing.state = ApprovalState.APPROVED if is_open else ApprovalState.PENDING
existing.justification = justification
existing.save()
existing_membership = DeviceNetworkMembership.query.filter(
DeviceNetworkMembership.user_network_approval_id == existing.id,
DeviceNetworkMembership.device_id == device_id,
DeviceNetworkMembership.deleted_at.is_(None),
).first()
if not existing_membership:
membership_state = MembershipState.APPROVED_INACTIVE if is_open else MembershipState.PENDING_DEVICE_REGISTRATION
membership = DeviceNetworkMembership(
organization_id=organization_id,
user_id=user_id,
device_id=device_id,
portal_network_id=portal_network_id,
user_network_approval_id=existing.id,
state=membership_state,
approved_for_activation=is_open,
)
membership.save()
_ensure_zerotier_member(device.node_id, portal_network_id, authorized=False)
return existing
is_open = network.request_mode.value == "open"
@@ -329,6 +351,23 @@ def list_user_approvals(user_id: str, organization_id: str) -> list[UserNetworkA
).all()
def list_all_org_approvals(
organization_id: str,
network_id: str | None = None,
state: str | None = None,
) -> list[UserNetworkApproval]:
"""List all approval records across all users in an org (admin use)."""
q = UserNetworkApproval.query.filter(
UserNetworkApproval.organization_id == organization_id,
UserNetworkApproval.deleted_at.is_(None),
)
if network_id:
q = q.filter(UserNetworkApproval.portal_network_id == network_id)
if state:
q = q.filter(UserNetworkApproval.state == state)
return q.order_by(UserNetworkApproval.created_at.desc()).all()
# ── Membership materialisation ───────────────────────────────────────────────
@@ -428,11 +467,12 @@ def activate_device_membership(
membership_id: str,
user_id: str,
lifetime_minutes: int | None = None,
admin_override: bool = False,
) -> ActivationSession:
"""Activate an approved device on a network. Creates an activation session and authorizes in ZT."""
membership = _get_membership(membership_id)
if membership.user_id != user_id:
if not admin_override and membership.user_id != user_id:
raise MembershipNotFoundError("Membership not found.")
# Check approval is still active
@@ -536,7 +576,8 @@ def deactivate_membership(
# Deauthorize in ZeroTier
device = Device.query.get(membership.device_id)
network = PortalNetwork.query.get(membership.portal_network_id)
_deauthorize_in_zerotier(device.node_id, network.zerotier_network_id)
_deauthorize_in_zerotier(device.node_id, network.zerotier_network_id,
organization_id=membership.organization_id)
membership.state = MembershipState.APPROVED_INACTIVE
membership.currently_authorized = False
@@ -567,6 +608,7 @@ def kill_switch(
target_user_id: str,
triggered_by_user_id: str,
scope: str,
organization_id: str | None = None,
reason: str | None = None,
network_ids: list[str] | None = None,
) -> KillSwitchEvent:
@@ -579,14 +621,18 @@ def kill_switch(
DeviceNetworkMembership.deleted_at.is_(None),
)
org_id = None
org_id = organization_id # Use caller-supplied org_id as the primary source
if scope_enum == KillSwitchScope.ORGANIZATION:
# Use the first membership's org
first = q.first()
org_id = first.organization_id if first else None
if not org_id:
# Fall back to deriving from first active membership
first = q.first()
org_id = first.organization_id if first else None
else:
# Scope query to the specified org
q = q.filter(DeviceNetworkMembership.organization_id == org_id)
elif scope_enum == KillSwitchScope.SELECTED_NETWORKS and network_ids:
q = q.filter(DeviceNetworkMembership.portal_network_id.in_(network_ids))
if network_ids:
if not org_id:
first_network = PortalNetwork.query.filter(
PortalNetwork.id.in_(network_ids),
PortalNetwork.deleted_at.is_(None),
@@ -594,7 +640,7 @@ def kill_switch(
org_id = first_network.organization_id if first_network else None
if not org_id:
org_id = network_ids[0] if network_ids else None
raise ValidationError("Cannot determine organization for kill switch event.")
# Create kill switch event
event = KillSwitchEvent(
@@ -608,14 +654,16 @@ def kill_switch(
event.save()
# Suspend all approvals
ApprovalState._value2member_map_ # just reference
approvals = UserNetworkApproval.query.filter(
UserNetworkApproval.user_id == target_user_id,
UserNetworkApproval.state == ApprovalState.APPROVED,
UserNetworkApproval.deleted_at.is_(None),
).all()
for approval in approvals:
if scope_enum == KillSwitchScope.SELECTED_NETWORKS and network_ids:
if scope_enum == KillSwitchScope.ORGANIZATION and org_id:
if approval.organization_id != org_id:
continue
elif scope_enum == KillSwitchScope.SELECTED_NETWORKS and network_ids:
if approval.portal_network_id not in network_ids:
continue
approval.state = ApprovalState.SUSPENDED
@@ -691,7 +739,8 @@ def _ensure_zerotier_member(
return
try:
zt.add_member(network.zerotier_network_id, node_id, authorized=authorized)
zt.add_member(network.zerotier_network_id, node_id, authorized=authorized,
organization_id=network.organization_id)
except Exception as exc:
logger.warning(
f"[_ensure_zerotier_member] Could not add member {node_id} "
@@ -705,7 +754,8 @@ def _authorize_in_zerotier(
membership: DeviceNetworkMembership,
) -> None:
try:
zt.authorize_member(zerotier_network_id, node_id)
zt.authorize_member(zerotier_network_id, node_id,
organization_id=membership.organization_id)
# Update zerotier_membership cache
zt_membership = ZeroTierMembership.query.filter(
@@ -740,6 +790,11 @@ def _authorize_in_zerotier(
success=True,
)
except ZeroTierAPIError as exc:
logger.warning(
f"[_authorize_in_zerotier] ZeroTier unavailable — skipping authorization "
f"for {node_id} on {zerotier_network_id}: {exc}"
)
except Exception as exc:
logger.error(
f"[_authorize_in_zerotier] Failed to authorize {node_id} "
@@ -748,9 +803,11 @@ def _authorize_in_zerotier(
raise
def _deauthorize_in_zerotier(node_id: str, zerotier_network_id: str) -> None:
def _deauthorize_in_zerotier(node_id: str, zerotier_network_id: str,
organization_id: str | None = None) -> None:
try:
zt.deauthorize_member(zerotier_network_id, node_id)
zt.deauthorize_member(zerotier_network_id, node_id,
organization_id=organization_id)
zt_membership = ZeroTierMembership.query.filter(
ZeroTierMembership.zerotier_network_id == zerotier_network_id,
@@ -940,7 +997,8 @@ def revoke_membership_soft(
if device and network:
try:
zt.deauthorize_member(network.zerotier_network_id, device.node_id)
zt.deauthorize_member(network.zerotier_network_id, device.node_id,
organization_id=membership.organization_id)
except Exception as exc:
logger.warning(f"[revoke_membership_soft] ZT deauthorize failed for {device.node_id}: {exc}")
@@ -984,7 +1042,8 @@ def hard_delete_membership(membership_id: str) -> None:
if device and network:
try:
zt.delete_network_member(network.zerotier_network_id, device.node_id)
zt.delete_network_member(network.zerotier_network_id, device.node_id,
organization_id=membership.organization_id)
logger.info(f"[hard_delete_membership] Deleted {device.node_id} from ZT network {network.zerotier_network_id}")
except Exception as exc:
logger.warning(f"[hard_delete_membership] ZT delete failed for {device.node_id}: {exc}")
gatehouse_app/services/notification_service.py
+78 -97
View File
@@ -17,6 +17,7 @@ from datetime import datetime, timezone
from typing import Optional, Dict, Any
import logging
import json
import threading
from gatehouse_app.extensions import db
from gatehouse_app.models.security.mfa_policy_compliance import MfaPolicyCompliance
@@ -78,29 +79,22 @@ class NotificationService:
)
# Send the notification
success = NotificationService._send_email(
NotificationService._send_email_async(
to_address=user.email,
subject=subject,
body=body,
)
if success:
logger.info(
f"Sent MFA deadline reminder to {user.email} "
f"({days_until_deadline} days remaining)"
)
AuditService.log_action(
action=AuditAction.MFA_POLICY_USER_COMPLIANT,
user_id=user.id,
organization_id=compliance.organization_id,
description=f"MFA deadline reminder sent. Days remaining: {days_until_deadline}",
)
else:
logger.warning(
f"Failed to send MFA deadline reminder to {user.email}"
)
return success
logger.info(
f"Sent MFA deadline reminder to {user.email} "
f"({days_until_deadline} days remaining)"
)
AuditService.log_action(
action=AuditAction.MFA_POLICY_USER_COMPLIANT,
user_id=user.id,
organization_id=compliance.organization_id,
description=f"MFA deadline reminder sent. Days remaining: {days_until_deadline}",
)
return True
except Exception as e:
logger.exception(f"Error sending MFA deadline reminder to {user.email}: {e}")
@@ -136,27 +130,19 @@ class NotificationService:
)
# Send the notification
success = NotificationService._send_email(
NotificationService._send_email_async(
to_address=user.email,
subject=subject,
body=body,
)
if success:
logger.info(f"Sent MFA suspension notification to {user.email}")
# Audit log
AuditService.log_action(
action=AuditAction.MFA_POLICY_USER_SUSPENDED,
user_id=user.id,
organization_id=compliance.organization_id,
description="MFA compliance suspension notification sent",
)
else:
logger.warning(
f"Failed to send MFA suspension notification to {user.email}"
)
return success
logger.info(f"Sent MFA suspension notification to {user.email}")
AuditService.log_action(
action=AuditAction.MFA_POLICY_USER_SUSPENDED,
user_id=user.id,
organization_id=compliance.organization_id,
description="MFA compliance suspension notification sent",
)
return True
except Exception as e:
logger.exception(
@@ -285,89 +271,84 @@ Gatehouse Security Team
return body
@staticmethod
def _send_email(
def _send_email_async(
to_address: str,
subject: str,
body: str,
html_body: Optional[str] = None,
) -> bool:
"""Send an email via SMTP.
) -> None:
"""Send an email on a daemon thread so the calling request returns immediately.
Returns True if the email was sent successfully, False otherwise.
If EMAIL_ENABLED is False, logs the email body instead (simulation mode).
If EMAIL_ENABLED is False, logs instead of sending.
All SMTP exceptions are caught and logged — this method never raises.
The Flask app context is pushed inside the thread so current_app works correctly.
"""
import smtplib
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from flask import current_app
email_enabled = current_app.config.get(NotificationService.EMAIL_ENABLED_KEY, False)
app = current_app._get_current_object() # capture real app before leaving request context
if not email_enabled:
logger.info(
f"[EMAIL DISABLED] Would have sent to: {to_address} | Subject: {subject}\n"
f"Body: {body[:500]}"
)
return False
def _send():
with app.app_context():
email_enabled = app.config.get(NotificationService.EMAIL_ENABLED_KEY, False)
if not email_enabled:
logger.info(
f"[EMAIL DISABLED] Would have sent to: {to_address} | Subject: {subject}\n"
f"Body: {body[:500]}"
)
return
smtp_host = current_app.config.get(NotificationService.SMTP_HOST_KEY, "")
smtp_port_raw = current_app.config.get(NotificationService.SMTP_PORT_KEY, 587)
smtp_username = current_app.config.get(NotificationService.SMTP_USERNAME_KEY)
smtp_password = current_app.config.get(NotificationService.SMTP_PASSWORD_KEY)
from_address = current_app.config.get(
NotificationService.FROM_ADDRESS_KEY, ""
)
smtp_host = app.config.get(NotificationService.SMTP_HOST_KEY, "")
smtp_port_raw = app.config.get(NotificationService.SMTP_PORT_KEY, 587)
smtp_username = app.config.get(NotificationService.SMTP_USERNAME_KEY)
smtp_password = app.config.get(NotificationService.SMTP_PASSWORD_KEY)
from_address = app.config.get(NotificationService.FROM_ADDRESS_KEY, "")
# Guard: refuse to attempt a connection when critical config is missing.
# This surfaces a clear log message instead of a confusing socket error.
missing = [k for k, v in [
("SMTP_HOST", smtp_host),
("FROM_ADDRESS", from_address),
] if not v]
if missing:
logger.error(
f"[EMAIL] Cannot send — missing config: {', '.join(missing)}. "
f"Would have sent to: {to_address} | Subject: {subject}"
)
return False
missing = [k for k, v in [("SMTP_HOST", smtp_host), ("FROM_ADDRESS", from_address)] if not v]
if missing:
logger.error(
f"[EMAIL] Cannot send — missing config: {', '.join(missing)}. "
f"Would have sent to: {to_address} | Subject: {subject}"
)
return
try:
smtp_port = int(smtp_port_raw)
except (TypeError, ValueError):
logger.error(f"[EMAIL] Invalid SMTP_PORT value: {smtp_port_raw!r}")
return False
try:
smtp_port = int(smtp_port_raw)
except (TypeError, ValueError):
logger.error(f"[EMAIL] Invalid SMTP_PORT value: {smtp_port_raw!r}")
return
smtp_use_tls = current_app.config.get(
NotificationService.SMTP_USE_TLS_KEY,
smtp_port not in (25, 1025),
)
smtp_use_tls = app.config.get(
NotificationService.SMTP_USE_TLS_KEY,
smtp_port not in (25, 1025),
)
try:
msg = MIMEMultipart("alternative")
msg["Subject"] = subject
msg["From"] = from_address
msg["To"] = to_address
msg.attach(MIMEText(body, "plain"))
if html_body:
msg.attach(MIMEText(html_body, "html"))
try:
msg = MIMEMultipart("alternative")
msg["Subject"] = subject
msg["From"] = from_address
msg["To"] = to_address
msg.attach(MIMEText(body, "plain"))
if html_body:
msg.attach(MIMEText(html_body, "html"))
with smtplib.SMTP(smtp_host, smtp_port) as server:
server.ehlo()
if smtp_use_tls:
server.starttls()
server.ehlo()
if smtp_username and smtp_password:
server.login(smtp_username, smtp_password)
server.send_message(msg)
with smtplib.SMTP(smtp_host, smtp_port) as server:
server.ehlo()
if smtp_use_tls:
server.starttls()
server.ehlo()
if smtp_username and smtp_password:
server.login(smtp_username, smtp_password)
server.send_message(msg)
logger.info(f"[EMAIL] Sent to {to_address} | Subject: {subject}")
return True
logger.info(f"[EMAIL] Sent to {to_address} | Subject: {subject}")
except Exception as e:
logger.error(f"[EMAIL] Failed to send to {to_address}: {e}")
return False
except Exception as e:
logger.error(f"[EMAIL] Failed to send to {to_address}: {e}")
threading.Thread(target=_send, daemon=True).start()
@staticmethod
def get_notification_stats(user_id: str) -> Dict[str, Any]:
gatehouse_app/services/portal_network_service.py
+112 -10
View File
@@ -9,7 +9,7 @@ from gatehouse_app.models.organization import Organization
from gatehouse_app.models.user import User
from gatehouse_app.services.audit_service import AuditService
from gatehouse_app.services import zerotier_api_service as zt
from gatehouse_app.utils.constants import NetworkRequestMode
from gatehouse_app.utils.constants import NetworkRequestMode, NetworkEnvironment
from gatehouse_app.exceptions import (
NetworkNotFoundError,
InvalidNetworkIdError,
@@ -57,23 +57,74 @@ def create_network(
default_activation_lifetime_minutes: Default session length
max_activation_lifetime_minutes: Cap on activation lifetime
"""
from gatehouse_app.utils.constants import NetworkEnvironment
zerotier_network_id = _validate_network_id(zerotier_network_id)
existing = PortalNetwork.query.filter(
existing_active = PortalNetwork.query.filter(
PortalNetwork.organization_id == organization_id,
PortalNetwork.zerotier_network_id == zerotier_network_id,
PortalNetwork.deleted_at.is_(None),
).first()
if existing:
if existing_active:
raise ValidationError(
f"A portal network already exists for ZT network {zerotier_network_id} "
f"in this organization."
)
env = NetworkEnvironment(environment) if environment else NetworkEnvironment.DEVELOPMENT
mode = NetworkRequestMode(request_mode)
# Normalize to lowercase so callers may pass "PRODUCTION" or "production" interchangeably
env_str = environment.lower() if environment else None
mode_str = request_mode.lower() if request_mode else "approval_required"
try:
env = NetworkEnvironment(env_str) if env_str else NetworkEnvironment.DEVELOPMENT
except ValueError:
valid = [e.value for e in NetworkEnvironment]
raise ValidationError(f"Invalid environment '{environment}'. Must be one of: {valid}")
try:
mode = NetworkRequestMode(mode_str)
except ValueError:
valid = [e.value for e in NetworkRequestMode]
raise ValidationError(f"Invalid request_mode '{request_mode}'. Must be one of: {valid}")
# If a soft-deleted record for the same (org, zt_network_id) pair exists, restore it
# rather than inserting a new row (which would violate the unique constraint).
deleted = PortalNetwork.query.filter(
PortalNetwork.organization_id == organization_id,
PortalNetwork.zerotier_network_id == zerotier_network_id,
PortalNetwork.deleted_at.isnot(None),
).first()
if deleted:
logger.info(
f"[PortalNetwork] Restoring soft-deleted portal network {deleted.id} "
f"for ZT network {zerotier_network_id}"
)
deleted.deleted_at = None
deleted.name = name
deleted.description = description
deleted.owner_user_id = owner_user_id
deleted.environment = env
deleted.request_mode = mode
deleted.default_activation_lifetime_minutes = default_activation_lifetime_minutes
deleted.max_activation_lifetime_minutes = max_activation_lifetime_minutes
deleted.is_active = True
deleted.save()
AuditService.log_action(
action="zt.network.restored",
user_id=owner_user_id,
organization_id=organization_id,
resource_type="portal_network",
resource_id=deleted.id,
metadata={
"zerotier_network_id": zerotier_network_id,
"name": name,
"environment": env.value,
"request_mode": mode.value,
},
description=f"Portal network '{name}' restored (ZT: {zerotier_network_id})",
success=True,
)
return deleted
network = PortalNetwork(
organization_id=organization_id,
@@ -90,7 +141,7 @@ def create_network(
# Try to verify the network exists in ZeroTier
try:
zt_network = zt.get_network(zerotier_network_id)
zt_network = zt.get_network(zerotier_network_id, organization_id=organization_id)
logger.info(
f"[PortalNetwork] Verified ZT network {zerotier_network_id} "
f"exists in ZeroTier: {zt_network.name}"
@@ -100,7 +151,7 @@ def create_network(
f"[PortalNetwork] ZT network {zerotier_network_id} not found "
"in ZeroTier — will be reconciled later."
)
except ZeroTierAPIError as exc:
except (ZeroTierAPIError, Exception) as exc:
logger.warning(
f"[PortalNetwork] Could not verify ZT network {zerotier_network_id}: {exc}"
)
@@ -175,6 +226,23 @@ def update_network(
if key not in allowed:
raise ValidationError(f"Cannot update field: {key}")
# Normalize environment / request_mode strings to lowercase enum values
if "environment" in kwargs and isinstance(kwargs["environment"], str):
env_str = kwargs["environment"].lower()
try:
kwargs["environment"] = NetworkEnvironment(env_str)
except ValueError:
valid = [e.value for e in NetworkEnvironment]
raise ValidationError(f"Invalid environment '{kwargs['environment']}'. Must be one of: {valid}")
if "request_mode" in kwargs and isinstance(kwargs["request_mode"], str):
mode_str = kwargs["request_mode"].lower()
try:
kwargs["request_mode"] = NetworkRequestMode(mode_str)
except ValueError:
valid = [e.value for e in NetworkRequestMode]
raise ValidationError(f"Invalid request_mode '{kwargs['request_mode']}'. Must be one of: {valid}")
network.update(**kwargs)
AuditService.log_action(
@@ -192,7 +260,11 @@ def update_network(
def delete_network(network_id: str, user_id: str) -> None:
"""Soft-delete a portal network and deactivate all memberships."""
"""Soft-delete a portal network and deactivate/clean up all related records."""
from datetime import datetime, timezone
from gatehouse_app.models import UserNetworkApproval
from gatehouse_app.extensions import db
network = get_network(network_id)
# Deauthorize all active memberships in ZeroTier
@@ -203,6 +275,36 @@ def delete_network(network_id: str, user_id: str) -> None:
network.delete(soft=True)
# Cascade soft-delete all active approvals and memberships for this network.
now = datetime.now(timezone.utc)
db.session.execute(
db.text(
"UPDATE user_network_approvals AS a "
"SET deleted_at = :now + (s.rn * interval '1 microsecond') "
"FROM ("
" SELECT id, row_number() OVER () AS rn "
" FROM user_network_approvals "
" WHERE portal_network_id = :network_id AND deleted_at IS NULL"
") s "
"WHERE a.id = s.id"
),
{"now": now, "network_id": network_id},
)
db.session.execute(
db.text(
"UPDATE device_network_memberships AS m "
"SET deleted_at = :now + (s.rn * interval '1 microsecond') "
"FROM ("
" SELECT id, row_number() OVER () AS rn "
" FROM device_network_memberships "
" WHERE portal_network_id = :network_id AND deleted_at IS NULL"
") s "
"WHERE m.id = s.id"
),
{"now": now, "network_id": network_id},
)
db.session.commit()
AuditService.log_action(
action="zt.network.deleted",
user_id=user_id,
gatehouse_app/services/zerotier_api_service.py
+83 -29
View File
@@ -1,7 +1,11 @@
"""ZeroTier API service — thin Flask adapter around the ZeroTierClient SDK.
Reads configuration from app config and translates SDK exceptions to
Secuird typed exceptions.
ZeroTier is managed exclusively at the organization level. Each organization
configures its own ZeroTier credentials (token, URL, mode) via the web UI
(ZeroTier Config page → stored in the organizations table).
Every call that interacts with ZeroTier must supply an organization_id so the correct org credentials
can be loaded from the database.
"""
import logging
@@ -19,97 +23,147 @@ from gatehouse_app.utils.zerotier_client import (
logger = logging.getLogger(__name__)
def _get_client(app=None) -> ZeroTierClient:
"""Build a ZeroTierClient from current app config."""
from flask import current_app
def _get_client(organization_id: Optional[str] = None, app=None) -> ZeroTierClient:
"""Build a ZeroTierClient using the organization's stored ZeroTier credentials.
app = app or current_app
Credentials are read exclusively from the organization record
(org.zt_api_token / org.zt_api_url / org.zt_api_mode).
Args:
organization_id: The org whose credentials should be used.
Required for any ZeroTier operation.
app: Flask app instance (defaults to current_app, only needed for
background tasks that run outside a request context).
Raises:
ZeroTierAPIError: If organization_id is missing, the org is not found,
or the org has incomplete ZeroTier credentials.
"""
if not organization_id:
raise ZeroTierAPIError(
"organization_id is required — ZeroTier credentials are managed "
"per-organization. Configure them via the ZeroTier Config page."
)
try:
from gatehouse_app.models.organization.organization import Organization
from gatehouse_app.extensions import db
org = db.session.get(Organization, organization_id)
except Exception as exc:
logger.error(f"[ZT] Failed to load org {organization_id} from DB: {exc}")
raise ZeroTierAPIError(
f"Could not load organization {organization_id}: {exc}"
) from exc
if not org:
raise ZeroTierAPIError(f"Organization {organization_id} not found.")
token: Optional[str] = org.zt_api_token or None
if not token:
raise ZeroTierAPIError(
f"Organization '{org.name}' has no ZeroTier credentials configured. "
"Go to Settings → ZeroTier Config to add a token, mode, and controller URL."
)
mode_str = (org.zt_api_mode or "").strip().lower()
if mode_str not in ("central", "controller"):
raise ZeroTierAPIError(
f"Organization '{org.name}' has no ZeroTier mode set. "
"Go to Settings → ZeroTier Config and select 'Central' or 'Controller'."
)
url: str = (org.zt_api_url or "").strip()
if not url:
raise ZeroTierAPIError(
f"Organization '{org.name}' has no ZeroTier controller/API URL set. "
"Go to Settings → ZeroTier Config and enter the URL for your ZeroTier "
"controller (e.g. http://host:9993) or Central API."
)
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,
logger.debug(
f"[ZT] Client for org:{organization_id} mode={mode_str} url={url}"
)
return ZeroTierClient(api_token=token, base_url=url, mode=mode)
def get_status() -> dict:
def get_status(organization_id: Optional[str] = None) -> dict:
"""Verify connectivity to the ZeroTier controller."""
client = _get_client()
client = _get_client(organization_id)
try:
return client.get_status()
except SDKZeroTierAPIError as exc:
raise ZeroTierAPIError(str(exc), status_code=exc.status_code) from exc
def list_networks():
def list_networks(organization_id: Optional[str] = None):
"""List all networks accessible to the configured token."""
client = _get_client()
client = _get_client(organization_id)
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):
def get_network(network_id: str, organization_id: Optional[str] = None):
"""Fetch a single network by ID."""
client = _get_client()
client = _get_client(organization_id)
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):
def list_members(network_id: str, organization_id: Optional[str] = None):
"""List all members on a network."""
client = _get_client()
client = _get_client(organization_id)
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):
def get_member(network_id: str, node_id: str, organization_id: Optional[str] = None):
"""Fetch a single member on a network."""
client = _get_client()
client = _get_client(organization_id)
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):
def authorize_member(network_id: str, node_id: str, organization_id: Optional[str] = None):
"""Authorize a member on a network. Returns updated member."""
client = _get_client()
client = _get_client(organization_id)
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):
def deauthorize_member(network_id: str, node_id: str, organization_id: Optional[str] = None):
"""De-authorize a member on a network. Returns updated member."""
client = _get_client()
client = _get_client(organization_id)
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):
def add_member(network_id: str, node_id: str, authorized: bool = False, organization_id: Optional[str] = None):
"""Manually add/pre-provision a member on a network."""
client = _get_client()
client = _get_client(organization_id)
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):
def delete_network_member(network_id: str, node_id: str, organization_id: Optional[str] = None):
"""Remove a member entirely from a ZeroTier network."""
client = _get_client()
client = _get_client(organization_id)
try:
return client.delete_member(network_id, node_id)
except SDKZeroTierAPIError as exc:
gatehouse_app/services/zerotier_reconciliation_service.py
+194 -31
View File
@@ -1,6 +1,7 @@
"""ZeroTier reconciliation service — polling loop to sync state with the controller."""
import logging
import time
from datetime import datetime, timezone
from gatehouse_app.extensions import db
@@ -34,16 +35,24 @@ def reconcile_expired_activations() -> int:
ActivationSession.deleted_at.is_(None),
).all()
logger.debug(f"[Reconciliation] Expiry check: {len(expired)} overdue session(s) found.")
count = 0
for session in expired:
try:
_expire_session(session)
count += 1
except Exception as exc:
logger.error(f"[Reconciliation] Failed to expire session {session.id}: {exc}")
logger.error(
f"[Reconciliation] Failed to expire session {session.id} "
f"(user={session.user_id} membership={session.device_network_membership_id}): {exc}",
exc_info=True,
)
if count > 0:
logger.info(f"[Reconciliation] Expired {count} activation sessions.")
logger.info(f"[Reconciliation] Expired {count} activation session(s).")
else:
logger.debug("[Reconciliation] No activation sessions to expire.")
return count
@@ -55,9 +64,14 @@ def reconcile_network(portal_network_id: str) -> dict:
"""
network = PortalNetwork.query.get(portal_network_id)
if not network or not network.is_active:
logger.debug(
f"[Reconciliation] Skipping portal_network_id={portal_network_id}: "
f"{'not found' if not network else 'inactive or deleted'}."
)
return {"skipped": True, "reason": "network_inactive_or_deleted"}
zerotier_network_id = network.zerotier_network_id
network_label = f"{network.name} ({zerotier_network_id})"
actions = {
"zt_members_checked": 0,
"zt_members_added": 0,
@@ -67,15 +81,25 @@ def reconcile_network(portal_network_id: str) -> dict:
"unknown_members": [],
}
t_start = time.monotonic()
logger.debug(f"[Reconciliation] Starting network reconciliation for {network_label}.")
# Get current ZT members
try:
zt_members = {m.node_id: m for m in zt.list_members(zerotier_network_id)}
zt_members = {m.node_id: m for m in zt.list_members(zerotier_network_id,
organization_id=network.organization_id)}
except Exception as exc:
logger.error(f"[Reconciliation] Failed to list ZT members for {zerotier_network_id}: {exc}")
logger.error(
f"[Reconciliation] Failed to list ZT members for {network_label}: {exc}",
exc_info=True,
)
actions["error"] = str(exc)
return actions
actions["zt_members_checked"] = len(zt_members)
logger.debug(
f"[Reconciliation] {network_label}: {len(zt_members)} member(s) fetched from ZT controller."
)
# Get our portal memberships for this network
our_memberships = {
@@ -87,13 +111,21 @@ def reconcile_network(portal_network_id: str) -> dict:
if m.device and m.device.deleted_at is None
}
logger.debug(
f"[Reconciliation] {network_label}: {len(our_memberships)} portal membership(s) to reconcile."
)
# Reconcile each portal membership
for node_id, membership in our_memberships.items():
zt_member = zt_members.pop(node_id, None)
device = membership.device
if not zt_member:
# Member not seen in ZT yet
# Member not seen in ZT yet — could be freshly joined or never connected
logger.debug(
f"[Reconciliation] {network_label}: node {node_id} "
f"(device={device.display_name!r}, state={membership.state}) not yet seen in ZT controller."
)
continue
actions["join_seen_updated"] += 1
@@ -104,31 +136,67 @@ def reconcile_network(portal_network_id: str) -> dict:
# Sync authorization state
if membership.state == MembershipState.ACTIVE_AUTHORIZED:
if not zt_member.is_authorized:
# We think it's active but ZT says it's not — re-authorize
# Portal says active but ZT disagrees — drift, re-authorize
logger.warning(
f"[Reconciliation] {network_label}: DRIFT detected — portal=ACTIVE_AUTHORIZED "
f"but ZT says unauthorized for node {node_id} (device={device.display_name!r}). Re-authorizing."
)
try:
zt.authorize_member(zerotier_network_id, node_id)
zt.authorize_member(zerotier_network_id, node_id,
organization_id=network.organization_id)
actions["authorized"] += 1
logger.info(
f"[Reconciliation] {network_label}: Re-authorized node {node_id} (device={device.display_name!r})."
)
except Exception as exc:
logger.warning(f"[Reconciliation] Re-authorize failed for {node_id}: {exc}")
logger.warning(
f"[Reconciliation] {network_label}: Re-authorize failed for node {node_id}: {exc}"
)
else:
logger.debug(
f"[Reconciliation] {network_label}: node {node_id} — portal=ACTIVE_AUTHORIZED, ZT=authorized. OK."
)
else:
if zt_member.is_authorized:
# We think it's not authorized but ZT says it is — deauthorize
# (could be manual override in ZT console)
# ZT says authorized but portal doesn't — could be manual override in ZT console
logger.warning(
f"[Reconciliation] {network_label}: DRIFT detected — portal state={membership.state} "
f"but ZT says authorized for node {node_id} (device={device.display_name!r}). Deauthorizing."
)
try:
zt.deauthorize_member(zerotier_network_id, node_id)
zt.deauthorize_member(zerotier_network_id, node_id,
organization_id=network.organization_id)
actions["deauthorized"] += 1
logger.info(
f"[Reconciliation] {network_label}: Deauthorized node {node_id} (device={device.display_name!r})."
)
except Exception as exc:
logger.warning(f"[Reconciliation] Deauthorize failed for {node_id}: {exc}")
logger.warning(
f"[Reconciliation] {network_label}: Deauthorize failed for node {node_id}: {exc}"
)
else:
logger.debug(
f"[Reconciliation] {network_label}: node {node_id} — "
f"portal={membership.state}, ZT=unauthorized. OK."
)
# Unknown ZT members not in our portal
actions["unknown_members"] = list(zt_members.keys())
# Unknown ZT members not in our portal — log only, do not touch
unknown = list(zt_members.keys())
actions["unknown_members"] = unknown
if unknown:
logger.warning(
f"[Reconciliation] {network_label}: {len(unknown)} ZT member(s) not in portal — "
f"node IDs: {', '.join(unknown)}"
)
elapsed_ms = int((time.monotonic() - t_start) * 1000)
logger.info(
f"[Reconciliation] Network {zerotier_network_id}: "
f"[Reconciliation] Network {network_label}: "
f"checked={actions['zt_members_checked']} "
f"authorized={actions['authorized']} "
f"deauthorized={actions['deauthorized']} "
f"unknown={len(actions['unknown_members'])}"
f"unknown={len(actions['unknown_members'])} "
f"elapsed={elapsed_ms}ms"
)
return actions
@@ -144,16 +212,34 @@ def reconcile_all() -> dict:
PortalNetwork.deleted_at.is_(None),
).all()
results = {"networks_processed": 0, "errors": 0}
logger.info(f"[Reconciliation] reconcile_all: {len(networks)} active network(s) to process.")
results = {"networks_processed": 0, "errors": 0, "authorized": 0, "deauthorized": 0, "unknown_members": []}
for network in networks:
try:
result = reconcile_network(network.id)
if "error" in result:
logger.error(
f"[Reconciliation] Network {network.name} ({network.zerotier_network_id}) "
f"failed: {result['error']}"
)
results["errors"] += 1
elif result.get("skipped"):
logger.debug(
f"[Reconciliation] Network {network.name} ({network.zerotier_network_id}) "
f"skipped: {result.get('reason')}"
)
else:
results["networks_processed"] += 1
results["authorized"] += result.get("authorized", 0)
results["deauthorized"] += result.get("deauthorized", 0)
results["unknown_members"].extend(result.get("unknown_members", []))
except Exception as exc:
logger.error(f"[Reconciliation] Failed to reconcile network {network.id}: {exc}")
logger.error(
f"[Reconciliation] Unhandled error reconciling network "
f"{network.name} ({network.id}): {exc}",
exc_info=True,
)
results["errors"] += 1
deleted_result = reconcile_deleted_memberships()
@@ -161,8 +247,11 @@ def reconcile_all() -> dict:
results["delete_errors"] = deleted_result.get("errors", 0)
logger.info(
f"[Reconciliation] Complete: {results['networks_processed']} networks processed, "
f"{results['errors']} errors, {results.get('deleted_memberships', 0)} memberships purged."
f"[Reconciliation] Complete: "
f"networks={results['networks_processed']} "
f"errors={results['errors']} "
f"purged={results.get('deleted_memberships', 0)} "
f"purge_errors={results.get('delete_errors', 0)}"
)
return results
@@ -180,8 +269,11 @@ def reconcile_deleted_memberships() -> dict:
).all()
if not deleted:
logger.debug("[Reconciliation] No soft-deleted memberships to purge.")
return {"deleted": 0, "errors": 0}
logger.info(f"[Reconciliation] Purging {len(deleted)} soft-deleted membership(s) from ZT and DB.")
results = {"deleted": 0, "errors": 0}
for membership in deleted:
try:
@@ -189,30 +281,49 @@ def reconcile_deleted_memberships() -> dict:
network = PortalNetwork.query.get(membership.portal_network_id)
if not device or not network:
logger.warning(
f"[Reconciliation] Membership {membership.id}: missing "
f"{'device' if not device else 'network'} — hard-deleting record only."
)
db.session.delete(membership)
db.session.commit()
results["deleted"] += 1
continue
node_id = device.node_id
zt_network_id = network.zerotier_network_id
network_label = f"{network.name} ({zt_network_id})"
try:
zt.delete_network_member(network.zerotier_network_id, device.node_id)
logger.info(f"[Reconciliation] Deleted {device.node_id} from ZT network {network.zerotier_network_id}")
zt.delete_network_member(zt_network_id, node_id,
organization_id=network.organization_id)
logger.info(
f"[Reconciliation] Removed node {node_id} (device={device.display_name!r}) "
f"from ZT network {network_label}."
)
except Exception as zt_exc:
logger.warning(
f"[Reconciliation] ZT delete failed for {device.node_id} "
f"on {network.zerotier_network_id}: {zt_exc}"
f"[Reconciliation] ZT delete failed for node {node_id} "
f"on {network_label}: {zt_exc} — proceeding with DB hard-delete."
)
db.session.delete(membership)
db.session.commit()
results["deleted"] += 1
logger.debug(
f"[Reconciliation] Hard-deleted membership {membership.id} "
f"(node={node_id}, network={network_label})."
)
except Exception as exc:
logger.error(f"[Reconciliation] Failed to hard-delete membership {membership.id}: {exc}")
logger.error(
f"[Reconciliation] Failed to hard-delete membership {membership.id}: {exc}",
exc_info=True,
)
results["errors"] += 1
if results["deleted"] > 0:
logger.info(f"[Reconciliation] Purged {results['deleted']} memberships.")
logger.info(f"[Reconciliation] Purged {results['deleted']} membership(s).")
return results
@@ -228,7 +339,12 @@ def _sync_zt_membership(membership: DeviceNetworkMembership, zt_member) -> None:
ZeroTierMembership.deleted_at.is_(None),
).first()
if not zt_membership:
is_new = zt_membership is None
if is_new:
logger.debug(
f"[Reconciliation] Creating new ZeroTierMembership cache record for "
f"node {device.node_id} on network {network.zerotier_network_id}."
)
zt_membership = ZeroTierMembership(
organization_id=membership.organization_id,
device_network_membership_id=membership.id,
@@ -236,6 +352,8 @@ def _sync_zt_membership(membership: DeviceNetworkMembership, zt_member) -> None:
node_id=device.node_id,
)
prev_authorized = zt_membership.authorized if not is_new else None
zt_membership.member_seen = True
zt_membership.authorized = zt_member.is_authorized
zt_membership.last_synced_at = datetime.now(timezone.utc)
@@ -248,11 +366,27 @@ def _sync_zt_membership(membership: DeviceNetworkMembership, zt_member) -> None:
zt_membership.save()
if not is_new and prev_authorized != zt_member.is_authorized:
logger.info(
f"[Reconciliation] ZT auth state changed for node {device.node_id} "
f"(device={device.display_name!r}): {prev_authorized} → {zt_member.is_authorized}"
)
# Update membership join_seen flag
if not membership.join_seen:
logger.info(
f"[Reconciliation] First join seen for node {device.node_id} "
f"(device={device.display_name!r}, membership={membership.id}). "
f"State: {membership.state} → {MembershipState.JOINED_DEAUTHORIZED}"
)
membership.join_seen = True
membership.state = MembershipState.JOINED_DEAUTHORIZED
membership.save()
else:
logger.debug(
f"[Reconciliation] Synced ZT membership for node {device.node_id} "
f"(device={device.display_name!r}, authorized={zt_member.is_authorized})."
)
def _expire_session(session: ActivationSession) -> None:
@@ -261,8 +395,19 @@ def _expire_session(session: ActivationSession) -> None:
session.end_reason = ActivationEndReason.EXPIRED
session.save()
logger.info(
f"[Reconciliation] Expiring activation session {session.id} "
f"(user={session.user_id}, membership={session.device_network_membership_id}, "
f"expired_at={session.expires_at.isoformat()})."
)
membership = DeviceNetworkMembership.query.get(session.device_network_membership_id)
if membership:
if not membership:
logger.warning(
f"[Reconciliation] Session {session.id}: membership "
f"{session.device_network_membership_id} not found — skipping ZT deauth."
)
else:
membership.state = MembershipState.ACTIVATION_EXPIRED
membership.currently_authorized = False
membership.save()
@@ -270,8 +415,14 @@ def _expire_session(session: ActivationSession) -> None:
device = Device.query.get(membership.device_id)
network = PortalNetwork.query.get(membership.portal_network_id)
if device and network:
network_label = f"{network.name} ({network.zerotier_network_id})"
try:
zt.deauthorize_member(network.zerotier_network_id, device.node_id)
zt.deauthorize_member(network.zerotier_network_id, device.node_id,
organization_id=network.organization_id)
logger.info(
f"[Reconciliation] Deauthorized expired node {device.node_id} "
f"(device={device.display_name!r}) on {network_label}."
)
# Update ZT membership cache
zt_membership = ZeroTierMembership.query.filter(
@@ -283,12 +434,24 @@ def _expire_session(session: ActivationSession) -> None:
zt_membership.authorized = False
zt_membership.last_synced_at = datetime.now(timezone.utc)
zt_membership.save()
else:
logger.debug(
f"[Reconciliation] No ZeroTierMembership cache record found for "
f"node {device.node_id} on {network_label} — nothing to update."
)
except Exception as exc:
logger.warning(
f"[_expire_session] Failed to deauthorize {device.node_id} "
f"on {network.zerotier_network_id}: {exc}"
f"[_expire_session] Failed to deauthorize node {device.node_id} "
f"on {network_label}: {exc}",
exc_info=True,
)
else:
logger.warning(
f"[Reconciliation] Session {session.id}: missing "
f"{'device' if not device else 'network'} for membership "
f"{membership.id} — ZT deauth skipped."
)
from gatehouse_app.services.audit_service import AuditService
AuditService.log_action(