LLeMbas/src/lembas/security/permissions.py

"""Permission vocabulary and resolution.

The model is deliberately small: a flat set of named booleans, granted by an
instance-wide baseline and widened by group membership. Permissions are a union
across groups -- being in a second group can only ever grant more, never take
away. That is the behaviour people expect, and the alternative (a deny that
wins) makes "why can this user not do X" unanswerable without simulating every
group.

Administrators bypass the whole thing. There is no permission that can be
withheld from an admin, because an admin can grant it back to themselves in two
clicks; pretending otherwise would be theatre.

Model *access* is separate and lives in models_visible_to(): a permission says
what a user may do, model access says which models they may do it with.
"""

from __future__ import annotations

from dataclasses import dataclass

from sqlalchemy import select
from sqlalchemy.orm import Session as DBSession

from lembas.db.models import Connection, Model, User


@dataclass(frozen=True)
class PermissionDef:
    key: str
    label: str
    description: str
    default: bool
    group: str


# The order here is the order they render in the admin UI.
PERMISSION_DEFS: tuple[PermissionDef, ...] = (
    PermissionDef(
        "chat.create", "Start chats", "Create new conversations.", True, "Chat"
    ),
    PermissionDef(
        "chat.delete", "Delete chats", "Delete their own conversations.", True, "Chat"
    ),
    PermissionDef(
        "chat.system_prompt",
        "Set system prompts",
        "Give an individual chat its own system prompt.",
        True,
        "Chat",
    ),
    PermissionDef(
        "chat.params",
        "Adjust sampling",
        "Change temperature, top-p and similar per chat.",
        False,
        "Chat",
    ),
    PermissionDef(
        "chat.model_select",
        "Choose the model",
        "Switch a chat to a different model. Without this, chats use the default.",
        True,
        "Chat",
    ),
    PermissionDef(
        "folder.manage",
        "Manage folders",
        "Create, rename, nest and delete folders.",
        True,
        "Workspace",
    ),
    PermissionDef(
        "files.upload",
        "Attach files",
        "Attach images, PDFs and text files to a message. Images only reach "
        "models marked as having vision.",
        True,
        "Workspace",
    ),
    PermissionDef(
        "tools.web_search",
        "Search the web",
        "Let a model look things up while it answers. Only offered to models "
        "marked as supporting tools, and only when web search is configured.",
        True,
        "Chat",
    ),
    PermissionDef(
        "tools.fetch",
        "Fetch a page",
        "Let a model retrieve one web page and read it, given its address. "
        "Addresses on this machine and this network are refused unless an "
        "administrator has allowed them.",
        True,
        "Chat",
    ),
    PermissionDef(
        "tools.image",
        "Generate images",
        "Let a model draw a picture and show it in the conversation. Only "
        "offered when an image generator has been configured, and every "
        "generation spends time on whatever machine is running it.",
        True,
        "Chat",
    ),
    PermissionDef(
        "tools.custom",
        "Use custom tools",
        "Let a model call the HTTP tools an administrator has defined. Which "
        "ones depends on the groups each tool is restricted to.",
        True,
        "Chat",
    ),
    PermissionDef(
        "tools.mcp",
        "Use MCP servers",
        "Let a model call tools from the MCP servers an administrator has "
        "added. Which ones depends on the groups each server is restricted to.",
        True,
        "Chat",
    ),
    PermissionDef(
        "agent.ssh",
        "Save SSH connections",
        "Keep connection profiles for machines of their own. The credential is "
        "encrypted here, and whoever saves it decides which host it opens.",
        False,
        "Agent",
    ),
    PermissionDef(
        "tools.agent",
        "Run commands",
        "Let a model read files, write files and run commands on one of their "
        "SSH connections. What it may do without asking depends on the chat's "
        "mode. Nothing runs on this server.",
        False,
        "Agent",
    ),
    PermissionDef(
        "agent.terminal",
        "Open a terminal",
        "Open an interactive shell on one of their own SSH connections, from "
        "inside the chat. What they type there is theirs: the chat's mode "
        "governs the model, not the person at the keyboard.",
        False,
        "Agent",
    ),
    PermissionDef(
        "tools.subagent",
        "Delegate to a helper",
        "Let a model hand a self-contained piece of work to a second one that "
        "runs on its own and reports back — reading and searching in parallel "
        "rather than one thing at a time. A helper cannot ask questions, "
        "cannot spawn helpers of its own, and can only do what this chat could "
        "already do without stopping to ask.",
        False,
        "Chat",
    ),
    PermissionDef(
        "tools.persona",
        "Have a personality of its own",
        "Let a model keep and rewrite its own character, and keep its own read of "
        "how this person works — carried into every conversation rather than "
        "forgotten at the end of one. Every version is kept, both are visible, "
        "and either can be put back or deleted. A model cannot do this while "
        "running as somebody's helper or on a schedule.",
        False,
        "Chat",
    ),
    PermissionDef(
        "tools.friend",
        "Ask another model",
        "Let a model put a question to one of the other models here and use the "
        "answer — a second opinion from something good at what it is bad at. "
        "It is told which models exist and what each is for, and it can only "
        "reach the ones this person could use themselves. The model answering "
        "cannot ask questions and cannot ask anyone else in turn.",
        False,
        "Chat",
    ),
    PermissionDef(
        "tools.ask",
        "Be asked questions",
        "Let a model stop mid-reply and ask you something, with answers to pick "
        "from or a box to write your own.",
        True,
        "Chat",
    ),
    PermissionDef(
        "tools.scratch",
        "Write in the canvas",
        "Let a model build something up in this chat's scratch document, which "
        "sits open beside the conversation and can be edited and attached to a "
        "message. It belongs to the chat and is not searchable afterwards.",
        True,
        "Chat",
    ),
    PermissionDef(
        "schedule.use",
        "Schedule work",
        "Set things to run later, on their own — once, or on a repeating "
        "timetable. This spends model time with nobody at the keyboard, so it "
        "is a capability chosen on purpose rather than one everybody has.",
        False,
        "Scheduling",
    ),
    PermissionDef(
        "reports.use",
        "Keep reports",
        "Read the Reports section: finished pieces of work filed for them to "
        "read later, by a model that was asked for one or by something that ran "
        "while they were away.",
        True,
        "Reports",
    ),
    PermissionDef(
        "tools.report",
        "File reports",
        "Let a model write a report when it finishes a piece of work, and read "
        "back ones it filed earlier. A report is addressed to the reader and "
        "cannot be replied to, so this costs nothing but a place to put things.",
        True,
        "Reports",
    ),
    PermissionDef(
        "audio.transcribe",
        "Dictate messages",
        "Speak a message instead of typing it. Needs a transcription endpoint.",
        True,
        "Audio",
    ),
    PermissionDef(
        "audio.listen",
        "Play replies aloud",
        "Have a reply read out. Needs a speech endpoint.",
        True,
        "Audio",
    ),
    PermissionDef(
        "library.use",
        "Use the library",
        "Keep knowledge documents, notes, memories and skills of their own.",
        True,
        "Library",
    ),
    PermissionDef(
        "library.share",
        "Share library items",
        "Give other people, or a group, access to their knowledge bases, notes, "
        "skills and reports. Sharing grants reading only — never changing, and "
        "never sharing on.",
        # On. It was off, which meant sharing shipped documented as done and
        # unreachable: the panel is only rendered for somebody who holds this,
        # so out of the box nobody could share anything and nothing said why.
        # An instance that wants it off can say so; one that never looked should
        # get the feature it was told it had.
        True,
        "Library",
    ),
    PermissionDef(
        "tools.knowledge",
        "Search their knowledge",
        "Let a model search the documents this user has collected.",
        True,
        "Library",
    ),
    PermissionDef(
        "tools.notes",
        "Read and write notes",
        "Let a model keep its own notes for this user, and read them back later.",
        True,
        "Library",
    ),
    PermissionDef(
        "tools.memory",
        "Remember things",
        "Let a model record short facts about this user, shown to it on every "
        "turn.",
        True,
        "Library",
    ),
    PermissionDef(
        "tools.skills",
        "Use and write skills",
        "Let a model follow saved instructions, and write new ones. Every "
        "change is recorded and can be rolled back.",
        True,
        "Library",
    ),
    # --- Reading and writing, split where the difference matters ------------
    # Three gates cover both, and for these three the two halves are genuinely
    # different decisions: a model that may *read* somebody's notes and not add
    # to them is a reasonable thing to want, and until now `tools.notes` was one
    # switch over five tools.
    #
    # Not split for every gate. `tools.web_search` has no write half; `report`
    # is a write with no read worth withholding; `agent` has modes, which are a
    # finer instrument than a permission and are per chat. A permission that
    # answers "the same as that one" is a permission nobody should be asked
    # about -- the reasoning `schedule.use` already carries.
    #
    # **All three default on**, so an instance that never looks behaves exactly
    # as it did: `_family_allowed` reads them only to *narrow* what the gate
    # already allowed.
    PermissionDef(
        "tools.notes.write",
        "Write notes",
        "Let a model create, change and delete notes. Without it, it can still "
        "search and read the ones that are there.",
        True,
        "Library",
    ),
    PermissionDef(
        "tools.memory.write",
        "Record memories",
        "Let a model add and forget short facts about this person. Without it, "
        "the memories it already has are still shown to it every turn.",
        True,
        "Library",
    ),
    PermissionDef(
        "tools.skills.write",
        "Write skills",
        "Let a model write new skills and change existing ones. Without it, it "
        "follows the skills that are there and cannot add to them — which is "
        "the setting for an instance whose skills are curated by hand.",
        True,
        "Library",
    ),
)

