Merge branch 'master' of https://github.com/italia/spid-testenv2

.travis.yml

@@ -25,3 +25,10 @@ after_success:
 
 before_install:
   - sudo apt-get install -y xmlsec1 libffi6
+
+matrix:
+  include:
+    - python: 3.6
+      env: TOXENV=flake8
+    - python: 3.6
+      env: TOXENV=isort

requirements.txt

@@ -5,3 +5,5 @@ passlib==1.7.1
 lxml==4.2.3
 Faker==0.8.16
 exrex
+
+importlib-resources==1.0.1

setup.cfg

@@ -0,0 +1,9 @@
+[flake8]
+exclude = *.egg-info
+
+[isort]
+line_length = 79
+combine_as_imports = true
+default_section = THIRDPARTY
+known_first_party = testenv
+not_skip = __init__.py

spid-testenv.py

@@ -6,6 +6,7 @@
 import os.path
 
 from flask import Flask
+
 from testenv.exceptions import BadConfiguration
 from testenv.server import IdpServer
 from testenv.utils import get_config

templates/login.html

@@ -7,7 +7,7 @@ <h1 class="u-text-r-xl u-margin-bottom-l">Login</h1>
             <input type="hidden" name="relay_state" value="{{relay_state}}" />
             <div class="Form-field form-animate-fields">
                 <input class="Form-input u-border-top-none u-border-right-none u-border-left-none u-border-bottom-xxs u-color-80 u-text-r-xs u-textWeight-600"
-                       id="username" name="username" aria-required="true" minlength="4" type="text" required>
+                       id="username" name="username" aria-required="true" minlength="4" type="text" required autofocus>
                 <label class="Form-label is-required u-text-r-xs u-textWeight-400 u-color-grey-40 " for="username">
                     <span class="form-label-content">Username</span>
                 </label>
@@ -26,4 +26,4 @@ <h1 class="u-text-r-xl u-margin-bottom-l">Login</h1>
             </div>
         </form>
     </article>
-{% endblock %}
+{% endblock %}

testenv/parser.py

@@ -4,12 +4,16 @@
 from datetime import datetime, timedelta
 from functools import reduce
 
+import importlib_resources
 from flask import escape
+from lxml import etree
 from saml2 import BINDING_HTTP_POST, BINDING_HTTP_REDIRECT
 from saml2.saml import NAMEID_FORMAT_ENTITY, NAMEID_FORMAT_TRANSIENT
-from testenv.settings import COMPARISONS, SPID_LEVELS, TIMEDELTA
+
+from testenv.settings import COMPARISONS, SPID_LEVELS, TIMEDELTA, XML_SCHEMAS
 from testenv.spid import Observer
-from testenv.utils import check_url, check_utc_date, str_to_time
+from testenv.translation import Libxml2Translator
+from testenv.utils import XMLError, check_url, check_utc_date, str_to_time
 
 
 class Attr(object):
@@ -239,10 +243,13 @@ def validate(self, data):
             if self._required and data is None:
                 # check if the element is required, if not provide and example
                 _error_msg = self.MANDATORY_ERROR
-                _example = '<br>Esempio:<br>'
-                lines = self._example.splitlines()
-                for line in lines:
-                    _example = '{}<pre>{}</pre>'.format(_example, escape(line))
+                if self._example:
+                    _example = '<br>Esempio:<br>'
+                    lines = self._example.splitlines()
+                    for line in lines:
+                        _example = '{}<pre>{}</pre>'.format(_example, escape(line))
+                else:
+                    _example = ''
                 _error_msg = '{} {}'.format(_error_msg, _example)
                 res['errors']['required_error'] = _error_msg
                 self._errors.update(res['errors'])
@@ -286,6 +293,8 @@ class SpidParser(object):
     """
 
     def __init__(self, *args, **kwargs):
+        from testenv.parser import XMLValidator
+        self.xml_validator = XMLValidator()
         self.schema = None
 
     def get_schema(self, action, binding, **kwargs):
@@ -453,10 +462,87 @@ def parse(self, obj, action, binding, schema=None, **kwargs):
         :param binding:
         :param schema: custom schema (None by default)
         """
+
+        errors = {}
+        # Validate xml against its XSD schema
+        validation_errors = self.xml_validator.validate_request(obj.xmlstr)
+        if validation_errors:
+            errors['validation_errors'] = validation_errors
+        # Validate xml against SPID rules
         _schema = self.get_schema(action, binding, **kwargs)\
             if schema is None else schema
         self.observer = Observer()
         self.observer.attach(_schema)
-        validated = _schema.validate(obj)
-        errors = self.observer.evaluate()
+        validated = _schema.validate(obj.message)
+        spid_errors = self.observer.evaluate()
+        if spid_errors:
+            errors['spid_errors'] = spid_errors
         return validated, errors
