Add optional permission scopes to API tokens

A token may carry a scopes list: it then grants only those permissions,
intersected with what the owner holds at use time, with the admin role
bypass suspended and role-gated routes denied - a scoped token from an
admin account is genuinely limited. Scope ceiling enforced at
create/update too (only permissions the owner holds; 400 lists
violations) and the picker only offers what you hold. Token management
itself now requires the new apitokens.create permission (admin by
default, grantable via roles). Unscoped tokens keep the exact prior
act-as-owner behavior; imports need an unscoped admin token.
Migration 7d22.

756 tests pass; live-verified scoped 201/403 matrix.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

tests/test_core/test_apitokens.py

@@ -10,7 +10,9 @@ admin.
 
 from datetime import datetime, timedelta, timezone
 
-from shopdb.core.models import ApiToken, Vendor
+from werkzeug.security import generate_password_hash
+
+from shopdb.core.models import ApiToken, Application, Vendor
 from shopdb.extensions import db as _db
 
 
@@ -18,10 +20,12 @@ def _naive_utcnow():
     return datetime.now(timezone.utc).replace(tzinfo=None)
 
 
-def _create_token(client, headers, name='test token', expiresat=None):
+def _create_token(client, headers, name='test token', expiresat=None, scopes=None):
     body = {'name': name}
     if expiresat is not None:
         body['expiresat'] = expiresat
+    if scopes is not None:
+        body['scopes'] = scopes
     response = client.post('/api/apitokens', json=body, headers=headers)
     return response
 
@@ -30,6 +34,31 @@ def _pat_headers(secret):
     return {'Authorization': f'Bearer {secret}'}
 
 
+def _user_with_perms(client, db, username, permnames):
+    """Create a non-admin user holding permnames, return (user, login headers).
+
+    Seeds the catalog so the named Permission rows exist, links them to a fresh
+    role, and logs the user in. Used to prove the apitokens.create gate and the
+    owner scope ceiling for non-admin owners.
+    """
+    from shopdb.core.models import User, Role, Permission
+    Permission.seed()
+    role = Role(rolename=username + 'role')
+    db.session.add(role)
+    db.session.flush()
+    role.permissions = Permission.query.filter(
+        Permission.name.in_(permnames)).all()
+    user = User(username=username, email=username + '@test.local',
+                passwordhash=generate_password_hash('testpass'))
+    user.roles.append(role)
+    db.session.add(user)
+    db.session.commit()
+    login = client.post('/api/auth/login',
+                        json={'username': username, 'password': 'testpass'})
+    headers = {'Authorization': f"Bearer {login.get_json()['data']['access_token']}"}
+    return user, headers
+
+
 def test_create_returns_secret_once_and_stores_hash(client, db, auth_headers):
     response = _create_token(client, auth_headers, name='import runner')
     assert response.status_code == 201
@@ -57,10 +86,12 @@ def test_pat_authenticates_permission_write_as_owner(client, db, admin_user,
     assert Vendor.query.filter_by(vendor='PAT Vendor').first() is not None
 
 
-def test_pat_403_when_owner_lacks_permission(client, db, member_user,
-                                             member_headers):
-    """A PAT owned by a role-less member is forbidden from an admin write."""
-    secret = _create_token(client, member_headers).get_json()['data']['secret']
+def test_pat_403_when_owner_lacks_permission(client, db):
+    """A PAT owned by a non-admin (who can mint tokens but is not admin) is
+    forbidden from an admin-role write."""
+    _user, headers = _user_with_perms(client, db, 'tokenmaker',
+                                      ['apitokens.create'])
+    secret = _create_token(client, headers).get_json()['data']['secret']
 
     response = client.post('/api/vendors', json={'vendor': 'Nope'},
                            headers=_pat_headers(secret))
@@ -110,32 +141,45 @@ def test_lastusedat_updates_on_use(client, db, admin_user, auth_headers):
 
 
 def test_member_cannot_revoke_other_users_token(client, db, admin_user,
-                                                auth_headers, member_headers):
-    """A role-less member cannot revoke a token owned by a different user."""
+                                                auth_headers):
+    """A token-capable non-admin cannot revoke a token owned by someone else."""
     secret = _create_token(client, auth_headers).get_json()['data']['secret']
     tokenid = ApiToken.query.filter_by(
         tokenhash=ApiToken.hash_secret(secret)).first().tokenid
 
