| Gabriel Cuba | f734996 | 2023-04-26 12:01:25 -0500 | [diff] [blame^] | 1 | from kubernetes import client, config |
| 2 | from kubernetes.client.rest import ApiException |
| 3 | |
| 4 | |
| 5 | def get_secret_data(name) -> dict: |
| 6 | # assume that we are executing in a kubernetes pod |
| 7 | try: |
| 8 | config.load_incluster_config() |
| 9 | except config.ConfigException: |
| 10 | # we are not running in kubernetes |
| 11 | return {} |
| 12 | # Read the namespace from the service account |
| 13 | current_namespace = open("/var/run/secrets/kubernetes.io/serviceaccount/namespace").read() |
| 14 | |
| 15 | v1 = client.CoreV1Api() |
| 16 | try: |
| 17 | secret = v1.read_namespaced_secret(name, current_namespace) |
| 18 | except ApiException as e: |
| 19 | if e.reason == 'Not Found': # Backwards compatibility: we run in k8s but certs don't exist |
| 20 | return {} |
| 21 | else: |
| 22 | raise |
| 23 | return secret.data |