Prepare ERP source for Gitea deployment

This commit is contained in:
A R R R Associates
2026-06-20 15:01:44 +05:30
commit 5c75eb6bd9
450 changed files with 67698 additions and 0 deletions
View File
+195
View File
@@ -0,0 +1,195 @@
from __future__ import annotations
from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel
from sqlalchemy import select
from sqlalchemy.orm import Session
from app.core.db.deps import get_common_db
from app.core.security.session_auth import require_login
from app.modules.core.audit.service import write_audit_log
from app.modules.core.iam.models import User
from app.modules.core.iam.scope import build_scope, scope_to_http, assert_can_manage_role_object
from app.modules.core.rbac.models import Permission, Role, RolePermission
router = APIRouter(prefix="/rbac", tags=["rbac"])
class RoleCreateRequest(BaseModel):
name: str
is_active: bool = True
class PermissionCreateRequest(BaseModel):
code: str
name: str
is_active: bool = True
class RolePermissionUpdateRequest(BaseModel):
permission_ids: list[int] = []
def _require_system_admin_scope(db: Session, current_user: User):
scope = build_scope(db, current_user)
if not scope.is_system_admin:
raise HTTPException(status_code=403, detail="Only System Admin can access RBAC.")
return scope
@router.get("/roles")
def list_roles(
current_user: User = Depends(require_login),
db: Session = Depends(get_common_db),
):
_require_system_admin_scope(db, current_user)
roles = db.execute(select(Role).order_by(Role.name)).scalars().all()
return [{"id": r.id, "name": r.name, "is_active": r.is_active} for r in roles]
@router.post("/roles")
def create_role(
payload: RoleCreateRequest,
current_user: User = Depends(require_login),
db: Session = Depends(get_common_db),
):
_require_system_admin_scope(db, current_user)
name = payload.name.strip()
if not name:
raise HTTPException(status_code=400, detail="Role name is required")
exists = db.execute(select(Role).where(Role.name == name)).scalar_one_or_none()
if exists:
raise HTTPException(status_code=400, detail="Role already exists")
role = Role(name=name, is_active=payload.is_active)
db.add(role)
db.commit()
db.refresh(role)
write_audit_log(
db,
action="role.create.api",
entity_type="role",
actor=current_user,
entity_id=role.id,
entity_name=role.name,
target_tenant_id=current_user.tenant_id,
target_branch_id=current_user.branch_id,
details={"name": role.name, "is_active": role.is_active},
)
return {"status": "ok", "id": role.id}
@router.get("/permissions")
def list_permissions(
current_user: User = Depends(require_login),
db: Session = Depends(get_common_db),
):
_require_system_admin_scope(db, current_user)
permissions = db.execute(select(Permission).order_by(Permission.code)).scalars().all()
return [{"id": p.id, "code": p.code, "name": p.name, "is_active": p.is_active} for p in permissions]
@router.post("/permissions")
def create_permission(
payload: PermissionCreateRequest,
current_user: User = Depends(require_login),
db: Session = Depends(get_common_db),
):
_require_system_admin_scope(db, current_user)
code = payload.code.strip()
if not code:
raise HTTPException(status_code=400, detail="Permission code is required")
exists = db.execute(select(Permission).where(Permission.code == code)).scalar_one_or_none()
if exists:
raise HTTPException(status_code=400, detail="Permission already exists")
permission = Permission(code=code, name=payload.name.strip(), is_active=payload.is_active)
db.add(permission)
db.commit()
db.refresh(permission)
write_audit_log(
db,
action="permission.create.api",
entity_type="permission",
actor=current_user,
entity_id=permission.id,
entity_name=permission.code,
target_tenant_id=current_user.tenant_id,
target_branch_id=current_user.branch_id,
details={"code": permission.code, "name": permission.name, "is_active": permission.is_active},
)
return {"status": "ok", "id": permission.id}
@router.get("/roles/{role_id}")
def role_detail(
role_id: int,
current_user: User = Depends(require_login),
db: Session = Depends(get_common_db),
):
scope = _require_system_admin_scope(db, current_user)
role = db.execute(select(Role).where(Role.id == role_id)).scalar_one_or_none()
if not role:
raise HTTPException(status_code=404, detail="Role not found")
try:
assert_can_manage_role_object(scope, role)
except Exception as exc:
raise scope_to_http(exc)
permission_ids = db.execute(
select(RolePermission.permission_id).where(RolePermission.role_id == role.id)
).scalars().all()
return {
"id": role.id,
"name": role.name,
"is_active": role.is_active,
"permission_ids": list(permission_ids),
}
@router.put("/roles/{role_id}/permissions")
def update_role_permissions(
role_id: int,
payload: RolePermissionUpdateRequest,
current_user: User = Depends(require_login),
db: Session = Depends(get_common_db),
):
_require_system_admin_scope(db, current_user)
role = db.execute(select(Role).where(Role.id == role_id)).scalar_one_or_none()
if not role:
raise HTTPException(status_code=404, detail="Role not found")
old_permission_ids = db.execute(
select(RolePermission.permission_id).where(RolePermission.role_id == role.id)
).scalars().all()
db.execute(RolePermission.__table__.delete().where(RolePermission.role_id == role.id))
for permission_id in payload.permission_ids:
db.add(RolePermission(role_id=role.id, permission_id=permission_id))
db.commit()
write_audit_log(
db,
action="role.permissions.update.api",
entity_type="role",
actor=current_user,
entity_id=role.id,
entity_name=role.name,
target_tenant_id=current_user.tenant_id,
target_branch_id=current_user.branch_id,
details={
"old_permission_ids": list(old_permission_ids),
"new_permission_ids": list(payload.permission_ids),
},
)
return {"status": "ok"}
+72
View File
@@ -0,0 +1,72 @@
from __future__ import annotations
from fastapi import Depends, HTTPException
from sqlalchemy import select
from sqlalchemy.orm import Session
from app.core.db.deps import get_common_db
from app.core.security.session_auth import require_login
from app.modules.core.rbac.models import Permission, Role, RolePermission, UserRole
def get_user_roles(db: Session, user_id: int) -> list[str]:
q = (
select(Role.name)
.join(UserRole, UserRole.role_id == Role.id)
.where(UserRole.user_id == user_id)
.order_by(Role.name)
)
return [name for (name,) in db.execute(q).all()]
def get_user_permissions(db: Session, user_id: int) -> list[str]:
q = (
select(Permission.code)
.join(RolePermission, RolePermission.permission_id == Permission.id)
.join(Role, Role.id == RolePermission.role_id)
.join(UserRole, UserRole.role_id == Role.id)
.where(UserRole.user_id == user_id, Permission.is_active.is_(True), Role.is_active.is_(True))
.distinct()
.order_by(Permission.code)
)
return [code for (code,) in db.execute(q).all()]
def user_has_role(db: Session, user_id: int, role_name: str) -> bool:
q = (
select(UserRole.id)
.join(Role, Role.id == UserRole.role_id)
.where(UserRole.user_id == user_id, Role.name == role_name)
)
return db.execute(q).first() is not None
def user_has_permission(db: Session, user_id: int, permission_code: str) -> bool:
if user_has_role(db, user_id, "System Admin"):
return True
q = (
select(Permission.id)
.join(RolePermission, RolePermission.permission_id == Permission.id)
.join(Role, Role.id == RolePermission.role_id)
.join(UserRole, UserRole.role_id == Role.id)
.where(UserRole.user_id == user_id, Permission.code == permission_code)
)
return db.execute(q).first() is not None
def require_role(role_name: str):
def _dep(user=Depends(require_login), db: Session = Depends(get_common_db)):
if user_has_role(db, user.id, role_name):
return True
raise HTTPException(status_code=403, detail=f"Missing role: {role_name}")
return _dep
def require_permission(permission_code: str):
def _dep(user=Depends(require_login), db: Session = Depends(get_common_db)):
if user_has_permission(db, user.id, permission_code):
return True
raise HTTPException(status_code=403, detail=f"Missing permission: {permission_code}")
return _dep
+33
View File
@@ -0,0 +1,33 @@
from __future__ import annotations
from sqlalchemy import String, Boolean, Integer, ForeignKey, UniqueConstraint
from sqlalchemy.orm import Mapped, mapped_column
from app.core.db.common import CommonBase
class Role(CommonBase):
__tablename__ = "roles"
__table_args__ = (UniqueConstraint("name", name="uq_role_name"),)
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
name: Mapped[str] = mapped_column(String(100), index=True)
is_active: Mapped[bool] = mapped_column(Boolean, default=True)
class Permission(CommonBase):
__tablename__ = "permissions"
__table_args__ = (UniqueConstraint("code", name="uq_permission_code"),)
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
code: Mapped[str] = mapped_column(String(150), index=True)
name: Mapped[str] = mapped_column(String(255), default="")
is_active: Mapped[bool] = mapped_column(Boolean, default=True)
class RolePermission(CommonBase):
__tablename__ = "role_permissions"
__table_args__ = (UniqueConstraint("role_id","permission_id", name="uq_role_perm"),)
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
role_id: Mapped[int] = mapped_column(ForeignKey("roles.id", ondelete="CASCADE"), index=True)
permission_id: Mapped[int] = mapped_column(ForeignKey("permissions.id", ondelete="CASCADE"), index=True)
class UserRole(CommonBase):
__tablename__ = "user_roles"
__table_args__ = (UniqueConstraint("user_id","role_id", name="uq_user_role"),)
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
user_id: Mapped[int] = mapped_column(ForeignKey("users.id", ondelete="CASCADE"), index=True)
role_id: Mapped[int] = mapped_column(ForeignKey("roles.id", ondelete="CASCADE"), index=True)
+26
View File
@@ -0,0 +1,26 @@
from __future__ import annotations
from fastapi import HTTPException
from sqlalchemy.orm import Session
from app.modules.core.iam.models import User
from app.modules.core.rbac.deps import get_user_permissions, get_user_roles
from app.modules.core.rbac.permissions_registry import expand_permission_codes
def has_permission(db: Session, user: User | None, permission_code: str) -> bool:
if not user:
return False
roles = set(get_user_roles(db, user.id))
if "System Admin" in roles:
return True
granted = set(get_user_permissions(db, user.id))
for code in expand_permission_codes(permission_code):
if code in granted:
return True
return False
def require_permission(db: Session, user: User | None, permission_code: str) -> None:
if not has_permission(db, user, permission_code):
raise HTTPException(status_code=403, detail=f"Missing permission: {permission_code}")
@@ -0,0 +1,172 @@
from __future__ import annotations
PERMISSIONS = {
"system.settings.view": "View System Settings",
"system.settings.edit": "Edit System Settings",
"system.settings.manage": "Manage System Settings",
"users.view": "View Users",
"users.manage": "Create and Edit Users",
"rbac.view": "View Roles and Permissions",
"rbac.manage": "Manage Roles and Permissions",
"audit.view": "View Audit Logs",
"alerts.view_self": "View Own Alerts",
"alerts.manage": "Create and Manage Alerts",
"users.invite": "Invite Users",
"users.reset_password": "Reset User Passwords",
"services.view": "View Services",
"services.create": "Create Services",
"services.edit": "Edit Services",
"services.deactivate": "Deactivate Services",
"services.cross_branch": "Manage Services Across Branches",
"services.cross_tenant": "Manage Services Across Tenants",
"services.catalogue.manage": "Manage System Service Catalogue",
"services.selection.manage": "Enable or Disable Firm Services",
"service_tasks.view": "View Service Task Templates",
"service_tasks.create": "Create Service Task Templates",
"service_tasks.edit": "Edit Service Task Templates",
"service_tasks.deactivate": "Deactivate Service Task Templates",
"clients.view": "View Clients",
"clients.create": "Create Clients",
"clients.import": "Import Clients from Excel",
"clients.edit": "Edit Clients",
"clients.deactivate": "Deactivate Clients",
"clients.activate": "Activate Clients",
"clients.archive": "Archive Clients",
"clients.restore": "Restore Clients",
"clients.assign_partner": "Assign Partner to Clients",
"clients.cross_branch": "Manage Clients Across Branches",
"clients.cross_tenant": "Manage Clients Across Tenants",
"clients.export": "Export Clients",
"clients.audit_log.view": "View Client Audit Logs",
"clients.view.own_only": "View Only Own Clients",
"employees.dashboard.view": "View HR Dashboard and Reports",
"employees.view": "View Employees",
"employees.create": "Create Employees",
"employees.edit": "Edit Employees",
"employees.status": "Change Employee Status",
"employees.cross_branch": "Manage Employees Across Branches",
"employees.cross_tenant": "Manage Employees Across Tenants",
"employees.ess.view": "View Employee Self Service Portal",
"employees.ess.profile.edit": "Edit Own Employee Profile",
"employees.work.view_self": "View Own Engagement Work Dashboard",
"employees.work.manage": "Manage Employee Engagement Work Allocation",
"employees.progress.view": "View Engagement Progress Dashboard",
"employees.registration.request": "Request Employee Profile Linkage",
"employees.registration.approve": "Approve Employee Registration Requests",
"employees.attendance.punch": "Punch Employee Attendance",
"employees.attendance.view_self": "View Own Attendance",
"employees.attendance.view_all": "View Employee Attendance",
"employees.attendance.approve": "Approve or Manage Employee Attendance",
"employees.leave.apply": "Apply Employee Leave",
"employees.leave.view_self": "View Own Leave",
"employees.leave.view_all": "View Employee Leave Requests",
"employees.leave.approve": "Approve Employee Leave Requests",
"employees.leave_type.manage": "Manage Employee Leave Types",
"employees.leave_balance.manage": "Manage Employee Leave Balances",
"employees.document_type.manage": "Manage Employee Document Types",
"employees.documents.delete": "Archive Employee Documents",
"employees.documents.verify": "Verify Employee Documents",
"employees.documents.manage": "Upload and Manage Employee Documents",
"employees.documents.view_all": "View Employee Documents",
"employees.documents.upload_self": "Upload Own Employee Documents",
"employees.documents.view_self": "View Own Employee Documents",
"employees.onboarding.view": "View Employee Onboarding",
"employees.onboarding.manage": "Manage Employee Onboarding Checklist and Tasks",
"employees.onboarding.approve": "Complete or Approve Employee Onboarding Tasks",
"employees.offboarding.view": "View Employee Offboarding",
"employees.offboarding.manage": "Manage Employee Offboarding",
"employees.offboarding.approve": "Approve and Complete Employee Offboarding",
"employees.offboarding.request_self": "Request Own Employee Offboarding",
"employees.payroll.structure.manage": "Manage Employee Salary Structures",
"employees.payroll.run": "Run Employee Payroll",
"employees.payroll.view": "View Employee Payroll and Payslips",
"employees.payroll.view_self": "View Own Payslips",
"employees.payroll.payout": "Approve or Mark Payroll Paid",
"employees.import": "Import Employee HR Data from Excel",
"employees.import.employee": "Import Employees from Excel",
"employees.import.leave_type": "Import Employee Leave Types from Excel",
"employees.import.leave_balance": "Import Employee Leave Balances from Excel",
"employees.import.salary_structure": "Import Employee Salary Structures from Excel",
"consultants.view": "View Consultants",
"consultants.manage": "Create and Edit Consultants",
"consultants.link_clients": "Link Consultants to Clients",
"consultants.cross_branch": "Manage Consultants Across Branches",
"consultants.portal.view": "View Consultant Portal",
"consultants.managed_clients.manage": "Manage Own Consultant Portal Clients",
"consultants.workspace.manage": "Manage Own Consultant SaaS Workspace",
"consultants.service_requests.manage": "Review Consultant Service Requests",
"consultants.conversions.manage": "Approve Consultant Managed Client Conversions",
"billing.view": "View Billing Invoices",
"billing.create": "Create Billing Invoices",
"billing.edit": "Edit Billing Invoices",
"billing.approve": "Approve Billing Invoices",
"billing.post": "Post Billing Invoices",
"billing.cancel": "Cancel Billing Invoices",
"billing.payment.create": "Record Billing Payments",
"billing.payment.view": "View Billing Payments",
"billing.reports": "View Billing Reports",
"billing.cross_branch": "Manage Billing Across Branches",
"billing.cross_tenant": "Manage Billing Across Audit Firms",
"billing.view_own": "View Own Client Billing Only",
"billing_fee_structure.view": "View Billing Fee Structure",
"billing_fee_structure.import": "Import Billing Fee Structure",
"billing_fee_structure.edit": "Edit Billing Fee Structure",
"billing_fee_structure.delete": "Delete Billing Fee Structure",
"billing_invoice.generate": "Generate Billing Invoices",
"billing_invoice.bulk_generate": "Bulk Generate Billing Invoices",
"platform_billing.view": "View Platform Billing",
"platform_billing.create": "Create Platform Billing Accounts and Invoices",
"platform_billing.edit": "Edit Platform Billing Records",
"platform_billing.generate": "Generate Platform Billing Invoices",
"platform_billing.post": "Post Platform Invoices",
"platform_billing.cancel": "Cancel Platform Invoices",
"platform_billing.payment.create": "Record Platform Billing Payments",
"platform_billing.payment.view": "View Platform Billing Payments",
"platform_billing.reports": "View Platform Billing Reports",
"platform_plans.manage": "Manage Platform Billing Plans",
"platform_subscriptions.manage": "Manage Platform Subscriptions",
# Marketplace / public lead permissions
"marketplace_leads.view": "View Marketplace Leads",
"marketplace_leads.create": "Create Marketplace Leads",
"marketplace_leads.assign": "Assign Marketplace Leads to Audit Firms",
"marketplace_leads.update": "Update Marketplace Lead Status",
"marketplace_leads.convert": "Convert Marketplace Leads to Clients",
"marketplace_leads.reports": "View Marketplace Lead Reports",
"marketplace_leads.view_assigned": "View Assigned Marketplace Leads",
"documents.view": "View Engagement Documents",
"documents.upload": "Upload Engagement Documents",
"documents.download": "Download Engagement Documents",
"documents.delete": "Archive Engagement Documents",
"documents.audit.view": "View Document Access Logs",
"notice_cases.view": "View Notice and Case Management",
"notice_cases.create": "Create Notices and Cases",
"notice_cases.edit": "Edit Notices and Cases",
"notice_cases.events.manage": "Manage Notice/Case Timeline Events",
"notice_cases.hearings.manage": "Manage Notice/Case Hearings",
"notice_cases.orders.manage": "Manage Notice/Case Orders",
"notice_cases.documents.upload": "Upload Notice/Case Documents",
"notice_cases.documents.download": "Download Notice/Case Documents",
"notice_cases.documents.delete": "Archive Notice/Case Documents",
"notice_cases.cross_branch": "Manage Notices/Cases Across Branches",
"notice_cases.cross_tenant": "Manage Notices/Cases Across Tenants",
"notice_cases.view.own_only": "View Only Own Assigned Notices/Cases",
}
PERMISSION_ALIASES = {
"system.settings.manage": ["system.settings.edit"],
"system.settings.edit": ["system.settings.manage"],
}
def expand_permission_codes(code: str) -> list[str]:
codes = [code]
for alias in PERMISSION_ALIASES.get(code, []):
if alias not in codes:
codes.append(alias)
return codes
+53
View File
@@ -0,0 +1,53 @@
from __future__ import annotations
from sqlalchemy import or_, select
from sqlalchemy.orm import Session
from app.modules.core.iam.scope import UserScope, get_manageable_roles
from app.modules.core.rbac.models import Permission, Role, RolePermission
from app.modules.core.iam.services import paginate_list
def search_roles(db: Session, scope: UserScope, q: str | None = None) -> list[Role]:
roles = db.execute(select(Role).order_by(Role.name)).scalars().all() if scope.is_system_admin else get_manageable_roles(db, scope)
query = (q or '').strip().lower()
if not query:
return roles
return [r for r in roles if query in (r.name or '').lower()]
def permissions_by_role(db: Session, roles: list[Role]) -> dict[int, list[str]]:
return {
role.id: [code for (code,) in db.execute(
select(Permission.code)
.join(RolePermission, RolePermission.permission_id == Permission.id)
.where(RolePermission.role_id == role.id)
.order_by(Permission.code)
).all()]
for role in roles
}
def build_roles_payload(db: Session, scope: UserScope, q: str | None = None, page: int = 1, per_page: int = 10) -> dict:
roles = search_roles(db, scope, q=q)
paged = paginate_list(roles, page=page, per_page=per_page)
return {
'roles': paged.items,
'roles_page': paged,
'permissions_by_role': permissions_by_role(db, paged.items),
'filters': {'q': (q or '').strip(), 'per_page': paged.per_page},
}
def search_permissions(db: Session, q: str | None = None) -> list[Permission]:
permissions = db.execute(select(Permission).order_by(Permission.code)).scalars().all()
query = (q or '').strip().lower()
if not query:
return permissions
return [p for p in permissions if query in (p.code or '').lower() or query in (p.name or '').lower()]
def build_permissions_payload(db: Session, q: str | None = None, page: int = 1, per_page: int = 15) -> dict:
permissions = search_permissions(db, q=q)
paged = paginate_list(permissions, page=page, per_page=per_page)
return {'permissions': paged.items, 'permissions_page': paged, 'filters': {'q': (q or '').strip(), 'per_page': paged.per_page}}
@@ -0,0 +1,21 @@
{% extends "ui/templates/base/layout.html" %}
{% import "ui/templates/components/macros.html" as ui %}
{% block content %}
<div class="grid gap-6 lg:grid-cols-[minmax(0,1fr)_360px]">
<div class="rounded-3xl bg-white p-6 shadow-soft">
<h2 class="text-xl font-semibold">Permissions</h2>
<p class="mt-1 text-sm text-slate-500">Searchable permission registry used by API guards and UI gate checks.</p>
<div class="mt-6 overflow-hidden rounded-2xl border border-slate-200">
{{ ui.search_bar('/system-settings/rbac/permissions', filters.q, filters.per_page) }}
{% if permissions %}
<table class="min-w-full divide-y divide-slate-200 text-sm">
<thead class="bg-slate-50"><tr><th class="px-4 py-3 text-left font-semibold text-slate-600">Code</th><th class="px-4 py-3 text-left font-semibold text-slate-600">Name</th><th class="px-4 py-3 text-left font-semibold text-slate-600">Status</th></tr></thead>
<tbody class="divide-y divide-slate-100">{% for permission in permissions %}<tr><td class="px-4 py-4 font-mono text-xs text-slate-700">{{ permission.code }}</td><td class="px-4 py-4 text-slate-700">{{ permission.name }}</td><td class="px-4 py-4">{{ ui.badge('Active','emerald') if permission.is_active else ui.badge('Inactive','rose') }}</td></tr>{% endfor %}</tbody>
</table>
{{ ui.pagination(permissions_page, '/system-settings/rbac/permissions', request.url.query) }}
{% else %}<div class="p-6">{{ ui.empty_state('No permissions matched the current filter.') }}</div>{% endif %}
</div>
</div>
<div class="rounded-3xl bg-white p-6 shadow-soft"><h3 class="text-lg font-semibold">Create Permission</h3>{% if "rbac.manage" in current_user_permissions and "System Admin" in current_user_roles %}<form method="post" action="/system-settings/rbac/permissions/new" class="mt-5 space-y-4"><input type="hidden" name="csrf_token" value="{{ csrf_token }}" /><div><label class="mb-2 block text-sm font-medium text-slate-700">Permission Code</label><input type="text" name="code" class="w-full rounded-2xl border border-slate-300 px-4 py-2.5 text-sm" placeholder="Ex: clients.manage" required /></div><div><label class="mb-2 block text-sm font-medium text-slate-700">Display Name</label><input type="text" name="name" class="w-full rounded-2xl border border-slate-300 px-4 py-2.5 text-sm" placeholder="Ex: Manage Clients" required /></div><label class="inline-flex items-center gap-3 rounded-2xl border border-slate-200 px-4 py-3 text-sm font-medium text-slate-700"><input type="checkbox" name="is_active" checked class="h-4 w-4 rounded border-slate-300" />Permission is active</label><button type="submit" class="w-full rounded-xl bg-brand-600 px-4 py-2.5 text-sm font-medium text-white hover:bg-brand-700">Create Permission</button></form>{% else %}<div class="mt-5">{{ ui.alert('Only System Admin can create permissions.', 'amber') }}</div>{% endif %}</div>
</div>
{% endblock %}
@@ -0,0 +1,31 @@
{% extends "ui/templates/base/layout.html" %}
{% block content %}
<div class="rounded-3xl bg-white p-6 shadow-soft">
<div class="flex flex-col gap-4 sm:flex-row sm:items-center sm:justify-between">
<div>
<h2 class="text-xl font-semibold">Configure Role Permissions</h2>
<p class="mt-1 text-sm text-slate-500">Role: <span class="font-medium text-slate-800">{{ role.name }}</span></p>
</div>
<a href="/system-settings/rbac/roles" class="rounded-xl border border-slate-300 px-4 py-2 text-sm font-medium text-slate-700 hover:bg-slate-50">Back to roles</a>
</div>
<form method="post" class="mt-6 space-y-6">
<input type="hidden" name="csrf_token" value="{{ csrf_token }}" />
<div class="grid gap-3 md:grid-cols-2 xl:grid-cols-3">
{% for permission in permissions %}
<label class="flex items-start gap-3 rounded-2xl border border-slate-200 p-4 hover:border-brand-300 hover:bg-brand-50/50">
<input type="checkbox" name="permission_ids" value="{{ permission.id }}" {% if permission.id in assigned_permission_ids %}checked{% endif %} class="mt-1 h-4 w-4 rounded border-slate-300" />
<span>
<span class="block font-mono text-xs text-brand-700">{{ permission.code }}</span>
<span class="mt-1 block text-sm font-medium text-slate-800">{{ permission.name }}</span>
</span>
</label>
{% endfor %}
</div>
<div class="flex justify-end border-t border-slate-200 pt-5">
{% if "rbac.manage" in current_user_permissions and "System Admin" in current_user_roles %}<button type="submit" class="rounded-xl bg-brand-600 px-4 py-2.5 text-sm font-medium text-white hover:bg-brand-700">Save Role Permissions</button>{% else %}<span class="text-sm text-slate-500">View-only access</span>{% endif %}
</div>
</form>
</div>
{% endblock %}
@@ -0,0 +1,23 @@
{% extends "ui/templates/base/layout.html" %}
{% import "ui/templates/components/macros.html" as ui %}
{% block content %}
<div class="grid gap-6 lg:grid-cols-[minmax(0,1fr)_360px]">
<div class="rounded-3xl bg-white p-6 shadow-soft">
<div class="flex items-center justify-between gap-4">
<div><h2 class="text-xl font-semibold">Roles</h2><p class="mt-1 text-sm text-slate-500">Core RBAC matrix with search and pagination for safer maintenance.</p></div>
<a href="/system-settings/rbac/permissions" class="rounded-xl border border-slate-300 px-4 py-2 text-sm font-medium text-slate-700 hover:bg-slate-50">View permissions</a>
</div>
<div class="mt-6 overflow-hidden rounded-2xl border border-slate-200">
{{ ui.search_bar('/system-settings/rbac/roles', filters.q, filters.per_page) }}
{% if roles %}
<table class="min-w-full divide-y divide-slate-200 text-sm">
<thead class="bg-slate-50"><tr><th class="px-4 py-3 text-left font-semibold text-slate-600">Role</th><th class="px-4 py-3 text-left font-semibold text-slate-600">Permissions</th><th class="px-4 py-3 text-right font-semibold text-slate-600">Action</th></tr></thead>
<tbody class="divide-y divide-slate-100">{% for role in roles %}<tr><td class="px-4 py-4"><div class="font-medium text-slate-900">{{ role.name }}</div><div class="text-xs text-slate-500">{{ 'Active' if role.is_active else 'Inactive' }}</div></td><td class="px-4 py-4"><div class="flex flex-wrap gap-2">{% for code in permissions_by_role.get(role.id, []) %}{{ ui.badge(code) }}{% else %}<span class="text-slate-400">No permissions</span>{% endfor %}</div></td><td class="px-4 py-4 text-right"><a href="/system-settings/rbac/roles/{{ role.id }}" class="rounded-xl border border-slate-300 px-3 py-2 text-xs font-medium text-slate-700 hover:bg-slate-50">{{ "Configure" if "rbac.manage" in current_user_permissions else "View" }}</a></td></tr>{% endfor %}</tbody>
</table>
{{ ui.pagination(roles_page, '/system-settings/rbac/roles', request.url.query) }}
{% else %}<div class="p-6">{{ ui.empty_state('No roles matched the current filter.') }}</div>{% endif %}
</div>
</div>
<div class="rounded-3xl bg-white p-6 shadow-soft"><h3 class="text-lg font-semibold">Create Role</h3>{% if "rbac.manage" in current_user_permissions and "System Admin" in current_user_roles %}<form method="post" action="/system-settings/rbac/roles/new" class="mt-5 space-y-4"><input type="hidden" name="csrf_token" value="{{ csrf_token }}" /><div><label class="mb-2 block text-sm font-medium text-slate-700">Role Name</label><input type="text" name="name" class="w-full rounded-2xl border border-slate-300 px-4 py-2.5 text-sm" placeholder="Ex: ComplianceAdmin" required /></div><label class="inline-flex items-center gap-3 rounded-2xl border border-slate-200 px-4 py-3 text-sm font-medium text-slate-700"><input type="checkbox" name="is_active" checked class="h-4 w-4 rounded border-slate-300" />Role is active</label><button type="submit" class="w-full rounded-xl bg-brand-600 px-4 py-2.5 text-sm font-medium text-white hover:bg-brand-700">Create Role</button></form>{% else %}<div class="mt-5">{{ ui.alert('Only System Admin can create roles.', 'amber') }}</div>{% endif %}</div>
</div>
{% endblock %}
+267
View File
@@ -0,0 +1,267 @@
from __future__ import annotations
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.session_auth import get_current_user
from app.core.templating import templates
from app.modules.core.audit.service import write_audit_log
from app.modules.core.iam.scope import build_scope, assert_can_manage_role_object
from app.modules.core.rbac.deps import get_user_permissions, get_user_roles
from app.modules.core.rbac.models import Permission, Role, RolePermission
from app.modules.core.rbac.services import build_permissions_payload, build_roles_payload
router = APIRouter(prefix="/system-settings/rbac", tags=["rbac-ui"])
def _redirect_login():
return RedirectResponse(url="/login", status_code=303)
def _redirect_denied():
return RedirectResponse(url="/system-settings", status_code=303)
def _is_system_admin(db, current_user) -> bool:
return "System Admin" in get_user_roles(db, current_user.id)
def _base_ctx(request: Request, current_user, db, **ctx):
base = {
"request": request,
"current_user": current_user,
"current_user_roles": get_user_roles(db, current_user.id),
"current_user_permissions": get_user_permissions(db, current_user.id),
"csrf_token": get_or_create_csrf_token(request),
}
base.update(ctx)
return base
@router.get("")
def rbac_dashboard(request: Request):
return RedirectResponse(url="/system-settings/rbac/roles", status_code=303)
@router.get("/roles")
def roles_list(request: Request, q: str = "", page: int = 1, per_page: int = 10):
db = CommonSessionLocal()
try:
current_user = get_current_user(request, db=db)
if not current_user:
return _redirect_login()
if not _is_system_admin(db, current_user):
return _redirect_denied()
scope = build_scope(db, current_user)
payload = build_roles_payload(db, scope, q=q, page=page, per_page=per_page)
return templates.TemplateResponse(
"modules/core/rbac/templates/roles_list.html",
_base_ctx(request, current_user, db, title="RBAC Roles", **payload),
)
finally:
db.close()
@router.post("/roles/new")
def role_create_submit(
request: Request,
name: str = Form(...),
is_active: str | None = Form(None),
csrf_token: str = Form(...),
):
validate_csrf(request, csrf_token)
db = CommonSessionLocal()
try:
current_user = get_current_user(request, db=db)
if not current_user:
return _redirect_login()
if not _is_system_admin(db, current_user):
return _redirect_denied()
name = name.strip()
if name and not db.execute(select(Role).where(Role.name == name)).scalar_one_or_none():
role = Role(name=name, is_active=is_active is not None)
db.add(role)
db.commit()
db.refresh(role)
write_audit_log(
db,
action="role.create",
entity_type="role",
actor=current_user,
request=request,
entity_id=role.id,
entity_name=role.name,
target_tenant_id=current_user.tenant_id,
target_branch_id=current_user.branch_id,
details={"name": role.name, "is_active": role.is_active},
)
return RedirectResponse(url="/system-settings/rbac/roles", status_code=303)
finally:
db.close()
@router.get("/roles/{role_id}")
def role_permissions_edit(request: Request, role_id: int):
db = CommonSessionLocal()
try:
current_user = get_current_user(request, db=db)
if not current_user:
return _redirect_login()
if not _is_system_admin(db, current_user):
return _redirect_denied()
role = db.execute(select(Role).where(Role.id == role_id)).scalar_one_or_none()
if not role:
return RedirectResponse(url="/system-settings/rbac/roles", status_code=303)
scope = build_scope(db, current_user)
try:
assert_can_manage_role_object(scope, role)
except Exception:
return RedirectResponse(url="/system-settings/rbac/roles", status_code=303)
permissions = db.execute(select(Permission).order_by(Permission.code)).scalars().all()
assigned_permission_ids = db.execute(
select(RolePermission.permission_id).where(RolePermission.role_id == role_id)
).scalars().all()
return templates.TemplateResponse(
"modules/core/rbac/templates/role_permissions_form.html",
_base_ctx(
request,
current_user,
db,
title=f"Role Permissions - {role.name}",
role=role,
permissions=permissions,
assigned_permission_ids=list(assigned_permission_ids),
),
)
finally:
db.close()
@router.post("/roles/{role_id}")
def role_permissions_submit(
request: Request,
role_id: int,
permission_ids: list[int] = Form([]),
csrf_token: str = Form(...),
):
validate_csrf(request, csrf_token)
db = CommonSessionLocal()
try:
current_user = get_current_user(request, db=db)
if not current_user:
return _redirect_login()
if not _is_system_admin(db, current_user):
return _redirect_denied()
role = db.execute(select(Role).where(Role.id == role_id)).scalar_one_or_none()
if not role:
return RedirectResponse(url="/system-settings/rbac/roles", status_code=303)
old_permission_ids = db.execute(
select(RolePermission.permission_id).where(RolePermission.role_id == role.id)
).scalars().all()
db.execute(RolePermission.__table__.delete().where(RolePermission.role_id == role.id))
for permission_id in permission_ids:
db.add(RolePermission(role_id=role.id, permission_id=permission_id))
db.commit()
write_audit_log(
db,
action="role.permissions.update",
entity_type="role",
actor=current_user,
request=request,
entity_id=role.id,
entity_name=role.name,
target_tenant_id=current_user.tenant_id,
target_branch_id=current_user.branch_id,
details={
"old_permission_ids": list(old_permission_ids),
"new_permission_ids": list(permission_ids),
},
)
return RedirectResponse(url="/system-settings/rbac/roles", status_code=303)
finally:
db.close()
@router.get("/permissions")
def permissions_list(request: Request, q: str = "", page: int = 1, per_page: int = 15):
db = CommonSessionLocal()
try:
current_user = get_current_user(request, db=db)
if not current_user:
return _redirect_login()
if not _is_system_admin(db, current_user):
return _redirect_denied()
payload = build_permissions_payload(db, q=q, page=page, per_page=per_page)
return templates.TemplateResponse(
"modules/core/rbac/templates/permissions_list.html",
_base_ctx(request, current_user, db, title="RBAC Permissions", **payload),
)
finally:
db.close()
@router.post("/permissions/new")
def permission_create_submit(
request: Request,
code: str = Form(...),
name: str = Form(...),
is_active: str | None = Form(None),
csrf_token: str = Form(...),
):
validate_csrf(request, csrf_token)
db = CommonSessionLocal()
try:
current_user = get_current_user(request, db=db)
if not current_user:
return _redirect_login()
if not _is_system_admin(db, current_user):
return _redirect_denied()
code = code.strip()
if code and not db.execute(select(Permission).where(Permission.code == code)).scalar_one_or_none():
permission = Permission(code=code, name=name.strip(), is_active=is_active is not None)
db.add(permission)
db.commit()
db.refresh(permission)
write_audit_log(
db,
action="permission.create",
entity_type="permission",
actor=current_user,
request=request,
entity_id=permission.id,
entity_name=permission.code,
target_tenant_id=current_user.tenant_id,
target_branch_id=current_user.branch_id,
details={"code": permission.code, "name": permission.name, "is_active": permission.is_active},
)
return RedirectResponse(url="/system-settings/rbac/permissions", status_code=303)
finally:
db.close()
+493
View File
@@ -0,0 +1,493 @@
from __future__ import annotations
from app.modules.core.rbac.permissions_registry import expand_permission_codes
def _has(permissions: list[str] | None, code: str) -> bool:
granted = set(permissions or [])
for candidate in expand_permission_codes(code):
if candidate in granted:
return True
return False
def _roles(role_names: list[str] | None) -> set[str]:
return set(role_names or [])
def is_system_admin(role_names: list[str] | None = None) -> bool:
return "System Admin" in _roles(role_names)
def is_firm_admin(role_names: list[str] | None = None) -> bool:
return "Firm Admin" in _roles(role_names)
def can_view_own_alerts(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "alerts.view_self") or bool(user)
def can_manage_alerts(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "alerts.manage") and (is_system_admin(role_names) or is_firm_admin(role_names) or "Partner" in _roles(role_names) or "Branch Manager" in _roles(role_names))
def can_view_users(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "users.view")
def can_manage_users(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "users.manage") and (
is_system_admin(role_names) or is_firm_admin(role_names)
)
def can_view_settings(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "system.settings.view")
def can_edit_settings(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "system.settings.edit")
def can_manage_settings(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return is_system_admin(role_names)
def can_view_tenants(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return is_system_admin(role_names)
def can_manage_tenants(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return is_system_admin(role_names)
def can_view_branches(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return is_system_admin(role_names) or is_firm_admin(role_names)
def can_manage_branches(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return is_system_admin(role_names) or is_firm_admin(role_names)
def can_change_branch_tenant(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return is_system_admin(role_names)
def can_view_rbac(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return is_system_admin(role_names)
def can_manage_rbac(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return is_system_admin(role_names)
def can_view_audit(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "audit.view") and (
is_system_admin(role_names) or is_firm_admin(role_names)
)
def can_view_services(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "services.view")
def can_manage_services(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "services.create") or _has(permissions, "services.edit")
def can_manage_service_tasks(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "service_tasks.create") or _has(permissions, "service_tasks.edit")
def can_import_services(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "services.import")
def can_import_service_tasks(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "service_tasks.import")
def can_switch_service_tenant(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return is_system_admin(role_names) and _has(permissions, "services.cross_tenant")
def can_switch_service_branch(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "services.cross_branch") or can_switch_service_tenant(
user, permissions, role_names
)
def can_view_clients(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "clients.view")
def can_manage_clients(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "clients.create") or _has(permissions, "clients.edit")
def can_export_clients(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "clients.export")
def can_switch_client_tenant(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return is_system_admin(role_names) and _has(permissions, "clients.cross_tenant")
def can_switch_client_branch(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "clients.cross_branch") or can_switch_client_tenant(
user, permissions, role_names
)
def can_view_consultants(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "consultants.view")
def can_manage_consultants(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "consultants.manage") and (
is_system_admin(role_names) or is_firm_admin(role_names)
)
def can_link_consultant_clients(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "consultants.link_clients")
def can_manage_consultant_service_requests(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "consultants.service_requests.manage") and (
is_system_admin(role_names) or is_firm_admin(role_names)
)
def can_manage_consultant_conversions(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "consultants.conversions.manage") and (
is_system_admin(role_names) or is_firm_admin(role_names)
)
def can_view_consultant_portal(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "consultants.portal.view") or "Consultant" in _roles(role_names)
def can_manage_own_consultant_workspace(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return can_view_consultant_portal(user, permissions, role_names) and _has(permissions, "consultants.workspace.manage")
def can_view_employee_dashboard(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "employees.dashboard.view") and (
is_system_admin(role_names) or is_firm_admin(role_names) or "Partner" in _roles(role_names) or "Branch Manager" in _roles(role_names)
)
def can_view_employees(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "employees.view")
def can_manage_employees(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "employees.create") or _has(permissions, "employees.edit")
def can_change_employee_status(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "employees.status")
def can_switch_employee_tenant(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return is_system_admin(role_names) and _has(permissions, "employees.cross_tenant")
def can_switch_employee_branch(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "employees.cross_branch") or can_switch_employee_tenant(user, permissions, role_names)
def can_view_employee_portal(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "employees.ess.view") or bool({"System Admin", "Firm Admin", "Partner", "Branch Manager", "Staff"}.intersection(_roles(role_names)))
def can_edit_own_employee_profile(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return can_view_employee_portal(user, permissions, role_names) and _has(permissions, "employees.ess.profile.edit")
def can_view_own_employee_work(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "employees.work.view_self") or can_view_employee_portal(user, permissions, role_names)
def can_request_employee_registration(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "employees.registration.request") or can_view_employee_portal(user, permissions, role_names)
def can_approve_employee_registrations(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "employees.registration.approve") and (
is_system_admin(role_names) or is_firm_admin(role_names) or "Partner" in _roles(role_names) or "Branch Manager" in _roles(role_names)
)
def can_punch_employee_attendance(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "employees.attendance.punch") or can_view_employee_portal(user, permissions, role_names)
def can_view_own_employee_attendance(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "employees.attendance.view_self") or can_view_employee_portal(user, permissions, role_names)
def can_view_all_employee_attendance(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "employees.attendance.view_all") and (
is_system_admin(role_names) or is_firm_admin(role_names) or "Partner" in _roles(role_names) or "Branch Manager" in _roles(role_names)
)
def can_approve_employee_attendance(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "employees.attendance.approve") and (
is_system_admin(role_names) or is_firm_admin(role_names) or "Partner" in _roles(role_names) or "Branch Manager" in _roles(role_names)
)
def can_apply_employee_leave(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "employees.leave.apply") or can_view_employee_portal(user, permissions, role_names)
def can_view_own_employee_leave(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "employees.leave.view_self") or can_view_employee_portal(user, permissions, role_names)
def can_view_all_employee_leave(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "employees.leave.view_all") and (
is_system_admin(role_names) or is_firm_admin(role_names) or "Partner" in _roles(role_names) or "Branch Manager" in _roles(role_names)
)
def can_approve_employee_leave(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "employees.leave.approve") and (
is_system_admin(role_names) or is_firm_admin(role_names) or "Partner" in _roles(role_names) or "Branch Manager" in _roles(role_names)
)
def can_manage_employee_leave_types(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "employees.leave_type.manage") and (
is_system_admin(role_names) or is_firm_admin(role_names) or "Partner" in _roles(role_names) or "Branch Manager" in _roles(role_names)
)
def can_manage_employee_leave_balances(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "employees.leave_balance.manage") and (
is_system_admin(role_names) or is_firm_admin(role_names) or "Partner" in _roles(role_names) or "Branch Manager" in _roles(role_names)
)
def can_view_own_employee_documents(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "employees.documents.view_self") or can_view_employee_portal(user, permissions, role_names)
def can_upload_own_employee_documents(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "employees.documents.upload_self") or can_view_employee_portal(user, permissions, role_names)
def can_view_all_employee_documents(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "employees.documents.view_all") and (
is_system_admin(role_names) or is_firm_admin(role_names) or "Partner" in _roles(role_names) or "Branch Manager" in _roles(role_names)
)
def can_manage_employee_documents(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "employees.documents.manage") and (
is_system_admin(role_names) or is_firm_admin(role_names) or "Partner" in _roles(role_names) or "Branch Manager" in _roles(role_names)
)
def can_verify_employee_documents(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "employees.documents.verify") and (
is_system_admin(role_names) or is_firm_admin(role_names) or "Partner" in _roles(role_names) or "Branch Manager" in _roles(role_names)
)
def can_manage_employee_document_types(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "employees.document_type.manage") and (
is_system_admin(role_names) or is_firm_admin(role_names) or "Partner" in _roles(role_names) or "Branch Manager" in _roles(role_names)
)
def can_view_employee_onboarding(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "employees.onboarding.view") and (
is_system_admin(role_names) or is_firm_admin(role_names) or "Partner" in _roles(role_names) or "Branch Manager" in _roles(role_names)
)
def can_manage_employee_onboarding(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "employees.onboarding.manage") and (
is_system_admin(role_names) or is_firm_admin(role_names) or "Partner" in _roles(role_names) or "Branch Manager" in _roles(role_names)
)
def can_approve_employee_onboarding(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "employees.onboarding.approve") and (
is_system_admin(role_names) or is_firm_admin(role_names) or "Partner" in _roles(role_names) or "Branch Manager" in _roles(role_names)
)
def can_view_employee_offboarding(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "employees.offboarding.view") and (
is_system_admin(role_names) or is_firm_admin(role_names) or "Partner" in _roles(role_names) or "Branch Manager" in _roles(role_names)
)
def can_manage_employee_offboarding(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "employees.offboarding.manage") and (
is_system_admin(role_names) or is_firm_admin(role_names) or "Partner" in _roles(role_names) or "Branch Manager" in _roles(role_names)
)
def can_approve_employee_offboarding(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "employees.offboarding.approve") and (
is_system_admin(role_names) or is_firm_admin(role_names) or "Partner" in _roles(role_names) or "Branch Manager" in _roles(role_names)
)
def can_request_own_employee_offboarding(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "employees.offboarding.request_self") or can_view_employee_portal(user, permissions, role_names)
def can_import_employee_hr(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "employees.import") and (
is_system_admin(role_names) or is_firm_admin(role_names) or "Partner" in _roles(role_names) or "Branch Manager" in _roles(role_names)
)
def can_manage_employee_payroll_structures(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "employees.payroll.structure.manage") and (
is_system_admin(role_names) or is_firm_admin(role_names) or "Partner" in _roles(role_names) or "Branch Manager" in _roles(role_names)
)
def can_run_employee_payroll(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "employees.payroll.run") and (
is_system_admin(role_names) or is_firm_admin(role_names) or "Partner" in _roles(role_names) or "Branch Manager" in _roles(role_names)
)
def can_view_employee_payroll(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "employees.payroll.view") and (
is_system_admin(role_names) or is_firm_admin(role_names) or "Partner" in _roles(role_names) or "Branch Manager" in _roles(role_names)
)
def can_view_own_employee_payslips(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "employees.payroll.view_self") or can_view_employee_portal(user, permissions, role_names)
def can_approve_employee_payroll(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "employees.payroll.payout") and (
is_system_admin(role_names) or is_firm_admin(role_names) or "Partner" in _roles(role_names)
)
def can_view_employee_progress(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "employees.progress.view") and (
is_system_admin(role_names) or is_firm_admin(role_names) or "Partner" in _roles(role_names) or "Branch Manager" in _roles(role_names)
)
def can_manage_employee_work(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "employees.work.manage") and (
is_system_admin(role_names) or is_firm_admin(role_names) or "Partner" in _roles(role_names) or "Branch Manager" in _roles(role_names)
)
def can_view_billing(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "billing.view")
def can_create_billing(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "billing.create")
def can_import_billing_fee_structure(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "billing_fee_structure.import")
def can_view_billing_fee_structure(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "billing_fee_structure.view")
def can_generate_billing_invoices(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "billing_invoice.generate") or _has(permissions, "billing_invoice.bulk_generate")
def can_view_platform_billing(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return is_system_admin(role_names) and _has(permissions, "platform_billing.view")
def can_manage_platform_billing(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return is_system_admin(role_names) and (_has(permissions, "platform_billing.create") or _has(permissions, "platform_billing.edit"))
def can_generate_platform_billing(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return is_system_admin(role_names) and _has(permissions, "platform_billing.generate")
def can_manage_platform_plans(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return is_system_admin(role_names) and _has(permissions, "platform_plans.manage")
def can_manage_platform_subscriptions(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return is_system_admin(role_names) and _has(permissions, "platform_subscriptions.manage")
def can_view_marketplace_leads(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "marketplace_leads.view") or _has(permissions, "marketplace_leads.view_assigned")
def can_create_marketplace_leads(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return is_system_admin(role_names) and _has(permissions, "marketplace_leads.create")
def can_assign_marketplace_leads(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return is_system_admin(role_names) and _has(permissions, "marketplace_leads.assign")
def can_update_marketplace_leads(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "marketplace_leads.update")
def can_convert_marketplace_leads(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "marketplace_leads.convert")
def can_view_documents(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "documents.view")
def can_upload_documents(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "documents.upload")
def can_download_documents(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "documents.download")
def can_delete_documents(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "documents.delete") and (is_firm_admin(role_names) or "Partner" in _roles(role_names))
def can_view_notice_cases(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "notice_cases.view")
def can_manage_notice_cases(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "notice_cases.create") or _has(permissions, "notice_cases.edit")
def can_upload_notice_case_documents(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "notice_cases.documents.upload")
def can_download_notice_case_documents(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "notice_cases.documents.download")
def can_delete_notice_case_documents(user=None, permissions: list[str] | None = None, role_names: list[str] | None = None) -> bool:
return _has(permissions, "notice_cases.documents.delete")