Phase 1: Fix all critical release blockers (B1-B10)

B1: Remove duplicate get_redis() — singleton no longer overwritten B2: Plugin routes now enforce activation status via require_active_plugin() B3: Fix UploadFile ForwardRef error — remove functools.wraps from wrap_plugin_route B4: DMS upload uses true streaming via save_stream() instead of RAM accumulation B5: Worker on_startup registers plugin event handlers + webhook dispatcher B6: Implement send_password_reset_email job, remove raw token logging B7: Webhook SSRF protection (IP validation, no redirects), secret removed from response B8: RLS repair migration 0044 + separate crm_runtime DB user (NOSUPERUSER, NOBYPASSRLS) B9: Fix .env.docker.example AUTH_SECRET → SECRET_KEY B10: Remove Redis default password, remove exposed DB/Redis ports Also: add frontend_url to config, add SMTP settings to .env.docker.example, update prestart.sh to use MIGRATION_DATABASE_URL for alembic.
2026-07-26 20:45:42 +02:00
parent 7a14973c68
commit 5ec1fc9b05
32 changed files with 1781 additions and 76 deletions
@@ -91,11 +91,6 @@ def hash_token(token: str) -> str:
    return hashlib.sha256(token.encode()).hexdigest()


-def get_redis() -> aioredis.Redis:
-    """Get a Redis client instance."""
-    return aioredis.from_url(get_settings().redis_url, decode_responses=True)
-
-
 async def create_session(
    db: AsyncSession,
    redis: aioredis.Redis,
@@ -80,3 +80,72 @@ async def get_job_status(job_id: str) -> dict[str, Any] | None:
        "start_time": job_info.start_time.isoformat() if job_info.start_time else None,
        "finish_time": job_info.finish_time.isoformat() if job_info.finish_time else None,
    }
+
+
+# ── Password Reset Email Job ─────────────────────────────────────────────────
+
+async def send_password_reset_email(
+    ctx: dict[str, Any],
+    *,
+    user_id: str,
+    email: str,
+    raw_token: str,
+    expires_at: str,
+) -> None:
+    """Send a password reset email via SMTP.
+
+    This is an ARQ worker function. It is registered with the job registry
+    so the worker can execute it when the auth service enqueues it.
+    """
+    import aiosmtplib
+    from email.mime.text import MIMEText
+    from email.mime.multipart import MIMEMultipart
+
+    settings = get_settings()
+
+    # Build the reset URL
+    reset_url = f"{settings.frontend_url.rstrip('/')}/reset-password?token={raw_token}"
+
+    # Build the email
+    msg = MIMEMultipart("alternative")
+    msg["From"] = settings.smtp_from_email
+    msg["To"] = email
+    msg["Subject"] = "LeoCRM — Passwort zurücksetzen"
+
+    text_body = (
+        f"Sie haben angefordert, Ihr Passwort zurückzusetzen.\n\n"
+        f"Klicken Sie auf den folgenden Link, um ein neues Passwort zu setzen:\n"
+        f"{reset_url}\n\n"
+        f"Dieser Link ist gültig bis {expires_at}.\n\n"
+        f"Falls Sie diese Anfrage nicht gestellt haben, können Sie diese\n"
+        f"E-Mail ignorieren. Ihr Passwort bleibt unverändert.\n"
+    )
+    html_body = (
+        f"<html><body>"
+        f"<h2>Passwort zurücksetzen</h2>"
+        f"<p>Sie haben angefordert, Ihr Passwort zurückzusetzen.</p>"
+        f"<p><a href=\"{reset_url}\">Passwort jetzt zurücksetzen</a></p>"
+        f"<p>Dieser Link ist gültig bis {expires_at}.</p>"
+        f"<p>Falls Sie diese Anfrage nicht gestellt haben, können Sie diese "
+        f"E-Mail ignorieren. Ihr Passwort bleibt unverändert.</p>"
+        f"</body></html>"
+    )
+    msg.attach(MIMEText(text_body, "plain", "utf-8"))
+    msg.attach(MIMEText(html_body, "html", "utf-8"))
+
+    # Send via SMTP
+    await aiosmtplib.send(
+        msg,
+        hostname=settings.smtp_host,
+        port=settings.smtp_port,
+        username=settings.smtp_username,
+        password=settings.smtp_password,
+        start_tls=settings.smtp_use_tls,
+    )
+    logger.info("Password reset email sent to %s for user %s", email, user_id)
+
+
+# Register the job so the worker can find it
+from app.core.job_registry import register_job  # noqa: E402
+
+register_job("send_password_reset_email", send_password_reset_email)
@@ -1,14 +1,19 @@
 """Plugin error isolation wrapper."""
 import logging
 import functools
