Stuart D. Gathman
621 lines
24 KiB
Python
621 lines
24 KiB
Python
# This software is provided 'as-is', without any express or implied
|
|
# warranty. In no event will the author be held liable for any damages
|
|
# arising from the use of this software.
|
|
#
|
|
# Permission is granted to anyone to use this software for any purpose,
|
|
# including commercial applications, and to alter it and redistribute it
|
|
# freely, subject to the following restrictions:
|
|
#
|
|
# 1. The origin of this software must not be misrepresented; you must not
|
|
# claim that you wrote the original software. If you use this software
|
|
# in a product, an acknowledgment in the product documentation would be
|
|
# appreciated but is not required.
|
|
# 2. Altered source versions must be plainly marked as such, and must not be
|
|
# misrepresented as being the original software.
|
|
# 3. This notice may not be removed or altered from any source distribution.
|
|
#
|
|
# Copyright (c) 2008 Greg Hewgill http://hewgill.com
|
|
#
|
|
# This has been modified from the original software.
|
|
# Copyright (c) 2011 William Grant <me@williamgrant.id.au>
|
|
|
|
import base64
|
|
import hashlib
|
|
import logging
|
|
import re
|
|
import time
|
|
|
|
from dkim.canonicalization import (
|
|
CanonicalizationPolicy,
|
|
InvalidCanonicalizationPolicyError,
|
|
)
|
|
from dkim.crypto import (
|
|
DigestTooLargeError,
|
|
HASH_ALGORITHMS,
|
|
parse_pem_private_key,
|
|
parse_public_key,
|
|
RSASSA_PKCS1_v1_5_sign,
|
|
RSASSA_PKCS1_v1_5_verify,
|
|
UnparsableKeyError,
|
|
)
|
|
try:
|
|
from dkim.dnsplug import get_txt
|
|
except:
|
|
def get_txt(s):
|
|
raise RuntimeError("DKIM.verify requires DNS or dnspython module")
|
|
from dkim.util import (
|
|
get_default_logger,
|
|
InvalidTagValueList,
|
|
parse_tag_value,
|
|
)
|
|
|
|
__all__ = [
|
|
"DKIMException",
|
|
"InternalError",
|
|
"KeyFormatError",
|
|
"MessageFormatError",
|
|
"ParameterError",
|
|
"Relaxed",
|
|
"Simple",
|
|
"DKIM",
|
|
"sign",
|
|
"verify",
|
|
]
|
|
|
|
Relaxed = b'relaxed' # for clients passing dkim.Relaxed
|
|
Simple = b'simple' # for clients passing dkim.Simple
|
|
|
|
def bitsize(x):
|
|
"""Return size of long in bits."""
|
|
return len(bin(x)) - 2
|
|
|
|
class DKIMException(Exception):
|
|
"""Base class for DKIM errors."""
|
|
pass
|
|
|
|
class InternalError(DKIMException):
|
|
"""Internal error in dkim module. Should never happen."""
|
|
pass
|
|
|
|
class KeyFormatError(DKIMException):
|
|
"""Key format error while parsing an RSA public or private key."""
|
|
pass
|
|
|
|
class MessageFormatError(DKIMException):
|
|
"""RFC822 message format error."""
|
|
pass
|
|
|
|
class ParameterError(DKIMException):
|
|
"""Input parameter error."""
|
|
pass
|
|
|
|
class ValidationError(DKIMException):
|
|
"""Validation error."""
|
|
pass
|
|
|
|
def select_headers(headers, include_headers):
|
|
"""Select message header fields to be signed/verified.
|
|
|
|
>>> h = [('from','biz'),('foo','bar'),('from','baz'),('subject','boring')]
|
|
>>> i = ['from','subject','to','from']
|
|
>>> select_headers(h,i)
|
|
[('from', 'baz'), ('subject', 'boring'), ('from', 'biz')]
|
|
>>> h = [('From','biz'),('Foo','bar'),('Subject','Boring')]
|
|
>>> i = ['from','subject','to','from']
|
|
>>> select_headers(h,i)
|
|
[('From', 'biz'), ('Subject', 'Boring')]
|
|
"""
|
|
sign_headers = []
|
|
lastindex = {}
|
|
for h in include_headers:
|
|
assert h == h.lower()
|
|
i = lastindex.get(h, len(headers))
|
|
while i > 0:
|
|
i -= 1
|
|
if h == headers[i][0].lower():
|
|
sign_headers.append(headers[i])
|
|
break
|
|
lastindex[h] = i
|
|
return sign_headers
|
|
|
|
# FWS = ([*WSP CRLF] 1*WSP) / obs-FWS ; Folding white space [RFC5322]
|
|
FWS = r'(?:(?:\s*\r?\n)?\s+)?'
|
|
RE_BTAG = re.compile(r'([;\s]b'+FWS+r'=)(?:'+FWS+r'[a-zA-Z0-9+/=])*(?:\r?\n\Z)?')
|
|
|
|
def hash_headers(hasher, canonicalize_headers, headers, include_headers,
|
|
sigheader, sig):
|
|
"""Update hash for signed message header fields."""
|
|
sign_headers = select_headers(headers,include_headers)
|
|
# The call to _remove() assumes that the signature b= only appears
|
|
# once in the signature header
|
|
cheaders = canonicalize_headers.canonicalize_headers(
|
|
[(sigheader[0], RE_BTAG.sub(b'\\1',sigheader[1]))])
|
|
# the dkim sig is hashed with no trailing crlf, even if the
|
|
# canonicalization algorithm would add one.
|
|
for x,y in sign_headers + [(x, y.rstrip()) for x,y in cheaders]:
|
|
hasher.update(x)
|
|
hasher.update(b":")
|
|
hasher.update(y)
|
|
return sign_headers
|
|
|
|
def validate_signature_fields(sig):
|
|
"""Validate DKIM-Signature fields.
|
|
|
|
Basic checks for presence and correct formatting of mandatory fields.
|
|
Raises a ValidationError if checks fail, otherwise returns None.
|
|
|
|
@param sig: A dict mapping field keys to values.
|
|
"""
|
|
mandatory_fields = (b'v', b'a', b'b', b'bh', b'd', b'h', b's')
|
|
for field in mandatory_fields:
|
|
if field not in sig:
|
|
raise ValidationError("signature missing %s=" % field)
|
|
|
|
if sig[b'v'] != b"1":
|
|
raise ValidationError("v= value is not 1 (%s)" % sig[b'v'])
|
|
if re.match(br"[\s0-9A-Za-z+/]+=*$", sig[b'b']) is None:
|
|
raise ValidationError("b= value is not valid base64 (%s)" % sig[b'b'])
|
|
if re.match(br"[\s0-9A-Za-z+/]+=*$", sig[b'bh']) is None:
|
|
raise ValidationError(
|
|
"bh= value is not valid base64 (%s)" % sig[b'bh'])
|
|
# Nasty hack to support both str and bytes... check for both the
|
|
# character and integer values.
|
|
if b'i' in sig and (
|
|
not sig[b'i'].endswith(sig[b'd']) or
|
|
sig[b'i'][-len(sig[b'd'])-1] not in ('@', '.', 64, 46)):
|
|
raise ValidationError(
|
|
"i= domain is not a subdomain of d= (i=%s d=%s)" %
|
|
(sig[b'i'], sig[b'd']))
|
|
if b'l' in sig and re.match(br"\d{,76}$", sig['l']) is None:
|
|
raise ValidationError(
|
|
"l= value is not a decimal integer (%s)" % sig[b'l'])
|
|
if b'q' in sig and sig[b'q'] != b"dns/txt":
|
|
raise ValidationError("q= value is not dns/txt (%s)" % sig[b'q'])
|
|
now = int(time.time())
|
|
slop = 36000 # 10H leeway for mailers with inaccurate clocks
|
|
t_sign = 0
|
|
if b't' in sig
|
|
if re.match(br"\d+$", sig[b't']) is None:
|
|
raise ValidationError(
|
|
"t= value is not a decimal integer (%s)" % sig[b't'])
|
|
t_sign = int(sig[b't'])
|
|
if t_sign > now + slop:
|
|
raise ValidationError(
|
|
"t= value is in the future (%s)" % sig[b't'])
|
|
if b'x' in sig:
|
|
if re.match(br"\d+$", sig[b'x']) is None:
|
|
raise ValidationError(
|
|
"x= value is not a decimal integer (%s)" % sig[b'x'])
|
|
x_sign = int(sig[b'x'])
|
|
if x_sign < now - slop:
|
|
raise ValidationError(
|
|
"x= value is past (%s)" % sig[b'x'])
|
|
if x_sign < t_sign:
|
|
raise ValidationError(
|
|
"x= value is less than t= value (x=%s t=%s)" %
|
|
(sig[b'x'], sig[b't']))
|
|
|
|
|
|
def rfc822_parse(message):
|
|
"""Parse a message in RFC822 format.
|
|
|
|
@param message: The message in RFC822 format. Either CRLF or LF is an accepted line separator.
|
|
@return: Returns a tuple of (headers, body) where headers is a list of (name, value) pairs.
|
|
The body is a CRLF-separated string.
|
|
"""
|
|
headers = []
|
|
lines = re.split(b"\r?\n", message)
|
|
i = 0
|
|
while i < len(lines):
|
|
if len(lines[i]) == 0:
|
|
# End of headers, return what we have plus the body, excluding the blank line.
|
|
i += 1
|
|
break
|
|
if lines[i][0] in ("\x09", "\x20", 0x09, 0x20):
|
|
headers[-1][1] += lines[i]+b"\r\n"
|
|
else:
|
|
m = re.match(br"([\x21-\x7e]+?):", lines[i])
|
|
if m is not None:
|
|
headers.append([m.group(1), lines[i][m.end(0):]+b"\r\n"])
|
|
elif lines[i].startswith(b"From "):
|
|
pass
|
|
else:
|
|
raise MessageFormatError("Unexpected characters in RFC822 header: %s" % lines[i])
|
|
i += 1
|
|
return (headers, b"\r\n".join(lines[i:]))
|
|
|
|
|
|
|
|
def fold(header):
|
|
"""Fold a header line into multiple crlf-separated lines at column 72.
|
|
|
|
>>> fold(b'foo')
|
|
'foo'
|
|
>>> fold(b'foo '+b'foo'*24).splitlines()[0]
|
|
'foo '
|
|
>>> fold(b'foo'*25).splitlines()[-1]
|
|
' foo'
|
|
>>> len(fold(b'foo'*25).splitlines()[0])
|
|
72
|
|
"""
|
|
i = header.rfind(b"\r\n ")
|
|
if i == -1:
|
|
pre = b""
|
|
else:
|
|
i += 3
|
|
pre = header[:i]
|
|
header = header[i:]
|
|
while len(header) > 72:
|
|
i = header[:72].rfind(b" ")
|
|
if i == -1:
|
|
j = 72
|
|
else:
|
|
j = i + 1
|
|
pre += header[:j] + b"\r\n "
|
|
header = header[j:]
|
|
return pre + header
|
|
|
|
#: Hold messages and options during DKIM signing and verification.
|
|
class DKIM(object):
|
|
# NOTE - the first 2 indentation levels are 2 instead of 4
|
|
# to minimize changed lines from the function only version.
|
|
|
|
#: The U{RFC5322<http://tools.ietf.org/html/rfc5322#section-3.6>}
|
|
#: complete list of singleton headers (which should
|
|
#: appear at most once). This can be used for a "paranoid" or
|
|
#: "strict" signing mode.
|
|
#: Bcc in this list is in the SHOULD NOT sign list, the rest could
|
|
#: be in the default FROZEN list, but that could also make signatures
|
|
#: more fragile than necessary.
|
|
#: @since: 0.5
|
|
RFC5322_SINGLETON = ('date','from','sender','reply-to','to','cc','bcc',
|
|
'message-id','in-reply-to','references')
|
|
|
|
#: Header fields to protect from additions by default.
|
|
#:
|
|
#: The short list below is the result more of instinct than logic.
|
|
#: @since: 0.5
|
|
FROZEN = ('from','date','subject')
|
|
|
|
#: The rfc4871 recommended header fields to sign
|
|
#: @since: 0.5
|
|
SHOULD = (
|
|
'sender', 'reply-to', 'subject', 'date', 'message-id', 'to', 'cc',
|
|
'mime-version', 'content-type', 'content-transfer-encoding', 'content-id',
|
|
'content- description', 'resent-date', 'resent-from', 'resent-sender',
|
|
'resent-to', 'resent-cc', 'resent-message-id', 'in-reply-to', 'references',
|
|
'list-id', 'list-help', 'list-unsubscribe', 'list-subscribe', 'list-post',
|
|
'list-owner', 'list-archive'
|
|
)
|
|
|
|
#: The rfc4871 recommended header fields not to sign.
|
|
#: @since: 0.5
|
|
SHOULD_NOT = (
|
|
'return-path', 'received', 'comments', 'keywords', 'bcc', 'resent-bcc',
|
|
'dkim-signature'
|
|
)
|
|
|
|
#: Create a DKIM instance to sign and verify rfc5322 messages.
|
|
#:
|
|
#: @param message: an RFC822 formatted message to be signed or verified
|
|
#: (with either \\n or \\r\\n line endings)
|
|
#: @param logger: a logger to which debug info will be written (default None)
|
|
#: @param signature_algorithm: the signing algorithm to use when signing
|
|
def __init__(self,message=None,logger=None,signature_algorithm=b'rsa-sha256',
|
|
minkey=1024):
|
|
self.set_message(message)
|
|
if logger is None:
|
|
logger = get_default_logger()
|
|
self.logger = logger
|
|
if signature_algorithm not in HASH_ALGORITHMS:
|
|
raise ParameterError(
|
|
"Unsupported signature algorithm: "+signature_algorithm)
|
|
self.signature_algorithm = signature_algorithm
|
|
#: Header fields which should be signed. Default from RFC4871
|
|
self.should_sign = set(DKIM.SHOULD)
|
|
#: Header fields which should not be signed. The default is from RFC4871.
|
|
#: Attempting to sign these headers results in an exception.
|
|
#: If it is necessary to sign one of these, it must be removed
|
|
#: from this list first.
|
|
self.should_not_sign = set(DKIM.SHOULD_NOT)
|
|
#: Header fields to sign an extra time to prevent additions.
|
|
self.frozen_sign = set(DKIM.FROZEN)
|
|
#: Minimum public key size. Shorter keys raise KeyFormatError. The
|
|
#: default is 1024
|
|
self.minkey = minkey
|
|
|
|
def add_frozen(self,s):
|
|
""" Add headers not in should_not_sign to frozen_sign.
|
|
@param s: list of headers to add to frozen_sign
|
|
@since: 0.5
|
|
|
|
>>> dkim = DKIM()
|
|
>>> dkim.add_frozen(DKIM.RFC5322_SINGLETON)
|
|
>>> sorted(dkim.frozen_sign)
|
|
['cc', 'date', 'from', 'in-reply-to', 'message-id', 'references', 'reply-to', 'sender', 'subject', 'to']
|
|
"""
|
|
self.frozen_sign.update(x.lower() for x in s
|
|
if x.lower() not in self.should_not_sign)
|
|
|
|
#: Load a new message to be signed or verified.
|
|
#: @param message: an RFC822 formatted message to be signed or verified
|
|
#: (with either \\n or \\r\\n line endings)
|
|
#: @since: 0.5
|
|
def set_message(self,message):
|
|
if message:
|
|
self.headers, self.body = rfc822_parse(message)
|
|
else:
|
|
self.headers, self.body = [],''
|
|
#: The DKIM signing domain last signed or verified.
|
|
self.domain = None
|
|
#: The DKIM key selector last signed or verified.
|
|
self.selector = 'default'
|
|
#: Signature parameters of last sign or verify. To parse
|
|
#: a DKIM-Signature header field that you have in hand,
|
|
#: use L{dkim.util.parse_tag_value}.
|
|
self.signature_fields = {}
|
|
#: The list of headers last signed or verified. Each header
|
|
#: is a name,value tuple. FIXME: The headers are canonicalized.
|
|
#: This could be more useful as original headers.
|
|
self.signed_headers = []
|
|
#: The public key size last verified.
|
|
self.keysize = 0
|
|
|
|
def default_sign_headers(self):
|
|
"""Return the default list of headers to sign: those in should_sign or
|
|
frozen_sign, with those in frozen_sign signed an extra time to prevent
|
|
additions.
|
|
@since: 0.5"""
|
|
hset = self.should_sign | self.frozen_sign
|
|
include_headers = [ x for x,y in self.headers
|
|
if x.lower() in hset ]
|
|
return include_headers + [ x for x in include_headers
|
|
if x.lower() in self.frozen_sign]
|
|
|
|
def all_sign_headers(self):
|
|
"""Return header list of all existing headers not in should_not_sign.
|
|
@since: 0.5"""
|
|
return [x for x,y in self.headers if x.lower() not in self.should_not_sign]
|
|
|
|
#: Sign an RFC822 message and return the DKIM-Signature header line.
|
|
#:
|
|
#: The include_headers option gives full control over which header fields
|
|
#: are signed. Note that signing a header field that doesn't exist prevents
|
|
#: that field from being added without breaking the signature. Repeated
|
|
#: fields (such as Received) can be signed multiple times. Instances
|
|
#: of the field are signed from bottom to top. Signing a header field more
|
|
#: times than are currently present prevents additional instances
|
|
#: from being added without breaking the signature.
|
|
#:
|
|
#: The length option allows the message body to be appended to by MTAs
|
|
#: enroute (e.g. mailing lists that append unsubscribe information)
|
|
#: without breaking the signature.
|
|
#:
|
|
#: The default include_headers for this method differs from the backward
|
|
#: compatible sign function, which signs all headers not
|
|
#: in should_not_sign. The default list for this method can be modified
|
|
#: by tweaking should_sign and frozen_sign (or even should_not_sign).
|
|
#: It is only necessary to pass an include_headers list when precise control
|
|
#: is needed.
|
|
#:
|
|
#: @param selector: the DKIM selector value for the signature
|
|
#: @param domain: the DKIM domain value for the signature
|
|
#: @param privkey: a PKCS#1 private key in base64-encoded text form
|
|
#: @param identity: the DKIM identity value for the signature
|
|
#: (default "@"+domain)
|
|
#: @param canonicalize: the canonicalization algorithms to use
|
|
#: (default (Simple, Simple))
|
|
#: @param include_headers: a list of strings indicating which headers
|
|
#: are to be signed (default rfc4871 recommended headers)
|
|
#: @param length: true if the l= tag should be included to indicate
|
|
#: body length signed (default False).
|
|
#: @return: DKIM-Signature header field terminated by '\r\n'
|
|
#: @raise DKIMException: when the message, include_headers, or key are badly
|
|
#: formed.
|
|
def sign(self, selector, domain, privkey, identity=None,
|
|
canonicalize=(b'relaxed',b'simple'), include_headers=None, length=False):
|
|
try:
|
|
pk = parse_pem_private_key(privkey)
|
|
except UnparsableKeyError as e:
|
|
raise KeyFormatError(str(e))
|
|
|
|
if identity is not None and not identity.endswith(domain):
|
|
raise ParameterError("identity must end with domain")
|
|
|
|
canon_policy = CanonicalizationPolicy.from_c_value(
|
|
b'/'.join(canonicalize))
|
|
headers = canon_policy.canonicalize_headers(self.headers)
|
|
|
|
if include_headers is None:
|
|
include_headers = self.default_sign_headers()
|
|
|
|
# rfc4871 says FROM is required
|
|
if 'from' not in ( x.lower() for x in include_headers ):
|
|
raise ParameterError("The From header field MUST be signed")
|
|
|
|
# raise exception for any SHOULD_NOT headers, call can modify
|
|
# SHOULD_NOT if really needed.
|
|
for x in include_headers:
|
|
if x.lower() in self.should_not_sign:
|
|
raise ParameterError("The %s header field SHOULD NOT be signed"%x)
|
|
|
|
body = canon_policy.canonicalize_body(self.body)
|
|
|
|
hasher = HASH_ALGORITHMS[self.signature_algorithm]
|
|
h = hasher()
|
|
h.update(body)
|
|
bodyhash = base64.b64encode(h.digest())
|
|
|
|
sigfields = [x for x in [
|
|
(b'v', b"1"),
|
|
(b'a', self.signature_algorithm),
|
|
(b'c', canon_policy.to_c_value()),
|
|
(b'd', domain),
|
|
(b'i', identity or b"@"+domain),
|
|
length and (b'l', len(body)),
|
|
(b'q', b"dns/txt"),
|
|
(b's', selector),
|
|
(b't', str(int(time.time())).encode('ascii')),
|
|
(b'h', b" : ".join(include_headers)),
|
|
(b'bh', bodyhash),
|
|
# Force b= to fold onto it's own line so that refolding after
|
|
# adding sig doesn't change whitespace for previous tags.
|
|
(b'b', b'0'*60),
|
|
] if x]
|
|
include_headers = [x.lower() for x in include_headers]
|
|
# record what verify should extract
|
|
self.include_headers = tuple(include_headers)
|
|
|
|
sig_value = fold(b"; ".join(b"=".join(x) for x in sigfields))
|
|
sig_value = RE_BTAG.sub(b'\\1',sig_value)
|
|
dkim_header = (b'DKIM-Signature', b' ' + sig_value)
|
|
h = hasher()
|
|
sig = dict(sigfields)
|
|
self.signed_headers = hash_headers(
|
|
h, canon_policy, headers, include_headers, dkim_header,sig)
|
|
self.logger.debug("sign headers: %r" % self.signed_headers)
|
|
|
|
try:
|
|
sig2 = RSASSA_PKCS1_v1_5_sign(h, pk)
|
|
except DigestTooLargeError:
|
|
raise ParameterError("digest too large for modulus")
|
|
# Folding b= is explicity allowed, but yahoo and live.com are broken
|
|
#sig_value += base64.b64encode(bytes(sig2))
|
|
# Instead of leaving unfolded (which lets an MTA fold it later and still
|
|
# breaks yahoo and live.com), we change the default signing mode to
|
|
# relaxed/simple (for broken receivers), and fold now.
|
|
sig_value = fold(sig_value + base64.b64encode(bytes(sig2)))
|
|
|
|
self.domain = domain
|
|
self.selector = selector
|
|
self.signature_fields = sig
|
|
return b'DKIM-Signature: ' + sig_value + b"\r\n"
|
|
|
|
#: Verify a DKIM signature.
|
|
#: @type idx: int
|
|
#: @param idx: which signature to verify. The first (topmost) signature is 0.
|
|
#: @type dnsfunc: callable
|
|
#: @param dnsfunc: an option function to lookup TXT resource records
|
|
#: for a DNS domain. The default uses dnspython or pydns.
|
|
#: @return: True if signature verifies or False otherwise
|
|
#: @raise DKIMException: when the message, signature, or key are badly formed
|
|
def verify(self,idx=0,dnsfunc=get_txt):
|
|
|
|
sigheaders = [(x,y) for x,y in self.headers if x.lower() == b"dkim-signature"]
|
|
if len(sigheaders) <= idx:
|
|
return False
|
|
|
|
# By default, we validate the first DKIM-Signature line found.
|
|
try:
|
|
sig = parse_tag_value(sigheaders[idx][1])
|
|
self.signature_fields = sig
|
|
except InvalidTagValueList as e:
|
|
raise MessageFormatError(e)
|
|
|
|
logger = self.logger
|
|
logger.debug("sig: %r" % sig)
|
|
|
|
validate_signature_fields(sig)
|
|
self.domain = sig[b'd']
|
|
self.selector = sig[b's']
|
|
|
|
try:
|
|
canon_policy = CanonicalizationPolicy.from_c_value(sig.get(b'c'))
|
|
except InvalidCanonicalizationPolicyError as e:
|
|
raise MessageFormatError("invalid c= value: %s" % e.args[0])
|
|
headers = canon_policy.canonicalize_headers(self.headers)
|
|
body = canon_policy.canonicalize_body(self.body)
|
|
|
|
try:
|
|
hasher = HASH_ALGORITHMS[sig[b'a']]
|
|
except KeyError as e:
|
|
raise MessageFormatError("unknown signature algorithm: %s" % e.args[0])
|
|
|
|
if b'l' in sig:
|
|
body = body[:int(sig[b'l'])]
|
|
|
|
h = hasher()
|
|
h.update(body)
|
|
bodyhash = h.digest()
|
|
logger.debug("bh: %s" % base64.b64encode(bodyhash))
|
|
try:
|
|
bh = base64.b64decode(re.sub(br"\s+", b"", sig[b'bh']))
|
|
except TypeError as e:
|
|
raise MessageFormatError(str(e))
|
|
if bodyhash != bh:
|
|
raise ValidationError(
|
|
"body hash mismatch (got %s, expected %s)" %
|
|
(base64.b64encode(bodyhash), sig[b'bh']))
|
|
|
|
name = sig[b's'] + b"._domainkey." + sig[b'd'] + b"."
|
|
s = dnsfunc(name)
|
|
if not s:
|
|
raise KeyFormatError("missing public key: %s"%name)
|
|
try:
|
|
pub = parse_tag_value(s)
|
|
except InvalidTagValueList as e:
|
|
raise KeyFormatError(e)
|
|
try:
|
|
pk = parse_public_key(base64.b64decode(pub[b'p']))
|
|
self.keysize = bitsize(pk['modulus'])
|
|
except KeyError:
|
|
raise KeyFormatError("incomplete public key: %s" % s)
|
|
except (TypeError,UnparsableKeyError) as e:
|
|
raise KeyFormatError("could not parse public key (%s): %s" % (pub[b'p'],e))
|
|
include_headers = [x.lower() for x in re.split(br"\s*:\s*", sig[b'h'])]
|
|
self.include_headers = tuple(include_headers)
|
|
# address bug#644046 by including any additional From header
|
|
# fields when verifying. Since there should be only one From header,
|
|
# this shouldn't break any legitimate messages. This could be
|
|
# generalized to check for extras of other singleton headers.
|
|
if 'from' in include_headers:
|
|
include_headers.append('from')
|
|
h = hasher()
|
|
self.signed_headers = hash_headers(
|
|
h, canon_policy, headers, include_headers, sigheaders[idx], sig)
|
|
try:
|
|
signature = base64.b64decode(re.sub(br"\s+", b"", sig[b'b']))
|
|
res = RSASSA_PKCS1_v1_5_verify(h, signature, pk)
|
|
if res and self.keysize < self.minkey:
|
|
raise KeyFormatError("public key too small: %d" % self.keysize)
|
|
return res
|
|
except (TypeError,DigestTooLargeError) as e:
|
|
raise KeyFormatError("digest too large for modulus: %s"%e)
|
|
|
|
def sign(message, selector, domain, privkey, identity=None,
|
|
canonicalize=(b'relaxed', b'simple'),
|
|
signature_algorithm=b'rsa-sha256',
|
|
include_headers=None, length=False, logger=None):
|
|
"""Sign an RFC822 message and return the DKIM-Signature header line.
|
|
@param message: an RFC822 formatted message (with either \\n or \\r\\n line endings)
|
|
@param selector: the DKIM selector value for the signature
|
|
@param domain: the DKIM domain value for the signature
|
|
@param privkey: a PKCS#1 private key in base64-encoded text form
|
|
@param identity: the DKIM identity value for the signature (default "@"+domain)
|
|
@param canonicalize: the canonicalization algorithms to use (default (Simple, Simple))
|
|
@param include_headers: a list of strings indicating which headers are to be signed (default all headers not listed as SHOULD NOT sign)
|
|
@param length: true if the l= tag should be included to indicate body length (default False)
|
|
@param logger: a logger to which debug info will be written (default None)
|
|
@return: DKIM-Signature header field terminated by \\r\\n
|
|
@raise DKIMException: when the message, include_headers, or key are badly formed.
|
|
"""
|
|
|
|
d = DKIM(message,logger=logger)
|
|
if not include_headers:
|
|
include_headers = d.default_sign_headers()
|
|
return d.sign(selector, domain, privkey, identity=identity, canonicalize=canonicalize, include_headers=include_headers, length=length)
|
|
|
|
def verify(message, logger=None, dnsfunc=get_txt, minkey=1024):
|
|
"""Verify the first (topmost) DKIM signature on an RFC822 formatted message.
|
|
@param message: an RFC822 formatted message (with either \\n or \\r\\n line endings)
|
|
@param logger: a logger to which debug info will be written (default None)
|
|
@return: True if signature verifies or False otherwise
|
|
"""
|
|
d = DKIM(message,logger=logger,minkey=minkey)
|
|
try:
|
|
return d.verify(dnsfunc=dnsfunc)
|
|
except DKIMException as x:
|
|
if logger is not None:
|
|
logger.error("%s" % x)
|
|
return False
|