Add kubernetes token-review identity tests

testsuite/openshift/client.py

@@ -57,6 +57,11 @@ def token(self):
         with self.context:
             return oc.whoami("-t")
 
+    def get_serviceaccount_auth_token(self, serviceaccount: str, audiences: list[str] = None) -> str:
+        """Returns bound token for given service account"""
+        token = self.do_action("create", "token", serviceaccount, *[f"--audience={a}" for a in audiences or []]).out()
+        return token.strip()
+
     @cached_property
     def apps_url(self):
         """Return URL under which all routes are routed"""

testsuite/policy/authorization/sections.py

@@ -115,13 +115,13 @@ def add_mtls(self, name: str, selector: Selector, **common_features):
         self.add_item(name, {"x509": {"selector": asdict(selector)}, **common_features})
 
     @modify
-    def add_kubernetes(self, name: str, audiences: list[str], **common_features):
+    def add_kubernetes(self, name: str, audiences: list[str] = None, **common_features):
         """Adds Kubernetes identity
         Args:
             :param name: name of the identity
             :param audiences: token audiences
         """
-        self.add_item(name, {"kubernetesTokenReview": {"audiences": audiences}}, **common_features)
+        self.add_item(name, {"kubernetesTokenReview": {"audiences": audiences} if audiences else {}}, **common_features)
 
     @modify
     def add_oidc(self, name, endpoint, *, ttl: int = 0, credentials: Credentials = None, **common_features):

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

@@ -0,0 +1,17 @@
+"""Conftest for kubernetes token-review tests"""
+
+import pytest
+
+
+@pytest.fixture(scope="module")
+def create_service_account(request, openshift, blame):
+    """Creates service account and returns its unique name"""
+
+    def _create_service_account(name):
+        sa_name = blame(name)
+        sa = openshift.do_action("create", "sa", sa_name, "-o", "json", "--dry-run=client", parse_output=True)
+        request.addfinalizer(lambda: sa.delete(ignore_not_found=True))
+        sa.create()
+        return sa_name
+
+    return _create_service_account

testsuite/tests/kuadrant/authorino/identity/token_review/test_audiences.py

@@ -0,0 +1,31 @@
+"""Test kubernetes token-review authorization with bound sa token that should contain all specified audiences"""
+
+import pytest
+
+
+pytestmark = [pytest.mark.authorino]
+
+TEST_AUDIENCES = ["test-aud1", "test-aud2", "test-aud3"]
+
+
+@pytest.fixture(scope="module")
+def authorization(authorization):
+    """Add kubernetes token-review identity with custom audiences specified"""
+    authorization.identity.add_kubernetes("token-review-aud", TEST_AUDIENCES)
+    return authorization
+
+
+@pytest.fixture(scope="module")
+def service_account_token(openshift, create_service_account):
+    """Create service account and request its bound token with the custom audiences"""
+    service_account = create_service_account("tkn-rev")
+    return openshift.get_serviceaccount_auth_token(service_account, TEST_AUDIENCES)
+
+
+def test_custom_audience(client, service_account_token):
+    """Test kubernetes token-review by adding custom audiences to the sa token and using it for the request"""
+    response = client.get("/get")
+    assert response.status_code == 401
+
+    response = client.get("/get", headers={"Authorization": "Bearer " + service_account_token})
+    assert response.status_code == 200

testsuite/tests/kuadrant/authorino/identity/token_review/test_host.py

@@ -0,0 +1,29 @@
+"""Test kubernetes token-review authorization with bound sa token that should contain host as audience by default"""
+
+import pytest
+
+
+pytestmark = [pytest.mark.authorino]
+
+
+@pytest.fixture(scope="module")
+def authorization(authorization):
+    """Add kubernetes token-review identity without any audiences"""
+    authorization.identity.add_kubernetes("token-review-host")
+    return authorization
+
+
+@pytest.fixture(scope="module")
+def service_account_token(openshift, hostname, create_service_account):
+    """Create service account and request its bound token with the hostname as audience"""
+    service_account = create_service_account("tkn-rev")
+    return openshift.get_serviceaccount_auth_token(service_account, [hostname.hostname])
+
+
+def test_host_audience(client, service_account_token):
+    """Test kubernetes token-review by adding hostname audience to the sa token and using it for the request"""
+    response = client.get("/get")
+    assert response.status_code == 401
+
+    response = client.get("/get", headers={"Authorization": "Bearer " + service_account_token})
+    assert response.status_code == 200