Refactor Policy to check the registration authority for restrictions

Signed-off-by: Ivan Kanakarakis <ivan.kanak@gmail.com>

2 files changed, 174 insertions, 132 deletions

diff --git a/tests/test_20_assertion.py b/tests/test_20_assertion.py
index dc501291..da7e70fd 100644
--- a/tests/test_20_assertion.py
+++ b/tests/test_20_assertion.py
@@ -218,7 +218,7 @@ def test_ava_filter_1():
            "surName": "Jeter",
            "mail": "derek@example.com"}
 
-    ava = r.filter(ava, "urn:mace:umu.se:saml:roland:sp", None, None)
+    ava = r.filter(ava, "urn:mace:umu.se:saml:roland:sp")
     assert _eq(list(ava.keys()), ["givenName", "surName"])
 
     ava = {"givenName": "Derek",
@@ -247,8 +247,7 @@ def test_ava_filter_2():
     ava = {"givenName": "Derek", "sn": "Jeter", "mail": "derek@example.com"}
 
     # mail removed because it doesn't match the regular expression
-    _ava = policy.filter(ava, 'urn:mace:umu.se:saml:roland:sp', None, [mail],
-                         [gn, sn])
+    _ava = policy.filter(ava, 'urn:mace:umu.se:saml:roland:sp', required=[mail], optional=[gn, sn])
 
     assert _eq(sorted(list(_ava.keys())), ["givenName", 'sn'])
 
@@ -256,8 +255,7 @@ def test_ava_filter_2():
 
     # it wasn't there to begin with
     try:
-        policy.filter(ava, 'urn:mace:umu.se:saml:roland:sp', None,
-                      [gn, sn, mail])
+        policy.filter(ava, 'urn:mace:umu.se:saml:roland:sp', required=[gn, sn, mail])
     except MissingValue:
         pass
 
@@ -287,8 +285,7 @@ def test_ava_filter_dont_fail():
 
     # mail removed because it doesn't match the regular expression
     # So it should fail if the 'fail_on_ ...' flag wasn't set
-    _ava = policy.filter(ava, 'urn:mace:umu.se:saml:roland:sp', None,
-                         [mail], [gn, sn])
+    _ava = policy.filter(ava, 'urn:mace:umu.se:saml:roland:sp', required=[mail], optional=[gn, sn])
 
     assert _ava
 
@@ -296,8 +293,7 @@ def test_ava_filter_dont_fail():
            "surName": "Jeter"}
 
     # it wasn't there to begin with
-    _ava = policy.filter(ava, 'urn:mace:umu.se:saml:roland:sp',
-                         None, [gn, sn, mail])
+    _ava = policy.filter(ava, 'urn:mace:umu.se:saml:roland:sp', required=[gn, sn, mail])
 
     assert _ava
 
@@ -633,7 +629,7 @@ def test_filter_ava_0():
            "mail": ["derek@nyy.mlb.com"]}
 
     # No restrictions apply
-    ava = policy.filter(ava, "urn:mace:example.com:saml:roland:sp", [], [])
+    ava = policy.filter(ava, "urn:mace:example.com:saml:roland:sp")
 
     assert _eq(sorted(list(ava.keys())), ["givenName", "mail", "surName"])
     assert ava["givenName"] == ["Derek"]
@@ -660,7 +656,7 @@ def test_filter_ava_1():
            "mail": ["derek@nyy.mlb.com"]}
 
     # No restrictions apply
-    ava = policy.filter(ava, "urn:mace:example.com:saml:roland:sp", [], [])
+    ava = policy.filter(ava, "urn:mace:example.com:saml:roland:sp")
 
     assert _eq(sorted(list(ava.keys())), ["givenName", "surName"])
     assert ava["givenName"] == ["Derek"]
@@ -685,7 +681,7 @@ def test_filter_ava_2():
            "mail": ["derek@nyy.mlb.com"]}
 
     # No restrictions apply
-    ava = policy.filter(ava, "urn:mace:example.com:saml:roland:sp", [], [])
+    ava = policy.filter(ava, "urn:mace:example.com:saml:roland:sp")
 
     assert _eq(list(ava.keys()), ["mail"])
     assert ava["mail"] == ["derek@nyy.mlb.com"]
@@ -709,7 +705,7 @@ def test_filter_ava_3():
            "mail": ["derek@nyy.mlb.com", "dj@example.com"]}
 
     # No restrictions apply
