~ar-python-hackers/authentication-results-python/trunk

#!/usr/bin/env python
# coding: utf-8

# Copyright © 2011 Julian Mehnle <julian@mehnle.net>
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#  http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

'Module for parsing "Authentication-Results" headers as defined in RFC 5451.'

MODULE = 'authres_parse'

__author__  = 'Julian Mehnle'
__email__   = 'julian@mehnle.net'
__version__ = '0.1'

import re

# XXX Not sure about those meta declarations.  Are they Python 3 specific?
#__metaclass__ = type
#__all__ = ['build', 'MakeHeader', 'UnknownVersionError']
#ptypes = ['smtp', 'header', 'body', 'policy']

# Helper functions
###############################################################################

retype = type(re.compile(''))

def isre(obj):
    return isinstance(obj, retype)

# Exceptions
###############################################################################

class AuthResError(Exception):
    "Generic exception generated by the authres_parse module"

    def __init__(self, message = None):
        Exception.__init__(self, message)
        self.message = message

class SyntaxError(AuthResError):
    "Syntax error while parsing Authentication-Results header"

    def __init__(self, message = None, parse_text = None):
        AuthResError.__init__(self, message)
        if parse_text is None or len(parse_text) <= 40:
            self.parse_text = parse_text
        else:
            self.parse_text = parse_text[0:40] + '...'

    def __str__(self):
        if self.message and self.parse_text:
            return 'Syntax error: {0} at: {1}'.format(self.message, self.parse_text)
        elif self.message:
            return 'Syntax error: {0}'.format(self.message)
        elif self.parse_text:
            return 'Syntax error at: {0}'.format(self.parse_text)
        else:
            return 'Syntax error'

# Main classes
###############################################################################

class AuthenticationResultProperty:
    def __init__(self, type, name, value):
        self.type  = type.lower()
        self.name  = name.lower()
        self.value = value

    def __str__(self):
        return '%s.%s=%s' % (self.type, self.name, self.value)

class AuthenticationResult: pass

class DefinedAuthenticationResult(AuthenticationResult):
    def __init__(self, method, version, result, reason = None, properties = []):
        self.method     = method.lower()
        self.version    = version and version.lower()
        self.result     = result.lower()
        self.reason     = reason
        self.properties = properties

    def __str__(self):
        strs = []
        strs.append(self.method)
        if self.version:
            strs.append('/')
            strs.append(self.version)
        strs.append('=')
        strs.append(self.result)
        if self.reason:
            strs.append(' reason=')
            strs.append(self.reason)
        for property_ in self.properties:
            strs.append(' ')
            strs.append(str(property_))
        return ''.join(strs)

class NoneAuthenticationResult(AuthenticationResult):
    def __init__(self):
        pass

    def __str__(self):
        return 'none'

