Mailu/core/admin/mailu/schemas.py

""" Mailu marshmallow fields and schema
"""

from copy import deepcopy
from collections import Counter
from datetime import timezone

import inspect
import json
import logging
import yaml

import sqlalchemy

from marshmallow import pre_load, post_load, post_dump, fields, Schema
from marshmallow.utils import ensure_text_type
from marshmallow.exceptions import ValidationError
from marshmallow_sqlalchemy import SQLAlchemyAutoSchemaOpts
from marshmallow_sqlalchemy.fields import RelatedList

from flask_marshmallow import Marshmallow

from cryptography.hazmat.primitives import serialization

from pygments import highlight
from pygments.token import Token
from pygments.lexers import get_lexer_by_name
from pygments.lexers.data import YamlLexer
from pygments.formatters import get_formatter_by_name

from mailu import models, dkim


ma = Marshmallow()


### import logging and schema colorization ###

_model2schema = {}

def get_schema(cls=None):
    """ return schema class for model """
    if cls is None:
        return _model2schema.values()
    return _model2schema.get(cls)

def mapped(cls):
    """ register schema in model2schema map """
    _model2schema[cls.Meta.model] = cls
    return cls

class Logger:
    """ helps with counting and colorizing
        imported and exported data
    """

    class MyYamlLexer(YamlLexer):
        """ colorize yaml constants and integers """
        def get_tokens(self, text, unfiltered=False):
            for typ, value in super().get_tokens(text, unfiltered):
                if typ is Token.Literal.Scalar.Plain:
                    if value in {'true', 'false', 'null'}:
                        typ = Token.Keyword.Constant
                    elif value == HIDDEN:
                        typ = Token.Error
                    else:
                        try:
                            int(value, 10)
                        except ValueError:
                            try:
                                float(value)
                            except ValueError:
                                pass
                            else:
                                typ = Token.Literal.Number.Float
                        else:
                            typ = Token.Literal.Number.Integer
                yield typ, value

    def __init__(self, want_color=None, can_color=False, debug=False, secrets=False):

        self.lexer = 'yaml'
        self.formatter = 'terminal'
        self.strip = False
        self.verbose = 0
        self.quiet = False
        self.secrets = secrets
        self.debug = debug
        self.print = print

        self.color = want_color or can_color

        self._counter = Counter()
        self._schemas = {}

        # log contexts
        self._diff_context = {
            'full': True,
            'secrets': secrets,
        }
        log_context = {
            'secrets': secrets,
        }

        # register listeners
        for schema in get_schema():
            model = schema.Meta.model
            self._schemas[model] = schema(context=log_context)
            sqlalchemy.event.listen(model, 'after_insert', self._listen_insert)
            sqlalchemy.event.listen(model, 'after_update', self._listen_update)
            sqlalchemy.event.listen(model, 'after_delete', self._listen_delete)

        # special listener for dkim_key changes
        # TODO: _listen_dkim can be removed when dkim keys are stored in database
        self._dedupe_dkim = set()
        sqlalchemy.event.listen(models.db.session, 'after_flush', self._listen_dkim)

        # register debug logger for sqlalchemy
        if self.debug:
            logging.basicConfig()
            logging.getLogger('sqlalchemy.engine').setLevel(logging.INFO)

    def _log(self, action, target, message=None):
        if message is None:
            try:
                message = self._schemas[target.__class__].dump(target)
            except KeyError:
                message = target
        if not isinstance(message, str):
            message = repr(message)
        self.print(f'{action} {target.__table__}: {self.colorize(message)}')

    def _listen_insert(self, mapper, connection, target): # pylint: disable=unused-argument
        """ callback method to track import """
        self._counter.update([('Created', target.__table__.name)])
        if self.verbose:
            self._log('Created', target)

    def _listen_update(self, mapper, connection, target): # pylint: disable=unused-argument
        """ callback method to track import """

        changes = {}
        inspection = sqlalchemy.inspect(target)
        for attr in sqlalchemy.orm.class_mapper(target.__class__).column_attrs:
            history = getattr(inspection.attrs, attr.key).history
            if history.has_changes() and history.deleted:
                before = history.deleted[-1]
                after = getattr(target, attr.key)
                # we don't have ordered lists
                if isinstance(before, list):
                    before = set(before)
                if isinstance(after, list):
                    after = set(after)
                # TODO: this can be removed when comment is not nullable in model
                if attr.key == 'comment' and not before and not after:
                    pass
                # only remember changed keys
                elif before != after:
                    if self.verbose:
                        changes[str(attr.key)] = (before, after)
                    else:
                        break

        if self.verbose:
            # use schema to log changed attributes
            schema = get_schema(target.__class__)
            only = set(changes.keys()) & set(schema().fields.keys())
            if only:
                for key, value in schema(
                    only=only,
                    context=self._diff_context
                ).dump(target).items():
                    before, after = changes[key]
                    if value == HIDDEN:
                        before = HIDDEN if before else before
                        after = HIDDEN if after else after
                    else:
                        # also hide this
                        after = value
                    self._log('Modified', target, f'{str(target)!r} {key}: {before!r} -> {after!r}')

        if changes:
            self._counter.update([('Modified', target.__table__.name)])

    def _listen_delete(self, mapper, connection, target): # pylint: disable=unused-argument
        """ callback method to track import """
        self._counter.update([('Deleted', target.__table__.name)])
        if self.verbose:
            self._log('Deleted', target)

    # TODO: _listen_dkim can be removed when dkim keys are stored in database
    def _listen_dkim(self, session, flush_context): # pylint: disable=unused-argument
        """ callback method to track import """
        for target in session.identity_map.values():
            # look at Domains originally loaded from db
            if not isinstance(target, models.Domain) or not target._sa_instance_state.load_path:
                continue
            before = target._dkim_key_on_disk
            after = target._dkim_key
            # "de-dupe" messages; this event is fired at every flush
            if before == after or (target, before, after) in self._dedupe_dkim:
                continue
            self._dedupe_dkim.add((target, before, after))
            self._counter.update([('Modified', target.__table__.name)])
            if self.verbose:
                if self.secrets:
                    before = before.decode('ascii', 'ignore')
                    after = after.decode('ascii', 'ignore')
                else:
                    before = HIDDEN if before else ''
                    after = HIDDEN if after else ''
                self._log('Modified', target, f'{str(target)!r} dkim_key: {before!r} -> {after!r}')

    def track_serialize(self, obj, item, backref=None):
        """ callback method to track import """
        # called for backref modification?
        if backref is not None:
            self._log(
                'Modified', item, '{target!r} {key}: {before!r} -> {after!r}'.format_map(backref))
            return
        # show input data?
        if self.verbose < 2:
            return
        # hide secrets in data
        if not self.secrets:
            item = self._schemas[obj.opts.model].hide(item)
            if 'hash_password' in item:
                item['password'] = HIDDEN
            if 'fetches' in item:
                for fetch in item['fetches']:
                    fetch['password'] = HIDDEN
        self._log('Handling', obj.opts.model, item)

    def changes(self, *messages, **kwargs):
        """ show changes gathered in counter """
        if self.quiet:
            return
        if self._counter:
            changes = []
            last = None
            for (action, what), count in sorted(self._counter.items()):
                if action != last:
                    if last:
                        changes.append('/')
                    changes.append(f'{action}:')
                    last = action
                changes.append(f'{what}({count})')
        else:
            changes = ['No changes.']
        self.print(*messages, *changes, **kwargs)

    def _format_errors(self, store, path=None):

        res = []
        if path is None:
            path = []
        for key in sorted(store):
            location = path + [str(key)]
            value = store[key]
            if isinstance(value, dict):
                res.extend(self._format_errors(value, location))
            else:
                for message in value:
                    res.append((".".join(location), message))

        if path:
            return res

        maxlen = max(len(loc) for loc, msg in res)
        res = [f'     - {loc.ljust(maxlen)} : {msg}' for loc, msg in res]
        errors = f'{len(res)} error{["s",""][len(res)==1]}'
        res.insert(0, f'[ValidationError] {errors} occurred during input validation')

        return '\n'.join(res)

    def _is_validation_error(self, exc):
        """ walk traceback to extract invalid field from marshmallow """
        path = []
        trace = exc.__traceback__
        while trace:
            if trace.tb_frame.f_code.co_name == '_serialize':
                if 'attr' in trace.tb_frame.f_locals:
                    path.append(trace.tb_frame.f_locals['attr'])
            elif trace.tb_frame.f_code.co_name == '_init_fields':
                spec = ', '.join(
                    '.'.join(path + [key])
                    for key in trace.tb_frame.f_locals['invalid_fields'])
                return f'Invalid filter: {spec}'
            trace = trace.tb_next
        return None

    def format_exception(self, exc):
        """ format ValidationErrors and other exceptions when not debugging """
        if isinstance(exc, ValidationError):
            return self._format_errors(exc.messages)
        if isinstance(exc, ValueError):
            if msg := self._is_validation_error(exc):
                return msg
        if self.debug:
            return None
        msg = ' '.join(str(exc).split())
        return f'[{exc.__class__.__name__}] {msg}'

    colorscheme = {
        Token:                  ('',        ''),
        Token.Name.Tag:         ('cyan',    'cyan'),
        Token.Literal.Scalar:   ('green',   'green'),
        Token.Literal.String:   ('green',   'green'),
        Token.Name.Constant:    ('green',   'green'), # multiline strings
        Token.Keyword.Constant: ('magenta', 'magenta'),
        Token.Literal.Number:   ('magenta', 'magenta'),
        Token.Error:            ('red',     'red'),
        Token.Name:             ('red',     'red'),
        Token.Operator:         ('red',     'red'),
    }

    def colorize(self, data, lexer=None, formatter=None, color=None, strip=None):
        """ add ANSI color to data """

        if color is False or not self.color:
            return data

        lexer = lexer or self.lexer
        lexer = Logger.MyYamlLexer() if lexer == 'yaml' else get_lexer_by_name(lexer)
        formatter = get_formatter_by_name(formatter or self.formatter, colorscheme=self.colorscheme)
        if strip is None:
            strip = self.strip

        res = highlight(data, lexer, formatter)
        if strip:
            return res.rstrip('\n')
        return res


