feat(auth): add model.skills.write permission and user_has_permission helper

In-process permission check for model-facing tool exec paths that need
to gate a write capability without HTTP middleware in the loop. Foundation
for the upcoming skills tool refactor: the merged
skills(action=create|update|enable|disable) tool will gate on
model.skills.write before reaching storage.

- Add model.skills.write to _VALID_PERMISSIONS (default-ungranted on every
  role including builtin-admin — operators opt themselves in explicitly)
- Add user_has_permission(user_id, permission, *, storage=None) helper
  that fails-closed on storage outages and short-circuits on empty user_id
- Document service-scope asymmetry with require_permission (no AuthResult
  in the model-tool path → no bypass; explicit guidance if a legitimate
  service-scope caller ever needs to reach here)
- Pin the "no implicit cache" contract with a regression test asserting
  every helper call hits storage (call_count == 2 after two calls)
- Lock the "builtin-admin default-ungranted" invariant with an alembic
  migration test that drives the chain to head and asserts the role's
  permission string omits model.skills.write
- Plus the role-create end-to-end test proving the constant flows through
  the admin endpoint's validator

Roles admin UI changes deferred to the PR that lands the gated tool — no
operator action needed until the capability exists.

Per-call DB hit + warning-log spam on outage deferred to a follow-up PR;
the helper is dead code in this commit, so cache TTL would be sized
against guesswork — better to wait for a real call-rate signal from the
first caller.
This commit is contained in:
Patrick Buckley
2026-05-22 10:10:45 -07:00
parent 03afb82369
commit ecae0f8778
4 changed files with 211 additions and 0 deletions
+141
View File
@@ -1901,3 +1901,144 @@ class TestRequirePermissionServiceScope:
result = require_permission(request, "admin.users")
assert result is not None
assert result.status_code == 401
# ---------------------------------------------------------------------------
# TestUserHasPermission — in-process permission check for tool exec paths
# ---------------------------------------------------------------------------
class TestUserHasPermission:
"""In-process permission helper for model-facing tool exec paths.
Distinct from ``require_permission`` (HTTP-only, returns JSONResponse);
this helper returns a plain bool so tool callers can shape the denial
themselves. Loads permissions through storage on every call — there's
no per-session cache, by design: a permission revocation should take
effect on the next tool call, not require a session restart.
"""
def test_returns_true_when_user_holds_permission(self):
from turnstone.core.auth import user_has_permission
storage = MagicMock()
storage.get_user_permissions.return_value = {"model.skills.write", "read"}
assert user_has_permission("alice", "model.skills.write", storage=storage) is True
def test_returns_false_when_user_lacks_permission(self):
from turnstone.core.auth import user_has_permission
storage = MagicMock()
storage.get_user_permissions.return_value = {"read", "write"}
assert user_has_permission("alice", "model.skills.write", storage=storage) is False
def test_empty_user_id_returns_false_without_storage_lookup(self):
"""Empty user_id short-circuits — no anonymous permission holder."""
from turnstone.core.auth import user_has_permission
storage = MagicMock()
assert user_has_permission("", "model.skills.write", storage=storage) is False
storage.get_user_permissions.assert_not_called()
def test_storage_failure_returns_false_fail_closed(self):
"""Roles backend hiccups must deny, not allow (fail-closed)."""
from turnstone.core.auth import user_has_permission
storage = MagicMock()
storage.get_user_permissions.side_effect = RuntimeError("DB down")
assert user_has_permission("alice", "model.skills.write", storage=storage) is False
def test_unregistered_storage_returns_false(self, monkeypatch):
"""Storage registry returning None (pre-init) denies without raising.
Only the model-tool path can land here — HTTP handlers run after
the auth middleware which already requires storage.
"""
from turnstone.core import auth as _auth_mod
monkeypatch.setattr(
"turnstone.core.storage._registry.get_storage", lambda: None, raising=True
)
assert _auth_mod.user_has_permission("alice", "model.skills.write") is False
def test_each_call_hits_storage_no_implicit_cache(self):
"""Pin the load-bearing 'no caching' contract from the class docstring.
Future refactor that adds an ``lru_cache`` decorator, a per-session
cache, or any process-wide memoization would silently break
revocation latency (an admin revoking ``model.skills.write`` from a
role would see the model still able to write skills until cache
expiry / session restart). If a cache is added intentionally, this
test should be rewritten to assert the invalidation contract — not
deleted.
"""
from turnstone.core.auth import user_has_permission
storage = MagicMock()
storage.get_user_permissions.return_value = {"model.skills.write"}
user_has_permission("alice", "model.skills.write", storage=storage)
user_has_permission("alice", "model.skills.write", storage=storage)
assert storage.get_user_permissions.call_count == 2
# ---------------------------------------------------------------------------
# TestBuiltinAdminDefaultPermissions — lock the "ungranted by default" invariant
# ---------------------------------------------------------------------------
class TestBuiltinAdminDefaultPermissions:
"""Regression guards on what builtin-admin gets out of the box.
The migration chain (008 seed + 017 catch-up + later additive
migrations) is the source of truth for builtin-admin's permission
set. Permissions intentionally ungranted by default — currently
``model.skills.write`` — must stay absent from that chain, or
operators upgrading from older versions silently inherit a
capability they never consented to. Mirrors the
``tests/test_migration_049.py`` pattern: drive Alembic forward
against an isolated SQLite DB and inspect the resulting row.
"""
def _alembic_cfg(self, db_path):
from pathlib import Path
from alembic.config import Config
migrations_dir = str(
Path(__file__).resolve().parent.parent / "turnstone" / "core" / "storage" / "migrations"
)
cfg = Config()
cfg.set_main_option("script_location", migrations_dir)
cfg.set_main_option("sqlalchemy.url", f"sqlite:///{db_path}")
return cfg
def test_model_skills_write_not_in_builtin_admin_after_full_migration(self, tmp_path):
"""After every shipped migration, ``builtin-admin.permissions`` must
not contain ``model.skills.write``. A migration that grants it
breaks the explicit-opt-in security contract documented in the
``_VALID_PERMISSIONS`` block in ``console/server.py``.
"""
import sqlalchemy as sa
from alembic import command
db_path = tmp_path / "perm.db"
cfg = self._alembic_cfg(db_path)
command.upgrade(cfg, "head")
engine = sa.create_engine(f"sqlite:///{db_path}")
try:
with engine.connect() as conn:
row = conn.execute(
sa.text("SELECT permissions FROM roles WHERE role_id = 'builtin-admin'")
).fetchone()
finally:
engine.dispose()
assert row is not None, "builtin-admin role not seeded by migration chain"
perms = {p.strip() for p in (row[0] or "").split(",") if p.strip()}
assert "model.skills.write" not in perms, (
"builtin-admin must NOT hold model.skills.write by default — "
f"got perms={sorted(perms)}. If a migration intentionally "
"added this grant, update the security contract in "
"``console/server.py`` _VALID_PERMISSIONS docstring first."
)
+23
View File
@@ -205,6 +205,29 @@ class TestRoles:
assert resp.status_code == 400
assert "name" in resp.json()["error"].lower()
def test_create_role_with_model_skills_write_permission(self, client):
"""``model.skills.write`` is enumerated in ``_VALID_PERMISSIONS`` and
passes role-create validation. Catches the case where the constant
is added on the server but missed by the validator or the constant
list."""
resp = client.post(
"/v1/api/admin/roles",
json=_role_payload(name="skillwriter", permissions="read,model.skills.write"),
)
assert resp.status_code == 200, resp.json()
assert "model.skills.write" in resp.json()["permissions"]
def test_create_role_rejects_unknown_permission(self, client):
"""Unknown permission strings are rejected — guards the validator
against typos in the constant list and would-be capability inflation
via the admin API."""
resp = client.post(
"/v1/api/admin/roles",
json=_role_payload(name="bogus", permissions="read,model.does.not.exist"),
)
assert resp.status_code == 400
assert "invalid" in resp.json()["error"].lower()
def test_create_role_default_display_name(self, client):
resp = client.post(
"/v1/api/admin/roles",
+8
View File
@@ -5894,6 +5894,14 @@ _VALID_PERMISSIONS = frozenset(
# surface for GET /v1/api/cluster/ws/{ws_id}/detail. Granted
# to builtin-admin via migration 040.
"admin.cluster.inspect",
# Model-facing write capability over the skill catalog. Gates
# the ``skills(action=create|update|enable|disable)`` tool path
# (in-process, not HTTP — distinct from ``admin.skills`` which
# gates admin-UI traffic). Default-ungranted on every role
# including builtin-admin — operators must opt themselves in
# explicitly before their coordinator sessions can mutate the
# catalog.
"model.skills.write",
"tools.approve",
"workstreams.create",
"workstreams.close",
+39
View File
@@ -116,6 +116,45 @@ def _load_user_permissions(storage: Any, user_id: str) -> set[str]:
return set()
def user_has_permission(user_id: str, permission: str, *, storage: Any = None) -> bool:
"""Return True if *user_id* holds *permission*.
For in-process callers — specifically the model-facing tool exec
path — that need to gate a write capability without an HTTP
middleware in the loop. HTTP handlers stay on
:func:`require_permission`, which carries the JSONResponse-shaped
denial. This helper returns a plain bool so the tool layer can
surface the denial in whatever shape it already uses (typically a
``_coord_tool_error`` row).
Empty ``user_id`` returns False without a storage lookup — there's
no anonymous holder of any permission. Storage lookup failures
are swallowed (logged at warning by ``_load_user_permissions``) and
return False — fail-closed on the permission check rather than
fail-open if the roles backend is briefly unavailable.
No service-scope bypass. ``require_permission`` lets a service-
scoped JWT skip the check by default; this helper has no equivalent
because the in-process model-tool path doesn't carry an
:class:`AuthResult` (scopes are an HTTP-layer concept). A service
token reaching here either resolves to a real ``user_id`` with the
grant or has no ``user_id`` and short-circuits to False. If a
legitimate service-scope caller ever needs to bypass, expose
``allow_service_bypass`` here mirroring ``require_permission`` and
thread the originating scope through the call site — don't try to
infer it from the lone ``user_id``.
"""
if not user_id:
return False
if storage is None:
from turnstone.core.storage._registry import get_storage
storage = get_storage()
if storage is None:
return False
return permission in _load_user_permissions(storage, user_id)
def _permissions_to_scopes(permissions: set[str]) -> frozenset[str]:
"""Derive legacy scopes from a granular permission set."""
scopes: set[str] = set()