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      return re.sub(b"(\r\n)*$", b"\r\n", content) 
 45   
 46   
 47 -def unfold_header_value(content): 
 48      return re.sub(b"\r\n", b"", content) 
 49   
 50   
 51 -def correct_empty_body(content): 
 52      if content == b"\r\n": 
 53          return b"" 
 54      else: 
 55          return content 
 56   
 57   
 58 -class Simple: 
 59      """Class that represents the "simple" canonicalization algorithm.""" 
 60   
 61      name = b"simple" 
 62   
 63      @staticmethod 
 64 -    def canonicalize_headers(headers): 
 65          # No changes to headers. 
 66          return headers 
 67   
 68      @staticmethod 
 69 -    def canonicalize_body(body): 
 70          # Ignore all empty lines at the end of the message body. 
 71          return strip_trailing_lines(body) 
 72   
 73   
 74 -class Relaxed: 
 75      """Class that represents the "relaxed" canonicalization algorithm.""" 
 76   
 77      name = b"relaxed" 
 78   
 79      @staticmethod 
 80 -    def canonicalize_headers(headers): 
 81          # Convert all header field names to lowercase. 
 82          # Unfold all header lines. 
 83          # Compress WSP to single space. 
 84          # Remove all WSP at the start or end of the field value (strip). 
 85          return [ 
 86              (x[0].lower().rstrip(), 
 87               compress_whitespace(unfold_header_value(x[1])).strip() + b"\r\n") 
 88              for x in headers] 
 89   
 90      @staticmethod 
 91 -    def canonicalize_body(body): 
 92          # Remove all trailing WSP at end of lines. 
 93          # Compress non-line-ending WSP to single space. 
 94          # Ignore all empty lines at the end of the message body. 
 95          return correct_empty_body(strip_trailing_lines( 
 96              compress_whitespace(strip_trailing_whitespace(body)))) 
 97   
 98   
 99 -class CanonicalizationPolicy: 
100   
101 -    def __init__(self, header_algorithm, body_algorithm): 
102          self.header_algorithm = header_algorithm 
103          self.body_algorithm = body_algorithm 
104   
105      @classmethod 
106 -    def from_c_value(cls, c): 
107          """Construct the canonicalization policy described by a c= value. 
108   
109          May raise an C{InvalidCanonicalizationPolicyError} if the given 
110          value is invalid 
111   
112          @param c: c= value from a DKIM-Signature header field 
113          @return: a C{CanonicalizationPolicy} 
114          """ 
115          if c is None: 
116              c = b'simple/simple' 
117          m = c.split(b'/') 
118          if len(m) not in (1, 2): 
119              raise InvalidCanonicalizationPolicyError(c) 
120          if len(m) == 1: 
121              m.append(b'simple') 
122          can_headers, can_body = m 
123          try: 
124              header_algorithm = ALGORITHMS[can_headers] 
125              body_algorithm = ALGORITHMS[can_body] 
126          except KeyError as e: 
127              raise InvalidCanonicalizationPolicyError(e.args[0]) 
128          return cls(header_algorithm, body_algorithm) 
129   
130 -    def to_c_value(self): 
131          return b'/'.join( 
132              (self.header_algorithm.name, self.body_algorithm.name)) 
133   
134 -    def canonicalize_headers(self, headers): 
135          return self.header_algorithm.canonicalize_headers(headers) 
136   
137 -    def canonicalize_body(self, body): 
138          return self.body_algorithm.canonicalize_body(body) 
139   
140   
141  ALGORITHMS = dict((c.name, c) for c in (Simple, Relaxed)) 
142