310 lines
12 KiB
Go
310 lines
12 KiB
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))
|
|
|
|
err = c.reconcileOutpost(ctx, pp.Spec.Outpost, int32(pk), ReconcileOutpostModeAdd)
|
|
if err != nil {
|
|
return fmt.Errorf("error when calling `reconcileOutpost`: %w", err)
|
|
}
|
|
|
|
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
|
|
}
|