diff --git a/api/kubescape/v1/groupversion_info.go b/api/kubescape/v1/groupversion_info.go new file mode 100644 index 0000000..181fc98 --- /dev/null +++ b/api/kubescape/v1/groupversion_info.go @@ -0,0 +1,20 @@ +package v1 + +import ( + "k8s.io/apimachinery/pkg/runtime/schema" + "sigs.k8s.io/controller-runtime/pkg/scheme" +) + +const ( + Group = "kubescape.io" + Version = "v1" + + SecurityExceptionKind = "SecurityException" + ClusterSecurityExceptionKind = "ClusterSecurityException" +) + +var GroupVersion = schema.GroupVersion{Group: Group, Version: Version} + +var SchemeBuilder = &scheme.Builder{GroupVersion: GroupVersion} + +var AddToScheme = SchemeBuilder.AddToScheme diff --git a/api/kubescape/v1/securityexception_types.go b/api/kubescape/v1/securityexception_types.go new file mode 100644 index 0000000..7f68821 --- /dev/null +++ b/api/kubescape/v1/securityexception_types.go @@ -0,0 +1,214 @@ +package v1 + +import ( + "k8s.io/apimachinery/pkg/runtime" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" +) + +type SecurityExceptionMatch struct { + NamespaceSelector *metav1.LabelSelector `json:"namespaceSelector,omitempty"` + ObjectSelector *metav1.LabelSelector `json:"objectSelector,omitempty"` + Resources []ResourceMatch `json:"resources,omitempty"` + Images []string `json:"images,omitempty"` +} + +type ResourceMatch struct { + APIGroup string `json:"apiGroup,omitempty"` + Kind string `json:"kind"` + Name string `json:"name,omitempty"` +} + +type VulnerabilityReference struct { + ID string `json:"id,omitempty"` +} + +type VulnerabilityException struct { + Vulnerability VulnerabilityReference `json:"vulnerability,omitempty"` + Status string `json:"status,omitempty"` + ExpiredOnFix *bool `json:"expiredOnFix,omitempty"` +} + +type PostureException struct { + ControlID string `json:"controlID,omitempty"` + Action string `json:"action,omitempty"` +} + +type SecurityExceptionSpec struct { + Author string `json:"author,omitempty"` + Reason string `json:"reason,omitempty"` + ExpiresAt *metav1.Time `json:"expiresAt,omitempty"` + Match *SecurityExceptionMatch `json:"match,omitempty"` + Vulnerabilities []VulnerabilityException `json:"vulnerabilities,omitempty"` + Posture []PostureException `json:"posture,omitempty"` +} + +type SecurityException struct { + metav1.TypeMeta `json:",inline"` + metav1.ObjectMeta `json:"metadata,omitempty"` + + Spec SecurityExceptionSpec `json:"spec"` +} + +type SecurityExceptionList struct { + metav1.TypeMeta `json:",inline"` + metav1.ListMeta `json:"metadata,omitempty"` + Items []SecurityException `json:"items"` +} + +type ClusterSecurityException struct { + metav1.TypeMeta `json:",inline"` + metav1.ObjectMeta `json:"metadata,omitempty"` + + Spec SecurityExceptionSpec `json:"spec"` +} + +type ClusterSecurityExceptionList struct { + metav1.TypeMeta `json:",inline"` + metav1.ListMeta `json:"metadata,omitempty"` + Items []ClusterSecurityException `json:"items"` +} + +func init() { + SchemeBuilder.Register( + &SecurityException{}, + &SecurityExceptionList{}, + &ClusterSecurityException{}, + &ClusterSecurityExceptionList{}, + ) +} + +func (in *SecurityExceptionMatch) DeepCopyInto(out *SecurityExceptionMatch) { + *out = *in + if in.NamespaceSelector != nil { + out.NamespaceSelector = in.NamespaceSelector.DeepCopy() + } + if in.ObjectSelector != nil { + out.ObjectSelector = in.ObjectSelector.DeepCopy() + } + if in.Resources != nil { + out.Resources = make([]ResourceMatch, len(in.Resources)) + copy(out.Resources, in.Resources) + } + if in.Images != nil { + out.Images = make([]string, len(in.Images)) + copy(out.Images, in.Images) + } +} + +func (in *SecurityExceptionSpec) DeepCopyInto(out *SecurityExceptionSpec) { + *out = *in + if in.ExpiresAt != nil { + copy := *in.ExpiresAt + out.ExpiresAt = © + } + if in.Match != nil { + out.Match = &SecurityExceptionMatch{} + in.Match.DeepCopyInto(out.Match) + } + if in.Vulnerabilities != nil { + out.Vulnerabilities = make([]VulnerabilityException, len(in.Vulnerabilities)) + copy(out.Vulnerabilities, in.Vulnerabilities) + } + if in.Posture != nil { + out.Posture = make([]PostureException, len(in.Posture)) + copy(out.Posture, in.Posture) + } +} + +func (in *SecurityException) DeepCopyInto(out *SecurityException) { + *out = *in + in.ObjectMeta.DeepCopyInto(&out.ObjectMeta) + in.Spec.DeepCopyInto(&out.Spec) +} + +func (in *SecurityException) DeepCopy() *SecurityException { + if in == nil { + return nil + } + out := new(SecurityException) + in.DeepCopyInto(out) + return out +} + +func (in *SecurityException) DeepCopyObject() runtime.Object { + if c := in.DeepCopy(); c != nil { + return c + } + return nil +} + +func (in *SecurityExceptionList) DeepCopyInto(out *SecurityExceptionList) { + *out = *in + in.ListMeta.DeepCopyInto(&out.ListMeta) + if in.Items != nil { + out.Items = make([]SecurityException, len(in.Items)) + for i := range in.Items { + in.Items[i].DeepCopyInto(&out.Items[i]) + } + } +} + +func (in *SecurityExceptionList) DeepCopy() *SecurityExceptionList { + if in == nil { + return nil + } + out := new(SecurityExceptionList) + in.DeepCopyInto(out) + return out +} + +func (in *SecurityExceptionList) DeepCopyObject() runtime.Object { + if c := in.DeepCopy(); c != nil { + return c + } + return nil +} + +func (in *ClusterSecurityException) DeepCopyInto(out *ClusterSecurityException) { + *out = *in + in.ObjectMeta.DeepCopyInto(&out.ObjectMeta) + in.Spec.DeepCopyInto(&out.Spec) +} + +func (in *ClusterSecurityException) DeepCopy() *ClusterSecurityException { + if in == nil { + return nil + } + out := new(ClusterSecurityException) + in.DeepCopyInto(out) + return out +} + +func (in *ClusterSecurityException) DeepCopyObject() runtime.Object { + if c := in.DeepCopy(); c != nil { + return c + } + return nil +} + +func (in *ClusterSecurityExceptionList) DeepCopyInto(out *ClusterSecurityExceptionList) { + *out = *in + in.ListMeta.DeepCopyInto(&out.ListMeta) + if in.Items != nil { + out.Items = make([]ClusterSecurityException, len(in.Items)) + for i := range in.Items { + in.Items[i].DeepCopyInto(&out.Items[i]) + } + } +} + +func (in *ClusterSecurityExceptionList) DeepCopy() *ClusterSecurityExceptionList { + if in == nil { + return nil + } + out := new(ClusterSecurityExceptionList) + in.DeepCopyInto(out) + return out +} + +func (in *ClusterSecurityExceptionList) DeepCopyObject() runtime.Object { + if c := in.DeepCopy(); c != nil { + return c + } + return nil +} diff --git a/controllers/cooldownqueue.go b/controllers/cooldownqueue.go new file mode 100644 index 0000000..719bda5 --- /dev/null +++ b/controllers/cooldownqueue.go @@ -0,0 +1,185 @@ +package controllers + +import ( + "sync" + "time" +) + +const defaultCooldown = 5 * time.Second + +// CooldownQueue debounces keys and emits each key once after a quiet period. +type CooldownQueue struct { + quietPeriod time.Duration + resultCh chan string + wakeCh chan struct{} + closedCh chan struct{} + timer *time.Timer + + mu sync.Mutex + pending map[string]time.Time + stopped bool + stopOnce sync.Once +} + +// NewCooldownQueue creates a queue with the default quiet period. +func NewCooldownQueue() *CooldownQueue { + return NewCooldownQueueWithParams(defaultCooldown) +} + +// NewCooldownQueueWithParams creates a queue with a custom quiet period. +func NewCooldownQueueWithParams(quietPeriod time.Duration) *CooldownQueue { + if quietPeriod <= 0 { + quietPeriod = defaultCooldown + } + q := &CooldownQueue{ + quietPeriod: quietPeriod, + resultCh: make(chan string, 64), + wakeCh: make(chan struct{}, 1), + closedCh: make(chan struct{}), + pending: map[string]time.Time{}, + } + q.timer = time.NewTimer(time.Hour) + q.timer.Stop() + go q.run() + return q +} + +// ResultChan returns the channel of debounced keys. +func (q *CooldownQueue) ResultChan() <-chan string { + return q.resultCh +} + +// Enqueue schedules a key to fire after the quiet period. +func (q *CooldownQueue) Enqueue(key string) { + if key == "" { + return + } + q.mu.Lock() + if q.stopped { + q.mu.Unlock() + return + } + q.pending[key] = time.Now().Add(q.quietPeriod) + q.mu.Unlock() + + q.signal() +} + +// Stop stops the queue and closes the output channel. +func (q *CooldownQueue) Stop() { + q.stopOnce.Do(func() { + q.mu.Lock() + q.stopped = true + q.mu.Unlock() + close(q.closedCh) + }) +} + +func (q *CooldownQueue) signal() { + select { + case q.wakeCh <- struct{}{}: + default: + } +} + +func (q *CooldownQueue) nextDeadline() (time.Time, bool) { + q.mu.Lock() + defer q.mu.Unlock() + + if len(q.pending) == 0 { + return time.Time{}, false + } + + var next time.Time + for _, deadline := range q.pending { + if next.IsZero() || deadline.Before(next) { + next = deadline + } + } + + return next, true +} + +func (q *CooldownQueue) fireDue(now time.Time) []string { + q.mu.Lock() + defer q.mu.Unlock() + + if len(q.pending) == 0 { + return nil + } + + keys := make([]string, 0, len(q.pending)) + for key, deadline := range q.pending { + if !deadline.After(now) { + keys = append(keys, key) + delete(q.pending, key) + } + } + + return keys +} + +func (q *CooldownQueue) run() { + for { + select { + case <-q.closedCh: + q.stopTimer() + close(q.resultCh) + return + default: + } + + next, ok := q.nextDeadline() + if !ok { + select { + case <-q.wakeCh: + continue + case <-q.closedCh: + q.stopTimer() + close(q.resultCh) + return + } + } + + wait := time.Until(next) + if wait < 0 { + wait = 0 + } + + q.resetTimer(wait) + + select { + case <-q.timer.C: + for _, key := range q.fireDue(time.Now()) { + q.resultCh <- key + } + case <-q.wakeCh: + continue + case <-q.closedCh: + q.stopTimer() + close(q.resultCh) + return + } + } +} + +func (q *CooldownQueue) stopTimer() { + if q.timer == nil { + return + } + if !q.timer.Stop() { + select { + case <-q.timer.C: + default: + } + } +} + +func (q *CooldownQueue) resetTimer(d time.Duration) { + if q.timer == nil { + q.timer = time.NewTimer(d) + return + } + q.stopTimer() + q.timer.Reset(d) +} diff --git a/controllers/cooldownqueue_test.go b/controllers/cooldownqueue_test.go new file mode 100644 index 0000000..2850b11 --- /dev/null +++ b/controllers/cooldownqueue_test.go @@ -0,0 +1,89 @@ +package controllers + +import ( + "testing" + "time" + + "github.com/stretchr/testify/assert" +) + +func TestCooldownQueue(t *testing.T) { + tests := []struct { + name string + run func(t *testing.T) + }{ + { + name: "single enqueue fires", + run: func(t *testing.T) { + queue := NewCooldownQueueWithParams(20 * time.Millisecond) + defer queue.Stop() + + queue.Enqueue("alpha") + + select { + case key := <-queue.ResultChan(): + assert.Equal(t, "alpha", key) + case <-time.After(200 * time.Millisecond): + t.Fatalf("timed out waiting for key") + } + }, + }, + { + name: "coalesces rapid enqueues", + run: func(t *testing.T) { + queue := NewCooldownQueueWithParams(30 * time.Millisecond) + defer queue.Stop() + + queue.Enqueue("alpha") + time.Sleep(10 * time.Millisecond) + queue.Enqueue("alpha") + time.Sleep(10 * time.Millisecond) + queue.Enqueue("alpha") + + select { + case key := <-queue.ResultChan(): + assert.Equal(t, "alpha", key) + case <-time.After(200 * time.Millisecond): + t.Fatalf("timed out waiting for key") + } + + select { + case key := <-queue.ResultChan(): + t.Fatalf("unexpected second key: %s", key) + case <-time.After(80 * time.Millisecond): + } + }, + }, + { + name: "different keys are independent", + run: func(t *testing.T) { + queue := NewCooldownQueueWithParams(20 * time.Millisecond) + defer queue.Stop() + + queue.Enqueue("alpha") + queue.Enqueue("bravo") + + received := map[string]struct{}{} + deadline := time.After(200 * time.Millisecond) + + for len(received) < 2 { + select { + case key := <-queue.ResultChan(): + received[key] = struct{}{} + case <-deadline: + t.Fatalf("timed out waiting for keys") + } + } + + _, hasAlpha := received["alpha"] + _, hasBravo := received["bravo"] + assert.True(t, hasAlpha) + assert.True(t, hasBravo) + }, + }, + } + + for _, tc := range tests { + t.Run(tc.name, tc.run) + } +} diff --git a/controllers/securityexception_watch_handler.go b/controllers/securityexception_watch_handler.go new file mode 100644 index 0000000..a3ebd9e --- /dev/null +++ b/controllers/securityexception_watch_handler.go @@ -0,0 +1,493 @@ +package controllers + +import ( + "context" + "errors" + "fmt" + "strings" + "sync" + "time" + + "github.com/armosec/armoapi-go/apis" + "github.com/armosec/armoapi-go/identifiers" + "github.com/kubescape/go-logger" + "github.com/kubescape/go-logger/helpers" + securityexceptionv1 "github.com/kubescape/operator/api/kubescape/v1" + "github.com/kubescape/operator/config" + "github.com/kubescape/operator/utils" + utilsapisv1 "github.com/kubescape/opa-utils/httpserver/apis/v1" + utilsmetav1 "github.com/kubescape/opa-utils/httpserver/meta/v1" + "github.com/panjf2000/ants/v2" + corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/labels" + "k8s.io/utils/ptr" + ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/handler" + "sigs.k8s.io/controller-runtime/pkg/reconcile" + "sigs.k8s.io/controller-runtime/pkg/source" +) + +const defaultExpiryInterval = 5 * time.Minute + +type ScanKinds struct { + Posture bool + Vulnerability bool +} + +type RescanRequest struct { + Namespaces []string + ScanKinds ScanKinds +} + +type RescanDispatcher interface { + Dispatch(ctx context.Context, req RescanRequest) error +} + +type DefaultRescanDispatcher struct { + cfg config.IConfig + workerPool *ants.PoolWithFunc +} + +func NewRescanDispatcher(cfg config.IConfig, workerPool *ants.PoolWithFunc) *DefaultRescanDispatcher { + return &DefaultRescanDispatcher{cfg: cfg, workerPool: workerPool} +} + +func (d *DefaultRescanDispatcher) Dispatch(ctx context.Context, req RescanRequest) error { + var errs []error + + for _, ns := range req.Namespaces { + if d.cfg != nil && d.cfg.SkipNamespace(ns) { + continue + } + + if req.ScanKinds.Posture && d.cfg != nil && d.cfg.Components().Kubescape.Enabled { + if err := d.dispatchPosture(ctx, ns); err != nil { + errs = append(errs, err) + } + } + if req.ScanKinds.Vulnerability && d.cfg != nil && d.cfg.Components().Kubevuln.Enabled { + if err := d.dispatchVulnerability(ctx, ns); err != nil { + errs = append(errs, err) + } + } + } + + if len(errs) == 0 { + return nil + } + return errors.Join(errs...) +} + +func (d *DefaultRescanDispatcher) dispatchPosture(ctx context.Context, namespace string) error { + cmd := &apis.Command{ + CommandName: apis.TypeRunKubescape, + Args: map[string]interface{}{ + utils.KubescapeScanV1: utilsmetav1.PostScanRequest{ + IncludeNamespaces: []string{namespace}, + TargetNames: []string{"all"}, + TargetType: utilsapisv1.KindFramework, + HostScanner: ptr.To(false), + }, + }, + } + return utils.AddCommandToChannel(ctx, d.cfg, cmd, d.workerPool) +} + +func (d *DefaultRescanDispatcher) dispatchVulnerability(ctx context.Context, namespace string) error { + designator := identifiers.PortalDesignator{ + Attributes: map[string]string{ + identifiers.AttributeNamespace: namespace, + }, + } + cmd := &apis.Command{ + CommandName: apis.TypeScanImages, + Designators: []identifiers.PortalDesignator{designator}, + } + return utils.AddCommandToChannel(ctx, d.cfg, cmd, d.workerPool) +} + +type SecurityExceptionWatchHandler struct { + client client.Client + cfg config.IConfig + dispatcher RescanDispatcher + queue *CooldownQueue + expiryInterval time.Duration + now func() time.Time + + expiredMu sync.Mutex + expiredMap map[string]struct{} +} + +type HandlerOption func(*SecurityExceptionWatchHandler) + +func WithCooldownQueue(queue *CooldownQueue) HandlerOption { + return func(h *SecurityExceptionWatchHandler) { + if queue != nil { + h.queue = queue + } + } +} + +func WithExpiryInterval(interval time.Duration) HandlerOption { + return func(h *SecurityExceptionWatchHandler) { + if interval > 0 { + h.expiryInterval = interval + } + } +} + +func WithClock(now func() time.Time) HandlerOption { + return func(h *SecurityExceptionWatchHandler) { + if now != nil { + h.now = now + } + } +} + +func NewSecurityExceptionWatchHandler(k8sClient client.Client, cfg config.IConfig, dispatcher RescanDispatcher, opts ...HandlerOption) *SecurityExceptionWatchHandler { + h := &SecurityExceptionWatchHandler{ + client: k8sClient, + cfg: cfg, + dispatcher: dispatcher, + queue: NewCooldownQueue(), + expiryInterval: defaultExpiryInterval, + now: time.Now, + expiredMap: map[string]struct{}{}, + } + for _, opt := range opts { + opt(h) + } + return h +} + +func (h *SecurityExceptionWatchHandler) Reconcile(ctx context.Context, req reconcile.Request) (ctrl.Result, error) { + kind := securityexceptionv1.SecurityExceptionKind + if req.Namespace == "" { + kind = securityexceptionv1.ClusterSecurityExceptionKind + } + + h.queue.Enqueue(makeExceptionKey(kind, req.Namespace, req.Name)) + return ctrl.Result{}, nil +} + +func (h *SecurityExceptionWatchHandler) SetupWithManager(mgr ctrl.Manager) error { + if err := ctrl.NewControllerManagedBy(mgr). + For(&securityexceptionv1.SecurityException{}). + Watches(source.Kind(mgr.GetCache(), &securityexceptionv1.ClusterSecurityException{}, &handler.EnqueueRequestForObject{})). + Complete(h); err != nil { + return err + } + return nil +} + +// Start runs the debounce worker and expiry loop. +func (h *SecurityExceptionWatchHandler) Start(ctx context.Context) error { + go h.runQueue(ctx) + if h.expiryInterval > 0 { + go h.runExpiry(ctx) + } + <-ctx.Done() + h.queue.Stop() + return nil +} + +func (h *SecurityExceptionWatchHandler) runQueue(ctx context.Context) { + for { + select { + case <-ctx.Done(): + return + case key, ok := <-h.queue.ResultChan(): + if !ok { + return + } + h.handleKey(ctx, key) + } + } +} + +func (h *SecurityExceptionWatchHandler) runExpiry(ctx context.Context) { + if err := h.checkExpired(ctx); err != nil { + logger.L().Ctx(ctx).Warning("failed initial expiry check", helpers.Error(err)) + } + + ticker := time.NewTicker(h.expiryInterval) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + if err := h.checkExpired(ctx); err != nil { + logger.L().Ctx(ctx).Warning("expiry check failed", helpers.Error(err)) + } + } + } +} + +func (h *SecurityExceptionWatchHandler) handleKey(ctx context.Context, key string) { + kind, namespace, name, err := parseExceptionKey(key) + if err != nil { + logger.L().Ctx(ctx).Warning("invalid exception key", helpers.String("key", key), helpers.Error(err)) + return + } + + spec, exists, err := h.fetchSpec(ctx, kind, namespace, name) + if err != nil { + logger.L().Ctx(ctx).Warning("failed to fetch exception", helpers.String("key", key), helpers.Error(err)) + return + } + + req := RescanRequest{ + ScanKinds: scanKindsForSpec(spec, exists), + } + + namespaces, err := h.resolveNamespaces(ctx, kind, namespace, spec, exists) + if err != nil { + logger.L().Ctx(ctx).Warning("failed to resolve namespaces", helpers.String("key", key), helpers.Error(err)) + return + } + + if len(namespaces) == 0 || h.dispatcher == nil { + return + } + + req.Namespaces = namespaces + if err := h.dispatcher.Dispatch(ctx, req); err != nil { + logger.L().Ctx(ctx).Warning("failed to dispatch rescan", helpers.String("key", key), helpers.Error(err)) + } +} + +func (h *SecurityExceptionWatchHandler) checkExpired(ctx context.Context) error { + now := h.now() + + if err := h.checkExpiredSecurityExceptions(ctx, now); err != nil { + return err + } + if err := h.checkExpiredClusterSecurityExceptions(ctx, now); err != nil { + return err + } + + return nil +} + +func (h *SecurityExceptionWatchHandler) checkExpiredSecurityExceptions(ctx context.Context, now time.Time) error { + var list securityexceptionv1.SecurityExceptionList + if err := h.client.List(ctx, &list); err != nil { + return err + } + + for i := range list.Items { + item := list.Items[i] + if !isExpired(item.Spec.ExpiresAt, now) { + h.clearExpired(makeExceptionKey(securityexceptionv1.SecurityExceptionKind, item.Namespace, item.Name)) + continue + } + + key := makeExceptionKey(securityexceptionv1.SecurityExceptionKind, item.Namespace, item.Name) + if !h.markExpired(key) { + continue + } + + if h.dispatcher == nil { + continue + } + + namespaces := h.filterNamespaces([]string{item.Namespace}) + if len(namespaces) == 0 { + continue + } + + req := RescanRequest{ + Namespaces: namespaces, + ScanKinds: scanKindsForSpec(item.Spec, true), + } + if err := h.dispatcher.Dispatch(ctx, req); err != nil { + logger.L().Ctx(ctx).Warning("failed to dispatch expired rescan", helpers.String("key", key), helpers.Error(err)) + } + } + + return nil +} + +func (h *SecurityExceptionWatchHandler) checkExpiredClusterSecurityExceptions(ctx context.Context, now time.Time) error { + var list securityexceptionv1.ClusterSecurityExceptionList + if err := h.client.List(ctx, &list); err != nil { + return err + } + + for i := range list.Items { + item := list.Items[i] + if !isExpired(item.Spec.ExpiresAt, now) { + h.clearExpired(makeExceptionKey(securityexceptionv1.ClusterSecurityExceptionKind, "", item.Name)) + continue + } + + key := makeExceptionKey(securityexceptionv1.ClusterSecurityExceptionKind, "", item.Name) + if !h.markExpired(key) { + continue + } + + if h.dispatcher == nil { + continue + } + + namespaces, err := h.resolveNamespaces(ctx, securityexceptionv1.ClusterSecurityExceptionKind, "", item.Spec, true) + if err != nil { + logger.L().Ctx(ctx).Warning("failed to resolve namespaces", helpers.String("key", key), helpers.Error(err)) + continue + } + if len(namespaces) == 0 { + continue + } + + req := RescanRequest{ + Namespaces: namespaces, + ScanKinds: scanKindsForSpec(item.Spec, true), + } + if err := h.dispatcher.Dispatch(ctx, req); err != nil { + logger.L().Ctx(ctx).Warning("failed to dispatch expired rescan", helpers.String("key", key), helpers.Error(err)) + } + } + + return nil +} + +func (h *SecurityExceptionWatchHandler) fetchSpec(ctx context.Context, kind, namespace, name string) (securityexceptionv1.SecurityExceptionSpec, bool, error) { + switch kind { + case securityexceptionv1.SecurityExceptionKind: + obj := &securityexceptionv1.SecurityException{} + err := h.client.Get(ctx, client.ObjectKey{Namespace: namespace, Name: name}, obj) + if apierrors.IsNotFound(err) { + return securityexceptionv1.SecurityExceptionSpec{}, false, nil + } + if err != nil { + return securityexceptionv1.SecurityExceptionSpec{}, false, err + } + return obj.Spec, true, nil + case securityexceptionv1.ClusterSecurityExceptionKind: + obj := &securityexceptionv1.ClusterSecurityException{} + err := h.client.Get(ctx, client.ObjectKey{Name: name}, obj) + if apierrors.IsNotFound(err) { + return securityexceptionv1.SecurityExceptionSpec{}, false, nil + } + if err != nil { + return securityexceptionv1.SecurityExceptionSpec{}, false, err + } + return obj.Spec, true, nil + default: + return securityexceptionv1.SecurityExceptionSpec{}, false, fmt.Errorf("unsupported kind: %s", kind) + } +} + +func (h *SecurityExceptionWatchHandler) resolveNamespaces(ctx context.Context, kind, namespace string, spec securityexceptionv1.SecurityExceptionSpec, exists bool) ([]string, error) { + if kind == securityexceptionv1.SecurityExceptionKind { + if namespace == "" { + return nil, fmt.Errorf("missing namespace for SecurityException") + } + return h.filterNamespaces([]string{namespace}), nil + } + + if !exists || spec.Match == nil || spec.Match.NamespaceSelector == nil { + return h.listNamespaces(ctx, labels.Everything()) + } + + selector, err := metav1.LabelSelectorAsSelector(spec.Match.NamespaceSelector) + if err != nil { + return nil, err + } + return h.listNamespaces(ctx, selector) +} + +func (h *SecurityExceptionWatchHandler) listNamespaces(ctx context.Context, selector labels.Selector) ([]string, error) { + var list corev1.NamespaceList + if selector == nil { + selector = labels.Everything() + } + if err := h.client.List(ctx, &list, client.MatchingLabelsSelector{Selector: selector}); err != nil { + return nil, err + } + + namespaces := make([]string, 0, len(list.Items)) + for i := range list.Items { + ns := list.Items[i].Name + if h.cfg != nil && h.cfg.SkipNamespace(ns) { + continue + } + namespaces = append(namespaces, ns) + } + return namespaces, nil +} + +func (h *SecurityExceptionWatchHandler) filterNamespaces(namespaces []string) []string { + if h.cfg == nil { + return namespaces + } + filtered := make([]string, 0, len(namespaces)) + for _, ns := range namespaces { + if h.cfg.SkipNamespace(ns) { + continue + } + filtered = append(filtered, ns) + } + return filtered +} + +func (h *SecurityExceptionWatchHandler) markExpired(key string) bool { + h.expiredMu.Lock() + defer h.expiredMu.Unlock() + + if _, exists := h.expiredMap[key]; exists { + return false + } + if h.expiredMap == nil { + h.expiredMap = map[string]struct{}{} + } + h.expiredMap[key] = struct{}{} + return true +} + +func (h *SecurityExceptionWatchHandler) clearExpired(key string) { + h.expiredMu.Lock() + defer h.expiredMu.Unlock() + + delete(h.expiredMap, key) +} + +func scanKindsForSpec(spec securityexceptionv1.SecurityExceptionSpec, exists bool) ScanKinds { + if !exists { + return ScanKinds{Posture: true, Vulnerability: true} + } + + kinds := ScanKinds{ + Posture: len(spec.Posture) > 0, + Vulnerability: len(spec.Vulnerabilities) > 0, + } + if !kinds.Posture && !kinds.Vulnerability { + kinds.Posture = true + kinds.Vulnerability = true + } + return kinds +} + +func isExpired(expiresAt *metav1.Time, now time.Time) bool { + if expiresAt == nil { + return false + } + return expiresAt.Time.Before(now) +} + +func makeExceptionKey(kind, namespace, name string) string { + return strings.Join([]string{kind, namespace, name}, "/") +} + +func parseExceptionKey(key string) (string, string, string, error) { + parts := strings.SplitN(key, "/", 3) + if len(parts) != 3 { + return "", "", "", fmt.Errorf("invalid key: %s", key) + } + return parts[0], parts[1], parts[2], nil +} diff --git a/controllers/securityexception_watch_handler_test.go b/controllers/securityexception_watch_handler_test.go new file mode 100644 index 0000000..c77c536 --- /dev/null +++ b/controllers/securityexception_watch_handler_test.go @@ -0,0 +1,234 @@ +package controllers + +import ( + "context" + "sync" + "testing" + "time" + + securityexceptionv1 "github.com/kubescape/operator/api/kubescape/v1" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/types" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + "sigs.k8s.io/controller-runtime/pkg/reconcile" +) + +type mockRescanDispatcher struct { + mu sync.Mutex + requests []RescanRequest +} + +func (m *mockRescanDispatcher) Dispatch(_ context.Context, req RescanRequest) error { + m.mu.Lock() + defer m.mu.Unlock() + m.requests = append(m.requests, req) + return nil +} + +func (m *mockRescanDispatcher) Requests() []RescanRequest { + m.mu.Lock() + defer m.mu.Unlock() + + out := make([]RescanRequest, len(m.requests)) + copy(out, m.requests) + return out +} + +func newTestScheme(t *testing.T) *runtime.Scheme { + scheme := runtime.NewScheme() + require.NoError(t, corev1.AddToScheme(scheme)) + require.NoError(t, securityexceptionv1.AddToScheme(scheme)) + return scheme +} + +func waitForKey(t *testing.T, queue *CooldownQueue, timeout time.Duration) string { + t.Helper() + + select { + case key := <-queue.ResultChan(): + return key + case <-time.After(timeout): + t.Fatalf("timed out waiting for debounced key") + } + return "" +} + +func TestSecurityExceptionWatchHandlerReconcile(t *testing.T) { + tests := []struct { + name string + request reconcile.Request + objects []client.Object + expectNamespaces []string + expectKinds ScanKinds + }{ + { + name: "namespaced exception targets its namespace", + request: reconcile.Request{NamespacedName: types.NamespacedName{Namespace: "team-a", Name: "se-a"}}, + objects: []client.Object{ + &securityexceptionv1.SecurityException{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: "team-a", + Name: "se-a", + }, + Spec: securityexceptionv1.SecurityExceptionSpec{ + Posture: []securityexceptionv1.PostureException{{ControlID: "C-001", Action: "exclude"}}, + }, + }, + }, + expectNamespaces: []string{"team-a"}, + expectKinds: ScanKinds{Posture: true, Vulnerability: false}, + }, + { + name: "cluster exception without selector targets all namespaces", + request: reconcile.Request{NamespacedName: types.NamespacedName{Name: "cse-all"}}, + objects: []client.Object{ + &securityexceptionv1.ClusterSecurityException{ + ObjectMeta: metav1.ObjectMeta{Name: "cse-all"}, + Spec: securityexceptionv1.SecurityExceptionSpec{ + Vulnerabilities: []securityexceptionv1.VulnerabilityException{{ + Vulnerability: securityexceptionv1.VulnerabilityReference{ID: "CVE-2026-0001"}, + Status: "approved", + }}, + }, + }, + &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: "team-a"}}, + &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: "team-b"}}, + }, + expectNamespaces: []string{"team-a", "team-b"}, + expectKinds: ScanKinds{Posture: false, Vulnerability: true}, + }, + { + name: "cluster exception namespace selector filters namespaces", + request: reconcile.Request{NamespacedName: types.NamespacedName{Name: "cse-filter"}}, + objects: []client.Object{ + &securityexceptionv1.ClusterSecurityException{ + ObjectMeta: metav1.ObjectMeta{Name: "cse-filter"}, + Spec: securityexceptionv1.SecurityExceptionSpec{ + Match: &securityexceptionv1.SecurityExceptionMatch{ + NamespaceSelector: &metav1.LabelSelector{MatchLabels: map[string]string{"team": "alpha"}}, + }, + Posture: []securityexceptionv1.PostureException{{ControlID: "C-002", Action: "exclude"}}, + }, + }, + &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: "alpha", Labels: map[string]string{"team": "alpha"}}}, + &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: "beta", Labels: map[string]string{"team": "beta"}}}, + }, + expectNamespaces: []string{"alpha"}, + expectKinds: ScanKinds{Posture: true, Vulnerability: false}, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + scheme := newTestScheme(t) + k8sClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(tc.objects...).Build() + dispatcher := &mockRescanDispatcher{} + queue := NewCooldownQueueWithParams(10 * time.Millisecond) + defer queue.Stop() + + h := NewSecurityExceptionWatchHandler(k8sClient, nil, dispatcher, WithCooldownQueue(queue)) + + _, err := h.Reconcile(context.Background(), tc.request) + require.NoError(t, err) + + key := waitForKey(t, queue, 200*time.Millisecond) + h.handleKey(context.Background(), key) + + requests := dispatcher.Requests() + require.Len(t, requests, 1) + assert.ElementsMatch(t, tc.expectNamespaces, requests[0].Namespaces) + assert.Equal(t, tc.expectKinds, requests[0].ScanKinds) + }) + } +} + +func TestSecurityExceptionWatchHandlerExpiry(t *testing.T) { + now := time.Date(2026, 5, 20, 10, 0, 0, 0, time.UTC) + + tests := []struct { + name string + objects []client.Object + expectNamespaces []string + expectKinds ScanKinds + expectCalls int + }{ + { + name: "expired cluster exception triggers rescan once", + objects: []client.Object{ + &securityexceptionv1.ClusterSecurityException{ + ObjectMeta: metav1.ObjectMeta{Name: "cse-expired"}, + Spec: securityexceptionv1.SecurityExceptionSpec{ + ExpiresAt: &metav1.Time{Time: now.Add(-1 * time.Hour)}, + Vulnerabilities: []securityexceptionv1.VulnerabilityException{{ + Vulnerability: securityexceptionv1.VulnerabilityReference{ID: "CVE-2026-0002"}, + Status: "approved", + }}, + }, + }, + &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: "team-a"}}, + &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: "team-b"}}, + }, + expectNamespaces: []string{"team-a", "team-b"}, + expectKinds: ScanKinds{Posture: false, Vulnerability: true}, + expectCalls: 1, + }, + { + name: "non-expired exception does not trigger rescan", + objects: []client.Object{ + &securityexceptionv1.SecurityException{ + ObjectMeta: metav1.ObjectMeta{Namespace: "team-a", Name: "se-future"}, + Spec: securityexceptionv1.SecurityExceptionSpec{ + ExpiresAt: &metav1.Time{Time: now.Add(1 * time.Hour)}, + Posture: []securityexceptionv1.PostureException{{ControlID: "C-003", Action: "exclude"}}, + }, + }, + }, + expectCalls: 0, + }, + { + name: "missing expiresAt does not trigger rescan", + objects: []client.Object{ + &securityexceptionv1.SecurityException{ + ObjectMeta: metav1.ObjectMeta{Namespace: "team-a", Name: "se-no-expiry"}, + Spec: securityexceptionv1.SecurityExceptionSpec{ + Posture: []securityexceptionv1.PostureException{{ControlID: "C-004", Action: "exclude"}}, + }, + }, + }, + expectCalls: 0, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + scheme := newTestScheme(t) + k8sClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(tc.objects...).Build() + dispatcher := &mockRescanDispatcher{} + queue := NewCooldownQueueWithParams(time.Hour) + defer queue.Stop() + + h := NewSecurityExceptionWatchHandler(k8sClient, nil, dispatcher, WithCooldownQueue(queue), WithClock(func() time.Time { + return now + })) + + require.NoError(t, h.checkExpired(context.Background())) + if tc.expectCalls > 0 { + require.NoError(t, h.checkExpired(context.Background())) + } + + requests := dispatcher.Requests() + assert.Len(t, requests, tc.expectCalls) + if tc.expectCalls == 0 { + return + } + + assert.ElementsMatch(t, tc.expectNamespaces, requests[0].Namespaces) + assert.Equal(t, tc.expectKinds, requests[0].ScanKinds) + }) + } +} diff --git a/main.go b/main.go index f058f5e..d5c1878 100644 --- a/main.go +++ b/main.go @@ -25,7 +25,9 @@ import ( rulebindingcachev1 "github.com/kubescape/operator/admission/rulebinding/cache" "github.com/kubescape/operator/admission/rulesupdate" "github.com/kubescape/operator/admission/webhook" + securityexceptionv1 "github.com/kubescape/operator/api/kubescape/v1" "github.com/kubescape/operator/config" + "github.com/kubescape/operator/controllers" "github.com/kubescape/operator/mainhandler" "github.com/kubescape/operator/nodeagentautoscaler" "github.com/kubescape/operator/objectcache" @@ -33,8 +35,11 @@ import ( "github.com/kubescape/operator/servicehandler" "github.com/kubescape/operator/utils" kssc "github.com/kubescape/storage/pkg/generated/clientset/versioned" + corev1 "k8s.io/api/core/v1" "k8s.io/apimachinery/pkg/runtime" restclient "k8s.io/client-go/rest" + ctrl "sigs.k8s.io/controller-runtime" + metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server" ) //go:generate swagger generate spec -o ./docs/swagger.yaml @@ -139,6 +144,40 @@ func main() { // setup main handler mainHandler := mainhandler.NewMainHandler(operatorConfig, k8sApi, exporter, ksStorageClient) + { + scheme := runtime.NewScheme() + if err := corev1.AddToScheme(scheme); err != nil { + logger.L().Ctx(ctx).Fatal("failed to add core scheme", helpers.Error(err)) + } + if err := securityexceptionv1.AddToScheme(scheme); err != nil { + logger.L().Ctx(ctx).Fatal("failed to add security exception scheme", helpers.Error(err)) + } + + mgr, err := ctrl.NewManager(k8sConfig, ctrl.Options{ + Scheme: scheme, + Metrics: metricsserver.Options{ + BindAddress: "0", + }, + }) + if err != nil { + logger.L().Ctx(ctx).Fatal("failed to create security exception manager", helpers.Error(err)) + } + + dispatcher := controllers.NewRescanDispatcher(operatorConfig, mainHandler.EventWorkerPool()) + handler := controllers.NewSecurityExceptionWatchHandler(mgr.GetClient(), operatorConfig, dispatcher) + if err := handler.SetupWithManager(mgr); err != nil { + logger.L().Ctx(ctx).Fatal("failed to setup security exception handler", helpers.Error(err)) + } + if err := mgr.Add(handler); err != nil { + logger.L().Ctx(ctx).Fatal("failed to register security exception handler", helpers.Error(err)) + } + + go func() { + if err := mgr.Start(ctx); err != nil { + logger.L().Ctx(ctx).Fatal("security exception manager stopped", helpers.Error(err)) + } + }() + } go func() { // open a REST API connection listener restAPIHandler := restapihandler.NewHTTPHandler(mainHandler.EventWorkerPool(), operatorConfig)