### marshmallow render modules ###

# hidden attributes
class _Hidden:
    def __bool__(self):
        return False
    def __copy__(self):
        return self
    def __deepcopy__(self, _):
        return self
    def __eq__(self, other):
        return str(other) == '<hidden>'
    def __repr__(self):
        return '<hidden>'
    __str__ = __repr__

yaml.add_representer(
    _Hidden,
    lambda dumper, data: dumper.represent_data(str(data))
)

HIDDEN = _Hidden()

# multiline attributes
class _Multiline(str):
    pass

yaml.add_representer(
    _Multiline,
    lambda dumper, data: dumper.represent_scalar(u'tag:yaml.org,2002:str', data, style='|')

)

# yaml render module
class RenderYAML:
    """ Marshmallow YAML Render Module
    """

    class SpacedDumper(yaml.Dumper):
        """ YAML Dumper to add a newline between main sections
            and double the indent used
        """

        def write_line_break(self, data=None):
            super().write_line_break(data)
            if len(self.indents) == 1:
                super().write_line_break()

        def increase_indent(self, flow=False, indentless=False):
            return super().increase_indent(flow, False)

    @staticmethod
    def _augment(kwargs, defaults):
        """ add defaults to kwargs if missing
        """
        for key, value in defaults.items():
            if key not in kwargs:
                kwargs[key] = value

    _load_defaults = {}
    @classmethod
    def loads(cls, *args, **kwargs):
        """ load yaml data from string
        """
        cls._augment(kwargs, cls._load_defaults)
        return yaml.safe_load(*args, **kwargs)

    _dump_defaults = {
        'Dumper': SpacedDumper,
        'default_flow_style': False,
        'allow_unicode': True,
        'sort_keys': False,
    }
    @classmethod
    def dumps(cls, *args, **kwargs):
        """ dump data to yaml string
        """
        cls._augment(kwargs, cls._dump_defaults)
        return yaml.dump(*args, **kwargs)