-    response = client.delete(f'/api/apitokens/{tokenid}', headers=member_headers)
+    _user, headers = _user_with_perms(client, db, 'tokenmaker',
+                                      ['apitokens.create'])
+    response = client.delete(f'/api/apitokens/{tokenid}', headers=headers)
     assert response.status_code == 403
     # Still active.
     assert _db.session.get(ApiToken, tokenid).isactive is True
 
 
-def test_member_can_manage_own_token(client, db, member_user, member_headers):
-    """By design any authed user manages their OWN tokens."""
-    create = _create_token(client, member_headers, name='mine')
+def test_roleless_member_cannot_create_token(client, db, member_user,
+                                             member_headers):
+    """Creating a token now requires apitokens.create; a role-less member is
+    denied with 403."""
+    response = _create_token(client, member_headers, name='nope')
+    assert response.status_code == 403
+
+
+def test_member_with_permission_can_manage_own_token(client, db):
+    """A non-admin granted apitokens.create manages their OWN tokens."""
+    _user, headers = _user_with_perms(client, db, 'tokenmaker',
+                                      ['apitokens.create'])
+    create = _create_token(client, headers, name='mine')
     assert create.status_code == 201
     tokenid = create.get_json()['data']['tokenid']
 
-    revoke = client.delete(f'/api/apitokens/{tokenid}', headers=member_headers)
+    revoke = client.delete(f'/api/apitokens/{tokenid}', headers=headers)
     assert revoke.status_code == 200
     assert _db.session.get(ApiToken, tokenid).isactive is False
 
 
-def test_admin_all_true_lists_everyone(client, db, admin_user, auth_headers,
-                                       member_user, member_headers):
+def test_admin_all_true_lists_everyone(client, db, admin_user, auth_headers):
     _create_token(client, auth_headers, name='admin token')
+    member, member_headers = _user_with_perms(client, db, 'tokenmaker',
+                                              ['apitokens.create'])
     _create_token(client, member_headers, name='member token')
 
     # Own-only (default) for admin: just the admin's token.
