Protect FastAPI Endpoints: Dependency Injection with Depends, Router-Level Security, HTTPBasic and OAuth2

Learn why FastAPI authentication doesn't auto-protect routes and how to secure endpoints using Depends, APIRouter dependencies, HTTPBasic, or OAuth2 tokens.

When you wire authentication into a FastAPI project but endpoints still respond without any checks, the usual culprit is missing dependency injection on those endpoints. Declaring a security backend or issuing cookies/tokens isn't enough; the framework enforces auth only where a dependency is explicitly applied—either per endpoint or at router level.

The unprotected endpoints

Below is a minimalized application where only the routes below the files list were meant to be protected, yet they are currently accessible without authentication. The program issues a cookie-based token on registration/login, but none of the file endpoints actually require any dependency that verifies the user. That is why everything stays open.

import uvicorn
from fastapi import FastAPI, HTTPException, Response, Depends, UploadFile
from fastapi.responses import StreamingResponse, FileResponse
from pydantic import BaseModel, Field
from typing import Annotated, List
from authx import AuthX, AuthXConfig
from sqlalchemy import select
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from passlib.context import CryptContext
# Password hashing
pwd_ctx = CryptContext(schemes=["bcrypt"], deprecated="auto")
# FastAPI app and DB session setup
api = FastAPI()  # FastAPI app instance
db_engine = create_async_engine("sqlite+aiosqlite:///users.db", echo=True)  # Async DB engine
session_factory = async_sessionmaker(db_engine, expire_on_commit=False)  # Async session factory
async def acquire_db():
    "Access DB session"
    async with session_factory() as s:
        yield s
DbSessionDep = Annotated[AsyncSession, Depends(acquire_db)]  # DB session dependency
class OrmBase(DeclarativeBase):
    pass
class AccountRow(OrmBase):  # Users table
    __tablename__ = "users"
    # Columns
    uid: Mapped[int] = mapped_column(primary_key=True, autoincrement=True, nullable=False)
    username: Mapped[str] = mapped_column(unique=True, index=True)
    password: Mapped[str]
class AccountIn(BaseModel):
    username: str = Field(max_length=10)
    password: str = Field(min_length=8, max_length=16)
# AuthX configuration
jwt_conf = AuthXConfig()
jwt_conf.JWT_SECRET_KEY = "kirieshki"
jwt_conf.JWT_ACCESS_COOKIE_NAME = "access_token"
jwt_conf.JWT_TOKEN_LOCATION = ['cookies']
authx_mgr = AuthX(config=jwt_conf)
@api.post("/register",
          tags=["Авторизация"],
          summary="Зарегистрировать аккаунт")
async def register_account(creds: AccountIn,
                           response: Response,
                           session: DbSessionDep):
    "User registration"
    check_stmt = select(AccountRow).where(AccountRow.username == creds.username)
    res = await session.execute(check_stmt)
    if res.scalar_one_or_none():
        raise HTTPException(status_code=400, detail="User already exist")
    hashed = pwd_ctx.hash(creds.password)
    row = AccountRow(username=creds.username, password=hashed)
    session.add(row)
    await session.commit()
    await session.refresh(row)
    token = authx_mgr.create_access_token(uid=str(row.uid))
    authx_mgr.set_access_cookies(token, response)
    return {"message": "User has been registered"}
@api.post("/login",
          tags=["Авторизация"],
          summary="Войти в аккаунт")
async def login_account(creds: AccountIn,
                        response: Response,
                        session: DbSessionDep):
    "User login"
    check_stmt = select(AccountRow).where(AccountRow.username == creds.username)
    res = await session.execute(check_stmt)
    user = res.scalar_one_or_none()
    if not user or not pwd_ctx.verify(creds.password, user.password):
        raise HTTPException(status_code=401, detail="Invalid username or password")
    token = authx_mgr.create_access_token(uid=str(user.uid))
    authx_mgr.set_access_cookies(token, response)
    return {"Success": True}
@api.post("/reset",
          tags=["Работа с БД"],
          summary="Сброс БД")
async def reset_db():
    "Reset DB"
    async with db_engine.begin() as conn:
        await conn.run_sync(OrmBase.metadata.drop_all)
        await conn.run_sync(OrmBase.metadata.create_all)
    return {"База данных успешно сброшена": True}
