When looking to secure your FastAPI application, integrating OAuth2 for third-party authentication stands out as a solid choice. OAuth2 is a popular standard that lets users give third-party apps access to their resources without handing over their passwords. Here’s a simple guide to help you set up OAuth2 with FastAPI for secure third-party authentication.

Understanding OAuth2

OAuth2 is built for scenarios where third-party apps access user resources. It has different flows to cater to various needs. For most cases, the Password Flow suffices. In this flow, the client sends the username and password to the server, which then returns an access token for future authentication requests.

Setting Up OAuth2 with FastAPI

First things first, you need a few libraries. FastAPI plays well with fastapi.security and pyjwt for OAuth2 and JWT tokens. So go ahead and install these:

pip install fastapi uvicorn pyjwt passlib[bcrypt]

Basic Setup

Start by creating a basic FastAPI app and define the OAuth2 scheme using OAuth2PasswordBearer. This handles the authentication flow for you.

from fastapi import FastAPI, Depends, HTTPException
from fastapi.security import OAuth2PasswordBearer
from pydantic import BaseModel

app = FastAPI()

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

@app.get("/items/")
async def read_items(token: str = Depends(oauth2_scheme)):
    return {"token": token}

In this snippet, tokenUrl="token" is where users will send their username and password to get a token. Make sure to replace it with your actual endpoint.

Implementing Authentication with JWT

For a more secure setup, you can use JSON Web Tokens (JWT). Let’s dive in:

from datetime import datetime, timedelta
import jwt
from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jwt.exceptions import InvalidTokenError
from passlib.context import CryptContext
from pydantic import BaseModel

SECRET_KEY = "your_secret_key"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

class Token(BaseModel):
    access_token: str
    token_type: str

class TokenData(BaseModel):
    username: str | None = None

class User(BaseModel):
    username: str
    email: str | None = None
    full_name: str | None = None
    disabled: bool | None = None

class UserInDB(User):
    hashed_password: str

fake_users_db = {
    "johndoe": {
        "username": "johndoe",
        "full_name": "John Doe",
        "email": "[email protected]",
        "hashed_password": pwd_context.hash("your_password"),
        "disabled": False,
    },
}

def verify_password(plain_password, hashed_password):
    return pwd_context.verify(plain_password, hashed_password)

def get_password_hash(password):
    return pwd_context.hash(password)

def get_user(db, username: str):
    if username in db:
        user_dict = db[username]
        return UserInDB(**user_dict)

def authenticate_user(fake_db, username: str, password: str):
    user = get_user(fake_db, username)
    if not user:
        return False
    if not verify_password(password, user.hashed_password):
        return False
    return user

def create_access_token(data: dict, expires_delta: timedelta | None = None):
    to_encode = data.copy()
    expire = datetime.utcnow() + (expires_delta if expires_delta else timedelta(minutes=15))
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

async def get_current_user(token: str = Depends(oauth2_scheme)):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
        token_data = TokenData(username=username)
    except InvalidTokenError:
        raise credentials_exception
    user = get_user(fake_users_db, username=token_data.username)
    if user is None:
        raise credentials_exception
    return user

async def get_current_active_user(current_user: User = Depends(get_current_user)):
    if current_user.disabled:
        raise HTTPException(status_code=400, detail="Inactive user")
    return current_user

@app.post("/token", response_model=Token)
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
    user = authenticate_user(fake_users_db, form_data.username, form_data.password)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    access_token = create_access_token(
        data={"sub": user.username}, expires_delta=timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    )
    return {"access_token": access_token, "token_type": "bearer"}

@app.get("/users/me")
async def read_users_me(current_user: User = Depends(get_current_active_user)):
    return current_user

How It Works

1. Authentication Flow: When a client hits the /token endpoint with a username and password, login_for_access_token checks the credentials and gives a JWT token if they match.
2. Token Verification: The get_current_user function checks the JWT token sent in the Authorization header of new requests. If it’s valid, it returns the user data; if not, it throws an error.
3. Active User Check: The get_current_active_user ensures the authenticated user isn’t disabled.

Security Best Practices

HTTPS Adoption: Always use HTTPS to encrypt data in transit. You can set SSL certificates while running your FastAPI app with Uvicorn:

if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=8000, ssl_keyfile="path/to/key.pem", ssl_certfile="path/to/cert.pem")

Dependency Management: Regularly update all external libraries and dependencies to fend off known vulnerabilities.

Safe Session Management: Secure cookies and encrypted session stores should be your go-to. This stops unauthorized access to user sessions.

Input Validation: Always validate and sanitize user inputs to thwart web attacks like SQL injection, XSS, and CSRF.

Security Headers and CORS: Set up appropriate security headers and tweak CORS settings to reduce cross-origin request risks and other web vulnerabilities.

Error Handling: Secure error handling and logging practices can prevent leakage of sensitive information.

Conclusion

Securing your FastAPI application with OAuth2 and JWT tokens is a sturdy method for handling authentication and authorization. Following these steps and best practices ensures your web service is protected and sturdy against possible threats. Security is an ever-evolving process, and keeping up with the latest practices is crucial for the integrity of your application.