From 8dacb739bdc8ab54388fc532fd192d3447540615 Mon Sep 17 00:00:00 2001 From: Agent Zero Date: Wed, 29 Jul 2026 12:33:46 +0200 Subject: [PATCH] P1 fixes: outbox no_handlers, HTML sanitization, WebSocket plugin check, fail-closed plugin gate, plugin admin-only --- app/commands/mail_commands.py | 3 ++- app/core/outbox.py | 20 +++++++++++++++++++- app/deps.py | 11 ++++++++--- app/main.py | 4 ++++ app/routes/plugins.py | 18 +++++++++--------- 5 files changed, 42 insertions(+), 14 deletions(-) diff --git a/app/commands/mail_commands.py b/app/commands/mail_commands.py index cdbf700..0b167be 100644 --- a/app/commands/mail_commands.py +++ b/app/commands/mail_commands.py @@ -13,6 +13,7 @@ from sqlalchemy.ext.asyncio import AsyncSession from app.commands.base import BaseCommand, CommandResult from app.core.outbox import enqueue_outbox_event +from app.plugins.builtins.mail.services import sanitize_html logger = logging.getLogger(__name__) @@ -62,7 +63,7 @@ class SendMailCommand(BaseCommand): cc_addr=",".join(self.cc) if self.cc else None, subject=self.subject, body_text=self.body_text, - body_html_sanitized=self.body_html, + body_html_sanitized=sanitize_html(self.body_html) if self.body_html else None, direction="outgoing", received_at=datetime.now(UTC), is_read=True, diff --git a/app/core/outbox.py b/app/core/outbox.py index 23c4beb..e59c426 100644 --- a/app/core/outbox.py +++ b/app/core/outbox.py @@ -174,12 +174,30 @@ async def process_outbox_batch( payload_dict = payload try: + # Enrich payload with event metadata for idempotency + payload_dict.setdefault("_event_id", str(event_id)) + payload_dict.setdefault("_event_name", event_name) + payload_dict.setdefault("_event_timestamp", datetime.now(timezone.utc).isoformat()) + results = await event_bus.publish_with_results(event_name, payload_dict) + + # Check if any handlers were registered at all + handler_count = len(results) # If any handler raised, treat as failure handler_errors = [r for r in results if r is not None] if handler_errors: raise handler_errors[0] - await db.execute(_MARK_PUBLISHED_SQL, {"id": str(event_id)}) + + if handler_count == 0: + # No handlers registered — mark as 'no_handlers' not 'published' + # This prevents events from silently disappearing + await db.execute( + text("UPDATE event_outbox SET status = 'no_handlers', published_at = now() WHERE id = :id"), + {"id": str(event_id)}, + ) + logger.warning("Outbox event %s (%s) had no handlers registered", event_id, event_name) + else: + await db.execute(_MARK_PUBLISHED_SQL, {"id": str(event_id)}) published_count += 1 except Exception as exc: logger.error( diff --git a/app/deps.py b/app/deps.py index 2ea937b..dadc062 100644 --- a/app/deps.py +++ b/app/deps.py @@ -312,8 +312,13 @@ def require_active_plugin(plugin_name: str): ) except HTTPException: raise - except Exception: - # If registry not initialized yet, allow request (startup race) - pass + except Exception as exc: + # Fail-closed: if registry check fails, deny access (P1.2 fix) + # Previously this was fail-open (pass) which allowed access on errors + logger.error("Plugin activation check failed for '%s': %s", plugin_name, exc) + raise HTTPException( + status_code=status.HTTP_503_SERVICE_UNAVAILABLE, + detail={"detail": f"Plugin activation check failed", "code": "plugin_check_error"}, + ) return _check diff --git a/app/main.py b/app/main.py index 42e3634..eaafe2a 100644 --- a/app/main.py +++ b/app/main.py @@ -458,6 +458,10 @@ def create_app() -> FastAPI: plugin_dep = Depends(require_active_plugin(plugin_name)) for route in router.routes: if isinstance(route, WebSocketRoute): + # WebSocket routes also need plugin check — don't skip (P1.9 fix) + if not hasattr(route, 'dependencies'): + route.dependencies = [] + route.dependencies.append(plugin_dep) continue # Add require_active_plugin to each HTTP route's dependencies if not hasattr(route, 'dependencies'): diff --git a/app/routes/plugins.py b/app/routes/plugins.py index b3ddd61..dfa6654 100644 --- a/app/routes/plugins.py +++ b/app/routes/plugins.py @@ -18,7 +18,7 @@ from pydantic import BaseModel from sqlalchemy.ext.asyncio import AsyncSession from app.core.db import get_db -from app.deps import require_permission +from app.deps import require_permission, require_admin from app.plugins.base import BasePlugin from app.plugins.manifest import PluginManifest from app.plugins.migration_runner import MigrationValidationError @@ -143,7 +143,7 @@ async def update_plugin_config( name: str, body: PluginConfigUpdate, db: AsyncSession = Depends(get_db), - current_user: dict = Depends(require_permission("plugins:configure")), + current_user: dict = Depends(require_admin), ): """Update the configuration for a specific plugin. @@ -171,7 +171,7 @@ async def update_plugin_config( async def install_plugin( name: str, db: AsyncSession = Depends(get_db), - current_user: dict = Depends(require_permission("plugins:configure")), + current_user: dict = Depends(require_admin), ): """Install a plugin by name. Runs migrations and creates DB record. @@ -204,7 +204,7 @@ async def install_plugin( async def activate_plugin( name: str, db: AsyncSession = Depends(get_db), - current_user: dict = Depends(require_permission("plugins:configure")), + current_user: dict = Depends(require_admin), ): """Activate a plugin by name. Registers event listeners and routes. @@ -237,7 +237,7 @@ async def activate_plugin( async def deactivate_plugin( name: str, db: AsyncSession = Depends(get_db), - current_user: dict = Depends(require_permission("plugins:configure")), + current_user: dict = Depends(require_admin), ): """Deactivate a plugin by name. Unregisters event listeners and routes. @@ -267,7 +267,7 @@ async def uninstall_plugin( name: str, remove_data: bool = Query(False), db: AsyncSession = Depends(get_db), - current_user: dict = Depends(require_permission("plugins:configure")), + current_user: dict = Depends(require_admin), ): """Uninstall a plugin by name. Deactivates, drops tables, removes DB record. @@ -477,7 +477,7 @@ def _install_plugin_from_dir( async def upload_plugin( file: UploadFile = File(...), db: AsyncSession = Depends(get_db), - current_user: dict = Depends(require_permission("plugins:configure")), + current_user: dict = Depends(require_admin), ): """Upload and install a plugin from a ZIP file. @@ -494,7 +494,7 @@ async def upload_plugin( async def install_plugin_from_url( body: PluginUrlInstall, db: AsyncSession = Depends(get_db), - current_user: dict = Depends(require_permission("plugins:configure")), + current_user: dict = Depends(require_admin), ): """Install a plugin from a URL (downloads ZIP and installs). @@ -522,7 +522,7 @@ class MarketplaceInstall(BaseModel): async def install_from_marketplace( body: MarketplaceInstall, db: AsyncSession = Depends(get_db), - current_user: dict = Depends(require_permission("plugins:configure")), + current_user: dict = Depends(require_admin), ): """Install a plugin from the marketplace.