API: refactor webapp to make use of FastAPI dependecy injection

unify all interaction with the kubernetes API and the kubeseal
binaries in reusable clients, which exist as mock and as productive
implementations. the clients are injected using the FastAPI
dependency injection system

.github/workflows/kind.yaml

@@ -131,8 +131,13 @@ jobs:
 
           for secret_name in strict-secret namespace-secret cluster-secret; do
             echo -n "Testing ${secret_name} "
-            test "$(kubectl get secret "${secret_name}" -n e2e \
-            -o go-template --template '{{ index .data "a-secret" }}')" = "YQ==" ||
-           { echo ERR; exit 1; } &&
-           echo OK
+            secret_value=$(kubectl get secret "${secret_name}" -n e2e \
+              -o go-template --template '{{ index .data "a-secret" }}')
+            if test "$secret_value" = "YQ=="; then 
+              echo OK
+            else
+              echo ERR
+              echo "Unexpected secret value '${secret_value}'" 1>&2
+              exit 1
+            fi
           done

Dockerfile.api

@@ -33,6 +33,7 @@ ENV UVICORN_PORT=5000 \
     UVICORN_HOST=0.0.0.0 \
     UVICORN_NO_DATE_HEADER=1 \
     UVICORN_NO_SERVER_HEADER=1 \
+    KUBESEAL_CERT=/tmp/cert.pem \
     KUBESEAL_BINARY=/tmp/kubeseal
 
 WORKDIR ${APP_PATH}

api/kubeseal_webgui_api/app.py

@@ -3,7 +3,8 @@
 import fastapi
 from fastapi.middleware.cors import CORSMiddleware
 
-from .app_config import fetch_sealed_secrets_cert
+from .app_config import settings
+from .dependencies import get_kubeseal_client
 from .routers import config, kubernetes, kubeseal
 
 LOGGER = logging.getLogger("kubeseal-webgui")
@@ -36,7 +37,10 @@
 @app.on_event("startup")
 def startup_event():
     LOGGER.info("Running startup tasks...")
-    fetch_sealed_secrets_cert()
+    if settings.kubeseal_autofetch:
+        with open(settings.kubeseal_cert, "w") as file:
+            LOGGER.info("Saving certificate in '%s'", settings.kubeseal_cert)
+            file.write(get_kubeseal_client().get_certificate())
     LOGGER.info("Startup tasks complete.")
 
 

api/kubeseal_webgui_api/app_config.py

@@ -1,85 +1,26 @@
-import logging
-import subprocess
-from os import environ
+from pathlib import Path
 
 from pydantic import BaseSettings
 
-binary = environ.get("KUBESEAL_BINARY", "/bin/false")
-mock = environ.get("MOCK_ENABLED", "False").lower() == "true"
-autofetch = environ.get("KUBESEAL_AUTOFETCH", "false")
-kubeseal_cert = environ.get("KUBESEAL_CERT", "/kubeseal-webgui/cert/kubeseal-cert.pem")
-
 
 class AppSettings(BaseSettings):
-    kubeseal_version: str
-    kubeseal_binary: str = binary
-    kubeseal_cert: str = environ.get("KUBESEAL_CERT", "/dev/null")
-    mock_enabled: bool = mock
+    kubeseal_binary: Path = Path("/bin/false")
+    kubeseal_cert: Path = Path("/dev/null")
+    kubeseal_autofetch: bool = False
+    mock_enabled: bool = False
+    sealed_secrets_namespace: str = "sealed-secrets"
+    sealed_secrets_controller_name: str = "sealed-secrets-controller"
     mock_namespace_count: int = 120
 
-
-LOGGER = logging.getLogger("kubeseal-webgui")
-
-
-def fetch_sealed_secrets_cert():
-    if mock or autofetch == "false":
-        return
-
-    sealed_secrets_namespace = environ.get(
-        "KUBESEAL_CONTROLLER_NAMESPACE", "sealed-secrets"
-    )
-    sealed_secrets_controller_name = environ.get(
-        "KUBESEAL_CONTROLLER_NAME", "sealed-secrets-controller"
-    )
-
-    LOGGER.info(
-        "Fetch certificate from sealed secrets controller '%s' in namespace '%s'",
-        sealed_secrets_controller_name,
-        sealed_secrets_namespace,
-    )
-    kubeseal_subprocess = subprocess.Popen(
-        [
-            binary,
-            "--fetch-cert",
-            "--controller-name",
-            sealed_secrets_controller_name,
-            "--controller-namespace",
-            sealed_secrets_namespace,
-        ],
-        stdout=subprocess.PIPE,
-        stderr=subprocess.PIPE,
-        encoding="utf-8",
-    )
-    output, error = kubeseal_subprocess.communicate()
-    if error:
-        error_message = f"Error in run_kubeseal: {error}"
-        LOGGER.error(error_message)
-        raise RuntimeError(error_message)
-    with open(kubeseal_cert, "w") as file:
-        LOGGER.info("Saving certificate in '%s'", kubeseal_cert)
-        file.write(output)
-
-
-def get_kubeseal_version() -> str:
-    """Retrieve the kubeseal binary version."""
-    LOGGER.debug("Retrieving kubeseal binary version.")
-    kubeseal_subprocess = subprocess.Popen(
-        [binary, "--version"],
-        stdout=subprocess.PIPE,
-        stderr=subprocess.PIPE,
-        encoding="utf-8",
-    )
-    output, error = kubeseal_subprocess.communicate()
-    if error:
-        error_message = f"Error in run_kubeseal: {error}"
-        LOGGER.error(error_message)
-        raise RuntimeError(error_message)
-
-    version = "".join(output.split("\n"))
-
-    return str(version).split(":")[1].replace('"', "").lstrip()
+    class Config:
+        fields = {
+            "sealed_secrets_namespace": {
+                "env": ["sealed_secrets_namespace", "KUBESEAL_CONTROLLER_NAMESPACE"],
+            },
+            "sealed_secrets_controller_name": {
+                "env": ["sealed_secrets_controller_name", "KUBESEAL_CONTROLLER_NAME"],
+            },
+        }
 
 
-settings = AppSettings(
-    kubeseal_version=get_kubeseal_version(),
-)
+settings = AppSettings()

