1
0
mirror of https://github.com/ijaric/voice_assistant.git synced 2025-05-24 14:33:26 +00:00

fix: token (token model TBU)

This commit is contained in:
Artem Litvinov 2023-09-29 18:11:11 +01:00
parent 46711d5172
commit 92380d4406
9 changed files with 42 additions and 37 deletions

View File

@ -10,5 +10,6 @@ API_HOST=0.0.0.0
API_PORT=8000 API_PORT=8000
JWT_SECRET_KEY=v9LctjUWwol4XbvczPiLFMDtZ8aal7mm JWT_SECRET_KEY=v9LctjUWwol4XbvczPiLFMDtZ8aal7mm
JWT_ALGORITHM=HS256
APP_RELOAD=True APP_RELOAD=True

View File

@ -1,6 +1,5 @@
from .base import * from .base import HealthResponseModel
__all__ = [ __all__ = [
"HealthResponseModel", "HealthResponseModel",
"TokenResponseModel",
] ]

View File

@ -1,12 +1,5 @@
import uuid
import pydantic import pydantic
class TokenResponseModel(pydantic.BaseModel):
sub: uuid.UUID
exp: int | None = None
class HealthResponseModel(pydantic.BaseModel): class HealthResponseModel(pydantic.BaseModel):
status: str = pydantic.Field(default=..., examples=["healthy"], description="Схема доступности сервиса") status: str = pydantic.Field(default=..., examples=["healthy"], description="Схема доступности сервиса")

View File

@ -1,26 +0,0 @@
import fastapi
from jose import JWTError, jwt
from pydantic import ValidationError
import lib.app.settings as app_settings
from lib.api.v1 import schemas as app_schemas
app = fastapi.FastAPI()
settings = app_settings.settings
security = fastapi.security.HTTPBearer()
def get_token_data(
authorization: fastapi.security.HTTPAuthorizationCredentials = fastapi.Security(security),
) -> app_schemas.entity.Token:
token = authorization.credentials
try:
secret_key = settings.project.jwt_secret_key
payload = jwt.decode(token, secret_key, algorithms=["HS256"])
return app_schemas.entity.Token(**payload)
except (JWTError, ValidationError):
raise fastapi.HTTPException(
status_code=fastapi.status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
)

View File

@ -5,6 +5,8 @@ import lib.app.split_settings.utils as app_split_settings_utils
class ProjectSettings(pydantic_settings.BaseSettings): class ProjectSettings(pydantic_settings.BaseSettings):
"""Project settings."""
model_config = pydantic_settings.SettingsConfigDict( model_config = pydantic_settings.SettingsConfigDict(
env_file=app_split_settings_utils.ENV_PATH, env_file=app_split_settings_utils.ENV_PATH,
env_file_encoding="utf-8", env_file_encoding="utf-8",
@ -12,7 +14,8 @@ class ProjectSettings(pydantic_settings.BaseSettings):
) )
debug: str = "false" debug: str = "false"
jwt_secret_key: pydantic.SecretStr = pydantic.Field(default=..., validation_alias="jwt_secret_key") jwt_secret_key: str = pydantic.Field(default=..., validation_alias="jwt_secret_key")
jwt_algorithm: str = "HS256"
@pydantic.field_validator("debug") @pydantic.field_validator("debug")
def validate_debug(cls, v: str) -> bool: def validate_debug(cls, v: str) -> bool:

View File

@ -1,3 +1,4 @@
from .base_sqlalchemy import Base from .base_sqlalchemy import Base
from .token import Token
__all__ = ["Base"] __all__ = ["Base", "Token"]

View File

@ -0,0 +1,9 @@
import uuid
import pydantic
# TODO: TBU
class Token(pydantic.BaseModel):
sub: uuid.UUID
exp: int | None = None

View File

@ -0,0 +1,25 @@
import fastapi
import fastapi.security
import jose
import jose.jwt
import pydantic
import lib.app.settings as app_settings
import lib.models as models
def get_token_data(
authorization: fastapi.security.HTTPAuthorizationCredentials = fastapi.Security(fastapi.security.HTTPBearer()),
) -> models.Token:
settings = app_settings.Settings()
token = authorization.credentials
try:
secret_key = settings.project.jwt_secret_key
payload = jose.jwt.decode(token, secret_key, algorithms=[settings.project.jwt_algorithm])
return models.Token(**payload)
except (jose.JWTError, pydantic.ValidationError) as error:
raise fastapi.HTTPException(
status_code=fastapi.status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
) from error