# json encoder
class JSONEncoder(json.JSONEncoder):
    """ JSONEncoder supporting serialization of HIDDEN """
    def default(self, o):
        """ serialize HIDDEN """
        if isinstance(o, _Hidden):
            return str(o)
        return json.JSONEncoder.default(self, o)

# json render module
class RenderJSON:
    """ Marshmallow JSON Render Module
    """

    @staticmethod
    def _augment(kwargs, defaults):
        """ add defaults to kwargs if missing
        """
        for key, value in defaults.items():
            if key not in kwargs:
                kwargs[key] = value

    _load_defaults = {}
    @classmethod
    def loads(cls, *args, **kwargs):
        """ load json data from string
        """
        cls._augment(kwargs, cls._load_defaults)
        return json.loads(*args, **kwargs)

    _dump_defaults = {
        'separators': (',',':'),
        'cls': JSONEncoder,
    }
    @classmethod
    def dumps(cls, *args, **kwargs):
        """ dump data to json string
        """
        cls._augment(kwargs, cls._dump_defaults)
        return json.dumps(*args, **kwargs)


### marshmallow: custom fields ###

def _rfc3339(datetime):
    """ dump datetime according to rfc3339 """
    if datetime.tzinfo is None:
        datetime = datetime.astimezone(timezone.utc)
    res = datetime.isoformat()
    if res.endswith('+00:00'):
        return f'{res[:-6]}Z'
    return res

fields.DateTime.SERIALIZATION_FUNCS['rfc3339'] = _rfc3339
fields.DateTime.DESERIALIZATION_FUNCS['rfc3339'] = fields.DateTime.DESERIALIZATION_FUNCS['iso']
fields.DateTime.DEFAULT_FORMAT = 'rfc3339'

