feat: add RBAC test for SA token

lugi0 · lugi0 · commit 9b5871d1c6b5 · 2025-05-06T18:41:51.000+02:00
Signed-off-by: lugi0 &lt;lgiorgi@redhat.com&gt;
diff --git a/tests/model_registry/conftest.py b/tests/model_registry/conftest.py
@@ -1,7 +1,11 @@
 import pytest
 import re
+import random
+import string
+import subprocess
 import schemathesis
-from typing import Generator, Any
+import shlex
+from typing import Generator, Any, List, Dict
 from kubernetes.dynamic.exceptions import ResourceNotFoundError
 from ocp_resources.pod import Pod
 from ocp_resources.secret import Secret
@@ -10,6 +14,9 @@
 from ocp_resources.persistent_volume_claim import PersistentVolumeClaim
 from ocp_resources.data_science_cluster import DataScienceCluster
 from ocp_resources.deployment import Deployment
+from ocp_resources.service_account import ServiceAccount
+from ocp_resources.role import Role
+from ocp_resources.role_binding import RoleBinding
 
 from ocp_resources.model_registry import ModelRegistry
 import schemathesis.schemas
@@ -23,6 +30,7 @@
 from simple_logger.logger import get_logger
 from kubernetes.dynamic import DynamicClient
 from pytest_testconfig import config as py_config
