grpc_authz_end2end_test.go

Documentation: google.golang.org/grpc/authz

     1  /*
     2   *
     3   * Copyright 2021 gRPC authors.
     4   *
     5   * Licensed under the Apache License, Version 2.0 (the "License");
     6   * you may not use this file except in compliance with the License.
     7   * You may obtain a copy of the License at
     8   *
     9   *     http://www.apache.org/licenses/LICENSE-2.0
    10   *
    11   * Unless required by applicable law or agreed to in writing, software
    12   * distributed under the License is distributed on an "AS IS" BASIS,
    13   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    14   * See the License for the specific language governing permissions and
    15   * limitations under the License.
    16   *
    17   */
    18  
    19  package authz_test
    20  
    21  import (
    22  	"context"
    23  	"crypto/tls"
    24  	"crypto/x509"
    25  	"io"
    26  	"net"
    27  	"os"
    28  	"testing"
    29  	"time"
    30  
    31  	"google.golang.org/grpc"
    32  	"google.golang.org/grpc/authz"
    33  	"google.golang.org/grpc/codes"
    34  	"google.golang.org/grpc/credentials"
    35  	"google.golang.org/grpc/credentials/insecure"
    36  	"google.golang.org/grpc/internal/grpctest"
    37  	"google.golang.org/grpc/metadata"
    38  	"google.golang.org/grpc/status"
    39  	"google.golang.org/grpc/testdata"
    40  
    41  	testgrpc "google.golang.org/grpc/interop/grpc_testing"
    42  	testpb "google.golang.org/grpc/interop/grpc_testing"
    43  )
    44  
    45  type testServer struct {
    46  	testgrpc.UnimplementedTestServiceServer
    47  }
    48  
    49  func (s *testServer) UnaryCall(ctx context.Context, req *testpb.SimpleRequest) (*testpb.SimpleResponse, error) {
    50  	return &testpb.SimpleResponse{}, nil
    51  }
    52  
    53  func (s *testServer) StreamingInputCall(stream testgrpc.TestService_StreamingInputCallServer) error {
    54  	for {
    55  		_, err := stream.Recv()
    56  		if err == io.EOF {
    57  			return stream.SendAndClose(&testpb.StreamingInputCallResponse{})
    58  		}
    59  		if err != nil {
    60  			return err
    61  		}
    62  	}
    63  }
    64  
    65  type s struct {
    66  	grpctest.Tester
    67  }
    68  
    69  func Test(t *testing.T) {
    70  	grpctest.RunSubTests(t, s{})
    71  }
    72  
    73  var authzTests = map[string]struct {
    74  	authzPolicy string
    75  	md          metadata.MD
    76  	wantStatus  *status.Status
    77  }{
    78  	"DeniesRPCMatchInDenyNoMatchInAllow": {
    79  		authzPolicy: `{
    80  				"name": "authz",
    81  				"allow_rules": 
    82  				[
    83  					{
    84  						"name": "allow_StreamingOutputCall",
    85  						"request": {
    86  							"paths":
    87  							[
    88  								"/grpc.testing.TestService/StreamingOutputCall"
    89  							]
    90  						}
    91  					}
    92  				],
    93  				"deny_rules":
    94  				[
    95  					{
    96  						"name": "deny_TestServiceCalls",
    97  						"request": {
    98  							"paths":
    99  							[
   100  								"/grpc.testing.TestService/*"
   101  							],
   102  							"headers":
   103  							[
   104  								{
   105  									"key": "key-abc",
   106  									"values":
   107  									[
   108  										"val-abc",
   109  										"val-def"
   110  									]
   111  								}
   112  							]
   113  						}
   114  					}
   115  				]
   116  			}`,
   117  		md:         metadata.Pairs("key-abc", "val-abc"),
   118  		wantStatus: status.New(codes.PermissionDenied, "unauthorized RPC request rejected"),
   119  	},
   120  	"DeniesRPCMatchInDenyAndAllow": {
   121  		authzPolicy: `{
   122  				"name": "authz",
   123  				"allow_rules":
   124  				[
   125  					{
   126  						"name": "allow_all",
   127  						"request": {
   128  							"paths":
   129  							[
   130  								"*"
   131  							]
   132  						}
   133  					}
   134  				],
   135  				"deny_rules":
   136  				[
   137  					{
   138  						"name": "deny_all",
   139  						"request": {
   140  							"paths":
   141  							[
   142  								"*"
   143  							]
   144  						}
   145  					}
   146  				]
   147  			}`,
   148  		wantStatus: status.New(codes.PermissionDenied, "unauthorized RPC request rejected"),
   149  	},
   150  	"AllowsRPCNoMatchInDenyMatchInAllow": {
   151  		authzPolicy: `{
   152  				"name": "authz",
   153  				"allow_rules":
   154  				[
   155  					{
   156  						"name": "allow_all"
   157  					}
   158  				],
   159  				"deny_rules":
   160  				[
   161  					{
   162  						"name": "deny_TestServiceCalls",
   163  						"request": {
   164  							"paths":
   165  							[
   166  								"/grpc.testing.TestService/UnaryCall",
   167  								"/grpc.testing.TestService/StreamingInputCall"
   168  							],
   169  							"headers": 
   170  							[
   171  								{
   172  									"key": "key-abc",
   173  									"values": 
   174  									[
   175  										"val-abc",
   176  										"val-def"
   177  									]
   178  								}
   179  							]
   180  						}
   181  					}
   182  				]
   183  			}`,
   184  		md:         metadata.Pairs("key-xyz", "val-xyz"),
   185  		wantStatus: status.New(codes.OK, ""),
   186  	},
   187  	"DeniesRPCNoMatchInDenyAndAllow": {
   188  		authzPolicy: `{
   189  				"name": "authz",
   190  				"allow_rules":
   191  				[
   192  					{
   193  						"name": "allow_some_user",
   194  						"source": {
   195  							"principals":
   196  							[
   197  								"some_user"
   198  							]
   199  						}
   200  					}
   201  				],
   202  				"deny_rules":
   203  				[
   204  					{
   205  						"name": "deny_StreamingOutputCall",
   206  						"request": {
   207  							"paths":
   208  							[
   209  								"/grpc.testing.TestService/StreamingOutputCall"
   210  							]
   211  						}
   212  					}
   213  				]
   214  			}`,
   215  		wantStatus: status.New(codes.PermissionDenied, "unauthorized RPC request rejected"),
   216  	},
   217  	"AllowsRPCEmptyDenyMatchInAllow": {
   218  		authzPolicy: `{
   219  				"name": "authz",
   220  				"allow_rules":
   221  				[
   222  					{
   223  						"name": "allow_UnaryCall",
   224  						"request":
   225  						{
   226  							"paths":
   227  							[
   228  								"/grpc.testing.TestService/UnaryCall"
   229  							]
   230  						}
   231  					},
   232  					{
   233  						"name": "allow_StreamingInputCall",
   234  						"request":
   235  						{
   236  							"paths":
   237  							[
   238  								"/grpc.testing.TestService/StreamingInputCall"
   239  							]
   240  						}
   241  					}
   242  				]
   243  			}`,
   244  		wantStatus: status.New(codes.OK, ""),
   245  	},
   246  	"DeniesRPCEmptyDenyNoMatchInAllow": {
   247  		authzPolicy: `{
   248  				"name": "authz",
   249  				"allow_rules":
   250  				[
   251  					{
   252  						"name": "allow_StreamingOutputCall",
   253  						"request": 
   254  						{
   255  							"paths":
   256  							[
   257  								"/grpc.testing.TestService/StreamingOutputCall"
   258  							]
   259  						}
   260  					}
   261  				]
   262  			}`,
   263  		wantStatus: status.New(codes.PermissionDenied, "unauthorized RPC request rejected"),
   264  	},
   265  	"DeniesRPCRequestWithPrincipalsFieldOnUnauthenticatedConnection": {
   266  		authzPolicy: `{
   267  				"name": "authz",
   268  				"allow_rules":
   269  				[
   270  					{
   271  						"name": "allow_authenticated",
   272  						"source": {
   273  							"principals": ["*", ""]
   274  						}
   275  					}
   276  				]
   277  			}`,
   278  		wantStatus: status.New(codes.PermissionDenied, "unauthorized RPC request rejected"),
   279  	},
   280  	"DeniesRPCRequestNoMatchInAllowFailsPresenceMatch": {
   281  		authzPolicy: `{
   282  				"name": "authz",
   283  				"allow_rules":
   284  				[
   285  					{
   286  						"name": "allow_TestServiceCalls",
   287  						"request": {
   288  							"paths":
   289  							[
   290  								"/grpc.testing.TestService/*"
   291  							],
   292  							"headers":
   293  							[
   294  								{
   295  									"key": "key-abc",
   296  									"values":
   297  									[
   298  										"*"
   299  									]
   300  								}
   301  							]
   302  						}
   303  					}
   304  				]
   305  			}`,
   306  		md:         metadata.Pairs("key-abc", ""),
   307  		wantStatus: status.New(codes.PermissionDenied, "unauthorized RPC request rejected"),
   308  	},
   309  }
   310  
   311  func (s) TestStaticPolicyEnd2End(t *testing.T) {
   312  	for name, test := range authzTests {
   313  		t.Run(name, func(t *testing.T) {
   314  			// Start a gRPC server with gRPC authz unary and stream server interceptors.
   315  			i, _ := authz.NewStatic(test.authzPolicy)
   316  			s := grpc.NewServer(
   317  				grpc.ChainUnaryInterceptor(i.UnaryInterceptor),
   318  				grpc.ChainStreamInterceptor(i.StreamInterceptor))
   319  			defer s.Stop()
   320  			testgrpc.RegisterTestServiceServer(s, &testServer{})
   321  
   322  			lis, err := net.Listen("tcp", "localhost:0")
   323  			if err != nil {
   324  				t.Fatalf("error listening: %v", err)
   325  			}
   326  			go s.Serve(lis)
   327  
   328  			// Establish a connection to the server.
   329  			clientConn, err := grpc.NewClient(lis.Addr().String(), grpc.WithTransportCredentials(insecure.NewCredentials()))
   330  			if err != nil {
   331  				t.Fatalf("grpc.NewClient(%v) failed: %v", lis.Addr().String(), err)
   332  			}
   333  			defer clientConn.Close()
   334  			client := testgrpc.NewTestServiceClient(clientConn)
   335  
   336  			ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
   337  			defer cancel()
   338  			ctx = metadata.NewOutgoingContext(ctx, test.md)
   339  
   340  			// Verifying authorization decision for Unary RPC.
   341  			_, err = client.UnaryCall(ctx, &testpb.SimpleRequest{})
   342  			if got := status.Convert(err); got.Code() != test.wantStatus.Code() || got.Message() != test.wantStatus.Message() {
   343  				t.Fatalf("[UnaryCall] error want:{%v} got:{%v}", test.wantStatus.Err(), got.Err())
   344  			}
   345  
   346  			// Verifying authorization decision for Streaming RPC.
   347  			stream, err := client.StreamingInputCall(ctx)
   348  			if err != nil {
   349  				t.Fatalf("failed StreamingInputCall err: %v", err)
   350  			}
   351  			req := &testpb.StreamingInputCallRequest{
   352  				Payload: &testpb.Payload{
   353  					Body: []byte("hi"),
   354  				},
   355  			}
   356  			if err := stream.Send(req); err != nil && err != io.EOF {
   357  				t.Fatalf("failed stream.Send err: %v", err)
   358  			}
   359  			_, err = stream.CloseAndRecv()
   360  			if got := status.Convert(err); got.Code() != test.wantStatus.Code() || got.Message() != test.wantStatus.Message() {
   361  				t.Fatalf("[StreamingCall] error want:{%v} got:{%v}", test.wantStatus.Err(), got.Err())
   362  			}
   363  		})
   364  	}
   365  }
   366  
   367  func (s) TestAllowsRPCRequestWithPrincipalsFieldOnTLSAuthenticatedConnection(t *testing.T) {
   368  	authzPolicy := `{
   369  				"name": "authz",
   370  				"allow_rules":
   371  				[
   372  					{
   373  						"name": "allow_authenticated",
   374  						"source": {
   375  							"principals": ["*", ""]
   376  						}
   377  					}
   378  				]
   379  			}`
   380  	// Start a gRPC server with gRPC authz unary server interceptor.
   381  	i, _ := authz.NewStatic(authzPolicy)
   382  	creds, err := credentials.NewServerTLSFromFile(testdata.Path("x509/server1_cert.pem"), testdata.Path("x509/server1_key.pem"))
   383  	if err != nil {
   384  		t.Fatalf("failed to generate credentials: %v", err)
   385  	}
   386  	s := grpc.NewServer(
   387  		grpc.Creds(creds),
   388  		grpc.ChainUnaryInterceptor(i.UnaryInterceptor))
   389  	defer s.Stop()
   390  	testgrpc.RegisterTestServiceServer(s, &testServer{})
   391  
   392  	lis, err := net.Listen("tcp", "localhost:0")
   393  	if err != nil {
   394  		t.Fatalf("error listening: %v", err)
   395  	}
   396  	go s.Serve(lis)
   397  
   398  	// Establish a connection to the server.
   399  	creds, err = credentials.NewClientTLSFromFile(testdata.Path("x509/server_ca_cert.pem"), "x.test.example.com")
   400  	if err != nil {
   401  		t.Fatalf("failed to load credentials: %v", err)
   402  	}
   403  	clientConn, err := grpc.NewClient(lis.Addr().String(), grpc.WithTransportCredentials(creds))
   404  	if err != nil {
   405  		t.Fatalf("grpc.NewClient(%v) failed: %v", lis.Addr().String(), err)
   406  	}
   407  	defer clientConn.Close()
   408  	client := testgrpc.NewTestServiceClient(clientConn)
   409  
   410  	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
   411  	defer cancel()
   412  
   413  	// Verifying authorization decision.
   414  	if _, err = client.UnaryCall(ctx, &testpb.SimpleRequest{}); err != nil {
   415  		t.Fatalf("client.UnaryCall(_, _) = %v; want nil", err)
   416  	}
   417  }
   418  
   419  func (s) TestAllowsRPCRequestWithPrincipalsFieldOnMTLSAuthenticatedConnection(t *testing.T) {
   420  	authzPolicy := `{
   421  				"name": "authz",
   422  				"allow_rules":
   423  				[
   424  					{
   425  						"name": "allow_authenticated",
   426  						"source": {
   427  							"principals": ["*", ""]
   428  						}
   429  					}
   430  				]
   431  			}`
   432  	// Start a gRPC server with gRPC authz unary server interceptor.
   433  	i, _ := authz.NewStatic(authzPolicy)
   434  	cert, err := tls.LoadX509KeyPair(testdata.Path("x509/server1_cert.pem"), testdata.Path("x509/server1_key.pem"))
   435  	if err != nil {
   436  		t.Fatalf("tls.LoadX509KeyPair(x509/server1_cert.pem, x509/server1_key.pem) failed: %v", err)
   437  	}
   438  	ca, err := os.ReadFile(testdata.Path("x509/client_ca_cert.pem"))
   439  	if err != nil {
   440  		t.Fatalf("os.ReadFile(x509/client_ca_cert.pem) failed: %v", err)
   441  	}
   442  	certPool := x509.NewCertPool()
   443  	if !certPool.AppendCertsFromPEM(ca) {
   444  		t.Fatal("failed to append certificates")
   445  	}
   446  	creds := credentials.NewTLS(&tls.Config{
   447  		ClientAuth:   tls.RequireAndVerifyClientCert,
   448  		Certificates: []tls.Certificate{cert},
   449  		ClientCAs:    certPool,
   450  	})
   451  	s := grpc.NewServer(
   452  		grpc.Creds(creds),
   453  		grpc.ChainUnaryInterceptor(i.UnaryInterceptor))
   454  	defer s.Stop()
   455  	testgrpc.RegisterTestServiceServer(s, &testServer{})
   456  
   457  	lis, err := net.Listen("tcp", "localhost:0")
   458  	if err != nil {
   459  		t.Fatalf("error listening: %v", err)
   460  	}
   461  	go s.Serve(lis)
   462  
   463  	// Establish a connection to the server.
   464  	cert, err = tls.LoadX509KeyPair(testdata.Path("x509/client1_cert.pem"), testdata.Path("x509/client1_key.pem"))
   465  	if err != nil {
   466  		t.Fatalf("tls.LoadX509KeyPair(x509/client1_cert.pem, x509/client1_key.pem) failed: %v", err)
   467  	}
   468  	ca, err = os.ReadFile(testdata.Path("x509/server_ca_cert.pem"))
   469  	if err != nil {
   470  		t.Fatalf("os.ReadFile(x509/server_ca_cert.pem) failed: %v", err)
   471  	}
   472  	roots := x509.NewCertPool()
   473  	if !roots.AppendCertsFromPEM(ca) {
   474  		t.Fatal("failed to append certificates")
   475  	}
   476  	creds = credentials.NewTLS(&tls.Config{
   477  		Certificates: []tls.Certificate{cert},
   478  		RootCAs:      roots,
   479  		ServerName:   "x.test.example.com",
   480  	})
   481  	clientConn, err := grpc.NewClient(lis.Addr().String(), grpc.WithTransportCredentials(creds))
   482  	if err != nil {
   483  		t.Fatalf("grpc.NewClient(%v) failed: %v", lis.Addr().String(), err)
   484  	}
   485  	defer clientConn.Close()
   486  	client := testgrpc.NewTestServiceClient(clientConn)
   487  
   488  	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
   489  	defer cancel()
   490  
   491  	// Verifying authorization decision.
   492  	if _, err = client.UnaryCall(ctx, &testpb.SimpleRequest{}); err != nil {
   493  		t.Fatalf("client.UnaryCall(_, _) = %v; want nil", err)
   494  	}
   495  }
   496  
   497  func (s) TestFileWatcherEnd2End(t *testing.T) {
   498  	for name, test := range authzTests {
   499  		t.Run(name, func(t *testing.T) {
   500  			file := createTmpPolicyFile(t, name, []byte(test.authzPolicy))
   501  			i, _ := authz.NewFileWatcher(file, 1*time.Second)
   502  			defer i.Close()
   503  
   504  			// Start a gRPC server with gRPC authz unary and stream server interceptors.
   505  			s := grpc.NewServer(
   506  				grpc.ChainUnaryInterceptor(i.UnaryInterceptor),
   507  				grpc.ChainStreamInterceptor(i.StreamInterceptor))
   508  			defer s.Stop()
   509  			testgrpc.RegisterTestServiceServer(s, &testServer{})
   510  
   511  			lis, err := net.Listen("tcp", "localhost:0")
   512  			if err != nil {
   513  				t.Fatalf("error listening: %v", err)
   514  			}
   515  			defer lis.Close()
   516  			go s.Serve(lis)
   517  
   518  			// Establish a connection to the server.
   519  			clientConn, err := grpc.NewClient(lis.Addr().String(), grpc.WithTransportCredentials(insecure.NewCredentials()))
   520  			if err != nil {
   521  				t.Fatalf("grpc.NewClient(%v) failed: %v", lis.Addr().String(), err)
   522  			}
   523  			defer clientConn.Close()
   524  			client := testgrpc.NewTestServiceClient(clientConn)
   525  
   526  			ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
   527  			defer cancel()
   528  			ctx = metadata.NewOutgoingContext(ctx, test.md)
   529  
   530  			// Verifying authorization decision for Unary RPC.
   531  			_, err = client.UnaryCall(ctx, &testpb.SimpleRequest{})
   532  			if got := status.Convert(err); got.Code() != test.wantStatus.Code() || got.Message() != test.wantStatus.Message() {
   533  				t.Fatalf("[UnaryCall] error want:{%v} got:{%v}", test.wantStatus.Err(), got.Err())
   534  			}
   535  
   536  			// Verifying authorization decision for Streaming RPC.
   537  			stream, err := client.StreamingInputCall(ctx)
   538  			if err != nil {
   539  				t.Fatalf("failed StreamingInputCall err: %v", err)
   540  			}
   541  			req := &testpb.StreamingInputCallRequest{
   542  				Payload: &testpb.Payload{
   543  					Body: []byte("hi"),
   544  				},
   545  			}
   546  			if err := stream.Send(req); err != nil && err != io.EOF {
   547  				t.Fatalf("failed stream.Send err: %v", err)
   548  			}
   549  			_, err = stream.CloseAndRecv()
   550  			if got := status.Convert(err); got.Code() != test.wantStatus.Code() || got.Message() != test.wantStatus.Message() {
   551  				t.Fatalf("[StreamingCall] error want:{%v} got:{%v}", test.wantStatus.Err(), got.Err())
   552  			}
   553  		})
   554  	}
   555  }
   556  
   557  func retryUntil(ctx context.Context, tsc testgrpc.TestServiceClient, want *status.Status) (lastErr error) {
   558  	for ctx.Err() == nil {
   559  		_, lastErr = tsc.UnaryCall(ctx, &testpb.SimpleRequest{})
   560  		if s := status.Convert(lastErr); s.Code() == want.Code() && s.Message() == want.Message() {
   561  			return nil
   562  		}
   563  		time.Sleep(20 * time.Millisecond)
   564  	}
   565  	return lastErr
   566  }
   567  
   568  func (s) TestFileWatcher_ValidPolicyRefresh(t *testing.T) {
   569  	valid1 := authzTests["DeniesRPCMatchInDenyAndAllow"]
   570  	file := createTmpPolicyFile(t, "valid_policy_refresh", []byte(valid1.authzPolicy))
   571  	i, _ := authz.NewFileWatcher(file, 100*time.Millisecond)
   572  	defer i.Close()
   573  
   574  	// Start a gRPC server with gRPC authz unary server interceptor.
   575  	s := grpc.NewServer(
   576  		grpc.ChainUnaryInterceptor(i.UnaryInterceptor))
   577  	defer s.Stop()
   578  	testgrpc.RegisterTestServiceServer(s, &testServer{})
   579  
   580  	lis, err := net.Listen("tcp", "localhost:0")
   581  	if err != nil {
   582  		t.Fatalf("error listening: %v", err)
   583  	}
   584  	defer lis.Close()
   585  	go s.Serve(lis)
   586  
   587  	// Establish a connection to the server.
   588  	clientConn, err := grpc.NewClient(lis.Addr().String(), grpc.WithTransportCredentials(insecure.NewCredentials()))
   589  	if err != nil {
   590  		t.Fatalf("grpc.NewClient(%v) failed: %v", lis.Addr().String(), err)
   591  	}
   592  	defer clientConn.Close()
   593  	client := testgrpc.NewTestServiceClient(clientConn)
   594  
   595  	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
   596  	defer cancel()
   597  
   598  	// Verifying authorization decision.
   599  	_, err = client.UnaryCall(ctx, &testpb.SimpleRequest{})
   600  	if got := status.Convert(err); got.Code() != valid1.wantStatus.Code() || got.Message() != valid1.wantStatus.Message() {
   601  		t.Fatalf("client.UnaryCall(_, _) = %v; want = %v", got.Err(), valid1.wantStatus.Err())
   602  	}
   603  
   604  	// Rewrite the file with a different valid authorization policy.
   605  	valid2 := authzTests["AllowsRPCEmptyDenyMatchInAllow"]
   606  	if err := os.WriteFile(file, []byte(valid2.authzPolicy), os.ModePerm); err != nil {
   607  		t.Fatalf("os.WriteFile(%q) failed: %v", file, err)
   608  	}
   609  
   610  	// Verifying authorization decision.
   611  	if got := retryUntil(ctx, client, valid2.wantStatus); got != nil {
   612  		t.Fatalf("client.UnaryCall(_, _) = %v; want = %v", got, valid2.wantStatus.Err())
   613  	}
   614  }
   615  
   616  func (s) TestFileWatcher_InvalidPolicySkipReload(t *testing.T) {
   617  	valid := authzTests["DeniesRPCMatchInDenyAndAllow"]
   618  	file := createTmpPolicyFile(t, "invalid_policy_skip_reload", []byte(valid.authzPolicy))
   619  	i, _ := authz.NewFileWatcher(file, 20*time.Millisecond)
   620  	defer i.Close()
   621  
   622  	// Start a gRPC server with gRPC authz unary server interceptors.
   623  	s := grpc.NewServer(
   624  		grpc.ChainUnaryInterceptor(i.UnaryInterceptor))
   625  	defer s.Stop()
   626  	testgrpc.RegisterTestServiceServer(s, &testServer{})
   627  
   628  	lis, err := net.Listen("tcp", "localhost:0")
   629  	if err != nil {
   630  		t.Fatalf("error listening: %v", err)
   631  	}
   632  	defer lis.Close()
   633  	go s.Serve(lis)
   634  
   635  	// Establish a connection to the server.
   636  	clientConn, err := grpc.NewClient(lis.Addr().String(), grpc.WithTransportCredentials(insecure.NewCredentials()))
   637  	if err != nil {
   638  		t.Fatalf("grpc.NewClient(%v) failed: %v", lis.Addr().String(), err)
   639  	}
   640  	defer clientConn.Close()
   641  	client := testgrpc.NewTestServiceClient(clientConn)
   642  
   643  	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
   644  	defer cancel()
   645  
   646  	// Verifying authorization decision.
   647  	_, err = client.UnaryCall(ctx, &testpb.SimpleRequest{})
   648  	if got := status.Convert(err); got.Code() != valid.wantStatus.Code() || got.Message() != valid.wantStatus.Message() {
   649  		t.Fatalf("client.UnaryCall(_, _) = %v; want = %v", got.Err(), valid.wantStatus.Err())
   650  	}
   651  
   652  	// Skips the invalid policy update, and continues to use the valid policy.
   653  	if err := os.WriteFile(file, []byte("{}"), os.ModePerm); err != nil {
   654  		t.Fatalf("os.WriteFile(%q) failed: %v", file, err)
   655  	}
   656  
   657  	// Wait 40 ms for background go routine to read updated files.
   658  	time.Sleep(40 * time.Millisecond)
   659  
   660  	// Verifying authorization decision.
   661  	_, err = client.UnaryCall(ctx, &testpb.SimpleRequest{})
   662  	if got := status.Convert(err); got.Code() != valid.wantStatus.Code() || got.Message() != valid.wantStatus.Message() {
   663  		t.Fatalf("client.UnaryCall(_, _) = %v; want = %v", got.Err(), valid.wantStatus.Err())
   664  	}
   665  }
   666  
   667  func (s) TestFileWatcher_RecoversFromReloadFailure(t *testing.T) {
   668  	valid1 := authzTests["DeniesRPCMatchInDenyAndAllow"]
   669  	file := createTmpPolicyFile(t, "recovers_from_reload_failure", []byte(valid1.authzPolicy))
   670  	i, _ := authz.NewFileWatcher(file, 100*time.Millisecond)
   671  	defer i.Close()
   672  
   673  	// Start a gRPC server with gRPC authz unary server interceptors.
   674  	s := grpc.NewServer(
   675  		grpc.ChainUnaryInterceptor(i.UnaryInterceptor))
   676  	defer s.Stop()
   677  	testgrpc.RegisterTestServiceServer(s, &testServer{})
   678  
   679  	lis, err := net.Listen("tcp", "localhost:0")
   680  	if err != nil {
   681  		t.Fatalf("error listening: %v", err)
   682  	}
   683  	defer lis.Close()
   684  	go s.Serve(lis)
   685  
   686  	// Establish a connection to the server.
   687  	clientConn, err := grpc.NewClient(lis.Addr().String(), grpc.WithTransportCredentials(insecure.NewCredentials()))
   688  	if err != nil {
   689  		t.Fatalf("grpc.NewClient(%v) failed: %v", lis.Addr().String(), err)
   690  	}
   691  	defer clientConn.Close()
   692  	client := testgrpc.NewTestServiceClient(clientConn)
   693  
   694  	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
   695  	defer cancel()
   696  
   697  	// Verifying authorization decision.
   698  	_, err = client.UnaryCall(ctx, &testpb.SimpleRequest{})
   699  	if got := status.Convert(err); got.Code() != valid1.wantStatus.Code() || got.Message() != valid1.wantStatus.Message() {
   700  		t.Fatalf("client.UnaryCall(_, _) = %v; want = %v", got.Err(), valid1.wantStatus.Err())
   701  	}
   702  
   703  	// Skips the invalid policy update, and continues to use the valid policy.
   704  	if err := os.WriteFile(file, []byte("{}"), os.ModePerm); err != nil {
   705  		t.Fatalf("os.WriteFile(%q) failed: %v", file, err)
   706  	}
   707  
   708  	// Wait 120 ms for background go routine to read updated files.
   709  	time.Sleep(120 * time.Millisecond)
   710  
   711  	// Verifying authorization decision.
   712  	_, err = client.UnaryCall(ctx, &testpb.SimpleRequest{})
   713  	if got := status.Convert(err); got.Code() != valid1.wantStatus.Code() || got.Message() != valid1.wantStatus.Message() {
   714  		t.Fatalf("client.UnaryCall(_, _) = %v; want = %v", got.Err(), valid1.wantStatus.Err())
   715  	}
   716  
   717  	// Rewrite the file with a different valid authorization policy.
   718  	valid2 := authzTests["AllowsRPCEmptyDenyMatchInAllow"]
   719  	if err := os.WriteFile(file, []byte(valid2.authzPolicy), os.ModePerm); err != nil {
   720  		t.Fatalf("os.WriteFile(%q) failed: %v", file, err)
   721  	}
   722  
   723  	// Verifying authorization decision.
   724  	if got := retryUntil(ctx, client, valid2.wantStatus); got != nil {
   725  		t.Fatalf("client.UnaryCall(_, _) = %v; want = %v", got, valid2.wantStatus.Err())
   726  	}
   727  }
   728
View as plain text