gian/imap-migration-tools
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Refactor authentication handling and improve documentation for IMAP email migration scripts

Browse Source 
- Updated README.md to clarify authentication options, including OAuth2 support.
- Modified backup_imap_emails.py to support OAuth2 credentials alongside traditional password authentication.
- Enhanced compare_imap_folders.py to conditionally require IMAP credentials based on local path usage.
- Updated count_imap_emails.py to improve argument parsing and error handling for missing credentials.
- Refactored migrate_imap_emails.py to streamline authentication logic and improve error messages.
- Enhanced restore_imap_emails.py to support OAuth2 and improve validation for required fields.
- Updated test cases across various modules to reflect changes in error handling and authentication requirements.
- Adjusted test configurations to ensure consistent exit codes for missing credentials.
This commit is contained in:
Javier Callico 2026-01-31 20:28:04 -08:00 committed by Nathan Moinvaziri
parent 06666529ee
commit 7ea0991930
15 changed files with 414 additions and 185 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

34
README.md
View File

@ -19,7 +19,7 @@ This repository contains a set of Python scripts designed to migrate emails betw
1. **`migrate_imap_emails.py`** (The Solution)
- Migrates emails folder-by-folder.
- **Multi-threaded**: Uses a thread pool to copy messages in parallel for high speed.
- **Smart De-duplication**: Checks if a message already exists in the destination (matching Message-ID and Size) and skips it if found.
- **Smart De-duplication**: Checks if a message already exists in the destination (matching Message-ID) and skips it if found.
- **Robust**: Preserves original dates and can preserve IMAP flags with `--preserve-flags`.
- **Gmail Mode**: For Gmail -> Gmail migrations, use `--gmail-mode` to migrate `[Gmail]/All Mail` (no duplicates) and apply Gmail labels by copying messages into label folders.
- **Cleanup**: Optionally deletes messages from the source after successful transfer (effectively a "Move" operation).
@ -49,7 +49,7 @@ This repository contains a set of Python scripts designed to migrate emails betw
- Uploads emails from a local backup to an IMAP server.
- **Format**: Reads `.eml` files and uploads them preserving original dates.
- **Structure**: Recreates the folder hierarchy on the destination server.
- **Incremental**: Skips emails that already exist (based on Message-ID and size), but still syncs labels and flags.
- **Incremental**: Skips emails that already exist (based on Message-ID), but still syncs labels and flags.
- **Sync Mode**: Optionally deletes emails from destination that no longer exist in local backup (`--dest-delete`).
- **Gmail Labels Restoration**: Applies labels from `labels_manifest.json` to recreate the original Gmail label structure.
@ -120,8 +120,8 @@ You can configure the scripts using **Environment Variables** (recommended for s
# Options (Optional)
export DELETE_FROM_SOURCE="false" # Set to "true" to delete from source after copy
export DEST_DELETE="false" # Set to "true" to delete orphans from destination (sync mode)
export PRESERVE_FLAGS="false" # Set to "true" to preserve IMAP flags (read/starred/etc)
export GMAIL_MODE="false" # Set to "true" for Gmail mode (All Mail + label application)
export PRESERVE_FLAGS="false" # Set to "true" to preserve IMAP flags (read/starred/etc)
export GMAIL_MODE="false" # Set to "true" for Gmail mode (All Mail + label application)
export MAX_WORKERS=4 # Number of parallel threads
export BATCH_SIZE=10 # Emails per batch
```
@ -191,10 +191,14 @@ python3 compare_imap_folders.py \
**Counting:**
```bash
python3 count_imap_emails.py --host "imap.gmail.com" --user "me@gmail.com" --pass "secret"
python3 count_imap_emails.py \
--host "imap.gmail.com" \
--user "me@gmail.com" \
--pass "secret"
# Count a local backup folder
python3 count_imap_emails.py --path "./my_backup"
python3 count_imap_emails.py \
--path "./my_backup"
# Or via environment variable
export BACKUP_LOCAL_PATH="./my_backup"
@ -203,8 +207,18 @@ python3 count_imap_emails.py
**Counting (OAuth2):**
```bash
python3 count_imap_emails.py --host "imap.gmail.com" --user "me@gmail.com" \
--client-id "id" --client-secret "secret"
python3 count_imap_emails.py \
--host "imap.gmail.com" \
--user "me@gmail.com" \
--client-id "id" \
--client-secret "secret"
# Or via environment variables (single-account script)
export IMAP_HOST="imap.gmail.com"
export IMAP_USERNAME="me@gmail.com"
export OAUTH2_CLIENT_ID="id"
export OAUTH2_CLIENT_SECRET="secret" # Required for Google
python3 count_imap_emails.py
```
**Backup:**
@ -595,6 +609,10 @@ All scripts support OAuth2 as an alternative to password-based authentication. T
To use OAuth2, pass `--client-id` (or `--src-client-id`/`--dest-client-id` for dual-account scripts) instead of the password argument.
Environment variable equivalents:
- Dual-account scripts: `SRC_OAUTH2_CLIENT_ID`, `SRC_OAUTH2_CLIENT_SECRET` and `DEST_OAUTH2_CLIENT_ID`, `DEST_OAUTH2_CLIENT_SECRET`.
- Single-account scripts (like `count_imap_emails.py`): `OAUTH2_CLIENT_ID`, `OAUTH2_CLIENT_SECRET` (also accepts `SRC_OAUTH2_CLIENT_ID`, `SRC_OAUTH2_CLIENT_SECRET`).
### Microsoft (Outlook / Office 365)
Requires the `msal` package (`pip install msal`). Uses the **device code flow** — no browser redirect needed. The tenant ID is auto-discovered from the user's email domain.

83
src/backup_imap_emails.py
View File

@ -13,7 +13,13 @@ Features:
- Gmail Labels Preservation: Creates a manifest mapping Message-IDs to Gmail labels for restoration.
Configuration (Environment Variables):
SRC_IMAP_HOST, SRC_IMAP_USERNAME, SRC_IMAP_PASSWORD: Source credentials.
SRC_IMAP_HOST, SRC_IMAP_USERNAME: Source credentials.
SRC_IMAP_PASSWORD: Source password (or App Password).
OAuth2 (Optional - instead of password):
SRC_OAUTH2_CLIENT_ID: OAuth2 Client ID
SRC_OAUTH2_CLIENT_SECRET: OAuth2 Client Secret (required for Google)
BACKUP_LOCAL_PATH: Destination local directory.
MAX_WORKERS: Number of concurrent threads (default: 10).
BATCH_SIZE: Number of emails to process per batch (default: 10).
@ -71,6 +77,12 @@ def safe_print(message):
def get_thread_connection(src_conf):
# Backwards-compatible: tests and some internal call sites may still pass
# (host, user, password) tuples. Normalize to the dict form used by
# imap_common.get_imap_connection_from_conf().
if isinstance(src_conf, tuple) and len(src_conf) == 3:
src_conf = {"host": src_conf[0], "user": src_conf[1], "password": src_conf[2]}
if not hasattr(thread_local, "src") or thread_local.src is None:
thread_local.src = imap_common.get_imap_connection_from_conf(src_conf)
try:
@ -87,6 +99,9 @@ def get_thread_connection(src_conf):
def process_batch(uids, folder_name, src_conf, local_folder_path):
if isinstance(src_conf, tuple) and len(src_conf) == 3:
src_conf = {"host": src_conf[0], "user": src_conf[1], "password": src_conf[2]}
src = get_thread_connection(src_conf)
if not src:
safe_print("Error: Could not establish connection for batch.")
@ -655,18 +670,49 @@ def main():
parser = argparse.ArgumentParser(description="Backup IMAP emails to local .eml files.")
# Source
parser.add_argument("--src-host", default=os.getenv("SRC_IMAP_HOST"), help="Source IMAP Server")
parser.add_argument("--src-user", default=os.getenv("SRC_IMAP_USERNAME"), help="Source Username")
parser.add_argument("--src-pass", default=os.getenv("SRC_IMAP_PASSWORD"), help="Source Password")
default_src_host = os.getenv("SRC_IMAP_HOST")
default_src_user = os.getenv("SRC_IMAP_USERNAME")
default_src_pass = os.getenv("SRC_IMAP_PASSWORD")
default_src_client_id = os.getenv("SRC_OAUTH2_CLIENT_ID")
parser.add_argument(
"--src-host",
default=default_src_host,
required=not bool(default_src_host),
help="Source IMAP Server (or SRC_IMAP_HOST)",
)
parser.add_argument(
"--src-user",
default=default_src_user,
required=not bool(default_src_user),
help="Source Username (or SRC_IMAP_USERNAME)",
)
# Authentication: require either password OR OAuth2 client-id (unless provided via env vars)
auth_required = not bool(default_src_pass or default_src_client_id)
auth_group = parser.add_mutually_exclusive_group(required=auth_required)
auth_group.add_argument("--src-pass", default=default_src_pass, help="Source Password (or SRC_IMAP_PASSWORD)")
auth_group.add_argument(
"--src-client-id",
default=default_src_client_id,
help="OAuth2 Client ID (or SRC_OAUTH2_CLIENT_ID)",
)
# OAuth2
parser.add_argument("--src-client-id", default=os.getenv("SRC_OAUTH2_CLIENT_ID"), help="OAuth2 Client ID")
parser.add_argument("--src-client-secret", default=os.getenv("SRC_OAUTH2_CLIENT_SECRET"),
help="OAuth2 Client Secret (if required)")
parser.add_argument(
"--src-client-secret",
default=os.getenv("SRC_OAUTH2_CLIENT_SECRET"),
help="OAuth2 Client Secret (if required) (or SRC_OAUTH2_CLIENT_SECRET)",
)
# Destination (Local Path)
env_path = os.getenv("BACKUP_LOCAL_PATH")
parser.add_argument("--dest-path", default=env_path, help="Local destination path (Mandatory)")
parser.add_argument(
"--dest-path",
default=env_path,
required=not bool(env_path),
help="Local destination path (or BACKUP_LOCAL_PATH)",
)
# Config
parser.add_argument("--workers", type=int, default=int(os.getenv("MAX_WORKERS", 10)), help="Thread count")
@ -715,24 +761,7 @@ def main():
args = parser.parse_args()
# Validate
missing = []
if not args.src_host:
missing.append("SRC_IMAP_HOST")
if not args.src_user:
missing.append("SRC_IMAP_USERNAME")
use_oauth2 = bool(args.src_client_id)
if not args.src_pass and not use_oauth2:
missing.append("SRC_IMAP_PASSWORD or OAuth2 credentials")
if missing:
print(f"Error: Missing credentials: {', '.join(missing)}")
sys.exit(1)
if not args.dest_path:
print("Error: Destination path is required.")
print("Please provide --dest-path or set environment variable BACKUP_LOCAL_PATH.")
sys.exit(1)
global MAX_WORKERS, BATCH_SIZE
MAX_WORKERS = args.workers
@ -766,7 +795,9 @@ def main():
"client_id": args.src_client_id,
"email": args.src_user,
"client_secret": args.src_client_secret,
} if use_oauth2 else None,
}
if use_oauth2
else None,
}
# Expand path (~/...)