stored_files = []  # In-memory list of uploaded files
@api.get("/get_filenames",
         tags=["Получение файлов"],
         summary="Получить названия файлов")
async def list_filenames():
    if len(stored_files) == 0:
        return {"File list is empty"}
    else:
        return stored_files
def stream_chunks(filename: str):
    "Chunked file streaming"
    with open(filename, "rb") as file:
        while chunk := file.read(1024 * 1024):
            yield chunk
@api.get("/get_files",
         tags=["Получение файлов"],
         summary="Получить файл")
async def fetch_file(filename: str):
    return FileResponse(filename)
@api.get("/streaming/{filename}",
         tags=["Получение файлов"],
         summary="Получение файла в стриминге")
async def stream_file(filename: str):
    return StreamingResponse(stream_chunks(filename))
@api.post("/upload",
          tags=["Добавление файлов"],
          summary="Загрузка файла")
async def upload_single(uploaded_file: UploadFile):
    "Upload single file"
    file = uploaded_file.file
    filename = uploaded_file.filename
    with open(f"1_{filename}", "wb") as f:
        f.write(file.read())
    stored_files.append(f"1_{filename}")
    return {"File was uploaded": True}
@api.post("/upload_multiple",
          tags=["Добавление файлов"],
          summary="Загрузка нескольких файлов")
async def upload_many(uploaded_files: list[UploadFile]):
    for uploaded_file in upload_many:
        file = uploaded_file.file
        filename = uploaded_file.filename
        with open(f"1_{filename}", "wb") as f:
            f.write(file.read())
        stored_files.append(f"1_{filename}")
    return {"Multiple files were uploaded": True}
if __name__ == "__main__":
    uvicorn.run("main:api", host="127.0.0.1", port=8000, reload=True)

Why the endpoints remain open

FastAPI does not implicitly protect routes just because a token is created or a security backend is configured. Protection happens only where you inject a dependency that performs authentication. If a handler doesn’t depend on a security dependency, it will be publicly accessible. In the code above, none of the routes below the files list are guarded by Depends, so every request goes through.

Two straightforward ways to protect routes

The first option is to add a security dependency to each endpoint that must be protected. This is the most explicit pattern—simple to reason about and easy to audit. For demonstration, the example below uses HTTPBasic as a guard, attached directly to a single endpoint. The logic is that the route cannot run unless the dependency resolves successfully.

from fastapi import FastAPI, Depends
from fastapi.security import HTTPBasic, HTTPBasicCredentials
svc = FastAPI()
basic_guard = HTTPBasic()
@svc.get("/")
async def index():
    return "OK"
@svc.get("/protected")
async def protected(creds: HTTPBasicCredentials = Depends(basic_guard)):
    return {"username": creds.username, "password": creds.password}

The second option is to group protected endpoints in a dedicated APIRouter and attach the dependency once to that router. That way, any route you register on that router is automatically protected. This pattern scales well and aligns with the idea of modular routers in bigger applications.

Working example with a protected router

The example below demonstrates a working setup that verifies credentials, stores the authenticated user on request.state, and exposes both open and protected routes. The protected router is created with a dependency that must pass for every endpoint attached to it. Test credentials used are admin for username and password for password.

from fastapi import FastAPI, APIRouter, Depends, Request, HTTPException, status
from fastapi.security import HTTPBasic, HTTPBasicCredentials
import hashlib
from secrets import compare_digest
from pydantic import BaseModel
app2 = FastAPI()
auth_scheme = HTTPBasic()
demo_users = {
    "admin": {
        "username": "admin",
        "full_name": "Test Admin",
        "email": "admin@example.com",
        "hashed_password": "c255f02d5cb0812495fcf301d3239f80693f349d6b423a4f1868997c6b211eda",  # Plain Password: password
        "salt": "fc96c333c19cfbf9f99b916c0dc82db1e2f8d88fa72e5f534147e25a19341fa3",
        "disabled": False,
        "priviliged": True,
    }
}
class UserOut(BaseModel):
    username: str
    full_name: str
    email: str
    disabled: bool
    priviliged: bool
class UserRecord(UserOut):
    hashed_password: str
    salt: str
