/usr/share/pyshared/twisted/mail/pb.py is in python-twisted-mail 12.0.0-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
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 | # Copyright (c) Twisted Matrix Laboratories.
# See LICENSE for details.
from twisted.spread import pb
from twisted.spread import banana
import os
import types
class Maildir(pb.Referenceable):
def __init__(self, directory, rootDirectory):
self.virtualDirectory = directory
self.rootDirectory = rootDirectory
self.directory = os.path.join(rootDirectory, directory)
def getFolderMessage(self, folder, name):
if '/' in name:
raise IOError("can only open files in '%s' directory'" % folder)
fp = open(os.path.join(self.directory, 'new', name))
try:
return fp.read()
finally:
fp.close()
def deleteFolderMessage(self, folder, name):
if '/' in name:
raise IOError("can only delete files in '%s' directory'" % folder)
os.rename(os.path.join(self.directory, folder, name),
os.path.join(self.rootDirectory, '.Trash', folder, name))
def deleteNewMessage(self, name):
return self.deleteFolderMessage('new', name)
remote_deleteNewMessage = deleteNewMessage
def deleteCurMessage(self, name):
return self.deleteFolderMessage('cur', name)
remote_deleteCurMessage = deleteCurMessage
def getNewMessages(self):
return os.listdir(os.path.join(self.directory, 'new'))
remote_getNewMessages = getNewMessages
def getCurMessages(self):
return os.listdir(os.path.join(self.directory, 'cur'))
remote_getCurMessages = getCurMessages
def getNewMessage(self, name):
return self.getFolderMessage('new', name)
remote_getNewMessage = getNewMessage
def getCurMessage(self, name):
return self.getFolderMessage('cur', name)
remote_getCurMessage = getCurMessage
def getSubFolder(self, name):
if name[0] == '.':
raise IOError("subfolder name cannot begin with a '.'")
name = name.replace('/', ':')
if self.virtualDirectoy == '.':
name = '.'+name
else:
name = self.virtualDirectory+':'+name
if not self._isSubFolder(name):
raise IOError("not a subfolder")
return Maildir(name, self.rootDirectory)
remote_getSubFolder = getSubFolder
def _isSubFolder(self, name):
return (not os.path.isdir(os.path.join(self.rootDirectory, name)) or
not os.path.isfile(os.path.join(self.rootDirectory, name,
'maildirfolder')))
class MaildirCollection(pb.Referenceable):
def __init__(self, root):
self.root = root
def getSubFolders(self):
return os.listdir(self.getRoot())
remote_getSubFolders = getSubFolders
def getSubFolder(self, name):
if '/' in name or name[0] == '.':
raise IOError("invalid name")
return Maildir('.', os.path.join(self.getRoot(), name))
remote_getSubFolder = getSubFolder
class MaildirBroker(pb.Broker):
def proto_getCollection(self, requestID, name, domain, password):
collection = self._getCollection()
if collection is None:
self.sendError(requestID, "permission denied")
else:
self.sendAnswer(requestID, collection)
def getCollection(self, name, domain, password):
if not self.domains.has_key(domain):
return
domain = self.domains[domain]
if (domain.dbm.has_key(name) and
domain.dbm[name] == password):
return MaildirCollection(domain.userDirectory(name))
class MaildirClient(pb.Broker):
def getCollection(self, name, domain, password, callback, errback):
requestID = self.newRequestID()
self.waitingForAnswers[requestID] = callback, errback
self.sendCall("getCollection", requestID, name, domain, password)
|