A R R R Associates
1037 lines
34 KiB
Python
1037 lines
34 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import datetime, timezone, timedelta
|
|
|
|
from fastapi import APIRouter, Form, Request
|
|
from fastapi.responses import RedirectResponse
|
|
from sqlalchemy import select
|
|
|
|
from app.core.db.common import CommonSessionLocal
|
|
from app.core.security.csrf import get_or_create_csrf_token, validate_csrf
|
|
from app.core.security.otp import start_otp, verify_otp
|
|
from app.core.security.passwords import verify_password, hash_password
|
|
from app.core.security.session_auth import (
|
|
SESSION_LOGIN_AT_KEY,
|
|
SESSION_USER_ID_KEY,
|
|
get_current_user,
|
|
)
|
|
from app.core.templating import templates
|
|
from app.core.settings import get_settings
|
|
from app.modules.core.iam.invite_service import accept_invite, reset_password_with_token
|
|
from app.modules.core.iam.models import LoginAttempt, User
|
|
from app.modules.client_identity.models import ClientPortalIdentity
|
|
from app.modules.client_identity.service import mark_identity_activated, resolve_login_user
|
|
from app.modules.core.rbac.models import Permission, Role, RolePermission, UserRole
|
|
from app.modules.core.tenancy.models import Branch, FinancialYear, Tenant
|
|
from app.modules.core.tenancy.settings_models import BranchSettings
|
|
from app.modules.email_integration.services import send_auth_otp_email, send_password_changed_email
|
|
|
|
router = APIRouter()
|
|
|
|
PENDING_POST_LOGIN_REDIRECT_KEY = "pending_post_login_redirect"
|
|
LOGIN_ENTRY_PATH_KEY = "login_entry_path"
|
|
PROFESSIONAL_LOGIN_PATH = "/login"
|
|
CLIENT_LOGIN_PATH = "/client/login"
|
|
PROFESSIONAL_ROLE_NAMES = {
|
|
"System Admin",
|
|
"Firm Admin",
|
|
"Partner",
|
|
"Manager",
|
|
"Branch Manager",
|
|
"Staff",
|
|
"Consultant",
|
|
}
|
|
SAFE_POST_LOGIN_REDIRECTS = {
|
|
"/mobile/attendance",
|
|
"/employee/attendance",
|
|
"/employee/dashboard",
|
|
}
|
|
|
|
# Verified active audit-firm domains are tenant authentication boundaries.
|
|
# Marketplace and consultant domain behaviour remains unchanged.
|
|
TENANT_BOUND_DOMAIN_TYPES = {"audit_firm_domain", "audit_firm_subdomain"}
|
|
|
|
|
|
def _bound_domain_tenant_id(request: Request) -> int | None:
|
|
"""Return the trusted tenant id bound to the current audit-firm domain.
|
|
|
|
DomainResolverMiddleware only marks exact, active and verified mappings as
|
|
resolved. The additional checks here make the authentication boundary
|
|
explicit and safe if the middleware evolves later.
|
|
"""
|
|
if not bool(getattr(request.state, "domain_resolved", False)):
|
|
return None
|
|
if not bool(getattr(request.state, "domain_is_verified", False)):
|
|
return None
|
|
if (getattr(request.state, "domain_status", None) or "").strip().lower() != "active":
|
|
return None
|
|
if (getattr(request.state, "domain_type", None) or "").strip() not in TENANT_BOUND_DOMAIN_TYPES:
|
|
return None
|
|
tenant_id = getattr(request.state, "domain_tenant_id", None)
|
|
try:
|
|
return int(tenant_id) if tenant_id not in (None, "", 0, "0") else None
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
|
|
def _clear_login_session(request: Request) -> None:
|
|
"""Remove authentication/context state without disturbing CSRF/session middleware."""
|
|
for key in (
|
|
SESSION_USER_ID_KEY,
|
|
SESSION_LOGIN_AT_KEY,
|
|
"user_email",
|
|
"tenant_id",
|
|
"branch_id",
|
|
"tenant_code",
|
|
"branch_code",
|
|
"active_tenant_id",
|
|
"active_branch_id",
|
|
"active_tenant_code",
|
|
"active_branch_code",
|
|
"active_financial_year",
|
|
"must_change_password",
|
|
"post_login_redirect",
|
|
"otp_verified",
|
|
LOGIN_ENTRY_PATH_KEY,
|
|
):
|
|
request.session.pop(key, None)
|
|
|
|
|
|
def _consume_safe_post_login_redirect(request: Request) -> str | None:
|
|
value = request.session.pop(PENDING_POST_LOGIN_REDIRECT_KEY, None)
|
|
value = (value or "").strip()
|
|
if value in SAFE_POST_LOGIN_REDIRECTS:
|
|
return value
|
|
return None
|
|
|
|
|
|
def _dev_otp_print_enabled() -> bool:
|
|
settings = get_settings()
|
|
return bool(getattr(settings, "DEV_AUTH_OTP_PRINT", False)) and (settings.ENV or "").lower() in {"dev", "local", "development"}
|
|
|
|
|
|
def _log_dev_otp(label: str, email: str, code: str) -> None:
|
|
if _dev_otp_print_enabled():
|
|
print(f"[DEV OTP] {label} user={email} code={code}")
|
|
|
|
|
|
def _client_ip(request: Request) -> str:
|
|
return request.client.host if request.client else "unknown"
|
|
|
|
|
|
def _attempt_key(email: str, ip: str) -> str:
|
|
return f"{email.lower().strip()}|{ip}"
|
|
|
|
|
|
def _get_branch_security_policy(db, user: User) -> BranchSettings | None:
|
|
if not getattr(user, "branch_id", None):
|
|
return None
|
|
return db.execute(
|
|
select(BranchSettings).where(BranchSettings.branch_id == user.branch_id)
|
|
).scalar_one_or_none()
|
|
|
|
|
|
def _user_roles(db, user_id: int) -> list[str]:
|
|
q = (
|
|
select(Role.name)
|
|
.join(UserRole, UserRole.role_id == Role.id)
|
|
.where(UserRole.user_id == user_id)
|
|
)
|
|
return [r for (r,) in db.execute(q).all()]
|
|
|
|
|
|
def _user_permissions(db, user_id: int) -> set[str]:
|
|
q = (
|
|
select(Permission.code)
|
|
.join(RolePermission, RolePermission.permission_id == Permission.id)
|
|
.join(UserRole, UserRole.role_id == RolePermission.role_id)
|
|
.where(UserRole.user_id == user_id, Permission.is_active.is_(True))
|
|
)
|
|
return set(db.execute(q).scalars().all())
|
|
|
|
|
|
def _otp_required(bs: BranchSettings | None, roles: list[str]) -> bool:
|
|
if not bs:
|
|
return False
|
|
required = {
|
|
x.strip() for x in (bs.otp_required_roles_csv or "").split(",") if x.strip()
|
|
}
|
|
return any(r in required for r in roles)
|
|
|
|
def _default_financial_year_code(db, tenant_id: int | None) -> str | None:
|
|
if not tenant_id:
|
|
return None
|
|
fy = db.execute(
|
|
select(FinancialYear).where(
|
|
FinancialYear.tenant_id == tenant_id,
|
|
FinancialYear.is_current.is_(True),
|
|
)
|
|
).scalar_one_or_none()
|
|
if fy:
|
|
return fy.year_code
|
|
fy = db.execute(
|
|
select(FinancialYear)
|
|
.where(FinancialYear.tenant_id == tenant_id)
|
|
.order_by(FinancialYear.start_date.desc(), FinancialYear.year_code.desc())
|
|
).scalars().first()
|
|
return fy.year_code if fy else None
|
|
|
|
|
|
def _tenant_code(db, tenant_id: int | None) -> str | None:
|
|
if not tenant_id:
|
|
return None
|
|
tenant = db.get(Tenant, int(tenant_id))
|
|
return tenant.code if tenant else None
|
|
|
|
|
|
def _branch_code(db, branch_id: int | None) -> str | None:
|
|
if not branch_id:
|
|
return None
|
|
branch = db.get(Branch, int(branch_id))
|
|
return branch.code if branch else None
|
|
|
|
|
|
def _post_login_redirect(must_change_password: bool, permissions: set[str], roles: list[str]) -> str:
|
|
if must_change_password:
|
|
return "/change-password-required"
|
|
|
|
role_set = set(roles or [])
|
|
|
|
# Platform owner always lands on platform control centre first.
|
|
if "System Admin" in role_set:
|
|
return "/system-admin/dashboard"
|
|
|
|
if "Client" in role_set:
|
|
return "/client/dashboard"
|
|
|
|
if "Consultant" in role_set:
|
|
return "/consultant/dashboard"
|
|
|
|
# If Firm Admin is also Partner, daily operations are more frequent;
|
|
# the workspace switcher exposes Firm Administration when required.
|
|
if "Firm Admin" in role_set and "Partner" in role_set:
|
|
return "/partner/dashboard"
|
|
|
|
if "Firm Admin" in role_set:
|
|
return "/firm-admin/dashboard"
|
|
|
|
if "Partner" in role_set:
|
|
return "/partner/dashboard"
|
|
|
|
if role_set.intersection({"Manager", "Branch Manager"}):
|
|
return "/manager/dashboard"
|
|
|
|
if (
|
|
"employees.ess.view" in permissions
|
|
or "employees.work.view_self" in permissions
|
|
or "employees.attendance.view_self" in permissions
|
|
or "employees.leave.view_self" in permissions
|
|
or "employees.documents.view_self" in permissions
|
|
or "employees.payroll.view_self" in permissions
|
|
or "Staff" in role_set
|
|
):
|
|
return "/employee/dashboard"
|
|
|
|
if "system.settings.view" in permissions or "users.view" in permissions:
|
|
return "/system-settings"
|
|
|
|
return "/employee/dashboard"
|
|
|
|
def _is_user_login_allowed(user: User) -> tuple[bool, str | None]:
|
|
if not user:
|
|
return False, "Invalid credentials"
|
|
|
|
if not getattr(user, "is_active", True):
|
|
return False, "User account is inactive."
|
|
|
|
if hasattr(user, "allow_login") and not bool(getattr(user, "allow_login", True)):
|
|
return False, "Login is disabled for this account."
|
|
|
|
if hasattr(user, "is_locked") and bool(getattr(user, "is_locked", False)):
|
|
return False, "User account is locked."
|
|
|
|
if hasattr(user, "deleted_at") and getattr(user, "deleted_at", None) is not None:
|
|
return False, "User account is deleted."
|
|
|
|
return True, None
|
|
|
|
|
|
def _template_context(
|
|
request: Request,
|
|
db=None,
|
|
*,
|
|
title: str,
|
|
flash: str | None = None,
|
|
extra: dict | None = None,
|
|
) -> dict:
|
|
ctx = {
|
|
"request": request,
|
|
"csrf_token": get_or_create_csrf_token(request),
|
|
"flash": flash,
|
|
"title": title,
|
|
}
|
|
|
|
if db is not None:
|
|
current_user = get_current_user(request, db=db)
|
|
if current_user:
|
|
ctx.update(
|
|
{
|
|
"current_user": current_user,
|
|
"current_user_roles": _user_roles(db, int(current_user.id)),
|
|
"current_user_permissions": list(
|
|
_user_permissions(db, int(current_user.id))
|
|
),
|
|
}
|
|
)
|
|
|
|
if extra:
|
|
ctx.update(extra)
|
|
|
|
return ctx
|
|
|
|
|
|
def _render_login(request: Request, flash: str | None = None, status_code: int = 200):
|
|
return templates.TemplateResponse(
|
|
"modules/core/iam/templates/login.html",
|
|
_template_context(request, title="Professional Login", flash=flash),
|
|
status_code=status_code,
|
|
)
|
|
|
|
|
|
def _render_client_login(request: Request, flash: str | None = None, status_code: int = 200):
|
|
return templates.TemplateResponse(
|
|
"modules/core/iam/templates/client_login.html",
|
|
_template_context(request, title="Client Login", flash=flash),
|
|
status_code=status_code,
|
|
)
|
|
|
|
|
|
def _is_client_only_roles(roles: list[str]) -> bool:
|
|
role_set = set(roles or [])
|
|
return "Client" in role_set and not bool(role_set.intersection(PROFESSIONAL_ROLE_NAMES))
|
|
|
|
|
|
def _client_identity_exists(db, user_id: int, bound_tenant_id: int | None) -> bool:
|
|
q = select(ClientPortalIdentity.id).where(ClientPortalIdentity.user_id == int(user_id))
|
|
if bound_tenant_id is not None:
|
|
q = q.where(ClientPortalIdentity.tenant_id == int(bound_tenant_id))
|
|
return db.execute(q).scalar_one_or_none() is not None
|
|
|
|
|
|
def _login_path_for_user(db, user: User | None) -> str:
|
|
if not user:
|
|
return PROFESSIONAL_LOGIN_PATH
|
|
return CLIENT_LOGIN_PATH if _is_client_only_roles(_user_roles(db, int(user.id))) else PROFESSIONAL_LOGIN_PATH
|
|
|
|
|
|
def _render_otp(request: Request, flash: str | None = None, status_code: int = 200):
|
|
db = CommonSessionLocal()
|
|
try:
|
|
return templates.TemplateResponse(
|
|
"modules/core/iam/templates/otp.html",
|
|
_template_context(
|
|
request,
|
|
db=db,
|
|
title="OTP Verification",
|
|
flash=flash,
|
|
),
|
|
status_code=status_code,
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def _render_change_password(
|
|
request: Request, flash: str | None = None, status_code: int = 200
|
|
):
|
|
db = CommonSessionLocal()
|
|
try:
|
|
return templates.TemplateResponse(
|
|
"modules/core/iam/templates/change_password.html",
|
|
_template_context(
|
|
request,
|
|
db=db,
|
|
title="Change Password",
|
|
flash=flash,
|
|
),
|
|
status_code=status_code,
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def _render_change_password_otp(
|
|
request: Request, flash: str | None = None, status_code: int = 200
|
|
):
|
|
db = CommonSessionLocal()
|
|
try:
|
|
return templates.TemplateResponse(
|
|
"modules/core/iam/templates/change_password_otp.html",
|
|
_template_context(
|
|
request,
|
|
db=db,
|
|
title="Confirm Password Change",
|
|
flash=flash,
|
|
),
|
|
status_code=status_code,
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def _render_forgot_password(
|
|
request: Request, flash: str | None = None, status_code: int = 200
|
|
):
|
|
return templates.TemplateResponse(
|
|
"modules/core/iam/templates/forgot_password.html",
|
|
_template_context(
|
|
request,
|
|
title="Forgot Password",
|
|
flash=flash,
|
|
),
|
|
status_code=status_code,
|
|
)
|
|
|
|
|
|
def _render_reset_password(
|
|
request: Request, flash: str | None = None, status_code: int = 200
|
|
):
|
|
return templates.TemplateResponse(
|
|
"modules/core/iam/templates/reset_password.html",
|
|
_template_context(
|
|
request,
|
|
title="Reset Password",
|
|
flash=flash,
|
|
),
|
|
status_code=status_code,
|
|
)
|
|
|
|
|
|
def _render_invite_accept(
|
|
request: Request, token: str, flash: str | None = None, status_code: int = 200
|
|
):
|
|
return templates.TemplateResponse(
|
|
"modules/core/iam/templates/invite_accept.html",
|
|
_template_context(
|
|
request,
|
|
title="Accept Invite",
|
|
flash=flash,
|
|
extra={"token": token},
|
|
),
|
|
status_code=status_code,
|
|
)
|
|
|
|
|
|
@router.get("/invite/accept")
|
|
def invite_accept_page(request: Request, token: str = ""):
|
|
if not token.strip():
|
|
return RedirectResponse(url="/login", status_code=303)
|
|
return _render_invite_accept(request, token=token.strip())
|
|
|
|
|
|
@router.post("/invite/accept")
|
|
def invite_accept_submit(
|
|
request: Request,
|
|
token: str = Form(...),
|
|
password: str = Form(...),
|
|
csrf_token: str = Form(...),
|
|
):
|
|
validate_csrf(request, csrf_token)
|
|
token_clean = token.strip()
|
|
if not token_clean:
|
|
return RedirectResponse(url="/login", status_code=303)
|
|
db = CommonSessionLocal()
|
|
try:
|
|
try:
|
|
user = accept_invite(db, token_clean, password.strip())
|
|
except ValueError as exc:
|
|
return _render_invite_accept(request, token=token_clean, flash=str(exc), status_code=400)
|
|
if not user:
|
|
return _render_invite_accept(request, token=token_clean, flash="Invalid or expired invite link.", status_code=400)
|
|
return RedirectResponse(url=_login_path_for_user(db, user), status_code=303)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.get("/login")
|
|
def login_page(request: Request):
|
|
return _render_login(request)
|
|
|
|
|
|
@router.get("/client/login")
|
|
def client_login_page(request: Request):
|
|
return _render_client_login(request)
|
|
|
|
|
|
def _submit_login(
|
|
request: Request,
|
|
*,
|
|
identifier: str,
|
|
password: str,
|
|
csrf_token: str,
|
|
client_portal: bool,
|
|
):
|
|
validate_csrf(request, csrf_token)
|
|
|
|
identifier_clean = identifier.strip().lower()
|
|
render_login = _render_client_login if client_portal else _render_login
|
|
login_path = CLIENT_LOGIN_PATH if client_portal else PROFESSIONAL_LOGIN_PATH
|
|
|
|
if not client_portal and "@" not in identifier_clean:
|
|
return render_login(
|
|
request,
|
|
flash="Professional users must sign in with their email address. Clients should use Client Login.",
|
|
status_code=400,
|
|
)
|
|
|
|
ip = _client_ip(request)
|
|
key = _attempt_key(identifier_clean, ip)
|
|
|
|
db = CommonSessionLocal()
|
|
try:
|
|
la = db.execute(
|
|
select(LoginAttempt).where(LoginAttempt.key == key)
|
|
).scalar_one_or_none()
|
|
now = datetime.now(timezone.utc)
|
|
|
|
if la and la.locked_until_utc and la.locked_until_utc.replace(
|
|
tzinfo=timezone.utc
|
|
) > now:
|
|
return render_login(
|
|
request,
|
|
flash=f"Account temporarily locked. Try again after {la.locked_until_utc}.",
|
|
status_code=429,
|
|
)
|
|
|
|
bound_tenant_id = _bound_domain_tenant_id(request)
|
|
user = resolve_login_user(db, identifier_clean, bound_tenant_id)
|
|
roles = _user_roles(db, int(user.id)) if user else []
|
|
is_client_only = _is_client_only_roles(roles)
|
|
has_client_identity = bool(
|
|
user and _client_identity_exists(db, int(user.id), bound_tenant_id)
|
|
)
|
|
|
|
portal_allowed = (
|
|
is_client_only and has_client_identity
|
|
if client_portal
|
|
else bool(user and not is_client_only)
|
|
)
|
|
|
|
can_login, blocked_reason = _is_user_login_allowed(user)
|
|
password_ok = bool(user and verify_password(password, user.password_hash))
|
|
|
|
if not user or not portal_allowed or not can_login or not password_ok:
|
|
lock_attempts = 5
|
|
lock_minutes = 15
|
|
|
|
if user:
|
|
bs = _get_branch_security_policy(db, user)
|
|
if bs:
|
|
lock_attempts = bs.lockout_attempts
|
|
lock_minutes = bs.lockout_minutes
|
|
|
|
if not la:
|
|
la = LoginAttempt(key=key, attempts=0, updated_at_utc=now)
|
|
db.add(la)
|
|
|
|
la.attempts = int(la.attempts or 0) + 1
|
|
la.updated_at_utc = now
|
|
|
|
if la.attempts >= lock_attempts:
|
|
la.locked_until_utc = now + timedelta(minutes=lock_minutes)
|
|
la.attempts = 0
|
|
|
|
db.commit()
|
|
|
|
if user and can_login and password_ok and not portal_allowed:
|
|
flash = (
|
|
"This login is for clients only. Please use Professional Login."
|
|
if client_portal
|
|
else "Client accounts must use the Client Login page."
|
|
)
|
|
else:
|
|
flash = blocked_reason or "Invalid credentials"
|
|
return render_login(request, flash=flash, status_code=400)
|
|
|
|
if la:
|
|
la.attempts = 0
|
|
la.locked_until_utc = None
|
|
la.updated_at_utc = now
|
|
db.commit()
|
|
|
|
user_id = int(user.id)
|
|
user_email = str(user.email)
|
|
tenant_id = getattr(user, "tenant_id", None)
|
|
branch_id = getattr(user, "branch_id", None)
|
|
must_change_password = bool(getattr(user, "must_change_password", False))
|
|
|
|
permissions = _user_permissions(db, user_id)
|
|
bs = _get_branch_security_policy(db, user)
|
|
|
|
# A verified active audit-firm domain is a hard tenant boundary.
|
|
# System Admin retains the existing platform-support capability, but all
|
|
# tenant users must belong to the tenant mapped to this hostname.
|
|
is_system_admin = "System Admin" in set(roles or [])
|
|
user_tenant_id = int(tenant_id) if tenant_id not in (None, "", 0, "0") else None
|
|
if bound_tenant_id is not None and not is_system_admin and user_tenant_id != bound_tenant_id:
|
|
_clear_login_session(request)
|
|
return render_login(
|
|
request,
|
|
flash="This account does not belong to the firm associated with this domain. Please use your firm's login URL.",
|
|
status_code=403,
|
|
)
|
|
|
|
request.session[SESSION_USER_ID_KEY] = user_id
|
|
request.session[SESSION_LOGIN_AT_KEY] = now.isoformat()
|
|
request.session["user_email"] = user_email
|
|
request.session[LOGIN_ENTRY_PATH_KEY] = login_path
|
|
tenant_code = _tenant_code(db, tenant_id)
|
|
branch_code = _branch_code(db, branch_id)
|
|
|
|
request.session["tenant_id"] = tenant_id
|
|
request.session["branch_id"] = branch_id
|
|
request.session["active_tenant_id"] = tenant_id
|
|
request.session["active_branch_id"] = branch_id
|
|
if tenant_code:
|
|
request.session["tenant_code"] = tenant_code
|
|
request.session["active_tenant_code"] = tenant_code
|
|
else:
|
|
request.session.pop("tenant_code", None)
|
|
request.session.pop("active_tenant_code", None)
|
|
if branch_code:
|
|
request.session["branch_code"] = branch_code
|
|
request.session["active_branch_code"] = branch_code
|
|
else:
|
|
request.session.pop("branch_code", None)
|
|
request.session.pop("active_branch_code", None)
|
|
|
|
active_financial_year = _default_financial_year_code(db, tenant_id)
|
|
if active_financial_year:
|
|
request.session["active_financial_year"] = active_financial_year
|
|
request.session["must_change_password"] = must_change_password
|
|
default_post_login_redirect = _post_login_redirect(
|
|
must_change_password, permissions, roles
|
|
)
|
|
pending_post_login_redirect = None if must_change_password else _consume_safe_post_login_redirect(request)
|
|
request.session["post_login_redirect"] = pending_post_login_redirect or default_post_login_redirect
|
|
|
|
if _otp_required(bs, roles):
|
|
code = start_otp(request)
|
|
try:
|
|
send_auth_otp_email(db, user=user, otp_code=code, purpose="login")
|
|
db.commit()
|
|
except Exception as exc:
|
|
print(f"[EMAIL OTP ERROR] user={user_email} error={exc}")
|
|
_log_dev_otp("login", user_email, code)
|
|
request.session["otp_verified"] = False
|
|
return RedirectResponse(url="/otp", status_code=303)
|
|
|
|
request.session["otp_verified"] = True
|
|
return RedirectResponse(
|
|
url=request.session.get("post_login_redirect", "/system-settings"),
|
|
status_code=303,
|
|
)
|
|
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post("/login")
|
|
def login_submit(
|
|
request: Request,
|
|
email: str = Form(...),
|
|
password: str = Form(...),
|
|
csrf_token: str = Form(...),
|
|
):
|
|
return _submit_login(
|
|
request,
|
|
identifier=email,
|
|
password=password,
|
|
csrf_token=csrf_token,
|
|
client_portal=False,
|
|
)
|
|
|
|
|
|
@router.post("/client/login")
|
|
def client_login_submit(
|
|
request: Request,
|
|
identifier: str = Form(...),
|
|
password: str = Form(...),
|
|
csrf_token: str = Form(...),
|
|
):
|
|
return _submit_login(
|
|
request,
|
|
identifier=identifier,
|
|
password=password,
|
|
csrf_token=csrf_token,
|
|
client_portal=True,
|
|
)
|
|
|
|
|
|
@router.get("/otp")
|
|
def otp_page(request: Request):
|
|
if not request.session.get(SESSION_USER_ID_KEY):
|
|
return RedirectResponse(url="/login", status_code=303)
|
|
return _render_otp(request)
|
|
|
|
|
|
@router.post("/otp")
|
|
def otp_submit(
|
|
request: Request,
|
|
otp: str = Form(...),
|
|
csrf_token: str = Form(...),
|
|
):
|
|
validate_csrf(request, csrf_token)
|
|
|
|
if not request.session.get(SESSION_USER_ID_KEY):
|
|
return RedirectResponse(url="/login", status_code=303)
|
|
|
|
if verify_otp(request, otp):
|
|
request.session["otp_verified"] = True
|
|
return RedirectResponse(
|
|
url=request.session.get("post_login_redirect", "/system-settings"),
|
|
status_code=303,
|
|
)
|
|
|
|
return _render_otp(
|
|
request,
|
|
flash="Invalid OTP. Please check the OTP sent to your registered email.",
|
|
status_code=400,
|
|
)
|
|
|
|
|
|
@router.get("/change-password")
|
|
def change_password_page(request: Request):
|
|
if not request.session.get(SESSION_USER_ID_KEY):
|
|
return RedirectResponse(url="/login", status_code=303)
|
|
return _render_change_password(request)
|
|
|
|
|
|
@router.post("/change-password")
|
|
def change_password_submit(
|
|
request: Request,
|
|
current_password: str = Form(...),
|
|
new_password: str = Form(...),
|
|
confirm_password: str = Form(...),
|
|
csrf_token: str = Form(...),
|
|
):
|
|
validate_csrf(request, csrf_token)
|
|
|
|
user_id = request.session.get(SESSION_USER_ID_KEY)
|
|
if not user_id:
|
|
return RedirectResponse(url="/login", status_code=303)
|
|
|
|
if new_password != confirm_password:
|
|
return _render_change_password(
|
|
request,
|
|
flash="New password and confirm password do not match.",
|
|
status_code=400,
|
|
)
|
|
|
|
if len(new_password.strip()) < 8:
|
|
return _render_change_password(
|
|
request,
|
|
flash="New password must be at least 8 characters.",
|
|
status_code=400,
|
|
)
|
|
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = db.execute(select(User).where(User.id == user_id)).scalar_one_or_none()
|
|
if not user:
|
|
request.session.clear()
|
|
return RedirectResponse(url="/login", status_code=303)
|
|
|
|
if not verify_password(current_password, user.password_hash):
|
|
return _render_change_password(
|
|
request,
|
|
flash="Current password is incorrect.",
|
|
status_code=400,
|
|
)
|
|
|
|
request.session["pending_password_change_hash"] = hash_password(
|
|
new_password.strip()
|
|
)
|
|
request.session["pending_password_change_user_id"] = int(user.id)
|
|
|
|
code = start_otp(request)
|
|
try:
|
|
send_auth_otp_email(db, user=user, otp_code=code, purpose="password_change")
|
|
db.commit()
|
|
except Exception as exc:
|
|
print(f"[EMAIL OTP ERROR] password-change user={user.email} error={exc}")
|
|
_log_dev_otp("password-change", str(user.email), code)
|
|
|
|
return RedirectResponse(url="/change-password/otp", status_code=303)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.get("/change-password/otp")
|
|
def change_password_otp_page(request: Request):
|
|
if not request.session.get(SESSION_USER_ID_KEY):
|
|
return RedirectResponse(url="/login", status_code=303)
|
|
|
|
if not request.session.get("pending_password_change_hash"):
|
|
return RedirectResponse(url="/change-password", status_code=303)
|
|
|
|
return _render_change_password_otp(request)
|
|
|
|
|
|
@router.post("/change-password/otp")
|
|
def change_password_otp_submit(
|
|
request: Request,
|
|
otp: str = Form(...),
|
|
csrf_token: str = Form(...),
|
|
):
|
|
validate_csrf(request, csrf_token)
|
|
|
|
user_id = request.session.get(SESSION_USER_ID_KEY)
|
|
pending_user_id = request.session.get("pending_password_change_user_id")
|
|
pending_hash = request.session.get("pending_password_change_hash")
|
|
|
|
if not user_id:
|
|
return RedirectResponse(url="/login", status_code=303)
|
|
|
|
if not pending_hash or not pending_user_id or int(user_id) != int(pending_user_id):
|
|
return RedirectResponse(url="/change-password", status_code=303)
|
|
|
|
if not verify_otp(request, otp):
|
|
return _render_change_password_otp(
|
|
request,
|
|
flash="Invalid OTP. Please check the OTP sent to your registered email.",
|
|
status_code=400,
|
|
)
|
|
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = db.execute(select(User).where(User.id == user_id)).scalar_one_or_none()
|
|
if not user:
|
|
request.session.clear()
|
|
return RedirectResponse(url="/login", status_code=303)
|
|
|
|
user.password_hash = pending_hash
|
|
user.must_change_password = False
|
|
user.password_changed_at_utc = datetime.now(timezone.utc)
|
|
try:
|
|
send_password_changed_email(db, user=user)
|
|
except Exception as exc:
|
|
print(f"[EMAIL PASSWORD CHANGED ERROR] user={user.email} error={exc}")
|
|
db.commit()
|
|
|
|
roles = _user_roles(db, int(user.id))
|
|
permissions = _user_permissions(db, int(user.id))
|
|
|
|
request.session.pop("pending_password_change_hash", None)
|
|
request.session.pop("pending_password_change_user_id", None)
|
|
request.session["must_change_password"] = False
|
|
request.session["post_login_redirect"] = _post_login_redirect(False, permissions, roles)
|
|
|
|
return RedirectResponse(url=request.session.get("post_login_redirect", "/system-settings"), status_code=303)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.get("/forgot-password")
|
|
def forgot_password_page(request: Request):
|
|
return _render_forgot_password(request)
|
|
|
|
|
|
@router.post("/forgot-password")
|
|
def forgot_password_submit(
|
|
request: Request,
|
|
email: str = Form(...),
|
|
csrf_token: str = Form(...),
|
|
):
|
|
validate_csrf(request, csrf_token)
|
|
|
|
email_clean = email.strip().lower()
|
|
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = resolve_login_user(db, email_clean, _bound_domain_tenant_id(request))
|
|
|
|
request.session.pop("password_reset_user_id", None)
|
|
request.session.pop("password_reset_email", None)
|
|
|
|
if not user:
|
|
return _render_forgot_password(
|
|
request,
|
|
flash="If the login ID exists, password reset instructions have been sent to the registered email.",
|
|
status_code=200,
|
|
)
|
|
|
|
can_login, _ = _is_user_login_allowed(user)
|
|
if not can_login:
|
|
return _render_forgot_password(
|
|
request,
|
|
flash="If the login ID exists, password reset instructions have been sent to the registered email.",
|
|
status_code=200,
|
|
)
|
|
|
|
request.session["password_reset_user_id"] = int(user.id)
|
|
request.session["password_reset_email"] = str(user.email)
|
|
|
|
code = start_otp(request)
|
|
try:
|
|
send_auth_otp_email(db, user=user, otp_code=code, purpose="password_reset")
|
|
db.commit()
|
|
except Exception as exc:
|
|
print(f"[EMAIL OTP ERROR] forgot-password user={user.email} error={exc}")
|
|
_log_dev_otp("forgot-password", str(user.email), code)
|
|
|
|
return RedirectResponse(url="/reset-password", status_code=303)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.get("/reset-password")
|
|
def reset_password_page(request: Request):
|
|
if not request.session.get("password_reset_user_id"):
|
|
return RedirectResponse(url="/forgot-password", status_code=303)
|
|
return _render_reset_password(request)
|
|
|
|
|
|
@router.post("/reset-password")
|
|
def reset_password_submit(
|
|
request: Request,
|
|
otp: str = Form(...),
|
|
new_password: str = Form(...),
|
|
confirm_password: str = Form(...),
|
|
csrf_token: str = Form(...),
|
|
):
|
|
validate_csrf(request, csrf_token)
|
|
|
|
user_id = request.session.get("password_reset_user_id")
|
|
if not user_id:
|
|
return RedirectResponse(url="/forgot-password", status_code=303)
|
|
|
|
if not verify_otp(request, otp):
|
|
return _render_reset_password(
|
|
request,
|
|
flash="Invalid OTP. Please check the OTP sent to your registered email.",
|
|
status_code=400,
|
|
)
|
|
|
|
if new_password != confirm_password:
|
|
return _render_reset_password(
|
|
request,
|
|
flash="New password and confirm password do not match.",
|
|
status_code=400,
|
|
)
|
|
|
|
if len(new_password.strip()) < 8:
|
|
return _render_reset_password(
|
|
request,
|
|
flash="New password must be at least 8 characters.",
|
|
status_code=400,
|
|
)
|
|
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user = db.execute(select(User).where(User.id == user_id)).scalar_one_or_none()
|
|
if not user:
|
|
request.session.pop("password_reset_user_id", None)
|
|
request.session.pop("password_reset_email", None)
|
|
return RedirectResponse(url="/forgot-password", status_code=303)
|
|
|
|
user.password_hash = hash_password(new_password.strip())
|
|
user.must_change_password = False
|
|
user.password_changed_at_utc = datetime.now(timezone.utc)
|
|
try:
|
|
send_password_changed_email(db, user=user)
|
|
except Exception as exc:
|
|
print(f"[EMAIL PASSWORD CHANGED ERROR] user={user.email} error={exc}")
|
|
db.commit()
|
|
|
|
request.session.pop("password_reset_user_id", None)
|
|
request.session.pop("password_reset_email", None)
|
|
|
|
return RedirectResponse(url="/login", status_code=303)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.get("/password-reset/accept")
|
|
def password_reset_token_page(request: Request, token: str = ""):
|
|
return templates.TemplateResponse(
|
|
"modules/core/iam/templates/reset_password_token.html",
|
|
_template_context(request, title="Reset Password", extra={"token": token.strip()}),
|
|
)
|
|
|
|
|
|
@router.post("/password-reset/accept")
|
|
def password_reset_token_submit(
|
|
request: Request,
|
|
token: str = Form(...),
|
|
new_password: str = Form(...),
|
|
confirm_password: str = Form(...),
|
|
csrf_token: str = Form(...),
|
|
):
|
|
validate_csrf(request, csrf_token)
|
|
token_clean = token.strip()
|
|
if new_password != confirm_password:
|
|
return templates.TemplateResponse(
|
|
"modules/core/iam/templates/reset_password_token.html",
|
|
_template_context(request, title="Reset Password", flash="New password and confirm password do not match.", extra={"token": token_clean}),
|
|
status_code=400,
|
|
)
|
|
if len(new_password.strip()) < get_settings().PASSWORD_MIN_LENGTH:
|
|
return templates.TemplateResponse(
|
|
"modules/core/iam/templates/reset_password_token.html",
|
|
_template_context(request, title="Reset Password", flash=f"New password must be at least {get_settings().PASSWORD_MIN_LENGTH} characters.", extra={"token": token_clean}),
|
|
status_code=400,
|
|
)
|
|
db = CommonSessionLocal()
|
|
try:
|
|
try:
|
|
user = reset_password_with_token(db, token_clean, new_password.strip())
|
|
except ValueError as exc:
|
|
return templates.TemplateResponse(
|
|
"modules/core/iam/templates/reset_password_token.html",
|
|
_template_context(request, title="Reset Password", flash=str(exc), extra={"token": token_clean}),
|
|
status_code=400,
|
|
)
|
|
if not user:
|
|
return templates.TemplateResponse(
|
|
"modules/core/iam/templates/reset_password_token.html",
|
|
_template_context(request, title="Reset Password", flash="Invalid or expired password reset link.", extra={"token": token_clean}),
|
|
status_code=400,
|
|
)
|
|
try:
|
|
send_password_changed_email(db, user=user)
|
|
except Exception as exc:
|
|
print(f"[EMAIL PASSWORD CHANGED ERROR] user={getattr(user, 'email', '')} error={exc}")
|
|
db.commit()
|
|
return RedirectResponse(url=_login_path_for_user(db, user), status_code=303)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.get("/change-password-required")
|
|
def change_password_required(request: Request):
|
|
return RedirectResponse(url="/change-password", status_code=303)
|
|
|
|
|
|
@router.get("/logout")
|
|
def logout(request: Request):
|
|
request.session.clear()
|
|
|
|
response = RedirectResponse(url="/login", status_code=303)
|
|
settings = get_settings()
|
|
response.delete_cookie(
|
|
key=settings.COOKIE_SESSION_NAME,
|
|
path="/",
|
|
secure=settings.COOKIE_SECURE,
|
|
httponly=True,
|
|
samesite=settings.COOKIE_SAMESITE,
|
|
)
|
|
response.headers["Cache-Control"] = (
|
|
"no-store, no-cache, must-revalidate, private, max-age=0"
|
|
)
|
|
response.headers["Pragma"] = "no-cache"
|
|
response.headers["Expires"] = "0"
|
|
response.headers["Clear-Site-Data"] = '"cache"'
|
|
return response
|
|
|