# Hash new plain password with salt; returns hash and salt
def get_password_hash_and_salt(plain_password: str):
    salt = secrets.token_hex(32)
    hashed_password = hashlib.pbkdf2_hmac('sha256', plain_password.encode('utf-8'), salt.encode('utf-8'), 100000)
    return hashed_password.hex(), salt
# Verify plain password against stored hash and salt
def verify_password(plain_password: str, user: UserRecord):
    derived = hashlib.pbkdf2_hmac('sha256', plain_password.encode('utf-8'), user.salt.encode('utf-8'), 100000)
    current_bytes = derived.hex().encode('utf-8')
    correct_bytes = user.hashed_password.encode('utf-8')
    return compare_digest(current_bytes, correct_bytes)
def fetch_user(username: str):
    user = demo_users.get(username)
    if user:
        return UserRecord(**user)
def assert_credentials(request: Request, creds: HTTPBasicCredentials = Depends(auth_scheme)):
    user = fetch_user(creds.username)
    if user and verify_password(creds.password, user):
        request.state.user = UserOut(**user.model_dump())
    else:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Basic"},
        )
open_router = APIRouter()
locked_router = APIRouter(dependencies=[Depends(assert_credentials)])
@open_router.get("/")
async def main_open():
    return "OK"
@locked_router.get("/protected")
async def main_locked(request: Request):
    return request.state.user
app2.include_router(open_router)
app2.include_router(locked_router)

Notice the use of request.state to make the user object available in endpoints when the dependency is attached to the router itself. Returning a value from a router-level dependency doesn’t pass it into endpoint parameters; storing it on request.state makes it accessible during the current request lifecycle. As discussed in resources referenced in the original explanation, the state exposed on the request is a shallow copy of the lifespan state, hence data you set on request.state is available to that single request.

Getting the data directly from the dependency

If you prefer the endpoint to receive the dependency’s return value directly—without going through request.state—declare the dependency on the endpoint itself. This means duplicating the Depends annotation for every protected route, but keeps the handler signature explicit.

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import HTTPBasic, HTTPBasicCredentials
from pydantic import BaseModel
from secrets import compare_digest
import hashlib
api3 = FastAPI()
auth_basic = HTTPBasic()
class UserPublic(BaseModel):
    username: str
    full_name: str
    email: str
    disabled: bool
    priviliged: bool
# ... reuse the same demo_users, verify_password, etc.
def verify_creds(creds: HTTPBasicCredentials = Depends(auth_basic)):
    # fetch_user and verify_password are assumed same as the previous example
    user = fetch_user(creds.username)
    if user and verify_password(creds.password, user):
        return UserPublic(**user.model_dump())
    else:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Basic"},
        )
@api3.get("/protected")
async def protected_here(user: UserPublic = Depends(verify_creds)):
    return user

OAuth2 option (token-based)

If you prefer token-based flows, you can switch to OAuth2. The snippet below shows how to attach OAuth2PasswordBearer as a dependency for a protected router. This is a sketch; the token endpoint and the end-to-end auth flow should follow the official tutorial.

from fastapi import FastAPI, APIRouter, Depends
from fastapi.security import OAuth2PasswordBearer
app4 = FastAPI()
oauth2 = OAuth2PasswordBearer(tokenUrl="token")
public_router = APIRouter()
secure_router = APIRouter(dependencies=[Depends(oauth2)])
@public_router.get("/")
async def alive():
    return "OK"
@secure_router.get("/protected")
async def show_token(token: str = Depends(oauth2)):
    return {"token": token}
app4.include_router(public_router)
app4.include_router(secure_router)

Why it’s important to know this

In FastAPI, security is composable. That power comes with an explicit requirement: you must attach the dependency where you want enforcement to happen. Relying on a global setup without binding it to endpoints leads to a false sense of protection and exposed routes. Organizing security at router level gives you a clean separation between open and protected areas, while endpoint-level dependencies make the contract of each route crystal clear.

Conclusion

If a route should be protected, inject a security dependency. Do it per endpoint for clarity or at the APIRouter level for better grouping and scalability. For cases where you attach the dependency to the router, use request.state to pass data to handlers; for endpoint-level dependencies, simply return the user object from the dependency and accept it as a parameter. Whether you pick HTTPBasic for demos or OAuth2 for tokens, the key is the same: wire Depends exactly where you need authentication enforced.