Make kubelet client optional for cluster-scoped brokers

conf/agent/agent_full.conf

@@ -534,8 +534,11 @@ plugins {
             # SubjectAccessReview username; no groups are set.
             # pod_reference_scope controls pod KubernetesObjectReference
             # resolution for the broker. Valid values are "agent_node"
-            # (default) and "cluster". agent_node only accepts pods whose
-            # spec.nodeName matches this agent's node name. Non-pod object
+            # (default) and "cluster". agent_node requires the kubelet client
+            # and only accepts pods returned by the local kubelet. cluster
+            # uses the local kubelet as a fast path when it is available, then
+            # falls back to the Kubernetes API server when the client is absent,
+            # the request fails, or the pod is not local. Non-pod object
             # references always use the Kubernetes API server.
             #
             # Additional Kubernetes authorization requirements when

doc/plugin_agent_workloadattestor_k8s.md

@@ -183,12 +183,14 @@ does not use the broker configuration or run this review.
 For `KubernetesObjectReference`, the reference identifies the target object by
 its resource (`<plural>.<group>`, with `core` as the group string for core
 resources) and either its namespaced name (`namespace` + `name`), its `uid`,
-or both. Pod references try the local kubelet pod list first. With the default
-`pod_reference_scope = "agent_node"`, pod references are limited to information
-returned by the local kubelet and do not fall back to the Kubernetes API
-server. With `pod_reference_scope = "cluster"`, pod references may fall back to
-the Kubernetes API server and resolve pods on any node. Non-pod object
-references are resolved through the Kubernetes API server. When
+or both. Pod references use the local kubelet pod list as a fast path when a
+kubelet client is available. With the default
+`pod_reference_scope = "agent_node"`, the kubelet client is required and pod
+references are limited to information returned by the local kubelet. With
+`pod_reference_scope = "cluster"`, an absent or unavailable kubelet client does
+not prevent resolution: the plugin falls back to the Kubernetes API server and
+can resolve pods on any node. Non-pod object references are resolved through
+the Kubernetes API server. When
 `experimental.broker.access_policy = "enforced"`, the plugin then creates the same
 `SubjectAccessReview` for the referenced object. The review uses the broker
 SPIFFE ID as the SAR username, no groups, the reference's resource group and

pkg/agent/plugin/workloadattestor/k8s/k8s.go

@@ -887,16 +887,16 @@ func brokerPodReferenceScope(brokerEntry *k8sBrokerEntry) podReferenceScope {
 	return brokerEntry.PodReferenceScope
 }
 
-// findPodByName resolves a single pod by its namespaced name. The kubelet
-// pod list is iterated first; this is O(n) over the node's pods (the list
-// is indexed by UID, not name) but n is small in practice and saves an API
-// server round-trip when the pod is local. Under agent_node scope, resolution
-// stops at the kubelet pod list. Under cluster scope, if the pod is not in the
-// kubelet list, the apiserver answers a precise Get directly — no list, no
-// client-side filter.
+// findPodByName resolves a single pod by its namespaced name. When configured,
+// the kubelet pod list is iterated first; this is O(n) over the node's pods (the
+// list is indexed by UID, not name) but n is small in practice and saves an API
+// server round-trip when the pod is local. Under agent_node scope, the kubelet
+// client is required and resolution stops at its pod list. Under cluster scope,
+// an unavailable kubelet client is ignored and the apiserver answers a precise
+// Get directly — no list, no client-side filter.
 func (p *Plugin) findPodByName(ctx context.Context, config *k8sConfig, namespace, name string, scope podReferenceScope) (*corev1.Pod, error) {
 	// Try kubelet pod list first; iterate to find a match by namespace+name.
-	podList, err := p.getPodList(ctx, config.Client, config.PollRetryInterval/2)
+	podList, err := p.getPodListForReference(ctx, config, scope)
 	if err != nil {
 		return nil, err
 	}
@@ -929,18 +929,19 @@ func (p *Plugin) findPodByName(ctx context.Context, config *k8sConfig, namespace
 	return pod, nil
 }
 
-// findPodByUID resolves a single pod by its Kubernetes UID. The kubelet pod
-// list is checked first because it's already keyed by UID and only contains
-// pods scheduled to this node — both common-case wins. Under agent_node scope,
-// resolution stops at the kubelet pod list. Under cluster scope, if the pod is
-// not in the kubelet list, it falls back to a cluster-wide
+// findPodByUID resolves a single pod by its Kubernetes UID. When configured,
+// the kubelet pod list is checked first because it's already keyed by UID and
+// only contains pods scheduled to this node — both common-case wins. Under
+// agent_node scope, the kubelet client is required and resolution stops at its
+// pod list. Under cluster scope, an unavailable kubelet client is ignored and
+// resolution falls back to a cluster-wide
 // PartialObjectMetadata List from the API server cache to resolve the pod
 // name+namespace, then fetches the full pod with a single live Get. Kubernetes
 // does not support `metadata.uid` as a field selector, so we list and filter
 // client-side regardless.
 func (p *Plugin) findPodByUID(ctx context.Context, config *k8sConfig, uid types.UID, scope podReferenceScope) (*corev1.Pod, error) {
 	// Try kubelet pod list first (already indexed by UID).
-	podList, err := p.getPodList(ctx, config.Client, config.PollRetryInterval/2)
+	podList, err := p.getPodListForReference(ctx, config, scope)
 	if err != nil {
 		return nil, err
 	}
@@ -980,6 +981,25 @@ func (p *Plugin) findPodByUID(ctx context.Context, config *k8sConfig, uid types.
 	return pod, nil
 }
 
+func (p *Plugin) getPodListForReference(ctx context.Context, config *k8sConfig, scope podReferenceScope) (map[string]*fastjson.Value, error) {
+	if config.Client == nil {
+		if scope != podReferenceScopeCluster {
+			return nil, status.Error(codes.FailedPrecondition, "kubelet client is not configured")
+		}
+		return nil, nil
+	}
+
+	podList, err := p.getPodList(ctx, config.Client, config.PollRetryInterval/2)
+	if err != nil {
+		if scope != podReferenceScopeCluster {
+			return nil, err
+		}
+		p.log.Warn("Unable to query kubelet for pod reference; falling back to Kubernetes API", "err", err)
+		return nil, nil
+	}
+	return podList, nil
+}
+
 // decodePodFromKubelet rehydrates a `corev1.Pod` from the partially-parsed
 // JSON the kubelet pod-list path keeps in fastjson form (the cache stores
 // pods as `*fastjson.Value` to avoid per-request unmarshalling when no pod

pkg/agent/plugin/workloadattestor/k8s/k8s_test.go

@@ -1262,8 +1262,8 @@ func testBrokerConfigWithEnforcedAccessPolicy() string {
 	return testBrokerConfigWithAccessPolicy("", string(brokerAccessPolicyEnforced))
 }
 
-func testBrokerConfigWithPodReferenceScope(scope string) string {
-	return testBrokerConfigWithAccessPolicy(scope, string(brokerAccessPolicyPermissive))
+func testBrokerConfigWithClusterPodReferenceScope() string {
+	return testBrokerConfigWithAccessPolicy(string(podReferenceScopeCluster), string(brokerAccessPolicyPermissive))
 }
 
 func testBrokerConfigWithAccessPolicy(scope, accessPolicy string) string {
@@ -1845,7 +1845,7 @@ func (s *Suite) TestAttestReferenceWithPodUID_FallbackToAPIServerWithClusterScop
 		max_poll_attempts = 5
 		poll_retry_interval = "1s"
 		%s
-`, s.kubeletPort(), testBrokerConfigWithPodReferenceScope("cluster"))
+`, s.kubeletPort(), testBrokerConfigWithClusterPodReferenceScope())
 	wa := s.loadPluginWithKubeClients(cfg, liveClient, metadataClient)
 
 	anyRef, err := anypb.New(&broker.KubernetesObjectReference{Type: &broker.KubernetesObjectType{Plural: "pods", Group: "core"}, Uid: testPodUID})
@@ -1871,7 +1871,7 @@ func (s *Suite) TestAttestReferenceWithPodName_FallbackToAPIServerWithClusterSco
 		max_poll_attempts = 5
 		poll_retry_interval = "1s"
 		%s
-`, s.kubeletPort(), testBrokerConfigWithPodReferenceScope("cluster"))
+`, s.kubeletPort(), testBrokerConfigWithClusterPodReferenceScope())
 	wa := s.loadPluginWithKubeClient(cfg, liveClient)
 
 	anyRef, err := anypb.New(&broker.KubernetesObjectReference{
@@ -1885,6 +1885,90 @@ func (s *Suite) TestAttestReferenceWithPodName_FallbackToAPIServerWithClusterSco
 	s.requireSelectorsEqual(testPodSelectors, selectors)
 }
 
+func (s *Suite) TestAttestReferenceWithPodUID_FallsBackToAPIServerWhenKubeletFails() {
+	s.startInsecureKubelet()
+
+	// Do not configure a pod list response. The kubelet returns an error.
+	liveClient := fakeKubeClientWithSubjectAccessReview(true, nil, testAPIServerBlogPod())
+	metadataClient := fakeKubeMetadataClient(testAPIServerBlogPodMetadata())
+	cfg := fmt.Sprintf(`
+		kubelet_read_only_port = %d
+		max_poll_attempts = 5
+		poll_retry_interval = "1s"
+		%s
+`, s.kubeletPort(), testBrokerConfigWithClusterPodReferenceScope())
+	wa := s.loadPluginWithKubeClients(cfg, liveClient, metadataClient)
+
+	anyRef, err := anypb.New(&broker.KubernetesObjectReference{Type: &broker.KubernetesObjectType{Plural: "pods", Group: "core"}, Uid: testPodUID})
+	s.Require().NoError(err)
+
+	selectors, err := wa.AttestReference(testBrokerContext(), anyRef)
+	s.Require().NoError(err)
+	s.requireSelectorsEqual(testPodSelectors, selectors)
+}
+
+func (s *Suite) TestAttestReferenceWithPodName_FallsBackToAPIServerWhenKubeletFails() {
+	s.startInsecureKubelet()
+
+	// Do not configure a pod list response. The kubelet returns an error.
+	liveClient := fakeKubeClientWithSubjectAccessReview(true, nil, testAPIServerBlogPod())
+	cfg := fmt.Sprintf(`
+		kubelet_read_only_port = %d
+		max_poll_attempts = 5
+		poll_retry_interval = "1s"
+		%s
+`, s.kubeletPort(), testBrokerConfigWithClusterPodReferenceScope())
+	wa := s.loadPluginWithKubeClient(cfg, liveClient)
+
+	anyRef, err := anypb.New(&broker.KubernetesObjectReference{
+		Type: &broker.KubernetesObjectType{Plural: "pods", Group: "core"},
+		Key:  &broker.KubernetesObjectKey{Namespace: "default", Name: "blog-24ck7"},
+	})
+	s.Require().NoError(err)
+
+	selectors, err := wa.AttestReference(testBrokerContext(), anyRef)
+	s.Require().NoError(err)
+	s.requireSelectorsEqual(testPodSelectors, selectors)
+}
+
+func (s *Suite) TestFindPodByUID_FallsBackToAPIServerWithoutKubeletClient() {
+	p := s.newPlugin()
+	p.kubeClient = fakeKubeClientWithSubjectAccessReview(true, nil, testAPIServerBlogPod())
+	p.kubeMetadataClient = fakeKubeMetadataClient(testAPIServerBlogPodMetadata())
+
+	pod, err := p.findPodByUID(context.Background(), &k8sConfig{}, testPodUID, podReferenceScopeCluster)
+	s.Require().NoError(err)
+	s.Require().Equal(testPodUID, string(pod.UID))
+}
+
+func (s *Suite) TestFindPodByName_FallsBackToAPIServerWithoutKubeletClient() {
+	p := s.newPlugin()
+	p.kubeClient = fakeKubeClientWithSubjectAccessReview(true, nil, testAPIServerBlogPod())
+
+	pod, err := p.findPodByName(context.Background(), &k8sConfig{}, "default", "blog-24ck7", podReferenceScopeCluster)
+	s.Require().NoError(err)
+	s.Require().Equal(testPodUID, string(pod.UID))
+}
+
+func (s *Suite) TestFindPodByUID_AgentNodeScopeRequiresKubeletClient() {
+	p := s.newPlugin()
+	p.kubeClient = fakeKubeClientWithSubjectAccessReview(true, nil, testAPIServerBlogPod())
+	p.kubeMetadataClient = fakeKubeMetadataClient(testAPIServerBlogPodMetadata())
+
+	pod, err := p.findPodByUID(context.Background(), &k8sConfig{}, testPodUID, podReferenceScopeAgentNode)
+	s.RequireGRPCStatusContains(err, codes.FailedPrecondition, "kubelet client is not configured")
+	s.Require().Nil(pod)
+}
+
+func (s *Suite) TestFindPodByName_AgentNodeScopeRequiresKubeletClient() {
+	p := s.newPlugin()
+	p.kubeClient = fakeKubeClientWithSubjectAccessReview(true, nil, testAPIServerBlogPod())
+
+	pod, err := p.findPodByName(context.Background(), &k8sConfig{}, "default", "blog-24ck7", podReferenceScopeAgentNode)
+	s.RequireGRPCStatusContains(err, codes.FailedPrecondition, "kubelet client is not configured")
+	s.Require().Nil(pod)
+}
+
 func (s *Suite) TestAttestReferenceWithPodUID_NotFound() {
 	s.startInsecureKubelet()