Add ResponseProperties object and example refactor

testsuite/objects/__init__.py

@@ -1,10 +1,10 @@
 """Module containing base classes for common objects"""
 import abc
-from dataclasses import dataclass, is_dataclass, fields
+from dataclasses import dataclass, is_dataclass, fields, field
 from copy import deepcopy
-from typing import Literal, Union
+from typing import Literal, Optional
 
-JSONValues = Union[None, str, int, bool, list["JSONValues"], dict[str, "JSONValues"]]
+JSONValues = None | str | int | bool | list["JSONValues"] | dict[str, "JSONValues"]
 
 
 def asdict(obj) -> dict[str, JSONValues]:
@@ -76,8 +76,15 @@ class ABCValue(abc.ABC):
     """
     Abstract Dataclass for specifying a Value in Authorization,
     can be either static or reference to value in AuthJson.
+
+    Optional features:
+        - name: for use as JsonProperty
+        - overwrite: for use in ExtendedProperty
     """
 
+    name: Optional[str] = field(default=None, kw_only=True)
+    overwrite: Optional[bool] = field(default=None, kw_only=True)
+
 
 @dataclass
 class Value(ABCValue):
@@ -86,15 +93,21 @@ class Value(ABCValue):
     value: JSONValues
 
 
+@dataclass
+class AuthJSON:
+    authJSON: str  # pylint: disable=invalid-name
+
+
 @dataclass
 class ValueFrom(ABCValue):
     """Dataclass for dynamic Value. It contains reference path to existing value in AuthJson."""
 
-    authJSON: str  # pylint: disable=invalid-name
+    valueFrom: AuthJSON | str  # pylint: disable=invalid-name
 
-    def asdict(self):
-        """Override `asdict` function"""
-        return {"valueFrom": {"authJSON": self.authJSON}}
+    def __post_init__(self):
+        """Set valueFrom to be an instance of _AuthJson"""
+        if not isinstance(self.valueFrom, AuthJSON):
+            self.valueFrom = AuthJSON(self.valueFrom)
 
 
 @dataclass
@@ -105,6 +118,65 @@ class Cache:
     key: ABCValue
 
 
+@dataclass
+class Wristband:
+    """Dataclass for Wristband used in ResponseProperties"""
+
+    issuer: str
+    secret_name: str
+    algorithm: str = "RS256"
+
+    def asdict(self):
+        """Override `asdict` function"""
+        return {
+            "issuer": self.issuer,
+            "signingKeyRefs": [
+                {
+                    "name": self.secret_name,
+                    "algorithm": self.algorithm,
+                }
+            ],
+        }
+
+
+@dataclass
+class Properties:
+    properties: Optional[list[ABCValue]]
+
+    def __post_init__(self):
+        for prop in self.properties:
+            if prop.name is None:
+                raise AttributeError("Values in properties list must have `name` attribute.")
+
+
+@dataclass
+class ResponseProperties:
+    """
+    Dataclass for dynamic response definition.
+
+    Attributes:
+        :param name:        Name of the property and by default name of the header
+        :param json:        List of Value objects with name attribute
+        :param wrapper:     Optional `httpHeader` or `envoyDynamicMetadata` value
+        :param wrapperKey:  Optional header name
+        :param wristband:   Optional wristband functionality
+    """
+
+    name: str
+    json: Optional[Properties] | Optional[list[ABCValue]] = None
+
+    wrapper: Optional[Literal["httpHeader", "envoyDynamicMetadata"]] = None
+    wrapperKey: Optional[str] = None  # pylint: disable=invalid-name
+    wristband: Optional[Wristband] = None
+
+    def __post_init__(self):
+        """Check for correct usage. And transform `self.json` to an instance of _Property."""
+        if not (self.wristband is None) ^ (self.json is None):
+            raise AttributeError("Exactly one of the `properties` and `wristband` argument must be specified")
+        if not isinstance(self.json, Properties):
+            self.json = Properties(self.json)
+
+
 @dataclass
 class PatternRef:
     """Dataclass for specifying Pattern reference in Authorization"""

testsuite/openshift/objects/auth_config/sections.py

@@ -7,6 +7,8 @@
     Rule,
     Cache,
     ABCValue,
+    ValueFrom,
+    ResponseProperties,
 )
 from testsuite.openshift.objects import modify
 
