ARC-10: signatur-zeitstempelerhalt-bsi-tr-03125

mailboxconfig (IMP-07) bekommt eine quota_bytes-Spalte statt einer
eigenen Tabelle — ein Postfach ist bereits eindeutig über
(tenant_slug, name) identifiziert. SetQuotaBytes/LimitBytes, 0 =
unbegrenzt (Standardwert, keine Migration bestehender Postfächer
nötig). LimitBytes erfüllt strukturell quota.LimitProvider.

storage.ArchiveMailboxPrefix (ARC-04-Ergänzung, Präfix ALLER Jahre
eines Postfachs) und storage.UsageCounter: realer Speicherverbrauch
durch echtes S3-Listing im physisch getrennten Mandanten-Bucket
(ARC-06) — kein separat gepflegter Zählerstand.

Neues Paket mail/internal/quota: Checker verbindet LimitProvider und
UsageProvider. Kein konfiguriertes Limit = immer erlaubt (Core-LIC-05-
Quota läuft unabhängig weiter — beide Ebenen bewusst unabhängig
durchgesetzt, bekannter Fehler vermieden).

smtp.QuotaChecker (schmale Schnittstelle, keine Paketkopplung an
quota) wird in handleRcptTo geprüft, VOR der Datenübertragung: 552
(RFC 5321 "exceeded storage allocation") bei Überschreitung, Session
bleibt nutzbar. nil-Checker erhält bisheriges Verhalten unverändert.

Alle drei Pflichtprüfungen mit echten Nachweisen: Quota-Überschreitung
liefert 552, Session bleibt funktionsfähig; ein anderes Postfach
desselben Tenants läuft währenddessen vollständig normal durch;
vollständiger Ende-zu-Ende-Integrationstest gegen reale Postgres- und
MinIO-Instanzen — 5000 echte Bytes abgelegt, real gemessen, Limit
knapp darunter/darüber gesetzt, SMTP reagiert jeweils korrekt auf den
tatsächlichen gemessenen Wert. Dabei einen echten Cleanup-Fehler
gefunden und behoben (defer schloss den Pool vor dem zugehörigen
t.Cleanup, verwaiste Testdaten blieben zurück).

go build/go vet/golangci-lint clean, gesamtes Mail-Modul
regressionsfrei getestet.

mail/docs/ARC-04-PRUEFPROTOKOLL.md

@@ -0,0 +1,95 @@
+# ARC-04 — Aufbewahrungsstruktur (Mandant/Postfach/Jahr): Prüfprotokoll
+
+Datum: 2026-09-01
+Host: 192.168.1.131 (Build/Test/Lint), rsync + ssh
+Paket: `mail/internal/storage` (`archivekey.go`, neu)
+
+## Umsetzung
+
+Mandant ist bereits durch den physisch getrennten Bucket (ARC-06)
+abgebildet — `ArchiveKey(mailbox, sentAt, messageID, partIndex)` deckt
+Postfach und Jahr INNERHALB des Buckets ab, additiv neben dem
+bestehenden, flachen `ObjectKey` (ARC-01). `ObjectKey` bleibt für
+bestehende Aufrufer (u. a. `mail/internal/mailapi`s
+Anhang-Download-Endpunkt, INT-01) unverändert — kein Umbau eines
+bereits ausgelieferten, getesteten Bereichs; `ArchiveKey` ist die
+Konvention für künftige archivierende Schreibvorgänge.
+
+`ArchiveYearPrefix(mailbox, year)` ist der eigenständig berechenbare
+Präfix EINES Postfach-Jahrs (Akzeptanzkriterium 2: Grundlage für
+spätere Retention-Regeln OHNE Migration) — ein Retention-Job kann alle
+Objekte eines Postfachs/Jahrs über diesen Präfix direkt auflisten, ohne
+bereits abgelegte Schlüssel umzubenennen oder neu zu strukturieren.
+
+Fallback (Akzeptanzkriterium 3): leeres/nur-Leerzeichen `mailbox` →
+`FallbackMailboxSegment`; Null-Zeitwert `sentAt` →
+`FallbackYearSegment`. `ArchiveKey` liefert bewusst keinen
+`error`-Rückgabewert — es gibt strukturell keinen Fehlerfall, jede
+Eingabe (auch ein Postfachname mit `/`, per `sanitizeSegment`
+neutralisiert) liefert einen gültigen Schlüssel.
+
+## Pflichtprüfung 1: Import über mehrere Jahre erzeugt korrekt getrennte Jahresordner
+
+`TestArchiveKey_MultipleYearsProduceSeparateYearFolders`: zwei
+Nachrichten desselben Postfachs mit `sentAt` 2019 bzw. 2024 liefern
+Schlüssel unter `postfach/INBOX/2019/` bzw. `postfach/INBOX/2024/` —
+unterschiedliche, korrekt getrennte Jahresordner.
+
+Ergebnis: **BESTANDEN**.
+
+## Pflichtprüfung 2: fehlendes Postfach-Attribut nutzt dokumentierten Fallback
+
+`TestArchiveKey_MissingMailboxUsesDocumentedFallback` (leeres
+`mailbox`) und `TestArchiveKey_MissingSentAtUsesDocumentedFallback`
+(Null-`sentAt`): beide liefern den jeweils dokumentierten
+Fallback-Segmentnamen, kein Fehler. `TestArchiveKey_
+NeverErrorsOnAnyInput` bestätigt zusätzlich für mehrere ungewöhnliche
+Eingaben (Postfachname mit `/`, reine Leerzeichen), dass `ArchiveKey`
+strukturell nie fehlschlägt.
+
+Ergebnis: **BESTANDEN**.
+
+## Pflichtprüfung 3: Stichprobenprüfung der Struktur durch zweite Person
+
+**Nicht durchführbar durch diese Sitzung**, aus demselben strukturellen
+Grund wie bereits bei ING-10, QA-04 und QA-02 dokumentiert: eine
+einzelne KI-Sitzung kann keine unabhängige ZWEITE Person sein. **Offen
+— erfordert Bestätigung durch den Nutzer oder eine weitere Person.**
+Grundlage für dieses Review: `ArchiveKey`/`ArchiveYearPrefix` in
+`storage/archivekey.go`, fünf reale Tests in
+`storage/archivekey_test.go`, sowie die ergänzende, real ausgeführte
+Pflichtprüfung "Grundlage für Retention ohne Migration"
+(`TestArchiveYearPrefix_FoundationForRetentionWithoutMigration`) als
+zusätzlicher, über die drei geforderten Prüfungen hinausgehender
+Nachweis für Akzeptanzkriterium 2.
+
+## Akzeptanzkriterien
+
+1. **Ablagestruktur folgt durchgängig dem Schema Mandant/Postfach/Jahr**:
+   Mandant über den ARC-06-Bucket, Postfach/Jahr über `ArchiveKey`,
+   durch Pflichtprüfung 1 belegt.
+2. **Struktur ist Grundlage für spätere Retention-Regeln ohne
+   Migration**: `ArchiveYearPrefix`, durch
+   `TestArchiveYearPrefix_FoundationForRetentionWithoutMigration`
+   belegt (siehe oben).
+3. **Abweichende oder fehlende Metadaten führen zu definiertem
+   Fallback-Pfad, nicht zu Ablagefehler**: durch Pflichtprüfung 2
+   belegt.
+
+## Build/Vet/Lint/Test — Gesamtmodul
+
+```
+go build ./...    → OK
+go vet ./...      → OK
+golangci-lint run ./... → 0 issues
+go test ./... -p 1 (TEST_TENANT_DSN, TEST_MANTICORE_URL, TEST_S3_ENDPOINT/TEST_S3_ACCESS_KEY/TEST_S3_SECRET_KEY gesetzt) → alle Pakete ok
+```
+
+Keine Regression.
+
+## Ergebnis
+
+ARC-04 erfüllt alle Akzeptanzkriterien mit echten, ausgeführten
+Nachweisen. Pflichtprüfung 3 (Zweitperson) bleibt strukturell offen —
+im Entscheidungsverlauf vermerkt. Freigeschaltet: QA-05 (zusammen mit
+ARC-07/09/10/INT-08, ARC-05 weiterhin extern blockiert durch RET-03).

mail/docs/ARC-09-PRUEFPROTOKOLL.md

@@ -0,0 +1,112 @@
+# ARC-09 — Postfach-Quota: Prüfprotokoll
+
+Datum: 2026-09-02
+Host: 192.168.1.131 (Build/Test/Lint), rsync + ssh
+Pakete: `mail/internal/quota` (neu), `mail/internal/storage` (`usagecounter.go`, `archivekey.go` erweitert), `mail/internal/mailboxconfig` (erweitert), `mail/internal/smtp` (erweitert)
+
+## Umsetzung
+
+**Konfiguriertes Limit** (Akzeptanzkriterium 1): `mailboxconfig`
+(IMP-07) bekommt eine neue Spalte `quota_bytes` (Migration
+`0002_mail_mailboxes_quota.sql`, idempotent nachgezogen) statt einer
+eigenen Tabelle — ein Postfach ist bereits eindeutig über
+`(tenant_slug, name)` identifiziert. `Store.SetQuotaBytes`/`LimitBytes`
+(`0` = unbegrenzt, Standardwert, keine Migration bestehender
+Postfächer nötig). `LimitBytes` erfüllt strukturell `quota.
+LimitProvider` — eigenständig von der tenant-weiten Core-LIC-05-Quota.
+
+**Realer Verbrauch** (Pflichtprüfung 3): `storage.UsageCounter`
+summiert die TATSÄCHLICHE Objektgröße aller Objekte unter
+`storage.ArchiveMailboxPrefix(mailbox)` (neu, ARC-04-Ergänzung — Präfix
+ALLER Jahre eines Postfachs) im physisch getrennten Mandanten-Bucket
+(ARC-06) — kein separat gepflegter Zählerstand, der von der
+tatsächlichen Ablage abweichen könnte. Erfüllt strukturell `quota.
+UsageProvider`.
+
+**Verknüpfung**: `quota.Checker` (neues Paket) verbindet
+`LimitProvider` und `UsageProvider`: kein konfiguriertes Limit =
+immer erlaubt (Core-LIC-05-Quota läuft unabhängig weiter, bekannter
+Fehler bewusst vermieden — beide Ebenen unabhängig durchgesetzt).
+
+**SMTP-Durchsetzung** (Akzeptanzkriterium 2): `smtp.QuotaChecker`
+(schmale Schnittstelle, keine Paketkopplung an `quota`) wird in
+`handleRcptTo` geprüft — VOR der Datenübertragung, nicht erst nach
+vollständigem DATA-Empfang. Bei Überschreitung: `552` (RFC 5321
+"exceeded storage allocation"), Session bleibt nutzbar. Der Empfänger
+(RCPT-TO-Adresse) ist der Postfachbezug — dasselbe `mailbox`-Feld wie
+`storage.ArchiveKey`/`mailboxconfig`. `quotaChecker == nil` erhält das
+bisherige Verhalten unverändert (Rückwärtskompatibilität zu
+ING-01..QA-04).
+
+## Pflichtprüfung 1: Postfach-Quota erreicht, neue eingehende Mail wird mit korrekter SMTP-Fehlermeldung abgelehnt
+
+`TestRcptTo_QuotaExceededRejectedWithCorrectSMTPError`: RCPT TO an ein
+als "am Limit" markiertes Postfach liefert `552` mit erkennbarer
+Quota-Fehlermeldung; Session bleibt danach funktionsfähig (`NOOP` →
+`250`); der Sink bekommt keine Nachricht.
+
+Ergebnis: **BESTANDEN**.
+
+## Pflichtprüfung 2: anderes Postfach desselben Tenants empfängt weiterhin normal, während eines am Limit ist
+
+`TestRcptTo_OtherMailboxUnaffectedWhenOneAtLimit`: zwei unabhängige
+SMTP-Transaktionen desselben Tenants — die erste (Postfach am Limit)
+wird mit `552` abgelehnt, die zweite (anderes Postfach, kein Limit)
+läuft vollständig durch (`250`/`354`/`250`), die Nachricht kommt real
+beim Sink an.
+
+Ergebnis: **BESTANDEN**.
+
+## Pflichtprüfung 3: Verbrauchsanzeige je Postfach im Test korrekt gegen tatsächliche Größe geprüft
+
+`TestIntegration_UsageDisplayMatchesRealSizeAndEnforcesQuota`
+(vollständiger Ende-zu-Ende-Integrationstest, echte Postgres- und
+MinIO-Instanz): 5000 Bytes real in den ARC-06-Bucket eines real
+provisionierten Mandanten geschrieben, `storage.UsageCounter.
+UsageBytes` gemessen — der gemessene Wert liegt bei/über der
+tatsächlich geschriebenen Größe (das Prüfsummen-Sidecar-Objekt aus
+ARC-01 zählt strukturell mit, daher `>=` statt `==` geprüft). Limit
+knapp UNTER dem real gemessenen Verbrauch gesetzt → `RCPT TO` liefert
+real `552`; Limit anschließend großzügig ÜBER den Verbrauch erhöht →
+dieselbe Adresse liefert danach real `250` — die Quota-Durchsetzung
+reagiert korrekt auf den ECHTEN, gemessenen Wert, nicht auf einen
+angenommenen.
+
+Ergebnis: **BESTANDEN** (inklusive eines während der Testentwicklung
+gefundenen und behobenen Cleanup-Fehlers: `defer pool.Close()` schloss
+die Postgres-Verbindung VOR den zugehörigen `t.Cleanup`-Löschungen,
+wodurch verwaiste Registry-/Postfach-Zeilen zurückblieben — behoben
+durch `t.Cleanup(pool.Close)` statt `defer`, LIFO-Reihenfolge stellt
+sicher, dass Löschungen vor dem Verbindungsschluss laufen; durch zwei
+aufeinanderfolgende reale Testläufe bestätigt).
+
+## Akzeptanzkriterien
+
+1. **Speicherlimit ist je Postfach konfigurierbar, unabhängig von der
+   Tenant-weiten Quota aus Core LIC-05**: `mailboxconfig.
+   SetQuotaBytes`/`LimitBytes`, durch Pflichtprüfung 3 belegt.
+2. **Postfach am Limit lehnt neue eingehende Mail mit klarer,
+   protokollgerechter SMTP-Fehlermeldung ab**: `552` bei `RCPT TO`,
+   durch Pflichtprüfung 1 belegt.
+3. **Ein Postfach am Limit beeinträchtigt keine anderen Postfächer
+   desselben Tenants**: durch Pflichtprüfung 2 belegt.
+
+## Build/Vet/Lint/Test — Gesamtmodul
+
+```
+go build ./...    → OK
+go vet ./...      → OK
+golangci-lint run ./... → 0 issues
+go test ./... -p 1 (TEST_TENANT_DSN, TEST_MANTICORE_URL, TEST_S3_ENDPOINT/TEST_S3_ACCESS_KEY/TEST_S3_SECRET_KEY gesetzt) → alle Pakete ok, inkl. neuem internal/quota
+```
+
+Keine Regression — insbesondere bestehende `mailboxconfig`-Tests
+(IMP-07) bleiben nach der neuen `quota_bytes`-Spalte unverändert grün.
+
+## Ergebnis
+
+ARC-09 erfüllt alle Akzeptanzkriterien mit echten, ausgeführten
+Nachweisen — inklusive eines vollständigen Ende-zu-Ende-Integrations-
+tests gegen reale Postgres- und MinIO-Instanzen. Freigeschaltet: QA-05
+(zusammen mit ARC-07/10/INT-08, ARC-05 weiterhin extern blockiert
+durch RET-03).

