#!/usr/bin/env python3 """Standalone Roundcube -> RustBox contacts/identities migrator. The tool does not import or start RustBox. By default it reads RustBox's database configuration from the same TOML file as the application, detects SQLite/MySQL/ PostgreSQL, and writes directly to RustBox per-account tables. With --provision-missing it can create missing SQLite account tenants before importing data. """ from __future__ import annotations import argparse import hashlib import html import json import os import re import sqlite3 import sys from dataclasses import dataclass from pathlib import Path from typing import Any, Iterable, Mapping, Sequence from urllib.parse import parse_qs, quote, unquote, urlparse try: import tomllib except ModuleNotFoundError: # pragma: no cover - Python < 3.11 try: import tomli as tomllib # type: ignore[no-redef] except ModuleNotFoundError as exc: # pragma: no cover raise SystemExit( "Python 3.11+ is recommended. On older Python install 'tomli'." ) from exc EMAIL_RE = re.compile( r"[a-z0-9.!#$%&'*+/=?^_`{|}~-]+@" r"[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?" r"(?:\.[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?)+", re.IGNORECASE, ) IDENT_RE = re.compile(r"^[A-Za-z0-9_]+$") BREAK_RE = re.compile(r"||", re.IGNORECASE) TAG_RE = re.compile(r"<[^>]*>", re.DOTALL) @dataclass class Contact: id: str email: str name: str company: str notes: str phone: str job_title: str address: str website: str birthday: str emails_json: str phones_json: str addresses_json: str @dataclass class Identity: display_name: str email: str signature_html: str signature_text: str is_default: bool reply_to: str bcc: str organization: str @dataclass class VCard: name: str = "" emails: list[str] | None = None phones: list[str] | None = None addresses: list[str] | None = None company: str = "" notes: str = "" job_title: str = "" website: str = "" birthday: str = "" def __post_init__(self) -> None: self.emails = self.emails or [] self.phones = self.phones or [] self.addresses = self.addresses or [] class Db: """Small DB-API adapter for SQLite, PyMySQL and psycopg/psycopg2.""" def __init__(self, backend: str, conn: Any): self.backend = backend self.conn = conn @property def placeholder(self) -> str: return "?" if self.backend == "sqlite" else "%s" def ident(self, value: str) -> str: validate_identifier(value, "SQL identifier") if self.backend == "mysql": return f"`{value}`" return f'"{value}"' def execute(self, sql: str, params: Sequence[Any] = ()) -> Any: cur = self.conn.cursor() cur.execute(sql, tuple(params)) return cur def fetchone(self, sql: str, params: Sequence[Any] = ()) -> Any | None: cur = self.execute(sql, params) row = cur.fetchone() cur.close() return row def fetchall(self, sql: str, params: Sequence[Any] = ()) -> list[Any]: cur = self.execute(sql, params) rows = list(cur.fetchall()) cur.close() return rows def commit(self) -> None: self.conn.commit() def rollback(self) -> None: self.conn.rollback() def close(self) -> None: self.conn.close() def parser() -> argparse.ArgumentParser: p = argparse.ArgumentParser( description="Migrate Roundcube MySQL contacts and identities directly into RustBox.", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" Default RustBox config precedence: --config > RUSTBOX_CONFIG > /etc/rustbox/rustbox.toml Target DB precedence: --target-backend/--target-url > DATABASE_BACKEND/DATABASE_URL > config.toml Examples: # all users, RustBox target read from config.toml scripts/migrate_roundcube.py --roundcube-url mysql://rc:secret@db/roundcube --dry-run scripts/migrate_roundcube.py --roundcube-url mysql://rc:secret@db/roundcube # override both sides scripts/migrate_roundcube.py \\ --config ./config.toml \\ --target-backend mysql \\ --target-url mysql://rustbox:secret@db/rustbox \\ --roundcube-host rc-db --roundcube-database roundcube \\ --roundcube-db-user rc --roundcube-db-password secret Dependencies: MySQL source/target: PyMySQL (pip install pymysql) PostgreSQL target: psycopg (pip install 'psycopg[binary]') """, ) p.add_argument("--config", help="RustBox TOML config path") p.add_argument( "--target-backend", choices=("sqlite", "mysql", "pgsql", "postgres", "postgresql"), help="override RustBox DB backend", ) p.add_argument("--target-url", help="override RustBox DB URL/path") rc = p.add_argument_group("Roundcube MySQL source") rc.add_argument("--roundcube-url", help="mysql://user:password@host:3306/database") rc.add_argument("--roundcube-host", help="Roundcube MySQL host") rc.add_argument("--roundcube-port", type=int, default=None, help="Roundcube MySQL port (default 3306)") rc.add_argument("--roundcube-database", help="Roundcube database name") rc.add_argument("--roundcube-db-user", help="Roundcube database user") rc.add_argument("--roundcube-db-password", help="Roundcube database password") rc.add_argument("--roundcube-prefix", default=os.getenv("ROUNDCUBE_TABLE_PREFIX", ""), help="table prefix, e.g. rc_") mode = p.add_mutually_exclusive_group() mode.add_argument("--all-users", action="store_true", help="migrate every matching Roundcube user (default)") mode.add_argument("--roundcube-account", help="migrate only this Roundcube users.username") p.add_argument("--rustbox-user", help="single-user target login if different from Roundcube username") p.add_argument("--user-hash", help="single-user RustBox user hash override") p.add_argument("--target-prefix", help="single-user RustBox tenant prefix override") p.add_argument("--dry-run", action="store_true", help="read/map only; do not write") p.add_argument( "--provision-missing", action="store_true", help="create missing RustBox account tenants before import (SQLite target only)", ) p.add_argument("--fail-fast", action="store_true", help="stop all-users migration after first user failure") return p def validate_identifier(value: str, label: str, allow_empty: bool = False) -> None: if value == "" and allow_empty: return if not value or not IDENT_RE.fullmatch(value): raise ValueError(f"{label} may contain only ASCII letters, digits and underscore") def normalize_backend(raw: str | None) -> str | None: if raw is None: return None value = raw.strip().lower() aliases = {"postgres": "pgsql", "postgresql": "pgsql"} value = aliases.get(value, value) if value not in {"sqlite", "mysql", "pgsql"}: raise ValueError(f"unsupported database backend: {raw}") return value def infer_backend_from_url(raw: str | None) -> str | None: if not raw: return None value = raw.strip().lower() if value.startswith("mysql://"): return "mysql" if value.startswith("postgres://") or value.startswith("postgresql://"): return "pgsql" if value.startswith("sqlite:"): return "sqlite" if "://" not in value: return "sqlite" return None def load_config(path: Path) -> dict[str, Any]: if not path.is_file(): return {} with path.open("rb") as fh: data = tomllib.load(fh) if not isinstance(data, dict): raise ValueError(f"invalid TOML root in {path}") return data def sqlite_path(raw: str) -> Path: value = raw.strip() if value.startswith("sqlite://"): value = value[len("sqlite://") :] elif value.startswith("sqlite:"): value = value[len("sqlite:") :] value = value.split("?", 1)[0].strip() if not value or value == ":memory:": raise ValueError("SQLite target must point to a persistent database file") return Path(value) def resolve_target(args: argparse.Namespace) -> tuple[Path, dict[str, Any], str, str]: config_path = Path( args.config or os.getenv("RUSTBOX_CONFIG") or "/etc/rustbox/rustbox.toml" ) config = load_config(config_path) env_backend = os.getenv("DATABASE_BACKEND") or os.getenv("RUSTBOX_DATABASE_BACKEND") env_url = os.getenv("DATABASE_URL") or os.getenv("RUSTBOX_DATABASE_URL") env_data_dir = os.getenv("DATA_DIR") explicit_url_backend = infer_backend_from_url(args.target_url) if args.target_url else None backend = normalize_backend( args.target_backend or explicit_url_backend or env_backend or config.get("database_backend") or "sqlite" ) assert backend is not None target_url = args.target_url or env_url or config.get("database_url") if backend == "sqlite": if target_url: target_url = str(sqlite_path(str(target_url))) else: data_dir = str(env_data_dir or config.get("data_dir") or "/data") target_url = str(Path(data_dir) / "rustbox.sqlite") elif not target_url: raise ValueError( f"database_url is required for {backend}; set it in config.toml or pass --target-url" ) inferred = infer_backend_from_url(str(target_url)) if inferred is not None and inferred != backend: raise ValueError( f"target backend is {backend}, but target URL looks like {inferred}; " "override both --target-backend and --target-url together" ) if not config and not (args.target_url or env_url): raise FileNotFoundError( f"RustBox config not found: {config_path}; pass --config or --target-url" ) return config_path, config, backend, str(target_url) def build_roundcube_url(args: argparse.Namespace) -> str: env_url = os.getenv("ROUNDCUBE_DATABASE_URL") if args.roundcube_url: return args.roundcube_url if env_url and not any( [ args.roundcube_host, args.roundcube_database, args.roundcube_db_user, args.roundcube_db_password, args.roundcube_port, ] ): return env_url host = args.roundcube_host or os.getenv("ROUNDCUBE_HOST") port = args.roundcube_port or int(os.getenv("ROUNDCUBE_PORT", "3306")) database = args.roundcube_database or os.getenv("ROUNDCUBE_DATABASE") user = args.roundcube_db_user or os.getenv("ROUNDCUBE_DB_USER") password = ( args.roundcube_db_password if args.roundcube_db_password is not None else os.getenv("ROUNDCUBE_DB_PASSWORD", "") ) if not all([host, database, user]): raise ValueError( "Roundcube source is required: pass --roundcube-url or " "--roundcube-host/--roundcube-database/--roundcube-db-user" ) return ( f"mysql://{quote(str(user), safe='')}:{quote(str(password), safe='')}@" f"{host}:{port}/{quote(str(database), safe='')}" ) def parse_mysql_url(raw: str) -> dict[str, Any]: parsed = urlparse(raw) if parsed.scheme.lower() not in {"mysql", "mariadb"}: raise ValueError("MySQL URL must start with mysql://") database = unquote(parsed.path.lstrip("/")) if not parsed.hostname or not parsed.username or not database: raise ValueError("MySQL URL must contain user, host and database") query = parse_qs(parsed.query) out: dict[str, Any] = { "host": parsed.hostname, "port": parsed.port or 3306, "user": unquote(parsed.username), "password": unquote(parsed.password or ""), "database": database, "charset": query.get("charset", ["utf8mb4"])[0], "autocommit": False, } if "unix_socket" in query: out["unix_socket"] = query["unix_socket"][0] if "connect_timeout" in query: out["connect_timeout"] = int(query["connect_timeout"][0]) return out def connect_mysql(raw: str, *, dict_rows: bool) -> Db: try: import pymysql # type: ignore except ModuleNotFoundError as exc: raise RuntimeError( "PyMySQL is required for Roundcube/MySQL. Install: python3 -m pip install pymysql" ) from exc kwargs = parse_mysql_url(raw) if dict_rows: kwargs["cursorclass"] = pymysql.cursors.DictCursor conn = pymysql.connect(**kwargs) return Db("mysql", conn) def connect_pgsql(raw: str) -> Db: try: import psycopg # type: ignore return Db("pgsql", psycopg.connect(raw, autocommit=False)) except ModuleNotFoundError: try: import psycopg2 # type: ignore return Db("pgsql", psycopg2.connect(raw)) except ModuleNotFoundError as exc: raise RuntimeError( "PostgreSQL target requires psycopg. Install: python3 -m pip install 'psycopg[binary]'" ) from exc def connect_target(backend: str, raw: str) -> Db: if backend == "sqlite": path = sqlite_path(raw) conn = sqlite3.connect(path, timeout=10) conn.execute("PRAGMA foreign_keys=ON") return Db("sqlite", conn) if backend == "mysql": return connect_mysql(raw, dict_rows=False) if backend == "pgsql": return connect_pgsql(raw) raise ValueError(f"unsupported target backend: {backend}") def row_first(row: Any) -> Any: if row is None: return None if isinstance(row, Mapping): return next(iter(row.values())) return row[0] def row_tuple(row: Any) -> tuple[Any, ...]: if isinstance(row, Mapping): return tuple(row.values()) return tuple(row) def hash_user(value: str) -> str: return hashlib.sha256(value.encode("utf-8")).hexdigest() def tenant_prefix(user_hash: str) -> str: """Match RustBox db::pool::tenant_prefix exactly.""" if not user_hash: raise ValueError("empty user hash for database isolation") if len(user_hash) >= 24 and all(ch in "0123456789abcdefABCDEF" for ch in user_hash): normalized = user_hash.lower() else: normalized = hash_user(user_hash) return f"u_{normalized[:24]}__" def migrations_dir() -> Path: override = os.getenv("RUSTBOX_MIGRATIONS_DIR") if override: return Path(override) return Path(__file__).resolve().parent.parent / "migrations" def sqlite_tenant_schema() -> tuple[str, int]: root = migrations_dir() / "sqlite" schema_path = root / "schema.sql" if not schema_path.is_file(): raise FileNotFoundError( f"RustBox SQLite schema not found: {schema_path}; " "set RUSTBOX_MIGRATIONS_DIR to the migrations directory" ) schema = schema_path.read_text(encoding="utf-8") begin = "-- RUSTBOX:TENANT-BEGIN" end = "-- RUSTBOX:TENANT-END" if begin not in schema or end not in schema: raise ValueError(f"invalid RustBox tenant schema markers in {schema_path}") tenant_sql = schema.split(begin, 1)[1].split(end, 1)[0] versions = [52] for path in root.glob("V[0-9][0-9][0-9]__*.sql"): try: versions.append(int(path.name[1:4])) except ValueError: pass return tenant_sql, max(versions) def split_sql_statements(sql: str) -> list[str]: # RustBox schema files contain ordinary DDL/DML statements and no stored # procedures; splitting on semicolons keeps provisioning transactional in SQLite. return [statement.strip() for statement in sql.split(";") if statement.strip()] def provision_sqlite_account(target: Db, user_hash: str) -> str: if target.backend != "sqlite": raise ValueError("--provision-missing currently supports SQLite targets only") prefix = tenant_prefix(user_hash) existing = target_prefix_for_hash(target, user_hash) if existing is not None: if existing != prefix: raise ValueError("database account-prefix collision detected") return existing ph = target.placeholder collision = target.fetchone( f"SELECT user_hash FROM rustbox_accounts WHERE table_prefix={ph}", [prefix] ) if collision is not None: raise ValueError("database account-prefix collision detected") tenant_sql, schema_version = sqlite_tenant_schema() rendered = tenant_sql.replace("{{p}}", prefix) seed = ( f"INSERT OR IGNORE INTO {prefix}user_meta (key, value) VALUES ('schema_version', '1');" f"INSERT OR IGNORE INTO {prefix}notification_preferences (id) VALUES (1);" f"INSERT OR IGNORE INTO {prefix}mail_preferences (id, messages_per_page) VALUES (1, 300);" f"INSERT OR IGNORE INTO {prefix}pgp_preferences (id) VALUES (1);" ) try: target.execute("BEGIN IMMEDIATE").close() for statement in split_sql_statements(rendered): target.execute(statement).close() for statement in split_sql_statements(seed): target.execute(statement).close() target.execute( f"INSERT INTO rustbox_accounts(user_hash, table_prefix, schema_version) VALUES({ph}, {ph}, {ph})", [user_hash, prefix, schema_version], ).close() target.commit() except Exception: target.rollback() raise return prefix def dedupe(values: Iterable[str]) -> list[str]: result: list[str] = [] seen: set[str] = set() for raw in values: value = str(raw).strip() if not value: continue key = value.lower() if key not in seen: seen.add(key) result.append(value) return result def first_nonempty(*values: str) -> str: for value in values: if value and value.strip(): return value.strip() return "" def row_mapping(row: Any) -> Mapping[str, Any]: if isinstance(row, Mapping): return row if isinstance(row, sqlite3.Row): return {key: row[key] for key in row.keys()} raise TypeError("Roundcube query did not return named columns") def row_text(row: Mapping[str, Any], name: str) -> str: value = row.get(name) if value is None: return "" if isinstance(value, bytes): return value.decode("utf-8", errors="replace") return str(value) def row_int(row: Mapping[str, Any], name: str) -> int | None: value = row.get(name) if value is None or value == "": return None try: return int(value) except (TypeError, ValueError): return None def unescape_vcard(value: str) -> str: return ( value.replace("\\n", "\n") .replace("\\N", "\n") .replace("\\,", ",") .replace("\\;", ";") .replace("\\\\", "\\") .strip() ) def parse_vcard(raw: str) -> VCard: out = VCard() if not raw.strip(): return out unfolded: list[str] = [] for line in raw.replace("\r\n", "\n").replace("\r", "\n").split("\n"): if line.startswith((" ", "\t")) and unfolded: unfolded[-1] += line.lstrip() else: unfolded.append(line) for line in unfolded: if ":" not in line: continue left, raw_value = line.split(":", 1) prop = left.split(";", 1)[0].rsplit(".", 1)[-1].upper() value = unescape_vcard(raw_value) if prop == "FN" and not out.name: out.name = value elif prop == "EMAIL": out.emails.append(value) elif prop == "TEL": out.phones.append(value) elif prop == "ADR": address = ", ".join(part.strip() for part in value.split(";") if part.strip()) if address: out.addresses.append(address) elif prop == "ORG" and not out.company: out.company = " / ".join(part.strip() for part in value.split(";") if part.strip()) elif prop == "NOTE" and not out.notes: out.notes = value elif prop == "TITLE" and not out.job_title: out.job_title = value elif prop == "URL" and not out.website: out.website = value elif prop == "BDAY" and not out.birthday: out.birthday = value out.emails = dedupe(out.emails) out.phones = dedupe(out.phones) out.addresses = dedupe(out.addresses) return out def email_candidates(raw: str) -> list[str]: return dedupe(match.group(0) for match in EMAIL_RE.finditer(raw or "")) def stable_contact_id(roundcube_user: str, source_id: str, email_addr: str) -> str: stable_key = source_id.strip() or email_addr digest = hashlib.sha256(f"{roundcube_user}:{stable_key}".encode("utf-8")).hexdigest() return f"rc-{digest[:32]}" def plain_to_html(value: str) -> str: return html.escape(value, quote=True).replace("\r\n", "
").replace("\n", "
") def html_to_plain(value: str) -> str: text = BREAK_RE.sub("\n", value) text = TAG_RE.sub("", text) return html.unescape(text.replace(" ", " ")).strip() def source_table(prefix: str, name: str) -> str: validate_identifier(prefix, "Roundcube table prefix", allow_empty=True) validate_identifier(name, "Roundcube table name") return f"`{prefix}{name}`" def load_roundcube_users(source: Db, prefix: str) -> list[str]: table = source_table(prefix, "users") rows = source.fetchall( f"SELECT username FROM {table} WHERE username IS NOT NULL AND username<>'' ORDER BY user_id" ) return [str(row_first(row)) for row in rows] def load_roundcube_data(source: Db, prefix: str, username: str) -> tuple[list[Contact], list[Identity], int]: users = source_table(prefix, "users") contacts_table = source_table(prefix, "contacts") identities_table = source_table(prefix, "identities") ph = source.placeholder row = source.fetchone(f"SELECT user_id FROM {users} WHERE username={ph} LIMIT 1", [username]) if row is None: raise ValueError(f"Roundcube user not found: {username}") user_id = row_first(row) contact_rows = source.fetchall(f"SELECT * FROM {contacts_table} WHERE user_id={ph}", [user_id]) contacts: list[Contact] = [] skipped_contacts = 0 for raw_row in contact_rows: row_map = row_mapping(raw_row) if row_int(row_map, "del") not in (None, 0): continue vcard = parse_vcard(row_text(row_map, "vcard")) emails = list(vcard.emails) emails.extend(email_candidates(row_text(row_map, "email"))) emails = dedupe(email_candidates(" ".join(emails))) if not emails: skipped_contacts += 1 continue primary_email = emails[0] full_name = " ".join( part for part in [row_text(row_map, "firstname"), row_text(row_map, "surname")] if part.strip() ) name = first_nonempty(vcard.name, row_text(row_map, "name"), full_name, primary_email) company = first_nonempty(vcard.company, row_text(row_map, "organization"), row_text(row_map, "company")) notes = first_nonempty(vcard.notes, row_text(row_map, "notes")) job_title = first_nonempty(vcard.job_title, row_text(row_map, "job_title"), row_text(row_map, "title")) website = first_nonempty(vcard.website, row_text(row_map, "website")) birthday = first_nonempty(vcard.birthday, row_text(row_map, "birthday")) phones = list(vcard.phones) direct_phone = row_text(row_map, "phone") if direct_phone.strip(): phones.append(direct_phone) phones = dedupe(phones) addresses = list(vcard.addresses) direct_address = row_text(row_map, "address") if direct_address.strip(): addresses.append(direct_address) addresses = dedupe(addresses) source_id = first_nonempty(row_text(row_map, "contact_id"), primary_email) contacts.append( Contact( id=stable_contact_id(username, source_id, primary_email), email=primary_email, name=name, company=company, notes=notes, phone=phones[0] if phones else "", job_title=job_title, address=addresses[0] if addresses else "", website=website, birthday=birthday, emails_json=json.dumps(emails[1:], ensure_ascii=False, separators=(",", ":")), phones_json=json.dumps(phones[1:], ensure_ascii=False, separators=(",", ":")), addresses_json=json.dumps(addresses[1:], ensure_ascii=False, separators=(",", ":")), ) ) identity_rows = source.fetchall(f"SELECT * FROM {identities_table} WHERE user_id={ph}", [user_id]) identities: list[Identity] = [] for raw_row in identity_rows: row_map = row_mapping(raw_row) if row_int(row_map, "del") not in (None, 0): continue candidates = email_candidates(row_text(row_map, "email")) if not candidates: continue signature = row_text(row_map, "signature") html_signature = (row_int(row_map, "html_signature") or 0) != 0 signature_html, signature_text = ( (signature, html_to_plain(signature)) if html_signature else (plain_to_html(signature), signature) ) identities.append( Identity( display_name=row_text(row_map, "name"), email=candidates[0], signature_html=signature_html, signature_text=signature_text, is_default=(row_int(row_map, "standard") or 0) != 0, reply_to=first_nonempty(row_text(row_map, "reply-to"), row_text(row_map, "reply_to")), bcc=row_text(row_map, "bcc"), organization=row_text(row_map, "organization"), ) ) return contacts, identities, skipped_contacts def target_prefix_for_hash(target: Db, user_hash: str) -> str | None: ph = target.placeholder row = target.fetchone( f"SELECT table_prefix FROM rustbox_accounts WHERE user_hash={ph}", [user_hash] ) if row is None: return None prefix = str(row_first(row)) validate_identifier(prefix, "RustBox table prefix") return prefix def target_account_for_login(target: Db, login: str) -> tuple[str, str, str] | None: exact_hash = hash_user(login) prefix = target_prefix_for_hash(target, exact_hash) if prefix is not None: return exact_hash, prefix, login ph = target.placeholder row = target.fetchone( "SELECT a.user_hash, a.table_prefix, s.email " "FROM sessions s JOIN rustbox_accounts a ON a.user_hash=s.user_hash " f"WHERE lower(s.email)=lower({ph}) ORDER BY s.last_accessed_unix DESC LIMIT 1", [login], ) if row is None: return None user_hash, prefix, email_addr = row_tuple(row)[:3] validate_identifier(str(prefix), "RustBox table prefix") return str(user_hash), str(prefix), str(email_addr) def now_expr(backend: str) -> str: if backend == "sqlite": return "datetime('now')" if backend == "mysql": return "CAST(CURRENT_TIMESTAMP AS CHAR)" return "CURRENT_TIMESTAMP::text" def upsert_contact(target: Db, prefix: str, contact: Contact) -> None: table = target.ident(f"{prefix}contacts") ph = target.placeholder existing = target.fetchone( f"SELECT id FROM {table} WHERE lower(email)=lower({ph}) ORDER BY created_at LIMIT 1", [contact.email], ) values = [ contact.email, contact.name, contact.company, contact.notes, contact.phone, contact.job_title, contact.address, contact.website, contact.birthday, contact.emails_json, contact.phones_json, contact.addresses_json, ] if existing is not None: contact_id = row_first(existing) assigns = ",".join( f"{column}={ph}" for column in ( "email", "name", "company", "notes", "phone", "job_title", "address", "website", "birthday", "emails_json", "phones_json", "addresses_json", ) ) target.execute( f"UPDATE {table} SET {assigns},source='roundcube',updated_at={now_expr(target.backend)} WHERE id={ph}", [*values, contact_id], ).close() return columns = ( "id,email,name,company,notes,phone,job_title,address,website,birthday," "emails_json,phones_json,addresses_json,is_favorite,contact_count,source" ) placeholders = ",".join([ph] * 13) target.execute( f"INSERT INTO {table} ({columns}) VALUES ({placeholders},0,0,'roundcube')", [contact.id, *values], ).close() def upsert_identity(target: Db, prefix: str, identity: Identity) -> None: table = target.ident(f"{prefix}identities") ph = target.placeholder if identity.is_default: target.execute(f"UPDATE {table} SET is_default=0 WHERE is_default<>0").close() existing = target.fetchone( f"SELECT id FROM {table} WHERE email={ph} AND display_name={ph} AND organization={ph} ORDER BY id LIMIT 1", [identity.email, identity.display_name, identity.organization], ) if existing is None: existing = target.fetchone( f"SELECT id FROM {table} WHERE email={ph} AND display_name='' AND organization='' " "AND signature_html='' AND signature_text='' ORDER BY id LIMIT 1", [identity.email], ) values = [ identity.display_name, identity.email, identity.signature_html, 1 if identity.is_default else 0, identity.reply_to, identity.bcc, identity.organization, identity.signature_text, ] if existing is not None: identity_id = row_first(existing) assignments = ",".join( f"{column}={ph}" for column in ( "display_name", "email", "signature_html", "is_default", "reply_to", "bcc", "organization", "signature_text", ) ) target.execute( f"UPDATE {table} SET {assignments},updated_at={now_expr(target.backend)} WHERE id={ph}", [*values, identity_id], ).close() return placeholders = ",".join([ph] * 8) target.execute( f"INSERT INTO {table} " "(display_name,email,signature_html,is_default,reply_to,bcc,organization,signature_text," "signature_new,signature_reply,signature_forward,signature_position) " f"VALUES ({placeholders},1,1,1,'above')", values, ).close() def migrate_one( source: Db, target: Db, roundcube_prefix: str, roundcube_user: str, rustbox_user: str, *, dry_run: bool, provision_missing: bool = False, user_hash_override: str | None = None, target_prefix_override: str | None = None, ) -> tuple[int, int, int]: contacts, identities, skipped_contacts = load_roundcube_data( source, roundcube_prefix, roundcube_user ) if target_prefix_override: validate_identifier(target_prefix_override, "RustBox table prefix") prefix = target_prefix_override else: user_hash = user_hash_override or hash_user(rustbox_user) prefix = target_prefix_for_hash(target, user_hash) if prefix is None: match = target_account_for_login(target, rustbox_user) if match is None: if not provision_missing: raise ValueError( f"RustBox account not provisioned for {rustbox_user!r}; " "use --provision-missing or log in once first" ) prefix = tenant_prefix(user_hash) if dry_run: print(f"WOULD PROVISION {rustbox_user}: {prefix}") else: prefix = provision_sqlite_account(target, user_hash) print(f"PROVISION {rustbox_user}: {prefix}") else: _, prefix, _ = match print( f"{roundcube_user} -> {rustbox_user}: {len(contacts)} contact(s), " f"{len(identities)} identity/identities, {skipped_contacts} contact(s) skipped without email" ) if dry_run: return len(contacts), len(identities), skipped_contacts try: for contact in contacts: upsert_contact(target, prefix, contact) for identity in identities: upsert_identity(target, prefix, identity) target.commit() except Exception: target.rollback() raise return len(contacts), len(identities), skipped_contacts def run(args: argparse.Namespace) -> int: validate_identifier(args.roundcube_prefix, "Roundcube table prefix", allow_empty=True) if args.target_prefix: validate_identifier(args.target_prefix, "RustBox table prefix") all_users = args.all_users or not args.roundcube_account if all_users and any([args.rustbox_user, args.user_hash, args.target_prefix]): raise ValueError( "--rustbox-user, --user-hash and --target-prefix are only valid with --roundcube-account" ) config_path, _config, backend, target_url = resolve_target(args) if args.provision_missing and backend != "sqlite": raise ValueError("--provision-missing currently supports SQLite targets only") roundcube_url = build_roundcube_url(args) print(f"RustBox config: {config_path}") print(f"RustBox target backend: {backend}") if backend == "sqlite": print(f"RustBox target database: {target_url}") print(f"Mode: {'all users' if all_users else 'single user'}{' (dry-run)' if args.dry_run else ''}") source = connect_mysql(roundcube_url, dict_rows=True) target = connect_target(backend, target_url) try: # Fail early with a clear error if this is not a RustBox database. target.fetchall("SELECT user_hash, table_prefix FROM rustbox_accounts LIMIT 1") if not all_users: roundcube_user = args.roundcube_account assert roundcube_user is not None rustbox_user = args.rustbox_user or roundcube_user migrate_one( source, target, args.roundcube_prefix, roundcube_user, rustbox_user, dry_run=args.dry_run, provision_missing=args.provision_missing, user_hash_override=args.user_hash, target_prefix_override=args.target_prefix, ) print("Dry run complete." if args.dry_run else "Migration complete.") return 0 users = load_roundcube_users(source, args.roundcube_prefix) print(f"Roundcube: found {len(users)} user(s).") migrated = skipped = failed = 0 total_contacts = total_identities = total_skipped_contacts = 0 for roundcube_user in users: try: match = target_account_for_login(target, roundcube_user) if match is None: if not args.provision_missing: skipped += 1 print( f"SKIP {roundcube_user}: matching RustBox account is not provisioned", file=sys.stderr, ) continue rustbox_user = roundcube_user user_hash = hash_user(rustbox_user) prefix = tenant_prefix(user_hash) if args.dry_run: print(f"WOULD PROVISION {rustbox_user}: {prefix}") else: prefix = provision_sqlite_account(target, user_hash) print(f"PROVISION {rustbox_user}: {prefix}") else: _, prefix, rustbox_user = match counts = migrate_one( source, target, args.roundcube_prefix, roundcube_user, rustbox_user, dry_run=args.dry_run, provision_missing=args.provision_missing, target_prefix_override=prefix, ) migrated += 1 total_contacts += counts[0] total_identities += counts[1] total_skipped_contacts += counts[2] except Exception as exc: target.rollback() failed += 1 print(f"FAIL {roundcube_user}: {exc}", file=sys.stderr) if args.fail_fast: raise label = "Dry run complete" if args.dry_run else "Migration complete" print( f"{label}: {migrated} user(s) processed, {skipped} skipped without RustBox account, " f"{failed} failed; {total_contacts} contact(s), {total_identities} identity/identities, " f"{total_skipped_contacts} contact(s) skipped without email." ) return 2 if failed else 0 finally: target.close() source.close() def main() -> int: args = parser().parse_args() try: return run(args) except KeyboardInterrupt: print("Migration interrupted.", file=sys.stderr) return 130 except Exception as exc: print(f"Roundcube migration failed: {exc}", file=sys.stderr) return 1 if __name__ == "__main__": raise SystemExit(main())