# Gates whose read and write halves are separate permissions. Keyed on the gate,
# with the permission derived as `tools.<gate>.write`, so adding a fourth is one
# entry here and one PermissionDef above.
SPLIT_GATES = ("notes", "memory", "skills")

PERMISSION_KEYS = tuple(d.key for d in PERMISSION_DEFS)
DEFAULT_PERMISSIONS = {d.key: d.default for d in PERMISSION_DEFS}


def permission_groups() -> dict[str, list[PermissionDef]]:
    """Definitions bucketed by their UI section, preserving declaration order."""
    grouped: dict[str, list[PermissionDef]] = {}
    for definition in PERMISSION_DEFS:
        grouped.setdefault(definition.group, []).append(definition)
    return grouped


def baseline_permissions(db: DBSession) -> dict[str, bool]:
    """Instance-wide permissions for a user in no group at all."""
    from lembas.services import settings_store

    stored = settings_store.get(db, "default_permissions") or {}
    return {key: bool(stored.get(key, DEFAULT_PERMISSIONS[key])) for key in PERMISSION_KEYS}


def resolve(db: DBSession, user: User | None) -> dict[str, bool]:
    """Effective permissions for a user."""
    if user is None:
        return dict.fromkeys(PERMISSION_KEYS, False)
    if user.is_admin:
        return dict.fromkeys(PERMISSION_KEYS, True)

    effective = baseline_permissions(db)
    for group in user.groups:
        granted = group.permissions_json or {}
        for key in PERMISSION_KEYS:
            # Union: a group can only widen. Absent means "no opinion", not
            # "deny", so a group need only list what it adds.
            if granted.get(key):
                effective[key] = True
    return effective


