MDS/openide
1
0
Fork 0
mirror of https://gitflic.ru/project/openide/openide.git synced 2026-09-27 10:03:11 +07:00
Code Issues Packages Projects Releases Wiki Activity

upgrade to Jython 2.5.0

Browse Source
This commit is contained in:
Dmitry Jemerov
2009-07-30 19:38:38 +04:00
parent f221494a43
commit bb24ece14b
487 changed files with 155647 additions and 18427 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/lib/Lib/imaplib.py
+408 -117
View File
@@ -15,12 +15,16 @@ Public functions: Internaldate2tuple
# Authentication code contributed by Donn Cave <donn@u.washington.edu> June 1998.
# String method conversion by ESR, February 2001.
# GET/SETACL contributed by Anthony Baxter <anthony@interlink.com.au> April 2001.
# IMAP4_SSL contributed by Tino Lange <Tino.Lange@isg.de> March 2002.
# GET/SETQUOTA contributed by Andreas Zeidler <az@kreativkombinat.de> June 2002.
# PROXYAUTH contributed by Rick Holbert <holbert.13@osu.edu> November 2002.
# GET/SETANNOTATION contributed by Tomas Lindroos <skitta@abo.fi> June 2005.
__version__ = "2.49"
__version__ = "2.58"
import binascii, re, socket, time, random, sys
import binascii, os, random, re, socket, sys, time
__all__ = ["IMAP4", "Internaldate2tuple",
__all__ = ["IMAP4", "IMAP4_SSL", "IMAP4_stream", "Internaldate2tuple",
"Int2AP", "ParseFlags", "Time2Internaldate"]
# Globals
@@ -28,6 +32,7 @@ __all__ = ["IMAP4", "Internaldate2tuple",
CRLF = '\r\n'
Debug = 0
IMAP4_PORT = 143
IMAP4_SSL_PORT = 993
AllowedVersions = ('IMAP4REV1', 'IMAP4') # Most recent first
# Commands
@@ -42,25 +47,34 @@ Commands = {
'COPY': ('SELECTED',),
'CREATE': ('AUTH', 'SELECTED'),
'DELETE': ('AUTH', 'SELECTED'),
'DELETEACL': ('AUTH', 'SELECTED'),
'EXAMINE': ('AUTH', 'SELECTED'),
'EXPUNGE': ('SELECTED',),
'FETCH': ('SELECTED',),
'GETACL': ('AUTH', 'SELECTED'),
'GETANNOTATION':('AUTH', 'SELECTED'),
'GETQUOTA': ('AUTH', 'SELECTED'),
'GETQUOTAROOT': ('AUTH', 'SELECTED'),
'MYRIGHTS': ('AUTH', 'SELECTED'),
'LIST': ('AUTH', 'SELECTED'),
'LOGIN': ('NONAUTH',),
'LOGOUT': ('NONAUTH', 'AUTH', 'SELECTED', 'LOGOUT'),
'LSUB': ('AUTH', 'SELECTED'),
'NAMESPACE': ('AUTH', 'SELECTED'),
'NOOP': ('NONAUTH', 'AUTH', 'SELECTED', 'LOGOUT'),
'PARTIAL': ('SELECTED',),
'PARTIAL': ('SELECTED',), # NB: obsolete
'PROXYAUTH': ('AUTH',),
'RENAME': ('AUTH', 'SELECTED'),
'SEARCH': ('SELECTED',),
'SELECT': ('AUTH', 'SELECTED'),
'SETACL': ('AUTH', 'SELECTED'),
'SETANNOTATION':('AUTH', 'SELECTED'),
'SETQUOTA': ('AUTH', 'SELECTED'),
'SORT': ('SELECTED',),
'STATUS': ('AUTH', 'SELECTED'),
'STORE': ('SELECTED',),
'SUBSCRIBE': ('AUTH', 'SELECTED'),
'THREAD': ('SELECTED',),
'UID': ('SELECTED',),
'UNSUBSCRIBE': ('AUTH', 'SELECTED'),
}
@@ -70,11 +84,12 @@ Commands = {
Continuation = re.compile(r'\+( (?P<data>.*))?')
Flags = re.compile(r'.*FLAGS \((?P<flags>[^\)]*)\)')
InternalDate = re.compile(r'.*INTERNALDATE "'
r'(?P<day>[ 123][0-9])-(?P<mon>[A-Z][a-z][a-z])-(?P<year>[0-9][0-9][0-9][0-9])'
r'(?P<day>[ 0123][0-9])-(?P<mon>[A-Z][a-z][a-z])-(?P<year>[0-9][0-9][0-9][0-9])'
r' (?P<hour>[0-9][0-9]):(?P<min>[0-9][0-9]):(?P<sec>[0-9][0-9])'
r' (?P<zonen>[-+])(?P<zoneh>[0-9][0-9])(?P<zonem>[0-9][0-9])'
r'"')
Literal = re.compile(r'.*{(?P<size>\d+)}$')
MapCRLF = re.compile(r'\r\n|\r|\n')
Response_code = re.compile(r'\[(?P<type>[A-Z-]+)( (?P<data>[^\]]*))?\]')
Untagged_response = re.compile(r'\* (?P<type>[A-Z-]+)( (?P<data>.*))?')
Untagged_status = re.compile(r'\* (?P<data>\d+) (?P<type>[A-Z-]+)( (?P<data2>.*))?')
@@ -105,7 +120,10 @@ class IMAP4:
Each command returns a tuple: (type, [data, ...]) where 'type'
is usually 'OK' or 'NO', and 'data' is either the text from the
tagged response, or untagged results from command.
tagged response, or untagged results from command. Each 'data'
is either a string, or a tuple. If a tuple, then the first part
is the header of the response, and the second part contains
the data (ie: 'literal' value).
Errors raise the exception class <instance>.error("<reason>").
IMAP4 server errors raise <instance>.abort("<reason>"),
@@ -118,10 +136,10 @@ class IMAP4:
the command re-tried.
"readonly" exceptions imply the command should be re-tried.
Note: to use this module, you must read the RFCs pertaining
to the IMAP4 protocol, as the semantics of the arguments to
each IMAP4 command are left to the invoker, not to mention
the results.
Note: to use this module, you must read the RFCs pertaining to the
IMAP4 protocol, as the semantics of the arguments to each IMAP4
command are left to the invoker, not to mention the results. Also,
most IMAP servers implement a sub-set of the commands available here.
"""
class error(Exception): pass # Logical errors - debug required
@@ -131,15 +149,13 @@ class IMAP4:
mustquote = re.compile(r"[^\w!#$%&'*+,.:;<=>?^`|~-]")
def __init__(self, host = '', port = IMAP4_PORT):
self.host = host
self.port = port
self.debug = Debug
self.state = 'LOGOUT'
self.literal = None # A literal argument to a command
self.tagged_commands = {} # Tagged commands awaiting response
self.untagged_responses = {} # {typ: [data, ...], ...}
self.continuation_response = '' # Last continuation response
self.is_readonly = None # READ-ONLY desired state
self.is_readonly = False # READ-ONLY desired state
self.tagnum = 0
# Open socket to server.
@@ -149,7 +165,7 @@ class IMAP4:
# Create unique tag for this session,
# and compile tagged response matcher.
self.tagpre = Int2AP(random.randint(0, 31999))
self.tagpre = Int2AP(random.randint(4096, 65535))
self.tagre = re.compile(r'(?P<tag>'
+ self.tagpre
+ r'\d+) (?P<type>[A-Z]+) (?P<data>.*)')
@@ -158,27 +174,29 @@ class IMAP4:
# request and store CAPABILITY response.
if __debug__:
self._cmd_log_len = 10
self._cmd_log_idx = 0
self._cmd_log = {} # Last `_cmd_log_len' interactions
if self.debug >= 1:
_mesg('imaplib version %s' % __version__)
_mesg('new IMAP4 connection, tag=%s' % self.tagpre)
self._mesg('imaplib version %s' % __version__)
self._mesg('new IMAP4 connection, tag=%s' % self.tagpre)
self.welcome = self._get_response()
if self.untagged_responses.has_key('PREAUTH'):
if 'PREAUTH' in self.untagged_responses:
self.state = 'AUTH'
elif self.untagged_responses.has_key('OK'):
elif 'OK' in self.untagged_responses:
self.state = 'NONAUTH'
else:
raise self.error(self.welcome)
cap = 'CAPABILITY'
self._simple_command(cap)
if not self.untagged_responses.has_key(cap):
typ, dat = self.capability()
if dat == [None]:
raise self.error('no CAPABILITY response from server')
self.capabilities = tuple(self.untagged_responses[cap][-1].upper().split())
self.capabilities = tuple(dat[-1].upper().split())
if __debug__:
if self.debug >= 3:
_mesg('CAPABILITIES: %s' % `self.capabilities`)
self._mesg('CAPABILITIES: %r' % (self.capabilities,))
for version in AllowedVersions:
if not version in self.capabilities:
@@ -191,7 +209,7 @@ class IMAP4:
def __getattr__(self, attr):
# Allow UPPERCASE variants of IMAP4 command methods.
if Commands.has_key(attr):
if attr in Commands:
return getattr(self, attr.lower())
raise AttributeError("Unknown IMAP4 command: '%s'" % attr)
@@ -200,13 +218,16 @@ class IMAP4:
# Overridable methods
def open(self, host, port):
"""Setup connection to remote server on "host:port".
def open(self, host = '', port = IMAP4_PORT):
"""Setup connection to remote server on "host:port"
(default: localhost:standard IMAP4 port).
This connection will be used by the routines:
read, readline, send, shutdown.
"""
self.host = host
self.port = port
self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.sock.connect((self.host, self.port))
self.sock.connect((host, port))
self.file = self.sock.makefile('rb')
@@ -224,6 +245,7 @@ class IMAP4:
"""Send data to remote."""
self.sock.sendall(data)
def shutdown(self):
"""Close I/O established in "open"."""
self.file.close()
@@ -292,7 +314,7 @@ class IMAP4:
date_time = Time2Internaldate(date_time)
else:
date_time = None
self.literal = message
self.literal = MapCRLF.sub(CRLF, message)
return self._simple_command(name, mailbox, flags, date_time)
@@ -313,9 +335,10 @@ class IMAP4:
be sent instead.
"""
mech = mechanism.upper()
cap = 'AUTH=%s' % mech
if not cap in self.capabilities:
raise self.error("Server doesn't allow %s authentication." % mech)
# XXX: shouldn't this code be removed, not commented out?
#cap = 'AUTH=%s' % mech
#if not cap in self.capabilities: # Let the server decide!
# raise self.error("Server doesn't allow %s authentication." % mech)
self.literal = _Authenticator(authobject).process
typ, dat = self._simple_command('AUTHENTICATE', mech)
if typ != 'OK':
@@ -324,6 +347,15 @@ class IMAP4:
return typ, dat
def capability(self):
"""(typ, [data]) = <instance>.capability()
Fetch capabilities list from server."""
name = 'CAPABILITY'
typ, dat = self._simple_command(name)
return self._untagged_response(typ, dat, name)
def check(self):
"""Checkpoint mailbox on server.
@@ -370,6 +402,12 @@ class IMAP4:
"""
return self._simple_command('DELETE', mailbox)
def deleteacl(self, mailbox, who):
"""Delete the ACLs (remove any rights) set for who on mailbox.
(typ, [data]) = <instance>.deleteacl(mailbox, who)
"""
return self._simple_command('DELETEACL', mailbox, who)
def expunge(self):
"""Permanently remove deleted items from selected mailbox.
@@ -409,6 +447,36 @@ class IMAP4:
return self._untagged_response(typ, dat, 'ACL')
def getannotation(self, mailbox, entry, attribute):
"""(typ, [data]) = <instance>.getannotation(mailbox, entry, attribute)
Retrieve ANNOTATIONs."""
typ, dat = self._simple_command('GETANNOTATION', mailbox, entry, attribute)
return self._untagged_response(typ, dat, 'ANNOTATION')
def getquota(self, root):
"""Get the quota root's resource usage and limits.
Part of the IMAP4 QUOTA extension defined in rfc2087.
(typ, [data]) = <instance>.getquota(root)
"""
typ, dat = self._simple_command('GETQUOTA', root)
return self._untagged_response(typ, dat, 'QUOTA')
def getquotaroot(self, mailbox):
"""Get the list of quota roots for the named mailbox.
(typ, [[QUOTAROOT responses...], [QUOTA responses]]) = <instance>.getquotaroot(mailbox)
"""
typ, dat = self._simple_command('GETQUOTAROOT', mailbox)
typ, quota = self._untagged_response(typ, dat, 'QUOTA')
typ, quotaroot = self._untagged_response(typ, dat, 'QUOTAROOT')
return typ, [quotaroot, quota]
def list(self, directory='""', pattern='*'):
"""List mailbox names in directory matching pattern.
@@ -428,8 +496,6 @@ class IMAP4:
NB: 'password' will be quoted.
"""
#if not 'AUTH=LOGIN' in self.capabilities:
# raise self.error("Server doesn't allow LOGIN authentication." % mech)
typ, dat = self._simple_command('LOGIN', user, self._quote(password))
if typ != 'OK':
raise self.error(dat[-1])
@@ -437,6 +503,21 @@ class IMAP4:
return typ, dat
def login_cram_md5(self, user, password):
""" Force use of CRAM-MD5 authentication.
(typ, [data]) = <instance>.login_cram_md5(user, password)
"""
self.user, self.password = user, password
return self.authenticate('CRAM-MD5', self._CRAM_MD5_AUTH)
def _CRAM_MD5_AUTH(self, challenge):
""" Authobject to use with CRAM-MD5 authentication. """
import hmac
return self.user + " " + hmac.HMAC(self.password, challenge).hexdigest()
def logout(self):
"""Shutdown connection to server.
@@ -448,7 +529,7 @@ class IMAP4:
try: typ, dat = self._simple_command('LOGOUT')
except: typ, dat = 'NO', ['%s: %s' % sys.exc_info()[:2]]
self.shutdown()
if self.untagged_responses.has_key('BYE'):
if 'BYE' in self.untagged_responses:
return 'BYE', self.untagged_responses['BYE']
return typ, dat
@@ -464,6 +545,13 @@ class IMAP4:
typ, dat = self._simple_command(name, directory, pattern)
return self._untagged_response(typ, dat, name)
def myrights(self, mailbox):
"""Show my ACLs for a mailbox (i.e. the rights that I have on mailbox).
(typ, [data]) = <instance>.myrights(mailbox)
"""
typ,dat = self._simple_command('MYRIGHTS', mailbox)
return self._untagged_response(typ, dat, 'MYRIGHTS')
def namespace(self):
""" Returns IMAP namespaces ala rfc2342
@@ -478,11 +566,11 @@ class IMAP4:
def noop(self):
"""Send NOOP command.
(typ, data) = <instance>.noop()
(typ, [data]) = <instance>.noop()
"""
if __debug__:
if self.debug >= 3:
_dump_ur(self.untagged_responses)
self._dump_ur(self.untagged_responses)
return self._simple_command('NOOP')
@@ -498,10 +586,23 @@ class IMAP4:
return self._untagged_response(typ, dat, 'FETCH')
def proxyauth(self, user):
"""Assume authentication as "user".
Allows an authorised administrator to proxy into any user's
mailbox.
(typ, [data]) = <instance>.proxyauth(user)
"""
name = 'PROXYAUTH'
return self._simple_command('PROXYAUTH', user)
def rename(self, oldmailbox, newmailbox):
"""Rename old mailbox name to new.
(typ, data) = <instance>.rename(oldmailbox, newmailbox)
(typ, [data]) = <instance>.rename(oldmailbox, newmailbox)
"""
return self._simple_command('RENAME', oldmailbox, newmailbox)
@@ -509,28 +610,30 @@ class IMAP4:
def search(self, charset, *criteria):
"""Search mailbox for matching messages.
(typ, [data]) = <instance>.search(charset, criterium, ...)
(typ, [data]) = <instance>.search(charset, criterion, ...)
'data' is space separated list of matching message numbers.
"""
name = 'SEARCH'
if charset:
typ, dat = apply(self._simple_command, (name, 'CHARSET', charset) + criteria)
typ, dat = self._simple_command(name, 'CHARSET', charset, *criteria)
else:
typ, dat = apply(self._simple_command, (name,) + criteria)
typ, dat = self._simple_command(name, *criteria)
return self._untagged_response(typ, dat, name)
def select(self, mailbox='INBOX', readonly=None):
def select(self, mailbox='INBOX', readonly=False):
"""Select a mailbox.
Flush all untagged responses.
(typ, [data]) = <instance>.select(mailbox='INBOX', readonly=None)
(typ, [data]) = <instance>.select(mailbox='INBOX', readonly=False)
'data' is count of messages in mailbox ('EXISTS' response).
Mandated responses are ('FLAGS', 'EXISTS', 'RECENT', 'UIDVALIDITY'), so
other responses should be obtained via <instance>.response('FLAGS') etc.
"""
# Mandated responses are ('FLAGS', 'EXISTS', 'RECENT', 'UIDVALIDITY')
self.untagged_responses = {} # Flush old responses.
self.is_readonly = readonly
if readonly:
@@ -542,11 +645,11 @@ class IMAP4:
self.state = 'AUTH' # Might have been 'SELECTED'
return typ, dat
self.state = 'SELECTED'
if self.untagged_responses.has_key('READ-ONLY') \
if 'READ-ONLY' in self.untagged_responses \
and not readonly:
if __debug__:
if self.debug >= 1:
_dump_ur(self.untagged_responses)
self._dump_ur(self.untagged_responses)
raise self.readonly('%s is not writable' % mailbox)
return typ, self.untagged_responses.get('EXISTS', [None])
@@ -554,11 +657,28 @@ class IMAP4:
def setacl(self, mailbox, who, what):
"""Set a mailbox acl.
(typ, [data]) = <instance>.create(mailbox, who, what)
(typ, [data]) = <instance>.setacl(mailbox, who, what)
"""
return self._simple_command('SETACL', mailbox, who, what)
def setannotation(self, *args):
"""(typ, [data]) = <instance>.setannotation(mailbox[, entry, attribute]+)
Set ANNOTATIONs."""
typ, dat = self._simple_command('SETANNOTATION', *args)
return self._untagged_response(typ, dat, 'ANNOTATION')
def setquota(self, root, limits):
"""Set the quota root's resource limits.
(typ, [data]) = <instance>.setquota(root, limits)
"""
typ, dat = self._simple_command('SETQUOTA', root, limits)
return self._untagged_response(typ, dat, 'QUOTA')
def sort(self, sort_criteria, charset, *search_criteria):
"""IMAP4rev1 extension SORT command.
@@ -569,7 +689,7 @@ class IMAP4:
# raise self.error('unimplemented extension command: %s' % name)
if (sort_criteria[0],sort_criteria[-1]) != ('(',')'):
sort_criteria = '(%s)' % sort_criteria
typ, dat = apply(self._simple_command, (name, sort_criteria, charset) + search_criteria)
typ, dat = self._simple_command(name, sort_criteria, charset, *search_criteria)
return self._untagged_response(typ, dat, name)
@@ -604,6 +724,16 @@ class IMAP4:
return self._simple_command('SUBSCRIBE', mailbox)
def thread(self, threading_algorithm, charset, *search_criteria):
"""IMAPrev1 extension THREAD command.
(type, [data]) = <instance>.thread(threading_alogrithm, charset, search_criteria, ...)
"""
name = 'THREAD'
typ, dat = self._simple_command(name, threading_algorithm, charset, *search_criteria)
return self._untagged_response(typ, dat, name)
def uid(self, command, *args):
"""Execute "command arg ..." with messages identified by UID,
rather than message number.
@@ -613,13 +743,13 @@ class IMAP4:
Returns response appropriate to 'command'.
"""
command = command.upper()
if not Commands.has_key(command):
if not command in Commands:
raise self.error("Unknown IMAP4 UID command: %s" % command)
if self.state not in Commands[command]:
raise self.error('command %s illegal in state %s'
% (command, self.state))
name = 'UID'
typ, dat = apply(self._simple_command, (name, command) + args)
typ, dat = self._simple_command(name, command, *args)
if command in ('SEARCH', 'SORT'):
name = command
else:
@@ -648,9 +778,9 @@ class IMAP4:
name = name.upper()
#if not name in self.capabilities: # Let the server decide!
# raise self.error('unknown extension command: %s' % name)
if not Commands.has_key(name):
if not name in Commands:
Commands[name] = (self.state,)
return apply(self._simple_command, (name,) + args)
return self._simple_command(name, *args)
@@ -663,9 +793,9 @@ class IMAP4:
ur = self.untagged_responses
if __debug__:
if self.debug >= 5:
_mesg('untagged_responses[%s] %s += ["%s"]' %
self._mesg('untagged_responses[%s] %s += ["%s"]' %
(typ, len(ur.get(typ,'')), dat))
if ur.has_key(typ):
if typ in ur:
ur[typ].append(dat)
else:
ur[typ] = [dat]
@@ -685,10 +815,10 @@ class IMAP4:
'command %s illegal in state %s' % (name, self.state))
for typ in ('OK', 'NO', 'BAD'):
if self.untagged_responses.has_key(typ):
if typ in self.untagged_responses:
del self.untagged_responses[typ]
if self.untagged_responses.has_key('READ-ONLY') \
if 'READ-ONLY' in self.untagged_responses \
and not self.is_readonly:
raise self.readonly('mailbox status changed to READ-ONLY')
@@ -709,9 +839,9 @@ class IMAP4:
if __debug__:
if self.debug >= 4:
_mesg('> %s' % data)
self._mesg('> %s' % data)
else:
_log('> %s' % data)
self._log('> %s' % data)
try:
self.send('%s%s' % (data, CRLF))
@@ -735,7 +865,7 @@ class IMAP4:
if __debug__:
if self.debug >= 4:
_mesg('write literal size %s' % len(literal))
self._mesg('write literal size %s' % len(literal))
try:
self.send(literal)
@@ -776,7 +906,7 @@ class IMAP4:
if self._match(self.tagre, resp):
tag = self.mo.group('tag')
if not self.tagged_commands.has_key(tag):
if not tag in self.tagged_commands:
raise self.abort('unexpected tagged response: %s' % resp)
typ = self.mo.group('type')
@@ -814,7 +944,7 @@ class IMAP4:
size = int(self.mo.group('size'))
if __debug__:
if self.debug >= 4:
_mesg('read literal size %s' % size)
self._mesg('read literal size %s' % size)
data = self.read(size)
# Store response with literal as tuple
@@ -834,7 +964,7 @@ class IMAP4:
if __debug__:
if self.debug >= 1 and typ in ('NO', 'BAD', 'BYE'):
_mesg('%s response: %s' % (typ, dat))
self._mesg('%s response: %s' % (typ, dat))
return resp
@@ -857,7 +987,7 @@ class IMAP4:
except self.abort, val:
if __debug__:
if self.debug >= 1:
print_log()
self.print_log()
raise
@@ -872,9 +1002,9 @@ class IMAP4:
line = line[:-2]
if __debug__:
if self.debug >= 4:
_mesg('< %s' % line)
self._mesg('< %s' % line)
else:
_log('< %s' % line)
self._log('< %s' % line)
return line
@@ -886,7 +1016,7 @@ class IMAP4:
self.mo = cre.match(s)
if __debug__:
if self.mo is not None and self.debug >= 5:
_mesg("\tmatched r'%s' => %s" % (cre.pattern, `self.mo.groups()`))
self._mesg("\tmatched r'%s' => %r" % (cre.pattern, self.mo.groups()))
return self.mo is not None
@@ -905,9 +1035,9 @@ class IMAP4:
if type(arg) is not type(''):
return arg
if (arg[0],arg[-1]) in (('(',')'),('"','"')):
if len(arg) >= 2 and (arg[0],arg[-1]) in (('(',')'),('"','"')):
return arg
if self.mustquote.search(arg) is None:
if arg and self.mustquote.search(arg) is None:
return arg
return self._quote(arg)
@@ -922,23 +1052,203 @@ class IMAP4:
def _simple_command(self, name, *args):
return self._command_complete(name, apply(self._command, (name,) + args))
return self._command_complete(name, self._command(name, *args))
def _untagged_response(self, typ, dat, name):
if typ == 'NO':
return typ, dat
if not self.untagged_responses.has_key(name):
if not name in self.untagged_responses:
return typ, [None]
data = self.untagged_responses[name]
data = self.untagged_responses.pop(name)
if __debug__:
if self.debug >= 5:
_mesg('untagged_responses[%s] => %s' % (name, data))
del self.untagged_responses[name]
self._mesg('untagged_responses[%s] => %s' % (name, data))
return typ, data
if __debug__:
def _mesg(self, s, secs=None):
if secs is None:
secs = time.time()
tm = time.strftime('%M:%S', time.localtime(secs))
sys.stderr.write(' %s.%02d %s\n' % (tm, (secs*100)%100, s))
sys.stderr.flush()
def _dump_ur(self, dict):
# Dump untagged responses (in `dict').
l = dict.items()
if not l: return
t = '\n\t\t'
l = map(lambda x:'%s: "%s"' % (x[0], x[1][0] and '" "'.join(x[1]) or ''), l)
self._mesg('untagged responses dump:%s%s' % (t, t.join(l)))
def _log(self, line):
# Keep log of last `_cmd_log_len' interactions for debugging.
self._cmd_log[self._cmd_log_idx] = (line, time.time())
self._cmd_log_idx += 1
if self._cmd_log_idx >= self._cmd_log_len:
self._cmd_log_idx = 0
def print_log(self):
self._mesg('last %d IMAP4 interactions:' % len(self._cmd_log))
i, n = self._cmd_log_idx, self._cmd_log_len
while n:
try:
self._mesg(*self._cmd_log[i])
except:
pass
i += 1
if i >= self._cmd_log_len:
i = 0
n -= 1
class IMAP4_SSL(IMAP4):
"""IMAP4 client class over SSL connection
Instantiate with: IMAP4_SSL([host[, port[, keyfile[, certfile]]]])
host - host's name (default: localhost);
port - port number (default: standard IMAP4 SSL port).
keyfile - PEM formatted file that contains your private key (default: None);
certfile - PEM formatted certificate chain file (default: None);
for more documentation see the docstring of the parent class IMAP4.
"""
def __init__(self, host = '', port = IMAP4_SSL_PORT, keyfile = None, certfile = None):
self.keyfile = keyfile
self.certfile = certfile
IMAP4.__init__(self, host, port)
def open(self, host = '', port = IMAP4_SSL_PORT):
"""Setup connection to remote server on "host:port".
(default: localhost:standard IMAP4 SSL port).
This connection will be used by the routines:
read, readline, send, shutdown.
"""
self.host = host
self.port = port
self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.sock.connect((host, port))
self.sslobj = socket.ssl(self.sock, self.keyfile, self.certfile)
def read(self, size):
"""Read 'size' bytes from remote."""
# sslobj.read() sometimes returns < size bytes
chunks = []
read = 0
while read < size:
data = self.sslobj.read(min(size-read, 16384))
read += len(data)
chunks.append(data)
return ''.join(chunks)
def readline(self):
"""Read line from remote."""
# NB: socket.ssl needs a "readline" method, or perhaps a "makefile" method.
line = []
while 1:
char = self.sslobj.read(1)
line.append(char)
if char == "\n": return ''.join(line)
def send(self, data):
"""Send data to remote."""
# NB: socket.ssl needs a "sendall" method to match socket objects.
bytes = len(data)
while bytes > 0:
sent = self.sslobj.write(data)
if sent == bytes:
break # avoid copy
data = data[sent:]
bytes = bytes - sent
def shutdown(self):
"""Close I/O established in "open"."""
self.sock.close()
def socket(self):
"""Return socket instance used to connect to IMAP4 server.
socket = <instance>.socket()
"""
return self.sock
def ssl(self):
"""Return SSLObject instance used to communicate with the IMAP4 server.
ssl = <instance>.socket.ssl()
"""
return self.sslobj
class IMAP4_stream(IMAP4):
"""IMAP4 client class over a stream
Instantiate with: IMAP4_stream(command)
where "command" is a string that can be passed to os.popen2()
for more documentation see the docstring of the parent class IMAP4.
"""
def __init__(self, command):
self.command = command
IMAP4.__init__(self)
def open(self, host = None, port = None):
"""Setup a stream connection.
This connection will be used by the routines:
read, readline, send, shutdown.
"""
self.host = None # For compatibility with parent class
self.port = None
self.sock = None
self.file = None
self.writefile, self.readfile = os.popen2(self.command)
def read(self, size):
"""Read 'size' bytes from remote."""
return self.readfile.read(size)
def readline(self):
"""Read line from remote."""
return self.readfile.readline()
def send(self, data):
"""Send data to remote."""
self.writefile.write(data)
self.writefile.flush()
def shutdown(self):
"""Close I/O established in "open"."""
self.readfile.close()
self.writefile.close()
class _Authenticator:
@@ -1066,7 +1376,7 @@ def Time2Internaldate(date_time):
tt = time.localtime(date_time)
elif isinstance(date_time, (tuple, time.struct_time)):
tt = date_time
elif isinstance(date_time, str):
elif isinstance(date_time, str) and (date_time[0],date_time[-1]) == ('"','"'):
return date_time # Assume in correct format
else:
raise ValueError("date_time not of a known type")
@@ -1078,55 +1388,30 @@ def Time2Internaldate(date_time):
zone = -time.altzone
else:
zone = -time.timezone
return '"' + dt + " %+03d%02d" % divmod(zone/60, 60) + '"'
if __debug__:
def _mesg(s, secs=None):
if secs is None:
secs = time.time()
tm = time.strftime('%M:%S', time.localtime(secs))
sys.stderr.write(' %s.%02d %s\n' % (tm, (secs*100)%100, s))
sys.stderr.flush()
def _dump_ur(dict):
# Dump untagged responses (in `dict').
l = dict.items()
if not l: return
t = '\n\t\t'
l = map(lambda x:'%s: "%s"' % (x[0], x[1][0] and '" "'.join(x[1]) or ''), l)
_mesg('untagged responses dump:%s%s' % (t, t.join(l)))
_cmd_log = [] # Last `_cmd_log_len' interactions
_cmd_log_len = 10
def _log(line):
# Keep log of last `_cmd_log_len' interactions for debugging.
if len(_cmd_log) == _cmd_log_len:
del _cmd_log[0]
_cmd_log.append((time.time(), line))
def print_log():
_mesg('last %d IMAP4 interactions:' % len(_cmd_log))
for secs,line in _cmd_log:
_mesg(line, secs)
return '"' + dt + " %+03d%02d" % divmod(zone//60, 60) + '"'
if __name__ == '__main__':
# To test: invoke either as 'python imaplib.py [IMAP4_server_hostname]'
# or 'python imaplib.py -s "rsh IMAP4_server_hostname exec /etc/rimapd"'
# to test the IMAP4_stream class
import getopt, getpass
try:
optlist, args = getopt.getopt(sys.argv[1:], 'd:')
optlist, args = getopt.getopt(sys.argv[1:], 'd:s:')
except getopt.error, val:
pass
optlist, args = (), ()
stream_command = None
for opt,val in optlist:
if opt == '-d':
Debug = int(val)
elif opt == '-s':
stream_command = val
if not args: args = (stream_command,)
if not args: args = ('',)
@@ -1135,7 +1420,7 @@ if __name__ == '__main__':
USER = getpass.getuser()
PASSWD = getpass.getpass("IMAP password for %s on %s: " % (USER, host or "localhost"))
test_mesg = 'From: %(user)s@localhost%(lf)sSubject: IMAP4 test%(lf)s%(lf)sdata...%(lf)s' % {'user':USER, 'lf':CRLF}
test_mesg = 'From: %(user)s@localhost%(lf)sSubject: IMAP4 test%(lf)s%(lf)sdata...%(lf)s' % {'user':USER, 'lf':'\n'}
test_seq1 = (
('login', (USER, PASSWD)),
('create', ('/tmp/xxx 1',)),
@@ -1145,7 +1430,7 @@ if __name__ == '__main__':
('list', ('/tmp', 'yy*')),
('select', ('/tmp/yyz 2',)),
('search', (None, 'SUBJECT', 'test')),
('partial', ('1', 'RFC822', 1, 1024)),
('fetch', ('1', '(FLAGS INTERNALDATE RFC822)')),
('store', ('1', 'FLAGS', '(\Deleted)')),
('namespace', ()),
('expunge', ()),
@@ -1164,15 +1449,21 @@ if __name__ == '__main__':
)
def run(cmd, args):
_mesg('%s %s' % (cmd, args))
typ, dat = apply(getattr(M, cmd), args)
_mesg('%s => %s %s' % (cmd, typ, dat))
M._mesg('%s %s' % (cmd, args))
typ, dat = getattr(M, cmd)(*args)
M._mesg('%s => %s %s' % (cmd, typ, dat))
if typ == 'NO': raise dat[0]
return dat
try:
M = IMAP4(host)
_mesg('PROTOCOL_VERSION = %s' % M.PROTOCOL_VERSION)
_mesg('CAPABILITIES = %s' % `M.capabilities`)
if stream_command:
M = IMAP4_stream(stream_command)
else:
M = IMAP4(host)
if M.state == 'AUTH':
test_seq1 = test_seq1[1:] # Login not needed
M._mesg('PROTOCOL_VERSION = %s' % M.PROTOCOL_VERSION)
M._mesg('CAPABILITIES = %r' % (M.capabilities,))
for cmd,args in test_seq1:
run(cmd, args)