auth unit tests & ruff visual fix
This commit is contained in:
@@ -1,13 +1,15 @@
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from datetime import UTC, datetime, timedelta
|
||||
from uuid import UUID
|
||||
|
||||
from fastapi import Request
|
||||
from .jwt import Jwt, Hashes
|
||||
from src.database.users.crud import UsersCrudActions
|
||||
|
||||
from src.database.auth.refresh_tokens import JwtCrudActions
|
||||
from src.database.users.crud import UsersCrudActions
|
||||
from src.errors.http_errors.errors import Errors
|
||||
from src.models.pydantic_models.model import RefreshTokensCreate, UserOut
|
||||
from src.models.configs_read.env import env_settings
|
||||
from src.models.pydantic_models.model import RefreshTokensCreate, UserOut
|
||||
|
||||
from .jwt import Hashes, Jwt
|
||||
|
||||
|
||||
class CurrentUser:
|
||||
@@ -78,7 +80,7 @@ class CurrentUser:
|
||||
token_hash=self.hash.token_to_hash(token),
|
||||
device_info=request.headers.get("user-agent", "unknown"),
|
||||
ip_address=request.headers.get("x-forwarded-for", "").split(",")[0].strip() or (request.client.host if request.client else "unknown"),
|
||||
expires_at=datetime.now(timezone.utc)+timedelta(days=env_settings.REFRESH_TOKEN_EXPIRE_DAYS)
|
||||
expires_at=datetime.now(UTC)+timedelta(days=env_settings.REFRESH_TOKEN_EXPIRE_DAYS)
|
||||
)
|
||||
self.jwt_db_actions.create_token(RefreshTokensCreate.model_dump(token_record))
|
||||
|
||||
@@ -113,8 +115,8 @@ class CurrentUser:
|
||||
'''sqlite constraints about timezone'''
|
||||
expires_at=old_record.expires_at
|
||||
if expires_at.tzinfo is None:
|
||||
expires_at = expires_at.replace(tzinfo=timezone.utc)
|
||||
if expires_at<datetime.now(timezone.utc):
|
||||
expires_at = expires_at.replace(tzinfo=UTC)
|
||||
if expires_at<datetime.now(UTC):
|
||||
raise self.error.credentials_error(detail="Token expired")
|
||||
|
||||
'''user check'''
|
||||
@@ -142,7 +144,7 @@ class CurrentUser:
|
||||
token_hash=self.hash.token_to_hash(new_refresh_token),
|
||||
device_info=request.headers.get("user-agent", "unknown"),
|
||||
ip_address=request.headers.get("x-forwarded-for", "").split(",")[0].strip() or (request.client.host if request.client else "unknown"),
|
||||
expires_at=datetime.now(timezone.utc)+timedelta(days=env_settings.REFRESH_TOKEN_EXPIRE_DAYS),
|
||||
expires_at=datetime.now(UTC)+timedelta(days=env_settings.REFRESH_TOKEN_EXPIRE_DAYS),
|
||||
)
|
||||
self.jwt_db_actions.create_token(RefreshTokensCreate.model_dump(new_token_record))
|
||||
|
||||
|
||||
+12
-10
@@ -1,10 +1,12 @@
|
||||
from jose import JWTError, jwt
|
||||
import bcrypt
|
||||
from src.errors.http_errors.errors import Errors
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from src.models.configs_read.env import env_settings
|
||||
from uuid import uuid4
|
||||
import hashlib
|
||||
from datetime import UTC, datetime, timedelta
|
||||
from uuid import uuid4
|
||||
|
||||
import bcrypt
|
||||
from jose import JWTError, jwt
|
||||
|
||||
from src.errors.http_errors.errors import Errors
|
||||
from src.models.configs_read.env import env_settings
|
||||
|
||||
'''Hash/Check hash'''
|
||||
class Hashes:
|
||||
@@ -32,7 +34,7 @@ class Jwt:
|
||||
def create_access_token(self, data:dict)->str:
|
||||
|
||||
user_info=data.copy()
|
||||
user_info.update({"exp": datetime.now(timezone.utc)+timedelta(minutes=env_settings.ACCESS_TOKEN_EXPIRE_MINUTES),
|
||||
user_info.update({"exp": datetime.now(UTC)+timedelta(minutes=env_settings.ACCESS_TOKEN_EXPIRE_MINUTES),
|
||||
"token_type":"access"})
|
||||
return jwt.encode(user_info, env_settings.SECRET_KEY, env_settings.ALGORITHM)
|
||||
|
||||
@@ -41,7 +43,7 @@ class Jwt:
|
||||
|
||||
user_info=data.copy()
|
||||
jti=str(uuid4())
|
||||
user_info.update({"exp":datetime.now(timezone.utc)+timedelta(days=env_settings.REFRESH_TOKEN_EXPIRE_DAYS),
|
||||
user_info.update({"exp":datetime.now(UTC)+timedelta(days=env_settings.REFRESH_TOKEN_EXPIRE_DAYS),
|
||||
"token_type":"refresh",
|
||||
"jti":jti
|
||||
})
|
||||
@@ -53,9 +55,9 @@ class Jwt:
|
||||
def jwt_decode(self, token:str)->dict:
|
||||
|
||||
try:
|
||||
payload=jwt.decode(token, env_settings.SECRET_KEY, algorithms=[env_settings.ALGORITHM])
|
||||
payload=jwt.decode(token, env_settings.SECRET_KEY, algorithms=[env_settings.ALGORITHM], options={"require_exp": True} )
|
||||
|
||||
if (payload.get("sub")) is None:
|
||||
if not (payload.get("sub")) or not (payload.get("token_type")):
|
||||
raise self.error.credentials_error(detail="Jwt token is incorrect")
|
||||
|
||||
except JWTError as e:
|
||||
|
||||
Reference in New Issue
Block a user