test: add extension PipelinePolicy tests

Makefile

@@ -163,7 +163,8 @@ podmonitors.monitoring.coreos.com,$\
 apiservices.apiregistration.k8s.io,$\
 horizontalpodautoscalers.autoscaling,$\
 oidcpolicies.extensions.kuadrant.io,$\
-planpolicies.extensions.kuadrant.io
+planpolicies.extensions.kuadrant.io,$\
+pipelinepolicies.extensions.kuadrant.io
 
 clean: ## Clean all objects on cluster created by running this testsuite. Set the env variable USER to delete after someone else
 	@echo "Deleting objects for user: $(USER)"

config/settings.local.yaml.tpl

@@ -16,6 +16,8 @@
 #      password: "testPassword"
 #  spicedb:
 #    image: "SPICEDB_IMAGE"
+#  pipeline_policy_extension_service:
+#    image: "PIPELINE_POLICY_EXTENSION_SERVICE_IMAGE"
 #  auth0:
 #    client_id: "CLIENT_ID"
 #    client_secret: "CLIENT_SECRET"

config/settings.yaml

@@ -18,6 +18,8 @@ default:
     image: "quay.io/rhn_support_azgabur/mockserver:latest"
   spicedb:
     image: "quay.io/authzed/spicedb:latest"
+  pipeline_policy_extension_service:
+    image: "quay.io/kuadrant/threat-assessment-service:latest"
   prometheus:
     project: "openshift-monitoring"
     service: "thanos-querier"

testsuite/kuadrant/extensions/pipeline_policy.py

@@ -0,0 +1,133 @@
+"""Module containing classes related to PipelinePolicy"""
+
+from __future__ import annotations
+
+from functools import cached_property
+from typing import Dict, List, Optional
+
+from testsuite.gateway import Referencable
+from testsuite.kubernetes import modify
+from testsuite.kubernetes.client import KubernetesClient
+from testsuite.kuadrant.policy import Policy
+
+
+class ActionSection:
+    """Section for request/response actions in a PipelinePolicy, mirrors ActionSpec from the Go API"""
+
+    def __init__(self, obj: "PipelinePolicy", section_name: str) -> None:
+        self.obj = obj
+        self.section_name = section_name
+
+    def modify_and_apply(self, modifier_func, retries=2, cmd_args=None):
+        """Delegates modify_and_apply to the parent PipelinePolicy"""
+
+        def _new_modifier(obj):
+            modifier_func(ActionSection(obj, self.section_name))
+
+        return self.obj.modify_and_apply(_new_modifier, retries, cmd_args)
+
+    @property
+    def committed(self):
+        """Delegates committed check to the parent PipelinePolicy"""
+        return self.obj.committed
+
+    @property
+    def section(self):
+        """Returns the action list for this section"""
+        return self.obj.model.spec.setdefault(self.section_name, [])
+
+    @modify
+    def add_grpc_method(self, method: str, var: Optional[str] = None, predicate: Optional[str] = None):
+        """Add a grpc_method action that calls an upstream"""
+        action: Dict = {"type": "grpc_method", "method": method}
+        if var is not None:
+            action["var"] = var
+        if predicate is not None:
+            action["predicate"] = predicate
+        self.section.append(action)
+
+    @modify
+    def add_deny(
+        self,
+        predicate: Optional[str] = None,
+        with_status: Optional[int] = None,
+        with_headers: Optional[str] = None,
+        with_body: Optional[str] = None,
+    ):
+        """Add a deny action"""
+        action: Dict = {"type": "deny"}
+        if predicate is not None:
+            action["predicate"] = predicate
+        if with_status is not None:
+            action["withStatus"] = with_status
+        if with_headers is not None:
+            action["withHeaders"] = with_headers
+        if with_body is not None:
+            action["withBody"] = with_body
+        self.section.append(action)
+
+    @modify
+    def add_fail(self, log_message: str, predicate: Optional[str] = None):
+        """Add a fail action"""
+        action: Dict = {"type": "fail", "logMessage": log_message}
+        if predicate is not None:
+            action["predicate"] = predicate
+        self.section.append(action)
+
+    @modify
+    def add_headers(self, headers: List[List[str]], predicate: Optional[str] = None):
+        """Add an add_headers action"""
+        action: Dict = {"type": "add_headers", "headersToAdd": str(headers)}
+        if predicate is not None:
+            action["predicate"] = predicate
+        self.section.append(action)
+
+
+class PipelinePolicy(Policy):
+    """PipelinePolicy for defining declarative action pipelines (request/response actions) on routes"""
+
+    @classmethod
+    def create_instance(
+        cls,
+        cluster: KubernetesClient,
+        name: str,
+        target: Referencable,
+        labels: Dict[str, str] = None,
+        section_name: str = None,
+    ):
+        """Creates base instance"""
+        model: Dict = {
+            "apiVersion": "extensions.kuadrant.io/v1alpha1",
+            "kind": "PipelinePolicy",
+            "metadata": {"name": name, "namespace": cluster.project, "labels": labels},
+            "spec": {
+                "targetRef": target.reference,
+            },
+        }
+        if section_name:
+            model["spec"]["targetRef"]["sectionName"] = section_name
+
+        return cls(model, context=cluster.context)
+
+    @cached_property
+    def on_http_request(self) -> ActionSection:
+        """Gives access to request actions"""
+        return ActionSection(self, "request")
+
+    @cached_property
+    def on_http_response(self) -> ActionSection:
+        """Gives access to response actions"""
+        return ActionSection(self, "response")
+
+    @modify
+    def add_action_method(self, name: str, url: str, service: str, method: str, message_template: str):
+        """Add a gRPC upstream action method definition"""
+        self.model.spec.setdefault("actionMethods", []).append(
+            {
+                "name": name,
+                "url": url,
+                "service": service,
+                "method": method,
+                "messageTemplate": message_template,
+            }
+        )

