Implemented a foundation for gRPC streaming

.vscode/quickfeed-words.txt

@@ -50,6 +50,7 @@ Godkjent
 gofumports
 gofumpt
 golangci
+Gollum
 googleapis
 gopath
 gopb
@@ -139,9 +140,11 @@ qfconnect
 qlog
 qtest
 quickfeed
+Radagast
 Repos
 repotoken
 run
+Sauron
 scms
 scorelimit
 scriptfile

web/grpc_auth_test.go

@@ -42,11 +42,11 @@ func TestGrpcAuth(t *testing.T) {
 		t.Fatal(err)
 	}
 	shutdown := web.MockQuickFeedServer(t, logger, db, connect.WithInterceptors(
-		interceptor.Metrics(),
-		interceptor.Validation(logger),
-		interceptor.UnaryUserVerifier(logger, tm),
-		interceptor.AccessControl(tm),
-		interceptor.TokenRefresher(tm),
+		interceptor.NewMetricsInterceptor(),
+		interceptor.NewValidationInterceptor(logger),
+		interceptor.NewUserInterceptor(logger, tm),
+		interceptor.NewAccessControlInterceptor(tm),
+		interceptor.NewTokenInterceptor(tm),
 	))
 
 	client := qtest.QuickFeedClient("")

web/interceptor/access_control.go

@@ -76,89 +76,107 @@ var accessRolesFor = map[string]roles{
 	"CreateCourse":            {admin},
 }
 
