FastAPI поддерживает различные режимы и спецификации аутентификации через свой модуль fastapi.security. Чтобы использовать базовую схему аутентификации, нам необходимо создать экземпляр класса HTTPBasic модуля и внедрить его в каждую службу API, чтобы защитить доступ к конечной точке.
В файл auth.py добавим следующий код:
from fastapi.security import HTTPBasic, HTTPBasicCredentials
security = HTTPBasic()
async def get_current_username(db: Annotated[AsyncSession, Depends(get_db)], credentials: HTTPBasicCredentials = Depends(security)):
user = await db.scalar(select(User).where(User.username == credentials.username))
if not user or not bcrypt_context.verify(credentials.password, user.hashed_password):
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)
return user
@router.get('/users/me')
async def read_current_user(user: str = Depends(get_current_username)):
return {'User': user}
Экземпляр HTTPBasic, внедренный в службы API, вызывает в браузере всплывающую форму входа, через которую мы вводим имя пользователя и пароль. Вход в систему заставит браузер отправить заголовок с учетными данными в приложение. Если приложение сталкивается с проблемой при его получении, схема HTTPBasic выдает код состояния 401 с сообщением «Unauthorized». Если при обработке формы нет ошибок, приложение должно получить заголовок WWW-Authenticate со значением введенных данных.
Мы добавили конечную точку read_current_user, которая будет выводить все данные авторизованного пользователя. То есть доступ к ней мы можем получить, только после авторизации. В ней, через зависимости, мы вызываем метод get_current_username(), чтобы проверить логин и пароль пользователя.
Разберем функцию get_current_username. Мы подключаем две зависимости, первая это зависимость сессий БД, вторая это зависимость от экземпляра класса HTTPBasic, для извлечения HTTPBasicCredentials — объекта, содержащего сведения об имени пользователя и пароле из введенной формы в браузере.
Далее мы выполняем запрос в нашу БД, получая пользователя на основе введенного имени пользователя. Далее мы проверяем введенный пароль bcrypt_context.verify(credentials.password, user.hashed_password). Данный метод хэширует введенный пароль, и сверяет его с хэшем пароля этого пользователя из БД. В случае если пользователь не найден или проверка хэшей паролей не прошла, мы вызываем ошибку с кодом 401. В ином случае мы возвращаем объект пользователя, чтобы вывести его в конечной точке.
Запустим сервер и проверим работу. Перейдем в конечную точку Read Current User, мы видим что у данной конечной точки появился значок замочка.
Теперь попробуем отправить запрос:
Введем правильные данные, и нажмем кнопку Вход.
В результате мы получим все данные этого пользователя. Мы также можем посмотреть Cookie файлы:
Но если мы очистим куки и попробуем авторизоваться с не верными данными пользователя, то получим ошибку 401. Появление сообщения об ошибке 401 Unauthorized Error («отказ в доступе») при открытии страницы сайта означает неверную авторизацию или аутентификацию пользователя на стороне сервера при обращении к определенному url-адресу.
На этом мы закончим рассмотрение базовой аутентификации, в следующем разделе мы создадим более удобную систему на основе OAuth2 и JWT токенов.
Добрый день!
def read_current_user(user: str = Depends(get_current_username)):
Забыли async перед функцией или если принимается асинхронная функция async не нужен уже?
@Николай_Баранов, Спасибо, поправил.