#!/usr/bin/env python3 """Mailcows SOGo-Webeinstiege auf Allianz Webmail umleiten (Python 3, ohne pip). Vorschau/Pruefung: python3 mailcow-webmail-weiterleitung.py --check Einrichten: python3 mailcow-webmail-weiterleitung.py --apply Ruecknahme: python3 mailcow-webmail-weiterleitung.py --remove Anderer Pfad: zusaetzlich --mailcow-dir /pfad/zu/mailcow-dockerized Nur die exakten Pfade /SOGo, /SOGo/, /SOGo/so und /SOGo/so/ werden umgeleitet. Mailcow-Login, Passwortwechsel, Adminbereich und /user bleiben erhalten. Die vorhandene Mailcow-Option zur direkten SOGo-Weiterleitung muss fuer die betroffenen Konten eingeschaltet bleiben. Passwortwechsel-Pflicht und 2FA werden NICHT abgeschaltet oder neu gesetzt. Keine Anmeldung wird uebertragen: Im neuen Webmailer ist eine weitere Anmeldung mit dem neuen Passwort noetig. Die Regel gilt fuer alle Mailcow-Domains und ersetzt nur den Webeinstieg; CalDAV/CardDAV, ActiveSync und bestehende tiefe SOGo-URLs werden nicht umgebogen. Keine PHP-Kerndateien, Passwoerter, Datenbankeintraege, DNS- oder TLS-Einstellungen werden veraendert. Nur Nginx bekommt einen Reload, kein Container wird gestoppt. Der vorhandene MAILCOW_HOSTNAME wird aus mailcow.conf gelesen, nicht geaendert. Rueckgabecodes: 0 = erfolgreich/geprueft; 1 = abgebrochen/Fehler; 2 = Regel geschrieben und Reload angefordert, oeffentliche Pruefung ausstehend. Grundlage: Mailcow master, eingesehen am 27.08.2026: https://github.com/mailcow/mailcow-dockerized/blob/master/data/web/index.php https://github.com/mailcow/mailcow-dockerized/blob/master/data/web/inc/triggers.user.inc.php https://github.com/mailcow/mailcow-dockerized/blob/master/data/conf/nginx/templates/sites-default.conf.j2 """ import argparse import datetime as dt import json import os from pathlib import Path import re import shutil import ssl import subprocess import sys import tempfile import time import urllib.error import urllib.request TARGET = 'https://webmail.unheilige-allianz.de/' NAME = 'site.allianz-webmail-redirect.custom' MARKER = '# Managed by mailcow-webmail-weiterleitung.py v1\n' ENTRY_PATHS = ('/SOGo', '/SOGo/', '/SOGo/so', '/SOGo/so/') INCLUDE = re.compile(r'^\s*include\s+/etc/nginx/conf\.d/site\.\*\.custom\s*;', re.M) class SetupError(Exception): pass def configuration(): text = MARKER + '# Exact web entry paths only. Do not redirect the entire /SOGo prefix.\n' for path in ENTRY_PATHS: text += ( '\nlocation = ' + path + ' {\n' ' add_header Cache-Control "no-store" always;\n' ' add_header Referrer-Policy "no-referrer" always;\n' ' return 303 ' + TARGET + ';\n' '}\n' ) return text.encode('utf-8') def default_directory(): manifest = Path('/opt/allianz-webmail/state/installation.json') if manifest.is_file(): try: value = json.loads(manifest.read_text()) if isinstance(value, dict) and value.get('project') == 'allianz-webmail' and isinstance(value.get('mailcow_dir'), str): return value['mailcow_dir'] except (OSError, ValueError): pass return '/opt/mailcow-dockerized' def settings(path): # Never source/eval mailcow.conf; other entries can contain secrets or shell text. result = {} for line in path.read_text(encoding='utf-8').splitlines(): match = re.fullmatch(r'\s*(MAILCOW_HOSTNAME|SKIP_SOGO)\s*=\s*(.*?)\s*', line) if match: value = match[2].split('#', 1)[0].strip() if len(value) >= 2 and value[0] in ('"', "'") and value[-1] == value[0]: value = value[1:-1] result[match[1]] = value host = result.get('MAILCOW_HOSTNAME', '').lower() labels = host.split('.') if len(host) > 253 or len(labels) < 2 or any(not re.fullmatch(r'[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?', label) for label in labels): raise SetupError('MAILCOW_HOSTNAME ist kein gültiger Hostname. Keine Änderung.') if host == 'webmail.unheilige-allianz.de': raise SetupError('Mailcow- und Webmail-Hostname dürfen für diese Lösung nicht identisch sein.') result['MAILCOW_HOSTNAME'] = host return result def compose(directory, *command): result = subprocess.run( ['docker', 'compose', '--project-directory', str(directory), 'exec', '-T', 'nginx-mailcow', *command], text=True, capture_output=True, timeout=30, check=False, ) if result.returncode: # A configuration dump may contain unrelated private configuration. Do not print it. raise SetupError('Nginx-/Docker-Prüfung fehlgeschlagen (Exit ' + str(result.returncode) + '). ' 'Containerstatus und „docker compose exec nginx-mailcow nginx -t“ prüfen.') return result.stdout def check_nginx(directory, require_include=True): # -T performs the syntax test and returns the actually included configuration. content = compose(directory, 'nginx', '-T') if require_include and not INCLUDE.search(content): raise SetupError('Der unterstützte Include für site.*.custom fehlt in der laufenden Nginx-Konfiguration. Keine automatische Anpassung.') return content def current_file(path): if path.is_symlink(): raise SetupError('Die Zieldatei ist ein symbolischer Link. Nicht verändert.') if not path.exists(): return None if not path.is_file(): raise SetupError('Das Weiterleitungsziel ist keine normale Datei. Nicht verändert.') content = path.read_bytes() if not content.startswith(MARKER.encode()): raise SetupError('Eine fremde Datei mit demselben Namen existiert bereits. Nicht überschrieben.') return content, path.stat().st_mode & 0o777 def write_atomic(path, value): if path.is_symlink(): raise SetupError('Die Zieldatei wurde zwischenzeitlich durch einen Link ersetzt.') if value is None: path.unlink(missing_ok=True) return content, mode = value descriptor, temporary = tempfile.mkstemp(prefix='.allianz-redirect-', dir=path.parent) try: with os.fdopen(descriptor, 'wb') as stream: stream.write(content) stream.flush() os.fsync(stream.fileno()) os.fchmod(stream.fileno(), mode) os.replace(temporary, path) finally: Path(temporary).unlink(missing_ok=True) def change(directory, path, previous, desired): if previous == desired: return False if current_file(path) != previous: raise SetupError('Die Datei wurde gleichzeitig geändert. Bitte erneut prüfen.') backup_base = directory / 'data/assets/webmail-redirect-backups' if backup_base.is_symlink(): raise SetupError('Das Sicherungsverzeichnis ist ein symbolischer Link. Abbruch.') backup_base.mkdir(parents=True, exist_ok=True, mode=0o700) stamp = dt.datetime.now(dt.timezone.utc).strftime('%Y%m%d-%H%M%S-%f') backup = backup_base / stamp backup.mkdir(mode=0o700) if previous is None: (backup / 'vorher-nicht-vorhanden.txt').write_text(str(path) + '\n', encoding='utf-8') else: (backup / NAME).write_bytes(previous[0]) (backup / NAME).chmod(0o600) print('Sicherung: ' + str(backup), flush=True) changed = False reload_attempted = False try: write_atomic(path, desired) changed = True loaded = check_nginx(directory, require_include=desired is not None) if desired is not None and desired[0].decode() not in loaded: raise SetupError('Die neue Datei ist im Nginx-Container nicht eingebunden. Bind-Mount/Installationspfad prüfen.') if desired is None and MARKER.strip() in loaded: raise SetupError('Die bisherige Regel ist im Container noch eingebunden. Bind-Mount/Installationspfad prüfen.') reload_attempted = True compose(directory, 'nginx', '-s', 'reload') except Exception as error: if changed: try: if current_file(path) != desired: raise SetupError('Datei gleichzeitig verändert; nicht automatisch überschrieben.') write_atomic(path, previous) check_nginx(directory, require_include=False) if reload_attempted: compose(directory, 'nginx', '-s', 'reload') except Exception as rollback_error: raise SetupError('Automatische Wiederherstellung nicht bestätigt. Sicherung: ' + str(backup)) from rollback_error raise SetupError('Änderung zurückgenommen; vorheriger Dateistand wiederhergestellt. ' + str(error)) from error raise return True class NoRedirect(urllib.request.HTTPRedirectHandler): def redirect_request(self, request, response, code, message, headers, new_url): return None def head(url): opener = urllib.request.build_opener( urllib.request.ProxyHandler({}), NoRedirect(), urllib.request.HTTPSHandler(context=ssl.create_default_context()), ) request = urllib.request.Request(url, method='HEAD', headers={'User-Agent': 'Allianz-Webmail-Redirect-Check/1.0'}) try: response = opener.open(request, timeout=4) except urllib.error.HTTPError as response_error: response = response_error with response: return response.code, response.headers.get('Location', '') def verify_public(host): # No credentials/cookies, no followed redirects, and full TLS verification. pending = list(ENTRY_PATHS) deadline = time.monotonic() + 15 while pending and time.monotonic() < deadline: try: status, location = head('https://' + host + pending[0]) except (OSError, urllib.error.URLError): return False if status == 303 and location == TARGET: pending.pop(0) else: time.sleep(0.25) return not pending def main(argv=None): parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) parser.add_argument('--mailcow-dir', default=default_directory(), help='Bestehendes Mailcow-Verzeichnis') action = parser.add_mutually_exclusive_group() action.add_argument('--apply', dest='action', action='store_const', const='apply', help='Weiterleitung einrichten') action.add_argument('--remove', dest='action', action='store_const', const='remove', help='Nur diese Weiterleitung zurücknehmen') action.add_argument('--check', dest='action', action='store_const', const='check', help='Nur prüfen (Standard)') parser.set_defaults(action='check') args = parser.parse_args(argv) if os.geteuid() != 0: raise SetupError('Bitte mit sudo oder als root ausführen.') if not shutil.which('docker'): raise SetupError('Docker fehlt. Dieses Skript gehört auf den bestehenden Mailcow-Server.') directory = Path(args.mailcow_dir).expanduser().resolve() if not (directory / 'mailcow.conf').is_file() or not (directory / 'docker-compose.yml').is_file(): raise SetupError('Keine Mailcow-Installation gefunden. --mailcow-dir /dein/pfad angeben.') conf_dir = directory / 'data/conf/nginx' if conf_dir.is_symlink() or not conf_dir.is_dir(): raise SetupError('Das Nginx-Konfigurationsverzeichnis fehlt oder ist ein symbolischer Link.') config = settings(directory / 'mailcow.conf') host = config['MAILCOW_HOSTNAME'] path = conf_dir / NAME previous = current_file(path) print('Mailcow: ' + str(directory), flush=True) print('Bestehende Anmeldung: https://' + host + '/', flush=True) print('Neues Webmail-Ziel: ' + TARGET, flush=True) if args.action != 'remove' and config.get('SKIP_SOGO', 'n').lower() not in ('n', 'no', 'false', '0'): raise SetupError('SKIP_SOGO deaktiviert die normale SOGo-Weiterleitung. Dieses Skript schaltet SOGo nicht wieder ein und ändert nichts.') # Removal can repair a currently invalid configuration caused by this file; # validate after removing it, not before. It also works after include changes. if args.action == 'remove': if change(directory, path, previous, None): print('Eigene Weiterleitungsdatei entfernt; Nginx-Konfiguration geprüft und Reload angefordert.') else: print('Keine eigene Weiterleitungsdatei vorhanden; nichts geändert.') return 0 loaded_before = check_nginx(directory) if args.action == 'check': print('Prüfung bestanden. Vier exakte SOGo-Webeinstiege würden umgeleitet; noch nichts geändert.') print('Zum Einrichten denselben Aufruf mit --apply ausführen.') return 0 try: status, location = head(TARGET) except (OSError, urllib.error.URLError) as error: raise SetupError('Das HTTPS-Webmail-Ziel ist nicht sicher erreichbar. Keine Änderung.') from error if status != 200: raise SetupError('Das Webmail-Ziel liefert nicht HTTP 200. Zuerst den Webmailer prüfen; keine Änderung.') desired = (configuration(), previous[1] if previous else 0o644) if change(directory, path, previous, desired): print('Regel geschrieben und Nginx-Reload angefordert.', flush=True) else: # The file may be present after a previously interrupted installation. if desired[0].decode() not in loaded_before: raise SetupError('Die vorhandene Datei ist im Nginx-Container nicht eingebunden. Installationspfad prüfen.') compose(directory, 'nginx', '-s', 'reload') print('Regel bereits vorhanden; Nginx-Reload erneut angefordert.', flush=True) if not verify_public(host): print('ACHTUNG: Die öffentliche Weiterleitung ist noch nicht bestätigt. Regel bleibt gespeichert.') print('DNS, vorgeschalteten Proxy und Erreichbarkeit von https://' + host + '/SOGo/so/ prüfen.') print('Rücknahme jederzeit mit --remove. Es wurden keine Container gestoppt.') return 2 print('Alle vier HTTPS-Einstiege liefern die erwartete Weiterleitung.') print('Jetzt den echten Ablauf testen: Mailcow-Login → ggf. Passwortwechsel/2FA → Webmail-Login.') print('Die Konto-Option „Direkte Weiterleitung zu SOGo“ muss weiterhin aktiv sein.') return 0 if __name__ == '__main__': try: sys.exit(main()) except (SetupError, OSError, ValueError, subprocess.TimeoutExpired) as error: print('Abbruch: ' + str(error), file=sys.stderr) sys.exit(1)