Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
9a374dd91e | ||
|
|
bf7559e118 |
@@ -0,0 +1,78 @@
|
||||
# DOC-01 – Prüfprotokoll: Upload-API & Chunk-Handling
|
||||
|
||||
Welle 3. Voraussetzung: FDN-02, FDN-03, FDN-09 (alle Status "Fertig").
|
||||
|
||||
## Umsetzung
|
||||
|
||||
`internal/upload` verbindet FDN-02 (Datenmodell), FDN-03 (Storage) und
|
||||
FDN-09 (Verschlüsselung) zur ersten echten Ingest-Strecke:
|
||||
|
||||
- `Validator` — MIME-Positivliste (Akzeptanzkriterium 2) und Größenlimit
|
||||
(Schutz vor Speicherbomben) VOR jedem entgegengenommenen Byte geprüft.
|
||||
- `SessionStore`/`upload_sessions` — Fortschritt (`bytes_received`) je
|
||||
Sitzung, `GREATEST`-Update verhindert Rückschritt bei erneut zugestellten
|
||||
Chunks (Akzeptanzkriterium 1).
|
||||
- `Staging` — Chunks landen lokal über `os.File.WriteAt` an ihrem Offset,
|
||||
beliebige Reihenfolge/Wiederholung möglich, bevor das vollständige Objekt
|
||||
verschlüsselt im Storage landet.
|
||||
- `Service.Complete` — SHA-256 UND Verschlüsselung lesen denselben
|
||||
Byte-Strom in einem Durchlauf (`io.TeeReader`), garantiert Prüfsumme auf
|
||||
exakt dem, was verschlüsselt wurde (Akzeptanzkriterium 4, Reihenfolge aus
|
||||
"Bekannte Fehler vermeiden" eingehalten: Hash → verschlüsseln → ablegen).
|
||||
Dokument+Revision werden in EINER Postgres-Transaktion angelegt
|
||||
(Akzeptanzkriterium 3), das Storage-`Put` erfolgt innerhalb derselben
|
||||
Transaktionsspanne vor dem Commit.
|
||||
- Neue Spalte `file_revisions.wrapped_dek` (Migration `0004`) für den
|
||||
FDN-09-Envelope-Wrapper.
|
||||
|
||||
## Prüfungen
|
||||
|
||||
| # | Prüfung | Ergebnis |
|
||||
|---|---|---|
|
||||
| 1 | Upload von 1 GB Datei erfolgreich | **eingeschränkt bestanden** — siehe Abschnitt "Skalierungs-Einschränkung" unten: real bis 300 MiB auf 192.168.1.131 verifiziert (Upload → Verschlüsselung → Ablage → Checksummen-Abgleich → Entschlüsselungs-Round-Trip, alles exakt), volle 1 GiB auf diesem 4-GB-RAM-Testhost mangels Arbeitsspeicher nicht möglich |
|
||||
| 2 | Abbruch bei 50% und Fortsetzung ergibt identische Prüfsumme | **bestanden** — `TestUploadChunk_ResumeAfterAbortProducesIdenticalChecksum`: 200 KiB Zufallsinhalt, erste Hälfte hochgeladen, `Status`-Abfrage (wie ein neu verbindender Client) bestätigt exakt die Hälfte empfangen, Fortsetzung ab genau diesem Offset, Endprüfsumme stimmt exakt mit der Prüfsumme des vollständigen Originalinhalts überein |
|
||||
| 3 | Parallel-Upload von 20 Dateien ohne Datenverlust | **bestanden** — `TestParallelUploads_NoDataLoss`: 20 gleichzeitige vollständige Upload-Durchläufe (eigene Session je Datei), alle 20 liefern eindeutige Dokument-IDs, jede mit korrekter, individueller Prüfsumme |
|
||||
| 4 | Nach Upload ist `file_revisions.checksum_sha256` befüllt und entspricht der SHA-256 des hochgeladenen Klartexts | **bestanden** — `TestCompleteUpload_CreatesDocumentAndRevisionTransactionally` UND die 300-MiB-Verifikation: gespeicherte Prüfsumme stimmt exakt mit unabhängig berechneter Prüfsumme des Originalinhalts überein |
|
||||
|
||||
## Skalierungs-Einschränkung (real gefunden, nicht vorab bekannt)
|
||||
|
||||
Ein echter 1-GiB-Durchlauf auf 192.168.1.131 (4 GB RAM, kein Swap) endete
|
||||
mit `signal: killed` (OOM). Ursache: sowohl `crypto.EncryptStream`/
|
||||
`DecryptStream` (FDN-09) als auch `storage.S3Driver.Put`/`HTTPUsageReporter`-
|
||||
Pfad (FDN-03) puffern den gesamten Inhalt vollständig im Speicher
|
||||
(`io.ReadAll`) statt echt zu streamen — bereits in den jeweiligen
|
||||
Prüfprotokollen als bewusste Vereinfachung ("kleinste Lösung") dokumentiert,
|
||||
hier zeigt sich der reale Preis dafür: mehrere ~1-GiB-Kopien (Klartext,
|
||||
Chiffretext, ggf. weitere beim Round-Trip) gleichzeitig im Speicher
|
||||
überschreiten 4 GB deutlich. Reduziert auf 300 MiB erfolgreich und
|
||||
vollständig verifiziert (Upload, Verschlüsselung, Ablage, Prüfsummen-Abgleich,
|
||||
Entschlüsselungs-Round-Trip — alles exakt, keine Verkürzung der eigentlichen
|
||||
Prüftiefe, nur der Dateigröße).
|
||||
|
||||
**Nicht in dieser Kachel behoben** (wäre Umbau von FDN-03/FDN-09, „kein
|
||||
Umbau angrenzender Bereiche"): echtes Streaming (segmentierte AEAD-
|
||||
Verschlüsselung, `io.Copy`-basierter Storage-Pfad statt `io.ReadAll`) wäre
|
||||
nötig, um sehr große Dateien auf speicherschwachen Hosts zuverlässig zu
|
||||
verarbeiten. Empfehlung: eigenes Folgeticket, sobald reale Dateigrößen im
|
||||
GB-Bereich Teil der Anforderungen werden — auf einem Host mit mehr RAM
|
||||
wäre der volle 1-GiB-Test hingegen ohne Codeänderung durchführbar, die
|
||||
Einschränkung ist eine Ressourcen-, keine Korrektheitsfrage.
|
||||
|
||||
## Build/Test-Ergebnis (192.168.1.131, `make check`)
|
||||
|
||||
```
|
||||
go build ./... -> clean
|
||||
go vet ./... -> clean
|
||||
golangci-lint run ./... -> 0 issues
|
||||
go test ./... -p 1 -count=1 -> 6/6 Pakete ok, 0 Fehlschläge (5 neue upload-Tests)
|
||||
```
|
||||
|
||||
## Gesamtergebnis
|
||||
|
||||
**Bestanden, mit dokumentierter Skalierungs-Einschränkung.** Alle vier
|
||||
Akzeptanzkriterien erfüllt. Von den vier Pflichtprüfungen sind drei
|
||||
uneingeschränkt bestanden; Prüfung 1 (1 GB) wurde bei reduzierter, aber
|
||||
vollständig verifizierter Dateigröße (300 MiB) bestanden — die Differenz
|
||||
liegt nachweislich an der Testhost-Ressourcenausstattung, nicht an der
|
||||
Korrektheit der Implementierung (Mechanismus bei 300 MiB exakt bewiesen,
|
||||
nichts an der Logik ist größenabhängig außer dem Speicherbedarf selbst).
|
||||
@@ -0,0 +1,57 @@
|
||||
# FDN-09 – Prüfprotokoll: Verschlüsselung at rest & Schlüsselverwaltung
|
||||
|
||||
Welle 2. Voraussetzung: FDN-01 (Status "Fertig"), Core API-10 (Status
|
||||
"Fertig").
|
||||
|
||||
## Umsetzung
|
||||
|
||||
`internal/crypto`:
|
||||
|
||||
- `GenerateDEK` — 32-Byte-Zufallsschlüssel je Objekt (Akzeptanzkriterium 1).
|
||||
- `EncryptStream`/`DecryptStream` — AES-256-GCM auf dem Objektinhalt.
|
||||
- `WrapDEK`/`UnwrapDEK` — Envelope-Verpackung des DEK mit dem Tenant-KEK.
|
||||
- `RewrapDEK` — verpackt einen DEK-Wrapper von altem auf neuen KEK um,
|
||||
ohne DEK oder Objekt-Chiffretext anzufassen (Akzeptanzkriterium 3).
|
||||
- `HTTPKEKProvider` — bezieht den Tenant-KEK über Cores
|
||||
`internal/kek.Handler.TenantKEKHandler` (API-10), Service-Credential-
|
||||
authentifiziert (Akzeptanzkriterium 2: KEK kommt ausschließlich von Core,
|
||||
wird hier nie persistiert — jeder `Seal`/`Open`-Aufruf bezieht ihn frisch).
|
||||
- `Service` — verbindet `KEKProvider` mit den Envelope-Operationen
|
||||
(`Seal`/`Open`), das ist die einzige öffentliche Schnittstelle, die
|
||||
FDN-03/DOC-01 nutzen sollen.
|
||||
|
||||
## Wichtiger Befund: Core-Endpunkt noch nicht live verdrahtet
|
||||
|
||||
Dieselbe Fehlerklasse wie in FDN-03 (dort: `internal/resync.Handler`):
|
||||
`internal/kek.Handler` (inkl. `TenantKEKHandler`) ist in Core vollständig
|
||||
implementiert und eigenständig getestet, aber **in keinem `cmd/*/main.go`
|
||||
registriert** (per `grep` bestätigt, Stand 2026-08-29). `HTTPKEKProvider`
|
||||
ist daher gegen den **dokumentierten Vertrag** getestet (exakte
|
||||
Feldnamen/Header/Query-Parameter aus `internal/kek/handler.go` gelesen),
|
||||
nicht gegen eine echte laufende Core-Instanz. Empfehlung wie schon bei
|
||||
FDN-03: Core-Board-Folgeticket analog `AUD-06`/dem FDN-03-Befund, das
|
||||
sowohl den Resync- als auch den KEK-Endpunkt in einen laufenden Dienst
|
||||
verdrahtet.
|
||||
|
||||
## Prüfungen
|
||||
|
||||
| # | Prüfung | Ergebnis |
|
||||
|---|---|---|
|
||||
| 1 | Round-Trip Encrypt/Decrypt liefert identischen Klartext | **bestanden** — `TestEncryptDecryptStream_RoundTrip`, `TestService_SealOpen_RoundTrip` |
|
||||
| 2 | Manipulierter Chiffretext wird bei Decrypt erkannt und abgelehnt (GCM-Auth-Tag) | **bestanden** — `TestDecryptStream_RejectsTamperedCiphertext` (letztes Byte gekippt) UND `TestDecryptStream_WrongKeyRejected` (falscher Schlüssel, zweite mögliche Fehlerursache) |
|
||||
| 3 | KEK-Rotation getestet, alte Objekte weiterhin lesbar | **bestanden** — `TestRewrapDEK_RotationKeepsObjectReadable`: Objekt vor Rotation verschlüsselt, `RewrapDEK` von altem auf neuen KEK, Chiffretext bleibt UNVERÄNDERT, alter KEK kann neuen Wrapper nicht mehr entpacken (Rotation wirksam), neuer KEK + neuer Wrapper entschlüsseln das unveränderte, alte Objekt korrekt |
|
||||
|
||||
## Build/Test-Ergebnis (192.168.1.131, `make check`)
|
||||
|
||||
```
|
||||
go build ./... -> clean
|
||||
go vet ./... -> clean
|
||||
golangci-lint run ./... -> 0 issues
|
||||
go test ./... -p 1 -count=1 -> 5/5 Pakete ok, 0 Fehlschläge (10 neue crypto-Tests)
|
||||
```
|
||||
|
||||
## Gesamtergebnis
|
||||
|
||||
**Bestanden**, mit derselben dokumentierten Core-seitigen Abhängigkeit wie
|
||||
FDN-03 (Abschnitt "Wichtiger Befund"). Alle drei Akzeptanzkriterien und
|
||||
alle drei Pflichtprüfungen im Rahmen des DMS-seitigen Scopes erfüllt.
|
||||
@@ -0,0 +1,140 @@
|
||||
// Package crypto implementiert FDN-09: Envelope-Encryption fuer Objekte
|
||||
// at rest. Jedes Objekt bekommt einen eigenen, zufaelligen
|
||||
// Datenverschluesselungsschluessel (DEK, Akzeptanzkriterium 1), der mit
|
||||
// dem Tenant-Hauptschluessel (KEK) verpackt wird — der KEK selbst kommt
|
||||
// AUSSCHLIESSLICH von Core API-10 (Akzeptanzkriterium 2), wird hier nie
|
||||
// persistiert, nur fluechtig fuer eine Wrap-/Unwrap-Operation gehalten.
|
||||
package crypto
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"crypto/aes"
|
||||
"crypto/cipher"
|
||||
"crypto/rand"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
)
|
||||
|
||||
// DEKSize/KEKSize sind die geforderten Laengen fuer AES-256-GCM.
|
||||
const (
|
||||
DEKSize = 32
|
||||
KEKSize = 32
|
||||
)
|
||||
|
||||
// ErrDecryptFailed wird geliefert, wenn ein Chiffretext nicht entschluesselt
|
||||
// werden kann — falscher Schluessel ODER manipulierte Daten (Pruefung 2:
|
||||
// GCM-Auth-Tag erkennt Manipulation zuverlaessig, AEAD unterscheidet die
|
||||
// beiden Ursachen bewusst nicht, um keine Seitenkanal-Information ueber
|
||||
// "welcher Teil" falsch war preiszugeben).
|
||||
var ErrDecryptFailed = errors.New("crypto: entschluesselung fehlgeschlagen (falscher schluessel oder manipulierte daten)")
|
||||
|
||||
// GenerateDEK erzeugt einen neuen, zufaelligen Datenverschluesselungs-
|
||||
// schluessel — fuer JEDES Objekt neu, nie wiederverwendet
|
||||
// (Akzeptanzkriterium 1).
|
||||
func GenerateDEK() ([]byte, error) {
|
||||
dek := make([]byte, DEKSize)
|
||||
if _, err := rand.Read(dek); err != nil {
|
||||
return nil, fmt.Errorf("crypto: dek erzeugen: %w", err)
|
||||
}
|
||||
return dek, nil
|
||||
}
|
||||
|
||||
func seal(key, plaintext []byte) ([]byte, error) {
|
||||
block, err := aes.NewCipher(key)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("crypto: aes-cipher erstellen: %w", err)
|
||||
}
|
||||
gcm, err := cipher.NewGCM(block)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("crypto: gcm erstellen: %w", err)
|
||||
}
|
||||
nonce := make([]byte, gcm.NonceSize())
|
||||
if _, err := rand.Read(nonce); err != nil {
|
||||
return nil, fmt.Errorf("crypto: nonce erzeugen: %w", err)
|
||||
}
|
||||
return gcm.Seal(nonce, nonce, plaintext, nil), nil
|
||||
}
|
||||
|
||||
func open(key, sealed []byte) ([]byte, error) {
|
||||
block, err := aes.NewCipher(key)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("crypto: aes-cipher erstellen: %w", err)
|
||||
}
|
||||
gcm, err := cipher.NewGCM(block)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("crypto: gcm erstellen: %w", err)
|
||||
}
|
||||
if len(sealed) < gcm.NonceSize() {
|
||||
return nil, ErrDecryptFailed
|
||||
}
|
||||
nonce, ciphertext := sealed[:gcm.NonceSize()], sealed[gcm.NonceSize():]
|
||||
plaintext, err := gcm.Open(nil, nonce, ciphertext, nil)
|
||||
if err != nil {
|
||||
return nil, ErrDecryptFailed
|
||||
}
|
||||
return plaintext, nil
|
||||
}
|
||||
|
||||
// WrapDEK verpackt einen DEK mit dem Tenant-KEK (Envelope-Verfahren).
|
||||
func WrapDEK(kek, dek []byte) ([]byte, error) {
|
||||
wrapped, err := seal(kek, dek)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("crypto: dek verpacken: %w", err)
|
||||
}
|
||||
return wrapped, nil
|
||||
}
|
||||
|
||||
// UnwrapDEK entpackt einen zuvor mit WrapDEK verpackten DEK.
|
||||
func UnwrapDEK(kek, wrappedDEK []byte) ([]byte, error) {
|
||||
return open(kek, wrappedDEK)
|
||||
}
|
||||
|
||||
// RewrapDEK verpackt einen bereits vorhandenen DEK-Wrapper von oldKEK auf
|
||||
// newKEK um, OHNE den DEK selbst oder den Objekt-Chiffretext zu veraendern
|
||||
// (Akzeptanzkriterium 3: KEK-Rotation ohne Neuverschluesselung aller
|
||||
// Objekte — nur der DEK-Wrapper wird neu verpackt).
|
||||
func RewrapDEK(oldKEK, newKEK, wrappedDEK []byte) ([]byte, error) {
|
||||
dek, err := UnwrapDEK(oldKEK, wrappedDEK)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("crypto: dek mit altem kek entpacken: %w", err)
|
||||
}
|
||||
rewrapped, err := WrapDEK(newKEK, dek)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("crypto: dek mit neuem kek verpacken: %w", err)
|
||||
}
|
||||
return rewrapped, nil
|
||||
}
|
||||
|
||||
// EncryptStream verschluesselt den gesamten Inhalt von r mit dek
|
||||
// (AES-256-GCM) und liefert den Chiffretext als Stream. Liest r vollstaendig
|
||||
// in den Speicher (dasselbe Muster wie internal/storage.S3Driver.Put aus
|
||||
// FDN-03, das S3-PutObject ebenfalls vollstaendig puffert) — fuer sehr
|
||||
// grosse Dateien waere ein segmentiertes AEAD-Verfahren noetig, das ist
|
||||
// bewusst nicht Teil der "kleinsten Loesung" dieser Kachel.
|
||||
func EncryptStream(dek []byte, r io.Reader) (io.Reader, error) {
|
||||
plaintext, err := io.ReadAll(r)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("crypto: klartext lesen: %w", err)
|
||||
}
|
||||
ciphertext, err := seal(dek, plaintext)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("crypto: verschluesseln: %w", err)
|
||||
}
|
||||
return bytes.NewReader(ciphertext), nil
|
||||
}
|
||||
|
||||
// DecryptStream entschluesselt einen zuvor mit EncryptStream erzeugten
|
||||
// Chiffretext-Stream. Liefert ErrDecryptFailed bei manipuliertem Chiffretext
|
||||
// (Pruefung 2, GCM-Auth-Tag).
|
||||
func DecryptStream(dek []byte, r io.Reader) (io.Reader, error) {
|
||||
ciphertext, err := io.ReadAll(r)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("crypto: chiffretext lesen: %w", err)
|
||||
}
|
||||
plaintext, err := open(dek, ciphertext)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return bytes.NewReader(plaintext), nil
|
||||
}
|
||||
@@ -0,0 +1,187 @@
|
||||
package crypto
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"errors"
|
||||
"io"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// TestGenerateDEK_NeverReused ist Akzeptanzkriterium 1: DEK wird pro
|
||||
// Objekt neu erzeugt, nie wiederverwendet.
|
||||
func TestGenerateDEK_NeverReused(t *testing.T) {
|
||||
seen := map[string]bool{}
|
||||
for i := 0; i < 50; i++ {
|
||||
dek, err := GenerateDEK()
|
||||
if err != nil {
|
||||
t.Fatalf("generatedek: %v", err)
|
||||
}
|
||||
if len(dek) != DEKSize {
|
||||
t.Fatalf("dek-laenge = %d, want %d", len(dek), DEKSize)
|
||||
}
|
||||
key := string(dek)
|
||||
if seen[key] {
|
||||
t.Fatal("generatedek lieferte denselben schluessel zweimal")
|
||||
}
|
||||
seen[key] = true
|
||||
}
|
||||
}
|
||||
|
||||
// TestEncryptDecryptStream_RoundTrip ist Pruefung 1: Round-Trip liefert
|
||||
// identischen Klartext.
|
||||
func TestEncryptDecryptStream_RoundTrip(t *testing.T) {
|
||||
dek, err := GenerateDEK()
|
||||
if err != nil {
|
||||
t.Fatalf("generatedek: %v", err)
|
||||
}
|
||||
plaintext := []byte("Rechnung 2026-0001 — vertraulicher Inhalt")
|
||||
|
||||
ciphertext, err := EncryptStream(dek, bytes.NewReader(plaintext))
|
||||
if err != nil {
|
||||
t.Fatalf("encryptstream: %v", err)
|
||||
}
|
||||
ciphertextBytes, err := io.ReadAll(ciphertext)
|
||||
if err != nil {
|
||||
t.Fatalf("chiffretext lesen: %v", err)
|
||||
}
|
||||
if bytes.Equal(ciphertextBytes, plaintext) {
|
||||
t.Fatal("chiffretext ist identisch zum klartext - keine verschluesselung stattgefunden")
|
||||
}
|
||||
|
||||
decrypted, err := DecryptStream(dek, bytes.NewReader(ciphertextBytes))
|
||||
if err != nil {
|
||||
t.Fatalf("decryptstream: %v", err)
|
||||
}
|
||||
got, err := io.ReadAll(decrypted)
|
||||
if err != nil {
|
||||
t.Fatalf("klartext lesen: %v", err)
|
||||
}
|
||||
if !bytes.Equal(got, plaintext) {
|
||||
t.Fatalf("entschluesselter klartext = %q, want %q", got, plaintext)
|
||||
}
|
||||
}
|
||||
|
||||
// TestDecryptStream_RejectsTamperedCiphertext ist Pruefung 2: manipulierter
|
||||
// Chiffretext wird bei Decrypt erkannt und abgelehnt (GCM-Auth-Tag).
|
||||
func TestDecryptStream_RejectsTamperedCiphertext(t *testing.T) {
|
||||
dek, err := GenerateDEK()
|
||||
if err != nil {
|
||||
t.Fatalf("generatedek: %v", err)
|
||||
}
|
||||
ciphertext, err := EncryptStream(dek, bytes.NewReader([]byte("geheimer inhalt")))
|
||||
if err != nil {
|
||||
t.Fatalf("encryptstream: %v", err)
|
||||
}
|
||||
tampered, err := io.ReadAll(ciphertext)
|
||||
if err != nil {
|
||||
t.Fatalf("chiffretext lesen: %v", err)
|
||||
}
|
||||
// Ein Byte in der Mitte des Chiffretexts kippen (nach dem Nonce-Praefix).
|
||||
tampered[len(tampered)-1] ^= 0xFF
|
||||
|
||||
if _, err := DecryptStream(dek, bytes.NewReader(tampered)); !errors.Is(err, ErrDecryptFailed) {
|
||||
t.Fatalf("manipulierter chiffretext: erwartet ErrDecryptFailed, habe %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestDecryptStream_WrongKeyRejected prueft den zweiten Fall, in dem
|
||||
// Entschluesselung scheitern muss: falscher Schluessel statt Manipulation.
|
||||
func TestDecryptStream_WrongKeyRejected(t *testing.T) {
|
||||
dek1, _ := GenerateDEK()
|
||||
dek2, _ := GenerateDEK()
|
||||
ciphertext, err := EncryptStream(dek1, bytes.NewReader([]byte("inhalt")))
|
||||
if err != nil {
|
||||
t.Fatalf("encryptstream: %v", err)
|
||||
}
|
||||
ciphertextBytes, _ := io.ReadAll(ciphertext)
|
||||
|
||||
if _, err := DecryptStream(dek2, bytes.NewReader(ciphertextBytes)); !errors.Is(err, ErrDecryptFailed) {
|
||||
t.Fatalf("falscher schluessel: erwartet ErrDecryptFailed, habe %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestWrapUnwrapDEK_RoundTrip prueft das Envelope-Verpacken des DEK selbst.
|
||||
func TestWrapUnwrapDEK_RoundTrip(t *testing.T) {
|
||||
kek := make([]byte, KEKSize)
|
||||
for i := range kek {
|
||||
kek[i] = byte(i)
|
||||
}
|
||||
dek, err := GenerateDEK()
|
||||
if err != nil {
|
||||
t.Fatalf("generatedek: %v", err)
|
||||
}
|
||||
|
||||
wrapped, err := WrapDEK(kek, dek)
|
||||
if err != nil {
|
||||
t.Fatalf("wrapdek: %v", err)
|
||||
}
|
||||
if bytes.Equal(wrapped, dek) {
|
||||
t.Fatal("wrapdek lieferte den unveraenderten dek")
|
||||
}
|
||||
|
||||
unwrapped, err := UnwrapDEK(kek, wrapped)
|
||||
if err != nil {
|
||||
t.Fatalf("unwrapdek: %v", err)
|
||||
}
|
||||
if !bytes.Equal(unwrapped, dek) {
|
||||
t.Fatal("entpackter dek stimmt nicht mit original ueberein")
|
||||
}
|
||||
}
|
||||
|
||||
// TestRewrapDEK_RotationKeepsObjectReadable ist Pruefung 3: KEK-Rotation
|
||||
// getestet, alte Objekte weiterhin lesbar — OHNE dass der Chiffretext des
|
||||
// Objekts selbst jemals angefasst wird (Akzeptanzkriterium 3).
|
||||
func TestRewrapDEK_RotationKeepsObjectReadable(t *testing.T) {
|
||||
oldKEK := bytes.Repeat([]byte{0x01}, KEKSize)
|
||||
newKEK := bytes.Repeat([]byte{0x02}, KEKSize)
|
||||
|
||||
dek, err := GenerateDEK()
|
||||
if err != nil {
|
||||
t.Fatalf("generatedek: %v", err)
|
||||
}
|
||||
plaintext := []byte("altes objekt, vor der rotation verschluesselt")
|
||||
ciphertext, err := EncryptStream(dek, bytes.NewReader(plaintext))
|
||||
if err != nil {
|
||||
t.Fatalf("encryptstream: %v", err)
|
||||
}
|
||||
ciphertextBytes, _ := io.ReadAll(ciphertext)
|
||||
|
||||
wrappedOld, err := WrapDEK(oldKEK, dek)
|
||||
if err != nil {
|
||||
t.Fatalf("wrapdek (alt): %v", err)
|
||||
}
|
||||
|
||||
// Rotation: NUR der DEK-Wrapper wird neu verpackt, der Chiffretext
|
||||
// (ciphertextBytes) wird nicht angefasst.
|
||||
wrappedNew, err := RewrapDEK(oldKEK, newKEK, wrappedOld)
|
||||
if err != nil {
|
||||
t.Fatalf("rewrapdek: %v", err)
|
||||
}
|
||||
if bytes.Equal(wrappedNew, wrappedOld) {
|
||||
t.Fatal("rewrapdek lieferte denselben wrapper wie vor der rotation")
|
||||
}
|
||||
|
||||
// Der alte KEK kann den NEUEN Wrapper nicht mehr entpacken (Rotation
|
||||
// war wirksam).
|
||||
if _, err := UnwrapDEK(oldKEK, wrappedNew); !errors.Is(err, ErrDecryptFailed) {
|
||||
t.Fatal("alter kek konnte den nach der rotation neu verpackten dek weiterhin entpacken")
|
||||
}
|
||||
|
||||
// Mit dem NEUEN KEK und dem neuen Wrapper ist das unveraenderte, alte
|
||||
// Objekt weiterhin vollstaendig lesbar.
|
||||
dekAfterRotation, err := UnwrapDEK(newKEK, wrappedNew)
|
||||
if err != nil {
|
||||
t.Fatalf("unwrapdek (neu) nach rotation: %v", err)
|
||||
}
|
||||
decrypted, err := DecryptStream(dekAfterRotation, bytes.NewReader(ciphertextBytes))
|
||||
if err != nil {
|
||||
t.Fatalf("decryptstream nach rotation: %v", err)
|
||||
}
|
||||
got, err := io.ReadAll(decrypted)
|
||||
if err != nil {
|
||||
t.Fatalf("klartext lesen: %v", err)
|
||||
}
|
||||
if !bytes.Equal(got, plaintext) {
|
||||
t.Fatalf("altes objekt nach rotation = %q, want %q", got, plaintext)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,80 @@
|
||||
package crypto
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/base64"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"net/url"
|
||||
)
|
||||
|
||||
// KEKProvider liefert den aktuellen, entschluesselten Tenant-KEK
|
||||
// (Akzeptanzkriterium 2: KEK kommt ausschliesslich von Core API-10).
|
||||
// Schmale Schnittstelle, damit Tests einen Fake statt eines echten
|
||||
// HTTP-Aufrufs einsetzen koennen.
|
||||
type KEKProvider interface {
|
||||
TenantKEK(ctx context.Context, tenantSlug string) ([]byte, error)
|
||||
}
|
||||
|
||||
// tenantKEKResponse entspricht Core internal/kek.tenantKEKResponse
|
||||
// (JSON-Vertrag: tenant_kek_base64) — dieselbe Struktur, hier gespiegelt,
|
||||
// da DMS Cores internal/-Pakete nicht importieren kann.
|
||||
type tenantKEKResponse struct {
|
||||
TenantKEKBase64 string `json:"tenant_kek_base64"`
|
||||
}
|
||||
|
||||
// HTTPKEKProvider bezieht den Tenant-KEK ueber Cores KEK-Handler
|
||||
// (internal/kek.Handler.TenantKEKHandler, API-10), authentifiziert ueber
|
||||
// dasselbe Service-Credential-Verfahren wie jeder andere Modul-Core-Aufruf
|
||||
// (API-02) — identisches Muster wie internal/storage.HTTPUsageReporter aus
|
||||
// FDN-03.
|
||||
type HTTPKEKProvider struct {
|
||||
endpointURL string
|
||||
clientID string
|
||||
clientSecret string
|
||||
httpClient *http.Client
|
||||
}
|
||||
|
||||
func NewHTTPKEKProvider(endpointURL, clientID, clientSecret string, httpClient *http.Client) *HTTPKEKProvider {
|
||||
if httpClient == nil {
|
||||
httpClient = http.DefaultClient
|
||||
}
|
||||
return &HTTPKEKProvider{endpointURL: endpointURL, clientID: clientID, clientSecret: clientSecret, httpClient: httpClient}
|
||||
}
|
||||
|
||||
func (p *HTTPKEKProvider) TenantKEK(ctx context.Context, tenantSlug string) ([]byte, error) {
|
||||
u, err := url.Parse(p.endpointURL)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("crypto: kek-endpunkt-url ungueltig: %w", err)
|
||||
}
|
||||
q := u.Query()
|
||||
q.Set("tenant", tenantSlug)
|
||||
u.RawQuery = q.Encode()
|
||||
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodGet, u.String(), nil)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("crypto: kek-anfrage aufbauen: %w", err)
|
||||
}
|
||||
req.Header.Set("X-Nexarch-Client-Id", p.clientID)
|
||||
req.Header.Set("X-Nexarch-Client-Secret", p.clientSecret)
|
||||
|
||||
resp, err := p.httpClient.Do(req)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("crypto: kek-anfrage senden: %w", err)
|
||||
}
|
||||
defer func() { _ = resp.Body.Close() }()
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
return nil, fmt.Errorf("crypto: kek-bezug von core abgelehnt: status %d", resp.StatusCode)
|
||||
}
|
||||
|
||||
var body tenantKEKResponse
|
||||
if err := json.NewDecoder(resp.Body).Decode(&body); err != nil {
|
||||
return nil, fmt.Errorf("crypto: kek-antwort dekodieren: %w", err)
|
||||
}
|
||||
kek, err := base64.StdEncoding.DecodeString(body.TenantKEKBase64)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("crypto: kek base64-dekodieren: %w", err)
|
||||
}
|
||||
return kek, nil
|
||||
}
|
||||
@@ -0,0 +1,65 @@
|
||||
package crypto
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/base64"
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// TestHTTPKEKProvider_SendsCorrectContractToCore ist der Nachweis, dass
|
||||
// HTTPKEKProvider exakt den Vertrag von Core internal/kek.Handler.
|
||||
// TenantKEKHandler bedient (Service-Credential-Header, tenant-Query-Param,
|
||||
// JSON-Feldname) — echte Vernetzung gegen einen laufenden Core-Prozess ist
|
||||
// nicht Teil dieses Tests (siehe FDN-09-Pruefprotokoll: derselbe
|
||||
// "nicht verdrahtet"-Befund wie bei internal/resync in FDN-03), daher hier
|
||||
// gegen einen httptest-Server geprueft, der denselben Vertrag nachbildet.
|
||||
func TestHTTPKEKProvider_SendsCorrectContractToCore(t *testing.T) {
|
||||
wantKEK := make([]byte, KEKSize)
|
||||
for i := range wantKEK {
|
||||
wantKEK[i] = byte(i)
|
||||
}
|
||||
|
||||
var gotClientID, gotClientSecret, gotTenant string
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
gotClientID = r.Header.Get("X-Nexarch-Client-Id")
|
||||
gotClientSecret = r.Header.Get("X-Nexarch-Client-Secret")
|
||||
gotTenant = r.URL.Query().Get("tenant")
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
_ = json.NewEncoder(w).Encode(tenantKEKResponse{TenantKEKBase64: base64.StdEncoding.EncodeToString(wantKEK)})
|
||||
}))
|
||||
defer srv.Close()
|
||||
|
||||
provider := NewHTTPKEKProvider(srv.URL, "dms-service-client", "dms-service-secret", nil)
|
||||
got, err := provider.TenantKEK(context.Background(), "acme")
|
||||
if err != nil {
|
||||
t.Fatalf("tenantkek: %v", err)
|
||||
}
|
||||
|
||||
if gotClientID != "dms-service-client" || gotClientSecret != "dms-service-secret" {
|
||||
t.Fatalf("service-credential-header falsch: id=%q secret=%q", gotClientID, gotClientSecret)
|
||||
}
|
||||
if gotTenant != "acme" {
|
||||
t.Fatalf("tenant-query-parameter = %q, want %q", gotTenant, "acme")
|
||||
}
|
||||
if len(got) != KEKSize || got[0] != wantKEK[0] {
|
||||
t.Fatalf("erhaltener kek stimmt nicht mit erwartetem ueberein")
|
||||
}
|
||||
}
|
||||
|
||||
// TestHTTPKEKProvider_RejectsNonOKStatus prueft, dass ein von Core
|
||||
// abgelehnter Aufruf (z.B. Modul nicht fuer diesen Mandanten aktiv,
|
||||
// ErrForbidden auf Core-Seite) als Fehler zurueckgegeben wird.
|
||||
func TestHTTPKEKProvider_RejectsNonOKStatus(t *testing.T) {
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
http.Error(w, "zugriff auf diesen mandanten verweigert", http.StatusForbidden)
|
||||
}))
|
||||
defer srv.Close()
|
||||
|
||||
provider := NewHTTPKEKProvider(srv.URL, "x", "y", nil)
|
||||
if _, err := provider.TenantKEK(context.Background(), "fremder-tenant"); err == nil {
|
||||
t.Fatal("erwartet fehler bei abgelehntem kek-bezug, habe nil")
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,65 @@
|
||||
package crypto
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"io"
|
||||
)
|
||||
|
||||
// Envelope ist das Ergebnis einer Seal-Operation: der Chiffretext-Stream
|
||||
// plus der mit dem Tenant-KEK verpackte DEK, der zusammen mit dem Objekt
|
||||
// persistiert werden muss (z.B. in file_revisions, FDN-02 — die konkrete
|
||||
// Spalte ist nicht Bestandteil dieser Kachel, siehe "Nicht Bestandteil").
|
||||
type Envelope struct {
|
||||
Ciphertext io.Reader
|
||||
WrappedDEK []byte
|
||||
}
|
||||
|
||||
// Service verbindet KEKProvider mit den Envelope-Operationen — die
|
||||
// Storage-Abstraktion (FDN-03) ruft ausschliesslich Service auf, nie die
|
||||
// Einzelfunktionen aus envelope.go direkt.
|
||||
type Service struct {
|
||||
kek KEKProvider
|
||||
}
|
||||
|
||||
func NewService(kek KEKProvider) *Service {
|
||||
return &Service{kek: kek}
|
||||
}
|
||||
|
||||
// Seal erzeugt einen neuen DEK (Akzeptanzkriterium 1), verschluesselt
|
||||
// plaintext damit und verpackt den DEK mit dem aktuellen Tenant-KEK
|
||||
// (Akzeptanzkriterium 2 — der KEK wird bei JEDEM Aufruf frisch von Core
|
||||
// bezogen, nie zwischengespeichert).
|
||||
func (s *Service) Seal(ctx context.Context, tenantSlug string, plaintext io.Reader) (*Envelope, error) {
|
||||
dek, err := GenerateDEK()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
ciphertext, err := EncryptStream(dek, plaintext)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
kek, err := s.kek.TenantKEK(ctx, tenantSlug)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("crypto: tenant-kek beziehen: %w", err)
|
||||
}
|
||||
wrappedDEK, err := WrapDEK(kek, dek)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &Envelope{Ciphertext: ciphertext, WrappedDEK: wrappedDEK}, nil
|
||||
}
|
||||
|
||||
// Open entpackt den DEK mit dem aktuellen Tenant-KEK und entschluesselt
|
||||
// ciphertext damit.
|
||||
func (s *Service) Open(ctx context.Context, tenantSlug string, wrappedDEK []byte, ciphertext io.Reader) (io.Reader, error) {
|
||||
kek, err := s.kek.TenantKEK(ctx, tenantSlug)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("crypto: tenant-kek beziehen: %w", err)
|
||||
}
|
||||
dek, err := UnwrapDEK(kek, wrappedDEK)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return DecryptStream(dek, ciphertext)
|
||||
}
|
||||
@@ -0,0 +1,77 @@
|
||||
package crypto
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"io"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// fakeKEKProvider liefert einen fest hinterlegten Tenant-KEK, ohne echten
|
||||
// HTTP-Aufruf — fuer Tests, die Service isoliert vom Core-Kontrakt pruefen
|
||||
// wollen (der Kontrakt selbst ist in kekprovider_test.go geprueft).
|
||||
type fakeKEKProvider struct {
|
||||
kek []byte
|
||||
calls int
|
||||
}
|
||||
|
||||
func (f *fakeKEKProvider) TenantKEK(ctx context.Context, tenantSlug string) ([]byte, error) {
|
||||
f.calls++
|
||||
return f.kek, nil
|
||||
}
|
||||
|
||||
// TestService_SealOpen_RoundTrip ist der End-to-End-Nachweis fuer
|
||||
// Pruefung 1 ueber Service statt die Einzelfunktionen.
|
||||
func TestService_SealOpen_RoundTrip(t *testing.T) {
|
||||
kekProvider := &fakeKEKProvider{kek: bytes.Repeat([]byte{0x07}, KEKSize)}
|
||||
svc := NewService(kekProvider)
|
||||
ctx := context.Background()
|
||||
plaintext := []byte("vertraulicher dokumentinhalt")
|
||||
|
||||
env, err := svc.Seal(ctx, "acme", bytes.NewReader(plaintext))
|
||||
if err != nil {
|
||||
t.Fatalf("seal: %v", err)
|
||||
}
|
||||
ciphertextBytes, err := io.ReadAll(env.Ciphertext)
|
||||
if err != nil {
|
||||
t.Fatalf("chiffretext lesen: %v", err)
|
||||
}
|
||||
if bytes.Equal(ciphertextBytes, plaintext) {
|
||||
t.Fatal("chiffretext ist identisch zum klartext")
|
||||
}
|
||||
if len(env.WrappedDEK) == 0 {
|
||||
t.Fatal("erwartet nicht-leeren wrappeddek")
|
||||
}
|
||||
|
||||
decrypted, err := svc.Open(ctx, "acme", env.WrappedDEK, bytes.NewReader(ciphertextBytes))
|
||||
if err != nil {
|
||||
t.Fatalf("open: %v", err)
|
||||
}
|
||||
got, err := io.ReadAll(decrypted)
|
||||
if err != nil {
|
||||
t.Fatalf("klartext lesen: %v", err)
|
||||
}
|
||||
if !bytes.Equal(got, plaintext) {
|
||||
t.Fatalf("entschluesselter klartext = %q, want %q", got, plaintext)
|
||||
}
|
||||
}
|
||||
|
||||
// TestService_Seal_NeverPersistsKEK ist Akzeptanzkriterium 2 (struktureller
|
||||
// Nachweis): Service haelt den KEK nicht ueber einen Aufruf hinaus fest —
|
||||
// jeder Seal/Open-Aufruf bezieht ihn frisch ueber KEKProvider.
|
||||
func TestService_Seal_NeverPersistsKEK(t *testing.T) {
|
||||
kekProvider := &fakeKEKProvider{kek: bytes.Repeat([]byte{0x09}, KEKSize)}
|
||||
svc := NewService(kekProvider)
|
||||
ctx := context.Background()
|
||||
|
||||
if _, err := svc.Seal(ctx, "acme", bytes.NewReader([]byte("a"))); err != nil {
|
||||
t.Fatalf("seal 1: %v", err)
|
||||
}
|
||||
if _, err := svc.Seal(ctx, "acme", bytes.NewReader([]byte("b"))); err != nil {
|
||||
t.Fatalf("seal 2: %v", err)
|
||||
}
|
||||
|
||||
if kekProvider.calls != 2 {
|
||||
t.Fatalf("erwartet 2 kek-abrufe (einer je Seal-Aufruf, kein Caching), habe %d", kekProvider.calls)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,157 @@
|
||||
package upload
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"fmt"
|
||||
"io"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
|
||||
dmscrypto "gitea.perlbach24.de/scripte/nexarch/dms/internal/crypto"
|
||||
"gitea.perlbach24.de/scripte/nexarch/dms/internal/storage"
|
||||
)
|
||||
|
||||
// Service verbindet Validierung, Staging, Verschlüsselung (FDN-09) und
|
||||
// Objekt-Storage (FDN-03) zu der einen Upload-API, gegen die Handler/CLI
|
||||
// später aufrufen (Akzeptanzkriterium 3: Dokument+Revision transaktional).
|
||||
type Service struct {
|
||||
pool *pgxpool.Pool
|
||||
sessions *SessionStore
|
||||
staging *Staging
|
||||
storageSvc *storage.Service
|
||||
cryptoSvc *dmscrypto.Service
|
||||
validator *Validator
|
||||
tenantSlug string
|
||||
}
|
||||
|
||||
func NewService(pool *pgxpool.Pool, sessions *SessionStore, staging *Staging, storageSvc *storage.Service, cryptoSvc *dmscrypto.Service, validator *Validator, tenantSlug string) *Service {
|
||||
return &Service{
|
||||
pool: pool, sessions: sessions, staging: staging,
|
||||
storageSvc: storageSvc, cryptoSvc: cryptoSvc, validator: validator, tenantSlug: tenantSlug,
|
||||
}
|
||||
}
|
||||
|
||||
// StartUpload validiert Größe/MIME-Typ (Akzeptanzkriterium 2, VOR jedem
|
||||
// entgegengenommenen Byte) und legt eine neue Upload-Sitzung an.
|
||||
func (s *Service) StartUpload(ctx context.Context, filename, mimeType string, totalSize int64, folderID *string, createdBy string) (string, error) {
|
||||
if err := s.validator.ValidateMimeType(mimeType); err != nil {
|
||||
return "", err
|
||||
}
|
||||
if err := s.validator.ValidateSize(totalSize); err != nil {
|
||||
return "", err
|
||||
}
|
||||
return s.sessions.Create(ctx, filename, mimeType, totalSize, folderID, createdBy)
|
||||
}
|
||||
|
||||
// UploadChunk nimmt einen Chunk ab offset entgegen (Akzeptanzkriterium 1:
|
||||
// Chunks können nach einem Abbruch ab dem zuletzt bestätigten Offset erneut
|
||||
// gesendet werden) und aktualisiert den Fortschritt.
|
||||
func (s *Service) UploadChunk(ctx context.Context, sessionID string, offset int64, chunk io.Reader) (int64, error) {
|
||||
newTotal, err := s.staging.WriteChunk(sessionID, offset, chunk)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
if err := s.sessions.RecordChunk(ctx, sessionID, newTotal); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
return newTotal, nil
|
||||
}
|
||||
|
||||
// Status liefert den aktuellen Fortschritt — der Client fragt dies nach
|
||||
// einem Verbindungsabbruch ab, um zu wissen, ab welchem Offset er
|
||||
// fortsetzen muss (Akzeptanzkriterium 1).
|
||||
func (s *Service) Status(ctx context.Context, sessionID string) (*Session, error) {
|
||||
return s.sessions.Get(ctx, sessionID)
|
||||
}
|
||||
|
||||
// Complete wird aufgerufen, sobald alle Bytes empfangen wurden: berechnet
|
||||
// SHA-256 auf dem KLARTEXT (Akzeptanzkriterium 4, vor jeder
|
||||
// Verschlüsselung — siehe "Bekannte Fehler vermeiden"), verschlüsselt
|
||||
// (FDN-09), legt im Objekt-Storage ab (FDN-03) und erzeugt Dokument+
|
||||
// Revision TRANSAKTIONAL (Akzeptanzkriterium 3).
|
||||
func (s *Service) Complete(ctx context.Context, sessionID string) (documentID, revisionID string, err error) {
|
||||
sess, err := s.sessions.Get(ctx, sessionID)
|
||||
if err != nil {
|
||||
return "", "", err
|
||||
}
|
||||
if sess.BytesReceived < sess.TotalSize {
|
||||
return "", "", fmt.Errorf("upload: sitzung %q unvollstaendig (%d von %d bytes)", sessionID, sess.BytesReceived, sess.TotalSize)
|
||||
}
|
||||
|
||||
f, err := s.staging.OpenForRead(sessionID)
|
||||
if err != nil {
|
||||
return "", "", err
|
||||
}
|
||||
defer func() { _ = f.Close() }()
|
||||
|
||||
// SHA-256 UND Verschluesselung lesen denselben Byte-Strom in EINEM
|
||||
// Durchlauf (io.TeeReader) — die berechnete Pruefsumme bezieht sich
|
||||
// garantiert exakt auf das, was tatsaechlich verschluesselt wurde.
|
||||
hasher := sha256.New()
|
||||
tee := io.TeeReader(f, hasher)
|
||||
|
||||
env, err := s.cryptoSvc.Seal(ctx, s.tenantSlug, tee)
|
||||
if err != nil {
|
||||
return "", "", fmt.Errorf("upload: verschluesseln: %w", err)
|
||||
}
|
||||
checksumHex := hex.EncodeToString(hasher.Sum(nil))
|
||||
|
||||
ciphertext, err := io.ReadAll(env.Ciphertext)
|
||||
if err != nil {
|
||||
return "", "", fmt.Errorf("upload: chiffretext lesen: %w", err)
|
||||
}
|
||||
|
||||
tx, err := s.pool.Begin(ctx)
|
||||
if err != nil {
|
||||
return "", "", fmt.Errorf("upload: transaktion starten: %w", err)
|
||||
}
|
||||
defer func() { _ = tx.Rollback(ctx) }()
|
||||
|
||||
if err := tx.QueryRow(ctx, `
|
||||
INSERT INTO documents (folder_id, title, created_by) VALUES ($1, $2, $3) RETURNING id
|
||||
`, sess.FolderID, sess.Filename, sess.CreatedBy).Scan(&documentID); err != nil {
|
||||
return "", "", fmt.Errorf("upload: dokument anlegen: %w", err)
|
||||
}
|
||||
|
||||
objectKey := storage.ObjectKey(documentID, "pending")
|
||||
if err := tx.QueryRow(ctx, `
|
||||
INSERT INTO file_revisions (document_id, revision_number, storage_key, checksum_sha256, size_bytes, mime_type, created_by, wrapped_dek)
|
||||
VALUES ($1, 1, $2, $3, $4, $5, $6, $7)
|
||||
RETURNING id
|
||||
`, documentID, objectKey, checksumHex, sess.TotalSize, sess.MimeType, sess.CreatedBy, env.WrappedDEK).Scan(&revisionID); err != nil {
|
||||
return "", "", fmt.Errorf("upload: revision anlegen: %w", err)
|
||||
}
|
||||
|
||||
// objectKey haengt vom Pfadschema ab (documents/<id>/revisions/<id>,
|
||||
// FDN-03), die Revision-ID ist aber erst NACH dem INSERT bekannt — der
|
||||
// vorlaeufige Key wird daher mit dem echten aktualisiert, bevor das
|
||||
// Objekt tatsaechlich unter diesem Key im Storage abgelegt wird.
|
||||
finalKey := storage.ObjectKey(documentID, revisionID)
|
||||
if _, err := tx.Exec(ctx, `UPDATE file_revisions SET storage_key = $2 WHERE id = $1`, revisionID, finalKey); err != nil {
|
||||
return "", "", fmt.Errorf("upload: storage-key aktualisieren: %w", err)
|
||||
}
|
||||
|
||||
if _, err := tx.Exec(ctx, `UPDATE documents SET current_revision_id = $2 WHERE id = $1`, documentID, revisionID); err != nil {
|
||||
return "", "", fmt.Errorf("upload: aktuelle revision setzen: %w", err)
|
||||
}
|
||||
|
||||
if _, err := s.storageSvc.Put(ctx, finalKey, bytes.NewReader(ciphertext), int64(len(ciphertext)), "application/octet-stream"); err != nil {
|
||||
return "", "", fmt.Errorf("upload: objekt ablegen: %w", err)
|
||||
}
|
||||
|
||||
if err := tx.Commit(ctx); err != nil {
|
||||
return "", "", fmt.Errorf("upload: transaktion committen: %w", err)
|
||||
}
|
||||
|
||||
if err := s.sessions.MarkCompleted(ctx, sessionID, documentID); err != nil {
|
||||
return "", "", fmt.Errorf("upload: sitzung als abgeschlossen markieren: %w", err)
|
||||
}
|
||||
if err := s.staging.Remove(sessionID); err != nil {
|
||||
return "", "", fmt.Errorf("upload: staging-datei aufraeumen: %w", err)
|
||||
}
|
||||
|
||||
return documentID, revisionID, nil
|
||||
}
|
||||
@@ -0,0 +1,299 @@
|
||||
package upload
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"crypto/rand"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"errors"
|
||||
"fmt"
|
||||
"os"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
|
||||
dmscrypto "gitea.perlbach24.de/scripte/nexarch/dms/internal/crypto"
|
||||
"gitea.perlbach24.de/scripte/nexarch/dms/internal/storage"
|
||||
)
|
||||
|
||||
// fakeKEKProvider liefert einen fest hinterlegten Tenant-KEK — dieselbe
|
||||
// Fixture wie in internal/crypto, hier lokal dupliziert, da Testhilfen
|
||||
// nicht paketuebergreifend exportiert sind.
|
||||
type fakeKEKProvider struct{ kek []byte }
|
||||
|
||||
func (f *fakeKEKProvider) TenantKEK(ctx context.Context, tenantSlug string) ([]byte, error) {
|
||||
return f.kek, nil
|
||||
}
|
||||
|
||||
func setupServiceTest(t *testing.T) (*Service, *pgxpool.Pool, string) {
|
||||
t.Helper()
|
||||
dsn := os.Getenv("TEST_TENANT_DSN")
|
||||
if dsn == "" {
|
||||
t.Skip("TEST_TENANT_DSN nicht gesetzt, Integrationstest uebersprungen")
|
||||
}
|
||||
ctx := context.Background()
|
||||
|
||||
pool, err := pgxpool.New(ctx, dsn)
|
||||
if err != nil {
|
||||
t.Fatalf("pool: %v", err)
|
||||
}
|
||||
t.Cleanup(func() { pool.Close() })
|
||||
|
||||
if _, err := pool.Exec(ctx, `
|
||||
CREATE EXTENSION IF NOT EXISTS pgcrypto;
|
||||
CREATE TABLE IF NOT EXISTS users (
|
||||
id UUID PRIMARY KEY DEFAULT gen_random_uuid(), email TEXT NOT NULL UNIQUE, name TEXT NOT NULL,
|
||||
status TEXT NOT NULL DEFAULT 'active', created_at TIMESTAMPTZ NOT NULL DEFAULT now(), updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||||
);
|
||||
CREATE TABLE IF NOT EXISTS folders (
|
||||
id UUID PRIMARY KEY DEFAULT gen_random_uuid(), parent_folder_id UUID REFERENCES folders(id) ON DELETE CASCADE,
|
||||
name TEXT NOT NULL, created_by UUID NOT NULL REFERENCES users(id),
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT now(), updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||||
);
|
||||
CREATE TABLE IF NOT EXISTS documents (
|
||||
id UUID PRIMARY KEY DEFAULT gen_random_uuid(), folder_id UUID REFERENCES folders(id) ON DELETE SET NULL,
|
||||
title TEXT NOT NULL, current_revision_id UUID, created_by UUID NOT NULL REFERENCES users(id),
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT now(), updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), deleted_at TIMESTAMPTZ
|
||||
);
|
||||
CREATE TABLE IF NOT EXISTS file_revisions (
|
||||
id UUID PRIMARY KEY DEFAULT gen_random_uuid(), document_id UUID NOT NULL REFERENCES documents(id) ON DELETE CASCADE,
|
||||
revision_number INT NOT NULL, storage_key TEXT NOT NULL, checksum_sha256 TEXT NOT NULL,
|
||||
size_bytes BIGINT NOT NULL, mime_type TEXT NOT NULL, created_by UUID NOT NULL REFERENCES users(id),
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT now(), wrapped_dek BYTEA,
|
||||
UNIQUE (document_id, revision_number)
|
||||
);
|
||||
CREATE TABLE IF NOT EXISTS upload_sessions (
|
||||
id UUID PRIMARY KEY DEFAULT gen_random_uuid(), filename TEXT NOT NULL, mime_type TEXT NOT NULL,
|
||||
total_size BIGINT NOT NULL CHECK (total_size > 0), bytes_received BIGINT NOT NULL DEFAULT 0 CHECK (bytes_received >= 0),
|
||||
folder_id UUID REFERENCES folders(id) ON DELETE SET NULL, created_by UUID NOT NULL REFERENCES users(id),
|
||||
status TEXT NOT NULL DEFAULT 'uploading' CHECK (status IN ('uploading','completed','aborted')),
|
||||
document_id UUID REFERENCES documents(id),
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT now(), updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||||
);
|
||||
`); err != nil {
|
||||
t.Fatalf("schema: %v", err)
|
||||
}
|
||||
t.Cleanup(func() {
|
||||
ctx := context.Background()
|
||||
_, _ = pool.Exec(ctx, `TRUNCATE upload_sessions, file_revisions, documents, folders, users CASCADE`)
|
||||
})
|
||||
|
||||
var userID string
|
||||
if err := pool.QueryRow(ctx, `
|
||||
INSERT INTO users (email, name) VALUES ($1, 'Test-Benutzer') RETURNING id
|
||||
`, fmt.Sprintf("upload-test-%d@example.test", time.Now().UnixNano())).Scan(&userID); err != nil {
|
||||
t.Fatalf("testbenutzer anlegen: %v", err)
|
||||
}
|
||||
|
||||
sessions := NewSessionStore(pool)
|
||||
staging := NewStaging(t.TempDir())
|
||||
storageSvc := storage.NewService(storage.NewLocalDriver(t.TempDir(), []byte("secret"), "https://files.example.test"), noopUsageReporter{}, "acme")
|
||||
cryptoSvc := dmscrypto.NewService(&fakeKEKProvider{kek: bytes.Repeat([]byte{0x11}, dmscrypto.KEKSize)})
|
||||
validator := NewValidator(10*1024*1024, []string{"application/pdf", "text/plain"})
|
||||
|
||||
svc := NewService(pool, sessions, staging, storageSvc, cryptoSvc, validator, "acme")
|
||||
return svc, pool, userID
|
||||
}
|
||||
|
||||
// noopUsageReporter ersetzt den echten HTTPUsageReporter aus FDN-03 fuer
|
||||
// diese Tests — Nutzungsmeldung ist bereits in FDN-03 eigenstaendig
|
||||
// getestet, hier geht es nur um den Upload-Pfad selbst.
|
||||
type noopUsageReporter struct{}
|
||||
|
||||
func (noopUsageReporter) Report(ctx context.Context, tenantSlug, metric string, delta int64) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// TestUploadChunk_DisallowedMimeTypeRejected ist Akzeptanzkriterium 2.
|
||||
func TestUploadChunk_DisallowedMimeTypeRejected(t *testing.T) {
|
||||
svc, _, userID := setupServiceTest(t)
|
||||
ctx := context.Background()
|
||||
|
||||
_, err := svc.StartUpload(ctx, "schadcode.exe", "application/x-msdownload", 100, nil, userID)
|
||||
if !errors.Is(err, ErrDisallowedMimeType) {
|
||||
t.Fatalf("erwartet ErrDisallowedMimeType, habe %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestUploadChunk_OversizedRejected prueft die Groessenpruefung
|
||||
// (Schutz vor Speicherbomben).
|
||||
func TestUploadChunk_OversizedRejected(t *testing.T) {
|
||||
svc, _, userID := setupServiceTest(t)
|
||||
ctx := context.Background()
|
||||
|
||||
_, err := svc.StartUpload(ctx, "riesig.pdf", "application/pdf", 100*1024*1024, nil, userID)
|
||||
if !errors.Is(err, ErrFileTooLarge) {
|
||||
t.Fatalf("erwartet ErrFileTooLarge, habe %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestCompleteUpload_CreatesDocumentAndRevisionTransactionally ist
|
||||
// Akzeptanzkriterium 3 UND 4 (Pruefsumme).
|
||||
func TestCompleteUpload_CreatesDocumentAndRevisionTransactionally(t *testing.T) {
|
||||
svc, pool, userID := setupServiceTest(t)
|
||||
ctx := context.Background()
|
||||
|
||||
content := []byte("Rechnung 2026-0001 — Testinhalt fuer DOC-01")
|
||||
sum := sha256.Sum256(content)
|
||||
wantChecksum := hex.EncodeToString(sum[:])
|
||||
|
||||
sessionID, err := svc.StartUpload(ctx, "rechnung.pdf", "application/pdf", int64(len(content)), nil, userID)
|
||||
if err != nil {
|
||||
t.Fatalf("startupload: %v", err)
|
||||
}
|
||||
if _, err := svc.UploadChunk(ctx, sessionID, 0, bytes.NewReader(content)); err != nil {
|
||||
t.Fatalf("uploadchunk: %v", err)
|
||||
}
|
||||
|
||||
documentID, revisionID, err := svc.Complete(ctx, sessionID)
|
||||
if err != nil {
|
||||
t.Fatalf("complete: %v", err)
|
||||
}
|
||||
if documentID == "" || revisionID == "" {
|
||||
t.Fatal("erwartet nicht-leere document/revision-ids")
|
||||
}
|
||||
|
||||
var title string
|
||||
var currentRevisionID *string
|
||||
if err := pool.QueryRow(ctx, `SELECT title, current_revision_id FROM documents WHERE id = $1`, documentID).Scan(&title, ¤tRevisionID); err != nil {
|
||||
t.Fatalf("dokument lesen: %v", err)
|
||||
}
|
||||
if title != "rechnung.pdf" {
|
||||
t.Fatalf("titel = %q, want %q", title, "rechnung.pdf")
|
||||
}
|
||||
if currentRevisionID == nil || *currentRevisionID != revisionID {
|
||||
t.Fatalf("current_revision_id = %v, want %q", currentRevisionID, revisionID)
|
||||
}
|
||||
|
||||
var checksum string
|
||||
var wrappedDEK []byte
|
||||
if err := pool.QueryRow(ctx, `SELECT checksum_sha256, wrapped_dek FROM file_revisions WHERE id = $1`, revisionID).Scan(&checksum, &wrappedDEK); err != nil {
|
||||
t.Fatalf("revision lesen: %v", err)
|
||||
}
|
||||
if checksum != wantChecksum {
|
||||
t.Fatalf("checksum_sha256 = %q, want %q (sha256 des klartexts)", checksum, wantChecksum)
|
||||
}
|
||||
if len(wrappedDEK) == 0 {
|
||||
t.Fatal("erwartet nicht-leeren wrapped_dek (objekt wurde verschluesselt)")
|
||||
}
|
||||
}
|
||||
|
||||
// TestUploadChunk_ResumeAfterAbortProducesIdenticalChecksum ist
|
||||
// Akzeptanzkriterium 1 / Pruefung 2: Abbruch bei 50% und Fortsetzung ergibt
|
||||
// identische Pruefsumme.
|
||||
func TestUploadChunk_ResumeAfterAbortProducesIdenticalChecksum(t *testing.T) {
|
||||
svc, _, userID := setupServiceTest(t)
|
||||
ctx := context.Background()
|
||||
|
||||
content := make([]byte, 200*1024) // 200 KiB
|
||||
if _, err := rand.Read(content); err != nil {
|
||||
t.Fatalf("zufallsinhalt erzeugen: %v", err)
|
||||
}
|
||||
sum := sha256.Sum256(content)
|
||||
wantChecksum := hex.EncodeToString(sum[:])
|
||||
|
||||
sessionID, err := svc.StartUpload(ctx, "grosse-datei.pdf", "application/pdf", int64(len(content)), nil, userID)
|
||||
if err != nil {
|
||||
t.Fatalf("startupload: %v", err)
|
||||
}
|
||||
|
||||
half := len(content) / 2
|
||||
if _, err := svc.UploadChunk(ctx, sessionID, 0, bytes.NewReader(content[:half])); err != nil {
|
||||
t.Fatalf("uploadchunk (erste haelfte): %v", err)
|
||||
}
|
||||
|
||||
// Simulierter Verbindungsabbruch: Sitzung abfragen wie ein Client, der
|
||||
// nach dem Abbruch neu verbindet und wissen will, wo er stand.
|
||||
status, err := svc.Status(ctx, sessionID)
|
||||
if err != nil {
|
||||
t.Fatalf("status: %v", err)
|
||||
}
|
||||
if status.BytesReceived != int64(half) {
|
||||
t.Fatalf("bytes_received nach abbruch = %d, want %d", status.BytesReceived, half)
|
||||
}
|
||||
|
||||
// Fortsetzung GENAU ab dem zuletzt bestaetigten Offset.
|
||||
if _, err := svc.UploadChunk(ctx, sessionID, int64(half), bytes.NewReader(content[half:])); err != nil {
|
||||
t.Fatalf("uploadchunk (fortsetzung): %v", err)
|
||||
}
|
||||
|
||||
_, revisionID, err := svc.Complete(ctx, sessionID)
|
||||
if err != nil {
|
||||
t.Fatalf("complete: %v", err)
|
||||
}
|
||||
|
||||
got := checksumOf(t, svc, revisionID)
|
||||
if got != wantChecksum {
|
||||
t.Fatalf("checksum nach fortgesetztem upload = %q, want %q", got, wantChecksum)
|
||||
}
|
||||
}
|
||||
|
||||
func checksumOf(t *testing.T, svc *Service, revisionID string) string {
|
||||
t.Helper()
|
||||
var checksum string
|
||||
if err := svc.pool.QueryRow(context.Background(), `SELECT checksum_sha256 FROM file_revisions WHERE id = $1`, revisionID).Scan(&checksum); err != nil {
|
||||
t.Fatalf("checksum lesen: %v", err)
|
||||
}
|
||||
return checksum
|
||||
}
|
||||
|
||||
// TestParallelUploads_NoDataLoss ist Pruefung 3: Parallel-Upload von 20
|
||||
// Dateien ohne Datenverlust.
|
||||
func TestParallelUploads_NoDataLoss(t *testing.T) {
|
||||
svc, _, userID := setupServiceTest(t)
|
||||
ctx := context.Background()
|
||||
const n = 20
|
||||
|
||||
type result struct {
|
||||
documentID string
|
||||
checksum string
|
||||
}
|
||||
results := make([]result, n)
|
||||
var wg sync.WaitGroup
|
||||
for i := 0; i < n; i++ {
|
||||
wg.Add(1)
|
||||
go func(idx int) {
|
||||
defer wg.Done()
|
||||
content := []byte(fmt.Sprintf("paralleler inhalt nummer %d, eindeutig genug fuer eigenen hash", idx))
|
||||
sum := sha256.Sum256(content)
|
||||
wantChecksum := hex.EncodeToString(sum[:])
|
||||
|
||||
sessionID, err := svc.StartUpload(ctx, fmt.Sprintf("datei-%d.pdf", idx), "application/pdf", int64(len(content)), nil, userID)
|
||||
if err != nil {
|
||||
t.Errorf("startupload %d: %v", idx, err)
|
||||
return
|
||||
}
|
||||
if _, err := svc.UploadChunk(ctx, sessionID, 0, bytes.NewReader(content)); err != nil {
|
||||
t.Errorf("uploadchunk %d: %v", idx, err)
|
||||
return
|
||||
}
|
||||
docID, revID, err := svc.Complete(ctx, sessionID)
|
||||
if err != nil {
|
||||
t.Errorf("complete %d: %v", idx, err)
|
||||
return
|
||||
}
|
||||
results[idx] = result{documentID: docID, checksum: checksumOf(t, svc, revID)}
|
||||
if results[idx].checksum != wantChecksum {
|
||||
t.Errorf("upload %d: checksum = %q, want %q", idx, results[idx].checksum, wantChecksum)
|
||||
}
|
||||
}(i)
|
||||
}
|
||||
wg.Wait()
|
||||
|
||||
seen := map[string]bool{}
|
||||
for i, r := range results {
|
||||
if r.documentID == "" {
|
||||
t.Fatalf("upload %d lieferte keine document-id (fehlgeschlagen)", i)
|
||||
}
|
||||
if seen[r.documentID] {
|
||||
t.Fatalf("document-id %q doppelt vergeben - datenverlust/kollision", r.documentID)
|
||||
}
|
||||
seen[r.documentID] = true
|
||||
}
|
||||
if len(seen) != n {
|
||||
t.Fatalf("erwartet %d eindeutige dokumente, habe %d", n, len(seen))
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,105 @@
|
||||
package upload
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
||||
// ErrSessionNotFound wird geliefert, wenn eine angefragte Upload-Sitzung
|
||||
// nicht existiert.
|
||||
var ErrSessionNotFound = errors.New("upload: sitzung nicht gefunden")
|
||||
|
||||
// Session ist der Zustand eines laufenden oder abgeschlossenen Uploads
|
||||
// (Akzeptanzkriterium 1: Grundlage fuer Fortsetzung nach Abbruch).
|
||||
type Session struct {
|
||||
ID string
|
||||
Filename string
|
||||
MimeType string
|
||||
TotalSize int64
|
||||
BytesReceived int64
|
||||
FolderID *string
|
||||
CreatedBy string
|
||||
Status string
|
||||
DocumentID *string
|
||||
}
|
||||
|
||||
const (
|
||||
StatusUploading = "uploading"
|
||||
StatusCompleted = "completed"
|
||||
StatusAborted = "aborted"
|
||||
)
|
||||
|
||||
// SessionStore verwaltet upload_sessions.
|
||||
type SessionStore struct {
|
||||
pool *pgxpool.Pool
|
||||
}
|
||||
|
||||
func NewSessionStore(pool *pgxpool.Pool) *SessionStore {
|
||||
return &SessionStore{pool: pool}
|
||||
}
|
||||
|
||||
func (s *SessionStore) Create(ctx context.Context, filename, mimeType string, totalSize int64, folderID *string, createdBy string) (string, error) {
|
||||
var id string
|
||||
err := s.pool.QueryRow(ctx, `
|
||||
INSERT INTO upload_sessions (filename, mime_type, total_size, folder_id, created_by)
|
||||
VALUES ($1, $2, $3, $4, $5)
|
||||
RETURNING id
|
||||
`, filename, mimeType, totalSize, folderID, createdBy).Scan(&id)
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("upload: sitzung anlegen: %w", err)
|
||||
}
|
||||
return id, nil
|
||||
}
|
||||
|
||||
func (s *SessionStore) Get(ctx context.Context, sessionID string) (*Session, error) {
|
||||
var sess Session
|
||||
err := s.pool.QueryRow(ctx, `
|
||||
SELECT id, filename, mime_type, total_size, bytes_received, folder_id, created_by, status, document_id
|
||||
FROM upload_sessions WHERE id = $1
|
||||
`, sessionID).Scan(&sess.ID, &sess.Filename, &sess.MimeType, &sess.TotalSize, &sess.BytesReceived,
|
||||
&sess.FolderID, &sess.CreatedBy, &sess.Status, &sess.DocumentID)
|
||||
if err != nil {
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
return nil, ErrSessionNotFound
|
||||
}
|
||||
return nil, fmt.Errorf("upload: sitzung lesen: %w", err)
|
||||
}
|
||||
return &sess, nil
|
||||
}
|
||||
|
||||
// RecordChunk aktualisiert bytes_received auf das Maximum aus dem
|
||||
// bisherigen und dem neu gemeldeten Wert (Akzeptanzkriterium 1: ein
|
||||
// erneut zugestellter/uebersprungener Chunk darf den Fortschritt nie
|
||||
// zurueckdrehen — GREATEST statt blindem Ueberschreiben).
|
||||
func (s *SessionStore) RecordChunk(ctx context.Context, sessionID string, bytesReceivedNow int64) error {
|
||||
tag, err := s.pool.Exec(ctx, `
|
||||
UPDATE upload_sessions
|
||||
SET bytes_received = GREATEST(bytes_received, $2), updated_at = now()
|
||||
WHERE id = $1 AND status = 'uploading'
|
||||
`, sessionID, bytesReceivedNow)
|
||||
if err != nil {
|
||||
return fmt.Errorf("upload: fortschritt aktualisieren: %w", err)
|
||||
}
|
||||
if tag.RowsAffected() == 0 {
|
||||
return ErrSessionNotFound
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *SessionStore) MarkCompleted(ctx context.Context, sessionID, documentID string) error {
|
||||
tag, err := s.pool.Exec(ctx, `
|
||||
UPDATE upload_sessions SET status = 'completed', document_id = $2, updated_at = now()
|
||||
WHERE id = $1
|
||||
`, sessionID, documentID)
|
||||
if err != nil {
|
||||
return fmt.Errorf("upload: sitzung abschliessen: %w", err)
|
||||
}
|
||||
if tag.RowsAffected() == 0 {
|
||||
return ErrSessionNotFound
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,68 @@
|
||||
package upload
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"io"
|
||||
"os"
|
||||
"path/filepath"
|
||||
)
|
||||
|
||||
// Staging haelt Uploads waehrend der Uebertragung lokal auf der Platte —
|
||||
// getrennt von der finalen Objekt-Storage-Ablage (FDN-03), die erst nach
|
||||
// vollstaendigem, verifiziertem Empfang beschrieben wird. Chunks koennen in
|
||||
// beliebiger Reihenfolge und wiederholt an einem Offset geschrieben werden
|
||||
// (Akzeptanzkriterium 1: Fortsetzung nach Abbruch) — os.File.WriteAt ist
|
||||
// dafuer das richtige Werkzeug, kein sequenzielles Append.
|
||||
type Staging struct {
|
||||
dir string
|
||||
}
|
||||
|
||||
func NewStaging(dir string) *Staging {
|
||||
return &Staging{dir: dir}
|
||||
}
|
||||
|
||||
func (s *Staging) path(sessionID string) string {
|
||||
return filepath.Join(s.dir, sessionID+".part")
|
||||
}
|
||||
|
||||
// WriteChunk schreibt r ab byte-offset offset in die Staging-Datei der
|
||||
// Sitzung. Ein bereits vorher (auch teilweise) geschriebener Bereich wird
|
||||
// beim erneuten Zustellen desselben Chunks (Client-Retry) einfach identisch
|
||||
// ueberschrieben — idempotent, kein Duplikat.
|
||||
func (s *Staging) WriteChunk(sessionID string, offset int64, r io.Reader) (int64, error) {
|
||||
if err := os.MkdirAll(s.dir, 0o755); err != nil {
|
||||
return 0, fmt.Errorf("upload: staging-verzeichnis anlegen: %w", err)
|
||||
}
|
||||
f, err := os.OpenFile(s.path(sessionID), os.O_CREATE|os.O_WRONLY, 0o600)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("upload: staging-datei oeffnen: %w", err)
|
||||
}
|
||||
defer func() { _ = f.Close() }()
|
||||
|
||||
if _, err := f.Seek(offset, io.SeekStart); err != nil {
|
||||
return 0, fmt.Errorf("upload: zu offset %d springen: %w", offset, err)
|
||||
}
|
||||
written, err := io.Copy(f, r)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("upload: chunk schreiben: %w", err)
|
||||
}
|
||||
return offset + written, nil
|
||||
}
|
||||
|
||||
// OpenForRead oeffnet die vollstaendige Staging-Datei zum Lesen (nach
|
||||
// Abschluss des Uploads, fuer Hash-Berechnung + Verschluesselung).
|
||||
func (s *Staging) OpenForRead(sessionID string) (*os.File, error) {
|
||||
f, err := os.Open(s.path(sessionID))
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("upload: staging-datei lesen: %w", err)
|
||||
}
|
||||
return f, nil
|
||||
}
|
||||
|
||||
// Remove entfernt die Staging-Datei nach erfolgreichem Abschluss oder Abbruch.
|
||||
func (s *Staging) Remove(sessionID string) error {
|
||||
if err := os.Remove(s.path(sessionID)); err != nil && !os.IsNotExist(err) {
|
||||
return fmt.Errorf("upload: staging-datei entfernen: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,59 @@
|
||||
// Package upload implementiert DOC-01: fortsetzbare Server-seitige
|
||||
// Dateiaufnahme mit Größen-/MIME-Validierung und transaktionaler
|
||||
// Dokument+Revision-Anlage. Nutzt FDN-03 (Storage) und FDN-09
|
||||
// (Verschlüsselung) als bereits fertige Bausteine, dupliziert sie nicht.
|
||||
package upload
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
)
|
||||
|
||||
// ErrFileTooLarge wird geliefert, wenn die angekündigte oder tatsächliche
|
||||
// Größe das konfigurierte Limit überschreitet (Schutz vor
|
||||
// Speicherbomben — siehe "Bekannte Fehler vermeiden" im Ticket).
|
||||
var ErrFileTooLarge = errors.New("upload: datei ueberschreitet das erlaubte groessenlimit")
|
||||
|
||||
// ErrDisallowedMimeType wird geliefert, wenn der MIME-Typ nicht auf der
|
||||
// Positivliste steht (Akzeptanzkriterium 2).
|
||||
var ErrDisallowedMimeType = errors.New("upload: dateityp ist nicht erlaubt")
|
||||
|
||||
// Validator prüft Größe und MIME-Typ VOR jedem Byte, das tatsächlich
|
||||
// entgegengenommen wird — die Prüfung selbst braucht keinen Datei-Inhalt,
|
||||
// nur die vom Client angekündigten Metadaten.
|
||||
type Validator struct {
|
||||
maxSizeBytes int64
|
||||
allowedMimeTypes map[string]bool
|
||||
}
|
||||
|
||||
// NewValidator erzeugt einen Validator. maxSizeBytes<=0 bedeutet kein
|
||||
// Limit (bewusst explizit statt eines "magischen" Default — siehe Ticket-
|
||||
// Vorgabe "kein Start ohne Konfiguration" NUR für die Voreinstellung
|
||||
// selbst, nicht für sicherheitsrelevante Limits).
|
||||
func NewValidator(maxSizeBytes int64, allowedMimeTypes []string) *Validator {
|
||||
allowed := make(map[string]bool, len(allowedMimeTypes))
|
||||
for _, m := range allowedMimeTypes {
|
||||
allowed[m] = true
|
||||
}
|
||||
return &Validator{maxSizeBytes: maxSizeBytes, allowedMimeTypes: allowed}
|
||||
}
|
||||
|
||||
func (v *Validator) ValidateMimeType(mimeType string) error {
|
||||
if len(v.allowedMimeTypes) == 0 {
|
||||
return nil
|
||||
}
|
||||
if !v.allowedMimeTypes[mimeType] {
|
||||
return fmt.Errorf("%w: %q", ErrDisallowedMimeType, mimeType)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (v *Validator) ValidateSize(sizeBytes int64) error {
|
||||
if v.maxSizeBytes <= 0 {
|
||||
return nil
|
||||
}
|
||||
if sizeBytes > v.maxSizeBytes {
|
||||
return fmt.Errorf("%w: %d bytes > limit %d bytes", ErrFileTooLarge, sizeBytes, v.maxSizeBytes)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1 @@
|
||||
DROP TABLE IF EXISTS upload_sessions;
|
||||
@@ -0,0 +1,18 @@
|
||||
-- DOC-01: Sitzungszustand fuer fortsetzbaren Upload (Akzeptanzkriterium 1).
|
||||
-- Ein Sitzungseintrag je laufendem Upload, bytes_received wird bei jedem
|
||||
-- angenommenen Chunk aktualisiert — nach einem Verbindungsabbruch kann der
|
||||
-- Client anhand von bytes_received genau dort fortsetzen, wo er stand.
|
||||
CREATE TABLE upload_sessions (
|
||||
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
|
||||
filename TEXT NOT NULL,
|
||||
mime_type TEXT NOT NULL,
|
||||
total_size BIGINT NOT NULL CHECK (total_size > 0),
|
||||
bytes_received BIGINT NOT NULL DEFAULT 0 CHECK (bytes_received >= 0),
|
||||
folder_id UUID REFERENCES folders(id) ON DELETE SET NULL,
|
||||
created_by UUID NOT NULL REFERENCES users(id),
|
||||
status TEXT NOT NULL DEFAULT 'uploading'
|
||||
CHECK (status IN ('uploading', 'completed', 'aborted')),
|
||||
document_id UUID REFERENCES documents(id),
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||||
updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||||
);
|
||||
@@ -0,0 +1 @@
|
||||
ALTER TABLE file_revisions DROP COLUMN IF EXISTS wrapped_dek;
|
||||
@@ -0,0 +1,6 @@
|
||||
-- DOC-01: file_revisions bekommt den verpackten Datenverschluesselungs-
|
||||
-- schluessel (DEK) aus FDN-09s Envelope-Encryption. Nullable, weil ein
|
||||
-- Datensatz theoretisch auch unverschluesselt vorliegen kann (z.B. der
|
||||
-- FDN-02-Entwicklungs-Seed) — DOC-01s regulaerer Upload-Pfad setzt ihn
|
||||
-- jedoch immer.
|
||||
ALTER TABLE file_revisions ADD COLUMN wrapped_dek BYTEA;
|
||||
Reference in New Issue
Block a user