Add logical operations on Patterns and Named patterns + new test

testsuite/policy/authorization/__init__.py

@@ -45,7 +45,7 @@ def asdict(self):
 
 
 @dataclass
-class Rule:
+class Pattern:
     """
     Data class for rules represented by simple pattern-matching expressions.
     Args:
@@ -60,6 +60,33 @@ class Rule:
     value: str
 
 
+@dataclass
+class AnyPattern:
+    """Dataclass specifying *OR* operation on patterns. Any one needs to pass for this block to pass."""
+
+    any: list["Rule"]
+
+
+@dataclass
+class AllPattern:
+    """Dataclass specifying *AND* operation on patterns. All need to pass for this block to pass."""
+
+    all: list["Rule"]
+
+
+@dataclass
+class PatternRef:
+    """
+    Dataclass that references other pattern-matching expression by name.
+    Use authorization.add_patterns() function to define named pattern-matching expression.
+    """
+
+    patternRef: str
+
+
+Rule = Pattern | AnyPattern | AllPattern | PatternRef
+
+
 @dataclass
 class ABCValue(abc.ABC):
     """
@@ -158,10 +185,3 @@ class Cache:
 
     ttl: int
     key: ABCValue
-
-
-@dataclass
-class PatternRef:
-    """Dataclass for specifying Pattern reference in Authorization"""
-
-    patternRef: str

testsuite/policy/authorization/auth_config.py

@@ -5,7 +5,8 @@
 from testsuite.utils import asdict
 from testsuite.openshift import OpenShiftObject, modify
 from testsuite.openshift.client import OpenShiftClient
-from .sections import AuthorizationSection, IdentitySection, MetadataSection, ResponseSection, Rule
+from .sections import AuthorizationSection, IdentitySection, MetadataSection, ResponseSection
+from . import Rule, Pattern
 
 
 class AuthConfig(OpenShiftObject):
@@ -75,3 +76,10 @@ def add_rule(self, when: list[Rule]):
         """Add rule for the skip of entire AuthConfig"""
         self.auth_section.setdefault("when", [])
         self.auth_section["when"].extend([asdict(x) for x in when])
+
+    @modify
+    def add_patterns(self, patterns: dict[str, list[Pattern]]):
+        """Add named pattern-matching expressions to be referenced in other "when" rules."""
+        self.model.spec.setdefault("patterns", {})
+        for key, value in patterns.items():
+            self.model.spec["patterns"].update({key: [asdict(x) for x in value]})

testsuite/policy/authorization/sections.py

@@ -5,6 +5,7 @@
     Selector,
     Credentials,
     Rule,
+    Pattern,
     ABCValue,
     ValueFrom,
     JsonResponse,
