Correction token duration and added token validity check endpoint
parent
a0d1fe40a9
commit
81a93c5683
2
.env
2
.env
|
|
@ -4,7 +4,7 @@ DATABASE_URL=mysql+aiomysql://sywmtnsg_admin:EEy_>2JJS0@localhost:6033/sywmtnsg_
|
||||||
# Configuration pour JWT
|
# Configuration pour JWT
|
||||||
SECRET_KEY=LAGs7G8Sis9aQHcipROxpjYRxFZKjr4wNm-_O0pBTkjNYv1rgPUR87VcNswH_VYGpIrsyGdqnNa3vcVSH0f5Tg
|
SECRET_KEY=LAGs7G8Sis9aQHcipROxpjYRxFZKjr4wNm-_O0pBTkjNYv1rgPUR87VcNswH_VYGpIrsyGdqnNa3vcVSH0f5Tg
|
||||||
ALGORITHM=HS256
|
ALGORITHM=HS256
|
||||||
ACCESS_TOKEN_EXPIRE_MINUTES=30
|
ACCESS_TOKEN_EXPIRE_MINUTES=4320
|
||||||
|
|
||||||
# Configuration pour AWS S3 (à remplir si nécessaire)
|
# Configuration pour AWS S3 (à remplir si nécessaire)
|
||||||
#AWS_ACCESS_KEY_ID=your-aws-access-key
|
#AWS_ACCESS_KEY_ID=your-aws-access-key
|
||||||
|
|
|
||||||
|
|
@ -43,6 +43,23 @@ async def login_for_access_token( form_data: OAuth2PasswordRequestForm = Depends
|
||||||
access_token = AuthService.create_access_token(data={"sub": user["email"]})
|
access_token = AuthService.create_access_token(data={"sub": user["email"]})
|
||||||
return {"access_token": access_token, "token_type": "bearer"}
|
return {"access_token": access_token, "token_type": "bearer"}
|
||||||
|
|
||||||
|
@router.post("/isvalid-token", summary="Verify token validity")
|
||||||
|
async def verify_token(token: str = Body(...)):
|
||||||
|
try:
|
||||||
|
# Décoder le token pour vérifier sa validité
|
||||||
|
payload = jwt.decode(token, settings.secret_key, algorithms=[settings.algorithm])
|
||||||
|
return {"valid": True, "message": "Token is valid", "payload": payload}
|
||||||
|
except jwt.ExpiredSignatureError:
|
||||||
|
raise HTTPException(
|
||||||
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||||
|
detail="Token has expired",
|
||||||
|
)
|
||||||
|
except JWTError:
|
||||||
|
raise HTTPException(
|
||||||
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||||
|
detail="Invalid token",
|
||||||
|
)
|
||||||
|
|
||||||
@router.get("/me", summary="Get current user")
|
@router.get("/me", summary="Get current user")
|
||||||
async def read_users_me(token:str = Depends(oauth2_scheme) , db=Depends(get_db)):
|
async def read_users_me(token:str = Depends(oauth2_scheme) , db=Depends(get_db)):
|
||||||
return await AuthService.get_current_user(token, db)
|
return await AuthService.get_current_user(token, db)
|
||||||
|
|
|
||||||
|
|
@ -5,7 +5,7 @@ class Settings(BaseSettings):
|
||||||
database_url: str = "mysql+aiomysql://sywmtnsg_admin:EEy_>2JJS0@localhost:6033/sywmtnsg_dm_management"
|
database_url: str = "mysql+aiomysql://sywmtnsg_admin:EEy_>2JJS0@localhost:6033/sywmtnsg_dm_management"
|
||||||
secret_key: str = "LAGs7G8Sis9aQHcipROxpjYRxFZKjr4wNm-_O0pBTkjNYv1rgPUR87VcNswH_VYGpIrsyGdqnNa3vcVSH0f5Tg"
|
secret_key: str = "LAGs7G8Sis9aQHcipROxpjYRxFZKjr4wNm-_O0pBTkjNYv1rgPUR87VcNswH_VYGpIrsyGdqnNa3vcVSH0f5Tg"
|
||||||
algorithm: str = "HS256"
|
algorithm: str = "HS256"
|
||||||
access_token_expire_minutes: int = 30
|
access_token_expire_minutes: int = 4320
|
||||||
aws_access_key_id: str = ""
|
aws_access_key_id: str = ""
|
||||||
aws_secret_access_key: str = ""
|
aws_secret_access_key: str = ""
|
||||||
aws_bucket_name: str = ""
|
aws_bucket_name: str = ""
|
||||||
|
|
|
||||||
|
|
@ -62,6 +62,32 @@ class AuthService:
|
||||||
expire = datetime.now(timezone.utc) + timedelta(minutes=settings.access_token_expire_minutes)
|
expire = datetime.now(timezone.utc) + timedelta(minutes=settings.access_token_expire_minutes)
|
||||||
to_encode.update({"exp": expire})
|
to_encode.update({"exp": expire})
|
||||||
return jwt.encode(to_encode, settings.secret_key, algorithm=settings.algorithm)
|
return jwt.encode(to_encode, settings.secret_key, algorithm=settings.algorithm)
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def verify_token(token: str) -> dict:
|
||||||
|
try:
|
||||||
|
# Décoder le token
|
||||||
|
payload = jwt.decode(token, settings.secret_key, algorithms=[settings.algorithm])
|
||||||
|
|
||||||
|
# Vérifier si le token est expiré
|
||||||
|
expiration_time = payload.get("exp")
|
||||||
|
if expiration_time and datetime.now(timezone.utc).timestamp() > expiration_time:
|
||||||
|
raise HTTPException(
|
||||||
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||||
|
detail="Token has expired",
|
||||||
|
)
|
||||||
|
|
||||||
|
return payload
|
||||||
|
except jwt.ExpiredSignatureError:
|
||||||
|
raise HTTPException(
|
||||||
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||||
|
detail="Token has expired",
|
||||||
|
)
|
||||||
|
except JWTError:
|
||||||
|
raise HTTPException(
|
||||||
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||||
|
detail="Invalid token",
|
||||||
|
)
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
async def get_current_user(token: str, db):
|
async def get_current_user(token: str, db):
|
||||||
|
|
|
||||||
|
|
@ -13,6 +13,6 @@ def verify_password(plain_password: str, hashed_password: str) -> bool:
|
||||||
|
|
||||||
def create_access_token(data: dict) -> str:
|
def create_access_token(data: dict) -> str:
|
||||||
to_encode = data.copy()
|
to_encode = data.copy()
|
||||||
expire = datetime.utcnow() + timedelta(minutes=settings.access_token_expire_minutes)
|
expire = datetime.now(datetime.timezone.utc) + timedelta(minutes=settings.access_token_expire_minutes)
|
||||||
to_encode.update({"exp": expire})
|
to_encode.update({"exp": expire})
|
||||||
return jwt.encode(to_encode, settings.secret_key, algorithm=settings.algorithm)
|
return jwt.encode(to_encode, settings.secret_key, algorithm=settings.algorithm)
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue