Add test for Raw kserve auth

tests/model_serving/model_server/authentication/conftest.py

@@ -15,7 +15,7 @@
 from ocp_resources.serving_runtime import ServingRuntime
 from pyhelper_utils.shell import run_command
 
-from utilities.infra import create_isvc_view_role, create_ns, s3_endpoint_secret
+from utilities.infra import create_isvc_view_role, create_ns, s3_endpoint_secret, create_inference_token
 from tests.model_serving.model_server.utils import create_isvc
 from utilities.constants import (
     KServeDeploymentType,
@@ -25,6 +25,8 @@
     RuntimeTemplates,
 )
 from utilities.serving_runtime import ServingRuntimeFromTemplate
+from utilities.constants import Annotations
+from utilities.constants import Labels
 
 
 # GRPC model serving
@@ -92,6 +94,20 @@ def http_view_role(
         yield role
 
 
+@pytest.fixture(scope="class")
+def http_raw_view_role(
+    admin_client: DynamicClient,
+    http_s3_caikit_raw_inference_service: InferenceService,
+) -> Role:
+    with create_isvc_view_role(
+        client=admin_client,
+        isvc=http_s3_caikit_raw_inference_service,
+        name=f"{http_s3_caikit_raw_inference_service.name}-view",
+        resource_names=[http_s3_caikit_raw_inference_service.name],
+    ) as role:
+        yield role
+
+
 @pytest.fixture(scope="class")
 def http_role_binding(
     admin_client: DynamicClient,
@@ -111,11 +127,33 @@ def http_role_binding(
         yield rb
 
 
+@pytest.fixture(scope="class")
+def http_raw_role_binding(
+    admin_client: DynamicClient,
+    http_raw_view_role: Role,
+    model_service_account: ServiceAccount,
+    http_s3_caikit_raw_inference_service: InferenceService,
+) -> RoleBinding:
+    with RoleBinding(
+        client=admin_client,
+        namespace=model_service_account.namespace,
+        name=f"{Protocols.HTTP}-{model_service_account.name}-view",
+        role_ref_name=http_raw_view_role.name,
+        role_ref_kind=http_raw_view_role.kind,
+        subjects_kind=model_service_account.kind,
+        subjects_name=model_service_account.name,
+    ) as rb:
+        yield rb
+
+
 @pytest.fixture(scope="class")
 def http_inference_token(model_service_account: ServiceAccount, http_role_binding: RoleBinding) -> str:
-    return run_command(
-        command=shlex.split(f"oc create token -n {model_service_account.namespace} {model_service_account.name}")
-    )[1].strip()
+    return create_inference_token(model_service_account=model_service_account)
+
+
+@pytest.fixture(scope="class")
+def http_raw_inference_token(model_service_account: ServiceAccount, http_raw_role_binding: RoleBinding) -> str:
+    return create_inference_token(model_service_account=model_service_account)
 
 
 @pytest.fixture()
@@ -127,14 +165,31 @@ def patched_remove_authentication_isvc(
         patches={
             http_s3_caikit_serverless_inference_service: {
                 "metadata": {
-                    "annotations": {"security.opendatahub.io/enable-auth": "false"},
+                    "annotations": {Annotations.KserveAuth.SECURITY: "false"},
                 }
             }
         }
     ):
         yield http_s3_caikit_serverless_inference_service
 
 
+@pytest.fixture()
+def patched_remove_raw_authentication_isvc(
+    admin_client: DynamicClient,
+    http_s3_caikit_raw_inference_service: InferenceService,
+) -> InferenceService:
+    with ResourceEditor(
+        patches={
+            http_s3_caikit_raw_inference_service: {
+                "metadata": {
+                    "labels": {Labels.KserveAuth.SECURITY: "false"},
+                }
+            }
+        }
+    ):
+        yield http_s3_caikit_raw_inference_service
+
+
 @pytest.fixture(scope="class")
 def grpc_view_role(admin_client: DynamicClient, grpc_s3_inference_service: InferenceService) -> Role:
     with create_isvc_view_role(
@@ -197,6 +252,30 @@ def http_s3_caikit_serverless_inference_service(
         yield isvc
 
 
+@pytest.fixture(scope="class")
+def http_s3_caikit_raw_inference_service(
+    request: FixtureRequest,
+    admin_client: DynamicClient,
+    model_namespace: Namespace,
+    http_s3_caikit_tgis_serving_runtime: ServingRuntime,
+    s3_models_storage_uri: str,
+    model_service_account: ServiceAccount,
+) -> InferenceService:
+    with create_isvc(
+        client=admin_client,
+        name=f"{Protocols.HTTP}-{ModelFormat.CAIKIT}",
+        namespace=model_namespace.name,
+        runtime=http_s3_caikit_tgis_serving_runtime.name,
+        storage_uri=s3_models_storage_uri,
+        model_format=http_s3_caikit_tgis_serving_runtime.instance.spec.supportedModelFormats[0].name,
+        deployment_mode=KServeDeploymentType.RAW_DEPLOYMENT,
+        model_service_account=model_service_account.name,
+        enable_auth=True,
+        external_route=True,
+    ) as isvc:
+        yield isvc
+
+
 # Unprivileged user tests
 @pytest.fixture(scope="class")
 def unprivileged_model_namespace(

tests/model_serving/model_server/authentication/test_kserve_token_authentication_raw.py

@@ -0,0 +1,46 @@
+import pytest
+
+from tests.model_serving.model_server.utils import verify_inference_response
+from utilities.constants import ModelFormat, ModelStoragePath, Protocols, ModelInferenceRuntime
+from utilities.inference_utils import Inference
+
+pytestmark = pytest.mark.usefixtures("valid_aws_config")
+
+
+@pytest.mark.rawdeployment
+@pytest.mark.parametrize(
+    "model_namespace, s3_models_storage_uri",
+    [
+        pytest.param(
+            {"name": "kserve-raw-token-authentication"},
+            {"model-dir": ModelStoragePath.FLAN_T5_SMALL},
+        )
+    ],
+    indirect=True,
+)
+class TestKserveTokenAuthenticationRawForRest:
+    @pytest.mark.smoke
+    @pytest.mark.dependency(name="test_model_authentication_using_rest_raw")
+    def test_model_authentication_using_rest_raw(self, http_s3_caikit_raw_inference_service, http_raw_inference_token):
+        """Verify RAW Kserve model query with token using REST"""
+        verify_inference_response(
+            inference_service=http_s3_caikit_raw_inference_service,
+            runtime=ModelInferenceRuntime.CAIKIT_TGIS_RUNTIME,
+            inference_type=Inference.ALL_TOKENS,
+            protocol=Protocols.HTTPS,
+            model_name=ModelFormat.CAIKIT,
+            use_default_query=True,
+            token=http_raw_inference_token,
+        )
+
+    @pytest.mark.dependency(name="test_disabled_raw_model_authentication")
+    def test_disabled_raw_model_authentication(self, patched_remove_raw_authentication_isvc):
+        """Verify model query after authentication is disabled"""
+        verify_inference_response(
+            inference_service=patched_remove_raw_authentication_isvc,
+            runtime=ModelInferenceRuntime.CAIKIT_TGIS_RUNTIME,
+            inference_type=Inference.ALL_TOKENS,
+            protocol=Protocols.HTTP,
+            model_name=ModelFormat.CAIKIT,
+            use_default_query=True,
+        )

tests/model_serving/model_server/utils.py

@@ -12,6 +12,7 @@
 from utilities.constants import (
     Annotations,
     KServeDeploymentType,
+    Labels,
 )
 from utilities.exceptions import (
     FailedPodsError,
@@ -129,9 +130,9 @@ def create_isvc(
     if enable_auth:
         # model mesh auth is set in servingruntime
         if deployment_mode == KServeDeploymentType.SERVERLESS:
-            annotations["security.opendatahub.io/enable-auth"] = "true"
+            annotations[Annotations.KserveAuth.SECURITY] = "true"
         elif deployment_mode == KServeDeploymentType.RAW_DEPLOYMENT:
-            labels["security.openshift.io/enable-authentication"] = "true"
+            labels[Labels.KserveAuth.SECURITY] = "true"
 
     # default to True if deployment_mode is Serverless (default behavior of Serverless) if was not provided by the user
     # model mesh external route is set in servingruntime

utilities/constants.py

@@ -122,6 +122,9 @@ class KubernetesIo:
     class KserveIo:
         DEPLOYMENT_MODE: str = "serving.kserve.io/deploymentMode"
 
+    class KserveAuth:
+        SECURITY: str = "security.opendatahub.io/enable-auth"
+
 
 class StorageClassName:
     NFS: str = "nfs"
@@ -148,6 +151,12 @@ class ConditionType:
     }
 
 
+class Labels:
+    class KserveAuth:
+        SECURITY: str = "security.opendatahub.io/enable-auth"
+
+
+MODEL_REGISTRY: str = "model-registry"
 MODELMESH_SERVING: str = "modelmesh-serving"
 ISTIO_CA_BUNDLE_FILENAME: str = "istio_knative.crt"
 OPENSHIFT_CA_BUNDLE_FILENAME: str = "openshift_ca.crt"

utilities/infra.py

@@ -22,6 +22,7 @@
 from ocp_resources.route import Route
 from ocp_resources.secret import Secret
 from ocp_resources.service import Service
+from ocp_resources.service_account import ServiceAccount
 from ocp_resources.serving_runtime import ServingRuntime
 from pyhelper_utils.shell import run_command
 from pytest_testconfig import config as py_config
@@ -324,3 +325,19 @@ def get_model_mesh_route(client: DynamicClient, isvc: InferenceService) -> Route
         return routes[0]
 
     raise ResourceNotFoundError(f"{isvc.name} has no routes")
+
+
+def create_inference_token(model_service_account: ServiceAccount) -> str:
+    """
+    Generates an inference token for the given model service account.
+
+    Args:
+        model_service_account (ServiceAccount): An object containing the namespace and name
+                               of the service account.
+
+    Returns:
+        str: The generated inference token.
+    """
+    return run_command(
+        shlex.split(f"oc create token -n {model_service_account.namespace} {model_service_account.name}")
+    )[1].strip()