Merge pull request #550 from christianlouis/copilot/add-database-models-for-multi-tenant

feat(models): Generic UserIntegration model for multi-tenant sources/destinations + encrypt IMAP passwords

app/api/__init__.py

@@ -17,6 +17,7 @@ from app.api.duplicates import router as duplicates_router
 from app.api.files import router as files_router
 from app.api.google_drive import router as google_drive_router
 from app.api.imap_accounts import router as imap_accounts_router
+from app.api.integrations import router as integrations_router
 from app.api.logs import router as logs_router
 from app.api.onboarding import router as onboarding_router
 from app.api.onedrive import router as onedrive_router
@@ -70,3 +71,4 @@ router.include_router(onboarding_router)
 router.include_router(billing_router)
 router.include_router(pipelines_router)
 router.include_router(imap_accounts_router)
+router.include_router(integrations_router)

app/api/imap_accounts.py

@@ -16,6 +16,7 @@ from sqlalchemy.orm import Session
 
 from app.database import get_db
 from app.models import UserImapAccount
+from app.utils.encryption import decrypt_value, encrypt_value
 from app.utils.subscription import get_tier, get_user_tier_id
 from app.utils.user_scope import get_current_owner_id
 
@@ -217,7 +218,7 @@ def create_imap_account(
         host=body.host,
         port=body.port,
         username=body.username,
-        password=body.password,
+        password=encrypt_value(body.password),
         use_ssl=body.use_ssl,
         delete_after_process=body.delete_after_process,
         is_active=body.is_active,
@@ -269,7 +270,7 @@ def update_imap_account(
     if body.username is not None:
         acct.username = body.username
     if body.password is not None:
-        acct.password = body.password
+        acct.password = encrypt_value(body.password)
     if body.use_ssl is not None:
         acct.use_ssl = body.use_ssl
     if body.delete_after_process is not None:
@@ -320,7 +321,7 @@ def test_saved_imap_account(account_id: int, request: Request, db: DbSession, ow
     if not acct:
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="IMAP account not found")
 
-    return _test_imap_connection(acct.host, acct.port, acct.username, acct.password, acct.use_ssl)
+    return _test_imap_connection(acct.host, acct.port, acct.username, decrypt_value(acct.password), acct.use_ssl)
 
 
 @router.post("/test", summary="Test an IMAP connection without saving")

app/api/integrations.py

@@ -0,0 +1,344 @@
+"""API endpoints for managing per-user integrations (sources and destinations).
+
+Provides CRUD operations for :class:`~app.models.UserIntegration` records.
+Each record represents one ingestion source (e.g. IMAP, Watch Folder) or
+storage destination (e.g. S3, Dropbox, Google Drive) configured by a user.
+
+Sensitive credentials are encrypted at rest using Fernet symmetric encryption
+(keyed from ``SESSION_SECRET``) via :mod:`app.utils.encryption`.  Credential
+values are **never** returned in API responses.
+"""
+
+import json
+import logging
+from typing import Annotated, Any
+
+from fastapi import APIRouter, Depends, HTTPException, Request, status
+from pydantic import BaseModel, Field
+from sqlalchemy.orm import Session
+
+from app.database import get_db
+from app.models import IntegrationDirection, IntegrationType, UserIntegration
+from app.utils.encryption import decrypt_value, encrypt_value
+from app.utils.user_scope import get_current_owner_id
+
+logger = logging.getLogger(__name__)
+router = APIRouter(prefix="/integrations", tags=["integrations"])
+
+DbSession = Annotated[Session, Depends(get_db)]
+
+# ---------------------------------------------------------------------------
+# Auth helpers
+# ---------------------------------------------------------------------------
+
+
+def _get_owner_id(request: Request) -> str:
+    """Return the current user's owner ID, raising 401 if unauthenticated."""
+    owner_id = get_current_owner_id(request)
+    if owner_id is None:
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Not authenticated")
+    return owner_id
+
+
+CurrentOwner = Annotated[str, Depends(_get_owner_id)]
+
+# ---------------------------------------------------------------------------
+# Pydantic schemas
+# ---------------------------------------------------------------------------
+
+_VALID_DIRECTIONS = IntegrationDirection.ALL
+_VALID_TYPES = IntegrationType.ALL
+
+
+class IntegrationCreate(BaseModel):
+    """Schema for creating a new integration."""
+
+    direction: str = Field(..., description="'SOURCE' or 'DESTINATION'")
+    integration_type: str = Field(..., description="Integration type (e.g. 'IMAP', 'S3', 'DROPBOX')")
+    name: str = Field(..., min_length=1, max_length=255, description="Human-readable label")
+    config: dict[str, Any] | None = Field(default=None, description="Non-sensitive configuration (JSON object)")
+    credentials: dict[str, Any] | None = Field(
+        default=None, description="Sensitive credentials (JSON object, encrypted at rest)"
+    )
+    is_active: bool = Field(default=True, description="Whether the integration is active")
+
+
+class IntegrationUpdate(BaseModel):
+    """Schema for updating an existing integration (all fields optional)."""
+
+    name: str | None = Field(default=None, min_length=1, max_length=255)
+    config: dict[str, Any] | None = None
+    credentials: dict[str, Any] | None = None
+    is_active: bool | None = None
+
+
+# ---------------------------------------------------------------------------
+# Validation helpers
+# ---------------------------------------------------------------------------
+
+
+def _validate_direction(direction: str) -> None:
+    """Raise 400 if *direction* is not a known value."""
+    if direction not in _VALID_DIRECTIONS:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail=f"Invalid direction '{direction}'. Must be one of: {sorted(_VALID_DIRECTIONS)}",
+        )
+
+
+def _validate_integration_type(integration_type: str) -> None:
+    """Raise 400 if *integration_type* is not a known value."""
+    if integration_type not in _VALID_TYPES:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail=f"Invalid integration_type '{integration_type}'. Must be one of: {sorted(_VALID_TYPES)}",
+        )
+
+
+# ---------------------------------------------------------------------------
+# Serialisation helpers
+# ---------------------------------------------------------------------------
+
+
+def _to_response(integration: UserIntegration) -> dict[str, Any]:
+    """Serialise a :class:`UserIntegration` row to a response dict.
+
+    Credentials are **never** included; only a boolean flag indicating
+    whether credentials have been configured is returned.
+    """
+    config_data: dict[str, Any] | None = None
+    if integration.config:
+        try:
+            config_data = json.loads(integration.config)
+        except (json.JSONDecodeError, TypeError):
+            config_data = None
+
+    return {
+        "id": integration.id,
+        "owner_id": integration.owner_id,
+        "direction": integration.direction,
+        "integration_type": integration.integration_type,
+        "name": integration.name,
+        "config": config_data,
+        "has_credentials": bool(integration.credentials),
+        "is_active": integration.is_active,
+        "last_used_at": integration.last_used_at.isoformat() if integration.last_used_at else None,
+        "last_error": integration.last_error,
+        "created_at": integration.created_at.isoformat() if integration.created_at else None,
+        "updated_at": integration.updated_at.isoformat() if integration.updated_at else None,
+    }
+
+
+def _encode_credentials(credentials: dict[str, Any] | None) -> str | None:
+    """Serialise *credentials* dict to an encrypted JSON string for storage."""
+    if not credentials:
+        return None
+    plaintext = json.dumps(credentials)
+    return encrypt_value(plaintext)
+
+
+def _decode_credentials(stored: str | None) -> dict[str, Any] | None:
+    """Decrypt and deserialise stored credentials back to a dict.
+
+    Returns ``None`` when *stored* is empty or cannot be decoded.
+    """
+    if not stored:
+        return None
+    plaintext = decrypt_value(stored)
+    if not plaintext:
+        return None
+    try:
+        return json.loads(plaintext)
+    except (json.JSONDecodeError, TypeError):
+        logger.error("Failed to decode credentials JSON after decryption")
+        return None
+
+
+# ---------------------------------------------------------------------------
+# Endpoints
+# ---------------------------------------------------------------------------
+
+
+@router.get("/", summary="List integrations for the current user")
+def list_integrations(
+    request: Request,
+    db: DbSession,
+    owner_id: CurrentOwner,
+    direction: str | None = None,
+    integration_type: str | None = None,
+) -> list[dict[str, Any]]:
+    """Return all integrations belonging to the authenticated user.
+
+    Optional query-string filters:
+
+    - ``direction`` — ``SOURCE`` or ``DESTINATION``
+    - ``integration_type`` — e.g. ``IMAP``, ``S3``, ``DROPBOX``
+    """
+    query = db.query(UserIntegration).filter(UserIntegration.owner_id == owner_id)
+
+    if direction is not None:
+        _validate_direction(direction)
+        query = query.filter(UserIntegration.direction == direction)
+
+    if integration_type is not None:
+        _validate_integration_type(integration_type)
+        query = query.filter(UserIntegration.integration_type == integration_type)
+
+    integrations = query.order_by(UserIntegration.id).all()
+    return [_to_response(i) for i in integrations]
+
+
+@router.post("/", status_code=status.HTTP_201_CREATED, summary="Create a new integration")
+def create_integration(
+    request: Request,
+    body: IntegrationCreate,
+    db: DbSession,
+    owner_id: CurrentOwner,
+) -> dict[str, Any]:
+    """Create a new source or destination integration for the current user.
+
+    ``credentials`` are encrypted at rest using Fernet symmetric encryption
+    before being persisted and are **never** returned in API responses.
+    """
+    _validate_direction(body.direction)
+    _validate_integration_type(body.integration_type)
+
+    integration = UserIntegration(
+        owner_id=owner_id,
+        direction=body.direction,
+        integration_type=body.integration_type,
+        name=body.name,
+        config=json.dumps(body.config) if body.config is not None else None,
+        credentials=_encode_credentials(body.credentials),
+        is_active=body.is_active,
+    )
+
+    try:
+        db.add(integration)
+        db.commit()
+        db.refresh(integration)
+    except Exception:
+        db.rollback()
+        raise
+
+    logger.info(
+        "User %s created %s integration %d (%s)",
+        owner_id,
+        body.direction,
+        integration.id,
+        body.integration_type,
+    )
+    return _to_response(integration)
+
+
+@router.get("/{integration_id}", summary="Get a single integration")
+def get_integration(
+    integration_id: int,
+    request: Request,
+    db: DbSession,
+    owner_id: CurrentOwner,
+) -> dict[str, Any]:
+    """Return a single integration by ID (must belong to the current user)."""
+    integration = (
+        db.query(UserIntegration)
+        .filter(UserIntegration.id == integration_id, UserIntegration.owner_id == owner_id)
+        .first()
+    )
+    if not integration:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Integration not found")
+    return _to_response(integration)
+
+
+@router.put("/{integration_id}", summary="Update an integration")
+def update_integration(
+    integration_id: int,
+    request: Request,
+    body: IntegrationUpdate,
+    db: DbSession,
+    owner_id: CurrentOwner,
+) -> dict[str, Any]:
+    """Update an existing integration.  Only provided fields are changed.
+
+    When ``credentials`` is supplied the stored value is replaced in full
+    with the freshly encrypted version of the new credentials dict.
+    """
+    integration = (
+        db.query(UserIntegration)
+        .filter(UserIntegration.id == integration_id, UserIntegration.owner_id == owner_id)
+        .first()
+    )
+    if not integration:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Integration not found")
+
+    if body.name is not None:
+        integration.name = body.name
+    if body.config is not None:
+        integration.config = json.dumps(body.config)
+    if body.credentials is not None:
+        integration.credentials = _encode_credentials(body.credentials)
+    if body.is_active is not None:
+        integration.is_active = body.is_active
+
+    # Reset last_error so the next operation gives a fresh result
+    integration.last_error = None
+
+    try:
+        db.commit()
+        db.refresh(integration)
+    except Exception:
+        db.rollback()
+        raise
+
+    logger.info("User %s updated integration %d", owner_id, integration_id)
+    return _to_response(integration)
+
+
+@router.delete("/{integration_id}", status_code=status.HTTP_204_NO_CONTENT, summary="Delete an integration")
+def delete_integration(
+    integration_id: int,
+    request: Request,
+    db: DbSession,
+    owner_id: CurrentOwner,
+) -> None:
+    """Delete an integration permanently."""
+    integration = (
+        db.query(UserIntegration)
+        .filter(UserIntegration.id == integration_id, UserIntegration.owner_id == owner_id)
+        .first()
+    )
+    if not integration:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Integration not found")
+
+    try:
+        db.delete(integration)
+        db.commit()
+    except Exception:
+        db.rollback()
+        raise
+
+    logger.info("User %s deleted integration %d", owner_id, integration_id)
+
+
+@router.get("/{integration_id}/credentials", summary="Retrieve decrypted credentials for an integration")
+def get_integration_credentials(
+    integration_id: int,
+    request: Request,
+    db: DbSession,
+    owner_id: CurrentOwner,
+) -> dict[str, Any]:
+    """Return the decrypted credentials dict for a saved integration.
+
+    This endpoint is intended for internal use by background tasks that need
+    to authenticate with a third-party service.  Treat the response as
+    sensitive — it contains plaintext secrets.
+    """
+    integration = (
+        db.query(UserIntegration)
+        .filter(UserIntegration.id == integration_id, UserIntegration.owner_id == owner_id)
+        .first()
+    )
+    if not integration:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Integration not found")
+
+    credentials = _decode_credentials(integration.credentials)
+    return {"credentials": credentials or {}}