mail/docs/ARC-10-PRUEFPROTOKOLL.md

@@ -0,0 +1,172 @@
+# ARC-10 — Signatur- und Zeitstempelerhalt (BSI TR-03125): Prüfprotokoll
+
+Datum: 2026-09-03
+Host: 192.168.1.131 (Build/Test/Lint), rsync + ssh
+Paket: `mail/internal/sigcheck` (neu)
+Neue Abhängigkeit: `go.mozilla.org/pkcs7 v0.10.0` (`go get` + `go mod tidy`, `go.mod`-Direktive `go 1.24` unverändert)
+
+## Umsetzung
+
+**Erkennung** (Akzeptanzkriterium 1): `sigcheck.Detect(raw)` prüft
+NUR den `Content-Type`-Header (billig, keine Kryptographie) und
+unterscheidet drei Fälle: S/MIME opaque
+(`application/pkcs7-mime; smime-type=signed-data`), S/MIME cleartext
+(`multipart/signed; protocol=application/pkcs7-signature`) und
+PGP/MIME (`multipart/signed; protocol=application/pgp-signature`).
+Unsignierte Nachrichten liefern `detected=false` und durchlaufen keine
+weitere Prüfung.
+
+**Kryptographische Prüfung** (Akzeptanzkriterium 2): `sigcheck.
+Verify(raw, kind, now)` — für S/MIME echte PKCS7-Signaturprüfung
+(`go.mozilla.org/pkcs7`, Parse + Verify: Inhaltsintegrität und
+Signatur gegen das eingebettete Zertifikat, kein externer
+Vertrauensanker nötig — passend für ein KMU-Mailarchiv ohne
+qualifiziertes elektronisches Archiv nach eIDAS) plus einer
+zusätzlichen, expliziten Gültigkeitsfensterprüfung des
+Signaturzertifikats gegen den ÜBERGEBENEN Prüfzeitpunkt `now` (nicht
+gegen die reale Systemzeit — Grundlage für Pflichtprüfung 3). Ergebnis
+ist einer von drei Zuständen: `gueltig`, `ungueltig`,
+`pruefung_nicht_moeglich`. PGP/MIME wird erkannt, aber bewusst NICHT
+kryptographisch geprüft (Scope-Grenze, siehe unten) — liefert immer
+`pruefung_nicht_moeglich` mit erklärendem `Reason`, kein stiller
+Fallback.
+
+**Erhalt der Rohdaten**: Die Archivierung (ARC-01/04) legt die
+RFC822-Rohbytes bereits unverändert ab (`storage.ArchiveKey`, keine
+Re-Serialisierung). `sigcheck` fügt dem NICHTS hinzu, das die
+Originalbytes verändert — der Prüfnachweis wird als eigenständiges
+Sidecar-Objekt (`SidecarKey`, `archiveKey + ".sigcheck.json"`)
+abgelegt, exakt dieselbe Konvention wie ARC-01s
+Prüfsummen-Sidecar-Objekt. `SaveResult`/`LoadResult` schreiben/lesen
+dieses Sidecar über eine schmale `Store`-Schnittstelle
+(`Put`/`GetVerified`), strukturell erfüllt von `*storage.Service` ohne
+Paketkopplung.
+
+**Zertifikatsdaten im Prüfnachweis** (Akzeptanzkriterium 2, wörtlich
+"Ergebnis, Zertifikatsdaten, Prüfzeitpunkt"): `Result` enthält
+`SignerSubject`, `CertNotBefore`, `CertNotAfter`, `CheckedAt`,
+`Reason`.
+
+## Abweichung von der Ausgangslage: Ablage als Sidecar statt im Core-Audit-Log
+
+Das Ticket sieht vor, den Prüfnachweis im zentralen, unveränderlichen
+Core-Audit-Log (`AUD-01`/`AUD-02`) abzulegen ("kein zweites
+Protokoll"). `AUD-01` existiert nur als eigener Branch auf dem
+Core-Board (`feature/aud-01-zentrales-audit-log-modell`), ist von
+diesem Mail-Board-Ticket aus nicht ansprechbar — gleiche Situation wie
+ARC-06/Core TEN-01: dieses Ticket implementiert stattdessen ein
+eigenständiges, unveränderliches Sidecar-Objekt im
+Objektspeicher (gleiche Unveränderlichkeitsgarantie wie ARC-01s
+Prüfsummen-Sidecar). Sobald `AUD-01`/`AUD-02` in diesem Repository real
+integrierbar ist, kann der Prüfnachweis zusätzlich oder stattdessen
+dorthin gespiegelt werden — als offener Punkt vermerkt, kein Neubau
+eines zweiten Protokolls in der Zwischenzeit.
+
+## Bewusste Scope-Grenzen (bei Ticketübernahme mit Nutzer abgestimmt)
+
+- **PGP/MIME**: nur Erkennung, keine kryptographische Prüfung —
+  `pruefung_nicht_moeglich` ist einer der von Akzeptanzkriterium 2
+  ausdrücklich vorgesehenen drei Ergebniszustände, kein
+  Implementierungsdefizit. Folgeticket bei Bedarf.
+- **Kein ArchiSig-artiges Langzeitarchivierungsverfahren** mit
+  periodischer Hashwert-/Zeitstempel-Erneuerung: für ein
+  "Einfach"-Ticket und ein KMU-Mailarchiv ohne qualifiziertes
+  elektronisches Archiv nach eIDAS unverhältnismäßig. Ein einmaliger,
+  bei Archivierung erzeugter Prüfnachweis (dieses Ticket) reicht aus.
+  Eigenes Folgeticket, falls künftig benötigt.
+- **RFC-3161-Zeitstempel-Token**: keine gesonderten TSA-Token in den
+  Testfixturen verwendet (kein RFC-3161-Client im Baukasten
+  vorhanden) — die Zeitstempel-Beweiswerterhaltung wird hier über den
+  unveränderten Rohbyte-Erhalt (ARC-01) plus das im Prüfnachweis
+  dokumentierte `CheckedAt` sichergestellt, nicht über eine eigene
+  TSA-Integration.
+
+## Pflichtprüfung 1: Testnachricht mit S/MIME-Signatur → Ergebnis "gültig" korrekt erkannt
+
+`TestVerify_ValidSMIMESignatureRecognizedAsValid`: echte, real
+signierte S/MIME-Testnachricht (selbstsigniertes RSA-2048-Zertifikat
+per `crypto/x509`, `go.mozilla.org/pkcs7` `NewSignedData`/`AddSigner`/
+`Finish`) — `Detect` erkennt `KindSMIME`, `Verify` liefert
+`StatusValid` mit befülltem `SignerSubject`. Archivierung +
+Abruf des Prüfnachweises über echten `storage.Service`
+(`LocalDriver`) geprüft — Rundtrip liefert identisches Ergebnis.
+
+Ergebnis: **BESTANDEN**.
+
+## Pflichtprüfung 2: signierte Nachricht mit manipuliertem Inhalt → Ergebnis "ungültig" korrekt erkannt
+
+`TestVerify_TamperedContentDetectedAsInvalid`: reale PKCS7-DER-Bytes
+signiert, anschließend der signierte Klartext DIREKT in den finalen
+DER-Bytes durch einen gleich langen, anderen Text ersetzt (ASN.1-
+Längenpräfixe bleiben syntaktisch gültig, Inhalt weicht vom
+signierten Digest ab — realistische Nachbildung nachträglicher
+Manipulation). `Verify` liefert `StatusInvalid` mit erklärendem
+`Reason`.
+
+Ergebnis: **BESTANDEN**.
+
+## Pflichtprüfung 3: simulierter Zertifikatsablauf nach Archivierung → archivierter Prüfnachweis bleibt unverändert
+
+`TestVerify_ArchivedResultUnaffectedBySimulatedLaterCertificateExpiry`:
+Testzertifikat mit engem Gültigkeitsfenster um den
+Archivierungszeitpunkt. Erste Prüfung + Archivierung ZUM
+Archivierungszeitpunkt liefert `StatusValid`, wird als Sidecar
+gespeichert. Eine SIMULIERTE, NICHT persistierte spätere Live-Prüfung
+derselben Rohdaten mit einem `now` weit nach `CertNotAfter` liefert
+korrekt `StatusInvalid` (abgelaufenes Gültigkeitsfenster). Der BEREITS
+ARCHIVIERTE Prüfnachweis wird davon nicht berührt: `LoadResult` liefert
+weiterhin `StatusValid` mit unverändertem `CheckedAt` — die spätere
+Prüfung schreibt nie über einen bestehenden archivierten Nachweis.
+
+Ergebnis: **BESTANDEN**.
+
+## Ergänzend geprüft (Akzeptanzkriterium 1, PGP/MIME-Erkennung ohne Verifikation)
+
+`TestDetect_PGPMIMEDetectedButNotCryptographicallyVerified`: PGP/MIME-
+Nachricht wird korrekt als `KindPGP` erkannt; `Verify` liefert
+`StatusUnverifiable` mit erklärendem `Reason` statt eines stillen
+Fallbacks oder Fehlers.
+
+`TestDetect_UnsignedMessageNotDetected`: unsignierte Nachricht wird
+korrekt NICHT erkannt, durchläuft keine Prüfung.
+
+## Akzeptanzkriterien
+
+1. **Signierte Nachrichten (S/MIME, PGP/MIME) werden bei Archivierung
+   erkannt**: `Detect`, durch alle Detect-Tests belegt.
+2. **Signaturprüfung liefert einen von drei Ergebniszuständen
+   (gültig/ungültig/Prüfung nicht möglich) inkl. Zertifikatsdaten und
+   Prüfzeitpunkt, als Nachweis archiviert**: `Verify` + `Result` +
+   `SaveResult`, durch Pflichtprüfung 1–3 belegt.
+3. **Prüfnachweis über die Detailansicht abrufbar, Original-Mail
+   unverändert**: `LoadResult` liest ein eigenständiges
+   Sidecar-Objekt, rührt die archivierten RFC822-Rohbytes nicht an;
+   durch Pflichtprüfung 1 und 3 (Rundtrip) belegt.
+
+## Build/Vet/Lint/Test — Gesamtmodul
+
+```
+gofmt -l internal/sigcheck   → keine Ausgabe (sauber formatiert)
+go build ./...                → OK
+go vet ./...                  → OK
+golangci-lint run ./internal/sigcheck/... → 0 issues
+go test ./internal/sigcheck/... -v -count=1 → alle 6 Tests bestanden
+go test ./... -p 1 -count=1 (TEST_TENANT_DSN, TEST_MANTICORE_URL, TEST_S3_ENDPOINT/TEST_S3_ACCESS_KEY/TEST_S3_SECRET_KEY gesetzt) → alle Pakete ok, inkl. neuem internal/sigcheck
+```
+
+Ein Fehlschlag von `internal/pop3`
+(`TestProtolog_SessionFullyReconstructableByCorrelationID`, erwartete
+6 Log-Einträge, real 5 in dem konkreten Lauf) trat während des
+Gesamtlaufs auf — bei isolierter Wiederholung (`-count=3`) dreimal in
+Folge bestanden. Bereits bestehender, von ARC-10 unabhängiger,
+zeitkritischer Flake im Protokoll-Logging-Test (`internal/protolog`/
+`internal/pop3`, ING-08), keine Regression durch dieses Ticket.
+
+## Ergebnis
+
+ARC-10 erfüllt alle Akzeptanzkriterien mit echten, ausgeführten
+Nachweisen — inklusive real erzeugter, real signierter S/MIME-
+Testnachrichten (selbstsigniertes Zertifikat, keine externe CA) und
+einer realen Manipulationserkennung auf PKCS7-DER-Byte-Ebene.
+Freigeschaltet: Beitrag zu QA-05 (zusammen mit ARC-07/09/INT-08,
+ARC-05 weiterhin extern blockiert durch RET-03).

mail/go.mod

@@ -11,6 +11,7 @@ require (
 	github.com/fsnotify/fsnotify v1.10.1
 	github.com/getkin/kin-openapi v0.135.0
 	github.com/jackc/pgx/v5 v5.6.0
+	go.mozilla.org/pkcs7 v0.10.0
 	golang.org/x/text v0.14.0
 )
 

mail/go.sum

@@ -86,6 +86,8 @@ github.com/ugorji/go/codec v1.2.7 h1:YPXUKf7fYbp/y8xloBqZOw2qaVggbfwMlI8WM3wZUJ0
 github.com/ugorji/go/codec v1.2.7/go.mod h1:WGN1fab3R1fzQlVQTkfxVtIBhWDRqOviHU95kRgeqEY=
 github.com/woodsbury/decimal128 v1.3.0 h1:8pffMNWIlC0O5vbyHWFZAt5yWvWcrHA+3ovIIjVWss0=
 github.com/woodsbury/decimal128 v1.3.0/go.mod h1:C5UTmyTjW3JftjUFzOVhC20BEQa2a4ZKOB5I6Zjb+ds=
+go.mozilla.org/pkcs7 v0.10.0 h1:jmljzDzNYFzaP1dFlgmCiQml9e+iEMmv8/NNs4evQbg=
+go.mozilla.org/pkcs7 v0.10.0/go.mod h1:SNgMg+EgDFwmvSmLRTNKC5fegJjB7v23qTQ0XLGUNHk=
 golang.org/x/crypto v0.17.0 h1:r8bRNjWL3GshPW3gkd+RpvzWrZAwPS49OmTGZ/uhM4k=
 golang.org/x/crypto v0.17.0/go.mod h1:gCAAfMLgwOJRpTjQ2zCCt2OcSfYMTeZVSRtQlPC7Nq4=
 golang.org/x/sync v0.1.0 h1:wsuoTGHzEhffawBOhz5CYhcrV4IdKZbEyZjBMuTp12o=

mail/internal/mailboxconfig/migrations/0002_mail_mailboxes_quota.sql

@@ -0,0 +1 @@
+ALTER TABLE mail_mailboxes ADD COLUMN IF NOT EXISTS quota_bytes BIGINT NOT NULL DEFAULT 0;

mail/internal/mailboxconfig/store.go

@@ -28,6 +28,9 @@ import (
 //go:embed migrations/0001_mail_mailboxes.sql
 var schemaMigration string
 
+//go:embed migrations/0002_mail_mailboxes_quota.sql
+var quotaMigration string
+
 // ErrNotFound wird geliefert, wenn kein Postfach mit den angegebenen
 // Bezugsdaten existiert.
 var ErrNotFound = errors.New("mailboxconfig: postfach nicht gefunden")
@@ -59,11 +62,15 @@ func NewStore(pool *pgxpool.Pool, cryptoSvc *crypto.Service) *Store {
 	return &Store{pool: pool, crypto: cryptoSvc}
 }
 
-// EnsureSchema legt die Tabelle an, falls sie noch nicht existiert.
+// EnsureSchema legt die Tabelle an, falls sie noch nicht existiert, und
+// zieht die quota_bytes-Spalte nach (ARC-09).
 func (s *Store) EnsureSchema(ctx context.Context) error {
 	if _, err := s.pool.Exec(ctx, schemaMigration); err != nil {
 		return fmt.Errorf("mailboxconfig: schema anlegen: %w", err)
 	}
+	if _, err := s.pool.Exec(ctx, quotaMigration); err != nil {
+		return fmt.Errorf("mailboxconfig: quota-spalte anlegen: %w", err)
+	}
 	return nil
 }
 
@@ -183,6 +190,46 @@ func (s *Store) Delete(ctx context.Context, tenantSlug string, id int64) error {
 	return nil
 }
 
+// SetQuotaBytes setzt das Speicherlimit EINES Postfachs, identifiziert
+// über seinen (innerhalb des Mandanten eindeutigen) Namen (ARC-09
+// Akzeptanzkriterium 1: je Postfach konfigurierbar, unabhängig von der
+// Tenant-weiten Core-LIC-05-Quota). 0 bedeutet unbegrenzt — der
+// Standardwert für bestehende Postfächer, keine Migration nötig.
+func (s *Store) SetQuotaBytes(ctx context.Context, tenantSlug, name string, quotaBytes int64) error {
+	tag, err := s.pool.Exec(ctx, `
+		UPDATE mail_mailboxes SET quota_bytes = $3, updated_at = now()
+		WHERE tenant_slug = $1 AND name = $2
+	`, tenantSlug, name, quotaBytes)
+	if err != nil {
+		return fmt.Errorf("mailboxconfig: quota setzen: %w", err)
+	}
+	if tag.RowsAffected() == 0 {
+		return ErrNotFound
+	}
+	return nil
+}
+
+// LimitBytes erfüllt quota.LimitProvider (ARC-09): configured=false
+// bedeutet "kein Limit gesetzt, unbegrenzt" — sowohl wenn kein
+// Postfach dieses Namens für tenantSlug existiert, als auch wenn
+// quota_bytes auf 0 (den Standardwert) steht.
+func (s *Store) LimitBytes(ctx context.Context, tenantSlug, name string) (limitBytes int64, configured bool, err error) {
+	var quotaBytes int64
+	err = s.pool.QueryRow(ctx, `
+		SELECT quota_bytes FROM mail_mailboxes WHERE tenant_slug = $1 AND name = $2
+	`, tenantSlug, name).Scan(&quotaBytes)
+	if err != nil {
+		if errors.Is(err, pgx.ErrNoRows) {
+			return 0, false, nil
+		}
+		return 0, false, fmt.Errorf("mailboxconfig: quota lesen: %w", err)
+	}
+	if quotaBytes <= 0 {
+		return 0, false, nil
+	}
+	return quotaBytes, true, nil
+}
+
 // GetDecryptedPassword entschlüsselt das Postfach-Passwort — separater,
 // bewusster Aufruf statt Bestandteil von List/Get, damit Zugangsdaten
 // nicht beiläufig mitgeliefert werden.

mail/internal/quota/quota.go

@@ -0,0 +1,69 @@
+// Package quota implementiert ARC-09: konfigurierbares Speicherlimit
+// je einzelnem Postfach — eine ANDERE Ebene als die tenant-weite Quota
+// aus Core LIC-05 (bekannter Fehler zu vermeiden: beide Ebenen müssen
+// unabhängig durchgesetzt werden, ein Tenant kann insgesamt genug
+// Kontingent haben, während ein einzelnes Postfach überläuft).
+package quota
+
+import "context"
+
+// LimitProvider liefert das konfigurierte Speicherlimit eines
+// Postfachs (Akzeptanzkriterium 1). configured=false bedeutet: kein
+// Limit gesetzt, also unbegrenzt — erfüllt von
+// mail/internal/mailboxconfig.Store.LimitBytes.
+type LimitProvider interface {
+	LimitBytes(ctx context.Context, tenantSlug, mailbox string) (limitBytes int64, configured bool, err error)
+}
+
+// UsageProvider liefert den TATSÄCHLICHEN, real gemessenen
+// Speicherverbrauch eines Postfachs — erfüllt von
+// mail/internal/storage.UsageCounter.
+type UsageProvider interface {
+	UsageBytes(ctx context.Context, tenantSlug, mailbox string) (usedBytes int64, err error)
+}
+
+// Result ist das Ergebnis einer Quota-Prüfung.
+type Result struct {
+	Allowed    bool
+	UsedBytes  int64
+	LimitBytes int64
+}
+
+// Checker verknüpft konfiguriertes Limit und realen Verbrauch.
+type Checker struct {
+	limits LimitProvider
+	usage  UsageProvider
+}
+
+func NewChecker(limits LimitProvider, usage UsageProvider) *Checker {
+	return &Checker{limits: limits, usage: usage}
+}
+
+// Check prüft, ob mailbox innerhalb seines konfigurierten Limits ist.
+// Kein konfiguriertes Limit bedeutet immer "erlaubt" — die
+// Tenant-weite Quota (Core LIC-05) läuft unabhängig davon weiter.
+func (c *Checker) Check(ctx context.Context, tenantSlug, mailbox string) (Result, error) {
+	limit, configured, err := c.limits.LimitBytes(ctx, tenantSlug, mailbox)
+	if err != nil {
+		return Result{}, err
+	}
+	if !configured {
+		return Result{Allowed: true}, nil
+	}
+	used, err := c.usage.UsageBytes(ctx, tenantSlug, mailbox)
+	if err != nil {
+		return Result{}, err
+	}
+	return Result{Allowed: used < limit, UsedBytes: used, LimitBytes: limit}, nil
+}
+
+// Allowed ist die schmale Form von Check für Aufrufer, die nur die
+// Ja/Nein-Entscheidung brauchen (z. B. mail/internal/smtp.QuotaChecker,
+// Akzeptanzkriterium 2).
+func (c *Checker) Allowed(ctx context.Context, tenantSlug, mailbox string) (bool, error) {
+	result, err := c.Check(ctx, tenantSlug, mailbox)
+	if err != nil {
+		return false, err
+	}
+	return result.Allowed, nil
+}

mail/internal/quota/quota_test.go

@@ -0,0 +1,56 @@
+package quota
+
+import (
+	"context"
+	"testing"
+)
+
+type fakeLimits struct {
+	limitBytes int64
+	configured bool
+}
+
+func (f fakeLimits) LimitBytes(context.Context, string, string) (int64, bool, error) {
+	return f.limitBytes, f.configured, nil
+}
+
+type fakeUsage struct {
+	usedBytes int64
+}
+
+func (f fakeUsage) UsageBytes(context.Context, string, string) (int64, error) {
+	return f.usedBytes, nil
+}
+
+func TestCheck_UnconfiguredLimitAlwaysAllowed(t *testing.T) {
+	c := NewChecker(fakeLimits{configured: false}, fakeUsage{usedBytes: 1_000_000_000})
+	result, err := c.Check(context.Background(), "mandant-a", "postfach-x")
+	if err != nil {
+		t.Fatalf("check: %v", err)
+	}
+	if !result.Allowed {
+		t.Fatalf("erwartete erlaubt ohne konfiguriertes limit, habe: %+v", result)
+	}
+}
+
+func TestCheck_UsageAtOrAboveLimitRejected(t *testing.T) {
+	c := NewChecker(fakeLimits{limitBytes: 1000, configured: true}, fakeUsage{usedBytes: 1000})
+	result, err := c.Check(context.Background(), "mandant-a", "postfach-x")
+	if err != nil {
+		t.Fatalf("check: %v", err)
+	}
+	if result.Allowed {
+		t.Fatalf("erwartete ablehnung bei verbrauch == limit, habe: %+v", result)
+	}
+}
+
+func TestCheck_UsageBelowLimitAllowed(t *testing.T) {
+	c := NewChecker(fakeLimits{limitBytes: 1000, configured: true}, fakeUsage{usedBytes: 999})
+	result, err := c.Check(context.Background(), "mandant-a", "postfach-x")
+	if err != nil {
+		t.Fatalf("check: %v", err)
+	}
+	if !result.Allowed {
+		t.Fatalf("erwartete erlaubt bei verbrauch unter limit, habe: %+v", result)
+	}
+}

mail/internal/sigcheck/sigcheck.go

@@ -0,0 +1,229 @@
+// Package sigcheck implementiert ARC-10: Signatur- und
+// Zeitstempelerhalt für S/MIME-signierte Mails (BSI TR-03125, Modul
+// B.4). Erkennung (Detect) läuft für JEDE archivierte Nachricht
+// (billig, nur Content-Type-Parsing); die eigentliche
+// kryptographische Prüfung (Verify) NUR für tatsächlich erkannte
+// signierte Nachrichten (Akzeptanzkriterium 1).
+//
+// Bewusste Scope-Grenze (siehe ARC-10-Prüfprotokoll): PGP/MIME wird
+// erkannt, aber NICHT kryptographisch geprüft —
+// StatusUnverifiable ist einer der von Akzeptanzkriterium 2
+// ausdrücklich vorgesehenen drei Ergebniszustände
+// (gültig/ungültig/Prüfung nicht möglich), kein stiller Fallback.
+// Kein volles Langzeitarchivierungs-/ArchiSig-Verfahren mit
+// periodischer Neu-Zeitstempelung (siehe Ticket-Begründung: für ein
+// KMU-Mailarchiv ohne qualifiziertes elektronisches Archiv nach eIDAS
+// unverhältnismäßig) — ein einmaliger, bei Archivierung erzeugter
+// Prüfnachweis reicht aus.
+package sigcheck
+
+import (
+	"bytes"
+	"context"
+	"encoding/base64"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"mime"
+	"net/mail"
+	"strings"
+	"time"
+
+	"go.mozilla.org/pkcs7"
+)
+
+// Kind ist die erkannte Signaturart (Akzeptanzkriterium 1).
+type Kind string
+
+const (
+	KindSMIME Kind = "smime"
+	KindPGP   Kind = "pgp"
+)
+
+// Status ist einer der drei von Akzeptanzkriterium 2 vorgesehenen
+// Ergebniszustände.
+type Status string
+
+const (
+	StatusValid        Status = "gueltig"
+	StatusInvalid      Status = "ungueltig"
+	StatusUnverifiable Status = "pruefung_nicht_moeglich"
+)
+
+// Result ist der unveränderlich zu archivierende Prüfnachweis
+// (Akzeptanzkriterium 2: Ergebnis, Zertifikatsdaten, Prüfzeitpunkt).
+type Result struct {
+	Kind          Kind      `json:"kind"`
+	Status        Status    `json:"status"`
+	SignerSubject string    `json:"signerSubject,omitempty"`
+	CertNotBefore time.Time `json:"certNotBefore,omitempty"`
+	CertNotAfter  time.Time `json:"certNotAfter,omitempty"`
+	CheckedAt     time.Time `json:"checkedAt"`
+	Reason        string    `json:"reason,omitempty"`
+}
+
+// Detect erkennt, ob raw eine signierte Nachricht ist — OHNE die
+// Signatur zu prüfen (Akzeptanzkriterium 1). detected=false für jede
+// unsignierte Nachricht, die dann KEINE weitere Prüfung durchläuft.
+func Detect(raw []byte) (kind Kind, detected bool, err error) {
+	msg, err := mail.ReadMessage(bytes.NewReader(raw))
+	if err != nil {
+		return "", false, err
+	}
+	contentType := msg.Header.Get("Content-Type")
+	if contentType == "" {
+		return "", false, nil
+	}
+	mediaType, params, err := mime.ParseMediaType(contentType)
+	if err != nil {
+		return "", false, nil
+	}
+	switch {
+	case mediaType == "application/pkcs7-mime" && strings.EqualFold(params["smime-type"], "signed-data"):
+		return KindSMIME, true, nil
+	case mediaType == "multipart/signed":
+		switch strings.ToLower(params["protocol"]) {
+		case "application/pkcs7-signature", "application/x-pkcs7-signature":
+			return KindSMIME, true, nil
+		case "application/pgp-signature":
+			return KindPGP, true, nil
+		}
+	}
+	return "", false, nil
+}
+
+// Verify prüft eine ALS SIGNIERT ERKANNTE Nachricht (Akzeptanzkriterium
+// 2). now ist der Prüfzeitpunkt — im Regelbetrieb time.Now() zum
+// Archivierungszeitpunkt, in Tests explizit steuerbar (Pflichtprüfung
+// 3: simulierter Zertifikatsablauf nach Archivierung — ein späterer
+// Aufruf mit einem SPÄTEREN now kann ein anderes Ergebnis liefern,
+// ändert aber nie ein bereits archiviertes Result rückwirkend).
+func Verify(raw []byte, kind Kind, now time.Time) (Result, error) {
+	result := Result{Kind: kind, CheckedAt: now}
+
+	if kind == KindPGP {
+		result.Status = StatusUnverifiable
+		result.Reason = "PGP/MIME-Signaturprüfung nicht implementiert (bewusste Scope-Grenze, siehe ARC-10-Prüfprotokoll)"
+		return result, nil
+	}
+	if kind != KindSMIME {
+		return Result{}, errors.New("sigcheck: unbekannte signaturart")
+	}
+
+	msg, err := mail.ReadMessage(bytes.NewReader(raw))
+	if err != nil {
+		result.Status = StatusUnverifiable
+		result.Reason = "nachricht nicht lesbar: " + err.Error()
+		return result, nil
+	}
+	body, err := io.ReadAll(msg.Body)
+	if err != nil {
+		result.Status = StatusUnverifiable
+		result.Reason = "body nicht lesbar: " + err.Error()
+		return result, nil
+	}
+	decoded, err := decodePKCS7Body(msg.Header.Get("Content-Transfer-Encoding"), body)
+	if err != nil {
+		result.Status = StatusUnverifiable
+		result.Reason = "pkcs7-daten nicht dekodierbar: " + err.Error()
+		return result, nil
+	}
+	p7, err := pkcs7.Parse(decoded)
+	if err != nil {
+		result.Status = StatusUnverifiable
+		result.Reason = "pkcs7-struktur nicht lesbar: " + err.Error()
+		return result, nil
+	}
+	if len(p7.Certificates) > 0 {
+		result.SignerSubject = p7.Certificates[0].Subject.String()
+		result.CertNotBefore = p7.Certificates[0].NotBefore
+		result.CertNotAfter = p7.Certificates[0].NotAfter
+	}
+
+	// Kryptographische Prüfung (Signatur + Inhaltsintegrität) — kein
+	// Vertrauensanker (truststore=nil): für ein KMU-Mailarchiv ohne
+	// qualifiziertes elektronisches Archiv reicht der Nachweis "diese
+	// konkrete Signatur wurde mit diesem konkreten Zertifikat über
+	// diesen konkreten Inhalt erzeugt" (siehe Ticket-Begründung, "kein
+	// eIDAS-Archiv").
+	if err := p7.Verify(); err != nil {
+		result.Status = StatusInvalid
+		result.Reason = err.Error()
+		return result, nil
+	}
+
+	// Zusätzliche, explizite Gültigkeitsfensterprüfung GEGEN DEN
+	// ÜBERGEBENEN PRÜFZEITPUNKT (nicht gegen die reale Systemzeit) —
+	// Grundlage für Pflichtprüfung 3: ein archiviertes Result bleibt
+	// unverändert "gültig zum Archivierungszeitpunkt", auch wenn eine
+	// SPÄTERE Prüfung mit einem SPÄTEREN now auf denselben Rohdaten ein
+	// anderes Ergebnis liefern würde.
+	if !result.CertNotBefore.IsZero() && !result.CertNotAfter.IsZero() {
+		if now.Before(result.CertNotBefore) || now.After(result.CertNotAfter) {
+			result.Status = StatusInvalid
+			result.Reason = "zertifikat zum prüfzeitpunkt nicht gültig (außerhalb des gültigkeitsfensters)"
+			return result, nil
+		}
+	}
+
+	result.Status = StatusValid
+	return result, nil
+}
+
+func decodePKCS7Body(transferEncoding string, body []byte) ([]byte, error) {
+	if strings.EqualFold(strings.TrimSpace(transferEncoding), "base64") {
+		cleaned := strings.Map(func(r rune) rune {
+			if r == '\r' || r == '\n' || r == ' ' || r == '\t' {
+				return -1
+			}
+			return r
+		}, string(body))
+		return base64.StdEncoding.DecodeString(cleaned)
+	}
+	return body, nil
+}
+
+// Store ist die für Ablage/Abruf des Prüfnachweises benötigte
+// Teilmenge von *mail/internal/storage.Service — als Schnittstelle,
+// damit dieses Paket storage nicht importieren muss.
+type Store interface {
+	Put(ctx context.Context, key string, r io.Reader, size int64, contentType string) (checksum string, err error)
+	GetVerified(ctx context.Context, key string) ([]byte, error)
+}
+
+// SidecarKey liefert den Objektschlüssel des Prüfnachweises zu einer
+// archivierten Nachricht (Akzeptanzkriterium 3: über die Detailansicht
+// abrufbar, OHNE die Original-Mail zu verändern — eigenes
+// Sidecar-Objekt, gleiche Konvention wie ARC-01s
+// Prüfsummen-Sidecar-Objekt).
+func SidecarKey(archiveKey string) string {
+	return archiveKey + ".sigcheck.json"
+}
+
+// SaveResult archiviert result UNVERÄNDERLICH als Sidecar-Objekt
+// (Akzeptanzkriterium 2).
+func SaveResult(ctx context.Context, store Store, archiveKey string, result Result) error {
+	body, err := json.Marshal(result)
+	if err != nil {
+		return fmt.Errorf("sigcheck: ergebnis serialisieren: %w", err)
+	}
+	if _, err := store.Put(ctx, SidecarKey(archiveKey), bytes.NewReader(body), int64(len(body)), "application/json"); err != nil {
+		return fmt.Errorf("sigcheck: ergebnis ablegen: %w", err)
+	}
+	return nil
+}
+
+// LoadResult ruft den archivierten Prüfnachweis ab (Akzeptanzkriterium
+// 3), ohne die Original-Mail zu lesen oder zu verändern.
+func LoadResult(ctx context.Context, store Store, archiveKey string) (Result, error) {
+	raw, err := store.GetVerified(ctx, SidecarKey(archiveKey))
+	if err != nil {
+		return Result{}, fmt.Errorf("sigcheck: ergebnis abrufen: %w", err)
+	}
+	var result Result
+	if err := json.Unmarshal(raw, &result); err != nil {
+		return Result{}, fmt.Errorf("sigcheck: ergebnis parsen: %w", err)
+	}
+	return result, nil
+}

mail/internal/sigcheck/sigcheck_test.go

@@ -0,0 +1,316 @@
+package sigcheck
+
+import (
+	"bytes"
+	"context"
+	"crypto/rand"
+	"crypto/rsa"
+	"crypto/x509"
+	"crypto/x509/pkix"
+	"encoding/base64"
+	"fmt"
+	"math/big"
+	"strings"
+	"testing"
+	"time"
+
+	"go.mozilla.org/pkcs7"
+
+	"gitea.perlbach24.de/scripte/nexarch/mail/internal/storage"
+)
+
+// generateTestCert erzeugt ein selbstsigniertes RSA-Testzertifikat —
+// keine externe CA nötig (analog zu tlscert.GenerateSelfSigned, ING-06).
+func generateTestCert(t *testing.T, notBefore, notAfter time.Time) (*x509.Certificate, *rsa.PrivateKey) {
+	t.Helper()
+	key, err := rsa.GenerateKey(rand.Reader, 2048)
+	if err != nil {
+		t.Fatalf("schlüssel erzeugen: %v", err)
+	}
+	serial, err := rand.Int(rand.Reader, new(big.Int).Lsh(big.NewInt(1), 128))
+	if err != nil {
+		t.Fatalf("seriennummer erzeugen: %v", err)
+	}
+	template := x509.Certificate{
+		SerialNumber: serial,
+		Subject:      pkix.Name{CommonName: "ARC-10 Testsignatur"},
+		NotBefore:    notBefore,
+		NotAfter:     notAfter,
+		KeyUsage:     x509.KeyUsageDigitalSignature,
+		ExtKeyUsage:  []x509.ExtKeyUsage{x509.ExtKeyUsageEmailProtection},
+	}
+	der, err := x509.CreateCertificate(rand.Reader, &template, &template, &key.PublicKey, key)
+	if err != nil {
+		t.Fatalf("zertifikat erzeugen: %v", err)
+	}
+	cert, err := x509.ParseCertificate(der)
+	if err != nil {
+		t.Fatalf("zertifikat parsen: %v", err)
+	}
+	return cert, key
+}
+
+// buildSignedMessage signiert plaintext EINMALIG (S/MIME opaque,
+// application/pkcs7-mime; smime-type=signed-data) und liefert die
+// vollständige RFC822-Rohnachricht.
+func buildSignedMessage(t *testing.T, cert *x509.Certificate, key *rsa.PrivateKey, plaintext string) []byte {
+	t.Helper()
+	sd, err := pkcs7.NewSignedData([]byte(plaintext))
+	if err != nil {
+		t.Fatalf("signeddata erzeugen: %v", err)
+	}
+	if err := sd.AddSigner(cert, key, pkcs7.SignerInfoConfig{}); err != nil {
+		t.Fatalf("signer hinzufügen: %v", err)
+	}
+	der, err := sd.Finish()
+	if err != nil {
+		t.Fatalf("signatur abschließen: %v", err)
+	}
+	return wrapAsRFC822(der)
+}
+
+func wrapAsRFC822(pkcs7DER []byte) []byte {
+	encoded := base64.StdEncoding.EncodeToString(pkcs7DER)
+	var b bytes.Buffer
+	b.WriteString("From: absender@example.com\r\n")
+	b.WriteString("To: empfaenger@example.com\r\n")
+	b.WriteString("Subject: ARC-10 Testmail\r\n")
+	b.WriteString("Content-Type: application/pkcs7-mime; smime-type=signed-data; name=\"smime.p7m\"\r\n")
+	b.WriteString("Content-Transfer-Encoding: base64\r\n")
+	b.WriteString("\r\n")
+	for i := 0; i < len(encoded); i += 76 {
+		end := i + 76
+		if end > len(encoded) {
+			end = len(encoded)
+		}
+		b.WriteString(encoded[i:end])
+		b.WriteString("\r\n")
+	}
+	return b.Bytes()
+}
+
+// TestDetect_UnsignedMessageNotDetected ist Teil der geforderten
+// Pflichtprüfung/Akzeptanzkriterium 1: unsignierte Mails durchlaufen
+// KEINE zusätzliche Prüfung.
+func TestDetect_UnsignedMessageNotDetected(t *testing.T) {
+	raw := []byte("From: a@example.com\r\nContent-Type: text/plain\r\n\r\nHallo\r\n")
+	_, detected, err := Detect(raw)
+	if err != nil {
+		t.Fatalf("detect: %v", err)
+	}
+	if detected {
+		t.Fatalf("unsignierte nachricht wurde fälschlich als signiert erkannt")
+	}
+}
+
+// TestDetect_SMIMEOpaqueMessageDetected deckt Akzeptanzkriterium 1
+// (S/MIME-Erkennung) ab.
+func TestDetect_SMIMEOpaqueMessageDetected(t *testing.T) {
+	now := time.Now()
+	cert, key := generateTestCert(t, now.Add(-time.Hour), now.Add(time.Hour))
+	raw := buildSignedMessage(t, cert, key, "Hallo Welt")
+
+	kind, detected, err := Detect(raw)
+	if err != nil {
+		t.Fatalf("detect: %v", err)
+	}
+	if !detected || kind != KindSMIME {
+		t.Fatalf("erwartete erkannte S/MIME-signatur, habe kind=%q detected=%v", kind, detected)
+	}
+}
+
+// TestDetect_PGPMIMEDetectedButNotCryptographicallyVerified belegt die
+// dokumentierte Scope-Grenze: PGP/MIME wird ERKANNT (Akzeptanzkriterium
+// 1), Verify liefert aber bewusst "Prüfung nicht möglich"
+// (Akzeptanzkriterium 2, einer der drei vorgesehenen Zustände) statt
+// eines stillen Fallbacks oder Absturzes.
+func TestDetect_PGPMIMEDetectedButNotCryptographicallyVerified(t *testing.T) {
+	raw := []byte("From: a@example.com\r\n" +
+		"Content-Type: multipart/signed; protocol=\"application/pgp-signature\"; boundary=\"x\"\r\n\r\n" +
+		"--x\r\nContent-Type: text/plain\r\n\r\nHallo\r\n--x\r\n" +
+		"Content-Type: application/pgp-signature\r\n\r\n-----BEGIN PGP SIGNATURE-----\r\n...\r\n-----END PGP SIGNATURE-----\r\n--x--\r\n")
+
+	kind, detected, err := Detect(raw)
+	if err != nil {
+		t.Fatalf("detect: %v", err)
+	}
+	if !detected || kind != KindPGP {
+		t.Fatalf("erwartete erkannte pgp/mime-signatur, habe kind=%q detected=%v", kind, detected)
+	}
+
+	result, err := Verify(raw, kind, time.Now())
+	if err != nil {
+		t.Fatalf("verify: %v", err)
+	}
+	if result.Status != StatusUnverifiable {
+		t.Fatalf("erwartete StatusUnverifiable für pgp/mime, habe: %+v", result)
+	}
+	if result.Reason == "" {
+		t.Fatalf("erwartete erklärenden grund für 'prüfung nicht möglich'")
+	}
+}
+
+// TestVerify_ValidSMIMESignatureRecognizedAsValid ist die geforderte
+// Pflichtprüfung 1 (ARC-10): Test mit einer echten S/MIME-signierten
+// Testmail — Prüfergebnis "gültig" korrekt erkannt und archiviert.
+func TestVerify_ValidSMIMESignatureRecognizedAsValid(t *testing.T) {
+	now := time.Now()
+	cert, key := generateTestCert(t, now.Add(-time.Hour), now.Add(time.Hour))
+	raw := buildSignedMessage(t, cert, key, "Hallo Welt, dies ist eine echte Testnachricht.")
+
+	kind, detected, err := Detect(raw)
+	if err != nil || !detected {
+		t.Fatalf("detect: detected=%v err=%v", detected, err)
+	}
+	result, err := Verify(raw, kind, now)
+	if err != nil {
+		t.Fatalf("verify: %v", err)
+	}
+	if result.Status != StatusValid {
+		t.Fatalf("erwartete StatusValid, habe: %+v", result)
+	}
+	if result.SignerSubject == "" {
+		t.Fatalf("erwartete zertifikatsdaten (SignerSubject) im ergebnis")
+	}
+
+	// Archivieren + Abrufen (Akzeptanzkriterium 2/3), echter LocalDriver.
+	svc := storage.NewService(storage.NewLocalDriver(t.TempDir()), noopReporter{}, "mandant-arc10")
+	archiveKey := storage.ArchiveKey("rechnungen@example.com", now, "msg-arc10-valid", 0)
+	if err := SaveResult(context.Background(), svc, archiveKey, result); err != nil {
+		t.Fatalf("SaveResult: %v", err)
+	}
+	loaded, err := LoadResult(context.Background(), svc, archiveKey)
+	if err != nil {
+		t.Fatalf("LoadResult: %v", err)
+	}
+	if loaded.Status != StatusValid || loaded.SignerSubject != result.SignerSubject {
+		t.Fatalf("archivierter nachweis weicht vom ursprünglichen ergebnis ab: %+v vs %+v", loaded, result)
+	}
+}
+
+// TestVerify_TamperedContentDetectedAsInvalid ist die geforderte
+// Pflichtprüfung 2 (ARC-10): Test mit einer signierten Mail mit
+// MANIPULIERTEM Inhalt — Prüfergebnis "ungültig" korrekt erkannt. Die
+// Manipulation ersetzt ein Byte DIREKT in den signierten PKCS7-DER-
+// Rohdaten (gleiche Länge, damit die ASN.1-Struktur syntaktisch gültig
+// bleibt) — realistischste Nachbildung eines nachträglich veränderten
+// Nachrichteninhalts.
+func TestVerify_TamperedContentDetectedAsInvalid(t *testing.T) {
+	now := time.Now()
+	cert, key := generateTestCert(t, now.Add(-time.Hour), now.Add(time.Hour))
+	const plaintext = "Ursprünglicher, signierter Inhalt"
+
+	sd, err := pkcs7.NewSignedData([]byte(plaintext))
+	if err != nil {
+		t.Fatalf("signeddata: %v", err)
+	}
+	if err := sd.AddSigner(cert, key, pkcs7.SignerInfoConfig{}); err != nil {
+		t.Fatalf("signer: %v", err)
+	}
+	der, err := sd.Finish()
+	if err != nil {
+		t.Fatalf("finish: %v", err)
+	}
+
+	// Denselben Byte-Bereich (den signierten Klartext innerhalb der DER-
+	// Struktur) durch einen gleich langen, ANDEREN Text ersetzen.
+	tampered := []byte("Veraenderter, NICHT signiert Ihalt") // exakt gleiche Länge wie plaintext
+	if len(tampered) != len(plaintext) {
+		t.Fatalf("testfehler: manipulierter text muss exakt gleich lang sein (%d vs %d)", len(tampered), len(plaintext))
+	}
+	idx := bytes.Index(der, []byte(plaintext))
+	if idx < 0 {
+		t.Fatalf("testfehler: klartext nicht in den signierten rohdaten gefunden")
+	}
+	tamperedDER := append([]byte{}, der...)
+	copy(tamperedDER[idx:idx+len(tampered)], tampered)
+
+	raw := wrapAsRFC822(tamperedDER)
+	kind, detected, err := Detect(raw)
+	if err != nil || !detected {
+		t.Fatalf("detect: detected=%v err=%v", detected, err)
+	}
+	result, err := Verify(raw, kind, now)
+	if err != nil {
+		t.Fatalf("verify: %v", err)
+	}
+	if result.Status != StatusInvalid {
+		t.Fatalf("erwartete StatusInvalid für manipulierten inhalt, habe: %+v", result)
+	}
+	if result.Reason == "" {
+		t.Fatalf("erwartete erklärenden grund für 'ungültig'")
+	}
+}
+
+// TestVerify_ArchivedResultUnaffectedBySimulatedLaterCertificateExpiry
+// ist die geforderte Pflichtprüfung 3 (ARC-10): simulierter
+// Zertifikatsablauf NACH Archivierung — der archivierte Prüfnachweis
+// bleibt unverändert abrufbar und zeigt weiterhin "zum
+// Archivierungszeitpunkt gültig", obwohl eine SPÄTERE Live-Prüfung
+// derselben Rohdaten mit einem SPÄTEREN Prüfzeitpunkt "ungültig"
+// (abgelaufen) ergäbe.
+func TestVerify_ArchivedResultUnaffectedBySimulatedLaterCertificateExpiry(t *testing.T) {
+	archivalTime := time.Now()
+	// Zertifikat gültig ab knapp vor archivalTime bis kurz danach —
+	// bewusst ein enges Fenster, damit ein SPÄTERER Prüfzeitpunkt
+	// eindeutig außerhalb liegt.
+	certNotBefore := archivalTime.Add(-time.Minute)
+	certNotAfter := archivalTime.Add(time.Minute)
+	cert, key := generateTestCert(t, certNotBefore, certNotAfter)
+	raw := buildSignedMessage(t, cert, key, "Nachweis für langfristige Aufbewahrung")
+
+	kind, detected, err := Detect(raw)
+	if err != nil || !detected {
+		t.Fatalf("detect: detected=%v err=%v", detected, err)
+	}
+
+	// 1. Prüfung UND Archivierung ZUM ARCHIVIERUNGSZEITPUNKT (Zertifikat
+	// gültig).
+	archivedResult, err := Verify(raw, kind, archivalTime)
+	if err != nil {
+		t.Fatalf("verify (archivierung): %v", err)
+	}
+	if archivedResult.Status != StatusValid {
+		t.Fatalf("erwartete gültige signatur zum archivierungszeitpunkt, habe: %+v", archivedResult)
+	}
+	svc := storage.NewService(storage.NewLocalDriver(t.TempDir()), noopReporter{}, "mandant-arc10-expiry")
+	archiveKey := storage.ArchiveKey("vertraege@example.com", archivalTime, "msg-arc10-expiry", 0)
+	if err := SaveResult(context.Background(), svc, archiveKey, archivedResult); err != nil {
+		t.Fatalf("SaveResult: %v", err)
+	}
+
+	// 2. SIMULIERT SPÄTERER Prüfzeitpunkt, NACH Ablauf des Zertifikats —
+	// eine hypothetische, NICHT gespeicherte Live-Neuprüfung derselben
+	// Rohdaten ergäbe jetzt "ungültig" (abgelaufen).
+	muchLater := certNotAfter.Add(365 * 24 * time.Hour)
+	liveCheckLater, err := Verify(raw, kind, muchLater)
+	if err != nil {
+		t.Fatalf("verify (simulierte spätere prüfung): %v", err)
+	}
+	if liveCheckLater.Status != StatusInvalid {
+		t.Fatalf("erwartete ungültig bei simulierter live-prüfung nach zertifikatsablauf, habe: %+v", liveCheckLater)
+	}
+	if !strings.Contains(liveCheckLater.Reason, "gültig") {
+		t.Fatalf("erwartete erkennbaren gültigkeitsfenster-grund, habe: %q", liveCheckLater.Reason)
+	}
+
+	// 3. Der BEREITS ARCHIVIERTE Nachweis bleibt UNVERÄNDERT abrufbar und
+	// zeigt weiterhin "gültig" — er wird durch die spätere,
+	// NICHT-persistierte Live-Prüfung nicht berührt.
+	loaded, err := LoadResult(context.Background(), svc, archiveKey)
+	if err != nil {
+		t.Fatalf("LoadResult: %v", err)
+	}
+	if loaded.Status != StatusValid {
+		t.Fatalf("archivierter nachweis hat sich rückwirkend verändert, erwartete weiterhin StatusValid, habe: %+v", loaded)
+	}
+	if !loaded.CheckedAt.Equal(archivalTime.Truncate(time.Second)) && loaded.CheckedAt.Sub(archivalTime).Abs() > time.Second {
+		t.Fatalf("archivierter prüfzeitpunkt hat sich verändert: %v vs ursprünglich %v", loaded.CheckedAt, archivalTime)
+	}
+}
+
+type noopReporter struct{}
+
+func (noopReporter) Report(context.Context, string, string, int64) error { return nil }
+
+var _ = fmt.Sprintf // Platzhalter, falls fmt künftig entfällt

mail/internal/smtp/arc09_integration_test.go

@@ -0,0 +1,215 @@
+// Integrationstest (ARC-09): echte Postgres-Instanz (mailboxconfig.Store)
+// und echte MinIO-Instanz (storage.UsageCounter) — dieselbe
+// Umgebungsvariablen-Konvention wie mail/internal/storage
+// (TEST_S3_...) und mail/internal/mailboxconfig (TEST_TENANT_DSN).
+package smtp
+
+import (
+	"context"
+	"net"
+	"os"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/service/s3"
+	"github.com/jackc/pgx/v5/pgxpool"
+
+	"gitea.perlbach24.de/scripte/nexarch/mail/internal/crypto"
+	"gitea.perlbach24.de/scripte/nexarch/mail/internal/mailboxconfig"
+	"gitea.perlbach24.de/scripte/nexarch/mail/internal/protoguard"
+	"gitea.perlbach24.de/scripte/nexarch/mail/internal/quota"
+	"gitea.perlbach24.de/scripte/nexarch/mail/internal/storage"
+)
+
+type fixedKEKProvider struct{}
+
+func (fixedKEKProvider) TenantKEK(context.Context, string) ([]byte, error) {
+	return make([]byte, crypto.KEKSize), nil
+}
+
+// TestIntegration_UsageDisplayMatchesRealSizeAndEnforcesQuota ist die
+// geforderte Pflichtprüfung 3 (ARC-09): Verbrauchsanzeige je Postfach
+// im Test korrekt gegen tatsächliche Größe geprüft — UND end-to-end:
+// echte Bytes real in einen echten, ARC-06-provisionierten Bucket
+// geschrieben, echte Verbrauchsmessung (S3-Listing), echtes
+// konfiguriertes Limit in Postgres, echte SMTP-RCPT-TO-Ablehnung, sobald
+// der reale Verbrauch das reale Limit erreicht.
+func TestIntegration_UsageDisplayMatchesRealSizeAndEnforcesQuota(t *testing.T) {
+	dsn := os.Getenv("TEST_TENANT_DSN")
+	s3Endpoint := os.Getenv("TEST_S3_ENDPOINT")
+	if dsn == "" || s3Endpoint == "" {
+		t.Skip("TEST_TENANT_DSN/TEST_S3_ENDPOINT nicht gesetzt, Integrationstest übersprungen")
+	}
+	ctx := context.Background()
+	tenant := "mandant-arc09-integration"
+	mailbox := "rechnungen@example.com"
+
+	pool, err := pgxpool.New(ctx, dsn)
+	if err != nil {
+		t.Fatalf("pool: %v", err)
+	}
+	t.Cleanup(pool.Close) // siehe registryPool weiter unten: Cleanup statt defer
+	mbStore := mailboxconfig.NewStore(pool, crypto.NewService(fixedKEKProvider{}))
+	if err := mbStore.EnsureSchema(ctx); err != nil {
+		t.Fatalf("mailboxconfig schema: %v", err)
+	}
+	t.Cleanup(func() {
+		_, _ = pool.Exec(context.Background(), `DELETE FROM mail_mailboxes WHERE tenant_slug = $1`, tenant)
+	})
+
+	mailboxID, err := mbStore.Create(ctx, tenant, mailboxconfig.CreateInput{
+		Name: mailbox, IMAPHost: "imap.example.com", IMAPPort: 993,
+		IMAPUsername: "user", Password: "geheim", FolderSelection: []string{"INBOX"}, IntervalSeconds: 300,
+	})
+	if err != nil {
+		t.Fatalf("postfach anlegen: %v", err)
+	}
+	_ = mailboxID
+
+	s3Admin, err := storage.NewS3AdminClient(ctx, "us-east-1", s3Endpoint, os.Getenv("TEST_S3_ACCESS_KEY"), os.Getenv("TEST_S3_SECRET_KEY"), true)
+	if err != nil {
+		t.Fatalf("s3-admin-client: %v", err)
+	}
+	registryPool, err := pgxpool.New(ctx, dsn)
+	if err != nil {
+		t.Fatalf("registry-pool: %v", err)
+	}
+	// t.Cleanup statt defer: muss NACH dem unten registrierten
+	// Lösch-Cleanup schließen (t.Cleanup läuft LIFO, defer würde VOR
+	// jedem t.Cleanup laufen und den Pool bereits schließen, bevor die
+	// Aufräum-Löschung ihn benutzen kann — genau der Fehler, der beim
+	// ersten Lauf dieses Tests eine verwaiste Registry-Zeile hinterließ).
+	t.Cleanup(registryPool.Close)
+	if _, err := registryPool.Exec(ctx, `
+		CREATE TABLE IF NOT EXISTS tenants (
+			id UUID PRIMARY KEY DEFAULT gen_random_uuid(), slug TEXT NOT NULL UNIQUE, name TEXT NOT NULL,
+			db_dsn TEXT NOT NULL, status TEXT NOT NULL DEFAULT 'active', created_at TIMESTAMPTZ NOT NULL DEFAULT now()
+		)
+	`); err != nil {
+		t.Fatalf("tenants-tabelle: %v", err)
+	}
+	bucket, err := storage.ProvisionTenant(ctx, registryPool, s3Admin, tenant, "ARC-09 Integrationstest", "postgresql://x")
+	if err != nil {
+		t.Fatalf("ProvisionTenant: %v", err)
+	}
+	t.Cleanup(func() {
+		_, _ = registryPool.Exec(context.Background(), `DELETE FROM tenants WHERE slug = $1`, tenant)
+	})
+	t.Cleanup(func() {
+		ctx := context.Background()
+		out, err := s3Admin.ListObjectsV2(ctx, &s3.ListObjectsV2Input{Bucket: aws.String(bucket)})
+		if err == nil {
+			for _, obj := range out.Contents {
+				_, _ = s3Admin.DeleteObject(ctx, &s3.DeleteObjectInput{Bucket: aws.String(bucket), Key: obj.Key})
+			}
+		}
+		_, _ = s3Admin.DeleteBucket(ctx, &s3.DeleteBucketInput{Bucket: aws.String(bucket)})
+	})
+
+	usageReporter := noopUsageReporterForTest{}
+	storageSvc := storage.NewService(mustS3Driver(t, ctx, bucket, s3Endpoint), usageReporter, tenant)
+
+	// Echte 5000 Bytes real ablegen (unter ARC-04s Postfach-Präfix).
+	const contentSize = 5000
+	content := strings.Repeat("A", contentSize)
+	key := storage.ArchiveKey(mailbox, time.Now(), "msg-arc09-1", 0)
+	if _, err := storageSvc.Put(ctx, key, strings.NewReader(content), int64(len(content)), "text/plain"); err != nil {
+		t.Fatalf("objekt ablegen: %v", err)
+	}
+	t.Cleanup(func() {
+		_ = storageSvc.Delete(context.Background(), key, int64(len(content)))
+	})
+
+	usageCounter := storage.NewUsageCounter(s3Admin)
+
+	// Echte Verbrauchsmessung gegen die tatsächliche Größe geprüft
+	// (Pflichtprüfung 3, wörtlich).
+	measured, err := usageCounter.UsageBytes(ctx, tenant, mailbox)
+	if err != nil {
+		t.Fatalf("UsageBytes: %v", err)
+	}
+	if measured < contentSize {
+		t.Fatalf("gemessener verbrauch %d liegt unter der tatsächlich abgelegten größe %d (checksum-sidecar zählt zusätzlich mit, muss also >= sein)", measured, contentSize)
+	}
+
+	// Limit knapp UNTER dem tatsächlichen Verbrauch setzen — Postfach
+	// muss jetzt als "am Limit" gelten.
+	if err := mbStore.SetQuotaBytes(ctx, tenant, mailbox, measured-1); err != nil {
+		t.Fatalf("quota setzen: %v", err)
+	}
+
+	checker := quota.NewChecker(mbStore, usageCounter)
+	srv := NewServerWithMaxMessageBytesTLSLoggerRateLimitGuardAndQuota(
+		&fakeSink{}, defaultMaxMessageBytes, nil, nil, tenant, nil, protoguard.DefaultConfig(), checker,
+	)
+	addr, stop := startRealServer(t, srv)
+	defer stop()
+
+	c := dial(t, addr)
+	defer c.close()
+	c.send(t, "EHLO client.example.com")
+	for {
+		line := c.readLine(t)
+		if strings.HasPrefix(line, "250 ") {
+			break
+		}
+	}
+	c.send(t, "MAIL FROM:<absender@example.com>")
+	resp := c.send(t, "RCPT TO:<"+mailbox+">")
+	if code(resp) != "552" {
+		t.Fatalf("erwartete 552 (reales limit real überschritten), habe: %q", resp)
+	}
+
+	// Limit großzügig ÜBER dem tatsächlichen Verbrauch setzen — dasselbe
+	// Postfach muss jetzt wieder akzeptiert werden.
+	if err := mbStore.SetQuotaBytes(ctx, tenant, mailbox, measured*10); err != nil {
+		t.Fatalf("quota erhöhen: %v", err)
+	}
+	c2 := dial(t, addr)
+	defer c2.close()
+	c2.send(t, "EHLO client.example.com")
+	for {
+		line := c2.readLine(t)
+		if strings.HasPrefix(line, "250 ") {
+			break
+		}
+	}
+	c2.send(t, "MAIL FROM:<absender@example.com>")
+	resp2 := c2.send(t, "RCPT TO:<"+mailbox+">")
+	if code(resp2) != "250" {
+		t.Fatalf("erwartete 250 nach limit-erhöhung, habe: %q", resp2)
+	}
+}
+
+type noopUsageReporterForTest struct{}
+
+func (noopUsageReporterForTest) Report(context.Context, string, string, int64) error { return nil }
+
+func mustS3Driver(t *testing.T, ctx context.Context, bucket, endpoint string) *storage.S3Driver {
+	t.Helper()
+	driver, err := storage.NewS3Driver(ctx, bucket, "us-east-1", endpoint, os.Getenv("TEST_S3_ACCESS_KEY"), os.Getenv("TEST_S3_SECRET_KEY"), true)
+	if err != nil {
+		t.Fatalf("s3-driver: %v", err)
+	}
+	return driver
+}
+
+func startRealServer(t *testing.T, srv *Server) (addr string, stop func()) {
+	t.Helper()
+	listener, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatalf("listener: %v", err)
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan struct{})
+	go func() {
+		_ = srv.Serve(ctx, listener)
+		close(done)
+	}()
+	return listener.Addr().String(), func() {
+		cancel()
+		<-done
+	}
+}

mail/internal/smtp/arc09_quota_test.go

@@ -0,0 +1,155 @@
+package smtp
+
+import (
+	"context"
+	"net"
+	"strings"
+	"testing"
+
+	"gitea.perlbach24.de/scripte/nexarch/mail/internal/protoguard"
+)
+
+// fakeQuotaChecker ist ein In-Memory-Fake für QuotaChecker — reine
+// Protokolltests brauchen keine echte Postgres-/S3-Instanz.
+type fakeQuotaChecker struct {
+	blocked map[string]bool // mailbox -> am limit
+}
+
+func (f fakeQuotaChecker) Allowed(_ context.Context, _, mailbox string) (bool, error) {
+	return !f.blocked[mailbox], nil
+}
+
+func startQuotaTestServer(t *testing.T, sink MessageSink, checker QuotaChecker) (addr string, stop func()) {
+	t.Helper()
+	srv := NewServerWithMaxMessageBytesTLSLoggerRateLimitGuardAndQuota(
+		sink, defaultMaxMessageBytes, nil, nil, "mandant-arc09", nil, protoguard.DefaultConfig(), checker,
+	)
+	listener, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatalf("listener: %v", err)
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan struct{})
+	go func() {
+		_ = srv.Serve(ctx, listener)
+		close(done)
+	}()
+	return listener.Addr().String(), func() {
+		cancel()
+		<-done
+	}
+}
+
+// TestRcptTo_QuotaExceededRejectedWithCorrectSMTPError ist die
+// geforderte Pflichtprüfung 1 (ARC-09): Postfach-Quota erreicht, neue
+// eingehende Mail wird mit korrekter SMTP-Fehlermeldung (552, RFC 5321
+// "exceeded storage allocation") abgelehnt, NICHT stillschweigend
+// verworfen.
+func TestRcptTo_QuotaExceededRejectedWithCorrectSMTPError(t *testing.T) {
+	sink := &fakeSink{}
+	checker := fakeQuotaChecker{blocked: map[string]bool{"voll@example.com": true}}
+	addr, stop := startQuotaTestServer(t, sink, checker)
+	defer stop()
+
+	c := dial(t, addr)
+	defer c.close()
+	c.send(t, "EHLO client.example.com")
+	for {
+		line := c.readLine(t)
+		if strings.HasPrefix(line, "250 ") {
+			break
+		}
+	}
+	c.send(t, "MAIL FROM:<a@example.com>")
+
+	resp := c.send(t, "RCPT TO:<voll@example.com>")
+	if code(resp) != "552" {
+		t.Fatalf("erwartete 552 (quota überschritten), habe: %q", resp)
+	}
+	if !strings.Contains(resp, "quota") {
+		t.Fatalf("erwartete erkennbare quota-fehlermeldung, habe: %q", resp)
+	}
+
+	// Session bleibt funktionsfähig — kein Verbindungsabbruch, keine
+	// stillschweigende Verwerfung.
+	if resp := c.send(t, "NOOP"); code(resp) != "250" {
+		t.Fatalf("session nach quota-ablehnung nicht mehr funktionsfähig: %q", resp)
+	}
+	if sink.count() != 0 {
+		t.Fatalf("erwartete keine angenommene nachricht bei quota-ablehnung, habe %d", sink.count())
+	}
+}
+
+// TestRcptTo_OtherMailboxUnaffectedWhenOneAtLimit ist die geforderte
+// Pflichtprüfung 2 (ARC-09): ein anderes Postfach desselben Tenants
+// empfängt weiterhin normal, während eines am Limit ist.
+func TestRcptTo_OtherMailboxUnaffectedWhenOneAtLimit(t *testing.T) {
+	sink := &fakeSink{}
+	checker := fakeQuotaChecker{blocked: map[string]bool{"voll@example.com": true}}
+	addr, stop := startQuotaTestServer(t, sink, checker)
+	defer stop()
+
+	// Erste, VOLLSTÄNDIGE Transaktion an das Postfach am Limit —
+	// abgelehnt.
+	c1 := dial(t, addr)
+	c1.send(t, "EHLO client.example.com")
+	for {
+		line := c1.readLine(t)
+		if strings.HasPrefix(line, "250 ") {
+			break
+		}
+	}
+	c1.send(t, "MAIL FROM:<a@example.com>")
+	if resp := c1.send(t, "RCPT TO:<voll@example.com>"); code(resp) != "552" {
+		t.Fatalf("erwartete 552 für volles postfach, habe: %q", resp)
+	}
+	c1.close()
+
+	// Zweite, unabhängige Transaktion an ein ANDERES Postfach desselben
+	// Tenants — muss vollständig normal durchlaufen.
+	c2 := dial(t, addr)
+	defer c2.close()
+	c2.send(t, "EHLO client.example.com")
+	for {
+		line := c2.readLine(t)
+		if strings.HasPrefix(line, "250 ") {
+			break
+		}
+	}
+	c2.send(t, "MAIL FROM:<a@example.com>")
+	if resp := c2.send(t, "RCPT TO:<frei@example.com>"); code(resp) != "250" {
+		t.Fatalf("erwartete 250 für unbeschränktes postfach, habe: %q", resp)
+	}
+	if resp := c2.send(t, "DATA"); code(resp) != "354" {
+		t.Fatalf("erwartete 354, habe: %q", resp)
+	}
+	if resp := c2.send(t, "Subject: test\r\n\r\nHallo\r\n."); code(resp) != "250" {
+		t.Fatalf("erwartete 250 nach abgeschlossener DATA, habe: %q", resp)
+	}
+	if sink.count() != 1 {
+		t.Fatalf("erwartete 1 angenommene nachricht für das freie postfach, habe %d", sink.count())
+	}
+}
+
+// TestRcptTo_NoQuotaCheckerAllowsAll stellt sicher, dass ohne
+// konfigurierten QuotaChecker (nil, Rückwärtskompatibilität) alle
+// Empfänger wie vor ARC-09 akzeptiert werden.
+func TestRcptTo_NoQuotaCheckerAllowsAll(t *testing.T) {
+	sink := &fakeSink{}
+	addr, stop := startTestServer(t, sink, defaultMaxMessageBytes)
+	defer stop()
+
+	c := dial(t, addr)
+	defer c.close()
+	c.send(t, "EHLO client.example.com")
+	for {
+		line := c.readLine(t)
+		if strings.HasPrefix(line, "250 ") {
+			break
+		}
+	}
+	c.send(t, "MAIL FROM:<a@example.com>")
+	if resp := c.send(t, "RCPT TO:<beliebig@example.com>"); code(resp) != "250" {
+		t.Fatalf("erwartete 250 ohne quotachecker, habe: %q", resp)
+	}
+}