@@ -59,6 +61,23 @@ def add_item(
 class Identities(Section):
     """Section which contains identity configuration"""
 
+    def add_item(
+        self,
+        name,
+        value,
+        priority: int = None,
+        when: Iterable[Rule] = None,
+        metrics: bool = None,
+        cache: Cache = None,
+        extended_properties: list[ABCValue] = None,
+    ):
+        """
+        Adds optional extendedProperties feature specific to IdentitySection and then calls parent add_idem() method
+        """
+        if extended_properties:
+            value["extendedProperties"] = [asdict(i) for i in extended_properties]
+        super().add_item(name, value, priority, when, metrics, cache)
+
     @modify
     def mtls(self, name: str, selector_key: str, selector_value: str, **common_features):
         """Adds mTLS identity
@@ -135,7 +154,7 @@ def anonymous(self, name, **common_features):
     @modify
     def plain(self, name, auth_json, **common_features):
         """Adds plain identity"""
-        self.add_item(name, {"plain": {"authJSON": auth_json}, **common_features})
+        self.add_item(name, {"plain": {"authJSON": auth_json}}, **common_features)
 
     @modify
     def remove_all(self):
@@ -176,18 +195,13 @@ class Responses(Section):
     """Section which contains response configuration"""
 
     def add_simple(self, auth_json, name="simple", key="data", **common_features):
-        """Adds simple response to AuthConfig"""
-        self.add(
-            {
-                "name": name,
-                "json": {"properties": [{"name": key, "valueFrom": {"authJSON": auth_json}}]},
-                **common_features,
-            }
-        )
+        self.add(asdict(ResponseProperties(name=name, json=[ValueFrom(auth_json, name=key)])), **common_features)
 
     @modify
     def add(self, response, **common_features):
         """Adds response section to AuthConfig."""
+        if isinstance(response, ResponseProperties):
+            response = asdict(response)
         self.add_item(response.pop("name"), response, **common_features)
 
 

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

@@ -0,0 +1,49 @@
+"""Basic tests for extended properties"""
+import pytest
+
+from testsuite.objects import Value, ValueFrom
+from testsuite.utils import extract_response
+
+
+@pytest.fixture(scope="module")
+def authorization(authorization, rhsso):
+    """
+    Add new identity with list of extended properties. This list contains:
+        - Static `value` and dynamic `jsonPath` properties
+        - Dynamic chaining properties which point to another extended property location before its created
+    Add simple response to inspect 'auth.identity' part of authJson where the properties will be created.
+    """
+    authorization.identity.oidc(
+        "rhsso",
+        rhsso.well_known["issuer"],
+        extended_properties=[
+            Value("static", name="property_static"),
+            # ValueFrom points to the request uri
+            ValueFrom("context.request.http.path", name="property_dynamic"),
+            ValueFrom("auth.identity.property_static", name="property_chain_static"),
+            ValueFrom("auth.identity.property_dynamic", name="property_chain_dynamic"),
+            ValueFrom("auth.identity.property_chain_self", name="property_chain_self", overwrite=True),
+        ],
+    )
+    authorization.responses.add_simple("auth.identity")
+    return authorization
+
+
+def test_basic(client, auth):
+    """
+    This test checks if static and dynamic extended properties are created and have the right value.
+    """
+    response = client.get("/anything/abc", auth=auth)
+    assert extract_response(response)["property_static"] % "MISSING" == "static"
+    assert extract_response(response)["property_dynamic"] % "MISSING" == "/anything/abc"
+
+
+def test_chain(client, auth):
+    """
+    This test checks if chaining extended properties have value None as chaining is not supported.
+    This behavior is undocumented but confirmed to be correct with dev team.
+    """
+    response = client.get("/anything/abc", auth=auth)
+    assert extract_response(response)["property_chain_static"] % "MISSING" is None
+    assert extract_response(response)["property_chain_dynamic"] % "MISSING" is None
+    assert extract_response(response)["property_chain_self"] % "MISSING" is None

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

@@ -0,0 +1,36 @@
+"""https://github.com/Kuadrant/authorino/pull/399"""
+import pytest
+
+from testsuite.objects import Value
+from testsuite.utils import extract_response
+
+
+@pytest.fixture(scope="module")
+def authorization(authorization):
+    """
+    Add plain authentication with three extended properties:
+    explicit False, explicit True and missing which should be default False.
+    Add simple response to expose `auth.identity` part of AuthJson
+    """
+    authorization.identity.plain(
+        "plain",
+        "context.request.http.headers.x-user|@fromstr",
+        extended_properties=[
+            Value("bar", name="name", overwrite=False),
+            Value(35, name="age", overwrite=True),
+            Value("admin", name="group"),
+        ],
+    )
+    authorization.responses.add_simple("auth.identity")
+
+    return authorization
+
+
+def test_overwrite(client):
+    """
+    Test the ExtendedProperty overwrite functionality overwriting the value in headers when True.
+    """
+    response = client.get("/get", headers={"x-user": '{"name":"foo","age":30,"group":"guest"}'})
+    assert extract_response(response)["name"] % "MISSING" == "foo"
+    assert extract_response(response)["age"] % 0 == 35
+    assert extract_response(response)["group"] % "MISSING" == "guest"

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

@@ -0,0 +1,59 @@
+"""https://github.com/Kuadrant/authorino/blob/main/docs/user-guides/token-normalization.md"""
+import pytest
+from testsuite.objects import Value, ValueFrom, Rule
+from testsuite.httpx.auth import HeaderApiKeyAuth, HttpxOidcClientAuth
+
+
+@pytest.fixture(scope="module")
+def auth_api_key(create_api_key, module_label):
+    """Creates API key Secret and returns auth for it."""
+    api_key = create_api_key("api-key", module_label, "api_key_value")
+    return HeaderApiKeyAuth(api_key)
+
+
+@pytest.fixture(scope="module")
+def auth_oidc_admin(rhsso, blame):
+    """Creates new user with new 'admin' role and return auth for it."""
+    realm_role = rhsso.realm.create_realm_role("admin")
+    user = rhsso.realm.create_user(blame("someuser"), blame("password"))
+    user.assign_realm_role(realm_role)
+    return HttpxOidcClientAuth.from_user(rhsso.get_token, user, "authorization")
+
+
+@pytest.fixture(scope="module")
+def authorization(authorization, rhsso, module_label):
+    """
+    Add rhsso identity provider with extended property "roles" which is dynamically mapped to
+    list of granted realm roles 'auth.identity.realm_access.roles'
+    Add api_key identity with extended property "roles" which is static list of one role 'admin'.
+
+    Add authorization rule allowing DELETE method only to users with role 'admin' in 'auth.identity.roles'
+    """
+    authorization.identity.oidc(
+        "rhsso",
+        rhsso.well_known["issuer"],
+        extended_properties=[ValueFrom("auth.identity.realm_access.roles", name="roles")],
+    )
+    authorization.identity.api_key(
+        "api_key", match_label=module_label, extended_properties=[Value(["admin"], name="roles")]
+    )
+
+    rule = Rule(selector="auth.identity.roles", operator="incl", value="admin")
+    when = Rule(selector="context.request.http.method", operator="eq", value="DELETE")
+    authorization.authorization.auth_rule("only-admins-can-delete", rule=rule, when=[when])
+    return authorization
+
+
+def test_token_normalization(client, auth, auth_oidc_admin, auth_api_key):
+    """
+    Tests token normalization scenario where three users with different types of authentication have "roles" value
+    normalized via extended_properties. Only user with an 'admin' role can use method DELETE.
+    - auth: oidc user without 'admin' role
+    - auth_oidc_admin: oidc user with 'admin' role
+    - auth_api_key: api key user which has static 'admin' role
+    """
+
+    assert client.get("/get", auth=auth).status_code == 200
+    assert client.delete("/delete", auth=auth).status_code == 403
+    assert client.delete("/delete", auth=auth_oidc_admin).status_code == 200
+    assert client.delete("/delete", auth=auth_api_key).status_code == 200

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

@@ -3,6 +3,7 @@
 import json
 
 import pytest
+from testsuite.objects import ResponseProperties, Value
 
 
 @pytest.fixture(scope="module")
@@ -27,7 +28,7 @@ def path_and_value(request):
 def responses(path_and_value):
     """Returns response to be added to the AuthConfig"""
     path, _ = path_and_value
-    return [{"name": "header", "json": {"properties": [{"name": "anything", "valueFrom": {"authJSON": path}}]}}]
+    return [ResponseProperties("header", json=[Value(path, name="anything")])]
 
 
 def test_auth_json_path(auth, client, path_and_value):

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

@@ -2,14 +2,15 @@
 import json
 
 import pytest
+from testsuite.objects import ResponseProperties, Value
 
 
 @pytest.fixture(scope="module")
 def responses():
     """Returns response to be added to the AuthConfig"""
     return [
-        {"name": "Header", "json": {"properties": [{"name": "anything", "value": "one"}]}},
-        {"name": "X-Test", "json": {"properties": [{"name": "anything", "value": "two"}]}},
+        ResponseProperties("Header", json=[Value("one", name="anything")]),
+        ResponseProperties("X-Test", json=[Value("two", name="anything")]),
     ]
 
 

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

@@ -2,12 +2,13 @@
 import json
 
 import pytest
+from testsuite.objects import ResponseProperties, Value
 
 
 @pytest.fixture(scope="module")
 def responses():
     """Returns response to be added to the AuthConfig"""
-    return [{"name": "header", "json": {"properties": [{"name": "anything", "value": "one"}]}}]
+    return [ResponseProperties(name="header", json=[Value("one", name="anything")])]
 
 
 def test_simple_response_with(auth, client):