120
src/compare_imap_folders.py
View File

@ -12,11 +12,19 @@ Configuration (Environment Variables):
SRC_IMAP_USERNAME : Source Username/Email
SRC_IMAP_PASSWORD : Source Password
OAuth2 (Optional - instead of password):
SRC_OAUTH2_CLIENT_ID : OAuth2 Client ID
SRC_OAUTH2_CLIENT_SECRET : OAuth2 Client Secret (required for Google)
Destination Account:
DEST_IMAP_HOST : Destination IMAP Host
DEST_IMAP_USERNAME : Destination Username/Email
DEST_IMAP_PASSWORD : Destination Password
OAuth2 (Optional - instead of password):
DEST_OAUTH2_CLIENT_ID : OAuth2 Client ID
DEST_OAUTH2_CLIENT_SECRET : OAuth2 Client Secret (required for Google)
Also supports local folders as source and/or destination:
SRC_LOCAL_PATH : Source local folder (backup root)
DEST_LOCAL_PATH : Destination local folder (backup root)
@ -129,40 +137,114 @@ def get_local_email_count(local_root: str, folder_name: str) -> Optional[int]:
def main():
parser = argparse.ArgumentParser(description="Compare email counts between two IMAP accounts.")
default_src_path = os.getenv("SRC_LOCAL_PATH")
default_dest_path = os.getenv("DEST_LOCAL_PATH")
# Phase 1: determine whether each side is local
pre_parser = argparse.ArgumentParser(add_help=False)
pre_parser.add_argument("--src-path", default=default_src_path)
pre_parser.add_argument("--dest-path", default=default_dest_path)
pre_args, _ = pre_parser.parse_known_args()
src_requires_imap = not bool(pre_args.src_path)
dest_requires_imap = not bool(pre_args.dest_path)
# Phase 2: full parser with conditional requirements
parser = argparse.ArgumentParser(description="Compare email counts between two IMAP accounts.")
parser.add_argument(
"--src-path",
default=os.getenv("SRC_LOCAL_PATH"),
default=default_src_path,
help="Source local folder (backup root). If set, IMAP source args are ignored.",
)
parser.add_argument(
"--dest-path",
default=os.getenv("DEST_LOCAL_PATH"),
default=default_dest_path,
help="Destination local folder (backup root). If set, IMAP destination args are ignored.",
)
# Source args
parser.add_argument("--src-host", default=os.getenv("SRC_IMAP_HOST"), help="Source IMAP Server")
parser.add_argument("--src-user", default=os.getenv("SRC_IMAP_USERNAME"), help="Source Username")
parser.add_argument("--src-pass", default=os.getenv("SRC_IMAP_PASSWORD"), help="Source Password")
parser.add_argument("--src-client-id", default=os.getenv("SRC_OAUTH2_CLIENT_ID"), help="Source OAuth2 Client ID")
parser.add_argument("--src-client-secret", default=os.getenv("SRC_OAUTH2_CLIENT_SECRET"),
help="Source OAuth2 Client Secret (if required)")
default_src_host = os.getenv("SRC_IMAP_HOST")
default_src_user = os.getenv("SRC_IMAP_USERNAME")
default_src_pass = os.getenv("SRC_IMAP_PASSWORD")
default_src_client_id = os.getenv("SRC_OAUTH2_CLIENT_ID")
parser.add_argument(
"--src-host",
default=default_src_host,
required=src_requires_imap and not bool(default_src_host),
help="Source IMAP Server (or SRC_IMAP_HOST)",
)
parser.add_argument(
"--src-user",
default=default_src_user,
required=src_requires_imap and not bool(default_src_user),
help="Source Username (or SRC_IMAP_USERNAME)",
)
src_auth_required = src_requires_imap and not bool(default_src_pass or default_src_client_id)
src_auth = parser.add_mutually_exclusive_group(required=src_auth_required)
src_auth.add_argument("--src-pass", default=default_src_pass, help="Source Password (or SRC_IMAP_PASSWORD)")
src_auth.add_argument(
"--src-client-id",
default=default_src_client_id,
help="Source OAuth2 Client ID (or SRC_OAUTH2_CLIENT_ID)",
)
parser.add_argument(
"--src-client-secret",
default=os.getenv("SRC_OAUTH2_CLIENT_SECRET"),
help="Source OAuth2 Client Secret (if required) (or SRC_OAUTH2_CLIENT_SECRET)",
)
# Dest args
parser.add_argument("--dest-host", default=os.getenv("DEST_IMAP_HOST"), help="Destination IMAP Server")
parser.add_argument("--dest-user", default=os.getenv("DEST_IMAP_USERNAME"), help="Destination Username")
parser.add_argument("--dest-pass", default=os.getenv("DEST_IMAP_PASSWORD"), help="Destination Password")
parser.add_argument("--dest-client-id", default=os.getenv("DEST_OAUTH2_CLIENT_ID"), help="Destination OAuth2 Client ID")
parser.add_argument("--dest-client-secret", default=os.getenv("DEST_OAUTH2_CLIENT_SECRET"),
help="Destination OAuth2 Client Secret (if required)")
default_dest_host = os.getenv("DEST_IMAP_HOST")
default_dest_user = os.getenv("DEST_IMAP_USERNAME")
default_dest_pass = os.getenv("DEST_IMAP_PASSWORD")
default_dest_client_id = os.getenv("DEST_OAUTH2_CLIENT_ID")
parser.add_argument(
"--dest-host",
default=default_dest_host,
required=dest_requires_imap and not bool(default_dest_host),
help="Destination IMAP Server (or DEST_IMAP_HOST)",
)
parser.add_argument(
"--dest-user",
default=default_dest_user,
required=dest_requires_imap and not bool(default_dest_user),
help="Destination Username (or DEST_IMAP_USERNAME)",
)
dest_auth_required = dest_requires_imap and not bool(default_dest_pass or default_dest_client_id)
dest_auth = parser.add_mutually_exclusive_group(required=dest_auth_required)
dest_auth.add_argument(
"--dest-pass",
default=default_dest_pass,
help="Destination Password (or DEST_IMAP_PASSWORD)",
)
dest_auth.add_argument(
"--dest-client-id",
default=default_dest_client_id,
help="Destination OAuth2 Client ID (or DEST_OAUTH2_CLIENT_ID)",
)
parser.add_argument(
"--dest-client-secret",
default=os.getenv("DEST_OAUTH2_CLIENT_SECRET"),
help="Destination OAuth2 Client Secret (if required) (or DEST_OAUTH2_CLIENT_SECRET)",
)
args = parser.parse_args()
src_is_local = bool(args.src_path)
dest_is_local = bool(args.dest_path)
SRC_HOST = args.src_host
SRC_USER = args.src_user
SRC_PASS = args.src_pass
DEST_HOST = args.dest_host
DEST_USER = args.dest_user
DEST_PASS = args.dest_pass
src_use_oauth2 = bool(args.src_client_id) and not src_is_local
dest_use_oauth2 = bool(args.dest_client_id) and not dest_is_local
# Acquire OAuth2 tokens if configured
src_oauth2_token = None
src_oauth2_provider = None
@ -202,14 +284,18 @@ def main():
else:
print(f"Source Host : {args.src_host}")
print(f"Source User : {args.src_user}")
print(f"Source Auth : {'OAuth2/' + src_oauth2_provider + ' (XOAUTH2)' if src_use_oauth2 else 'Basic (password)'}")
print(
f"Source Auth : {'OAuth2/' + src_oauth2_provider + ' (XOAUTH2)' if src_use_oauth2 else 'Basic (password)'}"
)
if dest_is_local:
print(f"Destination (Local): {args.dest_path}")
else:
print(f"Destination Host: {args.dest_host}")
print(f"Destination User: {args.dest_user}")
print(f"Destination Auth: {'OAuth2/' + dest_oauth2_provider + ' (XOAUTH2)' if dest_use_oauth2 else 'Basic (password)'}")
print(
f"Destination Auth: {'OAuth2/' + dest_oauth2_provider + ' (XOAUTH2)' if dest_use_oauth2 else 'Basic (password)'}"
)
print("-----------------------------\n")
src = None

