fix(security): redact URL query values and secret headers in user-fac…

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,26 +693,26 @@ 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__)
         progress.error(str(exc))
         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)

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] + "…"

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.

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 = "<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<sep>[?#])(?P<query>[^\s\"`#]*)")
+_PARAM_RE = re.compile(r"(?P<lead>^|&)(?P<name>[^=&]+)=(?P<value>[^&]*)")
+_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()}

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
 

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=<redacted>" in out
+        assert "cdn.example/x1080.mp4?sig=<redacted>" in out
+        assert "embed.example/e/1?token=<redacted>" in out
+        assert "<redacted>" 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

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=<redacted>" 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

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=<redacted>" in combined
+        assert "cdn.example/master.m3u8?token=<redacted>" in combined
+        assert mock_resolve.call_args[0][0] == "https://vidawra.cc/e/abc123?t=EMBED_SECRET"

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=<redacted>" 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=<redacted>&expires=<redacted>" in text
+
 
 def _http_error(status_code: int) -> cffi_exceptions.HTTPError:
     resp = MagicMock(status_code=status_code)