Security & Compliance

Audit-Trails & Revisionssicherheit: Wer hat wann was geändert?

Carola Schulte

27. Oktober 2025

25 min

"Wer hat den Preis geändert?" "War die Rechnung gestern noch anders?" "Können Sie beweisen, dass der Vertrag unverändert ist?" – Drei Fragen, die ohne Audit-Trail unbeantwortbar sind. In regulierten Branchen (Finanzen, Gesundheit, Behörden) ist Revisionssicherheit Pflicht. Aber auch jede ernsthafte Business-App braucht sie.

TL;DR – Die Kurzfassung

Audit-Trail: Lückenlose, unveränderliche Protokollierung aller Datenänderungen
Revisionssicherheit: Nachweisbare Unversehrtheit über lange Zeiträume (GoBD: 10 Jahre)
Drei Ansätze: Trigger-basiert (DB), Event-Sourcing (App), Dedicated Audit-Tables
Goldene Regel: Audit-Daten sind append-only. Niemals UPDATE, niemals DELETE.
Kryptografische Verkettung: Hash-Chain wie bei Blockchain – Manipulation wird erkennbar

Warum Audit-Trails unverzichtbar sind

Ein Audit-Trail beantwortet die 5 W-Fragen für jede Datenänderung:

┌─────────────────────────────────────────────────────────────────┐
│                    DIE 5 W DES AUDIT-TRAILS                     │
├─────────────────────────────────────────────────────────────────┤
│                                                                  │
│  WER?     → User-ID, Name, IP-Adresse, Session                  │
│  WAS?     → Welches Objekt (Tabelle, ID)                        │
│  WANN?    → Timestamp (UTC, mit Zeitzone!)                      │
│  WIE?     → CREATE, UPDATE, DELETE + alte/neue Werte            │
│  WARUM?   → Kontext (Request-ID, Feature, Business-Grund)       │
│                                                                  │
└─────────────────────────────────────────────────────────────────┘

Rechtliche Anforderungen

Regelwerk	Anforderung	Aufbewahrung
GoBD (DE)	Unveränderbarkeit, Nachvollziehbarkeit	10 Jahre
DSGVO Art. 30	Verzeichnis von Verarbeitungstätigkeiten	Während Verarbeitung
SOX (US)	Internal Controls, Financial Reporting	7 Jahre
HIPAA (US)	Access Logs für Gesundheitsdaten	6 Jahre
ISO 27001	A.12.4 Logging and Monitoring	Nach Risikoanalyse

Drei Implementierungs-Strategien

Strategie 1: Database Triggers (Transparentes Tracking)

Die Datenbank protokolliert automatisch – die Applikation muss nichts tun. Robust, aber weniger Kontext.

-- Audit-Log Tabelle (eine für alle Entities)
CREATE TABLE audit_log (
    id BIGSERIAL PRIMARY KEY,

    -- WAS wurde geändert?
    table_name VARCHAR(100) NOT NULL,
    record_id UUID NOT NULL,

    -- WIE wurde es geändert?
    action VARCHAR(10) NOT NULL CHECK (action IN ('INSERT', 'UPDATE', 'DELETE')),
    old_values JSONB,
    new_values JSONB,
    changed_fields TEXT[],  -- Bei UPDATE: welche Felder

    -- WANN?
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),

    -- WER? (aus Session-Variable)
    user_id UUID,
    user_email VARCHAR(255),
    ip_address INET,

    -- Kontext
    request_id UUID,

    -- Für Hash-Chain (Revisionssicherheit)
    prev_hash BYTEA,
    row_hash BYTEA NOT NULL
);

-- Indices für typische Queries
CREATE INDEX idx_audit_table_record ON audit_log(table_name, record_id);
CREATE INDEX idx_audit_created ON audit_log(created_at);
CREATE INDEX idx_audit_user ON audit_log(user_id);

-- Partitionierung für Performance bei großen Datenmengen
-- CREATE TABLE audit_log (...) PARTITION BY RANGE (created_at);

Generic Audit Trigger Function

-- WICHTIG: pgcrypto Extension für digest() aktivieren!
CREATE EXTENSION IF NOT EXISTS pgcrypto;

