API Authentication

import os
import httpx

# Reading API keys from environment
api_key = os.environ.get("MY_API_KEY")
if not api_key:
    raise ValueError("MY_API_KEY environment variable not set")

# API key in header (most common)
resp = httpx.get(
    "https://api.example.com/data",
    headers={"X-API-Key": api_key},
)

# API key as query parameter (some services)
resp = httpx.get(
    "https://api.example.com/data",
    params={"api_key": api_key},
)

# GitHub style — token prefix
resp = httpx.get(
    "https://api.github.com/user",
    headers={"Authorization": f"token {api_key}"},
)

import httpx

def get_bearer_token(client_id: str, client_secret: str, token_url: str) -> str:
    """Get a bearer token using client credentials."""
    resp = httpx.post(
        token_url,
        data={
            "grant_type": "client_credentials",
            "client_id": client_id,
            "client_secret": client_secret,
        },
    )
    resp.raise_for_status()
    return resp.json()["access_token"]

# Use the token
token = get_bearer_token(
    client_id="my-client-id",
    client_secret="my-client-secret",
    token_url="https://auth.example.com/oauth/token",
)

resp = httpx.get(
    "https://api.example.com/data",
    headers={"Authorization": f"Bearer {token}"},
)

from fastapi import FastAPI
from fastapi.responses import RedirectResponse
import httpx

app = FastAPI()

CLIENT_ID = "my-client-id"
CLIENT_SECRET = "my-client-secret"
AUTH_URL = "https://auth.example.com/authorize"
TOKEN_URL = "https://auth.example.com/token"
REDIRECT_URI = "http://localhost:8000/callback"

@app.get("/login")
def login():
    """Redirect to OAuth2 provider."""
    params = {
        "response_type": "code",
        "client_id": CLIENT_ID,
        "redirect_uri": REDIRECT_URI,
        "scope": "read write",
        "state": "random-state-string",
    }
    query = "&".join(f"{k}={v}" for k, v in params.items())
    return RedirectResponse(f"{AUTH_URL}?{query}")

@app.get("/callback")
def callback(code: str, state: str):
    """Exchange authorization code for tokens."""
    resp = httpx.post(TOKEN_URL, data={
        "grant_type": "authorization_code",
        "code": code,
        "redirect_uri": REDIRECT_URI,
        "client_id": CLIENT_ID,
        "client_secret": CLIENT_SECRET,
    })
    resp.raise_for_status()
    tokens = resp.json()
    return {
        "access_token": tokens["access_token"],
        "refresh_token": tokens.get("refresh_token"),
    }

import jwt
from datetime import datetime

SECRET_KEY = "your-secret-key"  # In production, use an asymmetric key

def validate_jwt(token: str) -> dict:
    """Validate a JWT token and return the payload."""
    try:
        payload = jwt.decode(
            token,
            SECRET_KEY,
            algorithms=["HS256"],
            options={"require": ["exp", "sub", "iss"]},
        )
        return payload
    except jwt.ExpiredSignatureError:
        raise ValueError("Token has expired")
    except jwt.InvalidTokenError as e:
        raise ValueError(f"Invalid token: {e}")

# FastAPI dependency
from fastapi import Depends, HTTPException
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials

security = HTTPBearer()

def get_current_user(
    credentials: HTTPAuthorizationCredentials = Depends(security),
) -> dict:
    try:
        return validate_jwt(credentials.credentials)
    except ValueError as e:
        raise HTTPException(status_code=401, detail=str(e))

import time
import httpx
from pathlib import Path
import json

class TokenManager:
    def __init__(self, token_url: str, client_id: str, client_secret: str):
        self.token_url = token_url
        self.client_id = client_id
        self.client_secret = client_secret
        self.token_file = Path.home() / ".config" / "myapp" / "tokens.json"
        self._tokens = self._load_tokens()

    def _load_tokens(self) -> dict:
        if self.token_file.exists():
            return json.loads(self.token_file.read_text())
        return {}

    def _save_tokens(self) -> None:
        self.token_file.parent.mkdir(parents=True, exist_ok=True)
        self.token_file.write_text(json.dumps(self._tokens))
        self.token_file.chmod(0o600)  # Restrict permissions

    def get_token(self) -> str:
        if self._is_expired():
            self._refresh()
        return self._tokens["access_token"]

    def _is_expired(self) -> bool:
        expires_at = self._tokens.get("expires_at", 0)
        return time.time() >= expires_at - 60  # Refresh 1 min early

    def _refresh(self) -> None:
        resp = httpx.post(self.token_url, data={
            "grant_type": "refresh_token",
            "refresh_token": self._tokens["refresh_token"],
            "client_id": self.client_id,
            "client_secret": self.client_secret,
        })
        resp.raise_for_status()
        data = resp.json()
        self._tokens = {
            "access_token": data["access_token"],
            "refresh_token": data.get("refresh_token", self._tokens["refresh_token"]),
            "expires_at": time.time() + data["expires_in"],
        }
        self._save_tokens()