""" Role Permissions Services — PRD-008 RBAC Multi-User Merchant Accounts Business logic layer. All DB access goes through crud.py. Redis cache key pattern: perms:{user_id}:{merchant_id} TTL 60 s. """ from __future__ import annotations import hashlib import json import logging import secrets import uuid from datetime import datetime, timedelta, timezone from typing import Optional from sqlalchemy.orm import Session from src.apps.role_permissions import crud from src.apps.role_permissions.models.merchant_audit_log import MerchantAuditLog from src.apps.role_permissions.models.merchant_invite import MerchantInvite from src.apps.role_permissions.models.permission import Permission from src.apps.role_permissions.models.role import Role from src.apps.role_permissions.models.user_role import UserRole from src.core.exceptions import APIException, NotFoundError, ConflictError logger = logging.getLogger(__name__) PERM_CACHE_TTL = 60 # seconds INVITE_EXPIRY_HOURS = 72 # ─── Redis helpers ──────────────────────────────────────────────────────────── def _get_redis(): try: import redis as redis_lib from src.core.config import settings url = getattr(settings, "CELERY_RESULT_BACKEND", None) or getattr( settings, "REDIS_URL", None ) if url and "redis" in url: return redis_lib.from_url(url, decode_responses=True) except Exception: pass return None def _invalidate_perm_cache(user_id: int, merchant_id: int) -> None: try: r = _get_redis() if r: r.delete(f"perms:{user_id}:{merchant_id}") except Exception: pass # ─── Hierarchy guard ───────────────────────────────────────────────────────── def assert_hierarchy(caller_rank: int, target_rank: int, action: str = "manage") -> None: """Raise 403 if the caller's rank is not strictly greater than the target's.""" if target_rank >= caller_rank: raise APIException( status_code=403, message=f"You cannot {action} a user with an equal or higher role.", ) # ─── Bootstrap ─────────────────────────────────────────────────────────────── def bootstrap_merchant_roles( db: Session, merchant_id: int, owner_user_id: int ) -> None: """ Idempotent. Ensures the system roles exist and assigns the Owner role to the given user for this merchant. Called from admin merchant creation. """ crud.seed_system_roles(db) owner_role = crud.get_role_by_slug(db, "owner") if not owner_role: raise APIException( status_code=500, message="System roles not found — run seed first." ) existing = crud.get_user_role(db, owner_user_id, merchant_id) if not existing: crud.assign_role_to_user(db, owner_user_id, owner_role.id, merchant_id) db.flush() # ─── Permissions ───────────────────────────────────────────────────────────── def get_permissions_list(db: Session) -> list[Permission]: return crud.get_all_permissions(db) def get_user_permissions( db: Session, user_id: int, merchant_id: int, redis_client=None, ) -> set[str]: """ Return the set of permission slugs for user in this merchant context. Owner (rank=3) → return all slugs. Otherwise: check Redis cache first, fallback to DB. """ user_role = crud.get_user_role(db, user_id, merchant_id) if not user_role: return set() # Owner bypass — all permissions if user_role.role and user_role.role.role_rank == 3: all_perms = crud.get_all_permissions(db) return {p.slug for p in all_perms} cache_key = f"perms:{user_id}:{merchant_id}" r = redis_client or _get_redis() if r: try: cached = r.get(cache_key) if cached: return set(json.loads(cached)) except Exception: pass perms = crud.get_user_permissions_from_db(db, user_id, merchant_id) if r: try: r.setex(cache_key, PERM_CACHE_TTL, json.dumps(list(perms))) except Exception: pass return perms def get_user_role(db: Session, user_id: int, merchant_id: int) -> Optional[UserRole]: return crud.get_user_role(db, user_id, merchant_id) # ─── Roles ─────────────────────────────────────────────────────────────────── def get_roles_for_merchant( db: Session, merchant_id: int, caller_rank: int ) -> list[Role]: return crud.get_roles_for_merchant(db, merchant_id, caller_rank) def create_role( db: Session, merchant_id: int, label: str, permission_ids: list[int], created_by_id: int, ) -> Role: """ Create a custom role scoped to this merchant. Validates that the slug would not collide with a system role slug, and that no existing role (system or custom) for this merchant has the same label. """ import re tentative_slug = f"custom_{re.sub(r'[^a-z0-9]+', '_', label.lower()).strip('_')}_{merchant_id}" _SYSTEM_SLUGS = {"owner", "admin", "staff"} if tentative_slug in _SYSTEM_SLUGS: raise APIException( status_code=400, message="Role name conflicts with a system role.", ) existing = crud.get_roles_for_merchant(db, merchant_id=merchant_id, caller_rank=99) if any(r.label.lower() == label.strip().lower() for r in existing): raise APIException(status_code=400, message="A role with this name already exists.") role = crud.create_custom_role(db, merchant_id, label, permission_ids, created_by_id) db.commit() return role def update_role( db: Session, merchant_id: int, role_id: int, label: Optional[str], permission_ids: Optional[list[int]], redis_client=None, ) -> Role: """ Update a custom role. Raises 403 for system roles. Invalidates Redis cache for all users assigned to this role. """ role = crud.get_role_by_id(db, role_id) if not role: raise NotFoundError(message="Role not found.") if role.is_system: raise APIException(status_code=403, message="System roles cannot be modified.") if role.merchant_id != merchant_id: raise APIException(status_code=403, message="Role does not belong to this merchant.") if label and label.strip().lower() != role.label.lower(): existing = crud.get_roles_for_merchant(db, merchant_id=merchant_id, caller_rank=99) if any(r.id != role_id and r.label.lower() == label.strip().lower() for r in existing): raise APIException(status_code=400, message="A role with this name already exists.") role = crud.update_role(db, role_id, label, permission_ids) # Invalidate cache for all users with this role _invalidate_role_cache(db, role_id, merchant_id, redis_client) db.commit() return role def delete_role( db: Session, merchant_id: int, role_id: int, redis_client=None, ) -> None: """ Delete a custom role. Raises 403 for system roles. Reassigns affected users to Staff and invalidates cache. """ role = crud.get_role_by_id(db, role_id) if not role: raise NotFoundError(message="Role not found.") if role.is_system: raise APIException(status_code=403, message="System roles cannot be deleted.") if role.merchant_id != merchant_id: raise APIException(status_code=403, message="Role does not belong to this merchant.") _invalidate_role_cache(db, role_id, merchant_id, redis_client) crud.delete_role(db, role_id) db.commit() def _invalidate_role_cache(db, role_id: int, merchant_id: int, redis_client=None) -> None: """Invalidate Redis cache for all users assigned to a given role.""" from sqlalchemy import select as sa_select r = redis_client or _get_redis() if not r: return try: from src.apps.role_permissions.models.user_role import UserRole as _UR members = db.execute( sa_select(_UR).where( _UR.role_id == role_id, _UR.merchant_id == merchant_id, ) ).scalars().all() for ur in members: r.delete(f"perms:{ur.user_id}:{merchant_id}") except Exception: pass # ─── Team members ───────────────────────────────────────────────────────────── def assign_role_to_user( db: Session, user_id: int, role_id: int, merchant_id: int, redis_client=None, ) -> UserRole: """Upsert UserRole. Invalidates Redis cache.""" user_role = crud.assign_role_to_user(db, user_id, role_id, merchant_id) _invalidate_perm_cache(user_id, merchant_id) if redis_client: try: redis_client.delete(f"perms:{user_id}:{merchant_id}") except Exception: pass db.commit() return user_role def get_members( db: Session, merchant_id: int, caller_rank: int ) -> list[UserRole]: return crud.get_members_for_merchant(db, merchant_id, caller_rank) def remove_member( db: Session, merchant_id: int, user_id: int, actor_id: int, redis_client=None, ) -> None: """ Remove a member from the merchant. Revokes their sessions. Writes audit log and invalidates cache. """ user_role = crud.get_user_role(db, user_id, merchant_id) if not user_role: raise NotFoundError(message="User is not a member of this merchant.") crud.remove_member(db, merchant_id, user_id) _invalidate_perm_cache(user_id, merchant_id) if redis_client: try: redis_client.delete(f"perms:{user_id}:{merchant_id}") except Exception: pass crud.write_audit_log( db, merchant_id=merchant_id, actor_id=actor_id, action="member.removed", target_user_id=user_id, ) db.commit() # ─── Ownership transfer ─────────────────────────────────────────────────────── def transfer_ownership( db: Session, merchant_id: int, from_user_id: int, to_user_id: int, actor_id: int, redis_client=None, ) -> None: """ Atomic ownership transfer. Old owner → Admin, new owner → Owner. Writes two audit log rows and invalidates both cache keys. """ owner_role = crud.get_role_by_slug(db, "owner") admin_role = crud.get_role_by_slug(db, "admin") if not owner_role or not admin_role: raise APIException(status_code=500, message="System roles not found.") # Verify from_user currently owns from_ur = crud.get_user_role(db, from_user_id, merchant_id) if not from_ur or from_ur.role.role_rank != 3: raise APIException(status_code=403, message="Current user is not the owner.") to_ur = crud.get_user_role(db, to_user_id, merchant_id) if not to_ur: raise NotFoundError(message="Target user is not a member of this merchant.") # Old owner → Admin crud.assign_role_to_user(db, from_user_id, admin_role.id, merchant_id) # New owner → Owner crud.assign_role_to_user(db, to_user_id, owner_role.id, merchant_id) _invalidate_perm_cache(from_user_id, merchant_id) _invalidate_perm_cache(to_user_id, merchant_id) crud.write_audit_log( db, merchant_id=merchant_id, actor_id=actor_id, action="ownership.transferred_from", target_user_id=from_user_id, metadata={"to_user_id": to_user_id}, ) crud.write_audit_log( db, merchant_id=merchant_id, actor_id=actor_id, action="ownership.transferred_to", target_user_id=to_user_id, metadata={"from_user_id": from_user_id}, ) db.commit() # ─── Invites ───────────────────────────────────────────────────────────────── def create_invite( db: Session, merchant_id: int, email: str, role_id: int, invited_by_id: int, caller_rank: int, ) -> tuple[MerchantInvite, str]: """ Create a pending invite. Returns (invite, raw_token). - Validates role rank <= caller_rank. - Returns existing pending invite if a duplicate exists. """ role = crud.get_role_by_id(db, role_id) if not role: raise NotFoundError(message="Role not found.") if role.role_rank >= caller_rank: raise APIException( status_code=403, message="You cannot invite a user with an equal or higher role.", ) # Check if the email already belongs to an active member of this merchant. from sqlalchemy import select as sa_select from src.apps.users.models.user import User as _User from src.apps.role_permissions.models.user_role import UserRole as _UserRole existing_member = db.execute( sa_select(_UserRole).join(_User, _UserRole.user_id == _User.id).where( _User.email == email.lower(), _UserRole.merchant_id == merchant_id, _User.deleted_at.is_(None), ) ).scalar_one_or_none() if existing_member: raise ConflictError( message=f"{email} is already an active member of this account." ) # Check for existing pending invite to the same email address for this merchant. # We match on email only (not role_id) — a user can only have one pending invite # per merchant at a time. Allowing multiple pending invites for the same email # could let an attacker enumerate role assignments by requesting different roles. existing_invites = crud.get_pending_invites(db, merchant_id) for inv in existing_invites: if inv.email.lower() == email.lower(): raise ConflictError( message=f"A pending invitation has already been sent to {email}. Use 'Resend' to send it again." ) raw_token = secrets.token_urlsafe(32) token_hash = hashlib.sha256(raw_token.encode()).hexdigest() expires_at = datetime.now(timezone.utc).replace(tzinfo=None) + timedelta( hours=INVITE_EXPIRY_HOURS ) invite = crud.create_invite( db, merchant_id=merchant_id, email=email, role_id=role_id, token_hash=token_hash, invited_by=invited_by_id, expires_at=expires_at, ) crud.write_audit_log( db, merchant_id=merchant_id, actor_id=invited_by_id, action="member.invited", target_user_id=None, target_role_id=role_id, metadata={"email": email}, ) db.commit() return invite, raw_token def get_pending_invites(db: Session, merchant_id: int) -> list[MerchantInvite]: return crud.get_pending_invites(db, merchant_id) def resend_invite( db: Session, merchant_id: int, invite_id: int, actor_id: int ) -> tuple[MerchantInvite, str]: """ Regenerate the invite token and reset the expiry, then re-dispatch the email. Returns (invite, raw_token) so the caller can fire the email task. """ from sqlalchemy import update as sa_update invite = crud.get_invite_by_id(db, invite_id, merchant_id) if not invite: raise NotFoundError(message="Invite not found.") if not invite.is_pending: raise APIException(status_code=400, message="Only pending invites can be re-sent.") raw_token = secrets.token_urlsafe(32) token_hash = hashlib.sha256(raw_token.encode()).hexdigest() expires_at = datetime.now(timezone.utc).replace(tzinfo=None) + timedelta( hours=INVITE_EXPIRY_HOURS ) db.execute( sa_update(MerchantInvite) .where(MerchantInvite.id == invite_id) .values(token_hash=token_hash, expires_at=expires_at) ) crud.write_audit_log( db, merchant_id=merchant_id, actor_id=actor_id, action="member.invite_resent", metadata={"invite_id": invite_id, "email": invite.email}, ) db.commit() db.refresh(invite) return invite, raw_token def cancel_invite( db: Session, merchant_id: int, invite_id: int, actor_id: int ) -> None: invite = crud.get_invite_by_id(db, invite_id, merchant_id) if not invite: raise NotFoundError(message="Invite not found.") crud.cancel_invite(db, invite_id, merchant_id) crud.write_audit_log( db, merchant_id=merchant_id, actor_id=actor_id, action="member.invite_cancelled", metadata={"invite_id": invite_id, "email": invite.email}, ) db.commit() def get_invite_info(db: Session, raw_token: str) -> dict: """ Public endpoint: return invite metadata for display on the accept-invite page. Raises 404 if not found, 410 if expired or cancelled. """ token_hash = hashlib.sha256(raw_token.encode()).hexdigest() invite = crud.get_invite_by_token_hash(db, token_hash) if not invite: raise NotFoundError(message="Invite not found.") if invite.cancelled_at is not None or invite.accepted_at is not None or invite.is_expired: raise APIException(status_code=410, message="This invite has expired or been cancelled.") # Load merchant info from sqlalchemy import select as sa_select from src.apps.merchants.models.merchant import Merchant from src.core.config import settings merchant = db.execute( sa_select(Merchant).where( Merchant.id == invite.merchant_id, Merchant.deleted_at.is_(None), ) ).scalar_one_or_none() merchant_name = merchant.name if merchant else "" merchant_logo_url = None if merchant and merchant.brand_logo: merchant_logo_url = ( f"{settings.SERVER_HOST}/api/v1/files/download" f"?filepath=/uploads/{merchant.merchant_id}/{merchant.brand_logo.name}" ) return { "merchant_name": merchant_name, "merchant_logo_url": merchant_logo_url, "role_name": invite.role.label if invite.role else "", "email": invite.email, "expires_at": invite.expires_at, } def accept_invite_service( db: Session, raw_token: str, first_name: str, last_name: str, password: str, request_ip: Optional[str] = None, ) -> dict: """ Accept a pending invite. - Validates invite is still pending. - Creates User + UserRole. - Marks invite as accepted. - Creates AuthSession → returns tokens. - Writes audit log. """ from datetime import timedelta from sqlalchemy import select as sa_select from src.apps.auth.models.auth_session import AuthSession from src.apps.auth.utils.jwt import jwt_manager from src.apps.merchants.models.merchant_users import MerchantUsers from src.apps.users.models.user import User from src.core.config import settings from src.core.utils.password import encrypt_password token_hash = hashlib.sha256(raw_token.encode()).hexdigest() invite = crud.get_invite_by_token_hash(db, token_hash) if not invite: raise NotFoundError(message="Invite not found.") if invite.accepted_at is not None or invite.cancelled_at is not None or invite.is_expired: raise APIException( status_code=410, message="This invite has expired or is no longer valid.", ) # Check no existing active user with this email existing = db.execute( sa_select(User).where(User.email == invite.email, User.deleted_at.is_(None)) ).scalar_one_or_none() if existing: raise APIException( status_code=409, message="A user with this email already exists. Please log in instead.", ) # Hash password hashed_bytes = encrypt_password(password) if not hashed_bytes: raise APIException(status_code=500, message="Password hashing failed.") hashed_str = hashed_bytes.decode("utf-8") if isinstance(hashed_bytes, bytes) else hashed_bytes username = invite.email.split("@")[0] + "_" + str(uuid.uuid4())[:8] user = User( email=invite.email, username=username, hashed_password=hashed_str, first_name=first_name, last_name=last_name, is_active=True, is_verified=True, user_type="merchant", ) if hasattr(user, "update_full_name"): user.first_name = first_name user.last_name = last_name user.update_full_name() from src.apps.base.utils.functions import generate_secure_id try: user.user_id = generate_secure_id(prepend="usr", length=20) except Exception: user.user_id = f"usr_{str(uuid.uuid4())[:20]}" db.add(user) db.flush() # Create MerchantUsers row so /me can find the merchant mu = MerchantUsers( user_id=user.id, merchant_id=invite.merchant_id, is_owner=False, ) db.add(mu) db.flush() # Assign role crud.assign_role_to_user(db, user.id, invite.role_id, invite.merchant_id) # Mark invite accepted — atomic guard prevents double-acceptance under concurrent requests claimed = crud.accept_invite(db, invite.id) if not claimed: # Another concurrent request accepted this invite first; rollback the user we just created db.rollback() raise APIException( status_code=410, message="This invite has already been accepted. Please log in instead.", ) # Create tokens user_data = { "user_id": user.id, "email": user.email, "first_name": user.first_name, "last_name": user.last_name, "is_active": user.is_active, "is_verified": user.is_verified, "is_superuser": False, } token_payload = {"user": user_data} access_token_expires = timedelta(minutes=settings.jwt_expires_minutes) refresh_token_expires = timedelta(days=settings.jwt_refresh_expires_days) access_token = jwt_manager.create_access_token(token_payload, access_token_expires) refresh_token = jwt_manager.create_refresh_token(token_payload, refresh_token_expires) now = datetime.now(timezone.utc) auth_session = AuthSession( user_id=user.id, access_token=access_token, refresh_token=refresh_token, access_token_expires_at=now + access_token_expires, refresh_token_expires_at=now + refresh_token_expires, ip_address=request_ip, is_active=True, is_revoked=False, ) db.add(auth_session) crud.write_audit_log( db, merchant_id=invite.merchant_id, actor_id=user.id, action="member.invite_accepted", target_user_id=user.id, target_role_id=invite.role_id, metadata={"invite_id": invite.id, "email": invite.email}, ) db.commit() return { "access_token": access_token, "refresh_token": refresh_token, "token_type": "bearer", "expires_at": (now + access_token_expires).isoformat(), "user": { "id": user.id, "email": user.email, "first_name": user.first_name, "last_name": user.last_name, }, } # ─── Audit log ──────────────────────────────────────────────────────────────── def write_audit_log( db: Session, merchant_id: int, actor_id: int, action: str, target_user_id: Optional[int] = None, target_role_id: Optional[int] = None, metadata: Optional[dict] = None, ) -> MerchantAuditLog: return crud.write_audit_log( db, merchant_id=merchant_id, actor_id=actor_id, action=action, target_user_id=target_user_id, target_role_id=target_role_id, metadata=metadata, )