app/models.py

@@ -402,9 +402,11 @@ class UserImapAccount(Base):
     host = Column(String(255), nullable=False)
     port = Column(Integer, nullable=False, default=993)
     username = Column(String(255), nullable=False)
-    # Password stored in plain text — the admin is responsible for access control.
-    # TODO: Encrypt at rest using cryptography.fernet before deploying in high-security
-    # environments. See SECURITY_AUDIT.md for full risk assessment and mitigation notes.
+    # Password stored encrypted using Fernet symmetric encryption via
+    # app.utils.encryption.encrypt_value / decrypt_value (keyed from SESSION_SECRET).
+    # New records are always encrypted; legacy plaintext records are transparently
+    # handled by decrypt_value which returns the value unchanged when no "enc:" prefix
+    # is present.
     password = Column(String(1024), nullable=False)
     use_ssl = Column(Boolean, nullable=False, default=True)
 
@@ -467,3 +469,134 @@ class BackupRecord(Base):
     remote_path = Column(String(1024), nullable=True)
 
     created_at = Column(DateTime(timezone=True), server_default=func.now(), index=True)
+
+
+# ---------------------------------------------------------------------------
+# Integration direction / type constants
+# ---------------------------------------------------------------------------
+
+
+class IntegrationDirection:
+    """Direction of data flow for a UserIntegration."""
+
+    SOURCE = "SOURCE"
+    DESTINATION = "DESTINATION"
+
+    ALL = {SOURCE, DESTINATION}
+
+
+class IntegrationType:
+    """Supported integration types for UserIntegration."""
+
+    # Source integrations (ingestion)
+    IMAP = "IMAP"
+    WATCH_FOLDER = "WATCH_FOLDER"
+    WEBHOOK = "WEBHOOK"
+
+    # Destination integrations (storage / output)
+    S3 = "S3"
+    DROPBOX = "DROPBOX"
+    GOOGLE_DRIVE = "GOOGLE_DRIVE"
+    ONEDRIVE = "ONEDRIVE"
+    WEBDAV = "WEBDAV"
+    NEXTCLOUD = "NEXTCLOUD"
+    FTP = "FTP"
+    SFTP = "SFTP"
+    EMAIL = "EMAIL"
+    PAPERLESS = "PAPERLESS"
+    RCLONE = "RCLONE"
+
+    ALL = {
+        IMAP,
+        WATCH_FOLDER,
+        WEBHOOK,
+        S3,
+        DROPBOX,
+        GOOGLE_DRIVE,
+        ONEDRIVE,
+        WEBDAV,
+        NEXTCLOUD,
+        FTP,
+        SFTP,
+        EMAIL,
+        PAPERLESS,
+        RCLONE,
+    }
+
+
+class UserIntegration(Base):
+    """Generic per-user integration record (source or destination).
+
+    Replaces ad-hoc per-integration-type tables with a single, extensible
+    model that supports any combination of ingestion sources and storage
+    destinations without schema changes when new integrations are added.
+
+    ``config`` holds non-sensitive connection settings as a JSON string
+    (e.g. host, port, bucket name, folder path).
+
+    ``credentials`` holds sensitive secrets (passwords, tokens, API keys)
+    as a Fernet-encrypted JSON string.  Always use
+    ``app.utils.encryption.encrypt_value`` / ``decrypt_value`` when
+    writing / reading this field.
+
+    Example config + credentials shapes by integration type:
+
+    IMAP:
+      config      = {"host": "imap.example.com", "port": 993,
+                     "username": "user@example.com", "use_ssl": true,
+                     "delete_after_process": false}
+      credentials = {"password": "secret"}
+
+    S3:
+      config      = {"bucket": "my-bucket", "region": "us-east-1",
+                     "endpoint_url": null, "folder_prefix": ""}
+      credentials = {"access_key_id": "AKI…", "secret_access_key": "…"}
+
+    DROPBOX:
+      config      = {"folder": "/DocuElevate"}
+      credentials = {"refresh_token": "…", "app_key": "…",
+                     "app_secret": "…"}
+
+    GOOGLE_DRIVE:
+      config      = {"folder_id": "1abc…"}
+      credentials = {"credentials_json": "{…service-account or OAuth…}"}
+
+    WEBDAV / NEXTCLOUD:
+      config      = {"url": "https://cloud.example.com/dav/",
+                     "folder": "/Documents"}
+      credentials = {"username": "user", "password": "secret"}
+    """
+
+    __tablename__ = "user_integrations"
+
+    id = Column(Integer, primary_key=True, index=True)
+
+    # Stable owner identifier — matches FileRecord.owner_id / UserImapAccount.owner_id
+    owner_id = Column(String, nullable=False, index=True)
+
+    # "SOURCE" or "DESTINATION"  (see IntegrationDirection)
+    direction = Column(String(20), nullable=False, index=True)
+
+    # One of the IntegrationType constants (e.g. "IMAP", "S3", "DROPBOX")
+    integration_type = Column(String(50), nullable=False, index=True)
+
+    # Human-readable label chosen by the user (e.g. "Work Gmail", "S3 Archive")
+    name = Column(String(255), nullable=False)
+
+    # Non-sensitive connection configuration (JSON string)
+    config = Column(Text, nullable=True)
+
+    # Sensitive credentials — always stored encrypted via encrypt_value()
+    credentials = Column(Text, nullable=True)
+
+    # When False the integration is not polled / used by background tasks
+    is_active = Column(Boolean, nullable=False, default=True)
+
+    # Timestamp of the last successful use of this integration
+    last_used_at = Column(DateTime(timezone=True), nullable=True)
+
+    # Last error message if the most recent operation failed (NULL = last op succeeded)
+    last_error = Column(Text, nullable=True)
+
+    created_at = Column(DateTime(timezone=True), server_default=func.now())
+    updated_at = Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now())