@@ -261,8 +262,8 @@ def add_role_rule(self, name: str, role: str, path: str, **common_features):
             :param role: name of role
             :param path: path to apply this rule to
         """
-        rule = Rule("auth.identity.realm_access.roles", "incl", role)
-        when = Rule("context.request.http.path", "matches", path)
+        rule = Pattern("auth.identity.realm_access.roles", "incl", role)
+        when = Pattern("context.request.http.path", "matches", path)
         common_features.setdefault("when", [])
         common_features["when"].append(when)
         self.add_auth_rules(name, [rule], **common_features)

testsuite/policy/rate_limit_policy.py

@@ -5,7 +5,7 @@
 
 import openshift as oc
 
-from testsuite.policy.authorization import Rule
+from testsuite.policy.authorization import Pattern
 from testsuite.utils import asdict
 from testsuite.gateway import Referencable
 from testsuite.openshift.client import OpenShiftClient
@@ -40,7 +40,7 @@ def create_instance(cls, openshift: OpenShiftClient, name, target: Referencable,
         return cls(model, context=openshift.context)
 
     @modify
-    def add_limit(self, name, limits: Iterable[Limit], when: Iterable[Rule] = None, counters: list[str] = None):
+    def add_limit(self, name, limits: Iterable[Limit], when: Iterable[Pattern] = None, counters: list[str] = None):
         """Add another limit"""
         limit: dict = {
             "rates": [asdict(limit) for limit in limits],

testsuite/tests/kuadrant/authorino/conditions/section_conditions/test_authorization_condition.py

@@ -1,13 +1,13 @@
 """Test condition to skip the authorization section of AuthConfig"""
 import pytest
 
-from testsuite.policy.authorization import Rule
+from testsuite.policy.authorization import Pattern
 
 
 @pytest.fixture(scope="module")
 def authorization(authorization):
     """Add to the AuthConfig authorization with opa policy that will always reject POST requests"""
-    when_post = [Rule("context.request.http.method", "eq", "POST")]
+    when_post = [Pattern("context.request.http.method", "eq", "POST")]
     authorization.authorization.add_opa_policy("opa", "allow { false }", when=when_post)
     return authorization
 

testsuite/tests/kuadrant/authorino/conditions/section_conditions/test_identity_condition.py

@@ -1,7 +1,7 @@
 """Test condition to skip the identity section of AuthConfig"""
 import pytest
 
-from testsuite.policy.authorization import Rule
+from testsuite.policy.authorization import Pattern
 from testsuite.httpx.auth import HeaderApiKeyAuth
 
 
@@ -20,7 +20,7 @@ def auth(api_key):
 @pytest.fixture(scope="module")
 def authorization(authorization, api_key):
     """Add to the AuthConfig API key identity, which can only be used on requests to the /get path"""
-    when_get = [Rule("context.request.http.path", "eq", "/get")]
+    when_get = [Pattern("context.request.http.path", "eq", "/get")]
     authorization.identity.add_api_key("api-key", selector=api_key.selector, when=when_get)
     return authorization
 

testsuite/tests/kuadrant/authorino/conditions/section_conditions/test_metadata_condition.py

@@ -1,7 +1,7 @@
 """Test condition to skip the metadata section of AuthConfig"""
 import pytest
 
-from testsuite.policy.authorization import Rule
+from testsuite.policy.authorization import Pattern
 
 
 @pytest.fixture(scope="module")
@@ -17,7 +17,7 @@ def authorization(authorization, mockserver_expectation):
     Add to the AuthConfig metadata evaluator with get http request to the mockserver,
     which will be only triggered on POST requests to the endpoint
     """
-    when_post = [Rule("context.request.http.method", "eq", "POST")]
+    when_post = [Pattern("context.request.http.method", "eq", "POST")]
     authorization.metadata.add_http("mock", mockserver_expectation, "GET", when=when_post)
     return authorization
 

testsuite/tests/kuadrant/authorino/conditions/section_conditions/test_response_condition.py

@@ -1,15 +1,15 @@
 """Test condition to skip the response section of AuthConfig"""
 import pytest
 
-from testsuite.policy.authorization import Rule, Value, JsonResponse
+from testsuite.policy.authorization import Pattern, Value, JsonResponse
 from testsuite.utils import extract_response
 
 
 @pytest.fixture(scope="module")
 def authorization(authorization):
     """Add to the AuthConfig response, which will only trigger on POST requests"""
     authorization.responses.add_success_header(
-        "simple", JsonResponse({"data": Value("response")}), when=[Rule("context.request.http.method", "eq", "POST")]
+        "simple", JsonResponse({"data": Value("response")}), when=[Pattern("context.request.http.method", "eq", "POST")]
     )
     return authorization
 

testsuite/tests/kuadrant/authorino/conditions/test_patternref_expressions.py

