Deploy dex

tests/integration/conftest.py

@@ -3,13 +3,13 @@
 
 """Fixtures for Jenkins-k8s-operator charm integration tests."""
 
+import os
 import random
 import secrets
 import string
 import typing
 
 import jenkinsapi.jenkins
-import kubernetes.client
 import kubernetes.config
 import kubernetes.stream
 import pytest
@@ -20,16 +20,21 @@
 from juju.model import Controller, Model
 from juju.unit import Unit
 from keycloak import KeycloakAdmin, KeycloakOpenIDConnection
+from lightkube import Client, KubeConfig
+from lightkube.core.exceptions import ApiError
 from pytest import FixtureRequest
 from pytest_operator.plugin import OpsTest
 
 import jenkins
 import state
 
 from .constants import ALLOWED_PLUGINS
+from .dex import apply_dex_resources, create_dex_resources, get_dex_manifest, get_dex_service_url
 from .helpers import get_pod_ip
 from .types_ import KeycloakOIDCMetadata, LDAPSettings, ModelAppUnit, UnitWebClient
 
+KUBECONFIG = os.environ.get("TESTING_KUBECONFIG", "~/.kube/config")
+
 
 @pytest.fixture(scope="module", name="model")
 def model_fixture(ops_test: OpsTest) -> Model:
@@ -843,6 +848,15 @@ async def oathkeeper_application_related_fixture(application: Application):
         "traefik-public:receive-ca-cert", "self_signed_certificates"
     )
     await application.model.add_relation(f"{oathkeeper.name}:kratos-endpoint-info", "kratos")