CREATE OR REPLACE FUNCTION audit_trigger_func()
RETURNS TRIGGER AS $$
DECLARE
    v_old_values JSONB;
    v_new_values JSONB;
    v_changed_fields TEXT[];
    v_user_id UUID;
    v_prev_hash BYTEA;
    v_row_hash BYTEA;
    v_hash_input TEXT;
    v_created_at TIMESTAMPTZ;  -- Konsistenter Timestamp!
BEGIN
    -- Timestamp EINMAL setzen – denselben Wert für Hash UND Insert verwenden!
    v_created_at := clock_timestamp();

    -- User-Context aus Session (von PHP gesetzt)
    v_user_id := NULLIF(current_setting('app.current_user_id', true), '')::UUID;

    -- Alte/neue Werte als JSON
    IF TG_OP = 'DELETE' THEN
        v_old_values := to_jsonb(OLD);
        v_new_values := NULL;
    ELSIF TG_OP = 'INSERT' THEN
        v_old_values := NULL;
        v_new_values := to_jsonb(NEW);
    ELSE -- UPDATE
        v_old_values := to_jsonb(OLD);
        v_new_values := to_jsonb(NEW);

        -- Nur geänderte Felder tracken
        SELECT array_agg(key)
        INTO v_changed_fields
        FROM jsonb_each(v_new_values) AS n(key, value)
        WHERE v_old_values->key IS DISTINCT FROM value;
    END IF;

    -- Letzten Hash für Chain holen (pro Tabelle, nicht pro Entity!)
    SELECT row_hash INTO v_prev_hash
    FROM audit_log
    WHERE table_name = TG_TABLE_NAME
    ORDER BY id DESC LIMIT 1;

    -- Neuen Hash berechnen (Chain)
    -- WICHTIG: BYTEA kanonisch als hex kodieren, nicht ::TEXT (Output-Format variiert!)
    -- WICHTIG: Exakt denselben Timestamp verwenden, den wir auch speichern!
    v_hash_input := COALESCE(encode(v_prev_hash, 'hex'), 'GENESIS') ||
                    TG_TABLE_NAME ||
                    COALESCE(OLD.id::TEXT, NEW.id::TEXT) ||  -- Demo: erwartet "id" als PK!
                    TG_OP ||
                    COALESCE(v_old_values::TEXT, '') ||
                    COALESCE(v_new_values::TEXT, '') ||
                    v_created_at::TEXT;
    -- sha256() existiert nicht out-of-the-box! → pgcrypto digest()
    v_row_hash := digest(v_hash_input, 'sha256');

    -- Audit-Eintrag schreiben (mit dem konsistenten Timestamp!)
    INSERT INTO audit_log (
        table_name, record_id, action,
        old_values, new_values, changed_fields,
        created_at,  -- Explizit setzen, nicht DEFAULT!
        user_id, user_email, ip_address, request_id,
        prev_hash, row_hash
    ) VALUES (
        TG_TABLE_NAME,
        COALESCE(NEW.id, OLD.id),
        TG_OP,
        v_old_values,
        v_new_values,
        v_changed_fields,
        v_created_at,  -- Derselbe Wert wie im Hash!
        v_user_id,
        NULLIF(current_setting('app.current_user_email', true), ''),
        NULLIF(current_setting('app.current_ip', true), '')::INET,
        NULLIF(current_setting('app.request_id', true), '')::UUID,
        v_prev_hash,
        v_row_hash
    );

    RETURN COALESCE(NEW, OLD);
END;
$$ LANGUAGE plpgsql;

