2026-07-21 10:29:31 +02:00

221 lines
5.3 KiB
Python

from __future__ import annotations
import json
import os
from dataclasses import dataclass
from functools import lru_cache
from typing import FrozenSet
@dataclass(frozen=True)
class UserConfig:
username: str
display_name: str
password_hash: str
roles: FrozenSet[str]
@dataclass(frozen=True)
class Settings:
app_title: str
tenant: str
site: str
pg_host: str
pg_port: int
pg_database: str
pg_user: str
pg_password: str
session_secret: str
session_max_age_s: int
cookie_secure: bool
write_enabled: bool
users: dict[str, UserConfig]
def env_first(*names: str, default: str | None = None) -> str | None:
for name in names:
value = os.getenv(name)
if value not in (None, ""):
return value
return default
def parse_bool(value: str | None, default: bool = False) -> bool:
if value is None:
return default
return value.strip().lower() in {
"1",
"true",
"yes",
"on",
}
def load_users(raw: str) -> dict[str, UserConfig]:
try:
payload = json.loads(raw)
except json.JSONDecodeError as exc:
raise RuntimeError(
"GOVERNANCE_UI_USERS_JSON no contiene JSON válido."
) from exc
if not isinstance(payload, list) or not payload:
raise RuntimeError(
"GOVERNANCE_UI_USERS_JSON debe ser una lista no vacía."
)
users: dict[str, UserConfig] = {}
for item in payload:
if not isinstance(item, dict):
raise RuntimeError(
"Cada usuario debe ser un objeto JSON."
)
username = str(item.get("username", "")).strip()
display_name = str(
item.get("display_name", "")
).strip()
password_hash = str(
item.get("password_hash", "")
).strip()
roles_raw = item.get("roles", [])
if (
not username
or not display_name
or not password_hash
or not isinstance(roles_raw, list)
):
raise RuntimeError(
"Cada usuario requiere username, display_name, "
"password_hash y roles."
)
roles = frozenset(
str(role).strip().upper()
for role in roles_raw
if str(role).strip()
)
allowed_roles = {
"PRODUCTION",
"TECHNICAL",
"JOINT",
"ADMIN",
}
if not roles or not roles.issubset(allowed_roles):
raise RuntimeError(
f"Roles inválidos para {username!r}: {sorted(roles)}"
)
if username in users:
raise RuntimeError(
f"Usuario duplicado: {username!r}"
)
users[username] = UserConfig(
username=username,
display_name=display_name,
password_hash=password_hash,
roles=roles,
)
return users
@lru_cache(maxsize=1)
def get_settings() -> Settings:
database = env_first(
"PGDATABASE",
"POSTGRES_DB",
)
user = env_first(
"PGUSER",
"POSTGRES_USER",
)
password = env_first(
"PGPASSWORD",
"POSTGRES_PASSWORD",
)
session_secret = env_first(
"GOVERNANCE_UI_SESSION_SECRET"
)
users_json = env_first(
"GOVERNANCE_UI_USERS_JSON"
)
missing = [
name
for name, value in (
("PGDATABASE/POSTGRES_DB", database),
("PGUSER/POSTGRES_USER", user),
("PGPASSWORD/POSTGRES_PASSWORD", password),
("GOVERNANCE_UI_SESSION_SECRET", session_secret),
("GOVERNANCE_UI_USERS_JSON", users_json),
)
if not value
]
if missing:
raise RuntimeError(
"Faltan variables: " + ", ".join(missing)
)
return Settings(
app_title=env_first(
"GOVERNANCE_UI_TITLE",
default="MESAVAULT · Revisión de gobernanza",
)
or "MESAVAULT · Revisión de gobernanza",
tenant=env_first(
"TENANT_OVERRIDE",
"TENANT",
default="ucepsa",
)
or "ucepsa",
site=env_first(
"SITE_OVERRIDE",
"SITE",
default="ucepsa_onpremise",
)
or "ucepsa_onpremise",
pg_host=env_first(
"PGHOST",
default="mv_ucepsa_postgres_hot",
)
or "mv_ucepsa_postgres_hot",
pg_port=int(
env_first(
"PGPORT",
default="5432",
)
or "5432"
),
pg_database=str(database),
pg_user=str(user),
pg_password=str(password),
session_secret=str(session_secret),
session_max_age_s=int(
env_first(
"GOVERNANCE_UI_SESSION_MAX_AGE_S",
default="28800",
)
or "28800"
),
cookie_secure=parse_bool(
env_first(
"GOVERNANCE_UI_COOKIE_SECURE",
default="false",
)
),
write_enabled=parse_bool(
env_first(
"GOVERNANCE_UI_WRITE_ENABLED",
default="false",
)
),
users=load_users(str(users_json)),
)