ARC-08: verschluesselungsschluessel-rotation

Tenant-KEK-Rotation ohne Neuverschlüsselung des Archivbestands
(Envelope-Encryption bleibt aus ARC-02 unverändert, Objekt-DEKs werden
nicht angefasst).

Core (API-10, RotateTenantKEK) ersetzt den Tenant-KEK durch einen neuen
Wert und hält keine Historie vor — TenantKEKHandler liefert immer nur den
aktuellen Schlüssel. Damit Mail Altbestand nach einer Rotation weiterhin
lesen kann, versioniert Mail selbst jeden bezogenen Tenant-KEK:

- crypto/kekversions.go: KEKVersionStore, lokal verschlüsselt mit
  eigenem Wrap-Schlüssel (nur über Umgebungsvariable), erkennt Rotation
  automatisch (RecordIfNew), erlaubt gezieltes Sperren einer Version
  (Revoke).
- crypto/service.go: Service.WithVersionStore (optional, Open bleibt für
  Rückwärtskompatibilität unverändert), Seal zeichnet die verwendete
  KEK-Version auf, neue Methode OpenAtVersion liest mit historischer
  statt aktueller Version.
- encstorage.go: neuer .dek.version-Sidecar (gleiches Muster wie der
  bestehende .dek-Sidecar), GetDecrypted nutzt OpenAtVersion; fehlender
  Sidecar (Altobjekte vor ARC-08) fällt auf Version 0 zurück, identisches
  Verhalten wie vorher.

Prüfungen (alle real durchgeführt, siehe mail/docs/ARC-08-PRUEFPROTOKOLL.md):
1. TestRotation_OldArchiveStaysReadableAfterMasterKeyRotation: Altbestand
   nach realer Rotation weiterhin lesbar über OpenAtVersion, naives Open
   mit dem neuen Schlüssel schlägt für das alte Objekt real fehl.
2. TestRotation_CompromisedOldKeyCanBeRevoked: gesperrte Version blockiert
   Lesezugriff real, andere Versionen bleiben unberührt.
3. Rotationsvorgang vollständig durchgespielt (siehe Prüfprotokoll).

Kein Umbau: storage/dedup/indexworker/search unverändert, bestehende
ARC-02-Tests (encstorage_test.go) unverändert weiterhin grün.

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

mail/internal/crypto/kekversions.go

