547 lines
24 KiB
Python
547 lines
24 KiB
Python
import logging
|
|
logger = logging.getLogger(__name__)
|
|
"""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, get_user_perms, can_view, can_edit
|
|
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", False),
|
|
("qualys_bypass_proxy", "Bypass proxy (true/false)", 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),
|
|
],
|
|
"security": [
|
|
("security_session_timeout", "Timeout session (minutes)", False),
|
|
("security_max_login_attempts", "Max tentatives login", False),
|
|
],
|
|
"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),
|
|
],
|
|
"ldap": [
|
|
("ldap_enabled", "Activer LDAP/AD (true/false)", False),
|
|
("ldap_server", "Serveur (ex: ldaps://ad.sanef.com:636)", False),
|
|
("ldap_base_dn", "Base DN (ex: DC=sanef,DC=com)", False),
|
|
("ldap_bind_dn", "Compte de bind (DN complet)", False),
|
|
("ldap_bind_pwd", "Mot de passe compte de bind", True),
|
|
("ldap_user_filter", "Filtre user (ex: (sAMAccountName={username}))", False),
|
|
("ldap_email_attr", "Attribut email (ex: mail)", False),
|
|
("ldap_name_attr", "Attribut nom affiché (ex: displayName)", False),
|
|
("ldap_tls", "TLS (true/false)", False),
|
|
("ldap_required_group", "Groupe AD autorise (DN complet, vide = tous)", False),
|
|
("ldap_default_role", "Role par defaut auto-provision (admin/operator/viewer)", False),
|
|
],
|
|
"itop_contacts": [
|
|
("itop_contact_teams", "Teams iTop à synchroniser (séparées par ,)", 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"]},
|
|
"itop_contacts": {"visible": ["admin"], "editable": ["admin"]},
|
|
"ldap": {"visible": ["admin"], "editable": ["admin"]},
|
|
"security": {"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()
|
|
allowed_nets = db.execute(text("SELECT * FROM allowed_networks ORDER BY cidr")).fetchall()
|
|
teams_channels = db.execute(text("""
|
|
SELECT id, name, webhook_url, description, is_active, is_default, created_at
|
|
FROM teams_channels
|
|
ORDER BY is_default DESC, name
|
|
""")).fetchall()
|
|
server_clusters = db.execute(text("""
|
|
SELECT c.id, c.name, c.description, c.reboot_strategy, c.is_active, c.created_at,
|
|
COUNT(s.id) AS server_count
|
|
FROM server_clusters c
|
|
LEFT JOIN servers s ON s.cluster_id = c.id
|
|
GROUP BY c.id, c.name, c.description, c.reboot_strategy, c.is_active, c.created_at
|
|
ORDER BY c.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),
|
|
"allowed_nets": allowed_nets,
|
|
"q_tags": q_tags, "q_assets": q_assets, "q_linked": q_linked,
|
|
"vcenters": vcenters, "saved": saved,
|
|
"teams_channels": teams_channels,
|
|
"server_clusters": server_clusters,
|
|
"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")
|
|
perms = get_user_perms(db, user)
|
|
if not can_view(perms, "settings"):
|
|
return RedirectResponse(url="/dashboard")
|
|
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("<p>Section inconnue</p>", status_code=400)
|
|
perms = get_user_perms(db, user)
|
|
if not can_edit(perms, "settings"):
|
|
return RedirectResponse(url="/settings")
|
|
role = user.get("role", "viewer")
|
|
if section in SECTION_ACCESS and role not in SECTION_ACCESS[section]["editable"]:
|
|
return RedirectResponse(url="/settings")
|
|
|
|
form = await request.form()
|
|
for key, label, is_secret in SECTIONS[section]:
|
|
val = form.get(key, "")
|
|
if is_secret and val == "********":
|
|
continue
|
|
# Checkbox: si absent du form = "false"
|
|
if key.endswith("_bypass_proxy") or key.endswith("_verify_ssl"):
|
|
val = "true" if val else "false"
|
|
set_secret(db, key, val, label)
|
|
elif val:
|
|
set_secret(db, key, val, label)
|
|
|
|
ctx = _build_context(db, user, saved=section)
|
|
ctx["request"] = request
|
|
return templates.TemplateResponse("settings.html", ctx)
|
|
|
|
|
|
@router.post("/settings/ldap/test")
|
|
async def settings_ldap_test(request: Request, db=Depends(get_db)):
|
|
"""Teste la connexion LDAP avec le compte de bind."""
|
|
from fastapi.responses import JSONResponse
|
|
user = get_current_user(request)
|
|
if not user:
|
|
return JSONResponse({"ok": False, "msg": "Non authentifié"}, status_code=401)
|
|
perms = get_user_perms(db, user)
|
|
if not can_edit(perms, "settings"):
|
|
return JSONResponse({"ok": False, "msg": "Permission refusée"}, status_code=403)
|
|
from ..services.ldap_service import test_connection
|
|
return JSONResponse(test_connection(db))
|
|
|
|
|
|
# --- 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")
|
|
perms = get_user_perms(db, user)
|
|
if not can_edit(perms, "settings"):
|
|
return RedirectResponse(url="/settings")
|
|
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")
|
|
perms = get_user_perms(db, user)
|
|
if not can_edit(perms, "settings"):
|
|
return RedirectResponse(url="/settings")
|
|
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)
|
|
|
|
|
|
# --- Teams channels CRUD ---
|
|
|
|
@router.post("/settings/teams-channel/add", response_class=HTMLResponse)
|
|
async def teams_channel_add(request: Request, db=Depends(get_db),
|
|
tc_name: str = Form(...),
|
|
tc_webhook_url: str = Form(...),
|
|
tc_description: str = Form(""),
|
|
tc_is_default: 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, "settings"):
|
|
return RedirectResponse(url="/settings")
|
|
is_def = bool(tc_is_default)
|
|
if is_def:
|
|
# un seul canal par défaut
|
|
db.execute(text("UPDATE teams_channels SET is_default=false WHERE is_default=true"))
|
|
db.execute(text("""
|
|
INSERT INTO teams_channels (name, webhook_url, description, is_default)
|
|
VALUES (:n, :u, :d, :df)
|
|
"""), {"n": tc_name.strip(), "u": tc_webhook_url.strip(),
|
|
"d": (tc_description or None), "df": is_def})
|
|
db.commit()
|
|
ctx = _build_context(db, user, saved="teams")
|
|
ctx["request"] = request
|
|
return templates.TemplateResponse("settings.html", ctx)
|
|
|
|
|
|
@router.post("/settings/teams-channel/{tc_id}/edit", response_class=HTMLResponse)
|
|
async def teams_channel_edit(request: Request, tc_id: int, db=Depends(get_db),
|
|
tc_name: str = Form(...),
|
|
tc_webhook_url: str = Form(...),
|
|
tc_description: str = Form(""),
|
|
tc_is_active: str = Form(""),
|
|
tc_is_default: 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, "settings"):
|
|
return RedirectResponse(url="/settings")
|
|
is_def = bool(tc_is_default)
|
|
is_act = bool(tc_is_active)
|
|
if is_def:
|
|
db.execute(text("UPDATE teams_channels SET is_default=false WHERE is_default=true AND id<>:id"),
|
|
{"id": tc_id})
|
|
db.execute(text("""
|
|
UPDATE teams_channels
|
|
SET name=:n, webhook_url=:u, description=:d, is_default=:df, is_active=:ia
|
|
WHERE id=:id
|
|
"""), {"n": tc_name.strip(), "u": tc_webhook_url.strip(),
|
|
"d": (tc_description or None), "df": is_def, "ia": is_act, "id": tc_id})
|
|
db.commit()
|
|
ctx = _build_context(db, user, saved="teams")
|
|
ctx["request"] = request
|
|
return templates.TemplateResponse("settings.html", ctx)
|
|
|
|
|
|
@router.post("/settings/teams-channel/{tc_id}/delete", response_class=HTMLResponse)
|
|
async def teams_channel_delete(request: Request, tc_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, "settings"):
|
|
return RedirectResponse(url="/settings")
|
|
db.execute(text("DELETE FROM teams_channels WHERE id=:id"), {"id": tc_id})
|
|
db.commit()
|
|
ctx = _build_context(db, user, saved="teams")
|
|
ctx["request"] = request
|
|
return templates.TemplateResponse("settings.html", ctx)
|
|
|
|
|
|
@router.post("/settings/teams-channel/{tc_id}/test")
|
|
async def teams_channel_test(request: Request, tc_id: int, db=Depends(get_db)):
|
|
"""Envoie un message test sur le webhook pour valider la config."""
|
|
from fastapi.responses import JSONResponse
|
|
user = get_current_user(request)
|
|
if not user:
|
|
return JSONResponse({"ok": False, "msg": "Non authentifié"}, status_code=401)
|
|
perms = get_user_perms(db, user)
|
|
if not can_edit(perms, "settings"):
|
|
return JSONResponse({"ok": False, "msg": "Permission refusée"}, status_code=403)
|
|
row = db.execute(text("SELECT name, webhook_url FROM teams_channels WHERE id=:id"),
|
|
{"id": tc_id}).fetchone()
|
|
if not row:
|
|
return JSONResponse({"ok": False, "msg": "Canal introuvable"}, status_code=404)
|
|
try:
|
|
from ..services.teams_service import send_test_message
|
|
result = send_test_message(row.webhook_url, row.name, user.get("username") or "PatchCenter")
|
|
return JSONResponse(result)
|
|
except Exception as e:
|
|
return JSONResponse({"ok": False, "msg": f"Erreur: {e}"}, status_code=500)
|
|
|
|
|
|
# --- Server clusters CRUD ---
|
|
|
|
@router.post("/settings/server-cluster/add", response_class=HTMLResponse)
|
|
async def server_cluster_add(request: Request, db=Depends(get_db),
|
|
sc_name: str = Form(...),
|
|
sc_description: str = Form(""),
|
|
sc_reboot_strategy: str = Form("sequential")):
|
|
user = get_current_user(request)
|
|
if not user:
|
|
return RedirectResponse(url="/login")
|
|
perms = get_user_perms(db, user)
|
|
if not can_edit(perms, "settings"):
|
|
return RedirectResponse(url="/settings")
|
|
strat = sc_reboot_strategy if sc_reboot_strategy in ("sequential", "parallel") else "sequential"
|
|
try:
|
|
db.execute(text("""
|
|
INSERT INTO server_clusters (name, description, reboot_strategy)
|
|
VALUES (:n, :d, :s)
|
|
"""), {"n": sc_name.strip(), "d": (sc_description or None), "s": strat})
|
|
db.commit()
|
|
except Exception as e:
|
|
db.rollback()
|
|
# Probablement violation unique sur name
|
|
logger.warning(f"server_cluster_add failed: {e}")
|
|
ctx = _build_context(db, user, saved="cluster")
|
|
ctx["request"] = request
|
|
return templates.TemplateResponse("settings.html", ctx)
|
|
|
|
|
|
@router.post("/settings/server-cluster/{sc_id}/edit", response_class=HTMLResponse)
|
|
async def server_cluster_edit(request: Request, sc_id: int, db=Depends(get_db),
|
|
sc_name: str = Form(...),
|
|
sc_description: str = Form(""),
|
|
sc_reboot_strategy: str = Form("sequential"),
|
|
sc_is_active: 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, "settings"):
|
|
return RedirectResponse(url="/settings")
|
|
strat = sc_reboot_strategy if sc_reboot_strategy in ("sequential", "parallel") else "sequential"
|
|
db.execute(text("""
|
|
UPDATE server_clusters
|
|
SET name=:n, description=:d, reboot_strategy=:s, is_active=:ia
|
|
WHERE id=:id
|
|
"""), {"n": sc_name.strip(), "d": (sc_description or None), "s": strat,
|
|
"ia": bool(sc_is_active), "id": sc_id})
|
|
db.commit()
|
|
ctx = _build_context(db, user, saved="cluster")
|
|
ctx["request"] = request
|
|
return templates.TemplateResponse("settings.html", ctx)
|
|
|
|
|
|
@router.post("/settings/server-cluster/{sc_id}/delete", response_class=HTMLResponse)
|
|
async def server_cluster_delete(request: Request, sc_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, "settings"):
|
|
return RedirectResponse(url="/settings")
|
|
# Check qu'aucun serveur n'y est encore rattaché
|
|
cnt = db.execute(text("SELECT COUNT(*) FROM servers WHERE cluster_id=:id"),
|
|
{"id": sc_id}).scalar() or 0
|
|
if cnt > 0:
|
|
# Désactivation seule (ne pas casser les FK et l'historique)
|
|
db.execute(text("UPDATE server_clusters SET is_active=false WHERE id=:id"), {"id": sc_id})
|
|
else:
|
|
db.execute(text("DELETE FROM server_clusters WHERE id=:id"), {"id": sc_id})
|
|
db.commit()
|
|
ctx = _build_context(db, user, saved="cluster")
|
|
ctx["request"] = request
|
|
return templates.TemplateResponse("settings.html", ctx)
|
|
|
|
|
|
@router.get("/settings/server-cluster/{sc_id}/servers")
|
|
async def server_cluster_servers(request: Request, sc_id: int, db=Depends(get_db)):
|
|
"""Liste des serveurs du cluster avec leur cluster_order — JSON pour panneau détail."""
|
|
from fastapi.responses import JSONResponse
|
|
user = get_current_user(request)
|
|
if not user:
|
|
return JSONResponse({"ok": False, "msg": "Non authentifié"}, status_code=401)
|
|
perms = get_user_perms(db, user)
|
|
if not can_view(perms, "settings"):
|
|
return JSONResponse({"ok": False, "msg": "Permission refusée"}, status_code=403)
|
|
rows = db.execute(text("""
|
|
SELECT id, hostname, cluster_order, machine_type, etat
|
|
FROM servers
|
|
WHERE cluster_id = :id
|
|
ORDER BY cluster_order NULLS LAST, hostname
|
|
"""), {"id": sc_id}).fetchall()
|
|
return JSONResponse({"ok": True, "count": len(rows),
|
|
"servers": [{"id": r.id, "hostname": str(r.hostname),
|
|
"order": r.cluster_order,
|
|
"machine_type": r.machine_type,
|
|
"etat": r.etat} for r in rows]})
|
|
|
|
|
|
# --- 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")
|
|
perms = get_user_perms(db, user)
|
|
if not can_edit(perms, "settings"):
|
|
return RedirectResponse(url="/settings")
|
|
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)
|
|
|
|
|
|
# --- Réseaux autorisés ---
|
|
|
|
def _regen_nginx_acl(db):
|
|
"""Régénère le fichier ACL nginx depuis la base"""
|
|
nets = db.execute(text("SELECT cidr FROM allowed_networks WHERE is_active = true ORDER BY cidr")).fetchall()
|
|
lines = ["# Généré par PatchCenter — ne pas éditer manuellement"]
|
|
for n in nets:
|
|
lines.append(f"allow {n.cidr};")
|
|
lines.append("deny all;")
|
|
try:
|
|
with open("/etc/nginx/patchcenter_acl.conf", "w") as f:
|
|
f.write("\n".join(lines) + "\n")
|
|
import subprocess
|
|
subprocess.run(["nginx", "-t"], capture_output=True, check=True)
|
|
subprocess.run(["systemctl", "reload", "nginx"], capture_output=True)
|
|
return True
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
@router.post("/settings/network/add")
|
|
async def network_add(request: Request, db=Depends(get_db),
|
|
cidr: str = Form(...), description: 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, "settings"):
|
|
return RedirectResponse(url="/settings")
|
|
db.execute(text("INSERT INTO allowed_networks (cidr, description) VALUES (:c, :d)"),
|
|
{"c": cidr.strip(), "d": description or None})
|
|
db.commit()
|
|
_regen_nginx_acl(db)
|
|
ctx = _build_context(db, user, saved="security")
|
|
ctx["request"] = request
|
|
return templates.TemplateResponse("settings.html", ctx)
|
|
|
|
|
|
@router.post("/settings/network/{net_id}/delete")
|
|
async def network_delete(request: Request, net_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, "settings"):
|
|
return RedirectResponse(url="/settings")
|
|
db.execute(text("DELETE FROM allowed_networks WHERE id = :id"), {"id": net_id})
|
|
db.commit()
|
|
_regen_nginx_acl(db)
|
|
ctx = _build_context(db, user, saved="security")
|
|
ctx["request"] = request
|
|
return templates.TemplateResponse("settings.html", ctx)
|
|
|
|
|
|
@router.post("/settings/network/{net_id}/toggle")
|
|
async def network_toggle(request: Request, net_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, "settings"):
|
|
return RedirectResponse(url="/settings")
|
|
db.execute(text("UPDATE allowed_networks SET is_active = NOT is_active WHERE id = :id"), {"id": net_id})
|
|
db.commit()
|
|
_regen_nginx_acl(db)
|
|
ctx = _build_context(db, user, saved="security")
|
|
ctx["request"] = request
|
|
return templates.TemplateResponse("settings.html", ctx)
|