Mock Google OAuth Implementation

Welcome to the second lesson of the "OAuth Fundamentals & Mock Implementation" course! In this lesson, we'll focus on creating a mock Google OAuth implementation. This approach allows us to simulate real-world OAuth interactions in a controlled environment, reinforcing your understanding of OAuth processes without needing actual third-party services. By the end of this lesson, you'll be equipped to create a mock OAuth flow, enhancing your grasp of how OAuth—an authorization framework—can be leveraged for authentication purposes. Let's get started! 🚀

🔒 Security Note for This Unit:

In this lesson, we're implementing a simplified OAuth flow to focus on understanding the core concepts. We'll use basic patterns like /success routes and tokens in URLs. In Unit 4, we'll upgrade to production-ready security with CSRF protection (state parameters). For now, focus on understanding how OAuth connects the pieces together.

⚠️ CRITICAL SECURITY WARNING - Tokens in URL Parameters:

Our current implementation passes tokens like /?token=xxx in URL query parameters. This is unsafe for production:

• ❌ Tokens appear in browser history
• ❌ Tokens leak through server logs
• ❌ Tokens exposed via HTTP Referer headers
• ❌ Tokens can be bookmarked and shared

Production solutions:

• ✅ Use HttpOnly, Secure cookies (recommended)
• ✅ Use Authorization Code + PKCE and deliver session via HttpOnly, Secure, SameSite cookies; avoid fragment/implicit flows
• ✅ Never use query parameters for sensitive tokens

We use query parameters in this demo for educational clarity only. In Unit 5, we'll discuss secure token delivery patterns.

In a mock Google OAuth implementation, we simulate the process of authenticating users through Google without actually connecting to Google's servers. This is useful for learning and testing purposes, as it allows us to understand the OAuth flow without the complexity of real-world integrations.

We implement a basic mock Google OAuth provider using pre-built HTML templates to focus on the OAuth authorization code flow logic. The mock implementation involves template loading, OAuth route structure, and JWT token integration with OAuth-authenticated users.

The first step is loading and serving OAuth consent screens. We use pre-built HTML templates that simulate the Google OAuth interface:

from fastapi import APIRouter
from fastapi.responses import FileResponse
import os

router = APIRouter()

# Helper function to get template path
def get_template_path(filename: str) -> str:
    return os.path.join(os.path.dirname(__file__), '..', 'templates', filename)

This code provides a helper function to construct the path to our HTML template files. The template file contains the mock Google consent screen that users will interact with. By using a separate template file, we keep the OAuth logic clean and focused, while the HTML handles the user interface. The os.path.join() function safely constructs the path relative to the current file location.

OAuth implementations follow a standard route pattern for handling the authorization flow:

# Mock Google OAuth consent screen
@router.get("/google")
async def google_oauth_login():
    template_path = get_template_path('google-oauth.html')
    return FileResponse(template_path)

# Google OAuth success handler
@router.get("/google/success")
async def google_oauth_success():
    # Handle OAuth callback logic
    pass

This structure provides two essential endpoints:

• /google - Serves the consent screen template.
• /google/success - Handles the OAuth callback after user consent.

The clean route structure keeps the focus on OAuth flow logic rather than HTML details. We use FastAPI's @router.get() decorator to define route handlers and FileResponse to serve the HTML template directly to the user's browser.

The OAuth authorization code flow follows these steps in our mock implementation:

1. User visits /google. The consent screen is displayed.
2. User grants permission. The template form submits to /google/success.
3. Server processes callback. The user is created/found and a token is generated.
4. User is redirected. Back to the application with the authentication token.

Important Security Note: In production OAuth implementations, you must include a state parameter to protect against CSRF attacks. This parameter should be:

• Generated randomly when displaying the consent screen
• Stored temporarily (e.g., in session or Redis)
• Validated when processing the callback
• Removed after successful validation

We're omitting state validation in this mock implementation to keep the focus on understanding the basic OAuth flow, but never skip this in real applications—it's a critical security requirement.

The callback handler processes the OAuth response with mock data:

from fastapi import Depends
from fastapi.responses import RedirectResponse
from sqlalchemy.future import select
from sqlalchemy.ext.asyncio import AsyncSession
from jose import jwt
from datetime import datetime, timedelta

from ..database import get_db
from ..models.user import User
from ..config import get_jwt_secret

@router.get("/google/success")
async def google_oauth_success(db: AsyncSession = Depends(get_db)):
    try:
        # Create or find a mock Google user
        result = await db.execute(select(User).where(User.username == 'google_user'))
        user = result.scalar_one_or_none()
        
        if not user:
            user = User(
                username='google_user',
                password=None  # OAuth users don't need passwords
            )
            db.add(user)
            await db.commit()
            await db.refresh(user)
        
        # Generate JWT token with expiration (24 hours)
        token_data = {
            'userId': user.id,
            'exp': datetime.utcnow() + timedelta(hours=24)
        }
        token = jwt.encode(token_data, get_jwt_secret(), algorithm='HS256')
        
        # ⚠️ SECURITY WARNING: Using query parameters for tokens is unsafe!
        # This is for educational purposes only - see security note above
        return RedirectResponse(url=f"/?token={token}&provider=google")
    except Exception as error:
        print(f'OAuth error: {error}')
        return RedirectResponse(url='/login?error=oauth_failed')

