VMM/mailbox: Added to the repository.
VMM/Handler: Integrated mailbox module. Code cleanups.
# -*- coding: UTF-8 -*-# Copyright (c) 2007 - 2010, Pascal Volk# See COPYING for distribution information.""" VirtualMailManager.Domain Virtual Mail Manager's Domain class to manage e-mail domains."""importosimportrefromrandomimportchoicefromVirtualMailManager.constants.ERRORimport \ACCOUNT_AND_ALIAS_PRESENT,ACCOUNT_PRESENT,ALIAS_PRESENT, \DOMAIN_ALIAS_EXISTS,DOMAIN_EXISTS,DOMAIN_INVALID,DOMAIN_TOO_LONG, \NO_SUCH_DOMAINfromVirtualMailManager.errorsimportDomainErrorasDomErrfromVirtualMailManager.TransportimportTransportMAILDIR_CHARS='0123456789abcdefghijklmnopqrstuvwxyz'RE_DOMAIN=re.compile(r"^(?:[a-z0-9-]{1,63}\.){1,}[a-z]{2,6}$")_=lambdamsg:msgclassDomain(object):"""Class to manage e-mail domains."""__slots__=('_directory','_gid','_name','_transport','_dbh','_new')def__init__(self,dbh,domainname):"""Creates a new Domain instance. Loads all relevant data from the database, if the domain could be found. To create a new domain call the methods set_directory() and set_transport() before save(). A DomainError will be thrown when the *domainname* is the name of an alias domain. Arguments: `dbh` : pyPgSQL.PgSQL.Connection a database connection for the database access `domainname` : basestring The name of the domain """self._name=check_domainname(domainname)self._dbh=dbhself._gid=0self._transport=Noneself._directory=Noneself._new=Trueself._load()def_load(self):"""Load information from the database and checks if the domain name is the primary one. Raises a DomainError if Domain._name isn't the primary name of the domain. """dbc=self._dbh.cursor()dbc.execute('SELECT dd.gid, tid, domaindir, is_primary FROM ''domain_data dd, domain_name dn WHERE domainname = %s AND ''dn.gid = dd.gid',self._name)result=dbc.fetchone()dbc.close()ifresult:ifnotresult[3]:raiseDomErr(_(u"The domain '%s' is an alias domain.")%self._name,DOMAIN_ALIAS_EXISTS)self._gid,self._directory=result[0],result[2]self._transport=Transport(self._dbh,tid=result[1])self._new=Falsedef_set_gid(self):"""Sets the ID of the domain - if not set yet."""assertself._gid==0dbc=self._dbh.cursor()dbc.execute("SELECT nextval('domain_gid')")self._gid=dbc.fetchone()[0]dbc.close()def_has(self,what):"""Checks if aliases or accounts are assigned to the domain. If there are assigned accounts or aliases True will be returned, otherwise False will be returned. Argument: `what` : basestring "alias" or "users" """assertwhatin('alias','users')dbc=self._dbh.cursor()ifwhat=='users':dbc.execute("SELECT count(gid) FROM users WHERE gid=%s",self._gid)else:dbc.execute("SELECT count(gid) FROM alias WHERE gid=%s",self._gid)count=dbc.fetchone()dbc.close()returncount[0]>0def_chk_delete(self,deluser,delalias):"""Checks dependencies for deletion. Arguments: deluser -- ignore available accounts (bool) delalias -- ignore available aliases (bool) """ifnotdeluser:hasuser=self._has('users')else:hasuser=Falseifnotdelalias:hasalias=self._has('alias')else:hasalias=Falseifhasuserandhasalias:raiseDomErr(_(u'There are accounts and aliases.'),ACCOUNT_AND_ALIAS_PRESENT)elifhasuser:raiseDomErr(_(u'There are accounts.'),ACCOUNT_PRESENT)elifhasalias:raiseDomErr(_(u'There are aliases.'),ALIAS_PRESENT)def_chk_state(self):"""Throws a DomainError if the Domain is new - not saved in the database."""ifself._new:raiseDomErr(_(u"The domain '%s' doesn't exist.")%self._name,NO_SUCH_DOMAIN)@propertydefgid(self):"""The GID of the Domain."""returnself._gid@propertydefname(self):"""The Domain's name."""returnself._name@propertydefdirectory(self):"""The Domain's directory."""returnself._directorydefset_directory(self,basedir):"""Set the path value of the Domain's directory, inside *basedir*. Argument: `basedir` : basestring The base directory of all domains """assertself._newandself._directoryisNoneself._set_gid()self._directory=os.path.join(basedir,choice(MAILDIR_CHARS),str(self._gid))@propertydeftransport(self):"""The Domain's transport."""returnself._transportdefset_transport(self,transport):"""Set the transport for the new Domain. Argument: `transport` : VirtualMailManager.Transport The transport of the new Domain """assertself._newandisinstance(transport,Transport)self._transport=transportdefsave(self):"""Stores the new domain in the database."""ifnotself._new:raiseDomErr(_(u"The domain '%s' already exists.")%self._name,DOMAIN_EXISTS)assertself._directoryisnotNoneandself._transportisnotNonedbc=self._dbh.cursor()dbc.execute("INSERT INTO domain_data VALUES (%s, %s, %s)",self._gid,self._transport.tid,self._directory)dbc.execute("INSERT INTO domain_name VALUES (%s, %s, %s)",self._name,self._gid,True)self._dbh.commit()dbc.close()self._new=Falsedefdelete(self,deluser=False,delalias=False):"""Deletes the domain. Arguments: `deluser` : bool force deletion of all available accounts, default `False` `delalias` : bool force deletion of all available aliases, default `False` """self._chk_state()self._chk_delete(deluser,delalias)dbc=self._dbh.cursor()fortblin('alias','users','relocated','domain_name','domain_data'):dbc.execute("DELETE FROM %s WHERE gid = %d"%(tbl,self._gid))self._dbh.commit()dbc.close()self._gid=0self._directory=self._transport=Noneself._new=Truedefupdate_transport(self,transport,force=False):"""Sets a new transport for the Domain. If *force* is `True` the new *transport* will be assigned to all existing accounts. Otherwise the *transport* will be only used for accounts created from now on. Arguments: `transport` : VirtualMailManager.Transport the new transport `force` : bool enforce new transport setting for all accounts, default `False` """self._chk_state()assertisinstance(transport,Transport)iftransport==self._transport:returndbc=self._dbh.cursor()dbc.execute("UPDATE domain_data SET tid = %s WHERE gid = %s",transport.tid,self._gid)ifdbc.rowcount>0:self._dbh.commit()ifforce:dbc.execute("UPDATE users SET tid = %s WHERE gid = %s",transport.tid,self._gid)ifdbc.rowcount>0:self._dbh.commit()dbc.close()self._transport=transportdefget_info(self):"""Returns a dictionary with information about the domain."""self._chk_state()dbc=self._dbh.cursor()dbc.execute('SELECT gid, domainname, transport, domaindir, ''aliasdomains accounts, aliases, relocated FROM ''vmm_domain_info WHERE gid = %s',self._gid)info=dbc.fetchone()dbc.close()keys=('gid','domainname','transport','domaindir','aliasdomains','accounts','aliases','relocated')returndict(zip(keys,info))defget_accounts(self):"""Returns a list with all accounts of the domain."""self._chk_state()dbc=self._dbh.cursor()dbc.execute('SELECT local_part from users where gid = %s ORDER BY ''local_part',self._gid)users=dbc.fetchall()dbc.close()accounts=[]ifusers:addr=u'@'.join_dom=self._nameaccounts=[addr((account[0],_dom))foraccountinusers]returnaccountsdefget_aliases(self):"""Returns a list with all aliases e-mail addresses of the domain."""self._chk_state()dbc=self._dbh.cursor()dbc.execute('SELECT DISTINCT address FROM alias WHERE gid = %s ORDER ''BY address',self._gid)addresses=dbc.fetchall()dbc.close()aliases=[]ifaddresses:addr=u'@'.join_dom=self._namealiases=[addr((alias[0],_dom))foraliasinaddresses]returnaliasesdefget_relocated(self):"""Returns a list with all addresses of relocated users."""self._chk_state()dbc=self._dbh.cursor()dbc.execute('SELECT address FROM relocated WHERE gid = %s ORDER BY ''address',self._gid)addresses=dbc.fetchall()dbc.close()relocated=[]ifaddresses:addr=u'@'.join_dom=self._namerelocated=[addr((address[0],_dom))foraddressinaddresses]returnrelocateddefget_aliase_names(self):"""Returns a list with all alias domain names of the domain."""self._chk_state()dbc=self._dbh.cursor()dbc.execute('SELECT domainname FROM domain_name WHERE gid = %s AND ''NOT is_primary ORDER BY domainname',self._gid)anames=dbc.fetchall()dbc.close()aliasdomains=[]ifanames:aliasdomains=[aname[0]foranameinanames]returnaliasdomainsdefcheck_domainname(domainname):"""Returns the validated domain name `domainname`. Throws an `DomainError`, if the domain name is too long or doesn't look like a valid domain name (label.label.label). """ifnotRE_DOMAIN.match(domainname):domainname=domainname.encode('idna')iflen(domainname)>255:raiseDomErr(_(u'The domain name is too long'),DOMAIN_TOO_LONG)ifnotRE_DOMAIN.match(domainname):raiseDomErr(_(u"The domain name '%s' is invalid")%domainname,DOMAIN_INVALID)returndomainnamedefget_gid(dbh,domainname):"""Returns the group id of the domain *domainname*. If the domain couldn't be found in the database 0 will be returned. """domainname=check_domainname(domainname)dbc=dbh.cursor()dbc.execute('SELECT gid FROM domain_name WHERE domainname=%s',domainname)gid=dbc.fetchone()dbc.close()ifgid:returngid[0]return0defsearch(dbh,pattern=None,like=False):"""'Search' for domains by *pattern* in the database. *pattern* may be a domain name or a partial domain name - starting and/or ending with a '%' sign. When the *pattern* starts or ends with a '%' sign *like* has to be `True` to perform a wildcard search. To retrieve all available domains use the arguments' default values. This function returns a tuple with a list and a dict: (order, domains). The order list contains the domains' gid, alphabetical sorted by the primary domain name. The domains dict's keys are the gids of the domains. The value of item is a list. The first list element contains the primary domain name or `None`. The elements [1:] contains the names of alias domains. Arguments: `pattern` : basestring a (partial) domain name (starting and/or ending with a "%" sign) `like` : bool should be `True` when *pattern* starts/ends with a "%" sign """ifpatternandnotlike:pattern=check_domainname(pattern)sql='SELECT gid, domainname, is_primary FROM domain_name'ifpattern:iflike:sql+=" WHERE domainname LIKE '%s'"%patternelse:sql+=" WHERE domainname = '%s'"%patternsql+=' ORDER BY is_primary DESC, domainname'dbc=dbh.cursor()dbc.execute(sql)result=dbc.fetchall()dbc.close()gids=[domain[0]fordomaininresultifdomain[2]]domains={}forgid,domain,is_primaryinresult:ifis_primary:ifnotgidindomains:domains[gid]=[domain]else:domains[gid].insert(0,domain)else:ifgidingids:ifgidindomains:domains[gid].append(domain)else:domains[gid]=[domain]else:gids.append(gid)domains[gid]=[None,domain]returngids,domainsdel_