Auth0 support

Author: rodneykinney

.claude-plugin/marketplace.json

@@ -11,7 +11,7 @@
       "name": "asta",
       "source": "./",
       "description": "Paper search, citations, literature reports, and Semantic Scholar API tools",
-      "version": "0.5.1",
+      "version": "0.6",
       "author": {
         "name": "AI2 Asta Team"
       },

.claude-plugin/plugin.json

@@ -1,6 +1,6 @@
 {
   "name": "asta",
-  "version": "0.5.1",
+  "version": "0.6",
   "description": "Asta science tools for Claude Code - paper search, citations, and more",
   "author": {
     "name": "AI2 Asta Team"

README.md

@@ -48,6 +48,23 @@ uv tool upgrade asta
 
 See `asta --help` for usage instructions.
 
+## Authentication
+
+Asta supports secure authentication via Auth0 Device Authorization Flow:
+
+```bash
+# Login (opens browser for authentication)
+asta auth login
+
+# Check authentication status
+asta auth status
+
+# Logout
+asta auth logout
+```
+
+For detailed authentication documentation, see [docs/AUTHENTICATION.md](docs/AUTHENTICATION.md).
+
 ## Development
 
 See [DEVELOPER.md](DEVELOPER.md) for contributor guidelines, architecture details, and development setup.

pyproject.toml

@@ -4,18 +4,24 @@ build-backend = "hatchling.build"
 
 [project]
 name = "asta"
-version = "0.5.1"
+version = "0.6"
 description = "Asta CLI for scientific literature review"
 readme = "README.md"
 requires-python = ">=3.11"
 license = "Apache-2.0"
 dependencies = [
     "click>=8.0",
     "pydantic>=2.0",
+    "pydantic-settings>=2.1.0",
     "pyhocon>=0.3.60",
     "pymupdf>=1.27.0",
     "pymupdf-layout>=1.27.0",
     "pymupdf4llm>=0.3.0",
+    "httpx>=0.26.0",
+    "rich>=13.7.0",
+    "python-jose[cryptography]>=3.3.0",
+    "keyring>=24.3.0",
+    "platformdirs>=4.1.0",
 ]
 
 [project.scripts]

src/asta/__init__.py

@@ -1,3 +1,3 @@
 """Asta - Science literature research tools"""
 
-__version__ = "0.5.1"
+__version__ = "0.6"

src/asta/auth/__init__.py

@@ -0,0 +1,5 @@
+"""Authentication module for Asta CLI."""
+
+from asta.auth.token_manager import TokenManager
+
+__all__ = ["TokenManager"]

src/asta/auth/device_flow.py

@@ -0,0 +1,193 @@
+"""
+Auth0 Device Authorization Flow implementation.
+
+Follows RFC 8628: https://tools.ietf.org/html/rfc8628
+"""
+
+import asyncio
+import time
+from dataclasses import dataclass
+
+import httpx
+
+from .exceptions import AuthenticationError, AuthenticationTimeout
+
+
+@dataclass
+class DeviceCodeResponse:
+    """Response from /oauth/device/code endpoint."""
+
+    device_code: str
+    user_code: str
+    verification_uri: str
+    verification_uri_complete: str
+    expires_in: int
+    interval: int
+
+
+@dataclass
+class TokenResponse:
+    """Response from /oauth/token endpoint."""
+
+    access_token: str
+    refresh_token: str | None
+    id_token: str | None
+    token_type: str
+    expires_in: int
+    scope: str
+
+
+class DeviceAuthFlow:
+    """Implements Auth0 Device Authorization Flow."""
+
+    def __init__(
+        self,
+        domain: str,
+        client_id: str,
+        audience: str,
+        scopes: str = "openid profile email offline_access",
+    ):
+        self.domain = domain
+        self.client_id = client_id
+        self.audience = audience
+        self.scopes = scopes
+        self.device_code_url = f"https://{domain}/oauth/device/code"
+        self.token_url = f"https://{domain}/oauth/token"
+
+    async def initiate(self) -> DeviceCodeResponse:
+        """
+        Step 1: Initiate device authorization flow.
+
+        Returns device code and user instructions.
+        """
+        async with httpx.AsyncClient() as client:
+            response = await client.post(
+                self.device_code_url,
+                data={  # Use form data instead of JSON for better compatibility
+                    "client_id": self.client_id,
+                    "scope": self.scopes,
+                    "audience": self.audience,
+                },
+            )
+            response.raise_for_status()
+            data = response.json()
+
+            return DeviceCodeResponse(
+                device_code=data["device_code"],
+                user_code=data["user_code"],
+                verification_uri=data["verification_uri"],
+                verification_uri_complete=data.get(
+                    "verification_uri_complete",
+                    f"{data['verification_uri']}?user_code={data['user_code']}",
+                ),
+                expires_in=data["expires_in"],
+                interval=data["interval"],
+            )
+
+    async def poll_for_token(
+        self, device_code: str, interval: int, timeout: int = 900
+    ) -> TokenResponse:
+        """
+        Step 2: Poll for access token.
+
+        Args:
+            device_code: Device code from initiate()
+            interval: Polling interval in seconds
+            timeout: Max time to wait in seconds
+
+        Returns:
+            Token response with access_token and refresh_token
+
+        Raises:
+            AuthenticationTimeout: If user doesn't complete auth in time
+            AuthenticationError: If auth fails or user denies
+        """
+        start_time = time.time()
+        current_interval = interval
+
+        async with httpx.AsyncClient() as client:
+            while time.time() - start_time < timeout:
+                await asyncio.sleep(current_interval)
+
+                try:
+                    response = await client.post(
+                        self.token_url,
+                        data={  # Use form data for better OAuth compatibility
+                            "grant_type": "urn:ietf:params:oauth:grant-type:device_code",
+                            "device_code": device_code,
+                            "client_id": self.client_id,
+                        },
+                    )
+
+                    # Success!
+                    if response.status_code == 200:
+                        data = response.json()
+                        return TokenResponse(
+                            access_token=data["access_token"],
+                            refresh_token=data.get("refresh_token"),
+                            id_token=data.get("id_token"),
+                            token_type=data["token_type"],
+                            expires_in=data["expires_in"],
+                            scope=data.get("scope", ""),
+                        )
+
+                    # Handle errors
+                    error_data = response.json()
+                    error = error_data.get("error")
+
+                    if error == "authorization_pending":
+                        # User hasn't completed auth yet
+                        continue
+                    elif error == "slow_down":
+                        # Increase polling interval
+                        current_interval += 5
+                        continue
+                    elif error == "expired_token":
+                        raise AuthenticationTimeout("Device code expired")
+                    elif error == "access_denied":
+                        raise AuthenticationError("User denied authorization")
+                    else:
+                        raise AuthenticationError(f"Authentication failed: {error}")
+
+                except httpx.HTTPError as e:
+                    raise AuthenticationError(f"Network error: {e}")
+
+        raise AuthenticationTimeout("Authentication timeout")
+
+    async def refresh_token(self, refresh_token: str) -> TokenResponse:
+        """
+        Refresh an expired access token.
+
+        Args:
+            refresh_token: Refresh token from previous authentication
+
+        Returns:
+            New token response
+        """
+        async with httpx.AsyncClient() as client:
+            response = await client.post(
+                self.token_url,
+                data={  # Use form data for better OAuth compatibility
+                    "grant_type": "refresh_token",
+                    "client_id": self.client_id,
+                    "refresh_token": refresh_token,
+                },
+            )
+
+            if response.status_code != 200:
+                error_data = response.json()
+                raise AuthenticationError(
+                    f"Token refresh failed: {error_data.get('error')}"
+                )
+
+            data = response.json()
+            return TokenResponse(
+                access_token=data["access_token"],
+                refresh_token=data.get(
+                    "refresh_token", refresh_token
+                ),  # May return same
+                id_token=data.get("id_token"),
+                token_type=data["token_type"],
+                expires_in=data["expires_in"],
+                scope=data.get("scope", ""),
+            )

src/asta/auth/exceptions.py

@@ -0,0 +1,13 @@
+"""Authentication exceptions."""
+
+
+class AuthenticationError(Exception):
+    """Base authentication error."""
+
+    pass
+
+
+class AuthenticationTimeout(AuthenticationError):
+    """Authentication timed out."""
+
+    pass

src/asta/auth/storage.py

@@ -0,0 +1,89 @@
+"""
+Secure token storage using platform-specific secure storage.
+
+Falls back to file-based storage with restrictive permissions.
+"""
+
+import json
+import os
+from pathlib import Path
+
+import keyring
+from platformdirs import user_config_dir
+
+APP_NAME = "asta-cli"
+TOKEN_FILE_NAME = "tokens.json"
+
+
+class TokenStorage:
+    """Manages secure storage of authentication tokens."""
+
+    def __init__(self, use_keyring: bool = True):
+        self.use_keyring = use_keyring
+        self.config_dir = Path(user_config_dir(APP_NAME, appauthor="AI2"))
+        self.token_file = self.config_dir / TOKEN_FILE_NAME
+
+        # Ensure config directory exists
+        self.config_dir.mkdir(parents=True, exist_ok=True)
+
+    def save_tokens(self, tokens: dict[str, str]) -> None:
+        """
+        Save tokens securely.
+
+        Tries system keyring first, falls back to file with restrictive permissions.
+        """
+        if self.use_keyring:
+            try:
+                keyring.set_password(APP_NAME, "tokens", json.dumps(tokens))
+                return
+            except Exception:
+                # Keyring failed, fall back to file
+                pass
+
+        # File-based storage
+        with open(self.token_file, "w") as f:
+            json.dump(tokens, f, indent=2)
+
+        # Set restrictive permissions (owner read/write only)
+        os.chmod(self.token_file, 0o600)
+
+    def load_tokens(self) -> dict[str, str] | None:
+        """Load tokens from storage."""
+        if self.use_keyring:
+            try:
+                token_json = keyring.get_password(APP_NAME, "tokens")
+                if token_json:
+                    return json.loads(token_json)
+            except Exception:
+                pass
+
+        # Try file-based storage
+        if self.token_file.exists():
+            try:
+                with open(self.token_file) as f:
+                    return json.load(f)
+            except Exception:
+                return None
+
+        return None
+
+    def delete_tokens(self) -> None:
+        """Delete stored tokens."""
+        if self.use_keyring:
+            try:
+                keyring.delete_password(APP_NAME, "tokens")
+            except Exception:
+                pass
+
+        if self.token_file.exists():
+            self.token_file.unlink()
+
+    def get_access_token(self) -> str | None:
+        """Get just the access token."""
+        tokens = self.load_tokens()
+        return tokens.get("access_token") if tokens else None
+
+    def get_refresh_token(self) -> str | None:
+        """Get just the refresh token."""
+        tokens = self.load_tokens()
+        return tokens.get("refresh_token") if tokens else None