@@ -0,0 +1,78 @@
+"""Test patterns reference functionality and All/Any logical expressions."""
+import pytest
+
+from testsuite.policy.authorization import Pattern, PatternRef, AnyPattern, AllPattern
+
+
+@pytest.fixture(scope="module")
+def authorization(authorization):
+    """
+    Add multiple named patterns to AuthConfig to be referenced in later authorization rules.
+    Create authorization rule which:
+    1. For a GET requests allows only paths "/anything/dog" and "/anything/cat"
+    2. For a POST requests allows only paths "/anything/apple" and "/anything/pear"
+    3. For requests that contain header "x-special" it will get authorized regardless.
+    """
+    authorization.add_patterns(
+        {
+            "apple": [Pattern("context.request.http.path", "eq", "/anything/apple")],
+            "pear": [Pattern("context.request.http.path", "eq", "/anything/pear")],
+            "dog": [Pattern("context.request.http.path", "eq", "/anything/dog")],
+            "cat": [Pattern("context.request.http.path", "eq", "/anything/cat")],
+            "get": [Pattern("context.request.http.method", "eq", "GET")],
+            "post": [Pattern("context.request.http.method", "eq", "POST")],
+        }
+    )
+
+    authorization.authorization.add_auth_rules(
+        "auth_rules",
+        [
+            AnyPattern(
+                [
+                    AllPattern([AnyPattern([PatternRef("dog"), PatternRef("cat")]), PatternRef("get")]),
+                    AllPattern([AnyPattern([PatternRef("apple"), PatternRef("pear")]), PatternRef("post")]),
+                    Pattern("context.request.http.headers.@keys", "incl", "x-special"),
+                ]
+            )
+        ],
+    )
+
+    return authorization
+
+
+@pytest.mark.parametrize(
+    "path, expected_code",
+    [
+        ("/get", 403),
+        ("/anything/rock", 403),
+        ("/anything/apple", 403),
+        ("/anything/pear", 403),
+        ("/anything/dog", 200),
+        ("/anything/cat", 200),
+    ],
+)
+def test_get_rule(client, auth, path, expected_code):
+    """Test if doing GET request adheres to specified auth rule."""
+    assert client.get(path, auth=auth).status_code == expected_code
+
+
+@pytest.mark.parametrize(
+    "path, expected_code",
+    [
+        ("/post", 403),
+        ("/anything/rock", 403),
+        ("/anything/apple", 200),
+        ("/anything/pear", 200),
+        ("/anything/dog", 403),
+        ("/anything/cat", 403),
+    ],
+)
+def test_post_rule(client, auth, path, expected_code):
+    """Test if doing POST request adheres to specified auth rule."""
+    assert client.post(path, auth=auth).status_code == expected_code
+
+
+def test_special_header_rule(client, auth):
+    """Test if using the "x-special" header adheres to specified auth rule."""
+    assert client.get("/get", auth=auth, headers={"x-special": "value"}).status_code == 200
+    assert client.post("/post", auth=auth, headers={"x-special": "value"}).status_code == 200

testsuite/tests/kuadrant/authorino/conditions/test_top_level_condition.py

@@ -1,13 +1,13 @@
 """Test condition to skip the entire AuthConfig"""
 import pytest
 
-from testsuite.policy.authorization import Rule
+from testsuite.policy.authorization import Pattern
 
 
 @pytest.fixture(scope="module")
 def authorization(authorization, module_label):
     """Add rule to the AuthConfig to skip entire authn/authz with certain request header"""
-    authorization.add_rule([Rule("context.request.http.headers.key", "neq", module_label)])
+    authorization.add_rule([Pattern("context.request.http.headers.key", "neq", module_label)])
     return authorization
 
 

testsuite/tests/kuadrant/authorino/identity/extended_properties/test_token_normalization.py

@@ -1,6 +1,6 @@
 """https://github.com/Kuadrant/authorino/blob/main/docs/user-guides/token-normalization.md"""
 import pytest
-from testsuite.policy.authorization import Rule, Value, ValueFrom
+from testsuite.policy.authorization import Pattern, Value, ValueFrom
 from testsuite.httpx.auth import HeaderApiKeyAuth, HttpxOidcClientAuth
 
 
@@ -45,8 +45,8 @@ def authorization(authorization, rhsso, api_key):
         defaults_properties={"roles": Value(["admin"])},
     )
 
-    rule = Rule(selector="auth.identity.roles", operator="incl", value="admin")
-    when = Rule(selector="context.request.http.method", operator="eq", value="DELETE")
+    rule = Pattern(selector="auth.identity.roles", operator="incl", value="admin")
+    when = Pattern(selector="context.request.http.method", operator="eq", value="DELETE")
     authorization.authorization.add_auth_rules("only-admins-can-delete", rules=[rule], when=[when])
     return authorization
 

testsuite/tests/kuadrant/authorino/metadata/test_user_info.py

@@ -5,7 +5,7 @@
 import pytest
 
 from testsuite.httpx.auth import HttpxOidcClientAuth
-from testsuite.policy.authorization import Rule
+from testsuite.policy.authorization import Pattern
 
 
 @pytest.fixture(scope="module")
