diff --git a/internal/openbao/credentials.go b/internal/openbao/credentials.go new file mode 100644 index 0000000..688f006 --- /dev/null +++ b/internal/openbao/credentials.go @@ -0,0 +1,133 @@ +/* +Copyright 2026. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +// Package openbao supplies credentials using one shared client and Kubernetes identity. +package openbao + +import ( + "context" + "errors" + "sync" + "time" + + api "git.ddupan.top/panxiao81/postgresql-tenant-operator/api/v1alpha1" + "git.ddupan.top/panxiao81/postgresql-tenant-operator/internal/instance" + kubernetesauth "github.com/openbao/openbao/api/auth/kubernetes/v2" + bao "github.com/openbao/openbao/api/v2" +) + +// Credentials owns authentication; database connections never depend on this client. +type Credentials struct { + client *bao.Client + mount string + authenticate func(context.Context) (*bao.Secret, error) + mu sync.Mutex + expires time.Time +} + +func NewCredentials(client *bao.Client, mount, authMount, role string) *Credentials { + source := &Credentials{client: client, mount: mount} + source.authenticate = func(ctx context.Context) (*bao.Secret, error) { + // Constructing auth here rereads the projected JWT after Kubernetes rotates it. + auth, err := kubernetesauth.NewKubernetesAuth(role, kubernetesauth.WithMountPath(authMount)) + if err != nil { + return nil, instance.Failure{Reason: api.ReasonAuthenticationFailed, Operation: "read Kubernetes identity"} + } + return client.Auth().Login(ctx, auth) + } + return source +} + +func (s *Credentials) login(ctx context.Context) error { + secret, err := s.authenticate(ctx) + if err != nil { + return classify("authenticate to OpenBao", err, true) + } + if secret == nil || secret.Auth == nil || secret.Auth.ClientToken == "" || secret.Auth.LeaseDuration <= 0 { + return instance.Failure{Reason: api.ReasonAuthenticationFailed, Operation: "OpenBao returned no leased identity"} + } + s.client.SetToken(secret.Auth.ClientToken) + ttl := time.Duration(secret.Auth.LeaseDuration) * time.Second + s.expires = time.Now().Add(ttl - ttl/10) + return nil +} + +func (s *Credentials) Read(ctx context.Context, reference api.OpenBaoSecretReference) (instance.Credentials, error) { + // Login and reads share a lock so concurrent first-use requests cannot race token replacement. + s.mu.Lock() + defer s.mu.Unlock() + if !time.Now().Before(s.expires) { + if err := s.login(ctx); err != nil { + return instance.Credentials{}, err + } + } + secret, err := s.client.KVv2(s.mount).Get(ctx, reference.Path) + var response *bao.ResponseError + if errors.As(err, &response) && response.StatusCode == 403 { + // A token may be revoked before its TTL. Reauthenticate once, never loop on denied policies. + s.expires = time.Time{} + if err := s.login(ctx); err != nil { + return instance.Credentials{}, err + } + secret, err = s.client.KVv2(s.mount).Get(ctx, reference.Path) + } + if err != nil { + return instance.Credentials{}, classify("read administrative credential", err, false) + } + usernameKey, passwordKey := reference.UsernameKey, reference.PasswordKey + if usernameKey == "" { + usernameKey = "username" + } + if passwordKey == "" { + passwordKey = "password" + } + if secret == nil { + return instance.Credentials{}, invalidCredential() + } + username, usernameOK := secret.Data[usernameKey].(string) + password, passwordOK := secret.Data[passwordKey].(string) + if !usernameOK || !passwordOK || username == "" || password == "" { + return instance.Credentials{}, invalidCredential() + } + return instance.Credentials{Username: username, Password: password}, nil +} + +func invalidCredential() error { + return instance.Failure{Reason: api.ReasonAuthenticationFailed, Operation: "administrative credential fields are missing"} +} + +func classify(operation string, err error, login bool) error { + reason := api.ReasonDependencyUnavailable + var safe instance.Failure + if errors.As(err, &safe) { + return safe + } + var response *bao.ResponseError + if errors.As(err, &response) { + switch response.StatusCode { + case 403: + reason = api.ReasonInsufficientPrivileges + if login { + reason = api.ReasonAuthenticationFailed + } + case 400, 401: + reason = api.ReasonAuthenticationFailed + case 404: + reason = api.ReasonInvalidSpec + } + } + return instance.Failure{Reason: reason, Operation: operation} +} diff --git a/internal/openbao/credentials_test.go b/internal/openbao/credentials_test.go new file mode 100644 index 0000000..62fba9a --- /dev/null +++ b/internal/openbao/credentials_test.go @@ -0,0 +1,125 @@ +/* +Copyright 2026. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package openbao + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "sync" + "sync/atomic" + "testing" + "time" + + api "git.ddupan.top/panxiao81/postgresql-tenant-operator/api/v1alpha1" + bao "github.com/openbao/openbao/api/v2" +) + +const testToken = "test-token" +const testAdmin = "admin" + +func TestSharedIdentityReauthenticatesOnExpiryAndRevocation(t *testing.T) { + var logins atomic.Int32 + var revoked atomic.Bool + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/v1/secret/data/admin" { + http.NotFound(w, r) + return + } + if revoked.Swap(false) { + w.WriteHeader(http.StatusForbidden) + return + } + if r.Header.Get("X-Vault-Token") != testToken { + w.WriteHeader(http.StatusForbidden) + return + } + _ = json.NewEncoder(w).Encode(map[string]any{ + "data": map[string]any{ + "data": map[string]any{"username": testAdmin, "password": "test-password"}, + "metadata": map[string]any{"version": 1}, + }, + }) + })) + defer server.Close() + config := bao.NewConfig() + config.Address = server.URL + config.MaxRetries = 0 + client, err := bao.NewClient(config) + if err != nil { + t.Fatal(err) + } + source := NewCredentials(client, "secret", "kubernetes", "controller") + source.authenticate = func(context.Context) (*bao.Secret, error) { + logins.Add(1) + return &bao.Secret{Auth: &bao.SecretAuth{ClientToken: testToken, LeaseDuration: 300}}, nil + } + reference := api.OpenBaoSecretReference{Path: testAdmin} + var wg sync.WaitGroup + for range 8 { + wg.Go(func() { + if _, err := source.Read(context.Background(), reference); err != nil { + t.Error(err) + } + }) + } + wg.Wait() + if logins.Load() != 1 { + t.Fatal("concurrent reads repeatedly authenticated") + } + source.expires = time.Time{} + if _, err := source.Read(context.Background(), reference); err != nil { + t.Fatal(err) + } + if logins.Load() != 2 { + t.Fatal("expired identity was not refreshed") + } + revoked.Store(true) + if _, err := source.Read(context.Background(), reference); err != nil { + t.Fatal(err) + } + if logins.Load() != 3 { + t.Fatal("revoked identity was not refreshed") + } +} + +func TestDeniedPolicyRetriesAuthenticationOnlyOnce(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusForbidden) + })) + defer server.Close() + config := bao.NewConfig() + config.Address = server.URL + config.MaxRetries = 0 + client, err := bao.NewClient(config) + if err != nil { + t.Fatal(err) + } + source := NewCredentials(client, "secret", "kubernetes", "controller") + logins := 0 + source.authenticate = func(context.Context) (*bao.Secret, error) { + logins++ + return &bao.Secret{Auth: &bao.SecretAuth{ClientToken: testToken, LeaseDuration: 300}}, nil + } + if _, err := source.Read(context.Background(), api.OpenBaoSecretReference{Path: testAdmin}); err == nil { + t.Fatal("denied policy was accepted") + } + if logins != 2 { + t.Fatal("authentication did not stop after one retry") + } +}