Hash-Chain Scope: Diese Implementierung baut eine Chain pro Tabelle (alle invoices-Einträge verkettet, alle contracts-Einträge verkettet, etc.). Für eine Chain pro Entity (nur Änderungen an invoice #123 verkettet) müsste die WHERE-Klausel WHERE table_name = ... AND record_id = ... lauten. Beides ist legitim – pro Tabelle ist einfacher zu verifizieren.

Demo-Annahme: Der Trigger erwartet id als Primary Key (UUID). Bei anderen PK-Namen (invoice_id, BIGSERIAL, Composite Keys) muss COALESCE(OLD.id, NEW.id) angepasst werden – z.B. über TG_ARGV oder eine Mapping-Tabelle.

Trigger an Tabellen anhängen

-- Für jede zu überwachende Tabelle
CREATE TRIGGER audit_invoices
    AFTER INSERT OR UPDATE OR DELETE ON invoices
    FOR EACH ROW EXECUTE FUNCTION audit_trigger_func();

CREATE TRIGGER audit_contracts
    AFTER INSERT OR UPDATE OR DELETE ON contracts
    FOR EACH ROW EXECUTE FUNCTION audit_trigger_func();

CREATE TRIGGER audit_users
    AFTER INSERT OR UPDATE OR DELETE ON users
    FOR EACH ROW EXECUTE FUNCTION audit_trigger_func();

PHP: Session-Variablen setzen

Wichtig zu SET LOCAL:

SET LOCAL funktioniert nur innerhalb einer Transaktion. Ohne aktive Transaktion gibt PostgreSQL einen Fehler.
Option A (empfohlen): Request-Transaktion (Unit-of-Work) – dann passt SET LOCAL.
Option B: Ohne Request-Transaktion: SET + am Ende RESET (bei Connection Pooling / Long-lived Prozessen unbedingt resetten!).

<?php

declare(strict_types=1);

// Option A: Mit Request-Transaktion (SET LOCAL)
final class AuditContextMiddleware
{
    public function __construct(
        private PDO $db,
    ) {}

    public function __invoke(Request $request, callable $next): Response
    {
        $user = $request->getAttribute('user');
        $requestId = $request->getAttribute('request_id') ?? Uuid::v4();

        // Transaktion starten – erst dann funktioniert SET LOCAL!
        $this->db->beginTransaction();

        try {
            $this->db->exec(sprintf(
                "SET LOCAL app.current_user_id = %s",
                $this->db->quote($user?->id ?? '')
            ));
            $this->db->exec(sprintf(
                "SET LOCAL app.current_user_email = %s",
                $this->db->quote($user?->email ?? '')
            ));
            $this->db->exec(sprintf(
                "SET LOCAL app.current_ip = %s",
                $this->db->quote($request->getServerParams()['REMOTE_ADDR'] ?? '')
            ));
            $this->db->exec(sprintf(
                "SET LOCAL app.request_id = %s",
                $this->db->quote($requestId)
            ));

            $response = $next($request);

            $this->db->commit();
            return $response;
        } catch (\Throwable $e) {
            $this->db->rollBack();
            throw $e;
        }
    }
}

// Option B: Ohne Request-Transaktion (SET + RESET)
// Für Setups ohne Unit-of-Work Pattern

$this->db->exec("SET app.current_user_id = " . $this->db->quote($userId));
// ... weitere SETs ...

// AM ENDE DES REQUESTS (finally-Block oder Destruktor):
$this->db->exec("RESET app.current_user_id");
$this->db->exec("RESET app.current_user_email");
// ... oder alle auf einmal:
$this->db->exec("RESET ALL");  // Vorsicht: resettet ALLE Session-Variablen!

Strategie 2: Application-Level Event Sourcing

Events sind die Wahrheit – der aktuelle Zustand wird daraus berechnet. Maximale Flexibilität, aber komplexer.

<?php

declare(strict_types=1);

// Events sind immutable Value Objects
final readonly class InvoicePriceChanged implements DomainEvent
{
    public function __construct(
        public string $eventId,
        public string $invoiceId,
        public int $oldPriceCents,
        public int $newPriceCents,
        public string $reason,
        public string $changedBy,
        public DateTimeImmutable $occurredAt,
    ) {}

    public function toArray(): array
    {
        return [
            'event_type' => 'invoice.price_changed',
            'event_id' => $this->eventId,
            'aggregate_id' => $this->invoiceId,
            'payload' => [
                'old_price_cents' => $this->oldPriceCents,
                'new_price_cents' => $this->newPriceCents,
                'reason' => $this->reason,
            ],
            'metadata' => [
                'changed_by' => $this->changedBy,
                'occurred_at' => $this->occurredAt->format('c'),
            ],
        ];
    }
}

<?php

final class EventStore
{
    public function __construct(
        private PDO $db,
    ) {}

    public function append(DomainEvent $event): void
    {
        $data = $event->toArray();

        // Vorherigen Hash für Chain holen
        $prevHash = $this->getLastHash($data['aggregate_id']);

        // Neuen Hash berechnen
        $hashInput = ($prevHash ?? 'GENESIS') . json_encode($data);
        $rowHash = hash('sha256', $hashInput, true);

        $sql = "INSERT INTO event_store (
                    event_id, aggregate_type, aggregate_id,
                    event_type, payload, metadata,
                    prev_hash, row_hash, created_at
                ) VALUES (
                    :event_id, :aggregate_type, :aggregate_id,
                    :event_type, :payload, :metadata,
                    :prev_hash, :row_hash, NOW()
                )";

        $stmt = $this->db->prepare($sql);
        $stmt->execute([
            'event_id' => $data['event_id'],
            'aggregate_type' => 'Invoice',
            'aggregate_id' => $data['aggregate_id'],
            'event_type' => $data['event_type'],
            'payload' => json_encode($data['payload']),
            'metadata' => json_encode($data['metadata']),
            'prev_hash' => $prevHash,
            'row_hash' => $rowHash,
        ]);
    }

    public function getEvents(string $aggregateId): array
    {
        $sql = "SELECT * FROM event_store
                WHERE aggregate_id = :id
                ORDER BY created_at ASC";

        $stmt = $this->db->prepare($sql);
        $stmt->execute(['id' => $aggregateId]);

        return $stmt->fetchAll(PDO::FETCH_ASSOC);
    }

    private function getLastHash(string $aggregateId): ?string
    {
        $sql = "SELECT row_hash FROM event_store
                WHERE aggregate_id = :id
                ORDER BY created_at DESC LIMIT 1";

        $stmt = $this->db->prepare($sql);
        $stmt->execute(['id' => $aggregateId]);

        return $stmt->fetchColumn() ?: null;
    }
}