@@ -22,7 +22,7 @@ def authorization(authorization, rhsso):
     """
     authorization.metadata.add_user_info("user-info", "rhsso")
     authorization.authorization.add_auth_rules(
-        "rule", [Rule("auth.metadata.user-info.email", "eq", rhsso.user.properties["email"])]
+        "rule", [Pattern("auth.metadata.user-info.email", "eq", rhsso.user.properties["email"])]
     )
     return authorization
 

testsuite/tests/kuadrant/authorino/operator/tls/mtls/conftest.py

@@ -3,15 +3,15 @@
 
 from testsuite.certificates import CertInfo
 from testsuite.utils import cert_builder
-from testsuite.policy.authorization import Rule
+from testsuite.policy.authorization import Pattern
 
 
 @pytest.fixture(scope="module", autouse=True)
 def authorization(authorization, blame, selector, cert_attributes):
     """Create AuthConfig with mtls identity and pattern matching rule"""
     authorization.identity.add_mtls(blame("mtls"), selector=selector)
 
-    rule_organization = Rule("auth.identity.Organization", "incl", cert_attributes["O"])
+    rule_organization = Pattern("auth.identity.Organization", "incl", cert_attributes["O"])
     authorization.authorization.add_auth_rules(blame("redhat"), [rule_organization])
 
     return authorization

testsuite/tests/kuadrant/authorino/operator/tls/mtls/test_mtls_attributes.py

@@ -1,13 +1,13 @@
 """Tests on mTLS authentication with multiple attributes"""
 import pytest
 
-from testsuite.policy.authorization import Rule
+from testsuite.policy.authorization import Pattern
 
 
 @pytest.fixture(scope="module", autouse=True)
 def authorization(authorization, blame, cert_attributes):
     """Add second pattern matching rule to the AuthConfig"""
-    rule_country = Rule("auth.identity.Country", "incl", cert_attributes["C"])
+    rule_country = Pattern("auth.identity.Country", "incl", cert_attributes["C"])
     authorization.authorization.add_auth_rules(blame("redhat"), [rule_country])
 
     return authorization

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

@@ -8,7 +8,7 @@
 import openshift as oc
 from openshift import OpenShiftPythonException
 
-from testsuite.policy.authorization import Rule, ValueFrom
+from testsuite.policy.authorization import Pattern, ValueFrom
 from testsuite.certificates import CertInfo
 from testsuite.policy.authorization.auth_config import AuthConfig
 from testsuite.utils import cert_builder
@@ -82,8 +82,8 @@ def authorization(authorization, openshift, module_label, authorino_domain) -> A
     user_value = ValueFrom("auth.identity.username")
 
     when = [
-        Rule("auth.authorization.features.allow", "eq", "true"),
-        Rule("auth.authorization.features.verb", "eq", "CREATE"),
+        Pattern("auth.authorization.features.allow", "eq", "true"),
+        Pattern("auth.authorization.features.verb", "eq", "CREATE"),
     ]
     kube_attrs = {
         "namespace": {"value": openshift.project},
@@ -97,8 +97,8 @@ def authorization(authorization, openshift, module_label, authorino_domain) -> A
     )
 
     when = [
-        Rule("auth.authorization.features.allow", "eq", "true"),
-        Rule("auth.authorization.features.verb", "eq", "DELETE"),
+        Pattern("auth.authorization.features.allow", "eq", "true"),
+        Pattern("auth.authorization.features.verb", "eq", "DELETE"),
     ]
     kube_attrs = {
         "namespace": {"value": openshift.project},

testsuite/tests/kuadrant/authorino/response/test_deny_with.py

@@ -2,7 +2,7 @@
 from json import loads
 import pytest
 
-from testsuite.policy.authorization import Rule, Value, ValueFrom, DenyResponse
+from testsuite.policy.authorization import Pattern, Value, ValueFrom, DenyResponse
 
 HEADERS = {
     "x-string-header": Value("abc"),
@@ -35,7 +35,7 @@ def authorization(authorization):
         )
     )
     # Authorize only when url path is "/allow"
-    authorization.authorization.add_auth_rules("Whitelist", [Rule("context.request.http.path", "eq", "/allow")])
+    authorization.authorization.add_auth_rules("Whitelist", [Pattern("context.request.http.path", "eq", "/allow")])
     return authorization