@@ -146,19 +190,20 @@ def test_admin_all_true_lists_everyone(client, db, admin_user, auth_headers,
     everyone = client.get('/api/apitokens?all=true',
                           headers=auth_headers).get_json()['data']
     userids = {t['userid'] for t in everyone}
-    assert admin_user.userid in userids and member_user.userid in userids
+    assert admin_user.userid in userids and member.userid in userids
     assert any(t.get('username') for t in everyone)
 
 
-def test_member_all_true_ignored(client, db, member_user, member_headers,
-                                 admin_user, auth_headers):
+def test_member_all_true_ignored(client, db, admin_user, auth_headers):
     """A non-admin passing ?all=true still only sees their own tokens."""
     _create_token(client, auth_headers, name='admin token')
+    member, member_headers = _user_with_perms(client, db, 'tokenmaker',
+                                              ['apitokens.create'])
     _create_token(client, member_headers, name='member token')
 
     result = client.get('/api/apitokens?all=true',
                         headers=member_headers).get_json()['data']
-    assert all(t['userid'] == member_user.userid for t in result)
+    assert all(t['userid'] == member.userid for t in result)
 
 
 def test_import_mode_works_over_pat(client, db, admin_user, auth_headers):
@@ -176,3 +221,183 @@ def test_import_mode_works_over_pat(client, db, admin_user, auth_headers):
     vendor = Vendor.query.filter_by(vendor='Legacy Vendor').first()
     assert vendor is not None
     assert vendor.createddate == datetime(2019, 1, 2, 3, 4, 5)
+
+
+# --- Scoped tokens -------------------------------------------------------
+
+def test_scoped_token_passes_listed_permission(client, db, admin_user,
+                                               auth_headers):
+    """A token scoped to applications.create can create an application."""
+    secret = _create_token(client, auth_headers,
+                           scopes=['applications.create']).get_json()['data']['secret']
+
+    response = client.post('/api/applications', json={'appname': 'Scoped App'},
+                           headers=_pat_headers(secret))
+    assert response.status_code == 201
+    assert Application.query.filter_by(appname='Scoped App').first() is not None
+
+
+def test_scoped_token_403_on_unlisted_permission(client, db, admin_user,
+                                                 auth_headers):
+    """A token scoped to only applications.create cannot delete (needs
+    applications.delete, which is not in scope)."""
+    app = Application(appname='Target App')
+    _db.session.add(app)
+    _db.session.commit()
+    appid = app.appid
+
+    secret = _create_token(client, auth_headers,
+                           scopes=['applications.create']).get_json()['data']['secret']
+    response = client.delete(f'/api/applications/{appid}',
+                             headers=_pat_headers(secret))
+    assert response.status_code == 403
+    assert _db.session.get(Application, appid) is not None
+
+
+def test_scoped_admin_token_does_not_bypass(client, db, admin_user, auth_headers):
+    """An admin owner's scoped token is genuinely limited: the admin permission
+    bypass is suspended, so an unlisted action is 403 even for an admin."""
+    secret = _create_token(client, auth_headers,
+                           scopes=['applications.create']).get_json()['data']['secret']
+
+    # settings.edit is not scoped -> 403 despite the owner being admin.
+    response = client.put('/api/settings/site_name',
+                          json={'value': 'Hacked'}, headers=_pat_headers(secret))
+    assert response.status_code == 403
+
+
+def test_scoped_token_denied_on_require_role_route(client, db, admin_user,
+                                                  auth_headers):
+    """Scoped tokens cannot pass a role gate. Vendor writes are require_role
+    admin, so even an admin-owned scoped token is denied."""
+    secret = _create_token(client, auth_headers,
+                           scopes=['applications.create']).get_json()['data']['secret']
+
+    response = client.post('/api/vendors', json={'vendor': 'RoleGated'},
+                           headers=_pat_headers(secret))
+    assert response.status_code == 403
+    assert Vendor.query.filter_by(vendor='RoleGated').first() is None
+
+
+def test_unscoped_token_unchanged(client, db, admin_user, auth_headers):
+    """Regression: an unscoped admin token still acts with full authority,
+    including role-gated writes."""
+    secret = _create_token(client, auth_headers).get_json()['data']['secret']
+
+    response = client.post('/api/vendors', json={'vendor': 'Unscoped Vendor'},
+                           headers=_pat_headers(secret))
+    assert response.status_code == 201
+    token = ApiToken.query.filter_by(
+        tokenhash=ApiToken.hash_secret(secret)).first()
+    assert token.scopelist is None
+
+
+def test_unknown_scope_name_rejected_at_create(client, db, admin_user,
+                                               auth_headers):
+    """An unknown permission name in scopes is rejected 400, listing it."""
+    response = _create_token(client, auth_headers, scopes=['not.a.permission'])
+    assert response.status_code == 400
+    assert 'not.a.permission' in response.get_json()['data']['error']['message']
+
+
+def test_scope_ceiling_rejects_permission_owner_lacks(client, db):
+    """The scope ceiling: a non-admin cannot mint a token scoped to a
+    permission they do not hold."""
+    _user, headers = _user_with_perms(client, db, 'tokenmaker',
+                                      ['apitokens.create'])
+    response = _create_token(client, headers, scopes=['applications.create'])
+    assert response.status_code == 400
+    assert 'applications.create' in response.get_json()['data']['error']['message']
+
+
+def test_non_admin_scoped_token_capped_at_own_permissions(client, db):
+    """A non-admin holding applications.create + apitokens.create may mint a
+    token scoped to applications.create and use it."""
+    _user, headers = _user_with_perms(
+        client, db, 'tokenmaker', ['apitokens.create', 'applications.create'])
+    secret = _create_token(client, headers,
+                           scopes=['applications.create']).get_json()['data']['secret']
+
+    response = client.post('/api/applications', json={'appname': 'Capped App'},
+                           headers=_pat_headers(secret))
+    assert response.status_code == 201
+
+
+def test_scoped_token_intersects_at_use_time(client, db):
+    """Defense in depth: if the owner loses a scoped permission after the token
+    is minted, the token can no longer use it (use-time intersection)."""
+    from shopdb.core.models import Role
+    _user, headers = _user_with_perms(
+        client, db, 'tokenmaker', ['apitokens.create', 'applications.create'])
+    secret = _create_token(client, headers,
+                           scopes=['applications.create']).get_json()['data']['secret']
+
+    # Strip applications.create from the owner's role.
+    role = Role.query.filter_by(rolename='tokenmakerrole').first()
+    role.permissions = [p for p in role.permissions
+                        if p.name != 'applications.create']
+    _db.session.commit()
+
+    response = client.post('/api/applications', json={'appname': 'Gone'},
+                           headers=_pat_headers(secret))
+    assert response.status_code == 403
+    assert Application.query.filter_by(appname='Gone').first() is None
+
+
+def test_import_mode_denied_over_scoped_admin_token(client, db, admin_user,
+                                                   auth_headers):
+    """A scoped admin token does NOT get import mode (an admin-role capability):
+    the createddate in the body is ignored and 'now' is stamped instead."""
+    secret = _create_token(
+        client, auth_headers,
+        scopes=['applications.create']).get_json()['data']['secret']
+
+    headers = _pat_headers(secret)
+    headers['X-Import-Mode'] = 'true'
+    before = _naive_utcnow()
+    response = client.post(
+        '/api/applications',
+        json={'appname': 'No Backdate', 'createddate': '2019-01-02 03:04:05'},
+        headers=headers)
+    assert response.status_code == 201
+
+    app = Application.query.filter_by(appname='No Backdate').first()
+    assert app is not None
+    # Import mode was denied, so createddate is ~now, not the 2019 value.
+    assert app.createddate >= before
+
+
+def test_scopes_update_round_trip(client, db, admin_user, auth_headers):
+    """Scopes can be set, changed, and cleared via PUT, and surface in GET."""
+    create = _create_token(client, auth_headers, scopes=['applications.create'])
+    tokenid = create.get_json()['data']['tokenid']
+
+    got = client.get('/api/apitokens', headers=auth_headers).get_json()['data']
+    row = next(t for t in got if t['tokenid'] == tokenid)
+    assert row['scopes'] == ['applications.create']
+
+    # Change scopes.
+    updated = client.put(f'/api/apitokens/{tokenid}',
+                         json={'scopes': ['applications.edit']},
+                         headers=auth_headers)
+    assert updated.status_code == 200
+    assert updated.get_json()['data']['scopes'] == ['applications.edit']
+
+    # Clear scopes (back to unscoped/full).
+    cleared = client.put(f'/api/apitokens/{tokenid}', json={'scopes': None},
+                         headers=auth_headers)
+    assert cleared.status_code == 200
+    assert cleared.get_json()['data']['scopes'] is None
+
+
+def test_update_scopes_ceiling_enforced(client, db):
+    """PUT scopes is also capped at the owner's permissions."""
+    _user, headers = _user_with_perms(client, db, 'tokenmaker',
+                                      ['apitokens.create'])
+    tokenid = _create_token(client, headers).get_json()['data']['tokenid']
+
+    response = client.put(f'/api/apitokens/{tokenid}',
+                          json={'scopes': ['applications.delete']},
+                          headers=headers)
+    assert response.status_code == 400
+    assert 'applications.delete' in response.get_json()['data']['error']['message']

tests/test_core/test_authz.py

@@ -37,14 +37,10 @@ EXEMPT_BLUEPRINTS = {'auth', 'collector', 'setup'}
 #     role-less member MAY edit their own record, so it does not fit the
 #     403-for-every-member contract this sweep asserts. The other-user 403 is
 #     covered by test_member_cannot_update_other_user below.
-#   apitokens.create_apitoken / update_apitoken / revoke_apitoken - personal
-#     API tokens. By design ANY authenticated user may create and manage their
-#     OWN tokens (own-resource logic, not a flat deny), so a role-less member
-#     gets 201/200 here, not the 403 this sweep asserts. The non-owner 403 is
-#     covered by test_apitokens.py (member cannot revoke another user's token).
-EXEMPT_ENDPOINTS = {'knowledgebase.track_click', 'users.update_user',
-                    'apitokens.create_apitoken', 'apitokens.update_apitoken',
-                    'apitokens.revoke_apitoken'}
+# The apitokens create/update/revoke endpoints are NOT exempt: they now require
+# the apitokens.create permission, so a role-less member gets the 403 this sweep
+# asserts (ownership is still enforced inside the handler for non-admins).
+EXEMPT_ENDPOINTS = {'knowledgebase.track_click', 'users.update_user'}
 
 
 @pytest.fixture(autouse=True)