Strategie 3: Dedicated Audit-Service

Entkoppelt via Message Queue – Audit-Daten in separater Datenbank. Gut für Microservices.

<?php

final class AuditClient
{
    public function __construct(
        private MessageBroker $broker,
        private AuditContext $context,
    ) {}

    public function log(
        string $action,
        string $entityType,
        string $entityId,
        array $oldValues = [],
        array $newValues = [],
        array $extra = [],
    ): void {
        $message = [
            'audit_id' => Uuid::v4(),
            'timestamp' => (new DateTimeImmutable())->format('c'),
            'action' => $action,
            'entity_type' => $entityType,
            'entity_id' => $entityId,
            'old_values' => $oldValues,
            'new_values' => $newValues,
            'context' => [
                'user_id' => $this->context->getUserId(),
                'user_email' => $this->context->getUserEmail(),
                'ip_address' => $this->context->getIpAddress(),
                'request_id' => $this->context->getRequestId(),
                'user_agent' => $this->context->getUserAgent(),
                'tenant_id' => $this->context->getTenantId(),
                ...$extra,
            ],
        ];

        // Async an Audit-Service senden
        $this->broker->publish('audit.events', $message);
    }
}

// Nutzung im Repository
final class InvoiceRepository
{
    public function updatePrice(string $id, int $newPrice, string $reason): void
    {
        $old = $this->findById($id);

        $this->db->prepare("UPDATE invoices SET price_cents = ? WHERE id = ?")
                 ->execute([$newPrice, $id]);

        $this->audit->log(
            action: 'UPDATE',
            entityType: 'Invoice',
            entityId: $id,
            oldValues: ['price_cents' => $old['price_cents']],
            newValues: ['price_cents' => $newPrice],
            extra: ['reason' => $reason],
        );
    }
}

Revisionssicherheit durch Hash-Chain

Wie bei einer Blockchain: Jeder Eintrag enthält den Hash des vorherigen. Manipulation eines Eintrags zerstört die Kette.

┌──────────────────────────────────────────────────────────────────┐
│                         HASH-CHAIN                               │
├──────────────────────────────────────────────────────────────────┤
│                                                                   │
│  Entry #1          Entry #2          Entry #3                    │
│  ┌──────────┐      ┌──────────┐      ┌──────────┐               │
│  │ Data     │      │ Data     │      │ Data     │               │
│  │ prev: ∅  │─────▶│ prev: H1 │─────▶│ prev: H2 │               │
│  │ hash: H1 │      │ hash: H2 │      │ hash: H3 │               │
│  └──────────┘      └──────────┘      └──────────┘               │
│                                                                   │
│  Wenn Entry #2 manipuliert wird:                                 │
│  → H2' ≠ H2                                                      │
│  → Entry #3.prev (H2) ≠ H2'                                      │
│  → Kette ist gebrochen!                                          │
│                                                                   │
└──────────────────────────────────────────────────────────────────┘

