diff --git a/shelfmark/release_sources/audiobookbay/handler.py b/shelfmark/release_sources/audiobookbay/handler.py index e4fbf70b..723fbfcc 100644 --- a/shelfmark/release_sources/audiobookbay/handler.py +++ b/shelfmark/release_sources/audiobookbay/handler.py @@ -24,6 +24,8 @@ if TYPE_CHECKING: from shelfmark.core.models import DownloadTask logger = setup_logger(__name__) +DEFAULT_ABB_HOSTNAME = "audiobookbay.lu" +ALLOWED_DETAIL_URL_SCHEMES = {"https"} def _resolve_configured_hostname() -> str: @@ -32,6 +34,23 @@ def _resolve_configured_hostname() -> str: return normalize_hostname(configured_hostname if isinstance(configured_hostname, str) else "") +def _resolve_allowed_detail_hostname() -> str: + """Return the ABB hostname allowed for queued detail URLs.""" + return _resolve_configured_hostname() or DEFAULT_ABB_HOSTNAME + + +def _detail_url_matches_host(detail_url: str, hostname: str) -> bool: + """Return True when a detail URL uses the allowed ABB scheme and host.""" + parsed = urlparse(detail_url) + detail_hostname = normalize_hostname(parsed.hostname) + allowed_hostname = normalize_hostname(hostname).lower().rstrip(".") + return ( + parsed.scheme.lower() in ALLOWED_DETAIL_URL_SCHEMES + and bool(detail_hostname) + and detail_hostname.lower().rstrip(".") == allowed_hostname + ) + + @register_handler("audiobookbay") class AudiobookBayHandler(ExternalClientHandler): """Handler for AudiobookBay downloads via configured torrent client.""" @@ -69,9 +88,14 @@ class AudiobookBayHandler(ExternalClientHandler): logger.warning("Missing details URL for AudiobookBay task: %s", task.task_id) return None - hostname = _resolve_configured_hostname() - if not hostname: - hostname = normalize_hostname(urlparse(detail_url).hostname) + hostname = _resolve_allowed_detail_hostname() + if not _detail_url_matches_host(detail_url, hostname): + status_callback("error", "Invalid AudiobookBay details URL") + logger.warning( + "Rejected AudiobookBay details URL with invalid scheme or host: %s", + detail_url, + ) + return None status_callback("resolving", "Extracting magnet link") magnet_link = scraper.extract_magnet_link(detail_url, hostname) diff --git a/tests/audiobookbay/test_handler.py b/tests/audiobookbay/test_handler.py index 27af8b8e..8fdd1afe 100644 --- a/tests/audiobookbay/test_handler.py +++ b/tests/audiobookbay/test_handler.py @@ -120,6 +120,83 @@ class TestAudiobookBayHandlerDownload: ) assert "resolving" in recorder.statuses + @patch("shelfmark.release_sources.audiobookbay.handler.scraper.extract_magnet_link") + @patch("shelfmark.release_sources.audiobookbay.handler.get_client") + @patch("shelfmark.release_sources.audiobookbay.handler.config.get") + def test_download_rejects_source_url_host_mismatch( + self, mock_config_get, mock_get_client, mock_extract_magnet + ): + """Test hostile detail URLs are rejected before page fetch.""" + mock_config_get.side_effect = lambda key, default="": ( + "audiobookbay.lu" if key == "ABB_HOSTNAME" else default + ) + + handler = AudiobookBayHandler() + task = DownloadTask( + task_id="35f56a3e5734bfa69c3169ee8e605a60", + source="audiobookbay", + title="Test Book", + content_type="audiobook", + source_url="https://169.254.169.254/latest/meta-data/", + ) + cancel_flag = Event() + recorder = ProgressRecorder() + + result = handler.download( + task=task, + cancel_flag=cancel_flag, + progress_callback=recorder.progress_callback, + status_callback=recorder.status_callback, + ) + + assert result is None + assert recorder.last_status == "error" + assert "details url" in recorder.last_message.lower() + mock_extract_magnet.assert_not_called() + mock_get_client.assert_not_called() + + @patch("shelfmark.release_sources.audiobookbay.handler.scraper.extract_magnet_link") + @patch("shelfmark.release_sources.audiobookbay.handler.get_client") + @patch("shelfmark.release_sources.audiobookbay.handler.config.get") + def test_download_allows_configured_source_url_host( + self, mock_config_get, mock_get_client, mock_extract_magnet + ): + """Test configured ABB host remains allowed for queued release URLs.""" + mock_config_get.side_effect = lambda key, default="": ( + "https://audiobookbay.lu/" if key == "ABB_HOSTNAME" else default + ) + mock_extract_magnet.return_value = "magnet:?xt=urn:btih:abc123" + + mock_client = MagicMock() + mock_client.name = "qbittorrent" + mock_client.find_existing.return_value = None + mock_client.add_download.return_value = "download_id_123" + mock_get_client.return_value = mock_client + + handler = AudiobookBayHandler() + task = DownloadTask( + task_id="35f56a3e5734bfa69c3169ee8e605a60", + source="audiobookbay", + title="Test Book", + content_type="audiobook", + source_url="https://audiobookbay.lu/abss/test-book/", + ) + cancel_flag = Event() + recorder = ProgressRecorder() + with patch.object(AudiobookBayHandler, "_poll_and_complete", return_value=None): + result = handler.download( + task=task, + cancel_flag=cancel_flag, + progress_callback=recorder.progress_callback, + status_callback=recorder.status_callback, + ) + + assert result is None + mock_extract_magnet.assert_called_once_with( + "https://audiobookbay.lu/abss/test-book/", "audiobookbay.lu" + ) + mock_client.add_download.assert_called_once() + @patch("shelfmark.release_sources.audiobookbay.handler.scraper.extract_magnet_link") @patch("shelfmark.release_sources.audiobookbay.handler.get_client") def test_download_existing_complete(self, mock_get_client, mock_extract_magnet):