Rakseeibra/arrr-erp
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Separate professional and client login paths

Browse Source
This commit is contained in:
A R R R Associates
2026-08-02 09:27:06 +05:30
parent 181d4f2c2e
commit 4dda760ed4
3 changed files with 164 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/ui/routes/auth.py
+124 -18
View File
@@ -19,6 +19,7 @@ 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
@@ -28,6 +29,18 @@ from app.modules.email_integration.services import send_auth_otp_email, send_pas
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",
@@ -79,6 +92,7 @@ def _clear_login_session(request: Request) -> None:
"must_change_password",
"post_login_redirect",
"otp_verified",
LOGIN_ENTRY_PATH_KEY,
):
request.session.pop(key, None)
@@ -279,11 +293,37 @@ def _template_context(
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="Login", flash=flash),
_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:
@@ -408,7 +448,7 @@ def invite_accept_submit(
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", status_code=303)
return RedirectResponse(url=_login_path_for_user(db, user), status_code=303)
finally:
db.close()
@@ -418,18 +458,34 @@ def login_page(request: Request):
return _render_login(request)
@router.post("/login")
def login_submit(
@router.get("/client/login")
def client_login_page(request: Request):
return _render_client_login(request)
def _submit_login(
request: Request,
email: str = Form(...),
password: str = Form(...),
csrf_token: str = Form(...),
*,
identifier: str,
password: str,
csrf_token: str,
client_portal: bool,
):
validate_csrf(request, csrf_token)
email_clean = email.strip().lower()
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(email_clean, ip)
key = _attempt_key(identifier_clean, ip)
db = CommonSessionLocal()
try:
@@ -441,18 +497,30 @@ def login_submit(
if la and la.locked_until_utc and la.locked_until_utc.replace(
tzinfo=timezone.utc
) > now:
return _render_login(
return render_login(
request,
flash=f"Account temporarily locked. Try again after {la.locked_until_utc}.",
status_code=429,
)
user = resolve_login_user(db, email_clean, _bound_domain_tenant_id(request))
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 can_login or not password_ok:
if not user or not portal_allowed or not can_login or not password_ok:
lock_attempts = 5
lock_minutes = 15
@@ -475,8 +543,15 @@ def login_submit(
db.commit()
flash = blocked_reason or "Invalid credentials"
return _render_login(request, flash=flash, status_code=400)
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
@@ -490,19 +565,17 @@ def login_submit(
branch_id = getattr(user, "branch_id", None)
must_change_password = bool(getattr(user, "must_change_password", False))
roles = _user_roles(db, user_id)
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.
bound_tenant_id = _bound_domain_tenant_id(request)
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(
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,
@@ -511,6 +584,7 @@ def login_submit(
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)
@@ -562,6 +636,38 @@ def login_submit(
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):
@@ -897,7 +1003,7 @@ def password_reset_token_submit(
except Exception as exc:
print(f"[EMAIL PASSWORD CHANGED ERROR] user={getattr(user, 'email', '')} error={exc}")
db.commit()
return RedirectResponse(url="/login", status_code=303)
return RedirectResponse(url=_login_path_for_user(db, user), status_code=303)
finally:
db.close()