Это второй из Две части Серия по внедрению авторизации в приложении Fastapi с использованием DETA. В предыдущей статье мы немного узнали о JWT, создали проект и закончили строительные блоки логики авторизации. В этой статье давайте реализуем логику и разверните наше приложение на Deta Micro! Полный код доступен здесь.
Реализация логики аутентики
Перед тем, как мы реализуем логику AUTH, давайте создадим модель данных для входа в систему и информацию о регистрации.
В user_model.py
:
from pydantic import BaseModel class AuthModel(BaseModel): username: str password: str
Эта модель представляет данные, которые мы можем ожидать от клиента, когда они попадают /Вход
или /Регистрация
конечные точки.
Обновите main.py
, со следующими утверждениями импорта
from auth import Auth from user_model import AuthModel from fastapi import FastAPI, HTTPException, Security from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
Кроме того, мы создадим AUTH_HANDLER
Чтобы получить доступ к логике из Auth
сорт. Мы будем использовать Безопасность
В наших защищенных конечных точках доступа к токену из заголовка запроса.
security = HTTPBearer() auth_handler = Auth()
Вот код для /Регистрация
конечная точка.
@app.post('/signup') def signup(user_details: AuthModel): if users_db.get(user_details.username) != None: return 'Account already exists' try: hashed_password = auth_handler.encode_password(user_details.password) user = {'key': user_details.username, 'password': hashed_password} return users_db.put(user) except: error_msg = 'Failed to signup user' return error_msg
В этой функции мы проверяем, если пользователь с именем пользователя уже существует в нашем users_db
Отказ Если это так, мы можем просто вернуть сообщение, указывающее, что учетная запись уже существует. Если пользователь уже не существует, мы можем записать пароль, используя encode_password
Функция от Auth.py
и хранить пользователь в users_db
Отказ В случае любых ошибок при добавлении пользователя к базе мы можем вернуть сообщение об ошибке.
/Вход
Конечная точка довольно проста. Это также принимает аргумент user_details
, в котором имеется имя пользователя и пароль.
@app.post('/login') def login(user_details: AuthModel): user = users_db.get(user_details.username) if (user is None): return HTTPException(status_code=401, detail='Invalid username') if (not auth_handler.verify_password(user_details.password, user['password'])): return HTTPException(status_code=401, detail='Invalid password') access_token = auth_handler.encode_token(user['key']) refresh_token = auth_handler.encode_refresh_token(user['key']) return {'access_token': access_token, 'refresh_token': refresh_token}
Если учетная запись с именем пользователя не существует, или если Hashed пароль в users_db
Не совпадает с входным паролем, мы можем просто поднять HttpException
Отказ В противном случае мы можем вернуть Access_Token
и Refresh_token
Отказ
@app.post('/secret') def secret_data(credentials: HTTPAuthorizationCredentials = Security(security)): token = credentials.credentials if(auth_handler.decode_token(token)): return 'Top Secret data only authorized users can access this info' @app.get('/notsecret') def not_secret_data(): return 'Not secret data'
/секрет
Конечная точка возвращает только «секретные данные», если аргумент токена действителен. Однако, если токен недействителен или истекший токен, то decode_token
поднимает HttpException
Отказ Токен обычно передается в заголовке запроса как Авторизация: Носитель <Токен>
Отказ Поэтому, чтобы получить токен, мы можем обернуть ввод учетные данные
вокруг HttpauthorizationCredenties
ярлык. Теперь мы можем получить доступ к токену из заголовка запроса в учетные данные. Credentials
Отказ
/not_secret.
Конечная точка является примером незащищенной конечной точки, которая не требует аутентификации.
@app.get('/refresh_token') def refresh_token(credentials: HTTPAuthorizationCredentials = Security(security)): refresh_token = credentials.credentials new_token = auth_handler.refresh_token(refresh_token) return {'access_token': new_token}
/refresh_token
Конечная точка также довольно проста, она получает токен обновления, который затем передается на функцию от логики Authic, чтобы получить новый токен.
Вот взгляд на main.py
в конце:
from fastapi import FastAPI, HTTPException, Security from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer from auth import Auth from deta import Deta from user_model import AuthModel deta = Deta() users_db = deta.Base('users') app = FastAPI() security = HTTPBearer() auth_handler = Auth() @app.post('/signup') def signup(user_details: AuthModel): if users_db.get(user_details.username) != None: return 'Account already exists' try: hashed_password = auth_handler.encode_password(user_details.password) user = {'key': user_details.username, 'password': hashed_password} return users_db.put(user) except: error_msg = 'Failed to signup user' return error_msg @app.post('/login') def login(user_details: AuthModel): user = users_db.get(user_details.username) if (user is None): return HTTPException(status_code=401, detail='Invalid username') if (not auth_handler.verify_password(user_details.password, user['password'])): return HTTPException(status_code=401, detail='Invalid password') access_token = auth_handler.encode_token(user['key']) refresh_token = auth_handler.encode_refresh_token(user['key']) return {'access_token': access_token, 'refresh_token': refresh_token} @app.get('/refresh_token') def refresh_token(credentials: HTTPAuthorizationCredentials = Security(security)): refresh_token = credentials.credentials new_token = auth_handler.refresh_token(refresh_token) return {'access_token': new_token} @app.post('/secret') def secret_data(credentials: HTTPAuthorizationCredentials = Security(security)): token = credentials.credentials if(auth_handler.decode_token(token)): return 'Top Secret data only authorized users can access this info' @app.get('/notsecret') def not_secret_data(): return 'Not secret data'
Чтобы проверить приложение, перейдите к терминалу в одном каталоге и запустите Увикорн Главная: Приложение
, вы можете пойти пойти /Документы
на местной конечной точке (Для меня это было http://127.0.0.1:8000/docs
) для проверки приложения.
/зарегистрироваться
curl -X 'POST' \ 'http://127.0.0.1:8000/signup' \ -H 'accept: application/json' \ -H 'Content-Type: application/json' \ -d '{ "username": "flyingsponge", "password": "SpongePassword" }' Response Body { "key": "flyingsponge", "password": "$2b$12$Ml66gTN0j4sAOkoDJ1aKnOmP0ye1FBNNk1QzEt0/6LgemgOUj469e" }
/авторизоваться
curl -X 'POST' \ 'http://127.0.0.1:8000/login' \ -H 'accept: application/json' \ -H 'Content-Type: application/json' \ -d '{ "username": "flyingsponge", "password": "SpongePassword" }' Response Body { "access_token": "eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJleHAiOjE2MTkwMTY4MzgsImlhdCI6MTYxOTAxNTAzOCwic3ViIjoiZmx5aW5nc3BvbmdlIn0.XnPaDwmj30M3vMOaPUOrqESIBNx0mctbjNW5jY8hIjQ", "refresh_token": "eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJleHAiOjE2MTkwNTEwMzgsImlhdCI6MTYxOTAxNTAzOCwic3ViIjoiZmx5aW5nc3BvbmdlIn0.Pm68HQFM6NwUxnEjFUxxTiOYT3KBqchD_e2g0LfP5Co" }
/секрет
curl -X 'POST' \ 'http://127.0.0.1:8000/secret' \ -H 'accept: application/json' \ -H 'Authorization: Bearer eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJleHAiOjE2MTkwMTY2MTIsImlhdCI6MTYxOTAxNDgxMiwic3ViIjoicmFta2kifQ.U8fUN1x1Gz18f7oA_y7Z9DE-1FIH9Ps0sDilwORvmI8' \ -d '' Response body "Top Secret data only authorized users can access this info"
/notsecret.
curl -X 'GET' \ 'http://127.0.0.1:8000/notsecret' \ -H 'accept: application/json' Response Body "Not secret data"
/секрет
curl -X 'POST' \ 'http://127.0.0.1:8000/secret' \ -H 'accept: application/json' \ -H 'Authorization: Bearer eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJleHAiOjE2MTg5MzU3MzcsImlhdCI6MTYxODkzNTY3Nywic3ViIjoicm9oYW4ifQ.dja0E6SUaZfEvYVKySjLE9OLXOtob5pjpy3R_rlCD7c' \ -d '' Response Body { "detail": "Token expired" }
Теперь, когда токен истек, давайте получим новый
/refreeh_token.
curl -X 'GET' \ 'http://127.0.0.1:8000/refresh_token' \ -H 'accept: application/json' \ -H 'Authorization: Bearer eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJleHAiOjE2MTkwNTA3NzYsImlhdCI6MTYxOTAxNDc3Niwic3ViIjoicmFta2kifQ.2J0O4RKwEcABxe6hEX7ZshMo66J6D0dD6-hYeFLBFGg' Response Body { "access_token": "eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJleHAiOjE2MTkwMTY2MTIsImlhdCI6MTYxOTAxNDgxMiwic3ViIjoicmFta2kifQ.U8fUN1x1Gz18f7oA_y7Z9DE-1FIH9Ps0sDilwORvmI8" }
Развертывание на Deta Micros
Прежде чем начать, убедитесь, что Установите Deta CLI. После установки запустите следующие команды в том же каталоге, чтобы развернуть наше приложение на Deta Micic.
deta login
Нам также нужно добавить .env
файл с секретом.
APP_SECRET_STRING=SECRET_STRING
Теперь нам нужно обновить наш Micro, делая:
deta new deta update -e .env deta deploy deta visor disable
Резюме
Наше простое приложение fastapi с jwt auth теперь готов! Как вы можете сказать, мы ничего не делаем «секрет» с нашей авторизацией. Эта статья является просто шаблоном для реализации разрешения. Вы можете построить этот шаблон, чтобы построить приложение FullStack, которое опирается на авторизацию. Полный код доступен здесь.
Ссылка на первую статью!
Оригинал: “https://dev.to/deta/get-started-with-fastapi-jwt-authentication-part-2-18ok”