Add new config option requested_authn_context

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

Author: c00kiemon5ter

src/saml2/client.py

@@ -144,6 +144,10 @@ def prepare_for_negotiated_authenticate(
             sign_post = False if binding == BINDING_HTTP_REDIRECT else sign
             sign_redirect = False if binding == BINDING_HTTP_POST and sign else sign
 
+            requested_authn_context = (
+                kwargs.get("requested_authn_context")
+                or self.config.getattr("requested_authn_context")
+            )
             reqid, request = self.create_authn_request(
                 destination=destination,
                 vorg=vorg,
@@ -155,6 +159,7 @@ def prepare_for_negotiated_authenticate(
                 sign=sign_post,
                 sign_alg=sigalg,
                 digest_alg=digest_alg,
+                requested_authn_context=requested_authn_context,
                 **kwargs,
             )
 

src/saml2/client_base.py

@@ -17,7 +17,9 @@
 from saml2.profile import paos, ecp
 from saml2.saml import NAMEID_FORMAT_PERSISTENT
 from saml2.saml import NAMEID_FORMAT_TRANSIENT
-from saml2.samlp import AuthnQuery, RequestedAuthnContext
+from saml2.saml import AuthnContextClassRef
+from saml2.samlp import AuthnQuery
+from saml2.samlp import RequestedAuthnContext
 from saml2.samlp import NameIDMappingRequest
 from saml2.samlp import AttributeQuery
 from saml2.samlp import AuthzDecisionQuery
@@ -358,18 +360,30 @@ def create_authn_request(
             provider_name = self._my_name()
         args["provider_name"] = provider_name
 
+        requested_authn_context = (
+            kwargs.pop("requested_authn_context", None)
+            or self.config.getattr("requested_authn_context", "sp")
+            or {}
+        )
+        if requested_authn_context:
+            args["requested_authn_context"] = RequestedAuthnContext(
+                authn_context_class_ref=[
+                    AuthnContextClassRef(accr)
+                    for accr in requested_authn_context.get("authn_context_class_ref", [])
+                ],
+                comparison=requested_authn_context.get("comparison"),
+            )
+
         # Allow argument values either as class instances or as dictionaries
         # all of these have cardinality 0..1
         _msg = AuthnRequest()
-        for param in ["scoping", "requested_authn_context", "conditions", "subject"]:
+        for param in ["scoping", "conditions", "subject"]:
             _item = kwargs.pop(param, None)
             if not _item:
                 continue
 
             if isinstance(_item, _msg.child_class(param)):
                 args[param] = _item
-            elif isinstance(_item, dict):
-                args[param] = RequestedAuthnContext(**_item)
             else:
                 raise ValueError("Wrong type for param {name}".format(name=param))
 

tests/servera_conf.py

@@ -4,6 +4,8 @@
 from saml2 import BINDING_HTTP_POST
 from saml2 import BINDING_HTTP_REDIRECT
 from saml2 import BINDING_HTTP_ARTIFACT
+from saml2.authn_context import PASSWORDPROTECTEDTRANSPORT as AUTHN_PASSWORD_PROTECTED
+from saml2.authn_context import TIMESYNCTOKEN as AUTHN_TIME_SYNC_TOKEN
 from saml2.saml import NAMEID_FORMAT_TRANSIENT
 from saml2.saml import NAMEID_FORMAT_PERSISTENT
 
@@ -42,8 +44,17 @@
             "required_attributes": ["surName", "givenName", "mail"],
             "optional_attributes": ["title", "eduPersonAffiliation"],
             "idp": ["urn:mace:example.com:saml:roland:idp"],
-            "name_id_format": [NAMEID_FORMAT_TRANSIENT,
-                               NAMEID_FORMAT_PERSISTENT]
+            "name_id_format": [
+                NAMEID_FORMAT_TRANSIENT,
+                NAMEID_FORMAT_PERSISTENT,
+            ],
+            "requested_authn_context": {
+                "authn_context_class_ref": [
+                    AUTHN_PASSWORD_PROTECTED,
+                    AUTHN_TIME_SYNC_TOKEN,
+                ],
+                "comparison": "exact",
+            },
         }
     },
     "debug": 1,

tests/test_31_config.py

@@ -5,15 +5,19 @@
 import logging
 from saml2.mdstore import MetadataStore, name
 
-from saml2 import BINDING_HTTP_REDIRECT, BINDING_SOAP, BINDING_HTTP_POST
-from saml2.config import SPConfig, IdPConfig, Config
-from saml2.saml import AUTHN_PASSWORD_PROTECTED, AuthnContextClassRef
-from saml2.samlp import RequestedAuthnContext
+from saml2 import BINDING_HTTP_REDIRECT
+from saml2 import BINDING_SOAP
+from saml2.config import Config
+from saml2.config import IdPConfig
+from saml2.config import SPConfig
+from saml2.authn_context import PASSWORDPROTECTEDTRANSPORT as AUTHN_PASSWORD_PROTECTED
+from saml2.authn_context import TIMESYNCTOKEN as AUTHN_TIME_SYNC_TOKEN
 from saml2 import logger
 
 from pathutils import dotname, full_path
 from saml2.sigver import security_context, CryptoBackendXMLSecurity
 
+
 sp1 = {
     "entityid": "urn:mace:umu.se:saml:roland:sp",
     "service": {
@@ -29,12 +33,13 @@
                         {'urn:oasis:names:tc:SAML:2.0:bindings:HTTP-Redirect':
                             'http://localhost:8088/sso/'}},
             },
-            "requested_authn_context": RequestedAuthnContext(
-                    authn_context_class_ref=[
-                        AuthnContextClassRef(AUTHN_PASSWORD_PROTECTED),
-                    ],
-                    comparison="exact",
-                ),
+            "requested_authn_context": {
+                "authn_context_class_ref": [
+                    AUTHN_PASSWORD_PROTECTED,
+                    AUTHN_TIME_SYNC_TOKEN,
+                ],
+                "comparison": "exact",
+            },
         }
     },
     "key_file": full_path("test.key"),
