|
| 1 | +from fastapi import FastAPI, HTTPException, Request |
| 2 | +from pydantic import BaseModel |
| 3 | + |
| 4 | +from authx import AuthX, AuthXConfig |
| 5 | + |
| 6 | +# Create a FastAPI app |
| 7 | +app = FastAPI(title="AuthX Fresh Token Example") |
| 8 | + |
| 9 | +# Configure AuthX |
| 10 | +auth_config = AuthXConfig( |
| 11 | + JWT_ALGORITHM="HS256", |
| 12 | + JWT_SECRET_KEY="your-secret-key", # In production, use a secure key and store it in environment variables |
| 13 | + JWT_TOKEN_LOCATION=["headers"], |
| 14 | + JWT_HEADER_TYPE="Bearer", |
| 15 | +) |
| 16 | + |
| 17 | +# Initialize AuthX |
| 18 | +auth = AuthX(config=auth_config) |
| 19 | + |
| 20 | +# Register error handlers |
| 21 | +auth.handle_errors(app) |
| 22 | + |
| 23 | + |
| 24 | +# Define models |
| 25 | +class User(BaseModel): |
| 26 | + username: str |
| 27 | + password: str |
| 28 | + |
| 29 | + |
| 30 | +# Sample user database (in a real app, you would use a database) |
| 31 | +USERS = { |
| 32 | + "user1": {"password": "password1", "email": "user1@example.com"}, |
| 33 | + "user2": {"password": "password2", "email": "user2@example.com"}, |
| 34 | +} |
| 35 | + |
| 36 | + |
| 37 | +@app.post("/login") |
| 38 | +def login(user: User): |
| 39 | + """Login endpoint that validates credentials and returns a fresh token.""" |
| 40 | + # Check if user exists and password is correct |
| 41 | + if user.username in USERS and USERS[user.username]["password"] == user.password: |
| 42 | + # Create a fresh token with the username as the subject |
| 43 | + fresh_token = auth.create_access_token(user.username, fresh=True) |
| 44 | + return {"fresh_token": fresh_token, "token_type": "bearer"} |
| 45 | + |
| 46 | + # Return error if credentials are invalid |
| 47 | + raise HTTPException(status_code=401, detail="Invalid username or password") |
| 48 | + |
| 49 | + |
| 50 | +@app.post("/refresh") |
| 51 | +async def refresh_token(request: Request): |
| 52 | + """Refresh endpoint that creates a non-fresh token using a fresh token.""" |
| 53 | + try: |
| 54 | + # Get the token from the request |
| 55 | + token = await auth.get_token_from_request(request) |
| 56 | + |
| 57 | + # Verify the token |
| 58 | + payload = auth.verify_token(token) |
| 59 | + |
| 60 | + # Create a non-fresh token |
| 61 | + access_token = auth.create_access_token(payload.sub, fresh=False) |
| 62 | + |
| 63 | + return {"access_token": access_token, "token_type": "bearer"} |
| 64 | + except Exception as e: |
| 65 | + print(f"Refresh error: {str(e)}") |
| 66 | + raise HTTPException(status_code=401, detail=str(e)) from e |
| 67 | + |
| 68 | + |
| 69 | +@app.get("/protected") |
| 70 | +async def protected_route(request: Request): |
| 71 | + """Protected route that requires a valid token (fresh or non-fresh).""" |
| 72 | + try: |
| 73 | + # Get the token from the request |
| 74 | + token = await auth.get_token_from_request(request) |
| 75 | + |
| 76 | + # Verify the token |
| 77 | + payload = auth.verify_token(token) |
| 78 | + |
| 79 | + # Get the username from the token subject |
| 80 | + username = payload.sub |
| 81 | + |
| 82 | + # Return user information |
| 83 | + return { |
| 84 | + "message": "You have access to this protected resource", |
| 85 | + "username": username, |
| 86 | + "email": USERS.get(username, {}).get("email"), |
| 87 | + "fresh": payload.fresh, |
| 88 | + } |
| 89 | + except Exception as e: |
| 90 | + print(f"Authentication error: {str(e)}") |
| 91 | + raise HTTPException(status_code=401, detail=str(e)) from e |
| 92 | + |
| 93 | + |
| 94 | +@app.get("/fresh-required") |
| 95 | +async def fresh_required_route(request: Request): |
| 96 | + """Protected route that requires a fresh token.""" |
| 97 | + try: |
| 98 | + # Get the token from the request |
| 99 | + token = await auth.get_token_from_request(request) |
| 100 | + |
| 101 | + # Verify the token |
| 102 | + payload = auth.verify_token(token) |
| 103 | + |
| 104 | + # Check if the token is fresh |
| 105 | + if not payload.fresh: |
| 106 | + raise HTTPException(status_code=401, detail="Fresh token required") |
| 107 | + |
| 108 | + # Get the username from the token subject |
| 109 | + username = payload.sub |
| 110 | + |
| 111 | + # Return user information |
| 112 | + return { |
| 113 | + "message": "You have access to this fresh-required resource", |
| 114 | + "username": username, |
| 115 | + "email": USERS.get(username, {}).get("email"), |
| 116 | + } |
| 117 | + except Exception as e: |
| 118 | + print(f"Authentication error: {str(e)}") |
| 119 | + raise HTTPException(status_code=401, detail=str(e)) from e |
| 120 | + |
| 121 | + |
| 122 | +@app.get("/") |
| 123 | +def read_root(): |
| 124 | + """Public route that doesn't require authentication.""" |
| 125 | + return { |
| 126 | + "message": "Welcome to AuthX Fresh Token Example", |
| 127 | + "endpoints": { |
| 128 | + "login": "POST /login - Get a fresh token", |
| 129 | + "refresh": "POST /refresh - Get a non-fresh token using a fresh token", |
| 130 | + "protected": "GET /protected - Access protected resource (requires any token)", |
| 131 | + "fresh-required": "GET /fresh-required - Access protected resource (requires fresh token)", |
| 132 | + }, |
| 133 | + } |
| 134 | + |
| 135 | + |
| 136 | +if __name__ == "__main__": |
| 137 | + import os |
| 138 | + |
| 139 | + import uvicorn |
| 140 | + |
| 141 | + port = int(os.environ.get("PORT", 8000)) |
| 142 | + uvicorn.run(app, host="0.0.0.0", port=port) |
0 commit comments