authz: Add onPolicyUpdate callback to file watcher.

Signed-off-by: Keith Collister <kcollister@google.com>

Author: hnefatl

authz/grpc_authz_server_interceptors.go

@@ -92,24 +92,41 @@ func (i *StaticInterceptor) StreamInterceptor(srv any, ss grpc.ServerStream, _ *
 // FileWatcherInterceptor contains details used to make authorization decisions
 // by watching a file path that contains authorization policy in JSON format.
 type FileWatcherInterceptor struct {
+	options             FileWatcherOptions
 	internalInterceptor unsafe.Pointer // *StaticInterceptor
-	policyFile          string
 	policyContents      []byte
-	refreshDuration     time.Duration
 	cancel              context.CancelFunc
 }
 
+// FileWatcherOptions contains configuration options for the
+// FileWatcherInterceptor.
+type FileWatcherOptions struct {
+	// PolicyFile contains a JSON string of the authorization policy.
+	PolicyFile string
+	// RefreshDuration is the delay between policy refreshes.
+	RefreshDuration time.Duration
+	// OnPolicyUpdate is a callback to be invoked when a policy is
+	// loaded/updated. The loaded policy string is passed as an argument.
+	OnPolicyUpdate func(string)
+}
+
 // NewFileWatcher returns a new FileWatcherInterceptor from a policy file
 // that contains JSON string of authorization policy and a refresh duration to
 // specify the amount of time between policy refreshes.
 func NewFileWatcher(file string, duration time.Duration) (*FileWatcherInterceptor, error) {
-	if file == "" {
+	return NewFileWatcherWithOptions(FileWatcherOptions{PolicyFile: file, RefreshDuration: duration, OnPolicyUpdate: nil})
+}
+
+// NewFileWatcherWithOptions returns a new FileWatcherInterceptor from a set of
+// options.
+func NewFileWatcherWithOptions(options FileWatcherOptions) (*FileWatcherInterceptor, error) {
+	if options.PolicyFile == "" {
 		return nil, fmt.Errorf("authorization policy file path is empty")
 	}
-	if duration <= time.Duration(0) {
-		return nil, fmt.Errorf("requires refresh interval(%v) greater than 0s", duration)
+	if options.RefreshDuration <= time.Duration(0) {
+		return nil, fmt.Errorf("requires refresh interval(%v) greater than 0s", options.RefreshDuration)
 	}
-	i := &FileWatcherInterceptor{policyFile: file, refreshDuration: duration}
+	i := &FileWatcherInterceptor{options: options}
 	if err := i.updateInternalInterceptor(); err != nil {
 		return nil, err
 	}
@@ -121,7 +138,7 @@ func NewFileWatcher(file string, duration time.Duration) (*FileWatcherIntercepto
 }
 
 func (i *FileWatcherInterceptor) run(ctx context.Context) {
-	ticker := time.NewTicker(i.refreshDuration)
+	ticker := time.NewTicker(i.options.RefreshDuration)
 	for {
 		if err := i.updateInternalInterceptor(); err != nil {
 			logger.Warningf("authorization policy reload status err: %v", err)
@@ -140,9 +157,9 @@ func (i *FileWatcherInterceptor) run(ctx context.Context) {
 // constructor, if there is an error in reading the file or parsing the policy, the
 // previous internalInterceptors will not be replaced.
 func (i *FileWatcherInterceptor) updateInternalInterceptor() error {
-	policyContents, err := os.ReadFile(i.policyFile)
+	policyContents, err := os.ReadFile(i.options.PolicyFile)
 	if err != nil {
-		return fmt.Errorf("policyFile(%s) read failed: %v", i.policyFile, err)
+		return fmt.Errorf("policyFile(%s) read failed: %v", i.options.PolicyFile, err)
 	}
 	if bytes.Equal(i.policyContents, policyContents) {
 		return nil
@@ -155,6 +172,9 @@ func (i *FileWatcherInterceptor) updateInternalInterceptor() error {
 	}
 	atomic.StorePointer(&i.internalInterceptor, unsafe.Pointer(interceptor))
 	logger.Infof("authorization policy reload status: successfully loaded new policy %v", policyContentsString)
+	if i.options.OnPolicyUpdate != nil {
+		i.options.OnPolicyUpdate(policyContentsString)
+	}
 	return nil
 }
 

authz/grpc_authz_server_interceptors_test.go

@@ -19,6 +19,7 @@
 package authz_test
 
 import (
+	"context"
 	"fmt"
 	"os"
 	"path"
@@ -28,6 +29,8 @@ import (
 	"google.golang.org/grpc/authz"
 )
 
+const defaultTestTimeout = 10 * time.Second
+
 func createTmpPolicyFile(t *testing.T, dirSuffix string, policy []byte) string {
 	t.Helper()
 
@@ -118,3 +121,50 @@ func (s) TestNewFileWatcher(t *testing.T) {
 		})
 	}
 }
+
+func (s) TestOnPolicyUpdate(t *testing.T) {
+	ctx, cancel := context.WithTimeout(t.Context(), defaultTestTimeout)
+	defer cancel()
+
+	updates := make(chan string, 10)
+	onPolicyUpdate := func(s string) {
+		updates <- s
+	}
+
+	content := `{"name": "foo1", "allow_rules":[{"name":"bar"}]}`
+	file := createTmpPolicyFile(t, "onpolicyupdate", []byte(content))
+	i, err := authz.NewFileWatcherWithOptions(authz.FileWatcherOptions{PolicyFile: file, RefreshDuration: time.Millisecond, OnPolicyUpdate: onPolicyUpdate})
+	if err != nil {
+		t.Fatalf("NewFileWatcherWithCallback() returned err: %v", err)
+	}
+	defer i.Close()
+
+	select {
+	case <-ctx.Done():
+		t.Fatalf("timeout waiting for policy update")
+	case update := <-updates:
+		if update != content {
+			t.Fatalf("unexpected contents on first load of policy file: got=%v, want=%v", update, content)
+		}
+	}
+
+	// Tweak the file, expect an update.
+	content = `{"name": "foo2", "allow_rules":[{"name":"bar"}]}`
+	if err := os.WriteFile(file, []byte(content), os.ModePerm); err != nil {
+		t.Fatalf("os.WriteFile(%q) failed: %v", file, err)
+	}
+
+	select {
+	case <-ctx.Done():
+		t.Fatalf("timeout waiting for policy update")
+	case update := <-updates:
+		if update != content {
+			t.Fatalf("unexpected contents after policy file changed: got=%v, want=%v", update, content)
+		}
+	}
+
+	close(updates)
+	if len(updates) != 0 {
+		t.Fatalf("expected exactly 2 updates in channel")
+	}
+}