WatchResources implements runtimev1.RuntimeServiceServer
(req *runtimev1.WatchResourcesRequest, ss runtimev1.RuntimeService_WatchResourcesServer)
| 91 | |
| 92 | // WatchResources implements runtimev1.RuntimeServiceServer |
| 93 | func (s *Server) WatchResources(req *runtimev1.WatchResourcesRequest, ss runtimev1.RuntimeService_WatchResourcesServer) error { |
| 94 | observability.AddRequestAttributes(ss.Context(), |
| 95 | attribute.String("args.instance_id", req.InstanceId), |
| 96 | attribute.String("args.kind", req.Kind), |
| 97 | ) |
| 98 | |
| 99 | claims := auth.GetClaims(ss.Context(), req.InstanceId) |
| 100 | if !claims.Can(runtime.ReadObjects) { |
| 101 | return ErrForbidden |
| 102 | } |
| 103 | |
| 104 | ctrl, err := s.runtime.Controller(ss.Context(), req.InstanceId) |
| 105 | if err != nil { |
| 106 | return err |
| 107 | } |
| 108 | |
| 109 | if req.Replay { |
| 110 | rs, err := ctrl.List(ss.Context(), req.Kind, "", false) |
| 111 | if err != nil { |
| 112 | return err |
| 113 | } |
| 114 | |
| 115 | for _, r := range rs { |
| 116 | r, access, err := s.runtime.ApplySecurityPolicy(ss.Context(), req.InstanceId, claims, r) |
| 117 | if err != nil { |
| 118 | return mapGRPCErrorWithFallback(err, codes.InvalidArgument) |
| 119 | } |
| 120 | if !access { |
| 121 | continue |
| 122 | } |
| 123 | |
| 124 | err = ss.Send(&runtimev1.WatchResourcesResponse{ |
| 125 | Event: runtimev1.ResourceEvent_RESOURCE_EVENT_WRITE, |
| 126 | Resource: r, |
| 127 | }) |
| 128 | if err != nil { |
| 129 | return err |
| 130 | } |
| 131 | } |
| 132 | } |
| 133 | |
| 134 | return ctrl.Subscribe(ss.Context(), func(e runtimev1.ResourceEvent, n *runtimev1.ResourceName, r *runtimev1.Resource) { |
| 135 | if r != nil { // r is nil for deletion events |
| 136 | var access bool |
| 137 | var err error |
| 138 | r, access, err = s.runtime.ApplySecurityPolicy(ss.Context(), req.InstanceId, claims, r) |
| 139 | if err != nil { |
| 140 | s.logger.Info("failed to apply security policy", zap.String("name", n.Name), zap.Error(err)) |
| 141 | return |
| 142 | } |
| 143 | if !access { |
| 144 | return |
| 145 | } |
| 146 | } |
| 147 | |
| 148 | err = ss.Send(&runtimev1.WatchResourcesResponse{ |
| 149 | Event: e, |
| 150 | Name: n, |
no test coverage detected