From aa8a74ba67da1d28684769b47197dd4577cff104 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Joakim=20Hells=C3=A9n?= Date: Fri, 10 Apr 2026 21:14:01 +0200 Subject: [PATCH] Validate webhook URLs on addition and modification; enhance tests for invalid URL handling --- discord_rss_bot/main.py | 45 +++++++++++++++++++++++++------------ tests/test_main.py | 50 +++++++++++++++++++++++++++++++++++++++++ 2 files changed, 81 insertions(+), 14 deletions(-) diff --git a/discord_rss_bot/main.py b/discord_rss_bot/main.py index 16fcbe1..961c70e 100644 --- a/discord_rss_bot/main.py +++ b/discord_rss_bot/main.py @@ -221,11 +221,15 @@ async def post_add_webhook( # Example: [{"name": "webhook_name", "url": "webhook_url"}] webhooks = cast("list[dict[str, str]]", webhooks) + clean_webhook_url: str = webhook_url.strip() + if not is_url_valid(clean_webhook_url): + raise HTTPException(status_code=400, detail="Invalid webhook URL") + # Only add the webhook if it doesn't already exist. stripped_webhook_name = webhook_name.strip() if all(webhook["name"] != stripped_webhook_name for webhook in webhooks): # Add the new webhook to the list of webhooks. - webhooks.append({"name": webhook_name.strip(), "url": webhook_url.strip()}) + webhooks.append({"name": webhook_name.strip(), "url": clean_webhook_url}) reader.set_tag((), "webhooks", webhooks) # pyright: ignore[reportArgumentType] @@ -1368,20 +1372,30 @@ def get_data_from_hook_url(hook_name: str, hook_url: str) -> WebhookInfo: Returns: WebhookInfo: The webhook username, avatar, guild id, etc. """ - our_hook: WebhookInfo = WebhookInfo(custom_name=hook_name, url=hook_url) + clean_hook_url: str = hook_url.strip() + our_hook: WebhookInfo = WebhookInfo(custom_name=hook_name, url=clean_hook_url) - if hook_url: - response: Response = httpx.get(hook_url) - if response.is_success: - webhook_json = json.loads(response.text) - our_hook.webhook_type = webhook_json["type"] or None - our_hook.webhook_id = webhook_json["id"] or None - our_hook.name = webhook_json["name"] or None - our_hook.avatar = webhook_json["avatar"] or None - our_hook.channel_id = webhook_json["channel_id"] or None - our_hook.guild_id = webhook_json["guild_id"] or None - our_hook.token = webhook_json["token"] or None - our_hook.avatar_mod = int(webhook_json["channel_id"] or 0) % 5 + # Keep /webhooks usable even if a malformed webhook URL was saved. + if not clean_hook_url or not is_url_valid(clean_hook_url): + logger.warning("Skipping webhook metadata fetch for invalid URL: %s", clean_hook_url) + return our_hook + + try: + response: Response = httpx.get(clean_hook_url, timeout=10.0) + except httpx.HTTPError as e: + logger.warning("Failed to fetch webhook metadata for %s: %s", clean_hook_url, e) + return our_hook + + if response.is_success: + webhook_json = json.loads(response.text) + our_hook.webhook_type = webhook_json["type"] or None + our_hook.webhook_id = webhook_json["id"] or None + our_hook.name = webhook_json["name"] or None + our_hook.avatar = webhook_json["avatar"] or None + our_hook.channel_id = webhook_json["channel_id"] or None + our_hook.guild_id = webhook_json["guild_id"] or None + our_hook.token = webhook_json["token"] or None + our_hook.avatar_mod = int(webhook_json["channel_id"] or 0) % 5 return our_hook @@ -1717,6 +1731,9 @@ def modify_webhook( webhooks = cast("list[dict[str, str]]", webhooks) old_hook_clean: str = old_hook.strip() new_hook_clean: str = new_hook.strip() + if not is_url_valid(new_hook_clean): + raise HTTPException(status_code=400, detail="Invalid webhook URL") + webhook_modified: bool = False for hook in webhooks: diff --git a/tests/test_main.py b/tests/test_main.py index 86a8de7..d680a26 100644 --- a/tests/test_main.py +++ b/tests/test_main.py @@ -89,6 +89,56 @@ def test_add_webhook() -> None: assert webhook_name in response.text, f"Webhook not found in /webhooks: {response.text}" +def test_add_webhook_rejects_invalid_url() -> None: + """Adding a webhook with a non-URL value should fail validation.""" + response: Response = client.post( + url="/add_webhook", + data={"webhook_name": "Invalid URL Hook", "webhook_url": "not-a-url"}, + ) + + assert response.status_code == 400, f"Expected invalid webhook URL to be rejected: {response.text}" + assert "Invalid webhook URL" in response.text + + +def test_add_webhook_allows_valid_url_after_invalid_attempt() -> None: + """A rejected invalid webhook URL should not prevent a later valid add.""" + response: Response = client.post(url="/delete_webhook", data={"webhook_url": webhook_url}) + assert response.status_code == 200, f"Failed to delete webhook: {response.text}" + + response = client.post( + url="/add_webhook", + data={"webhook_name": "Invalid URL Hook", "webhook_url": "not-a-url"}, + ) + assert response.status_code == 400, f"Expected invalid webhook URL to be rejected: {response.text}" + assert "Invalid webhook URL" in response.text + + response = client.post( + url="/add_webhook", + data={"webhook_name": webhook_name, "webhook_url": webhook_url}, + ) + assert response.status_code == 200, f"Failed to add webhook after invalid attempt: {response.text}" + + response = client.get(url="/webhooks") + assert response.status_code == 200, f"Failed to get /webhooks: {response.text}" + assert webhook_name in response.text, f"Webhook not found in /webhooks: {response.text}" + + response = client.post(url="/delete_webhook", data={"webhook_url": webhook_url}) + assert response.status_code == 200, f"Failed to delete webhook: {response.text}" + + +def test_webhooks_page_handles_invalid_stored_webhook_url() -> None: + """/webhooks should render even if a malformed webhook URL is present in storage.""" + reader: Reader = get_reader_dependency() + malformed_webhook_name = "Malformed hook" + malformed_webhook_url = "definitely-not-a-url" + + reader.set_tag((), "webhooks", [{"name": malformed_webhook_name, "url": malformed_webhook_url}]) # pyright: ignore[reportArgumentType] + response: Response = client.get(url="/webhooks") + + assert response.status_code == 200, f"/webhooks should not crash for malformed URLs: {response.text}" + assert malformed_webhook_name in response.text + + def test_create_feed() -> None: """Test the /create_feed page.""" # Ensure webhook exists for this test regardless of test order.