support for attribute-based authorization

example/plugins/microservices/attribute_authz.yaml.example

@@ -0,0 +1,21 @@
+module: satosa.micro_services.attribute_authorization.AttributeAuthorization
+name: AttributeAuthorization
+config:
+    attribute_allow:
+        target_provider1:
+            requester1:
+                attr1:
+                   - "^foo:bar$"
+                   - "^kaka$"
+            default:
+                attr1:
+                   - "plupp@.+$"
+        "":
+            "":
+                attr2:
+                   - "^knytte:.*$"
+    attribute_deny:
+        default:
+            default:
+                eppn:
+                   - "^[^@]+$"

src/satosa/micro_services/attribute_authorization.py

@@ -0,0 +1,64 @@
+import re
+
+from .base import ResponseMicroService
+from ..exception import SATOSAAuthenticationError
+from ..util import get_dict_defaults
+
+class AttributeAuthorization(ResponseMicroService):
+
+    """
+A microservice that performs simple regexp-based authorization based on response
+attributes. The configuration assumes a dict with two keys: attributes_allow
+and attributes_deny. An examples speaks volumes:
+
+```yaml
+config:
+    attribute_allow:
+        target_provider1:
+            requester1:
+                attr1:
+                   - "^foo:bar$"
+                   - "^kaka$"
+            default:
+                attr1:
+                   - "plupp@.+$"
+        "":
+            "":
+                attr2:
+                   - "^knytte:.*$"
+    attribute_deny:
+        default:
+            default:
+                eppn:
+                   - "^[^@]+$"
+
+```
+
+The use of "" and 'default' is synonymous. Attribute rules are not overloaded
+or inherited. For instance a response from "provider2" would only be allowed
+through if the eppn attribute had all values containing an '@' (something
+perhaps best implemented via an allow rule in practice). Responses from
+target_provider1 bound for requester1 would be allowed through only if attr1
+contained foo:bar or kaka. Note that attribute filters (the leaves of the
+structure above) are ORed together - i.e any attribute match is sufficient.
+    """
+
+    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_filters in get_dict_defaults(self.attribute_allow, requester, provider).items():
+            if attribute_name in attributes:
+                if not any([any(filter(re.compile(af).search, attributes[attribute_name])) for af in attribute_filters]):
+                    raise SATOSAAuthenticationError(context.state, "Permission denied")
+
+        for attribute_name, attribute_filters in get_dict_defaults(self.attribute_deny, requester, provider).items():
+            if attribute_name in attributes:
+                if any([any(filter(re.compile(af).search, attributes[attribute_name])) for af in attribute_filters]):
+                    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,98 @@
+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_success(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": ["test@example.com"],
+        }
+        try:
+           ctx = Context()
+           ctx.state = dict()
+           authz_service.process(ctx, resp)
+        except SATOSAAuthenticationError as ex:
+           assert False
+
+    def test_authz_allow_fail(self):
+        attribute_allow = {
+           "": { "default": {"a0": ['foo1','foo2']} }
+        }
+        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)
+           assert False
+        except SATOSAAuthenticationError as ex:
+           assert True
+
+    def test_authz_allow_second(self):
+        attribute_allow = {
+           "": { "default": {"a0": ['foo1','foo2']} }
+        }
+        attribute_deny = {}
+        authz_service = self.create_authz_service(attribute_allow, attribute_deny)
+        resp = InternalResponse(AuthenticationInformation(None, None, None))
+        resp.attributes = {
+            "a0": ["foo2","kaka"],
+        }
+        try:
+           ctx = Context()
+           ctx.state = dict()
+           authz_service.process(ctx, resp)
+        except SATOSAAuthenticationError as ex:
+           assert False
+
+    def test_authz_deny_success(self):
+        attribute_deny = {
+           "": { "default": {"a0": ['foo1','foo2']} }
+        }
+        attribute_allow = {}
+        authz_service = self.create_authz_service(attribute_allow, attribute_deny)
+        resp = InternalResponse(AuthenticationInformation(None, None, None))
+        resp.attributes = {
+            "a0": ["foo2"],
+        }
+        try:
+           ctx = Context()
+           ctx.state = dict()
+           authz_service.process(ctx, resp)
+           assert False
+        except SATOSAAuthenticationError as ex:
+           assert True
+
+    def test_authz_deny_fail(self):
+        attribute_deny = {
+           "": { "default": {"a0": ['foo1','foo2']} }
+        }
+        attribute_allow = {}
+        authz_service = self.create_authz_service(attribute_allow, attribute_deny)
+        resp = InternalResponse(AuthenticationInformation(None, None, None))
+        resp.attributes = {
+            "a0": ["foo3"],
+        }
+        try:
+           ctx = Context()
+           ctx.state = dict()
+           authz_service.process(ctx, resp)
+        except SATOSAAuthenticationError as ex:
+           assert False