mule-image/backend/scripts/register_nc_webhooks.py

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