Ask what the caller may do, not what its owner may do

Wave 4, first half. A personal API token scoped to endpoints.command alone
could queue system_restart, run_script and update_app against a shop-floor PC.

require_permission has always suspended the admin bypass for a scoped token,
and tests pin that. What nothing covered was a route checking a SECOND
permission by hand: the endpoints command route asked
current_user.haspermission(elevated), which reads the token's OWNER. An admin
minting a narrow token for a script therefore handed out remote restart, script
execution and app deployment - the exact bypass require_permission exists to
prevent, reintroduced one hand-rolled check at a time.

caller_holds is now scope-aware, so the answer is about the caller rather than
the person behind it: the owner must hold the permission AND, when the request
carries a scoped token, the scope list must name it. An unscoped token is the
person and is unchanged; a session is unchanged; guests still answer from the
guest role.

Both hand-rolled checks in the tree now use it - the endpoints elevated gate and
the secure-send tools.secrets check, which had the same shape and the same
consequence at a smaller scale.

Two tests, one proving the refusal and one proving an unscoped token still
passes (a fix that quietly broke every service token would be worse than the
bug). Verified by planting the owner-based check back: the first fails.

Reported by the fleet-agent lane of the 2026-09-03 test run and confirmed by an
independent verifier that minted its own token and queued its own commands.

Scope: product

plugins/endpoints/api/routes.py

@@ -1376,8 +1376,13 @@ def enqueue_endpoint_command(asset_id: int):
     # Destructive actions (for example system_restart) demand a stricter grant
     # ON TOP OF endpoints.command, so a remote reboot cannot ride the routine
     # toggle permission.
+    # caller_holds, not current_user.haspermission: the question is what THIS
+    # caller may do, and a scoped personal token is a narrower caller than its
+    # owner. Asking the owner let a token scoped to endpoints.command queue a
+    # remote restart, a script run and an app deploy - the exact bypass
+    # require_permission exists to prevent, reintroduced by checking by hand.
     elevated = elevated_permission_for(action)
-    if elevated and not current_user.haspermission(elevated):
+    if elevated and not caller_holds(elevated):
         return error_response(
             ErrorCodes.FORBIDDEN,
             f'Action {action!r} requires the {elevated} permission',

plugins/tools/api/routes.py

@@ -39,6 +39,7 @@ from flask_jwt_extended import jwt_required, get_jwt_identity, current_user
 from shopdb.api import (
     db, cache, Setting, client_ip,
     success_response, error_response, ErrorCodes,
+    caller_holds,
 )
 
 from ..models import SecureSecret
@@ -155,7 +156,9 @@ def create_secret():
                 ErrorCodes.VALIDATION_ERROR,
                 'Too many links from this address. Try again shortly.',
                 http_code=429)
-    elif not current_user.haspermission('tools.secrets'):
+    # caller_holds, for the same reason as the endpoints elevated gate: a
+    # scoped token is a narrower caller than the person who minted it.
+    elif not caller_holds('tools.secrets'):
         return error_response(
             ErrorCodes.FORBIDDEN,
             'You do not have permission to create secret links.',

shopdb/utils/authz.py

@@ -102,14 +102,27 @@ def caller_holds(permission_name: str) -> bool:
     Guests are included on purpose: a signed-out visitor holds whatever the
     guest role was granted, so the answer must come from the same place
     require_permission gets it.
+
+    Scoped personal tokens are included for the same reason, and this is the
+    half that was missing: the answer has to be about the CALLER, and a scoped
+    token is a narrower caller than its owner. Reading the owner's permissions
+    let a token scoped to endpoints.command queue a remote restart, because the
+    elevated check asked the owner rather than the token - the one bypass
+    require_permission exists to prevent, reintroduced by a route that did its
+    own check.
     """
     user = currentuser()
     if user is None:
         return guestholds(permission_name)
     try:
-        return bool(user.haspermission(permission_name))
+        if not user.haspermission(permission_name):
+            return False
+        patscopes = get_jwt().get('patscopes')
     except Exception:
         return False
+    if patscopes is not None:
+        return permission_name in patscopes
+    return True
 
 
 def require_permission(permission_name: str):

tests/test_core/test_apitokens.py

@@ -417,3 +417,45 @@ def test_update_scopes_ceiling_enforced(client, db):
                           headers=headers)
     assert response.status_code == 400
     assert 'applications.delete' in response.get_json()['data']['error']['message']
+
+
+# --- a route that does its own permission check --------------------------
+#
+# require_permission suspends the admin bypass for a scoped token, and the
+# tests above pin that. What they could not see is a route that checks a
+# SECOND permission by hand: the endpoints command route asked
+# current_user.haspermission(elevated), which reads the OWNER. A token scoped
+# to endpoints.command alone could therefore queue system_restart, run_script
+# and update_app on a shop-floor PC - the exact bypass this file exists to
+# prevent, reintroduced one route at a time.
+#
+# The fix is in caller_holds, so this tests the helper rather than only the one
+# route: it is the answer every hand-rolled check should be using.
+
+def test_a_scoped_token_cannot_pass_a_hand_rolled_permission_check(
+        client, db, admin_user, auth_headers):
+    """The secure-send route checks tools.secrets itself rather than through
+    require_permission. A scoped token whose owner holds that permission must
+    still be refused, because the caller is the token."""
+    secret = _create_token(
+        client, auth_headers,
+        scopes=['applications.create']).get_json()['data']['secret']
+
+    refused = client.post('/api/tools/secrets', json={'content': 'hello'},
+                          headers=_pat_headers(secret))
+    assert refused.status_code == 403, refused.get_json()
+
+
+def test_an_unscoped_token_still_passes_that_check(client, db, admin_user,
+                                                   auth_headers):
+    """The narrowing applies ONLY to scoped tokens - an unscoped one is the
+    person, and must keep working."""
+    secret = _create_token(
+        client, auth_headers, scopes=None).get_json()['data']['secret']
+
+    allowed = client.post('/api/tools/secrets', json={'content': 'hello'},
+                          headers=_pat_headers(secret))
+    # Not 403 is the whole assertion: the payload is deliberately incomplete
+    # (secure send wants ciphertext and iv), so a 400 proves the request got
+    # PAST the permission check, which is what this is about.
+    assert allowed.status_code != 403, allowed.get_json()