""" Integration Credential model for storing external system authentication. This model securely stores OAuth tokens, API keys, and other credentials needed to authenticate with external integrations like SyncroMSP, MSP Backups, etc. """ from datetime import datetime from typing import Optional from sqlalchemy import ( Boolean, CheckConstraint, Index, LargeBinary, String, Text, ) from sqlalchemy.orm import Mapped, mapped_column from .base import Base, TimestampMixin, UUIDMixin class IntegrationCredential(Base, UUIDMixin, TimestampMixin): """ Integration credentials for external system authentication. Stores encrypted credentials (API keys, OAuth tokens) for integrations. Each integration type has one record with its authentication credentials. All sensitive data is encrypted using AES-256-GCM. Attributes: id: Unique identifier integration_name: Unique name of the integration (syncro, msp_backups, zapier) credential_type: Type of credential (oauth, api_key, basic_auth) api_key_encrypted: Encrypted API key (if credential_type is api_key) oauth_token_encrypted: Encrypted OAuth access token oauth_refresh_token_encrypted: Encrypted OAuth refresh token oauth_expires_at: When the OAuth token expires api_base_url: Base URL for API calls webhook_url: Webhook URL for receiving callbacks is_active: Whether this integration is currently active last_tested_at: When the connection was last tested last_test_status: Result of last connection test created_at: When the credential was created updated_at: When the credential was last updated """ __tablename__ = "integration_credentials" # Integration identification integration_name: Mapped[str] = mapped_column( String(100), unique=True, nullable=False, doc="Unique name of integration (syncro, msp_backups, zapier)", ) # Credential type and encrypted values credential_type: Mapped[Optional[str]] = mapped_column( String(50), nullable=True, doc="Type of credential", ) api_key_encrypted: Mapped[Optional[bytes]] = mapped_column( LargeBinary, nullable=True, doc="Encrypted API key (AES-256-GCM)", ) oauth_token_encrypted: Mapped[Optional[bytes]] = mapped_column( LargeBinary, nullable=True, doc="Encrypted OAuth access token", ) oauth_refresh_token_encrypted: Mapped[Optional[bytes]] = mapped_column( LargeBinary, nullable=True, doc="Encrypted OAuth refresh token", ) oauth_expires_at: Mapped[Optional[datetime]] = mapped_column( nullable=True, doc="When the OAuth token expires", ) # Endpoints api_base_url: Mapped[Optional[str]] = mapped_column( String(500), nullable=True, doc="Base URL for API calls", ) webhook_url: Mapped[Optional[str]] = mapped_column( String(500), nullable=True, doc="Webhook URL for receiving callbacks", ) # Status is_active: Mapped[bool] = mapped_column( Boolean, default=True, nullable=False, doc="Whether this integration is active", ) last_tested_at: Mapped[Optional[datetime]] = mapped_column( nullable=True, doc="When the connection was last tested", ) last_test_status: Mapped[Optional[str]] = mapped_column( String(50), nullable=True, doc="Result of last connection test", ) # Indexes and constraints __table_args__ = ( CheckConstraint( "credential_type IN ('oauth', 'api_key', 'basic_auth')", name="ck_integration_credential_type", ), Index("idx_int_cred_name", "integration_name"), ) def __repr__(self) -> str: """String representation of the integration credential.""" return ( f"" )