def has(db: DBSession, user: User | None, key: str) -> bool:
    return resolve(db, user).get(key, False)


def explain(db: DBSession, user: User | None) -> dict[str, dict]:
    """Every permission, whether this user has it, and **where it came from**.

    The question the admin screens could not answer. `resolve` has always
    computed the union and thrown the working away, so "why can this person do
    X?" meant opening every group they belong to and reading the grids by eye --
    which is exactly the simulation the union rule exists to avoid needing.

    `source` is "admin" (bypassing everything), "baseline", or the names of the
    groups that granted it. A permission that is off has no source, because
    nothing granted it -- there is no such thing as a deny here to point at.
    """
    keys = PERMISSION_KEYS
    if user is None:
        return {key: {"on": False, "source": []} for key in keys}
    if user.is_admin:
        return {key: {"on": True, "source": ["admin"]} for key in keys}

    baseline = baseline_permissions(db)
    out: dict[str, dict] = {}
    for key in keys:
        sources = ["baseline"] if baseline.get(key) else []
        sources += [
            group.name for group in user.groups if (group.permissions_json or {}).get(key)
        ]
        out[key] = {"on": bool(sources), "source": sources}
    return out


# --- Quotas -------------------------------------------------------------------
# What a group may raise, and what each number means. Every one of them is
# **zero for no limit**, which is the convention `max_completion_tokens` and
# `index_chars` already use here, and it is what makes "unlimited" sayable at all.
#
# Five axes rather than one, because they fail differently and a single "budget"
# would have to pick an exchange rate between a token and a minute of somebody's
# GPU. There isn't one.
LIMIT_DEFS: tuple[tuple[str, str, str], ...] = (
    (
        "monthly_tokens",
        "Tokens a month",
        "Prompt and completion together, across every chat, reset on the first "
        "of the month. Reached, a reply says so before it spends anything "
        "rather than stopping half way through.",
    ),
    (
        "concurrent_replies",
        "Replies at once",
        "How many of their chats may be writing at the same time. This is the "
        "one that stops one person queueing every other person's work behind "
        "them on a single endpoint.",
    ),
    (
        "agent_seconds",
        "Longest agent reply",
        "Seconds of wall clock for one reply in an agent chat, if lower than "
        "the instance's own. Waiting for somebody to approve something does "
        "not count.",
    ),
    (
        "images_per_day",
        "Images a day",
        "Each one is a minute of somebody's GPU and no tokens at all, so a "
        "token budget says nothing about it.",
    ),
    (
        "helpers_per_reply",
        "Helpers per reply",
        "How many subagents one reply may send, if lower than the instance's "
        "own.",
    ),
)

