From 00d539a85dbdccea0f879b5c38bdab7ec9c25825 Mon Sep 17 00:00:00 2001 From: Adriano Rocha Date: Wed, 6 May 2026 21:59:16 -0300 Subject: [PATCH 1/2] feat: add headless login flow --- README.md | 14 +++++++- src/ytstudio/api.py | 84 ++++++++++++++++++++++++++++++++++++-------- src/ytstudio/main.py | 10 ++++-- tests/test_api.py | 83 +++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 173 insertions(+), 18 deletions(-) diff --git a/README.md b/README.md index ce2cd4d..47c26b9 100644 --- a/README.md +++ b/README.md @@ -47,7 +47,19 @@ ytstudio init --client-secrets path/to/client_secret_.json ytstudio login ``` -Credentials stored in `~/.config/ytstudio/`. +Credentials stored in `~/.config/ytstudio-cli/`. + +### Headless Linux login + +If you are logging in from a server without a browser, run: + +```bash +ytstudio login --headless +``` + +The command prints a Google OAuth URL. Open that URL in a browser on any machine and approve +access. The browser will then fail to load a `127.0.0.1` page; this is expected. Copy the full URL +from the browser address bar, paste it back into the terminal, and ytstudio will finish the login. ## API quota diff --git a/src/ytstudio/api.py b/src/ytstudio/api.py index 3a4510c..07ce632 100644 --- a/src/ytstudio/api.py +++ b/src/ytstudio/api.py @@ -1,3 +1,5 @@ +from urllib.parse import parse_qs, urlparse + import typer from google.auth.exceptions import RefreshError from google.auth.transport.requests import Request @@ -5,6 +7,7 @@ from google_auth_oauthlib.flow import InstalledAppFlow from googleapiclient.discovery import build from googleapiclient.errors import HttpError +from rich.prompt import Prompt from ytstudio.config import ( CLIENT_SECRETS_FILE, @@ -60,27 +63,17 @@ def api(request): "https://www.googleapis.com/auth/yt-analytics.readonly", ] +HEADLESS_REDIRECT_URI = "http://127.0.0.1:9876/" -def authenticate() -> None: - if not CLIENT_SECRETS_FILE.exists(): - console.print("[red]No client secrets found. Run 'ytstudio init' first.[/red]") - raise SystemExit(1) from None - console.print("[bold]Authenticating with YouTube...[/bold]\n") - - flow = InstalledAppFlow.from_client_secrets_file( +def _create_flow() -> InstalledAppFlow: + return InstalledAppFlow.from_client_secrets_file( str(CLIENT_SECRETS_FILE), scopes=SCOPES, ) - # Run local server for OAuth callback - credentials = flow.run_local_server( - port=9876, - prompt="consent", - open_browser=True, - ) - # Save credentials +def _save_credentials(credentials: Credentials) -> None: creds_data = { "token": credentials.token, "refresh_token": credentials.refresh_token, @@ -91,7 +84,8 @@ def authenticate() -> None: } save_credentials(creds_data) - # Get channel info + +def _show_login_success(credentials: Credentials) -> None: service = build("youtube", "v3", credentials=credentials) response = service.channels().list(part="snippet", mine=True).execute() @@ -103,6 +97,66 @@ def authenticate() -> None: success_message("Authentication successful") +def _validate_authorization_response(authorization_response: str) -> None: + parsed_url = urlparse(authorization_response) + query = parse_qs(parsed_url.query) + + if error := query.get("error"): + error_description = query.get("error_description", [""])[0] + message = error_description or error[0] + console.print(f"[red]Authorization failed: {message}[/red]") + raise SystemExit(1) from None + + if not query.get("code"): + console.print("[red]Redirect URL is missing an authorization code.[/red]") + raise SystemExit(1) from None + + +def _authenticate_headless() -> Credentials: + flow = _create_flow() + flow.redirect_uri = HEADLESS_REDIRECT_URI + authorization_url, _ = flow.authorization_url(prompt="consent") + + console.print("Open this URL in a browser on any machine:\n") + console.print(f"[bold]{authorization_url}[/bold]\n") + console.print( + "After approving access, the browser will fail to load a 127.0.0.1 page. " + "Copy the full failed redirect URL from the address bar and paste it below." + ) + + authorization_response = Prompt.ask("Redirect URL").strip() + _validate_authorization_response(authorization_response) + + try: + flow.fetch_token(authorization_response=authorization_response) + except Exception as error: + console.print(f"[red]Could not complete OAuth token exchange: {error}[/red]") + raise SystemExit(1) from None + + return flow.credentials + + +def authenticate(headless: bool = False) -> None: + if not CLIENT_SECRETS_FILE.exists(): + console.print("[red]No client secrets found. Run 'ytstudio init' first.[/red]") + raise SystemExit(1) from None + + console.print("[bold]Authenticating with YouTube...[/bold]\n") + + if headless: + credentials = _authenticate_headless() + else: + flow = _create_flow() + credentials = flow.run_local_server( + port=9876, + prompt="consent", + open_browser=True, + ) + + _save_credentials(credentials) + _show_login_success(credentials) + + def get_credentials() -> Credentials | None: creds_data = load_credentials() if not creds_data: diff --git a/src/ytstudio/main.py b/src/ytstudio/main.py index 1978981..835562c 100644 --- a/src/ytstudio/main.py +++ b/src/ytstudio/main.py @@ -36,9 +36,15 @@ def init( @app.command() -def login(): +def login( + headless: bool = typer.Option( + False, + "--headless", + help="Authenticate by pasting a redirect URL from another browser", + ), +): """Authenticate with YouTube via OAuth""" - authenticate() + authenticate(headless=headless) @app.command() diff --git a/tests/test_api.py b/tests/test_api.py index 0c59984..ad6f581 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -5,6 +5,7 @@ from googleapiclient.errors import HttpError from typer.testing import CliRunner +from ytstudio import api as api_module from ytstudio.api import api, get_authenticated_service, handle_api_error from ytstudio.main import app @@ -54,7 +55,89 @@ def test_login_requires_client_secrets(self): result = runner.invoke(app, ["login"]) assert result.exit_code == 1 + def test_login_headless_passes_option(self): + with patch("ytstudio.main.authenticate") as authenticate: + result = runner.invoke(app, ["login", "--headless"]) + + assert result.exit_code == 0 + authenticate.assert_called_once_with(headless=True) + def test_status_not_authenticated(self): with patch("ytstudio.api.load_credentials", return_value=None): result = runner.invoke(app, ["status"]) assert "Not authenticated" in result.stdout + + +class TestAuthenticate: + def test_normal_login_uses_local_server(self): + credentials = MagicMock() + flow = MagicMock() + flow.run_local_server.return_value = credentials + + with ( + patch("ytstudio.api.CLIENT_SECRETS_FILE") as mock_file, + patch("ytstudio.api._create_flow", return_value=flow), + patch("ytstudio.api._save_credentials") as save_credentials, + patch("ytstudio.api._show_login_success") as show_login_success, + ): + mock_file.exists.return_value = True + + api_module.authenticate() + + flow.run_local_server.assert_called_once_with( + port=9876, + prompt="consent", + open_browser=True, + ) + save_credentials.assert_called_once_with(credentials) + show_login_success.assert_called_once_with(credentials) + + def test_headless_login_exchanges_pasted_redirect_url(self): + credentials = MagicMock() + authorization_url = "https://accounts.google.com/o/oauth2/auth?state=test-state" + redirect_url = "http://127.0.0.1:9876/?state=test-state&code=test-code" + flow = MagicMock() + flow.authorization_url.return_value = (authorization_url, "test-state") + flow.credentials = credentials + + with ( + patch("ytstudio.api.CLIENT_SECRETS_FILE") as mock_file, + patch("ytstudio.api._create_flow", return_value=flow), + patch("ytstudio.api.Prompt.ask", return_value=redirect_url), + patch("ytstudio.api._save_credentials") as save_credentials, + patch("ytstudio.api._show_login_success") as show_login_success, + ): + mock_file.exists.return_value = True + + api_module.authenticate(headless=True) + + assert flow.redirect_uri == api_module.HEADLESS_REDIRECT_URI + flow.authorization_url.assert_called_once_with(prompt="consent") + flow.fetch_token.assert_called_once_with(authorization_response=redirect_url) + save_credentials.assert_called_once_with(credentials) + show_login_success.assert_called_once_with(credentials) + + def test_headless_login_rejects_missing_code(self): + with pytest.raises(SystemExit): + api_module._validate_authorization_response("http://127.0.0.1:9876/?state=test-state") + + def test_headless_login_rejects_authorization_error(self): + with pytest.raises(SystemExit): + api_module._validate_authorization_response( + "http://127.0.0.1:9876/?error=access_denied" + ) + + def test_headless_login_exits_when_token_exchange_fails(self): + flow = MagicMock() + flow.authorization_url.return_value = ("https://accounts.google.com/o/oauth2/auth", "state") + flow.fetch_token.side_effect = ValueError("state mismatch") + + with ( + patch("ytstudio.api._create_flow", return_value=flow), + patch( + "ytstudio.api.Prompt.ask", + return_value="http://127.0.0.1:9876/?state=wrong&code=test-code", + ), + pytest.raises(SystemExit), + ): + api_module._authenticate_headless() From 7ae7f83b2077c92198204ea9f77ee61aa0ecef50 Mon Sep 17 00:00:00 2001 From: Adriano Rocha Date: Wed, 6 May 2026 22:09:54 -0300 Subject: [PATCH 2/2] fix: avoid insecure headless auth response --- src/ytstudio/api.py | 15 +++++++++++---- tests/test_api.py | 23 +++++++++++++++++------ 2 files changed, 28 insertions(+), 10 deletions(-) diff --git a/src/ytstudio/api.py b/src/ytstudio/api.py index 07ce632..876e23d 100644 --- a/src/ytstudio/api.py +++ b/src/ytstudio/api.py @@ -97,7 +97,7 @@ def _show_login_success(credentials: Credentials) -> None: success_message("Authentication successful") -def _validate_authorization_response(authorization_response: str) -> None: +def _parse_authorization_response(authorization_response: str, expected_state: str) -> str: parsed_url = urlparse(authorization_response) query = parse_qs(parsed_url.query) @@ -111,11 +111,18 @@ def _validate_authorization_response(authorization_response: str) -> None: console.print("[red]Redirect URL is missing an authorization code.[/red]") raise SystemExit(1) from None + state = query.get("state", [""])[0] + if state != expected_state: + console.print("[red]Redirect URL state does not match this login attempt.[/red]") + raise SystemExit(1) from None + + return query["code"][0] + def _authenticate_headless() -> Credentials: flow = _create_flow() flow.redirect_uri = HEADLESS_REDIRECT_URI - authorization_url, _ = flow.authorization_url(prompt="consent") + authorization_url, state = flow.authorization_url(prompt="consent") console.print("Open this URL in a browser on any machine:\n") console.print(f"[bold]{authorization_url}[/bold]\n") @@ -125,10 +132,10 @@ def _authenticate_headless() -> Credentials: ) authorization_response = Prompt.ask("Redirect URL").strip() - _validate_authorization_response(authorization_response) + code = _parse_authorization_response(authorization_response, state) try: - flow.fetch_token(authorization_response=authorization_response) + flow.fetch_token(code=code) except Exception as error: console.print(f"[red]Could not complete OAuth token exchange: {error}[/red]") raise SystemExit(1) from None diff --git a/tests/test_api.py b/tests/test_api.py index ad6f581..397a4ed 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -113,30 +113,41 @@ def test_headless_login_exchanges_pasted_redirect_url(self): assert flow.redirect_uri == api_module.HEADLESS_REDIRECT_URI flow.authorization_url.assert_called_once_with(prompt="consent") - flow.fetch_token.assert_called_once_with(authorization_response=redirect_url) + flow.fetch_token.assert_called_once_with(code="test-code") save_credentials.assert_called_once_with(credentials) show_login_success.assert_called_once_with(credentials) def test_headless_login_rejects_missing_code(self): with pytest.raises(SystemExit): - api_module._validate_authorization_response("http://127.0.0.1:9876/?state=test-state") + api_module._parse_authorization_response( + "http://127.0.0.1:9876/?state=test-state", + "test-state", + ) def test_headless_login_rejects_authorization_error(self): with pytest.raises(SystemExit): - api_module._validate_authorization_response( - "http://127.0.0.1:9876/?error=access_denied" + api_module._parse_authorization_response( + "http://127.0.0.1:9876/?error=access_denied", + "test-state", + ) + + def test_headless_login_rejects_state_mismatch(self): + with pytest.raises(SystemExit): + api_module._parse_authorization_response( + "http://127.0.0.1:9876/?state=wrong&code=test-code", + "test-state", ) def test_headless_login_exits_when_token_exchange_fails(self): flow = MagicMock() flow.authorization_url.return_value = ("https://accounts.google.com/o/oauth2/auth", "state") - flow.fetch_token.side_effect = ValueError("state mismatch") + flow.fetch_token.side_effect = ValueError("token exchange failed") with ( patch("ytstudio.api._create_flow", return_value=flow), patch( "ytstudio.api.Prompt.ask", - return_value="http://127.0.0.1:9876/?state=wrong&code=test-code", + return_value="http://127.0.0.1:9876/?state=state&code=test-code", ), pytest.raises(SystemExit), ):