Skip to content
Snippets Groups Projects
Commit 2fe8fa88 authored by Stuart Gathman's avatar Stuart Gathman
Browse files

Use latest pyspf verbatim. Will depend on package when pyspf-2.0 is packaged.

parent e0f58cce
Branches
Tags
No related merge requests found
...@@ -2,8 +2,8 @@ ...@@ -2,8 +2,8 @@
"""SPF (Sender Policy Framework) implementation. """SPF (Sender Policy Framework) implementation.
Copyright (c) 2003, Terence Way Copyright (c) 2003, Terence Way
Portions Copyright (c) 2004,2005 Stuart Gathman <stuart@bmsi.com> Portions Copyright (c) 2004,2005,2006 Stuart Gathman <stuart@bmsi.com>
Portions Copyright (c) 2005 Scott Kitterman <scott@kitterman.com> Portions Copyright (c) 2005,2006 Scott Kitterman <scott@kitterman.com>
This module is free software, and you may redistribute it and/or modify This module is free software, and you may redistribute it and/or modify
it under the same terms as Python itself, so long as this copyright message it under the same terms as Python itself, so long as this copyright message
and disclaimer are retained in their original form. and disclaimer are retained in their original form.
...@@ -44,249 +44,105 @@ For news, bugfixes, etc. visit the home page for this implementation at ...@@ -44,249 +44,105 @@ For news, bugfixes, etc. visit the home page for this implementation at
# 18-dec-2003, v1.6, Failures on Intel hardware: endianness. Use ! on # 18-dec-2003, v1.6, Failures on Intel hardware: endianness. Use ! on
# struct.pack(), struct.unpack(). # struct.pack(), struct.unpack().
# #
# Development taken over by Stuart Gathman <stuart@bmsi.com> since # Development taken over by Stuart Gathman <stuart@bmsi.com>.
# Terrence is not responding to email.
# #
# $Log$ # $Log$
# Revision 1.25 2006/07/31 15:25:39 customdesigned # Revision 1.96 2006/10/01 01:27:54 customdesigned
# Permerror for multiple TXT SPF records. # Switch to pymilter lax processing convention:
# Always return strict result, extended result in q.perm_error.ext
# #
# Revision 1.24 2006/07/28 01:21:33 customdesigned # Revision 1.95 2006/09/30 22:53:44 customdesigned
# Remove debug print # Fix getp to obey SHOULDs in RFC.
# #
# Revision 1.23 2006/07/28 01:21:02 customdesigned # Revision 1.94 2006/09/30 22:23:25 customdesigned
# More fixes from pyspf # p macro tests and fixes
# #
# Revision 1.22 2006/06/21 21:13:07 customdesigned # Revision 1.93 2006/09/30 20:57:06 customdesigned
# initialize perm_error # Remove generator expression for compatibility with python2.3.
# #
# Revision 1.21 2006/05/12 16:15:20 customdesigned # Revision 1.92 2006/09/30 19:52:52 customdesigned
# a:1.2.3.4 -> ip4:1.2.3.4 'lax' heuristic. # Removed redundant flag and unneeded global.
# #
# Revision 1.20 2006/03/21 18:48:51 customdesigned # Revision 1.91 2006/09/30 19:37:49 customdesigned
# Import note_error from pyspf. Handle timeout on type99 lookup # Missing L
# specially (sender actually has no SPF record and a braindead DNS server).
# #
# Revision 1.19 2006/02/24 02:12:54 customdesigned # Revision 1.90 2006/09/30 19:29:58 customdesigned
# Properly report hard PermError (lax mode fails also) by always setting # pydns returns AAAA RR as binary string
# perm_error attribute with PermError exception. Improve reporting of
# invalid domain PermError.
# #
# Revision 1.18 2005/12/29 19:15:35 customdesigned # Revision 1.89 2006/09/29 20:23:11 customdesigned
# Handle NULL MX # Optimize cidrmatch
# #
# Revision 1.17 2005/12/23 21:44:15 customdesigned # Revision 1.88 2006/09/29 19:44:10 customdesigned
# Always include keyword data in Received-SPF header. # Fix ptr with ip6 for harsh mode.
# #
# Revision 1.16 2005/12/01 22:42:32 customdesigned # Revision 1.87 2006/09/29 19:26:53 customdesigned
# improve gossip support. # Add PTR tests and fix ip6 ptr
# Initialize srs_domain from srs.srs config property. Should probably
# always block unsigned DSN when signing all.
# #
# Revision 1.15 2005/10/30 01:08:14 customdesigned # Revision 1.86 2006/09/29 17:55:22 customdesigned
# Ignore records missing spaces. # Pass ip6 tests
# #
# Revision 1.14 2005/08/12 17:36:51 customdesigned # Revision 1.85 2006/09/29 15:58:02 customdesigned
# Trean non-existant include as no match in "lax" mode. # Pass self test on non IP6 python.
# PTR accepts no cidr.
# #
# Revision 1.13 2005/07/22 16:00:23 customdesigned # Revision 1.83 2006/09/27 18:09:40 kitterma
# Limit CNAME chains independently of DNS lookup limit # Converted spf.check to return pre-MARID result codes for drop in
# compatibility with pySPF 1.6/1.7. Added new procedure, spf.check2 to
# return RFC4408 results in a two part answer (result, explanation).
# This is the external API for pySPF 2.0. No longer any need to branch
# for 'classic' and RFC compliant pySPF libraries.
# #
# Revision 1.31 2005/07/22 02:11:50 customdesigned # Revision 1.82 2006/09/27 18:02:21 kitterma
# Use dictionary to check for CNAME loops. Check limit independently for # Converted max MX limit to ambiguity warning for validator.
# each top level name, just like for PTR.
# #
# Revision 1.30 2005/07/21 20:07:31 customdesigned # Revision 1.81 2006/09/27 17:38:14 kitterma
# Translate DNS error in DNSLookup. This completely isolates DNS # Updated initial comments and moved pre-1.7 changes to spf_changelog.
# dependencies to the DNSLookup method.
# #
# Revision 1.29 2005/07/21 17:49:39 customdesigned # Revision 1.80 2006/09/27 17:33:53 kitterma
# My best guess at what RFC intended for limiting CNAME loops. # Fixed indentation error in check0.
# #
# Revision 1.28 2005/07/21 17:37:08 customdesigned # Revision 1.79 2006/09/26 18:05:44 kitterma
# Break out external DNSLookup method so that test suite can # Removed unused receiver policy definitions.
# duplicate CNAME loop bug. Test zone data dictionary now
# mirrors structure of real DNS.
# #
# Revision 1.27 2005/07/21 15:26:06 customdesigned # Revision 1.78 2006/09/26 16:15:50 kitterma
# First cut at updating docs. Test suite is obsolete. # added additional IP4 and CIDR validation tests - no code changes.
# #
# Revision 1.26 2005/07/20 03:12:40 customdesigned # Revision 1.77 2006/09/25 19:42:32 customdesigned
# When not in strict mode, don't give PermErr for bad mechanism until # Fix unknown macro sentinel
# encountered during evaluation.
# #
# Revision 1.25 2005/07/19 23:24:42 customdesigned # Revision 1.76 2006/09/25 19:10:40 customdesigned
# Validate all mechanisms before evaluating. # Fix exp= error and add another failing test.
# #
# Revision 1.24 2005/07/19 18:11:52 kitterma # Revision 1.75 2006/09/25 02:02:30 kitterma
# Fix to change that compares type TXT and type SPF records. Bug in the change # Fixed redirect-cancels-exp test suite failure.
# prevented records from being returned if it was published as TXT, but not SPF.
# #
# Revision 1.23 2005/07/19 15:22:50 customdesigned # Revision 1.74 2006/09/24 04:04:08 kitterma
# MX and PTR limits are MUST NOT check limits, and do not result in PermErr. # Implemented check for macro 'c' - Macro unimplimented.
# Also, check belongs in mx and ptr specific methods, not in dns() method.
# #
# Revision 1.22 2005/07/19 05:02:29 customdesigned # Revision 1.73 2006/09/24 02:08:35 kitterma
# FQDN test was broken. Added test case. Move FQDN test to after # Fixed invalid-macro-char test failure.
# macro expansion.
# #
# Revision 1.21 2005/07/18 20:46:27 kitterma # Revision 1.72 2006/09/23 05:45:52 kitterma
# Fixed reference problem in 1.20 # Fixed domain-name-truncation test failure
# #
# Revision 1.20 2005/07/18 20:21:47 kitterma # Revision 1.71 2006/09/22 01:02:54 kitterma
# Change to dns_spf to go ahead and check for a type 99 (SPF) record even if a # pySPF correction for nolocalpart in rfc4408-tests.yml failed, 4.3/2.
# TXT record is found and make sure if type SPF is present that they are # Added comments to testspf.py on where to get YAML.
# identical when using strict processing.
# #
# Revision 1.19 2005/07/18 19:36:00 kitterma # Revision 1.70 2006/09/18 02:13:27 kitterma
# Change to require at least one dot in a domain name. Added PermError # Worked through a large number of pylint issues - all 4 spaces, not a mix
# description to indicate FQDN should be used. This is a common error. # of 4 spaces, 2 spaces, and tabs. Caught a few minor errors in the process.
# All built in tests still pass.
# #
# Revision 1.18 2005/07/18 17:13:37 kitterma # Revision 1.69 2006/09/17 18:44:25 kitterma
# Change macro processing to raise PermError on an unknown macro. # Fixed validation mode only crash bug when rDNS check had no PTR record
# schlitt-spf-classic-02 para 8.1. Change exp modifier processing to ignore
# exp strings with syntax errors. schlitt-spf-classic-02 para 6.2.
# #
# Revision 1.17 2005/07/18 14:35:34 customdesigned
# Remove debugging printf
#
# Revision 1.16 2005/07/18 14:34:14 customdesigned
# Forgot to remove debugging print
#
# Revision 1.15 2005/07/15 21:17:36 customdesigned
# Recursion limit raises AssertionError in strict mode, PermError otherwise.
#
# Revision 1.14 2005/07/15 20:34:11 customdesigned
# Check whether DNS package already supports SPF before patching
#
# Revision 1.13 2005/07/15 20:01:22 customdesigned
# Allow extended results for MX limit
#
# Revision 1.12 2005/07/15 19:12:09 customdesigned
# Official IANA SPF record (type 99) support.
#
# Revision 1.11 2005/07/15 18:03:02 customdesigned
# Fix unknown Received-SPF header broken by result changes
#
# Revision 1.10 2005/07/15 16:17:05 customdesigned
# Start type99 support.
# Make Scott's "/" support in parse_mechanism more elegant as requested.
# Add test case for "/" support.
#
# Revision 1.9 2005/07/15 03:33:14 kitterma
# Fix for bug 1238403 - Crash if non-CIDR / present. Also added
# validation check for valid IPv4 CIDR range.
#
# Revision 1.8 2005/07/14 04:18:01 customdesigned
# Bring explanations and Received-SPF header into line with
# the unknown=PermErr and error=TempErr convention.
# Hope my case-sensitive mech fix doesn't clash with Scotts.
#
# Revision 1.7 2005/07/12 21:43:56 kitterma
# Added processing to clarify some cases of unknown
# qualifier errors (to distinguish between unknown qualifier and
# unknown mechanism).
# Also cleaned up comments from previous updates.
#
# Revision 1.6 2005/06/29 14:46:26 customdesigned
# Distinguish trivial recursion from missing arg for diagnostic purposes.
#
# Revision 1.5 2005/06/28 17:48:56 customdesigned
# Support extended processing results when a PermError should strictly occur.
#
# Revision 1.4 2005/06/22 15:54:54 customdesigned
# Correct spelling.
#
# Revision 1.3 2005/06/22 00:08:24 kitterma
# Changes from draft-mengwong overall DNS lookup and recursion
# depth limits to draft-schlitt-spf-classic-02 DNS lookup, MX lookup, and
# PTR lookup limits. Recursion code is still present and functioning, but
# it should be impossible to trip it.
#
# Revision 1.2 2005/06/21 16:46:09 kitterma
# Updated definition of SPF, added reference to the sourceforge project site,
# and deleted obsolete Microsoft Caller ID for Email XML translation routine.
#
# Revision 1.1.1.1 2005/06/20 19:57:32 customdesigned
# Move Python SPF to its own module.
#
# Revision 1.5 2005/06/14 20:31:26 customdesigned
# fix pychecker nits
#
# Revision 1.4 2005/06/02 04:18:55 customdesigned
# Update copyright notices after reading article on /.
#
# Revision 1.3 2005/06/02 02:08:12 customdesigned
# Reject on PermErr
#
# Revision 1.2 2005/05/31 18:57:59 customdesigned
# Clear unknown mechanism list at proper time.
#
# Revision 1.24 2005/03/16 21:58:39 stuart
# Change Milter module to package.
#
# Revision 1.22 2005/02/09 17:52:59 stuart
# Report DNS errors as PermError rather than unknown.
#
# Revision 1.21 2004/11/20 16:37:03 stuart
# Handle multi-segment TXT records.
#
# Revision 1.20 2004/11/19 06:10:30 stuart
# Use PermError exception instead of reporting unknown.
#
# Revision 1.19 2004/11/09 23:00:18 stuart
# Limit recursion and DNS lookups separately.
#
#
# Revision 1.17 2004/09/10 18:08:26 stuart
# Return unknown for null mechanism
#
# Revision 1.16 2004/09/04 23:27:06 stuart
# More mechanism aliases.
#
# Revision 1.15 2004/08/30 21:19:05 stuart
# Return unknown for invalid ip syntax in mechanism
#
# Revision 1.14 2004/08/23 02:28:24 stuart
# Remove Perl usage message.
#
# Revision 1.13 2004/07/23 19:23:12 stuart
# Always fail to match on ip6, until we support it properly.
#
# Revision 1.12 2004/07/23 18:48:15 stuart
# Fold CID parsing into spf
#
# Revision 1.11 2004/07/21 21:32:01 stuart
# Handle CID records (Microsoft XML format).
#
# Revision 1.10 2004/04/19 22:12:11 stuart
# Release 0.6.9
#
# Revision 1.9 2004/04/18 03:29:35 stuart
# Pass most tests except -local and -rcpt-to
#
# Revision 1.8 2004/04/17 22:17:55 stuart
# Header comment method.
#
# Revision 1.7 2004/04/17 18:22:48 stuart
# Support default explanation.
#
# Revision 1.6 2004/04/06 20:18:02 stuart
# Fix bug in include
#
# Revision 1.5 2004/04/05 22:29:46 stuart
# SPF best_guess,
#
# Revision 1.4 2004/03/25 03:27:34 stuart
# Support delegation of SPF records.
#
# Revision 1.3 2004/03/13 12:23:23 stuart
# Expanded result codes. Tolerate common method misspellings.
# #
# See spf_changelog.txt for earlier changes.
__author__ = "Terence Way" __author__ = "Terence Way"
__email__ = "terry@wayforward.net" __email__ = "terry@wayforward.net"
__version__ = "1.7: July 26, 2006" __version__ = "1.7: July 22, 2005"
MODULE = 'spf' MODULE = 'spf'
USAGE = """To check an incoming mail request: USAGE = """To check an incoming mail request:
...@@ -323,6 +179,9 @@ def DNSLookup(name,qtype): ...@@ -323,6 +179,9 @@ def DNSLookup(name,qtype):
resp = req.req() resp = req.req()
#resp.show() #resp.show()
# key k: ('wayforward.net', 'A'), value v # key k: ('wayforward.net', 'A'), value v
# FIXME: pydns returns AAAA RR as 16 byte binary string, but
# A RR as dotted quad. For consistency, this driver should
# return both as binary string.
return [((a['name'], a['typename']), a['data']) for a in resp.answers] return [((a['name'], a['typename']), a['data']) for a in resp.answers]
except IOError, x: except IOError, x:
raise TempError, 'DNS ' + str(x) raise TempError, 'DNS ' + str(x)
...@@ -333,14 +192,11 @@ def isSPF(txt): ...@@ -333,14 +192,11 @@ def isSPF(txt):
"Return True if txt has SPF record signature." "Return True if txt has SPF record signature."
return txt.startswith('v=spf1 ') or txt == 'v=spf1' return txt.startswith('v=spf1 ') or txt == 'v=spf1'
# 32-bit IPv4 address mask
MASK = 0xFFFFFFFFL
# Regular expression to look for modifiers # Regular expression to look for modifiers
RE_MODIFIER = re.compile(r'^([a-z][a-z0-9_\-\.]*)=', re.IGNORECASE) RE_MODIFIER = re.compile(r'^([a-z][a-z0-9_\-\.]*)=', re.IGNORECASE)
# Regular expression to find macro expansions # Regular expression to find macro expansions
RE_CHAR = re.compile(r'%(%|_|-|(\{[a-zA-Z][0-9]*r?[^\}]*\}))') RE_CHAR = re.compile(r'%(%|_|-|(\{[^\}]*\}))')
# Regular expression to break up a macro expansion # Regular expression to break up a macro expansion
RE_ARGS = re.compile(r'([0-9]*)(r?)([^0-9a-zA-Z]*)') RE_ARGS = re.compile(r'([0-9]*)(r?)([^0-9a-zA-Z]*)')
...@@ -393,41 +249,6 @@ EXPLANATIONS = {'pass': 'sender SPF authorized', ...@@ -393,41 +249,6 @@ EXPLANATIONS = {'pass': 'sender SPF authorized',
'ambiguous': 'No error, but results may vary' 'ambiguous': 'No error, but results may vary'
} }
#Default receiver policies - can be overridden.
POLICY = {'tfwl': False, #Check trusted-forwarder.org
'skip_localhost': True, #Don't check SPF on local connections
'always_helo': False, #Only works if helo_first is also True.
'spf_helo_mustpass': True, #Treat HELO test returning softfail or
#neutral as Fail - HELO should be a single IP per name. No reason to
#accept SPF relaxed provisions for HELO. No affect if None.
'reject_helo_fail': False,
'spf_reject_fail': True,
'spf_reject_neutral': False,
'spf_accept_softfail': True,
'spf_best_guess': True,
'spf_strict': True,
}
# Recommended SMTP codes for certain SPF results. For results not in
# this table the recommendation is to accept the message as authorized.
# An SPF result is never enough to recommend that a message be accepted for
# delivery. Additional checks are generally required.
# The softfail result requires special processing.
SMTP_CODES = {
'fail': [550,'5.7.1'],
'temperror': [451,'4.4.3'],
'permerror': [550,'5.5.2'],
'softfail': [451,'4.3.0']
}
if not POLICY['spf_accept_softfail']:
SMTP_CODES['softfail'] = (550,'5.7.1')
if POLICY['spf_reject_neutral']:
SMTP_CODES['neutral'] = (550,'5.7.1')
# if set to a domain name, search _spf.domain namespace if no SPF record
# found in source domain.
DELEGATE = None
# support pre 2.2.1.... # support pre 2.2.1....
try: try:
bool, True, False = bool, True, False bool, True, False = bool, True, False
...@@ -436,6 +257,8 @@ except NameError: ...@@ -436,6 +257,8 @@ except NameError:
def bool(x): return not not x def bool(x): return not not x
# ...pre 2.2.1 # ...pre 2.2.1
DELEGATE = None
# standard default SPF record for best_guess # standard default SPF record for best_guess
DEFAULT_SPF = 'v=spf1 a/24 mx/24 ptr' DEFAULT_SPF = 'v=spf1 a/24 mx/24 ptr'
...@@ -492,24 +315,50 @@ class PermError(Exception): ...@@ -492,24 +315,50 @@ class PermError(Exception):
return '%s: %s'%(self.msg, self.mech) return '%s: %s'%(self.msg, self.mech)
return self.msg return self.msg
def check2(i, s, h, local=None, receiver=None):
"""Test an incoming MAIL FROM:<s>, from a client with ip address i.
h is the HELO/EHLO domain name. This is the RFC4408 compliant pySPF2.0
interface. The interface returns an SPF result and explanation only.
SMTP response codes are not returned since RFC 4408 does not specify
receiver policy. Applications updated for RFC 4408 should use this
interface.
Returns (result, explanation) where result in
['pass', 'permerror', 'fail', 'temperror', 'softfail', 'none', 'neutral' ].
Example:
#>>> check2(i='61.51.192.42', s='liukebing@bcc.com', h='bmsi.com')
"""
res,_,exp = query(i=i, s=s, h=h, local=local, receiver=receiver).check()
return res,exp
def check(i, s, h, local=None, receiver=None): def check(i, s, h, local=None, receiver=None):
"""Test an incoming MAIL FROM:<s>, from a client with ip address i. """Test an incoming MAIL FROM:<s>, from a client with ip address i.
h is the HELO/EHLO domain name. h is the HELO/EHLO domain name. This is the pre-RFC SPF Classic interface.
Applications written for pySPF 1.6/1.7 can use this interface to allow
pySPF2 to be a drop in replacement for older versions. With the exception
of result codes, performance in RFC 4408 compliant.
Returns (result, code, explanation) where result in Returns (result, code, explanation) where result in
['pass', 'permerror', 'fail', 'temperror', 'softfail', 'none', 'neutral' ]. ['pass', 'unknown', 'fail', 'error', 'softfail', 'none', 'neutral' ].
Example: Example:
#>>> check(i='61.51.192.42', s='liukebing@bcc.com', h='bmsi.com') #>>> check(i='61.51.192.42', s='liukebing@bcc.com', h='bmsi.com')
""" """
return query(i=i, s=s, h=h,local=local,receiver=receiver).check() res,code,exp = query(i=i, s=s, h=h, local=local, receiver=receiver).check()
if res == 'permerror':
res = 'unknown'
elif res == 'tempfail':
res =='error'
return res, code, exp
class query(object): class query(object):
"""A query object keeps the relevant information about a single SPF """A query object keeps the relevant information about a single SPF
query: query:
i: ip address of SMTP client i: ip address of SMTP client in dotted notation
s: sender declared in MAIL FROM:<> s: sender declared in MAIL FROM:<>
l: local part of sender s l: local part of sender s
d: current domain, initially domain part of sender s d: current domain, initially domain part of sender s
...@@ -518,6 +367,8 @@ class query(object): ...@@ -518,6 +367,8 @@ class query(object):
t: current timestamp t: current timestamp
p: SMTP client domain name p: SMTP client domain name
o: domain part of sender s o: domain part of sender s
r: receiver
c: pretty ip address (different from i for IPv6)
This is also, by design, the same variables used in SPF macro This is also, by design, the same variables used in SPF macro
expansion. expansion.
...@@ -525,12 +376,11 @@ class query(object): ...@@ -525,12 +376,11 @@ class query(object):
Also keeps cache: DNS cache. Also keeps cache: DNS cache.
""" """
def __init__(self, i, s, h, local=None, receiver=None, strict=True): def __init__(self, i, s, h, local=None, receiver=None, strict=True):
self.i, self.s, self.h = i, s, h self.s, self.h = s, h
if not s and h: if not s and h:
self.s = 'postmaster@' + h self.s = 'postmaster@' + h
self.l, self.o = split_email(s, h) self.l, self.o = split_email(s, h)
self.t = str(int(time.time())) self.t = str(int(time.time()))
self.v = 'in-addr'
self.d = self.o self.d = self.o
self.p = None self.p = None
if receiver: if receiver:
...@@ -547,19 +397,64 @@ class query(object): ...@@ -547,19 +397,64 @@ class query(object):
self.lookups = 0 self.lookups = 0
# strict can be False, True, or 2 (numeric) for harsh # strict can be False, True, or 2 (numeric) for harsh
self.strict = strict self.strict = strict
self.set_ip(i)
def set_ip(self, i):
"Set connect ip, and ip6 or ip4 mode."
if RE_IP4.match(i):
self.ip = addr2bin(i)
ip6 = False
else:
assert socket.has_ipv6,"No IPv6 python support"
self.ip = bin2long6(socket.inet_pton(socket.AF_INET6, i))
if (self.ip >> 32) == 0xFFFF: # IP4 mapped address
self.ip = self.ip & 0xFFFFFFFFL
ip6 = False
else:
ip6 = True
# NOTE: self.A is not lowercase, so isn't a macro. See query.expand()
if ip6:
self.c = socket.inet_ntop(socket.AF_INET6,
struct.pack("!QQ", self.ip>>64, self.ip&0xFFFFFFFFFFFFFFFFL))
self.i = '.'.join(list('%032X'%self.ip))
self.A = 'AAAA'
self.v = 'ip6'
self.cidrmax = 128
else:
self.c = socket.inet_ntoa(struct.pack("!L", self.ip))
self.i = self.c
self.A = 'A'
self.v = 'in-addr'
self.cidrmax = 32
def set_default_explanation(self, exp): def set_default_explanation(self, exp):
exps = self.exps exps = self.exps
defexps = self.defexps
for i in 'softfail', 'fail', 'permerror': for i in 'softfail', 'fail', 'permerror':
exps[i] = exp exps[i] = exp
defexps[i] = exp
def set_explanation(self, exp):
exps = self.exps
for i in 'softfail', 'fail', 'permerror':
exps[i] = exp
# Compute p macro only if needed
def getp(self): def getp(self):
if not self.p: if not self.p:
p = self.dns_ptr(self.i) p = self.validated_ptrs()
if len(p) > 0: if not p:
self.p = p[0] self.p = "unknown"
elif self.d in p:
self.p = self.d
else:
sfx = '.' + self.d
for d in p:
if d.endswith(sfx):
self.p = d
break
else: else:
self.p = self.i self.p = p[0]
return self.p return self.p
def best_guess(self, spf=DEFAULT_SPF): def best_guess(self, spf=DEFAULT_SPF):
...@@ -578,6 +473,9 @@ class query(object): ...@@ -578,6 +473,9 @@ class query(object):
>>> q.check(spf='v=spf1 ?all') >>> q.check(spf='v=spf1 ?all')
('neutral', 250, 'access neither permitted nor denied') ('neutral', 250, 'access neither permitted nor denied')
>>> q.check(spf='v=spf1 redirect=controlledmail.com exp=_exp.controlledmail.com')
('fail', 550, 'SPF fail - not authorized')
>>> q.check(spf='v=spf1 ip4:192.0.0.0/8 ?all moo') >>> q.check(spf='v=spf1 ip4:192.0.0.0/8 ?all moo')
('permerror', 550, 'SPF Permanent Error: Unknown mechanism found: moo') ('permerror', 550, 'SPF Permanent Error: Unknown mechanism found: moo')
...@@ -595,8 +493,11 @@ class query(object): ...@@ -595,8 +493,11 @@ class query(object):
>>> q.strict = False >>> q.strict = False
>>> q.check(spf='v=spf1 ip4:192.0.0.0/8 -all moo') >>> q.check(spf='v=spf1 ip4:192.0.0.0/8 -all moo')
('permerror', 550, 'SPF Permanent Error: Unknown mechanism found: moo')
>>> q.perm_error.ext
('pass', 250, 'sender SPF authorized') ('pass', 250, 'sender SPF authorized')
>>> q.strict = True
>>> q.check(spf='v=spf1 ip4:192.1.0.0/16 moo -all') >>> q.check(spf='v=spf1 ip4:192.1.0.0/16 moo -all')
('permerror', 550, 'SPF Permanent Error: Unknown mechanism found: moo') ('permerror', 550, 'SPF Permanent Error: Unknown mechanism found: moo')
...@@ -615,6 +516,12 @@ class query(object): ...@@ -615,6 +516,12 @@ class query(object):
>>> q.libspf_local='ip4:192.0.2.3 a:example.org' >>> q.libspf_local='ip4:192.0.2.3 a:example.org'
>>> q.check(spf='v=spf1 ip4:1.2.3.4 -a:example.net -all') >>> q.check(spf='v=spf1 ip4:1.2.3.4 -a:example.net -all')
('pass', 250, 'sender SPF authorized') ('pass', 250, 'sender SPF authorized')
>>> q.check(spf='v=spf1 ip4:1.2.3.4 -all exp=_exp.controlledmail.com')
('fail', 550, 'Controlledmail.com does not send mail from itself.')
>>> q.check(spf='v=spf1 ip4:1.2.3.4 ?all exp=_exp.controlledmail.com')
('neutral', 250, 'access neither permitted nor denied')
""" """
self.mech = [] # unknown mechanisms self.mech = [] # unknown mechanisms
# If not strict, certain PermErrors (mispelled # If not strict, certain PermErrors (mispelled
...@@ -622,6 +529,7 @@ class query(object): ...@@ -622,6 +529,7 @@ class query(object):
# will continue processing. However, the exception # will continue processing. However, the exception
# that strict processing would raise is saved here # that strict processing would raise is saved here
self.perm_error = None self.perm_error = None
self.result = None
try: try:
self.lookups = 0 self.lookups = 0
...@@ -630,7 +538,14 @@ class query(object): ...@@ -630,7 +538,14 @@ class query(object):
if self.libspf_local and spf: if self.libspf_local and spf:
spf = insert_libspf_local_policy( spf = insert_libspf_local_policy(
spf, self.libspf_local) spf, self.libspf_local)
return self.check1(spf, self.d, 0) rc = self.check1(spf, self.d, 0)
self.result = rc[0]
if self.perm_error:
# lax processing encountered a permerror, but continued
self.perm_error.ext = rc
raise self.perm_error
return rc
except TempError, x: except TempError, x:
self.prob = x.msg self.prob = x.msg
if x.mech: if x.mech:
...@@ -680,8 +595,7 @@ class query(object): ...@@ -680,8 +595,7 @@ class query(object):
try: try:
raise PermError(*msg) raise PermError(*msg)
except PermError, x: except PermError, x:
# FIXME: keep a list of errors for even friendlier # FIXME: keep a list of errors for even friendlier diagnostics.
# diagnostics.
self.perm_error = x self.perm_error = x
return self.perm_error return self.perm_error
...@@ -707,6 +621,10 @@ class query(object): ...@@ -707,6 +621,10 @@ class query(object):
... except PermError,x: print x ... except PermError,x: print x
Invalid IP4 CIDR length: ip4:1.2.3.4/247 Invalid IP4 CIDR length: ip4:1.2.3.4/247
>>> try: q.validate_mechanism('ip4:1.2.3.4/33')
... except PermError,x: print x
Invalid IP4 CIDR length: ip4:1.2.3.4/33
>>> try: q.validate_mechanism('a:example.com:8080') >>> try: q.validate_mechanism('a:example.com:8080')
... except PermError,x: print x ... except PermError,x: print x
Invalid domain found (use FQDN): example.com:8080 Invalid domain found (use FQDN): example.com:8080
...@@ -715,6 +633,10 @@ class query(object): ...@@ -715,6 +633,10 @@ class query(object):
... except PermError,x: print x ... except PermError,x: print x
Invalid IP4 address: ip4:1.2.3.444/24 Invalid IP4 address: ip4:1.2.3.444/24
>>> try: q.validate_mechanism('ip4:1.2.03.4/24')
... except PermError,x: print x
Invalid IP4 address: ip4:1.2.03.4/24
>>> try: q.validate_mechanism('-all:3030') >>> try: q.validate_mechanism('-all:3030')
... except PermError,x: print x ... except PermError,x: print x
Invalid all mechanism format - only qualifier allowed with all: -all:3030 Invalid all mechanism format - only qualifier allowed with all: -all:3030
...@@ -750,7 +672,7 @@ class query(object): ...@@ -750,7 +672,7 @@ class query(object):
# validate cidr and dual-cidr # validate cidr and dual-cidr
if m in ('a', 'mx', 'ptr'): if m in ('a', 'mx'):
if cidrlength is None: if cidrlength is None:
cidrlength = 32; cidrlength = 32;
elif cidrlength > 32: elif cidrlength > 32:
...@@ -759,6 +681,8 @@ class query(object): ...@@ -759,6 +681,8 @@ class query(object):
cidr6length = 128 cidr6length = 128
elif cidr6length > 128: elif cidr6length > 128:
raise PermError('Invalid IP6 CIDR length', mech) raise PermError('Invalid IP6 CIDR length', mech)
if self.v == 'ip6':
cidrlength = cidr6length
elif m == 'ip4': elif m == 'ip4':
if cidr6length is not None: if cidr6length is not None:
raise PermError('Dual CIDR not allowed', mech) raise PermError('Dual CIDR not allowed', mech)
...@@ -779,8 +703,8 @@ class query(object): ...@@ -779,8 +703,8 @@ class query(object):
raise PermError('Invalid IP6 address', mech) raise PermError('Invalid IP6 address', mech)
else: else:
if cidrlength is not None or cidr6length is not None: if cidrlength is not None or cidr6length is not None:
raise PermError('Dual CIDR not allowed',mech) raise PermError('CIDR not allowed', mech)
cidrlength = 32 cidrlength = self.cidrmax
# validate domain-spec # validate domain-spec
if m in ('a', 'mx', 'ptr', 'exists', 'include'): if m in ('a', 'mx', 'ptr', 'exists', 'include'):
...@@ -850,18 +774,22 @@ class query(object): ...@@ -850,18 +774,22 @@ class query(object):
continue continue
if m[0] == 'exp': if m[0] == 'exp':
try: # always fetch explanation to check permerrors
self.set_default_explanation(self.get_explanation(m[1])) exp = self.get_explanation(m[1])
except PermError: if not recursion:
pass # only set explanation in base recursion level
self.set_explanation(exp)
elif m[0] == 'redirect': elif m[0] == 'redirect':
self.check_lookups() self.check_lookups()
redirect = self.expand(m[1]) redirect = self.expand(m[1])
elif m[0] == 'default': elif m[0] == 'default':
arg = self.expand(m[1])
# default=- is the same as default=fail # default=- is the same as default=fail
default = RESULTS.get(m[1], default) default = RESULTS.get(arg, default)
else:
# spf rfc: 3.6 Unrecognized Mechanisms and Modifiers # spf rfc: 3.6 Unrecognized Mechanisms and Modifiers
self.expand(m[1]) # syntax error on invalid macro
# Evaluate mechanisms # Evaluate mechanisms
# #
...@@ -885,7 +813,7 @@ class query(object): ...@@ -885,7 +813,7 @@ class query(object):
elif m == 'exists': elif m == 'exists':
self.check_lookups() self.check_lookups()
try: try:
if len(self.dns_a(arg)) > 0: if len(self.dns_a(arg,'A')) > 0:
break break
except AmbiguityWarning: except AmbiguityWarning:
# Exists wants no response sometimes so don't raise # Exists wants no response sometimes so don't raise
...@@ -894,39 +822,34 @@ class query(object): ...@@ -894,39 +822,34 @@ class query(object):
elif m == 'a': elif m == 'a':
self.check_lookups() self.check_lookups()
if cidrmatch(self.i, self.dns_a(arg), cidrlength): if self.cidrmatch(self.dns_a(arg,self.A), cidrlength):
break break
elif m == 'mx': elif m == 'mx':
self.check_lookups() self.check_lookups()
if cidrmatch(self.i, self.dns_mx(arg), cidrlength): if self.cidrmatch(self.dns_mx(arg), cidrlength):
break break
elif m == 'ip4': elif m == 'ip4':
if self.v == 'in-addr': # match own connection type only
if arg == self.d:
raise PermError('Missing IP4 arg',mech)
try: try:
if cidrmatch(self.i, [arg], cidrlength): if self.cidrmatch([arg], cidrlength): break
break
except socket.error: except socket.error:
raise PermError('syntax error', mech) raise PermError('syntax error', mech)
elif m == 'ip6': elif m == 'ip6':
if arg == self.d: if self.v == 'ip6': # match own connection type only
raise PermError('Missing IP6 arg',mech) try:
# Until we support IPV6, we should never arg = socket.inet_pton(socket.AF_INET6,arg)
# get an IPv6 connection. So this mech if self.cidrmatch([arg], cidrlength): break
# will never match. except socket.error:
pass raise PermError('syntax error', mech)
elif m == 'ptr': elif m == 'ptr':
self.check_lookups() self.check_lookups()
if domainmatch(self.validated_ptrs(self.i), arg): if domainmatch(self.validated_ptrs(), arg):
break break
else:
raise result
else: else:
# no matches # no matches
if redirect: if redirect:
...@@ -935,8 +858,8 @@ class query(object): ...@@ -935,8 +858,8 @@ class query(object):
if not redirect_record: if not redirect_record:
raise PermError('redirect domain has no SPF record', raise PermError('redirect domain has no SPF record',
redirect) redirect)
return self.check1(redirect_record, redirect, self.exps = dict(self.defexps)
recursion + 1) return self.check1(redirect_record, redirect, recursion)
else: else:
result = default result = default
...@@ -967,6 +890,7 @@ class query(object): ...@@ -967,6 +890,7 @@ class query(object):
>>> q = query(s='strong-bad@email.example.com', >>> q = query(s='strong-bad@email.example.com',
... h='mx.example.org', i='192.0.2.3') ... h='mx.example.org', i='192.0.2.3')
>>> q.p = 'mx.example.org' >>> q.p = 'mx.example.org'
>>> q.r = 'example.net'
>>> q.expand('%{d}') >>> q.expand('%{d}')
'email.example.com' 'email.example.com'
...@@ -1010,6 +934,12 @@ class query(object): ...@@ -1010,6 +934,12 @@ class query(object):
>>> q.expand('%{l1r-}') >>> q.expand('%{l1r-}')
'strong' 'strong'
>>> q.expand('%{c}',stripdot=False)
'192.0.2.3'
>>> q.expand('%{r}',stripdot=False)
'example.net'
>>> q.expand('%{ir}.%{v}._spf.%{d2}') >>> q.expand('%{ir}.%{v}._spf.%{d2}')
'3.2.0.192.in-addr._spf.example.com' '3.2.0.192.in-addr._spf.example.com'
...@@ -1022,15 +952,39 @@ class query(object): ...@@ -1022,15 +952,39 @@ class query(object):
>>> q.expand('%{ir}.%{v}.%{l1r-}.lp._spf.%{d2}') >>> q.expand('%{ir}.%{v}.%{l1r-}.lp._spf.%{d2}')
'3.2.0.192.in-addr.strong.lp._spf.example.com' '3.2.0.192.in-addr.strong.lp._spf.example.com'
>>> try: q.expand('%(ir).%{v}.%{l1r-}.lp._spf.%{d2}')
... except PermError,x: print x
invalid-macro-char : %(ir)
>>> q.expand('%{p2}.trusted-domains.example.net') >>> q.expand('%{p2}.trusted-domains.example.net')
'example.org.trusted-domains.example.net' 'example.org.trusted-domains.example.net'
>>> q.expand('%{p2}.trusted-domains.example.net.') >>> q.expand('%{p2}.trusted-domains.example.net.')
'example.org.trusted-domains.example.net' 'example.org.trusted-domains.example.net'
>>> q = query(s='@email.example.com',
... h='mx.example.org', i='192.0.2.3')
>>> q.p = 'mx.example.org'
>>> q.expand('%{l}')
'postmaster'
""" """
macro_delimiters = ['{', '%', '-', '_']
end = 0 end = 0
result = '' result = ''
macro_count = str.count('%')
if macro_count != 0:
labels = str.split('.')
for label in labels:
is_macro = False
if len(label) > 1:
if label[0] == '%':
for delimit in macro_delimiters:
if label[1] == delimit:
is_macro = True
if not is_macro:
raise PermError ('invalid-macro-char ', label)
break
for i in RE_CHAR.finditer(str): for i in RE_CHAR.finditer(str):
result += str[end:i.start()] result += str[end:i.start()]
macro = str[i.start():i.end()] macro = str[i.start():i.end()]
...@@ -1042,20 +996,26 @@ class query(object): ...@@ -1042,20 +996,26 @@ class query(object):
result += '%20' result += '%20'
else: else:
letter = macro[2].lower() letter = macro[2].lower()
# print letter
if letter == 'p': if letter == 'p':
self.getp() self.getp()
expansion = getattr(self, letter, 'Macro Error') elif letter in 'crt' and stripdot:
raise PermError(
'c,r,t macros allowed in exp= text only', macro)
expansion = getattr(self, letter, self)
if expansion: if expansion:
if expansion == 'Macro Error': if expansion == self:
raise PermError('Unknown Macro Encountered') raise PermError('Unknown Macro Encountered', macro)
result += expand_one(expansion, result += expand_one(expansion, macro[3:-1],
macro[3:-1],
JOINERS.get(letter)) JOINERS.get(letter))
end = i.end() end = i.end()
result += str[end:] result += str[end:]
if stripdot and result.endswith('.'): if stripdot and result.endswith('.'):
return result[:-1] result = result[:-1]
if result.count('.') != 0:
if len(result) > 253:
result = result[(result.index('.')+1):]
return result return result
def dns_spf(self, domain): def dns_spf(self, domain):
...@@ -1074,7 +1034,7 @@ class query(object): ...@@ -1074,7 +1034,7 @@ class query(object):
b = [t for t in self.dns_99(domain) if isSPF(t)] b = [t for t in self.dns_99(domain) if isSPF(t)]
except TempError,x: except TempError,x:
# some braindead DNS servers hang on type 99 query # some braindead DNS servers hang on type 99 query
if self.strict > 1: raise x if self.strict > 1: raise TempError(x)
b = [] b = []
if len(b) > 1: if len(b) > 1:
...@@ -1113,61 +1073,60 @@ class query(object): ...@@ -1113,61 +1073,60 @@ class query(object):
# RFC 4408 section 5.4 "mx" # RFC 4408 section 5.4 "mx"
# To prevent DoS attacks, more than 10 MX names MUST NOT be looked up # To prevent DoS attacks, more than 10 MX names MUST NOT be looked up
mxnames = self.dns(domainname, 'MX') mxnames = self.dns(domainname, 'MX')
if len(mxnames) > MAX_MX:
self.note_error('More than %d MX records returned'%MAX_MX)
if self.strict: if self.strict:
max = MAX_MX max = MAX_MX
if self.strict > 1 and len(mxnames) == 0: if self.strict > 1:
if len(mxnames) > MAX_MX:
raise AmbiguityWarning(
'More than %d MX records returned'%MAX_MX)
if len(mxnames) == 0:
raise AmbiguityWarning( raise AmbiguityWarning(
'No MX records found for mx mechanism', domainname) 'No MX records found for mx mechanism', domainname)
else: else:
max = MAX_MX * 4 max = MAX_MX * 4
return [a for mx in mxnames[:max] for a in self.dns_a(mx[1])] return [a for mx in mxnames[:max] for a in self.dns_a(mx[1],self.A)]
def dns_a(self, domainname): def dns_a(self, domainname, A='A'):
"""Get a list of IP addresses for a domainname.""" """Get a list of IP addresses for a domainname.
"""
if not domainname: return [] if not domainname: return []
if self.strict > 1: if self.strict > 1:
alist = self.dns(domainname, 'A') alist = self.dns(domainname, A)
if len(alist) == 0: if len(alist) == 0:
raise AmbiguityWarning('No A records found for', raise AmbiguityWarning(
domainname) 'No %s records found for'%A, domainname)
else: else:
return alist return alist
return self.dns(domainname, 'A') return self.dns(domainname, A)
def dns_aaaa(self, domainname):
"""Get a list of IPv6 addresses for a domainname."""
return self.dns(domainname, 'AAAA')
def validated_ptrs(self, i): def validated_ptrs(self):
"""Figure out the validated PTR domain names for a given IP """Figure out the validated PTR domain names for the connect IP."""
address.
"""
# To prevent DoS attacks, more than 10 PTR names MUST NOT be looked up # To prevent DoS attacks, more than 10 PTR names MUST NOT be looked up
if self.strict: if self.strict:
max = MAX_PTR max = MAX_PTR
if self.strict > 1: if self.strict > 1:
#Break out the number of PTR records returned for testing #Break out the number of PTR records returned for testing
try: try:
ptrnames = self.dns_ptr(i) ptrnames = self.dns_ptr(self.i)
ptrip = [p for p in ptrnames if i in self.dns_a(p)]
if len(ptrnames) > max: if len(ptrnames) > max:
warning = 'More than ' + str(max) + ' PTR records returned' warning = 'More than %d PTR records returned' % max
raise AmbiguityWarning(warning, i) raise AmbiguityWarning(warning, i)
else: else:
if len(ptrnames) == 0: if len(ptrnames) == 0:
raise AmbiguityWarning('No PTR records found for ptr mechanism', ptrnames) raise AmbiguityWarning(
return ptrip 'No PTR records found for ptr mechanism', self.c)
except: except:
raise AmbiguityWarning('No PTR records found for ptr mechanism', ptrnames) raise AmbiguityWarning(
'No PTR records found for ptr mechanism', i)
else: else:
max = MAX_PTR * 4 max = MAX_PTR * 4
return [p for p in self.dns_ptr(i)[:max] if i in self.dns_a(p)] cidrlength = self.cidrmax
return [p for p in self.dns_ptr(self.i)[:max]
if self.cidrmatch(self.dns_a(p,self.A),cidrlength)]
def dns_ptr(self, i): def dns_ptr(self, i):
"""Get a list of domain names for an IP address.""" """Get a list of domain names for an IP address."""
return self.dns(reverse_dots(i) + ".in-addr.arpa", 'PTR') return self.dns('%s.%s.arpa'%(reverse_dots(i),self.v), 'PTR')
def dns(self, name, qtype, cnames=None): def dns(self, name, qtype, cnames=None):
"""DNS query. """DNS query.
...@@ -1196,20 +1155,34 @@ class query(object): ...@@ -1196,20 +1155,34 @@ class query(object):
cnames = {} cnames = {}
elif len(cnames) >= MAX_CNAME: elif len(cnames) >= MAX_CNAME:
#return result # if too many == NX_DOMAIN #return result # if too many == NX_DOMAIN
raise PermError( raise PermError('Length of CNAME chain exceeds %d' % MAX_CNAME)
'Length of CNAME chain exceeds %d' % MAX_CNAME)
cnames[name] = cname cnames[name] = cname
if cname in cnames: if cname in cnames:
raise PermError, 'CNAME loop' raise PermError, 'CNAME loop'
result = self.dns(cname, qtype, cnames=cnames) result = self.dns(cname, qtype, cnames=cnames)
return result return result
def cidrmatch(self, ipaddrs, n):
"""Match connect IP against a list of other IP addresses."""
try:
if self.v == 'ip6':
MASK = 0xFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFL
bin = bin2long6
else:
MASK = 0xFFFFFFFFL
bin = addr2bin
c = ~(MASK >> n) & MASK & self.ip
for ip in [bin(ip) for ip in ipaddrs]:
if c == ~(MASK >> n) & MASK & ip: return True
except socket.error: pass
return False
def get_header(self, res, receiver=None): def get_header(self, res, receiver=None):
if not receiver: if not receiver:
receiver = self.r receiver = self.r
if res in ('pass', 'fail',' softfail'): if res in ('pass', 'fail',' softfail'):
return '%s (%s: %s) client-ip=%s; envelope-from=%s; helo=%s;' % ( return '%s (%s: %s) client-ip=%s; envelope-from=%s; helo=%s;' % (
res,receiver,self.get_header_comment(res),self.i, res, receiver, self.get_header_comment(res), self.c,
self.l + '@' + self.o, self.h) self.l + '@' + self.o, self.h)
if res == 'permerror': if res == 'permerror':
return '%s (%s: %s)' % (' '.join([res] + self.mech), return '%s (%s: %s)' % (' '.join([res] + self.mech),
...@@ -1223,16 +1196,16 @@ class query(object): ...@@ -1223,16 +1196,16 @@ class query(object):
if res == 'pass': if res == 'pass':
return \ return \
"domain of %s designates %s as permitted sender" \ "domain of %s designates %s as permitted sender" \
% (sender,self.i) % (sender, self.c)
elif res == 'softfail': return \ elif res == 'softfail': return \
"transitioning domain of %s does not designate %s as permitted sender" \ "transitioning domain of %s does not designate %s as permitted sender" \
% (sender,self.i) % (sender, self.c)
elif res == 'neutral': return \ elif res == 'neutral': return \
"%s is neither permitted nor denied by domain of %s" \ "%s is neither permitted nor denied by domain of %s" \
% (self.i,sender) % (self.c, sender)
elif res == 'none': return \ elif res == 'none': return \
"%s is neither permitted nor denied by domain of %s" \ "%s is neither permitted nor denied by domain of %s" \
% (self.i,sender) % (self.c, sender)
#"%s does not designate permitted sender hosts" % sender #"%s does not designate permitted sender hosts" % sender
elif res == 'permerror': return \ elif res == 'permerror': return \
"permanent error in processing domain of %s: %s" \ "permanent error in processing domain of %s: %s" \
...@@ -1241,7 +1214,7 @@ class query(object): ...@@ -1241,7 +1214,7 @@ class query(object):
"temporary error in processing during lookup of %s" % sender "temporary error in processing during lookup of %s" % sender
elif res == 'fail': return \ elif res == 'fail': return \
"domain of %s does not designate %s as permitted sender" \ "domain of %s does not designate %s as permitted sender" \
% (sender,self.i) % (sender, self.c)
raise ValueError("invalid SPF result for header comment: "+res) raise ValueError("invalid SPF result for header comment: "+res)
def split_email(s, h): def split_email(s, h):
...@@ -1262,6 +1235,8 @@ def split_email(s, h): ...@@ -1262,6 +1235,8 @@ def split_email(s, h):
return 'postmaster', h return 'postmaster', h
else: else:
parts = s.split('@', 1) parts = s.split('@', 1)
if parts[0] == '':
parts[0] = 'postmaster'
if len(parts) == 2: if len(parts) == 2:
return tuple(parts) return tuple(parts)
else: else:
...@@ -1352,47 +1327,6 @@ def domainmatch(ptrs, domainsuffix): ...@@ -1352,47 +1327,6 @@ def domainmatch(ptrs, domainsuffix):
return False return False
def cidrmatch(i, ipaddrs, cidr_length = 32):
"""Match an IP address against a list of other IP addresses.
Examples:
>>> cidrmatch('192.168.0.45', ['192.168.0.44', '192.168.0.45'])
1
>>> cidrmatch('192.168.0.43', ['192.168.0.44', '192.168.0.45'])
0
>>> cidrmatch('192.168.0.43', ['192.168.0.44', '192.168.0.45'], 24)
1
"""
try:
c = cidr(i, cidr_length)
for ip in ipaddrs:
if cidr(ip, cidr_length) == c:
return True
except socket.error: pass
return False
def cidr(i, n):
"""Convert an IP address string with a CIDR mask into a 32-bit
or 128-bit integer.
i must be a string of numbers 0..255 separated by dots '.'::
pre: forall([0 <= int(p) < 256 for p in i.split('.')])
n is a number of bits to mask::
pre: 0 <= n <= 32
Examples:
>>> bin2addr(cidr('192.168.5.45', 32))
'192.168.5.45'
>>> bin2addr(cidr('192.168.5.45', 24))
'192.168.5.0'
>>> bin2addr(cidr('192.168.0.45', 8))
'192.0.0.0'
"""
return ~(MASK >> n) & MASK & addr2bin(i)
def addr2bin(str): def addr2bin(str):
"""Convert a string IPv4 address into an unsigned integer. """Convert a string IPv4 address into an unsigned integer.
...@@ -1421,27 +1355,12 @@ def addr2bin(str): ...@@ -1421,27 +1355,12 @@ def addr2bin(str):
>>> 10 * (2 ** 24) + 93 * (2 ** 16) + 512 >>> 10 * (2 ** 24) + 93 * (2 ** 16) + 512
173867520 173867520
""" """
try:
return struct.unpack("!L", socket.inet_aton(str))[0] return struct.unpack("!L", socket.inet_aton(str))[0]
except socket.error:
if not socket.has_ipv6: raise
h,l = struct.unpack("!QQ", socket.inet_pton(socket.AF_INET6,str))
return h << 64 | l;
def bin2addr(addr):
"""Convert a numeric IPv4 address into string n.n.n.n form.
Examples:: if socket.has_ipv6:
>>> bin2addr(socket.INADDR_LOOPBACK) def bin2long6(str):
'127.0.0.1' h, l = struct.unpack("!QQ", str)
return h << 64 | l
>>> bin2addr(socket.INADDR_ANY)
'0.0.0.0'
>>> bin2addr(socket.INADDR_NONE)
'255.255.255.255'
"""
return socket.inet_ntoa(struct.pack("!L", addr))
def expand_one(expansion, str, joiner): def expand_one(expansion, str, joiner):
if not str: if not str:
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment