claude-botandClaude Opus 5.5
89771995b3
ci/woodpecker/pr/ci-scripts-test Pipeline was successful
A drifted proxy provider was PATCHed with only the drifted fields. ProxyProviderSerializer.validate (providers/proxy/api.py, 2026.8.3) checks internal_host against mode from the request attrs alone, defaulting mode to proxy, so a flow-only PATCH fails with 400 "Internal host cannot be empty when forward auth is disabled." Seen live after #1285: 8 gates could not move to the new flows and stayed on their old ones. The fake Authentik in the tests now applies the same validation, and a test covers a flow change on an existing gate. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
653 lines
29 KiB
Python
653 lines
29 KiB
Python
#!/usr/bin/env python3
|
|
"""authentik-gate-sync: an ingress class that puts an Authentik proxy gate in front of an app.
|
|
|
|
Source of truth: every Ingress with spec.ingressClassName "authentik-gate". Traefik
|
|
ignores that class (its IngressClass names another controller), so a source
|
|
Ingress is never served directly. For each one this controller builds:
|
|
|
|
Authentik proxy provider + application + outpost (+ group-membership
|
|
expression policy bound to the application).
|
|
Authentik's own Kubernetes integration then runs the outpost as
|
|
ak-outpost-<gate> in the authentik namespace.
|
|
app ns ExternalName Service <ingress>-gate -> the outpost Service,
|
|
Traefik Ingress <ingress>-gate: the source's hosts, paths, TLS,
|
|
labels and annotations, backend swapped for the gate,
|
|
CNP <ingress>-gate: the backend pods admit this outpost.
|
|
authentik CNP gate-<gate>: the outpost may reach only those backend pods.
|
|
|
|
Like any ingress controller it publishes the source's status.loadBalancer: the
|
|
address Traefik reports for the route, or none while the gate has no route.
|
|
|
|
The source Ingress's backend is the real upstream; it becomes the provider's
|
|
internal host and its host the external host. Annotations (all optional):
|
|
|
|
auth.renner.uno/group group whose members may pass; none = any user
|
|
auth.renner.uno/login-flow authentication flow slug; none = the brand's
|
|
auth.renner.uno/authorization-flow authorization flow slug (default below)
|
|
auth.renner.uno/skip-path provider skip_path_regex
|
|
|
|
Fail closed: the Traefik Ingress is applied last and only after the Authentik
|
|
side reconciled, and is removed when that fails (e.g. the named group is gone),
|
|
so there is never a route to an unguarded backend.
|
|
|
|
Ownership: Kubernetes objects carry app.kubernetes.io/managed-by; Authentik
|
|
applications carry meta_publisher. Anything without the marker is never touched.
|
|
"""
|
|
|
|
import hashlib
|
|
import json
|
|
import logging
|
|
import os
|
|
import sys
|
|
import time
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
|
|
LOG = logging.getLogger("authentik-gate-sync")
|
|
|
|
AK_URL = os.environ.get("AK_URL", "http://authentik-server.authentik.svc.cluster.local").rstrip("/")
|
|
AK_TOKEN_FILE = os.environ.get("AK_TOKEN_FILE", "/var/run/secrets/authentik-gate-sync/token")
|
|
INGRESS_CLASS = os.environ.get("INGRESS_CLASS", "authentik-gate")
|
|
TARGET_CLASS = os.environ.get("TARGET_INGRESS_CLASS", "traefik")
|
|
OUTPOST_NS = os.environ.get("OUTPOST_NAMESPACE", "authentik")
|
|
SERVICE_CONNECTION = os.environ.get("SERVICE_CONNECTION", "Local Kubernetes Cluster")
|
|
DEFAULT_AUTHZ_FLOW = os.environ.get("DEFAULT_AUTHORIZATION_FLOW", "default-provider-authorization-implicit-consent")
|
|
INVALIDATION_FLOW = os.environ.get("INVALIDATION_FLOW", "default-provider-invalidation-flow")
|
|
COOKIE_DOMAIN = os.environ.get("COOKIE_DOMAIN", "renner.uno")
|
|
AK_BROWSER_URL = os.environ.get("AUTHENTIK_HOST_BROWSER", "https://auth.renner.uno/")
|
|
INTERVAL = int(os.environ.get("INTERVAL_SECONDS", "60"))
|
|
HTTP_TIMEOUT = 20
|
|
|
|
ANN = "auth.renner.uno/"
|
|
MANAGER = "authentik-gate-sync"
|
|
MANAGED_BY = {"app.kubernetes.io/managed-by": MANAGER}
|
|
GATE_LABEL = "auth.renner.uno/gate"
|
|
OUTPOST_PORT = 9000
|
|
|
|
# Never copied from the source onto the generated Ingress: cert-manager would
|
|
# issue a second Certificate for the same secret, Flux would treat it as its own,
|
|
# and the gate annotations mean nothing to Traefik.
|
|
DROP_LABEL_PREFIXES = ("kustomize.toolkit.fluxcd.io/", "app.kubernetes.io/managed-by")
|
|
DROP_ANNOTATION_PREFIXES = (ANN, "cert-manager.io/", "acme.cert-manager.io/",
|
|
"kubectl.kubernetes.io/last-applied-configuration")
|
|
|
|
K8S_API = "https://kubernetes.default.svc"
|
|
K8S_TOKEN_FILE = "/var/run/secrets/kubernetes.io/serviceaccount/token"
|
|
K8S_CA_FILE = "/var/run/secrets/kubernetes.io/serviceaccount/ca.crt"
|
|
|
|
|
|
class ApiError(Exception):
|
|
def __init__(self, msg, code=None):
|
|
super().__init__(msg)
|
|
self.code = code
|
|
|
|
|
|
class GateError(Exception):
|
|
"""A source Ingress that cannot be turned into a gate; it gets no route."""
|
|
|
|
|
|
def http(method, url, body=None, headers=None, cafile=None, content_type="application/json"):
|
|
data = None if body is None else json.dumps(body).encode()
|
|
req = urllib.request.Request(url, data=data, method=method)
|
|
req.add_header("Accept", "application/json")
|
|
if data is not None:
|
|
req.add_header("Content-Type", content_type)
|
|
for k, v in (headers or {}).items():
|
|
req.add_header(k, v)
|
|
ctx = None
|
|
if cafile:
|
|
import ssl
|
|
ctx = ssl.create_default_context(cafile=cafile)
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=HTTP_TIMEOUT, context=ctx) as resp:
|
|
raw = resp.read()
|
|
return json.loads(raw) if raw else None
|
|
except urllib.error.HTTPError as e:
|
|
detail = e.read().decode(errors="replace")[:500]
|
|
raise ApiError(f"{method} {url} -> {e.code}: {detail}", e.code) from None
|
|
except (OSError, ValueError) as e: # URLError, timeouts, resets, bad JSON
|
|
raise ApiError(f"{method} {url} -> {e}") from None
|
|
|
|
|
|
# --- desired state (pure) ---------------------------------------------------
|
|
|
|
def gate_name(ing):
|
|
"""<ns>-<ingress>, truncated, plus a hash of both: unambiguous (a-b/c vs a/b-c)
|
|
and short enough for label values and ak-outpost-<name> (63)."""
|
|
ns, name = ing["metadata"]["namespace"], ing["metadata"]["name"]
|
|
digest = hashlib.sha256(f"{ns}/{name}".encode()).hexdigest()[:6]
|
|
return f"{ns}-{name}"[:45].rstrip("-") + "-" + digest
|
|
|
|
|
|
def group_expression(group):
|
|
# repr() makes the annotation value a Python string literal, never code.
|
|
# A group that does not exist matches nothing, so deleting it denies access.
|
|
return f"return ak_is_group_member(request.user, name={group!r})"
|
|
|
|
|
|
def gate_spec(ing):
|
|
"""Translate a source Ingress into the gate it asks for. Raises GateError."""
|
|
meta = ing["metadata"]
|
|
ann = meta.get("annotations") or {}
|
|
rules = ing.get("spec", {}).get("rules") or []
|
|
hosts = sorted({r.get("host") for r in rules if r.get("host")})
|
|
if len(hosts) != 1 or len(hosts) != len(rules):
|
|
raise GateError(f"needs exactly one rule with one host, has {[r.get('host') for r in rules]}")
|
|
if hosts[0].startswith("*"):
|
|
raise GateError(f"wildcard host {hosts[0]} cannot be a gate")
|
|
backends = set()
|
|
for p in rules[0].get("http", {}).get("paths") or []:
|
|
svc = (p.get("backend") or {}).get("service")
|
|
if not svc:
|
|
raise GateError("every path needs a service backend")
|
|
port = svc.get("port") or {}
|
|
backends.add((svc["name"], port.get("number") or port.get("name")))
|
|
if ing.get("spec", {}).get("defaultBackend"):
|
|
raise GateError("defaultBackend is not supported")
|
|
if len(backends) != 1:
|
|
raise GateError(f"all paths must share one backend, has {sorted(map(str, backends))}")
|
|
(service, port), = backends
|
|
tls_hosts = {h for t in ing.get("spec", {}).get("tls") or [] for h in t.get("hosts") or []}
|
|
return {
|
|
"name": gate_name(ing),
|
|
"namespace": meta["namespace"],
|
|
"ingress": meta["name"],
|
|
"uid": meta["uid"],
|
|
"host": hosts[0],
|
|
"scheme": "https" if hosts[0] in tls_hosts else "http",
|
|
"service": service,
|
|
"port": port,
|
|
"group": ann.get(ANN + "group") or None,
|
|
"login_flow": ann.get(ANN + "login-flow") or None,
|
|
"authorization_flow": ann.get(ANN + "authorization-flow") or DEFAULT_AUTHZ_FLOW,
|
|
"skip_path": ann.get(ANN + "skip-path", ""),
|
|
}
|
|
|
|
|
|
def owner_ref(gate):
|
|
return [{"apiVersion": "networking.k8s.io/v1", "kind": "Ingress", "name": gate["ingress"],
|
|
"uid": gate["uid"], "controller": True, "blockOwnerDeletion": False}]
|
|
|
|
|
|
def gate_labels(gate):
|
|
return {**MANAGED_BY, GATE_LABEL: gate["name"]}
|
|
|
|
|
|
def outpost_service(gate):
|
|
return f"ak-outpost-{gate['name']}"
|
|
|
|
|
|
def external_name_service(gate):
|
|
return {"apiVersion": "v1", "kind": "Service",
|
|
"metadata": {"name": f"{gate['ingress']}-gate", "namespace": gate["namespace"],
|
|
"labels": gate_labels(gate), "ownerReferences": owner_ref(gate)},
|
|
"spec": {"type": "ExternalName",
|
|
"externalName": f"{outpost_service(gate)}.{OUTPOST_NS}.svc.cluster.local",
|
|
"ports": [{"name": "http", "port": OUTPOST_PORT, "targetPort": OUTPOST_PORT}]}}
|
|
|
|
|
|
def route_ingress(ing, gate):
|
|
"""The Traefik Ingress: the source with its backend swapped for the gate."""
|
|
meta = ing["metadata"]
|
|
labels = {k: v for k, v in (meta.get("labels") or {}).items() if not k.startswith(DROP_LABEL_PREFIXES)}
|
|
annotations = {k: v for k, v in (meta.get("annotations") or {}).items()
|
|
if not k.startswith(DROP_ANNOTATION_PREFIXES)}
|
|
backend = {"service": {"name": f"{gate['ingress']}-gate", "port": {"number": OUTPOST_PORT}}}
|
|
rules = []
|
|
for r in ing["spec"]["rules"]:
|
|
paths = [{"path": p.get("path", "/"), "pathType": p.get("pathType", "Prefix"), "backend": backend}
|
|
for p in r["http"]["paths"]]
|
|
rules.append({"host": r["host"], "http": {"paths": paths}})
|
|
spec = {"ingressClassName": TARGET_CLASS, "rules": rules}
|
|
if ing["spec"].get("tls"):
|
|
spec["tls"] = ing["spec"]["tls"]
|
|
return {"apiVersion": "networking.k8s.io/v1", "kind": "Ingress",
|
|
"metadata": {"name": f"{gate['ingress']}-gate", "namespace": gate["namespace"],
|
|
"labels": {**labels, **gate_labels(gate)}, "annotations": annotations,
|
|
"ownerReferences": owner_ref(gate)},
|
|
"spec": spec}
|
|
|
|
|
|
def outpost_selector(gate):
|
|
return {"k8s:io.kubernetes.pod.namespace": OUTPOST_NS, "goauthentik.io/outpost-name": gate["name"]}
|
|
|
|
|
|
def backend_ingress_cnp(gate, pod_selector, target_port):
|
|
return {"apiVersion": "cilium.io/v2", "kind": "CiliumNetworkPolicy",
|
|
"metadata": {"name": f"{gate['ingress']}-gate", "namespace": gate["namespace"],
|
|
"labels": gate_labels(gate), "ownerReferences": owner_ref(gate)},
|
|
"spec": {"endpointSelector": {"matchLabels": pod_selector},
|
|
"ingress": [{"fromEndpoints": [{"matchLabels": outpost_selector(gate)}],
|
|
"toPorts": [{"ports": [{"port": str(target_port), "protocol": "TCP"}]}]}]}}
|
|
|
|
|
|
def outpost_egress_cnp(gate, pod_selector, target_port):
|
|
peer = {"k8s:io.kubernetes.pod.namespace": gate["namespace"], **pod_selector}
|
|
return {"apiVersion": "cilium.io/v2", "kind": "CiliumNetworkPolicy",
|
|
"metadata": {"name": f"gate-{gate['name']}", "namespace": OUTPOST_NS, "labels": gate_labels(gate)},
|
|
"spec": {"endpointSelector": {"matchLabels": {"goauthentik.io/outpost-name": gate["name"]}},
|
|
"egress": [{"toEndpoints": [{"matchLabels": peer}],
|
|
"toPorts": [{"ports": [{"port": str(target_port), "protocol": "TCP"}]}]}]}}
|
|
|
|
|
|
def resolve_backend(gate, service):
|
|
"""(service port number, pod selector, target port) for the gate's backend.
|
|
A named targetPort stays a name: Cilium resolves named ports against the
|
|
selected pods' container ports, so the policy does not depend on the pods
|
|
being up."""
|
|
selector = service.get("spec", {}).get("selector")
|
|
if not selector:
|
|
raise GateError(f"Service {gate['service']} has no selector; cannot scope its network policy")
|
|
sp = None
|
|
for p in service["spec"].get("ports") or []:
|
|
if p.get("port") == gate["port"] or (p.get("name") and p.get("name") == gate["port"]):
|
|
sp = p
|
|
break
|
|
if sp is None:
|
|
raise GateError(f"Service {gate['service']} has no port {gate['port']}")
|
|
target = sp.get("targetPort", sp["port"])
|
|
if isinstance(target, int):
|
|
return sp["port"], selector, target
|
|
if isinstance(target, str) and target.isdigit():
|
|
return sp["port"], selector, int(target)
|
|
return sp["port"], selector, target
|
|
|
|
|
|
def outpost_config():
|
|
return {
|
|
"log_level": "info",
|
|
"authentik_host": AK_URL,
|
|
"authentik_host_insecure": False,
|
|
"authentik_host_browser": AK_BROWSER_URL,
|
|
"refresh_interval": "minutes=5",
|
|
"container_image": None,
|
|
"object_naming_template": "ak-outpost-%(name)s",
|
|
"kubernetes_namespace": OUTPOST_NS,
|
|
"kubernetes_replicas": 1,
|
|
"kubernetes_service_type": "ClusterIP",
|
|
"kubernetes_disabled_components": ["ingress"],
|
|
"kubernetes_image_pull_secrets": [],
|
|
"kubernetes_ingress_annotations": {},
|
|
"kubernetes_ingress_class_name": None,
|
|
"kubernetes_ingress_secret_name": "authentik-outpost-tls",
|
|
"kubernetes_json_patches": None,
|
|
}
|
|
|
|
|
|
def provider_body(gate, svc_port, flows):
|
|
return {
|
|
"name": gate["name"],
|
|
"mode": "proxy",
|
|
"external_host": f"{gate['scheme']}://{gate['host']}",
|
|
"internal_host": f"http://{gate['service']}.{gate['namespace']}.svc.cluster.local:{svc_port}",
|
|
"internal_host_ssl_validation": False,
|
|
"cookie_domain": COOKIE_DOMAIN,
|
|
"access_token_validity": "hours=24",
|
|
"refresh_token_validity": "days=30",
|
|
"intercept_header_auth": True,
|
|
"skip_path_regex": gate["skip_path"],
|
|
"authorization_flow": flows[gate["authorization_flow"]],
|
|
"authentication_flow": flows[gate["login_flow"]] if gate["login_flow"] else None,
|
|
"invalidation_flow": flows[INVALIDATION_FLOW],
|
|
}
|
|
|
|
|
|
def lb_ingress(obj):
|
|
return ((obj or {}).get("status") or {}).get("loadBalancer", {}).get("ingress") or []
|
|
|
|
|
|
def status_patch(source, addresses):
|
|
"""Merge patch publishing `addresses` as the source Ingress's load-balancer
|
|
status, as an ingress controller does; None when it already matches."""
|
|
if lb_ingress(source) == addresses:
|
|
return None
|
|
return {"status": {"loadBalancer": {"ingress": addresses or None}}}
|
|
|
|
|
|
def drift(current, desired):
|
|
return {k: v for k, v in desired.items() if current.get(k) != v}
|
|
|
|
|
|
# --- Kubernetes -------------------------------------------------------------
|
|
|
|
class K8s:
|
|
def _h(self):
|
|
with open(K8S_TOKEN_FILE) as f:
|
|
return {"Authorization": f"Bearer {f.read().strip()}"}
|
|
|
|
def get(self, path):
|
|
return http("GET", K8S_API + path, headers=self._h(), cafile=K8S_CA_FILE)
|
|
|
|
def apply(self, obj):
|
|
"""Server-side apply; idempotent, so every pass re-asserts every object.
|
|
Refuses to take over a same-named object this controller does not own."""
|
|
try:
|
|
live = self.get(object_path(obj))
|
|
except ApiError as e:
|
|
if e.code != 404:
|
|
raise
|
|
else:
|
|
if (live["metadata"].get("labels") or {}).get("app.kubernetes.io/managed-by") != MANAGER:
|
|
raise GateError(f'{obj["kind"]} {obj["metadata"]["namespace"]}/{obj["metadata"]["name"]} '
|
|
f"exists and is not managed by {MANAGER}")
|
|
path = object_path(obj) + f"?fieldManager={MANAGER}&force=true"
|
|
return http("PATCH", K8S_API + path, obj, headers=self._h(), cafile=K8S_CA_FILE,
|
|
content_type="application/apply-patch+yaml")
|
|
|
|
def delete(self, obj):
|
|
try:
|
|
http("DELETE", K8S_API + object_path(obj), headers=self._h(), cafile=K8S_CA_FILE)
|
|
except ApiError as e:
|
|
if e.code != 404:
|
|
raise
|
|
|
|
def patch_status(self, ns, name, patch):
|
|
path = f"/apis/networking.k8s.io/v1/namespaces/{ns}/ingresses/{name}/status"
|
|
return http("PATCH", K8S_API + path, patch, headers=self._h(), cafile=K8S_CA_FILE,
|
|
content_type="application/merge-patch+json")
|
|
|
|
def delete_managed(self, obj):
|
|
try:
|
|
live = self.get(object_path(obj))
|
|
except ApiError as e:
|
|
if e.code == 404:
|
|
return
|
|
raise
|
|
if (live["metadata"].get("labels") or {}).get("app.kubernetes.io/managed-by") == MANAGER:
|
|
self.delete(obj)
|
|
LOG.info("deleted %s %s/%s", obj["kind"], obj["metadata"]["namespace"], obj["metadata"]["name"])
|
|
|
|
|
|
COLLECTIONS = {
|
|
"Ingress": "/apis/networking.k8s.io/v1/namespaces/{ns}/ingresses",
|
|
"Service": "/api/v1/namespaces/{ns}/services",
|
|
"CiliumNetworkPolicy": "/apis/cilium.io/v2/namespaces/{ns}/ciliumnetworkpolicies",
|
|
}
|
|
CLUSTER_COLLECTIONS = {
|
|
"Ingress": "/apis/networking.k8s.io/v1/ingresses",
|
|
"Service": "/api/v1/services",
|
|
"CiliumNetworkPolicy": "/apis/cilium.io/v2/ciliumnetworkpolicies",
|
|
}
|
|
|
|
|
|
def object_path(obj):
|
|
m = obj["metadata"]
|
|
return COLLECTIONS[obj["kind"]].format(ns=m["namespace"]) + "/" + m["name"]
|
|
|
|
|
|
def managed_objects(k8s):
|
|
sel = urllib.parse.quote(f"app.kubernetes.io/managed-by={MANAGER}")
|
|
out = []
|
|
for kind, path in CLUSTER_COLLECTIONS.items():
|
|
for item in k8s.get(f"{path}?labelSelector={sel}").get("items", []):
|
|
item["kind"] = kind
|
|
out.append(item)
|
|
return out
|
|
|
|
|
|
# --- Authentik --------------------------------------------------------------
|
|
|
|
class Authentik:
|
|
def __init__(self, token):
|
|
self.h = {"Authorization": f"Bearer {token}"}
|
|
|
|
def req(self, method, path, body=None):
|
|
return http(method, f"{AK_URL}/api/v3/{path}", body, headers=self.h)
|
|
|
|
def list(self, path, **query):
|
|
"""All pages of a list endpoint."""
|
|
query = {**query, "page_size": 100}
|
|
out, page = [], 1
|
|
while True:
|
|
res = self.req("GET", f"{path}?{urllib.parse.urlencode({**query, 'page': page})}")
|
|
out.extend(res["results"])
|
|
if not res["pagination"].get("next"):
|
|
return out
|
|
page += 1
|
|
|
|
def one(self, path, **query):
|
|
found = self.list(path, **query)
|
|
return found[0] if found else None
|
|
|
|
def get(self, path):
|
|
try:
|
|
return self.req("GET", path)
|
|
except ApiError as e:
|
|
if e.code == 404:
|
|
return None
|
|
raise
|
|
|
|
|
|
def get_application(ak, slug):
|
|
# Retrieve, never list: the list endpoint only shows applications the token's
|
|
# user passes policies for and caches that per user, ignoring the filters.
|
|
return ak.get(f"core/applications/{slug}/")
|
|
|
|
|
|
def ensure_gate(ak, gate, svc_port, flows, connection):
|
|
name = gate["name"]
|
|
if gate["group"] and ak.one("core/groups/", name=gate["group"]) is None:
|
|
raise GateError(f"group {gate['group']!r} not found in Authentik")
|
|
|
|
app = get_application(ak, name)
|
|
if app is not None and app.get("meta_publisher") != MANAGER:
|
|
raise GateError(f"application {name} exists and is not managed by {MANAGER}")
|
|
|
|
want = provider_body(gate, svc_port, flows)
|
|
provider = ak.one("providers/proxy/", name__iexact=name)
|
|
if provider is None:
|
|
provider = ak.req("POST", "providers/proxy/", want)
|
|
LOG.info("%s: created provider %s", name, provider["pk"])
|
|
elif provider.get("assigned_application_slug") not in (None, name) or \
|
|
(app is not None and app.get("provider") != provider["pk"]):
|
|
raise GateError(f"proxy provider {name} belongs to another application")
|
|
elif drift(provider, want):
|
|
# Also adopts a provider whose application an interrupted teardown removed.
|
|
# The full body: the serializer validates internal_host against mode
|
|
# from the request alone, so a partial PATCH without them is a 400.
|
|
ak.req("PATCH", f"providers/proxy/{provider['pk']}/", want)
|
|
LOG.info("%s: updated provider %s", name, sorted(drift(provider, want)))
|
|
|
|
want_app = {"name": f"{gate['host']} (gate)", "slug": name, "provider": provider["pk"],
|
|
"meta_launch_url": want["external_host"], "meta_publisher": MANAGER,
|
|
"meta_description": f"Gate for Ingress {gate['namespace']}/{gate['ingress']}"}
|
|
if app is None:
|
|
app = ak.req("POST", "core/applications/", want_app)
|
|
LOG.info("%s: created application", name)
|
|
elif drift(app, want_app):
|
|
ak.req("PATCH", f"core/applications/{name}/", drift(app, want_app))
|
|
LOG.info("%s: updated application %s", name, sorted(drift(app, want_app)))
|
|
|
|
ensure_access(ak, gate, app["pbm_uuid"])
|
|
|
|
want_outpost = {"name": name, "type": "proxy", "service_connection": connection,
|
|
"providers": [provider["pk"]], "config": outpost_config()}
|
|
outpost = ak.one("outposts/instances/", name__iexact=name)
|
|
if outpost is None:
|
|
ak.req("POST", "outposts/instances/", want_outpost)
|
|
LOG.info("%s: created outpost", name)
|
|
elif set(outpost.get("providers") or []) - {provider["pk"]}:
|
|
raise GateError(f"outpost {name} serves providers this controller does not own")
|
|
elif drift(outpost, want_outpost):
|
|
ak.req("PATCH", f"outposts/instances/{outpost['pk']}/", drift(outpost, want_outpost))
|
|
LOG.info("%s: updated outpost %s", name, sorted(drift(outpost, want_outpost)))
|
|
|
|
|
|
def ensure_access(ak, gate, target):
|
|
"""Group gates bind an expression policy, not the group itself: a deleted
|
|
group cascades its bindings away (leaving the app open to everyone), while
|
|
the policy keeps evaluating False. The new binding exists before any stale
|
|
one is removed, so a change never passes through "no binding"."""
|
|
name = gate["name"]
|
|
policy = None
|
|
if gate["group"]:
|
|
want = {"name": name, "expression": group_expression(gate["group"]), "execution_logging": False}
|
|
policy = ak.one("policies/expression/", name=name)
|
|
if policy is None:
|
|
policy = ak.req("POST", "policies/expression/", want)
|
|
LOG.info("%s: created policy for group %s", name, gate["group"])
|
|
elif drift(policy, want):
|
|
ak.req("PATCH", f"policies/expression/{policy['pk']}/", drift(policy, want))
|
|
LOG.info("%s: policy now requires group %s", name, gate["group"])
|
|
|
|
bindings = ak.list("policies/bindings/", target=target)
|
|
keep = next((b for b in bindings if policy and b.get("policy") == policy["pk"]), None)
|
|
want_binding = {"enabled": True, "negate": False, "failure_result": False}
|
|
if policy and keep is None:
|
|
ak.req("POST", "policies/bindings/", {"target": target, "policy": policy["pk"], "order": 0,
|
|
"timeout": 30, **want_binding})
|
|
LOG.info("%s: bound policy", name)
|
|
elif keep is not None and drift(keep, want_binding):
|
|
ak.req("PATCH", f"policies/bindings/{keep['pk']}/", want_binding)
|
|
for b in bindings:
|
|
if b is not keep:
|
|
ak.req("DELETE", f"policies/bindings/{b['pk']}/")
|
|
LOG.info("%s: removed binding %s", name, b["pk"])
|
|
if not gate["group"]:
|
|
stale = ak.one("policies/expression/", name=name)
|
|
if stale is not None:
|
|
ak.req("DELETE", f"policies/expression/{stale['pk']}/")
|
|
|
|
|
|
def remove_gate(ak, app):
|
|
"""Delete an owned gate. The outpost serving only its provider goes first
|
|
(Authentik tears its pods down), then the application (bindings cascade; its
|
|
provider is PROTECTed while referenced), the provider, the policy. Ownership
|
|
is found through the application, so until it is gone a failed step is
|
|
retried on the next pass."""
|
|
name, provider = app["slug"], app.get("provider")
|
|
for o in ak.list("outposts/instances/", name__iexact=name):
|
|
if o.get("providers") == [provider]:
|
|
ak.req("DELETE", f"outposts/instances/{o['pk']}/")
|
|
ak.req("DELETE", f"core/applications/{name}/")
|
|
if provider:
|
|
ak.req("DELETE", f"providers/proxy/{provider}/")
|
|
policy = ak.one("policies/expression/", name=name)
|
|
if policy is not None:
|
|
ak.req("DELETE", f"policies/expression/{policy['pk']}/")
|
|
LOG.info("%s: removed gate from Authentik", name)
|
|
|
|
|
|
def owned_applications(ak):
|
|
"""Applications this controller owns, found through the (unfiltered) proxy
|
|
provider list and confirmed one by one."""
|
|
for p in ak.list("providers/proxy/"):
|
|
slug = p.get("assigned_application_slug")
|
|
if not slug:
|
|
continue
|
|
app = get_application(ak, slug)
|
|
if app and app.get("meta_publisher") == MANAGER and app.get("provider") == p["pk"]:
|
|
yield app
|
|
|
|
|
|
# --- reconcile --------------------------------------------------------------
|
|
|
|
def publish_status(k8s, source, addresses):
|
|
"""Status is cosmetic for the gate itself: a failure is logged, never fatal."""
|
|
patch = status_patch(source, addresses)
|
|
if patch is None:
|
|
return
|
|
m = source["metadata"]
|
|
try:
|
|
k8s.patch_status(m["namespace"], m["name"], patch)
|
|
LOG.info("%s/%s: status %s", m["namespace"], m["name"],
|
|
[a.get("ip") or a.get("hostname") for a in addresses])
|
|
except Exception as e:
|
|
LOG.warning("%s/%s: publishing status: %s", m["namespace"], m["name"], e)
|
|
|
|
|
|
def reconcile():
|
|
k8s = K8s()
|
|
with open(AK_TOKEN_FILE) as f:
|
|
ak = Authentik(f.read().strip())
|
|
|
|
sources = [i for i in k8s.get("/apis/networking.k8s.io/v1/ingresses").get("items", [])
|
|
if i.get("spec", {}).get("ingressClassName") == INGRESS_CLASS]
|
|
flows = {f["slug"]: f["pk"] for f in ak.list("flows/instances/")}
|
|
connection = ak.one("outposts/service_connections/kubernetes/", name=SERVICE_CONNECTION)
|
|
if connection is None:
|
|
raise ApiError(f"Kubernetes service connection {SERVICE_CONNECTION!r} not found")
|
|
connection = connection["pk"]
|
|
|
|
desired = set() # (kind, ns, name) of every Kubernetes object some gate wants
|
|
gates = set() # gate names whose Authentik side should exist
|
|
held = set() # gates hit by an API error: their live objects stay as they are
|
|
failed = 0
|
|
for ing in sorted(sources, key=gate_name):
|
|
ref = f'{ing["metadata"]["namespace"]}/{ing["metadata"]["name"]}'
|
|
try:
|
|
gate = gate_spec(ing)
|
|
for slug in filter(None, (gate["authorization_flow"], gate["login_flow"], INVALIDATION_FLOW)):
|
|
if slug not in flows:
|
|
raise GateError(f"flow {slug!r} not found in Authentik")
|
|
ns = gate["namespace"]
|
|
try:
|
|
service = k8s.get(f"/api/v1/namespaces/{ns}/services/{gate['service']}")
|
|
except ApiError as e:
|
|
if e.code == 404:
|
|
raise GateError(f"backend Service {gate['service']} does not exist") from None
|
|
raise
|
|
svc_port, pods, target_port = resolve_backend(gate, service)
|
|
gates.add(gate["name"])
|
|
ensure_gate(ak, gate, svc_port, flows, connection)
|
|
objs = [outpost_egress_cnp(gate, pods, target_port), backend_ingress_cnp(gate, pods, target_port),
|
|
external_name_service(gate), route_ingress(ing, gate)] # the route goes last
|
|
for obj in objs:
|
|
live = k8s.apply(obj)
|
|
desired.add((obj["kind"], obj["metadata"]["namespace"], obj["metadata"]["name"]))
|
|
publish_status(k8s, ing, lb_ingress(live)) # the route's address, from Traefik
|
|
except GateError as e:
|
|
# The Ingress cannot be gated as written: its route goes now, the
|
|
# rest with the sweep below.
|
|
LOG.error("%s: %s", ref, e)
|
|
failed += 1
|
|
gates.discard(gate_name(ing))
|
|
route = {"kind": "Ingress", "metadata": {"namespace": ing["metadata"]["namespace"],
|
|
"name": f'{ing["metadata"]["name"]}-gate'}}
|
|
try:
|
|
k8s.delete_managed(route)
|
|
except Exception as e: # the sweep retries
|
|
LOG.error("%s: removing route: %s", ref, e)
|
|
publish_status(k8s, ing, [])
|
|
except Exception as e:
|
|
# Transient: leave this gate exactly as the last good pass left it.
|
|
LOG.error("%s: %s", ref, e)
|
|
gates.add(gate_name(ing))
|
|
held.add(gate_name(ing))
|
|
failed += 1
|
|
|
|
removed = 0
|
|
for obj in managed_objects(k8s):
|
|
key = (obj["kind"], obj["metadata"]["namespace"], obj["metadata"]["name"])
|
|
if key not in desired and obj["metadata"]["labels"].get(GATE_LABEL) not in held:
|
|
obj = {"kind": obj["kind"], "metadata": {"namespace": key[1], "name": key[2]}}
|
|
k8s.delete(obj)
|
|
LOG.info("deleted %s %s/%s", *key)
|
|
removed += 1
|
|
|
|
for app in list(owned_applications(ak)):
|
|
if app["slug"] not in gates:
|
|
remove_gate(ak, app)
|
|
removed += 1
|
|
|
|
LOG.info("reconciled: %d gates, %d failed, %d objects removed", len(sources), failed, removed)
|
|
|
|
|
|
def main():
|
|
logging.basicConfig(level=os.environ.get("LOG_LEVEL", "INFO"), stream=sys.stdout,
|
|
format="%(asctime)s %(levelname)s %(message)s")
|
|
if os.environ.get("ONCE") == "1":
|
|
reconcile()
|
|
return
|
|
while True:
|
|
try:
|
|
reconcile()
|
|
except Exception as e: # keep running; the next pass retries
|
|
LOG.error("reconcile failed: %s", e)
|
|
time.sleep(INTERVAL)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|