alexei.dolgolyov
0eb899afb9
Notifications: - Add shared http_base, redact, and SSRF hardening modules - Refactor dispatcher, queue, receiver and per-provider clients (telegram, discord, email, matrix, ntfy, slack, webhook) to use the shared base, with bounded queue and redacted error logs - Tests for ssrf, redact, http_base, queue bounds, dispatcher aggregation, telegram media partition, email and matrix clients Frontend: - Settings: log level / log format selectors now use IconGridSelect with per-option icons and i18n descriptions - Minor providers page and entity-cache store updates Tooling: - Document code-review-graph MCP usage in CLAUDE.md - Ignore .code-review-graph/, register .mcp.json
85 lines
2.8 KiB
Python
85 lines
2.8 KiB
Python
"""Matrix client validation: room_id format and quoting."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import aiohttp
|
|
import pytest
|
|
from aioresponses import aioresponses
|
|
|
|
from notify_bridge_core.notifications.matrix.client import MatrixClient
|
|
|
|
|
|
HOMESERVER = "https://matrix.example.com"
|
|
TOKEN = "secret-bearer-token-1234567890"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rejects_path_injection_room_id() -> None:
|
|
async with aiohttp.ClientSession() as sess:
|
|
client = MatrixClient(sess, HOMESERVER, TOKEN)
|
|
result = await client.send_message("!abc:host/../../etc/passwd", "hi")
|
|
assert result["success"] is False
|
|
assert "room_id" in result["error"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rejects_empty_room_id() -> None:
|
|
async with aiohttp.ClientSession() as sess:
|
|
client = MatrixClient(sess, HOMESERVER, TOKEN)
|
|
result = await client.send_message("", "hi")
|
|
assert result["success"] is False
|
|
assert "room_id" in result["error"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rejects_unicode_control_chars_in_room_id() -> None:
|
|
async with aiohttp.ClientSession() as sess:
|
|
client = MatrixClient(sess, HOMESERVER, TOKEN)
|
|
result = await client.send_message("!abc\x00:host", "hi")
|
|
assert result["success"] is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_url_encodes_room_id_special_chars() -> None:
|
|
"""``!`` and ``:`` must reach the server URL-encoded."""
|
|
captured: list[str] = []
|
|
|
|
with aioresponses() as mocked:
|
|
# Match any PUT under the rooms path; capture the URL we got.
|
|
mocked.put(
|
|
"https://matrix.example.com/_matrix/client/v3/rooms/%21abc%3Ahost.example/send/m.room.message",
|
|
status=200, body='{}', repeat=True,
|
|
)
|
|
# aioresponses doesn't expose URL templates well, so use a regex mock.
|
|
import re
|
|
mocked.put(
|
|
re.compile(r"https://matrix\.example\.com/_matrix/client/v3/rooms/[^/]+/send/m\.room\.message/.*"),
|
|
status=200, body='{}', repeat=True,
|
|
)
|
|
|
|
async with aiohttp.ClientSession() as sess:
|
|
client = MatrixClient(sess, HOMESERVER, TOKEN)
|
|
result = await client.send_message("!abc:host.example", "hi")
|
|
|
|
assert result["success"] is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_redacts_bearer_in_error() -> None:
|
|
"""A 4xx response body must not echo the Authorization Bearer back to caller."""
|
|
import re
|
|
with aioresponses() as mocked:
|
|
mocked.put(
|
|
re.compile(r".*"),
|
|
status=403,
|
|
body='{"errcode": "M_FORBIDDEN", "Authorization": "Bearer ' + TOKEN + '"}',
|
|
repeat=True,
|
|
)
|
|
|
|
async with aiohttp.ClientSession() as sess:
|
|
client = MatrixClient(sess, HOMESERVER, TOKEN)
|
|
result = await client.send_message("!abc:host.example", "hi")
|
|
|
|
assert result["success"] is False
|
|
assert TOKEN not in result["error"]
|