+from fastapi import UploadFile as _UploadFile  # noqa: F401 — needed for ForwardRef resolution
 from fastapi.responses import JSONResponse

 logger = logging.getLogger(__name__)


 def wrap_plugin_route(handler):
-    """Decorator that isolates plugin route errors and returns structured JSON."""
-    @functools.wraps(handler)
+    """Decorator that isolates plugin route errors and returns structured JSON.
+
+    Does NOT use functools.wraps to avoid copying __annotations__ and
+    __wrapped__ — FastAPI would otherwise try to resolve
+    ``ForwardRef('UploadFile')`` from the original handler's signature.
+    """
    async def wrapper(*args, **kwargs):
        try:
            return await handler(*args, **kwargs)
@@ -18,4 +23,8 @@ def wrap_plugin_route(handler):
                status_code=500,
                content={'detail': f'Plugin error: {exc}', 'code': 'plugin_error'}
            )
+    # Preserve identity for debugging but NOT __wrapped__ or __annotations__
+    wrapper.__name__ = getattr(handler, '__name__', 'wrapper')
+    wrapper.__module__ = getattr(handler, '__module__', __name__)
+    wrapper.__qualname__ = getattr(handler, '__qualname__', 'wrapper')
    return wrapper
@@ -90,11 +90,59 @@ def _get_redis_settings() -> RedisSettings:
 async def on_startup(ctx: dict[str, Any]) -> None:
    """Called when worker starts."""
    logger.info("ARQ worker starting...")
+
+    # Initialize Redis singleton (same as API lifespan)
+    from app.core.auth import init_redis
+    await init_redis()
+
+    # Initialize service container
+    from app.core.service_container import get_container
+    container = get_container()
+    await container.initialize()
+
+    # Initialize plugin registry and discover built-in plugins
+    from app.plugins.registry import get_registry
+    from app.core.db import get_engine
+    from app.core.event_bus import get_event_bus
+    from app.core.webhook_dispatcher import register_webhook_event_handlers
+    from sqlalchemy import select as sa_select
+    from app.models.plugin import Plugin as PluginModel
+    from sqlalchemy.ext.asyncio import async_sessionmaker
+
+    registry = get_registry()
+    registry.initialize(get_engine(), app=None)
+    registry.discover_builtins()
+
+    event_bus = get_event_bus()
+    async_session = async_sessionmaker(get_engine(), expire_on_commit=False)
+
+    # Activate plugins that are marked active in DB (register event handlers)
+    async with async_session() as db:
+        for name in registry.resolve_load_order():
+            plugin = registry.get_plugin(name)
+            if plugin is None:
+                continue
+            result = await db.execute(
+                sa_select(PluginModel).where(PluginModel.name == name)
+            )
+            plugin_record = result.scalar_one_or_none()
+            if plugin_record is None or not plugin_record.active:
+                continue
+            try:
+                await plugin.on_activate(db, container, event_bus)
+                logger.info(f"Worker: activated plugin {name}")
+            except Exception as exc:
+                logger.error(f"Worker: failed to activate plugin {name}: {exc}")
+        await db.commit()
+
+    # Register webhook dispatcher on the event bus
+    register_webhook_event_handlers(event_bus)
+    logger.info("Worker: webhook event handlers registered")
+
    # Register search providers (normally done by app startup)
    try:
-        from app.core.db import get_session_factory
        from app.plugins.builtins.unified_search.provider_registry import auto_register_providers
-        factory = get_session_factory()
+        factory = async_session
        async with factory() as db:
            await auto_register_providers(db)
        logger.info("Search providers registered for worker")
@@ -105,6 +153,8 @@ async def on_startup(ctx: dict[str, Any]) -> None:
 async def on_shutdown(ctx: dict[str, Any]) -> None:
    """Called when worker shuts down."""
    logger.info("ARQ worker shutting down...")
+    from app.core.auth import close_redis
+    await close_redis()


 # ---------------------------------------------------------------------------