-    ava = policy.filter(ava, "urn:mace:example.com:saml:roland:sp", [], [])
+    ava = policy.filter(ava, "urn:mace:example.com:saml:roland:sp")
 
     assert _eq(list(ava.keys()), ["mail"])
     assert ava["mail"] == ["dj@example.com"]
@@ -733,7 +729,7 @@ def test_filter_ava_4():
            "mail": ["derek@nyy.mlb.com", "dj@example.com"]}
 
     # No restrictions apply
-    ava = policy.filter(ava, "urn:mace:example.com:saml:curt:sp", [], [])
+    ava = policy.filter(ava, "urn:mace:example.com:saml:curt:sp")
 
     assert _eq(sorted(list(ava.keys())), ['mail', 'givenName', 'surName'])
     assert _eq(ava["mail"], ["derek@nyy.mlb.com", "dj@example.com"])
@@ -772,7 +768,7 @@ def test_req_opt():
            'uid': 'rohe0002', 'edupersonaffiliation': 'staff'}
 
     sp_entity_id = "urn:mace:example.com:saml:curt:sp"
-    fava = policy.filter(ava, sp_entity_id, None, req, opt)
+    fava = policy.filter(ava, sp_entity_id, required=req, optional=opt)
     assert fava
 
 
@@ -872,22 +868,27 @@ def test_assertion_with_noop_attribute_conv():
 
 
 def test_filter_ava_5():
-    mds = MetadataStore(ATTRCONV, sec_config,
-                        disable_ssl_certificate_validation=True)
+    mds = MetadataStore(ATTRCONV, sec_config, disable_ssl_certificate_validation=True)
     mds.imp(METADATACONF["1"])
 
-    policy = Policy({
+    policy_conf = {
         "default": {
             "lifetime": {"minutes": 15},
             "attribute_restrictions": None,  # means all I have
             "entity_categories": ["swamid", "edugain"]
         }
-    })
-
-    ava = {"givenName": ["Derek"], "surName": ["Jeter"],
-           "mail": ["derek@nyy.mlb.com", "dj@example.com"]}
-
-    ava = policy.filter(ava, "urn:mace:example.com:saml:curt:sp", mdstore=mds, required=[], optional=[])
+    }
+    policy = Policy(restrictions=policy_conf, mds=mds)
+
+    ava = {
+        "givenName": ["Derek"],
+        "surName": ["Jeter"],
+        "mail": [
+            "derek@nyy.mlb.com",
+            "dj@example.com",
+        ],
+    }
+    ava = policy.filter(ava, "urn:mace:example.com:saml:curt:sp")
 
     # using entity_categories means there *always* are restrictions
     # in this case the only allowed attribute is eduPersonTargetedID
@@ -896,37 +897,40 @@ def test_filter_ava_5():
 
 
 def test_filter_ava_registration_authority_1():
-    mds = MetadataStore(ATTRCONV, sec_config,
-                        disable_ssl_certificate_validation=True)
+    mds = MetadataStore(ATTRCONV, sec_config, disable_ssl_certificate_validation=True)
     mds.imp(METADATACONF["1"])
-    config.metadata = mds
 
