authz: Add onPolicyUpdate callback to file watcher.

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

Author: hnefatl

authz/grpc_authz_end2end_test.go

@@ -95,7 +95,7 @@ var authzTests = map[string]struct {
 				]
 			}`,
 		md:         metadata.Pairs("key-abc", "val-abc"),
-		wantStatus: status.New(codes.PermissionDenied, "unauthorized RPC request rejected"),
+		wantStatus: status.New(codes.PermissionDenied, "authz: unauthorized RPC request rejected"),
 	},
 	"DeniesRPCMatchInDenyAndAllow": {
 		authzPolicy: `{
@@ -125,7 +125,7 @@ var authzTests = map[string]struct {
 					}
 				]
 			}`,
-		wantStatus: status.New(codes.PermissionDenied, "unauthorized RPC request rejected"),
+		wantStatus: status.New(codes.PermissionDenied, "authz: unauthorized RPC request rejected"),
 	},
 	"AllowsRPCNoMatchInDenyMatchInAllow": {
 		authzPolicy: `{
@@ -192,7 +192,7 @@ var authzTests = map[string]struct {
 					}
 				]
 			}`,
-		wantStatus: status.New(codes.PermissionDenied, "unauthorized RPC request rejected"),
+		wantStatus: status.New(codes.PermissionDenied, "authz: unauthorized RPC request rejected"),
 	},
 	"AllowsRPCEmptyDenyMatchInAllow": {
 		authzPolicy: `{
@@ -240,7 +240,7 @@ var authzTests = map[string]struct {
 					}
 				]
 			}`,
-		wantStatus: status.New(codes.PermissionDenied, "unauthorized RPC request rejected"),
+		wantStatus: status.New(codes.PermissionDenied, "authz: unauthorized RPC request rejected"),
 	},
 	"DeniesRPCRequestWithPrincipalsFieldOnUnauthenticatedConnection": {
 		authzPolicy: `{
@@ -255,7 +255,7 @@ var authzTests = map[string]struct {
 					}
 				]
 			}`,
-		wantStatus: status.New(codes.PermissionDenied, "unauthorized RPC request rejected"),
+		wantStatus: status.New(codes.PermissionDenied, "authz: unauthorized RPC request rejected"),
 	},
 	"DeniesRPCRequestNoMatchInAllowFailsPresenceMatch": {
 		authzPolicy: `{
@@ -284,7 +284,7 @@ var authzTests = map[string]struct {
 				]
 			}`,
 		md:         metadata.Pairs("key-abc", ""),
-		wantStatus: status.New(codes.PermissionDenied, "unauthorized RPC request rejected"),
+		wantStatus: status.New(codes.PermissionDenied, "authz: unauthorized RPC request rejected"),
 	},
 }
 

authz/grpc_authz_server_interceptors.go

@@ -65,7 +65,7 @@ func (i *StaticInterceptor) UnaryInterceptor(ctx context.Context, req any, _ *gr
 			if logger.V(2) {
 				logger.Infof("unauthorized RPC request rejected: %v", err)
 			}
-			return nil, status.Errorf(codes.PermissionDenied, "unauthorized RPC request rejected")
+			return nil, status.Errorf(codes.PermissionDenied, "authz: unauthorized RPC request rejected")
 		}
 		return nil, err
 	}
@@ -82,7 +82,7 @@ func (i *StaticInterceptor) StreamInterceptor(srv any, ss grpc.ServerStream, _ *
 			if logger.V(2) {
 				logger.Infof("unauthorized RPC request rejected: %v", err)
 			}
-			return status.Errorf(codes.PermissionDenied, "unauthorized RPC request rejected")
+			return status.Errorf(codes.PermissionDenied, "authz: unauthorized RPC request rejected")
 		}
 		return err
 	}
@@ -92,24 +92,54 @@ 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.
+//
+// # Experimental
+//
+// Notice: This API is EXPERIMENTAL and may be changed or removed in a
+// later release.
+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.
+	//
+	// The callback is executed synchronously, so should complete quickly or
+	// risk blocking future updates.
+	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 nil, fmt.Errorf("authorization policy file path is empty")
+	return NewFileWatcherWithOptions(FileWatcherOptions{PolicyFile: file, RefreshDuration: duration, OnPolicyUpdate: nil})
+}
+
+// NewFileWatcherWithOptions returns a new FileWatcherInterceptor from a set of
+// options.
+//
+// # Experimental
+//
+// Notice: This API is EXPERIMENTAL and may be changed or removed in a
+// later release.
+func NewFileWatcherWithOptions(options FileWatcherOptions) (*FileWatcherInterceptor, error) {
+	if options.PolicyFile == "" {
+		return nil, fmt.Errorf("authz: 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("authz: 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 +151,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 +170,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 +185,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()
 
@@ -85,7 +88,7 @@ func (s) TestNewFileWatcher(t *testing.T) {
 	}{
 		"InvalidRefreshDurationFailsToCreateInterceptor": {
 			refreshDuration: time.Duration(0),
-			wantErr:         fmt.Errorf("requires refresh interval(0s) greater than 0s"),
+			wantErr:         fmt.Errorf("authz: requires refresh interval(0s) greater than 0s"),
 		},
 		"InvalidPolicyFailsToCreateInterceptor": {
 			authzPolicy:     `{}`,
@@ -118,3 +121,54 @@ func (s) TestNewFileWatcher(t *testing.T) {
 		})
 	}
 }
+
+func (s) TestOnPolicyUpdate(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+
+	updates := make(chan string, 1)
+	onPolicyUpdate := func(s string) {
+		updates <- s
+	}
+
+	content := `{"name": "foo1", "allow_rules":[{"name":"bar"}]}`
+	file := createTmpPolicyFile(t, "onpolicyupdate", []byte(content))
+	opts := authz.FileWatcherOptions{PolicyFile: file, RefreshDuration: 50 * time.Millisecond, OnPolicyUpdate: onPolicyUpdate}
+	i, err := authz.NewFileWatcherWithOptions(opts)
+	if err != nil {
+		t.Fatalf("NewFileWatcherWithOptions(%v) returned err: %v", opts, 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)
+		}
+	}
+
+	sCtx, sCancel := context.WithTimeout(ctx, 100*time.Millisecond)
+	defer sCancel()
+	select {
+	case <-updates:
+		t.Fatal("OnPolicyUpdate callback invoked more times than expected")
+	case <-sCtx.Done():
+	}
+}