+from pyhelper_utils.shell import run_command
 from model_registry.types import RegisteredModel
 from tests.model_registry.constants import (
     MR_OPERATOR_NAME,
@@ -44,6 +52,7 @@
 
 
 LOGGER = get_logger(name=__name__)
+DEFAULT_TOKEN_DURATION = "10m"
 
 
 @pytest.fixture(scope="class")
@@ -293,3 +302,194 @@ def model_registry_operator_pod(admin_client: DynamicClient) -> Pod:
     if not model_registry_operator_pods:
         raise ResourceNotFoundError("Model registry operator pod not found")
     return model_registry_operator_pods[0]
+
+
+# --- Fixture Helper Function ---
+
+
+def generate_random_name(prefix: str = "test", length: int = 8) -> str:
+    """Generates a random string for resource names."""
+    suffix = "".join(random.choices(string.ascii_lowercase + string.digits, k=length))
+    return f"{prefix}-{suffix}"
+
+
+# --- Service Account and Namespace Fixtures (Function Scoped for Isolation) ---
+
+
+@pytest.fixture(scope="function")
+def sa_namespace(admin_client: DynamicClient) -> Generator[Namespace, None, None]:
+    """
+    Creates a temporary namespace using a context manager for automatic cleanup.
+    Function scope ensures a fresh namespace for each test needing it.
+    """
+    ns_name = generate_random_name(prefix="mr-rbac-test-ns")
+    LOGGER.info(f"Creating temporary namespace: {ns_name}")
+    # Use context manager for creation and deletion
+    with Namespace(client=admin_client, name=ns_name) as ns:
+        try:
+            ns.wait_for_status(status=Namespace.Status.ACTIVE, timeout=120)
+            LOGGER.info(f"Namespace {ns_name} is active.")
+            yield ns
+            # Cleanup happens automatically when exiting 'with' block
+            LOGGER.info(f"Namespace {ns_name} deletion initiated by context manager.")
+            # Add a final wait within the fixture if immediate confirmation is needed,
+            # but context manager handles the delete call. Let's rely on the manager.
+            # Consider adding ns.wait_deleted(timeout=180) here if needed AFTER yield returns.
+        except Exception:
+            LOGGER.error(f"Timeout waiting for namespace {ns_name} to become active.")
+            pytest.fail(f"Namespace {ns_name} failed to become active.")
+
+
+@pytest.fixture(scope="function")
+def service_account(admin_client: DynamicClient, sa_namespace: Namespace) -> Generator[ServiceAccount, None, None]:
+    """
+    Creates a ServiceAccount within the temporary namespace using a context manager.
+    Function scope ensures it's tied to the lifetime of sa_namespace for that test.
+    """
+    sa_name = generate_random_name(prefix="mr-test-user")
+    LOGGER.info(f"Creating ServiceAccount: {sa_name} in namespace {sa_namespace.name}")
+    # Use context manager for creation and deletion
+    with ServiceAccount(client=admin_client, name=sa_name, namespace=sa_namespace.name) as sa:
+        try:
+            sa.wait(timeout=60)  # Wait for SA object to exist
+            LOGGER.info(f"ServiceAccount {sa_name} created.")
+            yield sa
+            # Cleanup happens automatically when exiting 'with' block
+            LOGGER.info(f"ServiceAccount {sa_name} deletion initiated by context manager.")
+        except Exception:
+            LOGGER.error(f"Timeout waiting for ServiceAccount {sa_name} to be created.")
+            pytest.fail(f"ServiceAccount {sa_name} failed to be created.")
+
+
+@pytest.fixture(scope="function")
+def sa_token(service_account: ServiceAccount) -> str:  # type: ignore[return]
+    """
+    Retrieves a short-lived token for the ServiceAccount using 'oc create token'.
+    Function scope because token is temporary and tied to the SA for that test.
+    """
+    sa_name = service_account.name
+    namespace = service_account.namespace  # Get namespace name from SA object
+    LOGGER.info(f"Retrieving token for ServiceAccount: {sa_name} in namespace {namespace}")
+    # (Keep the subprocess logic from previous version - it's appropriate here)
+    try:
+        cmd = f"oc create token {sa_name} -n {namespace} --duration={DEFAULT_TOKEN_DURATION}"
+        LOGGER.debug(f"Executing command: {cmd}")
+        res, out, err = run_command(command=shlex.split(cmd), verify_stderr=False, check=True)
+        token = out.strip()
+        if not token:
+            pytest.fail(f"Retrieved token is empty for SA {sa_name} in {namespace}.")
+        LOGGER.info(f"Successfully retrieved token for SA {sa_name}")
+        return token
+    except subprocess.CalledProcessError as e:
+        LOGGER.error(f"Failed to create token for SA {sa_name} ns {namespace}: {e.stderr}")
+        pytest.fail(f"Failed to create token for SA {sa_name}: {e.stderr}")
+    except subprocess.TimeoutExpired:
+        LOGGER.error(f"Timeout creating token for SA {sa_name} ns {namespace}")
+        pytest.fail(f"Timeout creating token for SA {sa_name}")
+    except Exception as e:
+        LOGGER.error(
+            f"An unexpected error occurred during token retrieval for SA {sa_name} ns {namespace}: {e}", exc_info=True
+        )
+        pytest.fail(f"Unexpected error getting token for SA {sa_name}: {e}")
+
+
+# --- RBAC Fixtures (Using Context Managers, Function Scoped) ---
+
+
+@pytest.fixture(scope="function")
+def mr_access_role(
+    admin_client: DynamicClient,
+    model_registry_namespace: str,  # Existing fixture from main conftest
+    sa_namespace: Namespace,  # Used for unique naming
+) -> Generator[Role, None, None]:
+    """
+    Creates the MR Access Role using direct constructor parameters and a context manager.
+    """
+    role_name = f"registry-user-{MR_INSTANCE_NAME}-{sa_namespace.name[:8]}"
+    LOGGER.info(f"Defining Role: {role_name} in namespace {model_registry_namespace}")
+
+    # Define rules directly as required by the Role constructor's 'rules' parameter
+    role_rules: List[Dict[str, Any]] = [
+        {
+            "apiGroups": [""],  # Core API group
+            "resources": ["services"],  # As per last refinement for REST access
+            "resourceNames": [MR_INSTANCE_NAME],  # Grant access only to the specific MR service object
+            "verbs": ["get"],
+        }
+    ]
+
+    # Define labels, to be passed via **kwargs
+    role_labels = {
+        "app.kubernetes.io/component": "model-registry-test-rbac",
+        "test.opendatahub.io/namespace": sa_namespace.name,
+    }
+
+    LOGGER.info(f"Attempting to create Role: {role_name} with rules and labels.")
+    # Use context manager for creation and deletion
+    # Pass rules and labels directly
+    with Role(
+        client=admin_client,
+        name=role_name,
+        namespace=model_registry_namespace,
+        rules=role_rules,
+        label=role_labels,  # Pass labels via kwargs
+    ) as role:
+        try:
+            role.wait(timeout=60)  # Wait for role object to exist
+            LOGGER.info(f"Role {role.name} created successfully.")
+            yield role
+            LOGGER.info(f"Role {role.name} deletion initiated by context manager.")
+        except Exception as e:  # Catch other potential errors during Role instantiation or wait
+            LOGGER.error(f"Error during Role {role_name} creation or wait: {e}", exc_info=True)
+            pytest.fail(f"Failed during Role {role_name} creation: {e}")
+
+
+@pytest.fixture(scope="function")
+def mr_access_role_binding(
+    admin_client: DynamicClient,
+    model_registry_namespace: str,  # Existing fixture from main conftest
+    mr_access_role: Role,  # Depend on the role fixture to get its name
+    sa_namespace: Namespace,  # The namespace containing the test SA
+) -> Generator[RoleBinding, None, None]:
+    """
+    Creates the MR Access RoleBinding using direct constructor parameters and a context manager.
+    """
+    binding_name = f"{mr_access_role.name}-binding"  # Simplify name slightly, role name is already unique
+    role_name_ref = mr_access_role.name  # Get the actual name from the created Role object
+
+    LOGGER.info(
+        f"Defining RoleBinding: {binding_name} linking Group 'system:serviceaccounts:{sa_namespace.name}' "
+        f"to Role '{role_name_ref}' in namespace {model_registry_namespace}"
+    )
+
+    # Define labels, to be passed via **kwargs
+    binding_labels = {
+        "app.kubernetes.io/component": "model-registry-test-rbac",
+        "test.opendatahub.io/namespace": sa_namespace.name,
+    }
+
+    LOGGER.info(f"Attempting to create RoleBinding: {binding_name} with labels.")
+    # Use context manager for creation and deletion
+    # Pass subject and role_ref details directly to constructor
+    with RoleBinding(
+        client=admin_client,
+        name=binding_name,
+        namespace=model_registry_namespace,
+        # Subject parameters
+        subjects_kind="Group",
+        subjects_name=f"system:serviceaccounts:{sa_namespace.name}",
+        subjects_api_group="rbac.authorization.k8s.io",  # This is the default apiGroup for Group kind
+        # Role reference parameters
+        role_ref_kind="Role",
+        role_ref_name=role_name_ref,
+        # role_ref_api_group="rbac.authorization.k8s.io", # This is automatically set by the class
+        label=binding_labels,  # Pass labels via kwargs
+    ) as binding:
+        try:
+            binding.wait(timeout=60)  # Wait for binding object to exist
+            LOGGER.info(f"RoleBinding {binding.name} created successfully.")
+            yield binding
+            LOGGER.info(f"RoleBinding {binding.name} deletion initiated by context manager.")
+        except Exception as e:  # Catch other potential errors
+            LOGGER.error(f"Error during RoleBinding {binding_name} creation or wait: {e}", exc_info=True)
+            pytest.fail(f"Failed during RoleBinding {binding_name} creation: {e}")
diff --git a/tests/model_registry/rbac/test_mr_rbac_sa.py b/tests/model_registry/rbac/test_mr_rbac_sa.py
@@ -0,0 +1,145 @@
+# AI Disclaimer: Google Gemini 2.5 pro has been used to generate a majority of this code, with human review and editing.
+import pytest
+from typing import Self, Dict, Any
+from simple_logger.logger import get_logger
+
+# Framework / Client specific imports
+from model_registry import ModelRegistry as ModelRegistryClient
+from mr_openapi.exceptions import ForbiddenException
+
+from utilities.constants import DscComponents, Protocols  # Need Protocols.HTTPS
+from tests.model_registry.constants import MR_NAMESPACE
+
+
+LOGGER = get_logger(name=__name__)
+
+
+# --- Helper to build client args ---
+# Based on the 'model_registry_client' fixture
+def build_mr_client_args(rest_endpoint: str, token: str, author: str) -> Dict[str, Any]:
+    """Builds arguments for ModelRegistryClient based on REST endpoint and token."""
+    try:
+        server, port = rest_endpoint.split(":")
+
+        # Conftest example uses is_secure=False - this implies TLS verification might be off
+        # Adjust if your environment requires strict TLS verification
+        return {
+            "server_address": f"{Protocols.HTTPS}://{server}",
+            "port": port,
+            "user_token": token,
+            "is_secure": False,  # Match conftest example (disable TLS verification)
+            "author": author,
+        }
+    except Exception as e:
+        LOGGER.error(f"Error parsing REST endpoint '{rest_endpoint}': {e}")
+        raise ValueError(f"Could not parse REST endpoint: {rest_endpoint}") from e
+
+
+# --- Test Class ---
+
+
+# Parametrize to ensure Model Registry component is enabled via DSC fixture
+@pytest.mark.parametrize(
+    "updated_dsc_component_state_scope_class",
+    [
+        pytest.param(
+            {
+                "component_patch": {
+                    DscComponents.MODELREGISTRY: {
+                        "managementState": DscComponents.ManagementState.MANAGED,
+                        "registriesNamespace": MR_NAMESPACE,
+                    },
+                }
+            },
+            id="enable_modelregistry_default_ns",
+        )
+    ],
+    indirect=True,
+    scope="class",
+)
+@pytest.mark.usefixtures("updated_dsc_component_state_scope_class")
+class TestModelRegistryRBAC:
+    """
+    Tests RBAC for Model Registry REST endpoint using ServiceAccount tokens.
+    """
+
+    @pytest.mark.smoke
+    @pytest.mark.usefixtures("sa_namespace", "service_account")  # Need SA and NS for token
+    def test_service_account_access_denied(
+        self: Self,
+        sa_token: str,  # Get token for the test SA
+        model_registry_instance_rest_endpoint: str,  # REST endpoint fixture
+    ):
+        """
+        Verifies SA access is DENIED (403 Forbidden) by default via REST.
+        Does NOT use mr_access_role or mr_access_role_binding fixtures.
+        """
+        LOGGER.info("--- Starting RBAC Test: Access Denied ---")
+        LOGGER.info(f"Targeting Model Registry REST endpoint: {model_registry_instance_rest_endpoint}")
+        LOGGER.info("Expecting initial access DENIAL (403 Forbidden)")
+
+        try:
+            client_args = build_mr_client_args(
+                rest_endpoint=model_registry_instance_rest_endpoint, token=sa_token, author="rbac-test-denied"
+            )
+            LOGGER.debug(f"Attempting client connection with args: {client_args}")
+
+            # Expect an exception related to HTTP 403
+            # Adjust exception type based on ModelRegistryClient's behavior
+            with pytest.raises(ForbiddenException) as exc_info:  # Or client's custom error e.g. ApiClientError
+                _ = ModelRegistryClient(**client_args)
+
+            # Verify the status code from the caught exception
+            http_error = exc_info.value
+            assert http_error.body is not None, "HTTPError should have a response object"
+            LOGGER.info(f"Received expected HTTP error: Status Code {http_error.status}")
+            assert http_error.status == 403, f"Expected HTTP 403 Forbidden, but got {http_error.status}"
+            LOGGER.info("Successfully received expected HTTP 403 status code.")
+
+        except Exception as e:
+            LOGGER.error(f"Received unexpected error during 'access denied' check: {e}", exc_info=True)
+            pytest.fail(f"'Access denied' check failed unexpectedly: {e}")
+
+    @pytest.mark.smoke
+    # Use fixtures for SA/NS/Token AND the RBAC Role/Binding
+    @pytest.mark.usefixtures("sa_namespace", "service_account", "mr_access_role", "mr_access_role_binding")
+    def test_service_account_access_granted(
+        self: Self,
+        sa_token: str,  # Get token for the test SA
+        model_registry_instance_rest_endpoint: str,  # REST endpoint fixture
+        # mr_access_role and mr_access_role_binding are activated by @usefixtures
+    ):
+        """
+        Verifies SA access is GRANTED via REST after applying Role and RoleBinding fixtures.
+        """
+        LOGGER.info("--- Starting RBAC Test: Access Granted ---")
+        LOGGER.info(f"Targeting Model Registry REST endpoint: {model_registry_instance_rest_endpoint}")
+        LOGGER.info("Applied RBAC Role/Binding via fixtures. Expecting access GRANT.")
+
+        try:
+            client_args = build_mr_client_args(
+                rest_endpoint=model_registry_instance_rest_endpoint, token=sa_token, author="rbac-test-granted"
+            )
+            LOGGER.debug(f"Attempting client connection with args: {client_args}")
+
+            # Instantiate the client - this should now succeed or fail for non-auth reasons
+            mr_client_success = ModelRegistryClient(**client_args)
+            assert mr_client_success is not None, "Client initialization failed after granting permissions"
+            LOGGER.info("Client instantiated successfully after granting permissions.")
+
+        except ForbiddenException as e:
+            # If we get an HTTP error here, it's unexpected, especially 403
+            LOGGER.error(
+                f"Received unexpected HTTP error after granting access: {e.status if e.body else 'No Response'} - {e}",
+                exc_info=True,
+            )
+            if e.body is not None and e.status == 403:
+                pytest.fail(f"Still received HTTP 403 Forbidden after applying Role/RoleBinding: {e}")
+            else:
+                pytest.fail(f"Client interaction failed with HTTP error after granting permissions: {e}")
+
+        except Exception as e:
+            LOGGER.error(f"Received unexpected general error after granting access: {e}", exc_info=True)
+            pytest.fail(f"Client interaction failed unexpectedly after granting permissions: {e}")
+
+        LOGGER.info("--- RBAC Test Completed Successfully ---")
