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