
Research
Two Malicious Rust Crates Impersonate Popular Logger to Steal Wallet Keys
Socket uncovers malicious Rust crates impersonating fast_log to steal Solana and Ethereum wallet keys from source code.
auth-middleware
Advanced tools
Async Authentication and Authorization Middleware for FastAPI/Starlette Applications
Auth Middleware is a comprehensive, production-ready authentication and authorization solution for FastAPI and Starlette applications. It provides a pluggable architecture that supports multiple identity providers and authorization backends with async-first design principles.
pip install auth-middleware
poetry add auth-middleware
uv add auth-middleware
from fastapi import FastAPI, Depends, Request
from auth_middleware import JwtAuthMiddleware
from auth_middleware.functions import require_user, require_groups
from auth_middleware.providers.authn.cognito_provider import CognitoProvider
from auth_middleware.providers.authn.cognito_authz_provider_settings import CognitoAuthzProviderSettings
app = FastAPI()
# Configure authentication provider
auth_settings = CognitoAuthzProviderSettings(
user_pool_id="us-east-1_abcdef123",
user_pool_region="us-east-1",
jwt_token_verification_disabled=False,
)
# Add authentication middleware
app.add_middleware(
JwtAuthMiddleware,
auth_provider=CognitoProvider(settings=auth_settings),
)
# Protected endpoint - requires valid authentication
@app.get("/protected", dependencies=[Depends(require_user())])
async def protected_endpoint(request: Request):
user = request.state.current_user
return {
"message": f"Hello {user.name}",
"user_id": user.id,
"email": user.email
}
# Role-based endpoint - requires specific groups
@app.get("/admin", dependencies=[Depends(require_groups(["admin", "moderator"]))])
async def admin_endpoint(request: Request):
user = request.state.current_user
groups = await user.groups
return {
"message": "Admin access granted",
"user_groups": groups
}
from auth_middleware.providers.authn.entra_id_provider import EntraIdProvider
from auth_middleware.providers.authn.entra_id_provider_settings import EntraIdProviderSettings
# Configure Azure Entra ID
entra_settings = EntraIdProviderSettings(
tenant_id="your-tenant-id",
client_id="your-client-id",
issuer="https://sts.windows.net/your-tenant-id/",
)
app.add_middleware(
JwtAuthMiddleware,
auth_provider=EntraIdProvider(settings=entra_settings),
)
from auth_middleware.providers.authn.jwt_provider import JWTProvider
from auth_middleware.providers.authn.jwt_provider_settings import JWTProviderSettings
# Configure generic JWT provider
jwt_settings = JWTProviderSettings(
secret_key="your-secret-key",
algorithm="HS256",
issuer="your-issuer",
)
app.add_middleware(
JwtAuthMiddleware,
auth_provider=JWTProvider(settings=jwt_settings),
)
Configure the middleware using environment variables:
# Core middleware settings
AUTH_MIDDLEWARE_DISABLED=false
AUTH_MIDDLEWARE_LOG_LEVEL=INFO
# JWKS caching configuration
AUTH_MIDDLEWARE_JWKS_CACHE_INTERVAL_MINUTES=20
AUTH_MIDDLEWARE_JWKS_CACHE_USAGES=1000
# AWS Cognito settings (if using Cognito provider)
COGNITO_USER_POOL_ID=us-east-1_abcdef123
COGNITO_USER_POOL_REGION=us-east-1
# Azure Entra ID settings (if using Entra ID provider)
AZURE_TENANT_ID=your-tenant-id
AZURE_CLIENT_ID=your-client-id
AZURE_CLIENT_SECRET=your-client-secret
# JWT settings (if using generic JWT provider)
JWT_SECRET_KEY=your-secret-key
JWT_ALGORITHM=HS256
JWT_ISSUER=your-issuer
from auth_middleware.providers.authz.sql_groups_provider import SqlGroupsProvider
from auth_middleware.providers.authz.sql_permissions_provider import SqlPermissionsProvider
# Setup with database-backed authorization
auth_provider = CognitoProvider(
settings=auth_settings,
groups_provider=SqlGroupsProvider(),
permissions_provider=SqlPermissionsProvider(),
)
app.add_middleware(JwtAuthMiddleware, auth_provider=auth_provider)
# Use permission-based authorization
from auth_middleware.functions import require_permissions
@app.post("/admin/users", dependencies=[Depends(require_permissions(["user.create", "user.manage"]))])
async def create_user(request: Request):
user = request.state.current_user
permissions = await user.permissions
return {"message": "User creation allowed", "permissions": permissions}
# Get JWT token from your identity provider, then:
curl -H "Authorization: Bearer YOUR_JWT_TOKEN" \
http://localhost:8000/protected
@app.get("/profile")
async def get_profile(request: Request):
user = request.state.current_user
return {
"id": user.id,
"name": user.name,
"email": user.email,
"groups": await user.groups,
"permissions": await user.permissions
}
from auth_middleware.functions import require_user
@app.delete("/posts/{post_id}")
async def delete_post(post_id: str, request: Request, _: None = Depends(require_user())):
user = request.state.current_user
post = await get_post(post_id)
# Custom authorization logic
user_groups = await user.groups
if post.owner_id != user.id and "admin" not in user_groups:
raise HTTPException(status_code=403, detail="Not authorized to delete this post")
await delete_post_by_id(post_id)
return {"message": "Post deleted successfully"}
Component | Description |
---|---|
JwtAuthMiddleware | Main ASGI middleware for request authentication |
JWTProvider | Abstract base for authentication providers |
GroupsProvider | Abstract base for group-based authorization |
PermissionsProvider | Abstract base for permission-based authorization |
User | Pydantic model representing authenticated users |
The authenticated user is available in request.state.current_user
:
Property | Type | Description |
---|---|---|
id | str | Unique user identifier from identity provider |
name | str | None | User's display name |
email | EmailStr | None | User's email address |
groups | list[str] | User groups (async property) |
permissions | list[str] | User permissions (async property) |
Function | Description |
---|---|
require_user() | Ensures user is authenticated |
require_groups(groups) | Requires user to be in specified groups |
require_permissions(permissions) | Requires user to have specified permissions |
graph TD
A[HTTP Request] --> B[JwtAuthMiddleware]
B --> C[JWTBearerManager]
C --> D[JWT Provider]
D --> E[Token Validation]
E --> F[User Creation]
F --> G[Groups Provider]
F --> H[Permissions Provider]
G --> I[request.state.current_user]
H --> I
# Clone the repository
git clone https://github.com/impalah/auth-middleware.git
cd auth-middleware
# Create virtual environment and install dependencies
make venv
# Run tests
make test
# Run type checking
make type-check
# Run linting
make lint
# Run all quality checks
make check
# Build package
make build
# Publish to PyPI
make publish
# Build Docker image
make docker-build
We welcome contributions! Please see our Contributing Guide for details.
git checkout -b feature/amazing-feature
)git commit -m 'Add amazing feature'
)git push origin feature/amazing-feature
)This project is licensed under the MIT License - see the LICENSE file for details.
Created with ❤️ by impalah
Made for modern Python async applications 🐍⚡
FAQs
Async Auth Middleware for FastAPI/Starlette
We found that auth-middleware demonstrated a healthy version release cadence and project activity because the last version was released less than a year ago. It has 1 open source maintainer collaborating on the project.
Did you know?
Socket for GitHub automatically highlights issues in each pull request and monitors the health of all your open source dependencies. Discover the contents of your packages and block harmful activity before you install or update your dependencies.
Research
Socket uncovers malicious Rust crates impersonating fast_log to steal Solana and Ethereum wallet keys from source code.
Research
A malicious package uses a QR code as steganography in an innovative technique.
Research
/Security News
Socket identified 80 fake candidates targeting engineering roles, including suspected North Korean operators, exposing the new reality of hiring as a security function.