Add test for kubernetes subjectAccessReview with resource attributes

testsuite/kuadrant/policy/authorization/__init__.py

@@ -4,7 +4,7 @@
 from dataclasses import dataclass
 from typing import Literal, Optional
 from testsuite.utils import asdict, JSONValues
-
+from testsuite.kuadrant.policy import CelExpression
 
 # pylint: disable=invalid-name
 
@@ -109,6 +109,17 @@ class PlainResponse:
     plain: ABCValue
 
 
+@dataclass(kw_only=True)
+class ResourceAttributes:
+    """Dataclass for specifying Resource Attributes in the KubernetesSubjectAccessReview authorization"""
+
+    namespace: Optional[Value | ValueFrom | CelExpression] = None
+    group: Optional[Value | ValueFrom | CelExpression] = None
+    resource: Optional[Value | ValueFrom | CelExpression] = None
+    name: Optional[Value | ValueFrom | CelExpression] = None
+    verb: Optional[Value | ValueFrom | CelExpression] = None
+
+
 @dataclass
 class WristbandSigningKeyRef:
     """Name of Kubernetes secret and corresponding signing algorithm."""

testsuite/kuadrant/policy/authorization/sections.py

@@ -13,6 +13,7 @@
     WristbandResponse,
     DenyResponse,
     Cache,
+    ResourceAttributes,
 )
 from testsuite.utils import asdict
 from testsuite.kubernetes import modify, Selector
@@ -287,7 +288,9 @@ def add_external_opa_policy(self, name, endpoint, ttl=0, **common_features):
         self.add_item(name, {"opa": {"externalPolicy": {"url": endpoint, "ttl": ttl}}}, **common_features)
 
     @modify
