Grants that outlive what they name, and a rule you can read

sharing.forget_principal has existed since shares did, documented as the thing
that stops a recycled id inheriting somebody's grant, and was called by nobody.
Deleting a group left every grant naming it; deleting an account left both the
grants to it and the grants of its own work -- that second half is the one
nothing else could catch, since their rows cascade and the shares of those rows
have nothing to cascade from. Both now run before the delete, while the rows are
still findable, and a deleted resource forgets its own.

library.share defaulted to False, which meant sharing shipped documented as done
and unreachable: the panel only renders for somebody holding it, so out of the
box nobody could share anything and nothing said why. It is on.

The panel itself was checkboxes inside the resource's *save form*, listing every
group and every account on the instance, unpaginated, on every detail page -- and
a tick only took effect if you also saved the resource. It is its own routes now:
search, one grant per POST, the panel re-rendered from what is stored. Anything
already shared stays listed whatever the search says, or removing a grant would
mean searching for the name it was given to.

Reports join the shareable set and memories still do not: a finished piece of
work is the thing somebody most wants to hand over, and a record about a person
is not content to pass round. reports.visible became sharing.visible_to, which is
the one line its own docstring predicted. Two things fell out: `owned` beside
`get`, because sharing grants reading and deleting is the owner's alone; and
reading somebody else's report no longer clears their unread dot.

Permissions gained the answer to "what can this person actually do?" --
explain() is resolve()'s working shown rather than thrown away, naming admin, the
baseline, or the groups that granted each one. That is the simulation the union
rule exists to make unnecessary, and until now the only way to get it was to open
every group and read the grids by eye. Users and groups are list-plus-detail, and
membership is edited from one side: it was on both, and a full-form POST from
either overwrote what the other had shown.

Read and write are split for notes, memory and skills -- checked on the tool's
declared risk, after the gate so it can only narrow, and defaulting on.

