Merge pull request #542 from christianlouis/copilot/debug-local-login-issues
feat(auth): add diagnostic debug logging to local login flow
This commit is contained in:
+55
-4
@@ -312,20 +312,51 @@ async def auth(request: Request, db: Session = Depends(get_db)):
|
||||
username = form_data.get("username")
|
||||
password = form_data.get("password")
|
||||
|
||||
logger.debug(
|
||||
"[AUTH] Login attempt: username=%r password_provided=%s multi_user_enabled=%s",
|
||||
username,
|
||||
bool(password),
|
||||
settings.multi_user_enabled,
|
||||
)
|
||||
|
||||
# --- LocalUser check (multi-user mode only) ---
|
||||
if settings.multi_user_enabled:
|
||||
if not username:
|
||||
logger.warning(
|
||||
"[AUTH] LOGIN_FAILURE reason=empty_username multi_user_enabled=%s form_keys=%s content_type=%s",
|
||||
settings.multi_user_enabled,
|
||||
list(form_data.keys()),
|
||||
request.headers.get("content-type", "<missing>"),
|
||||
)
|
||||
return RedirectResponse(url="/login?error=Invalid+username+or+password", status_code=302)
|
||||
|
||||
local_user = (
|
||||
db.query(_LocalUser).filter((_LocalUser.username == username) | (_LocalUser.email == username)).first()
|
||||
)
|
||||
logger.debug(
|
||||
"[AUTH] LocalUser lookup: username=%r found=%s",
|
||||
username,
|
||||
local_user is not None,
|
||||
)
|
||||
if local_user is not None:
|
||||
if not local_user.is_active:
|
||||
logger.warning("[SECURITY] LOCAL_LOGIN_UNVERIFIED user=%s", username)
|
||||
logger.warning(
|
||||
"[SECURITY] LOCAL_LOGIN_UNVERIFIED user=%s is_active=%s",
|
||||
username,
|
||||
local_user.is_active,
|
||||
)
|
||||
return RedirectResponse(
|
||||
url="/login?error=Please+verify+your+email+address+before+logging+in",
|
||||
status_code=302,
|
||||
)
|
||||
if not _verify_password(password or "", local_user.hashed_password):
|
||||
logger.warning("[SECURITY] LOCAL_LOGIN_FAILURE user=%s", username)
|
||||
pw_ok = _verify_password(password or "", local_user.hashed_password)
|
||||
logger.debug(
|
||||
"[AUTH] Password verification: user=%s ok=%s",
|
||||
username,
|
||||
pw_ok,
|
||||
)
|
||||
if not pw_ok:
|
||||
logger.warning("[SECURITY] LOCAL_LOGIN_FAILURE reason=wrong_password user=%s", username)
|
||||
return RedirectResponse(url="/login?error=Invalid+username+or+password", status_code=302)
|
||||
user_data = _build_session_user(local_user)
|
||||
request.session["user"] = user_data
|
||||
@@ -339,12 +370,25 @@ async def auth(request: Request, db: Session = Depends(get_db)):
|
||||
redirect_url = request.session.pop("redirect_after_login", "/upload")
|
||||
return RedirectResponse(url=redirect_url, status_code=302)
|
||||
|
||||
# Local user not found; fall through to admin credential check below.
|
||||
logger.debug(
|
||||
"[AUTH] No LocalUser matched username=%r; falling through to admin credential check",
|
||||
username,
|
||||
)
|
||||
|
||||
# --- Admin credentials (always available as a fallback / single-user mode) ---
|
||||
# Guard: only attempt the match when credentials are actually configured.
|
||||
# Without this guard, Python's `None == None` would be True when neither
|
||||
# ADMIN_USERNAME nor ADMIN_PASSWORD is set, allowing any request that omits
|
||||
# those form fields to be authenticated as an admin — creating a phantom
|
||||
# "None@local.docuelevate" admin profile with full privileges.
|
||||
admin_configured = bool(settings.admin_username and settings.admin_password)
|
||||
logger.debug(
|
||||
"[AUTH] Admin credential check: admin_configured=%s username_match=%s multi_user_enabled=%s",
|
||||
admin_configured,
|
||||
username == settings.admin_username if admin_configured else False,
|
||||
settings.multi_user_enabled,
|
||||
)
|
||||
if (
|
||||
settings.admin_username
|
||||
and settings.admin_password
|
||||
@@ -365,7 +409,14 @@ async def auth(request: Request, db: Session = Depends(get_db)):
|
||||
redirect_url = request.session.pop("redirect_after_login", "/upload")
|
||||
return RedirectResponse(url=redirect_url, status_code=302)
|
||||
else:
|
||||
logger.warning("[SECURITY] LOCAL_LOGIN_FAILURE user=%s", username)
|
||||
logger.warning(
|
||||
"[SECURITY] LOCAL_LOGIN_FAILURE reason=no_match user=%r "
|
||||
"multi_user_enabled=%s admin_configured=%s form_empty=%s",
|
||||
username,
|
||||
settings.multi_user_enabled,
|
||||
admin_configured,
|
||||
not username and not password,
|
||||
)
|
||||
return RedirectResponse(url="/login?error=Invalid+username+or+password", status_code=302)
|
||||
|
||||
|
||||
|
||||
@@ -148,10 +148,16 @@ class CSRFMiddleware(BaseHTTPMiddleware):
|
||||
|
||||
# 2. For URL-encoded form bodies only (plain HTML form submissions).
|
||||
content_type = request.headers.get("content-type", "")
|
||||
logger.debug("CSRF: content_type=%r method=%s path=%s", content_type, request.method, request.url.path)
|
||||
if "application/x-www-form-urlencoded" in content_type:
|
||||
try:
|
||||
form = await request.form()
|
||||
token = form.get("csrf_token")
|
||||
logger.debug(
|
||||
"CSRF: form_keys=%s csrf_token_present=%s",
|
||||
list(form.keys()),
|
||||
bool(token),
|
||||
)
|
||||
if token:
|
||||
return str(token)
|
||||
except Exception as exc:
|
||||
|
||||
@@ -32,8 +32,12 @@ def hash_password(plain: str) -> str:
|
||||
def verify_password(plain: str, hashed: str) -> bool:
|
||||
"""Return True when *plain* matches the stored bcrypt *hashed* string."""
|
||||
try:
|
||||
return bcrypt.checkpw(plain.encode("utf-8"), hashed.encode("utf-8"))
|
||||
except Exception:
|
||||
result = bcrypt.checkpw(plain.encode("utf-8"), hashed.encode("utf-8"))
|
||||
if not result:
|
||||
logger.debug("verify_password: mismatch password_provided=%s", bool(plain))
|
||||
return result
|
||||
except Exception as exc:
|
||||
logger.warning("verify_password: exception type=%s msg=%s", type(exc).__name__, exc)
|
||||
return False
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user