authentik.outposts.controllers.k8s.utils

 1"""k8s utils"""
 2
 3from pathlib import Path
 4
 5from kubernetes.client.models.v1_container_port import V1ContainerPort
 6from kubernetes.client.models.v1_service_port import V1ServicePort
 7from kubernetes.config.incluster_config import SERVICE_TOKEN_FILENAME
 8
 9from authentik.outposts.controllers.k8s.triggers import NeedsRecreate
10
11
12def get_namespace() -> str:
13    """Get the namespace if we're running in a pod, otherwise default to default"""
14    path = Path(SERVICE_TOKEN_FILENAME.replace("token", "namespace"))
15    if path.exists():
16        with open(path, encoding="utf8") as _namespace_file:
17            return _namespace_file.read()
18    return "default"
19
20
21def compare_port(
22    current: V1ServicePort | V1ContainerPort, reference: V1ServicePort | V1ContainerPort
23) -> bool:
24    """Compare a single port"""
25    if current.name != reference.name:
26        return False
27    if current.protocol != reference.protocol:
28        return False
29    if isinstance(current, V1ServicePort) and isinstance(reference, V1ServicePort):
30        # We only care about the target port
31        if current.target_port != reference.target_port:
32            return False
33    if isinstance(current, V1ContainerPort) and isinstance(reference, V1ContainerPort):
34        # We only care about the target port
35        if current.container_port != reference.container_port:
36            return False
37    return True
38
39
40def compare_ports(
41    current: list[V1ServicePort | V1ContainerPort] | None,
42    reference: list[V1ServicePort | V1ContainerPort] | None,
43):
44    """Compare ports of a list"""
45    if not current or not reference:
46        raise NeedsRecreate()
47    if len(current) != len(reference):
48        raise NeedsRecreate()
49    for port in reference:
50        if not any(compare_port(port, current_port) for current_port in current):
51            raise NeedsRecreate()