PYTHON migrate_encryption.py

#!/usr/bin/env python3
"""One-time migration: encrypt existing plaintext config_json values.

Run this once after deploying the encryption feature to migrate existing
connector records from plaintext `config_json` to Fernet-encrypted
`config_encrypted`.

Usage:
    # From the project root:
    cd /path/to/command-sovereignty
    source venv/bin/activate

    # Set the encryption key first (IMPORTANT — use the SAME key as production)
    export CONNECTOR_ENCRYPTION_KEY='your-44-char-fernet-key-here'

    # Run migration
    python -m app.utils.migrate_encryption

    # Or run directly:
    python app/utils/migrate_encryption.py
"""

from __future__ import annotations

import sys
import logging

logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s %(message)s')
logger = logging.getLogger(__name__)


def main() -> None:
    from app import create_app
    from app.models import db, Connector
    from app.utils.encryption import encrypt_config

    app = create_app()

    with app.app_context():
        # Find all connectors that still have plaintext config
        connectors = Connector.query.filter(Connector.config_json.isnot(None)).all()
        logger.info("Found %d connector(s) with plaintext config to migrate", len(connectors))

        migrated = 0
        errors = 0

        for c in connectors:
            try:
                if c.config_json:
                    c.config_encrypted = encrypt_config(c.config_json)
                    c.config_json = None  # Clear plaintext after successful encryption
                    migrated += 1
                    logger.info("  Encrypted connector %s (%s)", c.id[:8], c.service)
                else:
                    logger.warning("  Skipping connector %s — config_json is None", c.id[:8])
            except Exception:
                errors += 1
                logger.exception("  Failed to encrypt connector %s (%s)", c.id[:8], c.service)

        if errors == 0:
            db.session.commit()
            logger.info("Migration complete: %d connector(s) encrypted successfully", migrated)
        else:
            db.session.rollback()
            logger.error(
                "Migration aborted: %d error(s) occurred. No changes committed.",
                errors,
            )
            sys.exit(1)


if __name__ == "__main__":
    main()