diff --git a/app/api/local_auth.py b/app/api/local_auth.py index 162c07d1..8316f7d6 100644 --- a/app/api/local_auth.py +++ b/app/api/local_auth.py @@ -86,7 +86,9 @@ class PasswordResetBody(BaseModel): @router.get("/signup", include_in_schema=False) async def signup_page(request: Request) -> Any: - """Render the signup page, or redirect to login when signup is disabled.""" + """Render the signup page, or redirect to login when multi-user / signup is disabled.""" + if not settings.multi_user_enabled: + return RedirectResponse(url="/login?error=Multi-user+mode+is+not+enabled", status_code=302) if not settings.allow_local_signup: return RedirectResponse(url="/login?error=Registration+is+not+enabled", status_code=302) return templates.TemplateResponse( @@ -130,13 +132,16 @@ async def signup(request: Request, body: SignupBody, db: DbSession) -> dict[str, """Create a new local user account and send a verification email. The account is inactive until the user clicks the email link. + Both ``MULTI_USER_ENABLED`` and ``ALLOW_LOCAL_SIGNUP`` must be ``True``. Raises: - 403: Local signup is disabled. + 403: Multi-user mode or local signup is disabled. 503: SMTP is not configured. 422: Passwords do not match. 409: Email or username already registered. """ + if not settings.multi_user_enabled: + raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Multi-user mode is not enabled.") if not settings.allow_local_signup: raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Registration is not enabled.") if not settings.email_host: @@ -170,8 +175,12 @@ async def signup(request: Request, body: SignupBody, db: DbSession) -> dict[str, ) db.add(profile) + # Flush to the DB so constraint violations (duplicate key etc.) surface NOW, + # before we attempt to send the email. We do NOT commit yet — the commit only + # happens after the email is sent successfully so that a failed email leaves + # no orphan records in the database. try: - db.commit() + db.flush() except Exception: db.rollback() raise @@ -180,20 +189,16 @@ async def signup(request: Request, body: SignupBody, db: DbSession) -> dict[str, try: send_verification_email(body.email, body.username, token, base_url) except Exception as exc: - # Clean up orphan records — don't leave an unverifiable account - try: - db.delete(user) - db.delete(profile) - db.commit() - except Exception: - db.rollback() - logger.exception("Failed to clean up orphan records for %s after email send failure", body.email) + # Email failed — roll back so no unverifiable user row persists. + # The user can simply try registering again once SMTP is fixed. + db.rollback() logger.warning("Signup email failed for %s: %s", body.email, exc) raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail=("Failed to send verification email. Please check that SMTP is correctly configured and try again."), ) from exc + db.commit() logger.info("New local user registered: %s", body.email) return {"message": "Verification email sent. Please check your inbox."} diff --git a/app/auth.py b/app/auth.py index 5ecbe584..699ddbbc 100644 --- a/app/auth.py +++ b/app/auth.py @@ -13,10 +13,18 @@ from starlette.responses import RedirectResponse from app.config import settings from app.database import get_db -oauth = OAuth() +# Conditional imports: only used when multi_user_enabled=True. Imported here at +# module level (not inside auth()) so they don't incur repeated import overhead. +# Guards at call-sites ensure they are never *called* in single-user mode. +from app.models import LocalUser as _LocalUser +from app.models import UserProfile as _UserProfile +from app.utils.local_auth import build_session_user as _build_session_user +from app.utils.local_auth import verify_password as _verify_password logger = logging.getLogger(__name__) +oauth = OAuth() + AUTH_ENABLED = settings.auth_enabled # Set up templates for authentication @@ -83,7 +91,8 @@ async def login(request: Request): "oauth_provider_name": OAUTH_PROVIDER_NAME, "app_version": settings.version, "csrf_token": getattr(request.state, "csrf_token", ""), - "allow_signup": settings.allow_local_signup, + # "Create account" link is only shown when multi-user mode AND local signup are both enabled + "allow_signup": settings.multi_user_enabled and settings.allow_local_signup, }, ) @@ -173,8 +182,6 @@ async def oauth_callback(request: Request, db: Session = Depends(get_db)): user_data.get("sub") or user_data.get("preferred_username") or user_data.get("email") or user_data.get("id") ) if user_id: - from app.models import UserProfile as _UserProfile - profile = db.query(_UserProfile).filter(_UserProfile.user_id == user_id).first() if profile and not profile.onboarding_completed: post_onboarding = request.session.pop("redirect_after_login", "/upload") @@ -192,41 +199,46 @@ async def oauth_callback(request: Request, db: Session = Depends(get_db)): async def auth(request: Request, db: Session = Depends(get_db)): """Handle local username/password authentication. - Checks LocalUser accounts first, then falls back to admin credentials. + In multi-user mode (``MULTI_USER_ENABLED=True``) local registered users are + checked first; if no matching LocalUser is found the request falls through to + the single admin-credential check so that single-user deployments continue to + work without any database involvement. + + In single-user mode (``MULTI_USER_ENABLED=False``, the default) the LocalUser + table is never queried — only the configured ADMIN_USERNAME / ADMIN_PASSWORD + are accepted, preserving full backward compatibility. """ form_data = await request.form() username = form_data.get("username") password = form_data.get("password") - # --- LocalUser check --- - from app.models import LocalUser as _LocalUser - from app.models import UserProfile as _UserProfile - from app.utils.local_auth import build_session_user as _build_session_user - from app.utils.local_auth import verify_password as _verify_password + # --- LocalUser check (multi-user mode only) --- + if settings.multi_user_enabled: + local_user = ( + db.query(_LocalUser).filter((_LocalUser.username == username) | (_LocalUser.email == username)).first() + ) + if local_user is not None: + if not _verify_password(password or "", local_user.hashed_password): + logger.warning("[SECURITY] LOCAL_LOGIN_FAILURE user=%s", username) + return RedirectResponse(url="/login?error=Invalid+username+or+password", status_code=302) + if not local_user.is_active: + logger.warning("[SECURITY] LOCAL_LOGIN_UNVERIFIED user=%s", username) + return RedirectResponse( + url="/login?error=Please+verify+your+email+address+before+logging+in", + status_code=302, + ) + user_data = _build_session_user(local_user) + request.session["user"] = user_data + logger.info("[SECURITY] LOCAL_LOGIN_SUCCESS user=%s", local_user.email) + profile = db.query(_UserProfile).filter(_UserProfile.user_id == local_user.email).first() + if profile and not profile.onboarding_completed: + post_onboarding = request.session.pop("redirect_after_login", "/upload") + request.session["post_onboarding_redirect"] = post_onboarding + return RedirectResponse(url="/onboarding", status_code=302) + redirect_url = request.session.pop("redirect_after_login", "/upload") + return RedirectResponse(url=redirect_url, status_code=302) - local_user = db.query(_LocalUser).filter((_LocalUser.username == username) | (_LocalUser.email == username)).first() - if local_user is not None: - if not _verify_password(password or "", local_user.hashed_password): - logger.warning("[SECURITY] LOCAL_LOGIN_FAILURE user=%s", username) - return RedirectResponse(url="/login?error=Invalid+username+or+password", status_code=302) - if not local_user.is_active: - logger.warning("[SECURITY] LOCAL_LOGIN_UNVERIFIED user=%s", username) - return RedirectResponse( - url="/login?error=Please+verify+your+email+address+before+logging+in", - status_code=302, - ) - user_data = _build_session_user(local_user) - request.session["user"] = user_data - logger.info("[SECURITY] LOCAL_LOGIN_SUCCESS user=%s", local_user.email) - profile = db.query(_UserProfile).filter(_UserProfile.user_id == local_user.email).first() - if profile and not profile.onboarding_completed: - post_onboarding = request.session.pop("redirect_after_login", "/upload") - request.session["post_onboarding_redirect"] = post_onboarding - return RedirectResponse(url="/onboarding", status_code=302) - redirect_url = request.session.pop("redirect_after_login", "/upload") - return RedirectResponse(url=redirect_url, status_code=302) - - # --- Admin credentials fallback --- + # --- Admin credentials (always available as a fallback / single-user mode) --- if username == settings.admin_username and password == settings.admin_password: request.session["user"] = { "id": "admin", diff --git a/app/config.py b/app/config.py index c59c31c4..848846a6 100644 --- a/app/config.py +++ b/app/config.py @@ -177,8 +177,9 @@ class Settings(BaseSettings): default=False, description=( "Allow users to self-register with email and password. " - "Requires email (SMTP) to be configured for verification emails. " - "Default: False (registration disabled, admin creates users)." + "Has no effect unless MULTI_USER_ENABLED is also True. " + "Requires SMTP to be configured so verification emails can be sent. " + "Default: False (registration disabled — admin creates users manually)." ), ) diff --git a/tests/test_local_auth.py b/tests/test_local_auth.py index 502e445e..bed05ea0 100644 --- a/tests/test_local_auth.py +++ b/tests/test_local_auth.py @@ -22,6 +22,7 @@ from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker from sqlalchemy.pool import StaticPool +from app.config import settings from app.database import Base, get_db from app.models import LocalUser, UserProfile from app.utils.local_auth import ( @@ -210,6 +211,7 @@ def test_signup_smtp_not_configured(la_client): """POST /api/auth/signup returns 503 when SMTP is not configured.""" with patch("app.api.local_auth.settings") as mock_settings: mock_settings.allow_local_signup = True + mock_settings.multi_user_enabled = True mock_settings.email_host = None resp = la_client.post( "/api/auth/signup", @@ -228,6 +230,7 @@ def test_signup_password_mismatch(la_client): """POST /api/auth/signup returns 422 when passwords do not match.""" with patch("app.api.local_auth.settings") as mock_settings: mock_settings.allow_local_signup = True + mock_settings.multi_user_enabled = True mock_settings.email_host = "smtp.example.com" resp = la_client.post( "/api/auth/signup", @@ -249,6 +252,7 @@ def test_signup_success(la_client): patch("app.api.local_auth.send_verification_email") as mock_send, ): mock_settings.allow_local_signup = True + mock_settings.multi_user_enabled = True mock_settings.email_host = "smtp.example.com" mock_settings.version = "test" resp = la_client.post( @@ -273,6 +277,7 @@ def test_signup_duplicate_email(la_client, active_user): patch("app.api.local_auth.send_verification_email"), ): mock_settings.allow_local_signup = True + mock_settings.multi_user_enabled = True mock_settings.email_host = "smtp.example.com" resp = la_client.post( "/api/auth/signup", @@ -295,6 +300,7 @@ def test_signup_duplicate_username(la_client, active_user): patch("app.api.local_auth.send_verification_email"), ): mock_settings.allow_local_signup = True + mock_settings.multi_user_enabled = True mock_settings.email_host = "smtp.example.com" resp = la_client.post( "/api/auth/signup", @@ -317,6 +323,7 @@ def test_signup_smtp_failure_cleans_up(la_client, la_session): patch("app.api.local_auth.send_verification_email", side_effect=RuntimeError("SMTP down")), ): mock_settings.allow_local_signup = True + mock_settings.multi_user_enabled = True mock_settings.email_host = "smtp.example.com" resp = la_client.post( "/api/auth/signup", @@ -521,6 +528,7 @@ def test_signup_page_enabled(la_client): """GET /signup returns 200 when allow_local_signup is True.""" with patch("app.api.local_auth.settings") as mock_settings: mock_settings.allow_local_signup = True + mock_settings.multi_user_enabled = True mock_settings.version = "test" resp = la_client.get("/signup") assert resp.status_code == 200 @@ -551,6 +559,7 @@ def test_reset_password_page(la_client): @pytest.mark.unit @pytest.mark.asyncio +@patch.object(settings, "multi_user_enabled", True) async def test_local_login_success(la_session, active_user): """auth() with valid LocalUser credentials sets session and redirects.""" from unittest.mock import AsyncMock, MagicMock @@ -571,6 +580,7 @@ async def test_local_login_success(la_session, active_user): @pytest.mark.unit @pytest.mark.asyncio +@patch.object(settings, "multi_user_enabled", True) async def test_local_login_by_email(la_session, active_user): """auth() accepts email as username for LocalUser lookup.""" from unittest.mock import AsyncMock, MagicMock @@ -590,6 +600,7 @@ async def test_local_login_by_email(la_session, active_user): @pytest.mark.unit @pytest.mark.asyncio +@patch.object(settings, "multi_user_enabled", True) async def test_local_login_wrong_password(la_session, active_user): """auth() with wrong password redirects to login with error.""" from unittest.mock import AsyncMock, MagicMock @@ -610,6 +621,7 @@ async def test_local_login_wrong_password(la_session, active_user): @pytest.mark.unit @pytest.mark.asyncio +@patch.object(settings, "multi_user_enabled", True) async def test_local_login_unverified(la_session, pending_user): """auth() for unverified user redirects with verification message.""" from unittest.mock import AsyncMock, MagicMock @@ -625,3 +637,41 @@ async def test_local_login_unverified(la_session, pending_user): result = await auth(mock_request, db=la_session) assert result.status_code == 302 assert "verify" in result.headers["location"].lower() + + +# --------------------------------------------------------------------------- +# Single-user backward-compatibility: LocalUser table must NOT be queried +# --------------------------------------------------------------------------- + + +@pytest.mark.unit +@pytest.mark.asyncio +@patch.object(settings, "multi_user_enabled", False) +async def test_single_user_mode_skips_local_user_table(la_session, active_user): + """In single-user mode auth() must not query LocalUser even when a matching + row exists. It should fall through to the admin-credential check.""" + from unittest.mock import AsyncMock, MagicMock + from unittest.mock import patch as _patch + + from fastapi import Request + + from app.auth import auth + + mock_request = MagicMock(spec=Request) + # Use the active LocalUser's credentials — they must NOT work in single-user mode + # because the whole LocalUser block is skipped. + mock_request.form = AsyncMock(return_value={"username": "activeuser", "password": "password123"}) + mock_request.session = {} + + with ( + _patch.object(settings, "admin_username", "activeuser"), + _patch.object(settings, "admin_password", "password123"), + ): + result = await auth(mock_request, db=la_session) + + # Should succeed via admin-credentials path (is_admin=True), not LocalUser path + assert result.status_code == 302 + assert "user" in mock_request.session + # Admin path sets is_admin=True and id="admin" + assert mock_request.session["user"]["is_admin"] is True + assert mock_request.session["user"]["id"] == "admin"