13290c1ebb
- auth: verification is_active au login (compte desactive = bloque) - settings: enforcement backend can_edit(settings) + role/section - servers: can_view/can_edit(servers) sur toutes les routes - planning: can_view/can_edit(planning) sur toutes les routes - specifics: can_view/can_edit(specifics) sur toutes les routes - contacts: rattache au module servers (can_view/can_edit) - campaigns: can_view/can_edit(campaigns) sur toutes les routes manquantes - audit/audit_full: can_view/can_edit(audit) sur toutes les routes - qualys: can_view/can_edit(qualys) sur toutes les routes - safe_patching: perm checks + authentification sur SSE stream - quickwin: can_view/can_edit(campaigns|quickwin) sur toutes les routes 97 points d'injection securises, 0 route sans controle Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
319 lines
12 KiB
Python
319 lines
12 KiB
Python
"""Router QuickWin — Campagnes patching rapide avec exclusions par serveur"""
|
|
import json
|
|
from datetime import datetime
|
|
from fastapi import APIRouter, Request, Depends, Query, Form
|
|
from fastapi.responses import HTMLResponse, RedirectResponse, JSONResponse
|
|
from fastapi.templating import Jinja2Templates
|
|
from ..dependencies import get_db, get_current_user, get_user_perms, can_view, can_edit, base_context
|
|
from ..services.quickwin_service import (
|
|
get_server_configs, upsert_server_config, delete_server_config,
|
|
get_eligible_servers, list_runs, get_run, get_run_entries,
|
|
create_run, delete_run, update_entry_field,
|
|
can_start_prod, get_run_stats, inject_yum_history,
|
|
DEFAULT_GENERAL_EXCLUDES,
|
|
)
|
|
from ..config import APP_NAME
|
|
|
|
router = APIRouter()
|
|
templates = Jinja2Templates(directory="app/templates")
|
|
|
|
|
|
@router.get("/quickwin", response_class=HTMLResponse)
|
|
async def quickwin_page(request: Request, db=Depends(get_db)):
|
|
user = get_current_user(request)
|
|
if not user:
|
|
return RedirectResponse(url="/login")
|
|
perms = get_user_perms(db, user)
|
|
if not can_view(perms, "campaigns") and not can_view(perms, "quickwin"):
|
|
return RedirectResponse(url="/dashboard")
|
|
|
|
runs = list_runs(db)
|
|
configs = get_server_configs(db)
|
|
now = datetime.now()
|
|
|
|
ctx = base_context(request, db, user)
|
|
ctx.update({
|
|
"app_name": APP_NAME,
|
|
"runs": runs,
|
|
"configs": configs,
|
|
"config_count": len(configs),
|
|
"current_week": now.isocalendar()[1],
|
|
"current_year": now.isocalendar()[0],
|
|
"can_create": can_edit(perms, "campaigns"),
|
|
"msg": request.query_params.get("msg"),
|
|
})
|
|
return templates.TemplateResponse("quickwin.html", ctx)
|
|
|
|
|
|
# -- Config exclusions par serveur --
|
|
|
|
@router.get("/quickwin/config", response_class=HTMLResponse)
|
|
async def quickwin_config_page(request: Request, db=Depends(get_db),
|
|
page: int = Query(1),
|
|
per_page: int = Query(14),
|
|
search: str = Query(""),
|
|
env: str = Query(""),
|
|
domain: str = Query(""),
|
|
zone: str = Query("")):
|
|
user = get_current_user(request)
|
|
if not user:
|
|
return RedirectResponse(url="/login")
|
|
perms = get_user_perms(db, user)
|
|
if not can_view(perms, "campaigns") and not can_view(perms, "quickwin"):
|
|
return RedirectResponse(url="/dashboard")
|
|
|
|
configs = get_server_configs(db)
|
|
|
|
# Filtres
|
|
filtered = configs
|
|
if search:
|
|
filtered = [s for s in filtered if search.lower() in s.hostname.lower()]
|
|
if env:
|
|
filtered = [s for s in filtered if s.environnement == env]
|
|
if domain:
|
|
filtered = [s for s in filtered if s.domaine == domain]
|
|
if zone:
|
|
filtered = [s for s in filtered if (s.zone or '') == zone]
|
|
|
|
# Pagination
|
|
per_page = max(5, min(per_page, 100))
|
|
total = len(filtered)
|
|
total_pages = max(1, (total + per_page - 1) // per_page)
|
|
page = max(1, min(page, total_pages))
|
|
start = (page - 1) * per_page
|
|
page_servers = filtered[start:start + per_page]
|
|
|
|
ctx = base_context(request, db, user)
|
|
ctx.update({
|
|
"app_name": APP_NAME,
|
|
"all_servers": page_servers,
|
|
"all_configs": configs,
|
|
"default_excludes": DEFAULT_GENERAL_EXCLUDES,
|
|
"total_count": total,
|
|
"page": page,
|
|
"per_page": per_page,
|
|
"total_pages": total_pages,
|
|
"filters": {"search": search, "env": env, "domain": domain, "zone": zone},
|
|
"msg": request.query_params.get("msg"),
|
|
})
|
|
return templates.TemplateResponse("quickwin_config.html", ctx)
|
|
|
|
|
|
@router.post("/quickwin/config/save")
|
|
async def quickwin_config_save(request: Request, db=Depends(get_db),
|
|
server_id: int = Form(0),
|
|
general_excludes: str = Form(""),
|
|
specific_excludes: str = Form(""),
|
|
notes: str = Form("")):
|
|
user = get_current_user(request)
|
|
if not user:
|
|
return RedirectResponse(url="/login")
|
|
perms = get_user_perms(db, user)
|
|
if not can_edit(perms, "campaigns") and not can_edit(perms, "quickwin"):
|
|
return RedirectResponse(url="/quickwin/config")
|
|
if server_id:
|
|
upsert_server_config(db, server_id, general_excludes.strip(),
|
|
specific_excludes.strip(), notes.strip())
|
|
return RedirectResponse(url="/quickwin/config?msg=saved", status_code=303)
|
|
|
|
|
|
@router.post("/quickwin/config/delete")
|
|
async def quickwin_config_delete(request: Request, db=Depends(get_db),
|
|
config_id: int = Form(0)):
|
|
user = get_current_user(request)
|
|
if not user:
|
|
return RedirectResponse(url="/login")
|
|
perms = get_user_perms(db, user)
|
|
if not can_edit(perms, "campaigns") and not can_edit(perms, "quickwin"):
|
|
return RedirectResponse(url="/quickwin/config")
|
|
if config_id:
|
|
delete_server_config(db, config_id)
|
|
return RedirectResponse(url="/quickwin/config?msg=deleted", status_code=303)
|
|
|
|
|
|
@router.post("/quickwin/config/bulk-add")
|
|
async def quickwin_config_bulk_add(request: Request, db=Depends(get_db),
|
|
server_ids: str = Form(""),
|
|
general_excludes: str = Form("")):
|
|
"""Ajouter plusieurs serveurs d'un coup avec les memes exclusions generales"""
|
|
user = get_current_user(request)
|
|
if not user:
|
|
return RedirectResponse(url="/login")
|
|
perms = get_user_perms(db, user)
|
|
if not can_edit(perms, "campaigns") and not can_edit(perms, "quickwin"):
|
|
return RedirectResponse(url="/quickwin/config")
|
|
ids = [int(x) for x in server_ids.split(",") if x.strip().isdigit()]
|
|
for sid in ids:
|
|
upsert_server_config(db, sid, general_excludes.strip(), "", "")
|
|
return RedirectResponse(url=f"/quickwin/config?msg=added_{len(ids)}", status_code=303)
|
|
|
|
|
|
# -- Runs QuickWin --
|
|
|
|
@router.post("/quickwin/create")
|
|
async def quickwin_create(request: Request, db=Depends(get_db),
|
|
label: str = Form(""),
|
|
week_number: int = Form(0),
|
|
year: int = Form(0),
|
|
server_ids: str = Form(""),
|
|
notes: str = Form("")):
|
|
user = get_current_user(request)
|
|
if not user:
|
|
return RedirectResponse(url="/login")
|
|
perms = get_user_perms(db, user)
|
|
if not can_edit(perms, "campaigns"):
|
|
return RedirectResponse(url="/quickwin")
|
|
|
|
if not label:
|
|
label = f"Quick Win S{week_number:02d} {year}"
|
|
|
|
ids = [int(x) for x in server_ids.split(",") if x.strip().isdigit()]
|
|
if not ids:
|
|
# Prendre tous les serveurs configures, sinon tous les eligibles
|
|
configs = get_server_configs(db)
|
|
ids = [c.server_id for c in configs]
|
|
if not ids:
|
|
eligible = get_eligible_servers(db)
|
|
ids = [s.id for s in eligible]
|
|
|
|
if not ids:
|
|
return RedirectResponse(url="/quickwin?msg=no_servers", status_code=303)
|
|
|
|
try:
|
|
run_id = create_run(db, year, week_number, label, user.get("uid"), ids, notes)
|
|
return RedirectResponse(url=f"/quickwin/{run_id}", status_code=303)
|
|
except Exception as e:
|
|
db.rollback()
|
|
return RedirectResponse(url=f"/quickwin?msg=error", status_code=303)
|
|
|
|
|
|
@router.get("/quickwin/{run_id}", response_class=HTMLResponse)
|
|
async def quickwin_detail(request: Request, run_id: int, db=Depends(get_db),
|
|
search: str = Query(""),
|
|
status: str = Query(""),
|
|
domain: str = Query(""),
|
|
hp_page: int = Query(1),
|
|
p_page: int = Query(1),
|
|
per_page: int = Query(14)):
|
|
user = get_current_user(request)
|
|
if not user:
|
|
return RedirectResponse(url="/login")
|
|
perms = get_user_perms(db, user)
|
|
if not can_view(perms, "campaigns") and not can_view(perms, "quickwin"):
|
|
return RedirectResponse(url="/dashboard")
|
|
|
|
run = get_run(db, run_id)
|
|
if not run:
|
|
return RedirectResponse(url="/quickwin")
|
|
|
|
entries = get_run_entries(db, run_id)
|
|
stats = get_run_stats(db, run_id)
|
|
prod_ok = can_start_prod(db, run_id)
|
|
|
|
hprod_all = [e for e in entries if e.branch == "hprod"]
|
|
prod_all = [e for e in entries if e.branch == "prod"]
|
|
|
|
# Filtres
|
|
def apply_filters(lst):
|
|
filtered = lst
|
|
if search:
|
|
filtered = [e for e in filtered if search.lower() in e.hostname.lower()]
|
|
if status:
|
|
filtered = [e for e in filtered if e.status == status]
|
|
if domain:
|
|
filtered = [e for e in filtered if e.domaine == domain]
|
|
return filtered
|
|
|
|
hprod = apply_filters(hprod_all)
|
|
prod = apply_filters(prod_all)
|
|
|
|
# Pagination
|
|
per_page = max(5, min(per_page, 100))
|
|
|
|
hp_total = len(hprod)
|
|
hp_total_pages = max(1, (hp_total + per_page - 1) // per_page)
|
|
hp_page = max(1, min(hp_page, hp_total_pages))
|
|
hp_start = (hp_page - 1) * per_page
|
|
hprod_page = hprod[hp_start:hp_start + per_page]
|
|
|
|
p_total = len(prod)
|
|
p_total_pages = max(1, (p_total + per_page - 1) // per_page)
|
|
p_page = max(1, min(p_page, p_total_pages))
|
|
p_start = (p_page - 1) * per_page
|
|
prod_page = prod[p_start:p_start + per_page]
|
|
|
|
ctx = base_context(request, db, user)
|
|
ctx.update({
|
|
"app_name": APP_NAME,
|
|
"run": run, "entries": entries, "stats": stats,
|
|
"hprod": hprod_page, "prod": prod_page,
|
|
"hprod_total": hp_total, "prod_total": p_total,
|
|
"hp_page": hp_page, "hp_total_pages": hp_total_pages,
|
|
"p_page": p_page, "p_total_pages": p_total_pages,
|
|
"per_page": per_page,
|
|
"prod_ok": prod_ok,
|
|
"filters": {"search": search, "status": status, "domain": domain},
|
|
"msg": request.query_params.get("msg"),
|
|
})
|
|
return templates.TemplateResponse("quickwin_detail.html", ctx)
|
|
|
|
|
|
@router.post("/quickwin/{run_id}/delete")
|
|
async def quickwin_delete(request: Request, run_id: int, db=Depends(get_db)):
|
|
user = get_current_user(request)
|
|
if not user:
|
|
return RedirectResponse(url="/login")
|
|
perms = get_user_perms(db, user)
|
|
if not can_edit(perms, "campaigns"):
|
|
return RedirectResponse(url="/quickwin")
|
|
delete_run(db, run_id)
|
|
return RedirectResponse(url="/quickwin?msg=deleted", status_code=303)
|
|
|
|
|
|
# -- API JSON --
|
|
|
|
@router.post("/api/quickwin/entry/update")
|
|
async def quickwin_entry_update(request: Request, db=Depends(get_db)):
|
|
user = get_current_user(request)
|
|
if not user:
|
|
return JSONResponse({"error": "unauthorized"}, 401)
|
|
perms = get_user_perms(db, user)
|
|
if not can_edit(perms, "campaigns") and not can_edit(perms, "quickwin"):
|
|
return JSONResponse({"error": "forbidden"}, 403)
|
|
body = await request.json()
|
|
entry_id = body.get("id")
|
|
field = body.get("field")
|
|
value = body.get("value")
|
|
if not entry_id or not field:
|
|
return JSONResponse({"error": "id and field required"}, 400)
|
|
ok = update_entry_field(db, entry_id, field, value)
|
|
return JSONResponse({"ok": ok})
|
|
|
|
|
|
@router.post("/api/quickwin/inject-yum-history")
|
|
async def quickwin_inject_yum(request: Request, db=Depends(get_db)):
|
|
user = get_current_user(request)
|
|
if not user:
|
|
return JSONResponse({"error": "unauthorized"}, 401)
|
|
perms = get_user_perms(db, user)
|
|
if not can_edit(perms, "campaigns"):
|
|
return JSONResponse({"error": "forbidden"}, 403)
|
|
body = await request.json()
|
|
if not isinstance(body, list):
|
|
return JSONResponse({"error": "expected list"}, 400)
|
|
updated, inserted = inject_yum_history(db, body)
|
|
return JSONResponse({"ok": True, "updated": updated, "inserted": inserted})
|
|
|
|
|
|
@router.get("/api/quickwin/prod-check/{run_id}")
|
|
async def quickwin_prod_check(request: Request, run_id: int, db=Depends(get_db)):
|
|
"""Verifie si le prod peut demarrer (tous hprod termines)"""
|
|
user = get_current_user(request)
|
|
if not user:
|
|
return JSONResponse({"error": "unauthorized"}, 401)
|
|
perms = get_user_perms(db, user)
|
|
if not can_view(perms, "campaigns") and not can_view(perms, "quickwin"):
|
|
return JSONResponse({"error": "forbidden"}, 403)
|
|
ok = can_start_prod(db, run_id)
|
|
return JSONResponse({"can_start_prod": ok})
|