Рубрики
Без рубрики

Начните с аутентификации fastapi jwt – часть 2

Это второй из двух частей серии о внедрении авторизации в приложении Fastapi, используя … Теги с Python, Fastapi, Deta, Jwt.

Это второй из Две части Серия по внедрению авторизации в приложении 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”