class LazyStringField(fields.String):
    """ Field that serializes a "false" value to the empty string
    """

    def _serialize(self, value, attr, obj, **kwargs):
        """ serialize None to the empty string
        """
        return value if value else ''

class CommaSeparatedListField(fields.Raw):
    """ Deserialize a string containing comma-separated values to
        a list of strings
    """

    default_error_messages = {
        "invalid": "Not a valid string or list.",
        "invalid_utf8": "Not a valid utf-8 string or list.",
    }

    def _deserialize(self, value, attr, data, **kwargs):
        """ deserialize comma separated string to list of strings
        """

        # empty
        if not value:
            return []

        # handle list
        if isinstance(value, list):
            try:
                value = [ensure_text_type(item) for item in value]
            except UnicodeDecodeError as exc:
                raise self.make_error("invalid_utf8") from exc

        # handle text
        else:
            if not isinstance(value, (str, bytes)):
                raise self.make_error("invalid")
            try:
                value = ensure_text_type(value)
            except UnicodeDecodeError as exc:
                raise self.make_error("invalid_utf8") from exc
            else:
                value = filter(bool, (item.strip() for item in value.split(',')))

        return list(value)


class DkimKeyField(fields.String):
    """ Serialize a dkim key to a multiline string and
        deserialize a dkim key data as string or list of strings
        to a valid dkim key
    """

    default_error_messages = {
        "invalid": "Not a valid string or list.",
        "invalid_utf8": "Not a valid utf-8 string or list.",
    }

    def _serialize(self, value, attr, obj, **kwargs):
        """ serialize dkim key as multiline string
        """

        # map empty string and None to None
        if not value:
            return ''

        # return multiline string
        return _Multiline(value.decode('utf-8'))

    def _wrap_key(self, begin, data, end):
        """ generator to wrap key into RFC 7468 format """
        yield begin
        pos = 0
        while pos < len(data):
            yield data[pos:pos+64]
            pos += 64
        yield end
        yield ''

    def _deserialize(self, value, attr, data, **kwargs):
        """ deserialize a string or list of strings to dkim key data
            with verification
        """

        # convert list to str
        if isinstance(value, list):
            try:
                value = ''.join(ensure_text_type(item) for item in value).strip()
            except UnicodeDecodeError as exc:
                raise self.make_error("invalid_utf8") from exc

        # only text is allowed
        else:
            if not isinstance(value, (str, bytes)):
                raise self.make_error("invalid")
            try:
                value = ensure_text_type(value).strip()
            except UnicodeDecodeError as exc:
                raise self.make_error("invalid_utf8") from exc

        # generate new key?
        if value.lower() == '-generate-':
            return dkim.gen_key()

        # no key or key is <hidden>?
        if not value or str(value) == '<hidden>':
            return None

        # remember part of value for ValidationError
        bad_key = value

        # strip header and footer, clean whitespace and wrap to 64 characters
        try:
            if value.startswith('-----BEGIN '):
                end = value.index('-----', 11) + 5
                header = value[:end]
                value = value[end:]
            else:
                header = '-----BEGIN PRIVATE KEY-----'

            if (pos := value.find('-----END ')) >= 0:
                end = value.index('-----', pos+9) + 5
                footer = value[pos:end]
                value = value[:pos]
            else:
                footer = '-----END PRIVATE KEY-----'
        except ValueError as exc:
            raise ValidationError(f'invalid dkim key {bad_key!r}') from exc

        # remove whitespace from key data
        value = ''.join(value.split())

        # remember part of value for ValidationError
        bad_key = f'{value[:25]}...{value[-10:]}' if len(value) > 40 else value

        # wrap key according to RFC 7468
        value = ('\n'.join(self._wrap_key(header, value, footer))).encode('ascii')

        # check key validity
        try:
            serialization.load_pem_private_key(value, password=None)
        except (UnicodeEncodeError, ValueError) as exc:
            raise ValidationError(f'invalid dkim key {bad_key!r}') from exc
        else:
            return value