@@ -0,0 +1,144 @@
+// ARC-08: Tenant-KEK-Rotation. Core (API-10, `internal/kek.Store.
+// RotateTenantKEK`) ersetzt den Tenant-KEK durch einen komplett neuen
+// Wert — Core selbst hält KEINE Historie vor, `TenantKEKHandler` liefert
+// immer nur den AKTUELLEN Schlüssel (siehe kekprovider.go). Damit ARC-08s
+// Akzeptanzkriterium 3 ("alte Schlüsselversionen bleiben für
+// Lesezugriff kontrolliert verfügbar") erfüllbar ist, muss Mail selbst
+// jeden von Core bezogenen Tenant-KEK versioniert zwischenspeichern —
+// KEKVersionStore übernimmt genau das, lokal mit einem eigenen,
+// ausschließlich über Umgebungsvariable bezogenen Wrap-Schlüssel
+// verschlüsselt (kein Klartext-KEK in der Datenbank).
+package crypto
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"fmt"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgxpool"
+)
+
+// ErrKEKVersionNotFound wird geliefert, wenn die angefragte Version für
+// den Mandanten nicht existiert.
+var ErrKEKVersionNotFound = errors.New("crypto: kek-version nicht gefunden")
+
+// ErrKEKVersionRevoked wird geliefert, wenn die angefragte Version gezielt
+// gesperrt wurde (Pflichtprüfung 2: kompromittierter alter Schlüssel kann
+// gezielt gesperrt werden) — der Lesezugriff auf mit dieser Version
+// verschlüsselte Altobjekte ist dann bewusst blockiert.
+var ErrKEKVersionRevoked = errors.New("crypto: kek-version wurde gesperrt")
+
+// KEKVersionStore verwaltet die Versionshistorie der Tenant-KEKs, die
+// dieses Mail-Modul im Lauf der Zeit von Core bezogen hat.
+type KEKVersionStore struct {
+	pool         *pgxpool.Pool
+	localWrapKey []byte
+}
+
+// NewKEKVersionStore erzeugt einen Store. localWrapKey verschlüsselt die
+// zwischengespeicherten Tenant-KEKs lokal at rest (KEKSize Bytes,
+// ausschließlich über Umgebungsvariable bezogen — nie im Code).
+func NewKEKVersionStore(pool *pgxpool.Pool, localWrapKey []byte) *KEKVersionStore {
+	return &KEKVersionStore{pool: pool, localWrapKey: localWrapKey}
+}
+
+// EnsureSchema legt die Tabelle an, falls sie noch nicht existiert —
+// gleiches Muster wie mail/internal/dedup/indexworker (kein zentraler
+// Migrationsläufer für Mandanten-Datenbanken im Mail-Modul vorhanden).
+func (s *KEKVersionStore) EnsureSchema(ctx context.Context) error {
+	if _, err := s.pool.Exec(ctx, `
+		CREATE TABLE IF NOT EXISTS mail_kek_versions (
+			tenant_slug  TEXT NOT NULL,
+			version      INT NOT NULL,
+			wrapped_kek  BYTEA NOT NULL,
+			revoked      BOOLEAN NOT NULL DEFAULT false,
+			created_at   TIMESTAMPTZ NOT NULL DEFAULT now(),
+			PRIMARY KEY (tenant_slug, version)
+		)
+	`); err != nil {
+		return fmt.Errorf("crypto: kek-versionsschema anlegen: %w", err)
+	}
+	return nil
+}
+
+// RecordIfNew merkt sich plainKEK als neue Version für tenantSlug, FALLS
+// er sich vom zuletzt gespeicherten Wert unterscheidet (Rotation
+// erkannt) — bei unverändertem KEK wird keine neue Version angelegt,
+// sondern die bestehende Versionsnummer zurückgegeben (Akzeptanzkriterium
+// 1: rotierbar verwaltet, nicht bei jedem Aufruf eine neue Version).
+func (s *KEKVersionStore) RecordIfNew(ctx context.Context, tenantSlug string, plainKEK []byte) (version int, err error) {
+	var latestVersion int
+	var latestWrapped []byte
+	err = s.pool.QueryRow(ctx, `
+		SELECT version, wrapped_kek FROM mail_kek_versions
+		WHERE tenant_slug = $1 ORDER BY version DESC LIMIT 1
+	`, tenantSlug).Scan(&latestVersion, &latestWrapped)
+	switch {
+	case errors.Is(err, pgx.ErrNoRows):
+		return s.insertVersion(ctx, tenantSlug, 1, plainKEK)
+	case err != nil:
+		return 0, fmt.Errorf("crypto: letzte kek-version lesen: %w", err)
+	}
+
+	latestPlain, err := open(s.localWrapKey, latestWrapped)
+	if err != nil {
+		return 0, fmt.Errorf("crypto: zwischengespeicherten kek entpacken: %w", err)
+	}
+	if bytes.Equal(latestPlain, plainKEK) {
+		return latestVersion, nil
+	}
+	return s.insertVersion(ctx, tenantSlug, latestVersion+1, plainKEK)
+}
+
+func (s *KEKVersionStore) insertVersion(ctx context.Context, tenantSlug string, version int, plainKEK []byte) (int, error) {
+	wrapped, err := seal(s.localWrapKey, plainKEK)
+	if err != nil {
+		return 0, fmt.Errorf("crypto: kek für zwischenspeicherung verpacken: %w", err)
+	}
+	if _, err := s.pool.Exec(ctx, `
+		INSERT INTO mail_kek_versions (tenant_slug, version, wrapped_kek) VALUES ($1, $2, $3)
+	`, tenantSlug, version, wrapped); err != nil {
+		return 0, fmt.Errorf("crypto: kek-version speichern: %w", err)
+	}
+	return version, nil
+}
+
+// Get liefert den entschlüsselten historischen Tenant-KEK einer
+// bestimmten Version. Liefert ErrKEKVersionRevoked, wenn die Version
+// gezielt gesperrt wurde (Pflichtprüfung 2).
+func (s *KEKVersionStore) Get(ctx context.Context, tenantSlug string, version int) ([]byte, error) {
+	var wrapped []byte
+	var revoked bool
+	err := s.pool.QueryRow(ctx, `
+		SELECT wrapped_kek, revoked FROM mail_kek_versions
+		WHERE tenant_slug = $1 AND version = $2
+	`, tenantSlug, version).Scan(&wrapped, &revoked)
+	if err != nil {
+		if errors.Is(err, pgx.ErrNoRows) {
+			return nil, ErrKEKVersionNotFound
+		}
+		return nil, fmt.Errorf("crypto: kek-version lesen: %w", err)
+	}
+	if revoked {
+		return nil, ErrKEKVersionRevoked
+	}
+	return open(s.localWrapKey, wrapped)
+}
+
+// Revoke sperrt eine Tenant-KEK-Version gezielt (Pflichtprüfung 2):
+// nachfolgende Get-Aufrufe für genau diese Version schlagen mit
+// ErrKEKVersionRevoked fehl, andere Versionen bleiben unberührt.
+func (s *KEKVersionStore) Revoke(ctx context.Context, tenantSlug string, version int) error {
+	tag, err := s.pool.Exec(ctx, `
+		UPDATE mail_kek_versions SET revoked = true WHERE tenant_slug = $1 AND version = $2
+	`, tenantSlug, version)
+	if err != nil {
+		return fmt.Errorf("crypto: kek-version sperren: %w", err)
+	}
+	if tag.RowsAffected() == 0 {
+		return ErrKEKVersionNotFound
+	}
+	return nil
+}