Tests for Open Policy Agent (OPA) Rego policies

testsuite/objects/__init__.py

@@ -56,6 +56,10 @@ def remove_host(self, hostname):
     def remove_all_hosts(self):
         """Remove host"""
 
+    @abc.abstractmethod
+    def add_opa_policy(self, name, rego_policy):
+        """Adds OPA inline Rego policy"""
+
 
 class PreexistingAuthorino(Authorino):
     """Authorino which is already deployed prior to the testrun"""

testsuite/openshift/objects/auth_config.py

@@ -107,3 +107,14 @@ def remove_all_identities(self):
         """Removes all identities from AuthConfig"""
         identities = self.model.spec.setdefault("identity", [])
         identities.clear()
+
+    @modify
+    def add_opa_policy(self, name, rego_policy):
+        """Adds Opa (https://www.openpolicyagent.org/docs/latest/) policy to the AuthConfig"""
+        policy = self.model.spec.setdefault("authorization", [])
+        policy.append({
+            "name": name,
+            "opa": {
+                "inlineRego": rego_policy
+            }
+        })

testsuite/tests/kuadrant/authorino/authorization/test_opa.py

@@ -0,0 +1,33 @@
+"""Tests for Open Policy Agent (OPA) Rego policies"""
+import pytest
+
+
+@pytest.fixture(scope="module")
+def header():
+    """Header used by OPA policy"""
+    return "opa", "opa-test"
+
+
+@pytest.fixture(scope="module")
+def authorization(authorization, header):
+    """
+    Creates AuthConfig with API key identity and configures it with OPA policy
+    that accepts only those requests that contain header correct header
+    """
+    key, value = header
+    rego_inline = f"allow {{ input.context.request.http.headers.{key} == \"{value}\" }}"
+    authorization.add_opa_policy("opa", rego_inline)
+    return authorization
+
+
+def test_authorized_by_opa(client, auth, header):
+    """Tests a request that should be authorized by OPA"""
+    key, value = header
+    response = client.get("/get", auth=auth, headers={key: value})
+    assert response.status_code == 200
+
+
+def test_rejected_by_opa(client, auth):
+    """Tests a request that does not have the correct header for OPA policy"""
+    response = client.get("/get", auth=auth)
+    assert response.status_code == 403

testsuite/tests/kuadrant/authorino/conftest.py

@@ -3,6 +3,7 @@
 from weakget import weakget
 
 from testsuite.httpx.auth import HttpxOidcClientAuth
+from testsuite.openshift.objects.api_key import APIKey
 from testsuite.openshift.objects.auth_config import AuthConfig
 from testsuite.objects import Authorino, Authorization, PreexistingAuthorino
 from testsuite.openshift.objects.authorino import AuthorinoCR
@@ -51,3 +52,15 @@ def client(authorization, envoy):
     client = envoy.client()
     yield client
     client.close()
+
+
+@pytest.fixture(scope="module")
+def create_api_key(blame, request, openshift):
+    """Creates API key Secret"""
+    def _create_secret(name, label_selector, api_key):
+        secret_name = blame(name)
+        secret = APIKey.create_instance(openshift, secret_name, label_selector, api_key)
+        request.addfinalizer(secret.delete)
+        secret.commit()
+        return secret_name
+    return _create_secret

testsuite/tests/kuadrant/authorino/identity/api_key/conftest.py

@@ -2,19 +2,6 @@
 import pytest
 
 from testsuite.httpx.auth import HeaderApiKeyAuth
-from testsuite.openshift.objects.api_key import APIKey
-
-
-@pytest.fixture(scope="module")
-def create_api_key(blame, request, openshift):
-    """Creates API key Secret"""
-    def _create_secret(name, label_selector, api_key):
-        secret_name = blame(name)
-        secret = APIKey.create_instance(openshift, secret_name, label_selector, api_key)
-        request.addfinalizer(secret.delete)
-        secret.commit()
-        return secret_name
-    return _create_secret
 
 
 @pytest.fixture(scope="module")