class PasswordField(fields.Str):
    """ Serialize a hashed password hash by stripping the obsolete {SCHEME}
        Deserialize a plain password or hashed password into a hashed password
    """

    _hashes = {'PBKDF2', 'BLF-CRYPT', 'SHA512-CRYPT', 'SHA256-CRYPT', 'MD5-CRYPT', 'CRYPT'}

    def _serialize(self, value, attr, obj, **kwargs):
        """ strip obsolete {password-hash} when serializing """
        # strip scheme spec if in database - it's obsolete
        if value.startswith('{') and (end := value.find('}', 1)) >= 0:
            if value[1:end] in self._hashes:
                return value[end+1:]
        return value

    def _deserialize(self, value, attr, data, **kwargs):
        """ hashes plain password or checks hashed password
            also strips obsolete {password-hash} when deserializing
        """

        # when hashing is requested: use model instance to hash plain password
        if data.get('hash_password'):
            # hash password using model instance
            inst = self.metadata['model']()
            inst.set_password(value)
            value = inst.password
            del inst

        # strip scheme spec when specified - it's obsolete
        if value.startswith('{') and (end := value.find('}', 1)) >= 0:
            if value[1:end] in self._hashes:
                value = value[end+1:]

        # check if algorithm is supported
        inst = self.metadata['model'](password=value)
        try:
            # just check against empty string to see if hash is valid
            inst.check_password('')
        except ValueError as exc:
            # ValueError: hash could not be identified
            raise ValidationError(f'invalid password hash {value!r}') from exc
        del inst

        return value


### base schema ###

class Storage:
    """ Storage class to save information in context
    """

    context = {}

    def store(self, key, value):
        """ store value under key """
        key = f'{self.__class__.__name__}.{key}'
        self.context.setdefault('_track', {})[key] = value

    def recall(self, key):
        """ recall value from key """
        key = f'{self.__class__.__name__}.{key}'
        return self.context['_track'][key]

class BaseOpts(SQLAlchemyAutoSchemaOpts):
    """ Option class with sqla session
    """
    def __init__(self, meta, ordered=False):
        if not hasattr(meta, 'sqla_session'):
            meta.sqla_session = models.db.session
        if not hasattr(meta, 'sibling'):
            meta.sibling = False
        super(BaseOpts, self).__init__(meta, ordered=ordered)