Quotas are the union rule applied to numbers, with the corner that makes it
interesting: zero means "no limit" and wins outright, or a group saying unlimited
would count for less than one saying a million. Absent means "no opinion".
_narrower folds a group's ceiling with the instance's and is deliberately not
min, for the same reason. Five axes, enforced where each is knowable -- before a
reply is built, before a second one starts, on an agent reply's clock, before a
minute of GPU, and beside the helper cap -- and usage is recorded even for a
reply that was stopped or errored, because an endpoint charges either way.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Jaroslav Beneš
2026-08-06 16:48:14 +02:00
parent 20bb569b00
commit 1b8c9f948c
31 changed files with 2226 additions and 377 deletions
+250
View File
@@ -0,0 +1,250 @@
"""What an account may spend, and the arithmetic that decides.
The rule to hold is the one the permissions already hold, applied to numbers:
**a second group can only ever grant more.** Its awkward corner is zero, which
means "no limit" — so the maximum has to be taken with zero winning outright, or
a group saying "unlimited" would count for less than one saying "a million".
"""
from __future__ import annotations
import pytest
from sqlalchemy import select
from lembas.db.models import ROLE_USER, Group, Usage, User
from lembas.security import permissions
from lembas.services import usage as usage_service
@pytest.fixture
def reader(db, registered) -> User:
person = User(
email="sam@shire.test", name="Sam", password_hash="x", role=ROLE_USER # noqa: S106
)
db.add(person)
db.commit()
return person
def _group(db, reader, **limits) -> Group:
group = Group(name=f"g{len(limits)}-{id(limits)}", limits_json=limits)
group.users.append(reader)
db.add(group)
db.commit()
return group
# --- Resolution -----------------------------------------------------------------
def test_nobody_is_limited_until_somebody_says_so(db, reader):
"""A quota that arrived with an upgrade and started refusing replies would
be the worst possible way to introduce one."""
assert permissions.limits_for(db, reader) == permissions.NO_LIMITS
def test_a_second_group_can_only_grant_more(db, reader):
_group(db, reader, monthly_tokens=1000)
_group(db, reader, monthly_tokens=5000)
db.refresh(reader)
assert permissions.limit(db, reader, "monthly_tokens") == 5000
def test_zero_means_no_limit_and_wins_outright(db, reader):
"""The union rule's awkward corner. A plain maximum would make "unlimited"
count for less than "a million", which is the rule inverted for exactly the
value somebody sets when they mean *stop limiting this person*."""
_group(db, reader, monthly_tokens=1000)
_group(db, reader, monthly_tokens=0)
db.refresh(reader)
assert permissions.limit(db, reader, "monthly_tokens") == 0
def test_a_group_with_no_opinion_contributes_nothing(db, reader):
"""Absent is not zero. If it were, adding a group that says nothing about
tokens would silently make somebody unlimited."""
_group(db, reader, monthly_tokens=1000)
_group(db, reader, images_per_day=5)
db.refresh(reader)
assert permissions.limit(db, reader, "monthly_tokens") == 1000
assert permissions.limit(db, reader, "images_per_day") == 5
def test_nonsense_in_the_column_is_ignored(db, reader):
"""A row can be written by hand or by an older version, and a limit nobody
can parse must not become a limit of zero — which is *unlimited* here, the
opposite of failing safe."""
_group(db, reader, monthly_tokens="lots")
_group(db, reader, monthly_tokens=1000)
db.refresh(reader)
assert permissions.limit(db, reader, "monthly_tokens") == 1000
def test_an_administrator_is_unlimited(db, registered):
admin = db.scalars(select(User).order_by(User.created_at)).first()
group = Group(name="tight", limits_json={"monthly_tokens": 1})
group.users.append(admin)
db.add(group)
db.commit()
assert permissions.limits_for(db, admin) == permissions.NO_LIMITS
# --- Recording ------------------------------------------------------------------
def test_usage_accumulates_into_one_row_per_month(db, reader):
usage_service.record(db, reader.id, prompt_tokens=100, completion_tokens=50)
usage_service.record(db, reader.id, prompt_tokens=10, completion_tokens=5)
db.commit()
rows = list(db.scalars(select(Usage).where(Usage.user_id == reader.id)))
assert len(rows) == 1
assert rows[0].prompt_tokens == 110
assert rows[0].completion_tokens == 55
assert rows[0].replies == 2
assert usage_service.month_tokens(db, reader.id) == 165
def test_recording_never_raises(db):
"""Bookkeeping that broke a reply would be worse than no bookkeeping."""
usage_service.record(db, "", prompt_tokens=10)
usage_service.record(db, "nobody-at-all", prompt_tokens=10)
def test_the_token_gate_answers_before_anything_is_spent(db, reader):
_group(db, reader, monthly_tokens=100)
db.refresh(reader)
assert usage_service.over_token_budget(db, reader) == ""
usage_service.record(db, reader.id, prompt_tokens=60, completion_tokens=60)
db.commit()
reason = usage_service.over_token_budget(db, reader)
assert "month" in reason
assert "100" in reason
def test_an_unlimited_account_is_never_over(db, reader):
usage_service.record(db, reader.id, prompt_tokens=10**9)
db.commit()
assert usage_service.over_token_budget(db, reader) == ""
# --- Narrowing ------------------------------------------------------------------
@pytest.mark.parametrize(
("instance", "quota", "expected"),
[
(900, 300, 300), # the group is tighter
(300, 900, 300), # the instance is tighter
(0, 300, 300), # the instance has no limit
(300, 0, 300), # the group has no opinion
(0, 0, 0), # neither
],
)
def test_two_ceilings_fold_to_the_tighter_one(instance, quota, expected):
"""Not `min`: a zero on either side would win and turn "no opinion" into
"no time at all". Written once because getting it wrong in one place is a
limit that silently stops working."""
from lembas.services.generation import _narrower
assert _narrower(instance, quota) == expected
# --- The screens ----------------------------------------------------------------
def test_a_group_stores_only_the_boxes_that_were_filled(db, client, registered, reader):
group = Group(name="team")
db.add(group)
db.commit()
client.post(
f"/admin/groups/{group.id}",
data={
"name": "team",
"limit_monthly_tokens": "5000",
"limit_images_per_day": "",
"limit_concurrent_replies": "not a number",
},
follow_redirects=False,
)
db.refresh(group)
assert group.limits_json == {"monthly_tokens": 5000}
def test_the_user_page_says_where_each_permission_came_from(db, client, registered, reader):
"""The question the grids could not answer without opening every group by
eye — which is exactly the simulation the union rule exists to avoid."""
group = Group(name="writers", permissions_json={"tools.notes.write": True})
group.users.append(reader)
db.add(group)
db.commit()
explained = permissions.explain(db, reader)
assert explained["tools.notes.write"]["on"] is True
assert "writers" in explained["tools.notes.write"]["source"]
# And something the baseline gives says baseline rather than naming a group.
assert explained["chat.create"]["source"] == ["baseline"]
page = client.get(f"/admin/users/{reader.id}").text
assert "writers" in page
assert "What this account can do" in page
def test_an_administrator_is_explained_as_bypassing(db, registered):
admin = db.scalars(select(User).order_by(User.created_at)).first()
explained = permissions.explain(db, admin)
assert all(entry["source"] == ["admin"] for entry in explained.values())
def test_membership_is_edited_from_the_group_and_not_the_user(db, client, registered, reader):
"""Two controls writing one value is how each becomes the answer to "why did
my change not stick?". The user page links to the group instead."""
group = Group(name="team")
db.add(group)
db.commit()
# No control for it on the user's page.
assert 'name="group_ids"' not in client.get(f"/admin/users/{reader.id}").text
# And a POST that tries anyway changes nothing.
client.post(
f"/admin/users/{reader.id}",
data={"name": "Sam", "role": "user", "active": "true", "group_ids": [group.id]},
follow_redirects=False,
)
db.refresh(reader)
assert reader.groups == []
client.post(
f"/admin/groups/{group.id}",
data={"name": "team", "user_ids": [reader.id]},
follow_redirects=False,
)
db.refresh(reader)
assert [g.name for g in reader.groups] == ["team"]
# Once they are in it, their page links to where it is edited.
assert f"/admin/groups/{group.id}" in client.get(f"/admin/users/{reader.id}").text
def test_the_user_list_is_paginated_and_searchable(db, client, registered):
for n in range(30):
db.add(
User(
email=f"p{n}@shire.test",
name=f"Person {n}",
password_hash="x", # noqa: S106
)
)
db.commit()
first = client.get("/admin/users").text
assert "Page 1 of" in first
found = client.get("/admin/users?q=Person+7").text
assert "Person 7" in found
assert "Person 12" not in found
+289
View File
@@ -0,0 +1,289 @@
"""Sharing: what it grants, what it does not, and what it leaves behind.
Half of this file is about grants that outlive what they name. `Share` carries
no foreign key in either direction — `principal_id` points at a user *or* a
group and `resource_id` at one of four tables, neither of which SQLite can
express — so nothing cascades, and every delete has to say so explicitly.
`sharing.forget_principal` existed for exactly this and was called by nobody.
"""
from __future__ import annotations
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import select
from lembas.db.models import (
ROLE_USER,
Group,
Note,
Report,
Share,
User,
)
from lembas.security import permissions
from lembas.services import reports as reports_service
from lembas.services import sharing
from lembas.services.library import notes as notes_service
@pytest.fixture
def owner(db, registered) -> User:
return db.scalars(select(User).order_by(User.created_at)).first()
@pytest.fixture
def reader(db) -> User:
person = User(
email="sam@shire.test", name="Sam", password_hash="x", role=ROLE_USER # noqa: S106
)
db.add(person)
db.commit()
return person
# --- Dangling grants ------------------------------------------------------------
def test_deleting_a_group_drops_the_shares_naming_it(db, client, registered, owner, reader):
"""It never did. A group id is a random hex string that nothing reissues
today and nothing promises not to reissue tomorrow."""
group = Group(name="team")
group.users.append(reader)
db.add(group)
db.commit()
note = notes_service.create(db, owner=owner, title="Secret", body="mellon")
sharing.set_grants(db, note, user_ids=[], group_ids=[group.id])
assert db.scalars(select(Share)).all()
client.post(f"/admin/groups/{group.id}/delete", follow_redirects=False)
db.expire_all()
assert db.scalars(select(Share)).all() == []
def test_deleting_an_account_drops_both_halves(db, client, registered, owner, reader):
"""Grants **to** them, and grants **of** their own work. The second is the
one nothing else could catch: their rows cascade, and the shares of those
rows have nothing to cascade from."""
theirs = notes_service.create(db, owner=reader, title="Theirs", body="x")
sharing.set_grants(db, theirs, user_ids=[owner.id], group_ids=[])
mine = notes_service.create(db, owner=owner, title="Mine", body="y")
sharing.set_grants(db, mine, user_ids=[reader.id], group_ids=[])
assert len(db.scalars(select(Share)).all()) == 2
client.post(f"/admin/users/{reader.id}/delete", follow_redirects=False)
db.expire_all()
assert db.scalars(select(Share)).all() == []
def test_deleting_a_report_drops_its_grants(db, owner, reader):
report = reports_service.create(db, owner=owner, title="Findings", body="x")
sharing.set_grants(db, report, user_ids=[reader.id], group_ids=[])
reports_service.delete(db, report)
db.expire_all()
assert db.scalars(select(Share)).all() == []
# --- Reports ---------------------------------------------------------------------
def test_a_shared_report_is_visible_and_not_deletable(db, owner, reader):
"""Sharing grants reading. Being able to see a report is not being able to
delete it out from under the person who filed it."""
report = reports_service.create(db, owner=owner, title="Findings", body="x")
sharing.set_grants(db, report, user_ids=[reader.id], group_ids=[])
assert reports_service.get(db, report.id, reader) is not None
assert reports_service.owned(db, report.id, reader) is None
assert reports_service.owned(db, report.id, owner) is not None
def test_an_unshared_report_stays_invisible(db, owner, reader):
report = reports_service.create(db, owner=owner, title="Findings", body="x")
assert reports_service.get(db, report.id, reader) is None
assert list(db.scalars(reports_service.visible(reader))) == []
def test_a_shared_report_appears_in_the_feed_and_the_filter(db, owner, reader):
report = reports_service.create(db, owner=owner, title="Findings", body="x")
theirs = reports_service.create(db, owner=reader, title="Mine", body="y")
sharing.set_grants(db, report, user_ids=[reader.id], group_ids=[])
everything = {r.id for r in db.scalars(reports_service.visible(reader))}
only_shared = {
r.id for r in db.scalars(select(Report).where(sharing.only_shared(Report, reader)))
}
assert everything == {report.id, theirs.id}
assert only_shared == {report.id}
def test_reading_somebody_elses_report_does_not_clear_their_dot(db, client, registered, owner):
"""`unread` is the owner's notification. Somebody it was shared with opening
it would silence a dot meant for a person who has not seen it."""
other = User(
email="sam@shire.test", name="Sam", password_hash="x", role=ROLE_USER # noqa: S106
)
db.add(other)
db.commit()
report = reports_service.create(db, owner=other, title="Theirs", body="x")
report.unread = True
db.commit()
sharing.set_grants(db, report, user_ids=[owner.id], group_ids=[])
assert client.get(f"/reports/{report.id}").status_code == 200
db.refresh(report)
assert report.unread is True
# --- Shared with me --------------------------------------------------------------
def test_only_shared_excludes_your_own(db, owner, reader):
mine = notes_service.create(db, owner=reader, title="Mine", body="x")
theirs = notes_service.create(db, owner=owner, title="Theirs", body="y")
sharing.set_grants(db, theirs, user_ids=[reader.id], group_ids=[])
rows = {n.id for n in db.scalars(select(Note).where(sharing.only_shared(Note, reader)))}
assert rows == {theirs.id}
assert mine.id not in rows
def test_the_filter_is_a_url_you_can_keep(db, client, registered):
for path in ("/library/notes", "/library/skills", "/library/knowledge", "/reports"):
page = client.get(f"{path}?shared=1")
assert page.status_code == 200, path
assert "Shared with me" in page.text, path
# --- The panel -------------------------------------------------------------------
def test_a_grant_is_stored_the_moment_it_is_made(db, client, registered, owner, reader):
"""It used to ride along with the resource's save form, so ticking a box and
navigating away did nothing — silently."""
note = notes_service.create(db, owner=owner, title="Note", body="x")
response = client.post(
f"/api/library/share/note/{note.id}",
data={"principal_type": "user", "principal_id": reader.id, "on": "true"},
)
assert response.status_code == 200
assert sharing.can_read(db, note, reader) is True
# And the answer is the panel, showing what is now true.
assert "Shared with" in response.text
def test_a_grant_can_be_taken_back(db, client, registered, owner, reader):
note = notes_service.create(db, owner=owner, title="Note", body="x")
sharing.set_grants(db, note, user_ids=[reader.id], group_ids=[])
client.post(
f"/api/library/share/note/{note.id}",
data={"principal_type": "user", "principal_id": reader.id, "on": "false"},
)
db.expire_all()
assert sharing.can_read(db, note, reader) is False
def test_the_panel_searches_rather_than_listing_everybody(db, client, registered, owner):
"""It rendered every group and every account on the instance, unpaginated,
on every detail page."""
for n in range(30):
db.add(
User(
email=f"p{n}@shire.test",
name=f"Person {n}",
password_hash="x", # noqa: S106
)
)
db.commit()
note = notes_service.create(db, owner=owner, title="Note", body="x")
everything = client.get(f"/api/library/share/note/{note.id}").text
assert everything.count('name="principal_id"') == 0 # values ride in hx-vals
assert "Person 29" not in everything
found = client.get(f"/api/library/share/note/{note.id}?q=Person+29").text
assert "Person 29" in found
def test_an_existing_grant_stays_listed_whatever_the_search_says(
db, client, registered, owner, reader
):
"""Otherwise the only way to remove a grant would be to search for the name
it was given to."""
note = notes_service.create(db, owner=owner, title="Note", body="x")
sharing.set_grants(db, note, user_ids=[reader.id], group_ids=[])
page = client.get(f"/api/library/share/note/{note.id}?q=nobody-matches-this").text
assert "sam@shire.test" in page
def test_somebody_it_was_shared_with_cannot_share_it_on(db, client, registered, owner, reader):
"""What keeps "who can see this?" answerable by asking one person."""
note = notes_service.create(db, owner=owner, title="Note", body="x")
sharing.set_grants(db, note, user_ids=[reader.id], group_ids=[])
other = TestClient(client.app)
other.post(
"/auth/register",
data={"name": "Merry", "email": "merry@shire.test", "password": "a-fine-second-breakfast"},
follow_redirects=False,
)
# Signed in as somebody else entirely: the panel is not theirs to open.
assert other.get(f"/api/library/share/note/{note.id}").status_code == 404
def test_a_grant_naming_nothing_is_refused(db, client, registered, owner):
"""A crafted id would write a grant that is invisible in the panel and
unremovable from it."""
note = notes_service.create(db, owner=owner, title="Note", body="x")
client.post(
f"/api/library/share/note/{note.id}",
data={"principal_type": "user", "principal_id": "not-a-real-id", "on": "true"},
)
assert db.scalars(select(Share)).all() == []
def test_sharing_is_on_by_default(db, registered, owner):
"""It was off, which meant sharing shipped documented as done and
unreachable: the panel only renders for somebody who holds this."""
assert permissions.has(db, owner, "library.share") is True
assert permissions.DEFAULT_PERMISSIONS["library.share"] is True
# --- Read and write, split -------------------------------------------------------
def test_a_reader_can_search_notes_and_not_write_them(db, reader):
from lembas.db.models import Chat, Connection, Model
from lembas.services import settings_store
from lembas.services import tools as tools_service
from lembas.services.crypto import encrypt
connection = Connection(name="c", base_url="http://127.0.0.1:1", api_key_encrypted=encrypt(""))
db.add(connection)
db.commit()
db.add(Model(connection_id=connection.id, model_id="m", capabilities_json={"tools": True}))
db.commit()
chat = Chat(user_id=reader.id, model_id="m", connection_id=connection.id)
db.add(chat)
db.commit()
settings_store.update(db, {"default_permissions": {"tools.notes.write": False}})
offered = {tool.name for tool in tools_service.resolve_tools(db, chat, reader).defs}
assert "notes_search" in offered
assert "notes_get" in offered
assert "notes_create" not in offered
assert "notes_edit" not in offered
assert "notes_delete" not in offered
def test_the_write_half_defaults_on(db, reader):
"""An instance that never looks behaves exactly as it did."""
for key in ("tools.notes.write", "tools.memory.write", "tools.skills.write"):
assert permissions.DEFAULT_PERMISSIONS[key] is True