dms/backup-tool: Quelle und Ziel waehlbar lokal oder S3

Backup und Restore nutzen NewDriverFromEnv mit den Praefixen
NEXARCH_DMS_BACKUP_ und NEXARCH_DMS_RESTORE_. Ohne STORAGE_DRIVER bleibt
es beim lokalen Verzeichnis. Der Drill-Test laeuft mit TEST_S3_SRC_BUCKET
und TEST_S3_BUCKET gegen S3.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>

connect/cmd/connect-app/main.go

@@ -18,9 +18,12 @@ import (
 	"gitea.perlbach24.de/scripte/nexarch/common/bearerauth"
 	"gitea.perlbach24.de/scripte/nexarch/common/moduleauth"
 	"gitea.perlbach24.de/scripte/nexarch/common/policyclient"
+	"gitea.perlbach24.de/scripte/nexarch/connect/internal/archiveclient"
 	"gitea.perlbach24.de/scripte/nexarch/connect/internal/connectorsetupapi"
 	"gitea.perlbach24.de/scripte/nexarch/connect/internal/credential"
 	"gitea.perlbach24.de/scripte/nexarch/connect/internal/crypto"
+	"gitea.perlbach24.de/scripte/nexarch/connect/internal/datevapi"
+	"gitea.perlbach24.de/scripte/nexarch/connect/internal/datevexport"
 	"gitea.perlbach24.de/scripte/nexarch/connect/internal/dms2import"
 	"gitea.perlbach24.de/scripte/nexarch/connect/internal/dms2importapi"
 	"gitea.perlbach24.de/scripte/nexarch/connect/internal/fieldmapping"
@@ -56,6 +59,25 @@ func buildPolicyClient() *policyclient.Client {
 	return policyclient.New(policyURL, token)
 }
 
+// buildDatevExports liefert nil ohne NEXARCH_CONNECT_APP_ARCHIVE_URL; dann ist
+// /datev/exports nicht registriert. Die DATEV-Uebertragung ist hier nicht verdrahtet.
+func buildDatevExports(pool *pgxpool.Pool) *datevexport.Store {
+	archiveURL := os.Getenv("NEXARCH_CONNECT_APP_ARCHIVE_URL")
+	if archiveURL == "" {
+		log.Print("connect-app: NEXARCH_CONNECT_APP_ARCHIVE_URL fehlt, /datev/exports nicht registriert")
+		return nil
+	}
+	token := os.Getenv("NEXARCH_CONNECT_APP_ARCHIVE_SERVICE_TOKEN")
+	if token == "" {
+		log.Fatal("connect-app: NEXARCH_CONNECT_APP_ARCHIVE_SERVICE_TOKEN ist erforderlich, wenn NEXARCH_CONNECT_APP_ARCHIVE_URL gesetzt ist")
+	}
+	return &datevexport.Store{
+		Pool:        pool,
+		Archive:     archiveclient.New(archiveURL, token),
+		CallbackURL: os.Getenv("NEXARCH_CONNECT_APP_DATEV_CALLBACK_URL"),
+	}
+}
+
 func main() {
 	if !killswitch.Enabled() {
 		log.Printf("connect-app: deaktiviert (%s=false), kein Start", killswitch.EnvEnabled)
@@ -98,6 +120,9 @@ func main() {
 	statusapi.NewServer(syncjob.NewStore(pool)).WithPolicy(policy).RegisterRoutes(mux)
 	dms2importapi.NewServer(mappings, dms2import.NewStore(pool)).RegisterRoutes(mux)
 	connectorsetupapi.NewServer(credentials, mappings).WithPolicy(policy).RegisterRoutes(mux)
+	if exports := buildDatevExports(pool); exports != nil {
+		datevapi.NewServer(exports, nil).WithPolicy(policy).RegisterRoutes(mux)
+	}
 
 	srv := &http.Server{
 		Addr:              addr,

connect/internal/datevapi/datevapi.go

@@ -9,6 +9,7 @@ import (
 	"net/http"
 	"time"
 
+	"gitea.perlbach24.de/scripte/nexarch/common/policyclient"
 	"gitea.perlbach24.de/scripte/nexarch/connect/internal/datevexport"
 	"gitea.perlbach24.de/scripte/nexarch/connect/internal/datevtransfer"
 )
@@ -16,6 +17,12 @@ import (
 type Server struct {
 	Exports   *datevexport.Store
 	Transfers *datevtransfer.Store
+	policy    *policyclient.Client
+}
+
+func (s *Server) WithPolicy(client *policyclient.Client) *Server {
+	s.policy = client
+	return s
 }
 
 func NewServer(exports *datevexport.Store, transfers *datevtransfer.Store) *Server {
@@ -23,8 +30,32 @@ func NewServer(exports *datevexport.Store, transfers *datevtransfer.Store) *Serv
 }
 
 func (s *Server) RegisterRoutes(mux *http.ServeMux) {
-	mux.HandleFunc("POST /datev/exports", s.requireIdentity(s.handleTrigger))
-	mux.HandleFunc("GET /datev/exports", s.requireIdentity(s.handleList))
+	mux.HandleFunc("POST /datev/exports", s.requireIdentity(s.requirePermission("connect.datev.write", s.handleTrigger)))
+	mux.HandleFunc("GET /datev/exports", s.requireIdentity(s.requirePermission("connect.datev.read", s.handleList)))
+}
+
+func (s *Server) requirePermission(permission string, next func(w http.ResponseWriter, r *http.Request, tenantSlug string)) func(w http.ResponseWriter, r *http.Request, tenantSlug string) {
+	return func(w http.ResponseWriter, r *http.Request, tenantSlug string) {
+		if s.policy != nil {
+			allowed, err := s.policy.Allowed(r.Context(), tenantSlug, roleOf(r), permission)
+			if err != nil {
+				writeError(w, http.StatusServiceUnavailable, "berechtigungsdienst nicht erreichbar")
+				return
+			}
+			if !allowed {
+				writeError(w, http.StatusForbidden, "keine berechtigung")
+				return
+			}
+		}
+		next(w, r, tenantSlug)
+	}
+}
+
+func roleOf(r *http.Request) string {
+	if role := r.Header.Get("X-User-Role"); role != "" {
+		return role
+	}
+	return "user"
 }
 
 func (s *Server) requireIdentity(next func(w http.ResponseWriter, r *http.Request, tenantSlug string)) http.HandlerFunc {

connect/internal/datevapi/datevapi_policy_test.go

@@ -0,0 +1,46 @@
+package datevapi
+
+import (
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+
+	"gitea.perlbach24.de/scripte/nexarch/common/policyclient"
+)
+
+func TestPolicy_PermissionAndRoleReachCore(t *testing.T) {
+	type ask struct{ role, permission string }
+	var got []ask
+	policyAPI := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		var body struct {
+			Role       string `json:"role"`
+			Permission string `json:"permission"`
+		}
+		_ = json.NewDecoder(r.Body).Decode(&body)
+		got = append(got, ask{body.Role, body.Permission})
+		_ = json.NewEncoder(w).Encode(map[string]bool{"allowed": false})
+	}))
+	defer policyAPI.Close()
+	mux := http.NewServeMux()
+	(&Server{}).WithPolicy(policyclient.New(policyAPI.URL, "tok")).RegisterRoutes(mux)
+
+	for _, c := range []struct{ method, role string }{{"GET", "tenant_admin"}, {"POST", ""}} {
+		req := httptest.NewRequest(c.method, "/datev/exports", strings.NewReader("{}"))
+		req.Header.Set("X-User-Id", "u1")
+		req.Header.Set("X-Tenant-Slug", "acme")
+		if c.role != "" {
+			req.Header.Set("X-User-Role", c.role)
+		}
+		rec := httptest.NewRecorder()
+		mux.ServeHTTP(rec, req)
+		if rec.Code != http.StatusForbidden {
+			t.Fatalf("%s: erwartet 403, bekam %d", c.method, rec.Code)
+		}
+	}
+	want := []ask{{"tenant_admin", "connect.datev.read"}, {"user", "connect.datev.write"}}
+	if len(got) != 2 || got[0] != want[0] || got[1] != want[1] {
+		t.Fatalf("erwartet %v, bekam %v", want, got)
+	}
+}

connect/internal/statusapi/statusapi.go

@@ -22,8 +22,7 @@ type Server struct {
 
 // WithPolicy aktiviert RBAC-12: echte Rechtepruefung gegen Core (POST
 // /authorize), zusaetzlich zur bestehenden X-User-Id/X-Tenant-Slug-Pruefung.
-// Rolle ist mangels eigenem Rollenkonzept in Connect fest "user" (Uebergang,
-// wie beim restlichen Header-basierten Identitaetsmodell dieses Pakets).
+// Die Rolle kommt aus X-User-Role (setzt bearerauth aus dem Token), sonst "user".
 func (s *Server) WithPolicy(client *policyclient.Client) *Server {
 	s.policy = client
 	return s
@@ -40,7 +39,7 @@ func (s *Server) RegisterRoutes(mux *http.ServeMux) {
 func (s *Server) requirePermission(permission string, next func(w http.ResponseWriter, r *http.Request, tenantSlug string)) func(w http.ResponseWriter, r *http.Request, tenantSlug string) {
 	return func(w http.ResponseWriter, r *http.Request, tenantSlug string) {
 		if s.policy != nil {
-			allowed, err := s.policy.Allowed(r.Context(), tenantSlug, "user", permission)
+			allowed, err := s.policy.Allowed(r.Context(), tenantSlug, roleOf(r), permission)
 			if err != nil {
 				writeError(w, http.StatusServiceUnavailable, "berechtigungsdienst nicht erreichbar")
 				return
@@ -54,6 +53,13 @@ func (s *Server) requirePermission(permission string, next func(w http.ResponseW
 	}
 }
 
+func roleOf(r *http.Request) string {
+	if role := r.Header.Get("X-User-Role"); role != "" {
+		return role
+	}
+	return "user"
+}
+
 func (s *Server) requireIdentity(next func(w http.ResponseWriter, r *http.Request, tenantSlug string)) http.HandlerFunc {
 	return func(w http.ResponseWriter, r *http.Request) {
 		if r.Header.Get("X-User-Id") == "" {

connect/internal/statusapi/statusapi_policy_test.go

@@ -60,3 +60,32 @@ func TestHandleList_PolicyAllows_Passes(t *testing.T) {
 		t.Fatalf("erwartet 200 mit berechtigung, bekam %d: %s", rec.Code, rec.Body.String())
 	}
 }
+
+func TestHandleList_RoleFromHeaderReachesPolicy(t *testing.T) {
+	var got []string
+	policyAPI := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		var body struct {
+			Role string `json:"role"`
+		}
+		_ = json.NewDecoder(r.Body).Decode(&body)
+		got = append(got, body.Role)
+		_ = json.NewEncoder(w).Encode(map[string]bool{"allowed": true})
+	}))
+	defer policyAPI.Close()
+	s := newTestServer(t).WithPolicy(policyclient.New(policyAPI.URL, "test-token"))
+	mux := http.NewServeMux()
+	s.RegisterRoutes(mux)
+
+	for _, role := range []string{"tenant_admin", ""} {
+		req := httptest.NewRequest("GET", "/connectors/status", nil)
+		req.Header.Set("X-User-Id", "u1")
+		req.Header.Set("X-Tenant-Slug", "acme")
+		if role != "" {
+			req.Header.Set("X-User-Role", role)
+		}
+		mux.ServeHTTP(httptest.NewRecorder(), req)
+	}
+	if len(got) != 2 || got[0] != "tenant_admin" || got[1] != "user" {
+		t.Fatalf("erwartet [tenant_admin user], bekam %v", got)
+	}
+}

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. Das `backup-tool` liest dieselben Variablen mit den Präfixen `NEXARCH_DMS_BACKUP_` (Quelle) und `NEXARCH_DMS_RESTORE_` (Ziel), auch mit `…_STORAGE_DRIVER=s3`.
+
 ## Struktur
 
 - `cmd/app` — Anfrage-Dienst (HTTP), blockiert nie durch lange Aufgaben

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"),

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
-}

dms/cmd/backup-tool/backup.go

@@ -14,6 +14,7 @@ import (
 type backupConfig struct {
 	DSN        string
 	StorageDir string
+	Driver     storage.Driver
 	TenantSlug string
 	OutDir     string
 	PgDumpPath string
@@ -45,7 +46,10 @@ func runBackup(ctx context.Context, cfg backupConfig) error {
 	}
 	defer pool.Close()
 
-	driver := storage.NewLocalDriver(cfg.StorageDir, []byte("backup-tool-unused"), "", 0)
+	driver := cfg.Driver
+	if driver == nil {
+		driver = storage.NewLocalDriver(cfg.StorageDir, []byte("backup-tool-unused"), "", 0)
+	}
 
 	rows, err := pool.Query(ctx, `
 		SELECT r.id, r.storage_key, r.size_bytes

dms/cmd/backup-tool/drill_test.go

@@ -32,6 +32,27 @@ const drillTenantSlug = "fdn27drill"
 // des Tickets): ein echter, unabhaengiger Quell- und Zielbestand auf dem
 // Entwicklungshost, kein simulierter Lauf. Er belegt AK1 bis AK3 in einem
 // zusammenhaengenden Durchlauf und misst RTO/RPO real.
+// drillDriver liefert den lokalen Treiber, oder einen S3-Treiber, wenn die
+// Umgebungsvariable bucketEnv einen Bucket nennt (TEST_S3_ENDPOINT usw. wie
+// in dms/internal/storage).
+func drillDriver(t *testing.T, bucketEnv, dir string) storage.Driver {
+	t.Helper()
+	bucket := os.Getenv(bucketEnv)
+	if bucket == "" {
+		return storage.NewLocalDriver(dir, []byte("secret"), "https://files.example.test", 0)
+	}
+	region := os.Getenv("TEST_S3_REGION")
+	if region == "" {
+		region = "us-east-1"
+	}
+	d, err := storage.NewS3Driver(context.Background(), bucket, region, os.Getenv("TEST_S3_ENDPOINT"),
+		os.Getenv("TEST_S3_ACCESS_KEY_ID"), os.Getenv("TEST_S3_SECRET_ACCESS_KEY"), true, 0)
+	if err != nil {
+		t.Fatalf("s3-treiber: %v", err)
+	}
+	return d
+}
+
 func TestBackupRestoreDrill_RealRoundTrip(t *testing.T) {
 	dsn := os.Getenv("TEST_TENANT_DSN")
 	if dsn == "" {
@@ -84,7 +105,8 @@ func TestBackupRestoreDrill_RealRoundTrip(t *testing.T) {
 	}
 
 	srcStorageDir := t.TempDir()
-	srcStorageSvc := storage.NewService(storage.NewLocalDriver(srcStorageDir, []byte("secret"), "https://files.example.test", 0), drillNoopUsage{}, drillTenantSlug)
+	srcDriver := drillDriver(t, "TEST_S3_SRC_BUCKET", srcStorageDir)
+	srcStorageSvc := storage.NewService(srcDriver, drillNoopUsage{}, drillTenantSlug)
 	uploadSvc := upload.NewService(srcPool, upload.NewSessionStore(srcPool), upload.NewStaging(t.TempDir()), srcStorageSvc, crypt, upload.NewValidator(10*1024*1024, nil), drillTenantSlug)
 
 	const wantDocs = 3
@@ -108,6 +130,7 @@ func TestBackupRestoreDrill_RealRoundTrip(t *testing.T) {
 	if err := runBackup(ctx, backupConfig{
 		DSN:        dsn,
 		StorageDir: srcStorageDir,
+		Driver:     srcDriver,
 		TenantSlug: drillTenantSlug,
 		OutDir:     backupDir,
 		PgSchema:   srcSchema,
@@ -128,11 +151,13 @@ func TestBackupRestoreDrill_RealRoundTrip(t *testing.T) {
 	dstSchema := fmt.Sprintf("fdn27dst_%d", time.Now().UnixNano())
 	t.Cleanup(func() { _, _ = admin.Exec(context.Background(), "DROP SCHEMA IF EXISTS "+dstSchema+" CASCADE") })
 	dstStorageDir := t.TempDir()
+	dstDriver := drillDriver(t, "TEST_S3_BUCKET", dstStorageDir)
 
 	restoreStart := time.Now()
 	if err := runRestore(ctx, restoreConfig{
 		DSN:          dsn,
 		StorageDir:   dstStorageDir,
+		Driver:       dstDriver,
 		InDir:        backupDir,
 		TargetSchema: dstSchema,
 		Crypto:       crypt,
@@ -162,10 +187,9 @@ func TestBackupRestoreDrill_RealRoundTrip(t *testing.T) {
 
 	// -- FDN-23-Integritaetsjob auf dem wiederhergestellten Bestand: muss
 	// "ok" liefern, sonst waere der Restore nur scheinbar vollstaendig --
-	dstStorageSvc := storage.NewLocalDriver(dstStorageDir, []byte("secret"), "https://files.example.test", 0)
 	verifier := &integrity.Verifier{
 		Pool:       dstPool,
-		Storage:    storage.NewService(dstStorageSvc, drillNoopUsage{}, drillTenantSlug),
+		Storage:    storage.NewService(dstDriver, drillNoopUsage{}, drillTenantSlug),
 		Crypto:     crypt,
 		TenantSlug: drillTenantSlug,
 		Config:     integrity.Config{Mode: integrity.ModeFull, Batch: 100},

dms/cmd/backup-tool/main.go

@@ -7,6 +7,7 @@ import (
 	"os"
 
 	dmscrypto "gitea.perlbach24.de/scripte/nexarch/dms/internal/crypto"
+	"gitea.perlbach24.de/scripte/nexarch/dms/internal/storage"
 )
 
 func main() {
@@ -18,9 +19,15 @@ func main() {
 	var err error
 	switch os.Args[1] {
 	case "backup":
-		err = runBackup(ctx, backupConfigFromEnv(os.Args[2]))
+		cfg := backupConfigFromEnv(os.Args[2])
+		if cfg.Driver, err = storage.NewDriverFromEnv(ctx, "NEXARCH_DMS_BACKUP"); err == nil {
+			err = runBackup(ctx, cfg)
+		}
 	case "restore":
-		err = runRestore(ctx, restoreConfigFromEnv(os.Args[2]))
+		cfg := restoreConfigFromEnv(os.Args[2])
+		if cfg.Driver, err = storage.NewDriverFromEnv(ctx, "NEXARCH_DMS_RESTORE"); err == nil {
+			err = runRestore(ctx, cfg)
+		}
 	default:
 		usage()
 		os.Exit(2)

dms/cmd/backup-tool/restore.go

@@ -23,6 +23,7 @@ var errTargetNotEmpty = errors.New("restore: zieldatenbank ist nicht leer (--for
 type restoreConfig struct {
 	DSN           string
 	StorageDir    string
+	Driver        storage.Driver
 	InDir         string
 	PgRestorePath string
 	PsqlPath      string
@@ -69,7 +70,10 @@ func runRestore(ctx context.Context, cfg restoreConfig) error {
 		}
 	}
 
-	driver := storage.NewLocalDriver(cfg.StorageDir, []byte("backup-tool-unused"), "", 0)
+	driver := cfg.Driver
+	if driver == nil {
+		driver = storage.NewLocalDriver(cfg.StorageDir, []byte("backup-tool-unused"), "", 0)
+	}
 	for _, obj := range m.Objects {
 		if err := restoreObject(ctx, driver, cfg.InDir, obj); err != nil {
 			return fmt.Errorf("restore: objekt %q (revision %s): %w", obj.StorageKey, obj.RevisionID, err)

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
-}

dms/internal/storage/fromenv.go

@@ -0,0 +1,42 @@
+package storage
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"strings"
+)
+
+// NewDriverFromEnv wählt den Treiber über <prefix>_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") != ""
+}

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")
+	}
+}