initial (somewhat tested) version

Author: leifj

src/satosa/micro_services/attribute_authorization.py

@@ -0,0 +1,35 @@
+import re
+
+from .base import ResponseMicroService
+from ..exception import SATOSAAuthenticationError
+
+def _filters(f, requester, provider):
+    pf = f.get(provider, f.get("", f.get("default", {})))
+    rf = pf.get(requester, pf.get("", pf.get("default", {})))
+    return rf.items()
+
+class AttributeAuthorization(ResponseMicroService):
+
+    def __init__(self, config, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.attribute_allow = config.get("attribute_allow", {})
+        self.attribute_deny = config.get("attribute_deny", {})
+
+    def _check_authz(self, context, attributes, requester, provider):
+        for attribute_name, attribute_filter in _filters(self.attribute_allow, requester, provider):
+            regex = re.compile(attribute_filter)
+            if attribute_name in attributes:
+                print(repr(regex))
+                print(list(filter(regex.search, attributes[attribute_name])))
+                if not list(filter(regex.search, attributes[attribute_name])):
+                    raise SATOSAAuthenticationError(context.state, "Permission denied")
+
+        for attribute_name, attribute_filter in _filters(self.attribute_deny, requester, provider):
+            regex = re.compile(attribute_filter)
+            if attribute_name in attributes:
+                if len(list(filter(regex.search, attributes[attribute_name]))) != len(attributes[attribute_name]):
+                    raise SATOSAAuthenticationError(context.state, "Permission denied")
+
+    def process(self, context, data):
+        self._check_authz(context, data.attributes, data.requester, data.auth_info.issuer)
+        return super().process(context, data)

tests/satosa/micro_services/test_attribute_authorization.py

@@ -0,0 +1,45 @@
+from satosa.internal_data import InternalResponse, AuthenticationInformation
+from satosa.micro_services.attribute_authorization import AttributeAuthorization
+from satosa.exception import SATOSAAuthenticationError
+from satosa.context import Context
+
+class TestAttributeAuthorization:
+    def create_authz_service(self, attribute_allow, attribute_deny):
+        authz_service = AttributeAuthorization(config=dict(attribute_allow=attribute_allow,attribute_deny=attribute_deny), name="test_authz",
+                                               base_url="https://satosa.example.com")
+        authz_service.next = lambda ctx, data: data
+        return authz_service
+
+    def test_authz_allow(self):
+        attribute_allow = {
+           "": { "default": {"a0": '.+@.+'} }
+        }
+        attribute_deny = {}
+        authz_service = self.create_authz_service(attribute_allow, attribute_deny)
+        resp = InternalResponse(AuthenticationInformation(None, None, None))
+        resp.attributes = {
+            "a0": ["[email protected]"],
+        }
+        try:
+           ctx = Context()
+           ctx.state = dict()
+           authz_service.process(ctx, resp)
+        except SATOSAAuthenticationError as ex:
+           assert False
+
+    def test_authz_not_allow(self):
+        attribute_allow = {
+           "": { "default": {"a0": 'foo'} }
+        }
+        attribute_deny = {}
+        authz_service = self.create_authz_service(attribute_allow, attribute_deny)
+        resp = InternalResponse(AuthenticationInformation(None, None, None))
+        resp.attributes = {
+            "a0": ["bar"],
+        }
+        try:
+           ctx = Context()
+           ctx.state = dict()
+           authz_service.process(ctx, resp)
+        except SATOSAAuthenticationError as ex:
+           assert True