Chain-Verifikation

<?php

final class AuditChainVerifier
{
    public function __construct(
        private PDO $db,
    ) {}

    public function verify(string $tableName = null): VerificationResult
    {
        // WICHTIG: BYTEA als hex selektieren! PDO liefert BYTEA sonst als \x... String,
        // der nicht direkt mit hash('sha256', ...) vergleichbar ist.
        $sql = "SELECT id, table_name, record_id, action,
                       old_values, new_values, created_at,
                       encode(prev_hash, 'hex') AS prev_hash,
                       encode(row_hash, 'hex') AS row_hash
                FROM audit_log";

        if ($tableName) {
            $sql .= " WHERE table_name = :table";
        }
        $sql .= " ORDER BY id ASC";

        $stmt = $this->db->prepare($sql);
        $stmt->execute($tableName ? ['table' => $tableName] : []);

        $errors = [];
        $lastHash = null;  // hex string oder null
        $count = 0;

        while ($row = $stmt->fetch(PDO::FETCH_ASSOC)) {
            $count++;

            // 1. Prev-Hash prüfen (beide sind jetzt hex strings)
            if ($row['prev_hash'] !== $lastHash) {
                $errors[] = [
                    'id' => $row['id'],
                    'type' => 'CHAIN_BROKEN',
                    'message' => "prev_hash mismatch at entry {$row['id']}",
                ];
            }

            // 2. Row-Hash verifizieren (neu berechnen)
            $expectedHash = $this->calculateHash(
                $lastHash,
                $row['table_name'],
                $row['record_id'],
                $row['action'],
                $row['old_values'],
                $row['new_values'],
                $row['created_at']
            );

            if ($row['row_hash'] !== $expectedHash) {
                $errors[] = [
                    'id' => $row['id'],
                    'type' => 'HASH_MISMATCH',
                    'message' => "Data integrity violation at entry {$row['id']}",
                ];
            }

            $lastHash = $row['row_hash'];
        }

        return new VerificationResult(
            valid: count($errors) === 0,
            entriesChecked: $count,
            errors: $errors,
        );
    }

    private function calculateHash(
        ?string $prevHash,
        string $tableName,
        string $recordId,
        string $action,
        ?string $oldValues,
        ?string $newValues,
        string $createdAt,
    ): string {
        // Muss exakt dem Trigger entsprechen!
        $input = ($prevHash ?? 'GENESIS') .
                 $tableName .
                 $recordId .
                 $action .
                 ($oldValues ?? '') .
                 ($newValues ?? '') .
                 $createdAt;

        // OHNE true → gibt hex string zurück (konsistent mit encode(..., 'hex'))
        return hash('sha256', $input);
    }
}

Umgang mit sensiblen Daten

Problem: Audit-Trail speichert alles – aber was ist mit Passwörtern, Kreditkarten, personenbezogenen Daten?

Strategie: Selective Field Masking

<?php

final class AuditFieldFilter
{
    // Felder, die NIE geloggt werden
    private const NEVER_LOG = [
        'password',
        'password_hash',
        'credit_card_number',
        'cvv',
        'secret_key',
        'api_secret',
    ];

    // Felder, die maskiert werden (Wert wird gehasht oder reduziert)
    private const MASK_FIELDS = [
        'email' => 'partial',      // max***@***.com
        'phone' => 'partial',      // +49 170 ***1234
        'iban' => 'partial',       // DE89 **** **** 1234
        'ip_address' => 'hash',    // SHA256
    ];

    public function filter(array $values): array
    {
        $filtered = [];

        foreach ($values as $key => $value) {
            if (in_array($key, self::NEVER_LOG, true)) {
                $filtered[$key] = '[REDACTED]';
                continue;
            }

            if (isset(self::MASK_FIELDS[$key])) {
                $filtered[$key] = $this->mask($value, self::MASK_FIELDS[$key]);
                continue;
            }

            $filtered[$key] = $value;
        }

        return $filtered;
    }

    private function mask(mixed $value, string $method): string
    {
        if ($value === null) {
            return '[NULL]';
        }

        return match ($method) {
            'hash' => hash('sha256', (string) $value),
            'partial' => $this->partialMask((string) $value),
            default => '[MASKED]',
        };
    }

