msb-public
/
helm
mirror of https://github.com/helm/helm
1
0
Code Issues Projects Releases Wiki Activity

Merge pull request #31610 from mohsenmottaghi/fix-helm-4-watch-strategy

Browse Source 
fix: use namespace-scoped watching to avoid cluster-wide LIST permiss…
pull/31574/head
George Jenkins 8 months ago committed by GitHub
parent f562a36d58 d6b35cedeb
commit 24e8297343
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 501 additions and 4 deletions
Show Stats Download Patch File Download Diff File

8
pkg/kube/statuswait.go
Unescape View File

@ -132,7 +132,9 @@ func (w *statusWaiter) waitForDelete(ctx context.Context, resourceList ResourceL
}
resources = append(resources, obj)
}
eventCh := sw.Watch(cancelCtx, resources, watcher.Options{})
eventCh := sw.Watch(cancelCtx, resources, watcher.Options{
RESTScopeStrategy: watcher.RESTScopeNamespace,
})
statusCollector := collector.NewResourceStatusCollector(resources)
done := statusCollector.ListenWithObserver(eventCh, statusObserver(cancel, status.NotFoundStatus))
<-done
@ -175,7 +177,9 @@ func (w *statusWaiter) wait(ctx context.Context, resourceList ResourceList, sw w
resources = append(resources, obj)
}
eventCh := sw.Watch(cancelCtx, resources, watcher.Options{})
eventCh := sw.Watch(cancelCtx, resources, watcher.Options{
RESTScopeStrategy: watcher.RESTScopeNamespace,
})
statusCollector := collector.NewResourceStatusCollector(resources)
done := statusCollector.ListenWithObserver(eventCh, statusObserver(cancel, status.CurrentStatus))
<-done

497
pkg/kube/statuswait_test.go
Unescape View File

