diff --git a/app/api/__init__.py b/app/api/__init__.py index 8817768f..66910467 100644 --- a/app/api/__init__.py +++ b/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) diff --git a/app/api/imap_accounts.py b/app/api/imap_accounts.py index ace727b3..69b26c72 100644 --- a/app/api/imap_accounts.py +++ b/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") diff --git a/app/api/integrations.py b/app/api/integrations.py new file mode 100644 index 00000000..d7aeb179 --- /dev/null +++ b/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 {}} diff --git a/app/models.py b/app/models.py index 309f76d8..733e5ed5 100644 --- a/app/models.py +++ b/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()) diff --git a/app/tasks/imap_tasks.py b/app/tasks/imap_tasks.py index 2b85dc9d..351d1e5b 100644 --- a/app/tasks/imap_tasks.py +++ b/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, ) diff --git a/migrations/versions/023_add_user_integrations.py b/migrations/versions/023_add_user_integrations.py new file mode 100644 index 00000000..632bf5d5 --- /dev/null +++ b/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") diff --git a/tests/conftest.py b/tests/conftest.py index 2364cc9b..87f66c12 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -67,6 +67,7 @@ from app.models import ( # noqa: F401, E402 ProcessingLog, SavedSearch, UserImapAccount, + UserIntegration, UserProfile, WebhookConfig, ) diff --git a/tests/test_api_integrations.py b/tests/test_api_integrations.py new file mode 100644 index 00000000..3b030cd0 --- /dev/null +++ b/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()