Authn Request Response plugins

- feat: `--authn-plugin` is the new optional parameter that allows to adopt a user-defined, installable and custom plugin, for handling authn request and responses

README.md

@@ -179,6 +179,8 @@ spid-sp-test offers the possibility to extend and configure new response tests t
   The templates are Jinja2 powered, so it's possible to
   extend `src/spid_sp_test/responses/templates/base.xml` with our preferred values
 
+- customize the way to get the SAML2 Authn Request, using plugins wrote by your own. If you're using a IAM Proxy with some OAuth2/OIDC frontends of a custom API, you can write your plugin and use it in the cli arguments, eg: `spid_sp_test --metadata-url https://localhost:8000/spid/metadata --extra  --authn-url https://localhost:8000/spid/login/?idp=http://localhost:8080 --debug INFO -tr --authn-plugin spid_sp_test.plugins.authn_request.Dummy`
+
 Looking at `src/spid_sp_test/responses/settings.py` or `tests/example.test-suite.json`
 we found that every test have a `response` attribute. Each element configured in would overload the
 value that will be rendered in the template. Each template can load these variable from its template context or

setup.py

@@ -10,7 +10,7 @@ def readme():
 
 setup(
     name='spid_sp_test',
-    version='0.7.0',
+    version='0.8.0',
     description="SAML2 SPID Service Provider validation tool that can be run from the command line",
     long_description=readme(),
     long_description_content_type='text/markdown',

src/spid_sp_test/authn_request.py

@@ -27,12 +27,14 @@
 from tempfile import NamedTemporaryFile
 
 from . exceptions import SAMLRequestNotFound
+from . utils import load_plugin
 
 
 logger = logging.getLogger(__name__)
 
 
-def get_authn_request(authn_request_url, verify_ssl=False):
+def get_authn_request(
+    authn_request_url:str, verify_ssl:bool=False, authn_plugin:str=None):
     """
         Detects the auth request url, if http/xml file or html file
     """
@@ -42,6 +44,12 @@ def get_authn_request(authn_request_url, verify_ssl=False):
     authn_request_str = None
     requests_session = None
 
+    # eg: auth_plugin = 'that.package.module.Class'
+    requests_session = requests.Session()
+    if authn_plugin:
+        func = load_plugin(authn_plugin)
+        authn_request_url = func(requests_session, authn_request_url).request()
+
     if authn_request_url[0:7] == 'file://':
         authn_request = open(authn_request_url[7:], 'rb').read().strip().strip(b'\n')
         # stupid test ... good enough for now
@@ -53,7 +61,6 @@ def get_authn_request(authn_request_url, verify_ssl=False):
         else:
             raise Exception(f"Can't detect authn request from f{authn_request_url}")
     else:
-        requests_session = requests.Session()
         request = requests_session.get(
             authn_request_url,
             verify=verify_ssl,
@@ -124,7 +131,8 @@ def __init__(self,
                  authn_request: dict = {},
                  xsds_files: list = None,
                  xsds_files_path: str = None,
-                 production: bool = False):
+                 production: bool = False,
+                 authn_plugin:str = None):
 
         super(SpidSpAuthnReqCheck, self).__init__(verify_ssl=production)
         self.category = 'authnrequest_strict'
@@ -133,7 +141,8 @@ def __init__(self,
         self.metadata = metadata
 
         self.authn_request = get_authn_request(authn_request_url,
-                                               verify_ssl=production)
+                                               verify_ssl=production,
+                                               authn_plugin=authn_plugin)
 
         try:
             self.authn_request_decoded = self.authn_request['SAMLRequest_xml']

src/spid_sp_test/plugins/authn_request.py

@@ -0,0 +1,18 @@
+
+
+class Dummy(object):
+
+    def __init__(self, requests_session, authn_request_url):
+        self.requests_session = requests_session
+        self.authn_request_url = authn_request_url
+
+    def request(self):
+        """
+        """
+        return self.authn_request_url
+
+    def response(self, url, data:dict={}, **kwargs):
+        """
+        """
+        res = self.requests_session.post(url, data=data, allow_redirects=True)
+        return res

src/spid_sp_test/response.py

@@ -25,6 +25,7 @@
 from tempfile import NamedTemporaryFile
 
 from . utils import get_xmlsec1_bin, html_absolute_paths
+from . utils import load_plugin
 
 logger = logging.getLogger(__name__)
 
@@ -147,14 +148,17 @@ def __init__(self, *args, **kwargs):
         self.kwargs = kwargs
         self.status_codes = None
 
+        self.authn_plugin = kwargs.get('authn_plugin')
+
     def get_acr(self):
         _acr = self.authnreq_etree.xpath(
             '//RequestedAuthnContext/AuthnContextClassRef')
         if _acr:
             return _acr[0].text
 
     def do_authnrequest(self):
-        self.authn_request_data = get_authn_request(self.authn_request_url)
+        self.authn_request_data = get_authn_request(
+                self.authn_request_url, authn_plugin=self.authn_plugin)
         self.authnreq_etree = etree.fromstring(
             self.authn_request_data['SAMLRequest_xml'])
         del_ns(self.authnreq_etree)
@@ -310,7 +314,11 @@ def send_response(self, xmlstr):
         }
         url = self.authnreq_attrs.get('AssertionConsumerURL', self.acs_url)
         ua = self.authn_request_data['requests_session']
-        res = ua.post(url, data=data, allow_redirects=True)
+        if self.authn_plugin:
+            func = load_plugin(self.authn_plugin)
+            res = func(ua, self.authn_request_url).response(url, data)
+        else:
+            res = ua.post(url, data=data, allow_redirects=True)
         msg = f'Response http status code [{res.status_code}]: {res.content.decode()}'
         self.logger.debug(msg)
         return res

src/spid_sp_test/spid_sp_test

@@ -215,6 +215,12 @@ if __name__ == '__main__':
         help="which profile to check"
     )
 
+    parser.add_argument(
+        '-ap', '--authn-plugin',
+        required=False,
+        help="use a custom plugin for authn requests, eg: for proxies/gateways"
+    )
+
     args = parser.parse_args()
     logging.basicConfig(level=getattr(logging, args.debug))
 
@@ -253,7 +259,8 @@ if __name__ == '__main__':
         data_ac = dict(
             metadata=metadata_check.metadata,
             authn_request_url=args.authn_url,
-            production=args.production
+            production=args.production,
+            authn_plugin=args.authn_plugin
         )
         authn_check = SpidSpAuthnReqCheck(**data_ac)
         selective_run(authn_check, profile, args.list)
@@ -275,7 +282,8 @@ if __name__ == '__main__':
             attr_json=args.attr_json,
             production=args.production,
             html_path=html_path if args.html_path else None,
-            no_send_response = args.no_send_response
+            no_send_response = args.no_send_response,
+            authn_plugin=args.authn_plugin
         )
 
         if args.html_path:

src/spid_sp_test/utils.py

@@ -1,4 +1,5 @@
 import base64
+import importlib
 import lxml.objectify
 import os
 import re
@@ -220,3 +221,10 @@ def report_to_html(o):
                         res.append('<p>Obtained value: %s</p>' % t['value'])
                         res.append('</div>')
     return '\n'.join(res)
+
+
+def load_plugin(plugin_name):
+    n1, _, n2 = plugin_name.rpartition('.')
+    module = importlib.import_module(n1)
+    func = getattr(module, n2)
+    return func