    private function partialMask(string $value): string
    {
        $len = strlen($value);
        if ($len <= 4) {
            return str_repeat('*', $len);
        }

        $visible = (int) ceil($len * 0.3);
        $start = substr($value, 0, $visible);
        $end = substr($value, -$visible);

        return $start . str_repeat('*', $len - 2 * $visible) . $end;
    }
}

DSGVO: Löschung vs. Revisionssicherheit

Konflikt: DSGVO Art. 17 (Recht auf Löschung) vs. GoBD (10 Jahre Aufbewahrung).

Problem mit UPDATE: Ein UPDATE audit_log würde die Hash-Chain zerstören und widerspricht dem "append-only" Prinzip! Zwei saubere Lösungen:

Lösung A: PII von Anfang an maskiert speichern

// Im Trigger/vor dem Speichern: PII nur gehasht speichern
$auditData['user_email'] = hash('sha256', $user->email . $salt);
$auditData['ip_address'] = hash('sha256', $request->ip . $salt);

// → Keine echten PII im Audit-Log = kein Löschproblem!

Lösung B: Redaction-Overlay (append-only bleibt erhalten)

-- Separate Tabelle für Redactions (NICHT die Originaldaten ändern!)
CREATE TABLE audit_redactions (
    id BIGSERIAL PRIMARY KEY,
    user_id UUID NOT NULL,
    redacted_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    reason VARCHAR(255) NOT NULL,  -- 'GDPR Art. 17'
    affected_entries BIGINT[]      -- Optional: welche audit_log IDs
);

-- View, die beim Anzeigen die Redaction "überlagert"
CREATE VIEW audit_log_display AS
SELECT
    a.id, a.table_name, a.record_id, a.action,
    a.old_values, a.new_values, a.created_at,
    CASE WHEN r.id IS NOT NULL THEN '[REDACTED]' ELSE a.user_email END as user_email,
    CASE WHEN r.id IS NOT NULL THEN NULL ELSE a.ip_address END as ip_address,
    a.user_id,
    a.row_hash  -- Hash bleibt original → Chain intakt!
FROM audit_log a
LEFT JOIN audit_redactions r ON a.user_id = r.user_id;

<?php

final class GdprCompliantAuditLog
{
    public function anonymizeUser(string $userId, string $reason): void
    {
        // NICHT die audit_log Tabelle ändern!
        // Stattdessen: Redaction-Eintrag (append-only)
        $sql = "INSERT INTO audit_redactions (user_id, reason)
                VALUES (:user_id, :reason)";

        $this->db->prepare($sql)->execute([
            'user_id' => $userId,
            'reason' => $reason,
        ]);

        // Und die Redaction selbst als Audit-Event loggen
        $this->audit->log(
            action: 'GDPR_REDACTION',
            entityType: 'User',
            entityId: $userId,
            extra: ['reason' => $reason],
        );
    }
}

Vorteil: Die Original-Hashes bleiben gültig, die Chain ist verifizierbar, aber die UI zeigt [REDACTED]. Prüfer sehen: "Eintrag existiert, wurde aber DSGVO-konform maskiert."

Typische Audit-Abfragen

-- Alle Änderungen an einer Rechnung
SELECT action, old_values, new_values, user_email, created_at
FROM audit_log
WHERE table_name = 'invoices' AND record_id = 'abc-123'
ORDER BY created_at DESC;

-- Wer hat in den letzten 24h Preise geändert?
SELECT DISTINCT user_email, COUNT(*) as changes
FROM audit_log
WHERE table_name = 'invoices'
  AND action = 'UPDATE'
  AND changed_fields @> ARRAY['price_cents']
  AND created_at > NOW() - INTERVAL '24 hours'
GROUP BY user_email;

-- Zustand einer Rechnung zu einem bestimmten Zeitpunkt rekonstruieren
WITH history AS (
    SELECT
        new_values,
        created_at,
        ROW_NUMBER() OVER (ORDER BY created_at DESC) as rn
    FROM audit_log
    WHERE table_name = 'invoices'
      AND record_id = 'abc-123'
      AND created_at <= '2025-10-15 14:00:00+02'
)
SELECT new_values FROM history WHERE rn = 1;

