Oauth

oauth

OAuth authentication API endpoints.

Provides endpoints for OAuth authentication flows including authorization URL generation, callback handling, and account linking.

Requires: pip install sreekarnv-fastauth[oauth] (for OAuth providers)

Classes

Functions

authorize

authorize(
    provider: str,
    request: Request,
    session: Session = Depends(get_session),
    current_user: User | None = None,
) -> OAuthAuthorizationResponse

Initiate OAuth authorization flow.

This endpoint generates a state token and authorization URL. If the user is logged in, this will be a linking flow. If not logged in, this will be a login/registration flow.

The code_verifier for PKCE is stored in the session.

Parameters:

Name	Type	Description	Default
provider_name		OAuth provider (e.g., 'google', 'github')	required
request	Request	FastAPI request object	required
session	Session	Database session	Depends(get_session)
current_user	User | None	Optional current user (if authenticated)	None

Returns:

Type	Description
OAuthAuthorizationResponse	OAuthAuthorizationResponse with authorization_url

Source code in fastauth/api/oauth.py

@router.get("/{provider}/authorize", response_model=OAuthAuthorizationResponse)
def authorize(
    provider: str,
    request: Request,
    session: Session = Depends(get_session),
    current_user: User | None = None,
) -> OAuthAuthorizationResponse:
    """
    Initiate OAuth authorization flow.

    This endpoint generates a state token and authorization URL.
    If the user is logged in, this will be a linking flow.
    If not logged in, this will be a login/registration flow.

    The code_verifier for PKCE is stored in the session.

    Args:
        provider_name: OAuth provider (e.g., 'google', 'github')
        request: FastAPI request object
        session: Database session
        current_user: Optional current user (if authenticated)

    Returns:
        OAuthAuthorizationResponse with authorization_url
    """
    provider_instance = _get_or_register_provider(provider)

    adapters = AdapterFactory(session=session)

    redirect_uri = str(request.url_for("oauth_callback", provider=provider))

    user_id = current_user.id if current_user else None

    authorization_url, state_token, code_verifier = initiate_oauth_flow(
        states=adapters.oauth_states,
        provider=provider_instance,
        redirect_uri=redirect_uri,
        user_id=user_id,
        use_pkce=True,
    )

    # NOTE: This requires SessionMiddleware to be configured
    # If SessionMiddleware is not available, we just won't store the code_verifier
    if code_verifier and "session" in request.scope:
        request.session["oauth_code_verifier"] = code_verifier
        request.session["oauth_state"] = state_token

    return OAuthAuthorizationResponse(authorization_url=authorization_url)

oauth_callback async

oauth_callback(
    provider: str,
    payload: OAuthCallbackRequest,
    request: Request,
    session: Session = Depends(get_session),
) -> TokenResponse

Handle OAuth callback after user authorization.

This endpoint: 1. Validates the state token (CSRF protection) 2. Exchanges authorization code for tokens 3. Fetches user info from provider 4. Creates or links user account 5. Issues FastAuth tokens (JWT + refresh token)

Parameters:

Name	Type	Description	Default
provider_name		OAuth provider (e.g., 'google', 'github')	required
payload	OAuthCallbackRequest	Callback request with code and state	required
request	Request	FastAPI request object	required
session	Session	Database session	Depends(get_session)

Returns:

Type	Description
TokenResponse	TokenResponse with access_token and refresh_token

Source code in fastauth/api/oauth.py

