byoi 0.1.0a1
Bring-Your-Own-Identity server library for FastAPI applications.
Requires Python: >=3.11
Keywords: fastapi, authentication, oidc, openid-connect, oauth2, identity, jwt, pkce, sso, single-sign-on, identity-provider, security

# BYOI - Bring Your Own Identity

A server library for FastAPI applications that provides OIDC (OpenID Connect) authentication with support for multiple identity providers.

## Features

- **PKCE (Proof Key for Code Exchange)** - Prevent authorization code interception attacks
- **ID Token Validation** - Full JWT validation with JWKS fetching and caching
- **Identity Linking** - Users can link multiple providers (Google + Microsoft, etc.)
- **Nonce Validation** - Prevents replay attacks
- **Proper Error Handling** - Clean error responses throughout
- **FastAPI Dependencies** - Easy integration with dependency injection
- **Multiple Client Types** - Support for web, desktop, and mobile applications
- **Flexible Caching** - In-memory and Redis cache implementations
- **OpenTelemetry Support** - Optional tracing integration for observability

## Installation

```bash
pip install byoi
```

For Redis cache support:
```bash
pip install byoi[redis]
```

For OpenTelemetry support:
```bash
pip install byoi[telemetry]
```

For all optional dependencies:
```bash
pip install byoi[all]
```

## Quick Start

### 1. Implement the Repository Protocols

BYOI requires you to implement three repository protocols for data persistence:

```python
from datetime import datetime
from typing import Any
from byoi import (
    UserProtocol,
    LinkedIdentityProtocol,
    AuthStateProtocol,
    UserRepositoryProtocol,
    LinkedIdentityRepositoryProtocol,
    AuthStateRepositoryProtocol,
)

# Example User implementation
class User:
    def __init__(self, id: str, email: str | None, is_active: bool = True):
        self.id = id
        self.email = email
        self.is_active = is_active

# Example UserRepository implementation
class UserRepository:
    def __init__(self):
        self._users: dict[str, User] = {}
    
    async def get_by_id(self, user_id: str) -> User | None:
        return self._users.get(user_id)
    
    async def get_by_email(self, email: str) -> User | None:
        for user in self._users.values():
            if user.email == email:
                return user
        return None
    
    async def create(
        self,
        email: str | None = None,
        is_active: bool = True,
        **extra_data: Any,
    ) -> User:
        import uuid
        user = User(id=str(uuid.uuid4()), email=email, is_active=is_active)
        self._users[user.id] = user
        return user
    
    async def update(self, user_id: str, **data: Any) -> User | None:
        user = self._users.get(user_id)
        if user:
            for key, value in data.items():
                setattr(user, key, value)
        return user

# Implement LinkedIdentityRepository and AuthStateRepository similarly...
```

### 2. Configure BYOI

```python
from fastapi import FastAPI
from byoi import BYOI, BYOIConfig, OIDCProviderConfig

# Create repositories
user_repo = UserRepository()
identity_repo = LinkedIdentityRepository()
auth_state_repo = AuthStateRepository()

# Configure BYOI
config = BYOIConfig(
    user_repository=user_repo,
    identity_repository=identity_repo,
    auth_state_repository=auth_state_repo,
    # cache is optional - defaults to InMemoryCache if not provided
    providers=[
        OIDCProviderConfig(
            name="google",
            display_name="Google",
            issuer="https://accounts.google.com",
            client_id="your-google-client-id",
            client_secret="your-google-client-secret",
        ),
        OIDCProviderConfig(
            name="microsoft",
            display_name="Microsoft",
            issuer="https://login.microsoftonline.com/{tenant}/v2.0",
            client_id="your-microsoft-client-id",
            client_secret="your-microsoft-client-secret",
        ),
    ],
)

# Create BYOI instance
byoi = BYOI(config)

# Create FastAPI app with BYOI lifespan
app = FastAPI(lifespan=byoi.lifespan)
```

