from fastapi import APIRouter, Depends, HTTPException
from app.api import schemas
from app.api.dependencies import get_current_admin
from app.config import settings
def create_admin_router() -> APIRouter:
router = APIRouter()
@router.put("/config/oidc", summary="Update OIDC Configuration")
async def update_oidc_config(
update: schemas.OIDCConfigUpdate,
admin = Depends(get_current_admin)
):
checking_disabled = False
if update.enabled is not None:
checking_disabled = not update.enabled
if update.allow_oidc_login is not None:
checking_disabled = not update.allow_oidc_login
if checking_disabled:
if not settings.ALLOW_PASSWORD_LOGIN:
raise HTTPException(status_code=400, detail="Cannot disable OIDC login while password login is also disabled.")
if not admin.password_hash:
raise HTTPException(status_code=400, detail="SafeGuard: Cannot disable OIDC! You do not have a local password set, which would lock you out of your Admin account.")
if update.enabled is not None:
settings.OIDC_ENABLED = update.enabled
if update.client_id is not None:
settings.OIDC_CLIENT_ID = update.client_id
if update.client_secret is not None:
settings.OIDC_CLIENT_SECRET = update.client_secret
if update.server_url is not None:
settings.OIDC_SERVER_URL = update.server_url
if update.redirect_uri is not None:
settings.OIDC_REDIRECT_URI = update.redirect_uri
if update.allow_oidc_login is not None:
settings.ALLOW_OIDC_LOGIN = update.allow_oidc_login
settings.save_to_yaml()
return {"message": "OIDC configuration updated successfully"}
@router.put("/config/app", summary="Update Application Configuration")
async def update_app_config(
update: schemas.AppConfigUpdate,
admin = Depends(get_current_admin)
):
if update.allow_password_login is not None:
is_oidc_active = settings.ALLOW_OIDC_LOGIN or settings.OIDC_ENABLED
if not update.allow_password_login and not is_oidc_active:
raise HTTPException(status_code=400, detail="Cannot disable password login while OIDC login is also disabled.")
settings.ALLOW_PASSWORD_LOGIN = update.allow_password_login
settings.save_to_yaml()
return {"message": "Application configuration updated successfully"}
@router.post("/config/oidc/test", summary="Test OIDC Discovery")
async def test_oidc_connection(
update: schemas.OIDCConfigUpdate,
admin = Depends(get_current_admin)
):
if not update.server_url:
raise HTTPException(status_code=400, detail="Server URL is required for testing.")
import httpx
try:
discovery_url = update.server_url.rstrip("/") + "/.well-known/openid-configuration"
async with httpx.AsyncClient() as client:
response = await client.get(discovery_url, timeout=5.0)
if response.status_code == 200:
return {"success": True, "message": "OIDC Identity Provider discovered successfully!"}
else:
return {"success": False, "message": f"Discovery failed with status {response.status_code}"}
except Exception as e:
return {"success": False, "message": f"Failed to reach OIDC provider: {str(e)}"}
@router.get("/config/swarm/test/{nonce}", summary="Echo Swarm Nonce")
async def echo_swarm_nonce(nonce: str):
return {"nonce": nonce}
@router.post("/config/swarm/test", summary="Test Swarm Connection")
async def test_swarm_connection(
update: schemas.SwarmConfigUpdate,
admin = Depends(get_current_admin)
):
if not update.external_endpoint:
raise HTTPException(status_code=400, detail="External endpoint is required for testing.")
import httpx
import uuid
try:
nonce = str(uuid.uuid4())
test_url = f"{update.external_endpoint.rstrip('/')}/api/v1/admin/config/swarm/test/{nonce}"
async with httpx.AsyncClient(verify=False) as client:
response = await client.get(test_url, timeout=10.0)
if response.status_code == 200:
data = response.json()
if data.get("nonce") == nonce:
return {"success": True, "message": "Successfully routed back to this hub instance!"}
else:
return {"success": False, "message": "Connected to an endpoint, but the verification signature did not match."}
else:
return {"success": False, "message": f"Endpoint reachable, but returned status {response.status_code}."}
except httpx.ConnectError:
return {"success": False, "message": "Failed to connect: Connection refused. Check if the domain/IP is correct and listening."}
except httpx.TimeoutException:
return {"success": False, "message": "Connection timed out. Check firewall or proxy settings."}
except Exception as e:
return {"success": False, "message": f"Verification failed: {str(e)}"}
@router.put("/config/swarm", summary="Update Swarm Configuration")
async def update_swarm_config(
update: schemas.SwarmConfigUpdate,
admin = Depends(get_current_admin)
):
if update.external_endpoint is not None:
settings.GRPC_EXTERNAL_ENDPOINT = update.external_endpoint
# Derived TLS enabled from endpoint protocol
endpoint = update.external_endpoint.lower()
settings.GRPC_TLS_ENABLED = endpoint.startswith("https://") or ":443" in endpoint
if update.grpc_endpoint is not None:
settings.GRPC_TARGET_ORIGIN = update.grpc_endpoint
settings.save_to_yaml()
return {"message": "Swarm configuration updated successfully"}
@router.get("/config/providers", summary="Get Global Providers Configuration")
async def get_global_providers(admin = Depends(get_current_admin)):
def mask_keys(providers_dict):
import copy
res = copy.deepcopy(providers_dict) if providers_dict else {}
for p_data in res.values():
if isinstance(p_data, dict) and p_data.get("api_key"):
k = str(p_data["api_key"])
p_data["api_key"] = k[:4] + "****" + k[-4:] if len(k) > 8 else "****"
return res
return {
"llm_providers": mask_keys(settings.LLM_PROVIDERS),
"active_llm_provider": settings.ACTIVE_LLM_PROVIDER,
"tts_providers": mask_keys(settings.TTS_PROVIDERS),
"active_tts_provider": settings.TTS_PROVIDER,
"stt_providers": mask_keys(settings.STT_PROVIDERS),
"active_stt_provider": settings.STT_PROVIDER
}
@router.put("/config/providers", summary="Update Global Providers Configuration")
async def update_global_providers(update: schemas.GlobalProvidersUpdate, admin = Depends(get_current_admin)):
def preserve_masked(new_dict, old_dict):
if not new_dict or not old_dict: return
for p_name, p_data in new_dict.items():
if isinstance(p_data, dict) and p_data.get("api_key") and "****" in str(p_data["api_key"]):
if p_name in old_dict and isinstance(old_dict[p_name], dict):
p_data["api_key"] = old_dict[p_name].get("api_key")
if update.llm_providers is not None:
preserve_masked(update.llm_providers, settings.LLM_PROVIDERS)
settings.LLM_PROVIDERS = update.llm_providers
if update.active_llm_provider is not None:
settings.ACTIVE_LLM_PROVIDER = update.active_llm_provider
if update.tts_providers is not None:
preserve_masked(update.tts_providers, settings.TTS_PROVIDERS)
settings.TTS_PROVIDERS = update.tts_providers
if update.active_tts_provider is not None:
settings.TTS_PROVIDER = update.active_tts_provider
if update.stt_providers is not None:
preserve_masked(update.stt_providers, settings.STT_PROVIDERS)
settings.STT_PROVIDERS = update.stt_providers
if update.active_stt_provider is not None:
settings.STT_PROVIDER = update.active_stt_provider
settings.save_to_yaml()
return {"message": "Global providers updated successfully"}
@router.get("/config", summary="Get Admin Configuration")
async def get_admin_config(
admin = Depends(get_current_admin)
):
return {
"app": {
"allow_password_login": settings.ALLOW_PASSWORD_LOGIN
},
"oidc": {
"enabled": settings.OIDC_ENABLED,
"client_id": settings.OIDC_CLIENT_ID,
"client_secret": settings.OIDC_CLIENT_SECRET,
"server_url": settings.OIDC_SERVER_URL,
"redirect_uri": settings.OIDC_REDIRECT_URI,
"allow_oidc_login": settings.ALLOW_OIDC_LOGIN
},
"swarm": {
"external_endpoint": settings.GRPC_EXTERNAL_ENDPOINT,
"grpc_endpoint": getattr(settings, "GRPC_TARGET_ORIGIN", None)
}
}
return router
