gitea-admin/gh-christianlouis-docuelevate
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: gate LocalUser machinery on multi_user_enabled for single-user backward compat

Browse Source 
- auth() only queries LocalUser table when multi_user_enabled=True
- login() only shows signup link when multi_user_enabled AND allow_local_signup
- signup page and POST endpoint both check multi_user_enabled first
- Move local-auth imports to module level in auth.py (no re-import overhead)
- Fix signup rollback: flush before email send, commit only on success
- Update allow_local_signup config description to document prerequisite
- Add test: single-user mode skips LocalUser table entirely
- Patch multi_user_enabled=True on all local-login integration tests

Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
This commit is contained in:
copilot-swe-agent[bot]
2026-03-07 13:35:10 +00:00
parent 43f3f6bdbe
commit b5b285ebe6
4 changed files with 114 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/api/local_auth.py
+16 -11
View File
@@ -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."}
app/auth.py
+45 -33
View File
@@ -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",
app/config.py
+3 -2
View File
@@ -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)."
),
)
tests/test_local_auth.py
+50
View File
@@ -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"