Source Code for Module dkim.canonicalization

  1  # This software is provided 'as-is', without any express or implied 
  2  # warranty.  In no event will the author be held liable for any damages 
  3  # arising from the use of this software. 
  4  # 
  5  # Permission is granted to anyone to use this software for any purpose, 
  6  # including commercial applications, and to alter it and redistribute it 
  7  # freely, subject to the following restrictions: 
  8  # 
  9  # 1. The origin of this software must not be misrepresented; you must not 
 10  #    claim that you wrote the original software. If you use this software 
 11  #    in a product, an acknowledgment in the product documentation would be 
 12  #    appreciated but is not required. 
 13  # 2. Altered source versions must be plainly marked as such, and must not be 
 14  #    misrepresented as being the original software. 
 15  # 3. This notice may not be removed or altered from any source distribution. 
 16  # 
 17  # Copyright (c) 2008 Greg Hewgill http://hewgill.com 
 18  # 
 19  # This has been modified from the original software. 
 20  # Copyright (c) 2011 William Grant <me@williamgrant.id.au> 
 21   
 22  import re 
 23   
 24  __all__ = [ 
 25      'CanonicalizationPolicy', 
 26      'InvalidCanonicalizationPolicyError', 
 27      ] 
 28   
 29   
 30 -class InvalidCanonicalizationPolicyError(Exception): 
 31      """The c= value could not be parsed.""" 
 32      pass 
 33   
 34   
 35 -def strip_trailing_whitespace(content): 
 36      return re.sub(b"[\t ]+\r\n", b"\r\n", content) 
 37   
 38   
 39 -def compress_whitespace(content): 
 40      return re.sub(b"[\t ]+", b" ", content) 
 41   
 42   
 43 -def strip_trailing_lines(content): 
 44      end = None 
 45      while content.endswith(b"\r\n", 0, end): 
 46          if end is None: 
 47              end = -2 
 48          else: 
 49              end -= 2 
 50   
 51      if end is None: 
 52          return content + b"\r\n" 
 53   
 54      end += 2 
 55      if end == 0: 
 56          return content 
 57   
 58      return content[:end] 
 59   
 60 -def unfold_header_value(content): 
 61      return re.sub(b"\r\n", b"", content) 
 62   
 63   
 64 -def correct_empty_body(content): 
 65      if content == b"\r\n": 
 66          return b"" 
 67      else: 
 68          return content 
 69   
 70   
 71 -class Simple: 
 72      """Class that represents the "simple" canonicalization algorithm.""" 
 73   
 74      name = b"simple" 
 75   
 76      @staticmethod 
 77 -    def canonicalize_headers(headers): 
 78          # No changes to headers. 
 79          return headers 
 80   
 81      @staticmethod 
 82 -    def canonicalize_body(body): 
 83          # Ignore all empty lines at the end of the message body. 
 84          return strip_trailing_lines(body) 
 85   
 86   
 87 -class Relaxed: 
 88      """Class that represents the "relaxed" canonicalization algorithm.""" 
 89   
 90      name = b"relaxed" 
 91   
 92      @staticmethod 
 93 -    def canonicalize_headers(headers): 
 94          # Convert all header field names to lowercase. 
 95          # Unfold all header lines. 
 96          # Compress WSP to single space. 
 97          # Remove all WSP at the start or end of the field value (strip). 
 98          return [ 
 99              (x[0].lower().rstrip(), 
100               compress_whitespace(unfold_header_value(x[1])).strip() + b"\r\n") 
101              for x in headers] 
102   
103      @staticmethod 
104 -    def canonicalize_body(body): 
105          # Remove all trailing WSP at end of lines. 
106          # Compress non-line-ending WSP to single space. 
107          # Ignore all empty lines at the end of the message body. 
108          return correct_empty_body(strip_trailing_lines( 
109              compress_whitespace(strip_trailing_whitespace(body)))) 
110   
111   
112 -class CanonicalizationPolicy: 
113   
114 -    def __init__(self, header_algorithm, body_algorithm): 
115          self.header_algorithm = header_algorithm 
116          self.body_algorithm = body_algorithm 
117   
118      @classmethod 
119 -    def from_c_value(cls, c): 
120          """Construct the canonicalization policy described by a c= value. 
121   
122          May raise an C{InvalidCanonicalizationPolicyError} if the given 
123          value is invalid 
124   
125          @param c: c= value from a DKIM-Signature header field 
126          @return: a C{CanonicalizationPolicy} 
127          """ 
128          if c is None: 
129              c = b'simple/simple' 
130          m = c.split(b'/') 
131          if len(m) not in (1, 2): 
132              raise InvalidCanonicalizationPolicyError(c) 
133          if len(m) == 1: 
134              m.append(b'simple') 
135          can_headers, can_body = m 
136          try: 
137              header_algorithm = ALGORITHMS[can_headers] 
138              body_algorithm = ALGORITHMS[can_body] 
139          except KeyError as e: 
140              raise InvalidCanonicalizationPolicyError(e.args[0]) 
141          return cls(header_algorithm, body_algorithm) 
142   
143 -    def to_c_value(self): 
144          return b'/'.join( 
145              (self.header_algorithm.name, self.body_algorithm.name)) 
146   
147 -    def canonicalize_headers(self, headers): 
148          return self.header_algorithm.canonicalize_headers(headers) 
149   
150 -    def canonicalize_body(self, body): 
151          return self.body_algorithm.canonicalize_body(body) 
152   
153   
154  ALGORITHMS = dict((c.name, c) for c in (Simple, Relaxed)) 
155