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
54 lines
1.6 KiB
Python
54 lines
1.6 KiB
Python
"""HttpProviderClient + safe_headers tests."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from notify_bridge_core.notifications.http_base import safe_headers
|
|
|
|
|
|
class TestSafeHeaders:
|
|
def test_drops_hop_by_hop(self) -> None:
|
|
out = safe_headers({
|
|
"X-Custom": "ok",
|
|
"Host": "evil.example.com",
|
|
"Content-Length": "999",
|
|
"Transfer-Encoding": "chunked",
|
|
"Connection": "close",
|
|
})
|
|
assert out == {"X-Custom": "ok"}
|
|
|
|
def test_rejects_crlf_in_value(self) -> None:
|
|
out = safe_headers({
|
|
"X-Custom": "ok",
|
|
"X-Bad": "value\r\nInjected: yes",
|
|
})
|
|
assert "X-Custom" in out
|
|
assert "X-Bad" not in out
|
|
|
|
def test_rejects_crlf_in_name(self) -> None:
|
|
out = safe_headers({
|
|
"X-Custom": "ok",
|
|
"X-Bad\r\nInject": "value",
|
|
})
|
|
assert out == {"X-Custom": "ok"}
|
|
|
|
def test_empty_input(self) -> None:
|
|
assert safe_headers(None) == {}
|
|
assert safe_headers({}) == {}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_http_base_returns_safe_error_on_invalid_url() -> None:
|
|
"""An obviously-bad URL must not panic or leak the URL verbatim."""
|
|
import aiohttp
|
|
|
|
from notify_bridge_core.notifications.http_base import HttpProviderClient
|
|
|
|
async with aiohttp.ClientSession() as sess:
|
|
client = HttpProviderClient(sess, provider_name="test")
|
|
# file:// is rejected by the SSRF guard before any HTTP call.
|
|
result = await client.request("POST", "file:///etc/passwd", json={})
|
|
assert result["success"] is False
|
|
assert "Unsafe URL" in result["error"]
|