Cortex-Inara/cortex/tests/test_webhooks.py

"""
Webhook auth tests — NC Talk HMAC, Google Chat JWT.

These tests verify that auth is enforced, not that full LLM responses work.

Architecture note: channel config (secrets, audience) lives in per-user channels.json,
not in settings. Tests mock get_user_channels() rather than patching settings fields.
Endpoints are per-user: /webhook/nextcloud/{username} and /channels/google-chat/{username}.
"""
import hashlib
import hmac
import json
import pytest
from unittest.mock import AsyncMock, patch


# ---------------------------------------------------------------------------
# Nextcloud Talk
# ---------------------------------------------------------------------------

_NC_SECRET = "test-bot-secret-12345"

_VALID_NC_PAYLOAD = {
    "type": "Create",
    "actor": {"type": "users", "id": "testuser", "name": "Test User"},
    "object": {
        "type": "Note",
        "content": json.dumps({"message": "Hello Inara"}),
    },
    "target": {"id": "abc123token"},
}

_NCT_CHANNELS = {
    "nextcloud": {
        "bot_secret":        _NC_SECRET,
        "notification_room": "abc123token",
        "url":               "https://nc.example.com",
    }
}


def _nc_headers(body: bytes, secret: str) -> dict:
    random_str = "abc123"
    sig = hmac.new(
        secret.encode(),
        (random_str + body.decode("utf-8")).encode(),
        hashlib.sha256,
    ).hexdigest()
    return {
        "X-Nextcloud-Talk-Random": random_str,
        "X-Nextcloud-Talk-Signature": sig,
    }


@pytest.mark.anyio
async def test_nct_valid_signature(client, mock_llm):
    body = json.dumps(_VALID_NC_PAYLOAD).encode()
    with patch("routers.nextcloud_talk.get_user_channels", return_value=_NCT_CHANNELS):
        with patch("routers.nextcloud_talk._send_reply", new_callable=AsyncMock):
            headers = _nc_headers(body, _NC_SECRET)
            r = await client.post(
                "/webhook/nextcloud/scott",
                content=body,
                headers={**headers, "Content-Type": "application/json"},
            )
    assert r.status_code == 200


@pytest.mark.anyio
async def test_nct_wrong_signature(client):
    body = json.dumps(_VALID_NC_PAYLOAD).encode()
    with patch("routers.nextcloud_talk.get_user_channels", return_value=_NCT_CHANNELS):
        r = await client.post(
            "/webhook/nextcloud/scott",
            content=body,
            headers={
                "Content-Type": "application/json",
                "X-Nextcloud-Talk-Random": "abc123",
                "X-Nextcloud-Talk-Signature": "badsignature",
            },
        )
    assert r.status_code == 401


@pytest.mark.anyio
async def test_nct_missing_signature(client):
    body = json.dumps(_VALID_NC_PAYLOAD).encode()
    with patch("routers.nextcloud_talk.get_user_channels", return_value=_NCT_CHANNELS):
        r = await client.post(
            "/webhook/nextcloud/scott",
            content=body,
            headers={"Content-Type": "application/json"},
        )
    assert r.status_code == 401


@pytest.mark.anyio
async def test_nct_no_secret_configured(client):
    """Service should return 500 if bot_secret is missing, not process the message."""
    body = json.dumps(_VALID_NC_PAYLOAD).encode()
    # cfg must be non-empty (truthy) to get past the 404 guard; missing bot_secret → 500
    empty_cfg = {"nextcloud": {"url": "https://nc.example.com"}}
    with patch("routers.nextcloud_talk.get_user_channels", return_value=empty_cfg):
        r = await client.post(
            "/webhook/nextcloud/scott",
            content=body,
            headers={"Content-Type": "application/json"},
        )
    assert r.status_code == 500


@pytest.mark.anyio
async def test_nct_bot_message_ignored(client):
    """Messages from other bots should be silently ignored (not processed)."""
    payload = {**_VALID_NC_PAYLOAD, "actor": {"type": "bots", "id": "otherbot", "name": "Bot"}}
    body = json.dumps(payload).encode()
    with patch("routers.nextcloud_talk.get_user_channels", return_value=_NCT_CHANNELS):
        headers = _nc_headers(body, _NC_SECRET)
        r = await client.post(
            "/webhook/nextcloud/scott",
            content=body,
            headers={**headers, "Content-Type": "application/json"},
        )
    assert r.status_code == 200


# ---------------------------------------------------------------------------
# Google Chat
# ---------------------------------------------------------------------------

_GCHAT_PAYLOAD = {
    "chat": {
        "messagePayload": {
            "message": {"text": "Hello", "argumentText": "Hello"},
            "space": {"name": "spaces/test123", "type": "ROOM"},
        },
        "user": {"displayName": "Test User"},
    }
}

_GCHAT_CHANNELS_NO_AUDIENCE = {
    # cfg must be non-empty (truthy) to pass the 404 guard; no audience → JWT skipped
    "google_chat": {"persona": "inara"}
}

_GCHAT_CHANNELS_WITH_AUDIENCE = {
    "google_chat": {"audience": "123456789"}
}


@pytest.mark.anyio
async def test_gchat_no_audience_configured(client, mock_llm):
    """When audience is not set, JWT check is skipped (current behaviour — documented bypass)."""
    with patch("routers.google_chat.get_user_channels", return_value=_GCHAT_CHANNELS_NO_AUDIENCE):
        r = await client.post("/channels/google-chat/scott", json=_GCHAT_PAYLOAD)
    assert r.status_code == 200


@pytest.mark.anyio
async def test_gchat_missing_token_with_audience(client):
    """When audience IS configured, requests without a token must be rejected."""
    with patch("routers.google_chat.get_user_channels", return_value=_GCHAT_CHANNELS_WITH_AUDIENCE):
        r = await client.post("/channels/google-chat/scott", json=_GCHAT_PAYLOAD)
    assert r.status_code == 401


@pytest.mark.anyio
async def test_gchat_invalid_token_with_audience(client):
    """A fake token should fail JWT verification."""
    payload_with_token = {
        **_GCHAT_PAYLOAD,
        "authorizationEventObject": {"systemIdToken": "not.a.valid.jwt"},
    }
    with patch("routers.google_chat.get_user_channels", return_value=_GCHAT_CHANNELS_WITH_AUDIENCE):
        r = await client.post("/channels/google-chat/scott", json=payload_with_token)
    assert r.status_code == 401


@pytest.mark.anyio
async def test_gchat_added_to_space(client, mock_llm):
    """Bot added to a space — should return a greeting, no auth when audience empty."""
    payload = {"chat": {"addedToSpacePayload": {"space": {"type": "ROOM"}}}}
    with patch("routers.google_chat.get_user_channels", return_value=_GCHAT_CHANNELS_NO_AUDIENCE):
        r = await client.post("/channels/google-chat/scott", json=payload)
    assert r.status_code == 200
    assert "hostAppDataAction" in r.json()