69
src/count_imap_emails.py
View File

@ -9,6 +9,12 @@ Configuration (Environment Variables):
IMAP_USERNAME : Username/Email
IMAP_PASSWORD : Password (or App Password)
OAuth2 (Optional - instead of password):
OAUTH2_CLIENT_ID : OAuth2 Client ID
OAUTH2_CLIENT_SECRET : OAuth2 Client Secret (required for Google)
SRC_OAUTH2_CLIENT_ID : Alternate OAuth2 client ID env var
SRC_OAUTH2_CLIENT_SECRET: Alternate OAuth2 client secret env var
Local backup counting:
BACKUP_LOCAL_PATH : Local backup root (preferred)
SRC_LOCAL_PATH : Alternate local backup root
@ -20,6 +26,13 @@ Examples:
export IMAP_PASSWORD="secretpassword"
python3 count_imap_emails.py
# Count an IMAP account using OAuth2
export IMAP_HOST="imap.gmail.com"
export IMAP_USERNAME="user@gmail.com"
export OAUTH2_CLIENT_ID="your-client-id"
export OAUTH2_CLIENT_SECRET="your-client-secret" # Required for Google
python3 count_imap_emails.py
# Count a local backup
python3 count_imap_emails.py --path "./my_backup"
@ -166,10 +179,17 @@ def count_local_emails(local_path: str) -> None:
print(f"{'TOTAL':<40} {total_all_folders:>10}")
if __name__ == "__main__":
def main(argv: list[str] | None = None) -> None:
# Phase 1: determine whether we're in local mode (--path)
default_path = os.getenv("BACKUP_LOCAL_PATH") or os.getenv("SRC_LOCAL_PATH")
pre_parser = argparse.ArgumentParser(add_help=False)
pre_parser.add_argument("--path", default=default_path)
pre_args, _ = pre_parser.parse_known_args(argv)
require_imap = not bool(pre_args.path)
# Phase 2: full parser with conditional requirements
parser = argparse.ArgumentParser(description="Count emails in IMAP account.")
default_path = os.getenv("BACKUP_LOCAL_PATH") or os.getenv("SRC_LOCAL_PATH")
parser.add_argument(
"--path",
default=default_path,
@ -180,17 +200,39 @@ if __name__ == "__main__":
default_host = os.getenv("IMAP_HOST") or os.getenv("SRC_IMAP_HOST")
default_user = os.getenv("IMAP_USERNAME") or os.getenv("SRC_IMAP_USERNAME")
default_pass = os.getenv("IMAP_PASSWORD") or os.getenv("SRC_IMAP_PASSWORD")
default_client_id = os.getenv("OAUTH2_CLIENT_ID") or os.getenv("SRC_OAUTH2_CLIENT_ID")
parser.add_argument("--host", default=default_host, help="IMAP Server")
parser.add_argument("--user", default=default_user, help="Username")
parser.add_argument("--pass", dest="password", default=default_pass, help="Password")
parser.add_argument(
"--host",
default=default_host,
required=require_imap and not bool(default_host),
help="IMAP Server (or IMAP_HOST / SRC_IMAP_HOST)",
)
parser.add_argument(
"--user",
default=default_user,
required=require_imap and not bool(default_user),
help="Username (or IMAP_USERNAME / SRC_IMAP_USERNAME)",
)
# OAuth2
parser.add_argument("--client-id", default=os.getenv("OAUTH2_CLIENT_ID") or os.getenv("SRC_OAUTH2_CLIENT_ID"), help="OAuth2 Client ID")
parser.add_argument("--client-secret", default=os.getenv("OAUTH2_CLIENT_SECRET") or os.getenv("SRC_OAUTH2_CLIENT_SECRET"),
help="OAuth2 Client Secret (if required)")
auth_required = require_imap and not bool(default_pass or default_client_id)
auth_group = parser.add_mutually_exclusive_group(required=auth_required)
auth_group.add_argument(
"--pass", dest="password", default=default_pass, help="Password (or IMAP_PASSWORD / SRC_IMAP_PASSWORD)"
)
auth_group.add_argument(
"--client-id",
default=default_client_id,
help="OAuth2 Client ID (or OAUTH2_CLIENT_ID / SRC_OAUTH2_CLIENT_ID)",
)
args = parser.parse_args()
parser.add_argument(
"--client-secret",
default=os.getenv("OAUTH2_CLIENT_SECRET") or os.getenv("SRC_OAUTH2_CLIENT_SECRET"),
help="OAuth2 Client Secret (if required) (or OAUTH2_CLIENT_SECRET / SRC_OAUTH2_CLIENT_SECRET)",
)
args = parser.parse_args(argv)
if args.path:
if not os.path.isdir(args.path):
@ -208,9 +250,6 @@ if __name__ == "__main__":
PASSWORD = args.password
use_oauth2 = bool(args.client_id)
if not IMAP_SERVER or not USERNAME or (not PASSWORD and not use_oauth2):
print("Error: Missing credentials.")
sys.exit(1)
# Acquire OAuth2 token if configured
oauth2_token = None
@ -236,3 +275,7 @@ if __name__ == "__main__":
print("-----------------------------\n")
count_emails(IMAP_SERVER, USERNAME, PASSWORD, oauth2_token)
if __name__ == "__main__":
main()

4
src/imap_common.py
View File

@ -83,9 +83,7 @@ def get_imap_connection_from_conf(conf):
"oauth2": dict or None # Contains provider, client_id, email, client_secret
}
"""
return get_imap_connection(
conf["host"], conf["user"], conf.get("password"), conf.get("oauth2_token")
)
return get_imap_connection(conf["host"], conf["user"], conf.get("password"), conf.get("oauth2_token"))
def get_imap_connection(host, user, password=None, oauth2_token=None):

15
src/imap_oauth2.py
View File

@ -4,13 +4,18 @@ IMAP OAuth2 Authentication
OAuth2 token acquisition and refresh for Microsoft and Google IMAP providers.
Supports device code flow (Microsoft) and installed app flow (Google),
with in-memory caching for silent token refresh.
Notes:
- Microsoft OAuth2 requires the `msal` package and uses device code flow.
- Google OAuth2 requires the `google-auth-oauthlib` package and uses an installed-app flow
(opens a browser and listens on a local HTTP redirect).
- Provider is auto-detected from the IMAP host string.
"""
import re
import sys
import threading
# Module-level caches for OAuth2 token refresh
_msal_app_cache = {} # (client_id, tenant_id) -> PublicClientApplication
_google_creds_cache = {} # (client_id, client_secret) -> credentials
@ -68,16 +73,16 @@ def acquire_microsoft_oauth2_token(client_id, email):
On subsequent calls, silently refreshes the token using the cached MSAL app
(which holds the refresh token in its in-memory cache).
"""
tenant_id = discover_microsoft_tenant(email)
if not tenant_id:
return None
try:
import msal
except ImportError:
print("Error: 'msal' package is required for Microsoft OAuth2. Install it with: pip install msal")
sys.exit(1)
tenant_id = discover_microsoft_tenant(email)
if not tenant_id:
return None
authority = f"https://login.microsoftonline.com/{tenant_id}"
scopes = ["https://outlook.office365.com/IMAP.AccessAsUser.All"]

120
src/migrate_imap_emails.py
View File

@ -23,11 +23,19 @@ Configuration (Environment Variables):
SRC_IMAP_USERNAME : Source Username/Email
SRC_IMAP_PASSWORD : Source Password (or App Password)
OAuth2 (Optional - instead of password):
SRC_OAUTH2_CLIENT_ID : OAuth2 Client ID
SRC_OAUTH2_CLIENT_SECRET : OAuth2 Client Secret (required for Google)
Destination Account:
DEST_IMAP_HOST : Destination IMAP Host
DEST_IMAP_USERNAME : Destination Username/Email
DEST_IMAP_PASSWORD : Destination Password
OAuth2 (Optional - instead of password):
DEST_OAUTH2_CLIENT_ID : OAuth2 Client ID
DEST_OAUTH2_CLIENT_SECRET : OAuth2 Client Secret (required for Google)
Options:
DELETE_FROM_SOURCE : Set to "true" to delete emails from source after successful transfer.
Default is "false" (Copy only).
@ -675,20 +683,72 @@ def main():
parser.add_argument("folder", nargs="?", help="Specific folder to migrate (e.g. '[Gmail]/Important')")
# Source args
parser.add_argument("--src-host", default=os.getenv("SRC_IMAP_HOST"), help="Source IMAP Host")
parser.add_argument("--src-user", default=os.getenv("SRC_IMAP_USERNAME"), help="Source Username")
parser.add_argument("--src-pass", default=os.getenv("SRC_IMAP_PASSWORD"), help="Source Password")
parser.add_argument("--src-client-id", default=os.getenv("SRC_OAUTH2_CLIENT_ID"), help="Source OAuth2 Client ID")
parser.add_argument("--src-client-secret", default=os.getenv("SRC_OAUTH2_CLIENT_SECRET"),
help="Source OAuth2 Client Secret (if required)")
default_src_host = os.getenv("SRC_IMAP_HOST")
default_src_user = os.getenv("SRC_IMAP_USERNAME")
default_src_pass = os.getenv("SRC_IMAP_PASSWORD")
default_src_client_id = os.getenv("SRC_OAUTH2_CLIENT_ID")
parser.add_argument(
"--src-host",
default=default_src_host,
required=not bool(default_src_host),
help="Source IMAP Host (or SRC_IMAP_HOST)",
)
parser.add_argument(
"--src-user",
default=default_src_user,
required=not bool(default_src_user),
help="Source Username (or SRC_IMAP_USERNAME)",
)
src_auth_required = not bool(default_src_pass or default_src_client_id)
src_auth = parser.add_mutually_exclusive_group(required=src_auth_required)
src_auth.add_argument("--src-pass", default=default_src_pass, help="Source Password (or SRC_IMAP_PASSWORD)")
src_auth.add_argument(
"--src-client-id",
default=default_src_client_id,
help="Source OAuth2 Client ID (or SRC_OAUTH2_CLIENT_ID)",
)
parser.add_argument(
"--src-client-secret",
default=os.getenv("SRC_OAUTH2_CLIENT_SECRET"),
help="Source OAuth2 Client Secret (if required) (or SRC_OAUTH2_CLIENT_SECRET)",
)
# Dest args
parser.add_argument("--dest-host", default=os.getenv("DEST_IMAP_HOST"), help="Destination IMAP Host")
parser.add_argument("--dest-user", default=os.getenv("DEST_IMAP_USERNAME"), help="Destination Username")
parser.add_argument("--dest-pass", default=os.getenv("DEST_IMAP_PASSWORD"), help="Destination Password")
parser.add_argument("--dest-client-id", default=os.getenv("DEST_OAUTH2_CLIENT_ID"), help="Destination OAuth2 Client ID")
parser.add_argument("--dest-client-secret", default=os.getenv("DEST_OAUTH2_CLIENT_SECRET"),
help="Destination OAuth2 Client Secret (if required)")
default_dest_host = os.getenv("DEST_IMAP_HOST")
default_dest_user = os.getenv("DEST_IMAP_USERNAME")
default_dest_pass = os.getenv("DEST_IMAP_PASSWORD")
default_dest_client_id = os.getenv("DEST_OAUTH2_CLIENT_ID")
parser.add_argument(
"--dest-host",
default=default_dest_host,
required=not bool(default_dest_host),
help="Destination IMAP Host (or DEST_IMAP_HOST)",
)
parser.add_argument(
"--dest-user",
default=default_dest_user,
required=not bool(default_dest_user),
help="Destination Username (or DEST_IMAP_USERNAME)",
)
dest_auth_required = not bool(default_dest_pass or default_dest_client_id)
dest_auth = parser.add_mutually_exclusive_group(required=dest_auth_required)
dest_auth.add_argument(
"--dest-pass",
default=default_dest_pass,
help="Destination Password (or DEST_IMAP_PASSWORD)",
)
dest_auth.add_argument(
"--dest-client-id",
default=default_dest_client_id,
help="Destination OAuth2 Client ID (or DEST_OAUTH2_CLIENT_ID)",
)
parser.add_argument(
"--dest-client-secret",
default=os.getenv("DEST_OAUTH2_CLIENT_SECRET"),
help="Destination OAuth2 Client Secret (if required) (or DEST_OAUTH2_CLIENT_SECRET)",
)
# Options
# Check env var for boolean default (msg "true" -> True)
@ -768,28 +828,8 @@ def main():
MAX_WORKERS = args.workers
BATCH_SIZE = args.batch
# Validation
# Validation
missing_vars = []
if not SRC_HOST:
missing_vars.append("SRC_IMAP_HOST")
if not SRC_USER:
missing_vars.append("SRC_IMAP_USERNAME")
src_use_oauth2 = bool(args.src_client_id)
if not SRC_PASS and not src_use_oauth2:
missing_vars.append("SRC_IMAP_PASSWORD")
if not DEST_HOST:
missing_vars.append("DEST_IMAP_HOST")
if not DEST_USER:
missing_vars.append("DEST_IMAP_USERNAME")
dest_use_oauth2 = bool(args.dest_client_id)
if not DEST_PASS and not dest_use_oauth2:
missing_vars.append("DEST_IMAP_PASSWORD")
if missing_vars:
print(f"Error: Missing configuration variables: {', '.join(missing_vars)}")
print("Please provide them via environment variables or command-line arguments.")
sys.exit(1)
# Acquire OAuth2 tokens if configured
src_oauth2_token = None
@ -827,10 +867,14 @@ def main():
print("\n--- Configuration Summary ---")
print(f"Source Host : {SRC_HOST}")
print(f"Source User : {SRC_USER}")
print(f"Source Auth : {'OAuth2/' + src_oauth2_provider + ' (XOAUTH2)' if src_use_oauth2 else 'Basic (password)'}")
print(
f"Source Auth : {'OAuth2/' + src_oauth2_provider + ' (XOAUTH2)' if src_use_oauth2 else 'Basic (password)'}"
)
print(f"Destination Host: {DEST_HOST}")
print(f"Destination User: {DEST_USER}")
print(f"Dest Auth : {'OAuth2/' + dest_oauth2_provider + ' (XOAUTH2)' if dest_use_oauth2 else 'Basic (password)'}")
print(
f"Destination Auth: {'OAuth2/' + dest_oauth2_provider + ' (XOAUTH2)' if dest_use_oauth2 else 'Basic (password)'}"
)
print(f"Delete fm Source: {DELETE_SOURCE}")
print(f"Dest Delete : {DEST_DELETE}")
print(f"Preserve Flags : {preserve_flags}")
@ -850,7 +894,9 @@ def main():
"client_id": args.src_client_id,
"email": SRC_USER,
"client_secret": args.src_client_secret,
} if src_use_oauth2 else None,
}
if src_use_oauth2
else None,
}
dest_conf = {
"host": DEST_HOST,
@ -862,7 +908,9 @@ def main():
"client_id": args.dest_client_id,
"email": DEST_USER,
"client_secret": args.dest_client_secret,
} if dest_use_oauth2 else None,
}
if dest_use_oauth2
else None,
}
try:

74
src/restore_imap_emails.py
View File

@ -7,12 +7,18 @@ Reads .eml files from a local directory and uploads them to the destination serv
Features:
- Folder Restoration: Recreates the folder structure from the backup.
- Gmail Labels Restoration: Uses labels_manifest.json to apply Gmail labels.
- Incremental Restore: Skips emails that already exist (based on Message-ID and size).
- Incremental Restore: Skips emails that already exist (based on Message-ID).
- Parallel Processing: Uses multithreading for fast uploads.
- Date Preservation: Restores emails with their original dates.
Configuration (Environment Variables):
DEST_IMAP_HOST, DEST_IMAP_USERNAME, DEST_IMAP_PASSWORD: Destination credentials.
DEST_IMAP_HOST, DEST_IMAP_USERNAME: Destination credentials.
DEST_IMAP_PASSWORD: Destination password (or App Password).
OAuth2 (Optional - instead of password):
DEST_OAUTH2_CLIENT_ID: OAuth2 Client ID
DEST_OAUTH2_CLIENT_SECRET: OAuth2 Client Secret (required for Google)
BACKUP_LOCAL_PATH: Source local directory containing the backup.
MAX_WORKERS: Number of concurrent threads (default: 4).
BATCH_SIZE: Number of emails to process per batch (default: 10).
@ -695,33 +701,48 @@ def main():
# Source (Local Path)
env_path = os.getenv("BACKUP_LOCAL_PATH")
parser.add_argument("--src-path", default=env_path, help="Local source path containing backup")
parser.add_argument(
"--src-path",
default=env_path,
required=not bool(env_path),
help="Local source path containing backup (or BACKUP_LOCAL_PATH)",
)
# Destination
default_dest_host = os.getenv("DEST_IMAP_HOST")
default_dest_user = os.getenv("DEST_IMAP_USERNAME")
default_dest_pass = os.getenv("DEST_IMAP_PASSWORD")
default_dest_client_id = os.getenv("DEST_OAUTH2_CLIENT_ID")
parser.add_argument(
"--dest-host",
default=os.getenv("DEST_IMAP_HOST"),
help="Destination IMAP Server",
default=default_dest_host,
required=not bool(default_dest_host),
help="Destination IMAP Server (or DEST_IMAP_HOST)",
)
parser.add_argument(
"--dest-user",
default=os.getenv("DEST_IMAP_USERNAME"),
help="Destination Username",
default=default_dest_user,
required=not bool(default_dest_user),
help="Destination Username (or DEST_IMAP_USERNAME)",
)
parser.add_argument(
dest_auth_required = not bool(default_dest_pass or default_dest_client_id)
dest_auth = parser.add_mutually_exclusive_group(required=dest_auth_required)
dest_auth.add_argument(
"--dest-pass",
default=os.getenv("DEST_IMAP_PASSWORD"),
help="Destination Password",
default=default_dest_pass,
help="Destination Password (or DEST_IMAP_PASSWORD)",
)
parser.add_argument(
dest_auth.add_argument(
"--dest-client-id",
default=os.getenv("DEST_OAUTH2_CLIENT_ID"),
help="Destination OAuth2 Client ID",
default=default_dest_client_id,
help="Destination OAuth2 Client ID (or DEST_OAUTH2_CLIENT_ID)",
)
parser.add_argument(
"--dest-client-secret",
default=os.getenv("DEST_OAUTH2_CLIENT_SECRET"),
help="Destination OAuth2 Client Secret (if required)",
help="Destination OAuth2 Client Secret (if required) (or DEST_OAUTH2_CLIENT_SECRET)",
)
# Config
@ -775,24 +796,7 @@ def main():
args = parser.parse_args()
# Validate
dest_use_oauth2 = bool(args.dest_client_id)
missing = []
if not args.dest_host:
missing.append("DEST_IMAP_HOST")
if not args.dest_user:
missing.append("DEST_IMAP_USERNAME")
if not args.dest_pass and not dest_use_oauth2:
missing.append("DEST_IMAP_PASSWORD (or --dest-client-id for OAuth2)")
if missing:
print(f"Error: Missing credentials: {', '.join(missing)}")
sys.exit(1)
if not args.src_path:
print("Error: Source path is required.")
print("Please provide --src-path or set environment variable BACKUP_LOCAL_PATH.")
sys.exit(1)
global MAX_WORKERS, BATCH_SIZE
MAX_WORKERS = args.workers
@ -826,7 +830,9 @@ def main():
"client_id": args.dest_client_id,
"email": args.dest_user,
"client_secret": args.dest_client_secret,
} if dest_use_oauth2 else None,
}
if dest_use_oauth2
else None,
}
# Expand path
@ -855,7 +861,9 @@ def main():
print(f"Source Path : {local_path}")
print(f"Destination Host: {args.dest_host}")
print(f"Destination User: {args.dest_user}")
print(f"Destination Auth: {'OAuth2/' + dest_oauth2_provider + ' (XOAUTH2)' if dest_use_oauth2 else 'Basic (password)'}")
print(
f"Destination Auth: {'OAuth2/' + dest_oauth2_provider + ' (XOAUTH2)' if dest_use_oauth2 else 'Basic (password)'}"
)
print(f"Workers : {args.workers}")
if args.gmail_mode:
print("Mode : Gmail Restore with Labels + Flags")

4
test/test_backup_imap_emails.py
View File

@ -201,7 +201,7 @@ class TestConfigValidation:
with pytest.raises(SystemExit) as exc_info:
backup_imap_emails.main()
assert exc_info.value.code == 1
assert exc_info.value.code == 2
def test_missing_dest_path(self, monkeypatch, capsys):
"""Test that missing destination path causes exit."""
@ -216,7 +216,7 @@ class TestConfigValidation:
with pytest.raises(SystemExit) as exc_info:
backup_imap_emails.main()
assert exc_info.value.code == 1
assert exc_info.value.code == 2
class TestGetExistingUids:

8
test/test_compare_imap_folders.py
View File

@ -227,7 +227,9 @@ class TestCompareFoldersErrorHandling:
mock_src.list.return_value = ("NO", [])
mock_dest = MagicMock()
monkeypatch.setattr("imap_common.get_imap_connection", lambda h, u, p, oauth2_token=None: mock_src if u == "s" else mock_dest)
monkeypatch.setattr(
"imap_common.get_imap_connection", lambda h, u, p, oauth2_token=None: mock_src if u == "s" else mock_dest
)
env = {
"SRC_IMAP_HOST": "h",
@ -294,7 +296,7 @@ class TestConfigValidation:
with pytest.raises(SystemExit) as exc_info:
compare_imap_folders.main()
assert exc_info.value.code == 1
assert exc_info.value.code == 2
def test_missing_dest_credentials(self, monkeypatch, capsys):
"""Test that missing destination credentials cause exit."""
@ -309,7 +311,7 @@ class TestConfigValidation:
with pytest.raises(SystemExit) as exc_info:
compare_imap_folders.main()
assert exc_info.value.code == 1
assert exc_info.value.code == 2
class TestTotals:

23
test/test_count_imap_emails.py
View File

@ -220,25 +220,14 @@ class TestMainFunction:
assert "INBOX" in captured.out
def test_missing_credentials(self, monkeypatch, capsys):
"""Test that missing credentials cause exit."""
env = {}
monkeypatch.setattr(os, "environ", env)
monkeypatch.setattr(sys, "argv", ["count_imap_emails.py"])
"""Test that missing auth is rejected by argparse (neither password nor OAuth2 client-id)."""
monkeypatch.setattr(os, "environ", {})
monkeypatch.setattr(sys, "argv", ["count_imap_emails.py", "--host", "localhost", "--user", "user"])
# Since the script uses if __name__ == "__main__", we need to test differently
# Test the validation path directly
with pytest.raises(SystemExit):
# Simulate running main
import argparse
with pytest.raises(SystemExit) as exc_info:
count_imap_emails.main()
parser = argparse.ArgumentParser()
parser.add_argument("--host", default=None)
parser.add_argument("--user", default=None)
parser.add_argument("--pass", dest="password", default=None)
args = parser.parse_args([])
if not all([args.host, args.user, args.password]):
sys.exit(1)
assert exc_info.value.code == 2
class TestSrcImapFallback:

2
test/test_imap_common.py
View File

@ -16,8 +16,6 @@ import os
import sys
from unittest.mock import Mock
import pytest
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../src")))
import imap_common

35
test/test_imap_oauth2.py
View File

@ -71,10 +71,12 @@ class TestDiscoverMicrosoftTenant:
def test_successful_discovery(self):
"""Test successful tenant ID extraction from OpenID config."""
tenant_id = "12345678-abcd-ef01-2345-67890abcdef0"
openid_response = json.dumps({
"issuer": f"https://sts.windows.net/{tenant_id}/",
"authorization_endpoint": f"https://login.microsoftonline.com/{tenant_id}/oauth2/v2.0/authorize",
}).encode("utf-8")
openid_response = json.dumps(
{
"issuer": f"https://sts.windows.net/{tenant_id}/",
"authorization_endpoint": f"https://login.microsoftonline.com/{tenant_id}/oauth2/v2.0/authorize",
}
).encode("utf-8")
mock_response = MagicMock()
mock_response.read.return_value = openid_response
@ -89,9 +91,9 @@ class TestDiscoverMicrosoftTenant:
def test_domain_extraction(self):
"""Test that domain is correctly extracted from email."""
mock_response = MagicMock()
mock_response.read.return_value = json.dumps({
"issuer": "https://sts.windows.net/abcdef01-2345-6789-abcd-ef0123456789/"
}).encode("utf-8")
mock_response.read.return_value = json.dumps(
{"issuer": "https://sts.windows.net/abcdef01-2345-6789-abcd-ef0123456789/"}
).encode("utf-8")
mock_response.__enter__ = lambda s: s
mock_response.__exit__ = MagicMock(return_value=False)
@ -126,9 +128,7 @@ class TestDiscoverMicrosoftTenant:
def test_no_tenant_in_issuer(self, capsys):
"""Test returns None when issuer has no tenant GUID."""
mock_response = MagicMock()
mock_response.read.return_value = json.dumps({
"issuer": "https://sts.windows.net/not-a-guid/"
}).encode("utf-8")
mock_response.read.return_value = json.dumps({"issuer": "https://sts.windows.net/not-a-guid/"}).encode("utf-8")
mock_response.__enter__ = lambda s: s
mock_response.__exit__ = MagicMock(return_value=False)
@ -339,12 +339,15 @@ class TestGoogleTokenRefresh:
imap_oauth2._google_creds_cache[("client-id", "client-secret")] = mock_creds
mock_request_module = MagicMock()
with patch.dict("sys.modules", {
"google": MagicMock(),
"google.auth": MagicMock(),
"google.auth.transport": MagicMock(),
"google.auth.transport.requests": mock_request_module,
}):
with patch.dict(
"sys.modules",
{
"google": MagicMock(),
"google.auth": MagicMock(),
"google.auth.transport": MagicMock(),
"google.auth.transport.requests": mock_request_module,
},
):
result = imap_oauth2.acquire_google_oauth2_token("client-id", "client-secret")
assert result == "refreshed_google_token"

4
test/test_migrate_imap_emails.py
View File

@ -372,7 +372,7 @@ class TestConfigValidation:
with pytest.raises(SystemExit) as exc_info:
migrate_imap_emails.main()
assert exc_info.value.code == 1
assert exc_info.value.code == 2
def test_missing_dest_credentials(self, monkeypatch, capsys):
"""Test that missing destination credentials cause exit."""
@ -386,7 +386,7 @@ class TestConfigValidation:
with pytest.raises(SystemExit) as exc_info:
migrate_imap_emails.main()
assert exc_info.value.code == 1
assert exc_info.value.code == 2
class TestMigrateErrorHandling:

4
test/test_restore_imap_emails.py
View File

@ -226,7 +226,7 @@ class TestConfigValidation:
with pytest.raises(SystemExit) as exc_info:
restore_imap_emails.main()
assert exc_info.value.code == 1
assert exc_info.value.code == 2
def test_missing_src_path(self, monkeypatch, capsys):
"""Test that missing source path causes exit."""
@ -241,7 +241,7 @@ class TestConfigValidation:
with pytest.raises(SystemExit) as exc_info:
restore_imap_emails.main()
assert exc_info.value.code == 1
assert exc_info.value.code == 2
def test_nonexistent_src_path(self, monkeypatch, capsys):
"""Test that non-existent source path causes exit."""