Merge pull request #130 from jsmolar/metadata

Added generic auth rule method for authconfig

testsuite/objects/__init__.py

@@ -1,5 +1,23 @@
 """Module containing base classes for common objects"""
 import abc
+from dataclasses import dataclass
+from typing import Literal
+
+
+@dataclass
+class Rule:
+    """
+    Data class for authorization rules represented by simple pattern-matching expressions.
+    Args:
+        :param selector: that is fetched from the Authorization JSON
+        :param operator: `eq` (equals), `neq` (not equal), `incl` (includes) and `excl` (excludes), for arrays
+                         `matches`, for regular expressions
+        :param value: a fixed comparable value
+    """
+
+    selector: str
+    operator: Literal["eq", "neq", "incl", "excl", "matches"]
+    value: str
 
 
 class LifecycleObject(abc.ABC):
@@ -72,6 +90,10 @@ def add_external_opa_policy(self, name, endpoint, ttl):
     def add_response(self, response):
         """Add response to AuthConfig"""
 
+    @abc.abstractmethod
+    def add_auth_rule(self, name: str, rule: Rule, when: Rule, metrics: bool, priority: int):
+        """Adds JSON pattern-matching authorization rule (authorization.json)"""
+
     @abc.abstractmethod
     def add_role_rule(self, name: str, role: str, path: str, metrics: bool, priority: int):
         """Adds a rule, which allows access to 'path' only to users with 'role'"""
@@ -84,6 +106,10 @@ def set_deny_with(self, code, value):
     def add_http_metadata(self, name, endpoint, method):
         """Set metadata http external auth feature"""
 
+    @abc.abstractmethod
+    def add_user_info_metadata(self, name, identity_source):
+        """Set metadata OIDC user info"""
+
 
 class PreexistingAuthorino(Authorino):
     """Authorino which is already deployed prior to the testrun"""

testsuite/openshift/objects/auth_config.py

@@ -2,7 +2,7 @@
 from dataclasses import dataclass, asdict
 from typing import Dict, Literal, List
 
-from testsuite.objects import Authorization
+from testsuite.objects import Authorization, Rule
 from testsuite.openshift.client import OpenShiftClient
 from testsuite.openshift.objects import OpenShiftObject, modify
 
@@ -119,33 +119,33 @@ def add_anonymous_identity(self, name):
         identities.append({"name": name, "anonymous": {}})
 
     @modify
-    def add_role_rule(self, name: str, role: str, path: str, metrics=False, priority=0):
-        """
-        Adds a rule, which allows access to 'path' only to users with 'role'
-        :param name: name of rule
-        :param role: name of role
-        :param path: path to apply this rule to
-        :param metrics: bool, allows metrics
-        :param priority: priority of rule
-        """
+    def add_auth_rule(self, name, rule: Rule, when: Rule = None, metrics=False, priority=0):
+        """Adds JSON pattern-matching authorization rule (authorization.json)"""
         authorization = self.model.spec.setdefault("authorization", [])
         authorization.append({
             "name": name,
             "metrics": metrics,
             "priority": priority,
             "json": {
-                "rules": [{
-                    "operator": "incl",
-                    "selector": "auth.identity.realm_access.roles",
-                    "value": role
-                }]
-            },
-            "when": [{
-                "operator": "matches",
-                "selector": "context.request.http.path",
-                "value": path
-            }]
+                "rules": [asdict(rule)]
+            }
         })
+        if when:
+            authorization[0].update({"when": [asdict(when)]})
+
+    def add_role_rule(self, name: str, role: str, path: str, metrics=False, priority=0):
+        """
+        Adds a rule, which allows access to 'path' only to users with 'role'
+        Args:
+            :param name: name of rule
+            :param role: name of role
+            :param path: path to apply this rule to
+            :param metrics: bool, allows metrics
+            :param priority: priority of rule
+        """
+        rule = Rule("auth.identity.realm_access.roles", "incl", role)
+        when = Rule("context.request.http.path", "matches", path)
+        self.add_auth_rule(name, rule, when, metrics, priority)
 
     @modify
     def remove_all_identities(self):
@@ -194,6 +194,7 @@ def set_deny_with(self, code, value):
 
     @modify
     def add_http_metadata(self, name, endpoint, method: Literal["GET", "POST"]):
+        """"Set metadata http external auth feature"""
         metadata = self.model.spec.setdefault("metadata", [])
         metadata.append({
             "name": name,
@@ -203,3 +204,14 @@ def add_http_metadata(self, name, endpoint, method: Literal["GET", "POST"]):
                 "headers": [{"name": "Accept", "value": "application/json"}]
             }
         })
+
+    @modify
+    def add_user_info_metadata(self, name, identity_source):
+        """Set metadata OIDC user info"""
+        metadata = self.model.spec.setdefault("metadata", [])
+        metadata.append({
+            "name": name,
+            "userInfo": {
+                "identitySource": identity_source
+            }
+        })

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

@@ -0,0 +1,32 @@
+"""
+Tests for external auth metadata. Online fetching of (OIDC) UserInfo data, associated with an OIDC identity source:
+https://github.com/Kuadrant/authorino/blob/main/docs/features.md#oidc-userinfo-metadatauserinfo
+"""
+import pytest
+
+from testsuite.openshift.objects.auth_config import Rule
+
+
+@pytest.fixture(scope="module")
+def authorization(authorization, rhsso):
+    """
+    Adds auth metadata OIDC UserInfo which fetches OIDC UserInfo in request-time.
+    Adds a simple rule that accepts only when fetched UserInfo contains the email address of the default RHSSO user.
+    """
+    user = rhsso.client.admin.get_user(rhsso.user)
+    authorization.add_user_info_metadata("user-info", "rhsso")
+    authorization.add_auth_rule("rule", Rule("auth.metadata.user-info.email", "eq", user["email"]))
+    return authorization
+
+
+def test_correct_auth(client, auth):
+    """Tests auth when UserInfo email matches the email address"""
+    response = client.get("get", auth=auth)
+    assert response.status_code == 200
+
+
+def test_incorrect_auth(client, auth, rhsso):
+    """Updates RHSSO user email address and tests incorrect auth"""
+    rhsso.client.admin.update_user(rhsso.user, {"email": "[email protected]"})
+    response = client.get("get", auth=auth)
+    assert response.status_code == 403