ruff and prettier fixes
Build and Push Docker Image / build-and-push (pull_request) Successful in 1m6s
Python Code Quality / python-code-quality (pull_request) Successful in 10s
Python Test / python-test (pull_request) Successful in 15s

This commit is contained in:
Brian Bjarke Jensen
2025-11-07 23:10:59 +01:00
parent 469fd722de
commit 2b956213aa
10 changed files with 644 additions and 655 deletions
+7 -7
View File
@@ -34,31 +34,31 @@ async def generate_invitation_link(
],
) -> InvitationResponse:
"""Generate a new invitation link for user registration.
The invitation token is valid for 24 hours and can be used once.
Requires admin role.
Args:
user_id: Admin user ID (from verify_admin)
invitation_repository: Invitation storage backend
Returns:
InvitationResponse with token and expiration details
"""
# Generate secure random token
invitation_token = secrets.token_urlsafe(32)
# Calculate expiration (24 hours from now)
expires_at = datetime.now(UTC) + timedelta(hours=24)
# Store invitation token in database
invitation_repository.create_invitation(
token=invitation_token,
created_by_user_id=user_id,
expires_at=expires_at,
)
return InvitationResponse(
token=invitation_token,
expires_at=expires_at.isoformat(),
+22 -40
View File
@@ -52,27 +52,23 @@ def verify_token(
def verify_admin(
user_id: Annotated[int, Depends(verify_token)],
user_repo: Annotated[
UserRepositoryInterface, Depends(get_user_repository)
],
user_repo: Annotated[UserRepositoryInterface, Depends(get_user_repository)],
) -> int:
"""Verify the user is an admin and return user_id."""
# First try to get user from database
user = user_repo.get_by_id(user_id)
if user:
# Database user - check is_admin field
if not user.get("is_admin", False):
raise HTTPException(
status_code=403, detail="Admin access required"
)
raise HTTPException(status_code=403, detail="Admin access required")
return user_id
# If not in database but has valid token with user_id=1,
# it's the environment-based admin (only assigned during env admin login)
if user_id == 1:
return user_id
# User not found and not environment admin
raise HTTPException(status_code=401, detail="User not found")
@@ -80,12 +76,8 @@ def verify_admin(
@router.post("/login", response_model=LoginResponse)
def login(
credentials: LoginRequest,
token_repo: Annotated[
TokenRepositoryInterface, Depends(get_token_repository)
],
user_repo: Annotated[
UserRepositoryInterface, Depends(get_user_repository)
],
token_repo: Annotated[TokenRepositoryInterface, Depends(get_token_repository)],
user_repo: Annotated[UserRepositoryInterface, Depends(get_user_repository)],
creds_repo: Annotated[
CredentialsRepositoryInterface, Depends(get_credentials_repository)
],
@@ -112,11 +104,9 @@ def login(
access_token=access_token,
is_admin=user.get("is_admin", False),
)
# Fallback to admin credentials from environment
if creds_repo.verify_admin_credentials(
credentials.username, credentials.password
):
if creds_repo.verify_admin_credentials(credentials.username, credentials.password):
# Generate a secure random token
access_token = secrets.token_urlsafe(32)
# Store token with user_id (hardcoded 1 for admin)
@@ -128,10 +118,8 @@ def login(
access_token=access_token,
is_admin=True,
)
raise HTTPException(
status_code=401, detail="Invalid username or password"
)
raise HTTPException(status_code=401, detail="Invalid username or password")
@router.post("/logout")
@@ -162,19 +150,15 @@ def verify_invitation(
@router.post("/register", response_model=RegisterResponse)
def register(
request: RegisterRequest,
user_repo: Annotated[
UserRepositoryInterface, Depends(get_user_repository)
],
user_repo: Annotated[UserRepositoryInterface, Depends(get_user_repository)],
invitation_repo: Annotated[
InvitationRepositoryInterface, Depends(get_invitation_repository)
],
token_repo: Annotated[
TokenRepositoryInterface, Depends(get_token_repository)
],
token_repo: Annotated[TokenRepositoryInterface, Depends(get_token_repository)],
) -> RegisterResponse:
"""
Register a new user with an invitation token.
Verifies the invitation, creates the user, consumes the invitation,
and returns an authentication token.
"""
@@ -184,7 +168,7 @@ def register(
status_code=400,
detail="Invalid or expired invitation token",
)
# Check if username already exists
existing_user = user_repo.get_by_username(request.username)
if existing_user:
@@ -192,21 +176,21 @@ def register(
status_code=400,
detail="Username already exists",
)
# Create the new user
# TODO: Hash password before storing (currently plain text)
user = user_repo.create(
username=request.username,
hashed_password=request.password,
)
# Consume the invitation token
invitation_repo.consume_invitation(request.invitation_token)
# Generate authentication token
access_token = secrets.token_urlsafe(32)
token_repo.store(access_token, user_id=user["id"], ttl=3600)
return RegisterResponse(
message="Registration successful",
username=user["username"],
@@ -218,16 +202,14 @@ def register(
@router.get("/me")
def get_current_user(
user_id: Annotated[int, Depends(verify_token)],
user_repo: Annotated[
UserRepositoryInterface, Depends(get_user_repository)
],
user_repo: Annotated[UserRepositoryInterface, Depends(get_user_repository)],
creds_repo: Annotated[
CredentialsRepositoryInterface, Depends(get_credentials_repository)
],
) -> dict:
"""Get current user info including admin status."""
user = user_repo.get_by_id(user_id)
# If user not found in DB, might be env-based admin
if not user:
# Return admin info from environment
@@ -236,7 +218,7 @@ def get_current_user(
"username": creds_repo.get_admin_username(),
"is_admin": True, # Env-based admin is always admin
}
return {
"id": user["id"],
"username": user["username"],