mirror of
https://github.com/k3s-io/kubernetes.git
synced 2025-08-01 07:47:56 +00:00
refactor: create generic policy plugin type similar to webhook
This commit is contained in:
Alexander Zielenski
parent
d697f43d73
commit
a6366573d5
@ -0,0 +1,38 @@
|
||||
/*
|
||||
Copyright 2024 The Kubernetes Authors.
|
||||
|
||||
Licensed under the Apache License, Version 2.0 (the "License");
|
||||
you may not use this file except in compliance with the License.
|
||||
You may obtain a copy of the License at
|
||||
|
||||
http://www.apache.org/licenses/LICENSE-2.0
|
||||
|
||||
Unless required by applicable law or agreed to in writing, software
|
||||
distributed under the License is distributed on an "AS IS" BASIS,
|
||||
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
See the License for the specific language governing permissions and
|
||||
limitations under the License.
|
||||
*/
|
||||
|
||||
package generic
|
||||
|
||||
import (
|
||||
"k8s.io/apimachinery/pkg/runtime/schema"
|
||||
"k8s.io/apimachinery/pkg/types"
|
||||
)
|
||||
|
||||
type PolicyAccessor interface {
|
||||
GetName() string
|
||||
GetNamespace() string
|
||||
GetParamKind() *schema.GroupVersionKind
|
||||
}
|
||||
|
||||
type BindingAccessor interface {
|
||||
GetName() string
|
||||
GetNamespace() string
|
||||
|
||||
// GetPolicyName returns the name of the (Validating/Mutating)AdmissionPolicy,
|
||||
// which is cluster-scoped, so namespace is usually left blank.
|
||||
// But we leave the door open to add a namespaced vesion in the future
|
||||
GetPolicyName() types.NamespacedName
|
||||
}
|
||||
@ -0,0 +1,64 @@
|
||||
/*
|
||||
Copyright 2024 The Kubernetes Authors.
|
||||
|
||||
Licensed under the Apache License, Version 2.0 (the "License");
|
||||
you may not use this file except in compliance with the License.
|
||||
You may obtain a copy of the License at
|
||||
|
||||
http://www.apache.org/licenses/LICENSE-2.0
|
||||
|
||||
Unless required by applicable law or agreed to in writing, software
|
||||
distributed under the License is distributed on an "AS IS" BASIS,
|
||||
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
See the License for the specific language governing permissions and
|
||||
limitations under the License.
|
||||
*/
|
||||
|
||||
package generic
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"k8s.io/apiserver/pkg/admission"
|
||||
)
|
||||
|
||||
// Hook represents a dynamic admission hook. The hook may be a webhook or a
|
||||
// policy. For webhook, the Hook may describe how to contact the endpoint, expected
|
||||
// cert, etc. For policies, the hook may describe a compiled policy-binding pair.
|
||||
type Hook interface {
|
||||
// All hooks are expected to contain zero or more match conditions, object
|
||||
// selectors, namespace selectors to help the dispatcher decide when to apply
|
||||
// the hook.
|
||||
//
|
||||
// Methods of matching logic is applied are specific to the hook and left up
|
||||
// to the implementation.
|
||||
}
|
||||
|
||||
// Source can list dynamic admission plugins.
|
||||
type Source[H Hook] interface {
|
||||
// Hooks returns the list of currently known admission hooks.
|
||||
Hooks() []H
|
||||
|
||||
// Run the source. This method should be called only once at startup.
|
||||
Run(ctx context.Context) error
|
||||
|
||||
// HasSynced returns true if the source has completed its initial sync.
|
||||
HasSynced() bool
|
||||
}
|
||||
|
||||
// Dispatcher dispatches evaluates an admission request against the currently
|
||||
// active hooks returned by the source.
|
||||
type Dispatcher[H Hook] interface {
|
||||
// Dispatch a request to the policies. Dispatcher may choose not to
|
||||
// call a hook, either because the rules of the hook does not match, or
|
||||
// the namespaceSelector or the objectSelector of the hook does not
|
||||
// match. A non-nil error means the request is rejected.
|
||||
Dispatch(ctx context.Context, a admission.Attributes, o admission.ObjectInterfaces, hooks []H) error
|
||||
}
|
||||
|
||||
// An evaluator represents a compiled CEL expression that can be evaluated a
|
||||
// given a set of inputs used by the generic PolicyHook for Mutating and
|
||||
// ValidatingAdmissionPolicy.
|
||||
// Mutating and Validating may have different forms of evaluators
|
||||
type Evaluator interface {
|
||||
}
|
||||
@ -0,0 +1,206 @@
|
||||
/*
|
||||
Copyright 2024 The Kubernetes Authors.
|
||||
|
||||
Licensed under the Apache License, Version 2.0 (the "License");
|
||||
you may not use this file except in compliance with the License.
|
||||
You may obtain a copy of the License at
|
||||
|
||||
http://www.apache.org/licenses/LICENSE-2.0
|
||||
|
||||
Unless required by applicable law or agreed to in writing, software
|
||||
distributed under the License is distributed on an "AS IS" BASIS,
|
||||
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
See the License for the specific language governing permissions and
|
||||
limitations under the License.
|
||||
*/
|
||||
|
||||
package generic
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
|
||||
"k8s.io/apimachinery/pkg/api/meta"
|
||||
"k8s.io/apimachinery/pkg/runtime/schema"
|
||||
utilruntime "k8s.io/apimachinery/pkg/util/runtime"
|
||||
"k8s.io/apiserver/pkg/admission"
|
||||
"k8s.io/apiserver/pkg/admission/initializer"
|
||||
"k8s.io/apiserver/pkg/admission/plugin/policy/matching"
|
||||
"k8s.io/apiserver/pkg/authorization/authorizer"
|
||||
"k8s.io/client-go/dynamic"
|
||||
"k8s.io/client-go/informers"
|
||||
"k8s.io/client-go/kubernetes"
|
||||
)
|
||||
|
||||
// H is the Hook type generated by the source and consumed by the dispatcher.
|
||||
type sourceFactory[H any] func(informers.SharedInformerFactory, kubernetes.Interface, dynamic.Interface, meta.RESTMapper) Source[H]
|
||||
type dispatcherFactory[H any] func(authorizer.Authorizer, *matching.Matcher) Dispatcher[H]
|
||||
|
||||
type Invocation struct {
|
||||
Resource schema.GroupVersionResource
|
||||
Subresource string
|
||||
Kind schema.GroupVersionKind
|
||||
}
|
||||
|
||||
// AdmissionPolicyManager is an abstract admission plugin with all the
|
||||
// infrastructure to define Admit or Validate on-top.
|
||||
type Plugin[H any] struct {
|
||||
*admission.Handler
|
||||
|
||||
sourceFactory sourceFactory[H]
|
||||
dispatcherFactory dispatcherFactory[H]
|
||||
|
||||
source Source[H]
|
||||
dispatcher Dispatcher[H]
|
||||
matcher *matching.Matcher
|
||||
|
||||
informerFactory informers.SharedInformerFactory
|
||||
client kubernetes.Interface
|
||||
restMapper meta.RESTMapper
|
||||
dynamicClient dynamic.Interface
|
||||
stopCh <-chan struct{}
|
||||
authorizer authorizer.Authorizer
|
||||
enabled bool
|
||||
}
|
||||
|
||||
var (
|
||||
_ initializer.WantsExternalKubeInformerFactory = &Plugin[any]{}
|
||||
_ initializer.WantsExternalKubeClientSet = &Plugin[any]{}
|
||||
_ initializer.WantsRESTMapper = &Plugin[any]{}
|
||||
_ initializer.WantsDynamicClient = &Plugin[any]{}
|
||||
_ initializer.WantsDrainedNotification = &Plugin[any]{}
|
||||
_ initializer.WantsAuthorizer = &Plugin[any]{}
|
||||
_ admission.InitializationValidator = &Plugin[any]{}
|
||||
)
|
||||
|
||||
func NewPlugin[H any](
|
||||
handler *admission.Handler,
|
||||
sourceFactory sourceFactory[H],
|
||||
dispatcherFactory dispatcherFactory[H],
|
||||
) *Plugin[H] {
|
||||
return &Plugin[H]{
|
||||
Handler: handler,
|
||||
sourceFactory: sourceFactory,
|
||||
dispatcherFactory: dispatcherFactory,
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Plugin[H]) SetExternalKubeInformerFactory(f informers.SharedInformerFactory) {
|
||||
c.informerFactory = f
|
||||
}
|
||||
|
||||
func (c *Plugin[H]) SetExternalKubeClientSet(client kubernetes.Interface) {
|
||||
c.client = client
|
||||
}
|
||||
|
||||
func (c *Plugin[H]) SetRESTMapper(mapper meta.RESTMapper) {
|
||||
c.restMapper = mapper
|
||||
}
|
||||
|
||||
func (c *Plugin[H]) SetDynamicClient(client dynamic.Interface) {
|
||||
c.dynamicClient = client
|
||||
}
|
||||
|
||||
func (c *Plugin[H]) SetDrainedNotification(stopCh <-chan struct{}) {
|
||||
c.stopCh = stopCh
|
||||
}
|
||||
|
||||
func (c *Plugin[H]) SetAuthorizer(authorizer authorizer.Authorizer) {
|
||||
c.authorizer = authorizer
|
||||
}
|
||||
|
||||
func (c *Plugin[H]) SetMatcher(matcher *matching.Matcher) {
|
||||
c.matcher = matcher
|
||||
}
|
||||
|
||||
func (c *Plugin[H]) SetEnabled(enabled bool) {
|
||||
c.enabled = enabled
|
||||
}
|
||||
|
||||
// ValidateInitialization - once clientset and informer factory are provided, creates and starts the admission controller
|
||||
func (c *Plugin[H]) ValidateInitialization() error {
|
||||
// By default enabled is set to false. It is up to types which embed this
|
||||
// struct to set it to true (if feature gate is enabled, or other conditions)
|
||||
if !c.enabled {
|
||||
return nil
|
||||
}
|
||||
if c.Handler == nil {
|
||||
return errors.New("missing handler")
|
||||
}
|
||||
if c.informerFactory == nil {
|
||||
return errors.New("missing informer factory")
|
||||
}
|
||||
if c.client == nil {
|
||||
return errors.New("missing kubernetes client")
|
||||
}
|
||||
if c.restMapper == nil {
|
||||
return errors.New("missing rest mapper")
|
||||
}
|
||||
if c.dynamicClient == nil {
|
||||
return errors.New("missing dynamic client")
|
||||
}
|
||||
if c.stopCh == nil {
|
||||
return errors.New("missing stop channel")
|
||||
}
|
||||
if c.authorizer == nil {
|
||||
return errors.New("missing authorizer")
|
||||
}
|
||||
|
||||
// Use default matcher
|
||||
namespaceInformer := c.informerFactory.Core().V1().Namespaces()
|
||||
c.matcher = matching.NewMatcher(namespaceInformer.Lister(), c.client)
|
||||
|
||||
if err := c.matcher.ValidateInitialization(); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
c.source = c.sourceFactory(c.informerFactory, c.client, c.dynamicClient, c.restMapper)
|
||||
c.dispatcher = c.dispatcherFactory(c.authorizer, c.matcher)
|
||||
|
||||
pluginContext, pluginContextCancel := context.WithCancel(context.Background())
|
||||
go func() {
|
||||
defer pluginContextCancel()
|
||||
<-c.stopCh
|
||||
}()
|
||||
|
||||
go func() {
|
||||
err := c.source.Run(pluginContext)
|
||||
if err != nil && !errors.Is(err, context.Canceled) {
|
||||
utilruntime.HandleError(fmt.Errorf("policy source context unexpectedly closed: %v", err))
|
||||
}
|
||||
}()
|
||||
|
||||
c.SetReadyFunc(func() bool {
|
||||
return namespaceInformer.Informer().HasSynced() && c.source.HasSynced()
|
||||
})
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *Plugin[H]) Dispatch(
|
||||
ctx context.Context,
|
||||
a admission.Attributes,
|
||||
o admission.ObjectInterfaces,
|
||||
) (err error) {
|
||||
if !c.enabled {
|
||||
return nil
|
||||
} else if isPolicyResource(a) {
|
||||
return nil
|
||||
} else if !c.WaitForReady() {
|
||||
return admission.NewForbidden(a, fmt.Errorf("not yet ready to handle request"))
|
||||
}
|
||||
|
||||
return c.dispatcher.Dispatch(ctx, a, o, c.source.Hooks())
|
||||
}
|
||||
|
||||
func isPolicyResource(attr admission.Attributes) bool {
|
||||
gvk := attr.GetResource()
|
||||
if gvk.Group == "admissionregistration.k8s.io" {
|
||||
if gvk.Resource == "validatingadmissionpolicies" || gvk.Resource == "validatingadmissionpolicybindings" {
|
||||
return true
|
||||
} else if gvk.Resource == "mutatingadmissionpolicies" || gvk.Resource == "mutatingadmissionpolicybindings" {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
@ -0,0 +1,461 @@
|
||||
/*
|
||||
Copyright 2024 The Kubernetes Authors.
|
||||
|
||||
Licensed under the Apache License, Version 2.0 (the "License");
|
||||
you may not use this file except in compliance with the License.
|
||||
You may obtain a copy of the License at
|
||||
|
||||
http://www.apache.org/licenses/LICENSE-2.0
|
||||
|
||||
Unless required by applicable law or agreed to in writing, software
|
||||
distributed under the License is distributed on an "AS IS" BASIS,
|
||||
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
See the License for the specific language governing permissions and
|
||||
limitations under the License.
|
||||
*/
|
||||
|
||||
package generic
|
||||
|
||||
import (
|
||||
"context"
|
||||
goerrors "errors"
|
||||
"fmt"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
corev1 "k8s.io/api/core/v1"
|
||||
"k8s.io/apimachinery/pkg/api/errors"
|
||||
"k8s.io/apimachinery/pkg/api/meta"
|
||||
"k8s.io/apimachinery/pkg/labels"
|
||||
"k8s.io/apimachinery/pkg/runtime"
|
||||
"k8s.io/apimachinery/pkg/runtime/schema"
|
||||
"k8s.io/apimachinery/pkg/types"
|
||||
utilruntime "k8s.io/apimachinery/pkg/util/runtime"
|
||||
"k8s.io/apimachinery/pkg/util/wait"
|
||||
"k8s.io/apiserver/pkg/admission/plugin/policy/internal/generic"
|
||||
"k8s.io/client-go/dynamic"
|
||||
"k8s.io/client-go/dynamic/dynamicinformer"
|
||||
"k8s.io/client-go/informers"
|
||||
"k8s.io/client-go/tools/cache"
|
||||
"k8s.io/klog/v2"
|
||||
)
|
||||
|
||||
type policySource[P runtime.Object, B runtime.Object, E Evaluator] struct {
|
||||
ctx context.Context
|
||||
policyInformer generic.Informer[P]
|
||||
bindingInformer generic.Informer[B]
|
||||
restMapper meta.RESTMapper
|
||||
newPolicyAccessor func(P) PolicyAccessor
|
||||
newBindingAccessor func(B) BindingAccessor
|
||||
|
||||
informerFactory informers.SharedInformerFactory
|
||||
dynamicClient dynamic.Interface
|
||||
|
||||
compiler func(P) E
|
||||
|
||||
// Currently compiled list of valid/active policy-binding pairs
|
||||
policies atomic.Pointer[[]PolicyHook[P, B, E]]
|
||||
// Whether the cache of policies is dirty and needs to be recompiled
|
||||
policiesDirty atomic.Bool
|
||||
|
||||
lock sync.Mutex
|
||||
compiledPolicies map[types.NamespacedName]compiledPolicyEntry[E]
|
||||
|
||||
// Temporary until we use the dynamic informer factory
|
||||
paramsCRDControllers map[schema.GroupVersionKind]*paramInfo
|
||||
}
|
||||
|
||||
type paramInfo struct {
|
||||
mapping meta.RESTMapping
|
||||
|
||||
// When the param is changed, or the informer is done being used, the cancel
|
||||
// func should be called to stop/cleanup the original informer
|
||||
cancelFunc func()
|
||||
|
||||
// The lister for this param
|
||||
informer informers.GenericInformer
|
||||
}
|
||||
|
||||
type compiledPolicyEntry[E Evaluator] struct {
|
||||
policyVersion string
|
||||
evaluator E
|
||||
}
|
||||
|
||||
type PolicyHook[P runtime.Object, B runtime.Object, E Evaluator] struct {
|
||||
Policy P
|
||||
Bindings []B
|
||||
ParamInformer informers.GenericInformer
|
||||
ParamScope meta.RESTScope
|
||||
|
||||
Evaluator E
|
||||
ConfigurationError error
|
||||
}
|
||||
|
||||
var _ Source[PolicyHook[runtime.Object, runtime.Object, Evaluator]] = &policySource[runtime.Object, runtime.Object, Evaluator]{}
|
||||
|
||||
func NewPolicySource[P runtime.Object, B runtime.Object, E Evaluator](
|
||||
policyInformer cache.SharedIndexInformer,
|
||||
bindingInformer cache.SharedIndexInformer,
|
||||
newPolicyAccessor func(P) PolicyAccessor,
|
||||
newBindingAccessor func(B) BindingAccessor,
|
||||
compiler func(P) E,
|
||||
paramInformerFactory informers.SharedInformerFactory,
|
||||
dynamicClient dynamic.Interface,
|
||||
restMapper meta.RESTMapper,
|
||||
) Source[PolicyHook[P, B, E]] {
|
||||
res := &policySource[P, B, E]{
|
||||
compiler: compiler,
|
||||
policyInformer: generic.NewInformer[P](policyInformer),
|
||||
bindingInformer: generic.NewInformer[B](bindingInformer),
|
||||
compiledPolicies: map[types.NamespacedName]compiledPolicyEntry[E]{},
|
||||
newPolicyAccessor: newPolicyAccessor,
|
||||
newBindingAccessor: newBindingAccessor,
|
||||
paramsCRDControllers: map[schema.GroupVersionKind]*paramInfo{},
|
||||
informerFactory: paramInformerFactory,
|
||||
dynamicClient: dynamicClient,
|
||||
restMapper: restMapper,
|
||||
}
|
||||
return res
|
||||
}
|
||||
|
||||
func (s *policySource[P, B, E]) Run(ctx context.Context) error {
|
||||
if s.ctx != nil {
|
||||
return fmt.Errorf("policy source already running")
|
||||
}
|
||||
|
||||
// Wait for initial cache sync of policies and informers before reconciling
|
||||
// any
|
||||
if !cache.WaitForNamedCacheSync(fmt.Sprintf("%T", s), ctx.Done(), s.UpstreamHasSynced) {
|
||||
err := ctx.Err()
|
||||
if err == nil {
|
||||
err = fmt.Errorf("initial cache sync for %T failed", s)
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
s.ctx = ctx
|
||||
|
||||
// Perform initial policy compilation after initial list has finished
|
||||
s.notify()
|
||||
s.refreshPolicies()
|
||||
|
||||
notifyFuncs := cache.ResourceEventHandlerFuncs{
|
||||
AddFunc: func(_ interface{}) {
|
||||
s.notify()
|
||||
},
|
||||
UpdateFunc: func(_, _ interface{}) {
|
||||
s.notify()
|
||||
},
|
||||
DeleteFunc: func(_ interface{}) {
|
||||
s.notify()
|
||||
},
|
||||
}
|
||||
handle, err := s.policyInformer.AddEventHandler(notifyFuncs)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer func() {
|
||||
if err := s.policyInformer.RemoveEventHandler(handle); err != nil {
|
||||
utilruntime.HandleError(fmt.Errorf("failed to remove policy event handler: %v", err))
|
||||
}
|
||||
}()
|
||||
|
||||
bindingHandle, err := s.bindingInformer.AddEventHandler(notifyFuncs)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer func() {
|
||||
if err := s.bindingInformer.RemoveEventHandler(bindingHandle); err != nil {
|
||||
utilruntime.HandleError(fmt.Errorf("failed to remove binding event handler: %v", err))
|
||||
}
|
||||
}()
|
||||
|
||||
// Start a worker that checks every second to see if policy data is dirty
|
||||
// and needs to be recompiled
|
||||
go func() {
|
||||
// Loop every 1 second until context is cancelled, refreshing policies
|
||||
wait.Until(s.refreshPolicies, 1*time.Second, ctx.Done())
|
||||
}()
|
||||
|
||||
<-ctx.Done()
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *policySource[P, B, E]) UpstreamHasSynced() bool {
|
||||
return s.policyInformer.HasSynced() && s.bindingInformer.HasSynced()
|
||||
}
|
||||
|
||||
// HasSynced implements Source.
|
||||
func (s *policySource[P, B, E]) HasSynced() bool {
|
||||
// As an invariant we never store `nil` into the atomic list of processed
|
||||
// policy hooks. If it is nil, then we haven't compiled all the policies
|
||||
// and stored them yet.
|
||||
return s.Hooks() != nil
|
||||
}
|
||||
|
||||
// Hooks implements Source.
|
||||
func (s *policySource[P, B, E]) Hooks() []PolicyHook[P, B, E] {
|
||||
res := s.policies.Load()
|
||||
|
||||
// Error case should not happen since evaluation function never
|
||||
// returns error
|
||||
if res == nil {
|
||||
// Not yet synced
|
||||
return nil
|
||||
}
|
||||
|
||||
return *res
|
||||
}
|
||||
|
||||
func (s *policySource[P, B, E]) refreshPolicies() {
|
||||
if !s.UpstreamHasSynced() {
|
||||
return
|
||||
} else if !s.policiesDirty.Swap(false) {
|
||||
return
|
||||
}
|
||||
|
||||
// It is ok the cache gets marked dirty again between us clearing the
|
||||
// flag and us calculating the policies. The dirty flag would be marked again,
|
||||
// and we'd have a no-op after comparing resource versions on the next sync.
|
||||
klog.Infof("refreshing policies")
|
||||
policies, err := s.calculatePolicyData()
|
||||
|
||||
// Intentionally store policy list regardless of error. There may be
|
||||
// an error returned if there was a configuration error in one of the policies,
|
||||
// but we would still want those policies evaluated
|
||||
// (for instance to return error on failaction). Or if there was an error
|
||||
// listing all policies at all, we would want to wipe the list.
|
||||
s.policies.Store(&policies)
|
||||
|
||||
if err != nil {
|
||||
// An error was generated while syncing policies. Mark it as dirty again
|
||||
// so we can retry later
|
||||
utilruntime.HandleError(fmt.Errorf("encountered error syncing policies: %v. Rescheduling policy sync", err))
|
||||
s.notify()
|
||||
}
|
||||
}
|
||||
|
||||
func (s *policySource[P, B, E]) notify() {
|
||||
s.policiesDirty.Store(true)
|
||||
}
|
||||
|
||||
// calculatePolicyData calculates the list of policies and bindings for each
|
||||
// policy. If there is an error in generation, it will return the error and
|
||||
// the partial list of policies that were able to be generated. Policies that
|
||||
// have an error will have a non-nil ConfigurationError field, but still be
|
||||
// included in the result.
|
||||
//
|
||||
// This function caches the result of the intermediate compilations
|
||||
func (s *policySource[P, B, E]) calculatePolicyData() ([]PolicyHook[P, B, E], error) {
|
||||
if !s.UpstreamHasSynced() {
|
||||
return nil, fmt.Errorf("cannot calculate policy data until upstream has synced")
|
||||
}
|
||||
|
||||
// Fat-fingered lock that can be made more fine-tuned if required
|
||||
s.lock.Lock()
|
||||
defer s.lock.Unlock()
|
||||
|
||||
// Create a local copy of all policies and bindings
|
||||
policiesToBindings := map[types.NamespacedName][]B{}
|
||||
bindingList, err := s.bindingInformer.List(labels.Everything())
|
||||
if err != nil {
|
||||
// This should never happen unless types are misconfigured
|
||||
// (can't use meta.accessor on them)
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// Gather a list of all active policy bindings
|
||||
for _, bindingSpec := range bindingList {
|
||||
bindingAccessor := s.newBindingAccessor(bindingSpec)
|
||||
policyKey := bindingAccessor.GetPolicyName()
|
||||
|
||||
// Add this binding to the list of bindings for this policy
|
||||
policiesToBindings[policyKey] = append(policiesToBindings[policyKey], bindingSpec)
|
||||
}
|
||||
|
||||
result := make([]PolicyHook[P, B, E], 0, len(bindingList))
|
||||
usedParams := map[schema.GroupVersionKind]struct{}{}
|
||||
var errs []error
|
||||
for policyKey, bindingSpecs := range policiesToBindings {
|
||||
var inf generic.NamespacedLister[P] = s.policyInformer
|
||||
if len(policyKey.Namespace) > 0 {
|
||||
inf = s.policyInformer.Namespaced(policyKey.Namespace)
|
||||
}
|
||||
policySpec, err := inf.Get(policyKey.Name)
|
||||
if errors.IsNotFound(err) {
|
||||
// Policy for bindings doesn't exist. This can happen if the policy
|
||||
// was deleted before the binding, or the binding was created first.
|
||||
//
|
||||
// Just skip bindings that refer to non-existent policies
|
||||
// If the policy is recreated, the cache will be marked dirty and
|
||||
// this function will run again.
|
||||
continue
|
||||
} else if err != nil {
|
||||
// This should never happen since fetching from a cache should never
|
||||
// fail and this function checks that the cache was synced before
|
||||
// even getting to this point.
|
||||
errs = append(errs, err)
|
||||
continue
|
||||
}
|
||||
|
||||
policyAccessor := s.newPolicyAccessor(policySpec)
|
||||
paramInformer, paramScope, configurationError := s.ensureParamsForPolicyLocked(policyAccessor.GetParamKind())
|
||||
result = append(result, PolicyHook[P, B, E]{
|
||||
Policy: policySpec,
|
||||
Bindings: bindingSpecs,
|
||||
Evaluator: s.compilePolicyLocked(policySpec),
|
||||
ParamInformer: paramInformer,
|
||||
ParamScope: paramScope,
|
||||
ConfigurationError: configurationError,
|
||||
})
|
||||
|
||||
// TEMPORARY UNTIL WE HAVE SHARED PARAM INFORMERS
|
||||
if paramKind := policyAccessor.GetParamKind(); paramKind != nil {
|
||||
usedParams[*paramKind] = struct{}{}
|
||||
}
|
||||
|
||||
// Should queue a re-sync for policy sync error. If our shared param
|
||||
// informer can notify us when CRD discovery changes we can remove this
|
||||
// and just rely on the informer to notify us when the CRDs change
|
||||
if configurationError != nil {
|
||||
errs = append(errs, configurationError)
|
||||
}
|
||||
}
|
||||
|
||||
// Clean up orphaned policies by replacing the old cache of compiled policies
|
||||
// (the map of used policies is updated by `compilePolicy`)
|
||||
for policyKey := range s.compiledPolicies {
|
||||
if _, wasSeen := policiesToBindings[policyKey]; !wasSeen {
|
||||
delete(s.compiledPolicies, policyKey)
|
||||
}
|
||||
}
|
||||
|
||||
// Clean up orphaned param informers
|
||||
for paramKind, info := range s.paramsCRDControllers {
|
||||
if _, wasSeen := usedParams[paramKind]; !wasSeen {
|
||||
info.cancelFunc()
|
||||
delete(s.paramsCRDControllers, paramKind)
|
||||
}
|
||||
}
|
||||
|
||||
err = nil
|
||||
if len(errs) > 0 {
|
||||
err = goerrors.Join(errs...)
|
||||
}
|
||||
return result, err
|
||||
}
|
||||
|
||||
// ensureParamsForPolicyLocked ensures that the informer for the paramKind is
|
||||
// started and returns the informer and the scope of the paramKind.
|
||||
//
|
||||
// Must be called under write lock
|
||||
func (s *policySource[P, B, E]) ensureParamsForPolicyLocked(paramSource *schema.GroupVersionKind) (informers.GenericInformer, meta.RESTScope, error) {
|
||||
if paramSource == nil {
|
||||
return nil, nil, nil
|
||||
} else if info, ok := s.paramsCRDControllers[*paramSource]; ok {
|
||||
return info.informer, info.mapping.Scope, nil
|
||||
}
|
||||
|
||||
mapping, err := s.restMapper.RESTMapping(schema.GroupKind{
|
||||
Group: paramSource.Group,
|
||||
Kind: paramSource.Kind,
|
||||
}, paramSource.Version)
|
||||
|
||||
if err != nil {
|
||||
// Failed to resolve. Return error so we retry again (rate limited)
|
||||
// Save a record of this definition with an evaluator that unconditionally
|
||||
return nil, nil, fmt.Errorf("failed to find resource referenced by paramKind: '%v'", *paramSource)
|
||||
}
|
||||
|
||||
// We are not watching this param. Start an informer for it.
|
||||
instanceContext, instanceCancel := context.WithCancel(s.ctx)
|
||||
|
||||
var informer informers.GenericInformer
|
||||
|
||||
// Try to see if our provided informer factory has an informer for this type.
|
||||
// We assume the informer is already started, and starts all types associated
|
||||
// with it.
|
||||
if genericInformer, err := s.informerFactory.ForResource(mapping.Resource); err == nil {
|
||||
informer = genericInformer
|
||||
|
||||
// Start the informer
|
||||
s.informerFactory.Start(instanceContext.Done())
|
||||
|
||||
} else {
|
||||
// Dynamic JSON informer fallback.
|
||||
// Cannot use shared dynamic informer since it would be impossible
|
||||
// to clean CRD informers properly with multiple dependents
|
||||
// (cannot start ahead of time, and cannot track dependencies via stopCh)
|
||||
informer = dynamicinformer.NewFilteredDynamicInformer(
|
||||
s.dynamicClient,
|
||||
mapping.Resource,
|
||||
corev1.NamespaceAll,
|
||||
// Use same interval as is used for k8s typed sharedInformerFactory
|
||||
// https://github.com/kubernetes/kubernetes/blob/7e0923899fed622efbc8679cca6b000d43633e38/cmd/kube-apiserver/app/server.go#L430
|
||||
10*time.Minute,
|
||||
cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc},
|
||||
nil,
|
||||
)
|
||||
go informer.Informer().Run(instanceContext.Done())
|
||||
}
|
||||
|
||||
klog.Infof("informer started for %v", *paramSource)
|
||||
ret := ¶mInfo{
|
||||
mapping: *mapping,
|
||||
cancelFunc: instanceCancel,
|
||||
informer: informer,
|
||||
}
|
||||
s.paramsCRDControllers[*paramSource] = ret
|
||||
return ret.informer, mapping.Scope, nil
|
||||
}
|
||||
|
||||
// For testing
|
||||
func (s *policySource[P, B, E]) getParamInformer(param schema.GroupVersionKind) (informers.GenericInformer, meta.RESTScope) {
|
||||
s.lock.Lock()
|
||||
defer s.lock.Unlock()
|
||||
|
||||
if info, ok := s.paramsCRDControllers[param]; ok {
|
||||
return info.informer, info.mapping.Scope
|
||||
}
|
||||
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
// compilePolicyLocked compiles the policy and returns the evaluator for it.
|
||||
// If the policy has not changed since the last compilation, it will return
|
||||
// the cached evaluator.
|
||||
//
|
||||
// Must be called under write lock
|
||||
func (s *policySource[P, B, E]) compilePolicyLocked(policySpec P) E {
|
||||
policyMeta, err := meta.Accessor(policySpec)
|
||||
if err != nil {
|
||||
// This should not happen if P, and B have ObjectMeta, but
|
||||
// unfortunately there is no way to express "able to call
|
||||
// meta.Accessor" as a type constraint
|
||||
utilruntime.HandleError(err)
|
||||
var emptyEvaluator E
|
||||
return emptyEvaluator
|
||||
}
|
||||
key := types.NamespacedName{
|
||||
Namespace: policyMeta.GetNamespace(),
|
||||
Name: policyMeta.GetName(),
|
||||
}
|
||||
|
||||
compiledPolicy, wasCompiled := s.compiledPolicies[key]
|
||||
|
||||
// If the policy or binding has changed since it was last compiled,
|
||||
// and if there is no configuration error (like a missing param CRD)
|
||||
// then we recompile
|
||||
if !wasCompiled ||
|
||||
compiledPolicy.policyVersion != policyMeta.GetResourceVersion() {
|
||||
|
||||
compiledPolicy = compiledPolicyEntry[E]{
|
||||
policyVersion: policyMeta.GetResourceVersion(),
|
||||
evaluator: s.compiler(policySpec),
|
||||
}
|
||||
s.compiledPolicies[key] = compiledPolicy
|
||||
}
|
||||
|
||||
return compiledPolicy.evaluator
|
||||
}
|
||||
@ -0,0 +1,233 @@
|
||||
/*
|
||||
Copyright 2024 The Kubernetes Authors.
|
||||
|
||||
Licensed under the Apache License, Version 2.0 (the "License");
|
||||
you may not use this file except in compliance with the License.
|
||||
You may obtain a copy of the License at
|
||||
|
||||
http://www.apache.org/licenses/LICENSE-2.0
|
||||
|
||||
Unless required by applicable law or agreed to in writing, software
|
||||
distributed under the License is distributed on an "AS IS" BASIS,
|
||||
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
See the License for the specific language governing permissions and
|
||||
limitations under the License.
|
||||
*/
|
||||
|
||||
package generic_test
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/require"
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
"k8s.io/apimachinery/pkg/runtime"
|
||||
"k8s.io/apimachinery/pkg/runtime/schema"
|
||||
"k8s.io/apimachinery/pkg/types"
|
||||
"k8s.io/apiserver/pkg/admission/plugin/policy/generic"
|
||||
"k8s.io/apiserver/pkg/admission/plugin/policy/matching"
|
||||
"k8s.io/apiserver/pkg/authorization/authorizer"
|
||||
"k8s.io/client-go/tools/cache"
|
||||
)
|
||||
|
||||
func makeTestDispatcher(authorizer.Authorizer, *matching.Matcher) generic.Dispatcher[generic.PolicyHook[*FakePolicy, *FakeBinding, generic.Evaluator]] {
|
||||
return nil
|
||||
}
|
||||
|
||||
func TestPolicySourceHasSyncedEmpty(t *testing.T) {
|
||||
testContext, testCancel, err := generic.NewPolicyTestContext(
|
||||
func(fp *FakePolicy) generic.PolicyAccessor { return fp },
|
||||
func(fb *FakeBinding) generic.BindingAccessor { return fb },
|
||||
func(fp *FakePolicy) generic.Evaluator { return nil },
|
||||
makeTestDispatcher,
|
||||
nil,
|
||||
nil,
|
||||
)
|
||||
require.NoError(t, err)
|
||||
defer testCancel()
|
||||
require.NoError(t, testContext.Start())
|
||||
|
||||
// Should be able to wait for cache sync
|
||||
require.True(t, cache.WaitForCacheSync(testContext.Done(), testContext.Source.HasSynced), "cache should sync after informer running")
|
||||
}
|
||||
|
||||
func TestPolicySourceHasSyncedInitialList(t *testing.T) {
|
||||
// Create a list of fake policies
|
||||
initialObjects := []runtime.Object{
|
||||
&FakePolicy{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
Name: "policy1",
|
||||
},
|
||||
},
|
||||
&FakeBinding{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
Name: "binding1",
|
||||
},
|
||||
PolicyName: "policy1",
|
||||
},
|
||||
}
|
||||
|
||||
testContext, testCancel, err := generic.NewPolicyTestContext(
|
||||
func(fp *FakePolicy) generic.PolicyAccessor { return fp },
|
||||
func(fb *FakeBinding) generic.BindingAccessor { return fb },
|
||||
func(fp *FakePolicy) generic.Evaluator { return nil },
|
||||
makeTestDispatcher,
|
||||
initialObjects,
|
||||
nil,
|
||||
)
|
||||
require.NoError(t, err)
|
||||
defer testCancel()
|
||||
require.NoError(t, testContext.Start())
|
||||
// Should be able to wait for cache sync
|
||||
require.Len(t, testContext.Source.Hooks(), 1, "should have one policy")
|
||||
require.NoError(t, testContext.UpdateAndWait(
|
||||
&FakePolicy{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
Name: "policy2",
|
||||
},
|
||||
},
|
||||
&FakeBinding{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
Name: "binding2",
|
||||
},
|
||||
PolicyName: "policy2",
|
||||
},
|
||||
))
|
||||
require.Len(t, testContext.Source.Hooks(), 2, "should have two policies")
|
||||
require.NoError(t, testContext.UpdateAndWait(
|
||||
&FakePolicy{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
Name: "policy3",
|
||||
},
|
||||
},
|
||||
&FakeBinding{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
Name: "binding3",
|
||||
},
|
||||
PolicyName: "policy3",
|
||||
},
|
||||
&FakePolicy{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
Name: "policy2",
|
||||
},
|
||||
ParamKind: &schema.GroupVersionKind{
|
||||
Group: "policy.example.com",
|
||||
Version: "v1",
|
||||
Kind: "FakeParam",
|
||||
},
|
||||
},
|
||||
))
|
||||
require.Len(t, testContext.Source.Hooks(), 3, "should have 3 policies")
|
||||
|
||||
}
|
||||
|
||||
func TestPolicySourceBindsToPolicies(t *testing.T) {
|
||||
// Create a list of fake policies
|
||||
initialObjects := []runtime.Object{
|
||||
&FakePolicy{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
Name: "policy1",
|
||||
},
|
||||
},
|
||||
&FakeBinding{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
Name: "binding1",
|
||||
},
|
||||
PolicyName: "policy1",
|
||||
},
|
||||
}
|
||||
|
||||
testContext, testCancel, err := generic.NewPolicyTestContext(
|
||||
func(fp *FakePolicy) generic.PolicyAccessor { return fp },
|
||||
func(fb *FakeBinding) generic.BindingAccessor { return fb },
|
||||
func(fp *FakePolicy) generic.Evaluator { return nil },
|
||||
makeTestDispatcher,
|
||||
initialObjects,
|
||||
nil,
|
||||
)
|
||||
require.NoError(t, err)
|
||||
defer testCancel()
|
||||
require.NoError(t, testContext.Start())
|
||||
|
||||
require.Len(t, testContext.Source.Hooks(), 1, "should have one policy")
|
||||
require.Len(t, testContext.Source.Hooks()[0].Bindings, 1, "should have one binding")
|
||||
require.Equal(t, "binding1", testContext.Source.Hooks()[0].Bindings[0].GetName(), "should have one binding")
|
||||
|
||||
// Change the binding to another policy (policies without bindings should
|
||||
// be ignored, so it should remove the first
|
||||
require.NoError(t, testContext.UpdateAndWait(
|
||||
&FakePolicy{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
Name: "policy2",
|
||||
},
|
||||
},
|
||||
&FakeBinding{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
Name: "binding1",
|
||||
},
|
||||
PolicyName: "policy2",
|
||||
}))
|
||||
require.Len(t, testContext.Source.Hooks(), 1, "should have one policy")
|
||||
require.Equal(t, "policy2", testContext.Source.Hooks()[0].Policy.GetName(), "policy name should be policy2")
|
||||
require.Len(t, testContext.Source.Hooks()[0].Bindings, 1, "should have one binding")
|
||||
require.Equal(t, "binding1", testContext.Source.Hooks()[0].Bindings[0].GetName(), "binding name should be binding1")
|
||||
|
||||
}
|
||||
|
||||
type FakePolicy struct {
|
||||
metav1.TypeMeta
|
||||
metav1.ObjectMeta
|
||||
|
||||
ParamKind *schema.GroupVersionKind
|
||||
}
|
||||
|
||||
var _ generic.PolicyAccessor = &FakePolicy{}
|
||||
|
||||
type FakeBinding struct {
|
||||
metav1.TypeMeta
|
||||
metav1.ObjectMeta
|
||||
|
||||
PolicyName string
|
||||
}
|
||||
|
||||
var _ generic.BindingAccessor = &FakeBinding{}
|
||||
|
||||
func (fp *FakePolicy) GetName() string {
|
||||
return fp.Name
|
||||
}
|
||||
|
||||
func (fp *FakePolicy) GetNamespace() string {
|
||||
return fp.Namespace
|
||||
}
|
||||
|
||||
func (fp *FakePolicy) GetParamKind() *schema.GroupVersionKind {
|
||||
return fp.ParamKind
|
||||
}
|
||||
|
||||
func (fb *FakeBinding) GetName() string {
|
||||
return fb.Name
|
||||
}
|
||||
|
||||
func (fb *FakeBinding) GetNamespace() string {
|
||||
return fb.Namespace
|
||||
}
|
||||
|
||||
func (fb *FakeBinding) GetPolicyName() types.NamespacedName {
|
||||
return types.NamespacedName{
|
||||
Name: fb.PolicyName,
|
||||
}
|
||||
}
|
||||
|
||||
func (fp *FakePolicy) DeepCopyObject() runtime.Object {
|
||||
// totally fudged deepcopy
|
||||
newFP := &FakePolicy{}
|
||||
*newFP = *fp
|
||||
return newFP
|
||||
}
|
||||
|
||||
func (fb *FakeBinding) DeepCopyObject() runtime.Object {
|
||||
// totally fudged deepcopy
|
||||
newFB := &FakeBinding{}
|
||||
*newFB = *fb
|
||||
return newFB
|
||||
}
|
||||
@ -0,0 +1,608 @@
|
||||
/*
|
||||
Copyright 2024 The Kubernetes Authors.
|
||||
|
||||
Licensed under the Apache License, Version 2.0 (the "License");
|
||||
you may not use this file except in compliance with the License.
|
||||
You may obtain a copy of the License at
|
||||
|
||||
http://www.apache.org/licenses/LICENSE-2.0
|
||||
|
||||
Unless required by applicable law or agreed to in writing, software
|
||||
distributed under the License is distributed on an "AS IS" BASIS,
|
||||
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
See the License for the specific language governing permissions and
|
||||
limitations under the License.
|
||||
*/
|
||||
|
||||
package generic
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
corev1 "k8s.io/api/core/v1"
|
||||
"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/runtime/serializer"
|
||||
"k8s.io/apimachinery/pkg/types"
|
||||
"k8s.io/apimachinery/pkg/util/uuid"
|
||||
"k8s.io/apimachinery/pkg/util/wait"
|
||||
"k8s.io/apimachinery/pkg/watch"
|
||||
"k8s.io/client-go/dynamic"
|
||||
dynamicfake "k8s.io/client-go/dynamic/fake"
|
||||
"k8s.io/client-go/informers"
|
||||
"k8s.io/client-go/kubernetes"
|
||||
"k8s.io/client-go/kubernetes/fake"
|
||||
clienttesting "k8s.io/client-go/testing"
|
||||
"k8s.io/client-go/tools/cache"
|
||||
"k8s.io/component-base/featuregate"
|
||||
|
||||
"k8s.io/apiserver/pkg/admission"
|
||||
"k8s.io/apiserver/pkg/admission/initializer"
|
||||
"k8s.io/apiserver/pkg/authorization/authorizer"
|
||||
"k8s.io/apiserver/pkg/features"
|
||||
)
|
||||
|
||||
// PolicyTestContext is everything you need to unit test a policy plugin
|
||||
type PolicyTestContext[P runtime.Object, B runtime.Object, E Evaluator] struct {
|
||||
context.Context
|
||||
Plugin *Plugin[PolicyHook[P, B, E]]
|
||||
Source Source[PolicyHook[P, B, E]]
|
||||
Start func() error
|
||||
|
||||
scheme *runtime.Scheme
|
||||
restMapper *meta.DefaultRESTMapper
|
||||
policyGVR schema.GroupVersionResource
|
||||
bindingGVR schema.GroupVersionResource
|
||||
|
||||
policyGVK schema.GroupVersionKind
|
||||
bindingGVK schema.GroupVersionKind
|
||||
|
||||
nativeTracker clienttesting.ObjectTracker
|
||||
policyAndBindingTracker clienttesting.ObjectTracker
|
||||
unstructuredTracker clienttesting.ObjectTracker
|
||||
}
|
||||
|
||||
func NewPolicyTestContext[P, B runtime.Object, E Evaluator](
|
||||
newPolicyAccessor func(P) PolicyAccessor,
|
||||
newBindingAccessor func(B) BindingAccessor,
|
||||
compileFunc func(P) E,
|
||||
dispatcher dispatcherFactory[PolicyHook[P, B, E]],
|
||||
initialObjects []runtime.Object,
|
||||
paramMappings []meta.RESTMapping,
|
||||
) (*PolicyTestContext[P, B, E], func(), error) {
|
||||
var Pexample P
|
||||
var Bexample B
|
||||
|
||||
// Create a fake resource and kind for the provided policy and binding types
|
||||
fakePolicyGVR := schema.GroupVersionResource{
|
||||
Group: "policy.example.com",
|
||||
Version: "v1",
|
||||
Resource: "fakepolicies",
|
||||
}
|
||||
fakeBindingGVR := schema.GroupVersionResource{
|
||||
Group: "policy.example.com",
|
||||
Version: "v1",
|
||||
Resource: "fakebindings",
|
||||
}
|
||||
fakePolicyGVK := fakePolicyGVR.GroupVersion().WithKind("FakePolicy")
|
||||
fakeBindingGVK := fakeBindingGVR.GroupVersion().WithKind("FakeBinding")
|
||||
|
||||
policySourceTestScheme, err := func() (*runtime.Scheme, error) {
|
||||
scheme := runtime.NewScheme()
|
||||
|
||||
if err := fake.AddToScheme(scheme); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
scheme.AddKnownTypeWithName(fakePolicyGVK, Pexample)
|
||||
scheme.AddKnownTypeWithName(fakeBindingGVK, Bexample)
|
||||
scheme.AddKnownTypeWithName(fakePolicyGVK.GroupVersion().WithKind(fakePolicyGVK.Kind+"List"), &FakeList[P]{})
|
||||
scheme.AddKnownTypeWithName(fakeBindingGVK.GroupVersion().WithKind(fakeBindingGVK.Kind+"List"), &FakeList[B]{})
|
||||
|
||||
for _, mapping := range paramMappings {
|
||||
// Skip if it is in the scheme already
|
||||
if scheme.Recognizes(mapping.GroupVersionKind) {
|
||||
continue
|
||||
}
|
||||
scheme.AddKnownTypeWithName(mapping.GroupVersionKind, &unstructured.Unstructured{})
|
||||
scheme.AddKnownTypeWithName(mapping.GroupVersionKind.GroupVersion().WithKind(mapping.GroupVersionKind.Kind+"List"), &unstructured.UnstructuredList{})
|
||||
}
|
||||
|
||||
return scheme, nil
|
||||
}()
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
|
||||
fakeRestMapper := func() *meta.DefaultRESTMapper {
|
||||
res := meta.NewDefaultRESTMapper([]schema.GroupVersion{
|
||||
{
|
||||
Group: "",
|
||||
Version: "v1",
|
||||
},
|
||||
})
|
||||
|
||||
res.Add(fakePolicyGVK, meta.RESTScopeRoot)
|
||||
res.Add(fakeBindingGVK, meta.RESTScopeRoot)
|
||||
res.Add(corev1.SchemeGroupVersion.WithKind("ConfigMap"), meta.RESTScopeNamespace)
|
||||
|
||||
for _, mapping := range paramMappings {
|
||||
res.AddSpecific(mapping.GroupVersionKind, mapping.Resource, mapping.Resource, mapping.Scope)
|
||||
}
|
||||
|
||||
return res
|
||||
}()
|
||||
|
||||
nativeClient := fake.NewSimpleClientset()
|
||||
dynamicClient := dynamicfake.NewSimpleDynamicClient(policySourceTestScheme)
|
||||
fakeInformerFactory := informers.NewSharedInformerFactory(nativeClient, 30*time.Second)
|
||||
|
||||
// Make an object tracker specifically for our policies and bindings
|
||||
policiesAndBindingsTracker := clienttesting.NewObjectTracker(
|
||||
policySourceTestScheme,
|
||||
serializer.NewCodecFactory(policySourceTestScheme).UniversalDecoder())
|
||||
|
||||
// Make an informer for our policies and bindings
|
||||
|
||||
policyInformer := cache.NewSharedIndexInformer(
|
||||
&cache.ListWatch{
|
||||
ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {
|
||||
return policiesAndBindingsTracker.List(fakePolicyGVR, fakePolicyGVK, "")
|
||||
},
|
||||
WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
|
||||
return policiesAndBindingsTracker.Watch(fakePolicyGVR, "")
|
||||
},
|
||||
},
|
||||
Pexample,
|
||||
30*time.Second,
|
||||
cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc},
|
||||
)
|
||||
bindingInformer := cache.NewSharedIndexInformer(
|
||||
&cache.ListWatch{
|
||||
ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {
|
||||
return policiesAndBindingsTracker.List(fakeBindingGVR, fakeBindingGVK, "")
|
||||
},
|
||||
WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
|
||||
return policiesAndBindingsTracker.Watch(fakeBindingGVR, "")
|
||||
},
|
||||
},
|
||||
Bexample,
|
||||
30*time.Second,
|
||||
cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc},
|
||||
)
|
||||
|
||||
var source Source[PolicyHook[P, B, E]]
|
||||
plugin := NewPlugin[PolicyHook[P, B, E]](
|
||||
admission.NewHandler(admission.Connect, admission.Create, admission.Delete, admission.Update),
|
||||
func(sif informers.SharedInformerFactory, i1 kubernetes.Interface, i2 dynamic.Interface, r meta.RESTMapper) Source[PolicyHook[P, B, E]] {
|
||||
source = NewPolicySource[P, B, E](
|
||||
policyInformer,
|
||||
bindingInformer,
|
||||
newPolicyAccessor,
|
||||
newBindingAccessor,
|
||||
compileFunc,
|
||||
sif,
|
||||
i2,
|
||||
r,
|
||||
)
|
||||
return source
|
||||
}, dispatcher)
|
||||
plugin.SetEnabled(true)
|
||||
|
||||
featureGate := featuregate.NewFeatureGate()
|
||||
err = featureGate.Add(map[featuregate.Feature]featuregate.FeatureSpec{
|
||||
//!TODO: move this to validating specific tests
|
||||
features.ValidatingAdmissionPolicy: {
|
||||
Default: true, PreRelease: featuregate.Beta}})
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
err = featureGate.SetFromMap(map[string]bool{string(features.ValidatingAdmissionPolicy): true})
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
|
||||
testContext, testCancel := context.WithCancel(context.Background())
|
||||
genericInitializer := initializer.New(
|
||||
nativeClient,
|
||||
dynamicClient,
|
||||
fakeInformerFactory,
|
||||
fakeAuthorizer{},
|
||||
featureGate,
|
||||
testContext.Done(),
|
||||
)
|
||||
genericInitializer.Initialize(plugin)
|
||||
plugin.SetRESTMapper(fakeRestMapper)
|
||||
|
||||
if err := plugin.ValidateInitialization(); err != nil {
|
||||
testCancel()
|
||||
return nil, nil, err
|
||||
}
|
||||
|
||||
res := &PolicyTestContext[P, B, E]{
|
||||
Context: testContext,
|
||||
Plugin: plugin,
|
||||
Source: source,
|
||||
|
||||
restMapper: fakeRestMapper,
|
||||
scheme: policySourceTestScheme,
|
||||
policyGVK: fakePolicyGVK,
|
||||
bindingGVK: fakeBindingGVK,
|
||||
policyGVR: fakePolicyGVR,
|
||||
bindingGVR: fakeBindingGVR,
|
||||
nativeTracker: nativeClient.Tracker(),
|
||||
policyAndBindingTracker: policiesAndBindingsTracker,
|
||||
unstructuredTracker: dynamicClient.Tracker(),
|
||||
}
|
||||
|
||||
for _, obj := range initialObjects {
|
||||
err := res.updateOne(obj)
|
||||
if err != nil {
|
||||
testCancel()
|
||||
return nil, nil, err
|
||||
}
|
||||
}
|
||||
|
||||
res.Start = func() error {
|
||||
fakeInformerFactory.Start(res.Done())
|
||||
go policyInformer.Run(res.Done())
|
||||
go bindingInformer.Run(res.Done())
|
||||
|
||||
if !cache.WaitForCacheSync(res.Done(), res.Source.HasSynced) {
|
||||
return fmt.Errorf("timed out waiting for initial cache sync")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
return res, testCancel, nil
|
||||
}
|
||||
|
||||
// UpdateAndWait updates the given object in the test, or creates it if it doesn't exist
|
||||
// Depending upon object type, waits afterward until the object is synced
|
||||
// by the policy source
|
||||
//
|
||||
// Be aware the UpdateAndWait will modify the ResourceVersion of the
|
||||
// provided objects.
|
||||
func (p *PolicyTestContext[P, B, E]) UpdateAndWait(objects ...runtime.Object) error {
|
||||
return p.update(true, objects...)
|
||||
}
|
||||
|
||||
// Update updates the given object in the test, or creates it if it doesn't exist
|
||||
//
|
||||
// Be aware the Update will modify the ResourceVersion of the
|
||||
// provided objects.
|
||||
func (p *PolicyTestContext[P, B, E]) Update(objects ...runtime.Object) error {
|
||||
return p.update(false, objects...)
|
||||
}
|
||||
|
||||
// Objects the given object in the test, or creates it if it doesn't exist
|
||||
// Depending upon object type, waits afterward until the object is synced
|
||||
// by the policy source
|
||||
func (p *PolicyTestContext[P, B, E]) update(wait bool, objects ...runtime.Object) error {
|
||||
for _, object := range objects {
|
||||
if err := p.updateOne(object); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
if wait {
|
||||
timeoutCtx, timeoutCancel := context.WithTimeout(p, 3*time.Second)
|
||||
defer timeoutCancel()
|
||||
|
||||
for _, object := range objects {
|
||||
if err := p.WaitForReconcile(timeoutCtx, object); err != nil {
|
||||
return fmt.Errorf("error waiting for reconcile of %v: %v", object, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Depending upon object type, waits afterward until the object is synced
|
||||
// by the policy source. Note that policies that are not bound are skipped,
|
||||
// so you should not try to wait for an unbound policy. Create both the binding
|
||||
// and policy, then wait.
|
||||
func (p *PolicyTestContext[P, B, E]) WaitForReconcile(timeoutCtx context.Context, object runtime.Object) error {
|
||||
if !p.Source.HasSynced() {
|
||||
return nil
|
||||
}
|
||||
|
||||
objectMeta, err := meta.Accessor(object)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
objectGVK := object.GetObjectKind().GroupVersionKind()
|
||||
switch objectGVK {
|
||||
case p.policyGVK:
|
||||
return wait.PollUntilContextCancel(timeoutCtx, 100*time.Millisecond, true, func(ctx context.Context) (done bool, err error) {
|
||||
policies := p.Source.Hooks()
|
||||
for _, policy := range policies {
|
||||
policyMeta, err := meta.Accessor(policy.Policy)
|
||||
if err != nil {
|
||||
return true, err
|
||||
} else if policyMeta.GetName() == objectMeta.GetName() && policyMeta.GetResourceVersion() == objectMeta.GetResourceVersion() {
|
||||
return true, nil
|
||||
}
|
||||
}
|
||||
return false, nil
|
||||
})
|
||||
case p.bindingGVK:
|
||||
return wait.PollUntilContextCancel(timeoutCtx, 100*time.Millisecond, true, func(ctx context.Context) (done bool, err error) {
|
||||
policies := p.Source.Hooks()
|
||||
for _, policy := range policies {
|
||||
for _, binding := range policy.Bindings {
|
||||
bindingMeta, err := meta.Accessor(binding)
|
||||
if err != nil {
|
||||
return true, err
|
||||
} else if bindingMeta.GetName() == objectMeta.GetName() && bindingMeta.GetResourceVersion() == objectMeta.GetResourceVersion() {
|
||||
return true, nil
|
||||
}
|
||||
}
|
||||
}
|
||||
return false, nil
|
||||
})
|
||||
|
||||
default:
|
||||
// Do nothing, params are visible immediately
|
||||
// Loop until one of the params is visible via get of the param informer
|
||||
return wait.PollUntilContextCancel(timeoutCtx, 100*time.Millisecond, true, func(ctx context.Context) (done bool, err error) {
|
||||
informer, scope := p.Source.(*policySource[P, B, E]).getParamInformer(objectGVK)
|
||||
if informer == nil {
|
||||
// Informer does not exist yet, keep waiting for sync
|
||||
return false, nil
|
||||
}
|
||||
|
||||
if !cache.WaitForCacheSync(timeoutCtx.Done(), informer.Informer().HasSynced) {
|
||||
return false, fmt.Errorf("timed out waiting for cache sync of param informer")
|
||||
}
|
||||
|
||||
var lister cache.GenericNamespaceLister = informer.Lister()
|
||||
if scope == meta.RESTScopeNamespace {
|
||||
lister = informer.Lister().ByNamespace(objectMeta.GetNamespace())
|
||||
}
|
||||
|
||||
fetched, err := lister.Get(objectMeta.GetName())
|
||||
if err != nil {
|
||||
if errors.IsNotFound(err) {
|
||||
return false, nil
|
||||
}
|
||||
return true, err
|
||||
}
|
||||
|
||||
// Ensure RV matches
|
||||
fetchedMeta, err := meta.Accessor(fetched)
|
||||
if err != nil {
|
||||
return true, err
|
||||
} else if fetchedMeta.GetResourceVersion() != objectMeta.GetResourceVersion() {
|
||||
return false, nil
|
||||
}
|
||||
|
||||
return true, nil
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func (p *PolicyTestContext[P, B, E]) waitForDelete(ctx context.Context, objectGVK schema.GroupVersionKind, name types.NamespacedName) error {
|
||||
srce := p.Source.(*policySource[P, B, E])
|
||||
|
||||
return wait.PollUntilContextCancel(ctx, 100*time.Millisecond, true, func(ctx context.Context) (done bool, err error) {
|
||||
switch objectGVK {
|
||||
case p.policyGVK:
|
||||
for _, hook := range p.Source.Hooks() {
|
||||
accessor := srce.newPolicyAccessor(hook.Policy)
|
||||
if accessor.GetName() == name.Name && accessor.GetNamespace() == name.Namespace {
|
||||
return false, nil
|
||||
}
|
||||
}
|
||||
|
||||
return true, nil
|
||||
case p.bindingGVK:
|
||||
for _, hook := range p.Source.Hooks() {
|
||||
for _, binding := range hook.Bindings {
|
||||
accessor := srce.newBindingAccessor(binding)
|
||||
if accessor.GetName() == name.Name && accessor.GetNamespace() == name.Namespace {
|
||||
return false, nil
|
||||
}
|
||||
}
|
||||
}
|
||||
return true, nil
|
||||
default:
|
||||
// Do nothing, params are visible immediately
|
||||
// Loop until one of the params is visible via get of the param informer
|
||||
informer, scope := p.Source.(*policySource[P, B, E]).getParamInformer(objectGVK)
|
||||
if informer == nil {
|
||||
return true, nil
|
||||
}
|
||||
|
||||
var lister cache.GenericNamespaceLister = informer.Lister()
|
||||
if scope == meta.RESTScopeNamespace {
|
||||
lister = informer.Lister().ByNamespace(name.Namespace)
|
||||
}
|
||||
|
||||
_, err = lister.Get(name.Name)
|
||||
if err != nil {
|
||||
if errors.IsNotFound(err) {
|
||||
return true, nil
|
||||
}
|
||||
return false, err
|
||||
}
|
||||
return false, nil
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
func (p *PolicyTestContext[P, B, E]) updateOne(object runtime.Object) error {
|
||||
objectMeta, err := meta.Accessor(object)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
objectMeta.SetResourceVersion(string(uuid.NewUUID()))
|
||||
objectGVK := object.GetObjectKind().GroupVersionKind()
|
||||
|
||||
if objectGVK.Empty() {
|
||||
// If the object doesn't have a GVK, ask the schema for preferred GVK
|
||||
knownKinds, _, err := p.scheme.ObjectKinds(object)
|
||||
if err != nil {
|
||||
return err
|
||||
} else if len(knownKinds) == 0 {
|
||||
return fmt.Errorf("no known GVKs for object in schema: %T", object)
|
||||
}
|
||||
toTake := 0
|
||||
|
||||
// Prefer GVK if it is our fake policy or binding
|
||||
for i, knownKind := range knownKinds {
|
||||
if knownKind == p.policyGVK || knownKind == p.bindingGVK {
|
||||
toTake = i
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
objectGVK = knownKinds[toTake]
|
||||
object.GetObjectKind().SetGroupVersionKind(objectGVK)
|
||||
}
|
||||
|
||||
// Make sure GVK is known to the fake rest mapper. To prevent cryptic error
|
||||
mapping, err := p.restMapper.RESTMapping(objectGVK.GroupKind(), objectGVK.Version)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
switch objectGVK {
|
||||
case p.policyGVK:
|
||||
err := p.policyAndBindingTracker.Update(p.policyGVR, object, objectMeta.GetNamespace())
|
||||
if errors.IsNotFound(err) {
|
||||
err = p.policyAndBindingTracker.Create(p.policyGVR, object, objectMeta.GetNamespace())
|
||||
}
|
||||
|
||||
return err
|
||||
case p.bindingGVK:
|
||||
err := p.policyAndBindingTracker.Update(p.bindingGVR, object, objectMeta.GetNamespace())
|
||||
if errors.IsNotFound(err) {
|
||||
err = p.policyAndBindingTracker.Create(p.bindingGVR, object, objectMeta.GetNamespace())
|
||||
}
|
||||
|
||||
return err
|
||||
default:
|
||||
if _, ok := object.(*unstructured.Unstructured); ok {
|
||||
if err := p.unstructuredTracker.Create(mapping.Resource, object, objectMeta.GetNamespace()); err != nil {
|
||||
if errors.IsAlreadyExists(err) {
|
||||
return p.unstructuredTracker.Update(mapping.Resource, object, objectMeta.GetNamespace())
|
||||
}
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
} else if err := p.nativeTracker.Create(mapping.Resource, object, objectMeta.GetNamespace()); err != nil {
|
||||
if errors.IsAlreadyExists(err) {
|
||||
return p.nativeTracker.Update(mapping.Resource, object, objectMeta.GetNamespace())
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
// Depending upon object type, waits afterward until the object is synced
|
||||
// by the policy source
|
||||
func (p *PolicyTestContext[P, B, E]) DeleteAndWait(object ...runtime.Object) error {
|
||||
for _, object := range object {
|
||||
if err := p.deleteOne(object); err != nil && !errors.IsNotFound(err) {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
timeoutCtx, timeoutCancel := context.WithTimeout(p, 3*time.Second)
|
||||
defer timeoutCancel()
|
||||
|
||||
for _, object := range object {
|
||||
accessor, err := meta.Accessor(object)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if err := p.waitForDelete(
|
||||
timeoutCtx,
|
||||
object.GetObjectKind().GroupVersionKind(),
|
||||
types.NamespacedName{Name: accessor.GetName(), Namespace: accessor.GetNamespace()}); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (p *PolicyTestContext[P, B, E]) deleteOne(object runtime.Object) error {
|
||||
objectMeta, err := meta.Accessor(object)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
objectMeta.SetResourceVersion(string(uuid.NewUUID()))
|
||||
objectGVK := object.GetObjectKind().GroupVersionKind()
|
||||
|
||||
if objectGVK.Empty() {
|
||||
// If the object doesn't have a GVK, ask the schema for preferred GVK
|
||||
knownKinds, _, err := p.scheme.ObjectKinds(object)
|
||||
if err != nil {
|
||||
return err
|
||||
} else if len(knownKinds) == 0 {
|
||||
return fmt.Errorf("no known GVKs for object in schema: %T", object)
|
||||
}
|
||||
toTake := 0
|
||||
|
||||
// Prefer GVK if it is our fake policy or binding
|
||||
for i, knownKind := range knownKinds {
|
||||
if knownKind == p.policyGVK || knownKind == p.bindingGVK {
|
||||
toTake = i
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
objectGVK = knownKinds[toTake]
|
||||
object.GetObjectKind().SetGroupVersionKind(objectGVK)
|
||||
}
|
||||
|
||||
// Make sure GVK is known to the fake rest mapper. To prevent cryptic error
|
||||
mapping, err := p.restMapper.RESTMapping(objectGVK.GroupKind(), objectGVK.Version)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
switch objectGVK {
|
||||
case p.policyGVK:
|
||||
return p.policyAndBindingTracker.Delete(p.policyGVR, objectMeta.GetNamespace(), objectMeta.GetName())
|
||||
case p.bindingGVK:
|
||||
return p.policyAndBindingTracker.Delete(p.bindingGVR, objectMeta.GetNamespace(), objectMeta.GetName())
|
||||
default:
|
||||
if _, ok := object.(*unstructured.Unstructured); ok {
|
||||
return p.unstructuredTracker.Delete(mapping.Resource, objectMeta.GetNamespace(), objectMeta.GetName())
|
||||
}
|
||||
return p.nativeTracker.Delete(mapping.Resource, objectMeta.GetNamespace(), objectMeta.GetName())
|
||||
}
|
||||
}
|
||||
|
||||
type FakeList[T runtime.Object] struct {
|
||||
metav1.TypeMeta
|
||||
metav1.ListMeta
|
||||
Items []T
|
||||
}
|
||||
|
||||
func (fl *FakeList[P]) DeepCopyObject() runtime.Object {
|
||||
copiedItems := make([]P, len(fl.Items))
|
||||
for i, item := range fl.Items {
|
||||
copiedItems[i] = item.DeepCopyObject().(P)
|
||||
}
|
||||
return &FakeList[P]{
|
||||
TypeMeta: fl.TypeMeta,
|
||||
ListMeta: fl.ListMeta,
|
||||
Items: copiedItems,
|
||||
}
|
||||
}
|
||||
|
||||
type fakeAuthorizer struct{}
|
||||
|
||||
func (f fakeAuthorizer) Authorize(ctx context.Context, a authorizer.Attributes) (authorizer.Decision, string, error) {
|
||||
return authorizer.DecisionAllow, "", nil
|
||||
}
|
||||
Loading…
Reference in New Issue
Block a user