FabledScribe/src/fabledassistant/services/access.py

"""Access control service.

Single source of truth for permission resolution on projects and notes.
All human-facing routes that should honour shares call these functions.
LLM tool routes continue to use owner-scoped service functions directly.

Permission rank: owner > admin > editor > viewer
"""

import logging

from sqlalchemy import select

from fabledassistant.models import async_session
from fabledassistant.models.group import GroupMembership
from fabledassistant.models.note import Note
from fabledassistant.models.project import Project
from fabledassistant.models.share import NoteShare, ProjectShare

logger = logging.getLogger(__name__)

PERMISSION_RANK: dict[str, int] = {
    "viewer": 1,
    "editor": 2,
    "admin": 3,
    "owner": 4,
}


def _higher(a: str | None, b: str | None) -> str | None:
    """Return the higher-ranked permission, or None if both are None."""
    if a is None:
        return b
    if b is None:
        return a
    return a if PERMISSION_RANK[a] >= PERMISSION_RANK[b] else b


async def _user_group_ids(session, user_id: int) -> list[int]:
    rows = (
        await session.execute(
            select(GroupMembership.group_id).where(GroupMembership.user_id == user_id)
        )
    ).scalars().all()
    return list(rows)


# ---------------------------------------------------------------------------
# Project permissions
# ---------------------------------------------------------------------------

async def get_project_permission(user_id: int, project_id: int) -> str | None:
    """Return the effective permission string for user on project, or None."""
    async with async_session() as session:
        project = await session.get(Project, project_id)
        if project is None:
            return None
        if project.user_id == user_id:
            return "owner"

        # Direct share
        direct = (
            await session.execute(
                select(ProjectShare).where(
                    ProjectShare.project_id == project_id,
                    ProjectShare.shared_with_user_id == user_id,
                )
            )
        ).scalar_one_or_none()

        # Group shares
        group_ids = await _user_group_ids(session, user_id)
        group_perm: str | None = None
        if group_ids:
            group_shares = (
                await session.execute(
                    select(ProjectShare).where(
                        ProjectShare.project_id == project_id,
                        ProjectShare.shared_with_group_id.in_(group_ids),
                    )
                )
            ).scalars().all()
            for gs in group_shares:
                group_perm = _higher(group_perm, gs.permission)

        return _higher(direct.permission if direct else None, group_perm)


async def can_read_project(user_id: int, project_id: int) -> bool:
    return (await get_project_permission(user_id, project_id)) is not None


async def can_write_project(user_id: int, project_id: int) -> bool:
    perm = await get_project_permission(user_id, project_id)
    return perm in ("editor", "admin", "owner")


async def can_admin_project(user_id: int, project_id: int) -> bool:
    perm = await get_project_permission(user_id, project_id)
    return perm in ("admin", "owner")


# ---------------------------------------------------------------------------
# Note / task permissions
# ---------------------------------------------------------------------------

async def get_note_permission(user_id: int, note_id: int) -> str | None:
    """Return the effective permission for user on a note/task, or None.

    Resolution order:
    1. Ownership
    2. Direct note share
    3. Group-based note share
    4. Inherited from project share (if note belongs to a project)
    Highest rank wins.
    """
    async with async_session() as session:
        note = await session.get(Note, note_id)
        if note is None:
            return None
        if note.user_id == user_id:
            return "owner"

        direct = (
            await session.execute(
                select(NoteShare).where(
                    NoteShare.note_id == note_id,
                    NoteShare.shared_with_user_id == user_id,
                )
            )
        ).scalar_one_or_none()

        group_ids = await _user_group_ids(session, user_id)
        group_perm: str | None = None
        if group_ids:
            group_shares = (
                await session.execute(
                    select(NoteShare).where(
                        NoteShare.note_id == note_id,
                        NoteShare.shared_with_group_id.in_(group_ids),
                    )
                )
            ).scalars().all()
            for gs in group_shares:
                group_perm = _higher(group_perm, gs.permission)

        note_perm = _higher(direct.permission if direct else None, group_perm)

        # Inherit from project if note belongs to one
        if note.project_id is not None:
            project_perm = await get_project_permission(user_id, note.project_id)
            note_perm = _higher(note_perm, project_perm)

        return note_perm


async def can_read_note(user_id: int, note_id: int) -> bool:
    return (await get_note_permission(user_id, note_id)) is not None


async def can_write_note(user_id: int, note_id: int) -> bool:
    perm = await get_note_permission(user_id, note_id)
    return perm in ("editor", "admin", "owner")