File size: 5,562 Bytes
a152b95
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
# Security and Authentication in FastAPI

FastAPI provides integrated security utilities built on top of OpenAPI standards. It supports OAuth2, API keys, HTTP Basic/Bearer authentication, and OpenID Connect, with each scheme automatically reflected in the interactive documentation.

## OAuth2 with Password Flow

The most common authentication pattern uses OAuth2 "password" flow with JWT tokens:

```python
from datetime import datetime, timedelta
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from pydantic import BaseModel

app = FastAPI()

SECRET_KEY = "your-secret-key-at-least-32-characters-long"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

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

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

def create_access_token(data: dict, expires_delta: timedelta | None = None):
    to_encode = data.copy()
    expire = datetime.utcnow() + (expires_delta or 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
    except JWTError:
        raise credentials_exception
    user = get_user_from_db(username)
    if user is None:
        raise credentials_exception
    return user

@app.post("/token", response_model=Token)
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    user = authenticate_user(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"}
```

The `OAuth2PasswordBearer(tokenUrl="token")` declaration tells FastAPI that the client obtains a token by sending credentials to the `/token` endpoint. The `tokenUrl` is relative to the application root. The token is then sent in subsequent requests via the `Authorization: Bearer <token>` header.

## HTTP Bearer Authentication

For simpler token-based auth without the full OAuth2 flow:

```python
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials

security = HTTPBearer()

@app.get("/protected/")
async def protected_route(
    credentials: HTTPAuthorizationCredentials = Depends(security),
):
    token = credentials.credentials
    # validate token
    return {"token": token, "scheme": credentials.scheme}
```

`HTTPBearer` extracts the token from the `Authorization: Bearer <token>` header. If the header is missing or does not use the Bearer scheme, FastAPI returns a 403 Forbidden response automatically.

## API Key Authentication

API keys can be passed via headers, query parameters, or cookies:

```python
from fastapi.security import APIKeyHeader, APIKeyQuery

api_key_header = APIKeyHeader(name="X-API-Key", auto_error=True)
api_key_query = APIKeyQuery(name="api_key", auto_error=False)

async def get_api_key(
    header_key: str | None = Depends(api_key_header),
    query_key: str | None = Depends(api_key_query),
):
    if header_key == "valid-api-key-12345":
        return header_key
    if query_key == "valid-api-key-12345":
        return query_key
    raise HTTPException(status_code=403, detail="Invalid API key")

@app.get("/data/", dependencies=[Depends(get_api_key)])
async def read_data():
    return {"data": "sensitive information"}
```

The `auto_error=True` parameter (the default) causes FastAPI to return an automatic 403 error when the key is missing. Setting `auto_error=False` allows the dependency to return `None` instead, letting you check multiple sources.

## OAuth2 Scopes

Scopes provide fine-grained permission control:

```python
from fastapi.security import SecurityScopes

oauth2_scheme = OAuth2PasswordBearer(
    tokenUrl="token",
    scopes={
        "items:read": "Read items",
        "items:write": "Create and update items",
        "admin": "Full administrative access",
    },
)

async def get_current_user(
    security_scopes: SecurityScopes,
    token: str = Depends(oauth2_scheme),
):
    # Decode token and verify required scopes
    payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
    token_scopes = payload.get("scopes", [])
    for scope in security_scopes.scopes:
        if scope not in token_scopes:
            raise HTTPException(status_code=403, detail="Not enough permissions")
    return get_user_from_db(payload.get("sub"))

@app.get("/items/", dependencies=[Depends(Security(get_current_user, scopes=["items:read"]))])
async def read_items():
    return [{"item": "Widget"}]
```

Each endpoint declares the scopes it requires, and the dependency verifies the token contains all necessary permissions before allowing access.