app/tasks/imap_tasks.py

@@ -37,6 +37,19 @@ logger = logging.getLogger(__name__)
 # Initialize Redis connection using Celery's Redis settings
 redis_client = redis.StrictRedis.from_url(settings.redis_url, decode_responses=True)
 
+
+def _decrypt_imap_password(password: str | None) -> str | None:
+    """Decrypt an IMAP account password stored in the database.
+
+    Passwords are stored encrypted (Fernet, ``enc:`` prefix) for new records;
+    legacy plaintext records are returned unchanged so existing accounts
+    continue to work until they are next updated via the API.
+    """
+    from app.utils.encryption import decrypt_value
+
+    return decrypt_value(password)
+
+
 LOCK_KEY = "imap_lock"  # Unique key for locking
 LOCK_EXPIRE = 300  # Lock expires in 5 minutes
 
@@ -160,7 +173,7 @@ def _pull_user_imap_accounts() -> None:
                         host=acct.host,
                         port=acct.port,
                         username=acct.username,
-                        password=acct.password,
+                        password=_decrypt_imap_password(acct.password),
                         use_ssl=acct.use_ssl,
                         delete_after_process=acct.delete_after_process,
                     )

migrations/versions/023_add_user_integrations.py

@@ -0,0 +1,48 @@
+"""Add user_integrations table for generic multi-tenant source/destination integrations
+
+Revision ID: 023_add_user_integrations
+Revises: 022_add_user_imap_accounts
+Create Date: 2026-03-08
+"""
+
+from typing import Union
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "023_add_user_integrations"
+down_revision: Union[str, None] = "022_add_user_imap_accounts"
+depends_on: Union[str, None] = None
+
+
+def upgrade() -> None:
+    """Create user_integrations table."""
+    op.create_table(
+        "user_integrations",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("owner_id", sa.String(), nullable=False),
+        sa.Column("direction", sa.String(20), nullable=False),
+        sa.Column("integration_type", sa.String(50), nullable=False),
+        sa.Column("name", sa.String(255), nullable=False),
+        sa.Column("config", sa.Text(), nullable=True),
+        sa.Column("credentials", sa.Text(), nullable=True),
+        sa.Column("is_active", sa.Boolean(), nullable=False, server_default="1"),
+        sa.Column("last_used_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("last_error", sa.Text(), nullable=True),
+        sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now()),
+        sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now()),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    op.create_index("ix_user_integrations_id", "user_integrations", ["id"])
+    op.create_index("ix_user_integrations_owner_id", "user_integrations", ["owner_id"])
+    op.create_index("ix_user_integrations_direction", "user_integrations", ["direction"])
+    op.create_index("ix_user_integrations_integration_type", "user_integrations", ["integration_type"])
+
+
+def downgrade() -> None:
+    """Drop user_integrations table."""
+    op.drop_index("ix_user_integrations_integration_type", "user_integrations")
+    op.drop_index("ix_user_integrations_direction", "user_integrations")
+    op.drop_index("ix_user_integrations_owner_id", "user_integrations")
+    op.drop_index("ix_user_integrations_id", "user_integrations")
+    op.drop_table("user_integrations")