### 3. Create Authentication Routes

```python
from fastapi import FastAPI, HTTPException
from byoi import (
    AuthServiceDep,
    ProviderManagerDep,
    AuthorizationRequest,
    TokenExchangeRequest,
    ClientType,
    InvalidStateError,
    StateExpiredError,
    TokenExchangeError,
)

@app.get("/auth/providers")
async def list_providers(provider_manager: ProviderManagerDep):
    """List all configured identity providers."""
    return provider_manager.list_providers()


@app.post("/auth/authorize")
async def create_authorization_url(
    provider_name: str,
    redirect_uri: str,
    auth_service: AuthServiceDep,
):
    """Create an authorization URL for OAuth flow."""
    request = AuthorizationRequest(
        provider_name=provider_name,
        redirect_uri=redirect_uri,
        client_type=ClientType.WEB,
    )
    return await auth_service.create_authorization_url(request)


@app.post("/auth/callback")
async def handle_callback(
    code: str,
    state: str,
    redirect_uri: str,
    auth_service: AuthServiceDep,
):
    """Handle OAuth callback and authenticate user."""
    try:
        request = TokenExchangeRequest(
            code=code,
            state=state,
            redirect_uri=redirect_uri,
        )
        user = await auth_service.authenticate(request)
        # Create your session/JWT here
        return {"user_id": user.user_id, "is_new": user.is_new_user}
    except InvalidStateError:
        raise HTTPException(400, "Invalid state parameter")
    except StateExpiredError:
        raise HTTPException(400, "Authorization request expired")
    except TokenExchangeError as e:
        raise HTTPException(400, f"Authentication failed: {e.message}")
```

## Identity Linking

Allow users to link multiple identity providers to their account:

```python
@app.post("/auth/link")
async def link_identity(
    user_id: str,  # From your session/JWT
    code: str,
    state: str,
    redirect_uri: str,
    auth_service: AuthServiceDep,
):
    """Link a new identity provider to user's account."""
    from byoi import LinkIdentityRequest, IdentityAlreadyLinkedError
    
    try:
        request = LinkIdentityRequest(
            user_id=user_id,
            code=code,
            state=state,
            redirect_uri=redirect_uri,
        )
        return await auth_service.link_identity(request)
    except IdentityAlreadyLinkedError:
        raise HTTPException(400, "This identity is already linked to another account")


@app.get("/auth/identities/{user_id}")
async def get_user_identities(
    user_id: str,
    auth_service: AuthServiceDep,
):
    """Get all linked identities for a user."""
    return await auth_service.get_user_identities(user_id)


@app.delete("/auth/identities/{user_id}/{identity_id}")
async def unlink_identity(
    user_id: str,
    identity_id: str,
    auth_service: AuthServiceDep,
):
    """Unlink an identity from a user's account."""
    from byoi import UnlinkIdentityRequest
    
    request = UnlinkIdentityRequest(user_id=user_id, identity_id=identity_id)
    return await auth_service.unlink_identity(request)
```

## Caching

BYOI supports multiple cache implementations:

### In-Memory Cache (Default)

```python
from byoi import InMemoryCache

cache = InMemoryCache(cleanup_interval_seconds=300)
```

### Redis Cache

```python
from byoi import RedisCache

cache = RedisCache(
    url="redis://localhost:6379/0",
    key_prefix="byoi:",
    max_connections=10,  # optional
)
```

### Null Cache

A no-op cache implementation that doesn't cache anything. Useful for testing or when caching is not desired.

```python
from byoi import NullCache

cache = NullCache()
```

### Custom Cache

Implement the `CacheProtocol`:

```python
from byoi import CacheProtocol

class MyCache(CacheProtocol):
    async def get(self, key: str) -> Any | None:
        ...
    
    async def set(self, key: str, value: Any, ttl_seconds: int | None = None) -> None:
        ...
    
    async def delete(self, key: str) -> bool:
        ...
    
    async def clear(self) -> None:
        ...
    
    async def close(self) -> None:
        ...
```

