DandelionNStuff/pymilter
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Move parse_header to Milter.utils.

Browse Source 
Test case for delayed DSN parsing.
Fix plock when source missing or cannot set owner/group.
This commit is contained in:
Stuart Gathman
2007-01-19 23:31:38 +00:00
parent 393aa6140a
commit 4c72135b0e
8 changed files with 112 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files
MANIFEST.in
+1 -1
View File
@@ -8,7 +8,7 @@ include ChangeLog
include MANIFEST.in
include testsample.py
include testmime.py
include testcache.py
include testutils.py
include testbms.py
include rejects.py
include report.py
Milter/cache.py
+8 -1
View File
@@ -10,6 +10,9 @@
# CBV results.
#
# $Log$
# Revision 1.5 2007/01/11 19:59:40 customdesigned
# Purge old entries in auto_whitelist and send_dsn logs.
#
# Revision 1.4 2007/01/11 04:31:26 customdesigned
# Negative feedback for bad headers. Purge cache logs on startup.
#
@@ -51,7 +54,11 @@ class AddrCache(object):
changed = False
try:
too_old = now - age*24*60*60 # max age in days
for ln in open(self.fname):
try:
fp = open(self.fname)
except OSError:
fp = ()
for ln in fp:
try:
rcpt,ts = ln.strip().split(None,1)
l = time.strptime(ts,AddrCache.time_format)
Milter/plock.py
+9 -5
View File
@@ -11,24 +11,28 @@ class PLock(object):
self.basename = basename
self.fp = None
def lock(self,lockname=None):
def lock(self,lockname=None,mode=0660,strict_perms=False):
"Start an update transaction. Return FILE to write new version."
self.unlock()
if not lockname:
lockname = self.basename + '.lock'
self.lockname = lockname
st = os.stat(self.basename)
try:
st = os.stat(self.basename)
mode |= st.st_mode
except OSError: pass
u = os.umask(0002)
try:
fd = os.open(lockname,os.O_WRONLY+os.O_CREAT+os.O_EXCL,st.st_mode|0660)
fd = os.open(lockname,os.O_WRONLY+os.O_CREAT+os.O_EXCL,mode)
finally:
os.umask(u)
self.fp = os.fdopen(fd,'w')
try:
os.chown(self.lockname,-1,st.st_gid)
except:
self.unlock()
raise
if strict_perms:
self.unlock()
raise
return self.fp
def wlock(self,lockname=None):
Milter/utils.py
+27
View File
@@ -1,7 +1,9 @@
import re
import struct
import socket
import email.Errors
from fnmatch import fnmatchcase
from email.Header import decode_header
ip4re = re.compile(r'^[0-9]*\.[0-9]*\.[0-9]*\.[0-9]*$')
@@ -56,3 +58,28 @@ def parse_addr(t):
pos = t.find('"@')
if pos > 0: return [t[1:pos],t[pos+2:]]
return t.split('@')
def parse_header(val):
"""Decode headers gratuitously encoded to hide the content.
"""
try:
h = decode_header(val)
if not len(h) or (not h[0][1] and len(h) == 1): return val
u = []
for s,enc in h:
if enc:
try:
u.append(unicode(s,enc))
except LookupError:
u.append(unicode(s))
else:
u.append(unicode(s))
u = ''.join(u)
for enc in ('us-ascii','iso-8859-1','utf8'):
try:
return u.encode(enc)
except UnicodeError: continue
except UnicodeDecodeError: pass
except LookupError: pass
except email.Errors.HeaderParseError: pass
return val
bms.py
+39 -54
View File
@@ -1,6 +1,12 @@
#!/usr/bin/env python
# A simple milter that has grown quite a bit.
# $Log$
# Revision 1.87 2007/01/18 16:48:44 customdesigned
# Doc update.
# Parse From header for delayed failure detection.
# Don't check reputation of trusted host.
# Track IP reputation only when missing PTR.
#
# Revision 1.86 2007/01/16 05:17:29 customdesigned
# REJECT after data for blacklisted emails - so in case of mistakes, a
# legitimate sender will know what happened.
@@ -76,11 +82,10 @@ import gc
import anydbm
import Milter.dsn as dsn
from Milter.dynip import is_dynip as dynip
from Milter.utils import iniplist,parse_addr,ip4re
from Milter.utils import iniplist,parse_addr,parse_header,ip4re
from Milter.config import MilterConfigParser
from fnmatch import fnmatchcase
from email.Header import decode_header
from email.Utils import getaddresses,parseaddr
# Import gossip if available
@@ -328,30 +333,26 @@ def read_config(list):
srs_domain.add(cp.getdefault('srs','fwdomain'))
banned_users = cp.getlist('srs','banned_users')
def parse_header(val):
"""Decode headers gratuitously encoded to hide the content.
"""
try:
h = decode_header(val)
if not len(h) or (not h[0][1] and len(h) == 1): return val
u = []
for s,enc in h:
if enc:
try:
u.append(unicode(s,enc))
except LookupError:
u.append(unicode(s))
else:
u.append(unicode(s))
u = ''.join(u)
for enc in ('us-ascii','iso-8859-1','utf8'):
def findsrs(fp):
lastln = None
for ln in fp:
if lastln:
if ln[0].isspace() and ln[0] != '\n':
lastln += ln
continue
try:
return u.encode(enc)
except UnicodeError: continue
except UnicodeDecodeError: pass
except LookupError: pass
except email.Errors.HeaderParseError: pass
return val
name,val = lastln.rstrip().split(None,1)
pos = val.find('<SRS')
if pos >= 0:
return srs.reverse(val[pos+1:-1])
except: continue
lnl = ln.lower()
if lnl.startswith('action:'):
if lnl.split()[-1] != 'failed': break
for k in ('message-id:','x-mailer:','sender:'):
if lnl.startswith(k):
lastln = ln
break
class SPFPolicy(object):
"Get SPF policy by result from sendmail style access file."
@@ -1362,35 +1363,19 @@ class bmsMilter(Milter.Milter):
# check for delayed bounce
if self.delayed_failure:
self.fp.seek(0)
lastln = None
for ln in self.fp:
if lastln:
if ln[0].isspace() and ln[0] != '\n':
lastln += ln
continue
try:
name,val = lastln.rstrip().split(None,1)
pos = val.find('<SRS')
if pos >= 0:
sender = srs.reverse(val[pos+1:-1])
cbv_cache[sender] = 500,self.delayed_failure,time.time()
try:
# save message for debugging
fname = tempfile.mktemp(".dsn")
os.rename(self.tempname,fname)
except:
fname = self.tempname
self.tempname = None
self.log('BLACKLIST:',sender,fname)
return Milter.DISCARD
except: continue
lnl = ln.lower()
if lnl.startswith('action:'):
if lnl.split()[-1] != 'failed': break
for k in ('message-id:','x-mailer:','sender:'):
if lnl.startswith(k):
lastln = ln
break
sender = findsrs(self.fp)
if sender:
cbv_cache[sender] = 500,self.delayed_failure,time.time()
try:
# save message for debugging
fname = tempfile.mktemp(".dsn")
os.rename(self.tempname,fname)
except:
fname = self.tempname
self.tempname = None
self.log('BLACKLIST:',sender,fname)
return Milter.DISCARD
# analyze external mail for spam
spam_checked = self.check_spam() # tag or quarantine for spam
test.py
+2 -2
View File
@@ -2,7 +2,7 @@ import unittest
import testbms
import testmime
import testsample
import testcache
import testutils
import os
def suite():
@@ -10,7 +10,7 @@ def suite():
s.addTest(testbms.suite())
s.addTest(testmime.suite())
s.addTest(testsample.suite())
s.addTest(testcache.suite())
s.addTest(testutils.suite())
return s
if __name__ == '__main__':
testbms.py
+19
View File
@@ -277,6 +277,25 @@ class BMSMilterTestCase(unittest.TestCase):
fp = milter._body
open("test/test1.tstout","w").write(fp.getvalue())
def testFindsrs(self):
if not bms.srs:
import SRS
bms.srs = SRS.new(secret='test')
sender = bms.srs.forward('foo@bar.com','mail.example.com')
sndr = bms.findsrs(StringIO.StringIO(
"""Received: from [1.16.33.86] (helo=mail.example.com)
by bastion4.mail.zen.co.uk with smtp (Exim 4.50) id 1H3IBC-00013b-O9
for foo@bar.com; Sat, 06 Jan 2007 20:30:17 +0000
X-Mailer: "PyMilter-0.8.5"
<%s> foo
MIME-Version: 1.0
Content-Type: text/plain
To: foo@bar.com
From: postmaster@mail.example.com
""" % sender
))
self.assertEqual(sndr,'foo@bar.com')
# def testReject(self):
# "Test content based spam rejection."
# milter = TestMilter()
testcache.py → testutils.py
+7 -3
View File
@@ -1,6 +1,7 @@
import unittest
import doctest
import os
import Milter.utils
from Milter.cache import AddrCache
class AddrCacheTestCase(unittest.TestCase):
@@ -26,7 +27,10 @@ class AddrCacheTestCase(unittest.TestCase):
self.failUnless(s[0].startswith('foo@bar.com '))
self.assertEquals(s[1].strip(),'baz@bar.com')
def suite(): return unittest.makeSuite(AddrCacheTestCase,'test')
def suite():
s = unittest.makeSuite(AddrCacheTestCase,'test')
s.addTest(doctest.DocTestSuite(Milter.utils))
return s
if __name__ == '__main__':
unittest.main()
unittest.TextTestRunner().run(suite())