#!/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()