api/kubeseal_webgui_api/dependencies.py

@@ -0,0 +1,26 @@
+from kubeseal_webgui_api.app_config import settings
+from kubeseal_webgui_api.internal import (
+    KubernetesClient,
+    KubesealClient,
+    MockKubernetesClient,
+    MockKubesealClient,
+)
+
+
+def get_kubeseal_client() -> KubesealClient:
+    if settings.mock_enabled:
+        return MockKubesealClient()
+
+    return KubesealClient(
+        cert=settings.kubeseal_cert,
+        binary=settings.kubeseal_binary,
+        namespace=settings.sealed_secrets_namespace,
+        controller=settings.sealed_secrets_controller_name,
+    )
+
+
+def get_kubernetes_client() -> KubernetesClient:
+    if settings.mock_enabled:
+        return MockKubernetesClient(namespace_count=settings.mock_namespace_count)
+
+    return KubernetesClient()

api/kubeseal_webgui_api/internal/__init__.py

@@ -0,0 +1,4 @@
+# flake8: noqa
+
+from .kubernetes_client import KubernetesClient, MockKubernetesClient
+from .kubeseal_client import DEFAULT_KUBESEAL_CERT, DEFAULT_KUBESEAL_BINARY, KubesealClient, MockKubesealClient

api/kubeseal_webgui_api/internal/kubernetes_client.py

@@ -0,0 +1,129 @@
+import itertools
+import logging
+import random
+from typing import List, Optional
+
+from kubernetes import config
+from kubernetes.client import ApiClient, CoreV1Api, V1NamespaceList
+
+LOGGER = logging.getLogger("kubeseal-webgui")
+
+
+class KubernetesClient:
+    def __init__(self, api_client: Optional[ApiClient] = None):
+        self._api_client = api_client
+
+    def get_namespaces(self) -> List[str]:
+        """Retrieve a list of namespaces from current kubernetes cluster."""
+        namespaces_list = []
+
+        LOGGER.info("Resolving in-cluster Namespaces")
+        v1 = CoreV1Api(api_client=self.api_client())
+        namespaces = v1.list_namespace()
+        if isinstance(namespaces, V1NamespaceList) and namespaces.items:
+            for ns in namespaces.items:
+                namespaces_list.append(ns.metadata.name)
+        else:
+            LOGGER.warning("No valid namespace list available via %s", namespaces)
+
+        LOGGER.debug("Namespaces list %s", namespaces_list)
+        return namespaces_list
+
+    def api_client(self) -> ApiClient:
+        if self._api_client is not None:
+            return self._api_client
+
+        config.load_incluster_config()
+        return ApiClient()
+
+
+adjectives = [
+    "altered",
+    "angry",
+    "big",
+    "blinking",
+    "boring",
+    "broken",
+    "bubbling",
+    "calculating",
+    "cute",
+    "diffing",
+    "expensive",
+    "fresh",
+    "fierce",
+    "floating",
+    "generous",
+    "golden",
+    "green",
+    "growing",
+    "hidden",
+    "hideous",
+    "interesting",
+    "kubed",
+    "mumbling",
+    "rusty",
+    "singing",
+    "small",
+    "sniffing",
+    "squared",
+    "talking",
+    "trusty",
+    "wise",
+    "walking",
+    "zooming",
+]
+nouns = [
+    "ant",
+    "bike",
+    "bird",
+    "captain",
+    "cheese",
+    "clock",
+    "digit",
+    "gorilla",
+    "kraken",
+    "number",
+    "maven",
+    "monitor",
+    "moose",
+    "moon",
+    "mouse",
+    "news",
+    "newt",
+    "octopus",
+    "opossum",
+    "otter",
+    "paper",
+    "passenger",
+    "potato",
+    "ship",
+    "spaceship",
+    "spaghetti",
+    "spoon",
+    "store",
+    "tomcat",
+    "trombone",
+    "unicorn",
+    "vine",
+    "whale",
+]
+
+
+class MockKubernetesClient(KubernetesClient):
+    def __init__(self, namespace_count: int = 50):
+        super().__init__()
+
+        self.namespace_count = namespace_count
+
+    def get_namespaces(self) -> List[str]:
+        """Generate a list of namespaces."""
+        LOGGER.debug("Generating %d namespaces", self.namespace_count)
+
+        return sorted(
+            {
+                "-".join(words)
+                for words in random.choices(  # noqa: S311 no security needed here
+                    list(itertools.product(adjectives, nouns)), k=self.namespace_count
+                )
+            }
+        )