feat(historie): FILE-008 manipulationssicheres Audit-Journal (Hash-Kette je Ressource)

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CVgbozhYmuEhiEJHffRXCV

arbeitskacheln/03_digital_file.md

@@ -189,7 +189,13 @@ aus Open-Source-Vergleich (KP Front/InvenTree), FILE-009 (Notizen) ergänzt am
   in der DB) wird vom Verifikations-Lauf erkannt und gemeldet.
 - **Tests:** Kettenbruch-Erkennungstest (Eintrag nachträglich manipuliert → Verifikation
   schlägt fehl), Normalfall-Test (unveränderte Kette → Verifikation erfolgreich).
-- **DoD:** offen. **Referenz (Open-Source-Vergleich 2026-09-09):** KP Front
+- **DoD:** ✅ umgesetzt (2026-09-10). `historie.hash` (Migration 0038, inkl.
+  Backfill bestehender Einträge in derselben Migration), Kette je
+  (entitaet_typ, entitaet_id) über `app/services/historie_hash.py`
+  (Genesis-Wert `"0"*64`), `GET /historie/verifikation` (Administrator-only),
+  Akte-Badge „Integrität geprüft ✓"/Bruch-Warnung. `backend/tests/test_historie.py`
+  (Normalfall + Manipulationserkennungstest + Rechte-Test). **Referenz
+  (Open-Source-Vergleich 2026-09-09):** KP Front
   (`feuerwehr-oberwil/kp-front`, AGPL-3.0-or-later — nur Muster) nutzt genau dieses
   Hash-Ketten-Prinzip für sein Einsatzjournal. Kombiniert mit InvenTree-Vorbild
   (lückenloser Stock-History-Trail bei jeder Bestandsänderung) — WH-006-Korrekturbuchungen

backend/alembic/versions/0038_historie_hash.py

@@ -0,0 +1,72 @@
+"""FILE-008: Hash-Kette für historie (manipulationssicheres Audit-Journal).
+
+Revision ID: 0038_historie_hash
+Revises: 0037_readiness_snapshot
+Create Date: 2026-09-10
+"""
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+from app.services.historie_hash import GENESIS_HASH, berechne_hash
+
+revision: str = "0038_historie_hash"
+down_revision: Union[str, None] = "0037_readiness_snapshot"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+historie_tabelle = sa.table(
+    "historie",
+    sa.column("id"),
+    sa.column("zeitpunkt"),
+    sa.column("benutzer_id"),
+    sa.column("ereignistyp"),
+    sa.column("entitaet_typ"),
+    sa.column("entitaet_id"),
+    sa.column("alter_wert"),
+    sa.column("neuer_wert"),
+    sa.column("begruendung"),
+    sa.column("hash"),
+)
+
+
+def upgrade() -> None:
+    op.add_column("historie", sa.Column("hash", sa.String(), nullable=True))
+
+    bind = op.get_bind()
+    zeilen = bind.execute(
+        sa.text(
+            "SELECT id, zeitpunkt, benutzer_id, ereignistyp, entitaet_typ, entitaet_id, "
+            "alter_wert, neuer_wert, begruendung FROM historie "
+            "ORDER BY entitaet_typ, entitaet_id, zeitpunkt, id"
+        )
+    ).mappings().all()
+
+    letzter_hash: dict[tuple[str, str], str] = {}
+    for zeile in zeilen:
+        schluessel = (zeile["entitaet_typ"], zeile["entitaet_id"])
+        vorgaenger = letzter_hash.get(schluessel, GENESIS_HASH)
+        hash_wert = berechne_hash(
+            vorgaenger,
+            id=str(zeile["id"]),
+            zeitpunkt=zeile["zeitpunkt"],
+            benutzer_id=zeile["benutzer_id"],
+            ereignistyp=zeile["ereignistyp"],
+            entitaet_typ=zeile["entitaet_typ"],
+            entitaet_id=zeile["entitaet_id"],
+            alter_wert=zeile["alter_wert"],
+            neuer_wert=zeile["neuer_wert"],
+            begruendung=zeile["begruendung"],
+        )
+        letzter_hash[schluessel] = hash_wert
+        bind.execute(
+            historie_tabelle.update().where(historie_tabelle.c.id == zeile["id"]).values(hash=hash_wert)
+        )
+
+    op.alter_column("historie", "hash", nullable=False)
+
+
+def downgrade() -> None:
+    op.drop_column("historie", "hash")

