feat: add headless login flow

README.md

@@ -47,7 +47,19 @@ ytstudio init --client-secrets path/to/client_secret_<id>.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
 

src/ytstudio/api.py

@@ -1,10 +1,13 @@
+from urllib.parse import parse_qs, urlparse
+
 import typer
 from google.auth.exceptions import RefreshError
 from google.auth.transport.requests import Request
 from google.oauth2.credentials import Credentials
 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,73 @@ def authenticate() -> None:
         success_message("Authentication successful")
 
 
+def _parse_authorization_response(authorization_response: str, expected_state: str) -> str:
+    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
+
+    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, 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")
+    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()
+    code = _parse_authorization_response(authorization_response, state)
+
+    try:
+        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
+
+    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:

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()

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,100 @@ 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(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._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._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("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=state&code=test-code",
+            ),
+            pytest.raises(SystemExit),
+        ):
+            api_module._authenticate_headless()