| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112 |
- """
- Superadmin endpoints for system settings.
- """
- from datetime import datetime, timedelta, timezone
- from typing import Annotated
- from fastapi import APIRouter, Depends, HTTPException, status
- from pydantic import BaseModel
- from sqlalchemy import select
- from sqlalchemy.ext.asyncio import AsyncSession
- from app.api.deps import get_current_superadmin
- from app.core.database import get_db
- from app.models.settings import Settings
- from app.models.user import User
- router = APIRouter()
- class AutoRegistrationResponse(BaseModel):
- """Auto-registration status response."""
- enabled: bool
- last_device_at: str | None
- class AutoRegistrationToggleRequest(BaseModel):
- """Auto-registration toggle request."""
- enabled: bool
- @router.get("/auto-registration", response_model=AutoRegistrationResponse)
- async def get_auto_registration_status(
- db: Annotated[AsyncSession, Depends(get_db)],
- current_user: Annotated[User, Depends(get_current_superadmin)],
- ):
- """
- Get auto-registration status.
- Returns current state and last device registration timestamp.
- """
- result = await db.execute(
- select(Settings).where(Settings.key == "auto_registration")
- )
- setting = result.scalar_one_or_none()
- if not setting:
- # Create default setting if not exists
- setting = Settings(
- key="auto_registration",
- value={"enabled": False, "last_device_at": None},
- )
- db.add(setting)
- await db.commit()
- # Check if should auto-disable (1 hour since last device)
- enabled = setting.value.get("enabled", False)
- last_device_at_str = setting.value.get("last_device_at")
- if enabled and last_device_at_str:
- last_device_at = datetime.fromisoformat(last_device_at_str)
- if datetime.now(timezone.utc) - last_device_at > timedelta(hours=1):
- # Auto-disable
- setting.value["enabled"] = False
- await db.commit()
- enabled = False
- return AutoRegistrationResponse(
- enabled=enabled,
- last_device_at=last_device_at_str,
- )
- @router.post("/auto-registration", response_model=AutoRegistrationResponse)
- async def toggle_auto_registration(
- data: AutoRegistrationToggleRequest,
- db: Annotated[AsyncSession, Depends(get_db)],
- current_user: Annotated[User, Depends(get_current_superadmin)],
- ):
- """
- Enable or disable auto-registration of new devices.
- When enabled, devices that authenticate with unknown MAC addresses will be
- automatically registered in the system (unassigned to any organization).
- Auto-registration will automatically disable after 1 hour of the last registered device.
- """
- result = await db.execute(
- select(Settings).where(Settings.key == "auto_registration")
- )
- setting = result.scalar_one_or_none()
- if not setting:
- setting = Settings(
- key="auto_registration",
- value={"enabled": data.enabled, "last_device_at": None},
- )
- db.add(setting)
- else:
- setting.value["enabled"] = data.enabled
- if not data.enabled:
- # Reset last_device_at when disabling
- setting.value["last_device_at"] = None
- await db.commit()
- return AutoRegistrationResponse(
- enabled=setting.value["enabled"],
- last_device_at=setting.value.get("last_device_at"),
- )
|