-// AccessControl checks user information stored in the JWT claims against the list of roles required to call the method.
-func AccessControl(tm *auth.TokenManager) connect.Interceptor {
-	return connect.UnaryInterceptorFunc(func(next connect.UnaryFunc) connect.UnaryFunc {
-		return connect.UnaryFunc(func(ctx context.Context, request connect.AnyRequest) (connect.AnyResponse, error) {
-			procedure := request.Spec().Procedure
-			method := procedure[strings.LastIndex(procedure, "/")+1:]
-			req, ok := request.Any().(requestID)
-			if !ok {
-				return nil, connect.NewError(connect.CodeUnimplemented,
-					fmt.Errorf("%s failed: message type %T does not implement IDFor interface", method, request))
-			}
-			claims, ok := auth.ClaimsFromContext(ctx)
-			if !ok {
-				return nil, connect.NewError(connect.CodePermissionDenied,
-					fmt.Errorf("AccessControl(%s): failed to get claims from request context", method))
-			}
-			for _, role := range accessRolesFor[method] {
-				switch role {
-				case none:
-					return next(ctx, request)
-				case user:
-					if claims.SameUser(req) {
-						// Make sure the user is not updating own admin status.
-						if method == "UpdateUser" {
-							if req.(*qf.User).GetIsAdmin() && !claims.Admin {
-								return nil, connect.NewError(connect.CodePermissionDenied,
-									fmt.Errorf("AccessControl(%s): user %d attempted to change admin status from %v to %v",
-										method, claims.UserID, claims.Admin, req.(*qf.User).GetIsAdmin()))
-							}
-						}
-						return next(ctx, request)
-					}
-				case student:
-					// GetSubmissions is used to fetch individual and group submissions.
-					// For individual submissions needs an extra check for user ID in request.
-					if method == "GetSubmissions" && req.IDFor("group") == 0 {
-						if !claims.SameUser(req) {
-							return nil, connect.NewError(connect.CodePermissionDenied,
-								fmt.Errorf("AccessControl(%s): ID mismatch in claims (%d) and request (%d)",
-									method, claims.UserID, req.IDFor("user")))
-						}
-					}
-					if claims.HasCourseStatus(req, qf.Enrollment_STUDENT) {
-						return next(ctx, request)
-					}
-				case group:
-					// Request for CreateGroup will not have ID yet, need to check
-					// if the user is in the group (unless teacher).
-					if method == "CreateGroup" {
-						notMember := !req.(*qf.Group).Contains(&qf.User{ID: claims.UserID})
-						notTeacher := !claims.HasCourseStatus(req, qf.Enrollment_TEACHER)
-						if notMember && notTeacher {
+type AccessControlInterceptor struct {
+	tokenManager *auth.TokenManager
+}
+
+func NewAccessControlInterceptor(tm *auth.TokenManager) *AccessControlInterceptor {
+	return &AccessControlInterceptor{tokenManager: tm}
+}
+
+func (*AccessControlInterceptor) WrapStreamingHandler(next connect.StreamingHandlerFunc) connect.StreamingHandlerFunc {
+	return connect.StreamingHandlerFunc(func(ctx context.Context, conn connect.StreamingHandlerConn) error {
+		return next(ctx, conn)
+	})
+}
+
+func (*AccessControlInterceptor) WrapStreamingClient(next connect.StreamingClientFunc) connect.StreamingClientFunc {
+	return connect.StreamingClientFunc(func(ctx context.Context, spec connect.Spec) connect.StreamingClientConn {
+		return next(ctx, spec)
+	})
+}
+
+// WrapUnary checks user information stored in the JWT claims against the list of roles required to call the method.
+func (a *AccessControlInterceptor) WrapUnary(next connect.UnaryFunc) connect.UnaryFunc {
+	return connect.UnaryFunc(func(ctx context.Context, request connect.AnyRequest) (connect.AnyResponse, error) {
+		procedure := request.Spec().Procedure
+		method := procedure[strings.LastIndex(procedure, "/")+1:]
+		req, ok := request.Any().(requestID)
+		if !ok {
+			return nil, connect.NewError(connect.CodeUnimplemented,
+				fmt.Errorf("access denied for %s: message type %T does not implement 'requestID' interface", method, request))
+		}
+		claims, ok := auth.ClaimsFromContext(ctx)
+		if !ok {
+			return nil, connect.NewError(connect.CodePermissionDenied,
+				fmt.Errorf("access denied for %s: failed to get claims from request context", method))
+		}
+		for _, role := range accessRolesFor[method] {
+			switch role {
+			case none:
+				return next(ctx, request)
+			case user:
+				if claims.SameUser(req) {
+					// Make sure the user is not updating own admin status.
+					if method == "UpdateUser" {
+						if req.(*qf.User).GetIsAdmin() && !claims.Admin {
 							return nil, connect.NewError(connect.CodePermissionDenied,
-								fmt.Errorf("AccessControl(%s): user %d tried to create group while not teacher or group member", method, claims.UserID))
+								fmt.Errorf("access denied for %s: user %d attempted to change admin status from %v to %v",
+									method, claims.UserID, claims.Admin, req.(*qf.User).GetIsAdmin()))
 						}
-						// Otherwise, create the group.
-						return next(ctx, request)
 					}
-					groupID := req.IDFor("group")
-					for _, group := range claims.Groups {
-						if group == groupID {
-							return next(ctx, request)
-						}
+					return next(ctx, request)
+				}
+			case student:
+				// GetSubmissions is used to fetch individual and group submissions.
+				// For individual submissions needs an extra check for user ID in request.
+				if method == "GetSubmissions" && req.IDFor("group") == 0 {
+					if !claims.SameUser(req) {
+						return nil, connect.NewError(connect.CodePermissionDenied,
+							fmt.Errorf("access denied for %s: ID mismatch in claims (%d) and request (%d)",
+								method, claims.UserID, req.IDFor("user")))
 					}
-				case teacher:
-					if method == "GetUserByCourse" {
-						if err := claims.IsCourseTeacher(tm.Database(), request.Any().(*qf.CourseUserRequest)); err != nil {
-							return nil, connect.NewError(connect.CodePermissionDenied,
-								fmt.Errorf("AccessControl(%s): %w", method, err))
-						}
-						return next(ctx, request)
+				}
+				if claims.HasCourseStatus(req, qf.Enrollment_STUDENT) {
+					return next(ctx, request)
+				}
+			case group:
+				// Request for CreateGroup will not have ID yet, need to check
+				// if the user is in the group (unless teacher).
+				if method == "CreateGroup" {
+					notMember := !req.(*qf.Group).Contains(&qf.User{ID: claims.UserID})
+					notTeacher := !claims.HasCourseStatus(req, qf.Enrollment_TEACHER)
+					if notMember && notTeacher {
+						return nil, connect.NewError(connect.CodePermissionDenied,
+							fmt.Errorf("access denied for %s: user %d tried to create group while not teacher or group member", method, claims.UserID))
 					}
-					if claims.HasCourseStatus(req, qf.Enrollment_TEACHER) {
+					// Otherwise, create the group.
+					return next(ctx, request)
+				}
+				groupID := req.IDFor("group")
+				for _, group := range claims.Groups {
+					if group == groupID {
 						return next(ctx, request)
 					}
-				case admin:
-					if claims.Admin {
-						return next(ctx, request)
+				}
+			case teacher:
+				if method == "GetUserByCourse" {
+					if err := claims.IsCourseTeacher(a.tokenManager.Database(), request.Any().(*qf.CourseUserRequest)); err != nil {
+						return nil, connect.NewError(connect.CodePermissionDenied,
+							fmt.Errorf("access denied for %s: %w", method, err))
 					}
+					return next(ctx, request)
+				}
+				if claims.HasCourseStatus(req, qf.Enrollment_TEACHER) {
+					return next(ctx, request)
+				}
+			case admin:
+				if claims.Admin {
+					return next(ctx, request)
 				}
 			}
-			return nil, connect.NewError(connect.CodePermissionDenied,
-				fmt.Errorf("AccessDenied(%s): required roles %v not satisfied by claims: %s", method, accessRolesFor[method], claims))
-		})
+		}
+		return nil, connect.NewError(connect.CodePermissionDenied,
+			fmt.Errorf("access denied for %s: required roles %v not satisfied by claims: %s", method, accessRolesFor[method], claims))
 	})
 }

web/interceptor/access_control_test.go

@@ -31,8 +31,8 @@ func TestAccessControl(t *testing.T) {
 		t.Fatal(err)
 	}
 	shutdown := web.MockQuickFeedServer(t, logger, db, connect.WithInterceptors(
-		interceptor.UnaryUserVerifier(logger, tm),
-		interceptor.AccessControl(tm),
+		interceptor.NewUserInterceptor(logger, tm),
+		interceptor.NewAccessControlInterceptor(tm),
 	))
 
 	client := qtest.QuickFeedClient("")

web/interceptor/metrics.go

@@ -46,26 +46,50 @@ var (
 	}, []string{"user"})
 )
 
-func Metrics() connect.Interceptor {
-	return connect.UnaryInterceptorFunc(func(next connect.UnaryFunc) connect.UnaryFunc {
-		return connect.UnaryFunc(func(ctx context.Context, request connect.AnyRequest) (connect.AnyResponse, error) {
-			procedure := request.Spec().Procedure
-			methodName := procedure[strings.LastIndex(procedure, "/")+1:]
-			defer metricsTimer(methodName)()
-			resp, err := next(ctx, request)
-			accessedMethodsCounter.WithLabelValues(methodName).Inc()
-			if resp != nil {
-				respondedMethodsCounter.WithLabelValues(methodName).Inc()
-			}
-			if err != nil {
-				failedMethodsCounter.WithLabelValues(methodName).Inc()
-				if methodName == "GetUser" {
-					// Can't get the user ID from err; so just counting
-					loginCounter.WithLabelValues("").Inc()
-				}
+type MetricsInterceptor struct{}
+
+func NewMetricsInterceptor() *MetricsInterceptor {
+	return &MetricsInterceptor{}
+}
+
+func (*MetricsInterceptor) WrapStreamingHandler(next connect.StreamingHandlerFunc) connect.StreamingHandlerFunc {
+	return connect.StreamingHandlerFunc(func(ctx context.Context, conn connect.StreamingHandlerConn) error {
+		procedure := conn.Spec().Procedure
+		methodName := procedure[strings.LastIndex(procedure, "/")+1:]
+		defer metricsTimer(methodName)()
+		accessedMethodsCounter.WithLabelValues(methodName).Inc()
+		err := next(ctx, conn)
+		if err != nil {
+			failedMethodsCounter.WithLabelValues(methodName).Inc()
+		}
+		return err
+	})
+}
+
+func (*MetricsInterceptor) WrapStreamingClient(next connect.StreamingClientFunc) connect.StreamingClientFunc {
+	return connect.StreamingClientFunc(func(ctx context.Context, spec connect.Spec) connect.StreamingClientConn {
+		return next(ctx, spec)
+	})
+}
+
+func (*MetricsInterceptor) WrapUnary(next connect.UnaryFunc) connect.UnaryFunc {
+	return connect.UnaryFunc(func(ctx context.Context, request connect.AnyRequest) (connect.AnyResponse, error) {
+		procedure := request.Spec().Procedure
+		methodName := procedure[strings.LastIndex(procedure, "/")+1:]
+		defer metricsTimer(methodName)()
+		resp, err := next(ctx, request)
+		accessedMethodsCounter.WithLabelValues(methodName).Inc()
+		if resp != nil {
+			respondedMethodsCounter.WithLabelValues(methodName).Inc()
+		}
+		if err != nil {
+			failedMethodsCounter.WithLabelValues(methodName).Inc()
+			if methodName == "GetUser" {
+				// Can't get the user ID from err; so just counting
+				loginCounter.WithLabelValues("").Inc()
 			}
-			return resp, err
-		})
+		}
+		return resp, err
 	})
 }
 

web/interceptor/req_validation.go

@@ -22,32 +22,50 @@ type idCleaner interface {
 	RemoveRemoteID()
 }
 
-// Validation returns a new unary server interceptor that validates requests
+type ValidationInterceptor struct {
+	logger *zap.SugaredLogger
+}
+
+func NewValidationInterceptor(logger *zap.SugaredLogger) *ValidationInterceptor {
+	return &ValidationInterceptor{logger: logger}
+}
+
+func (*ValidationInterceptor) WrapStreamingHandler(next connect.StreamingHandlerFunc) connect.StreamingHandlerFunc {
+	return connect.StreamingHandlerFunc(func(ctx context.Context, conn connect.StreamingHandlerConn) error {
+		return next(ctx, conn)
+	})
+}
+
+func (*ValidationInterceptor) WrapStreamingClient(next connect.StreamingClientFunc) connect.StreamingClientFunc {
+	return connect.StreamingClientFunc(func(ctx context.Context, spec connect.Spec) connect.StreamingClientConn {
+		return next(ctx, spec)
+	})
+}
+
+// WrapUnary returns a new unary server interceptor that validates requests
 // that implements the validator interface.
 // Invalid requests are rejected without logging and before it reaches any
 // user-level code and returns an illegal argument to the client.
 // Further, the response values are cleaned of any remote IDs.
 // In addition, the interceptor also implements a cancellation mechanism.
-func Validation(logger *zap.SugaredLogger) connect.Interceptor {
-	return connect.UnaryInterceptorFunc(func(next connect.UnaryFunc) connect.UnaryFunc {
-		return connect.UnaryFunc(func(ctx context.Context, request connect.AnyRequest) (connect.AnyResponse, error) {
-			if request.Any() != nil {
-				if err := validate(logger, request.Any()); err != nil {
-					// Reject the request if it is invalid.
-					return nil, err
-				}
-			}
-			resp, err := next(ctx, request)
-			if err != nil {
-				// Do not return the message to the client if an error occurs.
-				// We log the error and return an empty response.
-				logger.Errorf("Method '%s' failed: %v", request.Spec().Procedure, err)
-				logger.Errorf("Request Message: %T: %v", request.Any(), request.Any())
+func (v *ValidationInterceptor) WrapUnary(next connect.UnaryFunc) connect.UnaryFunc {
+	return connect.UnaryFunc(func(ctx context.Context, request connect.AnyRequest) (connect.AnyResponse, error) {
+		if request.Any() != nil {
+			if err := validate(v.logger, request.Any()); err != nil {
+				// Reject the request if it is invalid.
 				return nil, err
 			}
-			clean(resp.Any())
-			return resp, err
-		})
+		}
+		resp, err := next(ctx, request)
+		if err != nil {
+			// Do not return the message to the client if an error occurs.
+			// We log the error and return an empty response.
+			v.logger.Errorf("Method '%s' failed: %v", request.Spec().Procedure, err)
+			v.logger.Errorf("Request Message: %T: %v", request.Any(), request.Any())
+			return nil, err
+		}
+		clean(resp.Any())
+		return resp, err
 	})
 }