🐍 A comprehensive Python implementation of OAuth 2.1 for educational purposes
This project provides a complete, working OAuth 2.1 implementation in Python using FastAPI, designed to help developers understand OAuth concepts through hands-on exploration. It mirrors the educational approach of the Go OAuth learning project while leveraging Python's ecosystem and modern web frameworks.
- OAuth 2.1 Authorization Code Flow with step-by-step visualization
- PKCE (Proof Key for Code Exchange) implementation and security benefits
- Three-component OAuth architecture (Client, Authorization Server, Resource Server)
- Python web development with FastAPI, Pydantic, and modern async patterns
- Security best practices including bcrypt password hashing and token validation
- Real-world OAuth integration patterns applicable to production systems
The system consists of three independent FastAPI applications that communicate via HTTP:
┌─────────────────┐ ┌──────────────────────┐ ┌─────────────────────┐
│ Client App │ │ Authorization Server │ │ Resource Server │
│ Port 8080 │◄──►│ Port 8081 │ │ Port 8082 │
│ │ │ │ │ │
│ • OAuth Flow │ │ • User Authentication│ │ • Protected Resource│
│ • PKCE Gen │ │ • Authorization Codes│ │ • Token Validation │
│ • Token Storage │ │ • Access Tokens │ │ • User Info API │
└─────────────────┘ └──────────────────────┘ └─────────────────────┘
- Initiates OAuth flow with PKCE challenge generation
- Handles authorization callbacks and token exchange
- Accesses protected resources with Bearer tokens
- Provides educational UI showing each OAuth step
- Validates authorization requests and PKCE challenges
- Authenticates users with demo accounts
- Issues authorization codes with 10-minute expiration
- Exchanges codes for access tokens after PKCE verification
- Validates Bearer tokens from Authorization headers
- Serves protected resources and user information
- Demonstrates proper token-based access control
- Provides detailed logging of resource access attempts
- Python 3.8+ (Check with
python --version) - uv package manager (recommended) or pip
# Install uv if not already installed
curl -LsSf https://astral.sh/uv/install.sh | sh
# Clone and setup project
git clone <repository-url>
cd python-oauth-learning
# Install dependencies
uv sync
# Activate virtual environment
source .venv/bin/activate # On macOS/Linux
# or
.venv\Scripts\activate # On Windows# Clone project
git clone <repository-url>
cd python-oauth-learning
# Create virtual environment
python -m venv venv
source venv/bin/activate # On macOS/Linux
# or
venv\Scripts\activate # On Windows
# Install dependencies
pip install -r requirements.txt
pip install -r requirements-dev.txt# Start all three servers with health checks and monitoring
python scripts/start_all.pyThis script will:
- ✅ Check that all ports are available
- 🚀 Start servers in the correct order with health checks
- 📊 Display real-time status and connection information
- 🔄 Monitor processes and restart if needed
- 🛑 Handle graceful shutdown with Ctrl+C
If you prefer to start servers individually:
# Terminal 1: Authorization Server
uvicorn src.auth_server.main:app --host 0.0.0.0 --port 8081 --reload
# Terminal 2: Resource Server
uvicorn src.resource_server.main:app --host 0.0.0.0 --port 8082 --reload
# Terminal 3: Client Application
uvicorn src.client.main:app --host 0.0.0.0 --port 8080 --reloadOnce all servers are running:
- 🌐 Open your browser to http://localhost:8080
- 🔑 Click "Start OAuth Flow" to begin the authorization process
- 👤 Login with demo account:
- Username:
alice, Password:password123 - Username:
bob, Password:secret456 - Username:
carol, Password:mypass789
- Username:
- 📋 Follow the step-by-step flow and observe the detailed console logging
- 🔒 Access protected resources to complete the demonstration
The client generates a PKCE challenge and redirects to the authorization server:
# Generate PKCE challenge pair
verifier, challenge = PKCEGenerator.generate_challenge()
# Build authorization URL
auth_params = {
'client_id': 'demo-client',
'redirect_uri': 'http://localhost:8080/callback',
'scope': 'read',
'state': 'demo-state-123',
'code_challenge': challenge,
'code_challenge_method': 'S256',
'response_type': 'code'
}🔍 What happens:
- Client generates cryptographically secure PKCE verifier (43 characters)
- SHA256 hash of verifier becomes the challenge
- User is redirected to authorization server with challenge
- State parameter prevents CSRF attacks
The authorization server presents a login form:
# Demo accounts with bcrypt-hashed passwords
demo_accounts = {
'alice': '$2b$12$...', # password123
'bob': '$2b$12$...', # secret456
'carol': '$2b$12$...' # mypass789
}
# Verify credentials
if verify_password(password, user['password_hash']):
# Generate authorization code
auth_code = generate_secure_token()🔍 What happens:
- User enters credentials on authorization server
- Server validates against bcrypt-hashed passwords
- Authorization code generated with 10-minute expiration
- Code tied to client_id, user, and PKCE challenge
The client exchanges the code + PKCE verifier for an access token:
# Token request with PKCE verification
token_request = {
'grant_type': 'authorization_code',
'code': authorization_code,
'redirect_uri': 'http://localhost:8080/callback',
'client_id': 'demo-client',
'code_verifier': pkce_verifier
}
# Server verifies PKCE
if PKCEGenerator.verify_challenge(verifier, stored_challenge):
access_token = generate_secure_token()🔍 What happens:
- Client sends authorization code + PKCE verifier
- Server verifies PKCE challenge matches verifier
- Access token issued only if PKCE verification succeeds
- Authorization code marked as used (one-time only)
The client uses the Bearer token to access protected resources:
# Access protected resource
headers = {'Authorization': f'Bearer {access_token}'}
response = httpx.get('http://localhost:8082/protected', headers=headers)
# Resource server validates token
def validate_bearer_token(authorization: str = Header(None)):
if not authorization or not authorization.startswith('Bearer '):
raise HTTPException(401, "Invalid Authorization header")
return authorization[7:] # Extract token🔍 What happens:
- Client includes Bearer token in Authorization header
- Resource server validates token format and presence
- Protected content served if token is valid
- All requests logged with security details
Run the complete OAuth flow programmatically:
# Test single account
python scripts/demo_flow.py --username alice
# Test all demo accounts
python scripts/demo_flow.py --test-all
# Save results to file
python scripts/demo_flow.py --test-all --output results.json
# Custom server URLs
python scripts/demo_flow.py --auth-url http://localhost:9081Generate bcrypt hashes for new demo accounts:
# Generate hashes for all demo accounts
python scripts/hash_passwords.py
# Generate hash for specific password
python scripts/hash_passwords.py --password "newpassword123"python-oauth-learning/
├── 📄 pyproject.toml # uv project configuration
├── 📄 requirements.txt # Production dependencies
├── 📄 requirements-dev.txt # Development dependencies
├── 📁 src/
│ ├── 📁 shared/ # Common utilities and models
│ │ ├── 🐍 oauth_models.py # Pydantic models for validation
│ │ ├── 🔐 crypto_utils.py # PKCE and token generation
│ │ ├── 📊 logging_utils.py # Colored console logging
│ │ └── 🛡️ security.py # Password hashing utilities
│ ├── 📁 client/ # Client application (Port 8080)
│ │ ├── 🐍 main.py # FastAPI app and configuration
│ │ ├── 🛣️ routes.py # OAuth flow endpoints
│ │ ├── 📁 templates/ # Jinja2 HTML templates
│ │ └── 📁 static/ # CSS and JavaScript files
│ ├── 📁 auth_server/ # Authorization server (Port 8081)
│ │ ├── 🐍 main.py # FastAPI app and configuration
│ │ ├── 🛣️ routes.py # OAuth authorization endpoints
│ │ ├── 💾 storage.py # In-memory user and code storage
│ │ └── 📁 templates/ # Login form templates
│ └── 📁 resource_server/ # Resource server (Port 8082)
│ ├── 🐍 main.py # FastAPI app and configuration
│ ├── 🛣️ routes.py # Protected resource endpoints
│ ├── 🛡️ middleware.py # Token validation middleware
│ └── 📁 data/ # Protected resource files
├── 📁 scripts/ # Utility and automation scripts
│ ├── 🚀 start_all.py # Multi-server startup with monitoring
│ ├── 🔐 hash_passwords.py # Password hash generation
│ └── 🤖 demo_flow.py # Automated OAuth flow testing
└── 📁 tests/ # Test suite
├── 🧪 test_crypto_utils.py # PKCE and crypto function tests
└── 🧪 test_oauth_flow.py # Integration tests
- FastAPI - Modern async web framework with automatic API docs
- Pydantic - Data validation using Python type hints
- uvicorn - ASGI server for running FastAPI applications
- Jinja2 - Template engine for HTML rendering
- passlib - Password hashing library with bcrypt support
- httpx - Async HTTP client for server-to-server communication
- colorama - Cross-platform colored terminal output
-
Generate password hash:
python scripts/hash_passwords.py --password "newpassword" -
Add to user store in
src/auth_server/storage.py:self._users = { 'alice': {'password_hash': '$2b$12$...', 'email': 'alice@example.com'}, 'newuser': {'password_hash': '$2b$12$...', 'email': 'newuser@example.com'}, }
-
Update demo account list in templates and documentation
-
Define new scopes in
src/shared/oauth_models.py:class OAuthScope(str, Enum): READ = "read" WRITE = "write" ADMIN = "admin"
-
Add scope validation in authorization server
-
Implement scope-based access control in resource server
Add new protected endpoints in src/resource_server/routes.py:
@app.get("/api/profile")
async def get_user_profile(token: str = Depends(validate_bearer_token)):
"""Get user profile information"""
# Implement profile logic
return {"profile": "user_data"}The system provides comprehensive, color-coded logging of all OAuth messages:
[2024-01-15 10:30:15] CLIENT → AUTH-SERVER
Authorization Request:
client_id: demo-client
redirect_uri: http://localhost:8080/callback
scope: read
state: demo-state-123
code_challenge: dBjftJeZ4CVP-mB92K27uhbUJU1p1r_wW1gFWFOEjXk
code_challenge_method: S256
response_type: code
--------------------------------------------------
The PKCE implementation demonstrates RFC 7636 compliance:
# Code verifier: 43-character base64url string
verifier = base64.urlsafe_b64encode(secrets.token_bytes(32)).decode('utf-8').rstrip('=')
# Code challenge: SHA256 hash of verifier
challenge = base64.urlsafe_b64encode(
hashlib.sha256(verifier.encode('utf-8')).digest()
).decode('utf-8').rstrip('=')
# Verification: constant-time comparison
def verify_challenge(verifier: str, challenge: str) -> bool:
expected = base64.urlsafe_b64encode(
hashlib.sha256(verifier.encode('utf-8')).digest()
).decode('utf-8').rstrip('=')
return secrets.compare_digest(expected, challenge)- 🔐 PKCE mandatory - All authorization flows require PKCE
- ⏰ Short-lived codes - Authorization codes expire in 10 minutes
- 🔒 Secure token generation - Cryptographically secure random tokens
- 🛡️ bcrypt password hashing - Industry-standard password protection
- 🚫 One-time code use - Authorization codes invalidated after use
- 🎲 CSRF protection - State parameter validation
- ⚡ Constant-time comparison - Prevents timing attacks
# Run all tests
pytest
# Run with coverage
pytest --cov=src
# Run specific test file
pytest tests/test_crypto_utils.py
# Run with verbose output
pytest -vUnit Tests - Test individual components:
- PKCE generation and verification
- Password hashing and validation
- OAuth model validation
- Token generation utilities
Integration Tests - Test complete flows:
- End-to-end OAuth authorization
- Multi-server communication
- Error handling scenarios
- Security validation
-
Happy Path Flow
- Complete OAuth flow with valid credentials
- Verify all steps complete successfully
- Check token-based resource access
-
Error Scenarios
- Invalid PKCE verifier
- Expired authorization code
- Missing or malformed tokens
- Invalid user credentials
-
Security Tests
- CSRF attack prevention (state parameter)
- Authorization code interception (PKCE protection)
- Token replay attacks
- Scope validation
# Check what's using the port
lsof -i :8080
# Kill the process
kill -9 <PID># Ensure you're in the project root and virtual environment is activated
pwd # Should show python-oauth-learning directory
which python # Should show virtual environment path
# Reinstall dependencies
pip install -r requirements.txt# Check server logs for startup errors
python scripts/start_all.py
# Test individual server health
curl http://localhost:8081/health
curl http://localhost:8082/health
curl http://localhost:8080/health- Ensure code verifier is stored correctly in session
- Check that challenge generation uses SHA256
- Verify base64url encoding (no padding)
- Verify demo account passwords match hashed values
- Check bcrypt hash generation
- Ensure password verification uses correct hash
Enable detailed debugging:
# Set debug environment variables
export OAUTH_DEBUG=true
export LOG_LEVEL=DEBUG
# Run with debug logging
python scripts/start_all.pySlow Startup:
- Check available system resources
- Ensure no port conflicts
- Verify network connectivity between servers
High Memory Usage:
- Monitor process memory with
toporhtop - Check for memory leaks in long-running processes
- Consider restarting servers periodically
-
Fork and clone the repository
-
Install development dependencies:
uv sync --dev # or pip install -r requirements-dev.txt -
Install pre-commit hooks:
pre-commit install
-
Run tests to ensure everything works:
pytest
The project uses:
- Black for code formatting
- Ruff for linting
- Type hints throughout the codebase
- Docstrings for all public functions and classes
# Format code
black src/ tests/ scripts/
# Lint code
ruff check src/ tests/ scripts/
# Type checking
mypy src/- Create feature branch:
git checkout -b feature/new-feature - Write tests first (TDD approach)
- Implement feature with proper documentation
- Update README if needed
- Submit pull request with clear description
This project is licensed under the MIT License - see the LICENSE file for details.
- Inspired by the Go OAuth learning implementation
- Built with modern Python web development best practices
- Designed for educational use and real-world understanding
🎓 Happy Learning! This implementation provides a solid foundation for understanding OAuth 2.1 concepts and building secure, modern web applications with Python.