testsuite/tests/singlecluster/extensions/pipeline_policy/conftest.py

@@ -0,0 +1,30 @@
+"""Shared fixtures for PipelinePolicy testing."""
+
+import pytest
+
+from openshift_client import OpenShiftPythonException
+
+from testsuite.kuadrant.extensions.pipeline_policy import PipelinePolicy
+
+
+@pytest.fixture(scope="session", autouse=True)
+def check_pipeline_policy_crd(cluster, skip_or_fail):
+    """Skip all PipelinePolicy tests if the CRD is not installed on the cluster."""
+    try:
+        cluster.do_action("get", "crd/pipelinepolicies.extensions.kuadrant.io")
+    except OpenShiftPythonException:
+        skip_or_fail("PipelinePolicy CRD is not installed on the cluster")
+
+
+@pytest.fixture(scope="module")
+def pipeline_policy(cluster, blame, route, module_label):
+    """PipelinePolicy targeting the test HTTPRoute"""
+    return PipelinePolicy.create_instance(cluster, blame("pipeline"), route, labels={"testRun": module_label})
+
+
+@pytest.fixture(scope="module", autouse=True)
+def commit(request, pipeline_policy):
+    """Commit and wait for PipelinePolicy to be ready."""
+    request.addfinalizer(pipeline_policy.delete)
+    pipeline_policy.commit()
+    pipeline_policy.wait_for_ready()

testsuite/tests/singlecluster/extensions/pipeline_policy/grpc/conftest.py