-    policy = Policy({
+    policy_conf = {
         "default": {
             "lifetime": {"minutes": 15},
             "attribute_restrictions": None,
         },
-        "registration_authorities": {
-            "http://rr.aai.switch.ch/": {
-                "attribute_restrictions": {
-                    "givenName": None,
-                    "surName": None,
-                }
+        "http://rr.aai.switch.ch/": {
+            "attribute_restrictions": {
+                "givenName": None,
+                "surName": None,
             }
         }
-    }, config=config)
-
-    attributes = {"givenName": ["Derek"], "surName": ["Jeter"],
-                  "mail": ["derek@nyy.mlb.com", "dj@example.com"]}
+    }
+    policy = Policy(restrictions=policy_conf, mds=mds)
+
+    attributes = {
+        "givenName": ["Derek"],
+        "surName": ["Jeter"],
+        "mail": [
+            "derek@nyy.mlb.com",
+            "dj@example.com",
+        ],
+    }
 
     # SP registered with http://rr.aai.switch.ch/
-    ava = policy.filter(attributes, "https://aai-idp.unibe.ch/idp/shibboleth", mdstore=mds, required=[], optional=[])
+    ava = policy.filter(attributes, "https://aai-idp.unibe.ch/idp/shibboleth")
     assert _eq(sorted(list(ava.keys())), ["givenName", "surName"])
     assert ava["givenName"] == ["Derek"]
     assert ava["surName"] == ["Jeter"]
 
     # SP not registered with http://rr.aai.switch.ch/
-    ava = policy.filter(attributes, "https://alpha.kib.ki.se/shibboleth", mdstore=mds, required=[], optional=[])
+    ava = policy.filter(attributes, "https://alpha.kib.ki.se/shibboleth")
     assert _eq(sorted(list(ava.keys())), ["givenName", "mail", "surName"])
     assert ava["givenName"] == ["Derek"]
     assert ava["surName"] == ["Jeter"]
@@ -936,13 +940,16 @@ def test_filter_ava_registration_authority_1():
 def test_assertion_with_zero_attributes():
     ava = {}
     ast = Assertion(ava)
-    policy = Policy({
+
+    policy_conf = {
         "default": {
             "lifetime": {"minutes": 240},
             "attribute_restrictions": None,  # means all I have
             "name_form": NAME_FORMAT_URI
         },
-    })
+    }
+    policy = Policy(policy_conf)
+
     name_id = NameID(format=NAMEID_FORMAT_TRANSIENT, text="foobar")
     issuer = Issuer(text="entityid", format=NAMEID_FORMAT_ENTITY)
     farg = add_path(
diff --git a/tests/test_37_entity_categories.py b/tests/test_37_entity_categories.py
index 839030fd..a24a4feb 100644
--- a/tests/test_37_entity_categories.py
+++ b/tests/test_37_entity_categories.py
@@ -25,37 +25,47 @@ def _eq(l1, l2):
 
 
 def test_filter_ava():
-    policy = Policy({
+    policy_conf = {
         "default": {
             "lifetime": {"minutes": 15},
             # "attribute_restrictions": None  # means all I have
             "entity_categories": ["swamid"]
         }
-    })
+    }
+    policy = Policy(policy_conf, MDS)
 
-    ava = {"givenName": ["Derek"], "sn": ["Jeter"],
-           "mail": ["derek@nyy.mlb.com", "dj@example.com"], "c": ["USA"]}
+    ava = {
+        "givenName": ["Derek"],
+        "sn": ["Jeter"],
+        "mail": ["derek@nyy.mlb.com", "dj@example.com"],
+        "c": ["USA"]
+    }
 
-    ava = policy.filter(ava, "https://connect.sunet.se/shibboleth", MDS)
+    ava = policy.filter(ava, "https://connect.sunet.se/shibboleth")
 
     assert _eq(list(ava.keys()), ['mail', 'givenName', 'sn', 'c'])
     assert _eq(ava["mail"], ["derek@nyy.mlb.com", "dj@example.com"])
 
 
 def test_filter_ava2():
-    policy = Policy({
+    policy_conf = {
         "default": {
             "lifetime": {"minutes": 15},
             # "attribute_restrictions": None  # means all I have
             "entity_categories": ["refeds", "edugain"]
         }
-    })
+    }
+    policy = Policy(policy_conf, MDS)
 
-    ava = {"givenName": ["Derek"], "sn": ["Jeter"],
-           "mail": ["derek@nyy.mlb.com"], "c": ["USA"],
-           "eduPersonTargetedID": "foo!bar!xyz"}
+    ava = {
+        "givenName": ["Derek"],
+        "sn": ["Jeter"],
+        "mail": ["derek@nyy.mlb.com"],
+        "c": ["USA"],
+        "eduPersonTargetedID": "foo!bar!xyz"
+    }
 
-    ava = policy.filter(ava, "https://connect.sunet.se/shibboleth", MDS)
+    ava = policy.filter(ava, "https://connect.sunet.se/shibboleth")
 
     # Mismatch, policy deals with eduGAIN, metadata says SWAMID
     # So only minimum should come out
@@ -63,96 +73,120 @@ def test_filter_ava2():
 
 
 def test_filter_ava3():
-    policy = Policy({
+    mds = MetadataStore(ATTRCONV, sec_config, disable_ssl_certificate_validation=True)
+    mds.imp(
+        [
+            {
+                "class": "saml2.mdstore.MetaDataFile",
+                "metadata": [(full_path("entity_cat_sfs_hei.xml"),)]
+            }
+        ]
+    )
+
+    policy_conf = {
         "default": {
             "lifetime": {"minutes": 15},
             # "attribute_restrictions": None  # means all I have
             "entity_categories": ["swamid"]
         }
-    })
+    }
+    policy = Policy(policy_conf, mds)
+
+    ava = {
+        "givenName": ["Derek"],
+        "sn": ["Jeter"],
+        "mail": ["derek@nyy.mlb.com"],
+        "c": ["USA"],
+        "eduPersonTargetedID": "foo!bar!xyz",
+        "norEduPersonNIN": "19800101134"
+    }
+
+    ava = policy.filter(ava, "urn:mace:example.com:saml:roland:sp")
+    assert _eq(list(ava.keys()), ['eduPersonTargetedID', "norEduPersonNIN"])
 
+
+def test_filter_ava4():
     mds = MetadataStore(ATTRCONV, sec_config,
                         disable_ssl_certificate_validation=True)
     mds.imp([{"class": "saml2.mdstore.MetaDataFile",
-              "metadata": [(full_path("entity_cat_sfs_hei.xml"),)]}])
-
-    ava = {"givenName": ["Derek"], "sn": ["Jeter"],
-           "mail": ["derek@nyy.mlb.com"], "c": ["USA"],
-           "eduPersonTargetedID": "foo!bar!xyz",
-           "norEduPersonNIN": "19800101134"}
-
-    ava = policy.filter(ava, "urn:mace:example.com:saml:roland:sp", mds)
-
-    assert _eq(list(ava.keys()), ['eduPersonTargetedID', "norEduPersonNIN"])
-
+              "metadata": [(full_path("entity_cat_re_nren.xml"),)]}])
 
