Files
gh-christianlouis-inboxconv…/backend/app/core/config.py
T
copilot-swe-agent[bot] 45a2dd8be2 fix: IPv4 preference + DNS cache fallback for stable POP3/IMAP connections
- Add DNS_CACHE_FALLBACK_ENABLED setting (default: true)
- Add in-process DNS cache (_dns_cache, _dns_cache_lock, _get/_set_cached_ipv4)
- Add _resolve_ipv4 (async) and _resolve_ipv4_sync: AF_INET lookup, cache on
  success, return cached IP on EAI_AGAIN when fallback enabled
- Add _IMAP4SSLwithSNI: connects to pre-resolved IPv4 but uses original
  hostname for TLS SNI/cert verification
- Add _POP3SSLWithIPv4Pref / _POP3WithIPv4Pref: override _create_socket to
  use pre-resolved IPv4 while keeping original host for POP3_SSL SNI
- Add _make_pop3_conn factory and _make_imap_client async factory
- Wire IPv4/cache helpers into all 6 connection call-sites (IMAP x3, POP3 x3)
- Distinguish transient DNS (EAI_AGAIN) from permanent in _format_connection_error
- Replace asyncio.get_event_loop() with get_running_loop() in 4 async sites
- All 75 existing unit tests pass unchanged

Agent-Logs-Url: https://github.com/christianlouis/InboxConverge/sessions/be0a3bbd-1af1-408e-b918-84753723154b

Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
2026-05-03 20:49:51 +00:00

