feat: 接入 OpenBao Kubernetes 认证与短期会话续期

internal/database/adapter/openbao/authentication.go

@@ -0,0 +1,141 @@
+/*
+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"
+	"errors"
+	"os"
+	"path/filepath"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	kubernetesauth "github.com/openbao/openbao/api/auth/kubernetes/v2"
+	bao "github.com/openbao/openbao/api/v2"
+)
+
+var ErrAuthenticationConfiguration = errors.New("invalid OpenBao Kubernetes authentication configuration")
+
+// KubernetesSession 为专用 SDK client 维护短期登录，不持久化或对外返回 token。
+// 登录失败或 lease 不再可用时清空 client token，依赖恢复后重新读取投射的 SA token。
+// 续期调度由官方 LifetimeWatcher 负责，不实现自己的 lease 算法。
+type KubernetesSession struct {
+	client    *bao.Client
+	mount     string
+	role      string
+	tokenPath string
+	running   sync.Mutex
+	ready     atomic.Bool
+}
+
+func NewKubernetesSession(client *bao.Client, mount, role, tokenPath string) (*KubernetesSession, error) {
+	if client == nil || !validPath(mount) || !pathSegment.MatchString(role) || !filepath.IsAbs(tokenPath) {
+		return nil, ErrAuthenticationConfiguration
+	}
+	client.ClearToken()
+	client.SetMaxRetries(0)
+	client.SetClientTimeout(15 * time.Second)
+	return &KubernetesSession{client: client, mount: mount, role: role, tokenPath: tokenPath}, nil
+}
+
+// Ready 仅代表当前登录 lease 正由 SDK 管理，不保证下一次后端请求一定成功。
+func (s *KubernetesSession) Ready() bool { return s.ready.Load() }
+
+// Start 可交给 manager 管理；关闭时清空本地 token，不撤销共享后端数据。
+// SDK 的 Stop 不取消已发出的续期 HTTP 请求，因此等待该请求结束后才退出，最长受 client timeout 限制。
+func (s *KubernetesSession) Start(ctx context.Context) error {
+	if !s.running.TryLock() {
+		return errors.New("OpenBao authentication is already running")
+	}
+	defer s.running.Unlock()
+	defer s.clear()
+	for ctx.Err() == nil {
+		s.clear()
+		secret := s.login(ctx)
+		if secret != nil {
+			s.watch(ctx, secret)
+		}
+		s.clear()
+		// 登录失败及不能继续续期均有限速，防止依赖故障时形成请求忙循环。
+		select {
+		case <-ctx.Done():
+			return nil
+		case <-time.After(5 * time.Second):
+		}
+	}
+	return nil
+}
+
+func (s *KubernetesSession) clear() {
+	s.ready.Store(false)
+	s.client.ClearToken()
+}
+
+func (s *KubernetesSession) login(ctx context.Context) *bao.Secret {
+	// 只读取指定投射位置；SDK 的空 token 选项会回退默认路径，必须在此拒绝空文件。
+	data, err := os.ReadFile(s.tokenPath)
+	if err != nil || strings.TrimSpace(string(data)) == "" {
+		return nil
+	}
+	// helper 会缓存 token，不能跨登录轮次复用。
+	method, err := kubernetesauth.NewKubernetesAuth(s.role,
+		kubernetesauth.WithMountPath(s.mount),
+		kubernetesauth.WithServiceAccountToken(strings.TrimSpace(string(data))),
+	)
+	if err != nil {
+		return nil
+	}
+	// 先检查短期 lease，再发布到共享 client，避免暴露不合规的登录结果。
+	secret, err := method.Login(ctx, s.client)
+	if err != nil || secret == nil || secret.Auth == nil || secret.Auth.ClientToken == "" || secret.Auth.LeaseDuration <= 0 {
+		return nil
+	}
+	if ctx.Err() != nil {
+		return nil
+	}
+	s.client.SetToken(secret.Auth.ClientToken)
+	return secret
+}
+
+func (s *KubernetesSession) watch(ctx context.Context, secret *bao.Secret) {
+	behavior := bao.RenewBehaviorErrorOnErrors
+	if !secret.Auth.Renewable {
+		behavior = bao.RenewBehaviorRenewDisabled
+	}
+	watcher, err := s.client.NewLifetimeWatcher(&bao.LifetimeWatcherInput{Secret: secret, RenewBehavior: behavior})
+	if err != nil {
+		return
+	}
+	s.ready.Store(true)
+	go watcher.Start()
+	for {
+		select {
+		case <-ctx.Done():
+			s.clear()
+			watcher.Stop()
+			<-watcher.DoneCh()
+			return
+		case <-watcher.DoneCh():
+			watcher.Stop()
+			return
+		case <-watcher.RenewCh():
+			// 不记录 SDK Secret 或 token；无需复制 SDK 已处理的续期数据。
+		}
+	}
+}

internal/database/adapter/openbao/authentication_integration_test.go

@@ -0,0 +1,244 @@
+//go:build integration
+
+/*
+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_test
+
+import (
+	"context"
+	"crypto/ecdsa"
+	"crypto/elliptic"
+	"crypto/rand"
+	"crypto/tls"
+	"crypto/x509"
+	"encoding/pem"
+	"math/big"
+	"net"
+	"net/http"
+	"net/http/httptest"
+	"net/http/httputil"
+	"net/url"
+	"os/exec"
+	"path/filepath"
+	"strconv"
+	"strings"
+	"testing"
+	"time"
+
+	bao "github.com/openbao/openbao/api/v2"
+	authenticationv1 "k8s.io/api/authentication/v1"
+	corev1 "k8s.io/api/core/v1"
+	rbacv1 "k8s.io/api/rbac/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/rest"
+	"sigs.k8s.io/controller-runtime/pkg/envtest"
+
+	"git.ddupan.top/panxiao81/ayatori/internal/database/adapter/openbao"
+)
+
+const (
+	authNamespace = "bao-controller"
+	authAudience  = "openbao"
+)
+
+// Bao 容器通过 Docker bridge 访问这个仅转发 TokenReview 的临时入口。
+// 上游仍是带 CA 验证的真实 envtest API；不模拟 JWT 签名、audience 或 RBAC 判定。
+func tokenReviewEndpoint(t *testing.T, config *rest.Config) (string, string) {
+	t.Helper()
+	upstream, err := url.Parse(config.Host)
+	if err != nil {
+		t.Fatal("invalid envtest address")
+	}
+	transport, err := rest.TransportFor(rest.AnonymousClientConfig(config))
+	if err != nil {
+		t.Fatal("cannot construct TokenReview transport")
+	}
+	proxy := httputil.NewSingleHostReverseProxy(upstream)
+	proxy.Transport = transport
+	proxy.ErrorHandler = func(w http.ResponseWriter, _ *http.Request, _ error) { w.WriteHeader(http.StatusBadGateway) }
+	server := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.Method != http.MethodPost || r.URL.Path != "/apis/authentication.k8s.io/v1/tokenreviews" {
+			w.WriteHeader(http.StatusForbidden)
+			return
+		}
+		proxy.ServeHTTP(w, r)
+	}))
+	if err := server.Listener.Close(); err != nil {
+		t.Fatal("cannot replace fixture listener")
+	}
+	server.Listener, err = net.Listen("tcp", "0.0.0.0:0")
+	if err != nil {
+		t.Fatal("cannot expose TokenReview fixture")
+	}
+	ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second)
+	defer cancel()
+	output, err := exec.CommandContext(ctx, "docker", "network", "inspect", "bridge", "--format",
+		`{{(index .IPAM.Config 0).Gateway}}`).Output()
+	if err != nil {
+		t.Fatal("cannot locate fixture Docker bridge")
+	}
+	gateway := strings.TrimSpace(string(output))
+	if net.ParseIP(gateway) == nil {
+		t.Fatal("invalid fixture bridge gateway")
+	}
+	certificate, caPEM := tokenReviewCertificate(t, net.ParseIP(gateway))
+	server.TLS = &tls.Config{Certificates: []tls.Certificate{certificate}, MinVersion: tls.VersionTLS12}
+	server.StartTLS()
+	t.Cleanup(server.Close)
+	port := server.Listener.Addr().(*net.TCPAddr).Port
+	return "https://" + net.JoinHostPort(gateway, strconv.Itoa(port)), caPEM
+}
+
+func tokenReviewCertificate(t *testing.T, address net.IP) (tls.Certificate, string) {
+	t.Helper()
+	key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
+	if err != nil {
+		t.Fatal("cannot create fixture TLS key")
+	}
+	template := &x509.Certificate{
+		SerialNumber: big.NewInt(1),
+		NotBefore:    time.Now().Add(-time.Minute),
+		NotAfter:     time.Now().Add(time.Hour),
+		IPAddresses:  []net.IP{address},
+		KeyUsage:     x509.KeyUsageDigitalSignature | x509.KeyUsageCertSign,
+		ExtKeyUsage:  []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth},
+		IsCA:         true, BasicConstraintsValid: true,
+	}
+	der, err := x509.CreateCertificate(rand.Reader, template, template, &key.PublicKey, key)
+	if err != nil {
+		t.Fatal("cannot create fixture TLS certificate")
+	}
+	certificate := tls.Certificate{Certificate: [][]byte{der}, PrivateKey: key}
+	return certificate, string(pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: der}))
+}
+
+func TestKubernetesSessionWithRealTokenReview(t *testing.T) {
+	environment := &envtest.Environment{}
+	config, err := environment.Start()
+	if err != nil {
+		t.Fatal("cannot start authentication API fixture", err)
+	}
+	t.Cleanup(func() {
+		if err := environment.Stop(); err != nil {
+			t.Error("cannot stop authentication API fixture")
+		}
+	})
+	clientset, err := kubernetes.NewForConfig(config)
+	if err != nil {
+		t.Fatal("cannot construct fixture API client")
+	}
+	ctx := t.Context()
+	for _, namespace := range []string{authNamespace, "bao-unrelated"} {
+		if _, err := clientset.CoreV1().Namespaces().Create(ctx, &corev1.Namespace{Name: namespace}, metav1.CreateOptions{}); err != nil {
+			t.Fatal("cannot create fixture namespace")
+		}
+		if _, err := clientset.CoreV1().ServiceAccounts(namespace).Create(ctx, &corev1.ServiceAccount{Name: authRole}, metav1.CreateOptions{}); err != nil {
+			t.Fatal("cannot create fixture service account")
+		}
+	}
+	if _, err := clientset.RbacV1().ClusterRoleBindings().Create(ctx, &rbacv1.ClusterRoleBinding{
+		Name:     "bao-fixture-reviewer",
+		RoleRef:  rbacv1.RoleRef{APIGroup: "rbac.authorization.k8s.io", Kind: "ClusterRole", Name: "system:auth-delegator"},
+		Subjects: []rbacv1.Subject{{Kind: "ServiceAccount", Namespace: authNamespace, Name: authRole}},
+	}, metav1.CreateOptions{}); err != nil {
+		t.Fatal("cannot authorize fixture TokenReview")
+	}
+	requestToken := func(namespace string, audiences []string) string {
+		t.Helper()
+		response, err := clientset.CoreV1().ServiceAccounts(namespace).CreateToken(ctx, authRole,
+			&authenticationv1.TokenRequest{Spec: authenticationv1.TokenRequestSpec{Audiences: audiences}}, metav1.CreateOptions{})
+		if err != nil {
+			t.Fatal("cannot issue fixture service account token")
+		}
+		return response.Status.Token
+	}
+	root := baoFixture(t)
+	if err := root.Sys().EnableAuthWithOptionsWithContext(ctx, "kubernetes", &bao.EnableAuthOptions{Type: "kubernetes"}); err != nil {
+		t.Fatal("cannot enable fixture Kubernetes auth")
+	}
+	reviewerToken := requestToken(authNamespace, nil)
+	reviewURL, reviewCA := tokenReviewEndpoint(t, config)
+	if _, err := root.Logical().WriteWithContext(ctx, "auth/kubernetes/config", map[string]any{
+		"kubernetes_host":      reviewURL,
+		"kubernetes_ca_cert":   reviewCA,
+		"token_reviewer_jwt":   reviewerToken,
+		"disable_local_ca_jwt": true,
+	}); err != nil {
+		t.Fatal("cannot configure fixture TokenReview:", strings.NewReplacer(reviewerToken, "[REDACTED]", fixtureToken, "[REDACTED]").Replace(err.Error()))
+	}
+	if err := root.Sys().PutPolicyWithContext(ctx, authRole, `path "secret/data/applications/*" { capabilities = ["create", "update", "read"] }`); err != nil {
+		t.Fatal("cannot configure fixture credential policy")
+	}
+	if _, err := root.Logical().WriteWithContext(ctx, "auth/kubernetes/role/controller", map[string]any{
+		"bound_service_account_names":      []string{authRole},
+		"bound_service_account_namespaces": []string{authNamespace},
+		"audience":                         authAudience,
+		"token_policies":                   []string{authRole},
+		"token_ttl":                        "3s", "token_max_ttl": "15s",
+	}); err != nil {
+		t.Fatal("cannot configure fixture auth role")
+	}
+	path := filepath.Join(t.TempDir(), "token")
+	writeAuthToken(t, path, requestToken(authNamespace, []string{authAudience}))
+	client := fixtureClient(t, root.Address())
+	session, err := openbao.NewKubernetesSession(client, "kubernetes", authRole, path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	cancel := startSession(t, session)
+	waitForAuthentication(t, session.Ready)
+	store := fixtureStore(t, client)
+	if err := store.Create(ctx, credentialPath, fixtureCredential(t)); err != nil {
+		t.Fatal("Kubernetes identity cannot create scoped credential", err)
+	}
+	initialToken := client.Token()
+	// 超过初始 TTL 后同一 token 仍可用，证明发生真实 renew-self，而非只登录一次。
+	start := time.Now()
+	waitForAuthentication(t, func() bool { return time.Since(start) > 4*time.Second })
+	if client.Token() != initialToken {
+		t.Fatal("token was replaced before renewal could be verified")
+	}
+	if _, err := client.Auth().Token().LookupSelfWithContext(ctx); err != nil {
+		t.Fatal("short-lived token was not renewed")
+	}
+	for _, invalid := range []struct {
+		namespace string
+		audience  string
+	}{
+		{"bao-unrelated", authAudience}, {authNamespace, "wrong-audience"},
+	} {
+		writeAuthToken(t, path, requestToken(invalid.namespace, []string{invalid.audience}))
+		if err := root.Auth().Token().RevokeOrphanWithContext(ctx, client.Token()); err != nil {
+			t.Fatal("cannot revoke fixture token")
+		}
+		waitForAuthentication(t, func() bool { return !session.Ready() && client.Token() == "" })
+		// 用相同真实登录入口直接确认 namespace/audience 拒绝，不依赖定时轮询推断。
+		if _, err := root.Logical().WriteWithContext(ctx, "auth/kubernetes/login", map[string]any{
+			"role": authRole, "jwt": requestToken(invalid.namespace, []string{invalid.audience}),
+		}); err == nil {
+			t.Fatal("invalid Kubernetes identity was accepted")
+		}
+		writeAuthToken(t, path, requestToken(authNamespace, []string{authAudience}))
+		waitForAuthentication(t, session.Ready)
+		if _, err := store.Read(ctx, credentialPath); err != nil {
+			t.Fatal("credential access did not recover", err)
+		}
+	}
+	cancel()
+	waitForAuthentication(t, func() bool { return !session.Ready() && client.Token() == "" })
+}

internal/database/adapter/openbao/authentication_test.go

@@ -0,0 +1,176 @@
+/*
+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_test
+
+import (
+	"context"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"path/filepath"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"git.ddupan.top/panxiao81/ayatori/internal/database/adapter/openbao"
+)
+
+const (
+	authRole = "controller"
+)
+
+func writeAuthToken(t *testing.T, path, token string) {
+	t.Helper()
+	if err := os.WriteFile(path, []byte(token), 0600); err != nil {
+		t.Fatal("cannot write temporary projected token")
+	}
+}
+
+func TestKubernetesSessionDoesNotFallbackFromMissingToken(t *testing.T) {
+	var requests atomic.Int32
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		requests.Add(1)
+		w.WriteHeader(http.StatusForbidden)
+	}))
+	defer server.Close()
+	for _, missing := range []bool{true, false} {
+		path := filepath.Join(t.TempDir(), "token")
+		if !missing {
+			writeAuthToken(t, path, " \n")
+		}
+		client := fixtureClient(t, server.URL)
+		session, err := openbao.NewKubernetesSession(client, "kubernetes", authRole, path)
+		if err != nil {
+			t.Fatal(err)
+		}
+		ctx, cancel := context.WithTimeout(t.Context(), 100*time.Millisecond)
+		err = session.Start(ctx)
+		cancel()
+		if err != nil || session.Ready() || client.Token() != "" || requests.Load() != 0 {
+			t.Fatal("missing or empty token must not fall back to another identity")
+		}
+	}
+}
+
+func waitForAuthentication(t *testing.T, check func() bool) {
+	t.Helper()
+	deadline := time.NewTimer(20 * time.Second)
+	defer deadline.Stop()
+	for !check() {
+		select {
+		case <-deadline.C:
+			t.Fatal("authentication condition timed out")
+		case <-time.After(20 * time.Millisecond):
+		}
+	}
+}
+
+func startSession(t *testing.T, session *openbao.KubernetesSession) context.CancelFunc {
+	t.Helper()
+	ctx, cancel := context.WithCancel(t.Context())
+	done := make(chan error, 1)
+	go func() { done <- session.Start(ctx) }()
+	t.Cleanup(func() {
+		cancel()
+		select {
+		case <-done:
+		case <-time.After(20 * time.Second):
+			t.Error("authentication did not stop")
+		}
+	})
+	return cancel
+}
+
+func TestKubernetesSessionRereadsTokenAfterFailure(t *testing.T) {
+	var attempts atomic.Int32
+	var accepted atomic.Bool
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path != "/v1/auth/kubernetes/login" {
+			w.WriteHeader(http.StatusNotFound)
+			return
+		}
+		var request struct{ JWT, Role string }
+		if json.NewDecoder(r.Body).Decode(&request) != nil || request.Role != authRole {
+			t.Error("unexpected authentication request")
+			w.WriteHeader(http.StatusBadRequest)
+			return
+		}
+		attempts.Add(1)
+		if request.JWT != "rotated-test-jwt" {
+			w.WriteHeader(http.StatusForbidden)
+			return
+		}
+		accepted.Store(true)
+		if err := json.NewEncoder(w).Encode(map[string]any{"auth": map[string]any{
+			"client_token": fixtureToken, "lease_duration": 60, "renewable": false,
+		}}); err != nil {
+			t.Error("cannot encode authentication fixture response")
+		}
+	}))
+	defer server.Close()
+	path := filepath.Join(t.TempDir(), "token")
+	writeAuthToken(t, path, "expired-test-jwt")
+	client := fixtureClient(t, server.URL)
+	session, err := openbao.NewKubernetesSession(client, "kubernetes", authRole, path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if client.Token() != "" {
+		t.Fatal("constructor retained preexisting static token")
+	}
+	cancel := startSession(t, session)
+	waitForAuthentication(t, func() bool { return attempts.Load() > 0 })
+	if session.Ready() || client.Token() != "" {
+		t.Fatal("failed login retained credentials")
+	}
+	writeAuthToken(t, path, "rotated-test-jwt")
+	waitForAuthentication(t, func() bool { return accepted.Load() && session.Ready() })
+	if client.Token() != fixtureToken {
+		t.Fatal("successful login did not configure the client")
+	}
+	if session.Start(t.Context()) == nil {
+		t.Fatal("allowed concurrent lifecycle owners")
+	}
+	cancel()
+	waitForAuthentication(t, func() bool { return !session.Ready() && client.Token() == "" })
+}
+
+func TestKubernetesSessionRejectsUnboundedLease(t *testing.T) {
+	var attempts atomic.Int32
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		attempts.Add(1)
+		if err := json.NewEncoder(w).Encode(map[string]any{"auth": map[string]any{
+			"client_token": fixtureToken, "lease_duration": 0,
+		}}); err != nil {
+			t.Error("cannot encode authentication fixture response")
+		}
+	}))
+	defer server.Close()
+	path := filepath.Join(t.TempDir(), "token")
+	writeAuthToken(t, path, "test-jwt")
+	client := fixtureClient(t, server.URL)
+	session, err := openbao.NewKubernetesSession(client, "kubernetes", authRole, path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	startSession(t, session)
+	waitForAuthentication(t, func() bool { return attempts.Load() > 0 })
+	if session.Ready() || client.Token() != "" {
+		t.Fatal("accepted a token without a finite lease")
+	}
+}