-- Chain-Integrität prüfen (einfacher Check)
SELECT
    a.id,
    CASE
        WHEN a.prev_hash = LAG(a.row_hash) OVER (ORDER BY a.id)
        THEN 'OK'
        ELSE 'BROKEN'
    END as chain_status
FROM audit_log a
WHERE table_name = 'invoices'
ORDER BY id;

Audit-Trail im UI darstellen

<?php

final class AuditLogPresenter
{
    public function formatForUI(array $entry): array
    {
        $changes = $this->diffValues(
            json_decode($entry['old_values'] ?? '{}', true),
            json_decode($entry['new_values'] ?? '{}', true)
        );

        return [
            'id' => $entry['id'],
            'action' => $this->translateAction($entry['action']),
            'timestamp' => $this->formatTimestamp($entry['created_at']),
            'user' => $entry['user_email'] ?? 'System',
            'changes' => $changes,
            'summary' => $this->generateSummary($entry, $changes),
        ];
    }

    private function diffValues(array $old, array $new): array
    {
        $changes = [];

        foreach ($new as $field => $newValue) {
            $oldValue = $old[$field] ?? null;

            if ($oldValue !== $newValue) {
                $changes[] = [
                    'field' => $this->translateField($field),
                    'old' => $this->formatValue($oldValue),
                    'new' => $this->formatValue($newValue),
                ];
            }
        }

        return $changes;
    }

    private function generateSummary(array $entry, array $changes): string
    {
        $user = $entry['user_email'] ?? 'System';
        $action = $this->translateAction($entry['action']);

        if (count($changes) === 0) {
            return "{$user} hat {$action}";
        }

        if (count($changes) === 1) {
            $field = $changes[0]['field'];
            return "{$user} hat {$field} geändert";
        }

        return "{$user} hat " . count($changes) . " Felder geändert";
    }

    private function translateAction(string $action): string
    {
        return match ($action) {
            'INSERT' => 'erstellt',
            'UPDATE' => 'geändert',
            'DELETE' => 'gelöscht',
            default => $action,
        };
    }

    private function translateField(string $field): string
    {
        return [
            'price_cents' => 'Preis',
            'status' => 'Status',
            'customer_id' => 'Kunde',
            // ...
        ][$field] ?? $field;
    }
}

Performance bei großen Audit-Logs

Partitionierung

-- Partitionierung nach Monat
CREATE TABLE audit_log (
    id BIGSERIAL,
    table_name VARCHAR(100) NOT NULL,
    record_id UUID NOT NULL,
    action VARCHAR(10) NOT NULL,
    old_values JSONB,
    new_values JSONB,
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    user_id UUID,
    prev_hash BYTEA,
    row_hash BYTEA NOT NULL,
    PRIMARY KEY (id, created_at)
) PARTITION BY RANGE (created_at);

-- Partitionen erstellen
CREATE TABLE audit_log_2025_10 PARTITION OF audit_log
    FOR VALUES FROM ('2025-10-01') TO ('2025-11-01');

CREATE TABLE audit_log_2025_11 PARTITION OF audit_log
    FOR VALUES FROM ('2025-11-01') TO ('2025-12-01');

-- Automatisierte Partition-Erstellung (pg_partman oder Cron)

Archivierung alter Einträge

<?php

final class AuditArchiver
{
    public function archiveOldEntries(int $retentionMonths = 24): void
    {
        $cutoff = new DateTimeImmutable("-{$retentionMonths} months");

        // 1. In Cold Storage exportieren (S3, Glacier)
        $entries = $this->fetchOldEntries($cutoff);
        $this->exportToStorage($entries, $cutoff->format('Y-m'));

        // 2. Hash-Chain Endpunkt dokumentieren
        $lastEntry = end($entries);
        $this->recordArchiveCheckpoint($lastEntry['row_hash'], $cutoff);

        // 3. Aus Hot-Tabelle löschen
        $this->deleteArchivedEntries($cutoff);
    }

    private function exportToStorage(array $entries, string $period): void
    {
        $filename = "audit-archive-{$period}.json.gz";

        $json = json_encode($entries, JSON_THROW_ON_ERROR);
        $compressed = gzencode($json, 9);

        $this->s3->putObject([
            'Bucket' => 'my-audit-archive',
            'Key' => $filename,
            'Body' => $compressed,
            'StorageClass' => 'GLACIER_IR',  // Instant Retrieval
        ]);
    }
}

