diff --git a/src/alt_ani_cli/cli.py b/src/alt_ani_cli/cli.py index 6da2dcc..8c480a4 100644 --- a/src/alt_ani_cli/cli.py +++ b/src/alt_ani_cli/cli.py @@ -28,6 +28,7 @@ from alt_ani_cli.health import HealthTransition, ResolverHealth, Signal, classify_failure from alt_ani_cli.models import EmbedURL, EpisodeRow, PlayerEntry, SeriesRef from alt_ani_cli.player import runner as player_runner +from alt_ani_cli.redaction import redact_headers, redact_text from alt_ani_cli.shinden import api as shinden_api from alt_ani_cli.shinden import episode as shinden_episode from alt_ani_cli.shinden import http as shinden_http @@ -196,8 +197,8 @@ def _resolve_embed_with_spinner(client, online_id: str) -> EmbedURL: def _extract_stream(embed: EmbedURL, cookies_file: str | None, cookies_browser: str | None) -> Stream: - # the only place the full embed URL is shown — failure messages use the host only - progress.info(_PROG["embed"].format(url=embed.url)) + # the only place the embed URL is shown — failure messages use the host only + progress.info(_PROG["embed"].format(url=redact_text(embed.url))) return extract.resolve( embed.url, embed.referer, @@ -265,7 +266,7 @@ def _record_extraction_failure( transition = health.record(host, candidate.online_id, signal) _emit_health_transition(transition, category=fields.get("category"), http_status=fields.get("http_status")) diagnostics.resolve_result(host, False, type(exc).__name__, time.monotonic() - start, **fields) - progress.warn(_PROG["player_failed_long"].format(player=repr(candidate.player), number=ep_number, exc=exc)) + progress.warn(_PROG["player_failed_long"].format(player=repr(candidate.player), number=ep_number, exc=redact_text(exc))) @dataclass @@ -463,19 +464,20 @@ def _print_debug(stream: Stream, embed) -> None: from alt_ani_cli.ui.progress import _get con = _get() - con.print(f"\n[bold]Embed URL:[/bold] {embed.url}") - con.print(f"[bold]Direct URL:[/bold] {stream.url}") + con.print(f"\n[bold]Embed URL:[/bold] {redact_text(embed.url)}") + con.print(f"[bold]Direct URL:[/bold] {redact_text(stream.url)}") con.print(f"[bold]Ext:[/bold] {stream.ext}") if stream.headers: t = Table("Header", "Value", title="HTTP headers") - for k, v in stream.headers.items(): + for k, v in redact_headers(stream.headers).items(): t.add_row(k, v) con.print(t) if stream.qualities: t = Table("Quality", "URL", title="Available qualities") - for q, u in sorted(stream.qualities.items(), key=lambda kv: kv[0]): + for q, raw in sorted(stream.qualities.items(), key=lambda kv: kv[0]): + u = redact_text(raw) t.add_row(q, u[:80] + "..." if len(u) > 80 else u) con.print(t) @@ -691,18 +693,18 @@ def main() -> None: # noqa: C901 if exc.response is not None: status = exc.response.status_code url = str(exc.response.url) - progress.error(_PROG["http_error"].format(status=status, url=url)) + progress.error(_PROG["http_error"].format(status=status, url=redact_text(url))) if status == 403 and SHINDEN_BASE in url: if FLARESOLVERR_URL: - progress.warn(_PROG["flaresolverr_unreachable"].format(url=FLARESOLVERR_URL)) + progress.warn(_PROG["flaresolverr_unreachable"].format(url=redact_text(FLARESOLVERR_URL))) else: progress.warn(_PROG["cloudflare_hint"]) else: - progress.error(str(exc)) + progress.error(redact_text(exc)) sys.exit(1) except (AntiBotError, NoStreamError, ParseError, FilterMismatchError, DownloadFailedError, DownloadTargetError) as exc: diagnostics.session_end("error", type(exc).__name__) - progress.error(str(exc)) + progress.error(redact_text(exc)) sys.exit(1) except PlayerNotFoundError as exc: diagnostics.session_end("error", type(exc).__name__) @@ -710,7 +712,7 @@ def main() -> None: # noqa: C901 sys.exit(1) except ShindenError as exc: diagnostics.session_end("error", type(exc).__name__) - progress.error(_PROG["shinden_error"].format(exc=exc)) + progress.error(_PROG["shinden_error"].format(exc=redact_text(exc))) sys.exit(1) else: diagnostics.session_end("ok", None) diff --git a/src/alt_ani_cli/download.py b/src/alt_ani_cli/download.py index 2efac38..f1c8759 100644 --- a/src/alt_ani_cli/download.py +++ b/src/alt_ani_cli/download.py @@ -10,6 +10,7 @@ from alt_ani_cli.errors import DownloadFailedError, DownloadTargetError from alt_ani_cli.extract.common import Stream from alt_ani_cli.models import EpisodeRow, SeriesRef +from alt_ani_cli.redaction import redact_text from alt_ani_cli.ui import progress _SUPPRESS_PREFIXES = ( @@ -28,6 +29,7 @@ class _TruncLogger: def _emit(self, msg: str) -> None: if any(msg.startswith(p) for p in _SUPPRESS_PREFIXES): return + msg = redact_text(msg) cols = shutil.get_terminal_size((120, 24)).columns if len(msg) > cols: msg = msg[: cols - 1] + "…" diff --git a/src/alt_ani_cli/extract/__init__.py b/src/alt_ani_cli/extract/__init__.py index 6f05d91..e766e17 100644 --- a/src/alt_ani_cli/extract/__init__.py +++ b/src/alt_ani_cli/extract/__init__.py @@ -21,6 +21,7 @@ ytdlp_resolver, ) from alt_ani_cli.extract.common import Stream +from alt_ani_cli.redaction import redact_text # ebd.cda.pl/800x450/{id} → yt-dlp does not understand the embed URL; rewrite to www.cda.pl/video/{id} _EBD_CDA_RE = re.compile(r"/\d+x\d+/([0-9a-z]+)$", re.IGNORECASE) @@ -155,7 +156,8 @@ def resolver_family(host: str) -> str: def _exc_text(exc: Exception, embed_url: str, host: str) -> str: """Format an exception for user-facing messages, replacing the embed URL with its host.""" - return f"{type(exc).__name__}: {exc}".replace(repr(embed_url), host).replace(embed_url, host) + text = f"{type(exc).__name__}: {exc}".replace(repr(embed_url), host).replace(embed_url, host) + return redact_text(text) # Classify by ExtractError.category or exception type only — messages may carry URLs/tokens. diff --git a/src/alt_ani_cli/redaction.py b/src/alt_ani_cli/redaction.py new file mode 100644 index 0000000..a01ea77 --- /dev/null +++ b/src/alt_ani_cli/redaction.py @@ -0,0 +1,48 @@ +"""Display-only redaction of URL query values and secret headers — never apply to request data.""" + +import re +from collections.abc import Mapping + +REDACTED = "" + +SENSITIVE_HEADERS = frozenset( + { + "authorization", + "proxy-authorization", + "cookie", + "set-cookie", + "x-api-key", + "api-key", + "x-auth-token", + } +) + +# Stops only at chars invalid unencoded in a URL (or `#`), so ambiguous tails get redacted; skips `\\?\` paths. +_QUERY_RE = re.compile(r"(?<=\S)(?[?#])(?P[^\s\"`#]*)") +_PARAM_RE = re.compile(r"(?P^|&)(?P[^=&]+)=(?P[^&]*)") +_USERINFO_RE = re.compile(r"(?<=://)[^/\s@?#]+@") + + +def _redact_query(match: re.Match[str]) -> str: + body = _PARAM_RE.sub(lambda m: f"{m['lead']}{m['name']}={REDACTED}", match["query"]) + return f"{match['sep']}{body}" + + +def redact_text(text: object) -> str: + """Redact all query values and userinfo, keeping names/host/path; idempotent on its own output.""" + try: + s = str(text) + except Exception: + return REDACTED + s = _USERINFO_RE.sub(f"{REDACTED}@", s) + return _QUERY_RE.sub(_redact_query, s) + + +def redact_header_value(name: str, value: object) -> str: + if str(name).strip().lower() in SENSITIVE_HEADERS: + return REDACTED + return redact_text(value) + + +def redact_headers(headers: Mapping[str, object]) -> dict[str, str]: + return {k: redact_header_value(k, v) for k, v in headers.items()} diff --git a/src/alt_ani_cli/ui/menus.py b/src/alt_ani_cli/ui/menus.py index 194a059..e4a3b8f 100644 --- a/src/alt_ani_cli/ui/menus.py +++ b/src/alt_ani_cli/ui/menus.py @@ -15,6 +15,7 @@ from alt_ani_cli.content import CONTENT from alt_ani_cli.models import EpisodeRow, PlayerEntry, PlayerSource, RelatedSeries, SeriesHit, SeriesMetadata, SeriesRef +from alt_ani_cli.redaction import redact_text from alt_ani_cli.ui import progress _RES_RE = re.compile(r"(\d+)") @@ -540,9 +541,9 @@ def format_player_source(p: PlayerEntry, resolved: PlayerSource | None) -> tuple if p.subs_author: lines.append(_pl["source_author"].format(author=p.subs_author)) if p.source: - lines.append(_pl["source_url"].format(url=p.source)) + lines.append(_pl["source_url"].format(url=redact_text(p.source))) if resolved: - lines.append(_pl["source_embed"].format(url=resolved.embed_url)) + lines.append(_pl["source_embed"].format(url=redact_text(resolved.embed_url))) body = "\n".join(lines) or _pl["source_empty"] return _pl["source_header"].format(player=p.player), body diff --git a/tests/cli/test_debug_output.py b/tests/cli/test_debug_output.py new file mode 100644 index 0000000..4f14132 --- /dev/null +++ b/tests/cli/test_debug_output.py @@ -0,0 +1,74 @@ +"""--debug rendering: secrets are redacted at display time only.""" + +import io +from unittest.mock import patch + +import pytest +from rich.console import Console + +from alt_ani_cli.cli import _print_debug +from alt_ani_cli.extract.common import Stream +from alt_ani_cli.models import EmbedURL +from alt_ani_cli.player import mpv + +_URL = "https://cdn.example/x.mp4?token=SECRET" +_QUALITIES = {"1080p": "https://cdn.example/x1080.mp4?sig=ABC"} +_HEADERS = { + "Authorization": "Bearer VERY_SECRET", + "Cookie": "session=VERY_SECRET_TOO", + "Referer": "https://embed.example/e/1?token=REF_SECRET", + "User-Agent": "test-agent", +} +_EMBED = EmbedURL(url="https://embed.example/e/1?token=EMBED_SECRET", referer="https://shinden.pl/") + + +def _stream() -> Stream: + return Stream(url=_URL, headers=dict(_HEADERS), qualities=dict(_QUALITIES), ext="mp4") + + +def _render(stream: Stream) -> str: + buf = io.StringIO() + console = Console(file=buf, width=300, legacy_windows=False, highlight=False) + with patch("alt_ani_cli.ui.progress._get", return_value=console): + _print_debug(stream, _EMBED) + return buf.getvalue() + + +@pytest.mark.unit +class TestPrintDebugRedaction: + def test_no_secret_in_output(self): + out = _render(_stream()) + for secret in ("SECRET", "ABC", "VERY_SECRET", "VERY_SECRET_TOO", "REF_SECRET", "EMBED_SECRET", "Bearer"): + assert secret not in out + + def test_hosts_and_paths_visible(self): + out = _render(_stream()) + assert "cdn.example/x.mp4?token=" in out + assert "cdn.example/x1080.mp4?sig=" in out + assert "embed.example/e/1?token=" in out + assert "" in out + + def test_user_agent_still_visible(self): + assert "test-agent" in _render(_stream()) + + +@pytest.mark.unit +class TestRedactionLeavesRuntimeDataIntact: + def test_print_debug_does_not_mutate_stream(self): + stream = _stream() + _render(stream) + assert stream.url == _URL + assert stream.qualities == _QUALITIES + assert stream.headers == _HEADERS + assert _EMBED.url == "https://embed.example/e/1?token=EMBED_SECRET" + + def test_player_command_still_gets_raw_url_and_headers(self): + stream = _stream() + _render(stream) + with patch("alt_ani_cli.player.mpv._find", return_value="mpv"): + cmd = mpv.build(stream, title="t") + assert _URL in cmd + assert "--referrer=https://embed.example/e/1?token=REF_SECRET" in cmd + fields = next(a for a in cmd if a.startswith("--http-header-fields=")) + assert "Authorization: Bearer VERY_SECRET" in fields + assert "Cookie: session=VERY_SECRET_TOO" in fields diff --git a/tests/cli/test_errors.py b/tests/cli/test_errors.py index 08c36f2..a04481a 100644 --- a/tests/cli/test_errors.py +++ b/tests/cli/test_errors.py @@ -107,3 +107,34 @@ def test_download_failure_exits_1_with_plain_message(self, capsys): combined = "".join(capsys.readouterr()) assert "Pobieranie nie powiodło się." in combined assert "Błąd shinden" not in combined + + +@pytest.mark.unit +class TestHTTPErrorUrlRedaction: + def test_signed_query_value_not_shown(self, capsys): + exc = _make_http_error(403, "https://example.test/path?token=SECRET") + with ( + patch("sys.argv", ["alt-ani-cli", "--url", "https://shinden.pl/series/1-test"]), + patch("alt_ani_cli.cli._run_noninteractive", side_effect=exc), + patch("alt_ani_cli.cli.shinden_http.make_client"), + pytest.raises(SystemExit), + ): + main() + combined = "".join(capsys.readouterr()) + assert "SECRET" not in combined + assert "example.test/path" in combined + assert "token=" in combined + + def test_shinden_detection_still_uses_full_url(self, capsys): + exc = _make_http_error(403, "https://shinden.pl/episode/1/view?sig=SECRET") + with ( + patch("sys.argv", ["alt-ani-cli", "--url", "https://shinden.pl/series/1-test"]), + patch("alt_ani_cli.cli._run_noninteractive", side_effect=exc), + patch("alt_ani_cli.cli.shinden_http.make_client"), + patch("alt_ani_cli.cli.FLARESOLVERR_URL", ""), + pytest.raises(SystemExit), + ): + main() + combined = "".join(capsys.readouterr()) + assert "Cloudflare" in combined + assert "SECRET" not in combined diff --git a/tests/cli/test_resolve_with_fallback.py b/tests/cli/test_resolve_with_fallback.py index ef44917..a21d2e4 100644 --- a/tests/cli/test_resolve_with_fallback.py +++ b/tests/cli/test_resolve_with_fallback.py @@ -612,3 +612,21 @@ def test_health_none_never_calls_learn_label(self): assert result is not None assert embed is _EMBED + + +@pytest.mark.unit +class TestResolveWithFallbackRedaction: + def test_embed_and_failure_output_redact_query_values(self, capsys): + embed = EmbedURL(url="https://vidawra.cc/e/abc123?t=EMBED_SECRET", referer="https://shinden.pl/") + exc = NoStreamError("all extractors failed: https://cdn.example/master.m3u8?token=SECRET") + with ( + patch("alt_ani_cli.cli.shinden_api.resolve_embed", return_value=embed), + patch("alt_ani_cli.cli.extract.resolve", side_effect=exc) as mock_resolve, + patch("alt_ani_cli.cli.diagnostics.resolve_result"), + ): + _resolve_with_fallback(MagicMock(), [_PLAYER], _PLAYER, False, 1.0) + combined = "".join(capsys.readouterr()) + assert "EMBED_SECRET" not in combined and "SECRET" not in combined + assert "vidawra.cc/e/abc123?t=" in combined + assert "cdn.example/master.m3u8?token=" in combined + assert mock_resolve.call_args[0][0] == "https://vidawra.cc/e/abc123?t=EMBED_SECRET" diff --git a/tests/extract/test_dispatch.py b/tests/extract/test_dispatch.py index 7727435..1e275cf 100644 --- a/tests/extract/test_dispatch.py +++ b/tests/extract/test_dispatch.py @@ -11,7 +11,7 @@ from yt_dlp.networking.exceptions import TransportError from alt_ani_cli.errors import JavaScriptRequiredError, NoStreamError, UnsupportedHostError -from alt_ani_cli.extract import HOST_RULES, HostRule, _normalize_url, mp4upload, resolve, resolver_family +from alt_ani_cli.extract import HOST_RULES, HostRule, _exc_text, _normalize_url, mp4upload, resolve, resolver_family from alt_ani_cli.extract.common import CATEGORY_NO_STREAM_URL, CATEGORY_PARSER_DRIFT, ExtractError, Stream _REFERER = "https://shinden.pl/" @@ -239,6 +239,29 @@ def test_failure_messages_use_host_not_embed_url(self): assert embed_url not in exc_text assert "unknownhost.tv" in exc_text + def test_exc_text_redacts_other_signed_urls(self): + embed_url = "https://player.example/e/abc" + exc = TimeoutError(f"Timeout while fetching {embed_url} -> https://cdn.example/master.m3u8?token=SECRET") + text = _exc_text(exc, embed_url, "player.example") + assert embed_url not in text + assert "-> " in text and text.startswith("TimeoutError: Timeout while fetching player.example") + assert "SECRET" not in text + assert "cdn.example/master.m3u8?token=" in text + + def test_signed_url_in_failure_messages_redacted(self): + embed_url = "https://unknownhost.tv/embed/abc" + signed = "https://cdn.example/master.m3u8?token=SECRET&expires=999" + on_fallback = MagicMock() + with ( + patch("alt_ani_cli.extract.jwplayer.resolve", side_effect=ValueError(f"Timeout while fetching {signed}")), + patch("alt_ani_cli.extract.ytdlp_resolver.resolve", side_effect=Exception(f"HTTP 403 for {signed}")), + pytest.raises(NoStreamError) as exc_info, + ): + resolve(embed_url, _REFERER, on_fallback=on_fallback) + for text in (str(exc_info.value), on_fallback.call_args[0][2]): + assert "SECRET" not in text and "999" not in text + assert "cdn.example/master.m3u8?token=&expires=" in text + def _http_error(status_code: int) -> cffi_exceptions.HTTPError: resp = MagicMock(status_code=status_code) diff --git a/tests/test_download.py b/tests/test_download.py index 3fe2c69..5395c71 100644 --- a/tests/test_download.py +++ b/tests/test_download.py @@ -1,3 +1,4 @@ +import os import sys from pathlib import Path from unittest.mock import patch @@ -487,3 +488,76 @@ def test_cleanup_after_failed_attempt_stays_best_effort(self, fake_ydl, tmp_path with _unlink_fails_for("Show - ep1.redownload.mp4.part"), pytest.raises(DownloadFailedError): _overwrite(tmp_path) assert final.read_bytes() == _OLD + + +def _terminal(columns: int): + return patch("alt_ani_cli.download.shutil.get_terminal_size", return_value=os.terminal_size((columns, 24))) + + +@pytest.mark.unit +class TestTruncLoggerRedaction: + def test_secret_query_value_never_printed(self, capsys): + with _terminal(200): + download._TruncLogger().info("[hlsnative] Downloading m3u8 manifest https://cdn.example/hls/master.m3u8?token=SECRET") + out = capsys.readouterr().out + assert "SECRET" not in out + assert "" in out + assert "cdn.example/hls/master.m3u8?token=" in out + + def test_redaction_happens_before_truncation(self, capsys): + head = "x" * 40 + " https://h.example/m.m3u8?token=" + # Truncating first would keep "token=SECRET_…" — and a cut value no longer looks like a full query. + with _terminal(len(head) + 8): + download._TruncLogger().info(head + "SECRET_TAIL_VALUE_THAT_IS_LONG") + out = capsys.readouterr().out + assert "SECRET" not in out + assert out.rstrip("\n") == head + "&expires=" in out + + def test_suppressed_prefixes_still_suppressed(self, capsys): + download._TruncLogger().info("[generic] Extracting URL: https://h/m?token=SECRET") + assert capsys.readouterr().out == "" + + def test_real_ytdlp_timeout_message_does_not_leak(self, capsys): + msg = ( + "ERROR: [generic] master: Unable to download webpage: " + "https://host.example/hls/master.m3u8?token=THIS_MUST_NOT_LEAK&expires=999 (caused by TransportError('timed out'))" + ) + with _terminal(400): + download._TruncLogger().error(msg) + out = capsys.readouterr().out + assert "THIS_MUST_NOT_LEAK" not in out + assert "999" not in out + assert "host.example/hls/master.m3u8?token=&expires=" in out + assert "timed out" in out + + def test_local_paths_untouched(self, capsys): + with _terminal(200): + download._TruncLogger().info(r"[download] Destination: C:\Users\Test\Videos\Show - ep4.mp4") + assert r"C:\Users\Test\Videos\Show - ep4.mp4" in capsys.readouterr().out + + +@pytest.mark.unit +class TestRedactionDoesNotTouchTransport: + def test_signed_url_and_headers_reach_ytdlp_verbatim(self, fake_ydl, tmp_path): + url = "https://cdn.example/hls/master.m3u8?token=SECRET&expires=999" + headers = {"Referer": "https://embed.example/e/1?token=REF", "Cookie": "session=S", "User-Agent": "UA/1.0"} + seen: list[str] = [] + original = _FakeYDL.extract_info + + def spy(self, u, download=True): + seen.append(u) + return original(self, u, download) + + with patch.object(_FakeYDL, "extract_info", spy): + ydl = _run(Stream(url=url, headers=dict(headers), ext="m3u8"), tmp_path) + assert seen == [url] + assert ydl.opts["http_headers"] == headers diff --git a/tests/test_redaction.py b/tests/test_redaction.py new file mode 100644 index 0000000..fffd3c2 --- /dev/null +++ b/tests/test_redaction.py @@ -0,0 +1,156 @@ +import pytest + +from alt_ani_cli.redaction import REDACTED, redact_header_value, redact_headers, redact_text + + +@pytest.mark.unit +class TestRedactText: + def test_single_query_value_redacted(self): + assert redact_text("https://cdn/x.m3u8?token=SECRET") == f"https://cdn/x.m3u8?token={REDACTED}" + + def test_all_parameter_values_redacted(self): + out = redact_text("https://cdn.example/a?a=1&token=2&expires=3") + assert out == f"https://cdn.example/a?a={REDACTED}&token={REDACTED}&expires={REDACTED}" + + def test_non_secret_looking_values_redacted_too(self): + out = redact_text("https://cdn.example/hls/master.m3u8?token=ABC&expires=123&quality=1080") + assert out == f"https://cdn.example/hls/master.m3u8?token={REDACTED}&expires={REDACTED}&quality={REDACTED}" + + def test_url_without_query_unchanged(self): + url = "https://player.example:8443/e/abc123/video.mp4" + assert redact_text(url) == url + + def test_scheme_host_port_path_preserved(self): + out = redact_text("https://cdn.example:8443/hls/v1/master.m3u8?sig=XYZ") + assert out.startswith("https://cdn.example:8443/hls/v1/master.m3u8?sig=") + assert "XYZ" not in out + + def test_bare_filename_with_query(self): + out = redact_text("ERROR: fragment failed:\nmaster.m3u8?token=SECRET&x=1") + assert out == f"ERROR: fragment failed:\nmaster.m3u8?token={REDACTED}&x={REDACTED}" + + def test_request_path_with_query(self): + assert redact_text("GET /hls/master.txt?sig=ABC") == f"GET /hls/master.txt?sig={REDACTED}" + + def test_aws_presigned_query(self): + out = redact_text("https://b.s3.amazonaws.com/k?X-Amz-Credential=AAA&X-Amz-Signature=BBB&X-Amz-Date=CCC") + for secret in ("AAA", "BBB", "CCC"): + assert secret not in out + assert "X-Amz-Credential=" in out and "X-Amz-Signature=" in out and "X-Amz-Date=" in out + + def test_cloudfront_signed_query(self): + out = redact_text("https://d1.cloudfront.net/v.mp4?Policy=AAA&Signature=BBB&Key-Pair-Id=CCC") + for secret in ("AAA", "BBB", "CCC"): + assert secret not in out + assert "Policy=" in out and "Signature=" in out and "Key-Pair-Id=" in out + + def test_plain_fragment_and_surrounding_text_stay_readable(self): + out = redact_text("ERROR: unable to download https://cdn.example/hls/master.m3u8?token=ABC#part1 HTTP Error 403") + assert out == f"ERROR: unable to download https://cdn.example/hls/master.m3u8?token={REDACTED}#part1 HTTP Error 403" + + def test_query_like_fragment_values_redacted(self): + out = redact_text("https://host.example/cb#access_token=SECRET&expires_in=3600") + assert out == f"https://host.example/cb#access_token={REDACTED}&expires_in={REDACTED}" + + def test_multiline_ytdlp_message_keeps_status_and_location(self): + msg = "ERROR: unable to download\nhttps://cdn.example/hls/master.m3u8?token=ABC&expires=123\nHTTP Error 403" + out = redact_text(msg) + assert "ABC" not in out and "=123" not in out + assert "cdn.example/hls/master.m3u8?token=" in out and "&expires=" in out + assert "HTTP Error 403" in out + + def test_punctuation_only_value_fully_redacted(self): + out = redact_text("https://h/x?token=!!!!") + assert out == f"https://h/x?token={REDACTED}" + assert "!" not in out + + @pytest.mark.parametrize("tail", [".", ")", ",", ";", ":", "!", "]", "}", ">", "'"]) + def test_trailing_punctuation_is_treated_as_part_of_value(self, tail): + out = redact_text(f"https://h/x?token=abc{tail}") + assert out == f"https://h/x?token={REDACTED}" + assert "abc" not in out + + def test_question_mark_inside_value_does_not_end_it(self): + out = redact_text("https://h/x?token=abc?MORE") + assert out == f"https://h/x?token={REDACTED}" + assert "abc" not in out and "MORE" not in out + + def test_nested_query_after_question_mark_redacted(self): + out = redact_text("https://h/x?token=abc?more=SECOND") + assert out == f"https://h/x?token={REDACTED}" + assert "abc" not in out and "SECOND" not in out + + def test_redaction_continues_to_next_whitespace(self): + out = redact_text("failed GET 'https://cdn/x?token=S') then retry") + assert out == f"failed GET 'https://cdn/x?token={REDACTED} then retry" + + def test_url_userinfo_redacted(self): + out = redact_text("https://user:hunter2@host.example/path") + assert out == f"https://{REDACTED}@host.example/path" + + @pytest.mark.parametrize( + "text", + [ + "https://cdn.example/hls/master.m3u8?token=ABC&expires=123&quality=1080", + "master.m3u8?token=SECRET&x=1 then (https://h/x?a=1). ", + "https://u:p@h.example/a?b=1#c=2", + "https://h/p?a=1&&flag&c=", + "https://h/x?token=abc?more=SECOND", + "https://h/x?token=!!!!", + ], + ) + def test_idempotent(self, text): + once = redact_text(text) + assert redact_text(once) == once + + @pytest.mark.parametrize( + "text", + ["", "?", "??==&&", "=&?#", "https://", "?=&=", "a?b=c?d=e#f=g#h", "\x00\xff?\ud800=1", "?"], + ) + def test_malformed_input_never_raises(self, text): + assert isinstance(redact_text(text), str) + + def test_non_string_input_never_raises(self): + class Boom: + def __str__(self): + raise RuntimeError("nope") + + assert redact_text(None) == "None" + assert redact_text(ValueError("x https://h/a?t=S")) == f"x https://h/a?t={REDACTED}" + assert redact_text(Boom()) == REDACTED + + @pytest.mark.parametrize( + "text", + [ + r"C:\Users\Test\Videos\Show - ep4.mp4", + r"\\?\C:\Users\Test\Videos\a=b\Show.mp4", + "Shingeki no Kyojin: The Final Season - Kanketsu-hen", + "Overwrite existing file? [y/N]", + "Pobieranie nie powiodło się.", + ], + ) + def test_text_without_url_query_unchanged(self, text): + assert redact_text(text) == text + + +@pytest.mark.unit +class TestRedactHeaders: + @pytest.mark.parametrize( + "name", + ["Authorization", "proxy-authorization", "COOKIE", "Set-Cookie", "X-Api-Key", "api-key", "X-Auth-Token"], + ) + def test_sensitive_header_fully_redacted(self, name): + assert redact_header_value(name, "Bearer abc https://h/x") == REDACTED + + def test_other_header_gets_query_redaction(self): + out = redact_header_value("Referer", "https://host/e?id=123&token=ABC") + assert out == f"https://host/e?id={REDACTED}&token={REDACTED}" + + def test_plain_header_value_unchanged(self): + assert redact_header_value("User-Agent", "Mozilla/5.0 (Windows NT 10.0)") == "Mozilla/5.0 (Windows NT 10.0)" + + def test_redact_headers_returns_new_dict(self): + headers = {"Cookie": "s=1", "Referer": "https://h/e?t=2"} + out = redact_headers(headers) + assert out == {"Cookie": REDACTED, "Referer": f"https://h/e?t={REDACTED}"} + assert headers == {"Cookie": "s=1", "Referer": "https://h/e?t=2"} diff --git a/tests/ui/test_menus.py b/tests/ui/test_menus.py index c0a32d6..08f729c 100644 --- a/tests/ui/test_menus.py +++ b/tests/ui/test_menus.py @@ -478,6 +478,19 @@ def test_full_info(self): assert "https://miorosubs.com/" in body assert "https://ebd.cda.pl/620x395/xyz" in body + def test_query_values_redacted_for_display_only(self): + source = "https://subs.example/release?id=7&token=SRC_SECRET" + embed_url = "https://player.example/e/abc123?sig=EMBED_SECRET" + p = PlayerEntry(online_id="p1", player="CDA", lang_audio="jp", lang_subs="pl", subs_author="Mioro-Subs", source=source) + resolved = PlayerSource(online_id="p1", host="player.example", embed_url=embed_url) + title, body = format_player_source(p, resolved) + assert "SRC_SECRET" not in body and "EMBED_SECRET" not in body and "=7" not in body + assert "Mioro-Subs" in body + assert "subs.example/release?id=&token=" in body + assert "player.example/e/abc123?sig=" in body + assert "Źródło:" in body and "Embed:" in body + assert p.source == source and resolved.embed_url == embed_url + def test_no_info_renders_empty_message(self): title, body = format_player_source(_PLAYER, None) assert "CDA" in title