JWT Token Authentication

import time
import requests
from typing import Optional

class TokenManager:
    """Manage JWT tokens with automatic refresh."""
    
    def __init__(self, email: str, password: str):
        self.email = email
        self.password = password
        self.base_url = "https://api.openwearables.com/api/v1"
        
        self.access_token: Optional[str] = None
        self.refresh_token: Optional[str] = None
        self.expires_at: Optional[float] = None
    
    def login(self) -> None:
        """Perform initial login."""
        response = requests.post(
            f"{self.base_url}/auth/login",
            data={
                "username": self.email,
                "password": self.password
            }
        )
        response.raise_for_status()
        
        data = response.json()
        self.access_token = data["access_token"]
        self.refresh_token = data["refresh_token"]
        self.expires_at = time.time() + data["expires_in"]
    
    def is_token_valid(self) -> bool:
        """Check if current token is still valid (with 5 min buffer)."""
        if not self.access_token or not self.expires_at:
            return False
        return time.time() < (self.expires_at - 300)  # 5 min buffer
    
    def refresh_access_token(self) -> None:
        """Refresh the access token using refresh token."""
        # Note: This endpoint is not yet implemented
        # When available, it will look like this:
        response = requests.post(
            f"{self.base_url}/auth/refresh",
            json={"refresh_token": self.refresh_token}
        )
        response.raise_for_status()
        
        data = response.json()
        self.access_token = data["access_token"]
        self.expires_at = time.time() + data["expires_in"]
    
    def get_valid_token(self) -> str:
        """Get a valid access token, refreshing if necessary."""
        if not self.is_token_valid():
            if self.refresh_token:
                try:
                    self.refresh_access_token()
                except Exception:
                    # Refresh failed, re-login
                    self.login()
            else:
                self.login()
        
        return self.access_token
    
    def logout(self) -> None:
        """Logout and clear tokens."""
        if self.access_token:
            try:
                requests.post(
                    f"{self.base_url}/auth/logout",
                    headers={"Authorization": f"Bearer {self.access_token}"}
                )
            except Exception:
                pass  # Logout failed, but clear tokens anyway
        
        self.access_token = None
        self.refresh_token = None
        self.expires_at = None

# Usage
token_manager = TokenManager(
    email="[email protected]",
    password="your-password"
)

# Get a valid token (automatically refreshes if needed)
access_token = token_manager.get_valid_token()

# Use the token
response = requests.get(
    "https://api.openwearables.com/api/v1/auth/me",
    headers={"Authorization": f"Bearer {access_token}"}
)

# Logout when done
token_manager.logout()

import { useState, useEffect } from 'react';

interface TokenData {
  access_token: string;
  refresh_token: string;
  expires_in: number;
}

interface Developer {
  id: string;
  email: string;
  first_name: string | null;
  last_name: string | null;
}

export function useAuth() {
  const [accessToken, setAccessToken] = useState<string | null>(null);
  const [developer, setDeveloper] = useState<Developer | null>(null);
  const [loading, setLoading] = useState(true);

  const baseUrl = 'https://api.openwearables.com/api/v1';

  // Check for existing token on mount
  useEffect(() => {
    const token = sessionStorage.getItem('access_token');
    if (token) {
      setAccessToken(token);
      fetchDeveloper(token);
    } else {
      setLoading(false);
    }
  }, []);

  const fetchDeveloper = async (token: string) => {
    try {
      const response = await fetch(`${baseUrl}/auth/me`, {
        headers: { 'Authorization': `Bearer ${token}` }
      });
      
      if (response.ok) {
        const dev = await response.json();
        setDeveloper(dev);
      } else {
        // Token invalid, clear it
        logout();
      }
    } catch (error) {
      console.error('Failed to fetch developer:', error);
    } finally {
      setLoading(false);
    }
  };

  const login = async (email: string, password: string): Promise<boolean> => {
    try {
      const response = await fetch(`${baseUrl}/auth/login`, {
        method: 'POST',
        headers: { 'Content-Type': 'application/x-www-form-urlencoded' },
        body: new URLSearchParams({ username: email, password })
      });

      if (!response.ok) {
        const error = await response.json();
        throw new Error(error.detail || 'Login failed');
      }

      const tokenData: TokenData = await response.json();
      
      // Store tokens
      sessionStorage.setItem('access_token', tokenData.access_token);
      sessionStorage.setItem('refresh_token', tokenData.refresh_token);
      
      setAccessToken(tokenData.access_token);
      await fetchDeveloper(tokenData.access_token);
      
      return true;
    } catch (error) {
      console.error('Login error:', error);
      return false;
    }
  };

  const logout = async () => {
    if (accessToken) {
      try {
        await fetch(`${baseUrl}/auth/logout`, {
          method: 'POST',
          headers: { 'Authorization': `Bearer ${accessToken}` }
        });
      } catch (error) {
        console.error('Logout error:', error);
      }
    }

    // Clear everything
    sessionStorage.removeItem('access_token');
    sessionStorage.removeItem('refresh_token');
    setAccessToken(null);
    setDeveloper(null);
  };

  return {
    accessToken,
    developer,
    loading,
    isAuthenticated: !!developer,
    login,
    logout
  };
}