"""Router settings — configuration modules externes + connexions""" from fastapi import APIRouter, Request, Depends, Form from fastapi.responses import HTMLResponse, RedirectResponse from fastapi.templating import Jinja2Templates from sqlalchemy import text from ..dependencies import get_db, get_current_user from ..services.secrets_service import get_secret, set_secret, list_secrets, init_secrets_from_config from ..config import APP_NAME router = APIRouter() templates = Jinja2Templates(directory="app/templates") SECTIONS = { "qualys": [ ("qualys_url", "URL API", False), ("qualys_user", "Utilisateur", False), ("qualys_pass", "Mot de passe", True), ("qualys_proxy", "Proxy (ex: http://proxy:3128)", False), ], "itop": [ ("itop_url", "URL API", False), ("itop_user", "Utilisateur", False), ("itop_pass", "Mot de passe", True), ], "ssh_key": [ ("ssh_key_default_user", "User SSH par defaut", False), ("ssh_key_default_port", "Port par defaut", False), ("ssh_key_private_key", "Cle privee (PEM)", True), ], "ssh_pwd": [ ("ssh_pwd_default_user", "User par defaut", False), ("ssh_pwd_default_pass", "Password par defaut", True), ], "ssh_psmp": [ ("psmp_host", "Adresse PSMP", False), ("psmp_port", "Port PSMP", False), ("psmp_user_format", "Format user", False), ("psmp_cyberark_user", "Compte CyberArk", False), ("psmp_target_user", "Utilisateur cible", False), ("psmp_default_safe", "Safe par defaut", False), ], "rdp_psm": [ ("rdp_psm_pvwa_url", "URL PVWA", False), ("rdp_psm_pvwa_user", "User PVWA", False), ("rdp_psm_pvwa_pass", "Password PVWA", True), ("rdp_psm_component", "Connection Component", False), ], "rdp_pwd": [ ("rdp_pwd_default_user", "User par defaut", False), ("rdp_pwd_default_pass", "Password par defaut", True), ("rdp_pwd_default_port", "Port RDP", False), ], "vsphere": [ ("vsphere_user", "Utilisateur vCenter", False), ("vsphere_pass", "Mot de passe vCenter", True), ], "splunk": [ ("splunk_hec_url", "URL HEC", False), ("splunk_hec_token", "Token HEC", True), ("splunk_index", "Index", False), ("splunk_sourcetype", "Sourcetype", False), ("splunk_verify_ssl", "Verifier SSL (true/false)", False), ], "teams": [ ("teams_webhook_url", "Webhook URL (canal)", False), ("teams_sp_site_url", "SharePoint Site URL", False), ("teams_sp_library", "SharePoint Library", False), ("teams_sp_folder", "SharePoint Folder", False), ("teams_sp_client_id", "App Client ID", False), ("teams_sp_client_secret", "App Client Secret", True), ("teams_sp_tenant_id", "Tenant ID", False), ], } def _load_section_values(db): vals = {} for section, fields in SECTIONS.items(): for key, label, is_secret in fields: v = get_secret(db, key) if is_secret and v: vals[key] = "********" else: vals[key] = v or "" return vals # Regles d'acces par section: visible = qui peut voir, editable = qui peut modifier SECTION_ACCESS = { "qualys": {"visible": ["admin"], "editable": ["admin"]}, "ssh_key": {"visible": ["admin"], "editable": ["admin"]}, "ssh_pwd": {"visible": ["admin", "operator"], "editable": ["admin", "operator"]}, "ssh_psmp": {"visible": ["admin", "operator"], "editable": ["admin", "operator"]}, "rdp_psm": {"visible": ["admin"], "editable": ["admin"]}, "rdp_pwd": {"visible": [], "editable": []}, "vsphere": {"visible": ["admin", "operator", "coordinator"], "editable": ["admin", "operator"]}, "splunk": {"visible": ["admin", "coordinator"], "editable": ["admin", "coordinator"]}, "teams": {"visible": ["admin", "coordinator"], "editable": ["admin", "coordinator"]}, "itop": {"visible": ["admin"], "editable": ["admin"]}, } def _build_context(db, user, saved=None): init_secrets_from_config(db) role = user.get("role", "viewer") q_tags = db.execute(text("SELECT COUNT(*) FROM qualys_tags")).scalar() q_assets = db.execute(text("SELECT COUNT(*) FROM qualys_assets")).scalar() q_linked = db.execute(text("SELECT COUNT(*) FROM servers WHERE qualys_asset_id IS NOT NULL")).scalar() vcenters = db.execute(text("SELECT * FROM vcenters ORDER BY name")).fetchall() # Filtrer les sections visibles selon le role visible = {s: s in SECTION_ACCESS and role in SECTION_ACCESS[s]["visible"] for s in SECTIONS} editable = {s: s in SECTION_ACCESS and role in SECTION_ACCESS[s]["editable"] for s in SECTIONS} return { "user": user, "app_name": APP_NAME, "role": role, "sections": SECTIONS, "vals": _load_section_values(db), "q_tags": q_tags, "q_assets": q_assets, "q_linked": q_linked, "vcenters": vcenters, "saved": saved, "visible": visible, "editable": editable, } @router.get("/settings", response_class=HTMLResponse) async def settings_page(request: Request, db=Depends(get_db)): user = get_current_user(request) if not user: return RedirectResponse(url="/login") ctx = _build_context(db, user) ctx["request"] = request return templates.TemplateResponse("settings.html", ctx) @router.post("/settings/{section}", response_class=HTMLResponse) async def settings_save(request: Request, section: str, db=Depends(get_db)): user = get_current_user(request) if not user: return RedirectResponse(url="/login") if section not in SECTIONS: return HTMLResponse("

