osm_ee/util/util_kubernetes.py

from kubernetes import client, config
from kubernetes.client.rest import ApiException


def get_secret_data(name) -> dict:
    # assume that we are executing in a kubernetes pod
    try:
        config.load_incluster_config()
    except config.ConfigException:
        # we are not running in kubernetes
        return {}
    # Read the namespace from the service account
    current_namespace = open(
        "/var/run/secrets/kubernetes.io/serviceaccount/namespace"
    ).read()

    v1 = client.CoreV1Api()
    try:
        secret = v1.read_namespaced_secret(name, current_namespace)
    except ApiException as e:
        if e.reason in ("Not Found", "Forbidden"):
            # Backwards compatibility: we run in k8s but certs don't exist, or we are running in a different namespace
            return {}
        else:
            raise
    return secret.data