LIMIT_KEYS = tuple(key for key, _, _ in LIMIT_DEFS)

# Nobody is limited until somebody says so. A quota that arrived with an upgrade
# and started refusing replies would be the worst possible way to introduce one.
NO_LIMITS: dict[str, int] = dict.fromkeys(LIMIT_KEYS, 0)


def limits_for(db: DBSession, user: User | None) -> dict[str, int]:
    """What this user may spend, resolved across their groups.

    **By maximum**, which is the union rule applied to numbers: being in a second
    group can only ever grant more, never less. That is the same promise the
    permissions make, and having one of the two work the other way round is how
    "why can this person not do X" stops being answerable.

    **Zero wins outright**, because zero means "no limit". Taking the plain
    maximum would make a group saying "unlimited" count for less than one saying
    "a million", which is the union rule inverted for exactly one value -- and it
    is the value somebody sets when they mean *stop limiting this person*.

    An administrator is unlimited, for the reason `resolve` gives them every
    permission: they can raise their own quota in two clicks, and pretending
    otherwise is theatre.
    """
    if user is None or user.is_admin:
        return dict(NO_LIMITS)

    resolved = dict(NO_LIMITS)
    for key in LIMIT_KEYS:
        values = []
        for group in user.groups:
            raw = (group.limits_json or {}).get(key)
            if raw is None:
                continue  # no opinion, contributes nothing
            try:
                values.append(max(0, int(raw)))
            except (TypeError, ValueError):
                continue
        if not values or 0 in values:
            resolved[key] = 0
        else:
            resolved[key] = max(values)
    return resolved


def limit(db: DBSession, user: User | None, key: str) -> int:
    return limits_for(db, user).get(key, 0)


def models_visible_to(db: DBSession, user: User | None) -> list[Model]:
    """Models a user may start a chat with, in display order.

    A model is visible when it is enabled, its connection is enabled, and
    either it is public or the user belongs to one of its groups.
    """
    query = (
        select(Model)
        .join(Connection)
        .where(Model.enabled.is_(True), Connection.enabled.is_(True))
        .order_by(Model.position, Model.model_id)
    )
    candidates = list(db.scalars(query))

    if user is not None and user.is_admin:
        return candidates

    if user is None:
        return []

    member_of = {group.id for group in user.groups}
    return [
        model
        for model in candidates
        if model.public or member_of.intersection({g.id for g in model.groups})
    ]


def can_use_model(db: DBSession, user: User | None, model_id: str) -> bool:
    return any(model.model_id == model_id for model in models_visible_to(db, user))