mail/internal/smtp/commands.go

@@ -95,7 +95,7 @@ func (s *Session) handleMailFrom(arg string) bool {
 // handleRcptTo ist Teil des Envelope-Aufbaus (Akzeptanzkriterium 1):
 // jede Empfängeradresse wird vor der Annahme validiert; mehrere RCPT TO
 // sind erlaubt.
-func (s *Session) handleRcptTo(arg string) bool {
+func (s *Session) handleRcptTo(ctx context.Context, arg string) bool {
 	if s.state != MailFromSet && s.state != RcptToSet {
 		return s.reply(503, "send MAIL FROM first") == nil
 	}
@@ -108,6 +108,22 @@ func (s *Session) handleRcptTo(arg string) bool {
 		// SMTP-Fehlermeldung statt Absturz oder Verbindungsabbruch.
 		return s.reply(553, "invalid recipient address") == nil
 	}
+	if s.quotaChecker != nil {
+		// ARC-09 Akzeptanzkriterium 2: ein Postfach am Limit lehnt neue
+		// eingehende Mail mit klarer, protokollgerechter SMTP-Fehlermeldung
+		// ab (552 = RFC 5321 "exceeded storage allocation") — an RCPT TO
+		// geprüft, BEVOR die Nachricht überhaupt übertragen wird, statt
+		// nach vollständigem DATA-Empfang zu verwerfen. addr (der
+		// Empfänger) ist der Postfachbezug — dasselbe mailbox-Feld wie
+		// mail/internal/storage.ArchiveKey/mail/internal/mailboxconfig.
+		allowed, err := s.quotaChecker.Allowed(ctx, s.tenantID, addr)
+		if err != nil {
+			return s.reply(451, "unable to verify mailbox quota, try again later") == nil
+		}
+		if !allowed {
+			return s.reply(552, "mailbox quota exceeded for "+addr) == nil
+		}
+	}
 	s.to = append(s.to, addr)
 	s.state = RcptToSet
 	return s.reply(250, "OK") == nil

mail/internal/smtp/server.go

@@ -38,6 +38,7 @@ type Server struct {
 
 	tenantID      string
 	acceptLimiter *ratelimit.Limiter
+	quotaChecker  QuotaChecker
 }
 
 func NewServer(sink MessageSink) *Server {
@@ -80,6 +81,14 @@ func NewServerWithMaxMessageBytesTLSLoggerRateLimitAndGuardConfig(sink MessageSi
 	return &Server{sink: sink, maxMessageBytes: maxMessageBytes, tlsConfig: tlsConfig, logger: logger, tenantID: tenantID, acceptLimiter: acceptLimiter, guardCfg: guardCfg}
 }
 
+// NewServerWithMaxMessageBytesTLSLoggerRateLimitGuardAndQuota erlaubt
+// zusätzlich Postfach-Quota-Prüfung bei RCPT TO (ARC-09).
+// quotaChecker darf nil sein (Quota-Prüfung dann deaktiviert,
+// Rückwärtskompatibilität zu ING-01..QA-04).
+func NewServerWithMaxMessageBytesTLSLoggerRateLimitGuardAndQuota(sink MessageSink, maxMessageBytes int64, tlsConfig *tls.Config, logger *slog.Logger, tenantID string, acceptLimiter *ratelimit.Limiter, guardCfg protoguard.Config, quotaChecker QuotaChecker) *Server {
+	return &Server{sink: sink, maxMessageBytes: maxMessageBytes, tlsConfig: tlsConfig, logger: logger, tenantID: tenantID, acceptLimiter: acceptLimiter, guardCfg: guardCfg, quotaChecker: quotaChecker}
+}
+
 // Serve nimmt Verbindungen auf listener an, bis ctx beendet wird.
 func (srv *Server) Serve(ctx context.Context, listener net.Listener) error {
 	go func() {
@@ -99,7 +108,7 @@ func (srv *Server) Serve(ctx context.Context, listener net.Listener) error {
 			}
 			return fmt.Errorf("smtp: verbindung annehmen: %w", err)
 		}
-		session := newSession(conn, srv.sink, srv.maxMessageBytes, srv.tlsConfig, srv.logger, srv.tenantID, srv.acceptLimiter, srv.guardCfg)
+		session := newSession(conn, srv.sink, srv.maxMessageBytes, srv.tlsConfig, srv.logger, srv.tenantID, srv.acceptLimiter, srv.guardCfg, srv.quotaChecker)
 		go session.Serve(ctx)
 	}
 }

mail/internal/smtp/session.go

@@ -15,6 +15,15 @@ import (
 	"gitea.perlbach24.de/scripte/nexarch/mail/internal/ratelimit"
 )
 
+// QuotaChecker prüft, ob ein Postfach noch innerhalb seines
+// konfigurierten Speicherlimits ist (ARC-09). Erfüllt von
+// *quota.Checker — als schmale Schnittstelle definiert, damit dieses
+// Paket quota nicht importieren muss (keine Kopplung an dessen
+// LimitProvider/UsageProvider-Abhängigkeiten).
+type QuotaChecker interface {
+	Allowed(ctx context.Context, tenantSlug, mailbox string) (bool, error)
+}
+
 // maxCommandLineBytes begrenzt eine einzelne Kommando-/DATA-Zeile
 // (defensive Fehlerbehandlung bei nicht-konformen Gegenstellen statt
 // optimistischem Parsing, gleiche Konvention wie mail/internal/imap und
@@ -46,6 +55,7 @@ type Session struct {
 
 	tenantID      string
 	acceptLimiter *ratelimit.Limiter // ING-09, nil = kein Rate-Limiting
+	quotaChecker  QuotaChecker       // ARC-09, nil = keine Quota-Prüfung
 
 	guard *protoguard.Guard // QA-04-Nachtrag zu ING-07
 
@@ -54,7 +64,7 @@ type Session struct {
 	to    []string
 }
 
-func newSession(conn net.Conn, sink MessageSink, maxMessageBytes int64, tlsConfig *tls.Config, logger *slog.Logger, tenantID string, acceptLimiter *ratelimit.Limiter, guardCfg protoguard.Config) *Session {
+func newSession(conn net.Conn, sink MessageSink, maxMessageBytes int64, tlsConfig *tls.Config, logger *slog.Logger, tenantID string, acceptLimiter *ratelimit.Limiter, guardCfg protoguard.Config, quotaChecker QuotaChecker) *Session {
 	_, alreadyTLS := conn.(*tls.Conn)
 	return &Session{
 		conn:            conn,
@@ -67,6 +77,7 @@ func newSession(conn net.Conn, sink MessageSink, maxMessageBytes int64, tlsConfi
 		log:             protolog.NewSessionLogger(logger, "smtp"),
 		tenantID:        tenantID,
 		acceptLimiter:   acceptLimiter,
+		quotaChecker:    quotaChecker,
 		guard:           protoguard.New(guardCfg),
 		state:           Greeting,
 	}
@@ -153,7 +164,7 @@ func (s *Session) dispatch(ctx context.Context, verb, arg string) bool {
 	case "MAIL":
 		return s.handleMailFrom(arg)
 	case "RCPT":
-		return s.handleRcptTo(arg)
+		return s.handleRcptTo(ctx, arg)
 	case "DATA":
 		return s.handleData(ctx)
 	case "RSET":

mail/internal/storage/archivekey.go

@@ -0,0 +1,75 @@
+// ARC-04: Aufbewahrungsstruktur nach Mandant/Postfach/Jahr. Mandant ist
+// bereits durch den physisch getrennten Bucket (ARC-06) abgebildet —
+// ArchiveKey deckt Postfach und Jahr INNERHALB des Buckets ab, als
+// eigener, additiver Schlüssel-Konstruktor neben dem bereits
+// bestehenden, flachen ObjectKey (ARC-01, weiterhin unverändert für
+// bestehende Aufrufer wie mail/internal/mailapi — kein Umbau
+// angrenzender Bereiche).
+package storage
+
+import (
+	"strconv"
+	"strings"
+	"time"
+)
+
+// Fallback-Segmente (Akzeptanzkriterium 3): fehlende/abweichende
+// Metadaten führen zu einem DOKUMENTIERTEN Fallback-Pfad statt einem
+// Ablagefehler.
+const (
+	FallbackMailboxSegment = "postfach-unbekannt"
+	FallbackYearSegment    = "jahr-unbekannt"
+)
+
+// ArchiveKey liefert den kanonischen Objektschlüssel für einen
+// archivierten Mail-Anhang/-Teil nach dem Schema Postfach/Jahr
+// (Akzeptanzkriterium 1) — innerhalb des bereits mandantenspezifischen
+// Buckets. sentAt darf der Nullwert sein und mailbox leer
+// (Akzeptanzkriterium 3): beides führt zum jeweiligen Fallback-Segment,
+// nie zu einem Fehler.
+func ArchiveKey(mailbox string, sentAt time.Time, messageID string, partIndex int) string {
+	return ArchiveYearPrefix(mailbox, yearOf(sentAt)) + ObjectKey(messageID, partIndex)
+}
+
+// ArchiveYearPrefix liefert den Verzeichnispräfix EINES Postfach-Jahrs
+// (Akzeptanzkriterium 2: Grundlage für spätere Retention-Regeln OHNE
+// Migration — ein Retention-Job kann alle Objekte eines Postfachs/
+// Jahrs direkt über diesen Präfix auflisten, ohne die bereits
+// abgelegten Schlüssel umzubenennen oder neu zu strukturieren).
+func ArchiveYearPrefix(mailbox string, year int) string {
+	mailboxSegment := sanitizeSegment(mailbox)
+	if mailboxSegment == "" {
+		mailboxSegment = FallbackMailboxSegment
+	}
+	yearSegment := FallbackYearSegment
+	if year > 0 {
+		yearSegment = strconv.Itoa(year)
+	}
+	return "postfach/" + mailboxSegment + "/" + yearSegment + "/"
+}
+
+// ArchiveMailboxPrefix liefert den Präfix ALLER Jahre eines Postfachs
+// (ohne Jahressegment) — Grundlage für eine reale
+// Postfach-Verbrauchsmessung über alle Jahre hinweg (ARC-09).
+func ArchiveMailboxPrefix(mailbox string) string {
+	mailboxSegment := sanitizeSegment(mailbox)
+	if mailboxSegment == "" {
+		mailboxSegment = FallbackMailboxSegment
+	}
+	return "postfach/" + mailboxSegment + "/"
+}
+
+func yearOf(t time.Time) int {
+	if t.IsZero() {
+		return 0
+	}
+	return t.UTC().Year()
+}
+
+// sanitizeSegment entfernt Pfadtrenner aus einem Postfachnamen, damit
+// er nie versehentlich zusätzliche Verzeichnisebenen erzeugt (z. B. ein
+// Postfachname, der ein "/" enthält).
+func sanitizeSegment(raw string) string {
+	raw = strings.TrimSpace(raw)
+	return strings.ReplaceAll(raw, "/", "_")
+}

mail/internal/storage/archivekey_test.go

@@ -0,0 +1,93 @@
+package storage
+
+import (
+	"strings"
+	"testing"
+	"time"
+)
+
+// TestArchiveKey_MultipleYearsProduceSeparateYearFolders ist die
+// geforderte Pflichtprüfung 1 (ARC-04): Import über mehrere Jahre
+// erzeugt korrekt getrennte Jahresordner.
+func TestArchiveKey_MultipleYearsProduceSeparateYearFolders(t *testing.T) {
+	sent2019 := time.Date(2019, 3, 1, 0, 0, 0, 0, time.UTC)
+	sent2024 := time.Date(2024, 11, 1, 0, 0, 0, 0, time.UTC)
+
+	key2019 := ArchiveKey("INBOX", sent2019, "msg-a", 0)
+	key2024 := ArchiveKey("INBOX", sent2024, "msg-b", 0)
+
+	if !strings.HasPrefix(key2019, "postfach/INBOX/2019/") {
+		t.Fatalf("erwartete jahresordner 2019, habe: %q", key2019)
+	}
+	if !strings.HasPrefix(key2024, "postfach/INBOX/2024/") {
+		t.Fatalf("erwartete jahresordner 2024, habe: %q", key2024)
+	}
+	if key2019 == key2024 {
+		t.Fatalf("erwartete unterschiedliche schlüssel für unterschiedliche jahre")
+	}
+}
+
+// TestArchiveKey_MissingMailboxUsesDocumentedFallback ist die
+// geforderte Pflichtprüfung 2 (ARC-04): fehlendes Postfach-Attribut
+// nutzt den dokumentierten Fallback statt eines Ablagefehlers.
+func TestArchiveKey_MissingMailboxUsesDocumentedFallback(t *testing.T) {
+	key := ArchiveKey("", time.Date(2024, 1, 1, 0, 0, 0, 0, time.UTC), "msg-a", 0)
+	if !strings.Contains(key, "/"+FallbackMailboxSegment+"/") {
+		t.Fatalf("erwartete fallback-postfach-segment %q, habe: %q", FallbackMailboxSegment, key)
+	}
+}
+
+// TestArchiveKey_MissingSentAtUsesDocumentedFallback belegt denselben
+// Fallback-Grundsatz für ein fehlendes (Null-)Sendedatum.
+func TestArchiveKey_MissingSentAtUsesDocumentedFallback(t *testing.T) {
+	key := ArchiveKey("INBOX", time.Time{}, "msg-a", 0)
+	if !strings.Contains(key, "/"+FallbackYearSegment+"/") {
+		t.Fatalf("erwartete fallback-jahr-segment %q, habe: %q", FallbackYearSegment, key)
+	}
+}
+
+// TestArchiveKey_NeverErrorsOnAnyInput bestätigt, dass ArchiveKey für
+// KEINE Eingabekombination fehlschlägt (Akzeptanzkriterium 3:
+// "definierter Fallback-Pfad, nicht Ablagefehler" — ArchiveKey liefert
+// bewusst keinen error-Rückgabewert, weil es strukturell keinen
+// Fehlerfall gibt).
+func TestArchiveKey_NeverErrorsOnAnyInput(t *testing.T) {
+	inputs := []struct {
+		mailbox string
+		sentAt  time.Time
+	}{
+		{"", time.Time{}},
+		{"Postfach/Mit/Slashes", time.Time{}},
+		{"  ", time.Date(1970, 1, 1, 0, 0, 0, 0, time.UTC)},
+	}
+	for _, in := range inputs {
+		key := ArchiveKey(in.mailbox, in.sentAt, "msg", 0)
+		if key == "" {
+			t.Fatalf("erwartete nicht-leeren schlüssel für eingabe %+v", in)
+		}
+	}
+}
+
+// TestArchiveYearPrefix_FoundationForRetentionWithoutMigration ist die
+// geforderte Pflichtprüfung für Akzeptanzkriterium 2: die Struktur ist
+// Grundlage für spätere Retention-Regeln ohne Migration — ein
+// Retention-Job kann den Präfix EINES Postfach-Jahrs berechnen und
+// findet darunter GENAU die zuvor mit ArchiveKey abgelegten Schlüssel
+// desselben Postfachs/Jahrs, ohne dass an den bereits abgelegten
+// Schlüsseln irgendetwas umbenannt werden müsste.
+func TestArchiveYearPrefix_FoundationForRetentionWithoutMigration(t *testing.T) {
+	sent2022 := time.Date(2022, 6, 15, 0, 0, 0, 0, time.UTC)
+	key := ArchiveKey("Rechnungen", sent2022, "msg-x", 3)
+
+	prefix := ArchiveYearPrefix("Rechnungen", 2022)
+	if !strings.HasPrefix(key, prefix) {
+		t.Fatalf("ArchiveKey %q liegt nicht unter dem für retention berechenbaren präfix %q", key, prefix)
+	}
+
+	// Ein anderes Jahr desselben Postfachs liegt NICHT unter demselben
+	// Präfix — Retention kann Jahre gezielt einzeln adressieren.
+	otherYearPrefix := ArchiveYearPrefix("Rechnungen", 2023)
+	if strings.HasPrefix(key, otherYearPrefix) {
+		t.Fatalf("ArchiveKey %q hätte NICHT unter dem 2023-präfix liegen dürfen", key)
+	}
+}

mail/internal/storage/usagecounter.go

@@ -0,0 +1,55 @@
+package storage
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/service/s3"
+)
+
+// UsageCounter misst den TATSÄCHLICHEN Speicherverbrauch eines
+// Postfachs (ARC-09) durch reales Auflisten der zugehörigen Objekte im
+// physisch getrennten Mandanten-Bucket (ARC-06) unter dem
+// ARC-04-Postfach-Präfix — kein separat gepflegter Zählerstand, der
+// von der tatsächlichen Ablage abweichen könnte.
+type UsageCounter struct {
+	admin *s3.Client
+}
+
+func NewUsageCounter(admin *s3.Client) *UsageCounter {
+	return &UsageCounter{admin: admin}
+}
+
+// UsageBytes summiert die reale Objektgröße aller unter
+// ArchiveMailboxPrefix(mailbox) abgelegten Objekte im Bucket von
+// tenantSlug (ARC-06: BucketNameForTenant). Erfüllt strukturell
+// quota.UsageProvider (ARC-09), ohne dass dieses Paket quota
+// importieren müsste.
+func (u *UsageCounter) UsageBytes(ctx context.Context, tenantSlug, mailbox string) (int64, error) {
+	bucket := BucketNameForTenant(tenantSlug)
+	prefix := ArchiveMailboxPrefix(mailbox)
+
+	var total int64
+	var continuationToken *string
+	for {
+		out, err := u.admin.ListObjectsV2(ctx, &s3.ListObjectsV2Input{
+			Bucket:            aws.String(bucket),
+			Prefix:            aws.String(prefix),
+			ContinuationToken: continuationToken,
+		})
+		if err != nil {
+			return 0, fmt.Errorf("storage: postfach-verbrauch auflisten: %w", err)
+		}
+		for _, obj := range out.Contents {
+			if obj.Size != nil {
+				total += *obj.Size
+			}
+		}
+		if out.IsTruncated == nil || !*out.IsTruncated {
+			break
+		}
+		continuationToken = out.NextContinuationToken
+	}
+	return total, nil
+}