fix auth lifetime

This commit is contained in:
2024-04-07 21:31:15 +00:00
parent f8f5bf80c1
commit d55e8d1df3
13 changed files with 212 additions and 42 deletions

View File

@@ -17,7 +17,7 @@ class JoseJwtTokenProcessor(JwtTokenProcessor):
def generate_token(self, user_id: UserId) -> str:
issued_at = self.date_time_provider.get_current_time()
expiration_time = issued_at + timedelta(hours=self.jwt_options.expires_in)
expiration_time = issued_at + timedelta(minutes=self.jwt_options.expires_in)
claims = {
"iat": issued_at,
@@ -30,8 +30,10 @@ class JoseJwtTokenProcessor(JwtTokenProcessor):
def validate_token(self, token: str) -> UserId | None:
try:
payload = decode(token, self.jwt_options.secret, [self.jwt_options.algorithm])
return UserId(UUID(payload["sub"]))
except (JWTError, ValueError, KeyError):
return None
def refresh_token(self, token: str) -> str:
return ""

View File

@@ -1,9 +1,12 @@
import os
from functools import lru_cache
from typing import Annotated
import yaml # type: ignore
from fastapi import Depends
from api.infrastructure.auth.jwt_settings import JwtSettings
from api.infrastructure.dependencies.stub import Stub
from api.infrastructure.persistence.db_setings import DBSettings
from api.infrastructure.settings import Settings
@@ -14,23 +17,36 @@ def yaml_loader(file: str) -> dict[str, dict[str, str]]:
return yaml_data
@lru_cache
def app_settings() -> Settings:
def get_db_settings() -> DBSettings:
config_data = yaml_loader(
file=os.getenv("CONFIG_PATH", "./config/api_config.yml"),
)
return Settings(
db=DBSettings(
pg_user=config_data["db"]["user"],
pg_pass=config_data["db"]["password"],
pg_host=config_data["db"]["host"],
pg_port=int(config_data["db"]["port"]),
pg_db=config_data["db"]["database"],
),
jwt=JwtSettings(
secret=config_data["jwt"]["secret_key"],
expires_in=int(config_data["jwt"]["expires_in"]),
algorithm=config_data["jwt"]["algorithm"],
),
return DBSettings(
pg_user=config_data["db"]["user"],
pg_pass=config_data["db"]["password"],
pg_host=config_data["db"]["host"],
pg_port=int(config_data["db"]["port"]),
pg_db=config_data["db"]["database"],
)
def get_jwt_settings() -> JwtSettings:
config_data = yaml_loader(
file=os.getenv("CONFIG_PATH", "./config/api_config.yml"),
)
return JwtSettings(
secret=config_data["jwt"]["secret_key"],
expires_in=int(config_data["jwt"]["expires_in"]),
algorithm=config_data["jwt"]["algorithm"],
)
@lru_cache
def app_settings(
db_conf: Annotated[DBSettings, Depends(Stub(DBSettings))],
jwt_conf: Annotated[JwtSettings, Depends(Stub(JwtSettings))],
) -> Settings:
return Settings(
db=db_conf,
jwt=jwt_conf,
)

View File

@@ -26,9 +26,7 @@ def get_jwt_token_processor(
settings: Annotated[Settings, Depends(Stub(Settings))],
date_time_provider: Annotated[DateTimeProvider, Depends(Stub(DateTimeProvider))],
) -> JwtTokenProcessor:
return JoseJwtTokenProcessor(
jwt_options=settings.jwt, date_time_provider=date_time_provider
)
return JoseJwtTokenProcessor(jwt_options=settings.jwt, date_time_provider=date_time_provider)
def get_user_login(

View File

@@ -28,9 +28,12 @@ class SqlAlchemyUserRepository(UserRepository):
async def get_user(self, filter: dict) -> User | None:
stmt = text("""SELECT * FROM users WHERE email = :val""")
result = await self.session.execute(stmt, {"val": filter["email"]})
if not result:
result = result.mappings().one_or_none()
if result is None:
return None
result = result.mappings().one()
return User(
id=UserId(result.id),
name=UserFirstName(result.name),