class BaseSchema(ma.SQLAlchemyAutoSchema, Storage):
    """ Marshmallow base schema with custom exclude logic
        and option to hide sqla defaults
    """

    OPTIONS_CLASS = BaseOpts

    class Meta:
        """ Schema config """
        include_by_context = {}
        exclude_by_value = {}
        hide_by_context = {}
        order = []
        sibling = False

    def __init__(self, *args, **kwargs):

        # prepare only to auto-include explicitly specified attributes
        only = set(kwargs.get('only') or [])

        # get context
        context = kwargs.get('context', {})
        flags = {key for key, value in context.items() if value is True}

        # compile excludes
        exclude = set(kwargs.get('exclude', []))

        # always exclude
        exclude.update({'created_at', 'updated_at'} - only)

        # add include_by_context
        if context is not None:
            for need, what in getattr(self.Meta, 'include_by_context', {}).items():
                if not flags & set(need):
                    exclude |= what - only

        # update excludes
        kwargs['exclude'] = exclude

        # init SQLAlchemyAutoSchema
        super().__init__(*args, **kwargs)

        # exclude_by_value
        self._exclude_by_value = {
            key: values for key, values in getattr(self.Meta, 'exclude_by_value', {}).items()
            if key not in only
        }

        # exclude default values
        if not context.get('full'):
            for column in self.opts.model.__table__.columns:
                if column.name not in exclude and column.name not in only:
                    self._exclude_by_value.setdefault(column.name, []).append(
                        None if column.default is None else column.default.arg
                    )

        # hide by context
        self._hide_by_context = set()
        if context is not None:
            for need, what in getattr(self.Meta, 'hide_by_context', {}).items():
                if not flags & set(need):
                    self._hide_by_context |= what - only

        # remember primary keys
        self._primary = str(self.opts.model.__table__.primary_key.columns.values()[0].name)

        # determine attribute order
        if hasattr(self.Meta, 'order'):
            # use user-defined order
            order = self.Meta.order
        else:
            # default order is: primary_key + other keys alphabetically
            order = list(sorted(self.fields.keys()))
            if self._primary in order:
                order.remove(self._primary)
                order.insert(0, self._primary)

        # order fieldlists
        for fieldlist in (self.fields, self.load_fields, self.dump_fields):
            for field in order:
                if field in fieldlist:
                    fieldlist[field] = fieldlist.pop(field)

        # move post_load hook "_add_instance" to the end (after load_instance mixin)
        hooks = self._hooks[('post_load', False)]
        hooks.remove('_add_instance')
        hooks.append('_add_instance')

    def hide(self, data):
        """ helper method to hide input data for logging """
        # always returns a copy of data
        return {
            key: HIDDEN if key in self._hide_by_context else deepcopy(value)
            for key, value in data.items()
        }

    def get_parent(self):
        """ helper to determine parent of current object """
        for x in inspect.stack():
            loc = x[0].f_locals
            if 'ret_d' in loc:
                if isinstance(loc['self'], MailuSchema):
                    return self.context.get('config'), loc['attr_name']
                else:
                    return loc['self'].get_instance(loc['ret_d']), loc['attr_name']
        return None, None

    # this is only needed to work around the declared attr "email" primary key in model
    def get_instance(self, data):
        """ lookup item by defined primary key instead of key(s) from model """
        if self.transient:
            return None
        if keys := getattr(self.Meta, 'primary_keys', None):
            filters = {key: data.get(key) for key in keys}
            if None not in filters.values():
                try:
                    res = self.session.query(self.opts.model).filter_by(**filters).first()
                except sqlalchemy.exc.StatementError as exc:
                    raise ValidationError(f'Invalid {keys[0]}: {data.get(keys[0])!r}', data.get(keys[0])) from exc
                else:
                    return res
        res = super().get_instance(data)
        return res

    @pre_load(pass_many=True)
    def _patch_many(self, items, many, **kwargs): # pylint: disable=unused-argument
        """ - flush sqla session before serializing a section when requested
              (make sure all objects that could be referred to later are created)
            - when in update mode: patch input data before deserialization
              - handle "prune" and "delete" items
              - replace values in keys starting with '-' with default
        """

        # flush sqla session
        if not self.Meta.sibling:
            self.opts.sqla_session.flush()

        # stop early when not updating
        if not self.context.get('update'):
            return items

        # patch "delete", "prune" and "default"
        want_prune = []
        def patch(count, data):

            # we only process objects here
            if type(data) is not dict:
                raise ValidationError(f'Invalid item. {self.Meta.model.__tablename__.title()} needs to be an object.', f'{data!r}')

            # don't allow __delete__ coming from input
            if '__delete__' in data:
                raise ValidationError('Unknown field.', f'{count}.__delete__')

            # fail when hash_password is specified without password
            if 'hash_password' in data and not 'password' in data:
                raise ValidationError(
                    'Nothing to hash. Field "password" is missing.',
                    field_name = f'{count}.hash_password',
                )

            # handle "prune list" and "delete item" (-pkey: none and -pkey: id)
            for key in data:
                if key.startswith('-'):
                    if key[1:] == self._primary:
                        # delete or prune
                        if data[key] is None:
                            # prune
                            want_prune.append(True)
                            return None
                        # mark item for deletion
                        return {key[1:]: data[key], '__delete__': count}

            # handle "set to default value" (-key: none)
            def set_default(key, value):
                if not key.startswith('-'):
                    return (key, value)
                key = key[1:]
                if not key in self.opts.model.__table__.columns:
                    return (key, None)
                if value is not None:
                    raise ValidationError(
                        'Value must be "null" when resetting to default.',
                        f'{count}.{key}'
                    )
                value = self.opts.model.__table__.columns[key].default
                if value is None:
                    raise ValidationError(
                        'Field has no default value.',
                        f'{count}.{key}'
                    )
                return (key, value.arg)

            return dict(set_default(key, value) for key, value in data.items())

        # convert items to "delete" and filter "prune" item
        items = [
            item for item in [
                patch(count, item) for count, item in enumerate(items)
            ] if item
        ]

        # remember if prune was requested for _prune_items@post_load
        self.store('prune', bool(want_prune))

        # remember original items to stabilize password-changes in _add_instance@post_load
        self.store('original', items)

        return items

    @pre_load
    def _patch_item(self, data, many, **kwargs): # pylint: disable=unused-argument
        """ - call callback function to track import
            - stabilize import of items with auto-increment primary key
            - delete items
            - delete/prune list attributes
            - add missing required attributes
        """

        # callback
        if callback := self.context.get('callback'):
            callback(self, data)

        # stop early when not updating
        if not self.opts.load_instance or not self.context.get('update'):
            return data

        # stabilize import of auto-increment primary keys (not required),
        # by matching import data to existing items and setting primary key
        if not self._primary in data:
            parent, field = self.get_parent()
            if parent is not None:
                for item in getattr(parent, field):
                    existing = self.dump(item, many=False)
                    this = existing.pop(self._primary)
                    if data == existing:
                        self.instance = item
                        data[self._primary] = this
                        break

        # try to load instance
        instance = self.instance or self.get_instance(data)
        if instance is None:

            if '__delete__' in data:
                # deletion of non-existent item requested
                raise ValidationError(
                    f'Item to delete not found: {data[self._primary]!r}.',
                    field_name = f'{data["__delete__"]}.{self._primary}',
                )

        else:

            if self.context.get('update'):
                # delete instance from session when marked
                if '__delete__' in data:
                    self.opts.sqla_session.delete(instance)
                # delete item from lists or prune lists
                # currently: domain.alternatives, user.forward_destination,
                # user.manager_of, aliases.destination
                for key, value in data.items():
                    if not isinstance(self.fields.get(key), (
                        RelatedList, CommaSeparatedListField, fields.Raw)
                    ) or not isinstance(value, list):
                        continue
                    # deduplicate new value
                    new_value = set(value)
                    # handle list pruning
                    if '-prune-' in value:
                        value.remove('-prune-')
                        new_value.remove('-prune-')
                    else:
                        for old in getattr(instance, key):
                            # using str() is okay for now (see above)
                            new_value.add(str(old))
                    # handle item deletion
                    for item in value:
                        if item.startswith('-'):
                            new_value.remove(item)
                            try:
                                new_value.remove(item[1:])
                            except KeyError as exc:
                                raise ValidationError(
                                    f'Item to delete not found: {item[1:]!r}.',
                                    field_name=f'?.{key}',
                                ) from exc
                    # sort list of new values
                    data[key] = sorted(new_value)
                    # log backref modification not caught by modify hook
                    if isinstance(self.fields[key], RelatedList):
                        if callback := self.context.get('callback'):
                            before = {str(v) for v in getattr(instance, key)}
                            after = set(data[key])
                            if before != after:
                                callback(self, instance, {
                                    'key': key,
                                    'target': str(instance),
                                    'before': before,
                                    'after': after,
                                })

            # add attributes required for validation from db
            for attr_name, field_obj in self.load_fields.items():
                if field_obj.required and attr_name not in data:
                    data[attr_name] = getattr(instance, attr_name)

        return data

    @post_load(pass_many=True)
    def _prune_items(self, items, many, **kwargs): # pylint: disable=unused-argument
        """ handle list pruning """

        # stop early when not updating
        if not self.context.get('update'):
            return items

        # get prune flag from _patch_many@pre_load
        want_prune = self.recall('prune')

        # prune: determine if existing items in db need to be added or marked for deletion
        add_items = False
        del_items = False
        if self.Meta.sibling:
            # parent prunes automatically
            if not want_prune:
                # no prune requested => add old items
                add_items = True
        else:
            # parent does not prune automatically
            if want_prune:
                # prune requested => mark old items for deletion
                del_items = True

        if add_items or del_items:
            parent, field = self.get_parent()
            if parent is not None:
                existing = {item[self._primary] for item in items if self._primary in item}
                for item in getattr(parent, field):
                    key = getattr(item, self._primary)
                    if key not in existing:
                        if add_items:
                            items.append({self._primary: key})
                        else:
                            if self.context.get('update'):
                                self.opts.sqla_session.delete(self.instance or self.get_instance({self._primary: key}))

        return items

    @post_load
    def _add_instance(self, item, many, **kwargs): # pylint: disable=unused-argument
        """ - undo password change in existing instances when plain password did not change
            - add new instances to sqla session
        """

        if not item in self.opts.sqla_session:
            self.opts.sqla_session.add(item)
            return item

        # stop early when not updating or item has no password attribute
        if not self.context.get('update') or not hasattr(item, 'password'):
            return item

        # did we hash a new plaintext password?
        original = None
        pkey = getattr(item, self._primary)
        for data in self.recall('original'):
            if 'hash_password' in data and data.get(self._primary) == pkey:
                original = data['password']
                break
        if original is None:
            # password was hashed by us
            return item

        # reset hash if plain password matches hash from db
        if attr := getattr(sqlalchemy.inspect(item).attrs, 'password', None):
            if attr.history.has_changes() and attr.history.deleted:
                try:
                    # reset password hash
                    inst = type(item)(password=attr.history.deleted[-1])
                    if inst.check_password(original):
                        item.password = inst.password
                except ValueError:
                    # hash in db is invalid
                    pass
                else:
                    del inst

        return item

    @post_dump
    def _hide_values(self, data, many, **kwargs): # pylint: disable=unused-argument
        """ hide secrets """

        # stop early when not excluding/hiding
        if not self._exclude_by_value and not self._hide_by_context:
            return data

        # exclude or hide values
        full = self.context.get('full')
        return type(data)(
            (key, HIDDEN if key in self._hide_by_context else value)
            for key, value in data.items()
            if full or key not in self._exclude_by_value or value not in self._exclude_by_value[key]
        )

    # this field is used to mark items for deletion
    mark_delete = fields.Boolean(data_key='__delete__', load_only=True)

    # TODO: this can be removed when comment is not nullable in model
    comment = LazyStringField()