tests/conftest.py

@@ -67,6 +67,7 @@ from app.models import (  # noqa: F401, E402
     ProcessingLog,
     SavedSearch,
     UserImapAccount,
+    UserIntegration,
     UserProfile,
     WebhookConfig,
 )

tests/test_api_integrations.py

@@ -0,0 +1,550 @@
+"""Tests for the per-user integrations API (app/api/integrations.py)."""
+
+import pytest
+from fastapi.testclient import TestClient
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+from sqlalchemy.pool import StaticPool
+
+from app.database import Base, get_db
+from app.models import UserIntegration
+
+# ---------------------------------------------------------------------------
+# Test data constants
+# ---------------------------------------------------------------------------
+
+_OWNER = "test_user@example.com"
+_OTHER_OWNER = "other_user@example.com"
+
+_IMAP_SOURCE = {
+    "direction": "SOURCE",
+    "integration_type": "IMAP",
+    "name": "Work Gmail",
+    "config": {"host": "imap.gmail.com", "port": 993, "username": "work@example.com", "use_ssl": True},
+    "credentials": {"password": "s3cr3t"},
+    "is_active": True,
+}
+
+_S3_DESTINATION = {
+    "direction": "DESTINATION",
+    "integration_type": "S3",
+    "name": "Archive Bucket",
+    "config": {"bucket": "my-bucket", "region": "us-east-1"},
+    "credentials": {
+        "access_key_id": "AKIAIOSFODNN7EXAMPLE",
+        "secret_access_key": "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY",
+    },
+    "is_active": True,
+}
+
+
+# ---------------------------------------------------------------------------
+# Shared fixture helpers
+# ---------------------------------------------------------------------------
+
+
+@pytest.fixture()
+def int_engine():
+    """In-memory SQLite engine for integration tests."""
+    engine = create_engine(
+        "sqlite:///:memory:",
+        connect_args={"check_same_thread": False},
+        poolclass=StaticPool,
+    )
+    Base.metadata.create_all(bind=engine)
+    yield engine
+    Base.metadata.drop_all(bind=engine)
+
+
+@pytest.fixture()
+def int_session(int_engine):
+    """DB session scoped to one test."""
+    Session = sessionmaker(bind=int_engine)
+    session = Session()
+    yield session
+    session.close()
+
+
+def _make_client(int_engine, owner_id: str = _OWNER):
+    """Return a TestClient with *owner_id* injected as the authenticated user."""
+    from app.api.integrations import _get_owner_id
+    from app.main import app
+
+    def override_db():
+        Session = sessionmaker(bind=int_engine)
+        session = Session()
+        try:
+            yield session
+        finally:
+            session.close()
+
+    def override_owner():
+        return owner_id
+
+    app.dependency_overrides[get_db] = override_db
+    app.dependency_overrides[_get_owner_id] = override_owner
+    with TestClient(app, base_url="http://localhost", raise_server_exceptions=False) as client:
+        yield client
+    app.dependency_overrides.clear()
+
+
+@pytest.fixture()
+def int_client(int_engine):
+    """TestClient authenticated as _OWNER."""
+    yield from _make_client(int_engine, _OWNER)
+
+
+# ---------------------------------------------------------------------------
+# CRUD tests
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.integration
+class TestCreateIntegration:
+    """Tests for POST /api/integrations/."""
+
+    def test_create_source_integration(self, int_client):
+        """Create a SOURCE integration and verify the response."""
+        resp = int_client.post("/api/integrations/", json=_IMAP_SOURCE)
+        assert resp.status_code == 201
+        data = resp.json()
+        assert data["direction"] == "SOURCE"
+        assert data["integration_type"] == "IMAP"
+        assert data["name"] == "Work Gmail"
+        assert data["config"]["host"] == "imap.gmail.com"
+        assert data["has_credentials"] is True
+        # Credentials must never appear in the response
+        assert "credentials" not in data
+        assert "password" not in data
+
+    def test_create_destination_integration(self, int_client):
+        """Create a DESTINATION integration and verify the response."""
+        resp = int_client.post("/api/integrations/", json=_S3_DESTINATION)
+        assert resp.status_code == 201
+        data = resp.json()
+        assert data["direction"] == "DESTINATION"
+        assert data["integration_type"] == "S3"
+        assert data["has_credentials"] is True
+
+    def test_create_without_credentials(self, int_client):
+        """Integration without credentials sets has_credentials to False."""
+        payload = dict(_IMAP_SOURCE)
+        payload["credentials"] = None
+        resp = int_client.post("/api/integrations/", json=payload)
+        assert resp.status_code == 201
+        assert resp.json()["has_credentials"] is False
+
+    def test_create_invalid_direction(self, int_client):
+        """An unknown direction returns 400."""
+        payload = dict(_IMAP_SOURCE, direction="INVALID")
+        resp = int_client.post("/api/integrations/", json=payload)
+        assert resp.status_code == 400
+
+    def test_create_invalid_type(self, int_client):
+        """An unknown integration_type returns 400."""
+        payload = dict(_IMAP_SOURCE, integration_type="UNKNOWN_TYPE")
+        resp = int_client.post("/api/integrations/", json=payload)
+        assert resp.status_code == 400
+
+    def test_credentials_are_encrypted_in_db(self, int_client, int_session):
+        """Stored credentials must be encrypted (enc: prefix)."""
+        resp = int_client.post("/api/integrations/", json=_IMAP_SOURCE)
+        assert resp.status_code == 201
+        rec = int_session.query(UserIntegration).first()
+        assert rec is not None
+        assert rec.credentials is not None
+        assert rec.credentials.startswith("enc:")
+
+
+@pytest.mark.integration
+class TestListIntegrations:
+    """Tests for GET /api/integrations/."""
+
+    def test_list_empty(self, int_client):
+        """No integrations returns empty list."""
+        resp = int_client.get("/api/integrations/")
+        assert resp.status_code == 200
+        assert resp.json() == []
+
+    def test_list_returns_own_records(self, int_client, int_session):
+        """Users only see their own integrations."""
+        # Create an integration for _OWNER via the API client
+        int_client.post("/api/integrations/", json=_IMAP_SOURCE)
+
+        # Create an integration for a different owner directly in the DB
+        other_integration = UserIntegration(
+            owner_id=_OTHER_OWNER,
+            direction="DESTINATION",
+            integration_type="S3",
+            name="Archive Bucket",
+            config='{"bucket": "other-bucket"}',
+            is_active=True,
+        )
+        int_session.add(other_integration)
+        int_session.commit()
+
+        resp = int_client.get("/api/integrations/")
+        assert resp.status_code == 200
+        assert len(resp.json()) == 1
+        assert resp.json()[0]["integration_type"] == "IMAP"
+
+    def test_filter_by_direction(self, int_client):
+        """direction query param filters results."""
+        int_client.post("/api/integrations/", json=_IMAP_SOURCE)
+        int_client.post("/api/integrations/", json=_S3_DESTINATION)
+
+        resp = int_client.get("/api/integrations/?direction=SOURCE")
+        assert resp.status_code == 200
+        results = resp.json()
+        assert len(results) == 1
+        assert results[0]["direction"] == "SOURCE"
+
+    def test_filter_by_integration_type(self, int_client):
+        """integration_type query param filters results."""
+        int_client.post("/api/integrations/", json=_IMAP_SOURCE)
+        int_client.post("/api/integrations/", json=_S3_DESTINATION)
+
+        resp = int_client.get("/api/integrations/?integration_type=S3")
+        assert resp.status_code == 200
+        results = resp.json()
+        assert len(results) == 1
+        assert results[0]["integration_type"] == "S3"
+
+    def test_filter_invalid_direction_returns_400(self, int_client):
+        """Unknown direction filter returns 400."""
+        resp = int_client.get("/api/integrations/?direction=BAD")
+        assert resp.status_code == 400
+
+    def test_filter_invalid_type_returns_400(self, int_client):
+        """Unknown integration_type filter returns 400."""
+        resp = int_client.get("/api/integrations/?integration_type=BAD")
+        assert resp.status_code == 400
+
+
+@pytest.mark.integration
+class TestGetIntegration:
+    """Tests for GET /api/integrations/{id}."""
+
+    def test_get_existing(self, int_client):
+        """Retrieve a single integration by ID."""
+        created = int_client.post("/api/integrations/", json=_IMAP_SOURCE).json()
+        resp = int_client.get(f"/api/integrations/{created['id']}")
+        assert resp.status_code == 200
+        assert resp.json()["id"] == created["id"]
+
+    def test_get_not_found(self, int_client):
+        """Non-existent ID returns 404."""
+        resp = int_client.get("/api/integrations/9999")
+        assert resp.status_code == 404
+
+    def test_get_other_users_integration(self, int_client, int_session):
+        """Cannot retrieve another user's integration."""
+        other_integration = UserIntegration(
+            owner_id=_OTHER_OWNER,
+            direction="SOURCE",
+            integration_type="IMAP",
+            name="Other Mailbox",
+            is_active=True,
+        )
+        int_session.add(other_integration)
+        int_session.commit()
+        resp = int_client.get(f"/api/integrations/{other_integration.id}")
+        assert resp.status_code == 404
+
+
+@pytest.mark.integration
+class TestUpdateIntegration:
+    """Tests for PUT /api/integrations/{id}."""
+
+    def test_update_name(self, int_client):
+        """Update the name of an integration."""
+        created = int_client.post("/api/integrations/", json=_IMAP_SOURCE).json()
+        resp = int_client.put(f"/api/integrations/{created['id']}", json={"name": "Personal Gmail"})
+        assert resp.status_code == 200
+        assert resp.json()["name"] == "Personal Gmail"
+
+    def test_update_config(self, int_client):
+        """Update the config dict."""
+        created = int_client.post("/api/integrations/", json=_IMAP_SOURCE).json()
+        new_config = {"host": "imap.new.com", "port": 143, "username": "new@example.com", "use_ssl": False}
+        resp = int_client.put(f"/api/integrations/{created['id']}", json={"config": new_config})
+        assert resp.status_code == 200
+        assert resp.json()["config"]["host"] == "imap.new.com"
+
+    def test_update_credentials_re_encrypts(self, int_client, int_session):
+        """Updating credentials stores the new value encrypted."""
+        created = int_client.post("/api/integrations/", json=_IMAP_SOURCE).json()
+        first_enc = int_session.query(UserIntegration).get(created["id"]).credentials
+
+        resp = int_client.put(f"/api/integrations/{created['id']}", json={"credentials": {"password": "new_password"}})
+        assert resp.status_code == 200
+        int_session.expire_all()
+        second_enc = int_session.query(UserIntegration).get(created["id"]).credentials
+        # Both must be encrypted
+        assert second_enc.startswith("enc:")
+        # They should be different ciphertexts (Fernet uses random IV)
+        assert first_enc != second_enc
+
+    def test_update_is_active(self, int_client):
+        """Deactivate an integration."""
+        created = int_client.post("/api/integrations/", json=_IMAP_SOURCE).json()
+        resp = int_client.put(f"/api/integrations/{created['id']}", json={"is_active": False})
+        assert resp.status_code == 200
+        assert resp.json()["is_active"] is False
+
+    def test_update_not_found(self, int_client):
+        """Updating a non-existent integration returns 404."""
+        resp = int_client.put("/api/integrations/9999", json={"name": "Ghost"})
+        assert resp.status_code == 404
+
+    def test_update_clears_last_error(self, int_client, int_session):
+        """Updating an integration resets last_error."""
+        created = int_client.post("/api/integrations/", json=_IMAP_SOURCE).json()
+        rec = int_session.query(UserIntegration).get(created["id"])
+        rec.last_error = "Previous failure"
+        int_session.commit()
+
+        int_client.put(f"/api/integrations/{created['id']}", json={"name": "Updated"})
+        int_session.expire_all()
+        rec = int_session.query(UserIntegration).get(created["id"])
+        assert rec.last_error is None
+
+
+@pytest.mark.integration
+class TestDeleteIntegration:
+    """Tests for DELETE /api/integrations/{id}."""
+
+    def test_delete_existing(self, int_client, int_session):
+        """Delete an integration and confirm it is removed from the DB."""
+        created = int_client.post("/api/integrations/", json=_IMAP_SOURCE).json()
+        resp = int_client.delete(f"/api/integrations/{created['id']}")
+        assert resp.status_code == 204
+        assert int_session.query(UserIntegration).get(created["id"]) is None
+
+    def test_delete_not_found(self, int_client):
+        """Deleting a non-existent integration returns 404."""
+        resp = int_client.delete("/api/integrations/9999")
+        assert resp.status_code == 404
+
+    def test_delete_other_users_integration(self, int_client, int_session):
+        """Cannot delete another user's integration."""
+        other_integration = UserIntegration(
+            owner_id=_OTHER_OWNER,
+            direction="SOURCE",
+            integration_type="IMAP",
+            name="Other Mailbox",
+            is_active=True,
+        )
+        int_session.add(other_integration)
+        int_session.commit()
+        resp = int_client.delete(f"/api/integrations/{other_integration.id}")
+        assert resp.status_code == 404
+
+
+@pytest.mark.integration
+class TestGetIntegrationCredentials:
+    """Tests for GET /api/integrations/{id}/credentials."""
+
+    def test_returns_decrypted_credentials(self, int_client):
+        """Credentials endpoint returns the decrypted dict."""
+        created = int_client.post("/api/integrations/", json=_IMAP_SOURCE).json()
+        resp = int_client.get(f"/api/integrations/{created['id']}/credentials")
+        assert resp.status_code == 200
+        creds = resp.json()["credentials"]
+        assert creds["password"] == "s3cr3t"  # noqa: S105
+
+    def test_returns_empty_dict_when_no_credentials(self, int_client):
+        """No credentials stored returns empty dict."""
+        payload = dict(_IMAP_SOURCE, credentials=None)
+        created = int_client.post("/api/integrations/", json=payload).json()
+        resp = int_client.get(f"/api/integrations/{created['id']}/credentials")
+        assert resp.status_code == 200
+        assert resp.json()["credentials"] == {}
+
+    def test_not_found(self, int_client):
+        """Non-existent integration returns 404."""
+        resp = int_client.get("/api/integrations/9999/credentials")
+        assert resp.status_code == 404
+
+    def test_other_users_credentials_returns_404(self, int_client, int_session):
+        """Cannot retrieve another user's credentials."""
+        other_integration = UserIntegration(
+            owner_id=_OTHER_OWNER,
+            direction="SOURCE",
+            integration_type="IMAP",
+            name="Other Mailbox",
+            credentials='{"password": "secret"}',
+            is_active=True,
+        )
+        int_session.add(other_integration)
+        int_session.commit()
+        resp = int_client.get(f"/api/integrations/{other_integration.id}/credentials")
+        assert resp.status_code == 404
+
+
+@pytest.mark.unit
+class TestIntegrationModel:
+    """Unit tests for the UserIntegration model and helper constants."""
+
+    def test_integration_direction_constants(self):
+        """IntegrationDirection has expected values."""
+        from app.models import IntegrationDirection
+
+        assert IntegrationDirection.SOURCE == "SOURCE"
+        assert IntegrationDirection.DESTINATION == "DESTINATION"
+        assert "SOURCE" in IntegrationDirection.ALL
+        assert "DESTINATION" in IntegrationDirection.ALL
+
+    def test_integration_type_constants(self):
+        """IntegrationType has expected values."""
+        from app.models import IntegrationType
+
+        assert IntegrationType.IMAP == "IMAP"
+        assert IntegrationType.S3 == "S3"
+        assert IntegrationType.DROPBOX == "DROPBOX"
+        assert IntegrationType.GOOGLE_DRIVE == "GOOGLE_DRIVE"
+        assert IntegrationType.ONEDRIVE == "ONEDRIVE"
+        assert IntegrationType.WEBDAV == "WEBDAV"
+        assert IntegrationType.NEXTCLOUD == "NEXTCLOUD"
+        assert IntegrationType.WATCH_FOLDER == "WATCH_FOLDER"
+        assert IntegrationType.WEBHOOK == "WEBHOOK"
+        assert IntegrationType.FTP == "FTP"
+        assert IntegrationType.SFTP == "SFTP"
+        assert IntegrationType.EMAIL == "EMAIL"
+        assert IntegrationType.PAPERLESS == "PAPERLESS"
+        assert IntegrationType.RCLONE == "RCLONE"
+
+    def test_user_integration_tablename(self):
+        """UserIntegration uses the correct table name."""
+        assert UserIntegration.__tablename__ == "user_integrations"
+
+    def test_user_integration_fields(self):
+        """UserIntegration has all required columns."""
+        cols = {c.key for c in UserIntegration.__table__.columns}
+        expected = {
+            "id",
+            "owner_id",
+            "direction",
+            "integration_type",
+            "name",
+            "config",
+            "credentials",
+            "is_active",
+            "last_used_at",
+            "last_error",
+            "created_at",
+            "updated_at",
+        }
+        assert expected <= cols
+
+
+@pytest.mark.unit
+class TestCredentialHelpers:
+    """Unit tests for the credential encode/decode helpers in integrations.py."""
+
+    def test_encode_decode_round_trip(self):
+        """Encoding then decoding returns the original dict."""
+        from app.api.integrations import _decode_credentials, _encode_credentials
+
+        original = {"password": "super_secret", "token": "abc123"}
+        encoded = _encode_credentials(original)
+        assert encoded is not None
+        decoded = _decode_credentials(encoded)
+        assert decoded == original
+
+    def test_encode_none_returns_none(self):
+        """Encoding None returns None."""
+        from app.api.integrations import _encode_credentials
+
+        assert _encode_credentials(None) is None
+
+    def test_encode_empty_dict_returns_none(self):
+        """Encoding an empty dict returns None."""
+        from app.api.integrations import _encode_credentials
+
+        assert _encode_credentials({}) is None
+
+    def test_decode_none_returns_none(self):
+        """Decoding None returns None."""
+        from app.api.integrations import _decode_credentials
+
+        assert _decode_credentials(None) is None
+
+    def test_decode_empty_string_returns_none(self):
+        """Decoding an empty string returns None."""
+        from app.api.integrations import _decode_credentials
+
+        assert _decode_credentials("") is None
+
+    def test_decode_invalid_json_returns_none(self):
+        """Decoding a non-JSON plaintext string returns None."""
+        from app.api.integrations import _decode_credentials
+
+        # A non-JSON plaintext string (no enc: prefix) that decrypt_value returns as-is
+        assert _decode_credentials("not-valid-json") is None
+
+
+@pytest.mark.unit
+class TestImapPasswordEncryption:
+    """Unit tests verifying that IMAP passwords are encrypted at rest."""
+
+    def test_create_encrypts_password(self, int_engine):
+        """Creating an IMAP account stores the password encrypted."""
+        from app.api.imap_accounts import _get_owner_id
+        from app.main import app
+        from app.models import SubscriptionPlan, UserImapAccount, UserProfile
+
+        Session = sessionmaker(bind=int_engine)
+
+        def override_db():
+            session = Session()
+            try:
+                yield session
+            finally:
+                session.close()
+
+        def override_owner():
+            return _OWNER
+
+        # Seed a paid plan (max_mailboxes=0 means unlimited for paid) and profile
+        setup_session = Session()
+        plan = SubscriptionPlan(
+            plan_id="paid",
+            name="Paid",
+            price_monthly=9.99,
+            price_yearly=99.99,
+            max_mailboxes=0,  # 0 = unlimited for paid plans
+            is_active=True,
+        )
+        setup_session.add(plan)
+        profile = UserProfile(user_id=_OWNER, subscription_tier="paid")
+        setup_session.add(profile)
+        setup_session.commit()
+        setup_session.close()
+
+        app.dependency_overrides[get_db] = override_db
+        app.dependency_overrides[_get_owner_id] = override_owner
+        try:
+            with TestClient(app, base_url="http://localhost", raise_server_exceptions=False) as client:
+                payload = {
+                    "name": "Test",
+                    "host": "imap.example.com",
+                    "port": 993,
+                    "username": "user@example.com",
+                    "password": "plaintext_password",
+                    "use_ssl": True,
+                    "delete_after_process": False,
+                    "is_active": True,
+                }
+                resp = client.post("/api/imap-accounts/", json=payload)
+                assert resp.status_code == 201
+
+                # Verify the stored password is encrypted
+                verify_session = Session()
+                acct = verify_session.query(UserImapAccount).first()
+                assert acct is not None
+                assert acct.password != "plaintext_password"
+                assert acct.password.startswith("enc:")
+                verify_session.close()
+        finally:
+            app.dependency_overrides.clear()