oauth

name	oauth
description	OAuth 2.0 and OpenID Connect implementation
license	MIT
compatibility	opencode
metadata	{"audience":"developers","category":"security"}

What I do

• Implement OAuth 2.0 flows
• Handle OpenID Connect authentication
• Manage tokens securely
• Implement authorization grants
• Handle refresh tokens
• Validate tokens
• Implement PKCE for mobile/SPA
• Secure OAuth implementations

When to use me

When implementing OAuth authentication or authorization.

OAuth 2.0 Flows

OAuth 2.0 Grant Types:

1. Authorization Code (Web Apps)
   Client → Auth Server → User Login → Auth Code → 
   Client → Token Exchange → Access Token

2. Authorization Code + PKCE (Mobile/SPA)
   Same as above, but with code_verifier

3. Client Credentials (M2M)
   Client → Token Exchange → Access Token

4. Device Code (IoT, Smart TVs)
   Device Code Flow → User authorization on second device

5. Refresh Token
   Client → Token Refresh → New Access Token

OAuth Implementation

from dataclasses import dataclass
from typing import Optional
from datetime import datetime, timedelta
import secrets
import hashlib


@dataclass
class OAuthConfig:
    """OAuth configuration."""
    client_id: str
    client_secret: str
    authorization_endpoint: str
    token_endpoint: str
    redirect_uri: str
    scopes: list[str]
    access_token_ttl: int = 3600  # 1 hour
    refresh_token_ttl: int = 2592000  # 30 days


class OAuthClient:
    """OAuth 2.0 client implementation."""
    
    def __init__(self, config: OAuthConfig):
        self.config = config
        self.code_store = {}  # In production, use Redis
        self.token_store = {}
    
    def generate_authorization_code(self, state: str, redirect_uri: str):
        """Generate authorization code for auth code flow."""
        code = secrets.token_urlsafe(32)
        
        self.code_store[code] = {
            'state': state,
            'redirect_uri': redirect_uri,
            'scopes': self.config.scopes,
            'created_at': datetime.utcnow(),
            'expires_in': 600,  # 10 minutes
        }
        
        return code
    
    def validate_authorization_code(
        self,
        code: str,
        redirect_uri: str
    ) -> Optional[dict]:
        """Validate and consume authorization code."""
        if code not in self.code_store:
            return None
        
        code_data = self.code_store[code]
        
        # Check expiration
        elapsed = (datetime.utcnow() - code_data['created_at']).total_seconds()
        if elapsed > code_data['expires_in']:
            del self.code_store[code]
            return None
        
        # Validate redirect URI
        if code_data['redirect_uri'] != redirect_uri:
            return None
        
        # Consume code
        del self.code_store[code]
        
        return code_data
    
    def exchange_code_for_tokens(
        self,
        code: str,
        redirect_uri: str,
        client_id: str = None,
        client_secret: str = None,
        code_verifier: str = None
    ) -> dict:
        """Exchange authorization code for tokens."""
        # Validate client
        if client_id != self.config.client_id:
            raise OAuthError("invalid_client", "Invalid client ID")
        
        # Validate client secret (if required)
        if client_secret and not self._validate_client_secret(client_secret):
            raise OAuthError("invalid_client", "Invalid client secret")
        
        # Validate code
        code_data = self.validate_authorization_code(code, redirect_uri)
        if not code_data:
            raise OAuthError("invalid_grant", "Invalid or expired code")
        
        # Validate PKCE if provided
        if code_verifier and not self._validate_pkce(code, code_verifier):
            raise OAuthError("invalid_grant", "PKCE validation failed")
        
        # Generate tokens
        access_token = self._generate_access_token()
        refresh_token = self._generate_refresh_token()
        
        # Store tokens
        self.token_store[access_token] = {
            'scopes': code_data['scopes'],
            'created_at': datetime.utcnow(),
            'expires_in': self.config.access_token_ttl,
        }
        
        return {
            'access_token': access_token,
            'token_type': 'Bearer',
            'expires_in': self.config.access_token_ttl,
            'refresh_token': refresh_token,
            'scope': ' '.join(code_data['scopes']),
        }
    
    def refresh_access_token(
        self,
        refresh_token: str,
        scopes: list[str] = None
    ) -> dict:
        """Refresh access token using refresh token."""
        if refresh_token not in self.token_store:
            raise OAuthError("invalid_grant", "Invalid refresh token")
        
        token_data = self.token_store[refresh_token]
        
        # Check expiration
        elapsed = (datetime.utcnow() - token_data['created_at']).total_seconds()
        if elapsed > self.config.refresh_token_ttl:
            del self.token_store[refresh_token]
            raise OAuthError("invalid_grant", "Refresh token expired")
        
        # Validate scopes (can't request more than original)
        requested_scopes = scopes or token_data['scopes']
        for scope in requested_scopes:
            if scope not in token_data['scopes']:
                raise OAuthError("invalid_scope", "Requested scope not allowed")
        
        # Generate new tokens
        access_token = self._generate_access_token()
        new_refresh_token = self._generate_refresh_token()
        
        # Invalidate old refresh token
        del self.token_store[refresh_token]
        
        # Store new tokens
        self.token_store[access_token] = {
            'scopes': requested_scopes,
            'created_at': datetime.utcnow(),
            'expires_in': self.config.access_token_ttl,
        }
        
        return {
            'access_token': access_token,
            'token_type': 'Bearer',
            'expires_in': self.config.access_token_ttl,
            'refresh_token': new_refresh_token,
            'scope': ' '.join(requested_scopes),
        }
    
    def validate_access_token(self, token: str) -> Optional[dict]:
        """Validate access token and return claims."""
        if token not in self.token_store:
            return None
        
        token_data = self.token_store[token]
        
        elapsed = (datetime.utcnow() - token_data['created_at']).total_seconds()
        if elapsed > token_data['expires_in']:
            del self.token_store[token]
            return None
        
        return token_data
    
    def _generate_access_token(self) -> str:
        """Generate secure access token."""
        return secrets.token_urlsafe(32)
    
    def _generate_refresh_token(self) -> str:
        """Generate secure refresh token."""
        return secrets.token_urlsafe(32)
    
    def _validate_client_secret(self, secret: str) -> bool:
        """Validate client secret."""
        return secrets.compare_digest(secret, self.config.client_secret)
    
    def _validate_pkce(self, code_challenge: str, code_verifier: str) -> bool:
        """Validate PKCE code verifier against challenge."""
        # In production, store code_challenge with code
        return True