198 lines
7.1 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""
Application configuration using Pydantic settings.
Supports a hybrid configuration model:
• **Bootstrap settings** (DATABASE_URL, SECRET_KEY, ENCRYPTION_KEY)
are always loaded from environment variables or ``.env`` files.
• **Application settings** (SMTP, processing, Gmail API, etc.)
can be managed in the database via the ``AppSetting`` model and
the ``/api/v1/settings`` admin endpoints. When a setting exists
in the database it takes precedence over environment variables.
See ``app.services.config_service.ConfigService`` for the runtime
lookup logic and ``app.models.database_models.AppSetting`` for the
database model.
"""
from typing import Optional, List
from pydantic_settings import BaseSettings, SettingsConfigDict
from pydantic import field_validator
class Settings(BaseSettings):
"""Application settings loaded from environment variables"""
model_config = SettingsConfigDict(
env_file=".env", env_file_encoding="utf-8", case_sensitive=False, extra="ignore"
)
# Application
APP_NAME: str = "InboxConverge"
APP_VERSION: str = "2.0.0"
BUILD_DATE: str = ""
APP_URL: str = "https://inboxconverge.com"
CONTACT_EMAIL: str = "christian@inboxconverge.com"
DEBUG: bool = False
API_V1_PREFIX: str = "/api/v1"
# Server
HOST: str = (
"0.0.0.0" # nosec B104 intentional: containerised service binds all interfaces
)
PORT: int = 8000
# Database
DATABASE_URL: str = (
"postgresql+asyncpg://user:password@localhost:5432/inbox_converge"
)
DATABASE_POOL_SIZE: int = 20
DATABASE_MAX_OVERFLOW: int = 10
# Security
SECRET_KEY: str = "change-this-to-a-secure-random-secret-key-in-production"
ALGORITHM: str = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
REFRESH_TOKEN_EXPIRE_DAYS: int = 7
# Encryption (for storing POP3/IMAP credentials)
ENCRYPTION_KEY: str = "change-this-to-a-secure-encryption-key"
# OAuth2 - Google
GOOGLE_CLIENT_ID: Optional[str] = None
GOOGLE_CLIENT_SECRET: Optional[str] = None
GOOGLE_REDIRECT_URI: str = "http://localhost:3000/auth/callback/google"
# Gmail API (for direct email injection)
GMAIL_API_ENABLED: bool = True
GMAIL_INJECT_LABEL_IDS: List[str] = ["INBOX"]
# CORS
CORS_ORIGINS: List[str] = ["http://localhost:3000", "http://localhost:8000"]
# Stripe Payment
STRIPE_API_KEY: Optional[str] = None
STRIPE_WEBHOOK_SECRET: Optional[str] = None
STRIPE_PUBLISHABLE_KEY: Optional[str] = None
# Subscription Tiers
TIER_FREE_MAX_ACCOUNTS: int = 1
TIER_BASIC_MAX_ACCOUNTS: int = 5
TIER_PRO_MAX_ACCOUNTS: int = 20
TIER_ENTERPRISE_MAX_ACCOUNTS: int = 100
# Email Processing
MAX_EMAILS_PER_RUN: int = 50
CHECK_INTERVAL_MINUTES: int = 5
THROTTLE_EMAILS_PER_MINUTE: int = 10
# Redis (for Celery and caching)
REDIS_URL: str = "redis://localhost:6379/0"
# Celery
CELERY_BROKER_URL: str = "redis://localhost:6379/0"
CELERY_RESULT_BACKEND: str = "redis://localhost:6379/0"
# Apprise (notifications)
APPRISE_ENABLED: bool = True
# Mail connectivity resilience
# When True the app caches the last successfully resolved IPv4 address for
# every mail host and uses that address as a fallback if DNS subsequently
# fails (e.g. EAI_AGAIN / Temporary failure in name resolution). This also
# makes every outgoing POP3/IMAP connection prefer IPv4, avoiding
# ENETUNREACH errors on Docker hosts where IPv6 is not routed to the
# internet.
DNS_CACHE_FALLBACK_ENABLED: bool = True
# Logging
LOG_LEVEL: str = "INFO"
# Admin
ADMIN_EMAIL: Optional[str] = "christian@inboxconverge.com"
ADMIN_PASSWORD: Optional[str] = None
# User defaults & access control
# Tier assigned to every new user on registration: free | basic | pro | enterprise
DEFAULT_USER_TIER: str = "free"
# Comma-separated list of allowed email domains (empty = no restriction).
# When set, only addresses from these domains may register or log in.
# Useful for B2B / Google Workspace installations.
# Example: "company.com,subsidiary.com"
ALLOWED_DOMAINS: List[str] = []
# Mail Server Presets
MAIL_SERVER_PRESETS_FILE: str = "app/data/mail_server_presets.json"
@field_validator("CORS_ORIGINS", mode="before")
@classmethod
def assemble_cors_origins(cls, v: str | List[str]) -> List[str]:
"""Parse CORS origins from environment variable"""
if isinstance(v, str):
return [i.strip() for i in v.split(",")]
return v
@field_validator("ALLOWED_DOMAINS", mode="before")
@classmethod
def assemble_allowed_domains(cls, v: str | List[str]) -> List[str]:
"""Parse allowed domains from a comma-separated environment variable"""
if isinstance(v, str):
return [d.strip().lower() for d in v.split(",") if d.strip()]
return [d.lower() for d in v if d]
@field_validator("DEFAULT_USER_TIER")
@classmethod
def validate_default_user_tier(cls, v: str) -> str:
"""Ensure DEFAULT_USER_TIER is one of the known tier values"""
valid = {"free", "basic", "pro", "enterprise"}
if v.lower() not in valid:
raise ValueError(f"DEFAULT_USER_TIER must be one of {valid}, got '{v}'")
return v.lower()
@field_validator("SECRET_KEY")
@classmethod
def validate_secret_key(cls, v: str) -> str:
"""Validate that SECRET_KEY is changed from default and is secure"""
default_keys = [
"change-this-to-a-secure-random-secret-key-in-production",
"secret",
"secret-key",
"secretkey",
]
if v.lower() in default_keys:
raise ValueError(
"SECRET_KEY must be changed from default value! "
"Generate a secure key with: python -c 'import secrets; print(secrets.token_urlsafe(32))'"
)
if len(v) < 32:
raise ValueError(
f"SECRET_KEY must be at least 32 characters long (current: {len(v)}). "
"Generate a secure key with: python -c 'import secrets; print(secrets.token_urlsafe(32))'"
)
return v
@field_validator("ENCRYPTION_KEY")
@classmethod
def validate_encryption_key(cls, v: str) -> str:
"""Validate that ENCRYPTION_KEY is changed from default and is secure"""
default_keys = [
"change-this-to-a-secure-encryption-key",
"encryption",
"encryption-key",
"encryptionkey",
]
if v.lower() in default_keys:
raise ValueError(
"ENCRYPTION_KEY must be changed from default value! "
"Generate a secure key with: python -c 'import secrets; print(secrets.token_urlsafe(32))'"
)
if len(v) < 32:
raise ValueError(
f"ENCRYPTION_KEY must be at least 32 characters long (current: {len(v)}). "
"Generate a secure key with: python -c 'import secrets; print(secrets.token_urlsafe(32))'"
)
return v
# Global settings instance
settings = Settings()