fix(historie): Audit-Lücke bei Mangel/Personal/Objekt-Änderungen geschlossen
CI / backend-tests (push) Failing after 1m47s
CI / frontend-build (push) Successful in 17s

Bisher waren nur Kontrolle/Fehlbestand/Nachfüllung/Mindermenge/Geräteinstanz
im append-only Audit-Log (historie.log() war seit Prompt 13 nie flächendeckend
verdrahtet, docstring sagte das bereits so). Drei neue Ereignisse ergänzt:

- mangel_gemeldet / mangel_status_geaendert (Mangel-Modul)
- qualifikation_erfasst (Personal-Modul, sicherheitsrelevant: "wer hat wem
  wann eine Qualifikation bestätigt" hängt fachlich direkt an "wer darf
  fahren")
- objekt_geaendert (Status-/Fahrzeug-Zuordnungsänderungen via PATCH /objekte)

Lagerbewegung bewusst NICHT zusätzlich in historie dupliziert - hat bereits
eigenes vollständiges Audit-Trail (Wer/Wann/Von/Nach/Grund in eigener
Tabelle).

Frontend: Änderungslog-Filter um geraet_instanz/mangel/benutzer_qualifikation/
objekt ergänzt (geraet_instanz fehlte dort zuvor ebenfalls).

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01KC8HYvv6UkCVYheYiTw9DD
This commit is contained in:
2026-09-05 13:50:48 +02:00
co-authored by Claude Sonnet 5
parent 417df2beb6
commit 11ad747136
9 changed files with 201 additions and 16 deletions
+25 -2
View File
@@ -5,7 +5,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
from app.api.deps import get_current_user, require_roles
from app.db.session import get_db
from app.models.auth import RolleTyp
from app.models.auth import KnotenTyp, RolleTyp, Systemknoten
from app.models.personal import (
BenutzerQualifikation,
Einheit,
@@ -25,10 +25,16 @@ from app.schemas.personal import (
QualifikationstypRead,
QualifikationstypUpdate,
)
from app.services import historie as historie_service
from app.services.personal import pruefe_berechtigung
router = APIRouter()
async def _hauptserver_id(db: AsyncSession) -> int:
result = await db.execute(select(Systemknoten.id).where(Systemknoten.typ == KnotenTyp.haupt))
return result.scalar_one()
_admin_only = require_roles(RolleTyp.administration)
@@ -126,7 +132,7 @@ async def erfasse_benutzer_qualifikation(
benutzer_id: int,
payload: BenutzerQualifikationCreate,
db: AsyncSession = Depends(get_db),
_=Depends(_admin_only),
current_user=Depends(_admin_only),
) -> BenutzerQualifikation:
if payload.benutzer_id != benutzer_id:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="benutzer_id im Pfad und Body weichen ab")
@@ -138,6 +144,23 @@ async def erfasse_benutzer_qualifikation(
raise HTTPException(
status_code=status.HTTP_409_CONFLICT, detail="Qualifikation mit diesem Erwerbsdatum existiert bereits"
) from exc
# Historie-Lücke (Roadmap-Review): wer hat wem wann eine Qualifikation
# bestätigt, ist sicherheitsrelevant ("wer darf fahren" hängt fachlich
# direkt daran) und war bisher nirgends protokolliert.
await historie_service.log(
db,
zustaendiger_server_id=await _hauptserver_id(db),
benutzer_id=current_user.id,
ereignistyp="qualifikation_erfasst",
entitaet_typ="benutzer_qualifikation",
entitaet_id=qualifikation.id,
neuer_wert={
"benutzer_id": benutzer_id,
"qualifikationstyp_id": payload.qualifikationstyp_id,
"gueltig_bis": payload.gueltig_bis.isoformat() if payload.gueltig_bis else None,
},
)
return qualifikation