--- a/VirtualMailManager/Handler.py Wed Jul 28 01:03:56 2010 +0000
+++ /dev/null Thu Jan 01 00:00:00 1970 +0000
@@ -1,657 +0,0 @@
-# -*- coding: UTF-8 -*-
-# Copyright (c) 2007 - 2010, Pascal Volk
-# See COPYING for distribution information.
-
-"""
- VirtualMailManager.Handler
-
- A wrapper class. It wraps round all other classes and does some
- dependencies checks.
-
- Additionally it communicates with the PostgreSQL database, creates
- or deletes directories of domains or users.
-"""
-
-import os
-import re
-
-from shutil import rmtree
-from subprocess import Popen, PIPE
-
-from pyPgSQL import PgSQL # python-pgsql - http://pypgsql.sourceforge.net
-
-from VirtualMailManager.Account import Account
-from VirtualMailManager.Alias import Alias
-from VirtualMailManager.AliasDomain import AliasDomain
-from VirtualMailManager.common import exec_ok
-from VirtualMailManager.Config import Config as Cfg
-from VirtualMailManager.constants import \
- ACCOUNT_EXISTS, ALIAS_EXISTS, CONF_NOFILE, CONF_NOPERM, CONF_WRONGPERM, \
- DATABASE_ERROR, DOMAINDIR_GROUP_MISMATCH, DOMAIN_INVALID, \
- FOUND_DOTS_IN_PATH, INVALID_ARGUMENT, MAILDIR_PERM_MISMATCH, \
- NOT_EXECUTABLE, NO_SUCH_ACCOUNT, NO_SUCH_ALIAS, NO_SUCH_BINARY, \
- NO_SUCH_DIRECTORY, NO_SUCH_RELOCATED, RELOCATED_EXISTS
-from VirtualMailManager.Domain import Domain, get_gid
-from VirtualMailManager.EmailAddress import EmailAddress
-from VirtualMailManager.errors import \
- DomainError, NotRootError, PermissionError, VMMError
-from VirtualMailManager.mailbox import new as new_mailbox
-from VirtualMailManager.pycompat import any
-from VirtualMailManager.Relocated import Relocated
-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 = {
- TYPE_ACCOUNT: (_(u'an account'), ACCOUNT_EXISTS),
- TYPE_ALIAS: (_(u'an alias'), ALIAS_EXISTS),
- TYPE_RELOCATED: (_(u'a relocated user'), RELOCATED_EXISTS),
-}
-
-
-class Handler(object):
- """Wrapper class to simplify the access on all the stuff from
- VirtualMailManager"""
- __slots__ = ('_cfg', '_cfg_fname', '_dbh', '_warnings')
-
- 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
- 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._cfg_fname = ''
- self._warnings = []
- self._cfg = None
- self._dbh = None
-
- if os.geteuid():
- raise NotRootError(_(u"You are not root.\n\tGood bye!\n"),
- CONF_NOPERM)
- if self._check_cfg_file():
- self._cfg = Cfg(self._cfg_fname)
- self._cfg.load()
- if not skip_some_checks:
- self._cfg.check()
- self._chkenv()
-
- def _find_cfg_file(self):
- """Search the CFG_FILE in CFG_PATH.
- 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._cfg_fname = tmp
- break
- if not self._cfg_fname:
- raise VMMError(_(u"Could not find '%(cfg_file)s' in: "
- u"'%(cfg_path)s'") % {'cfg_file': CFG_FILE,
- 'cfg_path': CFG_PATH}, CONF_NOFILE)
-
- def _check_cfg_file(self):
- """Checks the configuration file, returns bool"""
- self._find_cfg_file()
- 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._cfg_fname,
- 'perms': fmode}, CONF_WRONGPERM)
- else:
- return True
-
- def _chkenv(self):
- """Make sure our base_directory is a directory and that all
- 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"'%(path)s' is not a directory.\n(%(cfg_file)s: "
- u"section 'misc', option 'base_directory')") %
- {'path': basedir, 'cfg_file': self._cfg_fname},
- NO_SUCH_DIRECTORY)
- 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"
- u"(%(cfg_file)s: section 'bin', option "
- 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"(%(cfg_file)s: section 'bin', option "
- u"'%(option)s')") % {'binary': val,
- 'cfg_file': self._cfg_fname, 'option': opt},
- err.code)
- else:
- raise
-
- def _db_connect(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'),
- user=self._cfg.pget('database.user'),
- host=self._cfg.dget('database.host'),
- 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, err:
- raise VMMError(str(err), DATABASE_ERROR)
-
- def _chk_other_address_types(self, address, exclude):
- """Checks if the EmailAddress *address* is known as `TYPE_ACCOUNT`,
- `TYPE_ALIAS` or `TYPE_RELOCATED`, but not as the `TYPE_*` specified
- by *exclude*. If the *address* is known as one of the `TYPE_*`s
- the according `TYPE_*` constant will be returned. Otherwise 0 will
- be returned."""
- assert exclude in (TYPE_ACCOUNT, TYPE_ALIAS, TYPE_RELOCATED) and \
- isinstance(address, EmailAddress)
- if exclude is not TYPE_ACCOUNT:
- account = Account(self._dbh, address)
- if account:
- return TYPE_ACCOUNT
- if exclude is not TYPE_ALIAS:
- alias = Alias(self._dbh, address)
- if alias:
- return TYPE_ALIAS
- if exclude is not TYPE_RELOCATED:
- relocated = Relocated(self._dbh, address)
- if relocated:
- return TYPE_RELOCATED
- return 0
-
- def _is_other_address(self, address, exclude):
- """Checks if *address* is known for an Account (TYPE_ACCOUNT),
- Alias (TYPE_ALIAS) or Relocated (TYPE_RELOCATED), except for
- *exclude*. Returns `False` if the address is not known for other
- types.
-
- Raises a `VMMError` if the address is known.
- """
- other = self._chk_other_address_types(address, exclude)
- if not other:
- return False
- msg = _(u"There is already %(a_type)s with the address '%(address)s'.")
- raise VMMError(msg % {'a_type': OTHER_TYPES[other][0],
- 'address': address}, OTHER_TYPES[other][1])
-
- def _get_account(self, address):
- """Return an Account instances for the given address (str)."""
- address = EmailAddress(address)
- self._db_connect()
- return Account(self._dbh, address)
-
- def _get_alias(self, address):
- """Return an Alias instances for the given address (str)."""
- address = EmailAddress(address)
- self._db_connect()
- return Alias(self._dbh, address)
-
- def _get_relocated(self, address):
- """Return a Relocated instances for the given address (str)."""
- address = EmailAddress(address)
- self._db_connect()
- return Relocated(self._dbh, address)
-
- def _get_domain(self, domainname):
- """Return a Domain instances for the given domain name (str)."""
- self._db_connect()
- return Domain(self._dbh, domainname)
-
- def _get_disk_usage(self, directory):
- """Estimate file space usage for the given directory.
-
- Keyword arguments:
- directory -- the directory to summarize recursively disk usage for
- """
- if self._isdir(directory):
- return Popen([self._cfg.dget('bin.du'), "-hs", directory],
- stdout=PIPE).communicate()[0].split('\t')[0]
- else:
- return 0
-
- def _isdir(self, directory):
- """Check if `directory` is a directory. Returns bool.
- When `directory` isn't a directory, a warning will be appended to
- _warnings."""
- isdir = os.path.isdir(directory)
- if not isdir:
- self._warnings.append(_('No such directory: %s') % directory)
- return isdir
-
- def _make_domain_dir(self, domain):
- """Create a directory for the `domain` and its accounts."""
- cwd = os.getcwd()
- hashdir, domdir = domain.directory.split(os.path.sep)[-2:]
- 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'))
- 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.chown('%s' % account.uid, account.uid, account.gid)
-
- def _delete_home(self, domdir, uid, gid):
- """Delete a user's home directory."""
- if uid > 0 and gid > 0:
- userdir = '%s' % uid
- if userdir.count('..') or domdir.count('..'):
- raise VMMError(_(u'Found ".." in home directory path.'),
- 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 VMMError(_(u'Detected owner/group mismatch in '
- u'home directory.'),
- MAILDIR_PERM_MISMATCH)
- rmtree(userdir, ignore_errors=True)
- else:
- raise VMMError(_(u"No such directory: %s") %
- os.path.join(domdir, userdir),
- NO_SUCH_DIRECTORY)
-
- def _delete_domain_dir(self, domdir, gid):
- """Delete a domain's directory."""
- if gid > 0:
- if not self._isdir(domdir):
- return
- 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)
- if os.path.isdir(domdirparent):
- os.chdir(domdirparent)
- if os.lstat(domdirdirs[1]).st_gid != gid:
- raise VMMError(_(u'Detected group mismatch in domain '
- u'directory.'), DOMAINDIR_GROUP_MISMATCH)
- rmtree(domdirdirs[1], ignore_errors=True)
-
- def has_warnings(self):
- """Checks if warnings are present, returns bool."""
- return bool(len(self._warnings))
-
- def get_warnings(self):
- """Returns a list with all available warnings and resets all
- warnings.
- """
- ret_val = self._warnings[:]
- del self._warnings[:]
- return ret_val
-
- 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)
-
- def cfg_pget(self, option):
- """Get the configured value of the *option* (section.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
-
- def domain_add(self, domainname, transport=None):
- """Wrapper around Domain.set_transport() and Domain.save()"""
- dom = self._get_domain(domainname)
- if transport is None:
- dom.set_transport(Transport(self._dbh,
- 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.save()
- self._make_domain_dir(dom)
-
- def domain_transport(self, domainname, transport, force=None):
- """Wrapper around Domain.update_transport()"""
- if force is not None and force != 'force':
- raise DomainError(_(u"Invalid argument: '%s'") % force,
- INVALID_ARGUMENT)
- dom = self._get_domain(domainname)
- trsp = Transport(self._dbh, transport=transport)
- if force is None:
- dom.update_transport(trsp)
- else:
- dom.update_transport(trsp, force=True)
-
- def domain_delete(self, domainname, force=None):
- """Wrapper around Domain.delete()"""
- if force and force not in ('deluser', 'delalias', 'delall'):
- raise DomainError(_(u"Invalid argument: '%s'") % force,
- INVALID_ARGUMENT)
- dom = self._get_domain(domainname)
- gid = dom.gid
- domdir = dom.directory
- 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'):
- self._delete_domain_dir(domdir, gid)
-
- def domain_info(self, domainname, details=None):
- """Wrapper around Domain.get_info(), Domain.get_accounts(),
- Domain.get_aliase_names(), Domain.get_aliases() and
- Domain.get_relocated."""
- if details not in [None, 'accounts', 'aliasdomains', 'aliases', 'full',
- 'relocated']:
- raise VMMError(_(u'Invalid argument: “%s”') % details,
- INVALID_ARGUMENT)
- dom = self._get_domain(domainname)
- dominfo = dom.get_info()
- if dominfo['domainname'].startswith('xn--'):
- dominfo['domainname'] += ' (%s)' % \
- dominfo['domainname'].decode('idna')
- if details is None:
- return dominfo
- elif details == 'accounts':
- return (dominfo, dom.get_accounts())
- elif details == 'aliasdomains':
- return (dominfo, dom.get_aliase_names())
- elif details == 'aliases':
- return (dominfo, dom.get_aliases())
- elif details == 'relocated':
- return(dominfo, dom.get_relocated())
- else:
- return (dominfo, dom.get_aliase_names(), dom.get_accounts(),
- dom.get_aliases(), dom.get_relocated())
-
- def aliasdomain_add(self, aliasname, domainname):
- """Adds an alias domain to the domain.
-
- Arguments:
-
- `aliasname` : basestring
- The name of the alias domain
- `domainname` : basestring
- The name of the target domain
- """
- dom = self._get_domain(domainname)
- alias_dom = AliasDomain(self._dbh, aliasname)
- alias_dom.set_destination(dom)
- alias_dom.save()
-
- def aliasdomain_info(self, aliasname):
- """Returns a dict (keys: "alias" and "domain") with the names of
- the alias domain and its primary domain."""
- self._db_connect()
- alias_dom = AliasDomain(self._dbh, aliasname)
- return alias_dom.info()
-
- def aliasdomain_switch(self, aliasname, domainname):
- """Modifies the target domain of an existing alias domain.
-
- Arguments:
-
- `aliasname` : basestring
- The name of the alias domain
- `domainname` : basestring
- The name of the new target domain
- """
- dom = self._get_domain(domainname)
- alias_dom = AliasDomain(self._dbh, aliasname)
- alias_dom.set_destination(dom)
- alias_dom.switch()
-
- def aliasdomain_delete(self, aliasname):
- """Deletes the given alias domain.
-
- Argument:
-
- `aliasname` : basestring
- The name of the alias domain
- """
- self._db_connect()
- alias_dom = AliasDomain(self._dbh, aliasname)
- alias_dom.delete()
-
- def domain_list(self, pattern=None):
- """Wrapper around function search() from module Domain."""
- from VirtualMailManager.Domain import search
- like = False
- if pattern and (pattern.startswith('%') or pattern.endswith('%')):
- like = True
- if not re.match(RE_DOMAIN_SEARCH, pattern.strip('%')):
- raise VMMError(_(u"The pattern '%s' contains invalid "
- u"characters.") % pattern, DOMAIN_INVALID)
- self._db_connect()
- return search(self._dbh, pattern=pattern, like=like)
-
- def user_add(self, emailaddress, password):
- """Wrapper around Account.set_password() and Account.save()."""
- acc = self._get_account(emailaddress)
- acc.set_password(password)
- acc.save()
- oldpwd = os.getcwd()
- self._make_home(acc)
- mailbox = new_mailbox(acc)
- mailbox.create()
- folders = self._cfg.dget('mailbox.folders').split(':')
- if any(folders):
- bad = mailbox.add_boxes(folders,
- self._cfg.dget('mailbox.subscribe'))
- if bad:
- self._warnings.append(_(u"Skipped mailbox folders:") +
- '\n\t- ' + '\n\t- '.join(bad))
- os.chdir(oldpwd)
-
- def alias_add(self, aliasaddress, *targetaddresses):
- """Creates a new `Alias` entry for the given *aliasaddress* with
- the given *targetaddresses*."""
- alias = self._get_alias(aliasaddress)
- destinations = [EmailAddress(address) for address in targetaddresses]
- warnings = []
- destinations = alias.add_destinations(destinations, warnings)
- if warnings:
- self._warnings.append(_('Ignored destination addresses:'))
- self._warnings.extend((' * %s' % w for w in warnings))
- for destination in destinations:
- if get_gid(self._dbh, destination.domainname) and \
- not self._chk_other_address_types(destination, TYPE_RELOCATED):
- self._warnings.append(_(u"The destination account/alias '%s' "
- u"doesn't exist.") % destination)
-
- def user_delete(self, emailaddress, force=None):
- """Wrapper around Account.delete(...)"""
- if force not in (None, 'delalias'):
- raise VMMError(_(u"Invalid argument: '%s'") % force,
- INVALID_ARGUMENT)
- acc = self._get_account(emailaddress)
- if not acc:
- raise VMMError(_(u"The account '%s' doesn't exist.") %
- acc.address, NO_SUCH_ACCOUNT)
- 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'):
- try:
- self._delete_home(dom_dir, uid, gid)
- except VMMError, err:
- if err.code in (FOUND_DOTS_IN_PATH, MAILDIR_PERM_MISMATCH,
- 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_dir, 'reason': err.msg}
- self._warnings.append(warning)
- else:
- raise
-
- def alias_info(self, aliasaddress):
- """Returns an iterator object for all destinations (`EmailAddress`
- instances) for the `Alias` with the given *aliasaddress*."""
- alias = self._get_alias(aliasaddress)
- if alias:
- return alias.get_destinations()
- if not self._is_other_address(alias.address, TYPE_ALIAS):
- raise VMMError(_(u"The alias '%s' doesn't exist.") %
- alias.address, NO_SUCH_ALIAS)
-
- def alias_delete(self, aliasaddress, targetaddress=None):
- """Deletes the `Alias` *aliasaddress* with all its destinations from
- the database. If *targetaddress* is not ``None``, only this
- destination will be removed from the alias."""
- alias = self._get_alias(aliasaddress)
- if targetaddress is None:
- alias.delete()
- else:
- alias.del_destination(EmailAddress(targetaddress))
-
- def user_info(self, emailaddress, details=None):
- """Wrapper around Account.get_info(...)"""
- if details not in (None, 'du', 'aliases', 'full'):
- raise VMMError(_(u"Invalid argument: '%s'") % details,
- INVALID_ARGUMENT)
- acc = self._get_account(emailaddress)
- 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'):
- path = os.path.join(acc.home, acc.mail_location.directory)
- info['disk usage'] = self._get_disk_usage(path)
- if details in (None, 'du'):
- return info
- if details in ('aliases', 'full'):
- return (info, acc.get_aliases())
- return info
-
- def user_by_uid(self, uid):
- """Search for an Account by its *uid*.
- Returns a dict (address, uid and gid) if a user could be found."""
- from VirtualMailManager.Account import get_account_by_uid
- self._db_connect()
- return get_account_by_uid(uid, self._dbh)
-
- def user_password(self, emailaddress, password):
- """Wrapper for Account.modify('password' ...)."""
- if not isinstance(password, basestring) or not password:
- raise VMMError(_(u"Could not accept password: '%s'") % password,
- INVALID_ARGUMENT)
- acc = self._get_account(emailaddress)
- if not acc:
- raise VMMError(_(u"The account '%s' doesn't exist.") %
- acc.address, NO_SUCH_ACCOUNT)
- acc.modify('password', password)
-
- def user_name(self, emailaddress, name):
- """Wrapper for Account.modify('name', ...)."""
- if not isinstance(name, basestring) or not name:
- raise VMMError(_(u"Could not accept name: '%s'") % name,
- INVALID_ARGUMENT)
- acc = self._get_account(emailaddress)
- if not acc:
- raise VMMError(_(u"The account '%s' doesn't exist.") %
- acc.address, NO_SUCH_ACCOUNT)
- acc.modify('name', name)
-
- def user_transport(self, emailaddress, transport):
- """Wrapper for Account.modify('transport', ...)."""
- if not isinstance(transport, basestring) or not transport:
- raise VMMError(_(u"Could not accept transport: '%s'") % transport,
- INVALID_ARGUMENT)
- acc = self._get_account(emailaddress)
- if not acc:
- raise VMMError(_(u"The account '%s' doesn't exist.") %
- acc.address, NO_SUCH_ACCOUNT)
- acc.modify('transport', transport)
-
- def user_disable(self, emailaddress, service=None):
- """Wrapper for Account.disable(service)"""
- if service not in (None, 'all', 'imap', 'pop3', 'smtp', 'sieve'):
- raise VMMError(_(u"Could not accept service: '%s'") % service,
- INVALID_ARGUMENT)
- acc = self._get_account(emailaddress)
- if not acc:
- raise VMMError(_(u"The account '%s' doesn't exist.") %
- acc.address, NO_SUCH_ACCOUNT)
- acc.disable(service)
-
- def user_enable(self, emailaddress, service=None):
- """Wrapper for Account.enable(service)"""
- if service not in (None, 'all', 'imap', 'pop3', 'smtp', 'sieve'):
- raise VMMError(_(u"Could not accept service: '%s'") % service,
- INVALID_ARGUMENT)
- acc = self._get_account(emailaddress)
- if not acc:
- raise VMMError(_(u"The account '%s' doesn't exist.") %
- acc.address, NO_SUCH_ACCOUNT)
- acc.enable(service)
-
- def relocated_add(self, emailaddress, targetaddress):
- """Creates a new `Relocated` entry in the database. If there is
- already a relocated user with the given *emailaddress*, only the
- *targetaddress* for the relocated user will be updated."""
- relocated = self._get_relocated(emailaddress)
- relocated.set_destination(EmailAddress(targetaddress))
-
- def relocated_info(self, emailaddress):
- """Returns the target address of the relocated user with the given
- *emailaddress*."""
- relocated = self._get_relocated(emailaddress)
- if relocated:
- return relocated.get_info()
- if not self._is_other_address(relocated.address, TYPE_RELOCATED):
- raise VMMError(_(u"The relocated user '%s' doesn't exist.") %
- relocated.address, NO_SUCH_RELOCATED)
-
- def relocated_delete(self, emailaddress):
- """Deletes the relocated user with the given *emailaddress* from
- the database."""
- relocated = self._get_relocated(emailaddress)
- relocated.delete()
-
-del _