feat: folder permissions (ACLs) - share folders with users/groups, inherit to subfolders, permission dialog UI
This commit is contained in:
@@ -34,6 +34,7 @@ from app.routes import (
|
||||
auth,
|
||||
errors,
|
||||
contact_folders,
|
||||
contact_folder_permissions,
|
||||
contacts,
|
||||
dashboard,
|
||||
entity_history,
|
||||
@@ -376,6 +377,7 @@ def create_app() -> FastAPI:
|
||||
app.include_router(notifications.router)
|
||||
app.include_router(contacts.router)
|
||||
app.include_router(contact_folders.router)
|
||||
app.include_router(contact_folder_permissions.router)
|
||||
app.include_router(dashboard.router)
|
||||
app.include_router(entity_history.router)
|
||||
app.include_router(import_export.router)
|
||||
|
||||
@@ -8,6 +8,7 @@ from app.models.audit import AuditLog, DeletionLog
|
||||
from app.models.auth import ApiToken, PasswordResetToken
|
||||
from app.models.contact import Contact, ContactPerson
|
||||
from app.models.contact_folder import ContactFolder
|
||||
from app.models.contact_folder_permission import ContactFolderPermission
|
||||
from app.models.contact_merge import ContactMergeHistory
|
||||
from app.models.entity_history import EntityHistory
|
||||
from app.models.currency import Currency
|
||||
@@ -45,6 +46,7 @@ __all__ = [
|
||||
"Contact",
|
||||
"ContactPerson",
|
||||
"ContactFolder",
|
||||
"ContactFolderPermission",
|
||||
"ContactMergeHistory",
|
||||
"EntityHistory",
|
||||
"Currency",
|
||||
|
||||
@@ -0,0 +1,90 @@
|
||||
"""Contact folder permission model — ACLs for folder sharing."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import uuid
|
||||
from datetime import datetime
|
||||
|
||||
from sqlalchemy import (
|
||||
Boolean,
|
||||
CheckConstraint,
|
||||
DateTime,
|
||||
ForeignKey,
|
||||
Index,
|
||||
String,
|
||||
UniqueConstraint,
|
||||
func,
|
||||
)
|
||||
from sqlalchemy.dialects.postgresql import UUID as PGUUID
|
||||
from sqlalchemy.orm import Mapped, mapped_column
|
||||
|
||||
from app.core.db import Base, TenantMixin
|
||||
|
||||
|
||||
class ContactFolderPermission(Base, TenantMixin):
|
||||
"""ACL entry for a contact folder.
|
||||
|
||||
Grants a specific permission level to a user or group for a folder.
|
||||
When ``inherit_to_subfolders`` is True, the permission also applies
|
||||
to all descendant folders.
|
||||
|
||||
Permission levels:
|
||||
- ``none`` — no access (explicit deny)
|
||||
- ``read`` — view folder and its contacts
|
||||
- ``write`` — read + edit contacts, add contacts to folder
|
||||
- ``admin`` — read + write + delete contacts + manage folder permissions
|
||||
"""
|
||||
|
||||
__tablename__ = "contact_folder_permissions"
|
||||
__table_args__ = (
|
||||
UniqueConstraint(
|
||||
"folder_id",
|
||||
"user_id",
|
||||
"group_id",
|
||||
"tenant_id",
|
||||
name="uq_cfp_folder_user_group_tenant",
|
||||
),
|
||||
# Ensure exactly one of user_id or group_id is set (not both, not neither)
|
||||
CheckConstraint(
|
||||
"(user_id IS NOT NULL AND group_id IS NULL) OR "
|
||||
"(user_id IS NULL AND group_id IS NOT NULL)",
|
||||
name="ck_cfp_exactly_one_principal",
|
||||
),
|
||||
Index("ix_cfp_folder", "folder_id"),
|
||||
Index("ix_cfp_user", "user_id"),
|
||||
Index("ix_cfp_group", "group_id"),
|
||||
Index("ix_cfp_tenant", "tenant_id"),
|
||||
)
|
||||
|
||||
id: Mapped[uuid.UUID] = mapped_column(
|
||||
PGUUID(as_uuid=True), primary_key=True, default=uuid.uuid4
|
||||
)
|
||||
folder_id: Mapped[uuid.UUID] = mapped_column(
|
||||
PGUUID(as_uuid=True),
|
||||
ForeignKey("contact_folders.id", ondelete="CASCADE"),
|
||||
nullable=False,
|
||||
index=True,
|
||||
)
|
||||
user_id: Mapped[uuid.UUID | None] = mapped_column(
|
||||
PGUUID(as_uuid=True),
|
||||
ForeignKey("users.id", ondelete="CASCADE"),
|
||||
nullable=True,
|
||||
)
|
||||
group_id: Mapped[uuid.UUID | None] = mapped_column(
|
||||
PGUUID(as_uuid=True),
|
||||
ForeignKey("groups.id", ondelete="CASCADE"),
|
||||
nullable=True,
|
||||
)
|
||||
permission_level: Mapped[str] = mapped_column(
|
||||
String(20), nullable=False, default="read"
|
||||
) # none | read | write | admin
|
||||
inherit_to_subfolders: Mapped[bool] = mapped_column(
|
||||
Boolean, nullable=False, default=True, server_default="true"
|
||||
)
|
||||
created_at: Mapped[datetime] = mapped_column(
|
||||
DateTime(timezone=True), nullable=False, server_default=func.now()
|
||||
)
|
||||
updated_at: Mapped[datetime] = mapped_column(
|
||||
DateTime(timezone=True), nullable=False, server_default=func.now(),
|
||||
onupdate=func.now(),
|
||||
)
|
||||
@@ -0,0 +1,107 @@
|
||||
"""Contact folder permission routes — ACL management."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import uuid
|
||||
|
||||
from fastapi import APIRouter, Depends, HTTPException, status
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.core.db import get_db
|
||||
from app.deps import require_permission
|
||||
from app.schemas.contact_folder_permission import (
|
||||
FolderPermissionCreate,
|
||||
FolderPermissionUpdate,
|
||||
)
|
||||
from app.services import contact_folder_permission_service
|
||||
|
||||
router = APIRouter(prefix="/api/v1/contact-folders", tags=["contact-folder-permissions"])
|
||||
|
||||
|
||||
@router.get("/{folder_id}/permissions")
|
||||
async def list_folder_permissions(
|
||||
folder_id: str,
|
||||
db: AsyncSession = Depends(get_db),
|
||||
current_user: dict = Depends(require_permission("contacts:read")),
|
||||
):
|
||||
"""List all permission entries for a folder."""
|
||||
tenant_id = uuid.UUID(current_user["tenant_id"])
|
||||
try:
|
||||
items = await contact_folder_permission_service.list_permissions(db, tenant_id, folder_id)
|
||||
except ValueError as e:
|
||||
raise HTTPException(status_code=404, detail=str(e))
|
||||
return {"items": items, "total": len(items)}
|
||||
|
||||
|
||||
@router.post("/{folder_id}/permissions", status_code=status.HTTP_201_CREATED)
|
||||
async def create_folder_permission(
|
||||
folder_id: str,
|
||||
body: FolderPermissionCreate,
|
||||
db: AsyncSession = Depends(get_db),
|
||||
current_user: dict = Depends(require_permission("contacts:write")),
|
||||
):
|
||||
"""Grant or update a permission on a folder for a user or group."""
|
||||
tenant_id = uuid.UUID(current_user["tenant_id"])
|
||||
try:
|
||||
return await contact_folder_permission_service.create_permission(
|
||||
db,
|
||||
tenant_id,
|
||||
folder_id,
|
||||
body.user_id,
|
||||
body.group_id,
|
||||
body.permission_level,
|
||||
body.inherit_to_subfolders,
|
||||
)
|
||||
except ValueError as e:
|
||||
raise HTTPException(status_code=400, detail=str(e))
|
||||
|
||||
|
||||
@router.put("/{folder_id}/permissions/{permission_id}")
|
||||
async def update_folder_permission(
|
||||
folder_id: str,
|
||||
permission_id: str,
|
||||
body: FolderPermissionUpdate,
|
||||
db: AsyncSession = Depends(get_db),
|
||||
current_user: dict = Depends(require_permission("contacts:write")),
|
||||
):
|
||||
"""Update an existing permission entry."""
|
||||
tenant_id = uuid.UUID(current_user["tenant_id"])
|
||||
try:
|
||||
return await contact_folder_permission_service.update_permission(
|
||||
db,
|
||||
tenant_id,
|
||||
permission_id,
|
||||
body.permission_level,
|
||||
body.inherit_to_subfolders,
|
||||
)
|
||||
except ValueError as e:
|
||||
raise HTTPException(status_code=404, detail=str(e))
|
||||
|
||||
|
||||
@router.delete("/{folder_id}/permissions/{permission_id}", status_code=status.HTTP_204_NO_CONTENT)
|
||||
async def delete_folder_permission(
|
||||
folder_id: str,
|
||||
permission_id: str,
|
||||
db: AsyncSession = Depends(get_db),
|
||||
current_user: dict = Depends(require_permission("contacts:write")),
|
||||
):
|
||||
"""Revoke a permission entry."""
|
||||
tenant_id = uuid.UUID(current_user["tenant_id"])
|
||||
try:
|
||||
await contact_folder_permission_service.delete_permission(db, tenant_id, permission_id)
|
||||
except ValueError as e:
|
||||
raise HTTPException(status_code=404, detail=str(e))
|
||||
|
||||
|
||||
@router.get("/{folder_id}/access")
|
||||
async def get_folder_access(
|
||||
folder_id: str,
|
||||
db: AsyncSession = Depends(get_db),
|
||||
current_user: dict = Depends(require_permission("contacts:read")),
|
||||
):
|
||||
"""Get effective access level for the current user on a folder."""
|
||||
tenant_id = uuid.UUID(current_user["tenant_id"])
|
||||
user_id = uuid.UUID(current_user["user_id"])
|
||||
return await contact_folder_permission_service.get_effective_access(
|
||||
db, tenant_id, user_id, uuid.UUID(folder_id)
|
||||
)
|
||||
@@ -0,0 +1,43 @@
|
||||
"""Schemas for contact folder permissions (ACLs)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
|
||||
class FolderPermissionCreate(BaseModel):
|
||||
user_id: str | None = None
|
||||
group_id: str | None = None
|
||||
permission_level: str = Field("read", pattern="^(none|read|write|admin)$")
|
||||
inherit_to_subfolders: bool = True
|
||||
|
||||
|
||||
class FolderPermissionUpdate(BaseModel):
|
||||
permission_level: str = Field(..., pattern="^(none|read|write|admin)$")
|
||||
inherit_to_subfolders: bool | None = None
|
||||
|
||||
|
||||
class FolderPermissionResponse(BaseModel):
|
||||
id: str
|
||||
folder_id: str
|
||||
user_id: str | None = None
|
||||
group_id: str | None = None
|
||||
user_name: str | None = None
|
||||
group_name: str | None = None
|
||||
permission_level: str
|
||||
inherit_to_subfolders: bool
|
||||
created_at: str | None = None
|
||||
|
||||
|
||||
class FolderPermissionListResponse(BaseModel):
|
||||
items: list[FolderPermissionResponse]
|
||||
total: int
|
||||
|
||||
|
||||
class FolderAccessInfo(BaseModel):
|
||||
"""Effective access level for the current user on a folder."""
|
||||
folder_id: str
|
||||
access_level: str # none | read | write | admin | owner
|
||||
is_owner: bool
|
||||
is_shared: bool
|
||||
inherited_from: str | None = None # parent folder id if inherited
|
||||
@@ -0,0 +1,414 @@
|
||||
"""Contact folder permission service — ACL management and access resolution."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import uuid
|
||||
from typing import Any
|
||||
|
||||
from sqlalchemy import or_, select, func
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.models.contact_folder import ContactFolder
|
||||
from app.models.contact_folder_permission import ContactFolderPermission
|
||||
from app.models.group import Group, UserGroup
|
||||
from app.models.user import User
|
||||
|
||||
# Permission hierarchy: higher = more access
|
||||
_PERM_RANK = {"none": 0, "read": 1, "write": 2, "admin": 3, "owner": 4}
|
||||
|
||||
|
||||
def _rank(level: str) -> int:
|
||||
return _PERM_RANK.get(level, 0)
|
||||
|
||||
|
||||
def _serialize_permission(
|
||||
p: ContactFolderPermission,
|
||||
user_name: str | None = None,
|
||||
group_name: str | None = None,
|
||||
) -> dict:
|
||||
return {
|
||||
"id": str(p.id),
|
||||
"folder_id": str(p.folder_id),
|
||||
"user_id": str(p.user_id) if p.user_id else None,
|
||||
"group_id": str(p.group_id) if p.group_id else None,
|
||||
"user_name": user_name,
|
||||
"group_name": group_name,
|
||||
"permission_level": p.permission_level,
|
||||
"inherit_to_subfolders": p.inherit_to_subfolders,
|
||||
"created_at": p.created_at.isoformat() if p.created_at else None,
|
||||
}
|
||||
|
||||
|
||||
async def list_permissions(
|
||||
db: AsyncSession, tenant_id: uuid.UUID, folder_id: str
|
||||
) -> list[dict]:
|
||||
"""List all permission entries for a folder."""
|
||||
folder_uuid = uuid.UUID(folder_id)
|
||||
result = await db.execute(
|
||||
select(ContactFolderPermission)
|
||||
.where(ContactFolderPermission.folder_id == folder_uuid)
|
||||
.where(ContactFolderPermission.tenant_id == tenant_id)
|
||||
.order_by(ContactFolderPermission.created_at)
|
||||
)
|
||||
perms = result.scalars().all()
|
||||
|
||||
# Batch-load user and group names
|
||||
user_ids = [p.user_id for p in perms if p.user_id]
|
||||
group_ids = [p.group_id for p in perms if p.group_id]
|
||||
|
||||
user_names: dict[uuid.UUID, str] = {}
|
||||
if user_ids:
|
||||
users_q = await db.execute(
|
||||
select(User.id, User.name).where(User.id.in_(user_ids))
|
||||
)
|
||||
user_names = {row[0]: row[1] for row in users_q}
|
||||
|
||||
group_names: dict[uuid.UUID, str] = {}
|
||||
if group_ids:
|
||||
groups_q = await db.execute(
|
||||
select(Group.id, Group.name).where(Group.id.in_(group_ids))
|
||||
)
|
||||
group_names = {row[0]: row[1] for row in groups_q}
|
||||
|
||||
return [
|
||||
_serialize_permission(
|
||||
p,
|
||||
user_names.get(p.user_id) if p.user_id else None,
|
||||
group_names.get(p.group_id) if p.group_id else None,
|
||||
)
|
||||
for p in perms
|
||||
]
|
||||
|
||||
|
||||
async def create_permission(
|
||||
db: AsyncSession,
|
||||
tenant_id: uuid.UUID,
|
||||
folder_id: str,
|
||||
user_id: str | None,
|
||||
group_id: str | None,
|
||||
permission_level: str,
|
||||
inherit_to_subfolders: bool = True,
|
||||
) -> dict:
|
||||
"""Create or update a permission entry for a folder."""
|
||||
folder_uuid = uuid.UUID(folder_id)
|
||||
|
||||
# Verify folder exists and belongs to tenant
|
||||
folder_q = await db.execute(
|
||||
select(ContactFolder)
|
||||
.where(ContactFolder.id == folder_uuid)
|
||||
.where(ContactFolder.tenant_id == tenant_id)
|
||||
)
|
||||
if not folder_q.scalar_one_or_none():
|
||||
raise ValueError("Folder not found")
|
||||
|
||||
user_uuid = uuid.UUID(user_id) if user_id else None
|
||||
group_uuid = uuid.UUID(group_id) if group_id else None
|
||||
|
||||
if not user_uuid and not group_uuid:
|
||||
raise ValueError("Either user_id or group_id must be provided")
|
||||
if user_uuid and group_uuid:
|
||||
raise ValueError("Only one of user_id or group_id can be provided")
|
||||
|
||||
# Check for existing entry (upsert)
|
||||
existing_q = await db.execute(
|
||||
select(ContactFolderPermission)
|
||||
.where(ContactFolderPermission.folder_id == folder_uuid)
|
||||
.where(ContactFolderPermission.tenant_id == tenant_id)
|
||||
.where(
|
||||
ContactFolderPermission.user_id == user_uuid
|
||||
if user_uuid
|
||||
else ContactFolderPermission.user_id.is_(None)
|
||||
)
|
||||
.where(
|
||||
ContactFolderPermission.group_id == group_uuid
|
||||
if group_uuid
|
||||
else ContactFolderPermission.group_id.is_(None)
|
||||
)
|
||||
)
|
||||
existing = existing_q.scalar_one_or_none()
|
||||
|
||||
if existing:
|
||||
existing.permission_level = permission_level
|
||||
existing.inherit_to_subfolders = inherit_to_subfolders
|
||||
await db.commit()
|
||||
await db.refresh(existing)
|
||||
# Load names for response
|
||||
user_name = group_name = None
|
||||
if existing.user_id:
|
||||
u = await db.execute(select(User.name).where(User.id == existing.user_id))
|
||||
user_name = u.scalar_one_or_none()
|
||||
if existing.group_id:
|
||||
g = await db.execute(select(Group.name).where(Group.id == existing.group_id))
|
||||
group_name = g.scalar_one_or_none()
|
||||
return _serialize_permission(existing, user_name, group_name)
|
||||
|
||||
perm = ContactFolderPermission(
|
||||
tenant_id=tenant_id,
|
||||
folder_id=folder_uuid,
|
||||
user_id=user_uuid,
|
||||
group_id=group_uuid,
|
||||
permission_level=permission_level,
|
||||
inherit_to_subfolders=inherit_to_subfolders,
|
||||
)
|
||||
db.add(perm)
|
||||
await db.commit()
|
||||
await db.refresh(perm)
|
||||
|
||||
# Load names for response
|
||||
user_name = group_name = None
|
||||
if perm.user_id:
|
||||
u = await db.execute(select(User.name).where(User.id == perm.user_id))
|
||||
user_name = u.scalar_one_or_none()
|
||||
if perm.group_id:
|
||||
g = await db.execute(select(Group.name).where(Group.id == perm.group_id))
|
||||
group_name = g.scalar_one_or_none()
|
||||
return _serialize_permission(perm, user_name, group_name)
|
||||
|
||||
|
||||
async def update_permission(
|
||||
db: AsyncSession,
|
||||
tenant_id: uuid.UUID,
|
||||
permission_id: str,
|
||||
permission_level: str,
|
||||
inherit_to_subfolders: bool | None = None,
|
||||
) -> dict:
|
||||
"""Update an existing permission entry."""
|
||||
perm_uuid = uuid.UUID(permission_id)
|
||||
result = await db.execute(
|
||||
select(ContactFolderPermission)
|
||||
.where(ContactFolderPermission.id == perm_uuid)
|
||||
.where(ContactFolderPermission.tenant_id == tenant_id)
|
||||
)
|
||||
perm = result.scalar_one_or_none()
|
||||
if not perm:
|
||||
raise ValueError("Permission not found")
|
||||
|
||||
perm.permission_level = permission_level
|
||||
if inherit_to_subfolders is not None:
|
||||
perm.inherit_to_subfolders = inherit_to_subfolders
|
||||
|
||||
await db.commit()
|
||||
await db.refresh(perm)
|
||||
|
||||
user_name = group_name = None
|
||||
if perm.user_id:
|
||||
u = await db.execute(select(User.name).where(User.id == perm.user_id))
|
||||
user_name = u.scalar_one_or_none()
|
||||
if perm.group_id:
|
||||
g = await db.execute(select(Group.name).where(Group.id == perm.group_id))
|
||||
group_name = g.scalar_one_or_none()
|
||||
return _serialize_permission(perm, user_name, group_name)
|
||||
|
||||
|
||||
async def delete_permission(
|
||||
db: AsyncSession, tenant_id: uuid.UUID, permission_id: str
|
||||
) -> None:
|
||||
"""Delete a permission entry."""
|
||||
perm_uuid = uuid.UUID(permission_id)
|
||||
result = await db.execute(
|
||||
select(ContactFolderPermission)
|
||||
.where(ContactFolderPermission.id == perm_uuid)
|
||||
.where(ContactFolderPermission.tenant_id == tenant_id)
|
||||
)
|
||||
perm = result.scalar_one_or_none()
|
||||
if not perm:
|
||||
raise ValueError("Permission not found")
|
||||
|
||||
await db.delete(perm)
|
||||
await db.commit()
|
||||
|
||||
|
||||
async def get_effective_access(
|
||||
db: AsyncSession,
|
||||
tenant_id: uuid.UUID,
|
||||
user_id: uuid.UUID,
|
||||
folder_id: uuid.UUID,
|
||||
) -> dict:
|
||||
"""Get the effective access level for a user on a folder.
|
||||
|
||||
Resolution order (highest wins):
|
||||
1. Folder owner → "owner"
|
||||
2. Direct permission on this folder
|
||||
3. Inherited permission from ancestor folders (inherit_to_subfolders=True)
|
||||
4. Group membership permissions (direct + inherited)
|
||||
5. No access → "none"
|
||||
"""
|
||||
# Check ownership
|
||||
folder_q = await db.execute(
|
||||
select(ContactFolder)
|
||||
.where(ContactFolder.id == folder_id)
|
||||
.where(ContactFolder.tenant_id == tenant_id)
|
||||
)
|
||||
folder = folder_q.scalar_one_or_none()
|
||||
if not folder:
|
||||
return {"folder_id": str(folder_id), "access_level": "none", "is_owner": False, "is_shared": False, "inherited_from": None}
|
||||
|
||||
if folder.user_id == user_id:
|
||||
# Check if shared with anyone
|
||||
shared_q = await db.execute(
|
||||
select(func.count(ContactFolderPermission.id))
|
||||
.where(ContactFolderPermission.folder_id == folder_id)
|
||||
.where(ContactFolderPermission.tenant_id == tenant_id)
|
||||
)
|
||||
is_shared = (shared_q.scalar() or 0) > 0
|
||||
return {"folder_id": str(folder_id), "access_level": "owner", "is_owner": True, "is_shared": is_shared, "inherited_from": None}
|
||||
|
||||
# Build ancestor chain (folder → parent → grandparent → ...)
|
||||
ancestor_chain: list[uuid.UUID] = [folder_id]
|
||||
current = folder
|
||||
while current.parent_id:
|
||||
ancestor_chain.append(current.parent_id)
|
||||
parent_q = await db.execute(
|
||||
select(ContactFolder)
|
||||
.where(ContactFolder.id == current.parent_id)
|
||||
.where(ContactFolder.tenant_id == tenant_id)
|
||||
)
|
||||
current = parent_q.scalar_one_or_none()
|
||||
if not current:
|
||||
break
|
||||
|
||||
# Get user's group memberships
|
||||
groups_q = await db.execute(
|
||||
select(UserGroup.group_id)
|
||||
.where(UserGroup.user_id == user_id)
|
||||
.where(UserGroup.tenant_id == tenant_id)
|
||||
)
|
||||
group_ids = [row[0] for row in groups_q]
|
||||
|
||||
best_level = "none"
|
||||
inherited_from = None
|
||||
|
||||
# Walk ancestor chain from closest to furthest
|
||||
for i, ancestor_id in enumerate(ancestor_chain):
|
||||
# Direct user permission
|
||||
user_perm_q = await db.execute(
|
||||
select(ContactFolderPermission)
|
||||
.where(ContactFolderPermission.folder_id == ancestor_id)
|
||||
.where(ContactFolderPermission.tenant_id == tenant_id)
|
||||
.where(ContactFolderPermission.user_id == user_id)
|
||||
)
|
||||
for perm in user_perm_q.scalars():
|
||||
# If this is an ancestor (not the folder itself), only apply if inherit_to_subfolders
|
||||
if i > 0 and not perm.inherit_to_subfolders:
|
||||
continue
|
||||
if _rank(perm.permission_level) > _rank(best_level):
|
||||
best_level = perm.permission_level
|
||||
inherited_from = str(ancestor_id) if i > 0 else None
|
||||
|
||||
# Group permissions
|
||||
if group_ids:
|
||||
group_perm_q = await db.execute(
|
||||
select(ContactFolderPermission)
|
||||
.where(ContactFolderPermission.folder_id == ancestor_id)
|
||||
.where(ContactFolderPermission.tenant_id == tenant_id)
|
||||
.where(ContactFolderPermission.group_id.in_(group_ids))
|
||||
)
|
||||
for perm in group_perm_q.scalars():
|
||||
if i > 0 and not perm.inherit_to_subfolders:
|
||||
continue
|
||||
if _rank(perm.permission_level) > _rank(best_level):
|
||||
best_level = perm.permission_level
|
||||
inherited_from = str(ancestor_id) if i > 0 else None
|
||||
|
||||
# Check if folder is shared at all
|
||||
shared_q = await db.execute(
|
||||
select(func.count(ContactFolderPermission.id))
|
||||
.where(ContactFolderPermission.folder_id == folder_id)
|
||||
.where(ContactFolderPermission.tenant_id == tenant_id)
|
||||
)
|
||||
is_shared = (shared_q.scalar() or 0) > 0
|
||||
|
||||
return {
|
||||
"folder_id": str(folder_id),
|
||||
"access_level": best_level,
|
||||
"is_owner": False,
|
||||
"is_shared": is_shared,
|
||||
"inherited_from": inherited_from,
|
||||
}
|
||||
|
||||
|
||||
async def get_visible_folder_ids(
|
||||
db: AsyncSession,
|
||||
tenant_id: uuid.UUID,
|
||||
user_id: uuid.UUID,
|
||||
) -> tuple[list[uuid.UUID], dict[uuid.UUID, str]]:
|
||||
"""Get all folder IDs visible to a user and their access levels.
|
||||
|
||||
Returns (visible_folder_ids, access_map) where access_map is
|
||||
folder_id → access_level string.
|
||||
"""
|
||||
# Get all folders in tenant
|
||||
all_folders_q = await db.execute(
|
||||
select(ContactFolder)
|
||||
.where(ContactFolder.tenant_id == tenant_id)
|
||||
.where(ContactFolder.deleted_at.is_(None))
|
||||
)
|
||||
all_folders = all_folders_q.scalars().all()
|
||||
|
||||
visible: list[uuid.UUID] = []
|
||||
access_map: dict[uuid.UUID, str] = {}
|
||||
|
||||
# Get user's group memberships
|
||||
groups_q = await db.execute(
|
||||
select(UserGroup.group_id)
|
||||
.where(UserGroup.user_id == user_id)
|
||||
.where(UserGroup.tenant_id == tenant_id)
|
||||
)
|
||||
group_ids = [row[0] for row in groups_q]
|
||||
|
||||
# Get all permission entries for this tenant
|
||||
all_perms_q = await db.execute(
|
||||
select(ContactFolderPermission)
|
||||
.where(ContactFolderPermission.tenant_id == tenant_id)
|
||||
)
|
||||
all_perms = all_perms_q.scalars().all()
|
||||
|
||||
# Build permission lookup: folder_id → list of (principal_type, principal_id, level, inherit)
|
||||
perm_lookup: dict[uuid.UUID, list[tuple[str, uuid.UUID | None, str, bool]]] = {}
|
||||
for p in all_perms:
|
||||
if p.folder_id not in perm_lookup:
|
||||
perm_lookup[p.folder_id] = []
|
||||
if p.user_id:
|
||||
perm_lookup[p.folder_id].append(("user", p.user_id, p.permission_level, p.inherit_to_subfolders))
|
||||
if p.group_id:
|
||||
perm_lookup[p.folder_id].append(("group", p.group_id, p.permission_level, p.inherit_to_subfolders))
|
||||
|
||||
# Build parent map for ancestor traversal
|
||||
parent_map: dict[uuid.UUID, uuid.UUID | None] = {}
|
||||
for f in all_folders:
|
||||
parent_map[f.id] = f.parent_id
|
||||
|
||||
for folder in all_folders:
|
||||
# Owner sees everything they own
|
||||
if folder.user_id == user_id:
|
||||
visible.append(folder.id)
|
||||
access_map[folder.id] = "owner"
|
||||
continue
|
||||
|
||||
# Check effective access via permissions (including inherited)
|
||||
best_level = "none"
|
||||
|
||||
# Build ancestor chain
|
||||
chain: list[uuid.UUID] = [folder.id]
|
||||
current_parent = parent_map.get(folder.id)
|
||||
while current_parent:
|
||||
chain.append(current_parent)
|
||||
current_parent = parent_map.get(current_parent)
|
||||
|
||||
for i, ancestor_id in enumerate(chain):
|
||||
perms = perm_lookup.get(ancestor_id, [])
|
||||
for ptype, pid, level, inherit in perms:
|
||||
if i > 0 and not inherit:
|
||||
continue
|
||||
if ptype == "user" and pid == user_id:
|
||||
if _rank(level) > _rank(best_level):
|
||||
best_level = level
|
||||
elif ptype == "group" and pid in group_ids:
|
||||
if _rank(level) > _rank(best_level):
|
||||
best_level = level
|
||||
|
||||
if best_level != "none":
|
||||
visible.append(folder.id)
|
||||
access_map[folder.id] = best_level
|
||||
|
||||
return visible, access_map
|
||||
@@ -27,11 +27,24 @@ def _serialize_folder(f: ContactFolder, contact_count: int = 0) -> dict:
|
||||
async def list_folders(
|
||||
db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID
|
||||
) -> list[dict]:
|
||||
"""List all folders for a user (flat list, frontend builds tree)."""
|
||||
"""List all folders visible to a user (owned + shared via ACLs).
|
||||
|
||||
Returns a flat list; the frontend builds the tree.
|
||||
Each folder includes ``access_level`` (owner|admin|write|read) and
|
||||
``is_shared`` flag.
|
||||
"""
|
||||
from app.services.contact_folder_permission_service import get_visible_folder_ids
|
||||
|
||||
# Get visible folder IDs and access map
|
||||
visible_ids, access_map = await get_visible_folder_ids(db, tenant_id, user_id)
|
||||
|
||||
if not visible_ids:
|
||||
return []
|
||||
|
||||
result = await db.execute(
|
||||
select(ContactFolder)
|
||||
.where(ContactFolder.tenant_id == tenant_id)
|
||||
.where(ContactFolder.user_id == user_id)
|
||||
.where(ContactFolder.id.in_(visible_ids))
|
||||
.where(ContactFolder.deleted_at.is_(None))
|
||||
.order_by(ContactFolder.parent_id, ContactFolder.sort_order, ContactFolder.name)
|
||||
)
|
||||
@@ -50,8 +63,23 @@ async def list_folders(
|
||||
for fid, cnt in count_result:
|
||||
counts[fid] = cnt
|
||||
|
||||
# Check which folders have permissions (are shared)
|
||||
from app.models.contact_folder_permission import ContactFolderPermission
|
||||
shared_q = await db.execute(
|
||||
select(ContactFolderPermission.folder_id)
|
||||
.where(ContactFolderPermission.folder_id.in_(folder_ids))
|
||||
.where(ContactFolderPermission.tenant_id == tenant_id)
|
||||
.group_by(ContactFolderPermission.folder_id)
|
||||
)
|
||||
shared_ids = {row[0] for row in shared_q}
|
||||
|
||||
return [
|
||||
{**_serialize_folder(f), "contact_count": counts.get(f.id, 0)}
|
||||
{
|
||||
**_serialize_folder(f),
|
||||
"contact_count": counts.get(f.id, 0),
|
||||
"access_level": access_map.get(f.id, "none"),
|
||||
"is_shared": f.id in shared_ids,
|
||||
}
|
||||
for f in folders
|
||||
]
|
||||
|
||||
|
||||
Reference in New Issue
Block a user