-def test_filter_ava4():
-    policy = Policy({
+    policy_conf = {
         "default": {
             "lifetime": {"minutes": 15},
             # "attribute_restrictions": None  # means all I have
             "entity_categories": ["swamid"]
         }
-    })
-
-    mds = MetadataStore(ATTRCONV, sec_config,
-                        disable_ssl_certificate_validation=True)
-    mds.imp([{"class": "saml2.mdstore.MetaDataFile",
-              "metadata": [(full_path("entity_cat_re_nren.xml"),)]}])
+    }
+    policy = Policy(policy_conf, mds)
 
-    ava = {"givenName": ["Derek"], "sn": ["Jeter"],
-           "mail": ["derek@nyy.mlb.com"], "c": ["USA"],
-           "eduPersonTargetedID": "foo!bar!xyz",
-           "norEduPersonNIN": "19800101134"}
+    ava = {
+        "givenName": ["Derek"],
+        "sn": ["Jeter"],
+        "mail": ["derek@nyy.mlb.com"],
+        "c": ["USA"],
+        "eduPersonTargetedID": "foo!bar!xyz",
+        "norEduPersonNIN": "19800101134"
+    }
 
-    ava = policy.filter(ava, "urn:mace:example.com:saml:roland:sp", mds)
-
-    assert _eq(list(ava.keys()),
-               ['eduPersonTargetedID', "givenName", "c", "mail",
-                "sn"])
+    ava = policy.filter(ava, "urn:mace:example.com:saml:roland:sp")
+    assert _eq(
+        list(ava.keys()), ['eduPersonTargetedID', "givenName", "c", "mail", "sn"]
+    )
 
 
 def test_filter_ava5():
+    mds = MetadataStore(ATTRCONV, sec_config,
+                        disable_ssl_certificate_validation=True)
+    mds.imp([{"class": "saml2.mdstore.MetaDataFile",
+              "metadata": [(full_path("entity_cat_re.xml"),)]}])
+
     policy = Policy({
         "default": {
             "lifetime": {"minutes": 15},
             # "attribute_restrictions": None  # means all I have
             "entity_categories": ["swamid"]
         }
-    })
+    }, mds)
 
-    mds = MetadataStore(ATTRCONV, sec_config,
-                        disable_ssl_certificate_validation=True)
-    mds.imp([{"class": "saml2.mdstore.MetaDataFile",
-              "metadata": [(full_path("entity_cat_re.xml"),)]}])
-
-    ava = {"givenName": ["Derek"], "sn": ["Jeter"],
-           "mail": ["derek@nyy.mlb.com"], "c": ["USA"],
-           "eduPersonTargetedID": "foo!bar!xyz",
-           "norEduPersonNIN": "19800101134"}
+    ava = {
+        "givenName": ["Derek"],
+        "sn": ["Jeter"],
+        "mail": ["derek@nyy.mlb.com"],
+        "c": ["USA"],
+        "eduPersonTargetedID": "foo!bar!xyz",
+        "norEduPersonNIN": "19800101134"
+    }
 
-    ava = policy.filter(ava, "urn:mace:example.com:saml:roland:sp", mds)
+    ava = policy.filter(ava, "urn:mace:example.com:saml:roland:sp")
 
     assert _eq(list(ava.keys()), ['eduPersonTargetedID'])
 
 
 def test_idp_policy_filter():
     with closing(Server("idp_conf_ec")) as idp:
-        ava = {"givenName": ["Derek"], "sn": ["Jeter"],
-               "mail": ["derek@nyy.mlb.com"], "c": ["USA"],
-               "eduPersonTargetedID": "foo!bar!xyz",
-               "norEduPersonNIN": "19800101134"}
+        ava = {
+            "givenName": ["Derek"],
+            "sn": ["Jeter"],
+            "mail": ["derek@nyy.mlb.com"],
+            "c": ["USA"],
+            "eduPersonTargetedID": "foo!bar!xyz",
+            "norEduPersonNIN": "19800101134"
+        }
 
         policy = idp.config.getattr("policy", "idp")
-        ava = policy.filter(ava, "urn:mace:example.com:saml:roland:sp",
-                            idp.metadata)
-
-        print(ava)
-        assert list(ava.keys()) == [
-            "eduPersonTargetedID"]  # because no entity category
+        ava = policy.filter(ava, "urn:mace:example.com:saml:roland:sp")
+        # because no entity category
+        assert list(ava.keys()) == ["eduPersonTargetedID"]
 
 
 def test_entity_category_import_from_path():
+    mds = MetadataStore(ATTRCONV, sec_config, disable_ssl_certificate_validation=True)
+    # The file entity_cat_rs.xml contains the SAML metadata for an SP
+    # tagged with the REFEDs R&S entity category.
+    mds.imp([{"class": "saml2.mdstore.MetaDataFile",
+              "metadata": [(full_path("entity_cat_rs.xml"),)]}])
+
     # The entity category module myentitycategory.py is in the tests
     # directory which is on the standard module search path.
     # The module uses a custom interpretation of the REFEDs R&S entity category
@@ -162,34 +196,35 @@ def test_entity_category_import_from_path():
             "lifetime": {"minutes": 15},
             "entity_categories": ["myentitycategory"]
         }
-    })
-
-    mds = MetadataStore(ATTRCONV, sec_config,
-                        disable_ssl_certificate_validation=True)
-
-    # The file entity_cat_rs.xml contains the SAML metadata for an SP
-    # tagged with the REFEDs R&S entity category.
-    mds.imp([{"class": "saml2.mdstore.MetaDataFile",
-              "metadata": [(full_path("entity_cat_rs.xml"),)]}])
-
-    ava = {"givenName": ["Derek"], "sn": ["Jeter"],
-           "displayName": "Derek Jeter",
-           "mail": ["derek@nyy.mlb.com"], "c": ["USA"],
-           "eduPersonTargetedID": "foo!bar!xyz",
-           "eduPersonUniqueId": "R13ET7UD68K0HGR153KE@my.org",
-           "eduPersonScopedAffiliation": "member@my.org",
-           "eduPersonPrincipalName": "user01@my.org",
-           "norEduPersonNIN": "19800101134"}
-
-    ava = policy.filter(ava, "urn:mace:example.com:saml:roland:sp", mds)
+    }, mds)
+
+    ava = {
+        "givenName": ["Derek"],
+        "sn": ["Jeter"],
+        "displayName": "Derek Jeter",
+        "mail": ["derek@nyy.mlb.com"],
+        "c": ["USA"],
+        "eduPersonTargetedID": "foo!bar!xyz",
+        "eduPersonUniqueId": "R13ET7UD68K0HGR153KE@my.org",
+        "eduPersonScopedAffiliation": "member@my.org",
+        "eduPersonPrincipalName": "user01@my.org",
+        "norEduPersonNIN": "19800101134"
+    }
+
+    ava = policy.filter(ava, "urn:mace:example.com:saml:roland:sp")
 
     # We expect c and norEduPersonNIN to be filtered out since they are not
     # part of the custom entity category.
-    assert _eq(list(ava.keys()),
-               ["eduPersonTargetedID", "eduPersonPrincipalName",
-                "eduPersonUniqueId", "displayName", "givenName",
-                "eduPersonScopedAffiliation", "mail", "sn"])
-
-
-if __name__ == "__main__":
-    test_filter_ava3()
+    assert _eq(
+        list(ava.keys()),
+        [
+            "eduPersonTargetedID",
+            "eduPersonPrincipalName",
+            "eduPersonUniqueId",
+            "displayName",
+            "givenName",
+            "eduPersonScopedAffiliation",
+            "mail",
+            "sn"
+        ]
+    )