Merge branch 'feat/backup' into nightly

data/Dockerfiles/dockerapi/main.py

@@ -159,6 +159,9 @@ async def post_containers(container_id : str, post_action : str, request: Reques
       api_call_method = getattr(dockerapi, api_call_method_name, lambda container_id: Response(content=json.dumps({'type': 'danger', 'msg':'container_post - unknown api call' }, indent=4), media_type="application/json"))
 
       dockerapi.logger.info("api call: %s, container_id: %s" % (api_call_method_name, container_id))
+      if getattr(api_call_method, 'long_running', False):
+        # runs for minutes (e.g. sogo-tool of a backup restore): in a worker thread, the other api calls go on
+        return await asyncio.to_thread(api_call_method, request_json, container_id=container_id)
       return api_call_method(request_json, container_id=container_id)
     except Exception as e:
       dockerapi.logger.error("error - container_post: %s" % str(e))

data/Dockerfiles/dockerapi/modules/DockerApi.py

@@ -6,9 +6,15 @@ import time
 import json
 import asyncio
 import platform
+import docker
 from datetime import datetime
 from fastapi import FastAPI, Response, Request
 
+def long_running(method):
+  """Marks an api call that runs for minutes: main.py runs it in a worker thread, the other api calls go on"""
+  method.long_running = True
+  return method
+
 class DockerApi:
   def __init__(self, redis_client, sync_docker_client, async_docker_client, logger):
     self.redis_client = redis_client
@@ -445,6 +451,35 @@ class DockerApi:
         return Response(content=json.dumps({ 'type': 'danger', 'msg': 'users required' }, indent=4), media_type="application/json")
       init_return = container.exec_run(["/bin/bash", "-c", 'for user in "$@"; do sogo-tool init-user "$user" || exit 1; done', "init_user"] + users, user='sogo')
       return self.exec_run_handler('generic', init_return)
+  # api call: container_post - post_action: exec - cmd: sogo - task: restore
+  # Resets calendars, address books and preferences of a user to the sogo-tool backup file <dir>/<username> in a
+  # restore batch of the backup (/restore/<job>/b<batch>/sogo_backup, only the sogo user can read it); values are
+  # checked and passed as arguments, never through the shell. Runs in a worker thread (main.py) with its own client.
+  @long_running
+  def container_post__exec__sogo__restore(self, request_json, **kwargs):
+    if 'container_id' in kwargs:
+      filters = {"id": kwargs['container_id']}
+    elif 'container_name' in kwargs:
+      filters = {"name": kwargs['container_name']}
+
+    directory = str(request_json.get('dir', ''))
+    username = str(request_json.get('username', ''))
+    if (not re.fullmatch(r'/restore/[0-9]+/b[0-9]+/sogo_backup', directory)
+      or not re.fullmatch(r'[A-Za-z0-9_][A-Za-z0-9._+-]*@[A-Za-z0-9-]+(\.[A-Za-z0-9-]+)+', username)):
+      return Response(content=json.dumps({ 'type': 'danger', 'msg': 'dir and username required' }, indent=4), media_type="application/json")
+    # the folder: no symlink, uid 999, mode 0700; the file: plain, uid 999, mode 0600. sogo-tool restores folders and
+    # preferences in two runs, both stopped before the runner gives up waiting (30 minutes)
+    script = ('[ -d "$1" ] && [ ! -L "$1" ] && [ "$(realpath -e -- "$1")" = "$1" ] && [ "$(stat -c "%u %a" -- "$1")" = "999 700" ] '
+      '&& [ -f "$1/$2" ] && [ ! -L "$1/$2" ] && [ "$(stat -c "%u %a" -- "$1/$2")" = "999 600" ] '
+      '|| { echo "invalid directory or file"; exit 1; }; '
+      'timeout -k 30 850 sogo-tool restore -f ALL "$1" "$2" && exec timeout -k 30 850 sogo-tool restore -p "$1" "$2"')
+    client = docker.DockerClient(base_url='unix:///var/run/docker.sock', version='auto', timeout=1800)
+    try:
+      for container in client.containers.list(filters=filters):
+        restore_return = container.exec_run(["/bin/bash", "-c", script, "restore", directory, username], user='sogo')
+        return self.exec_run_handler('generic', restore_return)
+    finally:
+      client.close()
   # api call: container_post - post_action: exec - cmd: doveadm - task: subscribe_shared
   # Subscribes users to every folder of a mailbox shared with them (Shared/<owner>/...), or unsubscribes them
   def container_post__exec__doveadm__subscribe_shared(self, request_json, **kwargs):

data/Dockerfiles/dovecot/Dockerfile

@@ -151,6 +151,7 @@ COPY syslog-ng-redis_slave.conf /etc/syslog-ng/syslog-ng-redis_slave.conf
 COPY imapsync /usr/local/bin/imapsync
 COPY imapsync_runner.py /usr/local/bin/imapsync_runner.py
 COPY retention_runner.py /usr/local/bin/retention_runner.py
+COPY restore_runner.py /usr/local/bin/restore_runner.py
 COPY report-spam.sieve /usr/lib/dovecot/sieve/report-spam.sieve
 COPY report-ham.sieve /usr/lib/dovecot/sieve/report-ham.sieve
 COPY rspamd-pipe-ham /usr/lib/dovecot/sieve/rspamd-pipe-ham

data/Dockerfiles/dovecot/docker-entrypoint.sh

@@ -282,6 +282,7 @@ chmod +x /usr/lib/dovecot/sieve/rspamd-pipe-ham \
   /usr/lib/dovecot/sieve/rspamd-pipe-spam \
   /usr/local/bin/imapsync_runner.py \
   /usr/local/bin/retention_runner.py \
+  /usr/local/bin/restore_runner.py \
   /usr/local/bin/imapsync \
   /usr/local/bin/trim_logs.sh \
   /usr/local/bin/sa-rules.sh \

data/Dockerfiles/dovecot/restore_runner.py

@@ -0,0 +1,712 @@
+#!/usr/bin/python3
+# Mail part of restores (ofelia, dovecot container, as vmail): the restic container restores a batch of mailboxes
+# into /restore/<job id>/b<n> and sets the job to step 'mails'. Per mailbox this imports the mails folder by folder
+# (into "Restored <snapshot time>" or only the missing ones into the original folders), adds IMAP ACLs and missing
+# Sieve scripts, merges the results into the job and sets step 'mails_done'; the restic runner then removes the staging
+# and goes on with the next batch. Nothing that exists is changed or deleted.
+
+import base64
+import fcntl
+import json
+import os
+import re
+import smtplib
+import stat
+import subprocess
+import sys
+import threading
+import time
+from email.message import EmailMessage
+from email.utils import formatdate, make_msgid
+import MySQLdb
+import MySQLdb.cursors
+
+lock_file = '/tmp/restore_runner.lock'
+restore_dir = '/restore'
+max_log = 1000000
+# import, moves and the removal of the temporary folder ignore the quota (a restore may need more than is free),
+# the quota is recalculated afterwards
+no_quota = ['-o', 'plugin/quota=']
+# uid ranges per import (one argument stays far below the 128 KiB limit), folders per fetch and status
+uid_ranges = 2000
+folder_batch = 200
+# the restic runner waits for the batch while this heartbeat is fresh
+heartbeat_interval = 30
+# seconds to wait for the next batch of a restore
+batch_wait = 120
+rights_by_letter = {'l': 'lookup', 'r': 'read', 'w': 'write', 's': 'write-seen', 't': 'write-deleted', 'i': 'insert',
+  'p': 'post', 'e': 'expunge', 'k': 'create', 'x': 'delete', 'a': 'admin'}
+components_here = {'mails', 'imap_acl', 'filters', 'sieve_activate'}
+
+conn = None
+
+
+def connect():
+  return MySQLdb.connect(unix_socket='/var/run/mysqld/mysqld.sock', user=os.environ['DBUSER'],
+    passwd=os.environ['DBPASS'], db=os.environ['DBNAME'], charset='utf8mb4', autocommit=True)
+
+# Reconnect when the server dropped the connection (imports can outlast wait_timeout)
+def query(sql, args=None):
+  global conn
+  try:
+    conn.ping()
+  except (AttributeError, MySQLdb.Error):
+    conn = connect()
+  cur = conn.cursor(MySQLdb.cursors.DictCursor)
+  cur.execute(sql, args)
+  return cur
+
+# state_restore_seen while a batch runs, in its own connection
+class Heartbeat(threading.Thread):
+  def __init__(self):
+    super().__init__(daemon=True)
+    self.done = threading.Event()
+
+  def run(self):
+    db = None
+    while not self.done.is_set():
+      try:
+        if db is None:
+          db = connect()
+        db.cursor().execute("""INSERT INTO backup_setting (name, value) VALUES ('state_restore_seen', %s)
+          ON DUPLICATE KEY UPDATE value = VALUES(value)""", (str(int(time.time())),))
+      except MySQLdb.Error:
+        db = None
+      self.done.wait(heartbeat_interval)
+
+def log(job_id, text):
+  print('restore job %d: %s' % (job_id, text), flush=True)
+  line = '%s %s\n' % (time.strftime('%Y-%m-%d %H:%M:%S'), text)
+  query("UPDATE backup_job SET log = RIGHT(CONCAT(IFNULL(log, ''), %s), %s) WHERE id = %s", (line, max_log, job_id))
+
+def save(job_id, result):
+  query("UPDATE backup_job SET result = %s WHERE id = %s", (json.dumps(result), job_id))
+
+def doveadm(*args, data=None, check=True, partial=False):
+  result = subprocess.run(['doveadm', *args], input=data, stdin=None if data is not None else subprocess.DEVNULL,
+    capture_output=True, timeout=86400)
+  # mailbox status fails for folders that cannot be opened (implicit parents); the others are listed
+  if check and result.returncode != 0 and not (partial and result.stdout.strip()):
+    # mail_log writes an info line per message, the errors are the other lines
+    lines = [l for l in result.stderr.decode('utf-8', 'replace').splitlines() if ': Info: ' not in l]
+    raise RuntimeError('\n'.join(lines)[-500:] or 'doveadm exit code %d' % result.returncode)
+  return result.stdout.decode('utf-8', 'replace')
+
+def doveadm_json(*args, partial=False):
+  out = doveadm('-f', 'json', *args, partial=partial)
+  return json.loads(out) if out.strip() else []
+
+# free disk space that must stay after the import: max(2 GiB, 5 % of the filesystem), a full disk stops mail
+# (disk_reserve, mib and count_errors as in restic_runner.py of the backup image; tests/python/backup compares them)
+def disk_reserve(total):
+  return max(2 << 30, total // 20)
+
+def import_space_ok(need, free, total):
+  return need * 11 // 10 + disk_reserve(total) <= free
+
+def mib(size):
+  return '%.1f MiB' % (size / 1048576)
+
+def component(errors=None):
+  return {'added': 0, 'skipped': 0, 'errors': errors or []}
+
+# Folder names on disk: modified UTF-7, listescape writes . / \ as \xx
+def mutf7_decode(text):
+  def utf16(m):
+    if not m.group(1):
+      return '&'
+    data = m.group(1).replace(',', '/')
+    return base64.b64decode(data + '=' * (-len(data) % 4)).decode('utf-16-be')
+  return re.sub(r'&([A-Za-z0-9+,]*)-', utf16, text)
+
+def unescape(part):
+  return mutf7_decode(re.sub(r'\\([0-9a-fA-F]{2})', lambda m: chr(int(m.group(1), 16)), part))
+
+# Maildir++: the maildir is INBOX, ".A.B" is A/B
+def folder_name(entry):
+  return '/'.join(unescape(part) for part in entry[1:].split('.')) if entry else 'INBOX'
+
+def is_dir(path):
+  try:
+    return stat.S_ISDIR(os.lstat(path).st_mode)
+  except OSError:
+    return False
+
+def read_file(path):
+  fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW)
+  with os.fdopen(fd, 'rb') as f:
+    return f.read()
+
+# Prefixes of the shared and public namespaces, their folders belong to others
+def shared_prefixes():
+  prefixes, namespace, depth = {'Shared/'}, None, 0
+  # as vmail doveconf cannot read every included file (mail_crypt keys) and fails, but prints the namespaces
+  for line in subprocess.run(['doveconf', '-n'], capture_output=True, text=True).stdout.splitlines():
+    stripped = line.strip()
+    if stripped.endswith('{'):
+      depth += 1
+      if depth == 1 and stripped.startswith('namespace'):
+        namespace = {}
+    elif stripped == '}':
+      if depth == 1 and namespace is not None:
+        if namespace.get('type') in ('shared', 'public') and namespace.get('prefix'):
+          prefixes.add(namespace['prefix'].split('%')[0])
+        namespace = None
+      depth = max(depth - 1, 0)
+    elif depth == 1 and namespace is not None and '=' in stripped:
+      key, value = (part.strip() for part in stripped.split('=', 1))
+      namespace[key] = value
+  return tuple(prefix for prefix in prefixes if prefix)
+
+# name => guid of the user's own folders (implicit parents have none); the names are looked up literally
+def own_folders(user, prefixes):
+  names = [n for n in doveadm('mailbox', 'list', '-u', user).splitlines() if n and not n.startswith(prefixes)]
+  wanted, guids = set(names), {}
+  for i in range(0, len(names), folder_batch):
+    for row in doveadm_json('mailbox', 'status', '-u', user, 'guid', '--', *names[i:i + folder_batch], partial=True):
+      if row.get('mailbox') in wanted and re.fullmatch(r'[0-9a-f]{32}', row.get('guid') or ''):
+        guids[row['mailbox']] = row['guid']
+  return guids
+
+def masked(user):
+  return '*' in user or '?' in user
+
+def free_name(base, existing):
+  name, count = base, 1
+  while name in existing or any(folder.startswith(name + '/') for folder in existing):
+    count += 1
+    name = '%s (%d)' % (base, count)
+  return name
+
+# Path of the maildir now (/var/vmail/<domain>/<local>/Maildir), None for unknown or inactive mailboxes
+def mail_location(user):
+  result = subprocess.run(['doveadm', 'user', '-f', 'mail', '--', user], capture_output=True, text=True)
+  m = re.match(r'maildir:([^:]+)', result.stdout.strip()) if result.returncode == 0 else None
+  return m.group(1).rstrip('/') if m else None
+
+# The restored maildir of a mailbox (may hold only ACL files), None if the snapshot has nothing of it
+def staged_maildir(staging, home, location):
+  for path in (staging + location, home + '/' + os.environ.get('MAILDIR_SUB', 'Maildir'), home):
+    if is_dir(path):
+      return path
+  return None
+
+def uid_chunks(uids):
+  ranges, start, prev = [], None, None
+  for uid in sorted(uids):
+    if start is None:
+      start = prev = uid
+    elif uid == prev + 1:
+      prev = uid
+    else:
+      ranges.append('%d:%d' % (start, prev) if prev != start else str(start))
+      start = prev = uid
+  if start is not None:
+    ranges.append('%d:%d' % (start, prev) if prev != start else str(start))
+  return [','.join(ranges[i:i + uid_ranges]) for i in range(0, len(ranges), uid_ranges)]
+
+# Same message: same Message-ID, without one the same Date and size
+def message_key(row):
+  message_id = re.sub(r'\s+', '', row.get('hdr.message-id') or '').strip('<>')
+  if message_id:
+    return 'id:' + message_id
+  return 'date:%s:%s' % (row.get('date.sent'), row.get('size.physical'))
+
+def present_keys(user, guids):
+  keys = set()
+  guids = list(guids)
+  for i in range(0, len(guids), folder_batch):
+    search = []
+    for guid in guids[i:i + folder_batch]:
+      search += (['OR'] if search else []) + ['mailbox-guid', guid]
+    keys.update(message_key(row) for row in doveadm_json('fetch', '-u', user, 'hdr.message-id date.sent size.physical',
+      *search))
+  return keys
+
+def location(path):
+  if ':' in path:
+    raise RuntimeError('unsupported path')
+  return 'maildir:%s:INDEX=MEMORY' % path
+
+# The folders of a restored maildir as standalone maildirs (their messages are the INBOX there): the subfolders are
+# moved out of the maildir, so every folder is imported without the name rules of the source (listescape)
+def split_folders(maildir, home):
+  folders, failed = [('INBOX', maildir)], []
+  out = home + '/_restore_folders'
+  for entry in sorted(os.listdir(maildir)):
+    path = os.path.join(maildir, entry)
+    if not entry.startswith('.') or entry in ('.', '..') or not is_dir(path) or not is_dir(path + '/cur'):
+      continue
+    try:
+      name = folder_name(entry)
+    except (ValueError, UnicodeDecodeError):
+      failed.append(entry)
+      continue
+    if not os.path.isdir(out):
+      os.mkdir(out, 0o700)
+    target = '%s/f%d' % (out, len(folders))
+    os.rename(path, target)
+    folders.append((name, target))
+  return folders, failed
+
+def source_messages(user, root):
+  return doveadm_json('-o', 'mail_location=' + location(root), 'fetch', '-u', user,
+    'uid hdr.message-id date.sent size.physical', 'mailbox', 'INBOX')
+
+def remove_tree(user, name):
+  doveadm(*no_quota, 'mailbox', 'delete', '-u', user, '-r', '-s', '--', name, check=False)
+
+def folder_guid(user, name):
+  rows = doveadm_json('mailbox', 'status', '-u', user, 'guid', '--', name)
+  for row in rows:
+    if row.get('mailbox') == name:
+      return row['guid']
+  raise RuntimeError('folder not found after the import')
+
+def quota_room(user):
+  # free bytes of the storage quota, None without a limit
+  for row in doveadm_json('quota', 'get', '-u', user):
+    if row.get('type') == 'STORAGE' and str(row.get('limit', '-')).isdigit():
+      return (int(row['limit']) - int(row.get('value') or 0)) * 1024
+  return None
+
+# Subscriptions of the backup (subscriptions file: "V\t2", then names with \t between the levels)
+def backup_subscriptions(maildir):
+  try:
+    lines = read_file(maildir + '/subscriptions').decode('utf-8', 'replace').splitlines()
+  except OSError:
+    return []
+  if lines and lines[0].startswith('V\t'):
+    return ['/'.join(unescape(part) for part in line.split('\t')) for line in lines[1:] if line.strip()]
+  # version 1: storage names with the Maildir++ separator
+  return [folder_name('.' + line.strip()) for line in lines if line.strip()]
+
+def restore_subscriptions(user, wanted, prefixes):
+  if not wanted:
+    return 0
+  existing = set(own_folders(user, prefixes)) | {'INBOX'}
+  subscribed = set(doveadm('mailbox', 'list', '-u', user, '-s').splitlines())
+  missing = [name for name in dict.fromkeys(wanted) if name in existing and name not in subscribed]
+  if missing:
+    doveadm('mailbox', 'subscribe', '-u', user, '--', *missing)
+  return len(missing)
+
+# Mails folder by folder: original = only messages missing anywhere in the mailbox, into their original folders;
+# folder = all messages below "Restored <snapshot time>"
+def restore_mails(job_id, user, home, maildir, mode, snapshot_time, prefixes, check_quota):
+  result = component()
+  result.update(folders=0, folders_failed=0)
+  if not maildir or not is_dir(maildir + '/cur'):
+    # a key the UI translates
+    result['note'] = 'no_mails'
+    return result
+  tmp = 'Restoring %d' % job_id
+  # a leftover of an interrupted run goes first, it is never counted as present
+  remove_tree(user, tmp)
+  subscriptions = backup_subscriptions(maildir)
+  folders, broken = split_folders(maildir, home)
+  for entry in broken:
+    result['errors'].append('folder %s: name not readable' % entry)
+    result['folders_failed'] += 1
+  live = own_folders(user, prefixes)
+  present = present_keys(user, live.values()) if mode == 'original' else set()
+  plan, need = [], 0
+  for name, root in folders:
+    try:
+      rows = source_messages(user, root)
+    except RuntimeError as e:
+      result['errors'].append('%s: %s' % (name, e))
+      result['folders_failed'] += 1
+      continue
+    wanted = [row for row in rows if mode != 'original' or message_key(row) not in present]
+    result['skipped'] += len(rows) - len(wanted)
+    if wanted:
+      size = sum(int(row.get('size.physical') or 0) for row in wanted)
+      plan.append((name, root, [int(row['uid']) for row in wanted], size))
+      need += size
+  if not plan:
+    if mode == 'original':
+      result['subscribed'] = restore_subscriptions(user, subscriptions, prefixes)
+    return result
+  vfs = os.statvfs('/var/vmail')
+  free, total = vfs.f_bavail * vfs.f_frsize, vfs.f_blocks * vfs.f_frsize
+  if not import_space_ok(need, free, total):
+    raise RuntimeError('not enough free disk space for the mails: %s needed plus %s reserve, %s free' % (
+      mib(need * 11 // 10), mib(disk_reserve(total)), mib(free)))
+  if check_quota:
+    room = quota_room(user)
+    if room is not None and need > room:
+      raise RuntimeError('the mails do not fit into the quota of the mailbox: %s needed, %s free' % (
+        mib(need), mib(max(room, 0))))
+  names = set(live) | {'INBOX'}
+  dest = None
+  if mode != 'original':
+    dest = free_name('Restored ' + snapshot_time, names)
+    doveadm('mailbox', 'create', '-u', user, '-s', '--', dest)
+    names.add(dest)
+    result['folder'] = dest
+  created = []
+  try:
+    for number, (name, root, uids, size) in enumerate(plan):
+      target = dest + '/' + name if dest else name
+      stage = '%s/f%d' % (tmp, number)
+      try:
+        for chunk in uid_chunks(uids):
+          doveadm(*no_quota, 'import', '-u', user, '-s', location(root), stage, 'mailbox', 'INBOX', 'uid', chunk)
+        guid = folder_guid(user, stage + '/INBOX')
+        if target not in names:
+          doveadm('mailbox', 'create', '-u', user, '-s', '--', target)
+          names.add(target)
+          created.append(target)
+        doveadm(*no_quota, 'move', '-u', user, '--', target, 'mailbox-guid', guid, 'all')
+        result['added'] += len(uids)
+        result['folders'] += 1
+      except RuntimeError as e:
+        result['errors'].append('%s: %s' % (name, e))
+        result['folders_failed'] += 1
+  finally:
+    remove_tree(user, tmp)
+  result['folders_created'] = len(created)
+  if mode == 'original':
+    result['subscribed'] = restore_subscriptions(user, subscriptions, prefixes)
+  return result
+
+def acl_entries(path):
+  entries = []
+  try:
+    lines = read_file(path).decode('utf-8', 'replace').splitlines()
+  except OSError:
+    return entries
+  for line in lines:
+    parts = line.split()
+    # negative rights would take rights away
+    if len(parts) < 2 or parts[0].startswith('-'):
+      continue
+    rights = [rights_by_letter[c] for c in parts[1] if c in rights_by_letter] + [p for p in parts[2:] if p.startswith(':')]
+    if rights:
+      entries.append((parts[0], rights))
+  return entries
+
+# (folder, entries) of the dovecot-acl files of a restored maildir
+def maildir_acls(maildir):
+  acls = []
+  if not maildir or not is_dir(maildir):
+    return acls
+  for entry in [''] + sorted(e for e in os.listdir(maildir) if e.startswith('.') and e not in ('.', '..')):
+    path = os.path.join(maildir, entry, 'dovecot-acl') if entry else os.path.join(maildir, 'dovecot-acl')
+    if entry and not is_dir(os.path.join(maildir, entry)):
+      continue
+    try:
+      if stat.S_ISREG(os.lstat(path).st_mode):
+        acls.append((folder_name(entry), acl_entries(path)))
+    except (OSError, ValueError, UnicodeDecodeError):
+      continue
+  return acls
+
+# owner => (folder, entries) of every mailbox in the ACL staging (domain admins: only their domains)
+def staged_acls(acl_dir, allowed):
+  acls = {}
+  base = (acl_dir or '') + '/var/vmail'
+  sub = os.environ.get('MAILDIR_SUB', '')
+  if not acl_dir or not is_dir(base):
+    return acls
+  for domain in os.listdir(base):
+    if '.' not in domain or not is_dir(os.path.join(base, domain)) or (allowed is not None and domain.lower() not in allowed):
+      continue
+    for local in os.listdir(os.path.join(base, domain)):
+      home = os.path.join(base, domain, local)
+      maildir = os.path.join(home, sub) if sub and is_dir(os.path.join(home, sub)) else home
+      found = maildir_acls(maildir)
+      if found:
+        acls['%s@%s' % (local, domain)] = found
+  return acls
+
+class AclRestore:
+  def __init__(self, prefixes, allowed):
+    self.prefixes = prefixes
+    self.allowed = allowed
+    self.users, self.folders = {}, {}
+    # anyone/authenticated only when the configuration allows them
+    self.anyone = os.environ.get('ACL_ANYONE', 'disallow') == 'allow'
+
+  def in_scope(self, address):
+    return self.allowed is None or address.rpartition('@')[2].lower() in self.allowed
+
+  def user_exists(self, user):
+    if user not in self.users:
+      self.users[user] = query("SELECT 1 FROM mailbox WHERE username = %s", (user,)).fetchone() is not None
+    return self.users[user]
+
+  def identifier_ok(self, identifier):
+    if identifier.startswith('user='):
+      return self.user_exists(identifier[5:].lower())
+    # group=mcg_<id> belongs to the mailbox groups, the group grants cron keeps them
+    if identifier.startswith('group=mcg_'):
+      return False
+    if identifier in ('anyone', 'authenticated'):
+      return self.anyone
+    return True
+
+  def shown(self, identifier):
+    # domain admins do not see addresses of other domains
+    if identifier.startswith('user=') and not self.in_scope(identifier[5:]):
+      return 'user=(other domain)'
+    return identifier
+
+  def add(self, owner, folder, identifier, rights, result):
+    if not self.in_scope(owner) or masked(owner):
+      result['skipped'] += 1
+      return
+    try:
+      if owner not in self.folders:
+        self.folders[owner] = set(own_folders(owner, self.prefixes)) | {'INBOX'}
+      if folder not in self.folders[owner] or not self.identifier_ok(identifier):
+        result['skipped'] += 1
+        return
+      have = set()
+      for row in doveadm_json('acl', 'get', '-u', owner, '--', folder):
+        if row.get('id') == identifier:
+          have.update((row.get('rights') or '').split())
+      missing = [right for right in rights if right not in have]
+      if not missing:
+        result['skipped'] += 1
+        return
+      doveadm('acl', 'add', '-u', owner, '--', folder, identifier, *missing)
+      result['added'] += 1
+    except RuntimeError as e:
+      result['errors'].append('%s %s %s: %s' % (owner, folder, self.shown(identifier), e))
+
+# Rights the mailbox granted on its folders and rights other mailboxes granted to it; added, never removed
+def restore_acl(acl, user, own, granted):
+  result = component()
+  for folder, entries in own:
+    for identifier, rights in entries:
+      acl.add(user, folder, identifier, rights, result)
+  if acl.user_exists(user):
+    for owner, folders in granted.items():
+      if owner == user or not acl.in_scope(owner) or not acl.user_exists(owner):
+        continue
+      for folder, entries in folders:
+        for identifier, rights in entries:
+          if identifier.lower() == 'user=' + user:
+            acl.add(owner, folder, identifier, rights, result)
+  return result
+
+# Own Sieve scripts (incl. the SOGo vacation script): missing names are added; with activate the active script of the
+# backup is activated if none is active now
+def restore_sieve(user, home, active_link, activate):
+  result = component()
+  folder = home + '/sieve'
+  scripts = []
+  if is_dir(folder):
+    scripts = sorted(f[:-6] for f in os.listdir(folder) if f.endswith('.sieve') and stat.S_ISREG(os.lstat(os.path.join(folder, f)).st_mode))
+  current, active = set(), None
+  for line in doveadm('sieve', 'list', '-u', user).splitlines():
+    name, _, flag = line.rpartition(' ')
+    if name:
+      current.add(name)
+      if flag == 'ACTIVE':
+        active = name
+  for name in scripts:
+    if name in current:
+      result['skipped'] += 1
+      continue
+    try:
+      doveadm('sieve', 'put', '-u', user, '--', name, data=read_file(os.path.join(folder, name + '.sieve')))
+      current.add(name)
+      result['added'] += 1
+    except (OSError, RuntimeError) as e:
+      result['errors'].append('%s: %s' % (name, e))
+  if not activate:
+    return result
+  target = os.readlink(active_link) if os.path.islink(active_link) else ''
+  backup_active = os.path.basename(target)[:-6] if target.endswith('.sieve') else None
+  if backup_active and not active and backup_active in current:
+    try:
+      doveadm('sieve', 'activate', '-u', user, '--', backup_active)
+      result['activated'] = backup_active
+    except RuntimeError as e:
+      result['errors'].append('%s: %s' % (backup_active, e))
+  return result
+
+# filters: backup.php reported its part already, the counts are added up
+def merge(results, key, part):
+  old = results.get(key)
+  if not isinstance(old, dict):
+    results[key] = part
+    return
+  for count in ('added', 'skipped'):
+    old[count] = (old.get(count) or 0) + part.get(count, 0)
+  old['errors'] = (old.get('errors') or []) + part.get('errors', [])
+  for name, value in part.items():
+    if name not in ('added', 'skipped', 'errors'):
+      old[name] = value
+
+def count_errors(result):
+  count = sum(len(result.get(key, {}).get('errors') or []) for key in ('domain', 'aliases') if isinstance(result.get(key), dict))
+  for entry in (result.get('mailboxes') or {}).values():
+    count += int(bool(entry.get('error')))
+    count += sum(len(value.get('errors') or []) for value in entry.values() if isinstance(value, dict))
+  return count
+
+def restore_mailbox(job_id, user, components, mode, staging, snapshot_time, prefixes, acl, granted, check_quota, result):
+  local, _, domain = user.rpartition('@')
+  home = '%s/var/vmail/%s/%s' % (staging, domain, local)
+  if masked(user):
+    for key in components & {'mails', 'imap_acl', 'filters'}:
+      merge(result, key, component(['username with * or ?, doveadm reads it as a mask']))
+    return
+  location = mail_location(user)
+  if not location:
+    for key in components & {'mails', 'imap_acl', 'filters'}:
+      merge(result, key, component(['the mailbox does not exist now or is inactive']))
+    return
+  maildir = staged_maildir(staging, home, location)
+  # the ACL files are read before the folders of the maildir are moved for the import
+  own = maildir_acls(maildir) if 'imap_acl' in components else []
+  if 'mails' in components:
+    try:
+      result['mails'] = restore_mails(job_id, user, home, maildir, mode, snapshot_time, prefixes, check_quota)
+    except Exception as e:
+      result['mails'] = component([str(e)[:1000]])
+    try:
+      doveadm('quota', 'recalc', '-u', user)
+    except RuntimeError as e:
+      result['mails']['errors'].append('quota recalc: %s' % e)
+    # new folders
+    acl.folders.pop(user, None)
+  if 'imap_acl' in components:
+    try:
+      result['imap_acl'] = restore_acl(acl, user, own, granted)
+    except Exception as e:
+      result['imap_acl'] = component([str(e)[:1000]])
+  if 'filters' in components:
+    try:
+      part = restore_sieve(user, home, '%s/var/vmail/sieve/%s.sieve' % (staging, user), 'sieve_activate' in components)
+    except Exception as e:
+      part = component([str(e)[:1000]])
+    merge(result, 'filters', part)
+
+def fail(job_id, error):
+  query("""UPDATE backup_job SET state = IF(state = 'cancelled', state, 'failed'), error = %s, finished = NOW()
+    WHERE id = %s""", (error, job_id))
+  log(job_id, 'restore job failed: ' + error)
+  notify(job_id, error)
+
+def notify(job_id, error):
+  hostname = os.environ.get('MAILCOW_HOSTNAME', '')
+  row = query("SELECT value FROM backup_setting WHERE name = 'notify_email'").fetchone()
+  rcpts = [r.strip() for r in (row['value'] if row else '').split(',') if re.fullmatch(r'[^@\s<>,]+@[^@\s<>,]+', r.strip())]
+  if not rcpts:
+    return
+  msg = EmailMessage()
+  msg['From'] = 'backup@' + hostname
+  msg['To'] = ', '.join(rcpts)
+  msg['Subject'] = 'mailcow backup on %s: restore job %d failed' % (hostname, job_id)
+  msg['Date'] = formatdate(localtime=True)
+  msg['Message-ID'] = make_msgid(domain=hostname or None)
+  msg.set_content('The restore job %d failed:\n\n%s\n\nThe job log is in the backup section of the mailcow UI.\n'
+    % (job_id, error))
+  try:
+    with smtplib.SMTP('postfix', 588, hostname or None, timeout=60) as smtp:
+      smtp.send_message(msg)
+  except (OSError, smtplib.SMTPException) as e:
+    print('restore job %d: the notification could not be sent: %s' % (job_id, e), file=sys.stderr)
+
+def staging_dir(job_id, path):
+  # the batch folder of this job, never a symlink
+  if not isinstance(path, str) or not re.fullmatch(r'/restore/%d/[A-Za-z0-9_-]+' % job_id, path) or \
+      os.path.realpath(path) != path or not is_dir(path):
+    return None
+  return path
+
+def run_batch(job):
+  job_id = job['id']
+  params = json.loads(job['params'] or '{}')
+  result = json.loads(job['result'] or '{}')
+  batch = result.get('batch') or {}
+  staging = staging_dir(job_id, batch.get('dir'))
+  acl_dir = staging_dir(job_id, batch.get('acl_dir')) if batch.get('acl_dir') else None
+  mailboxes = [m for m in batch.get('mailboxes') or [] if isinstance(m, str)]
+  components = {c for c in params.get('components') or [] if c in components_here}
+  mode = 'original' if params.get('mail_target') == 'original' else 'folder'
+  allowed = {d.lower() for d in params['allowed_domains']} if isinstance(params.get('allowed_domains'), list) else None
+  check_quota = params.get('requested_role', 'admin') != 'admin'
+  snapshot_time = (result.get('snapshot') or {}).get('time') or time.strftime('%Y-%m-%d %H:%M')
+  results = result.setdefault('mailboxes', {})
+  prefixes = shared_prefixes()
+  acl = AclRestore(prefixes, allowed)
+  granted = staged_acls(acl_dir, allowed) if 'imap_acl' in components else {}
+  log(job_id, 'restoring %s for %d mailboxes%s' % (', '.join(sorted(components)), len(mailboxes),
+    ' (mails: %s)' % mode if 'mails' in components else ''))
+  for user in mailboxes:
+    entry = results.setdefault(user, {})
+    if entry.get('error'):
+      continue
+    started = time.monotonic()
+    if not staging:
+      for key in components & {'mails', 'imap_acl', 'filters'}:
+        merge(entry, key, component(['the restored files are missing']))
+    else:
+      restore_mailbox(job_id, user, components, mode, staging, snapshot_time, prefixes, acl, granted, check_quota, entry)
+    summary = ', '.join('%s +%d' % (key, entry[key].get('added', 0)) for key in sorted(components) if isinstance(entry.get(key), dict))
+    errors = sum(len(entry[key].get('errors') or []) for key in components if isinstance(entry.get(key), dict))
+    log(job_id, '%s: %s%s (%d s)' % (user, summary, ', %d errors' % errors if errors else '', time.monotonic() - started))
+    result['errors'] = count_errors(result)
+    save(job_id, result)
+  result['errors'] = count_errors(result)
+  save(job_id, result)
+  # the restic runner removes the staging and goes on
+  query("UPDATE backup_job SET step = 'mails_done' WHERE id = %s AND state = 'running' AND step = 'mails_running'", (job_id,))
+
+def main():
+  global conn
+  if os.geteuid() == 0:
+    # as vmail like the ofelia job: a run as root would leave a lock file the runs as vmail cannot open
+    os.execv('/usr/local/bin/gosu', ['gosu', 'vmail', sys.executable, os.path.abspath(__file__)] + sys.argv[1:])
+  os.umask(0o077)
+  # released by the kernel when the process ends, a crashed run leaves no stale lock
+  lock_fd = os.open(lock_file, os.O_RDWR | os.O_CREAT, 0o600)
+  try:
+    fcntl.flock(lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
+  except OSError:
+    sys.exit("restore: can't lock " + lock_file)
+  conn = connect()
+  heartbeat = Heartbeat()
+  heartbeat.start()
+  try:
+    # under the lock: a batch in mails_running was left by a crashed or stopped run, its temporary folders go
+    for row in query("""SELECT id, result FROM backup_job WHERE type = 'restore' AND state = 'running'
+        AND step = 'mails_running'""").fetchall():
+      try:
+        for user in (json.loads(row['result'] or '{}').get('batch') or {}).get('mailboxes') or []:
+          remove_tree(user, 'Restoring %d' % row['id'])
+          doveadm('quota', 'recalc', '-u', user, check=False)
+      except (ValueError, AttributeError, RuntimeError):
+        pass
+      fail(row['id'], 'interrupted: the restore of the mails was stopped')
+    last, waited = None, 0
+    while True:
+      job = query("""SELECT id, params, result FROM backup_job WHERE type = 'restore' AND state = 'running'
+        AND step = 'mails' ORDER BY id LIMIT 1""").fetchone()
+      if not job:
+        # the next batch of a restore follows within seconds: wait for it instead of the next ofelia run
+        row = query("SELECT state FROM backup_job WHERE id = %s", (last,)).fetchone() if last else None
+        if row and row['state'] == 'running' and waited < batch_wait:
+          time.sleep(2)
+          waited += 2
+          continue
+        break
+      last, waited = job['id'], 0
+      if query("UPDATE backup_job SET step = 'mails_running' WHERE id = %s AND state = 'running' AND step = 'mails'",
+          (job['id'],)).rowcount != 1:
+        continue
+      try:
+        run_batch(job)
+      except Exception as e:
+        fail(job['id'], str(e)[:1000])
+  finally:
+    heartbeat.done.set()
+
+if __name__ == '__main__':
+  main()

data/Dockerfiles/restic/Dockerfile

@@ -0,0 +1,28 @@
+FROM alpine:3.24
+
+LABEL maintainer="The Infrastructure Company GmbH <info@servercow.de>"
+
+ENV LANG=C.UTF-8
+ENV LC_ALL=C.UTF-8
+# MariaDB Connector/C >= 3.4 refuses DB servers without TLS, the runner connects through the socket
+ENV MARIADB_TLS_DISABLE_PEER_VERIFICATION=1
+
+RUN apk add --no-cache --update \
+  ca-certificates \
+  git \
+  mariadb-client \
+  openssh-client \
+  py3-mysqlclient \
+  py3-pynacl \
+  python3 \
+  restic \
+  tini \
+  tzdata \
+  # dump, export, manifest and the mailcow directory are below /mailcow (paths of the snapshot), only root may read them
+  && mkdir -m 700 /mailcow
+
+COPY restic_runner.py /usr/local/bin/restic_runner.py
+COPY sogo_restore.py /usr/local/bin/sogo_restore.py
+
+ENTRYPOINT ["/sbin/tini", "--"]
+CMD ["/usr/bin/python3", "-u", "/usr/local/bin/restic_runner.py"]

data/Dockerfiles/restic/sogo_restore.py

@@ -0,0 +1,118 @@
+"""SOGo part of a mailbox restore (component sogo, overwrites), called by the restic runner for one mailbox.
+
+Runs `sogo-tool restore -f ALL <dir> <user>` and `sogo-tool restore -p <dir> <user>` in the sogo container through
+dockerapi on the user's file of the nightly sogo-tool backup: missing calendars, address books and entries come back,
+the names and shares of existing folders and all SOGo preferences are replaced by the backup's, entries changed since
+stay. Then the user's entries in SOGo's cache (memcached) are removed, so the change shows at once.
+"""
+
+import json
+import os
+import re
+import socket
+import ssl
+import urllib.request
+
+dockerapi_url = 'https://dockerapi:443'
+memcached_addr = ('memcached-mailcow', 11211)
+# dockerapi stops each of the two sogo-tool runs after 850 s
+restore_timeout = 1800
+# accepted by the dockerapi handler as well
+user_re = re.compile(r'[A-Za-z0-9_][A-Za-z0-9._+-]*@[A-Za-z0-9-]+(\.[A-Za-z0-9-]+)+')
+
+
+def _short(text, limit=300):
+  # sogo-tool may print database URLs: no credentials, one line, the end has the error
+  text = re.sub(r'([A-Za-z][A-Za-z0-9+.-]*://)[^\s/@]*@', r'\1***@', str(text))
+  text = ' '.join(text.split())
+  return text if len(text) <= limit else '...' + text[-limit:]
+
+
+def _api(path, body=None, timeout=60):
+  # dockerapi uses the mail certificate, its name does not match
+  ctx = ssl.create_default_context()
+  ctx.check_hostname = False
+  ctx.verify_mode = ssl.CERT_NONE
+  data = None if body is None else json.dumps(body).encode()
+  req = urllib.request.Request(dockerapi_url + path, data=data, headers={'Content-Type': 'application/json'},
+    method='GET' if body is None else 'POST')
+  with urllib.request.urlopen(req, timeout=timeout, context=ctx) as response:
+    return json.load(response)
+
+
+def _sogo_container():
+  project = os.environ.get('COMPOSE_PROJECT_NAME', '').lower()
+  by_name = None
+  for cid, info in _api('/containers/json').items():
+    if not cid.isalnum() or not isinstance(info, dict):
+      continue
+    labels = (info.get('Config') or {}).get('Labels') or {}
+    if labels.get('com.docker.compose.service') == 'sogo-mailcow':
+      if not project or labels.get('com.docker.compose.project', '').lower() == project:
+        return cid
+    elif by_name is None and 'sogo-mailcow' in str(info.get('Name', '')):
+      by_name = cid
+  if by_name:
+    return by_name
+  raise RuntimeError('sogo container not found')
+
+
+def _memcached_delete(keys):
+  keys = [k for k in keys if 0 < len(k.encode()) <= 250 and not re.search(r'[\s\x00-\x1f\x7f]', k)]
+  if not keys:
+    return
+  with socket.create_connection(memcached_addr, timeout=10) as s:
+    for key in keys:
+      s.sendall(b'delete ' + key.encode() + b'\r\n')
+      reply = b''
+      while not reply.endswith(b'\r\n'):
+        chunk = s.recv(64)
+        if not chunk:
+          raise RuntimeError('memcached closed the connection')
+        reply += chunk
+      if reply.strip() not in (b'DELETED', b'NOT_FOUND'):
+        raise RuntimeError('memcached: %s' % _short(reply.decode('ascii', 'replace')))
+
+
+def _cache_keys(db, username):
+  # preferences and the ACLs of the user's folders (<uid>/Calendar/personal+acl); never read the c_*location
+  # columns of sogo_folder_info (database credentials)
+  keys = ['%s+defaults' % username, '%s+settings' % username]
+  cur = db.cursor()
+  try:
+    cur.execute('SELECT c_path FROM sogo_folder_info WHERE c_path2 = %s', (username,))
+    keys += ['%s+acl' % row[0][len('/Users/'):] for row in cur.fetchall() if row[0].startswith('/Users/')]
+  finally:
+    cur.close()
+  return keys
+
+
+def restore(db, backup_dir, username, log):
+  """Restores the SOGo data of one user from <backup_dir>/<username> (staged readable for the sogo user only).
+  db: open MySQLdb connection; log: callable for short progress lines. Returns {"added": 0|1, "errors": [...]}."""
+  result = {'added': 0, 'errors': []}
+  if not user_re.fullmatch(username):
+    result['errors'].append('username not supported by the SOGo restore')
+    return result
+  path = os.path.join(backup_dir, username)
+  if not os.path.isfile(path) or os.path.islink(path):
+    # sogo-tool backup ALL skips users without SOGo folders
+    # a key the UI translates
+    result['note'] = 'no_sogo_data'
+    log('sogo: no SOGo data in the snapshot')
+    return result
+  try:
+    res = _api('/containers/%s/exec' % _sogo_container(), {'cmd': 'sogo', 'task': 'restore', 'dir': backup_dir,
+      'username': username}, timeout=restore_timeout)
+    if not isinstance(res, dict) or res.get('type') != 'success':
+      raise RuntimeError('sogo-tool restore failed: %s' % _short(res.get('msg', '') if isinstance(res, dict) else res))
+    result['added'] = 1
+    log('sogo: calendars, address books and preferences reset to the backup')
+  except Exception as e:
+    result['errors'].append(_short(e))
+  try:
+    # SOGo caches ACLs and preferences for 15 minutes
+    _memcached_delete(_cache_keys(db, username))
+  except Exception as e:
+    result['errors'].append('SOGo cache not cleared, changes may show up after 15 minutes: %s' % _short(e))
+  return result