"""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()))