Audit-Trails testen

<?php

final class AuditTrailTest extends TestCase
{
    public function testUpdateCreatesAuditEntry(): void
    {
        // Arrange
        $invoice = $this->createInvoice(['price_cents' => 10000]);

        // Act
        $this->invoiceService->updatePrice($invoice->id, 12000, 'Anpassung');

        // Assert
        $auditEntries = $this->auditRepo->getForEntity('invoices', $invoice->id);

        $this->assertCount(2, $auditEntries); // INSERT + UPDATE

        $lastEntry = end($auditEntries);
        $this->assertEquals('UPDATE', $lastEntry['action']);
        $this->assertEquals(10000, json_decode($lastEntry['old_values'])->price_cents);
        $this->assertEquals(12000, json_decode($lastEntry['new_values'])->price_cents);
    }

    public function testAuditChainIntegrity(): void
    {
        // Arrange: Mehrere Änderungen durchführen
        $invoice = $this->createInvoice(['status' => 'draft']);
        $this->invoiceService->updateStatus($invoice->id, 'sent');
        $this->invoiceService->updateStatus($invoice->id, 'paid');

        // Act
        $result = $this->chainVerifier->verify('invoices');

        // Assert
        $this->assertTrue($result->valid);
        $this->assertEquals(3, $result->entriesChecked);
        $this->assertEmpty($result->errors);
    }

    public function testSensitiveFieldsAreMasked(): void
    {
        // Arrange
        $user = $this->createUser(['email' => 'secret@example.com']);

        // Act
        $this->userService->updatePassword($user->id, 'newpassword123');

        // Assert
        $entry = $this->auditRepo->getLatestForEntity('users', $user->id);

        $newValues = json_decode($entry['new_values'], true);
        $this->assertEquals('[REDACTED]', $newValues['password_hash']);
    }
}

Checkliste für Revisionssicherheit

┌─────────────────────────────────────────────────────────────────┐
│               REVISIONSSICHERHEIT CHECKLISTE                     │
├─────────────────────────────────────────────────────────────────┤
│                                                                  │
│  □ Audit-Tabelle ist append-only (kein UPDATE/DELETE)           │
│  □ Hash-Chain implementiert und verifizierbar                   │
│  □ Timestamps in UTC mit Zeitzone                               │
│  □ User-Context bei jedem Eintrag (wer, woher)                  │
│  □ Sensible Felder maskiert/redacted                            │
│  □ Partitionierung für Performance                              │
│  □ Archivierungsstrategie definiert                             │
│  □ Restore-Prozess getestet                                     │
│  □ Monitoring für Chain-Integrität                              │
│  □ Dokumentation für Prüfer/Auditoren                           │
│                                                                  │
└─────────────────────────────────────────────────────────────────┘

Fazit

Audit-Trails sind kein "Nice-to-have", sondern Grundausstattung jeder ernsthaften Business-App:

Trigger-basiert: Einfachster Start, transparent für die Applikation
Event Sourcing: Maximale Flexibilität, Events sind die Source of Truth
Hash-Chain: Macht Manipulation erkennbar – die "Blockchain für Arme"
Append-only: Niemals UPDATE, niemals DELETE auf Audit-Daten
DSGVO-Konflikt: Anonymisieren statt Löschen, Anonymisierung selbst loggen

Meine Empfehlung: Starte mit Database Triggers + Hash-Chain. Das funktioniert für 90% der Fälle. Event Sourcing nur wenn du es wirklich brauchst (komplexe Domäne, Event-Replay, CQRS).

Weiterführende Ressourcen

Über Carola Schulte

Software-Architektin mit 25+ Jahren Erfahrung. Spezialisiert auf robuste Business-Apps mit PHP/PostgreSQL, Security-by-Design und DSGVO-konforme Systeme. 1,8M+ Lines of Code in Produktion.

die@entwicklerin.net entwicklerin.net

Revisionssichere Systeme benötigt?

Lassen Sie uns besprechen, wie Ihre Business-App GoBD-konform und prüfungssicher wird – kostenlos und unverbindlich.

Kostenloses Erstgespräch

Impressum Datenschutz

Teilen

Inhaltsverzeichnis