242 lines
12 KiB
Python
242 lines
12 KiB
Python
from fastapi import HTTPException
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from role.models import PermissionTags, Permissions, Roles
|
|
from role.serializers import serialize_permission, serialize_permission_tag, serialize_role
|
|
|
|
|
|
class Role:
|
|
def __init__(self, session: AsyncSession):
|
|
self.session = session
|
|
|
|
async def _bundle_payload(self, permission: Permissions) -> dict:
|
|
tag_ids = [int(i) for i in (permission.permission_tags or [])]
|
|
tags = await PermissionTags.get_permission_tags_by_ids(self.session, tag_ids)
|
|
by_id = {t.id: t.tag_name for t in tags}
|
|
tag_names = [by_id[i] for i in tag_ids if i in by_id]
|
|
return serialize_permission(permission, tag_names=tag_names)
|
|
|
|
async def _role_payload(self, role: Roles) -> dict:
|
|
perm_ids = [int(i) for i in (role.permissions or [])]
|
|
bundles_orm = await Permissions.get_permissions_by_ids(self.session, perm_ids)
|
|
by_id = {b.id: b for b in bundles_orm}
|
|
bundles = []
|
|
for pid in perm_ids:
|
|
bundle = by_id.get(pid)
|
|
if bundle is not None:
|
|
bundles.append(await self._bundle_payload(bundle))
|
|
tags = await Roles.resolve_tags(self.session, role)
|
|
return serialize_role(role, bundles=bundles, permissions=list(tags))
|
|
|
|
async def get_roles(self, top, skip, search=None):
|
|
rows = await Roles.get_roles(self.session, top, skip, search)
|
|
return [await self._role_payload(r) for r in rows]
|
|
|
|
async def get_role_by_id(self, record_id):
|
|
role = await Roles.get_role_by_id(self.session, int(record_id))
|
|
if not role or role.is_deleted:
|
|
raise HTTPException(status_code=404, detail="Role not found")
|
|
return await self._role_payload(role)
|
|
|
|
async def count_roles(self, search=None):
|
|
return await Roles.count_roles(self.session, search)
|
|
|
|
async def create_role(self, payload):
|
|
name = (payload.get("role_name") or "").strip()
|
|
if not name:
|
|
raise HTTPException(status_code=400, detail="role_name is required")
|
|
if await Roles.get_role_by_name(self.session, name):
|
|
raise HTTPException(status_code=409, detail="Role name already exists")
|
|
fields = {
|
|
"role_name": name,
|
|
"description": payload.get("description"),
|
|
"permissions": list(payload.get("permissions") or []),
|
|
"is_system": False,
|
|
"is_active": payload.get("is_active", True),
|
|
"is_deleted": False,
|
|
}
|
|
role = await Roles.insert_role(self.session, fields)
|
|
return await self._role_payload(role)
|
|
|
|
async def update_role(self, record_id, payload):
|
|
role = await Roles.get_role_by_id(self.session, int(record_id))
|
|
if not role or role.is_deleted:
|
|
raise HTTPException(status_code=404, detail="Role not found")
|
|
fields = {}
|
|
if "role_name" in payload and payload["role_name"] is not None:
|
|
new_name = payload["role_name"].strip()
|
|
if role.is_system and new_name != role.role_name:
|
|
raise HTTPException(status_code=409, detail="System roles cannot be renamed")
|
|
if new_name != role.role_name:
|
|
clash = await Roles.get_role_by_name(self.session, new_name)
|
|
if clash:
|
|
raise HTTPException(status_code=409, detail="Role name already exists")
|
|
fields["role_name"] = new_name
|
|
if "description" in payload and payload["description"] is not None:
|
|
fields["description"] = payload["description"]
|
|
if "permissions" in payload and payload["permissions"] is not None:
|
|
fields["permissions"] = list(payload["permissions"])
|
|
if "is_active" in payload and payload["is_active"] is not None:
|
|
fields["is_active"] = payload["is_active"]
|
|
updated = await Roles.update_role(self.session, int(record_id), fields)
|
|
return await self._role_payload(updated)
|
|
|
|
async def set_role_matrix(self, record_id, permission_tags):
|
|
"""Write the Access Control grid onto one overlay bundle for this role.
|
|
|
|
Shared system bundles are not mutated. The role then points at that
|
|
overlay only, so a ticked cell is the grant and an unticked cell is not.
|
|
"""
|
|
role = await Roles.get_role_by_id(self.session, int(record_id))
|
|
if not role or role.is_deleted:
|
|
raise HTTPException(status_code=404, detail="Role not found")
|
|
tag_ids = sorted({int(i) for i in (permission_tags or [])})
|
|
found = await PermissionTags.get_permission_tags_by_ids(self.session, tag_ids)
|
|
unknown = sorted(set(tag_ids) - {t.id for t in found})
|
|
if unknown:
|
|
raise HTTPException(
|
|
status_code=422,
|
|
detail=f"Unknown or inactive permission tag ids: {unknown}",
|
|
)
|
|
overlay_name = f"role_{role.id}_matrix"
|
|
bundle = await Permissions.get_permission_by_name(self.session, overlay_name)
|
|
if bundle is None:
|
|
bundle = await Permissions.insert_permission(
|
|
self.session,
|
|
{
|
|
"name": overlay_name,
|
|
"description": f"Access Control matrix for {role.role_name}",
|
|
"permission_tags": tag_ids,
|
|
"is_system": False,
|
|
"is_active": True,
|
|
"is_deleted": False,
|
|
},
|
|
)
|
|
else:
|
|
await Permissions.update_permission(
|
|
self.session, int(bundle.id), {"permission_tags": tag_ids},
|
|
)
|
|
updated = await Roles.update_role(
|
|
self.session, int(record_id), {"permissions": [int(bundle.id)]},
|
|
)
|
|
return await self._role_payload(updated)
|
|
|
|
async def delete_role(self, record_id):
|
|
role = await Roles.get_role_by_id(self.session, int(record_id))
|
|
if not role or role.is_deleted:
|
|
raise HTTPException(status_code=404, detail="Role not found")
|
|
if role.is_system:
|
|
raise HTTPException(status_code=409, detail="System roles cannot be deleted")
|
|
deleted = await Roles.soft_delete_role(self.session, int(record_id))
|
|
return await self._role_payload(deleted)
|
|
|
|
async def get_permissions(self, top, skip, search=None):
|
|
rows = await Permissions.get_permissions(self.session, top, skip, search)
|
|
return [await self._bundle_payload(r) for r in rows]
|
|
|
|
async def get_permission_by_id(self, record_id):
|
|
row = await Permissions.get_permission_by_id(self.session, int(record_id))
|
|
if not row or row.is_deleted:
|
|
raise HTTPException(status_code=404, detail="Permission bundle not found")
|
|
return await self._bundle_payload(row)
|
|
|
|
async def count_permissions(self, search=None):
|
|
return await Permissions.count_permissions(self.session, search)
|
|
|
|
async def create_permission(self, payload):
|
|
name = (payload.get("name") or "").strip()
|
|
if not name:
|
|
raise HTTPException(status_code=400, detail="name is required")
|
|
if await Permissions.get_permission_by_name(self.session, name):
|
|
raise HTTPException(status_code=409, detail="Permission bundle name already exists")
|
|
fields = {
|
|
"name": name,
|
|
"description": payload.get("description"),
|
|
"permission_tags": list(payload.get("permission_tags") or []),
|
|
"is_system": False,
|
|
"is_active": payload.get("is_active", True),
|
|
"is_deleted": False,
|
|
}
|
|
row = await Permissions.insert_permission(self.session, fields)
|
|
return await self._bundle_payload(row)
|
|
|
|
async def update_permission(self, record_id, payload):
|
|
row = await Permissions.get_permission_by_id(self.session, int(record_id))
|
|
if not row or row.is_deleted:
|
|
raise HTTPException(status_code=404, detail="Permission bundle not found")
|
|
fields = {}
|
|
if "name" in payload and payload["name"] is not None:
|
|
new_name = payload["name"].strip()
|
|
if row.is_system and new_name != row.name:
|
|
raise HTTPException(
|
|
status_code=409, detail="System permission bundles cannot be renamed"
|
|
)
|
|
if new_name != row.name:
|
|
clash = await Permissions.get_permission_by_name(self.session, new_name)
|
|
if clash:
|
|
raise HTTPException(
|
|
status_code=409, detail="Permission bundle name already exists"
|
|
)
|
|
fields["name"] = new_name
|
|
if "description" in payload and payload["description"] is not None:
|
|
fields["description"] = payload["description"]
|
|
if "permission_tags" in payload and payload["permission_tags"] is not None:
|
|
fields["permission_tags"] = list(payload["permission_tags"])
|
|
if "is_active" in payload and payload["is_active"] is not None:
|
|
fields["is_active"] = payload["is_active"]
|
|
updated = await Permissions.update_permission(self.session, int(record_id), fields)
|
|
return await self._bundle_payload(updated)
|
|
|
|
async def get_permission_tags(self, top, skip, search=None):
|
|
rows = await PermissionTags.get_permission_tags(self.session, top, skip, search)
|
|
return [serialize_permission_tag(r) for r in rows]
|
|
|
|
async def get_permission_tag_by_id(self, record_id):
|
|
row = await PermissionTags.get_permission_tag_by_id(self.session, int(record_id))
|
|
if not row or row.is_deleted:
|
|
raise HTTPException(status_code=404, detail="Permission tag not found")
|
|
return serialize_permission_tag(row)
|
|
|
|
async def count_permission_tags(self, search=None):
|
|
return await PermissionTags.count_permission_tags(self.session, search)
|
|
|
|
|
|
class PermissionBundle:
|
|
"""Sets the exact permission-tag set on one bundle, for the permission matrix.
|
|
|
|
`id` is a permissions.id from /permissions/fetch. Every role holding that bundle
|
|
picks the change up on its next request, because Roles.resolve_tags runs live.
|
|
"""
|
|
|
|
def __init__(self,session:AsyncSession):
|
|
self.session=session
|
|
|
|
async def set_role_tags(self,payload):
|
|
bundle=await Permissions.get_permission_by_id(self.session,int(payload.get("id")))
|
|
if not bundle or bundle.is_deleted:
|
|
raise HTTPException(status_code=404,detail="Permission bundle not found")
|
|
tag_ids=sorted(set(payload.get("permission_tags") or []))
|
|
# an unknown or inactive id resolves to nothing, so the box would silently
|
|
# refuse to stay ticked. reject instead of granting less than was asked.
|
|
found=await PermissionTags.get_permission_tags_by_ids(self.session,tag_ids)
|
|
unknown=sorted(set(tag_ids)-{t.id for t in found})
|
|
if unknown:
|
|
raise HTTPException(status_code=422,detail=f"Unknown or inactive permission tag ids: {unknown}")
|
|
# build fields explicitly: update_permission setattr's whatever it is given, so
|
|
# passing the payload through would write name=None and hit the NOT NULL.
|
|
fields={"permission_tags":tag_ids}
|
|
if payload.get("name") is not None:
|
|
name=payload["name"].strip()
|
|
if bundle.is_system and name!=bundle.name:
|
|
raise HTTPException(status_code=409,detail="System permission bundles cannot be renamed")
|
|
clash=await Permissions.get_permission_by_name(self.session,name)
|
|
if clash and clash.id!=bundle.id:
|
|
raise HTTPException(status_code=409,detail="Permission bundle name already exists")
|
|
fields["name"]=name
|
|
if payload.get("description") is not None:
|
|
fields["description"]=payload["description"]
|
|
if payload.get("is_active") is not None:
|
|
fields["is_active"]=payload["is_active"]
|
|
await Permissions.update_permission(self.session,int(bundle.id),fields)
|
|
return await Role(session=self.session).get_permission_by_id(int(bundle.id))
|