### schema definitions ###

@mapped
class DomainSchema(BaseSchema):
    """ Marshmallow schema for Domain model """
    class Meta:
        """ Schema config """
        model = models.Domain
        load_instance = True
        include_relationships = True
        exclude = ['users', 'managers', 'aliases']

        include_by_context = {
            ('dns',): {'dkim_publickey', 'dns_mx', 'dns_spf', 'dns_dkim', 'dns_dmarc'},
        }
        hide_by_context = {
            ('secrets',): {'dkim_key'},
        }
        exclude_by_value = {
            'alternatives': [[]],
            'dkim_key': [None],
            'dkim_publickey': [None],
            'dns_mx': [None],
            'dns_spf': [None],
            'dns_dkim': [None],
            'dns_dmarc': [None],
        }

    dkim_key = DkimKeyField(allow_none=True)
    dkim_publickey = fields.String(dump_only=True)
    dns_mx = fields.String(dump_only=True)
    dns_spf = fields.String(dump_only=True)
    dns_dkim = fields.String(dump_only=True)
    dns_dmarc = fields.String(dump_only=True)


@mapped
class TokenSchema(BaseSchema):
    """ Marshmallow schema for Token model """
    class Meta:
        """ Schema config """
        model = models.Token
        load_instance = True

        sibling = True

    password = PasswordField(required=True, metadata={'model': models.User})
    hash_password = fields.Boolean(load_only=True, missing=False)