@@ -0,0 +1,61 @@
+"""Shared fixtures for PipelinePolicy gRPC tests."""
+
+import pytest
+
+from testsuite.kubernetes import Selector
+from testsuite.kubernetes.deployment import Deployment
+from testsuite.kubernetes.service import Service, ServicePort
+from testsuite.utils.constants import HTTP_API_PORT
+
+
+@pytest.fixture(scope="module")
+def threat_assessment_service(request, cluster, blame, module_label, testconfig):
+    """Deploys the ThreatAssessmentService gRPC backend"""
+    testconfig.validators.validate(only="pipeline_policy_extension_service")
+    name = blame("threat")
+    match_labels = {"app": module_label, "deployment": name}
+
+    deployment = Deployment.create_instance(
+        cluster,
+        name,
+        container_name="threat-assessment",
+        image=testconfig["pipeline_policy_extension_service"]["image"],
+        ports={"grpc": HTTP_API_PORT},
+        selector=Selector(matchLabels=match_labels),
+        labels={"app": module_label},
+        readiness_probe={"grpc": {"port": HTTP_API_PORT}, "initialDelaySeconds": 3, "periodSeconds": 5},
+    )
+    request.addfinalizer(deployment.delete)
+    deployment.commit()
+    deployment.wait_for_ready()
+
+    service = Service.create_instance(
+        cluster,
+        name,
+        selector=match_labels,
+        ports=[ServicePort(name="grpc", port=HTTP_API_PORT, targetPort="grpc")],
+        labels={"app": module_label},
+    )
+    request.addfinalizer(service.delete)
+    service.commit()
+    return service
+
+
+@pytest.fixture(scope="module")
+def threat_service_url(threat_assessment_service):
+    """gRPC URL for the threat assessment service."""
+    svc = threat_assessment_service
+    return f"grpc://{svc.name()}.{svc.namespace()}.svc.cluster.local:{HTTP_API_PORT}"
+
+
+@pytest.fixture(scope="module")
+def pipeline_policy(pipeline_policy, threat_service_url):
+    """PipelinePolicy with the threat assessment gRPC action method pre-registered."""
+    pipeline_policy.add_action_method(
+        name="assess",
+        url=threat_service_url,
+        service="threat.v1.ThreatAssessmentService",
+        method="AssessRequest",
+        message_template="threat.v1.ThreatRequest{uri: request.path}",
+    )
+    return pipeline_policy

testsuite/tests/singlecluster/extensions/pipeline_policy/grpc/test_grpc_basic.py

@@ -0,0 +1,50 @@
+"""Tests for PipelinePolicy grpc_method action: upstream calls and conditional execution."""
+
+import pytest
+
+from testsuite.utils.constants import THREAT_ASSESSMENT_THRESHOLD
+
+pytestmark = [pytest.mark.kuadrant_only, pytest.mark.extensions]
+
+
+@pytest.fixture(scope="module")
+def pipeline_policy(pipeline_policy):
+    """PipelinePolicy with conditional gRPC execution and threat-level deny."""
+    pipeline_policy.on_http_request.add_grpc_method(
+        method="assess",
+        var="threatResponse",
+        predicate='"x-assess-threat" in request.headers',
+    )
+    pipeline_policy.on_http_request.add_deny(predicate='request.url_path == "/blocked"', with_status=403)
+    pipeline_policy.on_http_request.add_deny(
+        predicate=f"threatResponse.threat_level >= {THREAT_ASSESSMENT_THRESHOLD}",
+        with_status=403,
+    )
+
+    pipeline_policy.on_http_response.add_headers(
+        [["x-threat-assessed", "true"]],
+        predicate='"x-assess-threat" in request.headers',
+    )
+    pipeline_policy.on_http_response.add_headers(
+        [["x-threat-assessed", "false"]],
+        predicate='!("x-assess-threat" in request.headers)',
+    )
+    pipeline_policy.on_http_response.add_headers([["x-threat-threshold", str(THREAT_ASSESSMENT_THRESHOLD)]])
+
+    return pipeline_policy
+
+
+def test_basic_grpc_upstream_call(client):
+    """gRPC upstream is called when predicate matches, response var is available to subsequent actions."""
+    response = client.get("/get", headers={"x-assess-threat": "true"})
+    assert response.status_code == 200
+    assert response.headers.get("x-threat-assessed") == "true"
+    assert response.headers.get("x-threat-threshold") == str(THREAT_ASSESSMENT_THRESHOLD)
+
+
+def test_conditional_grpc_call_skipped(client):
+    """gRPC upstream is not called when predicate does not match."""
+    response = client.get("/get")
+    assert response.status_code == 200
+    assert response.headers.get("x-threat-assessed") == "false"
+    assert response.headers.get("x-threat-threshold") == str(THREAT_ASSESSMENT_THRESHOLD)

