initial but somewhat tested

Author: leifj

example/plugins/microservices/attribute_generation.yaml.example

@@ -0,0 +1,11 @@
+module: satosa.micro_services.attribute_generation.AddSyntheticAttributes
+name: AddSyntheticAttributes
+config:
+    synthetic_attributes:
+        target_provider1:
+            requester1:
+                eduPersonAffiliation: member;employee
+        default:
+            default:
+                schacHomeOrganization: {{eduPersonPrincipalName.scope}}
+                schacHomeOrganizationType: tomfoolery provider

setup.py

@@ -22,10 +22,11 @@
         "PyYAML",
         "gunicorn",
         "Werkzeug",
-        "click"
+        "click",
     ],
     extras_require={
         "ldap": ["ldap3"],
+        "pystache": ["pystache"]
     },
     zip_safe=False,
     classifiers=[

src/satosa/micro_services/attribute_generation.py

@@ -0,0 +1,78 @@
+import re
+import pystache
+
+from .base import ResponseMicroService
+
+def _config(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 MustachAttrValue(object):
+    def __init__(self,values):
+       self.values = values
+       if any(['@' in v for v in values]):
+          local_parts = []
+          domain_parts = []
+          scopes = dict()
+          for v in values:
+             (local_part, sep, domain_part) = v.partition('@')
+             # probably not needed now...
+             local_parts.append(local_part)
+             domain_parts.append(domain_part)
+             scopes[domain_part] = True
+          self._scopes = list(scopes.keys())
+       else:
+          self._scopes = None
+
+    def __str__(self):
+        return ";".join(self.values)
+   
+    @property 
+    def value(self):
+        if 1 == len(self.values): 
+           return self.values[0]
+        else:
+           return self.values
+
+    @property
+    def first(self):
+        if len(self.values) > 0:
+           return self.values[0]
+        else:
+           return ""
+
+    @property
+    def scope(self):
+        if self._scopes is not None:
+           return self._scopes[0]
+        return ""
+     
+
+class AddSyntheticAttributes(ResponseMicroService):
+    """
+    Add synthetic attributes to the responses.
+    """
+
+    def __init__(self, config, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.synthetic_attributes = config["synthetic_attributes"]
+
+    def _synthesize(self, attributes, requester, provider):
+        syn_attributes = dict()
+        context = dict()
+        
+        for attr_name,values in attributes.items():
+           context[attr_name] = MustachAttrValue(values)
+
+        recipes = _config(self.synthetic_attributes, requester, provider)
+        print(context)
+        for attr_name, fmt in recipes:
+           print(fmt)
+           syn_attributes[attr_name] = re.split("[;\n]+", pystache.render(fmt, context))
+        print(syn_attributes)
+        return syn_attributes
+
+    def process(self, context, data):
+        data.attributes.update(self._synthesize(data.attributes, data.requester, data.auth_info.issuer))
+        return super().process(context, data)

tests/satosa/micro_services/test_attribute_generation.py

@@ -0,0 +1,64 @@
+from satosa.internal_data import InternalResponse, AuthenticationInformation
+from satosa.micro_services.attribute_generation import AddSyntheticAttributes
+from satosa.exception import SATOSAAuthenticationError
+from satosa.context import Context
+
+class TestAddSyntheticAttributes:
+    def create_syn_service(self, synthetic_attributes):
+        authz_service = AddSyntheticAttributes(config=dict(synthetic_attributes=synthetic_attributes),
+                                               name="test_gen",
+                                               base_url="https://satosa.example.com")
+        authz_service.next = lambda ctx, data: data
+        return authz_service
+
+    def test_generate_static(self):
+        synthetic_attributes = {
+           "": { "default": {"a0": "value1;value2" }}
+        }
+        authz_service = self.create_syn_service(synthetic_attributes)
+        resp = InternalResponse(AuthenticationInformation(None, None, None))
+        resp.attributes = {
+            "a1": ["[email protected]"],
+        }
+        ctx = Context()
+        ctx.state = dict()
+        authz_service.process(ctx, resp)
+        assert("value1" in resp.attributes['a0'])
+        assert("value2" in resp.attributes['a0'])
+        assert("[email protected]" in resp.attributes['a1'])
+
+    def test_generate_mustache1(self):
+        synthetic_attributes = {
+           "": { "default": {"a0": "{{kaka}}#{{eppn.scope}}" }}
+        }
+        authz_service = self.create_syn_service(synthetic_attributes)
+        resp = InternalResponse(AuthenticationInformation(None, None, None))
+        resp.attributes = {
+            "kaka": ["kaka1"],
+            "eppn": ["[email protected]","[email protected]"]
+        }
+        ctx = Context()
+        ctx.state = dict()
+        authz_service.process(ctx, resp)
+        assert("kaka1#example.com" in resp.attributes['a0'])
+        assert("kaka1" in resp.attributes['kaka'])
+        assert("[email protected]" in resp.attributes['eppn'])
+        assert("[email protected]" in resp.attributes['eppn'])
+
+    def test_generate_mustache2(self):
+        synthetic_attributes = {
+           "": { "default": {"a0": "{{kaka.first}}#{{eppn.scope}}" }}
+        }
+        authz_service = self.create_syn_service(synthetic_attributes)
+        resp = InternalResponse(AuthenticationInformation(None, None, None))
+        resp.attributes = {
+            "kaka": ["kaka1","kaka2"],
+            "eppn": ["[email protected]","[email protected]"]
+        }
+        ctx = Context()
+        ctx.state = dict()
+        authz_service.process(ctx, resp)
+        assert("kaka1#example.com" in resp.attributes['a0'])
+        assert("kaka1" in resp.attributes['kaka'])
+        assert("[email protected]" in resp.attributes['eppn'])
+        assert("[email protected]" in resp.attributes['eppn'])