Reisacher-Forderung/app/permissions.py

from dataclasses import dataclass

from fastapi import HTTPException, status
from sqlalchemy import text
from sqlalchemy.orm import Session


@dataclass(frozen=True)
class Entitlement:
    hauptbetrieb_id: str
    standort_id: str
    abteilung: str | None
    role: str
    priority: int


@dataclass(frozen=True)
class Principal:
    username: str
    name: str
    email: str
    role: str
    auth_source: str
    csrf: str
    entitlements: tuple[Entitlement, ...]


RIGHTS_SQL = text("""
    SELECT BR.Hauptbetrieb_ID,
           BR.Standort_ID,
           BR.Abteilung,
           BR.Rolle_Name,
           BR.Rolle_Prio
    FROM dbo.Benutzer_Rechte AS BR
    WHERE BR.Benutzer_ID = :username
    ORDER BY BR.Rolle_Prio DESC, BR.Rolle_Name ASC
    """)


def load_entitlements(db: Session, username: str) -> tuple[Entitlement, ...]:
    rows = db.execute(RIGHTS_SQL, {"username": username}).mappings().all()
    return tuple(
        Entitlement(
            hauptbetrieb_id=str(row["Hauptbetrieb_ID"]),
            standort_id=str(row["Standort_ID"]),
            abteilung=row["Abteilung"],
            role=row["Rolle_Name"],
            priority=int(row["Rolle_Prio"]),
        )
        for row in rows
    )


def highest_role(entitlements: tuple[Entitlement, ...]) -> str:
    if not entitlements:
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Keine Berechtigung für diese Anwendung")
    return sorted(entitlements, key=lambda item: (-item.priority, item.role))[0].role


FORDERUNG_ACCESS_SQL = text("""
    SELECT TOP 1 1
    FROM dbo.Forderungen_Kopf AS F
    WHERE F.Client_DB = :client_db
      AND F.Beleg_Nr = :beleg_nr
      AND EXISTS (
          SELECT 1
          FROM dbo.Benutzer_Rechte AS BR
          WHERE BR.Benutzer_ID = :auth_username
            AND BR.Hauptbetrieb_ID = F.Hauptbetrieb_ID
            AND BR.Standort_ID = F.Standort_ID
            AND BR.Abteilung = F.Abteilung
      )
    """)


def has_forderung_access(db: Session, principal: Principal, client_db: str, beleg_nr: str) -> bool:
    return (
        db.execute(
            FORDERUNG_ACCESS_SQL,
            {"client_db": client_db, "beleg_nr": beleg_nr, "auth_username": principal.username},
        ).first()
        is not None
    )


def ensure_forderung_access(db: Session, principal: Principal, client_db: str, beleg_nr: str) -> None:
    if not has_forderung_access(db, principal, client_db, beleg_nr):
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Forderung nicht gefunden")


def current_principal(request) -> Principal:
    principal = getattr(request.state, "principal", None)
    if principal is None:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Nicht angemeldet")
    return principal