Merge pull request #85 from jsmolar/opa_external

Tests for OPA external registry

config/settings.local.yaml.tpl

@@ -19,6 +19,8 @@
 #    client_id: "CLIENT_ID"
 #    client_secret: "CLIENT_SECRET"
 #    url: "AUTH0_URL"
+#  mockserver:
+#    url: "MOCKSERVER_URL"
 #  cfssl: "cfssl"  # Path to the CFSSL library for TLS tests
 #  authorino:
 #    image: "quay.io/kuadrant/authorino:latest"  # If specified will override the authorino image

testsuite/config/__init__.py

@@ -32,6 +32,7 @@ def __init__(self, name, default, **kwargs) -> None:
         DefaultValueValidator("rhsso.url", must_exist=True, default=fetch_route("no-ssl-sso")),
         DefaultValueValidator("rhsso.password",
                               must_exist=True, default=fetch_secret("credential-sso", "ADMIN_PASSWORD")),
+        DefaultValueValidator("mockserver.url", must_exist=True, default=fetch_route("no-ssl-mockserver")),
     ],
     loaders=["testsuite.config.openshift_loader", "dynaconf.loaders.env_loader"]
 )

testsuite/mockserver.py

@@ -0,0 +1,38 @@
+"""Module for Mockserver integration"""
+from urllib.parse import urljoin
+
+import httpx
+
+
+class Mockserver:
+    """Mockserver deployed in Openshift (located in Tools or self-managed)"""
+
+    def __init__(self, url):
+        self.url = url
+
+    def create_expectation(self, expectation_id, path, opa_policy):
+        """Creates an Expectation - response with body that contains OPA policy (Rego query)"""
+        response = httpx.put(
+            urljoin(self.url, "/mockserver/expectation"), verify=False, timeout=5, json={
+                    "id": expectation_id,
+                    "httpRequest": {
+                        "path": path
+                    },
+                    "httpResponse": {
+                        "headers": {
+                            "Content-Type": ["plain/text"]
+                        },
+                        "body": opa_policy
+                    }
+                }
+            )
+        response.raise_for_status()
+        return response
+
+    def clear_expectation(self, expectation_id):
+        """Clears Expectation with specific ID"""
+        httpx.put(
+            urljoin(self.url, "/mockserver/clear"), verify=False, timeout=5, json={
+                    "id": expectation_id
+                }
+            ).raise_for_status()

testsuite/objects/__init__.py

@@ -60,6 +60,10 @@ def remove_all_hosts(self):
     def add_opa_policy(self, name, rego_policy):
         """Adds OPA inline Rego policy"""
 
+    @abc.abstractmethod
+    def add_external_opa_policy(self, name, endpoint, ttl):
+        """Adds OPA policy from external registry"""
+
     @abc.abstractmethod
     def add_response(self, response):
         """Add response to AuthConfig"""

testsuite/openshift/objects/auth_config.py

@@ -151,8 +151,24 @@ def add_opa_policy(self, name, rego_policy):
             }
         })
 
+    @modify
+    def add_external_opa_policy(self, name, endpoint, ttl=0):
+        """
+        Adds OPA policy that is declared as an HTTP endpoint
+        """
+        policy = self.model.spec.setdefault("authorization", [])
+        policy.append({
+            "name": name,
+            "opa": {
+                "externalRegistry": {
+                    "endpoint": endpoint,
+                    "ttl": ttl
+                }
+            }
+        })
+
     @modify
     def add_response(self, response):
-        """Adds response section to authconfig."""
+        """Adds response section to AuthConfig."""
         responses = self.model.spec.setdefault("response", [])
         responses.append(response)

testsuite/tests/kuadrant/authorino/authorization/opa/conftest.py

@@ -0,0 +1,23 @@
+"""Conftest for Open Policy Agent (OPA)"""
+import pytest
+
+from testsuite.mockserver import Mockserver
+from testsuite.utils import rego_allow_header
+
+
+@pytest.fixture(scope="module")
+def header():
+    """Header used by OPA policy"""
+    return "opa", "opa-test"
+
+
+@pytest.fixture(scope="module")
+def mockserver(request, testconfig, module_label, header):
+    """Returns mockserver and creates Expectation that returns Rego query"""
+    try:
+        mockserver = Mockserver(testconfig["mockserver"]["url"])
+        request.addfinalizer(lambda: mockserver.clear_expectation(module_label))
+        mockserver.create_expectation(module_label, "/opa", rego_allow_header(*header))
+        return mockserver
+    except KeyError as exc:
+        return pytest.skip(f"Mockserver configuration item is missing: {exc}")