@@ -218,13 +223,23 @@ def test_1():
 
     assert len(c._sp_idp) == 1
     assert list(c._sp_idp.keys()) == ["urn:mace:example.com:saml:roland:idp"]
-    assert list(c._sp_idp.values()) == [{'single_sign_on_service':
-                                         {
-                                             'urn:oasis:names:tc:SAML:2.0:bindings:HTTP-Redirect':
-                                             'http://localhost:8088/sso/'}}]
+    assert list(c._sp_idp.values()) == [
+        {
+            'single_sign_on_service': {
+                'urn:oasis:names:tc:SAML:2.0:bindings:HTTP-Redirect': (
+                    'http://localhost:8088/sso/'
+                )
+            }
+        }
+    ]
 
     assert c.only_use_keys_in_metadata
-    assert 'PasswordProtectedTransport' in c._sp_requested_authn_context.to_string().decode()
+    assert type(c.getattr("requested_authn_context")) is dict
+    assert c.getattr("requested_authn_context").get("authn_context_class_ref") == [
+            AUTHN_PASSWORD_PROTECTED,
+            AUTHN_TIME_SYNC_TOKEN,
+    ]
+    assert c.getattr("requested_authn_context").get("comparison") == "exact"
 
 
 def test_2():

tests/test_71_authn_request.py

@@ -1,6 +1,7 @@
 from contextlib import closing
 from saml2.client import Saml2Client
 from saml2.server import Server
+from saml2.saml import AuthnContextClassRef
 
 
 def test_authn_request_with_acs_by_index():
@@ -15,22 +16,30 @@ def test_authn_request_with_acs_by_index():
     # instead of AssertionConsumerServiceURL. The index with label ACS_INDEX
     # exists in the SP metadata in servera.xml.
     request_id, authn_request = sp.create_authn_request(
-                                sp.config.entityid,
-                                assertion_consumer_service_index=ACS_INDEX)
+        sp.config.entityid, assertion_consumer_service_index=ACS_INDEX
+    )
 
-    # Make sure the authn_request contains AssertionConsumerServiceIndex.
-    acs_index = getattr(authn_request,
-                        'assertion_consumer_service_index', None)
+    assert authn_request.requested_authn_context.authn_context_class_ref == [
+        AuthnContextClassRef(accr)
+        for accr in sp.config.getattr("requested_authn_context").get("authn_context_class_ref")
+    ]
+    assert authn_request.requested_authn_context.comparison == (
+        sp.config.getattr("requested_authn_context").get("comparison")
+    )
 
+    # Make sure the authn_request contains AssertionConsumerServiceIndex.
+    acs_index = getattr(
+        authn_request, 'assertion_consumer_service_index', None
+    )
     assert acs_index == ACS_INDEX
 
     # Create IdP.
     with closing(Server(config_file="idp_all_conf")) as idp:
-
         # Ask the IdP to pick out the binding and destination from the
         # authn_request.
-        binding, destination = idp.pick_binding("assertion_consumer_service",
-                                                request=authn_request)
+        binding, destination = idp.pick_binding(
+            "assertion_consumer_service", request=authn_request
+        )
 
         # Make sure the IdP pick_binding method picks the correct location
         # or destination based on the ACS index in the authn request.