package main import ( "bytes" "context" "encoding/json" "errors" "fmt" "io" "maps" "net/http" "net/http/httptest" "os" "path/filepath" "slices" "strings" "sync" "testing" "time" "gopkg.in/yaml.v3" ) func servicePath(parts ...string) string { return filepath.Join(append([]string{"..", ".."}, parts...)...) } func serviceFiles(t *testing.T, paths ...string) []byte { t.Helper() var content []byte for _, path := range paths { value, err := os.ReadFile(servicePath(strings.Split(path, "/")...)) if err != nil { t.Fatal(err) } content = append(content, value...) } return content } type probeHTTP struct { Path string `yaml:"path"` Port any `yaml:"port"` Scheme string `yaml:"scheme"` HTTPHeaders []struct { Name string `yaml:"name"` Value string `yaml:"value"` } `yaml:"httpHeaders"` } type deploymentManifest struct { Spec struct { Replicas int `yaml:"replicas"` Strategy struct { Type string `yaml:"type"` RollingUpdate struct { MaxUnavailable int `yaml:"maxUnavailable"` MaxSurge int `yaml:"maxSurge"` } `yaml:"rollingUpdate"` } `yaml:"strategy"` Template struct { Metadata struct { Annotations map[string]string `yaml:"annotations"` } `yaml:"metadata"` Spec struct { Volumes []struct { Name string `yaml:"name"` Projected *struct { Sources []struct { ServiceAccountToken *struct { Audience string `yaml:"audience"` ExpirationSeconds *int64 `yaml:"expirationSeconds"` } `yaml:"serviceAccountToken"` } `yaml:"sources"` } `yaml:"projected"` } `yaml:"volumes"` TopologySpreadConstraints []struct { MaxSkew int `yaml:"maxSkew"` MinDomains int `yaml:"minDomains"` TopologyKey string `yaml:"topologyKey"` WhenUnsatisfiable string `yaml:"whenUnsatisfiable"` } `yaml:"topologySpreadConstraints"` Containers []struct { Name string `yaml:"name"` Image string `yaml:"image"` Command []string `yaml:"command"` Args []string `yaml:"args"` VolumeMounts []struct { MountPath string `yaml:"mountPath"` } `yaml:"volumeMounts"` ReadinessProbe struct { HTTPGet *probeHTTP `yaml:"httpGet"` TCPSocket *struct { Port any `yaml:"port"` } `yaml:"tcpSocket"` } `yaml:"readinessProbe"` } `yaml:"containers"` } `yaml:"spec"` } `yaml:"template"` } `yaml:"spec"` } func serviceDeployment(t *testing.T, path string) deploymentManifest { t.Helper() var deployment deploymentManifest if err := yaml.Unmarshal(serviceFiles(t, path), &deployment); err != nil { t.Fatal(err) } return deployment } func hasArgument(arguments []string, value string) bool { return slices.Contains(arguments, value) } func argumentValue(arguments []string, name string) string { for index := 0; index+1 < len(arguments); index++ { if arguments[index] == name { return arguments[index+1] } } return "" } func hexValue(value string, length int) bool { return len(value) == length && strings.Trim(value, "0123456789abcdef") == "" } func qualifiedImage(image, repository string) bool { prefix, digest, found := strings.Cut(image, "@sha256:") return found && strings.HasPrefix(prefix, repository+":") && hexValue(digest, 64) } func TestParseOptionsUsesProvidedArguments(t *testing.T) { value, err := parseOptions([]string{ "--plan", "desired.json", "--stalwart-url", "https://mail.example.invalid", "--once", }) if err != nil { t.Fatal(err) } if value.plan != "desired.json" || value.stalwartURL != "https://mail.example.invalid" || !value.once { t.Fatalf("options = %#v", value) } if err := validateOptions(value); err != nil { t.Fatal(err) } } func TestRunCLIReturnsRedactedDiagnostic(t *testing.T) { path := filepath.Join(t.TempDir(), "stalwart-cli") if err := os.WriteFile(path, []byte("#!/bin/sh\nprintf 'rejected token %s\\n' \"$STALWART_TOKEN\" >&2\nexit 1\n"), 0o700); err != nil { t.Fatal(err) } _, err := runCLI(t.Context(), path, []string{"STALWART_TOKEN=synthetic-sensitive-value"}, false, "apply", "--file") if err == nil { t.Fatal("failed command returned no error") } value := err.Error() if strings.Contains(value, "synthetic-sensitive-value") || !strings.Contains(value, "rejected token [redacted]") { t.Fatalf("diagnostic = %q", value) } } func TestNextBackoffIsBounded(t *testing.T) { maximum := 5 * time.Minute value := 5 * time.Second for range 20 { value = nextBackoff(value, maximum) } if value != maximum { t.Fatalf("backoff = %s, want %s", value, maximum) } } func TestPersistentLeaderFailureRelinquishesOwnership(t *testing.T) { store := &statusStore{status: controllerStatus{LastSuccess: "2026-01-01T00:00:00Z", Ready: true}} for range 2 { if recordLeaderFailure(store, 3) { t.Fatal("leader relinquished ownership before the failure threshold") } } if !recordLeaderFailure(store, 3) { t.Fatal("leader did not relinquish ownership at the failure threshold") } status := store.snapshot() if status.Ready || !status.PersistentFailure || status.ConsecutiveFailures != 3 || status.LastResult != "failure" { t.Fatalf("status = %#v", status) } } func TestOwnershipSessionUsesBoundedKeepalives(t *testing.T) { var statements []string if err := configureOwnershipSession(context.Background(), func(_ context.Context, statement string) error { statements = append(statements, statement) return nil }); err != nil { t.Fatal(err) } if !slices.Equal(statements, ownershipSessionSettings) { t.Fatalf("statements = %q, want %q", statements, ownershipSessionSettings) } } func TestOwnershipSessionStopsAtSettingFailure(t *testing.T) { failed := errors.New("setting failed") var statements []string err := configureOwnershipSession(context.Background(), func(_ context.Context, statement string) error { statements = append(statements, statement) if len(statements) == 2 { return failed } return nil }) if !errors.Is(err, failed) { t.Fatalf("error = %v, want %v", err, failed) } if !slices.Equal(statements, ownershipSessionSettings[:2]) { t.Fatalf("statements = %q, want %q", statements, ownershipSessionSettings[:2]) } } func TestRunCLIErrorNamesAction(t *testing.T) { path := filepath.Join(t.TempDir(), "stalwart-cli") if err := os.WriteFile(path, []byte("#!/bin/sh\nexit 17\n"), 0o700); err != nil { t.Fatal(err) } _, err := runCLI(t.Context(), path, nil, false, "apply", "--file", "synthetic-plan") if err == nil || !strings.Contains(err.Error(), "stalwart CLI apply --file") { t.Fatalf("error = %v", err) } } type recordedOwnership struct { steps *[]string } func (value recordedOwnership) Close(context.Context) error { *value.steps = append(*value.steps, "ownership") return nil } func (recordedOwnership) Ping(context.Context) error { return nil } type recordedServer struct { steps *[]string } type testOwnershipManager struct { mu sync.Mutex held bool } type testOwnership struct { manager *testOwnershipManager } func (value testOwnership) Close(context.Context) error { value.manager.mu.Lock() defer value.manager.mu.Unlock() value.manager.held = false return nil } func (testOwnership) Ping(context.Context) error { return nil } func (manager *testOwnershipManager) acquire(context.Context, string) (ownershipConnection, error) { manager.mu.Lock() defer manager.mu.Unlock() if manager.held { return nil, errOwnerExists } manager.held = true return testOwnership{manager: manager}, nil } func TestControllersHandOffOwnership(t *testing.T) { manager := &testOwnershipManager{} type runningController struct { cancel context.CancelFunc done chan error reconciled chan struct{} } controllers := make([]runningController, 2) for index := range controllers { ctx, cancel := context.WithCancel(t.Context()) done := make(chan error, 1) reconciled := make(chan struct{}, 1) runtime := controllerRuntime{ acquireOwnership: manager.acquire, planRevision: func(string) (string, error) { return "synthetic-revision", nil }, reconcile: func(context.Context, options) error { reconciled <- struct{}{} return nil }, } value := options{ dsnFile: "synthetic-dsn", plan: "synthetic-plan", stalwartURL: "http://stalwart.invalid", bootstrapUser: "recovery", bootstrapPasswordFile: "synthetic-password", statusAddress: "127.0.0.1:0", interval: time.Hour, initialBackoff: 10 * time.Millisecond, maxBackoff: 20 * time.Millisecond, reconcileTimeout: time.Second, persistentFailureAfter: 3, } go func() { done <- runControllerWithRuntime(ctx, value, runtime) }() controllers[index] = runningController{cancel: cancel, done: done, reconciled: reconciled} } leader := -1 select { case <-controllers[0].reconciled: leader = 0 case <-controllers[1].reconciled: leader = 1 case <-time.After(2 * time.Second): t.Fatal("neither controller reconciled") } standby := 1 - leader select { case <-controllers[standby].reconciled: t.Fatal("standby reconciled while the leader held ownership") case <-time.After(50 * time.Millisecond): } controllers[leader].cancel() if err := <-controllers[leader].done; err != nil { t.Fatalf("leader shutdown: %v", err) } select { case <-controllers[standby].reconciled: case <-time.After(2 * time.Second): t.Fatal("standby did not reconcile after leader shutdown") } controllers[standby].cancel() if err := <-controllers[standby].done; err != nil { t.Fatalf("standby shutdown: %v", err) } } func TestUnhealthyLeaderCoolsDownWhileStandbyTakesOwnership(t *testing.T) { manager := &testOwnershipManager{} baseOptions := options{ dsnFile: "synthetic-dsn", plan: "synthetic-plan", stalwartURL: "http://stalwart.invalid", bootstrapUser: "recovery", bootstrapPasswordFile: "synthetic-password", statusAddress: "127.0.0.1:0", interval: time.Hour, initialBackoff: 10 * time.Millisecond, maxBackoff: 200 * time.Millisecond, reconcileTimeout: time.Second, persistentFailureAfter: 1, } unhealthyContext, cancelUnhealthy := context.WithCancel(t.Context()) unhealthyDone := make(chan error, 1) unhealthyReconciled := make(chan struct{}, 1) unhealthyRuntime := controllerRuntime{ acquireOwnership: manager.acquire, planRevision: func(string) (string, error) { return "synthetic-revision", nil }, reconcile: func(context.Context, options) error { unhealthyReconciled <- struct{}{} return errors.New("synthetic reconciliation failure") }, } go func() { unhealthyDone <- runControllerWithRuntime(unhealthyContext, baseOptions, unhealthyRuntime) }() select { case <-unhealthyReconciled: case <-time.After(2 * time.Second): t.Fatal("unhealthy controller did not reconcile") } healthyContext, cancelHealthy := context.WithCancel(t.Context()) healthyDone := make(chan error, 1) healthyReconciled := make(chan struct{}, 1) healthyRuntime := controllerRuntime{ acquireOwnership: manager.acquire, planRevision: func(string) (string, error) { return "synthetic-revision", nil }, reconcile: func(context.Context, options) error { healthyReconciled <- struct{}{} return nil }, } go func() { healthyDone <- runControllerWithRuntime(healthyContext, baseOptions, healthyRuntime) }() select { case <-healthyReconciled: case <-time.After(2 * time.Second): t.Fatal("healthy standby did not take ownership") } select { case err := <-unhealthyDone: t.Fatalf("unhealthy controller exited during cooldown: %v", err) default: } cancelHealthy() if err := <-healthyDone; err != nil { t.Fatalf("healthy controller shutdown: %v", err) } cancelUnhealthy() if err := <-unhealthyDone; err != nil { t.Fatalf("unhealthy controller shutdown: %v", err) } } func (value recordedServer) Shutdown(context.Context) error { *value.steps = append(*value.steps, "server") return nil } func TestControllerShutdownReleasesOwnershipBeforeStatusServer(t *testing.T) { var steps []string closeOwnership(recordedOwnership{steps: &steps}) shutdownStatusServer(recordedServer{steps: &steps}) if !slices.Equal(steps, []string{"ownership", "server"}) { t.Fatalf("shutdown order = %v", steps) } } func TestStandbyStatusIsLiveButNotReady(t *testing.T) { store := &statusStore{status: controllerStatus{LastResult: "standby"}} handler := statusHandler(store) for _, check := range []struct { path string want int }{ {path: "/healthz", want: http.StatusOK}, {path: "/readyz", want: http.StatusServiceUnavailable}, } { response := httptest.NewRecorder() handler.ServeHTTP(response, httptest.NewRequest(http.MethodGet, check.path, nil)) if response.Code != check.want { t.Fatalf("%s status = %d, want %d", check.path, response.Code, check.want) } } } func TestPlanRevisionChangesWithReadableSource(t *testing.T) { path := filepath.Join(t.TempDir(), "plan.json") if err := os.WriteFile(path, []byte(`[{"revision":1}]`), 0o600); err != nil { t.Fatal(err) } first, err := planRevision(path) if err != nil { t.Fatal(err) } if err := os.WriteFile(path, []byte(`[{"revision":2}]`), 0o600); err != nil { t.Fatal(err) } second, err := planRevision(path) if err != nil { t.Fatal(err) } if first == second { t.Fatal("plan revision did not change") } } func TestRenderPlanInjectsSecretWithoutChangingSource(t *testing.T) { root := t.TempDir() source := filepath.Join(root, "plan.json") target := filepath.Join(root, "plan.ndjson") secretRoot := filepath.Join(root, "secrets") if err := os.Mkdir(secretRoot, 0o700); err != nil { t.Fatal(err) } if err := os.WriteFile(filepath.Join(secretRoot, "password"), []byte("projected-value"), 0o400); err != nil { t.Fatal(err) } original := `[{"credential":{"$secretFile":"password"},"name":"account"}]` if err := os.WriteFile(source, []byte(original), 0o600); err != nil { t.Fatal(err) } if err := renderPlan(source, target, secretRoot); err != nil { t.Fatal(err) } content, err := os.ReadFile(target) if err != nil { t.Fatal(err) } var operation map[string]any if err := json.Unmarshal(content, &operation); err != nil { t.Fatal(err) } if operation["credential"] != "projected-value" { t.Fatalf("credential = %v", operation["credential"]) } unchanged, err := os.ReadFile(source) if err != nil { t.Fatal(err) } if string(unchanged) != original { t.Fatal("readable source changed") } info, err := os.Stat(target) if err != nil { t.Fatal(err) } if info.Mode().Perm() != 0o600 { t.Fatalf("rendered mode = %o", info.Mode().Perm()) } } func TestRenderSecretSet(t *testing.T) { root := t.TempDir() source := filepath.Join(root, "plan.json") target := filepath.Join(root, "plan.ndjson") const audience = `synthetic/oidc"audience\client` if err := os.WriteFile(filepath.Join(root, "audience"), []byte(audience+"\n"), 0o600); err != nil { t.Fatal(err) } original := `[{"@type":"reconcile","object":"Directory","value":{"zitadel":{"@type":"Oidc","acceptedAudiences":{"$secretSet":"audience"}}}}]` if err := os.WriteFile(source, []byte(original), 0o600); err != nil { t.Fatal(err) } if err := renderPlan(source, target, root); err != nil { t.Fatal(err) } content, err := os.ReadFile(target) if err != nil { t.Fatal(err) } var operation struct { Value map[string]struct { Audiences map[string]bool `json:"acceptedAudiences"` } `json:"value"` } if err := json.Unmarshal(content, &operation); err != nil { t.Fatal(err) } audiences := operation.Value["zitadel"].Audiences if len(audiences) != 1 || !audiences[audience] { t.Fatalf("accepted audiences = %v", audiences) } unchanged, err := os.ReadFile(source) if err != nil { t.Fatal(err) } if string(unchanged) != original { t.Fatal("readable source changed") } } func TestSecretSetMissing(t *testing.T) { _, err := injectSecrets(map[string]any{"$secretSet": "missing"}, t.TempDir()) if !errors.Is(err, os.ErrNotExist) { t.Fatalf("missing secret error = %v", err) } } func TestProductionPlanRendersFromProjectedSecrets(t *testing.T) { secretRoot := t.TempDir() secrets := map[string]string{ "dragonfly_url": "redis://default:synthetic-cache-password@stalwart.dragonfly.svc.cluster.local:6379", "oidc_audience": "synthetic-oidc-audience", "oidc_controller_subject": "synthetic-controller-subject", "oidc_human_admin_subject": "synthetic-human-subject", "oidc_issuer_url": "https://auth.example.invalid", "oidc_notifier_subject": "synthetic-notifier-subject", "oidc_sender_subject": "synthetic-sender-subject", "oidc_test_subject": "synthetic-test-subject", "relay_endpoint": "smtp.email.ap-osaka-1.oci.oraclecloud.com", "relay_username": "synthetic-relay-user", } for name, value := range secrets { if err := os.WriteFile(filepath.Join(secretRoot, name), []byte(value+"\n"), 0o600); err != nil { t.Fatal(err) } } target := filepath.Join(t.TempDir(), "plan.ndjson") if err := renderPlan(servicePath("k8s", "base", "plan.json"), target, secretRoot); err != nil { t.Fatal(err) } content, err := os.ReadFile(target) if err != nil { t.Fatal(err) } if bytes.Contains(content, []byte(`"$secretFile"`)) { t.Fatal("rendered production plan contains an unresolved secret file") } for _, value := range secrets { if !bytes.Contains(content, []byte(value)) { t.Fatalf("rendered production plan does not contain projected value %q", value) } } decoder := json.NewDecoder(bytes.NewReader(content)) operations := 0 for { var operation map[string]any err := decoder.Decode(&operation) if errors.Is(err, io.EOF) { break } if err != nil { t.Fatal(err) } if operation["@type"] == nil || operation["object"] == nil { t.Fatalf("operation %d is incomplete", operations) } if operation["object"] == "Certificate" { t.Fatal("production plan owns certificate lifecycle outside the controller") } operations++ } if operations != 34 { t.Fatalf("rendered production plan has %d operations, want 34", operations) } } func TestProductionPlanOwnsOutboundSchedules(t *testing.T) { content, err := os.ReadFile(servicePath("k8s", "base", "plan.json")) if err != nil { t.Fatal(err) } var operations []struct { Type string `json:"@type"` Object string `json:"object"` Value json.RawMessage `json:"value"` } if err := json.Unmarshal(content, &operations); err != nil { t.Fatal(err) } queues := map[string]string{} schedules := map[string]string{} for _, operation := range operations { if operation.Object != "MtaVirtualQueue" && operation.Object != "MtaDeliverySchedule" { continue } if operation.Type != "reconcile" { t.Fatalf("%s operation is %q, want reconcile", operation.Object, operation.Type) } var values map[string]struct { Name string `json:"name"` QueueID string `json:"queueId"` } if err := json.Unmarshal(operation.Value, &values); err != nil { t.Fatal(err) } for reference, value := range values { if operation.Object == "MtaVirtualQueue" { queues[value.Name] = "#" + reference } else { schedules[value.Name] = value.QueueID } } } for _, name := range []string{"local", "remote", "dsn", "report"} { queueReference, queueExists := queues[name] scheduleReference, scheduleExists := schedules[name] if !queueExists || !scheduleExists { t.Fatalf("outbound inventory is missing %q", name) } if scheduleReference != queueReference { t.Fatalf("schedule %q references %q, want %q", name, scheduleReference, queueReference) } } if len(queues) != 4 || len(schedules) != 4 { t.Fatalf("outbound inventory has %d queues and %d schedules, want 4 each", len(queues), len(schedules)) } } func TestProductionPlanBindsZitadelSubjectsWithoutNativeCredentials(t *testing.T) { content, err := os.ReadFile(servicePath("k8s", "base", "plan.json")) if err != nil { t.Fatal(err) } value := string(content) for _, required := range []string{ `"claimUsername": "sub"`, `"usernameDomain": "mail-auth.quorate.internal"`, `"$secretFile": "oidc_human_admin_subject"`, `"$secretFile": "oidc_controller_subject"`, `"$secretFile": "oidc_notifier_subject"`, `"$secretFile": "oidc_sender_subject"`, `"$secretFile": "oidc_test_subject"`, } { if !strings.Contains(value, required) { t.Fatalf("production plan lacks %s", required) } } for _, forbidden := range []string{ `"$secretFile": "operator_password"`, `"$secretFile": "controller_password"`, `"$secretFile": "alerts_password"`, `"$secretFile": "test_password"`, `"claimUsername": "preferred_username"`, } { if strings.Contains(value, forbidden) { t.Fatalf("production plan contains %s", forbidden) } } } func TestProductionPlanExposesOnlyAccountAPIEndpointsOnUserHTTP(t *testing.T) { content, err := os.ReadFile(servicePath("k8s", "base", "plan.json")) if err != nil { t.Fatal(err) } value := string(content) required := `listener == \"user-http\" && starts_with(path, \"/api\") && path != \"/api/schema\" && !starts_with(path, \"/api/schema/\") && path != \"/api/account\" && !starts_with(path, \"/api/discover/\")` if !strings.Contains(value, required) { t.Fatal("production plan lacks the account API boundary") } } func TestProductionPlanReconcilesZitadelOwnedAccounts(t *testing.T) { content, err := os.ReadFile(servicePath("k8s", "base", "plan.json")) if err != nil { t.Fatal(err) } var operations []struct { Type string `json:"@type"` Object string `json:"object"` Value json.RawMessage `json:"value"` } if err := json.Unmarshal(content, &operations); err != nil { t.Fatal(err) } for _, operation := range operations { if operation.Object != "Account" { continue } if operation.Type != "upsert" { t.Fatalf("Account operation is %q", operation.Type) } var accounts map[string]struct { Credentials map[string]any `json:"credentials"` Name string `json:"name"` DomainID string `json:"domainId"` Aliases map[string]struct { Name any `json:"name"` DomainID string `json:"domainId"` } `json:"aliases"` Permissions struct { Type string `json:"@type"` Enabled map[string]bool `json:"enabledPermissions"` } `json:"permissions"` } if err := json.Unmarshal(operation.Value, &accounts); err != nil { t.Fatal(err) } for _, name := range []string{"human-admin", "test-client", "controller", "alert-sender", "notification-sender"} { if len(accounts[name].Credentials) != 0 { t.Fatalf("%s has native credentials", name) } } operatorAccount := accounts["operator"] if len(operatorAccount.Aliases) != 0 { t.Fatal("operator mailbox retains the test address alias") } operator := operatorAccount.Permissions if operator.Type != "Replace" || len(operator.Enabled) != 1 || !operator.Enabled["emailReceive"] { t.Fatal("operator mailbox cannot explicitly receive mail") } human := accounts["human-admin"].Permissions if human.Type != "Replace" || len(human.Enabled) != 625 || !human.Enabled["authenticate"] || !human.Enabled["emailReceive"] || !human.Enabled["emailSend"] || !human.Enabled["sysDomainQuery"] { t.Fatal("human administrator permissions are incomplete") } controller := accounts["controller"].Permissions if controller.Type != "Replace" || len(controller.Enabled) != len(human.Enabled) || !maps.Equal(controller.Enabled, human.Enabled) { t.Fatal("controller permissions are not explicit for full reconciliation") } sender := accounts["alert-sender"].Permissions if sender.Type != "Replace" || len(sender.Enabled) != 7 || !sender.Enabled["authenticate"] || !sender.Enabled["emailSend"] { t.Fatal("sender permissions are not restricted") } notifier := accounts["notification-sender"] notifierSubject, ok := notifier.Aliases["0"].Name.(map[string]any) if notifier.Name != "noreply" || notifier.DomainID != "#domain" || len(notifier.Aliases) != 1 || !ok || notifierSubject["$secretFile"] != "oidc_notifier_subject" || notifier.Aliases["0"].DomainID != "#authentication-domain" { t.Fatal("notification sender lacks its Zitadel subject alias") } notifierPermissions := notifier.Permissions if notifierPermissions.Type != "Replace" || len(notifierPermissions.Enabled) != 7 || !notifierPermissions.Enabled["authenticate"] || !notifierPermissions.Enabled["emailSend"] { t.Fatal("notification sender permissions are not restricted") } testClient := accounts["test-client"] testSubject, ok := testClient.Aliases["0"].Name.(map[string]any) if len(testClient.Aliases) != 1 || !ok || testSubject["$secretFile"] != "oidc_test_subject" || testClient.Aliases["0"].DomainID != "#authentication-domain" { t.Fatal("test client lacks its Zitadel subject alias") } testPermissions := testClient.Permissions if testPermissions.Type != "Replace" || !testPermissions.Enabled["authenticate"] || !testPermissions.Enabled["emailReceive"] || !testPermissions.Enabled["emailSend"] || !testPermissions.Enabled["imapIdle"] || !testPermissions.Enabled["imapSearch"] { t.Fatal("test client permissions are incomplete") } return } t.Fatal("Account operation is absent") } func TestBootstrapJobOwnsNoDesiredState(t *testing.T) { content, err := os.ReadFile(servicePath("k8s", "bootstrap", "server", "job.yaml")) if err != nil { t.Fatal(err) } value := string(content) for _, forbidden := range []string{"--plan", "/desired/", "$secretFile"} { if strings.Contains(value, forbidden) { t.Fatalf("bootstrap Job contains %q", forbidden) } } if !strings.Contains(value, "http://stalwart-controller:9090/readyz") { t.Fatal("bootstrap Job does not wait for controller readiness") } } func TestNormalControllerPrefersAStandbyOnAnotherNode(t *testing.T) { deployment := serviceDeployment(t, "k8s/normal/controller/deploy.yaml") if deployment.Spec.Replicas != 2 || deployment.Spec.Strategy.Type != "RollingUpdate" || deployment.Spec.Strategy.RollingUpdate.MaxUnavailable != 1 || deployment.Spec.Strategy.RollingUpdate.MaxSurge != 0 { t.Fatalf("normal controller rollout = %#v", deployment.Spec) } if len(deployment.Spec.Template.Spec.TopologySpreadConstraints) != 1 { t.Fatalf("topology spread constraints = %#v", deployment.Spec.Template.Spec.TopologySpreadConstraints) } spread := deployment.Spec.Template.Spec.TopologySpreadConstraints[0] if spread.MaxSkew != 1 || spread.MinDomains != 2 || spread.TopologyKey != "kubernetes.io/hostname" || spread.WhenUnsatisfiable != "DoNotSchedule" { t.Fatalf("topology spread constraint = %#v", spread) } if len(deployment.Spec.Template.Spec.Containers) != 1 || !hasArgument(deployment.Spec.Template.Spec.Containers[0].Args, "--standby-ready") { t.Fatal("normal controller does not configure a ready standby") } content := string(serviceFiles(t, "k8s/normal/controller/pdb.yaml", "k8s/normal/kustomization.yaml")) for _, required := range []string{"minAvailable: 1", "- controller/pdb.yaml", "- ../controller"} { if !strings.Contains(content, required) { t.Fatalf("normal controller HA manifests lack %q", required) } } } func TestControllerModesPatchOneSharedWorkload(t *testing.T) { for _, path := range []string{"k8s/bootstrap/kustomization.yaml", "k8s/normal/kustomization.yaml"} { var kustomization struct { Resources []string `yaml:"resources"` Patches []struct { Path string `yaml:"path"` } `yaml:"patches"` } if err := yaml.Unmarshal(serviceFiles(t, path), &kustomization); err != nil { t.Fatal(err) } if !slices.Contains(kustomization.Resources, "../controller") { t.Fatalf("%s does not use the shared controller", path) } controllerPatches := 0 for _, patch := range kustomization.Patches { if patch.Path == "controller/deploy.yaml" { controllerPatches++ } } if controllerPatches != 1 { t.Fatalf("%s has %d controller deployment patches", path, controllerPatches) } } } func TestReplicatedStalwartWorkloadsUseHardTopologySpreading(t *testing.T) { for _, path := range []string{"k8s/normal/server/deploy.yaml"} { deployment := serviceDeployment(t, path) if len(deployment.Spec.Template.Spec.TopologySpreadConstraints) != 1 { t.Fatalf("%s does not have one topology spread constraint", path) } spread := deployment.Spec.Template.Spec.TopologySpreadConstraints[0] if spread.MaxSkew != 1 || spread.MinDomains != 2 || spread.TopologyKey != "kubernetes.io/hostname" || spread.WhenUnsatisfiable != "DoNotSchedule" { t.Fatalf("%s topology spread constraint = %#v", path, spread) } } } func TestControllerManifestsPinQualifiedImage(t *testing.T) { const repository = "registry.quorate.internal/quorate/stalwart-controller" controller := serviceDeployment(t, "k8s/controller/deploy.yaml") if len(controller.Spec.Template.Spec.Containers) != 1 { t.Fatal("shared controller deployment does not contain one controller") } image := controller.Spec.Template.Spec.Containers[0].Image if !qualifiedImage(image, repository) { t.Fatal("shared controller deployment does not pin the qualified controller image") } revision := controller.Spec.Template.Metadata.Annotations["quorate.internal/source-revision"] if !hexValue(revision, 40) { t.Fatal("shared controller deployment does not declare the embedded source revision") } if !strings.Contains(string(serviceFiles(t, "k8s/normal/server/deploy.yaml")), image) { t.Fatal("normal server deployment does not pin the qualified controller image") } } func TestPublicAccountLoginRoutes(t *testing.T) { route := string(serviceFiles(t, "k8s/relay/gateway/route.yaml")) frontend, err := os.ReadFile(servicePath("..", "mail-keys", "k8s", "delivery", "nginx.conf")) if err != nil { t.Fatal(err) } for _, content := range []string{route, string(frontend)} { if !strings.Contains(content, "/login") || !strings.Contains(content, "/account/login") { t.Fatalf("account login redirect is absent from %q", content) } } } func TestRecoveryCredentialIsLimitedToBootstrapController(t *testing.T) { bootstrap := serviceDeployment(t, "k8s/bootstrap/controller/deploy.yaml").Spec.Template.Spec.Containers[0].Args normal := serviceDeployment(t, "k8s/normal/controller/deploy.yaml").Spec.Template.Spec.Containers[0].Args for _, required := range []string{"--bootstrap-user", "--bootstrap-password-file"} { if !hasArgument(bootstrap, required) { t.Fatalf("bootstrap controller lacks %q", required) } if hasArgument(normal, required) { t.Fatalf("normal controller retains %q", required) } } } func TestNormalControllerUsesOIDCWithoutNativeSteadyCredentials(t *testing.T) { normal := serviceDeployment(t, "k8s/normal/controller/deploy.yaml").Spec.Template.Spec.Containers[0].Args for _, required := range []string{ "--oidc-issuer-file", "--oidc-audience-file", "--oidc-client-id-file", "--oidc-client-secret-file", } { if !hasArgument(normal, required) { t.Fatalf("normal controller lacks %s", required) } } for _, forbidden := range []string{"--bootstrap-user", "--bootstrap-password-file"} { if hasArgument(normal, forbidden) { t.Fatalf("normal controller contains %s", forbidden) } } server := serviceFiles(t, "k8s/normal/server/deploy.yaml") if !strings.Contains(string(server), "wait-oidc") { t.Fatal("normal Stalwart deployment lacks the OIDC startup gate") } bootstrap := serviceDeployment(t, "k8s/bootstrap/controller/deploy.yaml").Spec.Template.Spec.Containers[0].Args if argumentValue(bootstrap, "--stalwart-url") != "http://stalwart-bootstrap.stalwart.svc.cluster.local:8080" { t.Fatal("bootstrap controller does not target the bootstrap Stalwart service") } for _, forbidden := range []string{"--oidc-client-id-file", "--oidc-client-secret-file", "--app-passwords", "--vault-address"} { if hasArgument(bootstrap, forbidden) { t.Fatalf("bootstrap controller contains %s", forbidden) } } } func TestServerReadiness(t *testing.T) { deployment := serviceDeployment(t, "k8s/normal/server/deploy.yaml") probes := make(map[string]*probeHTTP) for _, container := range deployment.Spec.Template.Spec.Containers { probes[container.Name] = container.ReadinessProbe.HTTPGet if (container.Name == "stalwart" || container.Name == "account-web") && container.ReadinessProbe.TCPSocket != nil { t.Fatalf("%s readiness checks only a socket", container.Name) } } backend := probes["stalwart"] if backend == nil || backend.Path != "/healthz/live" || backend.Port != "management" || backend.Scheme != "" && backend.Scheme != "HTTP" { t.Fatalf("Stalwart readiness = %#v", backend) } frontend := probes["account-web"] if frontend == nil || frontend.Path != "/account/oauth/callback" || frontend.Port != 8443 || frontend.Scheme != "HTTPS" { t.Fatalf("account frontend readiness = %#v", frontend) } if len(frontend.HTTPHeaders) != 1 || frontend.HTTPHeaders[0].Name != "Host" || frontend.HTTPHeaders[0].Value != "mail.quorate.dev" { t.Fatalf("account frontend probe headers = %#v", frontend.HTTPHeaders) } } func TestNormalControllerUsesKubernetesDefaultVaultTokenAudience(t *testing.T) { deployment := serviceDeployment(t, "k8s/normal/controller/deploy.yaml") for _, volume := range deployment.Spec.Template.Spec.Volumes { if volume.Name != "vault-token" || volume.Projected == nil { continue } for _, source := range volume.Projected.Sources { if source.ServiceAccountToken == nil { continue } if source.ServiceAccountToken.Audience != "" { t.Fatalf("Vault token audience = %q", source.ServiceAccountToken.Audience) } if source.ServiceAccountToken.ExpirationSeconds == nil || *source.ServiceAccountToken.ExpirationSeconds != 600 { t.Fatal("Vault token does not use the short-lived projection") } return } } t.Fatal("Vault service-account token projection is absent") } func TestStalwartNetworkPoliciesAllowDependencyResolution(t *testing.T) { for _, paths := range [][]string{ {"k8s/bootstrap/server/policy.yaml", "k8s/controller/policy.yaml"}, {"k8s/normal/server/policy.yaml", "k8s/controller/policy.yaml"}, } { content := serviceFiles(t, paths...) if strings.Count(string(content), `- matchPattern: "*"`) != 2 { t.Fatalf("%v does not allow DNS queries for both workloads", paths) } } } func TestPrivateSubmissionListenerUsesCiliumTransportEncryption(t *testing.T) { content, err := os.ReadFile(servicePath("k8s", "base", "plan.json")) if err != nil { t.Fatal(err) } var operations []struct { Object string `json:"object"` Value json.RawMessage `json:"value"` } if err := json.Unmarshal(content, &operations); err != nil { t.Fatal(err) } listenersFound := false authFound := false for _, operation := range operations { switch operation.Object { case "NetworkListener": listenersFound = true var listeners map[string]struct { UseTLS bool `json:"useTls"` TLSImplicit bool `json:"implicitTls"` } if err := json.Unmarshal(operation.Value, &listeners); err != nil { t.Fatal(err) } if listeners["submission"].UseTLS || listeners["submission"].TLSImplicit { t.Fatal("cluster-local submission listener requires TLS") } if !listeners["smtp"].UseTLS || !listeners["submissions"].UseTLS || !listeners["submissions"].TLSImplicit || !listeners["imaps"].UseTLS || !listeners["imaps"].TLSImplicit { t.Fatal("public mail listeners do not retain their TLS settings") } case "MtaStageAuth": authFound = true if !bytes.Contains(operation.Value, []byte(`"if": "local_port == 587 || (local_port != 25 && is_tls)"`)) { t.Fatal("cluster-local submission does not offer password authentication") } } } if !listenersFound || !authFound { t.Fatal("submission listener or authentication policy is absent") } } func TestStalwartUsesCombinedTrustBundle(t *testing.T) { for _, path := range []string{ "k8s/bootstrap/server/job.yaml", "k8s/normal/server/deploy.yaml", } { content := serviceFiles(t, path) for _, required := range []string{ "cat /etc/ssl/certs/ca-certificates.crt /run/runtime/internal_ca_cert > /run/trust/ca-certificates.crt", "value: /run/trust/ca-certificates.crt", "mountPath: /run/trust", } { if !bytes.Contains(content, []byte(required)) { t.Fatalf("%s does not contain %q", path, required) } } } } func TestRenderPlanRejectsSecretTraversal(t *testing.T) { root := t.TempDir() source := filepath.Join(root, "plan.json") if err := os.WriteFile(source, []byte(`[{"credential":{"$secretFile":"../password"}}]`), 0o600); err != nil { t.Fatal(err) } err := renderPlan(source, filepath.Join(root, "plan.ndjson"), root) if err == nil { t.Fatal("secret traversal was accepted") } } func TestCertificateInventoryPreservesDuplicateSANsForCleanup(t *testing.T) { output := strings.Join([]string{ `{"id":"other","subjectAlternativeNames":{"other.example.invalid":true}}`, `{"id":"declared","subjectAlternativeNames":{"*.example.invalid":true}}`, `{"id":"duplicate","subjectAlternativeNames":{"mail.example.invalid":true}}`, }, "\n") inventory, err := certificateInventory([]byte(output), "mail.example.invalid") if err != nil { t.Fatal(err) } if !slices.Equal(inventory.ids, []string{"other", "declared", "duplicate"}) { t.Fatalf("ids = %q", inventory.ids) } if !slices.Equal(inventory.matches, []string{"declared", "duplicate"}) { t.Fatalf("matches = %q", inventory.matches) } } func TestReconcileUpdatesDefaultCertificateBeforePruningDuplicates(t *testing.T) { root := t.TempDir() plan := filepath.Join(root, "plan.json") renderedPlan := filepath.Join(root, "plan.ndjson") cli := filepath.Join(root, "stalwart-cli") calls := filepath.Join(root, "calls") if err := os.WriteFile(plan, []byte(`[{"@type":"update","object":"DataStore"}]`), 0o600); err != nil { t.Fatal(err) } if err := os.WriteFile(cli, []byte(`#!/bin/sh printf '%s\n' "$*" >> "$STALWART_TEST_CALLS" if [ "$1 $2" = "query Certificate" ]; then printf '%s\n' '{"id":"obsolete","subjectAlternativeNames":{"mail.example.invalid":true}}' printf '%s\n' '{"id":"mail","subjectAlternativeNames":{"mail.example.invalid":true}}' fi `), 0o700); err != nil { t.Fatal(err) } t.Setenv("STALWART_TEST_CALLS", calls) if err := reconcileOnce(context.Background(), options{ plan: plan, renderedPlan: renderedPlan, secretRoot: root, cli: cli, stalwartURL: "http://stalwart.invalid", certificateSAN: "mail.example.invalid", updateApps: true, }); err != nil { t.Fatal(err) } content, err := os.ReadFile(calls) if err != nil { t.Fatal(err) } lines := strings.Split(strings.TrimSpace(string(content)), "\n") want := []string{ "apply --file " + renderedPlan + " --no-color", "create Action/ReloadSettings --json {} --no-color", "create Action/UpdateApps --json {} --no-color", "query Certificate --fields id,subjectAlternativeNames --json --no-color", `update Certificate mail --field certificate={"@type":"File","filePath":"/run/tls/./tls.crt"} --field privateKey={"@type":"File","filePath":"/run/tls/tls.key"} --no-color`, `update Certificate mail --field certificate={"@type":"File","filePath":"/run/tls/tls.crt"} --field privateKey={"@type":"File","filePath":"/run/tls/tls.key"} --no-color`, "create Action/ReloadTlsCertificates --json {} --no-color", "query Certificate --fields id,subjectAlternativeNames --json --no-color", "update SystemSettings --field defaultCertificateId=mail --no-color", "delete Certificate --ids obsolete --no-color", "create Action/ReloadTlsCertificates --json {} --no-color", } if strings.Join(lines, "\n") != strings.Join(want, "\n") { t.Fatalf("calls = %q, want %q", lines, want) } } func TestReconcileCertificateCreatesMissingCertificate(t *testing.T) { root := t.TempDir() cli := filepath.Join(root, "stalwart-cli") state := filepath.Join(root, "created") reloaded := filepath.Join(root, "reloaded") calls := filepath.Join(root, "calls") script := fmt.Sprintf(`#!/bin/sh printf '%%s\n' "$*" >> "$STALWART_TEST_CALLS" if [ "$1 $2" = "query Certificate" ] && [ -f %q ]; then if [ -f %q ]; then printf '%%s\n' '{"id":"mail","subjectAlternativeNames":{"mail.example.invalid":true}}' else printf '%%s\n' '{"id":"mail","subjectAlternativeNames":{}}' fi elif [ "$1 $2" = "create Certificate" ]; then touch %q elif [ "$1 $2" = "create Action/ReloadTlsCertificates" ]; then touch %q fi `, state, reloaded, state, reloaded) if err := os.WriteFile(cli, []byte(script), 0o700); err != nil { t.Fatal(err) } t.Setenv("STALWART_TEST_CALLS", calls) if err := reconcileCertificate(t.Context(), options{ cli: cli, certificateSAN: "mail.example.invalid", }, nil); err != nil { t.Fatal(err) } content, err := os.ReadFile(calls) if err != nil { t.Fatal(err) } lines := strings.Split(strings.TrimSpace(string(content)), "\n") if len(lines) != 8 || lines[0] != "query Certificate --fields id,subjectAlternativeNames --json --no-color" || !strings.HasPrefix(lines[1], "create Certificate ") || lines[2] != lines[0] || !strings.HasPrefix(lines[3], "update Certificate mail ") || !strings.HasPrefix(lines[4], "update Certificate mail ") || lines[5] != "create Action/ReloadTlsCertificates --json {} --no-color" || lines[6] != lines[0] || lines[7] != "update SystemSettings --field defaultCertificateId=mail --no-color" { t.Fatalf("calls = %q", lines) } } func TestBootstrapEnvironmentReadsProjectedCredential(t *testing.T) { path := filepath.Join(t.TempDir(), "password") if err := os.WriteFile(path, []byte("bootstrap-value\n"), 0o400); err != nil { t.Fatal(err) } environment, err := bootstrapEnvironment(options{ bootstrapUser: "recovery", bootstrapPasswordFile: path, }) if err != nil { t.Fatal(err) } if strings.Join(environment, "\n") != "STALWART_USER=recovery\nSTALWART_PASSWORD=bootstrap-value" { t.Fatal("bootstrap environment does not use the projected credential") } } func TestOIDCEnvironmentUsesDiscoveryAndProjectAudience(t *testing.T) { var server *httptest.Server server = httptest.NewServer(http.HandlerFunc(func(response http.ResponseWriter, request *http.Request) { switch request.URL.Path { case "/.well-known/openid-configuration": _ = json.NewEncoder(response).Encode(map[string]string{ "issuer": server.URL, "jwks_uri": server.URL + "/keys", "token_endpoint": server.URL + "/token", }) case "/token": clientID, clientSecret, ok := request.BasicAuth() if !ok || clientID != "controller-id" || clientSecret != "controller-secret" { response.WriteHeader(http.StatusUnauthorized) return } if err := request.ParseForm(); err != nil { t.Fatal(err) } if request.Form.Get("grant_type") != "client_credentials" || request.Form.Get("scope") != "openid urn:zitadel:iam:org:project:id:project-id:aud" { t.Fatalf("token form = %v", request.Form) } _ = json.NewEncoder(response).Encode(map[string]string{"access_token": "synthetic-token"}) default: response.WriteHeader(http.StatusNotFound) } })) defer server.Close() root := t.TempDir() files := map[string]string{ "issuer": server.URL, "audience": "project-id", "client-id": "controller-id", "client-secret": "controller-secret", } for name, content := range files { if err := os.WriteFile(filepath.Join(root, name), []byte(content), 0o400); err != nil { t.Fatal(err) } } environment, err := oidcEnvironment(context.Background(), options{ oidcIssuerFile: filepath.Join(root, "issuer"), oidcAudienceFile: filepath.Join(root, "audience"), oidcClientIDFile: filepath.Join(root, "client-id"), oidcClientSecretFile: filepath.Join(root, "client-secret"), }) if err != nil { t.Fatal(err) } if strings.Join(environment, "\n") != "STALWART_TOKEN=synthetic-token" { t.Fatalf("environment = %q", environment) } } func TestWaitForOIDCRequiresAdvertisedKeys(t *testing.T) { var server *httptest.Server server = httptest.NewServer(http.HandlerFunc(func(response http.ResponseWriter, request *http.Request) { switch request.URL.Path { case "/.well-known/openid-configuration": _ = json.NewEncoder(response).Encode(map[string]string{ "issuer": server.URL, "jwks_uri": server.URL + "/keys", "token_endpoint": server.URL + "/token", }) case "/keys": _, _ = io.WriteString(response, `{"keys":[{"kid":"synthetic"}]}`) default: response.WriteHeader(http.StatusNotFound) } })) defer server.Close() issuerFile := filepath.Join(t.TempDir(), "issuer") if err := os.WriteFile(issuerFile, []byte(server.URL), 0o400); err != nil { t.Fatal(err) } if err := waitForOIDC(context.Background(), issuerFile, time.Millisecond); err != nil { t.Fatal(err) } } func TestReconcileUsesOnlyBootstrapCredentials(t *testing.T) { root := t.TempDir() plan := filepath.Join(root, "plan.json") renderedPlan := filepath.Join(root, "plan.ndjson") cli := filepath.Join(root, "stalwart-cli") calls := filepath.Join(root, "calls") password := filepath.Join(root, "bootstrap-password") if err := os.WriteFile(plan, []byte(`[{"@type":"update","object":"DataStore"}]`), 0o600); err != nil { t.Fatal(err) } if err := os.WriteFile(password, []byte("bootstrap-value"), 0o400); err != nil { t.Fatal(err) } if err := os.WriteFile(cli, []byte(`#!/bin/sh printf '%s|%s|%s|%s\n' "$STALWART_URL" "$STALWART_USER" "$STALWART_PASSWORD" "$*" >> "$STALWART_TEST_CALLS" if [ "$STALWART_URL" != "http://stalwart-bootstrap:8080" ]; then exit 1 fi `), 0o700); err != nil { t.Fatal(err) } t.Setenv("STALWART_TEST_CALLS", calls) if err := reconcileOnce(context.Background(), options{ plan: plan, renderedPlan: renderedPlan, secretRoot: root, cli: cli, stalwartURL: "http://stalwart-bootstrap:8080", bootstrapUser: "recovery", bootstrapPasswordFile: password, }); err != nil { t.Fatal(err) } content, err := os.ReadFile(calls) if err != nil { t.Fatal(err) } lines := strings.Split(strings.TrimSpace(string(content)), "\n") if len(lines) != 2 { t.Fatalf("CLI calls = %q", lines) } for _, line := range lines { if !strings.HasPrefix(line, "http://stalwart-bootstrap:8080|recovery|bootstrap-value|") { t.Fatalf("bootstrap CLI call = %q", line) } } } func TestMergedEnvironmentReplacesCredential(t *testing.T) { environment := mergedEnvironment( []string{"PATH=/bin", "STALWART_PASSWORD=old"}, []string{"STALWART_USER=operator", "STALWART_PASSWORD=new"}, ) value := strings.Join(environment, "\n") if strings.Contains(value, "STALWART_PASSWORD=old") { t.Fatal("old credential remains in child environment") } if !strings.Contains(value, "STALWART_PASSWORD=new") { t.Fatal("new credential is absent from child environment") } } func TestStatusDoesNotContainCredentialFields(t *testing.T) { encoded, err := json.Marshal(controllerStatus{ Leader: true, Ready: true, PlanRevision: strings.Repeat("a", 64), BuildRevision: "test-revision", }) if err != nil { t.Fatal(err) } value := strings.ToLower(string(encoded)) if strings.Contains(value, "password") || strings.Contains(value, "secret") { t.Fatal("status exposes a credential field") } }