-    def add_kubernetes(self, name: str, user: ABCValue, resource_attributes: dict = None, **common_features):
+    def add_kubernetes(
+        self, name: str, user: ABCValue, resource_attributes: ResourceAttributes = None, **common_features
+    ):
         """Adds Kubernetes authorization
 
         :param name: name of kubernetes authorization
@@ -298,7 +301,10 @@ def add_kubernetes(self, name: str, user: ABCValue, resource_attributes: dict =
         self.add_item(
             name,
             {
-                "kubernetesSubjectAccessReview": {"user": asdict(user), "resourceAttributes": resource_attributes},
+                "kubernetesSubjectAccessReview": {
+                    "user": asdict(user),
+                    "resourceAttributes": asdict(resource_attributes) if resource_attributes else None,
+                },
             },
             **common_features,
         )

testsuite/kubernetes/cluster_role.py

@@ -1,9 +1,23 @@
 """ClusterRole and ClusterRoleBinding objects for Kubernetes"""
 
-from typing import Any
+from typing import Optional
+from dataclasses import dataclass
+
+from testsuite.utils import asdict
 from testsuite.kubernetes import KubernetesObject
 
 
+@dataclass(kw_only=True)
+class Rule:  # pylint: disable=invalid-name
+    """Dataclass for ClusterRole rule"""
+
+    verbs: list[str]
+    apiGroups: Optional[list[str]] = None
+    nonResourceURLs: Optional[list[str]] = None
+    resourceNames: Optional[list[str]] = None
+    resources: Optional[list[str]] = None
+
+
 class ClusterRole(KubernetesObject):
     """Kubernetes ClusterRole"""
 
@@ -12,7 +26,7 @@ def create_instance(
         cls,
         cluster,
         name,
-        rules: list[dict[str, Any]] = None,
+        rules: list[Rule] = None,
         labels: dict[str, str] = None,
     ):
         """Creates a new ClusterRole instance"""
@@ -23,7 +37,7 @@ def create_instance(
                 "name": name,
                 "labels": labels,
             },
-            "rules": rules,
+            "rules": [asdict(rule) for rule in rules] if rules else None,
         }
         return cls(model, context=cluster.context)
 

testsuite/tests/singlecluster/authorino/identity/subject_access_review/conftest.py

@@ -3,8 +3,7 @@
 import pytest
 
 from testsuite.httpx.auth import HeaderApiKeyAuth
-from testsuite.kuadrant.policy.authorization import ValueFrom
-from testsuite.kubernetes.cluster_role import ClusterRole, ClusterRoleBinding
+from testsuite.kubernetes.cluster_role import ClusterRoleBinding
 
 
 @pytest.fixture(scope="module")
@@ -17,8 +16,6 @@ def audience(hostname):
 def authorization(authorization):
     """Add kubernetes token-review and subject-access-review identity"""
     authorization.identity.add_kubernetes("token-review-host")
-    user = ValueFrom("auth.identity.user.username")
-    authorization.authorization.add_kubernetes("subject-access-review-host", user)
     return authorization
 
 
@@ -38,37 +35,27 @@ def _create_cluster_role_binding(cluster_role, service_accounts):
 
 
 @pytest.fixture(scope="module")
-def cluster_role(request, cluster, blame, module_label):
-    """Creates and returns a ClusterRole"""
-    rules = [{"nonResourceURLs": ["/get"], "verbs": ["get"]}]
-    cluster_role = ClusterRole.create_instance(cluster, blame("cr"), rules, labels={"app": module_label})
-    request.addfinalizer(cluster_role.delete)
-    cluster_role.commit()
-    return cluster_role
-
-
-@pytest.fixture(scope="module")
-def bound_service_account_token(cluster_role, create_service_account, create_cluster_role_binding, audience):
+def service_account_token(create_service_account, create_cluster_role_binding, cluster_role, audience):
     """Create a ServiceAccount, bind it to a ClusterRole and return its token with a given audience"""
-    service_account = create_service_account("tkn-auth")
-    create_cluster_role_binding(cluster_role.model.metadata.name, [service_account.model.metadata.name])
+    service_account = create_service_account("auth")
+    create_cluster_role_binding(cluster_role.name(), [service_account.name()])
     return service_account.get_auth_token(audience)
 
 
 @pytest.fixture(scope="module")
-def auth(bound_service_account_token):
+def auth(service_account_token):
     """Create request auth with service account token as API key"""
-    return HeaderApiKeyAuth(bound_service_account_token, "Bearer")
+    return HeaderApiKeyAuth(service_account_token, "Bearer")
 
 
 @pytest.fixture(scope="module")
-def service_account_token(create_service_account, audience):
+def service_account_token2(create_service_account, audience):
     """Create a non-authorized service account and request its bound token with the hostname as audience"""
-    service_account = create_service_account("tkn-non-auth")
+    service_account = create_service_account("no-auth")
     return service_account.get_auth_token(audience)
 
 
 @pytest.fixture(scope="module")
-def auth2(service_account_token):
+def auth2(service_account_token2):
     """Create request auth with service account token as API key"""
-    return HeaderApiKeyAuth(service_account_token, "Bearer")
+    return HeaderApiKeyAuth(service_account_token2, "Bearer")

testsuite/tests/singlecluster/authorino/identity/subject_access_review/test_non_resource_attributes.py

@@ -0,0 +1,38 @@
+"""Test kubernetes SubjectAccessReview non-resource attributes authorization by verifying only a
+ServiceAccount bound to a ClusterRole is authorized to access a resource"""
+
+import pytest
+
+from testsuite.kuadrant.policy.authorization import ValueFrom
+from testsuite.kubernetes.cluster_role import ClusterRole, Rule
+
+pytestmark = [pytest.mark.authorino]
+
+
+@pytest.fixture(scope="module")
+def authorization(authorization):
+    """Add kubernetes subject-access-review authorization with non-resource attributes (omit resource_attributes)"""
+    authorization.authorization.add_kubernetes(
+        "subject-access-review-username", ValueFrom("auth.identity.user.username")
+    )
+    return authorization
+
+
+@pytest.fixture(scope="module")
+def cluster_role(request, cluster, blame, module_label):
+    """Creates and returns a ClusterRole"""
+    rules = [Rule(verbs=["get"], nonResourceURLs=["/get"])]
+    cluster_role = ClusterRole.create_instance(cluster, blame("cr"), rules, labels={"app": module_label})
+    request.addfinalizer(cluster_role.delete)
+    cluster_role.commit()
+    return cluster_role
+
+
+def test_subject_access_review_non_resource_attributes(client, auth, auth2):
+    """Test Kubernetes SubjectAccessReview functionality by setting up authentication and authorization for an endpoint
+    and querying it with authorized and non-authorized ServiceAccount."""
+    response = client.get("/get", auth=auth)
+    assert response.status_code == 200
+
+    response = client.get("/get", auth=auth2)
+    assert response.status_code == 403

testsuite/tests/singlecluster/authorino/identity/subject_access_review/test_resource_attributes.py

@@ -0,0 +1,44 @@
+"""Test kubernetes SubjectAccessReview with resource attributes"""
+
+import pytest
+
+from testsuite.kuadrant.policy import CelExpression
+from testsuite.kuadrant.policy.authorization import ValueFrom, Value, ResourceAttributes
+from testsuite.kubernetes.cluster_role import ClusterRole, Rule
+
+pytestmark = [pytest.mark.authorino]
+
+
+@pytest.fixture(scope="module")
+def authorization(authorization):
+    """Add kubernetes subject-access-review identity with resource attributes for authpolicy resource"""
+    authorization.authorization.add_kubernetes(
+        "subject-access-review-host",
+        ValueFrom("auth.identity.user.username"),
+        ResourceAttributes(
+            resource=Value("authpolicy"), group=Value("kuadrant.io"), verb=CelExpression("request.method.lowerAscii()")
+        ),
+    )
+    return authorization
+
+
+@pytest.fixture(scope="module")
+def cluster_role(request, cluster, blame, module_label):
+    """Creates ClusterRole with rules only for accessing authpolicy resource"""
+    rules = [Rule(verbs=["get"], resources=["authpolicy"], apiGroups=["kuadrant.io"])]
+    cluster_role = ClusterRole.create_instance(cluster, blame("cr"), rules, labels={"app": module_label})
+    request.addfinalizer(cluster_role.delete)
+    cluster_role.commit()
+    return cluster_role
+
+
+def test_subject_access_review_resource_attributes(client, auth, auth2):
+    """Test if the client is authorized to access the api based on the service account token resource attributes"""
+    response = client.get("/get", auth=auth)
+    assert response.status_code == 200
+
+    response = client.post("/post", auth=auth)
+    assert response.status_code == 403
+
+    response = client.get("/get", auth=auth2)
+    assert response.status_code == 403

testsuite/tests/singlecluster/authorino/operator/tls/test_webhook.py

@@ -10,7 +10,7 @@
 
 from testsuite.certificates import CertInfo
 from testsuite.kubernetes.ingress import Ingress
-from testsuite.kuadrant.policy.authorization import ValueFrom, Pattern
+from testsuite.kuadrant.policy.authorization import ResourceAttributes, Value, ValueFrom, Pattern
 from testsuite.kuadrant.policy.authorization.auth_config import AuthConfig
 from testsuite.utils import cert_builder
 
@@ -86,30 +86,36 @@ def authorization(authorization, cluster, authorino_domain) -> AuthConfig:
         Pattern("auth.authorization.features.allow", "eq", "true"),
         Pattern("auth.authorization.features.verb", "eq", "CREATE"),
     ]
-    kube_attrs = {
-        "namespace": {"value": cluster.project},
-        "group": {"value": "networking.k8s.io"},
-        "resource": {"value": "Ingress"},
-        "verb": {"value": "create"},
-    }
     # add response for admission webhook for creating Ingress
     authorization.authorization.add_kubernetes(
-        "ingress-authn-k8s-binding-create", user_value, kube_attrs, when=when, priority=1
+        "ingress-authn-k8s-binding-create",
+        user_value,
+        ResourceAttributes(
+            namespace=Value(cluster.project),
+            group=Value("networking.k8s.io"),
+            resource=Value("Ingress"),
+            verb=Value("create"),
+        ),
+        when=when,
+        priority=1,
     )
 
     when = [
         Pattern("auth.authorization.features.allow", "eq", "true"),
         Pattern("auth.authorization.features.verb", "eq", "DELETE"),
     ]
-    kube_attrs = {
-        "namespace": {"value": cluster.project},
-        "group": {"value": "networking.k8s.io"},
-        "resource": {"value": "Ingress"},
-        "verb": {"value": "delete"},
-    }
     # add response for admission webhook for deleting Ingress
     authorization.authorization.add_kubernetes(
-        "ingress-authn-k8s-binding-delete", user_value, kube_attrs, when=when, priority=1
+        "ingress-authn-k8s-binding-delete",
+        user_value,
+        ResourceAttributes(
+            namespace=Value(cluster.project),
+            group=Value("networking.k8s.io"),
+            resource=Value("Ingress"),
+            verb=Value("delete"),
+        ),
+        when=when,
+        priority=1,
     )
     return authorization