backend/app/api/v1/endpoints/historie.py

@@ -9,12 +9,15 @@ from app.db.session import get_db
 from app.models.auth import Benutzer, RolleTyp
 from app.models.historie import Historie
 from app.schemas.historie import HistorieRead
+from app.schemas.historie_verifikation import HistorieVerifikation
+from app.services.historie import pruefe_kette
 
 router = APIRouter()
 
 _verantwortliche = require_roles(
     RolleTyp.administration, RolleTyp.materialverantwortlicher, RolleTyp.leitungsverantwortlicher
 )
+_admin_only = require_roles(RolleTyp.administration)
 
 
 @router.get("/historie", response_model=list[HistorieRead])
@@ -61,3 +64,17 @@ async def liste_historie(
         )
         for eintrag, benutzer_name in result.all()
     ]
+
+
+@router.get("/historie/verifikation", response_model=HistorieVerifikation)
+async def verifiziere_historie(
+    entitaet_typ: str,
+    entitaet_id: str,
+    db: AsyncSession = Depends(get_db),
+    _=Depends(_admin_only),
+) -> HistorieVerifikation:
+    """FILE-008: Hash-Kette EINER Ressource nachrechnen und mit den
+    gespeicherten Hashes vergleichen - Administrator-only, da dies ein
+    forensisches Werkzeug ist, kein Alltags-Lesezugriff."""
+    ergebnis = await pruefe_kette(db, entitaet_typ, entitaet_id)
+    return HistorieVerifikation(entitaet_typ=entitaet_typ, entitaet_id=entitaet_id, **ergebnis)

backend/app/models/historie.py

@@ -25,3 +25,6 @@ class Historie(Base):
     alter_wert: Mapped[dict | None] = mapped_column(JSONB)
     neuer_wert: Mapped[dict | None] = mapped_column(JSONB)
     begruendung: Mapped[str | None] = mapped_column(String)
+    # FILE-008: SHA-256 über Inhalt+Vorgänger-Hash, Kette läuft je
+    # (entitaet_typ, entitaet_id) separat - siehe app/services/historie_hash.py.
+    hash: Mapped[str] = mapped_column(String, nullable=False)

backend/app/schemas/historie_verifikation.py

@@ -0,0 +1,11 @@
+import uuid
+
+from pydantic import BaseModel
+
+
+class HistorieVerifikation(BaseModel):
+    entitaet_typ: str
+    entitaet_id: str
+    intakt: bool
+    geprueft: int
+    bruch_bei_eintrag_id: uuid.UUID | None = None

backend/app/services/historie.py

@@ -1,9 +1,60 @@
+import uuid
 from datetime import datetime, timezone
 from typing import Any
 