class OAuthError(Exception):
    """OAuth error."""
    
    def __init__(self, error: str, error_description: str):
        self.error = error
        self.error_description = error_description
        super().__init__(f"{error}: {error_description}")

PKCE Implementation

import base64
import hashlib
import secrets


class PKCE:
    """PKCE helper for OAuth 2.0."""
    
    @staticmethod
    def generate_code_verifier(length: int = 43) -> str:
        """
        Generate cryptographically secure code verifier.
        
        Must be 43-128 characters, using [A-Z], [a-z], [0-9], "-", ".", "_", "~"
        """
        alphabet = 'ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-._~'
        return ''.join(secrets.choice(alphabet) for _ in range(length))
    
    @staticmethod
    def generate_code_challenge(code_verifier: str) -> str:
        """
        Generate code challenge from code verifier.
        
        Uses S256 method: BASE64URL(SHA256(code_verifier))
        """
        sha256_hash = hashlib.sha256(code_verifier.encode('ascii')).digest()
        return base64.urlsafe_b64encode(sha256_hash).decode('ascii').rstrip('=')
    
    @staticmethod
    def generate_pair() -> tuple[str, str]:
        """Generate code verifier and challenge pair."""
        code_verifier = PKCE.generate_code_verifier()
        code_challenge = PKCE.generate_code_challenge(code_verifier)
        return code_verifier, code_challenge
    
    @staticmethod
    def () -> :
        
        expected_challenge = PKCE.generate_code_challenge(code_verifier)
         secrets.compare_digest(expected_challenge, code_challenge)



 ():
    
    code_verifier, code_challenge = PKCE.generate_pair()
    
    
    store_code_verifier(code_verifier)
    
    
    auth_url = (
        
        
        
        
        
        
        
        
    )
    
    
    
    tokens = client.exchange_code_for_tokens(
        code=authorization_code,
        redirect_uri=redirect_uri,
        code_verifier=code_verifier
    )

Token Validation

from jwt import jwt, JWTError
from jwt.algorithms import RSAAlgorithm


class TokenValidator:
    """Validate OAuth and JWT tokens."""
    
    def __init__(self, jwks_uri: str):
        self.jwks_uri = jwks_uri
        self.jwks_cache = {}
    
    def validate_token(self, token: str, audience: str) -> dict:
        """Validate JWT token."""
        try:
            # Get signing key
            unverified_header = jwt.get_unverified_header(token)
            kid = unverified_header.get('kid')
            
            key = self._get_signing_key(kid)
            
            # Decode and validate
            claims = jwt.decode(
                token,
                key,
                algorithms=['RS256'],
                audience=audience,
                options={
                    'verify_exp': True,
                    'verify_aud': True,
                    'verify_iat': True,
                }
            )
            
            return claims
        
        except jwt.ExpiredSignatureError:
            raise TokenValidationError("Token has expired")
        except jwt.InvalidAudienceError:
            raise TokenValidationError("Invalid audience")
        except JWTError  e:
             TokenValidationError()
    
     () -> :
        
         kid  .jwks_cache:
             .jwks_cache[kid]
        
        
        jwks = requests.get(.jwks_uri).json()
        
         key_data  jwks.get(, []):
             key_data.get() == kid:
                key = RSAAlgorithm.from_jwk(key_data)
                .jwks_cache[kid] = key
                 key
        
         TokenValidationError()

Best Practices

1. Use PKCE for all flows
   Required for mobile and SPA
   Recommended for all

2. Use short-lived access tokens
   Combine with refresh tokens

3. Validate everything
   Client ID, redirect URI, scopes, audience

4. Secure token storage
   Access tokens: memory or secure cookie
   Refresh tokens: server-side storage

5. Use state parameter
   Prevent CSRF attacks

6. Implement token binding
   Bind tokens to client

7. Log and monitor
   Track token usage

8. Rotate keys regularly
   Use JWKS for key rotation

9. Don't leak tokens in URLs
   Use POST body, not query params

10. Implement token revocation
    Support token revocation endpoint