+    await application.model.applications["kratos-external-idp-integrator"].set_config(
+        {
+            "client_id": "client_id",
+            "client_secret": "client_secret",
+            "provider": "generic",
+            "issuer_url": "https://path/to/dex",
+            "scope": "profile email",
+        }
+    )
     await application.model.wait_for_idle(
         status="active",
         apps=[
@@ -857,6 +871,39 @@ async def oathkeeper_application_related_fixture(application: Application):
     return oathkeeper
 
 
+@pytest.fixture(scope="session", name="client")
+def client_fixture() -> Client:
+    """k8s client."""
+    return Client(config=KubeConfig.from_file(KUBECONFIG), field_manager="dex-test")
+
+
+@pytest.fixture(scope="module")
+def ext_idp_service(ops_test: OpsTest, client: Client) -> typing.Generator[str, None, None]:
+    """Deploy a DEX service on top of k8s for authentication."""
+    # Use ops-lib-manifests?
+    try:
+        create_dex_resources(client)
+
+        # We need to set the dex issuer_url to be the IP that was assigned to
+        # the dex service by metallb. We can't know that before hand, so we
+        # reapply the dex manifests.
+        apply_dex_resources(client)
+
+        yield get_dex_service_url(client)
+    finally:
+        if not ops_test.keep_model:
+            for obj in get_dex_manifest():
+                try:
+                    # mypy doesn't work well with lightkube
+                    client.delete(
+                        type(obj),
+                        obj.metadata.name,  # type: ignore
+                        namespace=obj.metadata.namespace,  # type: ignore
+                    )
+                except ApiError:
+                    pass
+
+
 @pytest.fixture()
 def external_user_email() -> str:
     """Username for testing proxy authentication."""

tests/integration/dex.py

@@ -0,0 +1,216 @@
+#!/usr/bin/env python3
+# Copyright 2023 Canonical Ltd.
+# See LICENSE file for licensing details.
+
+"""DEX deployment and utilities for testing."""
+
+import logging
+from os.path import join
+from pathlib import Path
+from time import sleep
+from typing import List, Optional
+
+import requests
+from lightkube import Client, codecs
+from lightkube.core.exceptions import ApiError, ObjectDeleted
+from lightkube.resources.apps_v1 import Deployment
+from lightkube.resources.core_v1 import Pod, Service
+from requests.exceptions import RequestException
+
+logger = logging.getLogger(__name__)
+
+
+DEX_MANIFESTS = Path(__file__).parent / "dex.yaml"
+
+
+def get_dex_manifest(
+    client_id: Optional[str] = None,
+    client_secret: Optional[str] = None,
+    redirect_uri: Optional[str] = None,
+    issuer_url: Optional[str] = None,
+) -> List[codecs.AnyResource]:
+    """Get the DEX manifest interpolating the needed variables.
+
+    Args:
+        client_id: client ID.
+        client_secret: client secret.
+        redirect_uri: redirect URI.
+        issuer_url: issuer URL.
+
+    Returns:
+        the list of created resources.
+    """
+    with open(DEX_MANIFESTS, "r", encoding="utf-8") as file:
+        return codecs.load_all_yaml(
+            file,
+            context={
+                "client_id": client_id,
+                "client_secret": client_secret,
+                "redirect_uri": redirect_uri,
+                "issuer_url": issuer_url,
+            },
+        )
+
+
+def _restart_dex(client: Client) -> None:
+    """Restart the DEX pods.
+
+    Args:
+        client: k8s client.
+    """
+    for pod in client.list(Pod, namespace="dex", labels={"app": "dex"}):
+        # mypy doesn't work well with lightkube
+        client.delete(Pod, pod.metadata.name, namespace="dex")  # type: ignore
+
+
+def _wait_until_dex_is_ready(client: Client, issuer_url: Optional[str] = None) -> None:
+    """Wait for DEX to be up.
+
+    Args:
+        client: k8s client.
+        issuer_url: issuer URL.
+
+    Raises:
+        RuntimeError: if DEX fails to start.
+    """
+    for pod in client.list(Pod, namespace="dex", labels={"app": "dex"}):
+        # Some pods may be deleted, if we are restarting
+        try:
+            # mypy doesn't work well with lightkube
+            client.wait(
+                Pod,
+                pod.metadata.name,  # type: ignore
+                for_conditions=["Ready", "Deleted"],
+                namespace="dex",
+            )
+        except ObjectDeleted:
+            pass
+    client.wait(Deployment, "dex", namespace="dex", for_conditions=["Available"])
+    if not issuer_url:
+        issuer_url = get_dex_service_url(client)
+
+    resp = requests.get(join(issuer_url, ".well-known/openid-configuration"), timeout=5)
+    if resp.status_code != 200:
+        raise RuntimeError("Failed to deploy dex")
+
+
+def wait_until_dex_is_ready(client: Client, issuer_url: Optional[str] = None) -> None:
+    """Wait for DEX to be up.
+
+    Args:
+        client: k8s client.
+        issuer_url: issuer URL.
+    """
+    try:
+        _wait_until_dex_is_ready(client, issuer_url)
+    except (RuntimeError, RequestException):
+        # It may take some time for dex to restart, so we sleep a little
+        # and try again
+        sleep(3)
+        _wait_until_dex_is_ready(client, issuer_url)
+
+
+def _apply_dex_manifests(
+    client: Client,
+    client_id: str = "client_id",
+    client_secret: str = "client_secret",
+    redirect_uri: str = "",
+    issuer_url: Optional[str] = None,
+) -> None:
+    """Apply the DEX manifest definitions.
+
+    Args:
+        client: k8s client.
+        client_id: client ID.
+        client_secret: client secret.
+        redirect_uri: redirect URI.
+        issuer_url: issuer URL.
+    """
+    objs = get_dex_manifest(
+        client_id=client_id,
+        client_secret=client_secret,
+        redirect_uri=redirect_uri,
+        issuer_url=issuer_url,
+    )
+
+    for obj in objs:
+        client.apply(obj, force=True)
+
+
+def create_dex_resources(
+    client: Client,
+    client_id: str = "client_id",
+    client_secret: str = "client_secret",
+    redirect_uri: str = "",
+    issuer_url: Optional[str] = None,
+):
+    """Apply the DEX manifest definitions and wait for DEX to be up.
+
+    Args:
+        client: k8s client.
+        client_id: client ID.
+        client_secret: client secret.
+        redirect_uri: redirect URI.
+        issuer_url: issuer URL.
+    """
+    _apply_dex_manifests(
+        client,
+        client_id=client_id,
+        client_secret=client_secret,
+        redirect_uri=redirect_uri,
+        issuer_url=issuer_url,
+    )
+
+    logger.info("Waiting for dex to be ready")
+    wait_until_dex_is_ready(client, issuer_url)
+
+
+def apply_dex_resources(
+    client: Client,
+    client_id: str = "client_id",
+    client_secret: str = "client_secret",
+    redirect_uri: str = "",
+    issuer_url: Optional[str] = None,
+) -> None:
+    """Apply the DEX manifest definitions and wait for DEX to start up.
+
+    Args:
+        client: k8s client.
+        client_id: client ID.
+        client_secret: client secret.
+        redirect_uri: redirect URI.
+        issuer_url: issuer URL.
+    """
+    if not issuer_url:
+        try:
+            issuer_url = get_dex_service_url(client)
+        except ApiError:
+            logger.info("No service found for dex")
+
+    _apply_dex_manifests(
+        client,
+        client_id=client_id,
+        client_secret=client_secret,
+        redirect_uri=redirect_uri,
+        issuer_url=issuer_url,
+    )
+
+    logger.info("Restarting dex")
+    _restart_dex(client)
+
+    logger.info("Waiting for dex to be ready")
+    wait_until_dex_is_ready(client, issuer_url)
+
+
+def get_dex_service_url(client: Client) -> str:
+    """Get the DEX service URL.
+
+    Args:
+    client: k8s client.
+
+    Returns:
+        the service URL.
+    """
+    service = client.get(Service, "dex", namespace="dex")
+    # mypy doesn't work well with lightkube
+    return f"http://{service.status.loadBalancer.ingress[0].ip}:5556/"  # type: ignore