@mapped
class FetchSchema(BaseSchema):
    """ Marshmallow schema for Fetch model """
    class Meta:
        """ Schema config """
        model = models.Fetch
        load_instance = True

        sibling = True
        include_by_context = {
            ('full', 'import'): {'last_check', 'error'},
        }
        hide_by_context = {
            ('secrets',): {'password'},
        }


@mapped
class UserSchema(BaseSchema):
    """ Marshmallow schema for User model """
    class Meta:
        """ Schema config """
        model = models.User
        load_instance = True
        include_relationships = True
        exclude = ['_email', 'domain', 'localpart', 'domain_name', 'quota_bytes_used']

        primary_keys = ['email']
        exclude_by_value = {
            'forward_destination': [[]],
            'tokens':              [[]],
            'fetches':             [[]],
            'manager_of':          [[]],
            'reply_enddate':       ['2999-12-31'],
            'reply_startdate':     ['1900-01-01'],
        }

    email = fields.String(required=True)
    tokens = fields.Nested(TokenSchema, many=True)
    fetches = fields.Nested(FetchSchema, many=True)

    password = PasswordField(required=True, metadata={'model': models.User})
    hash_password = fields.Boolean(load_only=True, missing=False)


@mapped
class AliasSchema(BaseSchema):
    """ Marshmallow schema for Alias model """
    class Meta:
        """ Schema config """
        model = models.Alias
        load_instance = True
        exclude = ['_email', 'domain', 'localpart', 'domain_name']

        primary_keys = ['email']
        exclude_by_value = {
            'destination': [[]],
        }

    email = fields.String(required=True)
    destination = CommaSeparatedListField()


@mapped
class ConfigSchema(BaseSchema):
    """ Marshmallow schema for Config model """
    class Meta:
        """ Schema config """
        model = models.Config
        load_instance = True


@mapped
class RelaySchema(BaseSchema):
    """ Marshmallow schema for Relay model """
    class Meta:
        """ Schema config """
        model = models.Relay
        load_instance = True


@mapped
class MailuSchema(Schema, Storage):
    """ Marshmallow schema for complete Mailu config """
    class Meta:
        """ Schema config """
        model = models.MailuConfig
        render_module = RenderYAML

        order = ['domain', 'user', 'alias', 'relay'] # 'config'

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        # order fieldlists
        for fieldlist in (self.fields, self.load_fields, self.dump_fields):
            for field in self.Meta.order:
                if field in fieldlist:
                    fieldlist[field] = fieldlist.pop(field)

    @pre_load
    def _clear_config(self, data, many, **kwargs): # pylint: disable=unused-argument
        """ create config object in context if missing
            and clear it if requested
        """
        if 'config' not in self.context:
            self.context['config'] = models.MailuConfig()
        if self.context.get('clear'):
            self.context['config'].clear(
                models = {field.nested.opts.model for field in self.fields.values()}
            )
        return data

    @post_load
    def _make_config(self, data, many, **kwargs): # pylint: disable=unused-argument
        """ update and return config object """
        config = self.context['config']
        for section in self.Meta.order:
            if section in data:
                config.update(data[section], section)

        return config

    domain = fields.Nested(DomainSchema, many=True)
    user = fields.Nested(UserSchema, many=True)
    alias = fields.Nested(AliasSchema, many=True)
    relay = fields.Nested(RelaySchema, many=True)
#    config = fields.Nested(ConfigSchema, many=True)