8e62b1fb11
Qualys: - Recherche API temps réel + cache 24h base locale - Tags: liste DYN/STAT, mapping V3 (DOM-*, TYP-*, APP-*), nb assets cliquable - CRUD tags: créer STAT, supprimer, resync API - Détail asset: infos + décodage nomenclature V3 + tags assignés - Ajout/retrait tag unitaire avec autocomplete filtrable - Bulk add/remove tag en masse avec dropdown filtrable - Tags retirer: charge dynamiquement les STAT assignés aux assets sélectionnés - Resync assets sélectionnés + retour même recherche Contacts: - 50 contacts importés avec 93 scopes (domaine/app/serveur/zone par env) - 13 rôles (responsable_domaine, ra_prod, ra_recette, referent_technique...) - Recherche par nom/email/serveur (affiche contacts liés) - CRUD complet: éditer, scopes, activer/désactiver, supprimer - Serveurs liés calculés dynamiquement depuis les scopes Audit: - Restructuré: Audit général + sous-menu Spécifique - Dernier audit global affiché avec date - Lancer audit général avec exclusions (domaines/zones) et parallélisme - KPIs Qualys KO et S1 KO cliquables - Export CSV Serveurs: - Actions groupées bulk (domaine, env, tier, état, owner, licence) - Dashboard: KPI EOL ajouté - Filtre état: EOL + en décommissionnement ajoutés - 138 serveurs EOL importés depuis Qualys (owner=na, hors périmètre) Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
248 lines
10 KiB
Python
248 lines
10 KiB
Python
"""Router contacts — gestion des responsables applicatifs et scopes"""
|
|
from fastapi import APIRouter, Request, Depends, Query, Form
|
|
from fastapi.responses import HTMLResponse, RedirectResponse
|
|
from fastapi.templating import Jinja2Templates
|
|
from sqlalchemy import text
|
|
from ..dependencies import get_db, get_current_user, get_user_perms, can_view, can_edit, base_context
|
|
from ..config import APP_NAME
|
|
|
|
router = APIRouter()
|
|
templates = Jinja2Templates(directory="app/templates")
|
|
|
|
ROLES = [
|
|
("responsable_domaine", "Responsable domaine"),
|
|
("responsable_prod", "Responsable production"),
|
|
("responsable_applicatif", "Responsable applicatif"),
|
|
("referent_technique", "Référent technique"),
|
|
("ra_prod", "RA Production"),
|
|
("ra_recette", "RA Recette"),
|
|
("ra_preprod", "RA Pré-prod"),
|
|
("ra_test", "RA Test"),
|
|
("ra_dev", "RA Développement"),
|
|
("chef_projet", "Chef de projet"),
|
|
("contact_technique", "Contact technique"),
|
|
("editeur", "Éditeur"),
|
|
("autre", "Autre"),
|
|
]
|
|
|
|
SCOPE_TYPES = [
|
|
("domain", "Domaine"),
|
|
("application", "Application"),
|
|
("app_group", "Groupe applicatif"),
|
|
("server", "Serveur"),
|
|
("zone", "Zone"),
|
|
]
|
|
|
|
ENV_SCOPES = ["all", "prod", "recette", "preprod", "test", "dev"]
|
|
|
|
|
|
@router.get("/contacts", response_class=HTMLResponse)
|
|
async def contacts_page(request: Request, db=Depends(get_db),
|
|
search: str = Query(None), role: str = Query(None),
|
|
server: str = Query(None)):
|
|
user = get_current_user(request)
|
|
if not user:
|
|
return RedirectResponse(url="/login")
|
|
|
|
where = ["1=1"]
|
|
params = {}
|
|
if search:
|
|
where.append("(c.name ILIKE :q OR c.email ILIKE :q)")
|
|
params["q"] = f"%{search}%"
|
|
if role:
|
|
where.append("c.role = :role")
|
|
params["role"] = role
|
|
|
|
# Recherche par serveur : trouver les contacts liés à ce serveur
|
|
server_info = None
|
|
if server:
|
|
server_info = db.execute(text("""
|
|
SELECT s.id, s.hostname, d.code as domain_code, d.name as domain_name,
|
|
e.name as env_name, s.app_group, z.name as zone_name,
|
|
ss.app_type
|
|
FROM servers s
|
|
LEFT JOIN domain_environments de ON s.domain_env_id = de.id
|
|
LEFT JOIN domains d ON de.domain_id = d.id
|
|
LEFT JOIN environments e ON de.environment_id = e.id
|
|
LEFT JOIN zones z ON s.zone_id = z.id
|
|
LEFT JOIN server_specifics ss ON ss.server_id = s.id
|
|
WHERE LOWER(s.hostname) = LOWER(:h)
|
|
"""), {"h": server.strip()}).fetchone()
|
|
|
|
if server_info:
|
|
where.append("""c.id IN (
|
|
SELECT cs.contact_id FROM contact_scopes cs WHERE
|
|
(cs.scope_type = 'server' AND LOWER(cs.scope_value) = LOWER(:srv_hn))
|
|
OR (cs.scope_type = 'domain' AND cs.scope_value = :srv_dom)
|
|
OR (cs.scope_type = 'app_group' AND cs.scope_value = :srv_ag)
|
|
OR (cs.scope_type = 'application' AND UPPER(cs.scope_value) = UPPER(:srv_app))
|
|
OR (cs.scope_type = 'zone' AND cs.scope_value = :srv_zone)
|
|
)""")
|
|
params["srv_hn"] = server.strip()
|
|
params["srv_dom"] = server_info.domain_code or ""
|
|
params["srv_ag"] = server_info.app_group or ""
|
|
params["srv_app"] = server_info.app_type or ""
|
|
params["srv_zone"] = server_info.zone_name or ""
|
|
|
|
wc = " AND ".join(where)
|
|
|
|
contacts = db.execute(text(f"""
|
|
SELECT c.*,
|
|
(SELECT string_agg(cs.scope_type || ':' || cs.scope_value ||
|
|
CASE WHEN cs.env_scope != 'all' THEN '(' || cs.env_scope || ')' ELSE '' END, ', '
|
|
ORDER BY cs.scope_type, cs.scope_value)
|
|
FROM contact_scopes cs WHERE cs.contact_id = c.id) as scopes_summary
|
|
FROM contacts c WHERE {wc} ORDER BY c.name
|
|
"""), params).fetchall()
|
|
|
|
roles_in_db = db.execute(text(
|
|
"SELECT DISTINCT role FROM contacts ORDER BY role"
|
|
)).fetchall()
|
|
|
|
domains = db.execute(text("SELECT code, name FROM domains ORDER BY display_order")).fetchall()
|
|
app_types = db.execute(text(
|
|
"SELECT DISTINCT app_type FROM server_specifics WHERE app_type IS NOT NULL ORDER BY app_type"
|
|
)).fetchall()
|
|
|
|
ctx = base_context(request, db, user)
|
|
ctx.update({
|
|
"app_name": APP_NAME, "contacts": contacts,
|
|
"roles": ROLES, "roles_in_db": [r.role for r in roles_in_db],
|
|
"scope_types": SCOPE_TYPES, "env_scopes": ENV_SCOPES,
|
|
"domains": domains, "app_types": [r.app_type for r in app_types],
|
|
"search": search, "role_filter": role, "server": server, "server_info": server_info,
|
|
"msg": request.query_params.get("msg"),
|
|
})
|
|
return templates.TemplateResponse("contacts.html", ctx)
|
|
|
|
|
|
@router.get("/contacts/{contact_id}", response_class=HTMLResponse)
|
|
async def contact_detail(request: Request, contact_id: int, db=Depends(get_db)):
|
|
user = get_current_user(request)
|
|
if not user:
|
|
return HTMLResponse("<p>Non autorisé</p>")
|
|
|
|
contact = db.execute(text("SELECT * FROM contacts WHERE id = :id"),
|
|
{"id": contact_id}).fetchone()
|
|
if not contact:
|
|
return HTMLResponse("<p>Non trouvé</p>")
|
|
|
|
scopes = db.execute(text("""
|
|
SELECT cs.* FROM contact_scopes cs WHERE cs.contact_id = :cid ORDER BY cs.scope_type, cs.scope_value
|
|
"""), {"cid": contact_id}).fetchall()
|
|
|
|
# Serveurs liés via scopes
|
|
servers = db.execute(text("""
|
|
SELECT DISTINCT s.id, s.hostname, d.name as domaine, e.name as environnement
|
|
FROM servers s
|
|
LEFT JOIN domain_environments de ON s.domain_env_id = de.id
|
|
LEFT JOIN domains d ON de.domain_id = d.id
|
|
LEFT JOIN environments e ON de.environment_id = e.id
|
|
LEFT JOIN server_specifics ss ON ss.server_id = s.id
|
|
LEFT JOIN zones z ON s.zone_id = z.id
|
|
WHERE EXISTS (
|
|
SELECT 1 FROM contact_scopes cs WHERE cs.contact_id = :cid AND (
|
|
(cs.scope_type = 'domain' AND d.code = cs.scope_value)
|
|
OR (cs.scope_type = 'application' AND UPPER(ss.app_type) = UPPER(cs.scope_value))
|
|
OR (cs.scope_type = 'server' AND LOWER(s.hostname) = LOWER(cs.scope_value))
|
|
OR (cs.scope_type = 'app_group' AND s.app_group = cs.scope_value)
|
|
OR (cs.scope_type = 'zone' AND z.name = cs.scope_value)
|
|
)
|
|
)
|
|
ORDER BY d.name, s.hostname LIMIT 100
|
|
"""), {"cid": contact_id}).fetchall()
|
|
|
|
domains = db.execute(text("SELECT code, name FROM domains ORDER BY display_order")).fetchall()
|
|
app_types = db.execute(text(
|
|
"SELECT DISTINCT app_type FROM server_specifics WHERE app_type IS NOT NULL ORDER BY app_type"
|
|
)).fetchall()
|
|
|
|
return templates.TemplateResponse("partials/contact_detail.html", {
|
|
"request": request, "c": contact, "scopes": scopes, "servers": servers,
|
|
"roles": ROLES, "scope_types": SCOPE_TYPES, "env_scopes": ENV_SCOPES,
|
|
"domains": domains, "app_types": [r.app_type for r in app_types],
|
|
})
|
|
|
|
|
|
@router.post("/contacts/add")
|
|
async def contact_add(request: Request, db=Depends(get_db),
|
|
name: str = Form(...), email: str = Form(...), contact_role: str = Form("autre")):
|
|
user = get_current_user(request)
|
|
if not user:
|
|
return RedirectResponse(url="/login")
|
|
try:
|
|
db.execute(text("""
|
|
INSERT INTO contacts (name, email, role, is_active)
|
|
VALUES (:n, :e, :r, true)
|
|
"""), {"n": name.strip(), "e": email.strip().lower(), "r": contact_role})
|
|
db.commit()
|
|
return RedirectResponse(url="/contacts?msg=added", status_code=303)
|
|
except Exception:
|
|
db.rollback()
|
|
return RedirectResponse(url="/contacts?msg=exists", status_code=303)
|
|
|
|
|
|
@router.post("/contacts/{contact_id}/edit")
|
|
async def contact_edit(request: Request, contact_id: int, db=Depends(get_db),
|
|
name: str = Form(""), email: str = Form(""), contact_role: str = Form("")):
|
|
user = get_current_user(request)
|
|
if not user:
|
|
return RedirectResponse(url="/login")
|
|
updates = []; params = {"id": contact_id}
|
|
if name: updates.append("name = :n"); params["n"] = name
|
|
if email: updates.append("email = :e"); params["e"] = email.lower()
|
|
if contact_role: updates.append("role = :r"); params["r"] = contact_role
|
|
if updates:
|
|
db.execute(text(f"UPDATE contacts SET {', '.join(updates)} WHERE id = :id"), params)
|
|
db.commit()
|
|
return RedirectResponse(url="/contacts?msg=edited", status_code=303)
|
|
|
|
|
|
@router.post("/contacts/{contact_id}/toggle")
|
|
async def contact_toggle(request: Request, contact_id: int, db=Depends(get_db)):
|
|
user = get_current_user(request)
|
|
if not user:
|
|
return RedirectResponse(url="/login")
|
|
db.execute(text("UPDATE contacts SET is_active = NOT is_active WHERE id = :id"), {"id": contact_id})
|
|
db.commit()
|
|
return RedirectResponse(url="/contacts?msg=toggled", status_code=303)
|
|
|
|
|
|
@router.post("/contacts/{contact_id}/scope/add")
|
|
async def scope_add(request: Request, contact_id: int, db=Depends(get_db),
|
|
scope_type: str = Form(...), scope_value: str = Form(...),
|
|
env_scope: str = Form("all")):
|
|
user = get_current_user(request)
|
|
if not user:
|
|
return RedirectResponse(url="/login")
|
|
try:
|
|
db.execute(text("""
|
|
INSERT INTO contact_scopes (contact_id, scope_type, scope_value, env_scope)
|
|
VALUES (:cid, :st, :sv, :es)
|
|
"""), {"cid": contact_id, "st": scope_type, "sv": scope_value.strip(), "es": env_scope})
|
|
db.commit()
|
|
except Exception:
|
|
db.rollback()
|
|
return RedirectResponse(url=f"/contacts?msg=scope_added", status_code=303)
|
|
|
|
|
|
@router.post("/contacts/scope/{scope_id}/delete")
|
|
async def scope_delete(request: Request, scope_id: int, db=Depends(get_db)):
|
|
user = get_current_user(request)
|
|
if not user:
|
|
return RedirectResponse(url="/login")
|
|
db.execute(text("DELETE FROM contact_scopes WHERE id = :id"), {"id": scope_id})
|
|
db.commit()
|
|
return RedirectResponse(url="/contacts?msg=scope_deleted", status_code=303)
|
|
|
|
|
|
@router.post("/contacts/{contact_id}/delete")
|
|
async def contact_delete(request: Request, contact_id: int, db=Depends(get_db)):
|
|
user = get_current_user(request)
|
|
if not user:
|
|
return RedirectResponse(url="/login")
|
|
db.execute(text("DELETE FROM contact_scopes WHERE contact_id = :cid"), {"cid": contact_id})
|
|
db.execute(text("DELETE FROM contacts WHERE id = :cid"), {"cid": contact_id})
|
|
db.commit()
|
|
return RedirectResponse(url="/contacts?msg=deleted", status_code=303)
|