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:57 UTC
(skywalking-banyandb) branch fix/rbac-stream-symlink-reload created (now b0b421ed7)
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a change to branch fix/rbac-stream-symlink-reload
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
at b0b421ed7 fix(liaison): enforce RBAC on hot-enabled write streams and symlink configs
This branch includes the following new commits:
new b0b421ed7 fix(liaison): enforce RBAC on hot-enabled write streams and symlink configs
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(skywalking-banyandb) 01/01: fix(liaison): enforce RBAC on hot-enabled write streams and symlink configs
Posted by ha...@apache.org.
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")
+ }
+}