diff --git a/dms/README.md b/dms/README.md index 4e7d6238..42896720 100644 --- a/dms/README.md +++ b/dms/README.md @@ -29,6 +29,10 @@ Fachliche Routen (z. B. `POST /retention/destroy-callback`) liegen unter dem Pr `scripts/check-route-prefixes.sh`); nur `/healthz` und `/metrics` bleiben auf Wurzelebene. Die bei archive registrierte Rückruf-URL (`NEXARCH_DMS_APP_RETENTION_CALLBACK_URL`) muss den Präfix enthalten. +## Objekt-Speicher + +`NEXARCH_DMS_APP_STORAGE_DRIVER` und `NEXARCH_DMS_WORKER_STORAGE_DRIVER` wählen den Treiber: `local` (Standard, braucht `…_STORAGE_DIR`) oder `s3`. Für `s3` sind `…_S3_BUCKET`, `…_S3_ACCESS_KEY_ID` und `…_S3_SECRET_ACCESS_KEY` Pflicht; optional `…_S3_REGION` (Standard `us-east-1`), `…_S3_ENDPOINT_URL` (MinIO, Ceph) und `…_S3_PATH_STYLE=1`. App und Worker müssen denselben Bucket nutzen. + ## Struktur - `cmd/app` — Anfrage-Dienst (HTTP), blockiert nie durch lange Aufgaben diff --git a/dms/cmd/app/api.go b/dms/cmd/app/api.go index 11b2a25c..23868e76 100644 --- a/dms/cmd/app/api.go +++ b/dms/cmd/app/api.go @@ -41,10 +41,10 @@ type uploadStack struct { } func newUploadStack(pool *pgxpool.Pool, stagingDir string) uploadStack { - storageDir := os.Getenv("NEXARCH_DMS_APP_STORAGE_DIR") - signingSecret := os.Getenv("NEXARCH_DMS_APP_STORAGE_SIGNING_SECRET") - publicBaseURL := os.Getenv("NEXARCH_DMS_APP_STORAGE_PUBLIC_BASE_URL") - driver := storage.NewLocalDriver(storageDir, []byte(signingSecret), publicBaseURL, signedURLMaxTTL(os.Getenv("NEXARCH_DMS_APP_SIGNED_URL_MAX_TTL"))) + driver, err := storage.NewDriverFromEnv(context.Background(), "NEXARCH_DMS_APP") + if err != nil { + log.Fatalf("dms-app: %v", err) + } usageReporter := storage.NewHTTPUsageReporter( os.Getenv("NEXARCH_DMS_APP_USAGE_ENDPOINT_URL"), os.Getenv("NEXARCH_DMS_APP_USAGE_CLIENT_ID"), diff --git a/dms/cmd/app/main.go b/dms/cmd/app/main.go index c544c12c..fbc36a03 100644 --- a/dms/cmd/app/main.go +++ b/dms/cmd/app/main.go @@ -194,10 +194,10 @@ func setupRetention(mux *http.ServeMux, dsn string) { log.Fatalf("dms-app: datenbankverbindung: %v", err) } - storageDir := os.Getenv("NEXARCH_DMS_APP_STORAGE_DIR") - signingSecret := os.Getenv("NEXARCH_DMS_APP_STORAGE_SIGNING_SECRET") - publicBaseURL := os.Getenv("NEXARCH_DMS_APP_STORAGE_PUBLIC_BASE_URL") - driver := storage.NewLocalDriver(storageDir, []byte(signingSecret), publicBaseURL, signedURLMaxTTL(os.Getenv("NEXARCH_DMS_APP_SIGNED_URL_MAX_TTL"))) + driver, err := storage.NewDriverFromEnv(ctx, "NEXARCH_DMS_APP") + if err != nil { + log.Fatalf("dms-app: %v", err) + } usageEndpoint := os.Getenv("NEXARCH_DMS_APP_USAGE_ENDPOINT_URL") usageClientID := os.Getenv("NEXARCH_DMS_APP_USAGE_CLIENT_ID") @@ -360,11 +360,3 @@ func setupWatchFolder(ctx context.Context, dir string) { } }() } - -func signedURLMaxTTL(raw string) time.Duration { - ttl, err := storage.ParseMaxSignedURLTTL(raw) - if err != nil { - log.Fatalf("dms-app: %v", err) - } - return ttl -} diff --git a/dms/cmd/worker/main.go b/dms/cmd/worker/main.go index 7743e0bb..f5c92ff0 100644 --- a/dms/cmd/worker/main.go +++ b/dms/cmd/worker/main.go @@ -133,14 +133,16 @@ func processRetentionRegisterJobs(ctx context.Context, queue *jobqueue.Queue, cl // startJobWorkers verdrahtet alle Fachjobs (FDN-17). Fehlt die Storage-/Schluessel- // Konfiguration, werden sie nicht gestartet und das wird gemeldet. func startJobWorkers(ctx context.Context, pool *pgxpool.Pool, queue *jobqueue.Queue, tracker *errortracking.Tracker) { - storageDir := os.Getenv("NEXARCH_DMS_WORKER_STORAGE_DIR") tenantSlug := os.Getenv("NEXARCH_DMS_WORKER_TENANT_SLUG") kekURL := os.Getenv("NEXARCH_DMS_WORKER_KEK_ENDPOINT_URL") - if storageDir == "" || tenantSlug == "" || kekURL == "" { - log.Print("dms-worker: NEXARCH_DMS_WORKER_STORAGE_DIR/_TENANT_SLUG/_KEK_ENDPOINT_URL unvollstaendig, Fachjobs (Vorschau, Konvertierung, Virenscan, Export, Texterkennung) werden NICHT verarbeitet") + if !storage.ConfiguredFromEnv("NEXARCH_DMS_WORKER") || tenantSlug == "" || kekURL == "" { + log.Print("dms-worker: NEXARCH_DMS_WORKER_STORAGE_DIR (oder _STORAGE_DRIVER=s3 mit _S3_BUCKET)/_TENANT_SLUG/_KEK_ENDPOINT_URL unvollstaendig, Fachjobs (Vorschau, Konvertierung, Virenscan, Export, Texterkennung) werden NICHT verarbeitet") return } - driver := storage.NewLocalDriver(storageDir, []byte(os.Getenv("NEXARCH_DMS_WORKER_STORAGE_SIGNING_SECRET")), os.Getenv("NEXARCH_DMS_WORKER_STORAGE_PUBLIC_BASE_URL"), signedURLMaxTTL(os.Getenv("NEXARCH_DMS_WORKER_SIGNED_URL_MAX_TTL"))) + driver, err := storage.NewDriverFromEnv(ctx, "NEXARCH_DMS_WORKER") + if err != nil { + log.Fatalf("dms-worker: %v", err) + } usage := storage.NewHTTPUsageReporter(os.Getenv("NEXARCH_DMS_WORKER_USAGE_ENDPOINT_URL"), os.Getenv("NEXARCH_DMS_WORKER_USAGE_CLIENT_ID"), os.Getenv("NEXARCH_DMS_WORKER_USAGE_CLIENT_SECRET"), nil) cryptoSvc := dmscrypto.NewService(dmscrypto.NewHTTPKEKProvider(kekURL, os.Getenv("NEXARCH_DMS_WORKER_KEK_CLIENT_ID"), os.Getenv("NEXARCH_DMS_WORKER_KEK_CLIENT_SECRET"), nil)) @@ -304,11 +306,3 @@ func startJobWorkers(ctx context.Context, pool *pgxpool.Pool, queue *jobqueue.Qu go integrity.RunScheduler(ctx, pool, queue, integritySched, 30*time.Second) } } - -func signedURLMaxTTL(raw string) time.Duration { - ttl, err := storage.ParseMaxSignedURLTTL(raw) - if err != nil { - log.Fatalf("dms-worker: %v", err) - } - return ttl -} diff --git a/dms/internal/storage/fromenv.go b/dms/internal/storage/fromenv.go new file mode 100644 index 00000000..e88f1e01 --- /dev/null +++ b/dms/internal/storage/fromenv.go @@ -0,0 +1,42 @@ +package storage + +import ( + "context" + "fmt" + "os" + "strings" +) + +// NewDriverFromEnv wählt den Treiber über _STORAGE_DRIVER ("local", +// Standard, oder "s3"). Zugangsdaten kommen nur aus der Umgebung. +func NewDriverFromEnv(ctx context.Context, prefix string) (Driver, error) { + env := func(name string) string { return os.Getenv(prefix + "_" + name) } + ttl, err := ParseMaxSignedURLTTL(env("SIGNED_URL_MAX_TTL")) + if err != nil { + return nil, err + } + switch strings.ToLower(env("STORAGE_DRIVER")) { + case "", "local": + return NewLocalDriver(env("STORAGE_DIR"), []byte(env("STORAGE_SIGNING_SECRET")), env("STORAGE_PUBLIC_BASE_URL"), ttl), nil + case "s3": + bucket, key, secret := env("S3_BUCKET"), env("S3_ACCESS_KEY_ID"), env("S3_SECRET_ACCESS_KEY") + if bucket == "" || key == "" || secret == "" { + return nil, fmt.Errorf("storage: %s_S3_BUCKET, _S3_ACCESS_KEY_ID und _S3_SECRET_ACCESS_KEY sind erforderlich", prefix) + } + region := env("S3_REGION") + if region == "" { + region = "us-east-1" + } + return NewS3Driver(ctx, bucket, region, env("S3_ENDPOINT_URL"), key, secret, env("S3_PATH_STYLE") == "1", ttl) + default: + return nil, fmt.Errorf("storage: %s_STORAGE_DRIVER %q unbekannt (local oder s3)", prefix, env("STORAGE_DRIVER")) + } +} + +// ConfiguredFromEnv meldet, ob für prefix ein Speicher konfiguriert ist. +func ConfiguredFromEnv(prefix string) bool { + if strings.EqualFold(os.Getenv(prefix+"_STORAGE_DRIVER"), "s3") { + return os.Getenv(prefix+"_S3_BUCKET") != "" + } + return os.Getenv(prefix+"_STORAGE_DIR") != "" +} diff --git a/dms/internal/storage/fromenv_test.go b/dms/internal/storage/fromenv_test.go new file mode 100644 index 00000000..1e68d78b --- /dev/null +++ b/dms/internal/storage/fromenv_test.go @@ -0,0 +1,69 @@ +package storage + +import ( + "context" + "testing" +) + +func TestNewDriverFromEnv(t *testing.T) { + ctx := context.Background() + + t.Run("Standard ist lokal", func(t *testing.T) { + t.Setenv("X_STORAGE_DIR", t.TempDir()) + d, err := NewDriverFromEnv(ctx, "X") + if err != nil { + t.Fatal(err) + } + if _, ok := d.(*LocalDriver); !ok { + t.Fatalf("erwartet *LocalDriver, bekam %T", d) + } + }) + + t.Run("s3 mit Zugangsdaten", func(t *testing.T) { + t.Setenv("X_STORAGE_DRIVER", "S3") + t.Setenv("X_S3_BUCKET", "b") + t.Setenv("X_S3_ACCESS_KEY_ID", "k") + t.Setenv("X_S3_SECRET_ACCESS_KEY", "s") + t.Setenv("X_S3_ENDPOINT_URL", "http://127.0.0.1:9000") + d, err := NewDriverFromEnv(ctx, "X") + if err != nil { + t.Fatal(err) + } + if _, ok := d.(*S3Driver); !ok { + t.Fatalf("erwartet *S3Driver, bekam %T", d) + } + }) + + t.Run("s3 ohne Zugangsdaten schlaegt fehl", func(t *testing.T) { + t.Setenv("X_STORAGE_DRIVER", "s3") + t.Setenv("X_S3_BUCKET", "b") + if _, err := NewDriverFromEnv(ctx, "X"); err == nil { + t.Fatal("erwartet Fehler ohne Schluessel") + } + }) + + t.Run("unbekannter Treiber", func(t *testing.T) { + t.Setenv("X_STORAGE_DRIVER", "ftp") + if _, err := NewDriverFromEnv(ctx, "X"); err == nil { + t.Fatal("erwartet Fehler fuer unbekannten Treiber") + } + }) +} + +func TestConfiguredFromEnv(t *testing.T) { + if ConfiguredFromEnv("Y") { + t.Fatal("ohne Variablen nicht konfiguriert") + } + t.Setenv("Y_STORAGE_DIR", "/x") + if !ConfiguredFromEnv("Y") { + t.Fatal("STORAGE_DIR zaehlt als konfiguriert") + } + t.Setenv("Y_STORAGE_DRIVER", "s3") + if ConfiguredFromEnv("Y") { + t.Fatal("s3 braucht den Bucket") + } + t.Setenv("Y_S3_BUCKET", "b") + if !ConfiguredFromEnv("Y") { + t.Fatal("s3 mit Bucket ist konfiguriert") + } +}