diff --git a/backend/app/main.py b/backend/app/main.py index c8d5d1a..95280a2 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -12,7 +12,7 @@ import os from app.config import settings from app.database import init_db -from app.routers import photos, folders, heaps, tags, discard, library, search, auth, admin, sharing, upload, download, features, nextcloud +from app.routers import photos, folders, heaps, tags, discard, library, search, auth, admin, sharing, upload, download, features, nextcloud, nc_webhook from app.services.scanner import start_initial_scan, bootstrap_default_source_root from app.services.cleanup import cleanup_data_integrity @@ -113,6 +113,7 @@ app.include_router(upload.router, prefix="/api/v1/upload", tags=["upload"]) app.include_router(download.router, prefix="/api/v1/download", tags=["download"]) app.include_router(features.router, prefix="/api/v1/features", tags=["features"]) app.include_router(nextcloud.router, prefix="/api/v1/nextcloud", tags=["nextcloud"]) +app.include_router(nc_webhook.router, prefix="/api/v1/internal", tags=["nc-webhook"]) @app.get("/") async def root(): diff --git a/backend/app/routers/nc_webhook.py b/backend/app/routers/nc_webhook.py new file mode 100644 index 0000000..d8cc6a3 --- /dev/null +++ b/backend/app/routers/nc_webhook.py @@ -0,0 +1,179 @@ +"""Internal webhook receiver for Nextcloud file events. + +Replaces the watchfiles-based `watch_folders` Celery task: instead of +mule polling the bind-mount with inotify, Nextcloud's `webhook_listeners` +app POSTs here on every NodeCreated / NodeWritten / NodeDeleted / +NodeRenamed event, and we dispatch the same scan_folder / +handle_file_deletion machinery that the watcher used. + +Auth: `Authorization: Bearer ` header. +constant_time compare. 401 on mismatch, 401 also when the secret isn't +configured (fail closed). + +The route is intentionally outside `/api/v1/photos/...` so it doesn't +get caught by the per-user auth middleware — webhook requests come +from Nextcloud as a service principal, not as a logged-in user. They +get NO mule app session. +""" +from __future__ import annotations + +import hmac +import logging +import os +from pathlib import Path +from typing import Any + +from fastapi import APIRouter, Header, HTTPException, Request, status +from sqlalchemy import select + +from app.database import AsyncSessionLocal +from app.models.folders import SourceRoot +from app.services.nextcloud_dav import NEXTCLOUD_USERS_ROOT + +logger = logging.getLogger(__name__) + +router = APIRouter() + + +# Same list the on-disk watcher used (app.tasks.scan.SUPPORTED_EXTENSIONS). +# Imported lazily inside the handler so this module can load without +# pulling in the tasks package at startup. +def _supported_extensions() -> set[str]: + from app.tasks.scan import SUPPORTED_EXTENSIONS + return SUPPORTED_EXTENSIONS + + +def _expected_secret() -> str | None: + return os.environ.get("NEXTCLOUD_WEBHOOK_SECRET") or None + + +def _nc_path_to_abs(nc_path: str) -> str | None: + """Map a Nextcloud-internal path (`/admin/files/Photos/foo.jpg`) to + the absolute bind-mount path mule's workers operate on + (`/nextcloud-users/admin/files/Photos/foo.jpg`). + + Returns None for paths that don't sit under `/files/...` + (NC also emits events for trashbin, versions, etc — we ignore + those). + """ + if not nc_path or not nc_path.startswith("/"): + return None + parts = nc_path.lstrip("/").split("/", 2) + if len(parts) < 3 or parts[1] != "files": + return None + return os.path.join(NEXTCLOUD_USERS_ROOT, parts[0], "files", parts[2]) + + +def _classify(event_class: str) -> str | None: + """Bucket the full event class string into the four buckets we act + on. Returns None for events we don't care about (Before*, copy, + touched, etc).""" + short = event_class.rsplit("\\", 1)[-1] + return { + "NodeCreatedEvent": "created", + "NodeWrittenEvent": "written", + "NodeDeletedEvent": "deleted", + "NodeRenamedEvent": "renamed", + }.get(short) + + +async def _source_root_for(parent_dir: str) -> str | None: + """Find the SourceRoot id whose path contains `parent_dir`.""" + async with AsyncSessionLocal() as session: + result = await session.execute( + select(SourceRoot).where(SourceRoot.is_active == True) # noqa: E712 + ) + roots = result.scalars().all() + normalized = os.path.normpath(parent_dir) + for sr in roots: + root_path = os.path.normpath(sr.path) + if normalized == root_path or normalized.startswith(root_path + os.sep): + return sr.id + return None + + +@router.post("/nc-webhook") +async def nc_webhook( + request: Request, + authorization: str | None = Header(default=None), +): + """Receive a Nextcloud file event and dispatch the matching + scan_folder / handle_file_deletion task. Returns 204 on success + (Nextcloud doesn't care about the body).""" + expected = _expected_secret() + if not expected: + # Fail closed: a misconfigured server should reject webhooks + # rather than accept arbitrary POSTs. + logger.error("nc-webhook hit but NEXTCLOUD_WEBHOOK_SECRET is not set") + raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED) + + if not authorization or not authorization.startswith("Bearer "): + raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED) + presented = authorization[len("Bearer "):] + if not hmac.compare_digest(presented, expected): + raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED) + + try: + payload: dict[str, Any] = await request.json() + except Exception: + raise HTTPException(status_code=400, detail="malformed json") + + event = payload.get("event") or {} + event_class = event.get("class") or "" + bucket = _classify(event_class) + if bucket is None: + return {"status": "ignored", "reason": "unwanted event"} + + # Import lazily so this router can load before the celery app is + # ready — important when the backend boots before broker is up. + from app.tasks.scan import scan_folder, handle_file_deletion + + supported = _supported_extensions() + + if bucket in ("created", "written"): + node = event.get("node") or {} + nc_path = node.get("path") + abs_path = _nc_path_to_abs(nc_path) if nc_path else None + if not abs_path: + return {"status": "ignored", "reason": "non-user-files path"} + if Path(abs_path).suffix.lower() not in supported: + return {"status": "ignored", "reason": "unsupported extension"} + parent_dir = str(Path(abs_path).parent) + source_root_id = await _source_root_for(parent_dir) + if source_root_id is None: + # Outside any registered source root — we don't index this + # part of Nextcloud at all. + return {"status": "ignored", "reason": "outside source root"} + scan_folder.delay(parent_dir, source_root_id) + logger.info("nc-webhook %s: queued scan_folder for %s", bucket, parent_dir) + return {"status": "queued", "action": "scan_folder", "path": parent_dir} + + if bucket == "deleted": + node = event.get("node") or {} + nc_path = node.get("path") + abs_path = _nc_path_to_abs(nc_path) if nc_path else None + if not abs_path: + return {"status": "ignored", "reason": "non-user-files path"} + if Path(abs_path).suffix.lower() not in supported: + return {"status": "ignored", "reason": "unsupported extension"} + await handle_file_deletion(abs_path) + logger.info("nc-webhook deleted: marked %s as discarded", abs_path) + return {"status": "applied", "action": "discard", "path": abs_path} + + if bucket == "renamed": + source = event.get("source") or {} + target = event.get("target") or {} + old_abs = _nc_path_to_abs(source.get("path") or "") + new_abs = _nc_path_to_abs(target.get("path") or "") + if old_abs and Path(old_abs).suffix.lower() in supported: + await handle_file_deletion(old_abs) + if new_abs and Path(new_abs).suffix.lower() in supported: + parent_dir = str(Path(new_abs).parent) + source_root_id = await _source_root_for(parent_dir) + if source_root_id is not None: + scan_folder.delay(parent_dir, source_root_id) + logger.info("nc-webhook renamed: %s -> %s", old_abs, new_abs) + return {"status": "applied", "action": "rename", "from": old_abs, "to": new_abs} + + # Shouldn't reach here — classify() already filtered. + return {"status": "ignored"} diff --git a/backend/scripts/register_nc_webhooks.py b/backend/scripts/register_nc_webhooks.py new file mode 100644 index 0000000..ca71ed8 --- /dev/null +++ b/backend/scripts/register_nc_webhooks.py @@ -0,0 +1,173 @@ +"""Register (or re-register) the four file-event webhooks against the +Nextcloud webhook_listeners app, pointing them at mule's internal +receiver. + +Idempotent: deletes any existing webhooks whose URI matches the target +mule URL before posting fresh ones. Run after a config change (target +URL, secret) or after the NC stack is rebuilt fresh. + + docker exec mulita-backend python -m scripts.register_nc_webhooks + +Reads: +- NEXTCLOUD_BASE_URL — already set for the DAV client +- NEXTCLOUD_WEBHOOK_TARGET — http URL of mule's webhook endpoint + (default: http://192.168.8.136:8001/api/v1/internal/nc-webhook) +- NEXTCLOUD_WEBHOOK_SECRET — shared bearer secret; must match the + backend env var of the same name + +Registers as the first mule user with `is_admin=true` (or, failing +that, the first user with NC credentials configured). The OCS endpoint +itself only requires basic auth as that NC user. +""" +from __future__ import annotations + +import asyncio +import logging +import os +import sys +from typing import Any + +import httpx +from sqlalchemy import select + +from app.database import AsyncSessionLocal +from app.models.user import User +from app.services.secrets import decrypt + +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s %(levelname)s %(name)s: %(message)s", +) +logger = logging.getLogger("register_nc_webhooks") + +DEFAULT_TARGET = "http://192.168.8.136:8001/api/v1/internal/nc-webhook" + +EVENTS = [ + "OCP\\Files\\Events\\Node\\NodeCreatedEvent", + "OCP\\Files\\Events\\Node\\NodeWrittenEvent", + "OCP\\Files\\Events\\Node\\NodeDeletedEvent", + "OCP\\Files\\Events\\Node\\NodeRenamedEvent", +] + + +async def _pick_admin() -> User | None: + """Pick a mule user we can use to authenticate against NC's OCS + API. Prefer admin role, fall back to any user with NC creds set.""" + async with AsyncSessionLocal() as s: + # First try admins. + r = await s.execute( + select(User).where( + User.role == "admin", + User.nextcloud_app_password_enc.is_not(None), + ) + ) + u = r.scalars().first() + if u: + return u + # Fall back to any user with creds. + r = await s.execute( + select(User).where(User.nextcloud_app_password_enc.is_not(None)) + ) + return r.scalars().first() + + +def _ocs(base: str, path: str) -> str: + return f"{base.rstrip('/')}/ocs/v2.php/apps/webhook_listeners/api/v1{path}" + + +def _ocs_headers() -> dict[str, str]: + # OCS-APIRequest header is mandatory for OCS endpoints. + return { + "OCS-APIRequest": "true", + "Accept": "application/json", + } + + +def _list_existing(c: httpx.Client, base: str) -> list[dict[str, Any]]: + r = c.get(_ocs(base, "/webhooks"), headers=_ocs_headers()) + r.raise_for_status() + body = r.json() + return body.get("ocs", {}).get("data", []) or [] + + +def _delete(c: httpx.Client, base: str, webhook_id: str) -> None: + r = c.delete(_ocs(base, f"/webhooks/{webhook_id}"), headers=_ocs_headers()) + if r.status_code not in (200, 204): + logger.warning( + "delete webhook %s returned %s: %s", + webhook_id, r.status_code, r.text[:200], + ) + + +def _register( + c: httpx.Client, + base: str, + target: str, + secret: str, + event_class: str, +) -> dict[str, Any]: + body = { + "uri": target, + "httpMethod": "POST", + "event": event_class, + "authMethod": "header", + "authData": {"Authorization": f"Bearer {secret}"}, + "headers": {"Content-Type": "application/json"}, + } + r = c.post(_ocs(base, "/webhooks"), headers=_ocs_headers(), json=body) + r.raise_for_status() + return r.json().get("ocs", {}).get("data", {}) + + +async def main() -> int: + base = os.environ.get("NEXTCLOUD_BASE_URL", "").rstrip("/") + target = os.environ.get("NEXTCLOUD_WEBHOOK_TARGET", DEFAULT_TARGET) + secret = os.environ.get("NEXTCLOUD_WEBHOOK_SECRET", "") + if not base: + logger.error("NEXTCLOUD_BASE_URL is not set") + return 2 + if not secret: + logger.error("NEXTCLOUD_WEBHOOK_SECRET is not set") + return 2 + + user = await _pick_admin() + if user is None: + logger.error( + "no mule user has nextcloud_app_password_enc set; can't auth to OCS" + ) + return 2 + + nc_user = user.nextcloud_username or user.username + pw = decrypt(user.nextcloud_app_password_enc) + if not nc_user or not pw: + logger.error("user %s has incomplete NC credentials", user.username) + return 2 + + auth = httpx.BasicAuth(nc_user, pw) + with httpx.Client(timeout=30, auth=auth, follow_redirects=False) as c: + existing = _list_existing(c, base) + logger.info("found %d existing webhook(s)", len(existing)) + + # Delete any pointing at the same target URI — idempotent rerun. + for w in existing: + if w.get("uri") == target: + logger.info( + "removing existing webhook id=%s event=%s", + w.get("id"), w.get("event"), + ) + _delete(c, base, str(w["id"])) + + # Register fresh. + for event_class in EVENTS: + data = _register(c, base, target, secret, event_class) + logger.info( + "registered: id=%s event=%s", + data.get("id"), event_class, + ) + + logger.info("done — %d webhooks registered against %s", len(EVENTS), target) + return 0 + + +if __name__ == "__main__": + sys.exit(asyncio.run(main())) diff --git a/docker-compose.yml b/docker-compose.yml index 0ccf5a6..0a6db08 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -97,6 +97,7 @@ services: # but mutating endpoints fail with a clear error. - NEXTCLOUD_USERS_ROOT=${NEXTCLOUD_USERS_ROOT:-/nextcloud-users} - NEXTCLOUD_BASE_URL=${NEXTCLOUD_BASE_URL:-} + - NEXTCLOUD_WEBHOOK_SECRET=${NEXTCLOUD_WEBHOOK_SECRET:-} - LOG_LEVEL=${LOG_LEVEL:-INFO} - TZ=${TZ:-UTC} depends_on: @@ -160,6 +161,7 @@ services: - PHOTO_DIRS=/photos - NEXTCLOUD_USERS_ROOT=${NEXTCLOUD_USERS_ROOT:-/nextcloud-users} - NEXTCLOUD_BASE_URL=${NEXTCLOUD_BASE_URL:-} + - NEXTCLOUD_WEBHOOK_SECRET=${NEXTCLOUD_WEBHOOK_SECRET:-} - LOG_LEVEL=${LOG_LEVEL:-INFO} - TZ=${TZ:-UTC} # NullPool — see app/database.py for rationale. @@ -215,6 +217,7 @@ services: - PHOTO_DIRS=/photos - NEXTCLOUD_USERS_ROOT=${NEXTCLOUD_USERS_ROOT:-/nextcloud-users} - NEXTCLOUD_BASE_URL=${NEXTCLOUD_BASE_URL:-} + - NEXTCLOUD_WEBHOOK_SECRET=${NEXTCLOUD_WEBHOOK_SECRET:-} - LOG_LEVEL=${LOG_LEVEL:-INFO} - TZ=${TZ:-UTC} - MULITA_CELERY_WORKER=1 @@ -258,6 +261,7 @@ services: - PHOTO_DIRS=/photos - NEXTCLOUD_USERS_ROOT=${NEXTCLOUD_USERS_ROOT:-/nextcloud-users} - NEXTCLOUD_BASE_URL=${NEXTCLOUD_BASE_URL:-} + - NEXTCLOUD_WEBHOOK_SECRET=${NEXTCLOUD_WEBHOOK_SECRET:-} - LOG_LEVEL=${LOG_LEVEL:-INFO} - TZ=${TZ:-UTC} - MULITA_CELERY_WORKER=1