testsuite/tests/singlecluster/extensions/pipeline_policy/grpc/test_grpc_composition.py

@@ -0,0 +1,88 @@
+"""Tests for PipelinePolicy composition of gRPC actions with deny and fail."""
+
+import pytest
+
+from testsuite.kuadrant.extensions.pipeline_policy import PipelinePolicy
+
+pytestmark = [pytest.mark.kuadrant_only, pytest.mark.extensions]
+
+
+@pytest.fixture(scope="module", autouse=True)
+def commit():
+    """No module-level policy; each test creates its own."""
+
+
+@pytest.fixture(scope="module")
+def create_grpc_policy(cluster, blame, route, threat_service_url, module_label):
+    """Factory for creating a PipelinePolicy with the assess gRPC action pre-registered."""
+
+    def _create(name):
+        policy = PipelinePolicy.create_instance(cluster, blame(name), route, labels={"testRun": module_label})
+        policy.add_action_method(
+            name="assess",
+            url=threat_service_url,
+            service="threat.v1.ThreatAssessmentService",
+            method="AssessRequest",
+            message_template="threat.v1.ThreatRequest{uri: request.path}",
+        )
+        policy.on_http_request.add_grpc_method(method="assess", var="threat")
+        return policy
+
+    return _create
+
+
+def test_fail_before_deny(request, create_grpc_policy, client):
+    """Fail action terminates the chain before the deny action when gRPC response triggers the fail predicate."""
+    policy = create_grpc_policy("failord")
+    policy.on_http_request.add_fail("threat too high", predicate="threat.threat_level >= 4")
+    policy.on_http_request.add_deny(predicate='request.url_path == "/blocked"', with_status=403)
+    request.addfinalizer(policy.delete)
+    policy.commit()
+    policy.wait_for_ready()
+
+    response = client.get("/admin")
+    assert response.status_code == 500
+
+
+def test_deny_after_grpc_call(request, create_grpc_policy, client):
+    """Deny action after gRPC call works when the deny predicate matches."""
+    policy = create_grpc_policy("grpc-deny")
+    policy.on_http_request.add_deny(predicate='request.url_path == "/blocked"', with_status=403)
+    request.addfinalizer(policy.delete)
+    policy.commit()
+    policy.wait_for_ready()
+
+    response = client.get("/blocked")
+    assert response.status_code == 403
+
+
+def test_deny_based_on_grpc_var(request, create_grpc_policy, client):
+    """Deny action using gRPC response variable denies requests when threat level is high."""
+    policy = create_grpc_policy("grpc-var-deny")
+    policy.on_http_request.add_deny(predicate="threat.threat_level >= 4", with_status=403)
+    request.addfinalizer(policy.delete)
+    policy.commit()
+    policy.wait_for_ready()
+
+    response = client.get("/admin")
+    assert response.status_code == 403
+
+    response = client.get("/get")
+    assert response.status_code == 200
+
+
+def test_deny_with_dynamic_body(request, create_grpc_policy, client):
+    """Deny action with CEL expression in withBody interpolates gRPC response variable."""
+    policy = create_grpc_policy("dyn-body")
+    policy.on_http_request.add_deny(
+        predicate="threat.threat_level >= 4",
+        with_status=403,
+        with_body="'blocked: threat level ' + string(threat.threat_level)",
+    )
+    request.addfinalizer(policy.delete)
+    policy.commit()
+    policy.wait_for_ready()
+
+    response = client.get("/admin")
+    assert response.status_code == 403
+    assert "blocked: threat level" in response.text