testsuite/tests/kuadrant/authorino/authorization/opa/test_auto_refresh_policy.py

@@ -0,0 +1,47 @@
+"""
+Tests for Open Policy Agent (OPA) policy pulled from external registry.
+Registry is represented by Mockserver Expectation that returns Rego query.
+"""
+import time
+
+import pytest
+
+from testsuite.utils import rego_allow_header
+
+
+@pytest.fixture(scope="module")
+def updated_header():
+    """Header for updated OPA policy"""
+    return "updated", "updated-value"
+
+
+@pytest.fixture(scope="module", autouse=True)
+def update_external_opa(mockserver, module_label, updated_header):
+    """Updates Expectation with updated header"""
+    mockserver.create_expectation(module_label, "/opa", rego_allow_header(*updated_header))
+    # Sleeps for 1 second to compensate auto-refresh cycle `authorization.opa.externalRegistry.ttl = 1`
+    time.sleep(1)
+
+
+@pytest.fixture(scope="module")
+def authorization(authorization, mockserver):
+    """
+    Adds OPA policy. Rego query is located on external registry (Mockserver).
+    Policy accepts requests that contain `header`.
+    """
+    authorization.add_external_opa_policy("opa", mockserver.url + "/opa", 1)
+    return authorization
+
+
+def test_auto_refresh(client, auth, updated_header):
+    """Tests auto-refresh of OPA policy from external registry."""
+    key, value = updated_header
+    response = client.get("/get", auth=auth, headers={key: value})
+    assert response.status_code == 200
+
+
+def test_previous(client, auth, header):
+    """Tests invalidation of previous OPA policy"""
+    key, value = header
+    response = client.get("/get", auth=auth, headers={key: value})
+    assert response.status_code == 403

testsuite/tests/kuadrant/authorino/authorization/opa/test_external_registry.py

@@ -0,0 +1,26 @@
+"""Tests for Open Policy Agent (OPA) using Mockserver Expectations as http endpoint with Rego query"""
+
+import pytest
+
+
+@pytest.fixture(scope="module")
+def authorization(authorization, mockserver):
+    """
+    Adds OPA policy. Rego query is located on external registry (Mockserver).
+    Policy accepts requests that contain `header`.
+    """
+    authorization.add_external_opa_policy("opa", mockserver.url + "/opa")
+    return authorization
+
+
+def test_allowed_by_opa(client, auth, header):
+    """Tests a request that should be authorized by OPA external registry declaration"""
+    key, value = header
+    response = client.get("/get", auth=auth, headers={key: value})
+    assert response.status_code == 200
+
+
+def test_denied_by_opa(client, auth):
+    """Tests a request should be denied by OPA external registry declaration"""
+    response = client.get("/get", auth=auth)
+    assert response.status_code == 403

testsuite/tests/kuadrant/authorino/authorization/test_opa.py → testsuite/tests/kuadrant/authorino/authorization/opa/test_inline_rego.py

@@ -1,22 +1,13 @@
 """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"
+from testsuite.utils import rego_allow_header
 
 
 @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)
+    """Adds OPA policy that accepts all requests that contain `header`"""
+    authorization.add_opa_policy("opa", rego_allow_header(*header))
     return authorization
 
 

testsuite/utils.py

@@ -63,3 +63,8 @@ def cert_builder(cfssl: CFSSLClient, chain: dict, hosts: Union[str, Collection[s
             result.update(cert_builder(cfssl, info.children, parsed_hosts, cert))
         result[name] = cert
     return result
+
+
+def rego_allow_header(key, value):
+    """Rego query that allows all requests that contain specific header with`key` and `value`"""
+    return f"allow {{ input.context.request.http.headers.{key} == \"{value}\" }}"