@ -17,7 +17,10 @@ limitations under the License.
package kube // import "helm.sh/helm/v3/pkg/kube"
import (
"context"
"errors"
"fmt"
"strings"
"testing"
"time"
@ -27,11 +30,14 @@ import (
appsv1 "k8s.io/api/apps/v1"
batchv1 "k8s.io/api/batch/v1"
v1 "k8s.io/api/core/v1"
apierrors "k8s.io/apimachinery/pkg/api/errors"
"k8s.io/apimachinery/pkg/api/meta"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/runtime/schema"
"k8s.io/apimachinery/pkg/util/yaml"
"k8s.io/client-go/dynamic"
dynamicfake "k8s.io/client-go/dynamic/fake"
"k8s.io/kubectl/pkg/scheme"
)
@ -153,6 +159,83 @@ spec:
- containerPort: 80
`
var podNamespace1Manifest = `
apiVersion: v1
kind: Pod
metadata:
name: pod-ns1
namespace: namespace-1
status:
conditions:
- type: Ready
status: "True"
phase: Running
`
var podNamespace2Manifest = `
apiVersion: v1
kind: Pod
metadata:
name: pod-ns2
namespace: namespace-2
status:
conditions:
- type: Ready
status: "True"
phase: Running
`
var podNamespace1NoStatusManifest = `
apiVersion: v1
kind: Pod
metadata:
name: pod-ns1
namespace: namespace-1
`
var jobNamespace1CompleteManifest = `
apiVersion: batch/v1
kind: Job
metadata:
name: job-ns1
namespace: namespace-1
generation: 1
status:
succeeded: 1
active: 0
conditions:
- type: Complete
status: "True"
`
var podNamespace2SucceededManifest = `
apiVersion: v1
kind: Pod
metadata:
name: pod-ns2
namespace: namespace-2
status:
phase: Succeeded
`
var clusterRoleManifest = `
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRole
metadata:
name: test-cluster-role
rules:
- apiGroups: [""]
resources: ["pods"]
verbs: ["get", "list"]
`
var namespaceManifest = `
apiVersion: v1
kind: Namespace
metadata:
name: test-namespace
`
func getGVR(t *testing.T, mapper meta.RESTMapper, obj *unstructured.Unstructured) schema.GroupVersionResource {
t.Helper()
gvk := obj.GroupVersionKind()
@ -232,11 +315,11 @@ func TestStatusWaitForDelete(t *testing.T) {
for _, objToDelete := range objsToDelete {
u := objToDelete.(*unstructured.Unstructured)
gvr := getGVR(t, fakeMapper, u)
go func() {
go func(gvr schema.GroupVersionResource, u *unstructured.Unstructured) {
time.Sleep(timeUntilPodDelete)
err := fakeClient.Tracker().Delete(gvr, u.GetNamespace(), u.GetName())
assert.NoError(t, err)
}()
}(gvr, u)
}
resourceList := getResourceListFromRuntimeObjs(t, c, objsToCreate)
err := statusWaiter.WaitForDelete(resourceList, timeout)
@ -448,3 +531,413 @@ func TestWatchForReady(t *testing.T) {
})
}
}
func TestStatusWaitMultipleNamespaces(t *testing.T) {
t.Parallel()
tests := []struct {
name string
objManifests []string
expectErrs []error
testFunc func(statusWaiter, ResourceList, time.Duration) error
}{
{
name: "pods in multiple namespaces",
objManifests: []string{podNamespace1Manifest, podNamespace2Manifest},
testFunc: func(sw statusWaiter, rl ResourceList, timeout time.Duration) error {
return sw.Wait(rl, timeout)
},
},
{
name: "hooks in multiple namespaces",
objManifests: []string{jobNamespace1CompleteManifest, podNamespace2SucceededManifest},
testFunc: func(sw statusWaiter, rl ResourceList, timeout time.Duration) error {
return sw.WatchUntilReady(rl, timeout)
},
},
{
name: "error when resource not ready in one namespace",
objManifests: []string{podNamespace1NoStatusManifest, podNamespace2Manifest},
expectErrs: []error{errors.New("resource not ready, name: pod-ns1, kind: Pod, status: InProgress"), errors.New("context deadline exceeded")},
testFunc: func(sw statusWaiter, rl ResourceList, timeout time.Duration) error {
return sw.Wait(rl, timeout)
},
},
{
name: "delete resources in multiple namespaces",
objManifests: []string{podNamespace1Manifest, podNamespace2Manifest},
testFunc: func(sw statusWaiter, rl ResourceList, timeout time.Duration) error {
return sw.WaitForDelete(rl, timeout)
},
},
{
name: "cluster-scoped resources work correctly with unrestricted permissions",
objManifests: []string{podNamespace1Manifest, clusterRoleManifest},
testFunc: func(sw statusWaiter, rl ResourceList, timeout time.Duration) error {
return sw.Wait(rl, timeout)
},
},
{
name: "namespace-scoped and cluster-scoped resources work together",
objManifests: []string{podNamespace1Manifest, podNamespace2Manifest, clusterRoleManifest},
testFunc: func(sw statusWaiter, rl ResourceList, timeout time.Duration) error {
return sw.Wait(rl, timeout)
},
},
{
name: "delete cluster-scoped resources works correctly",
objManifests: []string{podNamespace1Manifest, namespaceManifest},
testFunc: func(sw statusWaiter, rl ResourceList, timeout time.Duration) error {
return sw.WaitForDelete(rl, timeout)
},
},
{
name: "watch cluster-scoped resources works correctly",
objManifests: []string{clusterRoleManifest},
testFunc: func(sw statusWaiter, rl ResourceList, timeout time.Duration) error {
return sw.WatchUntilReady(rl, timeout)
},
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
t.Parallel()
c := newTestClient(t)
fakeClient := dynamicfake.NewSimpleDynamicClient(scheme.Scheme)
fakeMapper := testutil.NewFakeRESTMapper(
v1.SchemeGroupVersion.WithKind("Pod"),
batchv1.SchemeGroupVersion.WithKind("Job"),
schema.GroupVersion{Group: "rbac.authorization.k8s.io", Version: "v1"}.WithKind("ClusterRole"),
v1.SchemeGroupVersion.WithKind("Namespace"),
)
sw := statusWaiter{
client: fakeClient,
restMapper: fakeMapper,
}
objs := getRuntimeObjFromManifests(t, tt.objManifests)
for _, obj := range objs {
u := obj.(*unstructured.Unstructured)
gvr := getGVR(t, fakeMapper, u)
err := fakeClient.Tracker().Create(gvr, u, u.GetNamespace())
assert.NoError(t, err)
}
if strings.Contains(tt.name, "delete") {
timeUntilDelete := time.Millisecond * 500
for _, obj := range objs {
u := obj.(*unstructured.Unstructured)
gvr := getGVR(t, fakeMapper, u)
go func(gvr schema.GroupVersionResource, u *unstructured.Unstructured) {
time.Sleep(timeUntilDelete)
err := fakeClient.Tracker().Delete(gvr, u.GetNamespace(), u.GetName())
assert.NoError(t, err)
}(gvr, u)
}
}
resourceList := getResourceListFromRuntimeObjs(t, c, objs)
err := tt.testFunc(sw, resourceList, time.Second*3)
if tt.expectErrs != nil {
assert.EqualError(t, err, errors.Join(tt.expectErrs...).Error())
return
}
assert.NoError(t, err)
})
}
}
type restrictedDynamicClient struct {
dynamic.Interface
allowedNamespaces map[string]bool
clusterScopedListAttempted bool
}
func newRestrictedDynamicClient(baseClient dynamic.Interface, allowedNamespaces []string) *restrictedDynamicClient {
allowed := make(map[string]bool)
for _, ns := range allowedNamespaces {
allowed[ns] = true
}
return &restrictedDynamicClient{
Interface: baseClient,
allowedNamespaces: allowed,
}
}
func (r *restrictedDynamicClient) Resource(resource schema.GroupVersionResource) dynamic.NamespaceableResourceInterface {
return &restrictedNamespaceableResource{
NamespaceableResourceInterface: r.Interface.Resource(resource),
allowedNamespaces: r.allowedNamespaces,
clusterScopedListAttempted: &r.clusterScopedListAttempted,
}
}
type restrictedNamespaceableResource struct {
dynamic.NamespaceableResourceInterface
allowedNamespaces map[string]bool
clusterScopedListAttempted *bool
}
func (r *restrictedNamespaceableResource) Namespace(ns string) dynamic.ResourceInterface {
return &restrictedResource{
ResourceInterface: r.NamespaceableResourceInterface.Namespace(ns),
namespace: ns,
allowedNamespaces: r.allowedNamespaces,
clusterScopedListAttempted: r.clusterScopedListAttempted,
}
}
func (r *restrictedNamespaceableResource) List(_ context.Context, _ metav1.ListOptions) (*unstructured.UnstructuredList, error) {
*r.clusterScopedListAttempted = true
return nil, apierrors.NewForbidden(
schema.GroupResource{Resource: "pods"},
"",
fmt.Errorf("user does not have cluster-wide LIST permissions for cluster-scoped resources"),
)
}
type restrictedResource struct {
dynamic.ResourceInterface
namespace string
allowedNamespaces map[string]bool
clusterScopedListAttempted *bool
}
func (r *restrictedResource) List(ctx context.Context, opts metav1.ListOptions) (*unstructured.UnstructuredList, error) {
if r.namespace == "" {
*r.clusterScopedListAttempted = true
return nil, apierrors.NewForbidden(
schema.GroupResource{Resource: "pods"},
"",
fmt.Errorf("user does not have cluster-wide LIST permissions for cluster-scoped resources"),
)
}
if !r.allowedNamespaces[r.namespace] {
return nil, apierrors.NewForbidden(
schema.GroupResource{Resource: "pods"},
"",
fmt.Errorf("user does not have LIST permissions in namespace %q", r.namespace),
)
}
return r.ResourceInterface.List(ctx, opts)
}
func TestStatusWaitRestrictedRBAC(t *testing.T) {
t.Parallel()
tests := []struct {
name string
objManifests []string
allowedNamespaces []string
expectErrs []error
testFunc func(statusWaiter, ResourceList, time.Duration) error
}{
{
name: "pods in multiple namespaces with namespace permissions",
objManifests: []string{podNamespace1Manifest, podNamespace2Manifest},
allowedNamespaces: []string{"namespace-1", "namespace-2"},
testFunc: func(sw statusWaiter, rl ResourceList, timeout time.Duration) error {
return sw.Wait(rl, timeout)
},
},
{
name: "delete pods in multiple namespaces with namespace permissions",
objManifests: []string{podNamespace1Manifest, podNamespace2Manifest},
allowedNamespaces: []string{"namespace-1", "namespace-2"},
testFunc: func(sw statusWaiter, rl ResourceList, timeout time.Duration) error {
return sw.WaitForDelete(rl, timeout)
},
},
{
name: "hooks in multiple namespaces with namespace permissions",
objManifests: []string{jobNamespace1CompleteManifest, podNamespace2SucceededManifest},
allowedNamespaces: []string{"namespace-1", "namespace-2"},
testFunc: func(sw statusWaiter, rl ResourceList, timeout time.Duration) error {
return sw.WatchUntilReady(rl, timeout)
},
},
{
name: "error when cluster-scoped resource included",
objManifests: []string{podNamespace1Manifest, clusterRoleManifest},
allowedNamespaces: []string{"namespace-1"},
expectErrs: []error{fmt.Errorf("user does not have cluster-wide LIST permissions for cluster-scoped resources")},
testFunc: func(sw statusWaiter, rl ResourceList, timeout time.Duration) error {
return sw.Wait(rl, timeout)
},
},
{
name: "error when deleting cluster-scoped resource",
objManifests: []string{podNamespace1Manifest, namespaceManifest},
allowedNamespaces: []string{"namespace-1"},
expectErrs: []error{fmt.Errorf("user does not have cluster-wide LIST permissions for cluster-scoped resources")},
testFunc: func(sw statusWaiter, rl ResourceList, timeout time.Duration) error {
return sw.WaitForDelete(rl, timeout)
},
},
{
name: "error when accessing disallowed namespace",
objManifests: []string{podNamespace1Manifest, podNamespace2Manifest},
allowedNamespaces: []string{"namespace-1"},
expectErrs: []error{fmt.Errorf("user does not have LIST permissions in namespace %q", "namespace-2")},
testFunc: func(sw statusWaiter, rl ResourceList, timeout time.Duration) error {
return sw.Wait(rl, timeout)
},
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
t.Parallel()
c := newTestClient(t)
baseFakeClient := dynamicfake.NewSimpleDynamicClient(scheme.Scheme)
fakeMapper := testutil.NewFakeRESTMapper(
v1.SchemeGroupVersion.WithKind("Pod"),
batchv1.SchemeGroupVersion.WithKind("Job"),
schema.GroupVersion{Group: "rbac.authorization.k8s.io", Version: "v1"}.WithKind("ClusterRole"),
v1.SchemeGroupVersion.WithKind("Namespace"),
)
restrictedClient := newRestrictedDynamicClient(baseFakeClient, tt.allowedNamespaces)
sw := statusWaiter{
client: restrictedClient,
restMapper: fakeMapper,
}
objs := getRuntimeObjFromManifests(t, tt.objManifests)
for _, obj := range objs {
u := obj.(*unstructured.Unstructured)
gvr := getGVR(t, fakeMapper, u)
err := baseFakeClient.Tracker().Create(gvr, u, u.GetNamespace())
assert.NoError(t, err)
}
if strings.Contains(tt.name, "delet") {
timeUntilDelete := time.Millisecond * 500
for _, obj := range objs {
u := obj.(*unstructured.Unstructured)
gvr := getGVR(t, fakeMapper, u)
go func(gvr schema.GroupVersionResource, u *unstructured.Unstructured) {
time.Sleep(timeUntilDelete)
err := baseFakeClient.Tracker().Delete(gvr, u.GetNamespace(), u.GetName())
assert.NoError(t, err)
}(gvr, u)
}
}
resourceList := getResourceListFromRuntimeObjs(t, c, objs)
err := tt.testFunc(sw, resourceList, time.Second*3)
if tt.expectErrs != nil {
require.Error(t, err)
for _, expectedErr := range tt.expectErrs {
assert.Contains(t, err.Error(), expectedErr.Error())
}
return
}
assert.NoError(t, err)
assert.False(t, restrictedClient.clusterScopedListAttempted)
})
}
}
func TestStatusWaitMixedResources(t *testing.T) {
t.Parallel()
tests := []struct {
name string
objManifests []string
allowedNamespaces []string
expectErrs []error
testFunc func(statusWaiter, ResourceList, time.Duration) error
}{
{
name: "wait succeeds with namespace-scoped resources only",
objManifests: []string{podNamespace1Manifest, podNamespace2Manifest},
allowedNamespaces: []string{"namespace-1", "namespace-2"},
testFunc: func(sw statusWaiter, rl ResourceList, timeout time.Duration) error {
return sw.Wait(rl, timeout)
},
},
{
name: "wait fails when cluster-scoped resource included",
objManifests: []string{podNamespace1Manifest, clusterRoleManifest},
allowedNamespaces: []string{"namespace-1"},
expectErrs: []error{fmt.Errorf("user does not have cluster-wide LIST permissions for cluster-scoped resources")},
testFunc: func(sw statusWaiter, rl ResourceList, timeout time.Duration) error {
return sw.Wait(rl, timeout)
},
},
{
name: "waitForDelete fails when cluster-scoped resource included",
objManifests: []string{podNamespace1Manifest, clusterRoleManifest},
allowedNamespaces: []string{"namespace-1"},
expectErrs: []error{fmt.Errorf("user does not have cluster-wide LIST permissions for cluster-scoped resources")},
testFunc: func(sw statusWaiter, rl ResourceList, timeout time.Duration) error {
return sw.WaitForDelete(rl, timeout)
},
},
{
name: "wait fails when namespace resource included",
objManifests: []string{podNamespace1Manifest, namespaceManifest},
allowedNamespaces: []string{"namespace-1"},
expectErrs: []error{fmt.Errorf("user does not have cluster-wide LIST permissions for cluster-scoped resources")},
testFunc: func(sw statusWaiter, rl ResourceList, timeout time.Duration) error {
return sw.Wait(rl, timeout)
},
},
{
name: "error when accessing disallowed namespace",
objManifests: []string{podNamespace1Manifest, podNamespace2Manifest},
allowedNamespaces: []string{"namespace-1"},
expectErrs: []error{fmt.Errorf("user does not have LIST permissions in namespace %q", "namespace-2")},
testFunc: func(sw statusWaiter, rl ResourceList, timeout time.Duration) error {
return sw.Wait(rl, timeout)
},
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
t.Parallel()
c := newTestClient(t)
baseFakeClient := dynamicfake.NewSimpleDynamicClient(scheme.Scheme)
fakeMapper := testutil.NewFakeRESTMapper(
v1.SchemeGroupVersion.WithKind("Pod"),
batchv1.SchemeGroupVersion.WithKind("Job"),
schema.GroupVersion{Group: "rbac.authorization.k8s.io", Version: "v1"}.WithKind("ClusterRole"),
v1.SchemeGroupVersion.WithKind("Namespace"),
)
restrictedClient := newRestrictedDynamicClient(baseFakeClient, tt.allowedNamespaces)
sw := statusWaiter{
client: restrictedClient,
restMapper: fakeMapper,
}
objs := getRuntimeObjFromManifests(t, tt.objManifests)
for _, obj := range objs {
u := obj.(*unstructured.Unstructured)
gvr := getGVR(t, fakeMapper, u)
err := baseFakeClient.Tracker().Create(gvr, u, u.GetNamespace())
assert.NoError(t, err)
}
if strings.Contains(tt.name, "delet") {
timeUntilDelete := time.Millisecond * 500
for _, obj := range objs {
u := obj.(*unstructured.Unstructured)
gvr := getGVR(t, fakeMapper, u)
go func(gvr schema.GroupVersionResource, u *unstructured.Unstructured) {
time.Sleep(timeUntilDelete)
err := baseFakeClient.Tracker().Delete(gvr, u.GetNamespace(), u.GetName())
assert.NoError(t, err)
}(gvr, u)
}
}
resourceList := getResourceListFromRuntimeObjs(t, c, objs)
err := tt.testFunc(sw, resourceList, time.Second*3)
if tt.expectErrs != nil {
require.Error(t, err)
for _, expectedErr := range tt.expectErrs {
assert.Contains(t, err.Error(), expectedErr.Error())
}
return
}
assert.NoError(t, err)
assert.False(t, restrictedClient.clusterScopedListAttempted)
})
}
}

Write Preview
Loading…
Cancel
Save