+from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.models.historie import Historie
+from app.services.historie_hash import GENESIS_HASH, berechne_hash
+
+
+async def pruefe_kette(db: AsyncSession, entitaet_typ: str, entitaet_id: str) -> dict:
+    """FILE-008: läuft die gespeicherte Kette (entitaet_typ, entitaet_id)
+    nochmal komplett nach und vergleicht jeden gespeicherten Hash mit dem
+    NEU berechneten Wert aus den aktuellen Spaltenwerten - jede nachträgliche
+    Änderung (auch direkt in der DB) lässt den Vergleich ab dieser Stelle
+    fehlschlagen."""
+    result = await db.execute(
+        select(Historie)
+        .where(Historie.entitaet_typ == entitaet_typ, Historie.entitaet_id == entitaet_id)
+        .order_by(Historie.zeitpunkt, Historie.id)
+    )
+    eintraege = list(result.scalars().all())
+
+    vorgaenger_hash = GENESIS_HASH
+    for eintrag in eintraege:
+        erwarteter_hash = berechne_hash(
+            vorgaenger_hash,
+            id=str(eintrag.id),
+            zeitpunkt=eintrag.zeitpunkt,
+            benutzer_id=eintrag.benutzer_id,
+            ereignistyp=eintrag.ereignistyp,
+            entitaet_typ=eintrag.entitaet_typ,
+            entitaet_id=eintrag.entitaet_id,
+            alter_wert=eintrag.alter_wert,
+            neuer_wert=eintrag.neuer_wert,
+            begruendung=eintrag.begruendung,
+        )
+        if erwarteter_hash != eintrag.hash:
+            return {
+                "intakt": False,
+                "geprueft": len(eintraege),
+                "bruch_bei_eintrag_id": eintrag.id,
+            }
+        vorgaenger_hash = eintrag.hash
+
+    return {"intakt": True, "geprueft": len(eintraege), "bruch_bei_eintrag_id": None}
+
+
+async def _letzter_hash(db: AsyncSession, entitaet_typ: str, entitaet_id: str) -> str:
+    result = await db.execute(
+        select(Historie.hash)
+        .where(Historie.entitaet_typ == entitaet_typ, Historie.entitaet_id == entitaet_id)
+        .order_by(Historie.zeitpunkt.desc(), Historie.id.desc())
+        .limit(1)
+    )
+    return result.scalar_one_or_none() or GENESIS_HASH
 
 
 async def log(
@@ -20,18 +71,41 @@ async def log(
 ) -> Historie:
     """Prompt 13: append-only Audit-Log. Deckt Sprint 5 die Kernkette ab (Kontrolle,
     Fehlbestand, Nachfüllung, Mindermenge) - Stammdaten-/Vorlagen-/Benutzerverwaltung
-    folgen bei Bedarf mit demselben Muster, hier noch nicht flächendeckend verdrahtet."""
-    eintrag = Historie(
-        erzeugt_von_server_id=zustaendiger_server_id,
-        zeitpunkt=datetime.now(timezone.utc),
+    folgen bei Bedarf mit demselben Muster, hier noch nicht flächendeckend verdrahtet.
+
+    FILE-008: jeder Eintrag verkettet sich per Hash mit dem Vorgänger DERSELBEN
+    (entitaet_typ, entitaet_id) - nicht global, siehe Kachel-Entscheidung
+    2026-09-09."""
+    entitaet_id_str = str(entitaet_id)
+    vorgaenger_hash = await _letzter_hash(db, entitaet_typ, entitaet_id_str)
+
+    eintrag_id = uuid.uuid4()
+    zeitpunkt = datetime.now(timezone.utc)
+    hash_wert = berechne_hash(
+        vorgaenger_hash,
+        id=str(eintrag_id),
+        zeitpunkt=zeitpunkt,
         benutzer_id=benutzer_id,
         ereignistyp=ereignistyp,
         entitaet_typ=entitaet_typ,
-        entitaet_id=str(entitaet_id),
+        entitaet_id=entitaet_id_str,
         alter_wert=alter_wert,
         neuer_wert=neuer_wert,
         begruendung=begruendung,
     )
+    eintrag = Historie(
+        id=eintrag_id,
+        erzeugt_von_server_id=zustaendiger_server_id,
+        zeitpunkt=zeitpunkt,
+        benutzer_id=benutzer_id,
+        ereignistyp=ereignistyp,
+        entitaet_typ=entitaet_typ,
+        entitaet_id=entitaet_id_str,
+        alter_wert=alter_wert,
+        neuer_wert=neuer_wert,
+        begruendung=begruendung,
+        hash=hash_wert,
+    )
     db.add(eintrag)
     await db.flush()
     return eintrag

backend/app/services/historie_hash.py

@@ -0,0 +1,50 @@
+import hashlib
+import json
+from datetime import datetime
+from typing import Any
+
+# FILE-008: fixer Genesis-Wert statt NULL/leerem String für den ersten Eintrag
+# einer Kette (entitaet_typ, entitaet_id) - macht "kein Vorgänger" explizit statt
+# über einen Sonderfall NULL zu behandeln, und ist stabil über Migration und
+# Laufzeit-Code hinweg identisch (beide importieren diese Konstante).
+GENESIS_HASH = "0" * 64
+
+# Keine SQLAlchemy-Modell-Importe hier (bewusst nur stdlib) - dieses Modul wird
+# sowohl vom laufenden Service als auch von der Backfill-Migration importiert.
+
+
+def _serialisiere(wert: Any) -> Any:
+    if isinstance(wert, datetime):
+        return wert.isoformat()
+    return wert
+
+
+def berechne_hash(
+    vorgaenger_hash: str,
+    *,
+    id: str,
+    zeitpunkt: datetime | str,
+    benutzer_id: int | None,
+    ereignistyp: str,
+    entitaet_typ: str,
+    entitaet_id: str,
+    alter_wert: dict | None,
+    neuer_wert: dict | None,
+    begruendung: str | None,
+) -> str:
+    """SHA-256 über (Vorgänger-Hash + kanonisch serialisierter Eintragsinhalt) -
+    jede nachträgliche Änderung eines historischen Eintrags oder seines
+    Vorgängers ändert den Hash und bricht damit die Kette erkennbar."""
+    inhalt = {
+        "id": str(id),
+        "zeitpunkt": _serialisiere(zeitpunkt),
+        "benutzer_id": benutzer_id,
+        "ereignistyp": ereignistyp,
+        "entitaet_typ": entitaet_typ,
+        "entitaet_id": entitaet_id,
+        "alter_wert": alter_wert,
+        "neuer_wert": neuer_wert,
+        "begruendung": begruendung,
+    }
+    kanonisch = json.dumps(inhalt, sort_keys=True, default=str, ensure_ascii=True)
+    return hashlib.sha256((vorgaenger_hash + kanonisch).encode("utf-8")).hexdigest()

backend/tests/test_historie.py

@@ -97,3 +97,82 @@ async def test_historie_mitarbeiter_ohne_zugriff(client, mitarbeiter_user):
     token = await login(client, "mitarbeiter1")
     response = await client.get("/api/v1/historie", headers=auth_header(token))
     assert response.status_code == 403
+
+
+@pytest.mark.asyncio
+async def test_hash_kette_intakt_im_normalfall(
+    client, db_session, objekt_mit_position, mitarbeiter_user, admin_user
+):
+    objekt, material = objekt_mit_position
+    mitarbeiter_token = await login(client, "mitarbeiter1")
+
+    start = await client.post(
+        f"/api/v1/objekte/{objekt.id}/kontrollen", json={"uebernehmen": False}, headers=auth_header(mitarbeiter_token)
+    )
+    kontrolle_id = start.json()["id"]
+    await client.put(
+        f"/api/v1/kontrollen/{kontrolle_id}/positionen/{material.id}",
+        json={"istmenge": "10"},
+        headers=auth_header(mitarbeiter_token),
+    )
+
+    admin_token = await login(client, "admin1")
+    response = await client.get(
+        "/api/v1/historie/verifikation",
+        params={"entitaet_typ": "kontrolle", "entitaet_id": kontrolle_id},
+        headers=auth_header(admin_token),
+    )
+    assert response.status_code == 200
+    body = response.json()
+    assert body["intakt"] is True
+    assert body["geprueft"] >= 2
+    assert body["bruch_bei_eintrag_id"] is None
+
+
+@pytest.mark.asyncio
+async def test_hash_kette_erkennt_manipulation(
+    client, db_session, objekt_mit_position, mitarbeiter_user, admin_user
+):
+    objekt, material = objekt_mit_position
+    mitarbeiter_token = await login(client, "mitarbeiter1")
+
+    start = await client.post(
+        f"/api/v1/objekte/{objekt.id}/kontrollen", json={"uebernehmen": False}, headers=auth_header(mitarbeiter_token)
+    )
+    kontrolle_id = start.json()["id"]
+    await client.put(
+        f"/api/v1/kontrollen/{kontrolle_id}/positionen/{material.id}",
+        json={"istmenge": "10"},
+        headers=auth_header(mitarbeiter_token),
+    )
+
+    erster_eintrag = (
+        await db_session.execute(
+            select(Historie)
+            .where(Historie.entitaet_typ == "kontrolle", Historie.entitaet_id == str(kontrolle_id))
+            .order_by(Historie.zeitpunkt)
+        )
+    ).scalars().first()
+    erster_eintrag.begruendung = "nachträglich manipuliert"
+    await db_session.commit()
+
+    admin_token = await login(client, "admin1")
+    response = await client.get(
+        "/api/v1/historie/verifikation",
+        params={"entitaet_typ": "kontrolle", "entitaet_id": kontrolle_id},
+        headers=auth_header(admin_token),
+    )
+    body = response.json()
+    assert body["intakt"] is False
+    assert body["bruch_bei_eintrag_id"] == str(erster_eintrag.id)
+
+
+@pytest.mark.asyncio
+async def test_hash_kette_verifikation_nur_admin(client, materialverantwortlicher_user):
+    token = await login(client, "materialverantwortlicher1")
+    response = await client.get(
+        "/api/v1/historie/verifikation",
+        params={"entitaet_typ": "kontrolle", "entitaet_id": "irgendwas"},
+        headers=auth_header(token),
+    )
+    assert response.status_code == 403

frontend/src/components/HistorieIntegritaetBadge.tsx

@@ -0,0 +1,43 @@
+import { useEffect, useState } from "react";
+
+import { apiRequest } from "../api/client";
+import { useAuth } from "../auth/AuthContext";
+
+interface Verifikation {
+  intakt: boolean;
+  geprueft: number;
+  bruch_bei_eintrag_id: string | null;
+}
+
+interface Props {
+  entitaetTyp: string;
+  entitaetId: string;
+}
+
+// FILE-008: Verifikation ist Administrator-only (Backend-Gate), Badge blendet
+// sich für alle anderen Rollen einfach aus statt einen 403-Fehler zu zeigen.
+export function HistorieIntegritaetBadge({ entitaetTyp, entitaetId }: Props) {
+  const { istAdmin } = useAuth();
+  const [ergebnis, setErgebnis] = useState<Verifikation | null>(null);
+
+  useEffect(() => {
+    if (!istAdmin) return;
+    apiRequest<Verifikation>(
+      `/historie/verifikation?entitaet_typ=${entitaetTyp}&entitaet_id=${encodeURIComponent(entitaetId)}`
+    )
+      .then(setErgebnis)
+      .catch(() => setErgebnis(null));
+  }, [istAdmin, entitaetTyp, entitaetId]);
+
+  if (!istAdmin || ergebnis === null || ergebnis.geprueft === 0) return null;
+
+  return ergebnis.intakt ? (
+    <span className="text-success" style={{ fontSize: "0.85rem" }}>
+      Integrität geprüft ✓ ({ergebnis.geprueft} Einträge)
+    </span>
+  ) : (
+    <span className="text-danger" style={{ fontSize: "0.85rem" }}>
+      ⚠ Hash-Kette gebrochen ab Eintrag {ergebnis.bruch_bei_eintrag_id}
+    </span>
+  );
+}

frontend/src/pages/AktePage.tsx

@@ -5,6 +5,7 @@ import { apiRequest, ladeObjektEtikett } from "../api/client";
 import { useAuth } from "../auth/AuthContext";
 import type { Akte } from "../api/types";
 import { DokumentePanel } from "../components/DokumentePanel";
+import { HistorieIntegritaetBadge } from "../components/HistorieIntegritaetBadge";
 import { NotizenPanel } from "../components/NotizenPanel";
 import { ReadinessVerlaufPanel } from "../components/ReadinessVerlaufPanel";
 import { TankbuchPanel } from "../components/TankbuchPanel";
@@ -304,7 +305,10 @@ export function AktePage() {
       </div>
 
       <div className="akte-historie card">
-        <h3 style={{ marginTop: 0 }}>Historie</h3>
+        <div className="row-between" style={{ alignItems: "baseline" }}>
+          <h3 style={{ marginTop: 0 }}>Historie</h3>
+          <HistorieIntegritaetBadge entitaetTyp="objekt" entitaetId={String(objekt.id)} />
+        </div>
         {akte.historie.length === 0 && <p className="text-muted">Noch keine Änderungen protokolliert.</p>}
         {akte.historie.length > 0 && (
           <ul>