class AuthenticationResultsHeader:
    NONE_RESULT = NoneAuthenticationResult()

    HEADER_FIELD_NAME = 'Authentication-Results'
    HEADER_FIELD_PATTERN = re.compile(r'^Authentication-Results:\s*', re.I)

    RFC2045_TOKEN_PATTERN       = r"[A-Za-z0-9!#$%&'*+.^_`{|}~-]+"   # Printable ASCII w/o tspecials
    RFC5234_WSP_PATTERN         = r'[\t ]'
    RFC5322_QUOTED_PAIR_PATTERN = r'\\[\t \x21-\x7e]'
    RFC5322_FWS_PATTERN         = r'(?:%s*(?:\r\n|\n))?%s+' % (RFC5234_WSP_PATTERN, RFC5234_WSP_PATTERN)
    RFC5322_CTEXT_PATTERN       = r'[\x21-\x27\x2a-\x5b\x5d-\x7e]'   # Printable ASCII w/o ()\
    RFC5322_ATEXT_PATTERN       = r"[A-Za-z0-9!#$%&'*+/=?^_`{|}~-]"  # Printable ASCII w/o specials
    RFC5322_QTEXT_PATTERN       = r'[\x21\x23-\x5b\x5d-\x7e]'        # Printable ASCII w/o "\
    KTEXT_PATTERN               = r"[A-Za-z0-9!#$%&'*+?^_`{|}~-]"    # Like atext, w/o /=
    PTEXT_PATTERN               = r"[A-Za-z0-9!#$%&'*+/=?^_`{|}~.@-]"

    @classmethod
    def parse(self, string):
        """
        Creates an authres_parse.Header object by parsing an "Authentication-Results"
        header (expecting the field name at the beginning).  Expects the header to have
        been unfolded and any RFC 5322 header field comments to have been removed.
        """
        string, n = self.HEADER_FIELD_PATTERN.subn('', string, 1)
        if n == 1:
            return self.parse_value(string)
        else:
            raise SyntaxError('parse_with_name', 'Not an "Authentication-Results" header field: {0}'.format(string))

    @classmethod
    def parse_value(self, string):
        """
        Creates an authres_parse.Header object by parsing an "Authentication-Results"
        header value.  Expects the header value to have been unfolded and any RFC 5322
        header field comments to have been removed.
        """
        header = self()
        header._parse_text = string.rstrip('\r\n\t ')
        header._parse()
        return header

    def __init__(self, authserv_id = None, version = None, results = []):
        """
        Examples:

        >>> str(AuthenticationResultsHeader('test.example.org'))
        'Authentication-Results: test.example.org; none'

        >>> str(AuthenticationResultsHeader('test.example.org', version=1))
        'Authentication-Results: test.example.org 1; none'
        """
        self.authserv_id = authserv_id and authserv_id.lower()
        self.version     = version     and str(version).lower()
        self.results     = results

    def __str__(self):
        strs = []
        strs.append(self.HEADER_FIELD_NAME)
        strs.append(': ')
        strs.append(self.authserv_id)
        if self.version:
            strs.append(' ')
            strs.append(self.version)
        if len(self.results):
            for result in self.results:
                strs.append('; ')
                strs.append(str(result))
        else:
            strs.append('; ')
            strs.append(str(self.NONE_RESULT))
        return ''.join(strs)

    # Principal parser methods
    # =========================================================================

    def _parse(self):
        authserv_id = self._parse_authserv_id()
        if not authserv_id:
            raise SyntaxError('Expected authserv-id', self._parse_text)

        self._parse_rfc5322_cfws()

        version = self._parse_version()

        self._parse_rfc5322_cfws()

        results = []
        result = True
        while result:
            result = self._parse_resinfo()
            if result:
                results.append(result)
                if result == self.NONE_RESULT:
                    break
        if not len(results):
            raise SyntaxError('Expected "none" or at least one resinfo', self._parse_text)
        elif results == [self.NONE_RESULT]:
            results = []

        self._parse_rfc5322_cfws()
        self._parse_end()

        self.authserv_id = authserv_id.lower()
        self.version     = version and version.lower()
        self.results     = results

    def _parse_authserv_id(self):
        return self._parse_rfc5322_dot_atom()

    def _parse_version(self):
        version_match = self._parse_pattern(r'\d+')
        self._parse_rfc5322_cfws()
        return version_match and version_match.group()

    def _parse_resinfo(self):
        self._parse_rfc5322_cfws()
        if not self._parse_pattern(r';'):
            return
        self._parse_rfc5322_cfws()
        if self._parse_pattern(r'none'):
            return self.NONE_RESULTS
        else:
            method, version, result = self._parse_methodspec()
            self._parse_rfc5322_cfws()
            reason = self._parse_reasonspec()
            properties = []
            property_ = True
            while property_:
                self._parse_rfc5322_cfws()
                property_ = self._parse_propspec()
                if property_:
                    properties.append(property_)
            return DefinedAuthenticationResult(method, version, result, reason, properties)

    def _parse_methodspec(self):
        self._parse_rfc5322_cfws()
        method, version = self._parse_method()
        self._parse_rfc5322_cfws()
        if not self._parse_pattern(r'='):
            raise SyntaxError('Expected "="', self._parse_text)
        self._parse_rfc5322_cfws()
        result = self._parse_rfc5322_dot_atom()
        if not result:
            raise SyntaxError('Expected result', self._parse_text)
        return (method, version, result)

    def _parse_method(self):
        method = self._parse_dot_key_atom()
        if not method:
            raise SyntaxError('Expected method', self._parse_text)
        self._parse_rfc5322_cfws()
        if not self._parse_pattern(r'/'):
            return (method, None)
        self._parse_rfc5322_cfws()
        version_match = self._parse_pattern(r'\d+')
        if not version_match:
            raise SyntaxError('Expected version', self._parse_text)
        return (method, version_match.group())

    def _parse_reasonspec(self):
        if self._parse_pattern(r'reason'):
            self._parse_rfc5322_cfws()
            if not self._parse_pattern(r'='):
                raise SyntaxError('Expected "="', self._parse_text)
            self._parse_rfc5322_cfws()
            reason_match = self._parse_rfc2045_value()
            if not reason_match:
                raise SyntaxError('Expected reason', self._parse_text)
            return reason_match.group()

    def _parse_propspec(self):
        ptype = self._parse_key_atom()
        if not ptype:
            return
        elif ptype.lower() not in ['smtp', 'header', 'body', 'policy']:
            raise SyntaxError('Invalid ptype; expected any of "smtp", "header", "body", "policy", got "%s"' % ptype, self._parse_text)
        self._parse_rfc5322_cfws()
        if not self._parse_pattern(r'\.'):
            raise SyntaxError('Expected "."', self._parse_text)
        self._parse_rfc5322_cfws()
        property_ = self._parse_dot_key_atom()
        self._parse_rfc5322_cfws()
        if not self._parse_pattern(r'='):
            raise SyntaxError('Expected "="', self._parse_text)
        pvalue = self._parse_pvalue()
        if not pvalue:
            raise SyntaxError('Expected pvalue', self._parse_text)
        return AuthenticationResultProperty(ptype, property_, pvalue)

    def _parse_pvalue(self):
        self._parse_rfc5322_cfws()

        # The original rule is (modulo CFWS):
        #
        #     pvalue = [ [local-part] "@" ] domain-name / value
        #     value  = token / quoted-string
        #
        # Distinguishing <token> and <domain-name> may require backtracking,
        # and in order to avoid the need for that, the following is a simpli-
        # fication of the <pvalue> rule from RFC 5451, erring on the side of
        # laxity.
        #
        # Since <local-part> is either a <quoted-string> or <dot-atom>, and
        # <value> is either a <quoted-string> or a <token>, and <dot-atom> and
        # <token> are very similar (<dot-atom> is a superset of <token> except
        # that it multiple dots may not be adjacent), we allow a union of ".",
        # "@" and <atext> characters (jointly denoted <ptext>) in the place of
        # <dot-atom> and <token>.
        #
        # We then allow four patterns:
        #
        #     pvalue = quoted-string                 /
        #              quoted-string "@" domain-name /
        #                            "@" domain-name /
        #              1*ptext

        quoted_string_match = self._parse_rfc5322_quoted_string()
        if quoted_string_match:
            if self._parse_pattern(r'@'):
                # quoted-string "@" domain-name
                domain_name = self._parse_rfc5322_dot_atom()
                self._parse_rfc5322_cfws()
                if domain_name:
                    return '%s@%s' % (quoted_string, domain_name)
            else:
                # quoted-string
                self._parse_rfc5322_cfws()
                # Look ahead to see whether pvalue terminates after quoted-string as expected:
                if re.match(r';|$', self._parse_text):
                    return quoted_string
        else:
            if self._parse_pattern(r'@'):
                # "@" domain-name
                domain_name = self._parse_rfc5322_dot_atom()
                self._parse_rfc5322_cfws()
                if domain_name:
                    return '@' + domain_name
            else:
                # 1*ptext
                pvalue_match = self._parse_pattern(r'%s+' % self.PTEXT_PATTERN)
                self._parse_rfc5322_cfws()
                if pvalue_match:
                    return pvalue_match.group()

    def _parse_end(self):
        if self._parse_text == '':
            return True
        else:
            raise SyntaxError('Expected end of text', self._parse_text)

    # Generic grammar parser methods
    # =========================================================================

    def _parse_pattern(self, pattern):
        match = [None]

        def matched(m):
            match[0] = m
            return ''

        regexp = pattern if isre(pattern) else re.compile(r'^' + pattern, re.I)  # XXX
        self._parse_text = regexp.sub(matched, self._parse_text, 1)
        return match[0]

    def _parse_rfc2045_value(self):
        return self._parse_rfc2045_token() or self._parse_rfc5322_quoted_string()

    def _parse_rfc2045_token(self):
        token_match = self._parse_pattern(self.RFC2045_TOKEN_PATTERN)
        return token_match and token_match.group()

    def _parse_rfc5322_quoted_string(self):
        self._parse_rfc5322_cfws()
        if not self._parse_pattern(r'^"'):
            return
        all_qcontent = ''
        qcontent_match = True
        while qcontent_match:
            fws_match = self._parse_pattern(self.RFC5322_FWS_PATTERN)
            if fws_match:
                all_qcontent += fws_match.group()
            qcontent_match = self._parse_rfc5322_qcontent()
            if qcontent_match:
                all_qcontent += qcontent_match.group()
        self._parse_pattern(self.RFC5322_FWS_PATTERN)
        if not self._parse_pattern(r'"'):
            raise SyntaxError('Expected <">', self._parse_text)
        self._parse_rfc5322_cfws()
        return all_qcontent

    def _parse_rfc5322_qcontent(self):
        if self._parse_pattern(r'%s+' % self.RFC5322_QTEXT_PATTERN):
            return True
        elif self._parse_pattern(self.RFC5322_QUOTED_PAIR_PATTERN):
            return True

    def _parse_rfc5322_dot_atom(self):
        self._parse_rfc5322_cfws()
        dot_atom_text_match = self._parse_pattern(r'%s+(?:\.%s+)*' %
            (self.RFC5322_ATEXT_PATTERN, self.RFC5322_ATEXT_PATTERN))
        self._parse_rfc5322_cfws()
        return dot_atom_text_match and dot_atom_text_match.group()

    def _parse_dot_key_atom(self):
        # Like _parse_rfc5322_dot_atom, but disallows "/" (forward slash) and
        # "=" (equal sign).
        self._parse_rfc5322_cfws()
        dot_atom_text_match = self._parse_pattern(r'%s+(?:\.%s+)*' %
            (self.KTEXT_PATTERN, self.KTEXT_PATTERN))
        self._parse_rfc5322_cfws()
        return dot_atom_text_match and dot_atom_text_match.group()

    def _parse_key_atom(self):
        # Like _parse_dot_key_atom, but also disallows "." (dot).
        self._parse_rfc5322_cfws()
        dot_atom_text_match = self._parse_pattern(r'%s+' % self.KTEXT_PATTERN)
        self._parse_rfc5322_cfws()
        return dot_atom_text_match and dot_atom_text_match.group()

    def _parse_rfc5322_cfws(self):
        fws_match     = False
        comment_match = True
        while comment_match:
            fws_match     = fws_match or self._parse_pattern(self.RFC5322_FWS_PATTERN)
            comment_match = self._parse_rfc5322_comment()
        fws_match = fws_match or self._parse_pattern(self.RFC5322_FWS_PATTERN)
        return fws_match or comment_match

    def _parse_rfc5322_comment(self):
        if self._parse_pattern(r'\('):
            while self._parse_pattern(self.RFC5322_FWS_PATTERN) or self._parse_rfc5322_ccontent(): pass
            if self._parse_pattern(r'^\)'):
                return True
            else:
                raise SyntaxError('comment: expected FWS or ccontent or ")"', self._parse_text)

    def _parse_rfc5322_ccontent(self):
        if self._parse_pattern(r'%s+' % self.RFC5322_CTEXT_PATTERN):
            return True
        elif self._parse_pattern(self.RFC5322_QUOTED_PAIR_PATTERN):
            return True
        elif self._parse_rfc5322_comment():
            return True

def _test():
    import doctest
    import authres_parse
    return doctest.testmod(authres_parse)

if __name__ == '__main__':
    _test()

# vim:sw=4 sts=4