Section inconnue

", status_code=400) form = await request.form() for key, label, is_secret in SECTIONS[section]: val = form.get(key, "") if is_secret and val == "********": continue if val: set_secret(db, key, val, label) ctx = _build_context(db, user, saved=section) ctx["request"] = request return templates.TemplateResponse("settings.html", ctx) # --- vCenter CRUD --- @router.post("/settings/vcenter/add", response_class=HTMLResponse) async def vcenter_add(request: Request, db=Depends(get_db), vc_name: str = Form(...), vc_endpoint: str = Form(...), vc_datacenter: str = Form(""), vc_description: str = Form(""), vc_responsable: str = Form("")): user = get_current_user(request) if not user: return RedirectResponse(url="/login") db.execute(text( "INSERT INTO vcenters (name, endpoint, datacenter, description, responsable) VALUES (:n, :e, :dc, :desc, :resp)" ), {"n": vc_name, "e": vc_endpoint, "dc": vc_datacenter or None, "desc": vc_description or None, "resp": vc_responsable or None}) db.commit() ctx = _build_context(db, user, saved="vsphere") ctx["request"] = request return templates.TemplateResponse("settings.html", ctx) @router.post("/settings/vcenter/{vc_id}/delete", response_class=HTMLResponse) async def vcenter_delete(request: Request, vc_id: int, db=Depends(get_db)): user = get_current_user(request) if not user: return RedirectResponse(url="/login") db.execute(text("UPDATE vcenters SET is_active = false WHERE id = :id"), {"id": vc_id}) db.commit() ctx = _build_context(db, user, saved="vsphere") ctx["request"] = request return templates.TemplateResponse("settings.html", ctx) # --- Secret individuel --- @router.post("/settings/secret/update", response_class=HTMLResponse) async def secret_update(request: Request, db=Depends(get_db), secret_key: str = Form(...), secret_value: str = Form(...)): user = get_current_user(request) if not user: return RedirectResponse(url="/login") if secret_value and secret_value != "********": # Recuperer la description existante existing = db.execute(text("SELECT description FROM app_secrets WHERE key = :k"), {"k": secret_key}).fetchone() desc = existing.description if existing else secret_key set_secret(db, secret_key, secret_value, desc) ctx = _build_context(db, user, saved="secret") ctx["request"] = request return templates.TemplateResponse("settings.html", ctx)