Validate webhook URLs on addition and modification; enhance tests for invalid URL handling

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,10 +1372,20 @@ 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)
+
+    # 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 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
@@ -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:

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.