vmm: comparison VirtualMailManager/Handler.py

equal deleted inserted replaced

-:1903d4ce97d7
+:db77501aeaed
 class Handler(object):
 """Wrapper class to simplify the access on all the stuff from
 VirtualMailManager"""
 __slots__ = ('_Cfg', '_cfgFileName', '__dbh', '_scheme', '__warnings',
 '_postconf')
 def __init__(self, skip_some_checks=False):
 """Creates a new Handler instance.
 ``skip_some_checks`` : bool
 When a derived class knows how to handle all checks this
 self.__findCfgFile()
 fstat = os.stat(self._cfgFileName)
 fmode = int(oct(fstat.st_mode & 0777))
 if fmode % 100 and fstat.st_uid != fstat.st_gid or \
 fmode % 10 and fstat.st_uid == fstat.st_gid:
 raise VMMPermException(_(
 u'fix permissions (%(perms)s) for “%(file)s”\n\
 `chmod 0600 %(file)s` would be great.') % {'file':
 self._cfgFileName, 'perms': fmode}, ERR.CONF_WRONGPERM)
 else:
 return True
 exec_ok(val)
 except VMMException, e:
 code = e.code()
 if code is ERR.NO_SUCH_BINARY:
 raise VMMException(_(u'“%(binary)s” doesn\'t exist.\n\
-(vmm.cfg: section "bin", option "%(option)s")') %{'binary': val,'option': opt},
+(vmm.cfg: section "bin", option "%(option)s")') %
+{'binary': val, 'option': opt},
 ERR.NO_SUCH_BINARY)
 elif code is ERR.NOT_EXECUTABLE:
-raise VMMException(_(u'“%(binary)s” is not executable.\n\
+raise VMMException(_(u'“%(binary)s” is not executable.\
-(vmm.cfg: section "bin", option "%(option)s")') %{'binary': val,'option': opt},
+\n(vmm.cfg: section "bin", option "%(option)s")') %
+{'binary': val, 'option': opt},
 ERR.NOT_EXECUTABLE)
 else:
 raise
 def __dbConnect(self):
 accountExists = staticmethod(accountExists)
 def aliasExists(dbh, address):
 sql = "SELECT DISTINCT gid FROM alias WHERE gid = (SELECT gid FROM\
 domain_name WHERE domainname = '%s') AND address = '%s'" % (
 address._domainname, address._localpart)
 return Handler._exists(dbh, sql)
 aliasExists = staticmethod(aliasExists)
 def relocatedExists(dbh, address):
 sql = "SELECT gid FROM relocated WHERE gid = (SELECT gid FROM\
 domain_name WHERE domainname = '%s') AND address = '%s'" % (
 address._domainname, address._localpart)
 return Handler._exists(dbh, sql)
 relocatedExists = staticmethod(relocatedExists)
 def __getAccount(self, address, password=None):
 self.__dbConnect()
 address = EmailAddress(address)
 if not password is None:
 address = EmailAddress(address)
 if destination is not None:
 destination = EmailAddress(destination)
 return Alias(self.__dbh, address, destination)
-def __getRelocated(self,address, destination=None):
+def __getRelocated(self, address, destination=None):
 self.__dbConnect()
 address = EmailAddress(address)
 if destination is not None:
 destination = EmailAddress(destination)
 return Relocated(self.__dbh, address, destination)
 def __domDirMake(self, domdir, gid):
 os.umask(0006)
 oldpwd = os.getcwd()
 basedir = self._Cfg.dget('misc.base_directory')
-domdirdirs = domdir.replace(basedir+'/', '').split('/')
+domdirdirs = domdir.replace(basedir + '/', '').split('/')
 os.chdir(basedir)
 if not os.path.isdir(domdirdirs[0]):
 self.__makedir(domdirdirs[0], 489, 0,
 self._Cfg.dget('misc.gid_mail'))
 os.umask(0007)
 self.__makedir(domdirdirs[1], self._Cfg.dget('domain.directory_mode'),
 0, gid)
 os.chdir(oldpwd)
-def __subscribeFL(self, folderlist, uid, gid):
+def __subscribe(self, folderlist, uid, gid):
+"""Creates a subscriptions file with the mailboxes from `folderlist`"""
 fname = os.path.join(self._Cfg.dget('maildir.name'), 'subscriptions')
-sf = file(fname, 'w')
+sf = open(fname, 'w')
 for f in folderlist:
-sf.write(f+'\n')
+sf.write('%s\n' % f)
 sf.flush()
 sf.close()
 os.chown(fname, uid, gid)
 os.chmod(fname, 384)
 oldpwd = os.getcwd()
 os.chdir(domdir)
 maildir = self._Cfg.dget('maildir.name')
 folders = [maildir]
+append = folders.append
 for folder in self._Cfg.dget('maildir.folders').split(':'):
 folder = folder.strip()
-if len(folder) and not folder.count('..')\
+if len(folder) and not folder.count('..'):
-and re.match(RE_MBOX_NAMES, folder):
+if re.match(RE_MBOX_NAMES, folder):
-folders.append('%s/.%s' % (maildir, folder))
+append('%s/.%s' % (maildir, folder))
+else:
+self.__warnings.append(_('Skipped mailbox folder: %r') %
+folder)
+else:
+self.__warnings.append(_('Skipped mailbox folder: %r') %
+folder)
 subdirs = ['cur', 'new', 'tmp']
 mode = self._Cfg.dget('account.directory_mode')
 self.__makedir('%s' % uid, mode, uid, gid)
 os.chdir('%s' % uid)
 for folder in folders:
 self.__makedir(folder, mode, uid, gid)
 for subdir in subdirs:
 self.__makedir(os.path.join(folder, subdir), mode, uid, gid)
-self.__subscribeFL([f.replace(maildir+'/.', '') for f in folders[1:]],
+self.__subscribe((f.replace(maildir + '/.', '') for f in folders[1:]),
 uid, gid)
 os.chdir(oldpwd)
 def __userDirDelete(self, domdir, uid, gid):
 if uid > 0 and gid > 0:
 userdir = '%s' % uid
 if userdir.count('..') or domdir.count('..'):
 raise VMMException(_(u'Found ".." in home directory path.'),
 ERR.FOUND_DOTS_IN_PATH)
 if os.path.isdir(domdir):
 os.chdir(domdir)
 if os.path.isdir(userdir):
 mdstat = os.stat(userdir)
 if (mdstat.st_uid, mdstat.st_gid) != (uid, gid):
-raise VMMException(
+raise VMMException(_(
-_(u'Detected owner/group mismatch in home directory.'),
+u'Detected owner/group mismatch in home directory.'),
 ERR.MAILDIR_PERM_MISMATCH)
 rmtree(userdir, ignore_errors=True)
 else:
 raise VMMException(_(u"No such directory: %s") %
 os.path.join(domdir, userdir), ERR.NO_SUCH_DIRECTORY)
 def __domDirDelete(self, domdir, gid):
 if gid > 0:
 if not self.__isdir(domdir):
 return
 basedir = self._Cfg.dget('misc.base_directory')
-domdirdirs = domdir.replace(basedir+'/', '').split('/')
+domdirdirs = domdir.replace(basedir + '/', '').split('/')
 domdirparent = os.path.join(basedir, domdirdirs[0])
 if basedir.count('..') or domdir.count('..'):
 raise VMMException(_(u'Found ".." in domain directory path.'),
 ERR.FOUND_DOTS_IN_PATH)
 if os.path.isdir(domdirparent):
 return '{%s}%s' % (self._scheme, self.__pwMD5(password, user))
 elif self._scheme == 'MD4':
 return '{%s}%s' % (self._scheme, self.__pwMD4(password))
 elif self._scheme in ['SMD5', 'SSHA', 'CRAM-MD5', 'HMAC-MD5',
 'LANMAN', 'NTLM', 'RPA']:
-return Popen([self._Cfg.dget('bin.dovecotpw'), '-s',
+return Popen([self._Cfg.dget('bin.dovecotpw'),
-self._scheme,'-p',password],stdout=PIPE).communicate()[0][:-1]
+'-s', self._scheme, '-p', password],
+stdout=PIPE).communicate()[0][:-1]
 else:
 return '{%s}%s' % (self._scheme, password)
 def hasWarnings(self):
 """Checks if warnings are present, returns bool."""
 dom.updateTransport(transport)
 else:
 dom.updateTransport(transport, force=True)
 def domainDelete(self, domainname, force=None):
-if not force is None and force not in ['deluser','delalias','delall']:
+if not force is None and force not in ['deluser', 'delalias',
-raise VMMDomainException(_(u"Invalid argument: “%s”") % force,
+'delall']:
-ERR.INVALID_OPTION)
+raise VMMDomainException(_(u'Invalid argument: “%s”') %
+force, ERR.INVALID_OPTION)
 dom = self.__getDomain(domainname)
 gid = dom.getID()
 domdir = dom.getDir()
 if self._Cfg.dget('domain.force_deletion') or force == 'delall':
 dom.delete(True, True)
 if self._Cfg.dget('domain.delete_directory'):
 self.__domDirDelete(domdir, gid)
 def domainInfo(self, domainname, details=None):
 if details not in [None, 'accounts', 'aliasdomains', 'aliases', 'full',
-'relocated', 'detailed']:
+'relocated']:
 raise VMMException(_(u'Invalid argument: “%s”') % details,
 ERR.INVALID_AGUMENT)
-if details == 'detailed':
-details = 'full'
-self.__warnings.append(_(u'\
-The keyword “detailed” is deprecated and will be removed in a future release.\n\
-Please use the keyword “full” to get full details.'))
 dom = self.__getDomain(domainname)
 dominfo = dom.getInfo()
 if dominfo['domainname'].startswith('xn--'):
 dominfo['domainname'] += ' (%s)' % ace2idna(dominfo['domainname'])
 if details is None:
 alias.save(long(self._postconf.read('virtual_alias_expansion_limit')))
 gid = self.__getDomain(alias._dest._domainname).getID()
 if gid > 0 and (not Handler.accountExists(self.__dbh, alias._dest) and
 not Handler.aliasExists(self.__dbh, alias._dest)):
 self.__warnings.append(
 _(u"The destination account/alias “%s” doesn't exist.") %
 alias._dest)
 def userDelete(self, emailaddress, force=None):
 if force not in [None, 'delalias']:
 raise VMMException(_(u"Invalid argument: “%s”") % force,
 ERR.MAILDIR_PERM_MISMATCH, ERR.NO_SUCH_DIRECTORY]:
 warning = _(u"""\
 The account has been successfully deleted from the database.
 But an error occurred while deleting the following directory:
 “%(directory)s”
-Reason: %(reason)s""") % {'directory': acc.getDir('home'),'reason': e.msg()}
+Reason: %(reason)s""") %
+{'directory': acc.getDir('home'), 'reason': e.msg()}
 self.__warnings.append(warning)
 else:
-raise e
+raise
 def aliasInfo(self, aliasaddress):
 alias = self.__getAlias(aliasaddress)
 return alias.getInfo()
 if password is None or (isinstance(password, basestring) and
 not len(password)):
 raise ValueError('could not accept password: %r' % password)
 acc = self.__getAccount(emailaddress)
 if acc.getUID() == 0:
-raise VMMException(_(u"Account doesn't exist"), ERR.NO_SUCH_ACCOUNT)
+raise VMMException(_(u"Account doesn't exist"),
+ERR.NO_SUCH_ACCOUNT)
 acc.modify('password', self.__pwhash(password, user=emailaddress))
 def userName(self, emailaddress, name):
 acc = self.__getAccount(emailaddress)
 acc.modify('name', name)

branch	v0.6.x
changeset 191	db77501aeaed
parent 190	1903d4ce97d7
child 192	0854fb9f3bc5