+
+
+class XMLSchemaFileLoader(object):
+    """
+    Load XML Schema instances from the filesystem.
+    """
+
+    def __init__(self, import_path=None):
+        self._import_path = import_path or 'testenv.xsd'
+
+    def load(self, name):
+        with importlib_resources.path(self._import_path, name) as path:
+            xmlschema_doc = etree.parse(str(path))
+            return etree.XMLSchema(xmlschema_doc)
+
+
+class XMLValidator(object):
+    """
+    Validate XML fragments against XML Schema (XSD).
+    """
+
+    def __init__(self, schema_loader=None, parser=None, translator=None):
+        self._schema_loader = schema_loader or XMLSchemaFileLoader()
+        self._parser = parser or etree.XMLParser()
+        self._translator = translator or Libxml2Translator()
+        self._load_schemas()
+
+    def _load_schemas(self):
+        self._schemas = {
+            type_: self._schema_loader.load(name)
+            for type_, name in XML_SCHEMAS.items()
+        }
+
+    def validate_request(self, xml):
+        return self._run(xml, 'protocol')
+
+    def _run(self, xml, schema_type):
+        xml_doc, parsing_errors = self._parse_xml(xml)
+        if parsing_errors:
+            return parsing_errors
+        return self._validate_xml(xml_doc, schema_type)
+
+    def _parse_xml(self, xml):
+        xml_doc, errors = None, []
+        try:
+            xml_doc = etree.fromstring(xml, parser=self._parser)
+        except SyntaxError:
+            error_log = self._parser.error_log
+            errors = self._handle_errors(error_log)
+        return xml_doc, errors
+
+    def _validate_xml(self, xml_doc, schema_type):
+        schema = self._schemas[schema_type]
+        errors = []
+        try:
+            schema.assertValid(xml_doc)
+        except Exception:
+            error_log = schema.error_log
+            errors = self._handle_errors(error_log)
+        return errors
+
+    def _handle_errors(self, errors):
+        original_errors = [
+            XMLError(err.line, err.column, err.domain_name,
+                     err.type_name, err.message, err.path)
+            for err in errors
+        ]
+        return self._translator.translate_many(original_errors)

testenv/server.py

@@ -22,6 +22,7 @@
 from saml2.s_utils import UnknownSystemEntity, UnsupportedBinding
 from saml2.saml import NAME_FORMAT_BASIC, NAMEID_FORMAT_TRANSIENT, Attribute
 from saml2.sigver import verify_redirect_signature
+
 from testenv.exceptions import BadConfiguration
 from testenv.parser import SpidParser
 from testenv.settings import (ALLOWED_SIG_ALGS, AUTH_NO_CONSENT, DIGEST_ALG,
@@ -337,7 +338,7 @@ def _raise_error(self, msg, extra=None):
 
     def _check_spid_restrictions(self, msg, action, binding, **kwargs):
         parsed_msg, errors = self.spid_parser.parse(
-            msg.message, action, binding, **kwargs
+            msg, action, binding, **kwargs
         )
         self.app.logger.debug('parsed authn_request: {}'.format(parsed_msg))
         return parsed_msg, errors
@@ -354,13 +355,14 @@ def _store_request(self, authnreq):
         self.ticket[key] = authnreq
         return key
 
-    def _handle_errors(self, errors, xmlstr):
+    def _handle_errors(self, xmlstr, errors={}):
         _escaped_xml = escape(prettify_xml(xmlstr.decode()))
         rendered_error_response = render_template_string(
             spid_error_table,
             **{
                 'lines': _escaped_xml.splitlines(),
-                'errors': errors
+                'spid_errors': errors.get('spid_errors', []),
+                'validation_errors': errors.get('validation_errors', [])
                 }
             )
         return rendered_error_response
@@ -521,7 +523,7 @@ def single_sign_on_service(self):
             )
 
         if errors:
-            return self._handle_errors(errors, req_info.xmlstr)
+            return self._handle_errors(req_info.xmlstr, errors=errors)
 
         if not req_info:
             self._raise_error('Processo di parsing del messaggio fallito.')
@@ -954,7 +956,7 @@ def single_logout_service(self):
             )
 
         if errors:
-            return self._handle_errors(errors, req_info.xmlstr)
+            return self._handle_errors(req_info.xmlstr, errors=errors)
 
         # Check if it is signed
         if _binding == BINDING_HTTP_REDIRECT:

testenv/settings.py

@@ -37,6 +37,9 @@
 SIGN_ALG = ds.SIG_RSA_SHA512
 DIGEST_ALG = ds.DIGEST_SHA512
 
+XML_SCHEMAS = {
+    'protocol': 'saml-schema-protocol-2.0.xsd',
+}
 
 spid_error_table = '''
 <html>
@@ -57,7 +60,7 @@
             </tr>
         </thead>
         <tbody>
-            {% for err in errors %}
+            {% for err in spid_errors %}
                 <tr>
                     <td class="spid-error__elem" id="{{err.1}}">{{err.1}}</td>
                     <td>
@@ -79,6 +82,14 @@
                     </td>
                 </tr>
             {% endfor %}
+            {% for err in validation_errors %}
+                <tr>
+                    <td class="spid-error__elem" id="{{err.path}}">{{err.path}}</td>
+                    <td>
+                        {{err.message}}
+                    </td>
+                </tr>
+            {% endfor %}
         </tbody>
     </table>