Work at SourceForge, help us to make it a better place! We have an immediate need for a Support Technician in our San Francisco or Denver office.

Close

[98dec2]: ForgeMail / forgemail / lib / util.py Maximize Restore History

Download this file

util.py    191 lines (170 with data), 6.8 kB

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
import re
import logging
import smtplib
import email.feedparser
from email.MIMEMultipart import MIMEMultipart
from email.MIMEText import MIMEText
from email import header
import tg
from paste.deploy.converters import asbool, asint
from formencode import validators as fev
from pylons import c
from allura.lib.helpers import push_config, find_project
from allura import model as M
from allura.lib.utils import ConfigProxy
from . import exc
log = logging.getLogger(__name__)
RE_MESSAGE_ID = re.compile(r'<(.*)>')
config = ConfigProxy(
common_suffix='forgemail.domain',
return_path='forgemail.return_path')
EMAIL_VALIDATOR=fev.Email(not_empty=True)
def Header(text, charset):
'''Helper to make sure we don't over-encode headers
(gmail barfs with encoded email addresses.)'''
if isinstance(text, header.Header):
return text
h = header.Header('', charset)
for word in text.split(' '):
h.append(word)
return h
def parse_address(addr):
userpart, domain = addr.split('@')
# remove common domain suffix
if not domain.endswith(config.common_suffix):
raise exc.AddressException, 'Unknown domain: ' + domain
domain = domain[:-len(config.common_suffix)]
path = '/'.join(reversed(domain.split('.')))
project, mount_point = find_project('/' + path)
if project is None:
raise exc.AddressException, 'Unknown project: ' + domain
if len(mount_point) != 1:
raise exc.AddressException, 'Unknown tool: ' + domain
with push_config(c, project=project):
app = project.app_instance(mount_point[0])
if not app:
raise exc.AddressException, 'Unknown tool: ' + domain
topic = '%s.msg.%s' % (app.config.tool_name, userpart)
return topic, project, app
def parse_message(data):
# Parse the email to its constituent parts
parser = email.feedparser.FeedParser()
parser.feed(data)
msg = parser.close()
# Extract relevant data
result = {}
result['multipart'] = multipart = msg.is_multipart()
result['headers'] = dict(msg)
result['message_id'] = _parse_message_id(msg.get('Message-ID'))[0]
result['in_reply_to'] = _parse_message_id(msg.get('In-Reply-To'))
result['references'] = _parse_message_id(msg.get('References'))
if multipart:
result['parts'] = []
for part in msg.walk():
dpart = dict(
headers=dict(part),
message_id=result['message_id'],
in_reply_to=result['in_reply_to'],
references=result['references'],
content_type=part.get_content_type(),
filename=part.get_filename(None),
payload=part.get_payload(decode=True))
charset = part.get_content_charset()
if charset:
dpart['payload'] = dpart['payload'].decode(charset)
result['parts'].append(dpart)
else:
result['payload'] = msg.get_payload(decode=True)
charset = msg.get_content_charset()
if charset:
result['payload'] = result['payload'].decode(charset)
return result
def identify_sender(peer, email_address, headers, msg):
# Dumb ID -- just look for email address claimed by a particular user
addr = M.EmailAddress.query.get(_id=M.EmailAddress.canonical(email_address))
if addr and addr.claimed_by_user_id:
return addr.claimed_by_user()
addr = M.EmailAddress.query.get(_id=M.EmailAddress.canonical(headers.get('From')))
if addr and addr.claimed_by_user_id:
return addr.claimed_by_user()
return M.User.anonymous()
def encode_email_part(content, content_type):
try:
return MIMEText(content.encode('iso-8859-1'), content_type, 'iso-8859-1')
except:
return MIMEText(content.encode('latin-1'), content_type, 'utf-8')
def make_multipart_message(*parts):
msg = MIMEMultipart('related')
msg.preamble = 'This is a multi-part message in MIME format.'
alt = MIMEMultipart('alternative')
msg.attach(alt)
for part in parts:
alt.attach(part)
return msg
def _parse_message_id(msgid):
if msgid is None: return []
return [ mo.group(1)
for mo in RE_MESSAGE_ID.finditer(msgid) ]
def _parse_smtp_addr(addr):
addr = str(addr)
addrs = _parse_message_id(addr)
if addrs and addrs[0]: return addrs[0]
if '@' in addr: return addr
return 'noreply@in.sf.net'
def isvalid(addr):
'''return True if addr is a (possibly) valid email address, false
otherwise'''
try:
EMAIL_VALIDATOR.to_python(addr, None)
return True
except fev.Invalid:
return False
class SMTPClient(object):
def __init__(self):
self._client = None
def sendmail(self, addrs, addrfrom, reply_to, subject, message_id, in_reply_to, message):
if not addrs: return
charset = message.get_charset()
if charset is None:
charset = 'iso-8859-1'
message['To'] = Header(reply_to, charset)
message['From'] = Header(addrfrom, charset)
message['Reply-To'] = Header(reply_to, charset)
message['Subject'] = Header(subject, charset)
message['Message-ID'] = Header('<' + message_id + '>', charset)
if in_reply_to:
if isinstance(in_reply_to, basestring):
in_reply_to = [ in_reply_to ]
in_reply_to = ','.join(('<' + irt + '>') for irt in in_reply_to)
message['In-Reply-To'] = Header(in_reply_to, charset)
content = message.as_string()
smtp_addrs = map(_parse_smtp_addr, addrs)
smtp_addrs = [ a for a in smtp_addrs if isvalid(a) ]
if not smtp_addrs:
log.warning('No valid addrs in %s, so not sending mail', addrs)
return
try:
self._client.sendmail(
config.return_path,
smtp_addrs,
content)
except:
self._connect()
self._client.sendmail(
config.return_path,
smtp_addrs,
content)
def _connect(self):
if asbool(tg.config.get('smtp_ssl', False)):
smtp_client = smtplib.SMTP_SSL(
tg.config.get('smtp_server', 'localhost'),
asint(tg.config.get('smtp_port', 25)))
else:
smtp_client = smtplib.SMTP(
tg.config.get('smtp_server', 'localhost'),
asint(tg.config.get('smtp_port', 465)))
if tg.config.get('smtp_user', None):
smtp_client.login(tg.config['smtp_user'], tg.config['smtp_password'])
if asbool(tg.config.get('smtp_tls', False)):
smtp_client.starttls()
self._client = smtp_client