## Adding Custom Providers

```python
from byoi import OIDCProviderConfig

# Standard OIDC provider
okta_config = OIDCProviderConfig(
    name="okta",
    display_name="Okta",
    issuer="https://your-domain.okta.com",
    client_id="your-client-id",
    client_secret="your-client-secret",
)

# Custom provider with overrides
custom_config = OIDCProviderConfig(
    name="custom",
    display_name="Custom IDP",
    issuer="https://idp.example.com",
    client_id="client-id",
    client_secret="client-secret",
    # Override discovered endpoints if needed
    authorization_endpoint_override="https://idp.example.com/oauth/authorize",
    token_endpoint_override="https://idp.example.com/oauth/token",
    jwks_uri_override="https://idp.example.com/.well-known/jwks.json",
    # Custom scopes
    scopes=["openid", "email", "profile", "custom_scope"],
    # Extra auth parameters
    extra_auth_params={"prompt": "consent"},
)

# Add providers after setup
byoi.add_provider(custom_config)
```

## Error Handling

BYOI provides specific exception types for different error scenarios:

```python
from byoi import (
    BYOIError,                    # Base exception for all BYOI errors
    ConfigurationError,           # Configuration problems
    # Provider errors
    ProviderError,                # Base class for provider-related errors
    ProviderNotFoundError,        # Unknown provider
    ProviderDiscoveryError,       # OIDC discovery failed
    # Authentication errors
    AuthenticationError,          # Base class for authentication errors
    InvalidStateError,            # Invalid OAuth state
    StateExpiredError,            # OAuth state expired
    InvalidCodeError,             # Invalid authorization code
    TokenExchangeError,           # Token exchange failed
    TokenRefreshError,            # Token refresh failed
    # Token validation errors
    TokenValidationError,         # ID token validation failed
    InvalidNonceError,            # Nonce mismatch
    InvalidIssuerError,           # Issuer mismatch
    InvalidAudienceError,         # Audience mismatch
    TokenExpiredError,            # Token has expired
    JWKSFetchError,               # Failed to fetch JWKS
    # Identity errors
    IdentityError,                # Base class for identity errors
    IdentityAlreadyLinkedError,   # Identity already linked to another account
    IdentityNotFoundError,        # Identity not found
    CannotUnlinkLastIdentityError,  # Cannot unlink the last identity
    UserNotFoundError,            # User not found
    # Cache errors
    CacheError,                   # Cache operation failed
)
```

## Protocols Reference

### UserProtocol

```python
class UserProtocol(Protocol):
    @property
    def id(self) -> str: ...
    
    @property
    def email(self) -> str | None: ...
    
    @property
    def is_active(self) -> bool: ...
```

### LinkedIdentityProtocol

```python
class LinkedIdentityProtocol(Protocol):
    @property
    def id(self) -> str: ...
    
    @property
    def user_id(self) -> str: ...
    
    @property
    def provider_name(self) -> str: ...
    
    @property
    def provider_subject(self) -> str: ...
    
    @property
    def email(self) -> str | None: ...
    
    @property
    def created_at(self) -> datetime: ...
    
    @property
    def last_used_at(self) -> datetime | None: ...
```

### AuthStateProtocol

```python
class AuthStateProtocol(Protocol):
    @property
    def state(self) -> str: ...
    
    @property
    def code_verifier(self) -> str: ...
    
    @property
    def nonce(self) -> str: ...
    
    @property
    def provider_name(self) -> str: ...
    
    @property
    def redirect_uri(self) -> str: ...
    
    @property
    def created_at(self) -> datetime: ...
    
    @property
    def expires_at(self) -> datetime: ...
    
    @property
    def client_type(self) -> str: ...
    
    @property
    def extra_data(self) -> dict[str, Any]: ...
```

## License

MIT License - see LICENSE file for details.