@router.post("/{provider}/callback", response_model=TokenResponse)
async def oauth_callback(
    provider: str,
    payload: OAuthCallbackRequest,
    request: Request,
    session: Session = Depends(get_session),
) -> TokenResponse:
    """
    Handle OAuth callback after user authorization.

    This endpoint:
    1. Validates the state token (CSRF protection)
    2. Exchanges authorization code for tokens
    3. Fetches user info from provider
    4. Creates or links user account
    5. Issues FastAuth tokens (JWT + refresh token)

    Args:
        provider_name: OAuth provider (e.g., 'google', 'github')
        payload: Callback request with code and state
        request: FastAPI request object
        session: Database session

    Returns:
        TokenResponse with access_token and refresh_token
    """
    provider_instance = _get_or_register_provider(provider)

    code_verifier = None
    try:
        if hasattr(request, "session") and "session" in request.scope:
            code_verifier = request.session.get("oauth_code_verifier")
            stored_state = request.session.get("oauth_state")

            if stored_state is not None:
                if stored_state != payload.state:
                    raise HTTPException(
                        status_code=status.HTTP_400_BAD_REQUEST,
                        detail="State mismatch - possible CSRF attack",
                    )

                request.session.pop("oauth_code_verifier", None)
                request.session.pop("oauth_state", None)
    except (AssertionError, KeyError):
        # SessionMiddleware not installed, skip session checks
        pass

    adapters = AdapterFactory(session=session)

    try:
        user, _ = await complete_oauth_flow(
            states=adapters.oauth_states,
            oauth_accounts=adapters.oauth_accounts,
            users=adapters.users,
            provider=provider_instance,
            code=payload.code,
            state=payload.state,
            code_verifier=code_verifier,
        )

    except OAuthStateError as e:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=str(e),
        )
    except OAuthAccountAlreadyLinkedError as e:
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail=str(e),
        )
    except OAuthError as e:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=str(e),
        )

    access_token = create_access_token(subject=str(user.id))
    refresh_token = create_refresh_token(
        refresh_tokens=adapters.refresh_tokens,
        user_id=user.id,
    )

    metadata = extract_request_metadata(request)
    create_session(
        sessions=adapters.sessions,
        users=adapters.users,
        user_id=user.id,
        ip_address=metadata.ip_address,
        user_agent=metadata.user_agent,
    )

    return TokenResponse(
        access_token=access_token,
        refresh_token=refresh_token,
    )

list_linked_accounts

list_linked_accounts(
    session: Session = Depends(get_session),
    current_user: User = Depends(get_current_user),
) -> list[OAuthLinkResponse]

List all OAuth accounts linked to the current user.

Requires authentication.

Parameters:

Name	Type	Description	Default
session	Session	Database session	Depends(get_session)
current_user	User	Current authenticated user	Depends(get_current_user)

Returns:

Type	Description
list[OAuthLinkResponse]	List of linked OAuth accounts

Source code in fastauth/api/oauth.py

@router.get("/linked", response_model=list[OAuthLinkResponse])
def list_linked_accounts(
    session: Session = Depends(get_session),
    current_user: User = Depends(get_current_user),
) -> list[OAuthLinkResponse]:
    """
    List all OAuth accounts linked to the current user.

    Requires authentication.

    Args:
        session: Database session
        current_user: Current authenticated user

    Returns:
        List of linked OAuth accounts
    """
    adapters = AdapterFactory(session=session)

    accounts = get_linked_accounts(
        oauth_accounts=adapters.oauth_accounts,
        user_id=current_user.id,
    )

    return [
        OAuthLinkResponse(
            provider=account.provider,
            email=account.email,
            name=account.name,
            linked_at=account.created_at,
        )
        for account in accounts
    ]

unlink_provider

unlink_provider(
    provider: str,
    session: Session = Depends(get_session),
    current_user: User = Depends(get_current_user),
) -> None

Unlink an OAuth provider from the current user's account.

Requires authentication.

Parameters:

Name	Type	Description	Default
provider_name		OAuth provider to unlink (e.g., 'google', 'github')	required
session	Session	Database session	Depends(get_session)
current_user	User	Current authenticated user	Depends(get_current_user)

Returns:

Type	Description
None	204 No Content on success

Source code in fastauth/api/oauth.py

@router.delete("/{provider}/unlink", status_code=status.HTTP_204_NO_CONTENT)
def unlink_provider(
    provider: str,
    session: Session = Depends(get_session),
    current_user: User = Depends(get_current_user),
) -> None:
    """
    Unlink an OAuth provider from the current user's account.

    Requires authentication.

    Args:
        provider_name: OAuth provider to unlink (e.g., 'google', 'github')
        session: Database session
        current_user: Current authenticated user

    Returns:
        204 No Content on success
    """
    adapters = AdapterFactory(session=session)

    try:
        unlink_oauth_account(
            oauth_accounts=adapters.oauth_accounts,
            user_id=current_user.id,
            provider=provider,
        )
    except OAuthError as e:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=str(e),
        )

    return None