OAuth users are handled differently than regular users since they authenticate through external providers:

# Create or find a mock Google user
result = await db.execute(select(User).where(User.username == 'google_user'))
user = result.scalar_one_or_none()

if not user:
    user = User(
        username='google_user',
        password=None  # OAuth users don't need passwords
    )
    db.add(user)
    await db.commit()
    await db.refresh(user)

Key aspects of OAuth user creation:

• No password required. OAuth users authenticate through Google.
• User lookup. Check if the user already exists before creating.
• Simple user data. The mock implementation uses basic user information.

We use SQLAlchemy's async operations with select() to query the database and scalar_one_or_none() to get either the user or None if not found.

After successful OAuth authentication, we generate a JWT token for the user session:

from jose import jwt
from datetime import datetime, timedelta
from ..config import get_jwt_secret

# Generate JWT token with expiration (security best practice)
# Tokens expire after 24 hours to prevent replay attacks
token_data = {
    'userId': user.id,
    'exp': datetime.utcnow() + timedelta(hours=24)
}
token = jwt.encode(token_data, get_jwt_secret(), algorithm='HS256')

# ⚠️ SECURITY WARNING: Passing tokens in URLs is unsafe (see note above)
return RedirectResponse(url=f"/?token={token}&provider=google")

This integration:

• Uses existing JWT system. The same token generation as regular authentication.
• Includes user ID. Links the token to the authenticated user.
• Adds expiration claim. Tokens expire after 24 hours for security.
• Uses environment-based secret. get_jwt_secret() loads from environment variables.

We use the jose library for JWT operations, which provides jwt.encode() to create tokens with the HS256 algorithm. The exp claim ensures tokens don't live forever, preventing replay attacks.

Why we use get_jwt_secret() instead of a constant:

• Secrets should never be hard-coded in source code
• Environment variables allow different secrets per environment (dev/staging/production)
• Supports secret rotation without code changes
• Follows the "12-factor app" methodology for configuration

Here's the complete OAuth route implementation:

from fastapi import APIRouter, Depends
from fastapi.responses import FileResponse, RedirectResponse
from sqlalchemy.future import select
from sqlalchemy.ext.asyncio import AsyncSession
from jose import jwt
from datetime import datetime, timedelta
import os

from ..database import get_db
from ..models.user import User
from ..config import get_jwt_secret

router = APIRouter()

# Load HTML template path
def get_template_path(filename: str) -> str:
    return os.path.join(os.path.dirname(__file__), '..', 'templates', filename)

# Mock Google OAuth consent screen
@router.get("/google")
async def google_oauth_login():
    template_path = get_template_path('google-oauth.html')
    return FileResponse(template_path)

# Google OAuth success handler
@router.get("/google/success")
async def google_oauth_success(db: AsyncSession = Depends(get_db)):
    try:
        # Create or find a mock Google user
        result = await db.execute(select(User).where(User.username == 'google_user'))
        user = result.scalar_one_or_none()
        
        if not user:
            user = User(
                username='google_user',
                password=None  # OAuth users don't need passwords
            )
            db.add(user)
            await db.commit()
            await db.refresh(user)
        
        # Generate JWT token with expiration (24 hours)
        token_data = {
            'userId': user.id,
            'exp': datetime.utcnow() + timedelta(hours=24)
        }
        token = jwt.encode(token_data, get_jwt_secret(), algorithm='HS256')
        
        # ⚠️ SECURITY WARNING: Passing tokens in URLs is unsafe!
        # Production: Use HttpOnly, Secure cookies instead
        return RedirectResponse(url=f"/?token={token}&provider=google")
    except Exception as error:
        print(f'OAuth error: {error}')
        return RedirectResponse(url='/login?error=oauth_failed')

To test the complete OAuth flow:

1. Navigate to /google. See the consent screen.
2. Click the authorization button. The form submits to /google/success.
3. User is created/found. Database interaction occurs.
4. JWT token is generated. The user session is established.
5. Redirect to main app. The user is authenticated with the token.

The HTML template handles the user interface, while your route logic manages the authentication flow.

In this lesson, we explored how to create a mock Google OAuth implementation focusing on the core OAuth authorization code flow logic. We learned about template loading, OAuth route structure, user creation for OAuth users, and JWT token integration with proper expiration.

Important security reminders:

• Tokens in URL parameters are unsafe - we'll discuss secure patterns in Unit 5
• Always use environment-based secrets (get_jwt_secret()) not hard-coded values
• JWT tokens should always include exp claim to prevent replay attacks

You now understand how these components work together to create a complete OAuth flow. As you move on to the practice exercises, you'll implement these concepts and reinforce your understanding by building the OAuth routes yourself. Keep up the great work! 🌟