panxiao81/ayatori
1
0
Fork 0
Code Issues Pull Requests 2 Actions Packages Projects Releases Wiki Activity

fix: 复用 manager kubeconfig 通过 TokenRequest 登录 OpenBao
Verify / lint (pull_request) Successful in 11m47s
Details
Verify / database-integration (pull_request) Successful in 15m24s
Details
Verify / test (pull_request) Successful in 16m41s
Details

Browse Source
This commit is contained in:
panxiao81
2026-09-25 21:02:06 +00:00
parent dcf9ab50df
commit f0aa86f676
10 changed files with 479 additions and 88 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/database/adapter/openbao/authentication.go
+46 -16
View File
@@ -19,8 +19,6 @@ package openbao
import (
"context"
"errors"
"os"
"path/filepath"
"strings"
"sync"
"sync/atomic"
@@ -28,30 +26,52 @@ import (
kubernetesauth "github.com/openbao/openbao/api/auth/kubernetes/v2"
bao "github.com/openbao/openbao/api/v2"
authenticationv1 "k8s.io/api/authentication/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/util/validation"
corev1 "k8s.io/client-go/kubernetes/typed/core/v1"
"k8s.io/client-go/rest"
)
var ErrAuthenticationConfiguration = errors.New("invalid OpenBao Kubernetes authentication configuration")
// KubernetesSession 为专用 SDK client 维护短期登录，不持久化或对外返回 token。
// 登录失败或 lease 不再可用时清空 client token，依赖恢复后重新读取投射的 SA token。
// 每次登录通过 manager 的 Kubernetes 身份申请新的 SA JWT，不依赖 controller 的部署位置。
// 续期调度由官方 LifetimeWatcher 负责，不实现自己的 lease 算法。
type KubernetesSession struct {
client *bao.Client
mount string
role string
tokenPath string
running sync.Mutex
ready atomic.Bool
client *bao.Client
mount string
role string
accounts corev1.ServiceAccountInterface
identity KubernetesIdentity
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) {
// KubernetesIdentity 是部署固定的登录目标，不由 Tenant 选择。
type KubernetesIdentity struct {
Namespace string
ServiceAccount string
Audience string
}
func NewKubernetesSession(client *bao.Client, config *rest.Config, mount, role string, identity KubernetesIdentity) (*KubernetesSession, error) {
if client == nil || config == nil || !validPath(mount) || !pathSegment.MatchString(role) ||
len(validation.IsDNS1123Label(identity.Namespace)) != 0 ||
len(validation.IsDNS1123Subdomain(identity.ServiceAccount)) != 0 || strings.TrimSpace(identity.Audience) == "" {
return nil, ErrAuthenticationConfiguration
}
api, err := corev1.NewForConfig(config)
if err != nil {
return nil, ErrAuthenticationConfiguration
}
client.ClearToken()
client.SetMaxRetries(0)
client.SetClientTimeout(15 * time.Second)
return &KubernetesSession{client: client, mount: mount, role: role, tokenPath: tokenPath}, nil
return &KubernetesSession{
client: client, mount: mount, role: role,
accounts: api.ServiceAccounts(identity.Namespace), identity: identity,
}, nil
}
// Ready 仅代表当前登录 lease 正由 SDK 管理，不保证下一次后端请求一定成功。
@@ -88,15 +108,25 @@ func (s *KubernetesSession) clear() {
}
func (s *KubernetesSession) login(ctx context.Context) *bao.Secret {
// 只读取指定投射位置；SDK 的空 token 选项会回退默认路径，必须在此拒绝空文件。
data, err := os.ReadFile(s.tokenPath)
if err != nil || strings.TrimSpace(string(data)) == "" {
ctx, cancel := context.WithTimeout(ctx, 15*time.Second)
defer cancel()
// JWT 只用于本次登录，不缓存或自行解析 kubeconfig 中的凭据。
// client-go 负责 kubeconfig/in-cluster 身份与凭据更新；API server 按 RBAC 签发。
expirationSeconds := int64(600)
token, err := s.accounts.CreateToken(ctx, s.identity.ServiceAccount, &authenticationv1.TokenRequest{
Spec: authenticationv1.TokenRequestSpec{
Audiences: []string{s.identity.Audience},
ExpirationSeconds: &expirationSeconds,
},
}, metav1.CreateOptions{})
if err != nil || token == nil || strings.TrimSpace(token.Status.Token) == "" ||
!token.Status.ExpirationTimestamp.After(time.Now()) {
return nil
}
// helper 会缓存 token，不能跨登录轮次复用。
method, err := kubernetesauth.NewKubernetesAuth(s.role,
kubernetesauth.WithMountPath(s.mount),
kubernetesauth.WithServiceAccountToken(strings.TrimSpace(string(data))),
kubernetesauth.WithServiceAccountToken(token.Status.Token),
)
if err != nil {
return nil
internal/database/adapter/openbao/authentication_integration_test.go
+108 -25
View File
@@ -33,7 +33,6 @@ import (
"net/http/httputil"
"net/url"
"os/exec"
"path/filepath"
"strconv"
"strings"
"testing"
@@ -43,17 +42,23 @@ import (
authenticationv1 "k8s.io/api/authentication/v1"
corev1 "k8s.io/api/core/v1"
rbacv1 "k8s.io/api/rbac/v1"
apierrors "k8s.io/apimachinery/pkg/api/errors"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/client-go/kubernetes"
"k8s.io/client-go/rest"
"k8s.io/client-go/tools/clientcmd"
ctrl "sigs.k8s.io/controller-runtime"
"sigs.k8s.io/controller-runtime/pkg/envtest"
metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server"
"git.ddupan.top/panxiao81/ayatori/internal/database/adapter/openbao"
)
const (
authNamespace = "bao-controller"
authAudience = "openbao"
unrelatedAuthNamespace = "bao-unrelated"
tokenRequestRole = "request-openbao-token"
authNamespace = "bao-controller"
authAudience = "openbao"
)
// Bao 容器通过 Docker bridge 访问这个仅转发 TokenReview 的临时入口。
@@ -127,7 +132,17 @@ func tokenReviewCertificate(t *testing.T, address net.IP) (tls.Certificate, stri
return certificate, string(pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: der}))
}
func TestKubernetesSessionWithRealTokenReview(t *testing.T) {
type kubernetesAuthFixture struct {
config *rest.Config
controllerConfig *rest.Config
admin *kubernetes.Clientset
controller *kubernetes.Clientset
grant func()
requestToken func(string, []string) string
}
func newKubernetesAuthFixture(t *testing.T) *kubernetesAuthFixture {
t.Helper()
environment := &envtest.Environment{}
config, err := environment.Start()
if err != nil {
@@ -143,7 +158,7 @@ func TestKubernetesSessionWithRealTokenReview(t *testing.T) {
t.Fatal("cannot construct fixture API client")
}
ctx := t.Context()
for _, namespace := range []string{authNamespace, "bao-unrelated"} {
for _, namespace := range []string{authNamespace, unrelatedAuthNamespace} {
if _, err := clientset.CoreV1().Namespaces().Create(ctx, &corev1.Namespace{Name: namespace}, metav1.CreateOptions{}); err != nil {
t.Fatal("cannot create fixture namespace")
}
@@ -151,9 +166,52 @@ func TestKubernetesSessionWithRealTokenReview(t *testing.T) {
t.Fatal("cannot create fixture service account")
}
}
// 集群外身份只有指定 SA 的 TokenRequest 权限，不使用 envtest 管理员凭据运行会话。
user, err := environment.AddUser(envtest.User{Name: "systemd-controller"}, config)
if err != nil {
t.Fatal("cannot create external controller identity")
}
kubeconfig, err := user.KubeConfig()
if err != nil {
t.Fatal("cannot build external kubeconfig")
}
externalConfig, err := clientcmd.RESTConfigFromKubeConfig(kubeconfig)
if err != nil {
t.Fatal("cannot load external kubeconfig")
}
if _, err := clientset.RbacV1().Roles(authNamespace).Create(ctx, &rbacv1.Role{
Name: tokenRequestRole,
Rules: []rbacv1.PolicyRule{{APIGroups: []string{""}, Resources: []string{"serviceaccounts/token"},
ResourceNames: []string{authRole}, Verbs: []string{"create"}}},
}, metav1.CreateOptions{}); err != nil {
t.Fatal("cannot create TokenRequest Role")
}
permission := &rbacv1.RoleBinding{
Name: tokenRequestRole,
RoleRef: rbacv1.RoleRef{APIGroup: rbacv1.GroupName, Kind: "Role", Name: tokenRequestRole},
Subjects: []rbacv1.Subject{{Kind: "User", APIGroup: rbacv1.GroupName, Name: "systemd-controller"}},
}
grant := func() {
t.Helper()
if _, err := clientset.RbacV1().RoleBindings(authNamespace).Create(ctx, permission.DeepCopy(), metav1.CreateOptions{}); err != nil {
t.Fatal("cannot grant TokenRequest permission")
}
}
grant()
externalClient, err := kubernetes.NewForConfig(externalConfig)
if err != nil {
t.Fatal("cannot construct restricted controller client")
}
for _, target := range []struct{ namespace, name string }{{unrelatedAuthNamespace, authRole}, {authNamespace, "another-account"}} {
_, err := externalClient.CoreV1().ServiceAccounts(target.namespace).CreateToken(ctx, target.name,
&authenticationv1.TokenRequest{Spec: authenticationv1.TokenRequestSpec{Audiences: []string{authAudience}}}, metav1.CreateOptions{})
if !apierrors.IsForbidden(err) {
t.Fatal("TokenRequest escaped Role namespace/resourceNames restriction")
}
}
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"},
RoleRef: rbacv1.RoleRef{APIGroup: rbacv1.GroupName, 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")
@@ -167,12 +225,22 @@ func TestKubernetesSessionWithRealTokenReview(t *testing.T) {
}
return response.Status.Token
}
return &kubernetesAuthFixture{
config: config, controllerConfig: externalConfig, admin: clientset, controller: externalClient,
grant: grant, requestToken: requestToken,
}
}
func TestKubernetesSessionWithRealTokenReview(t *testing.T) {
api := newKubernetesAuthFixture(t)
ctx := t.Context()
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)
reviewerToken := api.requestToken(authNamespace, nil)
reviewURL, reviewCA := tokenReviewEndpoint(t, api.config)
if _, err := root.Logical().WriteWithContext(ctx, "auth/kubernetes/config", map[string]any{
"kubernetes_host": reviewURL,
"kubernetes_ca_cert": reviewCA,
@@ -189,18 +257,23 @@ func TestKubernetesSessionWithRealTokenReview(t *testing.T) {
"bound_service_account_namespaces": []string{authNamespace},
"audience": authAudience,
"token_policies": []string{authRole},
"token_ttl": "3s", "token_max_ttl": "15s",
"token_ttl": "3s", "token_max_ttl": "60s",
}); 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)
manager, err := ctrl.NewManager(api.controllerConfig, ctrl.Options{Metrics: metricsserver.Options{BindAddress: "0"}, HealthProbeBindAddress: "0"})
if err != nil {
t.Fatal("cannot construct external controller manager")
}
session, err := openbao.NewKubernetesSession(client, manager.GetConfig(), "kubernetes", authRole, testIdentity)
if err != nil {
t.Fatal(err)
}
cancel := startSession(t, session)
if err := manager.Add(session); err != nil {
t.Fatal("cannot register authentication lifecycle")
}
cancel := startSession(t, manager)
waitForAuthentication(t, session.Ready)
store := fixtureStore(t, client)
if err := store.Create(ctx, credentialPath, fixtureCredential(t)); err != nil {
@@ -220,24 +293,34 @@ func TestKubernetesSessionWithRealTokenReview(t *testing.T) {
namespace string
audience string
}{
{"bao-unrelated", authAudience}, {authNamespace, "wrong-audience"},
{unrelatedAuthNamespace, 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}),
"role": authRole, "jwt": api.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)
}
}
if err := api.admin.RbacV1().RoleBindings(authNamespace).Delete(ctx, tokenRequestRole, metav1.DeleteOptions{}); err != nil {
t.Fatal("cannot revoke TokenRequest permission")
}
if err := root.Auth().Token().RevokeOrphanWithContext(ctx, client.Token()); err != nil {
t.Fatal("cannot revoke fixture OpenBao token")
}
waitForAuthentication(t, func() bool { return !session.Ready() && client.Token() == "" })
denied, err := api.controller.CoreV1().ServiceAccounts(authNamespace).CreateToken(ctx, authRole,
&authenticationv1.TokenRequest{Spec: authenticationv1.TokenRequestSpec{Audiences: []string{authAudience}}}, metav1.CreateOptions{})
if !apierrors.IsForbidden(err) || (denied != nil && denied.Status.Token != "") {
t.Fatal("revoked caller still obtained a token")
}
api.grant()
waitForAuthentication(t, session.Ready)
if client.Token() == initialToken {
t.Fatal("reauthentication reused revoked OpenBao token")
}
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
+59 -30
View File
@@ -21,12 +21,14 @@ import (
"encoding/json"
"net/http"
"net/http/httptest"
"os"
"path/filepath"
"sync/atomic"
"testing"
"time"
authenticationv1 "k8s.io/api/authentication/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/client-go/rest"
"git.ddupan.top/panxiao81/ayatori/internal/database/adapter/openbao"
)
@@ -34,27 +36,52 @@ const (
authRole = "controller"
)
func writeAuthToken(t *testing.T, path, token string) {
var testIdentity = openbao.KubernetesIdentity{Namespace: "bao-controller", ServiceAccount: authRole, Audience: "openbao"}
func authenticationConfig(address string) *rest.Config {
return &rest.Config{Host: address, ContentType: "application/json"}
}
func authenticationServer(t *testing.T, token func() (string, bool), login http.HandlerFunc) *httptest.Server {
t.Helper()
if err := os.WriteFile(path, []byte(token), 0600); err != nil {
t.Fatal("cannot write temporary projected token")
}
return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.URL.Path != "/api/v1/namespaces/bao-controller/serviceaccounts/controller/token" {
login(w, r)
return
}
var request authenticationv1.TokenRequest
if json.NewDecoder(r.Body).Decode(&request) != nil || r.Method != http.MethodPost ||
len(request.Spec.Audiences) != 1 || request.Spec.Audiences[0] != testIdentity.Audience ||
request.Spec.ExpirationSeconds == nil || *request.Spec.ExpirationSeconds != 600 {
t.Error("unexpected TokenRequest target or lifetime")
w.WriteHeader(http.StatusBadRequest)
return
}
jwt, allowed := token()
if !allowed {
w.WriteHeader(http.StatusForbidden)
return
}
w.Header().Set("Content-Type", "application/json")
if err := json.NewEncoder(w).Encode(authenticationv1.TokenRequest{
APIVersion: "authentication.k8s.io/v1", Kind: "TokenRequest",
Status: authenticationv1.TokenRequestStatus{Token: jwt, ExpirationTimestamp: metav1.NewTime(time.Now().Add(10 * time.Minute))},
}); err != nil {
t.Error("cannot encode fixture TokenRequest response")
}
}))
}
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")
}
server := authenticationServer(t, func() (string, bool) { return "", !missing }, func(w http.ResponseWriter, _ *http.Request) {
requests.Add(1)
w.WriteHeader(http.StatusForbidden)
})
defer server.Close()
client := fixtureClient(t, server.URL)
session, err := openbao.NewKubernetesSession(client, "kubernetes", authRole, path)
session, err := openbao.NewKubernetesSession(client, authenticationConfig(server.URL), "kubernetes", authRole, testIdentity)
if err != nil {
t.Fatal(err)
}
@@ -62,7 +89,7 @@ func TestKubernetesSessionDoesNotFallbackFromMissingToken(t *testing.T) {
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")
t.Fatal("denied or empty TokenRequest must not fall back to another identity")
}
}
}
@@ -80,7 +107,7 @@ func waitForAuthentication(t *testing.T, check func() bool) {
}
}
func startSession(t *testing.T, session *openbao.KubernetesSession) context.CancelFunc {
func startSession(t *testing.T, session interface{ Start(context.Context) error }) context.CancelFunc {
t.Helper()
ctx, cancel := context.WithCancel(t.Context())
done := make(chan error, 1)
@@ -96,10 +123,16 @@ func startSession(t *testing.T, session *openbao.KubernetesSession) context.Canc
return cancel
}
func TestKubernetesSessionRereadsTokenAfterFailure(t *testing.T) {
func TestKubernetesSessionRequestsNewTokenAfterFailure(t *testing.T) {
var attempts atomic.Int32
var accepted atomic.Bool
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
var rotated atomic.Bool
server := authenticationServer(t, func() (string, bool) {
if rotated.Load() {
return "rotated-test-jwt", true
}
return "expired-test-jwt", true
}, func(w http.ResponseWriter, r *http.Request) {
if r.URL.Path != "/v1/auth/kubernetes/login" {
w.WriteHeader(http.StatusNotFound)
return
@@ -121,12 +154,10 @@ func TestKubernetesSessionRereadsTokenAfterFailure(t *testing.T) {
}}); 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)
session, err := openbao.NewKubernetesSession(client, authenticationConfig(server.URL), "kubernetes", authRole, testIdentity)
if err != nil {
t.Fatal(err)
}
@@ -138,7 +169,7 @@ func TestKubernetesSessionRereadsTokenAfterFailure(t *testing.T) {
if session.Ready() || client.Token() != "" {
t.Fatal("failed login retained credentials")
}
writeAuthToken(t, path, "rotated-test-jwt")
rotated.Store(true)
waitForAuthentication(t, func() bool { return accepted.Load() && session.Ready() })
if client.Token() != fixtureToken {
t.Fatal("successful login did not configure the client")
@@ -152,19 +183,17 @@ func TestKubernetesSessionRereadsTokenAfterFailure(t *testing.T) {
func TestKubernetesSessionRejectsUnboundedLease(t *testing.T) {
var attempts atomic.Int32
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
server := authenticationServer(t, func() (string, bool) { return "test-jwt", true }, 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)
session, err := openbao.NewKubernetesSession(client, authenticationConfig(server.URL), "kubernetes", authRole, testIdentity)
if err != nil {
t.Fatal(err)
}