vmm: comparison VirtualMailManager/Handler.py

equal deleted inserted replaced

-:d619e97a8f18
+:4dc2edf02d11
 from VirtualMailManager.Transport import Transport
 _ = lambda msg: msg
+CFG_FILE = 'vmm.cfg'
+CFG_PATH = '/root:/usr/local/etc:/etc'
 RE_DOMAIN_SEARCH = """^[a-z0-9-\.]+$"""
 TYPE_ACCOUNT = 0x1
 TYPE_ALIAS = 0x2
 TYPE_RELOCATED = 0x4
 OTHER_TYPES = {
 class Handler(object):
 """Wrapper class to simplify the access on all the stuff from
 VirtualMailManager"""
-__slots__ = ('_Cfg', '_cfgFileName', '_dbh', '__warnings')
+__slots__ = ('_cfg', '_cfg_fname', '_dbh', '__warnings')
 def __init__(self, skip_some_checks=False):
 """Creates a new Handler instance.
 ``skip_some_checks`` : bool
 argument may be ``True``. By default it is ``False`` and
 all checks will be performed.
 Throws a NotRootError if your uid is greater 0.
 """
-self._cfgFileName = ''
+self._cfg_fname = ''
 self.__warnings = []
-self._Cfg = None
+self._cfg = None
 self._dbh = None
 if os.geteuid():
 raise NotRootError(_(u"You are not root.\n\tGood bye!\n"),
 CONF_NOPERM)
-if self.__chkCfgFile():
+if self.__check_cfg_file():
-self._Cfg = Cfg(self._cfgFileName)
+self._cfg = Cfg(self._cfg_fname)
-self._Cfg.load()
+self._cfg.load()
 if not skip_some_checks:
-self._Cfg.check()
+self._cfg.check()
 self._chkenv()
-def __findCfgFile(self):
+def __find_cfg_file(self):
-for path in ['/root', '/usr/local/etc', '/etc']:
+"""Search the CFG_FILE in CFG_PATH.
-tmp = os.path.join(path, 'vmm.cfg')
+Raise a VMMError when no vmm.cfg could be found.
+"""
+for path in CFG_PATH.split(':'):
+tmp = os.path.join(path, CFG_FILE)
 if os.path.isfile(tmp):
-self._cfgFileName = tmp
+self._cfg_fname = tmp
 break
-if not len(self._cfgFileName):
+if not self._cfg_fname:
-raise VMMError(_(u"No 'vmm.cfg' found in: "
+raise VMMError(_(u"Could not find '%(cfg_file)s' in: "
-u"/root:/usr/local/etc:/etc"), CONF_NOFILE)
+u"'%(cfg_path)s'") % {'cfg_file': CFG_FILE,
+'cfg_path' : CFG_PATH}, CONF_NOFILE)
-def __chkCfgFile(self):
+def __check_cfg_file(self):
 """Checks the configuration file, returns bool"""
-self.__findCfgFile()
+self.__find_cfg_file()
-fstat = os.stat(self._cfgFileName)
+fstat = os.stat(self._cfg_fname)
 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 PermissionError(_(u"wrong permissions for '%(file)s': "
 u"%(perms)s\n`chmod 0600 %(file)s` would "
-u"be great.") % {'file': self._cfgFileName,
+u"be great.") % {'file': self._cfg_fname,
 'perms': fmode}, CONF_WRONGPERM)
 else:
 return True
 def _chkenv(self):
-""""""
+"""Make sure our base_directory is a directory and that all
-basedir = self._Cfg.dget('misc.base_directory')
+required executables exists and are executable.
+If not, a VMMError will be raised"""
+basedir = self._cfg.dget('misc.base_directory')
 if not os.path.exists(basedir):
 old_umask = os.umask(0006)
 os.makedirs(basedir, 0771)
 os.chown(basedir, 0, 0)
 os.umask(old_umask)
 elif not os.path.isdir(basedir):
-raise VMMError(_(u"'%s' is not a directory.\n(vmm.cfg: section "
+raise VMMError(_(u"'%(path)s' is not a directory.\n(%(cfg_file)s: "
-u"'misc', option 'base_directory')") % basedir,
+u"section 'misc', option 'base_directory')") %
+{'path': basedir, 'cfg_file': self._cfg_fname},
 NO_SUCH_DIRECTORY)
-for opt, val in self._Cfg.items('bin'):
+for opt, val in self._cfg.items('bin'):
 try:
 exec_ok(val)
 except VMMError, err:
 if err.code is NO_SUCH_BINARY:
-raise VMMError(_(u"'%(binary)s' doesn't exist.\n(vmm.cfg: "
+raise VMMError(_(u"'%(binary)s' doesn't exist.\n"
-u"section 'bin', option '%(option)s')") %
+u"(%(cfg_file)s: section 'bin', option "
-{'binary': val, 'option': opt}, err.code)
+u"'%(option)s')") % {'binary': val,
+'cfg_file': self._cfg_fname, 'option': opt},
+err.code)
 elif err.code is NOT_EXECUTABLE:
 raise VMMError(_(u"'%(binary)s' is not executable.\n"
-u"(vmm.cfg: section 'bin', option "
+u"(%(cfg_file)s: section 'bin', option "
 u"'%(option)s')") % {'binary': val,
-'option': opt}, err.code)
+'cfg_file': self._cfg_fname, 'option': opt},
+err.code)
 else:
 raise
 def __dbConnect(self):
 """Creates a pyPgSQL.PgSQL.connection instance."""
 if self._dbh is None or (isinstance(self._dbh, PgSQL.Connection) and
 not self._dbh._isOpen):
 try:
 self._dbh = PgSQL.connect(
-database=self._Cfg.dget('database.name'),
+database=self._cfg.dget('database.name'),
-user=self._Cfg.pget('database.user'),
+user=self._cfg.pget('database.user'),
-host=self._Cfg.dget('database.host'),
+host=self._cfg.dget('database.host'),
-password=self._Cfg.pget('database.pass'),
+password=self._cfg.pget('database.pass'),
 client_encoding='utf8', unicode_results=True)
 dbc = self._dbh.cursor()
 dbc.execute("SET NAMES 'UTF8'")
 dbc.close()
 except PgSQL.libpq.DatabaseError, e:
 Keyword arguments:
 directory -- the directory to summarize recursively disk usage for
 """
 if self.__isdir(directory):
-return Popen([self._Cfg.dget('bin.du'), "-hs", directory],
+return Popen([self._cfg.dget('bin.du'), "-hs", directory],
 stdout=PIPE).communicate()[0].split('\t')[0]
 else:
 return 0
 def __isdir(self, directory):
 return isdir
 def __make_domain_dir(self, domain):
 cwd = os.getcwd()
 hashdir, domdir = domain.directory.split(os.path.sep)[-2:]
-os.chdir(self._Cfg.dget('misc.base_directory'))
+os.chdir(self._cfg.dget('misc.base_directory'))
 if not os.path.isdir(hashdir):
 os.mkdir(hashdir, 0711)
 os.chown(hashdir, 0, 0)
 os.mkdir(os.path.join(hashdir, domdir),
-self._Cfg.dget('domain.directory_mode'))
+self._cfg.dget('domain.directory_mode'))
 os.chown(domain.directory, 0, domain.gid)
 os.chdir(cwd)
 def __make_home(self, account):
 """Create a home directory for the new Account *account*."""
 os.umask(0007)
 os.chdir(account.domain_directory)
-os.mkdir('%s' % account.uid, self._Cfg.dget('account.directory_mode'))
+os.mkdir('%s' % account.uid, self._cfg.dget('account.directory_mode'))
 os.chown('%s' % account.uid, account.uid, account.gid)
 def __userDirDelete(self, domdir, uid, gid):
 if uid > 0 and gid > 0:
 userdir = '%s' % uid
 def __domDirDelete(self, domdir, gid):
 if gid > 0:
 if not self.__isdir(domdir):
 return
-basedir = self._Cfg.dget('misc.base_directory')
+basedir = self._cfg.dget('misc.base_directory')
 domdirdirs = domdir.replace(basedir + '/', '').split('/')
 domdirparent = os.path.join(basedir, domdirdirs[0])
 if basedir.count('..') or domdir.count('..'):
 raise VMMError(_(u'Found ".." in domain directory path.'),
 FOUND_DOTS_IN_PATH)
 def cfg_dget(self, option):
 """Get the configured value of the *option* (section.option).
 When the option was not configured its default value will be
 returned."""
-return self._Cfg.dget(option)
+return self._cfg.dget(option)
 def cfg_pget(self, option):
 """Get the configured value of the *option* (section.option)."""
-return self._Cfg.pget(option)
+return self._cfg.pget(option)
 def cfg_install(self):
 """Installs the cfg_dget method as ``cfg_dget`` into the built-in
 namespace."""
 import __builtin__
 assert 'cfg_dget' not in __builtin__.__dict__
-__builtin__.__dict__['cfg_dget'] = self._Cfg.dget
+__builtin__.__dict__['cfg_dget'] = self._cfg.dget
 def domainAdd(self, domainname, transport=None):
 dom = self.__getDomain(domainname)
 if transport is None:
 dom.set_transport(Transport(self._dbh,
-transport=self._Cfg.dget('misc.transport')))
+transport=self._cfg.dget('misc.transport')))
 else:
 dom.set_transport(Transport(self._dbh, transport=transport))
-dom.set_directory(self._Cfg.dget('misc.base_directory'))
+dom.set_directory(self._cfg.dget('misc.base_directory'))
 dom.save()
 self.__make_domain_dir(dom)
 def domainTransport(self, domainname, transport, force=None):
 if force is not None and force != 'force':
 raise DomainError(_(u"Invalid argument: '%s'") % force,
 INVALID_ARGUMENT)
 dom = self.__getDomain(domainname)
 gid = dom.gid
 domdir = dom.directory
-if self._Cfg.dget('domain.force_deletion') or force == 'delall':
+if self._cfg.dget('domain.force_deletion') or force == 'delall':
 dom.delete(True, True)
 elif force == 'deluser':
 dom.delete(deluser=True)
 elif force == 'delalias':
 dom.delete(delalias=True)
 else:
 dom.delete()
-if self._Cfg.dget('domain.delete_directory'):
+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']:
 acc.save()
 oldpwd = os.getcwd()
 self.__make_home(acc)
 mailbox = new_mailbox(acc)
 mailbox.create()
-folders = self._Cfg.dget('mailbox.folders').split(':')
+folders = self._cfg.dget('mailbox.folders').split(':')
 if any(folders):
 bad = mailbox.add_boxes(folders,
-self._Cfg.dget('mailbox.subscribe'))
+self._cfg.dget('mailbox.subscribe'))
 if bad:
 self.__warnings.append(_(u"Skipped mailbox folders:") +
 '\n\t- ' + '\n\t- '.join(bad))
 os.chdir(oldpwd)
 uid = acc.uid
 gid = acc.gid
 dom_dir = acc.domain_directory
 acc_dir = acc.home
 acc.delete(bool(force))
-if self._Cfg.dget('account.delete_directory'):
+if self._cfg.dget('account.delete_directory'):
 try:
 self.__userDirDelete(dom_dir, uid, gid)
 except VMMError, err:
 if err.code in (FOUND_DOTS_IN_PATH, MAILDIR_PERM_MISMATCH,
 NO_SUCH_DIRECTORY):
 if not acc:
 if not self._is_other_address(acc.address, TYPE_ACCOUNT):
 raise VMMError(_(u"The account '%s' doesn't exist.") %
 acc.address, NO_SUCH_ACCOUNT)
 info = acc.get_info()
-if self._Cfg.dget('account.disk_usage') or details in ('du', 'full'):
+if self._cfg.dget('account.disk_usage') or details in ('du', 'full'):
 path = os.path.join(acc.home, acc.mail_location.directory)
 info['disk usage'] = self.__getDiskUsage(path)
 if details in (None, 'du'):
 return info
 if details in ('aliases', 'full'):

branch	v0.6.x
changeset 318	4dc2edf02d11
parent 316	31d8931dc535
child 319	f4956b4ceba1