🐛 fix(auth.py): change create_user_longterm_token to use create_super_user to ensure super user exists before creating token

✨ feat(auth.py): add create_super_user function to create super user if it doesn't exist
🐛 fix(login.py): remove hardcoded user_id and use create_user_longterm_token without arguments
🐛 fix(users.py): remove redundant password hashing in add_super_user_for_testing_purposes_delete_me_before_merge_into_dev
🐛 fix(base.py): change AUTO_LOGIN default value to False and add FIRST_SUPERUSER and FIRST_SUPERUSER_PASSWORD settings
This commit is contained in:
gustavoschaedler 2023-08-15 20:32:53 +01:00 • committed by anovazzi1
commit 19c1eb6cf6
4 changed files with 34 additions and 12 deletions

View file

@ -83,18 +83,40 @@ def create_token(data: dict, expires_delta: timedelta):
) )
def create_user_longterm_token( def create_super_user(db: Session = Depends(get_session)) -> User:
user_id: UUID, db: Session = Depends(get_session), update_last_login: bool = False settings_manager = get_settings_manager()
) -> dict:
super_user = get_user_by_username(db, settings_manager.settings.FIRST_SUPERUSER)
if not super_user:
super_user = User(
username=settings_manager.settings.FIRST_SUPERUSER,
password=get_password_hash(
settings_manager.settings.FIRST_SUPERUSER_PASSWORD
),
is_superuser=True,
is_active=True,
last_login_at=None,
)
db.add(super_user)
db.commit()
db.refresh(super_user)
return super_user
def create_user_longterm_token(db: Session = Depends(get_session)) -> dict:
super_user = create_super_user(db)
access_token_expires_longterm = timedelta(days=365) access_token_expires_longterm = timedelta(days=365)
access_token = create_token( access_token = create_token(
data={"sub": str(user_id)}, data={"sub": str(super_user.id)},
expires_delta=access_token_expires_longterm, expires_delta=access_token_expires_longterm,
) )
# Update: last_login_at # Update: last_login_at
if update_last_login: update_user_last_login_at(super_user.id, db)
update_user_last_login_at(user_id, db)
return { return {
"access_token": access_token, "access_token": access_token,

View file

@ -1,4 +1,3 @@
from uuid import UUID
from sqlalchemy.orm import Session from sqlalchemy.orm import Session
from fastapi import APIRouter, Depends, HTTPException, status from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm from fastapi.security import OAuth2PasswordRequestForm
@ -38,8 +37,7 @@ async def auto_login(db: Session = Depends(get_session)):
settings_manager = get_settings_manager() settings_manager = get_settings_manager()
if settings_manager.settings.AUTO_LOGIN: if settings_manager.settings.AUTO_LOGIN:
user_id = UUID("3fa85f64-5717-4562-b3fc-2c963f66afa6") return create_user_longterm_token(db)
return create_user_longterm_token(user_id, db)
raise HTTPException( raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST, status_code=status.HTTP_400_BAD_REQUEST,

View file

@ -115,14 +115,13 @@ def add_super_user_for_testing_purposes_delete_me_before_merge_into_dev(
""" """
new_user = User( new_user = User(
username="superuser", username="superuser",
password="12345", password=get_password_hash("12345"),
is_active=True, is_active=True,
is_superuser=True, is_superuser=True,
last_login_at=None, last_login_at=None,
) )
try: try:
new_user.password = get_password_hash(new_user.password)
db.add(new_user) db.add(new_user)
db.commit() db.commit()
db.refresh(new_user) db.refresh(new_user)

View file

@ -41,9 +41,12 @@ class Settings(BaseSettings):
ALGORITHM: str = "HS256" ALGORITHM: str = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES: int = 60 ACCESS_TOKEN_EXPIRE_MINUTES: int = 60
REFRESH_TOKEN_EXPIRE_MINUTES: int = 70 REFRESH_TOKEN_EXPIRE_MINUTES: int = 70
# If AUTO_LOGIN = True # If AUTO_LOGIN = True
# > The application does not request login and logs in automatically as a super user. # > The application does not request login and logs in automatically as a super user.
AUTO_LOGIN: bool = True AUTO_LOGIN: bool = False
FIRST_SUPERUSER: str = "superuser"
FIRST_SUPERUSER_PASSWORD: str = "12345"
@validator("DATABASE_URL", pre=True) @validator("DATABASE_URL", pre=True)
def set_database_url(cls, value): def set_database_url(cls, value):