P1 fixes: outbox no_handlers, HTML sanitization, WebSocket plugin check, fail-closed plugin gate, plugin admin-only
This commit is contained in:
@@ -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,
|
||||
|
||||
+19
-1
@@ -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(
|
||||
|
||||
+8
-3
@@ -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
|
||||
|
||||
@@ -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'):
|
||||
|
||||
@@ -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.
|
||||
|
||||
|
||||
Reference in New Issue
Block a user