mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-10-18 11:56:01 +02:00
Add backend support for advanced authentication
Co-authored-by: cadtoolbox <12723486+cadtoolbox@users.noreply.github.com>
This commit is contained in:
copilot-swe-agent[bot]
co-authored by
cadtoolbox
cadtoolbox
parent
453565d9f8
commit
1058f3fd5c
@@ -8,16 +8,39 @@ from sqlalchemy.orm import selectinload
|
||||
from backend.app.core.auth import (
|
||||
ACCESS_TOKEN_EXPIRE_MINUTES,
|
||||
authenticate_user,
|
||||
authenticate_user_by_email,
|
||||
create_access_token,
|
||||
get_current_active_user,
|
||||
get_password_hash,
|
||||
get_user_by_email,
|
||||
get_user_by_username,
|
||||
)
|
||||
from backend.app.core.database import get_db
|
||||
from backend.app.models.group import Group
|
||||
from backend.app.models.settings import Settings
|
||||
from backend.app.models.user import User
|
||||
from backend.app.schemas.auth import GroupBrief, LoginRequest, LoginResponse, SetupRequest, SetupResponse, UserResponse
|
||||
from backend.app.schemas.auth import (
|
||||
ForgotPasswordRequest,
|
||||
ForgotPasswordResponse,
|
||||
GroupBrief,
|
||||
LoginRequest,
|
||||
LoginResponse,
|
||||
ResetPasswordRequest,
|
||||
ResetPasswordResponse,
|
||||
SetupRequest,
|
||||
SetupResponse,
|
||||
SMTPSettings,
|
||||
TestSMTPRequest,
|
||||
TestSMTPResponse,
|
||||
UserResponse,
|
||||
)
|
||||
from backend.app.services.email_service import (
|
||||
create_password_reset_email,
|
||||
generate_secure_password,
|
||||
get_smtp_settings,
|
||||
save_smtp_settings,
|
||||
send_email,
|
||||
)
|
||||
|
||||
|
||||
def _user_to_response(user: User) -> UserResponse:
|
||||
@@ -25,6 +48,7 @@ def _user_to_response(user: User) -> UserResponse:
|
||||
return UserResponse(
|
||||
id=user.id,
|
||||
username=user.username,
|
||||
email=user.email,
|
||||
role=user.role,
|
||||
is_active=user.is_active,
|
||||
is_admin=user.is_admin,
|
||||
@@ -46,6 +70,27 @@ async def is_auth_enabled(db: AsyncSession) -> bool:
|
||||
return setting.value.lower() == "true"
|
||||
|
||||
|
||||
async def is_advanced_auth_enabled(db: AsyncSession) -> bool:
|
||||
"""Check if advanced authentication is enabled."""
|
||||
result = await db.execute(select(Settings).where(Settings.key == "advanced_auth_enabled"))
|
||||
setting = result.scalar_one_or_none()
|
||||
if setting is None:
|
||||
return False
|
||||
return setting.value.lower() == "true"
|
||||
|
||||
|
||||
async def set_advanced_auth_enabled(db: AsyncSession, enabled: bool) -> None:
|
||||
"""Set advanced authentication enabled status."""
|
||||
from sqlalchemy import func
|
||||
from sqlalchemy.dialects.sqlite import insert as sqlite_insert
|
||||
|
||||
stmt = sqlite_insert(Settings).values(key="advanced_auth_enabled", value="true" if enabled else "false")
|
||||
stmt = stmt.on_conflict_do_update(
|
||||
index_elements=["key"], set_={"value": "true" if enabled else "false", "updated_at": func.now()}
|
||||
)
|
||||
await db.execute(stmt)
|
||||
|
||||
|
||||
async def set_auth_enabled(db: AsyncSession, enabled: bool) -> None:
|
||||
"""Set authentication enabled status."""
|
||||
from sqlalchemy import func
|
||||
@@ -216,7 +261,10 @@ async def disable_auth(
|
||||
|
||||
@router.post("/login", response_model=LoginResponse)
|
||||
async def login(request: LoginRequest, db: AsyncSession = Depends(get_db)):
|
||||
"""Login and get access token."""
|
||||
"""Login and get access token.
|
||||
|
||||
Supports username or email-based login. Username lookup is case-insensitive.
|
||||
"""
|
||||
# Check if auth is enabled
|
||||
auth_enabled = await is_auth_enabled(db)
|
||||
if not auth_enabled:
|
||||
@@ -225,7 +273,15 @@ async def login(request: LoginRequest, db: AsyncSession = Depends(get_db)):
|
||||
detail="Authentication is not enabled",
|
||||
)
|
||||
|
||||
# Try username-based authentication first
|
||||
user = await authenticate_user(db, request.username, request.password)
|
||||
|
||||
# If username auth failed and advanced auth is enabled, try email-based authentication
|
||||
if not user:
|
||||
advanced_auth = await is_advanced_auth_enabled(db)
|
||||
if advanced_auth:
|
||||
user = await authenticate_user_by_email(db, request.username, request.password)
|
||||
|
||||
if not user:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
@@ -263,3 +319,331 @@ async def get_current_user_info(
|
||||
async def logout():
|
||||
"""Logout (client should discard token)."""
|
||||
return {"message": "Logged out successfully"}
|
||||
|
||||
|
||||
# Advanced Authentication Endpoints
|
||||
|
||||
|
||||
@router.post("/smtp/test", response_model=TestSMTPResponse)
|
||||
async def test_smtp_connection(
|
||||
test_request: TestSMTPRequest,
|
||||
current_user: User = Depends(get_current_active_user),
|
||||
db: AsyncSession = Depends(get_db),
|
||||
):
|
||||
"""Test SMTP connection with provided settings (admin only)."""
|
||||
import logging
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Reload user with groups for proper is_admin check
|
||||
result = await db.execute(select(User).where(User.id == current_user.id).options(selectinload(User.groups)))
|
||||
user = result.scalar_one()
|
||||
|
||||
if not user.is_admin:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_403_FORBIDDEN,
|
||||
detail="Only admins can test SMTP settings",
|
||||
)
|
||||
|
||||
try:
|
||||
smtp_settings = SMTPSettings(
|
||||
smtp_host=test_request.smtp_host,
|
||||
smtp_port=test_request.smtp_port,
|
||||
smtp_username=test_request.smtp_username,
|
||||
smtp_password=test_request.smtp_password,
|
||||
smtp_use_tls=test_request.smtp_use_tls,
|
||||
smtp_from_email=test_request.smtp_from_email,
|
||||
)
|
||||
|
||||
# Send test email
|
||||
send_email(
|
||||
smtp_settings=smtp_settings,
|
||||
to_email=test_request.test_recipient,
|
||||
subject="BamBuddy SMTP Test",
|
||||
body_text="This is a test email from BamBuddy. If you received this, your SMTP settings are working correctly!",
|
||||
body_html="<p>This is a test email from <strong>BamBuddy</strong>.</p><p>If you received this, your SMTP settings are working correctly!</p>",
|
||||
)
|
||||
|
||||
logger.info(f"Test email sent successfully to {test_request.test_recipient}")
|
||||
return TestSMTPResponse(success=True, message="Test email sent successfully")
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to send test email: {e}")
|
||||
return TestSMTPResponse(success=False, message=f"Failed to send test email: {str(e)}")
|
||||
|
||||
|
||||
@router.get("/smtp", response_model=SMTPSettings | None)
|
||||
async def get_smtp_config(
|
||||
current_user: User = Depends(get_current_active_user),
|
||||
db: AsyncSession = Depends(get_db),
|
||||
):
|
||||
"""Get SMTP settings (admin only). Password is not returned."""
|
||||
# Reload user with groups for proper is_admin check
|
||||
result = await db.execute(select(User).where(User.id == current_user.id).options(selectinload(User.groups)))
|
||||
user = result.scalar_one()
|
||||
|
||||
if not user.is_admin:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_403_FORBIDDEN,
|
||||
detail="Only admins can view SMTP settings",
|
||||
)
|
||||
|
||||
smtp_settings = await get_smtp_settings(db)
|
||||
if smtp_settings:
|
||||
# Don't return password in response
|
||||
smtp_settings.smtp_password = None
|
||||
return smtp_settings
|
||||
|
||||
|
||||
@router.post("/smtp", response_model=dict)
|
||||
async def save_smtp_config(
|
||||
smtp_settings: SMTPSettings,
|
||||
current_user: User = Depends(get_current_active_user),
|
||||
db: AsyncSession = Depends(get_db),
|
||||
):
|
||||
"""Save SMTP settings (admin only)."""
|
||||
import logging
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Reload user with groups for proper is_admin check
|
||||
result = await db.execute(select(User).where(User.id == current_user.id).options(selectinload(User.groups)))
|
||||
user = result.scalar_one()
|
||||
|
||||
if not user.is_admin:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_403_FORBIDDEN,
|
||||
detail="Only admins can update SMTP settings",
|
||||
)
|
||||
|
||||
try:
|
||||
await save_smtp_settings(db, smtp_settings)
|
||||
await db.commit()
|
||||
logger.info(f"SMTP settings updated by admin user: {user.username}")
|
||||
return {"message": "SMTP settings saved successfully"}
|
||||
except Exception as e:
|
||||
await db.rollback()
|
||||
logger.error(f"Failed to save SMTP settings: {e}")
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||||
detail=f"Failed to save SMTP settings: {str(e)}",
|
||||
)
|
||||
|
||||
|
||||
@router.post("/advanced-auth/enable", response_model=dict)
|
||||
async def enable_advanced_auth(
|
||||
current_user: User = Depends(get_current_active_user),
|
||||
db: AsyncSession = Depends(get_db),
|
||||
):
|
||||
"""Enable advanced authentication (admin only).
|
||||
|
||||
Requires SMTP settings to be configured and tested first.
|
||||
"""
|
||||
import logging
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Reload user with groups for proper is_admin check
|
||||
result = await db.execute(select(User).where(User.id == current_user.id).options(selectinload(User.groups)))
|
||||
user = result.scalar_one()
|
||||
|
||||
if not user.is_admin:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_403_FORBIDDEN,
|
||||
detail="Only admins can enable advanced authentication",
|
||||
)
|
||||
|
||||
# Verify SMTP settings are configured
|
||||
smtp_settings = await get_smtp_settings(db)
|
||||
if not smtp_settings:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail="SMTP settings must be configured before enabling advanced authentication",
|
||||
)
|
||||
|
||||
try:
|
||||
await set_advanced_auth_enabled(db, True)
|
||||
await db.commit()
|
||||
logger.info(f"Advanced authentication enabled by admin user: {user.username}")
|
||||
return {"message": "Advanced authentication enabled successfully", "advanced_auth_enabled": True}
|
||||
except Exception as e:
|
||||
await db.rollback()
|
||||
logger.error(f"Failed to enable advanced authentication: {e}")
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||||
detail=f"Failed to enable advanced authentication: {str(e)}",
|
||||
)
|
||||
|
||||
|
||||
@router.post("/advanced-auth/disable", response_model=dict)
|
||||
async def disable_advanced_auth(
|
||||
current_user: User = Depends(get_current_active_user),
|
||||
db: AsyncSession = Depends(get_db),
|
||||
):
|
||||
"""Disable advanced authentication (admin only)."""
|
||||
import logging
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Reload user with groups for proper is_admin check
|
||||
result = await db.execute(select(User).where(User.id == current_user.id).options(selectinload(User.groups)))
|
||||
user = result.scalar_one()
|
||||
|
||||
if not user.is_admin:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_403_FORBIDDEN,
|
||||
detail="Only admins can disable advanced authentication",
|
||||
)
|
||||
|
||||
try:
|
||||
await set_advanced_auth_enabled(db, False)
|
||||
await db.commit()
|
||||
logger.info(f"Advanced authentication disabled by admin user: {user.username}")
|
||||
return {"message": "Advanced authentication disabled successfully", "advanced_auth_enabled": False}
|
||||
except Exception as e:
|
||||
await db.rollback()
|
||||
logger.error(f"Failed to disable advanced authentication: {e}")
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||||
detail=f"Failed to disable advanced authentication: {str(e)}",
|
||||
)
|
||||
|
||||
|
||||
@router.get("/advanced-auth/status")
|
||||
async def get_advanced_auth_status(db: AsyncSession = Depends(get_db)):
|
||||
"""Get advanced authentication status."""
|
||||
advanced_auth_enabled = await is_advanced_auth_enabled(db)
|
||||
smtp_configured = await get_smtp_settings(db) is not None
|
||||
return {
|
||||
"advanced_auth_enabled": advanced_auth_enabled,
|
||||
"smtp_configured": smtp_configured,
|
||||
}
|
||||
|
||||
|
||||
@router.post("/forgot-password", response_model=ForgotPasswordResponse)
|
||||
async def forgot_password(request: ForgotPasswordRequest, db: AsyncSession = Depends(get_db)):
|
||||
"""Request password reset via email (advanced auth only)."""
|
||||
import logging
|
||||
import os
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Check if advanced auth is enabled
|
||||
advanced_auth = await is_advanced_auth_enabled(db)
|
||||
if not advanced_auth:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail="Advanced authentication is not enabled",
|
||||
)
|
||||
|
||||
# Get SMTP settings
|
||||
smtp_settings = await get_smtp_settings(db)
|
||||
if not smtp_settings:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||||
detail="Email service is not configured",
|
||||
)
|
||||
|
||||
# Find user by email
|
||||
user = await get_user_by_email(db, request.email)
|
||||
|
||||
# Always return success message to prevent email enumeration
|
||||
# but only send email if user exists
|
||||
if user and user.is_active:
|
||||
try:
|
||||
# Generate new password
|
||||
new_password = generate_secure_password()
|
||||
user.password_hash = get_password_hash(new_password)
|
||||
await db.commit()
|
||||
|
||||
# Get login URL from environment or use default
|
||||
login_url = os.environ.get("APP_URL", "http://localhost:5173") + "/login"
|
||||
|
||||
# Send password reset email
|
||||
subject, text_body, html_body = create_password_reset_email(user.username, new_password, login_url)
|
||||
send_email(smtp_settings, user.email, subject, text_body, html_body)
|
||||
|
||||
logger.info(f"Password reset email sent to {user.email}")
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to send password reset email: {e}")
|
||||
# Don't reveal error to user for security
|
||||
|
||||
return ForgotPasswordResponse(
|
||||
message="If the email address is associated with an account, a password reset email has been sent."
|
||||
)
|
||||
|
||||
|
||||
@router.post("/reset-password", response_model=ResetPasswordResponse)
|
||||
async def reset_user_password(
|
||||
request: ResetPasswordRequest,
|
||||
current_user: User = Depends(get_current_active_user),
|
||||
db: AsyncSession = Depends(get_db),
|
||||
):
|
||||
"""Reset a user's password and send them an email (admin only, advanced auth only)."""
|
||||
import logging
|
||||
import os
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Reload user with groups for proper is_admin check
|
||||
result = await db.execute(select(User).where(User.id == current_user.id).options(selectinload(User.groups)))
|
||||
admin_user = result.scalar_one()
|
||||
|
||||
if not admin_user.is_admin:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_403_FORBIDDEN,
|
||||
detail="Only admins can reset user passwords",
|
||||
)
|
||||
|
||||
# Check if advanced auth is enabled
|
||||
advanced_auth = await is_advanced_auth_enabled(db)
|
||||
if not advanced_auth:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail="Advanced authentication is not enabled",
|
||||
)
|
||||
|
||||
# Get SMTP settings
|
||||
smtp_settings = await get_smtp_settings(db)
|
||||
if not smtp_settings:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||||
detail="Email service is not configured",
|
||||
)
|
||||
|
||||
# Find user to reset
|
||||
result = await db.execute(select(User).where(User.id == request.user_id))
|
||||
user = result.scalar_one_or_none()
|
||||
if not user:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail="User not found",
|
||||
)
|
||||
|
||||
if not user.email:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail="User does not have an email address configured",
|
||||
)
|
||||
|
||||
try:
|
||||
# Generate new password
|
||||
new_password = generate_secure_password()
|
||||
user.password_hash = get_password_hash(new_password)
|
||||
await db.commit()
|
||||
|
||||
# Get login URL from environment or use default
|
||||
login_url = os.environ.get("APP_URL", "http://localhost:5173") + "/login"
|
||||
|
||||
# Send password reset email
|
||||
subject, text_body, html_body = create_password_reset_email(user.username, new_password, login_url)
|
||||
send_email(smtp_settings, user.email, subject, text_body, html_body)
|
||||
|
||||
logger.info(f"Password reset by admin {admin_user.username} for user {user.username}")
|
||||
return ResetPasswordResponse(message=f"Password reset email sent to {user.email}")
|
||||
except Exception as e:
|
||||
await db.rollback()
|
||||
logger.error(f"Failed to reset password for user {user.username}: {e}")
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||||
detail=f"Failed to reset password: {str(e)}",
|
||||
)
|
||||
|
||||
@@ -15,8 +15,15 @@ from backend.app.models.archive import PrintArchive
|
||||
from backend.app.models.group import Group
|
||||
from backend.app.models.library import LibraryFile
|
||||
from backend.app.models.print_queue import PrintQueueItem
|
||||
from backend.app.models.settings import Settings
|
||||
from backend.app.models.user import User
|
||||
from backend.app.schemas.auth import ChangePasswordRequest, GroupBrief, UserCreate, UserResponse, UserUpdate
|
||||
from backend.app.services.email_service import (
|
||||
create_welcome_email,
|
||||
generate_secure_password,
|
||||
get_smtp_settings,
|
||||
send_email,
|
||||
)
|
||||
|
||||
router = APIRouter(prefix="/users", tags=["users"])
|
||||
|
||||
@@ -26,6 +33,7 @@ def _user_to_response(user: User) -> UserResponse:
|
||||
return UserResponse(
|
||||
id=user.id,
|
||||
username=user.username,
|
||||
email=user.email,
|
||||
role=user.role,
|
||||
is_active=user.is_active,
|
||||
is_admin=user.is_admin,
|
||||
@@ -54,9 +62,27 @@ async def create_user(
|
||||
_: User | None = RequirePermissionIfAuthEnabled(Permission.USERS_CREATE),
|
||||
db: AsyncSession = Depends(get_db),
|
||||
):
|
||||
"""Create a new user."""
|
||||
# Check if username already exists
|
||||
existing_user = await db.execute(select(User).where(User.username == user_data.username))
|
||||
"""Create a new user.
|
||||
|
||||
When advanced authentication is enabled:
|
||||
- Email is required
|
||||
- Password is auto-generated and emailed to user
|
||||
- Admin cannot set or see the password
|
||||
"""
|
||||
import logging
|
||||
import os
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Check if advanced auth is enabled
|
||||
result = await db.execute(select(Settings).where(Settings.key == "advanced_auth_enabled"))
|
||||
advanced_auth_setting = result.scalar_one_or_none()
|
||||
advanced_auth_enabled = advanced_auth_setting and advanced_auth_setting.value.lower() == "true"
|
||||
|
||||
# Check if username already exists (case-insensitive)
|
||||
existing_user = await db.execute(
|
||||
select(User).where(func.lower(User.username) == func.lower(user_data.username))
|
||||
)
|
||||
if existing_user.scalar_one_or_none():
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
@@ -70,9 +96,38 @@ async def create_user(
|
||||
detail="Role must be 'admin' or 'user'",
|
||||
)
|
||||
|
||||
# Advanced auth validation
|
||||
if advanced_auth_enabled:
|
||||
if not user_data.email:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail="Email is required when advanced authentication is enabled",
|
||||
)
|
||||
# Check if email already exists (case-insensitive)
|
||||
existing_email = await db.execute(
|
||||
select(User).where(func.lower(User.email) == func.lower(user_data.email))
|
||||
)
|
||||
if existing_email.scalar_one_or_none():
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail="Email already exists",
|
||||
)
|
||||
|
||||
# Generate password if advanced auth enabled, otherwise require password
|
||||
if advanced_auth_enabled:
|
||||
password = generate_secure_password()
|
||||
else:
|
||||
if not user_data.password:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail="Password is required when advanced authentication is disabled",
|
||||
)
|
||||
password = user_data.password
|
||||
|
||||
new_user = User(
|
||||
username=user_data.username,
|
||||
password_hash=get_password_hash(user_data.password),
|
||||
email=user_data.email,
|
||||
password_hash=get_password_hash(password),
|
||||
role=user_data.role,
|
||||
is_active=True,
|
||||
)
|
||||
@@ -92,6 +147,21 @@ async def create_user(
|
||||
await db.commit()
|
||||
await db.refresh(new_user)
|
||||
|
||||
# Send welcome email if advanced auth enabled
|
||||
if advanced_auth_enabled and new_user.email:
|
||||
try:
|
||||
smtp_settings = await get_smtp_settings(db)
|
||||
if smtp_settings:
|
||||
login_url = os.environ.get("APP_URL", "http://localhost:5173") + "/login"
|
||||
subject, text_body, html_body = create_welcome_email(new_user.username, password, login_url)
|
||||
send_email(smtp_settings, new_user.email, subject, text_body, html_body)
|
||||
logger.info(f"Welcome email sent to {new_user.email}")
|
||||
else:
|
||||
logger.warning(f"SMTP not configured, could not send welcome email to {new_user.email}")
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to send welcome email: {e}")
|
||||
# Don't fail user creation if email fails
|
||||
|
||||
return _user_to_response(new_user)
|
||||
|
||||
|
||||
@@ -161,8 +231,10 @@ async def update_user(
|
||||
)
|
||||
|
||||
if user_data.username is not None:
|
||||
# Check if new username already exists
|
||||
existing_user = await db.execute(select(User).where(User.username == user_data.username, User.id != user_id))
|
||||
# Check if new username already exists (case-insensitive)
|
||||
existing_user = await db.execute(
|
||||
select(User).where(func.lower(User.username) == func.lower(user_data.username), User.id != user_id)
|
||||
)
|
||||
if existing_user.scalar_one_or_none():
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
@@ -170,6 +242,18 @@ async def update_user(
|
||||
)
|
||||
user.username = user_data.username
|
||||
|
||||
if user_data.email is not None:
|
||||
# Check if new email already exists (case-insensitive)
|
||||
existing_email = await db.execute(
|
||||
select(User).where(func.lower(User.email) == func.lower(user_data.email), User.id != user_id)
|
||||
)
|
||||
if existing_email.scalar_one_or_none():
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail="Email already exists",
|
||||
)
|
||||
user.email = user_data.email
|
||||
|
||||
if user_data.password is not None:
|
||||
user.password_hash = get_password_hash(user_data.password)
|
||||
|
||||
|
||||
@@ -12,7 +12,7 @@ from fastapi import Depends, Header, HTTPException, status
|
||||
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
|
||||
from jwt.exceptions import PyJWTError as JWTError
|
||||
from passlib.context import CryptContext
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy import func, select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
from sqlalchemy.orm import selectinload
|
||||
|
||||
@@ -128,13 +128,26 @@ def create_access_token(data: dict, expires_delta: timedelta | None = None) -> s
|
||||
|
||||
|
||||
async def get_user_by_username(db: AsyncSession, username: str) -> User | None:
|
||||
"""Get a user by username with groups loaded for permission checks."""
|
||||
result = await db.execute(select(User).where(User.username == username).options(selectinload(User.groups)))
|
||||
"""Get a user by username (case-insensitive) with groups loaded for permission checks."""
|
||||
result = await db.execute(
|
||||
select(User).where(func.lower(User.username) == func.lower(username)).options(selectinload(User.groups))
|
||||
)
|
||||
return result.scalar_one_or_none()
|
||||
|
||||
|
||||
async def get_user_by_email(db: AsyncSession, email: str) -> User | None:
|
||||
"""Get a user by email (case-insensitive) with groups loaded for permission checks."""
|
||||
result = await db.execute(
|
||||
select(User).where(func.lower(User.email) == func.lower(email)).options(selectinload(User.groups))
|
||||
)
|
||||
return result.scalar_one_or_none()
|
||||
|
||||
|
||||
async def authenticate_user(db: AsyncSession, username: str, password: str) -> User | None:
|
||||
"""Authenticate a user by username and password."""
|
||||
"""Authenticate a user by username and password.
|
||||
|
||||
Username lookup is case-insensitive. Password is case-sensitive.
|
||||
"""
|
||||
user = await get_user_by_username(db, username)
|
||||
if not user:
|
||||
return None
|
||||
@@ -145,6 +158,21 @@ async def authenticate_user(db: AsyncSession, username: str, password: str) -> U
|
||||
return user
|
||||
|
||||
|
||||
async def authenticate_user_by_email(db: AsyncSession, email: str, password: str) -> User | None:
|
||||
"""Authenticate a user by email and password.
|
||||
|
||||
Email lookup is case-insensitive. Password is case-sensitive.
|
||||
"""
|
||||
user = await get_user_by_email(db, email)
|
||||
if not user:
|
||||
return None
|
||||
if not verify_password(password, user.password_hash):
|
||||
return None
|
||||
if not user.is_active:
|
||||
return None
|
||||
return user
|
||||
|
||||
|
||||
async def is_auth_enabled(db: AsyncSession) -> bool:
|
||||
"""Check if authentication is enabled."""
|
||||
try:
|
||||
|
||||
@@ -24,6 +24,7 @@ class User(Base):
|
||||
|
||||
id: Mapped[int] = mapped_column(primary_key=True)
|
||||
username: Mapped[str] = mapped_column(String(100), unique=True, index=True)
|
||||
email: Mapped[str | None] = mapped_column(String(255), unique=True, index=True, nullable=True)
|
||||
password_hash: Mapped[str] = mapped_column(String(255))
|
||||
role: Mapped[str] = mapped_column(
|
||||
String(20), default="user"
|
||||
|
||||
@@ -24,7 +24,8 @@ class LoginResponse(BaseModel):
|
||||
|
||||
class UserCreate(BaseModel):
|
||||
username: str
|
||||
password: str
|
||||
password: str | None = None # Optional when advanced auth is enabled
|
||||
email: str | None = None
|
||||
role: str = "user"
|
||||
group_ids: list[int] | None = None
|
||||
|
||||
@@ -32,6 +33,7 @@ class UserCreate(BaseModel):
|
||||
class UserUpdate(BaseModel):
|
||||
username: str | None = None
|
||||
password: str | None = None
|
||||
email: str | None = None
|
||||
role: str | None = None
|
||||
is_active: bool | None = None
|
||||
group_ids: list[int] | None = None
|
||||
@@ -40,6 +42,7 @@ class UserUpdate(BaseModel):
|
||||
class UserResponse(BaseModel):
|
||||
id: int
|
||||
username: str
|
||||
email: str | None = None
|
||||
role: str # Deprecated, kept for backward compatibility
|
||||
is_active: bool
|
||||
is_admin: bool # Computed from role and group membership
|
||||
@@ -65,3 +68,44 @@ class SetupRequest(BaseModel):
|
||||
class SetupResponse(BaseModel):
|
||||
auth_enabled: bool
|
||||
admin_created: bool | None = None
|
||||
|
||||
|
||||
class ForgotPasswordRequest(BaseModel):
|
||||
email: str
|
||||
|
||||
|
||||
class ForgotPasswordResponse(BaseModel):
|
||||
message: str
|
||||
|
||||
|
||||
class ResetPasswordRequest(BaseModel):
|
||||
user_id: int
|
||||
|
||||
|
||||
class ResetPasswordResponse(BaseModel):
|
||||
message: str
|
||||
|
||||
|
||||
class SMTPSettings(BaseModel):
|
||||
smtp_host: str
|
||||
smtp_port: int
|
||||
smtp_username: str
|
||||
smtp_password: str | None = None # Optional for read operations
|
||||
smtp_use_tls: bool = True
|
||||
smtp_from_email: str
|
||||
smtp_from_name: str = "BamBuddy"
|
||||
|
||||
|
||||
class TestSMTPRequest(BaseModel):
|
||||
smtp_host: str
|
||||
smtp_port: int
|
||||
smtp_username: str
|
||||
smtp_password: str
|
||||
smtp_use_tls: bool = True
|
||||
smtp_from_email: str
|
||||
test_recipient: str
|
||||
|
||||
|
||||
class TestSMTPResponse(BaseModel):
|
||||
success: bool
|
||||
message: str
|
||||
|
||||
@@ -0,0 +1,321 @@
|
||||
"""Email service for sending authentication-related emails."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import secrets
|
||||
import smtplib
|
||||
import string
|
||||
from email.mime.multipart import MIMEMultipart
|
||||
from email.mime.text import MIMEText
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from backend.app.models.settings import Settings
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from backend.app.schemas.auth import SMTPSettings
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def generate_secure_password(length: int = 16) -> str:
|
||||
"""Generate a secure random password.
|
||||
|
||||
Args:
|
||||
length: Length of the password (default: 16)
|
||||
|
||||
Returns:
|
||||
A secure random password containing uppercase, lowercase, digits, and special characters
|
||||
"""
|
||||
# Define character sets
|
||||
lowercase = string.ascii_lowercase
|
||||
uppercase = string.ascii_uppercase
|
||||
digits = string.digits
|
||||
special = "!@#$%^&*()_+-=[]{}|;:,.<>?"
|
||||
|
||||
# Ensure at least one character from each set
|
||||
password_chars = [
|
||||
secrets.choice(lowercase),
|
||||
secrets.choice(uppercase),
|
||||
secrets.choice(digits),
|
||||
secrets.choice(special),
|
||||
]
|
||||
|
||||
# Fill the rest with random characters from all sets
|
||||
all_chars = lowercase + uppercase + digits + special
|
||||
password_chars.extend(secrets.choice(all_chars) for _ in range(length - 4))
|
||||
|
||||
# Shuffle to avoid predictable patterns
|
||||
secrets.SystemRandom().shuffle(password_chars)
|
||||
|
||||
return "".join(password_chars)
|
||||
|
||||
|
||||
async def get_smtp_settings(db: AsyncSession) -> SMTPSettings | None:
|
||||
"""Get SMTP settings from database.
|
||||
|
||||
Args:
|
||||
db: Database session
|
||||
|
||||
Returns:
|
||||
SMTPSettings object or None if not configured
|
||||
"""
|
||||
from backend.app.schemas.auth import SMTPSettings
|
||||
|
||||
# Fetch all SMTP-related settings
|
||||
result = await db.execute(
|
||||
select(Settings).where(
|
||||
Settings.key.in_([
|
||||
"smtp_host",
|
||||
"smtp_port",
|
||||
"smtp_username",
|
||||
"smtp_password",
|
||||
"smtp_use_tls",
|
||||
"smtp_from_email",
|
||||
"smtp_from_name",
|
||||
])
|
||||
)
|
||||
)
|
||||
settings_dict = {s.key: s.value for s in result.scalars().all()}
|
||||
|
||||
# Check if minimum required settings are present
|
||||
required_keys = ["smtp_host", "smtp_port", "smtp_username", "smtp_from_email"]
|
||||
if not all(key in settings_dict for key in required_keys):
|
||||
return None
|
||||
|
||||
return SMTPSettings(
|
||||
smtp_host=settings_dict["smtp_host"],
|
||||
smtp_port=int(settings_dict["smtp_port"]),
|
||||
smtp_username=settings_dict["smtp_username"],
|
||||
smtp_password=settings_dict.get("smtp_password"),
|
||||
smtp_use_tls=settings_dict.get("smtp_use_tls", "true").lower() == "true",
|
||||
smtp_from_email=settings_dict["smtp_from_email"],
|
||||
smtp_from_name=settings_dict.get("smtp_from_name", "BamBuddy"),
|
||||
)
|
||||
|
||||
|
||||
async def save_smtp_settings(db: AsyncSession, smtp_settings: SMTPSettings) -> None:
|
||||
"""Save SMTP settings to database.
|
||||
|
||||
Args:
|
||||
db: Database session
|
||||
smtp_settings: SMTP settings to save
|
||||
"""
|
||||
from sqlalchemy import func
|
||||
from sqlalchemy.dialects.sqlite import insert as sqlite_insert
|
||||
|
||||
settings_data = {
|
||||
"smtp_host": smtp_settings.smtp_host,
|
||||
"smtp_port": str(smtp_settings.smtp_port),
|
||||
"smtp_username": smtp_settings.smtp_username,
|
||||
"smtp_use_tls": "true" if smtp_settings.smtp_use_tls else "false",
|
||||
"smtp_from_email": smtp_settings.smtp_from_email,
|
||||
"smtp_from_name": smtp_settings.smtp_from_name,
|
||||
}
|
||||
|
||||
# Only save password if provided
|
||||
if smtp_settings.smtp_password:
|
||||
settings_data["smtp_password"] = smtp_settings.smtp_password
|
||||
|
||||
for key, value in settings_data.items():
|
||||
stmt = sqlite_insert(Settings).values(key=key, value=value)
|
||||
stmt = stmt.on_conflict_do_update(
|
||||
index_elements=["key"],
|
||||
set_={"value": value, "updated_at": func.now()},
|
||||
)
|
||||
await db.execute(stmt)
|
||||
|
||||
|
||||
def send_email(
|
||||
smtp_settings: SMTPSettings,
|
||||
to_email: str,
|
||||
subject: str,
|
||||
body_text: str,
|
||||
body_html: str | None = None,
|
||||
) -> None:
|
||||
"""Send an email using SMTP.
|
||||
|
||||
Args:
|
||||
smtp_settings: SMTP configuration
|
||||
to_email: Recipient email address
|
||||
subject: Email subject
|
||||
body_text: Plain text body
|
||||
body_html: Optional HTML body
|
||||
|
||||
Raises:
|
||||
Exception: If email sending fails
|
||||
"""
|
||||
msg = MIMEMultipart("alternative")
|
||||
msg["From"] = f"{smtp_settings.smtp_from_name} <{smtp_settings.smtp_from_email}>"
|
||||
msg["To"] = to_email
|
||||
msg["Subject"] = subject
|
||||
|
||||
# Attach plain text part
|
||||
msg.attach(MIMEText(body_text, "plain"))
|
||||
|
||||
# Attach HTML part if provided
|
||||
if body_html:
|
||||
msg.attach(MIMEText(body_html, "html"))
|
||||
|
||||
# Send email
|
||||
try:
|
||||
if smtp_settings.smtp_use_tls:
|
||||
# Use TLS (port 587 typically)
|
||||
with smtplib.SMTP(smtp_settings.smtp_host, smtp_settings.smtp_port, timeout=10) as server:
|
||||
server.starttls()
|
||||
if smtp_settings.smtp_password:
|
||||
server.login(smtp_settings.smtp_username, smtp_settings.smtp_password)
|
||||
server.send_message(msg)
|
||||
else:
|
||||
# Use SSL (port 465 typically) or no encryption
|
||||
with smtplib.SMTP_SSL(smtp_settings.smtp_host, smtp_settings.smtp_port, timeout=10) as server:
|
||||
if smtp_settings.smtp_password:
|
||||
server.login(smtp_settings.smtp_username, smtp_settings.smtp_password)
|
||||
server.send_message(msg)
|
||||
logger.info(f"Email sent successfully to {to_email}")
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to send email to {to_email}: {e}")
|
||||
raise
|
||||
|
||||
|
||||
def create_welcome_email(username: str, password: str, login_url: str) -> tuple[str, str, str]:
|
||||
"""Create welcome email content for new user.
|
||||
|
||||
Args:
|
||||
username: Username of the new user
|
||||
password: Auto-generated password
|
||||
login_url: URL to login page
|
||||
|
||||
Returns:
|
||||
Tuple of (subject, text_body, html_body)
|
||||
"""
|
||||
subject = "Welcome to BamBuddy - Your Account Details"
|
||||
|
||||
text_body = f"""Welcome to BamBuddy!
|
||||
|
||||
Your account has been created. Here are your login details:
|
||||
|
||||
Username: {username}
|
||||
Password: {password}
|
||||
|
||||
You can login at: {login_url}
|
||||
|
||||
For security reasons, please change your password after your first login.
|
||||
|
||||
Best regards,
|
||||
BamBuddy Team
|
||||
"""
|
||||
|
||||
html_body = f"""<!DOCTYPE html>
|
||||
<html>
|
||||
<head>
|
||||
<meta charset="utf-8">
|
||||
<meta name="viewport" content="width=device-width, initial-scale=1.0">
|
||||
</head>
|
||||
<body style="font-family: Arial, sans-serif; line-height: 1.6; color: #333; max-width: 600px; margin: 0 auto; padding: 20px;">
|
||||
<div style="background: linear-gradient(135deg, #667eea 0%, #764ba2 100%); padding: 20px; border-radius: 8px 8px 0 0;">
|
||||
<h1 style="color: white; margin: 0; font-size: 24px;">Welcome to BamBuddy!</h1>
|
||||
</div>
|
||||
<div style="background: #f9f9f9; padding: 30px; border-radius: 0 0 8px 8px; border: 1px solid #ddd; border-top: none;">
|
||||
<p style="font-size: 16px;">Your account has been created. Here are your login details:</p>
|
||||
|
||||
<div style="background: white; padding: 20px; border-radius: 4px; margin: 20px 0; border-left: 4px solid #667eea;">
|
||||
<p style="margin: 0 0 10px 0;"><strong>Username:</strong> <code style="background: #f0f0f0; padding: 2px 6px; border-radius: 3px;">{username}</code></p>
|
||||
<p style="margin: 0;"><strong>Password:</strong> <code style="background: #f0f0f0; padding: 2px 6px; border-radius: 3px;">{password}</code></p>
|
||||
</div>
|
||||
|
||||
<div style="text-align: center; margin: 30px 0;">
|
||||
<a href="{login_url}" style="display: inline-block; background: #667eea; color: white; padding: 12px 30px; text-decoration: none; border-radius: 4px; font-weight: bold;">Login Now</a>
|
||||
</div>
|
||||
|
||||
<p style="font-size: 14px; color: #666; border-top: 1px solid #ddd; padding-top: 20px; margin-top: 20px;">
|
||||
<strong>Security Note:</strong> For security reasons, please change your password after your first login.
|
||||
</p>
|
||||
|
||||
<p style="font-size: 14px; color: #999; margin-top: 30px;">
|
||||
Best regards,<br>
|
||||
BamBuddy Team
|
||||
</p>
|
||||
</div>
|
||||
</body>
|
||||
</html>
|
||||
"""
|
||||
|
||||
return subject, text_body, html_body
|
||||
|
||||
|
||||
def create_password_reset_email(username: str, password: str, login_url: str) -> tuple[str, str, str]:
|
||||
"""Create password reset email content.
|
||||
|
||||
Args:
|
||||
username: Username of the user
|
||||
password: New auto-generated password
|
||||
login_url: URL to login page
|
||||
|
||||
Returns:
|
||||
Tuple of (subject, text_body, html_body)
|
||||
"""
|
||||
subject = "BamBuddy - Your Password Has Been Reset"
|
||||
|
||||
text_body = f"""Your BamBuddy password has been reset.
|
||||
|
||||
Your login details:
|
||||
|
||||
Username: {username}
|
||||
New Password: {password}
|
||||
|
||||
You can login at: {login_url}
|
||||
|
||||
For security reasons, please change your password after logging in.
|
||||
|
||||
If you did not request this password reset, please contact your administrator immediately.
|
||||
|
||||
Best regards,
|
||||
BamBuddy Team
|
||||
"""
|
||||
|
||||
html_body = f"""<!DOCTYPE html>
|
||||
<html>
|
||||
<head>
|
||||
<meta charset="utf-8">
|
||||
<meta name="viewport" content="width=device-width, initial-scale=1.0">
|
||||
</head>
|
||||
<body style="font-family: Arial, sans-serif; line-height: 1.6; color: #333; max-width: 600px; margin: 0 auto; padding: 20px;">
|
||||
<div style="background: linear-gradient(135deg, #667eea 0%, #764ba2 100%); padding: 20px; border-radius: 8px 8px 0 0;">
|
||||
<h1 style="color: white; margin: 0; font-size: 24px;">Password Reset</h1>
|
||||
</div>
|
||||
<div style="background: #f9f9f9; padding: 30px; border-radius: 0 0 8px 8px; border: 1px solid #ddd; border-top: none;">
|
||||
<p style="font-size: 16px;">Your BamBuddy password has been reset.</p>
|
||||
|
||||
<div style="background: white; padding: 20px; border-radius: 4px; margin: 20px 0; border-left: 4px solid #667eea;">
|
||||
<p style="margin: 0 0 10px 0;"><strong>Username:</strong> <code style="background: #f0f0f0; padding: 2px 6px; border-radius: 3px;">{username}</code></p>
|
||||
<p style="margin: 0;"><strong>New Password:</strong> <code style="background: #f0f0f0; padding: 2px 6px; border-radius: 3px;">{password}</code></p>
|
||||
</div>
|
||||
|
||||
<div style="text-align: center; margin: 30px 0;">
|
||||
<a href="{login_url}" style="display: inline-block; background: #667eea; color: white; padding: 12px 30px; text-decoration: none; border-radius: 4px; font-weight: bold;">Login Now</a>
|
||||
</div>
|
||||
|
||||
<div style="background: #fff3cd; border: 1px solid #ffc107; border-radius: 4px; padding: 15px; margin: 20px 0;">
|
||||
<p style="margin: 0; font-size: 14px; color: #856404;">
|
||||
<strong>⚠️ Security Alert:</strong> If you did not request this password reset, please contact your administrator immediately.
|
||||
</p>
|
||||
</div>
|
||||
|
||||
<p style="font-size: 14px; color: #666; border-top: 1px solid #ddd; padding-top: 20px; margin-top: 20px;">
|
||||
<strong>Security Note:</strong> For security reasons, please change your password after logging in.
|
||||
</p>
|
||||
|
||||
<p style="font-size: 14px; color: #999; margin-top: 30px;">
|
||||
Best regards,<br>
|
||||
BamBuddy Team
|
||||
</p>
|
||||
</div>
|
||||
</body>
|
||||
</html>
|
||||
"""
|
||||
|
||||
return subject, text_body, html_body
|
||||
Reference in New Issue
Block a user