From e9e3863b9b7be06c6df75f6a0b384a8e0fdd2c45 Mon Sep 17 00:00:00 2001 From: Carlo Teubner Date: Sat, 6 Jun 2026 15:18:57 +0100 Subject: [PATCH] agent: k8s: coordinate kubelet pod list fetching Move kubelet pod-list fetching, caching and backoff timing into a long-lived fetcher that versions pod-list responses, instead of using the singleflight package. This now enables us to cache successful kubelet responses until another request is eligible to start, which makes more sense. Previously, they were cached for _half_ the polling interval, which could lead to the following situation (illustrated with the default polling interval of 500 ms): - T = 0 ms: Attestation request 1 makes pod-list request, finds no result, sleeps for 500 ms. - T = 300 ms: Attestation request 2 finds cache expired, makes pod-list request that contains an entry that request 1 was looking for. - T = 500 ms: Only now does request 1 wake up and find the desired entry in the cached pod list. Fixing this with the singleflight approach would not have been easily possible, as it could cause retries to re-use the same cached pod list response as in the previous iteration. The new approach also enforces a minimum interval between kubelet request _start times_ across concurrent attestations. Unlike the previous fixed delay after each request, this allows the next request to start immediately when the previous one took longer than the configured interval. In addition, several other improvements (that could also have been implemented with the old singleflight mechanism): - kubelet requests: do not cancel an inflight request just because its _original_ caller is no longer interested, as other callers may be waiting for it. (With singleflight, this could have been fixed via the Group.DoChan method.) - kubelet requests: retry failed fetches via the same backoff mechanism, rather than immediately returning a failure response. Previously, it was only used when the pod list did not contain the workload. - kubelet requests: when reloading the client (whether through explicit Configure call or periodically via reload_interval), only create a fresh http.Transport object if necessary based on config/certificate changes, ensuring we don't trash its connection pool unnecessarily. - kubelet requests: treat invalid "items" JSON type in reponse as error instead of empty. - kubelet requests: replace "attempt" log value rather than appending a new one with each attempt. - kubelet requests: report kubelet and malformed-response failures as Unavailable instead of Internal. (Failure to find the workload in the pod list after attempts exhausted continues to be DeadlineExceeded.) - kubelet requests: use net.JoinHostPort for endpoints (enables IPv6 support). - kube API server requests: correctly reflect runtime changes to api_server.cache.enabled setting via Configure(). - Validate(): more thorough implementation that checks it can read and parse kubelet certificates, etc. - gRPC return codes: preserve whether caller termination was cancellation or deadline expiry. - gRPC return codes: report failed Sigstore policy verification as PermissionDenied instead of Internal. - tests: fix a flakey (order-dependent) Sigstore selector test failure caused by mutating shared expected selectors. - misc: small code cleanups. - docs: add missing descriptions of max_poll_attempts, poll_retry_interval, reload_interval. - docs: assorted fixes. Signed-off-by: Carlo Teubner --- doc/plugin_agent_workloadattestor_k8s.md | 47 +- pkg/agent/plugin/workloadattestor/k8s/k8s.go | 581 +++--------- .../plugin/workloadattestor/k8s/k8s_posix.go | 12 +- .../workloadattestor/k8s/k8s_posix_test.go | 2 +- .../plugin/workloadattestor/k8s/k8s_test.go | 267 ++++-- .../workloadattestor/k8s/kubeletclient.go | 164 ++++ .../workloadattestor/k8s/podlistfetcher.go | 457 ++++++++++ .../k8s/podlistfetcher_test.go | 842 ++++++++++++++++++ 8 files changed, 1854 insertions(+), 518 deletions(-) create mode 100644 pkg/agent/plugin/workloadattestor/k8s/kubeletclient.go create mode 100644 pkg/agent/plugin/workloadattestor/k8s/podlistfetcher.go create mode 100644 pkg/agent/plugin/workloadattestor/k8s/podlistfetcher_test.go diff --git a/doc/plugin_agent_workloadattestor_k8s.md b/doc/plugin_agent_workloadattestor_k8s.md index 8b6cb1535a..1f683843c7 100644 --- a/doc/plugin_agent_workloadattestor_k8s.md +++ b/doc/plugin_agent_workloadattestor_k8s.md @@ -46,24 +46,27 @@ server name validation against the kubelet certificate. **Note** To run on Windows containers, Kubernetes v1.24+ and containerd v1.6+ are required, since [hostprocess](https://kubernetes.io/docs/tasks/configure-pod-container/create-hostprocess-pod/) container is required on the agent container. -| Configuration | Description | -|-----------------------------------------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| -| `disable_container_selectors` | If true, container selectors are not produced. This can be used to produce pod selectors when the workload pod is known but the workload container is not ready at the time of attestation. | -| `enable_namespace_labels` | If true, namespace labels are fetched from the Kubernetes API server and produced as selectors. Requires RBAC `get`, `list` and `watch` on `namespaces`. Disabled by default. | -| `kubelet_read_only_port` | The kubelet read-only port. This is mutually exclusive with `kubelet_secure_port`. | -| `kubelet_secure_port` | The kubelet secure port. It defaults to `10250` unless `kubelet_read_only_port` is set. | -| `kubelet_ca_path` | The path on disk to a file containing CA certificates used to verify the kubelet certificate. Required unless `skip_kubelet_verification` is set. Defaults to the cluster CA bundle `/run/secrets/kubernetes.io/serviceaccount/ca.crt`. | -| `skip_kubelet_verification` | If true, kubelet certificate verification is skipped | -| `token_path` | The path on disk to the bearer token used for kubelet authentication. Defaults to the service account token `/run/secrets/kubernetes.io/serviceaccount/token` | -| `certificate_path` | The path on disk to client certificate used for kubelet authentication | -| `private_key_path` | The path on disk to client key used for kubelet authentication | -| `use_anonymous_authentication` | If true, use anonymous authentication for kubelet communication | -| `node_name_env` | The environment variable used to obtain the node name. Defaults to `MY_NODE_NAME`. | -| `node_name` | The name of the node. Overrides the value obtained by the environment variable specified by `node_name_env`. | -| `experimental` | The experimental options that are subject to change or removal (see below). | -| `sigstore` | Sigstore options. Options described below. See [Sigstore options](#sigstore-options). When set, enables verification of container image signatures and attestations. | -| `use_new_container_locator` | If true, enables the new container locator algorithm that has support for cgroups v2. Defaults to true. | -| `verbose_container_locator_logs` | If true, enables verbose logging of mountinfo and cgroup information used to locate containers. Defaults to false. | +| Configuration | Description | +|-----------------------------------------|---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| +| `disable_container_selectors` | If true, container selectors are not produced. This can be used to produce pod selectors when the workload pod is known but the workload container is not ready at the time of attestation. | +| `enable_namespace_labels` | If true, namespace labels are fetched from the Kubernetes API server and produced as selectors. Requires RBAC `get`, `list` and `watch` on `namespaces`. Disabled by default. | +| `kubelet_read_only_port` | The kubelet read-only port. This is mutually exclusive with `kubelet_secure_port`. | +| `kubelet_secure_port` | The kubelet secure port. It defaults to `10250` unless `kubelet_read_only_port` is set. | +| `kubelet_ca_path` | The path on disk to a file containing CA certificates used to verify the kubelet certificate. Required unless `skip_kubelet_verification` is set. Defaults to the cluster CA bundle `/var/run/secrets/kubernetes.io/serviceaccount/ca.crt`. | +| `skip_kubelet_verification` | If true, kubelet certificate verification is skipped. | +| `token_path` | The path on disk to the bearer token used for kubelet authentication. Defaults to the service account token `/var/run/secrets/kubernetes.io/serviceaccount/token`. | +| `certificate_path` | The path on disk to client certificate used for kubelet authentication. | +| `private_key_path` | The path on disk to client key used for kubelet authentication. | +| `use_anonymous_authentication` | If true, use anonymous authentication for kubelet communication. | +| `node_name_env` | The environment variable used to obtain the node name for kubelet requests. Defaults to `MY_NODE_NAME`. | +| `node_name` | The node name used for kubelet requests. Overrides the value obtained from the env var specified by `node_name_env`. | +| `max_poll_attempts` | The maximum number of kubelet polling attempts while locating the workload container/pod. Defaults to `60`. | +| `poll_retry_interval` | The interval between kubelet polling attempts while locating the workload container/pod. Defaults to `500ms`. | +| `reload_interval` | How often kubelet TLS and token configuration is reloaded from disk. Defaults to `1m`. | +| `experimental` | The experimental options that are subject to change or removal (see below). | +| `sigstore` | Sigstore options. Options described below. See [Sigstore options](#sigstore-options). When set, enables verification of container image signatures and attestations. | +| `use_new_container_locator` | If true, enables the new container locator algorithm that has support for cgroups v2. Defaults to true. | +| `verbose_container_locator_logs` | If true, enables verbose logging of mountinfo and cgroup information used to locate containers. Defaults to false. | These are the current experimental configurations. @@ -121,13 +124,13 @@ Sigstore enabled selectors (available when configured to use `sigstore`) |--------------------------------------------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| | k8s:image-signature:verified | When the image signature was verified and is valid. | | k8s:image-attestations:verified | When the image attestations were verified and are valid. | -| k8s:image-signature-value | The base64 encoded value of the signature (eg. `k8s:image-signature-content:MEUCIQCyem8Gcr0sPFMP7fTXazCN57NcN5+MjxJw9Oo0x2eM+AIgdgBP96BO1Te/NdbjHbUeb0BUye6deRgVtQEv5No5smA=`) | +| k8s:image-signature-value | The base64 encoded value of the signature (eg. `k8s:image-signature-value:MEUCIQCyem8Gcr0sPFMP7fTXazCN57NcN5+MjxJw9Oo0x2eM+AIg...`). | | k8s:image-signature-subject | The OIDC principal that signed the image (e.g., `k8s:image-signature-subject:spirex@example.com`) | | k8s:image-signature-issuer | The OIDC issuer of the signature (e.g., `k8s:image-signature-issuer:https://accounts.google.com`) | | k8s:image-signature-log-id | A unique LogID for the Rekor transparency log entry (eg. `k8s:image-signature-log-id:c0d23d6ad406973f9559f3ba2d1ca01f84147d8ffc5b8445c224f98b95918123`) | | k8s:image-signature-log-index | The log index for the Rekor transparency log entry (eg. `k8s:image-signature-log-index:105695637`) | | k8s:image-signature-integrated-time | The time (in Unix timestamp format) when the image signature was integrated into the signature transparency log (eg. `k8s:image-signature-integrated-time:1719237832`) | -| k8s:image-signature-signed-entry-timestamp | The base64 encoded signed entry (signature over the logID, logIndex, body and integratedTime) (eg. `k8s:image-signature-integrated-time:MEQCIDP77vB0/MEbR1QKZ7Ol8PgFwGEEvnQJiv5cO7ATDYRwAiB9eBLYZjclxRNaaNJVBdQfP9Y8vGVJjwdbisme2cKabc`) | +| k8s:image-signature-signed-entry-timestamp | The base64 encoded signed entry (signature over the logID, logIndex, body and integratedTime) (eg. `k8s:image-signature-signed-entry-timestamp:MEQCIDP77vB0/MEbR1QK...`). | If `ignore_tlog` is set to `true`, the selectors based on the Rekor bundle (`-log-id`, `-log-index`, `-integrated-time`, and `-signed-entry-timestamp`) are not generated. @@ -361,7 +364,7 @@ WorkloadAttestor "k8s" { } ``` -To use the secure kubelet port, verify via `/run/secrets/kubernetes.io/serviceaccount/ca.crt`, and authenticate via the default service account token: +To use the secure kubelet port, verify via `/var/run/secrets/kubernetes.io/serviceaccount/ca.crt`, and authenticate via the default service account token: ```hcl WorkloadAttestor "k8s" { @@ -421,7 +424,7 @@ WorkloadAttestor "k8s" { ## Platform support -This plugin is only supported on Unix systems. +This plugin is supported on Unix systems and Windows hostprocess containers. ## Known issues diff --git a/pkg/agent/plugin/workloadattestor/k8s/k8s.go b/pkg/agent/plugin/workloadattestor/k8s/k8s.go index 76d6bab90a..e9400839ab 100644 --- a/pkg/agent/plugin/workloadattestor/k8s/k8s.go +++ b/pkg/agent/plugin/workloadattestor/k8s/k8s.go @@ -2,18 +2,11 @@ package k8s import ( "context" - "crypto/tls" - "crypto/x509" "encoding/json" "errors" "fmt" - "io" - "net/http" "net/url" "os" - "path/filepath" - "strconv" - "strings" "sync" "time" @@ -28,11 +21,9 @@ import ( "github.com/spiffe/spire/pkg/agent/broker/brokercontext" "github.com/spiffe/spire/pkg/agent/common/sigstore" "github.com/spiffe/spire/pkg/common/catalog" - "github.com/spiffe/spire/pkg/common/pemutil" "github.com/spiffe/spire/pkg/common/pluginconf" "github.com/spiffe/spire/pkg/common/telemetry" "github.com/valyala/fastjson" - "golang.org/x/sync/singleflight" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" "google.golang.org/protobuf/types/known/anypb" @@ -232,33 +223,17 @@ type k8sAPIServerCacheHCLConfig struct { UnusedKeyPositions map[string][]hcltoken.Pos `hcl:",unusedKeyPositions"` } -type k8sAPIServerCacheConfig struct { - Enabled bool -} - // k8sConfig holds the configuration distilled from HCL type k8sConfig struct { - Secure bool - Port int - MaxPollAttempts int - PollRetryInterval time.Duration - SkipKubeletVerification bool - TokenPath string - CertificatePath string - PrivateKeyPath string - UseAnonymousAuthentication bool - KubeletCAPath string - NodeName string - ReloadInterval time.Duration - DisableContainerSelectors bool - EnableNamespaceLabels bool - ContainerHelper ContainerHelper - sigstoreConfig *sigstore.Config - APIServerCache k8sAPIServerCacheConfig - Broker *k8sBrokerConfig - - Client *kubeletClient - LastReload time.Time + MaxPollAttempts int + PollRetryInterval time.Duration + DisableContainerSelectors bool + EnableNamespaceLabels bool + ContainerHelper ContainerHelper + sigstoreConfig *sigstore.Config + APIServerCacheEnabled bool + Broker *k8sBrokerConfig + podListFetcherConfig podListFetcherConfig } func (p *Plugin) buildConfig(coreConfig catalog.CoreConfig, hclText string, status *pluginconf.Status) *k8sConfig { @@ -270,7 +245,19 @@ func (p *Plugin) buildConfig(coreConfig catalog.CoreConfig, hclText string, stat } pluginconf.ReportUnusedKeys(status, newConfig.UnusedKeyPositions) - apiServerCacheConfig, brokerConfig := buildExperimentalConfig(newConfig.Experimental, status) + var apiServerCacheEnabled bool + var brokerConfig *k8sBrokerConfig + if newConfig.Experimental != nil { + pluginconf.ReportUnusedKeys(status, newConfig.Experimental.UnusedKeyPositions) + if newConfig.Experimental.APIServer != nil { + pluginconf.ReportUnusedKeys(status, newConfig.Experimental.APIServer.UnusedKeyPositions) + if newConfig.Experimental.APIServer.Cache != nil { + pluginconf.ReportUnusedKeys(status, newConfig.Experimental.APIServer.Cache.UnusedKeyPositions) + apiServerCacheEnabled = newConfig.Experimental.APIServer.Cache.Enabled + } + } + brokerConfig = buildBrokerConfig("experimental.broker", newConfig.Experimental.Broker, status) + } // Determine max poll attempts with default maxPollAttempts := newConfig.MaxPollAttempts @@ -335,49 +322,39 @@ func (p *Plugin) buildConfig(coreConfig catalog.CoreConfig, hclText string, stat sigstoreConfig = sigstore.NewConfigFromHCL(newConfig.Sigstore, p.log) } - // return the kubelet client - return &k8sConfig{ - Secure: secure, - Port: port, - MaxPollAttempts: maxPollAttempts, - PollRetryInterval: pollRetryInterval, - SkipKubeletVerification: newConfig.SkipKubeletVerification, - TokenPath: newConfig.TokenPath, - CertificatePath: newConfig.CertificatePath, - PrivateKeyPath: newConfig.PrivateKeyPath, - UseAnonymousAuthentication: newConfig.UseAnonymousAuthentication, - KubeletCAPath: newConfig.KubeletCAPath, - NodeName: nodeName, - ReloadInterval: reloadInterval, - DisableContainerSelectors: newConfig.DisableContainerSelectors, - EnableNamespaceLabels: newConfig.EnableNamespaceLabels, - ContainerHelper: containerHelper, - sigstoreConfig: sigstoreConfig, - APIServerCache: apiServerCacheConfig, - Broker: brokerConfig, + kubeletCAPath := newConfig.KubeletCAPath + if kubeletCAPath == "" { + kubeletCAPath = p.defaultKubeletCAPath() } -} - -func buildAPIServerCacheConfig(hclConfig *k8sAPIServerHCLConfig, status *pluginconf.Status) k8sAPIServerCacheConfig { - if hclConfig == nil { - return k8sAPIServerCacheConfig{} - } - pluginconf.ReportUnusedKeys(status, hclConfig.UnusedKeyPositions) - if hclConfig.Cache == nil { - return k8sAPIServerCacheConfig{} - } - pluginconf.ReportUnusedKeys(status, hclConfig.Cache.UnusedKeyPositions) - return k8sAPIServerCacheConfig{ - Enabled: hclConfig.Cache.Enabled, + tokenPath := newConfig.TokenPath + if tokenPath == "" { + tokenPath = p.defaultTokenPath() } -} -func buildExperimentalConfig(hclConfig *k8sExperimentalHCLConfig, status *pluginconf.Status) (k8sAPIServerCacheConfig, *k8sBrokerConfig) { - if hclConfig == nil { - return k8sAPIServerCacheConfig{}, nil + // Return the plugin and pod list fetcher configuration. + return &k8sConfig{ + MaxPollAttempts: maxPollAttempts, + PollRetryInterval: pollRetryInterval, + DisableContainerSelectors: newConfig.DisableContainerSelectors, + EnableNamespaceLabels: newConfig.EnableNamespaceLabels, + ContainerHelper: containerHelper, + sigstoreConfig: sigstoreConfig, + APIServerCacheEnabled: apiServerCacheEnabled, + Broker: brokerConfig, + podListFetcherConfig: podListFetcherConfig{ + pollRetryInterval: pollRetryInterval, + secure: secure, + port: port, + skipKubeletVerification: newConfig.SkipKubeletVerification, + tokenPath: tokenPath, + certificatePath: newConfig.CertificatePath, + privateKeyPath: newConfig.PrivateKeyPath, + useAnonymousAuthentication: newConfig.UseAnonymousAuthentication, + kubeletCAPath: kubeletCAPath, + nodeName: nodeName, + reloadInterval: reloadInterval, + }, } - pluginconf.ReportUnusedKeys(status, hclConfig.UnusedKeyPositions) - return buildAPIServerCacheConfig(hclConfig.APIServer, status), buildBrokerConfig("experimental.broker", hclConfig.Broker, status) } type brokerAccessPolicy string @@ -509,20 +486,22 @@ type Plugin struct { kubeCacheCancel context.CancelFunc kubeCacheDone chan struct{} - cachedPodList map[string]*fastjson.Value - cachedPodListValidUntil time.Time - singleflight singleflight.Group + podListFetcher *podListFetcher } func New() *Plugin { - return &Plugin{ - clock: clock.New(), + pluginClock := clock.New() + p := &Plugin{ + clock: pluginClock, getenv: os.Getenv, } + p.podListFetcher = newPodListFetcher(pluginClock, p.rootDir) + return p } func (p *Plugin) SetLogger(log hclog.Logger) { p.log = log + p.podListFetcher.setLogger(log) } // Attest implements the PID-based workload attestor RPC. PID handling is @@ -605,11 +584,8 @@ func validateKubernetesObjectReference(objRef *broker.KubernetesObjectReference) if objKey == nil && objRef.GetUid() == "" { return status.Error(codes.InvalidArgument, "object reference is missing key and UID") } - if objKey != nil { - name := objKey.GetName() - if name == "" { - return status.Error(codes.InvalidArgument, "object reference key is missing name") - } + if objKey != nil && objKey.GetName() == "" { + return status.Error(codes.InvalidArgument, "object reference key is missing name") } return nil } @@ -644,23 +620,32 @@ func (p *Plugin) attestByPIDReference(ctx context.Context, pid int32) (*attestRe ) // Poll pod information and search for the pod with the container. If - // the pod is not found then delay for a little bit and try again. + // the pod is not found then wait for the fetcher to provide a newer + // result and try again. var scratch []byte + var podListVersion uint64 for attempt := 1; ; attempt++ { - log = log.With(telemetry.Attempt, attempt) + log := log.With(telemetry.Attempt, attempt) - podList, err := p.getPodList(ctx, config.Client, config.PollRetryInterval/2) - if err != nil { - return nil, err + // The pod list fetcher takes care of caching and rate-limiting / backoffing. + podList, podListErr := p.podListFetcher.fetchNext(ctx, podListVersion) + if podListErr != nil { + if ctx.Err() != nil { + return nil, ctx.Err() + } + if errors.Is(podListErr, errPodListFetcherClosed) { + return nil, status.Error(codes.Unavailable, podListErr.Error()) + } + // Otherwise, we'll log podListErr below, and we may retry. + } else { + podListVersion = podList.version } var result *attestReferenceResult - for podKey, podValue := range podList { - if podKnown { - if podKey != string(podUID) { - // The pod holding the container is known. Skip unrelated pods. - continue - } + for podKey, podValue := range podList.pods { + if podKnown && podKey != string(podUID) { + // The pod holding the container is known. Skip unrelated pods. + continue } // Reduce allocations by dumping to the same backing array on @@ -669,7 +654,7 @@ func (p *Plugin) attestByPIDReference(ctx context.Context, pid int32) (*attestRe pod := new(corev1.Pod) if err := json.Unmarshal(scratch, &pod); err != nil { - return nil, status.Errorf(codes.Internal, "unable to decode pod info from kubelet response: %v", err) + return nil, status.Errorf(codes.Unavailable, "unable to decode pod info from kubelet response: %v", err) } var selectorValues []string @@ -695,9 +680,12 @@ func (p *Plugin) attestByPIDReference(ctx context.Context, pid int32) (*attestRe if sigstoreVerifier != nil { log.Debug("Attempting to verify sigstore image signature", "image", containerStatus.Image) - sigstoreSelectors, err := p.sigstoreVerifier.Verify(ctx, containerStatus.ImageID) + sigstoreSelectors, err := sigstoreVerifier.Verify(ctx, containerStatus.ImageID) if err != nil { - return nil, status.Errorf(codes.Internal, "error verifying sigstore image signature for imageID %s: %v", containerStatus.ImageID, err) + if ctx.Err() != nil { + return nil, ctx.Err() + } + return nil, status.Errorf(codes.PermissionDenied, "error verifying sigstore image signature for imageID %s: %v", containerStatus.ImageID, err) } selectorValues = append(selectorValues, sigstoreSelectors...) } @@ -738,18 +726,22 @@ func (p *Plugin) attestByPIDReference(ctx context.Context, pid int32) (*attestRe } // if the container was not located after the maximum number of attempts then the search is over. - if attempt >= config.MaxPollAttempts { + switch { + case attempt >= config.MaxPollAttempts: + if podListErr != nil { + log.Warn("Unable to get pod list; giving up", telemetry.Error, podListErr) + return nil, status.Error(codes.Unavailable, podListErr.Error()) + } log.Warn("Container id not found; giving up") return nil, status.Error(codes.DeadlineExceeded, "no selectors found after max poll attempts") - } - - // wait a bit for containers to initialize before trying again. - log.Debug("Container id not found", telemetry.RetryInterval, config.PollRetryInterval) - - select { - case <-p.clock.After(config.PollRetryInterval): - case <-ctx.Done(): - return nil, status.Errorf(codes.Canceled, "no selectors found: %v", ctx.Err()) + case podListErr != nil: + log.Warn("Unable to get pod list; will retry after backoff", + telemetry.Error, podListErr, + telemetry.RetryInterval, config.PollRetryInterval) + default: + // wait a bit for containers to initialize before trying again. + log.Debug("Container id not found; will retry after backoff", + telemetry.RetryInterval, config.PollRetryInterval) } } } @@ -833,25 +825,25 @@ func (p *Plugin) getBrokerEntryIfPresent(ctx context.Context, config *k8sConfig) // can match pod-specific fields like container images and service accounts // that aren't present on a PartialObjectMetadata. func (p *Plugin) attestByPodReference(ctx context.Context, brokerEntry *k8sBrokerEntry, objRef *broker.KubernetesObjectReference) (*attestReferenceResult, error) { + config, _, _, err := p.getConfig() + if err != nil { + return nil, status.Errorf(codes.Internal, "unable to get config: %v", err) + } + key := objRef.GetKey() namespace := key.GetNamespace() name := key.GetName() uid := types.UID(objRef.GetUid()) - config, _, _, err := p.getConfig() - if err != nil { - return nil, err - } - var pod *corev1.Pod switch { case name != "": if namespace == "" { return nil, ErrNamespaceRequired } - pod, err = p.findPodByName(ctx, config, namespace, name, brokerPodReferenceScope(brokerEntry)) + pod, err = p.findPodByName(ctx, namespace, name, brokerPodReferenceScope(brokerEntry)) default: - pod, err = p.findPodByUID(ctx, config, uid, brokerPodReferenceScope(brokerEntry)) + pod, err = p.findPodByUID(ctx, uid, brokerPodReferenceScope(brokerEntry)) } if err != nil { return nil, err @@ -894,9 +886,9 @@ func brokerPodReferenceScope(brokerEntry *k8sBrokerEntry) podReferenceScope { // 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. -func (p *Plugin) findPodByName(ctx context.Context, config *k8sConfig, namespace, name string, scope podReferenceScope) (*corev1.Pod, error) { +func (p *Plugin) findPodByName(ctx context.Context, 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.fetchKubeletPodList(ctx) if err != nil { return nil, err } @@ -938,9 +930,9 @@ func (p *Plugin) findPodByName(ctx context.Context, config *k8sConfig, namespace // 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) { +func (p *Plugin) findPodByUID(ctx context.Context, 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.fetchKubeletPodList(ctx) if err != nil { return nil, err } @@ -980,6 +972,17 @@ func (p *Plugin) findPodByUID(ctx context.Context, config *k8sConfig, uid types. return pod, nil } +func (p *Plugin) fetchKubeletPodList(ctx context.Context) (map[string]*fastjson.Value, error) { + podList, err := p.podListFetcher.fetchNext(ctx, 0) + if err != nil { + if ctx.Err() != nil { + return nil, ctx.Err() + } + return nil, status.Error(codes.Unavailable, err.Error()) + } + return podList.pods, 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 @@ -1218,6 +1221,14 @@ func (p *Plugin) getOrCreateKubeClient() (client.Client, error) { // used for PartialObjectMetadata. When the API server cache is enabled, only // this client is cache-backed. func (p *Plugin) getOrCreateKubeMetadataClient(ctx context.Context) (client.Client, error) { + config, _, _, err := p.getConfig() + if err != nil { + return nil, err + } + if !config.APIServerCacheEnabled { + return p.getOrCreateKubeClient() + } + p.kubeMu.RLock() if p.kubeMetadataClient != nil { c := p.kubeMetadataClient @@ -1232,28 +1243,6 @@ func (p *Plugin) getOrCreateKubeMetadataClient(ctx context.Context) (client.Clie return p.kubeMetadataClient, nil } - config, _, _, err := p.getConfig() - if err != nil { - return nil, err - } - if !config.APIServerCache.Enabled { - if p.kubeClient != nil { - p.kubeMetadataClient = p.kubeClient - return p.kubeMetadataClient, nil - } - restConfig, clientOptions, err := buildKubeClientOptions() - if err != nil { - return nil, err - } - c, err := client.New(restConfig, clientOptions) - if err != nil { - return nil, fmt.Errorf("unable to create Kubernetes client: %w", err) - } - p.kubeClient = c - p.kubeMetadataClient = c - return c, nil - } - restConfig, clientOptions, err := buildKubeClientOptions() if err != nil { return nil, err @@ -1338,6 +1327,7 @@ func buildKubeClientOptions() (*rest.Config, client.Options, error) { } func (p *Plugin) Close() error { + p.podListFetcher.close() p.kubeMu.Lock() cacheCancel := p.kubeCacheCancel cacheDone := p.kubeCacheDone @@ -1360,10 +1350,6 @@ func (p *Plugin) Configure(ctx context.Context, req *configv1.ConfigureRequest) return nil, err } - if err := p.reloadKubeletClient(newConfig); err != nil { - return nil, err - } - var sigstoreVerifier sigstore.Verifier if newConfig.sigstoreConfig != nil { verifier := sigstore.NewVerifier(newConfig.sigstoreConfig) @@ -1374,6 +1360,13 @@ func (p *Plugin) Configure(ctx context.Context, req *configv1.ConfigureRequest) sigstoreVerifier = verifier } + if err := p.podListFetcher.configure(ctx, newConfig.podListFetcherConfig); err != nil { + if ctx.Err() != nil { + return nil, ctx.Err() + } + return nil, status.Error(codes.InvalidArgument, err.Error()) + } + p.mu.Lock() defer p.mu.Unlock() p.config = newConfig @@ -1384,7 +1377,13 @@ func (p *Plugin) Configure(ctx context.Context, req *configv1.ConfigureRequest) } func (p *Plugin) Validate(_ context.Context, req *configv1.ValidateRequest) (resp *configv1.ValidateResponse, err error) { - _, notes, err := pluginconf.Build(req, p.buildConfig) + newConfig, notes, err := pluginconf.Build(req, p.buildConfig) + if err == nil { + err = p.podListFetcher.validate(newConfig.podListFetcherConfig) + if err != nil { + notes = append(notes, err.Error()) + } + } return &configv1.ValidateResponse{ Valid: err == nil, @@ -1399,196 +1398,15 @@ func (p *Plugin) getConfig() (*k8sConfig, ContainerHelper, sigstore.Verifier, er if p.config == nil { return nil, nil, nil, status.Error(codes.FailedPrecondition, "not configured") } - if err := p.reloadKubeletClient(p.config); err != nil { - p.log.Warn("Unable to load kubelet client", "err", err) - } return p.config, p.containerHelper, p.sigstoreVerifier, nil } -func (p *Plugin) setPodListCache(podList map[string]*fastjson.Value, cacheFor time.Duration) { - p.mu.Lock() - defer p.mu.Unlock() - - p.cachedPodList = podList - p.cachedPodListValidUntil = p.clock.Now().Add(cacheFor) -} - -func (p *Plugin) getPodListCache() map[string]*fastjson.Value { - p.mu.RLock() - defer p.mu.RUnlock() - - if p.clock.Now().Sub(p.cachedPodListValidUntil) >= 0 { - return nil - } - - return p.cachedPodList -} - func (p *Plugin) setContainerHelper(c ContainerHelper) { p.mu.Lock() defer p.mu.Unlock() p.containerHelper = c } -func (p *Plugin) reloadKubeletClient(config *k8sConfig) (err error) { - // The insecure client only needs to be loaded once. - if !config.Secure { - if config.Client == nil { - config.Client = &kubeletClient{ - URL: url.URL{ - Scheme: "http", - Host: fmt.Sprintf("127.0.0.1:%d", config.Port), - }, - } - } - return nil - } - - // Is the client still fresh? - if config.Client != nil && p.clock.Now().Sub(config.LastReload) < config.ReloadInterval { - return nil - } - - tlsConfig := &tls.Config{ - InsecureSkipVerify: config.SkipKubeletVerification, //nolint: gosec // intentionally configurable - } - - var rootCAs *x509.CertPool - if !config.SkipKubeletVerification { - rootCAs, err = p.loadKubeletCA(config.KubeletCAPath) - if err != nil { - return err - } - } - - switch { - case config.SkipKubeletVerification: - - // When contacting the kubelet over localhost, skip the hostname validation. - // Unfortunately Go does not make this straightforward. We disable - // verification but supply a VerifyPeerCertificate that will be called - // with the raw kubelet certs that we can verify directly. - case config.NodeName == "": - tlsConfig.InsecureSkipVerify = true - tlsConfig.SessionTicketsDisabled = true - tlsConfig.VerifyPeerCertificate = func(rawCerts [][]byte, _ [][]*x509.Certificate) error { - var certs []*x509.Certificate - for _, rawCert := range rawCerts { - cert, err := x509.ParseCertificate(rawCert) - if err != nil { - return err - } - certs = append(certs, cert) - } - - // this is improbable. - if len(certs) == 0 { - return errors.New("no certs presented by kubelet") - } - - _, err := certs[0].Verify(x509.VerifyOptions{ - Roots: rootCAs, - Intermediates: newCertPool(certs[1:]), - }) - return err - } - default: - tlsConfig.RootCAs = rootCAs - } - - var token string - switch { - case config.UseAnonymousAuthentication: - // Don't load credentials if using anonymous authentication - case config.CertificatePath != "" && config.PrivateKeyPath != "": - kp, err := p.loadX509KeyPair(config.CertificatePath, config.PrivateKeyPath) - if err != nil { - return err - } - tlsConfig.Certificates = append(tlsConfig.Certificates, *kp) - case config.CertificatePath != "" && config.PrivateKeyPath == "": - return status.Error(codes.InvalidArgument, "the private key path is required with the certificate path") - case config.CertificatePath == "" && config.PrivateKeyPath != "": - return status.Error(codes.InvalidArgument, "the certificate path is required with the private key path") - case config.CertificatePath == "" && config.PrivateKeyPath == "": - token, err = p.loadToken(config.TokenPath) - if err != nil { - return err - } - } - - host := config.NodeName - if host == "" { - host = "127.0.0.1" - } - - config.Client = &kubeletClient{ - Transport: &http.Transport{ - TLSClientConfig: tlsConfig, - }, - URL: url.URL{ - Scheme: "https", - Host: fmt.Sprintf("%s:%d", host, config.Port), - }, - Token: token, - } - config.LastReload = p.clock.Now() - return nil -} - -func (p *Plugin) loadKubeletCA(path string) (*x509.CertPool, error) { - if path == "" { - path = p.defaultKubeletCAPath() - } - caPEM, err := p.readFile(path) - if err != nil { - return nil, status.Errorf(codes.InvalidArgument, "unable to load kubelet CA: %v", err) - } - certs, err := pemutil.ParseCertificates(caPEM) - if err != nil { - return nil, status.Errorf(codes.InvalidArgument, "unable to parse kubelet CA: %v", err) - } - - return newCertPool(certs), nil -} - -func (p *Plugin) loadX509KeyPair(cert, key string) (*tls.Certificate, error) { - certPEM, err := p.readFile(cert) - if err != nil { - return nil, status.Errorf(codes.InvalidArgument, "unable to load certificate: %v", err) - } - keyPEM, err := p.readFile(key) - if err != nil { - return nil, status.Errorf(codes.InvalidArgument, "unable to load private key: %v", err) - } - kp, err := tls.X509KeyPair(certPEM, keyPEM) - if err != nil { - return nil, status.Errorf(codes.InvalidArgument, "unable to load keypair: %v", err) - } - return &kp, nil -} - -func (p *Plugin) loadToken(path string) (string, error) { - if path == "" { - path = p.defaultTokenPath() - } - token, err := p.readFile(path) - if err != nil { - return "", status.Errorf(codes.InvalidArgument, "unable to load token: %v", err) - } - return strings.TrimSpace(string(token)), nil -} - -// readFile reads the contents of a file through the filesystem interface -func (p *Plugin) readFile(path string) ([]byte, error) { - f, err := os.Open(filepath.Join(p.rootDir, path)) - if err != nil { - return nil, err - } - defer f.Close() - return io.ReadAll(f) -} - func (p *Plugin) getNodeName(name string, env string) string { switch { case name != "": @@ -1600,95 +1418,6 @@ func (p *Plugin) getNodeName(name string, env string) string { } } -func (p *Plugin) getPodList(ctx context.Context, client *kubeletClient, cacheFor time.Duration) (map[string]*fastjson.Value, error) { - result := p.getPodListCache() - if result != nil { - return result, nil - } - - podList, err, _ := p.singleflight.Do("podList", func() (any, error) { - result := p.getPodListCache() - if result != nil { - return result, nil - } - - podListBytes, err := client.GetPodList(ctx) - if err != nil { - return nil, err - } - - var parser fastjson.Parser - podList, err := parser.ParseBytes(podListBytes) - if err != nil { - return nil, status.Errorf(codes.Internal, "unable to parse kubelet response: %v", err) - } - - items := podList.GetArray("items") - result = make(map[string]*fastjson.Value, len(items)) - - for _, podValue := range items { - uid := string(podValue.Get("metadata", "uid").GetStringBytes()) - - if uid == "" { - p.log.Warn("Pod has no UID", "pod", podValue) - continue - } - - result[uid] = podValue - } - - p.setPodListCache(result, cacheFor) - - return result, nil - }) - if err != nil { - return nil, err - } - - return podList.(map[string]*fastjson.Value), nil -} - -type kubeletClient struct { - Transport *http.Transport - URL url.URL - Token string -} - -func (c *kubeletClient) GetPodList(ctx context.Context) ([]byte, error) { - ctx, cancel := context.WithTimeout(ctx, 10*time.Second) - defer cancel() - - url := c.URL - url.Path = "/pods" - req, err := http.NewRequestWithContext(ctx, "GET", url.String(), nil) - if err != nil { - return nil, status.Errorf(codes.Internal, "unable to create request: %v", err) - } - if c.Token != "" { - req.Header.Set("Authorization", "Bearer "+c.Token) - } - - client := &http.Client{} - if c.Transport != nil { - client.Transport = c.Transport - } - resp, err := client.Do(req) - if err != nil { - return nil, status.Errorf(codes.Internal, "unable to perform request: %v", err) - } - defer resp.Body.Close() - - if resp.StatusCode != http.StatusOK { - return nil, status.Errorf(codes.Internal, "unexpected status code on pods response: %d %s", resp.StatusCode, tryRead(resp.Body)) - } - - out, err := io.ReadAll(resp.Body) - if err != nil { - return nil, status.Errorf(codes.Internal, "unable to read pods response: %v", err) - } - return out, nil -} - func lookUpContainerInPod(containerID string, status corev1.PodStatus, log hclog.Logger) (*corev1.ContainerStatus, bool) { for _, status := range status.ContainerStatuses { // TODO: should we be keying off of the status or is the lack of a @@ -1735,7 +1464,7 @@ func lookUpContainerInPod(containerID string, status corev1.PodStatus, log hclog func getPodImageIdentifiers(containerStatuses ...corev1.ContainerStatus) map[string]struct{} { // Map is used purely to exclude duplicate selectors, value is unused. - podImages := make(map[string]struct{}) + podImages := make(map[string]struct{}, 2*len(containerStatuses)) // Note that for each pod image we generate *2* matching selectors. // This is to support matching against ImageID, which has a SHA // docker.io/envoyproxy/envoy-alpine@sha256:bf862e5f5eca0a73e7e538224578c5cf867ce2be91b5eaed22afc153c00363eb @@ -1786,8 +1515,8 @@ func getSelectorValuesFromPodInfo(pod *corev1.Pod) []string { fmt.Sprintf("node-name:%s", pod.Spec.NodeName), fmt.Sprintf("pod-uid:%s", pod.UID), fmt.Sprintf("pod-name:%s", pod.Name), - fmt.Sprintf("pod-image-count:%s", strconv.Itoa(len(pod.Status.ContainerStatuses))), - fmt.Sprintf("pod-init-image-count:%s", strconv.Itoa(len(pod.Status.InitContainerStatuses))), + fmt.Sprintf("pod-image-count:%d", len(pod.Status.ContainerStatuses)), + fmt.Sprintf("pod-init-image-count:%d", len(pod.Status.InitContainerStatuses)), } for podImage := range getPodImageIdentifiers(pod.Status.ContainerStatuses...) { @@ -1815,17 +1544,3 @@ func getSelectorValuesFromWorkloadContainerStatus(status *corev1.ContainerStatus } return selectorValues } - -func tryRead(r io.Reader) string { - buf := make([]byte, 1024) - n, _ := r.Read(buf) - return string(buf[:n]) -} - -func newCertPool(certs []*x509.Certificate) *x509.CertPool { - certPool := x509.NewCertPool() - for _, cert := range certs { - certPool.AddCert(cert) - } - return certPool -} diff --git a/pkg/agent/plugin/workloadattestor/k8s/k8s_posix.go b/pkg/agent/plugin/workloadattestor/k8s/k8s_posix.go index 054f41bbdb..c55a007817 100644 --- a/pkg/agent/plugin/workloadattestor/k8s/k8s_posix.go +++ b/pkg/agent/plugin/workloadattestor/k8s/k8s_posix.go @@ -47,7 +47,7 @@ func (h *containerHelper) Configure(config *HCLConfig, log hclog.Logger) error { if h.useNewContainerLocator { log.Info("Using the new container locator") } else { - log.Warn("Using the legacy container locator. This option will removed in a future release.") + log.Warn("Using the legacy container locator. This option will be removed in a future release.") } return nil @@ -143,13 +143,7 @@ func reSubMatchMap(r *regexp.Regexp, str string) map[string]string { } func isValidCGroupPathMatches(matches map[string]string) bool { - if matches == nil { - return false - } - if matches["mustnotmatch"] != "" { - return false - } - return true + return matches != nil && matches["mustnotmatch"] == "" } func getPodUIDAndContainerIDFromCGroupPath(cgroupPath string) (types.UID, string, bool) { @@ -193,7 +187,7 @@ func getPodUIDAndContainerIDFromCGroupPath(cgroupPath string) (types.UID, string func canonicalizePodUID(uid string) types.UID { return types.UID(strings.Map(func(r rune) rune { if unicode.IsPunct(r) { - r = '-' + return '-' } return r }, uid)) diff --git a/pkg/agent/plugin/workloadattestor/k8s/k8s_posix_test.go b/pkg/agent/plugin/workloadattestor/k8s/k8s_posix_test.go index 483db73b23..6969b4368c 100644 --- a/pkg/agent/plugin/workloadattestor/k8s/k8s_posix_test.go +++ b/pkg/agent/plugin/workloadattestor/k8s/k8s_posix_test.go @@ -203,7 +203,7 @@ func (s *Suite) requireAttestSuccessWithCrioPod(p workloadattestor.WorkloadAttes func (s *Suite) requireAttestFailWithDuplicateContainerID(p workloadattestor.WorkloadAttestor) { s.addPodListResponse(crioPodListDuplicateContainerIDFilePath) s.addCgroupsResponse(cgPidInCrioPodFilePath) - s.requireAttestFailure(p, "two pods found with same container Id") + s.requireAttestFailure(p, codes.Internal, "two pods found with same container Id") } func (s *Suite) requireAttestSuccessWithPodSystemdCgroups(p workloadattestor.WorkloadAttestor) { diff --git a/pkg/agent/plugin/workloadattestor/k8s/k8s_test.go b/pkg/agent/plugin/workloadattestor/k8s/k8s_test.go index e9b69e78c4..65478b1364 100644 --- a/pkg/agent/plugin/workloadattestor/k8s/k8s_test.go +++ b/pkg/agent/plugin/workloadattestor/k8s/k8s_test.go @@ -1,6 +1,7 @@ package k8s import ( + "bytes" "context" "crypto/ecdsa" "crypto/rand" @@ -15,13 +16,17 @@ import ( "os" "path/filepath" "slices" + "strings" "sync" + "sync/atomic" "testing" "time" + "github.com/hashicorp/go-hclog" "github.com/spiffe/go-spiffe/v2/exp/proto/spiffe/broker" "github.com/spiffe/go-spiffe/v2/spiffeid" workloadattestorv1 "github.com/spiffe/spire-plugin-sdk/proto/spire/plugin/agent/workloadattestor/v1" + configv1 "github.com/spiffe/spire-plugin-sdk/proto/spire/service/common/config/v1" "github.com/spiffe/spire/pkg/agent/broker/brokercontext" "github.com/spiffe/spire/pkg/agent/common/sigstore" "github.com/spiffe/spire/pkg/agent/plugin/workloadattestor" @@ -112,6 +117,12 @@ type attestResult struct { err error } +// trackedKubeClient gives tests a pointer identity to compare without +// comparing controller-runtime fake client internals, which are not comparable. +type trackedKubeClient struct { + client.Client +} + func TestPlugin(t *testing.T) { spiretest.Run(t, new(Suite)) } @@ -223,9 +234,9 @@ func (s *Suite) TestAttestWithPidInPodAfterRetry() { resultCh := s.goAttest(p) - s.clock.WaitForAfter(time.Minute, "waiting for retry timer") + s.clock.WaitForTimer(time.Minute, "waiting for retry timer") s.clock.Add(testPollRetryInterval) - s.clock.WaitForAfter(time.Minute, "waiting for retry timer") + s.clock.WaitForTimer(time.Minute, "waiting for retry timer") s.clock.Add(testPollRetryInterval) select { @@ -237,6 +248,35 @@ func (s *Suite) TestAttestWithPidInPodAfterRetry() { } } +func (s *Suite) TestAttestRetriesTransientKubeletError() { + var requestCount atomic.Int32 + s.setServer(httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) { + if requestCount.Add(1) == 1 { + http.Error(w, "try again", http.StatusServiceUnavailable) + return + } + s.serveHTTP(w, req) + }))) + p := s.loadInsecurePlugin() + s.addPodListResponse(podListFilePath) + s.addGetContainerResponsePidInPod() + + resultCh := s.goAttest(p) + + s.clock.WaitForTimer(time.Minute, "waiting for retry timer") + s.Require().EqualValues(1, requestCount.Load()) + s.clock.Add(testPollRetryInterval) + + select { + case result := <-resultCh: + s.Require().NoError(result.err) + s.requireSelectorsEqual(testPodAndContainerSelectors, result.selectors) + s.Require().EqualValues(2, requestCount.Load()) + case <-time.After(time.Minute): + s.FailNow("timed out waiting for attest response") + } +} + func (s *Suite) TestAttestWithPidNotInPodCancelsEarly() { s.startInsecureKubelet() p := s.loadInsecurePlugin() @@ -251,6 +291,27 @@ func (s *Suite) TestAttestWithPidNotInPodCancelsEarly() { s.Require().Nil(selectors) } +func (s *Suite) TestAttestFailsFastWhenPodListFetcherClosed() { + var logs bytes.Buffer + p := s.newPlugin() + p.log = hclog.New(&hclog.LoggerOptions{ + Level: hclog.Warn, + Output: &logs, + }) + p.config = &k8sConfig{ + MaxPollAttempts: 5, + PollRetryInterval: testPollRetryInterval, + } + p.containerHelper = s.oc.getContainerHelper(p) + s.addGetContainerResponsePidInPod() + p.podListFetcher.close() + + resp, err := p.Attest(s.T().Context(), &workloadattestorv1.AttestRequest{Pid: pid}) + s.Require().Nil(resp) + s.RequireGRPCStatus(err, codes.Unavailable, errPodListFetcherClosed.Error()) + s.Require().NotContains(logs.String(), "will retry") +} + func (s *Suite) TestAttestPodListCache() { s.startInsecureKubelet() p := s.loadInsecurePlugin() @@ -270,7 +331,7 @@ func (s *Suite) TestAttestPodListCache() { s.Require().Equal(1, s.podListResponseCount()) // Now expire the cache, attest, and observe the last listing was consumed. - s.clock.Add(testPollRetryInterval / 2) + s.clock.Add(testPollRetryInterval) s.requireAttestSuccess(p, testPodAndContainerSelectors) s.Require().Equal(0, s.podListResponseCount()) } @@ -287,13 +348,13 @@ func (s *Suite) TestAttestWithPidNotInPodAfterRetry() { resultCh := s.goAttest(p) - s.clock.WaitForAfter(time.Minute, "waiting for retry timer") + s.clock.WaitForTimer(time.Minute, "waiting for retry timer") s.clock.Add(testPollRetryInterval) - s.clock.WaitForAfter(time.Minute, "waiting for retry timer") + s.clock.WaitForTimer(time.Minute, "waiting for retry timer") s.clock.Add(testPollRetryInterval) - s.clock.WaitForAfter(time.Minute, "waiting for retry timer") + s.clock.WaitForTimer(time.Minute, "waiting for retry timer") s.clock.Add(testPollRetryInterval) - s.clock.WaitForAfter(time.Minute, "waiting for retry timer") + s.clock.WaitForTimer(time.Minute, "waiting for retry timer") s.clock.Add(testPollRetryInterval) select { @@ -310,14 +371,16 @@ func (s *Suite) TestAttestOverSecurePortViaTokenAuth() { s.startSecureKubeletWithTokenAuth(true, "default-token") // use the service account token for auth - p := s.loadSecurePlugin(``) + p := s.loadSecurePlugin(` + max_poll_attempts = 1 + `) s.requireAttestSuccessWithPod(p) // write out a different token and make sure it is picked up on reload s.writeFile(defaultTokenPath, "bad-token") s.clock.Add(defaultReloadInterval) - s.requireAttestFailure(p, `expected "Bearer default-token", got "Bearer bad-token"`) + s.requireAttestFailure(p, codes.Unavailable, `expected "Bearer default-token", got "Bearer bad-token"`) } func (s *Suite) TestAttestOverSecurePortViaClientAuth() { @@ -328,6 +391,7 @@ func (s *Suite) TestAttestOverSecurePortViaClientAuth() { p := s.loadSecurePlugin(` certificate_path = "cert.pem" private_key_path = "key.pem" + max_poll_attempts = 1 `) s.requireAttestSuccessWithPod(p) @@ -337,7 +401,7 @@ func (s *Suite) TestAttestOverSecurePortViaClientAuth() { s.writeCert(certPath, clientCert) s.clock.Add(defaultReloadInterval) - s.requireAttestFailure(p, "remote error: tls") + s.requireAttestFailure(p, codes.Unavailable, "remote error: tls") } func (s *Suite) TestAttestOverSecurePortViaAnonymousAuth() { @@ -431,7 +495,7 @@ func (s *Suite) TestAttestWithNamespaceLabelsErrorFailsAttestation() { s.addPodListResponse(podListFilePath) s.addGetContainerResponsePidInPod() - s.requireAttestFailure(p, "unable to get namespace labels") + s.requireAttestFailure(p, codes.Internal, "unable to get namespace labels") } func (s *Suite) TestAttestWithSigstoreSelectors() { @@ -439,12 +503,46 @@ func (s *Suite) TestAttestWithSigstoreSelectors() { p := s.loadInsecurePluginWithSigstore() // Add the expected selectors from the Sigstore verifier - testPodAndContainerSelectors = append(testPodAndContainerSelectors, sigstoreSelectors...) + expectedSelectors := slices.Concat(testPodAndContainerSelectors, sigstoreSelectors) s.addPodListResponse(podListFilePath) s.addGetContainerResponsePidInPod() - s.requireAttestSuccess(p, testPodAndContainerSelectors) + s.requireAttestSuccess(p, expectedSelectors) +} + +func (s *Suite) TestValidate() { + p := s.newPlugin() + p.SetLogger(hclog.NewNullLogger()) + s.T().Cleanup(func() { + s.Require().NoError(p.Close()) + }) + + resp, err := p.Validate(s.T().Context(), &configv1.ValidateRequest{ + CoreConfiguration: &configv1.CoreConfiguration{ + TrustDomain: "example.org", + }, + HclConfiguration: "kubelet_read_only_port = 10255", + }) + s.Require().NoError(err) + s.Require().True(resp.Valid) + s.Require().Nil(p.podListFetcher.client) + s.Require().Nil(p.podListFetcher.config) + + resp, err = p.Validate(s.T().Context(), &configv1.ValidateRequest{ + CoreConfiguration: &configv1.CoreConfiguration{ + TrustDomain: "example.org", + }, + HclConfiguration: ` + skip_kubelet_verification = true + token_path = "no-such-file" + `, + }) + s.Require().NoError(err) + s.Require().False(resp.Valid) + s.Require().Contains(strings.Join(resp.Notes, " "), "unable to load token") + s.Require().Nil(p.podListFetcher.client) + s.Require().Nil(p.podListFetcher.config) } func (s *Suite) TestConfigure() { @@ -459,16 +557,16 @@ func (s *Suite) TestConfigure() { s.writeCert("some-other-ca", s.kubeletCert) type config struct { - Insecure bool - VerifyKubelet bool - HasNodeName bool - Token string - KubeletURL string - MaxPollAttempts int - PollRetryInterval time.Duration - ReloadInterval time.Duration - SigstoreConfig *sigstore.Config - APIServerCache bool + Insecure bool + VerifyKubelet bool + HasNodeName bool + Token string + KubeletURL string + MaxPollAttempts int + PollRetryInterval time.Duration + ReloadInterval time.Duration + SigstoreConfig *sigstore.Config + APIServerCacheEnabled bool } testCases := []struct { @@ -587,12 +685,12 @@ func (s *Suite) TestConfigure() { } `, config: &config{ - Insecure: true, - KubeletURL: "http://127.0.0.1:12345", - MaxPollAttempts: defaultMaxPollAttempts, - PollRetryInterval: defaultPollRetryInterval, - ReloadInterval: defaultReloadInterval, - APIServerCache: true, + Insecure: true, + KubeletURL: "http://127.0.0.1:12345", + MaxPollAttempts: defaultMaxPollAttempts, + PollRetryInterval: defaultPollRetryInterval, + ReloadInterval: defaultReloadInterval, + APIServerCacheEnabled: true, }, }, { @@ -762,35 +860,98 @@ func (s *Suite) TestConfigure() { c, _, _, err := p.getConfig() require.NoError(t, err) + require.Equal(t, c.podListFetcherConfig, *p.podListFetcher.config) + client := p.podListFetcher.client switch { case testCase.config.Insecure: - assert.Nil(t, c.Client.Transport) - case !assert.NotNil(t, c.Client.Transport): - case !assert.NotNil(t, c.Client.Transport.TLSClientConfig): + assert.Nil(t, client.transport) + case !assert.NotNil(t, client.transport): + case !assert.NotNil(t, client.transport.TLSClientConfig): case !testCase.config.VerifyKubelet: - assert.True(t, c.Client.Transport.TLSClientConfig.InsecureSkipVerify) - assert.Nil(t, c.Client.Transport.TLSClientConfig.VerifyPeerCertificate) + assert.True(t, client.transport.TLSClientConfig.InsecureSkipVerify) + assert.Nil(t, client.transport.TLSClientConfig.VerifyPeerCertificate) default: if testCase.config.HasNodeName { - if assert.NotNil(t, c.Client.Transport.TLSClientConfig.RootCAs) { - assert.True(t, c.Client.Transport.TLSClientConfig.RootCAs.Equal(kubeletCertPool)) + if assert.NotNil(t, client.transport.TLSClientConfig.RootCAs) { + assert.True(t, client.transport.TLSClientConfig.RootCAs.Equal(kubeletCertPool)) } } else { - assert.True(t, c.Client.Transport.TLSClientConfig.InsecureSkipVerify) - assert.NotNil(t, c.Client.Transport.TLSClientConfig.VerifyPeerCertificate) + assert.True(t, client.transport.TLSClientConfig.InsecureSkipVerify) + assert.NotNil(t, client.transport.TLSClientConfig.VerifyPeerCertificate) } } - assert.Equal(t, testCase.config.Token, c.Client.Token) - assert.Equal(t, testCase.config.KubeletURL, c.Client.URL.String()) + assert.Equal(t, testCase.config.Token, client.token) + assert.Equal(t, testCase.config.KubeletURL, client.endpoint.String()) assert.Equal(t, testCase.config.MaxPollAttempts, c.MaxPollAttempts) assert.Equal(t, testCase.config.PollRetryInterval, c.PollRetryInterval) - assert.Equal(t, testCase.config.ReloadInterval, c.ReloadInterval) - assert.Equal(t, testCase.config.APIServerCache, c.APIServerCache.Enabled) + assert.Equal(t, testCase.config.ReloadInterval, c.podListFetcherConfig.reloadInterval) + assert.Equal(t, testCase.config.APIServerCacheEnabled, c.APIServerCacheEnabled) }) } } +func (s *Suite) TestGetOrCreateKubeMetadataClientReturnsLiveClientWhenAPIServerCacheDisabled() { + p := s.newPlugin() + p.SetLogger(hclog.NewNullLogger()) + s.T().Cleanup(func() { + s.Require().NoError(p.Close()) + }) + + liveClient := &trackedKubeClient{Client: fakeKubeClientWithSubjectAccessReview(false, nil)} + metadataClient := &trackedKubeClient{Client: fakeKubeMetadataClient()} + + _, err := p.Configure(s.T().Context(), &configv1.ConfigureRequest{ + CoreConfiguration: &configv1.CoreConfiguration{ + TrustDomain: "example.org", + }, + HclConfiguration: ` + kubelet_read_only_port = 12345 + `, + }) + s.Require().NoError(err) + + p.kubeClient = liveClient + p.kubeMetadataClient = metadataClient + + c, err := p.getOrCreateKubeMetadataClient(s.T().Context()) + s.Require().NoError(err) + assert.Same(s.T(), liveClient, c) +} + +func (s *Suite) TestGetOrCreateKubeMetadataClientReturnsMetadataClientWhenAPIServerCacheEnabled() { + p := s.newPlugin() + p.SetLogger(hclog.NewNullLogger()) + s.T().Cleanup(func() { + s.Require().NoError(p.Close()) + }) + + metadataClient := &trackedKubeClient{Client: fakeKubeMetadataClient()} + + _, err := p.Configure(s.T().Context(), &configv1.ConfigureRequest{ + CoreConfiguration: &configv1.CoreConfiguration{ + TrustDomain: "example.org", + }, + HclConfiguration: ` + kubelet_read_only_port = 12345 + experimental { + api_server { + cache { + enabled = true + } + } + } + `, + }) + s.Require().NoError(err) + + p.kubeMetadataClient = metadataClient + + c, err := p.getOrCreateKubeMetadataClient(s.T().Context()) + s.Require().NoError(err) + assert.Same(s.T(), metadataClient, c) +} + func (s *Suite) TestConfigureBroker() { testCases := []struct { name string @@ -1081,14 +1242,14 @@ func (s *Suite) TestConfigureWithSigstore() { } func (s *Suite) newPlugin() *Plugin { - p := New() - p.rootDir = s.dir - p.clock = s.clock - p.getenv = func(key string) string { - return s.env[key] + return &Plugin{ + rootDir: s.dir, + clock: s.clock, + getenv: func(key string) string { + return s.env[key] + }, + podListFetcher: newPodListFetcher(s.clock, s.dir), } - - return p } func (s *Suite) setServer(server *httptest.Server) { @@ -1100,8 +1261,8 @@ func (s *Suite) setServer(server *httptest.Server) { func (s *Suite) writeFile(path, data string) { realPath := filepath.Join(s.dir, path) - s.Require().NoError(os.MkdirAll(filepath.Dir(realPath), 0755)) - s.Require().NoError(os.WriteFile(realPath, []byte(data), 0600)) + s.Require().NoError(os.MkdirAll(filepath.Dir(realPath), 0o755)) + s.Require().NoError(os.WriteFile(realPath, []byte(data), 0o600)) } func (s *Suite) serveHTTP(w http.ResponseWriter, _ *http.Request) { @@ -1519,9 +1680,9 @@ func (s *Suite) requireAttestSuccess(p workloadattestor.WorkloadAttestor, expect s.requireSelectorsEqual(expectedSelectors, selectors) } -func (s *Suite) requireAttestFailure(p workloadattestor.WorkloadAttestor, contains string) { +func (s *Suite) requireAttestFailure(p workloadattestor.WorkloadAttestor, code codes.Code, contains string) { selectors, err := p.Attest(context.Background(), pid) - s.RequireGRPCStatusContains(err, codes.Internal, contains) + s.RequireGRPCStatusContains(err, code, contains) s.Require().Nil(selectors) } diff --git a/pkg/agent/plugin/workloadattestor/k8s/kubeletclient.go b/pkg/agent/plugin/workloadattestor/k8s/kubeletclient.go new file mode 100644 index 0000000000..9bfdbe5b0b --- /dev/null +++ b/pkg/agent/plugin/workloadattestor/k8s/kubeletclient.go @@ -0,0 +1,164 @@ +package k8s + +import ( + "context" + "crypto/tls" + "crypto/x509" + "errors" + "fmt" + "io" + "net" + "net/http" + "net/url" + "strconv" + + "github.com/spiffe/spire/pkg/common/pemutil" +) + +// kubeletTransportConfig holds the config items from which the http.Transport object +// is created; if any of them changes, a new transport must be created. +// The relevant logic (in podListFetcher) compares this struct using '==', hence +// care must be taken when making changes to the set of fields included here. +type kubeletTransportConfig struct { + secure bool + skipKubeletVerification bool + nodeName string + port int + caPEM string + certificatePEM string + privateKeyPEM string +} + +type kubeletClient struct { + transport *http.Transport + transportConfig kubeletTransportConfig + endpoint url.URL + token string +} + +func newKubeletClient(config kubeletTransportConfig, token string) (*kubeletClient, error) { + if !config.secure { + return &kubeletClient{ + transportConfig: config, + endpoint: url.URL{ + Scheme: "http", + Host: net.JoinHostPort("127.0.0.1", strconv.Itoa(config.port)), + }, + }, nil + } + + tlsConfig := &tls.Config{ + InsecureSkipVerify: config.skipKubeletVerification, //nolint: gosec // intentionally configurable + } + + var rootCAs *x509.CertPool + if !config.skipKubeletVerification { + certs, err := pemutil.ParseCertificates([]byte(config.caPEM)) + if err != nil { + return nil, fmt.Errorf("unable to parse kubelet CA: %w", err) + } + rootCAs = newCertPool(certs) + } + + switch { + case config.skipKubeletVerification: + case config.nodeName == "": + tlsConfig.InsecureSkipVerify = true + tlsConfig.SessionTicketsDisabled = true + tlsConfig.VerifyPeerCertificate = verifyKubeletCertificate(rootCAs) + default: + tlsConfig.RootCAs = rootCAs + } + + if config.certificatePEM != "" { + kp, err := tls.X509KeyPair([]byte(config.certificatePEM), []byte(config.privateKeyPEM)) + if err != nil { + return nil, fmt.Errorf("unable to load keypair: %w", err) + } + tlsConfig.Certificates = append(tlsConfig.Certificates, kp) + } + + host := config.nodeName + if host == "" { + host = "127.0.0.1" + } + + return &kubeletClient{ + transport: &http.Transport{TLSClientConfig: tlsConfig}, + transportConfig: config, + endpoint: url.URL{ + Scheme: "https", + Host: net.JoinHostPort(host, strconv.Itoa(config.port)), + }, + token: token, + }, nil +} + +func (c *kubeletClient) getPodList(ctx context.Context) ([]byte, error) { + url := c.endpoint + url.Path = "/pods" + req, err := http.NewRequestWithContext(ctx, http.MethodGet, url.String(), nil) + if err != nil { + return nil, fmt.Errorf("unable to create request: %w", err) + } + if c.token != "" { + req.Header.Set("Authorization", "Bearer "+c.token) + } + + client := &http.Client{} + if c.transport != nil { + client.Transport = c.transport + } + resp, err := client.Do(req) + if err != nil { + return nil, fmt.Errorf("unable to perform request: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("unexpected status code on pods response: %d %s", resp.StatusCode, tryRead(resp.Body)) + } + + out, err := io.ReadAll(resp.Body) + if err != nil { + return nil, fmt.Errorf("unable to read pods response: %w", err) + } + return out, nil +} + +func tryRead(r io.Reader) string { + buf := make([]byte, 1024) + n, _ := r.Read(buf) + return string(buf[:n]) +} + +func newCertPool(certs []*x509.Certificate) *x509.CertPool { + certPool := x509.NewCertPool() + for _, cert := range certs { + certPool.AddCert(cert) + } + return certPool +} + +func verifyKubeletCertificate(rootCAs *x509.CertPool) func([][]byte, [][]*x509.Certificate) error { + return func(rawCerts [][]byte, _ [][]*x509.Certificate) error { + var certs []*x509.Certificate + for _, rawCert := range rawCerts { + cert, err := x509.ParseCertificate(rawCert) + if err != nil { + return err + } + certs = append(certs, cert) + } + + if len(certs) == 0 { + return errors.New("no certs presented by kubelet") + } + + _, err := certs[0].Verify(x509.VerifyOptions{ + Roots: rootCAs, + Intermediates: newCertPool(certs[1:]), + }) + return err + } +} diff --git a/pkg/agent/plugin/workloadattestor/k8s/podlistfetcher.go b/pkg/agent/plugin/workloadattestor/k8s/podlistfetcher.go new file mode 100644 index 0000000000..9581c55b49 --- /dev/null +++ b/pkg/agent/plugin/workloadattestor/k8s/podlistfetcher.go @@ -0,0 +1,457 @@ +package k8s + +import ( + "context" + "errors" + "fmt" + "os" + "path/filepath" + "strings" + "sync" + "time" + + "github.com/andres-erbsen/clock" + "github.com/hashicorp/go-hclog" + "github.com/valyala/fastjson" +) + +const kubeletRequestTimeout = 10 * time.Second + +var errPodListFetcherClosed = errors.New("pod list fetcher is closed") + +type podListFetcherConfig struct { + pollRetryInterval time.Duration + secure bool + port int + skipKubeletVerification bool + tokenPath string + certificatePath string + privateKeyPath string + useAnonymousAuthentication bool + kubeletCAPath string + nodeName string + reloadInterval time.Duration +} + +// podListFetcher coordinates access to the kubelet pod list. Concurrent callers +// share at most one kubelet request, with request starts separated by at least +// the configured poll interval. Successful results are cached until another +// request is eligible to start and assigned monotonically increasing versions +// so callers can wait for a pod list newer than one they have already examined. +// +// The run goroutine exclusively owns the mutable state identified below. +// Public methods and background operations communicate with it through +// channels. +type podListFetcher struct { + clock clock.Clock + log hclog.Logger + rootDir string + + actionCh chan func() + stopCh chan struct{} + wg sync.WaitGroup + + // The fields below are owned by the run goroutine. + config *podListFetcherConfig + client *kubeletClient + clientLoadedAt time.Time + lastFetchStart time.Time + fetchTimer *clock.Timer + fetchWaiters map[chan<- podListFetchResult]struct{} + fetchCancel context.CancelFunc + cachedPodList versionedPodList + cachedFetchStart time.Time + + // These are callbacks purely to facilitate testing. + fetch func(context.Context, *kubeletClient) (map[string]*fastjson.Value, error) + buildClient func(podListFetcherConfig, *kubeletClient) (*kubeletClient, error) +} + +type versionedPodList struct { + pods map[string]*fastjson.Value + version uint64 +} + +type podListFetchResult struct { + versionedPodList + err error +} + +// newPodListFetcher constructs a pod list fetcher and starts its run goroutine. +// The caller must eventually call close exactly once. +func newPodListFetcher(clock clock.Clock, rootDir string) *podListFetcher { + f := podListFetcher{ + clock: clock, + log: hclog.NewNullLogger(), + rootDir: rootDir, + actionCh: make(chan func()), + stopCh: make(chan struct{}), + fetchWaiters: make(map[chan<- podListFetchResult]struct{}), + } + f.fetch = f.fetchPodList + f.buildClient = f.buildKubeletClient + f.wg.Go(f.run) + return &f +} + +// setLogger replaces the logger used while processing kubelet responses. It +// must be called before the fetcher is otherwise used. +func (f *podListFetcher) setLogger(log hclog.Logger) { + f.log = log +} + +// configure recreates the internal kubelet client based on config. A failure +// leaves the current configuration unchanged. Calls to configure must not +// overlap. +func (f *podListFetcher) configure(ctx context.Context, config podListFetcherConfig) error { + resultCh := make(chan error, 1) + select { + case f.actionCh <- func() { f.startConfigure(config, resultCh) }: + case <-ctx.Done(): + return ctx.Err() + case <-f.stopCh: + return errPodListFetcherClosed + } + + // Once accepted, wait for the definitive result so configuration cannot + // be installed after the caller observes a context cancellation. + select { + case err := <-resultCh: + return err + case <-f.stopCh: + return errPodListFetcherClosed + } +} + +// validate validates the given config. +func (f *podListFetcher) validate(config podListFetcherConfig) error { + _, err := f.buildClient(config, nil) + return err +} + +// fetchNext returns a cached result newer than afterVersion while that result is +// fresh, or waits for the next shared kubelet request. +func (f *podListFetcher) fetchNext(ctx context.Context, afterVersion uint64) (versionedPodList, error) { + resultCh := make(chan podListFetchResult, 1) + + select { + case f.actionCh <- func() { f.registerFetchRequest(afterVersion, resultCh) }: + case <-ctx.Done(): + return versionedPodList{}, ctx.Err() + case <-f.stopCh: + return versionedPodList{}, errPodListFetcherClosed + } + + select { + case result := <-resultCh: + return result.versionedPodList, result.err + case <-ctx.Done(): + f.dispatch(func() { f.cancelFetchRequest(resultCh) }) + return versionedPodList{}, ctx.Err() + case <-f.stopCh: + return versionedPodList{}, errPodListFetcherClosed + } +} + +// close stops the run goroutine, cancels any in-flight kubelet request, and +// waits for all background operations to finish. It must be called exactly once. +func (f *podListFetcher) close() { + close(f.stopCh) + f.wg.Wait() +} + +func (f *podListFetcher) run() { + for { + select { + case action := <-f.actionCh: + action() + case <-f.timerChan(): + f.fetchTimerFired() + case <-f.stopCh: + f.stop() + return + } + } +} + +func (f *podListFetcher) dispatch(action func()) { + select { + case f.actionCh <- action: + case <-f.stopCh: + } +} + +func (f *podListFetcher) registerFetchRequest(afterVersion uint64, resultCh chan<- podListFetchResult) { + if f.cachedPodList.pods != nil && afterVersion < f.cachedPodList.version && + f.config != nil && f.clock.Now().Before(f.cachedFetchStart.Add(f.config.pollRetryInterval)) { + resultCh <- podListFetchResult{versionedPodList: f.cachedPodList} + return + } + + f.fetchWaiters[resultCh] = struct{}{} + f.scheduleFetch() +} + +func (f *podListFetcher) cancelFetchRequest(resultCh chan<- podListFetchResult) { + delete(f.fetchWaiters, resultCh) + if len(f.fetchWaiters) == 0 { + f.stopTimer() + // Could also call fetchCancel here, but another request + // might still come in and then see the cancellation. + // Also, future requests may benefit from having a cached result. + } +} + +func (f *podListFetcher) completeFetch(result podListFetchResult, originalClient, reloadedClient *kubeletClient) { + f.fetchCancel = nil + + if reloadedClient != nil && f.client == originalClient { + // Only install the new kubelet client if the client from which it was + // built is still installed. This prevents a reload started before a + // configure call from overwriting the newly configured client. + f.installKubeletClient(reloadedClient) + } + + if result.err == nil { + f.cachedPodList = result.versionedPodList + f.cachedFetchStart = f.lastFetchStart + } + + for resultCh := range f.fetchWaiters { + resultCh <- result + } + clear(f.fetchWaiters) +} + +func (f *podListFetcher) startConfigure(config podListFetcherConfig, resultCh chan<- error) { + previousClient := f.client + buildClient := f.buildClient + + f.wg.Go(func() { + client, err := buildClient(config, previousClient) + if err != nil { + resultCh <- err + return + } + + f.dispatch(func() { f.completeConfigure(config, client, resultCh) }) + }) +} + +func (f *podListFetcher) completeConfigure(config podListFetcherConfig, client *kubeletClient, resultCh chan<- error) { + f.config = &config + f.installKubeletClient(client) + // Reset any pending timer because pollRetryInterval may have changed, then + // schedule a fetch if there are waiters. In particular, waiters may have + // arrived before the initial configuration completed, when no timer could + // be scheduled. + f.stopTimer() + f.scheduleFetch() + resultCh <- nil +} + +func (f *podListFetcher) fetchTimerFired() { + f.stopTimer() + if len(f.fetchWaiters) > 0 { + f.startFetch() + } +} + +func (f *podListFetcher) stop() { + f.stopTimer() + if f.fetchInFlight() { + f.fetchCancel() + } +} + +func (f *podListFetcher) scheduleFetch() { + if f.config == nil || f.fetchInFlight() || f.fetchTimer != nil || len(f.fetchWaiters) == 0 { + return + } + + if f.lastFetchStart.IsZero() { + f.startFetch() + return + } + + delay := f.config.pollRetryInterval - f.clock.Now().Sub(f.lastFetchStart) + if delay <= 0 { + f.startFetch() + return + } + + f.fetchTimer = f.clock.Timer(delay) +} + +func (f *podListFetcher) startFetch() { + f.lastFetchStart = f.clock.Now() + fetchCtx, cancel := context.WithTimeout(context.Background(), kubeletRequestTimeout) + f.fetchCancel = cancel + + config := *f.config + version := f.cachedPodList.version + 1 + originalClient := f.client + shouldReloadClient := originalClient == nil || f.clock.Now().Sub(f.clientLoadedAt) >= config.reloadInterval + buildClient, fetch := f.buildClient, f.fetch + + f.wg.Go(func() { + defer cancel() + clientForFetch := originalClient + var reloadedClient *kubeletClient + if shouldReloadClient { + var err error + reloadedClient, err = buildClient(config, originalClient) + if err != nil { + f.dispatch(func() { f.completeFetch(podListFetchResult{err: err}, originalClient, nil) }) + return + } + clientForFetch = reloadedClient + } + + pods, err := fetch(fetchCtx, clientForFetch) + result := podListFetchResult{ + versionedPodList: versionedPodList{pods: pods, version: version}, + err: err, + } + f.dispatch(func() { f.completeFetch(result, originalClient, reloadedClient) }) + }) +} + +func (f *podListFetcher) stopTimer() { + if f.fetchTimer == nil { + return + } + + f.fetchTimer.Stop() + f.fetchTimer = nil +} + +func (f *podListFetcher) timerChan() <-chan time.Time { + if f.fetchTimer == nil { + return nil + } + return f.fetchTimer.C +} + +func (f *podListFetcher) fetchInFlight() bool { + return f.fetchCancel != nil +} + +func (f *podListFetcher) fetchPodList(ctx context.Context, client *kubeletClient) (map[string]*fastjson.Value, error) { + podListBytes, err := client.getPodList(ctx) + if err != nil { + return nil, err + } + return f.parsePodList(podListBytes) +} + +func (f *podListFetcher) parsePodList(podListBytes []byte) (map[string]*fastjson.Value, error) { + var parser fastjson.Parser + podList, err := parser.ParseBytes(podListBytes) + if err != nil { + return nil, fmt.Errorf("unable to parse kubelet response: %w", err) + } + + if podList.Type() != fastjson.TypeObject { + return nil, errors.New("invalid kubelet response: expected an object") + } + itemsValue := podList.Get("items") + if itemsValue == nil { + return nil, errors.New("invalid kubelet response: expected an items array") + } + var items []*fastjson.Value + switch itemsValue.Type() { + case fastjson.TypeArray: + items = itemsValue.GetArray() + case fastjson.TypeNull: + // encoding/json marshals a nil PodList.Items slice as null. + default: + return nil, errors.New("invalid kubelet response: expected an items array") + } + result := make(map[string]*fastjson.Value, len(items)) + + for _, podValue := range items { + uid := string(podValue.Get("metadata", "uid").GetStringBytes()) + if uid == "" { + f.log.Warn("Pod has no UID", "pod", podValue) + continue + } + result[uid] = podValue + } + + return result, nil +} + +func (f *podListFetcher) buildKubeletClient(config podListFetcherConfig, previousClient *kubeletClient) (*kubeletClient, error) { + transportConfig, token, err := f.loadKubeletTransportConfig(config) + if err != nil { + return nil, err + } + + if previousClient != nil && previousClient.transportConfig == transportConfig { + // An in-flight fetch may still reference the existing client. Copy it to + // update the token without a data race while reusing the transport. + client := *previousClient + client.token = token + return &client, nil + } + + return newKubeletClient(transportConfig, token) +} + +func (f *podListFetcher) installKubeletClient(client *kubeletClient) { + f.client = client + f.clientLoadedAt = f.clock.Now() +} + +func (f *podListFetcher) loadKubeletTransportConfig(config podListFetcherConfig) (kubeletTransportConfig, string, error) { + transportConfig := kubeletTransportConfig{ + secure: config.secure, + skipKubeletVerification: config.skipKubeletVerification, + nodeName: config.nodeName, + port: config.port, + } + if !config.secure { + return transportConfig, "", nil + } + + if !config.skipKubeletVerification { + caPEM, err := f.readFile(config.kubeletCAPath) + if err != nil { + return kubeletTransportConfig{}, "", fmt.Errorf("unable to load kubelet CA: %w", err) + } + transportConfig.caPEM = string(caPEM) + } + + var token string + switch { + case config.useAnonymousAuthentication: + case config.certificatePath != "" && config.privateKeyPath != "": + certPEM, err := f.readFile(config.certificatePath) + if err != nil { + return kubeletTransportConfig{}, "", fmt.Errorf("unable to load certificate: %w", err) + } + keyPEM, err := f.readFile(config.privateKeyPath) + if err != nil { + return kubeletTransportConfig{}, "", fmt.Errorf("unable to load private key: %w", err) + } + transportConfig.certificatePEM = string(certPEM) + transportConfig.privateKeyPEM = string(keyPEM) + case config.certificatePath != "": + return kubeletTransportConfig{}, "", errors.New("the private key path is required with the certificate path") + case config.privateKeyPath != "": + return kubeletTransportConfig{}, "", errors.New("the certificate path is required with the private key path") + default: + tokenBytes, err := f.readFile(config.tokenPath) + if err != nil { + return kubeletTransportConfig{}, "", fmt.Errorf("unable to load token: %w", err) + } + token = strings.TrimSpace(string(tokenBytes)) + } + + return transportConfig, token, nil +} + +func (f *podListFetcher) readFile(path string) ([]byte, error) { + return os.ReadFile(filepath.Join(f.rootDir, path)) +} diff --git a/pkg/agent/plugin/workloadattestor/k8s/podlistfetcher_test.go b/pkg/agent/plugin/workloadattestor/k8s/podlistfetcher_test.go new file mode 100644 index 0000000000..aa3adf0d92 --- /dev/null +++ b/pkg/agent/plugin/workloadattestor/k8s/podlistfetcher_test.go @@ -0,0 +1,842 @@ +package k8s + +import ( + "context" + "crypto/x509" + "errors" + "math/big" + "os" + "path/filepath" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/hashicorp/go-hclog" + "github.com/spiffe/spire/pkg/common/pemutil" + "github.com/spiffe/spire/test/clock" + "github.com/spiffe/spire/test/spiretest" + "github.com/stretchr/testify/require" + "github.com/valyala/fastjson" +) + +const testPodListFetchRetryInterval = time.Second + +func TestPodListFetcherCachesAndSpacesRequestsFromStart(t *testing.T) { + fetcher, mockClock := newTestPodListFetcher(t) + + var requestCount atomic.Int32 + configureTestFetcher(t, fetcher, func(context.Context) (map[string]*fastjson.Value, error) { + requestCount.Add(1) + return map[string]*fastjson.Value{}, nil + }) + + first, err := fetcher.fetchNext(t.Context(), 0) + require.NoError(t, err) + require.EqualValues(t, 1, first.version) + + cached, err := fetcher.fetchNext(t.Context(), 0) + require.NoError(t, err) + require.Equal(t, first.version, cached.version) + require.EqualValues(t, 1, requestCount.Load()) + + nextResult := make(chan podListFetchResult, 1) + go func() { + podList, err := fetcher.fetchNext(t.Context(), first.version) + nextResult <- podListFetchResult{versionedPodList: podList, err: err} + }() + + mockClock.WaitForTimer(time.Minute, "waiting for pod list retry timer") + mockClock.Add(testPodListFetchRetryInterval - time.Nanosecond) + require.Never(t, func() bool { + return requestCount.Load() > 1 + }, 10*time.Millisecond, time.Millisecond) + + mockClock.Add(time.Nanosecond) + next := <-nextResult + require.NoError(t, next.err) + require.EqualValues(t, 2, next.version) + require.EqualValues(t, 2, requestCount.Load()) +} + +func TestPodListFetcherSharesFetchAcrossConcurrentCallers(t *testing.T) { + fetcher, _ := newTestPodListFetcher(t) + + started := make(chan struct{}) + release := make(chan struct{}) + var requestCount atomic.Int32 + configureTestFetcher(t, fetcher, func(context.Context) (map[string]*fastjson.Value, error) { + requestCount.Add(1) + close(started) + <-release + return map[string]*fastjson.Value{}, nil + }) + + const callerCount = 16 + var ready sync.WaitGroup + ready.Add(callerCount) + start := make(chan struct{}) + results := make(chan podListFetchResult, callerCount) + for range callerCount { + go func() { + ready.Done() + <-start + podList, err := fetcher.fetchNext(t.Context(), 0) + results <- podListFetchResult{versionedPodList: podList, err: err} + }() + } + ready.Wait() + close(start) + <-started + close(release) + + for range callerCount { + result := <-results + require.NoError(t, result.err) + require.EqualValues(t, 1, result.version) + } + require.EqualValues(t, 1, requestCount.Load()) +} + +func TestPodListFetcherStartsOverdueRequestAfterSlowRequest(t *testing.T) { + fetcher, mockClock := newTestPodListFetcher(t) + + firstStarted := make(chan struct{}) + releaseFirst := make(chan struct{}) + var requestCount atomic.Int32 + configureTestFetcher(t, fetcher, func(context.Context) (map[string]*fastjson.Value, error) { + if requestCount.Add(1) == 1 { + close(firstStarted) + <-releaseFirst + } + return map[string]*fastjson.Value{}, nil + }) + + firstResult := make(chan podListFetchResult, 1) + go func() { + podList, err := fetcher.fetchNext(t.Context(), 0) + firstResult <- podListFetchResult{versionedPodList: podList, err: err} + }() + + <-firstStarted + mockClock.Add(2 * testPodListFetchRetryInterval) + close(releaseFirst) + first := <-firstResult + require.NoError(t, first.err) + + second, err := fetcher.fetchNext(t.Context(), first.version) + require.NoError(t, err) + require.EqualValues(t, 2, second.version) + require.EqualValues(t, 2, requestCount.Load()) +} + +func TestPodListFetcherDoesNotCacheFailures(t *testing.T) { + fetcher, mockClock := newTestPodListFetcher(t) + + fetchErr := errors.New("fetch failed") + var requestCount atomic.Int32 + configureTestFetcher(t, fetcher, func(context.Context) (map[string]*fastjson.Value, error) { + if requestCount.Add(1) == 1 { + return nil, fetchErr + } + return map[string]*fastjson.Value{}, nil + }) + + _, err := fetcher.fetchNext(t.Context(), 0) + require.ErrorIs(t, err, fetchErr) + + secondResult := make(chan podListFetchResult, 1) + go func() { + podList, err := fetcher.fetchNext(t.Context(), 0) + secondResult <- podListFetchResult{versionedPodList: podList, err: err} + }() + + mockClock.WaitForTimer(time.Minute, "waiting for retry after failed pod list request") + select { + case result := <-secondResult: + t.Fatalf("request returned before retry interval elapsed: %v", result.err) + default: + } + + mockClock.Add(testPodListFetchRetryInterval) + second := <-secondResult + require.NoError(t, second.err) + require.EqualValues(t, 1, second.version) + require.EqualValues(t, 2, requestCount.Load()) +} + +func TestPodListFetcherDoesNotRevalidateStaleCacheAfterFailure(t *testing.T) { + fetcher, mockClock := newTestPodListFetcher(t) + + fetchErr := errors.New("fetch failed") + var requestCount atomic.Int32 + configureTestFetcher(t, fetcher, func(context.Context) (map[string]*fastjson.Value, error) { + if requestCount.Add(1) == 2 { + return nil, fetchErr + } + return map[string]*fastjson.Value{}, nil + }) + + first, err := fetcher.fetchNext(t.Context(), 0) + require.NoError(t, err) + require.EqualValues(t, 1, first.version) + + mockClock.Add(testPodListFetchRetryInterval) + _, err = fetcher.fetchNext(t.Context(), 0) + require.ErrorIs(t, err, fetchErr) + + nextResult := make(chan podListFetchResult, 1) + go func() { + podList, err := fetcher.fetchNext(t.Context(), 0) + nextResult <- podListFetchResult{versionedPodList: podList, err: err} + }() + + mockClock.WaitForTimer(time.Minute, "waiting for retry after failed cache refresh") + select { + case result := <-nextResult: + t.Fatalf("request returned stale cached result after failed refresh: %v", result.err) + default: + } + + mockClock.Add(testPodListFetchRetryInterval) + next := <-nextResult + require.NoError(t, next.err) + require.EqualValues(t, 2, next.version) + require.EqualValues(t, 3, requestCount.Load()) +} + +func TestPodListFetcherDoesNotPollWithoutWaiters(t *testing.T) { + fetcher, mockClock := newTestPodListFetcher(t) + + var requestCount atomic.Int32 + configureTestFetcher(t, fetcher, func(context.Context) (map[string]*fastjson.Value, error) { + requestCount.Add(1) + return map[string]*fastjson.Value{}, nil + }) + + _, err := fetcher.fetchNext(t.Context(), 0) + require.NoError(t, err) + + mockClock.Add(10 * testPodListFetchRetryInterval) + require.EqualValues(t, 1, requestCount.Load()) + + _, err = fetcher.fetchNext(t.Context(), 0) + require.NoError(t, err) + require.EqualValues(t, 2, requestCount.Load()) +} + +func TestPodListFetcherLateWaiterDoesNotInheritPreviousCallerCancellation(t *testing.T) { + fetcher, _ := newTestPodListFetcher(t) + + started := make(chan struct{}) + release := make(chan struct{}) + var requestCount atomic.Int32 + configureTestFetcher(t, fetcher, func(ctx context.Context) (map[string]*fastjson.Value, error) { + requestCount.Add(1) + close(started) + <-release + if err := ctx.Err(); err != nil { + return nil, err + } + return map[string]*fastjson.Value{}, nil + }) + + ctx, cancel := context.WithCancel(t.Context()) + firstResult := make(chan podListFetchResult, 1) + go func() { + podList, err := fetcher.fetchNext(ctx, 0) + firstResult <- podListFetchResult{versionedPodList: podList, err: err} + }() + + <-started + + // Cancel the only waiter, leaving the kubelet request temporarily without + // any callers. The request must remain in flight so that a waiter arriving + // during this window does not receive a context cancellation it did not + // cause. + cancel() + require.ErrorIs(t, (<-firstResult).err, context.Canceled) + + // Submit directly to ensure the new waiter is registered before the + // kubelet request completes. + secondResult := make(chan podListFetchResult, 1) + fetcher.actionCh <- func() { fetcher.registerFetchRequest(0, secondResult) } + + close(release) + require.NoError(t, (<-secondResult).err) + require.EqualValues(t, 1, requestCount.Load()) +} + +func TestPodListFetcherCallerCancellationDoesNotCancelSharedFetch(t *testing.T) { + fetcher, _ := newTestPodListFetcher(t) + + started := make(chan struct{}) + release := make(chan struct{}) + var requestCount atomic.Int32 + configureTestFetcher(t, fetcher, func(context.Context) (map[string]*fastjson.Value, error) { + requestCount.Add(1) + close(started) + <-release + return map[string]*fastjson.Value{}, nil + }) + + ctx, cancel := context.WithCancel(t.Context()) + firstResult := make(chan podListFetchResult, 1) + go func() { + podList, err := fetcher.fetchNext(ctx, 0) + firstResult <- podListFetchResult{versionedPodList: podList, err: err} + }() + + <-started + + // Submit directly so the test knows the fetcher registered the second + // waiter before the first one is canceled. + secondResult := make(chan podListFetchResult, 1) + fetcher.actionCh <- func() { fetcher.registerFetchRequest(0, secondResult) } + + cancel() + require.ErrorIs(t, (<-firstResult).err, context.Canceled) + close(release) + require.NoError(t, (<-secondResult).err) + require.EqualValues(t, 1, requestCount.Load()) +} + +func TestPodListFetcherCloseWaitsForFetch(t *testing.T) { + mockClock := clock.NewMock(t) + fetcher := newPodListFetcher(mockClock, "") + + started := make(chan struct{}) + fetchDone := make(chan struct{}) + configureTestFetcher(t, fetcher, func(ctx context.Context) (map[string]*fastjson.Value, error) { + close(started) + <-ctx.Done() + close(fetchDone) + return nil, ctx.Err() + }) + + result := make(chan podListFetchResult, 1) + go func() { + podList, err := fetcher.fetchNext(t.Context(), 0) + result <- podListFetchResult{versionedPodList: podList, err: err} + }() + <-started + + fetcher.close() + <-fetchDone + require.ErrorIs(t, (<-result).err, errPodListFetcherClosed) +} + +func TestPodListFetcherCloseWaitsForKubeletClientBuild(t *testing.T) { + fetcher := newPodListFetcher(clock.NewMock(t), "") + + started := make(chan struct{}) + release := make(chan struct{}) + buildDone := make(chan struct{}) + fetcher.buildClient = func(podListFetcherConfig, *kubeletClient) (*kubeletClient, error) { + close(started) + <-release + close(buildDone) + return nil, errors.New("build failed") + } + + configureResult := make(chan error, 1) + go func() { + configureResult <- fetcher.configure(t.Context(), podListFetcherConfig{}) + }() + <-started + + closeResult := make(chan struct{}) + go func() { + fetcher.close() + close(closeResult) + }() + require.Never(t, func() bool { + select { + case <-closeResult: + return true + default: + return false + } + }, 10*time.Millisecond, time.Millisecond) + + close(release) + <-closeResult + <-buildDone + require.Error(t, <-configureResult) +} + +func TestPodListFetcherConfigureAfterClose(t *testing.T) { + fetcher := newPodListFetcher(clock.NewMock(t), "") + fetcher.close() + + err := fetcher.configure(t.Context(), podListFetcherConfig{}) + require.ErrorIs(t, err, errPodListFetcherClosed) +} + +func TestPodListFetcherConfigureWaitsForAcceptedRequestAfterContextCancellation(t *testing.T) { + fetcher, _ := newTestPodListFetcher(t) + + started := make(chan struct{}) + release := make(chan struct{}) + fetcher.buildClient = func(config podListFetcherConfig, previousClient *kubeletClient) (*kubeletClient, error) { + close(started) + <-release + return fetcher.buildKubeletClient(config, previousClient) + } + + ctx, cancel := context.WithCancel(t.Context()) + configResult := make(chan error, 1) + go func() { + configResult <- fetcher.configure(ctx, podListFetcherConfig{ + port: 1, + }) + }() + <-started + + cancel() + select { + case err := <-configResult: + t.Fatalf("configure returned before the accepted request completed: %v", err) + default: + } + + close(release) + require.NoError(t, <-configResult) + require.Equal(t, 1, fetcher.config.port) +} + +func TestPodListFetcherFailedConfigurePreservesClient(t *testing.T) { + fetcher, _ := newTestPodListFetcher(t) + require.NoError(t, fetcher.configure(t.Context(), podListFetcherConfig{port: 1})) + + buildErr := errors.New("build failed") + fetcher.buildClient = func(podListFetcherConfig, *kubeletClient) (*kubeletClient, error) { + return nil, buildErr + } + require.ErrorIs(t, fetcher.configure(t.Context(), podListFetcherConfig{port: 2}), buildErr) + require.Equal(t, 1, fetcher.config.port) + require.Equal(t, "http://127.0.0.1:1", fetcher.client.endpoint.String()) +} + +func TestPodListFetcherRunProcessesRequestsWhileConfiguringKubeletClient(t *testing.T) { + fetcher, _ := newTestPodListFetcher(t) + + require.NoError(t, fetcher.configure(t.Context(), podListFetcherConfig{ + pollRetryInterval: time.Second, + port: 1, + reloadInterval: time.Hour, + })) + + started := make(chan struct{}) + release := make(chan struct{}) + fetcher.buildClient = func(config podListFetcherConfig, previousClient *kubeletClient) (*kubeletClient, error) { + close(started) + <-release + return fetcher.buildKubeletClient(config, previousClient) + } + fetcher.fetch = func(ctx context.Context, _ *kubeletClient) (map[string]*fastjson.Value, error) { + <-ctx.Done() + return nil, ctx.Err() + } + + configResult := make(chan error, 1) + go func() { + configResult <- fetcher.configure(t.Context(), podListFetcherConfig{ + pollRetryInterval: time.Second, + port: 2, + reloadInterval: time.Hour, + }) + }() + <-started + + resultCh := make(chan podListFetchResult, 1) + requestReceived := make(chan struct{}) + go func() { + fetcher.actionCh <- func() { fetcher.registerFetchRequest(0, resultCh) } + close(requestReceived) + }() + require.Eventually(t, func() bool { + select { + case <-requestReceived: + return true + default: + return false + } + }, time.Second, time.Millisecond) + + cancellationReceived := make(chan struct{}) + go func() { + fetcher.actionCh <- func() { fetcher.cancelFetchRequest(resultCh) } + close(cancellationReceived) + }() + require.Eventually(t, func() bool { + select { + case <-cancellationReceived: + return true + default: + return false + } + }, time.Second, time.Millisecond) + + close(release) + require.NoError(t, <-configResult) +} + +func TestPodListFetcherDoesNotInstallClientFromStaleFetchResult(t *testing.T) { + transportConfig := kubeletTransportConfig{ + secure: true, + skipKubeletVerification: true, + port: 2, + } + newClient := func(token string) *kubeletClient { + client, err := newKubeletClient(transportConfig, token) + require.NoError(t, err) + return client + } + + fetcher := podListFetcher{ + clock: clock.NewMock(t), + client: newClient("new-token"), + } + + fetcher.completeFetch(podListFetchResult{ + versionedPodList: versionedPodList{pods: map[string]*fastjson.Value{}, version: 1}, + }, newClient("old-token"), newClient("old-token")) + require.Equal(t, "https://127.0.0.1:2", fetcher.client.endpoint.String()) + require.Equal(t, "new-token", fetcher.client.token) + require.EqualValues(t, 1, fetcher.cachedPodList.version) + + fetcher.completeFetch(podListFetchResult{ + versionedPodList: versionedPodList{pods: map[string]*fastjson.Value{}, version: 2}, + }, fetcher.client, newClient("newest-token")) + require.Equal(t, "newest-token", fetcher.client.token) + require.EqualValues(t, 2, fetcher.cachedPodList.version) +} + +func TestPodListFetcherReconfigurePreservesCache(t *testing.T) { + fetcher, mockClock := newTestPodListFetcher(t) + + var oldRequestCount atomic.Int32 + var newRequestCount atomic.Int32 + fetcher.fetch = func(_ context.Context, client *kubeletClient) (map[string]*fastjson.Value, error) { + if client.endpoint.Port() == "1" { + oldRequestCount.Add(1) + } else { + newRequestCount.Add(1) + } + return map[string]*fastjson.Value{}, nil + } + require.NoError(t, fetcher.configure(t.Context(), podListFetcherConfig{ + pollRetryInterval: testPodListFetchRetryInterval, + port: 1, + })) + + first, err := fetcher.fetchNext(t.Context(), 0) + require.NoError(t, err) + + require.NoError(t, fetcher.configure(t.Context(), podListFetcherConfig{ + pollRetryInterval: testPodListFetchRetryInterval, + port: 2, + })) + + cached, err := fetcher.fetchNext(t.Context(), 0) + require.NoError(t, err) + require.Equal(t, first.version, cached.version) + require.EqualValues(t, 1, oldRequestCount.Load()) + require.EqualValues(t, 0, newRequestCount.Load()) + + nextResult := make(chan podListFetchResult, 1) + go func() { + podList, err := fetcher.fetchNext(t.Context(), first.version) + nextResult <- podListFetchResult{versionedPodList: podList, err: err} + }() + + mockClock.WaitForTimer(time.Minute, "waiting for reconfigured pod list request") + require.EqualValues(t, 0, newRequestCount.Load()) + mockClock.Add(testPodListFetchRetryInterval) + + next := <-nextResult + require.NoError(t, next.err) + require.Greater(t, next.version, first.version) + require.EqualValues(t, 1, oldRequestCount.Load()) + require.EqualValues(t, 1, newRequestCount.Load()) +} + +func TestPodListFetcherReconfigureUpdatesCacheLifetime(t *testing.T) { + t.Run("Extend", func(t *testing.T) { + fetcher, mockClock := newTestPodListFetcher(t) + + var requestCount atomic.Int32 + fetcher.fetch = func(context.Context, *kubeletClient) (map[string]*fastjson.Value, error) { + requestCount.Add(1) + return map[string]*fastjson.Value{}, nil + } + require.NoError(t, fetcher.configure(t.Context(), podListFetcherConfig{ + pollRetryInterval: time.Second, + })) + + first, err := fetcher.fetchNext(t.Context(), 0) + require.NoError(t, err) + mockClock.Add(1500 * time.Millisecond) + + require.NoError(t, fetcher.configure(t.Context(), podListFetcherConfig{ + pollRetryInterval: 2 * time.Second, + })) + cached, err := fetcher.fetchNext(t.Context(), 0) + require.NoError(t, err) + require.Equal(t, first.version, cached.version) + require.EqualValues(t, 1, requestCount.Load()) + }) + + t.Run("Shorten", func(t *testing.T) { + fetcher, mockClock := newTestPodListFetcher(t) + + var requestCount atomic.Int32 + fetcher.fetch = func(context.Context, *kubeletClient) (map[string]*fastjson.Value, error) { + requestCount.Add(1) + return map[string]*fastjson.Value{}, nil + } + require.NoError(t, fetcher.configure(t.Context(), podListFetcherConfig{ + pollRetryInterval: 2 * time.Second, + })) + + first, err := fetcher.fetchNext(t.Context(), 0) + require.NoError(t, err) + mockClock.Add(time.Second) + + require.NoError(t, fetcher.configure(t.Context(), podListFetcherConfig{ + pollRetryInterval: 500 * time.Millisecond, + })) + refreshed, err := fetcher.fetchNext(t.Context(), 0) + require.NoError(t, err) + require.Greater(t, refreshed.version, first.version) + require.EqualValues(t, 2, requestCount.Load()) + }) +} + +func TestPodListFetcherReloadsKubeletClient(t *testing.T) { + mockClock := clock.NewMock(t) + firstConfig := podListFetcherConfig{ + port: 1, + reloadInterval: time.Second, + } + fetcher := &podListFetcher{ + clock: mockClock, + config: &firstConfig, + actionCh: make(chan func(), 1), + fetch: func(context.Context, *kubeletClient) (map[string]*fastjson.Value, error) { + return map[string]*fastjson.Value{}, nil + }, + } + fetcher.buildClient = fetcher.buildKubeletClient + runFetch := func() { + fetcher.startFetch() + (<-fetcher.actionCh)() + } + + runFetch() + firstClient := fetcher.client + + secondConfig := podListFetcherConfig{ + port: 2, + reloadInterval: time.Second, + } + fetcher.config = &secondConfig + runFetch() + require.Same(t, firstClient, fetcher.client) + + mockClock.Add(firstConfig.reloadInterval) + runFetch() + require.NotSame(t, firstClient, fetcher.client) +} + +func TestPodListFetcherBuildKubeletClientReusesTransportWhenOnlyTokenChanges(t *testing.T) { + mockClock := clock.NewMock(t) + rootDir := t.TempDir() + fetcher := newPodListFetcher(mockClock, rootDir) + t.Cleanup(fetcher.close) + + const tokenPath = "token" + require.NoError(t, os.WriteFile(filepath.Join(rootDir, tokenPath), []byte("old-token"), 0o600)) + + config := podListFetcherConfig{ + secure: true, + skipKubeletVerification: true, + tokenPath: tokenPath, + reloadInterval: time.Second, + } + client, err := fetcher.buildKubeletClient(config, fetcher.client) + require.NoError(t, err) + fetcher.installKubeletClient(client) + firstClient := fetcher.client + firstTransport := fetcher.client.transport + + require.NoError(t, os.WriteFile(filepath.Join(rootDir, tokenPath), []byte("new-token"), 0o600)) + client, err = fetcher.buildKubeletClient(config, fetcher.client) + require.NoError(t, err) + fetcher.installKubeletClient(client) + + require.NotSame(t, firstClient, fetcher.client) + require.Same(t, firstTransport, fetcher.client.transport) + require.Equal(t, "old-token", firstClient.token) + require.Equal(t, "new-token", fetcher.client.token) +} + +func TestPodListFetcherBuildKubeletClientReplacesTransportWhenCAChanges(t *testing.T) { + mockClock := clock.NewMock(t) + rootDir := t.TempDir() + fetcher := newPodListFetcher(mockClock, rootDir) + t.Cleanup(fetcher.close) + + const caPath = "ca.pem" + writeTestCA(t, filepath.Join(rootDir, caPath), 1) + + config := podListFetcherConfig{ + secure: true, + kubeletCAPath: caPath, + nodeName: "localhost", + useAnonymousAuthentication: true, + reloadInterval: time.Second, + } + client, err := fetcher.buildKubeletClient(config, fetcher.client) + require.NoError(t, err) + fetcher.installKubeletClient(client) + firstTransport := fetcher.client.transport + + writeTestCA(t, filepath.Join(rootDir, caPath), 2) + client, err = fetcher.buildKubeletClient(config, fetcher.client) + require.NoError(t, err) + fetcher.installKubeletClient(client) + + require.NotSame(t, firstTransport, fetcher.client.transport) +} + +func TestPodListFetcherCreatesAndInstallsConfiguredKubeletClient(t *testing.T) { + fetcher, _ := newTestPodListFetcher(t) + + config := podListFetcherConfig{ + pollRetryInterval: time.Second, + port: 1, + } + usedClient := make(chan *kubeletClient, 1) + fetcher.fetch = func(_ context.Context, client *kubeletClient) (map[string]*fastjson.Value, error) { + usedClient <- client + return map[string]*fastjson.Value{}, nil + } + + require.NoError(t, fetcher.configure(t.Context(), config)) + _, err := fetcher.fetchNext(t.Context(), 0) + require.NoError(t, err) + require.Equal(t, "http://127.0.0.1:1", (<-usedClient).endpoint.String()) +} + +func TestPodListFetcherConfiguresKubeletClientDuringFetch(t *testing.T) { + fetcher := podListFetcher{ + clock: clock.NewMock(t), + actionCh: make(chan func(), 1), + fetchCancel: func() {}, + } + fetcher.buildClient = fetcher.buildKubeletClient + config := podListFetcherConfig{port: 1} + resultCh := make(chan error, 1) + + fetcher.startConfigure(config, resultCh) + (<-fetcher.actionCh)() + require.NoError(t, <-resultCh) + require.Equal(t, "http://127.0.0.1:1", fetcher.client.endpoint.String()) +} + +func TestPodListFetcherParsePodList(t *testing.T) { + for _, testCase := range []struct { + name string + response string + wantUIDs []string + wantErr string + }{ + { + name: "indexes pods by UID", + response: `{"items":[ + {"metadata":{"uid":"pod-1"}}, + {"metadata":{"uid":"pod-2"}} + ]}`, + wantUIDs: []string{"pod-1", "pod-2"}, + }, + { + name: "accepts null items as an empty list", + response: `{"items":null}`, + }, + { + name: "ignores pods without a UID", + response: `{"items":[ + {"metadata":{}}, + {"metadata":{"uid":"pod-1"}} + ]}`, + wantUIDs: []string{"pod-1"}, + }, + { + name: "rejects malformed response", + response: `{"items":`, + wantErr: "unable to parse kubelet response", + }, + { + name: "rejects non-object response", + response: `[]`, + wantErr: "invalid kubelet response: expected an object", + }, + { + name: "rejects response without items", + response: `{}`, + wantErr: "invalid kubelet response: expected an items array", + }, + { + name: "rejects response with non-array items", + response: `{"items":{}}`, + wantErr: "invalid kubelet response: expected an items array", + }, + } { + t.Run(testCase.name, func(t *testing.T) { + fetcher := podListFetcher{log: hclog.NewNullLogger()} + pods, err := fetcher.parsePodList([]byte(testCase.response)) + if testCase.wantErr != "" { + require.ErrorContains(t, err, testCase.wantErr) + return + } + + require.NoError(t, err) + require.Len(t, pods, len(testCase.wantUIDs)) + for _, uid := range testCase.wantUIDs { + require.Contains(t, pods, uid) + } + }) + } +} + +func newTestPodListFetcher(t *testing.T) (*podListFetcher, *clock.Mock) { + t.Helper() + + mockClock := clock.NewMock(t) + fetcher := newPodListFetcher(mockClock, "") + t.Cleanup(fetcher.close) + return fetcher, mockClock +} + +func configureTestFetcher(t *testing.T, fetcher *podListFetcher, fetch func(context.Context) (map[string]*fastjson.Value, error)) { + t.Helper() + + fetcher.fetch = func(ctx context.Context, _ *kubeletClient) (map[string]*fastjson.Value, error) { + return fetch(ctx) + } + require.NoError(t, fetcher.configure(t.Context(), podListFetcherConfig{ + pollRetryInterval: testPodListFetchRetryInterval, + })) +} + +func writeTestCA(t *testing.T, path string, serial int64) { + cert, _ := spiretest.SelfSignCertificate(t, &x509.Certificate{ + SerialNumber: big.NewInt(serial), + IsCA: true, + BasicConstraintsValid: true, + KeyUsage: x509.KeyUsageCertSign, + }) + require.NoError(t, os.WriteFile(path, pemutil.EncodeCertificate(cert), 0o600)) +}