You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by ha...@apache.org on 2026/09/13 09:19:58 UTC
(skywalking-banyandb) 01/01: fix(liaison): enforce RBAC on hot-enabled write streams and symlink configs
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch fix/rbac-stream-symlink-reload
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
commit b0b421ed739e4ecc8a0dcedf55ff38cb93f09987
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Sun Sep 13 08:47:59 2026 +0000
fix(liaison): enforce RBAC on hot-enabled write streams and symlink configs
Wrap ScopeFrameGroups streams with FrameAuthorizer even while RBAC is off
so a later policy enable authorizes frames on already-open writes. Watch
symlink targets so ConfigMap-style auth file updates reload credentials.
Co-authored-by: Cursor <cu...@cursor.com>
---
CHANGES.md | 2 +
banyand/liaison/grpc/methodpolicy.go | 10 ++
banyand/liaison/grpc/rbac_stream_reload_test.go | 129 ++++++++++++++++++++++++
banyand/liaison/pkg/auth/reloader.go | 94 ++++++++++++++---
banyand/liaison/pkg/auth/reloader_test.go | 73 ++++++++++++++
5 files changed, 294 insertions(+), 14 deletions(-)
diff --git a/CHANGES.md b/CHANGES.md
index a81a972e5..aa31e32f3 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -15,6 +15,8 @@ Release Notes.
- Fix FODC proxy `/metrics` returning partial data or timing out when concurrent scrapes overlap.
- Enforce trace query time ranges independently of the sort index, skipping row timestamp checks when the query fully covers a part.
- Retry property schema registry initialization indefinitely, logging an error every 10 attempts.
+- Authorize frames on write streams that were opened before RBAC was hot-enabled, so enabling RBAC mid-stream cannot leave long-lived Measure/Stream/Trace writes unchecked.
+- Reload auth config when the watched path is a symlink and its target file changes (ConfigMap-style layouts).
### Document
diff --git a/banyand/liaison/grpc/methodpolicy.go b/banyand/liaison/grpc/methodpolicy.go
index e42397193..a41e4ee6d 100644
--- a/banyand/liaison/grpc/methodpolicy.go
+++ b/banyand/liaison/grpc/methodpolicy.go
@@ -520,6 +520,16 @@ func NewAuthorizationStreamInterceptor(
handlerContext := ContextWithSnapshot(context.WithValue(stream.Context(), principalContextKey{}, principal), snapshot)
trustedStream := &principalServerStream{ServerStream: stream, ctx: handlerContext}
if !snapshot.RBACEnabled() {
+ // Still wrap write streams so a later hot-enable of RBAC authorizes
+ // frames on already-open streams. FrameAuthorizer is a no-op while RBAC is off.
+ if policy.Scope == ScopeFrameGroups {
+ return handler(server, NewFrameAuthorizer(trustedStream, FrameAuthorization{
+ Snapshots: reloader,
+ Observer: observer,
+ Principal: principal,
+ Policy: policy,
+ }))
+ }
return handler(server, trustedStream)
}
diff --git a/banyand/liaison/grpc/rbac_stream_reload_test.go b/banyand/liaison/grpc/rbac_stream_reload_test.go
new file mode 100644
index 000000000..7ccb98806
--- /dev/null
+++ b/banyand/liaison/grpc/rbac_stream_reload_test.go
@@ -0,0 +1,129 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you 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 grpc_test
+
+import (
+ "context"
+ "os"
+ "path/filepath"
+ "testing"
+ "time"
+
+ grpclib "google.golang.org/grpc"
+ "google.golang.org/grpc/codes"
+ "google.golang.org/grpc/metadata"
+ "google.golang.org/grpc/status"
+ "google.golang.org/protobuf/proto"
+
+ commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
+ measurev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/measure/v1"
+ liaisongrpc "github.com/apache/skywalking-banyandb/banyand/liaison/grpc"
+ "github.com/apache/skywalking-banyandb/banyand/liaison/pkg/auth"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+)
+
+const streamReloadUsersOnlyYAML = `
+users:
+ - username: "alice"
+ password: "secret"
+`
+
+const streamReloadRBACEnabledYAML = `
+users:
+ - username: "alice"
+ password: "secret"
+rbac:
+ enabled: true
+ roles:
+ writer_alpha:
+ permissions: ["data:write", "data:read", "schema:read", "schema:write"]
+ bindings:
+ - principal: "alice"
+ role: "writer_alpha"
+ groups: ["alpha"]
+`
+
+type reloadScriptedStream struct {
+ grpclib.ServerStream
+ ctx context.Context
+ frames []proto.Message
+ idx int
+}
+
+func (s *reloadScriptedStream) Context() context.Context { return s.ctx }
+
+func (s *reloadScriptedStream) RecvMsg(m any) error {
+ if s.idx >= len(s.frames) {
+ return status.Error(codes.OutOfRange, "no more frames")
+ }
+ proto.Merge(m.(proto.Message), s.frames[s.idx])
+ s.idx++
+ return nil
+}
+
+// TestStreamOpenedWithoutRBACEnforcesAfterHotEnable proves that a Measure write
+// stream accepted while RBAC was off still authorizes frames after a reload turns
+// RBAC on. Without FrameAuthorizer wrapping at accept time, enabling RBAC left
+// long-lived write streams unchecked.
+func TestStreamOpenedWithoutRBACEnforcesAfterHotEnable(t *testing.T) {
+ dir := t.TempDir()
+ path := filepath.Join(dir, "auth.yaml")
+ if writeErr := os.WriteFile(path, []byte(streamReloadUsersOnlyYAML), 0o600); writeErr != nil {
+ t.Fatalf("writing users-only policy: %v", writeErr)
+ }
+
+ reloader := auth.InitAuthReloader()
+ if configErr := reloader.ConfigAuthReloader(path, false, logger.GetLogger("rbac-stream-reload-test")); configErr != nil {
+ t.Fatalf("ConfigAuthReloader: %v", configErr)
+ }
+ if startErr := reloader.Start(); startErr != nil {
+ t.Fatalf("Start: %v", startErr)
+ }
+ defer reloader.Stop()
+ if reloader.CurrentSnapshot().RBACEnabled() {
+ t.Fatal("RBACEnabled() = true, want false for users-only policy")
+ }
+
+ interceptor := liaisongrpc.NewAuthorizationStreamInterceptor(reloader, liaisongrpc.GlobalMethodPolicies(), nil)
+ info := &grpclib.StreamServerInfo{FullMethod: "/banyandb.measure.v1.MeasureService/Write"}
+ incoming := metadata.NewIncomingContext(context.Background(), metadata.Pairs("username", "alice", "password", "secret"))
+
+ base := &reloadScriptedStream{
+ ctx: incoming,
+ frames: []proto.Message{
+ &measurev1.WriteRequest{Metadata: &commonv1.Metadata{Group: "beta", Name: "m"}, MessageId: 1},
+ },
+ }
+ callErr := interceptor(nil, base, info, func(_ any, stream grpclib.ServerStream) error {
+ if writeErr := os.WriteFile(path, []byte(streamReloadRBACEnabledYAML), 0o600); writeErr != nil {
+ return writeErr
+ }
+ select {
+ case <-reloader.GetUpdateChannel():
+ case <-time.After(3 * time.Second):
+ return status.Error(codes.DeadlineExceeded, "reload did not land")
+ }
+ if !reloader.CurrentSnapshot().RBACEnabled() {
+ return status.Error(codes.FailedPrecondition, "RBAC still disabled after reload")
+ }
+ return stream.RecvMsg(&measurev1.WriteRequest{})
+ })
+ if status.Code(callErr) != codes.PermissionDenied {
+ t.Fatalf("beta write after hot-enable RBAC = %v (%s), want PermissionDenied", callErr, status.Code(callErr))
+ }
+}
diff --git a/banyand/liaison/pkg/auth/reloader.go b/banyand/liaison/pkg/auth/reloader.go
index a765d735e..cfdad5b8e 100644
--- a/banyand/liaison/pkg/auth/reloader.go
+++ b/banyand/liaison/pkg/auth/reloader.go
@@ -119,16 +119,17 @@ func (ar *Reloader) loadConfig(filePath string) error {
// Reloader manages dynamic reloading of auth config.
type Reloader struct {
- policyObserver PolicyObserver
- debounceTimer *time.Timer
- updateCh chan struct{}
- Config *Config
- watcher *fsnotify.Watcher
- log *logger.Logger
- snapshot atomic.Pointer[compiledSnapshot]
- configFile string
- lastConfigHash []byte
- mu sync.RWMutex
+ policyObserver PolicyObserver
+ debounceTimer *time.Timer
+ updateCh chan struct{}
+ Config *Config
+ watcher *fsnotify.Watcher
+ log *logger.Logger
+ snapshot atomic.Pointer[compiledSnapshot]
+ configFile string
+ resolvedConfigFile string
+ lastConfigHash []byte
+ mu sync.RWMutex
}
// InitAuthReloader returns Reloader with default values.
@@ -174,21 +175,82 @@ func (ar *Reloader) ConfigAuthReloader(configFile string, healthAuthEnabled bool
ar.updateCh = make(chan struct{}, 1)
ar.lastConfigHash = lastConfigHash
ar.mu.Unlock()
+ ar.refreshResolvedConfigPath()
return nil
}
// Start begins monitoring the config file.
func (ar *Reloader) Start() error {
- configDir := filepath.Dir(ar.configFile)
- if watchErr := ar.watcher.Add(configDir); watchErr != nil {
- return fmt.Errorf("failed to watch auth config directory %s: %w", configDir, watchErr)
+ if watchErr := ar.syncWatches(); watchErr != nil {
+ return watchErr
}
go ar.watchFiles()
return nil
}
+// refreshResolvedConfigPath records the symlink target when configFile is a symlink
+// so writes to the target (common ConfigMap/..data layouts) also trigger reload.
+func (ar *Reloader) refreshResolvedConfigPath() {
+ if ar == nil {
+ return
+ }
+ ar.mu.RLock()
+ configFile := ar.configFile
+ ar.mu.RUnlock()
+ if configFile == "" {
+ return
+ }
+ resolved := ""
+ if evaluated, evalErr := filepath.EvalSymlinks(configFile); evalErr == nil {
+ resolved = filepath.Clean(evaluated)
+ if resolved == filepath.Clean(configFile) {
+ resolved = ""
+ }
+ }
+ ar.mu.Lock()
+ ar.resolvedConfigFile = resolved
+ ar.mu.Unlock()
+}
+
+// syncWatches ensures the directories of the config path and any resolved symlink
+// target are watched.
+func (ar *Reloader) syncWatches() error {
+ ar.mu.RLock()
+ watcher := ar.watcher
+ configFile := ar.configFile
+ resolved := ar.resolvedConfigFile
+ ar.mu.RUnlock()
+ if watcher == nil {
+ return errors.New("watcher is nil")
+ }
+ dirs := map[string]struct{}{
+ filepath.Dir(configFile): {},
+ }
+ if resolved != "" {
+ dirs[filepath.Dir(resolved)] = struct{}{}
+ }
+ for dir := range dirs {
+ if watchErr := watcher.Add(dir); watchErr != nil {
+ return fmt.Errorf("failed to watch auth config directory %s: %w", dir, watchErr)
+ }
+ }
+ return nil
+}
+
+// isConfigPathEvent reports whether an fsnotify event path refers to the configured
+// auth file or the file it currently resolves to through symlinks.
+func (ar *Reloader) isConfigPathEvent(eventName string) bool {
+ cleaned := filepath.Clean(eventName)
+ ar.mu.RLock()
+ defer ar.mu.RUnlock()
+ if cleaned == filepath.Clean(ar.configFile) {
+ return true
+ }
+ return ar.resolvedConfigFile != "" && cleaned == ar.resolvedConfigFile
+}
+
// SetPolicyObserver installs reload observability and reports the current accepted policy.
func (ar *Reloader) SetPolicyObserver(observer PolicyObserver) {
if ar == nil {
@@ -245,7 +307,7 @@ func (ar *Reloader) watchFiles() {
if !ok {
return
}
- if filepath.Clean(event.Name) != filepath.Clean(ar.configFile) {
+ if !ar.isConfigPathEvent(event.Name) {
continue
}
ar.log.Debug().Str("file", event.Name).Str("op", event.Op.String()).Msg("Detected auth file event")
@@ -291,6 +353,10 @@ func (ar *Reloader) tryReload() {
ar.mu.Lock()
ar.lastConfigHash = newHash
ar.mu.Unlock()
+ ar.refreshResolvedConfigPath()
+ if watchErr := ar.syncWatches(); watchErr != nil {
+ ar.log.Error().Err(watchErr).Msg("failed to refresh auth config watches after reload")
+ }
// notify
select {
diff --git a/banyand/liaison/pkg/auth/reloader_test.go b/banyand/liaison/pkg/auth/reloader_test.go
index 43d19cf9f..58e9df8e5 100644
--- a/banyand/liaison/pkg/auth/reloader_test.go
+++ b/banyand/liaison/pkg/auth/reloader_test.go
@@ -183,3 +183,76 @@ users:
t.Error("expected bob/hunter2 to be valid after atomic replacement")
}
}
+
+func TestReloaderUpdatesWhenSymlinkTargetChanges(t *testing.T) {
+ dir := t.TempDir()
+ targetPath := filepath.Join(dir, "target.yaml")
+ linkPath := filepath.Join(dir, "auth.yaml")
+ initialYAML := `
+users:
+ - username: "alice"
+ password: "old"
+rbac:
+ enabled: true
+ bindings:
+ - principal: "alice"
+ role: "writer"
+ groups: ["payments"]
+`
+ if writeErr := os.WriteFile(targetPath, []byte(initialYAML), 0o600); writeErr != nil {
+ t.Fatalf("writing symlink target: %v", writeErr)
+ }
+ if linkErr := os.Symlink(targetPath, linkPath); linkErr != nil {
+ t.Fatalf("creating auth symlink: %v", linkErr)
+ }
+
+ reloader := InitAuthReloader()
+ log := logger.GetLogger("auth-symlink-target-test")
+ if configureErr := reloader.ConfigAuthReloader(linkPath, false, log); configureErr != nil {
+ t.Fatalf("ConfigAuthReloader failed: %v", configureErr)
+ }
+ if startErr := reloader.Start(); startErr != nil {
+ t.Fatalf("Start failed: %v", startErr)
+ }
+ defer reloader.Stop()
+
+ initialRevision := reloader.CurrentSnapshot().Revision()
+ if !reloader.CheckUsernameAndPassword("alice", "old") {
+ t.Fatal("expected alice/old to authenticate before target update")
+ }
+
+ updatedYAML := `
+users:
+ - username: "alice"
+ password: "new"
+rbac:
+ enabled: true
+ bindings: []
+`
+ if writeErr := os.WriteFile(targetPath, []byte(updatedYAML), 0o600); writeErr != nil {
+ t.Fatalf("updating symlink target: %v", writeErr)
+ }
+
+ select {
+ case <-reloader.GetUpdateChannel():
+ case <-time.After(3 * time.Second):
+ t.Fatal("timed out waiting for reload after symlink target write")
+ }
+
+ if reloader.CurrentSnapshot().Revision() <= initialRevision {
+ t.Fatalf("revision = %d after target write, want greater than %d", reloader.CurrentSnapshot().Revision(), initialRevision)
+ }
+ if reloader.CheckUsernameAndPassword("alice", "old") {
+ t.Error("alice/old should be rejected after target password rotation")
+ }
+ if !reloader.CheckUsernameAndPassword("alice", "new") {
+ t.Error("expected alice/new to authenticate after target update")
+ }
+ principal, ok := reloader.CurrentSnapshot().Authenticate("alice", "new")
+ if !ok {
+ t.Fatal("Authenticate(alice, new) failed after target update")
+ }
+ if reloader.CurrentSnapshot().Allows(principal, PermissionDataWrite, "payments") {
+ t.Error("writer grant on payments should be cleared after bindings removed")
+ }
+}