authentik-kubernetes-operator/pkg/controllers/proxyprovider/controller.go

/*
Copyright 2017 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 proxyprovider

import (
	"context"
	"fmt"
	"net/http"
	"slices"
	"strconv"
	"time"

	"golang.org/x/time/rate"

	corev1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/kubernetes/scheme"
	typedcorev1 "k8s.io/client-go/kubernetes/typed/core/v1"
	"k8s.io/client-go/tools/cache"
	"k8s.io/client-go/tools/record"
	"k8s.io/client-go/util/workqueue"
	"k8s.io/klog/v2"

	"gitea.t000-n.de/t.behrendt/authentik-kubernetes-operator/internal/baseController"
	v1alpha1 "gitea.t000-n.de/t.behrendt/authentik-kubernetes-operator/pkg/apis/proxyprovider/v1alpha1"
	clientset "gitea.t000-n.de/t.behrendt/authentik-kubernetes-operator/pkg/generated/clientset/versioned"
	operatorscheme "gitea.t000-n.de/t.behrendt/authentik-kubernetes-operator/pkg/generated/clientset/versioned/scheme"
	informers "gitea.t000-n.de/t.behrendt/authentik-kubernetes-operator/pkg/generated/informers/externalversions/proxyprovider/v1alpha1"
	listers "gitea.t000-n.de/t.behrendt/authentik-kubernetes-operator/pkg/generated/listers/proxyprovider/v1alpha1"
	authentikapi "goauthentik.io/api/v3"
)

const controllerAgentName = "proxy-provider-controller"

const (
	SuccessSynced         = "Synced"
	ErrResourceExists     = "ErrResourceExists"
	MessageResourceExists = "Resource %q already exists and is not managed by ProxyProvider"
	MessageResourceSynced = "ProxyProvider synced successfully"
	FieldManager          = controllerAgentName
)

// Finalizers
const (
	DeleteAuthentikProxyProviderFinalizer = "proxyprovider.t000-n.de/delete-authentik-proxyprovider"
)

type ProxyProviderController struct {
	kubeclientset          kubernetes.Interface
	proxyProviderClientset clientset.Interface
	authentik              *authentikapi.APIClient

	proxyLister listers.ProxyProviderLister

	controller *baseController.Controller
}

func NewController(
	ctx context.Context,
	kubeclientset kubernetes.Interface,
	proxyProviderClientset clientset.Interface,
	authentik *authentikapi.APIClient,
	proxyInformer informers.ProxyProviderInformer,
) *ProxyProviderController {
	logger := klog.FromContext(ctx)

	utilruntime.Must(operatorscheme.AddToScheme(scheme.Scheme))
	logger.V(4).Info("Creating event broadcaster")

	eventBroadcaster := record.NewBroadcaster(record.WithContext(ctx))
	eventBroadcaster.StartStructuredLogging(0)
	eventBroadcaster.StartRecordingToSink(&typedcorev1.EventSinkImpl{Interface: kubeclientset.CoreV1().Events("")})
	recorder := eventBroadcaster.NewRecorder(scheme.Scheme, corev1.EventSource{Component: controllerAgentName})
	ratelimiter := workqueue.NewTypedMaxOfRateLimiter(
		workqueue.NewTypedItemExponentialFailureRateLimiter[cache.ObjectName](5*time.Millisecond, 1000*time.Second),
		&workqueue.TypedBucketRateLimiter[cache.ObjectName]{Limiter: rate.NewLimiter(rate.Limit(50), 300)},
	)

	c := &ProxyProviderController{
		kubeclientset:          kubeclientset,
		proxyProviderClientset: proxyProviderClientset,
		authentik:              authentik,
		proxyLister:            proxyInformer.Lister(),
	}
	c.controller = baseController.NewController(
		ctx,
		workqueue.NewTypedRateLimitingQueue(ratelimiter),
		recorder,
		proxyInformer.Informer().HasSynced,
		c.syncHandler,
	)

	logger.Info("Setting up event handlers")
	proxyInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
		AddFunc: c.controller.Enqueue,
		UpdateFunc: func(_, newObj interface{}) {
			c.controller.Enqueue(newObj)
		},
	})

	return c
}

func (c *ProxyProviderController) Run(ctx context.Context, workers int) error {
	return c.controller.Run(ctx, workers)
}

func (c *ProxyProviderController) syncHandler(ctx context.Context, objectRef cache.ObjectName) error {
	logger := klog.LoggerWithValues(klog.FromContext(ctx), "objectRef", objectRef)

	pp, err := c.proxyLister.ProxyProviders(objectRef.Namespace).Get(objectRef.Name)
	if err != nil {
		if errors.IsNotFound(err) {
			logger.V(4).Info("ProxyProvider no longer exists")
			return nil
		}
		return err
	}
	logger.V(4).Info("sync ProxyProvider", "name", pp.Name)

	if !pp.ObjectMeta.DeletionTimestamp.IsZero() {
		logger.Info("Reconciling deletion of ProxyProvider", "name", pp.Name)
		return c.reconcileDelete(ctx, pp)
	}

	if pp.Status.PK == "" {
		logger.Info("Reconciling creation of ProxyProvider", "name", pp.Name)
		return c.reconcileCreate(ctx, pp)
	}

	// Check if all finalizers are present. If not, we add them. Same pattern as above, just needs a helper function to check for presence of a finalizer.
	if !slices.Contains(pp.ObjectMeta.Finalizers, DeleteAuthentikProxyProviderFinalizer) {
		logger.Info("Ensuring finalizers are present", "name", pp.Name)
		return c.ensureFinalizers(ctx, pp)
	}

	logger.Info("Reconciling update of ProxyProvider", "name", pp.Name)
	return c.reconcileUpdate(ctx, pp)
}

func (c *ProxyProviderController) ensureFinalizers(ctx context.Context, pp *v1alpha1.ProxyProvider) error {
	pp.ObjectMeta.Finalizers = append(pp.ObjectMeta.Finalizers, DeleteAuthentikProxyProviderFinalizer)
	return c.updateProxyProvider(ctx, pp)
}

func (c *ProxyProviderController) reconcileDelete(ctx context.Context, pp *v1alpha1.ProxyProvider) error {
	pk, err := strconv.ParseInt(pp.Status.PK, 10, 32)
	if err != nil {
		return fmt.Errorf("error parsing PK: %v", err)
	}

	err = c.reconcileOutpost(ctx, pp.Spec.Outpost, int32(pk), ReconcileOutpostModeRemove)
	if err != nil {
		return fmt.Errorf("error when calling `reconcileOutpost`: %w", err)
	}

	// Delete ProxyProvider
	r, err := c.authentik.ProvidersApi.ProvidersProxyDestroy(ctx, int32(pk)).Execute()
	if err != nil {
		// This handles an edge-case, where when the ProxyProvider on Authentik has already been deleted, but the finalizer is still present. We just remove the finalizer and return.
		if r != nil && r.StatusCode != http.StatusNotFound {
			return fmt.Errorf("error when calling `ProvidersAPI.ProvidersProxyDestroy`: %w with response %v", err, r)
		}
	}

	pp.ObjectMeta.Finalizers = slices.Delete(pp.ObjectMeta.Finalizers, slices.Index(pp.ObjectMeta.Finalizers, DeleteAuthentikProxyProviderFinalizer), 1)
	return c.updateProxyProvider(ctx, pp)
}

func (c *ProxyProviderController) reconcileUpdate(ctx context.Context, pp *v1alpha1.ProxyProvider) error {
	// We retrieve the existing PP from the API by slug.
	pk, err := strconv.ParseInt(pp.Status.PK, 10, 32)
	if err != nil {
		return fmt.Errorf("error parsing PK: %v", err)
	}
	_, r, err := c.authentik.ProvidersApi.ProvidersAllRetrieve(ctx, int32(pk)).Execute()
	if err != nil {
		if r != nil && r.StatusCode == http.StatusNotFound {
			// This handles an edge-case, where when the PorxyProvider on Authentik has been deleted, e.g. by mistake. We just remove the PK and return.
			// During the next reconciliation, the ProxyProvider will be re-created.
			pp.Status.PK = ""
			return c.updateProxyProviderStatus(ctx, pp)
		}
		return fmt.Errorf("error retrieving existing ProxyProvider: %v with response %v", err, r)
	}

	proxyProviderRequest := &authentikapi.PatchedProxyProviderRequest{
		Name:              &pp.Spec.Name,
		AuthorizationFlow: &pp.Spec.AuthorizationFlow,
		InvalidationFlow:  &pp.Spec.InvalidationFlow,
		ExternalHost:      &pp.Spec.ExternalHost,
		Mode:              authentikapi.PROXYMODE_FORWARD_SINGLE.Ptr(),
	}
	resp, r, err := c.authentik.ProvidersApi.ProvidersProxyPartialUpdate(ctx, int32(pk)).PatchedProxyProviderRequest(*proxyProviderRequest).Execute()
	if err != nil {
		return fmt.Errorf("error when calling `ProvidersAPI.ProvidersProxyPartialUpdate`: %w with response %v", err, r)
	}

	pp.Status.PK = strconv.Itoa(int(resp.Pk))

	return c.updateProxyProviderStatus(ctx, pp)
}

func (c *ProxyProviderController) reconcileCreate(ctx context.Context, pp *v1alpha1.ProxyProvider) error {
	proxyProviderRequest := &authentikapi.ProxyProviderRequest{
		Name:              pp.Spec.Name,
		AuthorizationFlow: pp.Spec.AuthorizationFlow,
		InvalidationFlow:  pp.Spec.InvalidationFlow,
		ExternalHost:      pp.Spec.ExternalHost,
		Mode:              authentikapi.PROXYMODE_FORWARD_SINGLE.Ptr(),
	}
	resp, r, err := c.authentik.ProvidersApi.ProvidersProxyCreate(ctx).ProxyProviderRequest(*proxyProviderRequest).Execute()
	if err != nil {
		return fmt.Errorf("error when calling `ProvidersAPI.ProvidersProxyCreate`: %w with response %v", err, r)
	}

	err = c.reconcileOutpost(ctx, pp.Spec.Outpost, resp.Pk, ReconcileOutpostModeAdd)
	if err != nil {
		return fmt.Errorf("error when calling `reconcileOutpost`: %w", err)
	}

	pp.Status.PK = strconv.Itoa(int(resp.Pk))
	return c.updateProxyProviderStatus(ctx, pp)
}

func (c *ProxyProviderController) updateProxyProviderStatus(ctx context.Context, pp *v1alpha1.ProxyProvider) error {
	ppCopy := pp.DeepCopy()
	_, err := c.proxyProviderClientset.ProxyproviderV1alpha1().ProxyProviders(ppCopy.Namespace).UpdateStatus(ctx, ppCopy, metav1.UpdateOptions{FieldManager: FieldManager})
	return err
}

// Update metadata, spec, etc. of the ProxyProvider object.
func (c *ProxyProviderController) updateProxyProvider(ctx context.Context, pp *v1alpha1.ProxyProvider) error {
	ppCopy := pp.DeepCopy()
	_, err := c.proxyProviderClientset.ProxyproviderV1alpha1().ProxyProviders(ppCopy.Namespace).Update(ctx, ppCopy, metav1.UpdateOptions{FieldManager: FieldManager})
	if err != nil {
		return fmt.Errorf("error updating ProxyProvider metadata: %v", err)
	}
	return nil
}

type ReconcileOutpostMode string

const (
	ReconcileOutpostModeAdd    ReconcileOutpostMode = "add"
	ReconcileOutpostModeRemove ReconcileOutpostMode = "remove"
)

func (c *ProxyProviderController) reconcileOutpost(ctx context.Context, outpostId string, providerPk int32, mode ReconcileOutpostMode) error {
	logger := klog.LoggerWithValues(klog.FromContext(ctx), "outpostId", outpostId, "providerPk", providerPk, "mode", mode)

	outpost, r, err := c.authentik.OutpostsApi.OutpostsInstancesRetrieve(ctx, outpostId).Execute()
	if err != nil {
		return fmt.Errorf("error when calling `OutpostsAPI.OutpostsInstancesRetrieve`: %w with response %v", err, r)
	}
	updated := false

	switch mode {
	case ReconcileOutpostModeAdd:
		if !slices.Contains(outpost.Providers, providerPk) {
			outpost.Providers = append(outpost.Providers, providerPk)
			updated = true
		} else {
			logger.V(4).Info("Provider already in outpost")
		}
	case ReconcileOutpostModeRemove:
		if slices.Contains(outpost.Providers, providerPk) {
			outpost.Providers = slices.Delete(outpost.Providers, slices.Index(outpost.Providers, providerPk), 1)
			updated = true
		}
	default:
		return fmt.Errorf("invalid mode: %s", mode)
	}

	if !updated {
		return nil
	}

	outpostPartialUpdateRequest := &authentikapi.PatchedOutpostRequest{
		Providers: outpost.Providers,
	}
	_, r, err = c.authentik.OutpostsApi.OutpostsInstancesPartialUpdate(ctx, outpostId).PatchedOutpostRequest(*outpostPartialUpdateRequest).Execute()
	if err != nil {
		return fmt.Errorf("error when calling `OutpostsAPI.OutpostsInstancesPartialUpdate`: %w with response %v", err, r)
	}

	return nil
}