From: Frank Brehm Date: Tue, 26 Mar 2024 17:27:19 +0000 (+0100) Subject: Extending class PostfixLogchainInfo X-Git-Url: https://git.uhu-banane.net/?a=commitdiff_plain;h=60d4d662fcc2fb7592f1d823ca184173d873524f;p=pixelpark%2Fpp-admin-tools.git Extending class PostfixLogchainInfo --- diff --git a/lib/pp_admintools/postfix_chain.py b/lib/pp_admintools/postfix_chain.py index 1cb4b34..aab306e 100644 --- a/lib/pp_admintools/postfix_chain.py +++ b/lib/pp_admintools/postfix_chain.py @@ -17,9 +17,11 @@ import re from collections import OrderedDict # Third party modules +from fb_tools.common import is_sequence from fb_tools.common import pp from fb_tools.errors import InvalidMailAddressError from fb_tools.mailaddress import MailAddress +from fb_tools.mailaddress import MailAddressList from fb_tools.obj import FbGenericBaseObject # Own modules @@ -28,7 +30,7 @@ from .xlate import XLATOR _ = XLATOR.gettext ngettext = XLATOR.ngettext -__version__ = '0.5.2' +__version__ = '0.6.0' LOG = logging.getLogger(__name__) @@ -114,16 +116,20 @@ class PostfixLogchainInfo(FbGenericBaseObject): attributes = ( 'client_host', 'client_addr', 'start', 'end', 'message_id', 'postfix_id', 'ehlo', 'starttls', 'sent_quit', 'auth', 'commands', 'rcpt', 'data', 'mail', 'from_address', - 'to_address', 'smtpd_pid', 'mailhost', + 'origin_to', 'smtpd_pid', 'mailhost', 'tls_version', 'tls_cipher', 'size', 'nr_rcpt', ) # ------------------------------------------------------------------------- - def __init__(self, **kwargs): + def __init__(self, to_addresses=[], **kwargs): """Initialize this object.""" for attr in self.attributes: priv_name = '_' + attr setattr(self, priv_name, None) + self.to_addresses = MailAddressList() + if to_addresses: + self.set_to_addresses(to_addresses) + for attr in kwargs.keys(): if attr not in self.attributes: msg = _('Unknown parameter {p!r} on calling {c}.__init__().').format( @@ -131,6 +137,28 @@ class PostfixLogchainInfo(FbGenericBaseObject): raise AttributeError(msg) setattr(self, attr, kwargs[attr]) + # ------------------------------------------------------------------------- + def set_to_addresses(self, addresses): + """Set the array with all TO addresses of the mail after address resolution.""" + self.to_addresses = [] + if addresses is None: + return + + if isinstance(addresses, MailAddressList): + self.to_addresses = copy.copy(addresses) + return + + if not is_sequence(addresses): + addresses = [addresses] + + for addr in addresses: + self.to_addresses.append(addr) + + # ------------------------------------------------------------------------- + def add_to_address(self, address): + """Append the given mail address to the list of TO addresses.""" + self.to_addresses.append(address) + # ------------------------------------------------------------------------- @classmethod def date_fromisoformat(cls, datestr): @@ -597,33 +625,121 @@ class PostfixLogchainInfo(FbGenericBaseObject): # ----------------------------------------------------------- @property - def to_address(self): - """Return the RCPT TO address in the SMTP dialogue envelope.""" - return self._to_address + def origin_to(self): + """Return the original RCPT TO address in the SMTP dialogue envelope.""" + return self._origin_to - @to_address.setter - def to_address(self, value): + @origin_to.setter + def origin_to(self, value): if value is None: - self._to_address = None + self._origin_to = None return if isinstance(value, MailAddress): - self._to_address = copy.copy(value) + self._origin_to = copy.copy(value) return val = str(value).strip() if val == '': - self._to_address = None + self._origin_to = None return try: - self._to_address = MailAddress(val) + self._origin_to = MailAddress(val) except InvalidMailAddressError as e: msg = _('Could not interprete to address {a!r}: {e}').format(a=val, e=e) if self.warn_on_parse_error: LOG.warn(msg) else: LOG.debug(msg) - self._to_address = val + self._origin_to = val + + # ------------------------------------------------------- + @property + def tls_version(self): + """Return the used SSL/TLS version in communication.""" + return self._tls_version + + @tls_version.setter + def tls_version(self, value): + if value is None: + self._tls_version = None + return + + val = str(value).strip() + if val == '': + self._tls_version = None + return + self._tls_version = val + + # ------------------------------------------------------- + @property + def tls_cipher(self): + """Return the used SSL/TLS version in communication.""" + return self._tls_cipher + + @tls_cipher.setter + def tls_cipher(self, value): + if value is None: + self._tls_cipher = None + return + + val = str(value).strip() + if val == '': + self._tls_cipher = None + return + self._tls_cipher = val + + # ----------------------------------------------------------- + @property + def size(self): + """Return the size of the mail in bytes.""" + return self._size + + @size.setter + def size(self, value): + if value is None: + self._size = None + return + + if isinstance(value, int): + self._size = value + return + try: + val = int(value) + self._size = val + except ValueErros as e: + msg = _('Could not interprete mail size {a!r}: {e}').format(a=val, e=e) + if self.warn_on_parse_error: + LOG.warn(msg) + else: + LOG.debug(msg) + self._size = None + + # ----------------------------------------------------------- + @property + def nr_rcpt(self): + """Return th enumber of recipients in the mail.""" + return self._nr_rcpt + + @nr_rcpt.setter + def nr_rcpt(self, value): + if value is None: + self._nr_rcpt = None + return + + if isinstance(value, int): + self._nr_rcpt = value + return + try: + val = int(value) + self._nr_rcpt = val + except ValueErros as e: + msg = _('Could not interprete number of recipients {a!r}: {e}').format(a=val, e=e) + if self.warn_on_parse_error: + LOG.warn(msg) + else: + LOG.debug(msg) + self._nr_rcpt = None # ------------------------------------------------------------------------- def __str__(self): @@ -690,6 +806,13 @@ class PostfixLogchainInfo(FbGenericBaseObject): # Catch all res[attrib] = value + if exportable: + res['to_addresses'] = None + if self.to_addresses: + res['to_addresses'] = self.to_addresses.as_list(as_str=True) + else: + res['to_addresses'] = copy.copy(self.to_addresses) + # Non init properties if not exportable: res['duration'] = self.duration