From d3f57aa8be2144d826954e238bdc7e9f4d5fa052 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Joakim=20Hells=C3=A9n?= Date: Thu, 23 Jul 2026 22:49:38 +0200 Subject: [PATCH] Remove redundant url unquoting in route handlers --- discord_rss_bot/main.py | 56 +++++++++++-------- tests/test_main.py | 119 +++++++++++++++++++++++++++++++++------- 2 files changed, 131 insertions(+), 44 deletions(-) diff --git a/discord_rss_bot/main.py b/discord_rss_bot/main.py index c053f76..ced82e6 100644 --- a/discord_rss_bot/main.py +++ b/discord_rss_bot/main.py @@ -709,7 +709,7 @@ async def post_attach_feed_webhook( Raises: HTTPException: If feed or webhook cannot be found. """ - clean_feed_url: str = urllib.parse.unquote(feed_url.strip()) + clean_feed_url: str = feed_url.strip() selected_webhook_name: str = webhook_dropdown.strip() try: @@ -834,7 +834,8 @@ async def get_whitelist( HTMLResponse: The whitelist page. """ clean_feed_url: str = feed_url.strip() - feed: Feed = reader.get_feed(urllib.parse.unquote(clean_feed_url)) + feed: Feed = reader.get_feed(clean_feed_url) + context = { "request": request, "feed": feed, @@ -876,7 +877,7 @@ async def get_whitelist_preview( Returns: HTMLResponse: Rendered filter preview fragment. """ - clean_feed_url: str = urllib.parse.unquote(feed_url.strip()) + clean_feed_url: str = feed_url.strip() feed: Feed = reader.get_feed(clean_feed_url) form_values: dict[str, str] = { @@ -963,7 +964,8 @@ async def get_blacklist( Returns: HTMLResponse: The blacklist page. """ - feed: Feed = reader.get_feed(urllib.parse.unquote(feed_url)) + clean_feed_url: str = feed_url.strip() + feed: Feed = reader.get_feed(clean_feed_url) context = { "request": request, @@ -1006,8 +1008,9 @@ async def get_blacklist_preview( Returns: HTMLResponse: Rendered filter preview fragment. """ - clean_feed_url: str = urllib.parse.unquote(feed_url.strip()) + clean_feed_url: str = feed_url.strip() feed: Feed = reader.get_feed(clean_feed_url) + form_values: dict[str, str] = { "blacklist_title": blacklist_title, "blacklist_summary": blacklist_summary, @@ -1384,7 +1387,7 @@ async def post_set_custom( reader.set_tag(feed_url, "message_avatar_url", typing.cast("JSONType", message_avatar_url.strip())) clean_feed_url: str = feed_url.strip() - feed: Feed = reader.get_feed(urllib.parse.unquote(clean_feed_url)) + feed: Feed = reader.get_feed(clean_feed_url) stored_custom_message: str = get_custom_message(reader, feed) if our_custom_message != stored_custom_message: @@ -1410,7 +1413,8 @@ async def get_custom( Returns: HTMLResponse: The custom message page. """ - feed: Feed = reader.get_feed(urllib.parse.unquote(feed_url.strip())) + clean_feed_url: str = feed_url.strip() + feed: Feed = reader.get_feed(clean_feed_url) # Collect extension variable names for the template. ext_registry: dict[str, type] = get_extension_registry() @@ -1466,7 +1470,8 @@ async def get_embed_page( Returns: HTMLResponse: The custom message page. """ - feed: Feed = reader.get_feed(urllib.parse.unquote(feed_url.strip())) + clean_feed_url: str = feed_url.strip() + feed: Feed = reader.get_feed(clean_feed_url) embed: CustomEmbed = get_embed(reader, feed) @@ -1562,7 +1567,7 @@ async def post_embed( # ruff:ignore[complex-structure, too-many-branches] RedirectResponse: Redirect to the embed page. """ clean_feed_url: str = feed_url.strip() - feed: Feed = reader.get_feed(urllib.parse.unquote(clean_feed_url)) + feed: Feed = reader.get_feed(clean_feed_url) custom_embed: CustomEmbed = get_embed(reader, feed) @@ -1619,7 +1624,9 @@ async def get_extensions( Returns: HTMLResponse: The extensions configuration page. """ - feed: Feed = reader.get_feed(urllib.parse.unquote(feed_url.strip())) + clean_feed_url: str = feed_url.strip() + feed: Feed = reader.get_feed(clean_feed_url) + registry: dict[str, type] = get_extension_registry() enabled: list[str] = get_enabled_extensions(reader, feed.url) @@ -2128,7 +2135,7 @@ async def get_feed( # ruff:ignore[complex-structure, too-many-branches, too-man """ entries_per_page: int = 20 - clean_feed_url: str = urllib.parse.unquote(feed_url.strip()) + clean_feed_url: str = feed_url.strip() try: feed: Feed = reader.get_feed(clean_feed_url) @@ -2557,8 +2564,8 @@ async def get_sent_webhooks( Returns: sent_webhooks.html HTML """ - clean_feed_url: str = urllib.parse.unquote(feed_url.strip()) - clean_webhook_url: str = urllib.parse.unquote(webhook_url.strip()) + clean_feed_url: str = feed_url.strip() + clean_webhook_url: str = webhook_url.strip() records: list[SentWebhookRecord] = get_sent_webhook_records(reader) if clean_feed_url: @@ -2673,12 +2680,13 @@ async def remove_feed( Raises: HTTPException: Feed not found """ + clean_feed_url: str = feed_url.strip() try: - reader.delete_feed(urllib.parse.unquote(feed_url)) + reader.delete_feed(clean_feed_url) except FeedNotFoundError as e: raise HTTPException(status_code=404, detail="Feed not found") from e - commit_state_change(reader, f"Remove feed {urllib.parse.unquote(feed_url)}") + commit_state_change(reader, f"Remove feed {clean_feed_url}") return RedirectResponse(url="/", status_code=303) @@ -2703,7 +2711,7 @@ async def update_feed( HTTPException: If the feed is not found. """ try: - clean_feed_url: str = urllib.parse.unquote(feed_url) + clean_feed_url: str = feed_url.strip() modified_entries: list[tuple[str, str]] = update_feed_and_collect_modified_entries(reader, clean_feed_url) except FeedNotFoundError as e: raise HTTPException(status_code=404, detail="Feed not found") from e @@ -3239,19 +3247,18 @@ async def post_entry( Returns: RedirectResponse: Redirect to the feed page. """ - unquoted_entry_id: str = urllib.parse.unquote(entry_id) - clean_feed_url: str = urllib.parse.unquote(feed_url.strip()) if feed_url else "" + clean_feed_url: str = feed_url.strip() if feed_url else "" # Prefer feed-scoped lookup when feed_url is provided. This avoids ambiguity when # multiple feeds contain entries with the same ID. entry: Entry | None = None if clean_feed_url: entry = next( - (entry for entry in reader.get_entries(feed=clean_feed_url) if entry.id == unquoted_entry_id), + (entry for entry in reader.get_entries(feed=clean_feed_url) if entry.id == entry_id), None, ) else: - entry = next((entry for entry in reader.get_entries() if entry.id == unquoted_entry_id), None) + entry = next((entry for entry in reader.get_entries() if entry.id == entry_id), None) if entry is None: return HTMLResponse(status_code=404, content=f"Entry '{entry_id}' not found.") @@ -3511,8 +3518,9 @@ async def get_webhook_entries_mass_update_preview( Returns: HTMLResponse: Rendered partial template containing summary + preview table. """ - clean_webhook_url: str = urllib.parse.unquote(webhook_url.strip()) + clean_webhook_url: str = webhook_url.strip() all_feeds: list[Feed] = list(reader.get_feeds()) + webhook_feeds: list[Feed] = [ feed for feed in all_feeds if str(reader.get_tag(feed.url, "webhook", "")) == clean_webhook_url ] @@ -3564,7 +3572,7 @@ async def get_webhook_entries( # ruff:ignore[complex-structure, too-many-locals HTTPException: If no feeds are found for this webhook or webhook doesn't exist. """ entries_per_page: int = 20 - clean_webhook_url: str = urllib.parse.unquote(webhook_url.strip()) + clean_webhook_url: str = webhook_url.strip() # Get the webhook name from the webhooks list webhooks: list[dict[str, str]] = cast("list[dict[str, str]]", list(reader.get_tag((), "webhooks", []))) @@ -3656,7 +3664,7 @@ async def get_webhook_entries( # ruff:ignore[complex-structure, too-many-locals "is_show_more_entries_button_visible": is_show_more_entries_button_visible, "total_entries": total_entries, "feeds_count": len(webhook_feeds), - "message": urllib.parse.unquote(message) if message else "", + "message": message, **mass_update_context, } return templates.TemplateResponse(request=request, name="webhook_entries.html", context=context) @@ -3687,7 +3695,7 @@ async def post_bulk_change_feed_urls( # ruff:ignore[complex-structure, too-many Raises: HTTPException: If webhook is missing or replace_from is empty. """ - clean_webhook_url: str = urllib.parse.unquote(webhook_url.strip()) + clean_webhook_url: str = webhook_url.strip() clean_replace_from: str = replace_from.strip() clean_replace_to: str = replace_to.strip() diff --git a/tests/test_main.py b/tests/test_main.py index 967ffd3..c0a44e6 100644 --- a/tests/test_main.py +++ b/tests/test_main.py @@ -71,7 +71,6 @@ def test_search() -> None: feeds: Response = client.get("/") if feed_url in feeds.text: client.post(url="/remove", data={"feed_url": feed_url}) - client.post(url="/remove", data={"feed_url": encoded_feed_url(feed_url)}) # Delete the webhook if it already exists before we run the test. response: Response = client.post(url="/delete_webhook", data={"webhook_url": webhook_url}) @@ -181,7 +180,6 @@ def test_create_feed() -> None: feeds: Response = client.get(url="/") if feed_url in feeds.text: client.post(url="/remove", data={"feed_url": feed_url}) - client.post(url="/remove", data={"feed_url": encoded_feed_url(feed_url)}) # Add the feed. response: Response = client.post(url="/add", data={"feed_url": feed_url, "webhook_dropdown": webhook_name}) @@ -255,7 +253,6 @@ def test_get() -> None: feeds: Response = client.get("/") if feed_url in feeds.text: client.post(url="/remove", data={"feed_url": feed_url}) - client.post(url="/remove", data={"feed_url": encoded_feed_url(feed_url)}) # Add the feed. response: Response = client.post(url="/add", data={"feed_url": feed_url, "webhook_dropdown": webhook_name}) @@ -272,16 +269,16 @@ def test_get() -> None: response: Response = client.get(url="/add_webhook") assert response.status_code == 200, f"/add_webhook failed: {response.text}" - response: Response = client.get(url="/blacklist", params={"feed_url": encoded_feed_url(feed_url)}) + response: Response = client.get(url="/blacklist", params={"feed_url": feed_url}) assert response.status_code == 200, f"/blacklist failed: {response.text}" - response: Response = client.get(url="/custom", params={"feed_url": encoded_feed_url(feed_url)}) + response: Response = client.get(url="/custom", params={"feed_url": feed_url}) assert response.status_code == 200, f"/custom failed: {response.text}" - response: Response = client.get(url="/embed", params={"feed_url": encoded_feed_url(feed_url)}) + response: Response = client.get(url="/embed", params={"feed_url": feed_url}) assert response.status_code == 200, f"/embed failed: {response.text}" - response: Response = client.get(url="/feed", params={"feed_url": encoded_feed_url(feed_url)}) + response: Response = client.get(url="/feed", params={"feed_url": feed_url}) assert response.status_code == 200, f"/feed failed: {response.text}" response: Response = client.get(url="/") @@ -293,7 +290,7 @@ def test_get() -> None: response = client.get(url="/webhook_entries", params={"webhook_url": webhook_url}) assert response.status_code == 200, f"/webhook_entries failed: {response.text}" - response: Response = client.get(url="/whitelist", params={"feed_url": encoded_feed_url(feed_url)}) + response: Response = client.get(url="/whitelist", params={"feed_url": feed_url}) assert response.status_code == 200, f"/whitelist failed: {response.text}" @@ -351,10 +348,98 @@ def test_feed_page_shows_steam_thumbnail_hint_for_steam_feeds() -> None: app.dependency_overrides = {} +@pytest.mark.parametrize( + "encoded_url", + [ + pytest.param( + "https://gazellegames.net/feeds.php?feed=test&name=Halo%3A+Campaign+Evolved", + id="percent-encoded-colon", + ), + pytest.param( + "https://example.com/path%2Fwith%2Fslashes?q=val", + id="percent-encoded-slash", + ), + pytest.param( + "https://example.com/feed?title=A%23B%26C%3D", + id="percent-encoded-special-chars", + ), + pytest.param( + "https://example.com/feed?name=with+plus+signs", + id="literal-plus-signs", + ), + pytest.param( + "https://example.com/feed?filter=%7B%22key%22%3A+%22val%22%7D", + id="percent-encoded-json", + ), + pytest.param( + "https://example.com/feed?redirect=https%3A%2F%2Fother.com%2F", + id="percent-encoded-nested-url", + ), + ], +) +def test_feed_page_with_percent_encoded_url(encoded_url: str) -> None: + """Feed lookups with percent-encoded URLs should not double-decode them. + + FastAPI already decodes query parameters. Previously, get_feed() called + urllib.parse.unquote() on the already-decoded value, which caused %3A to + become : and broke the lookup. This test verifies that all common + percent-encoded characters survive the query parameter round-trip. + """ + + @dataclass(slots=True) + class DummyFeed: + url: str + title: str + updates_enabled: bool = True + last_exception: None = None + added: None = None + last_updated: None = None + last_retrieved: None = None + update_after: None = None + + class StubReader: + def __init__(self) -> None: + self.feed = DummyFeed(url=encoded_url, title="Test Feed") + + def get_feed(self, feed_url: str) -> DummyFeed: + assert feed_url == self.feed.url, f"Expected {self.feed.url!r}, got {feed_url!r}" + return self.feed + + def get_tag(self, _resource: str | DummyFeed, key: str, default: TestTagValue = None) -> TestTagValue: + return { + "webhooks": [], + "delivery_mode": "embed", + "screenshot_layout": "desktop", + "media_gallery_image_limit": 0, + "webhook_text_length_limit": 4000, + "save_sent_webhooks": True, + }.get(key, default) + + def get_entry_counts(self, **_kwargs: TestKwargValue) -> SimpleNamespace: + return SimpleNamespace(total=0) + + def get_entries(self, **_kwargs: TestKwargValue) -> list[Entry]: + return [] + + def get_feed_counts(self, **_kwargs: TestKwargValue) -> SimpleNamespace: + return SimpleNamespace() + + stub = StubReader() + app.dependency_overrides[get_reader_dependency] = lambda: stub + + try: + with patch("discord_rss_bot.main.create_html_for_feed", return_value=""): + response: Response = client.get(url="/feed", params={"feed_url": encoded_url}) + + assert response.status_code == 200, f"/feed failed: {response.text}" + finally: + app.dependency_overrides = {} + + def test_blacklist_page_uses_live_preview_layout() -> None: ensure_preview_feed_exists() - response: Response = client.get(url="/blacklist", params={"feed_url": encoded_feed_url(feed_url)}) + response: Response = client.get(url="/blacklist", params={"feed_url": feed_url}) assert response.status_code == 200, f"/blacklist failed: {response.text}" assert 'hx-get="/blacklist_preview"' in response.text @@ -365,7 +450,7 @@ def test_blacklist_page_uses_live_preview_layout() -> None: def test_whitelist_page_uses_live_preview_layout() -> None: ensure_preview_feed_exists() - response: Response = client.get(url="/whitelist", params={"feed_url": encoded_feed_url(feed_url)}) + response: Response = client.get(url="/whitelist", params={"feed_url": feed_url}) assert response.status_code == 200, f"/whitelist failed: {response.text}" assert 'hx-get="/whitelist_preview"' in response.text @@ -919,7 +1004,6 @@ def test_pause_feed() -> None: feeds: Response = client.get(url="/") if feed_url in feeds.text: client.post(url="/remove", data={"feed_url": feed_url}) - client.post(url="/remove", data={"feed_url": encoded_feed_url(feed_url)}) # Add the feed. response: Response = client.post(url="/add", data={"feed_url": feed_url, "webhook_dropdown": webhook_name}) @@ -955,7 +1039,6 @@ def test_unpause_feed() -> None: feeds: Response = client.get("/") if feed_url in feeds.text: client.post(url="/remove", data={"feed_url": feed_url}) - client.post(url="/remove", data={"feed_url": encoded_feed_url(feed_url)}) # Add the feed. response: Response = client.post(url="/add", data={"feed_url": feed_url, "webhook_dropdown": webhook_name}) @@ -991,7 +1074,6 @@ def test_remove_feed() -> None: feeds: Response = client.get(url="/") if feed_url in feeds.text: client.post(url="/remove", data={"feed_url": feed_url}) - client.post(url="/remove", data={"feed_url": encoded_feed_url(feed_url)}) # Add the feed. response: Response = client.post(url="/add", data={"feed_url": feed_url, "webhook_dropdown": webhook_name}) @@ -1317,7 +1399,7 @@ def test_update_feed_not_found() -> None: nonexistent_feed_url = "https://nonexistent-feed.example.com/rss.xml" # Try to update the non-existent feed - response: Response = client.get(url="/update", params={"feed_url": urllib.parse.quote(nonexistent_feed_url)}) + response: Response = client.get(url="/update", params={"feed_url": nonexistent_feed_url}) # Check that it returns a 404 status code assert response.status_code == 404, f"Expected 404 for non-existent feed, got: {response.status_code}" @@ -1376,15 +1458,13 @@ def test_post_entry_send_to_discord() -> None: entries: list[Entry] = list(reader.get_entries(feed=feed_url, limit=1)) assert entries, "Feed should have at least one entry to send" entry_to_send: main_module.Entry = entries[0] - encoded_id: str = urllib.parse.quote(entry_to_send.id) - no_redirect_client = TestClient(app, follow_redirects=False) # Patch execute_webhook so no real HTTP requests are made to Discord. with patch("discord_rss_bot.feeds.execute_webhook") as mock_execute: response = no_redirect_client.get( url="/post_entry", - params={"entry_id": encoded_id, "feed_url": urllib.parse.quote(feed_url)}, + params={"entry_id": entry_to_send.id, "feed_url": feed_url}, ) assert response.status_code == 303, f"Expected redirect after sending, got {response.status_code}: {response.text}" @@ -1445,7 +1525,7 @@ def test_post_entry_uses_feed_url_to_disambiguate_duplicate_ids() -> None: with patch("discord_rss_bot.main.send_entry_to_discord", side_effect=fake_send_entry_to_discord): response: Response = no_redirect_client.get( url="/post_entry", - params={"entry_id": urllib.parse.quote(shared_id), "feed_url": urllib.parse.quote(feed_b)}, + params={"entry_id": shared_id, "feed_url": feed_b}, ) assert response.status_code == 303, f"Expected redirect after sending, got {response.status_code}" @@ -2075,10 +2155,9 @@ def test_webhook_entries_url_encoding() -> None: assert response.status_code == 200, f"Failed to add feed: {response.text}" # Get webhook_entries with URL-encoded webhook URL - encoded_webhook_url = urllib.parse.quote(webhook_url) response = client.get( url="/webhook_entries", - params={"webhook_url": encoded_webhook_url}, + params={"webhook_url": webhook_url}, ) assert response.status_code == 200, f"Failed to get /webhook_entries with encoded URL: {response.text}"