From 12175169e4820015f858d597773a1bd23c11e63d Mon Sep 17 00:00:00 2001
From: cproudlock <cproudlock@localhost>
Date: Sun, 12 Jul 2026 09:13:33 -0400
Subject: [PATCH] Accept managed collector service tokens on the collector API

A token scoped to the new collector.ingest permission is a collector
service token: the collector endpoints accept it via X-API-Key or
Bearer alongside the env fleet keys (which remain the fallback), giving
the fleet credential rotation, revocation, and last-used visibility
from the API Tokens page. Containment holds both ways: a collector
token authorizes nothing else, and no other credential gains collector
access. Shared token validation refactored out of the auth shim; a
Collector service token quick-preset in the create modal; integration
guide documents minting, rotation via site-config.json, and the
service-identity pattern.

765 tests pass; live acceptance matrix verified.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 CHANGELOG.md                                  |  23 ++++
 docs/COLLECTOR-INTEGRATION.md                 |  81 ++++++++++++
 docs/CONFIG.md                                |  10 +-
 frontend/src/views/settings/ApiTokensList.vue |  34 +++++
 shopdb/core/api/collector.py                  | 104 +++++++++++----
 shopdb/core/models/user.py                    |   6 +
 shopdb/utils/apitoken_auth.py                 |  15 ++-
 tests/test_core/test_apitokens.py             |  16 +++
 tests/test_core/test_collector_contract.py    | 121 ++++++++++++++++++
 9 files changed, 381 insertions(+), 29 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 3db0a64..4a0fa16 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -48,6 +48,29 @@ ADR-007 and ADR-002.
   create/edit modals gain a "Restrict permissions" section (a category-grouped
   checkbox grid limited to the permissions the creator holds) and the token
   lists show a full-access / N-permissions access chip.
+- Managed collector service tokens: the collector ingest API
+  (`/api/collector/<plugin>` + the legacy `/pc` `/apps` `/heartbeat` `/bulk`
+  `/status` endpoints) now ALSO accepts a managed API token scoped to the new
+  `collector.ingest` permission (category `collector`), on top of the existing
+  `COLLECTOR_API_KEY[_<PLUGIN>]` env keys (which stay supported as a
+  bootstrap/legacy fallback - nothing breaks). The token may be presented in
+  `X-API-Key` (as GE-Enforce sends today) OR as an `Authorization: Bearer`
+  token; both transports validate the PAT the same way the login shim does
+  (hash lookup, active, unexpired, active owner) via a shared
+  `resolve_api_token` helper refactored out of `apitoken_auth.py`, require
+  `collector.ingest` in the token's scope list AND that the owner holds it, and
+  stamp `lastusedat` (same 60s throttle). A token scoped to ONLY
+  `collector.ingest` is a collector service token: it authorizes the collector
+  API and NOTHING else - the existing scoped-token machinery denies it on every
+  permission- and role-gated route and on import mode, so a leaked collector
+  token cannot touch the regular API. Recommended flow (documented): an admin
+  mints the scoped token (the scope suspends the admin bypass, containing it);
+  rotate by minting a new one, deploying via `site-config.json`, watching
+  `lastusedat`, then revoking the old. The Settings > API Tokens create modal
+  gains a "Collector service token" quick-preset (pre-selects only
+  `collector.ingest`). Docs: `docs/COLLECTOR-INTEGRATION.md` (new "Managed
+  collector tokens" section) and `docs/CONFIG.md`. Core feature; no
+  plugin-contract change.
 - Vendor-model photos on asset detail heroes: computers and printers now
   surface the linked model's `imageurl` in their extension payloads (the
   field machines already exposed), and the machine, PC, printer, network
diff --git a/docs/COLLECTOR-INTEGRATION.md b/docs/COLLECTOR-INTEGRATION.md
index b6fd5cf..118de1d 100644
--- a/docs/COLLECTOR-INTEGRATION.md
+++ b/docs/COLLECTOR-INTEGRATION.md
@@ -30,6 +30,80 @@ a real caller (the GE-Enforce fleet agent) to it.
   HTTP 500 `Collector API key not configured` and rejects every request. An
   unconfigured server never silently accepts unauthenticated data.
 - A caller that sends the wrong key (or no key) gets HTTP 401 `Invalid API key`.
+- In addition to the env keys, a managed API token scoped to `collector.ingest`
+  is accepted as a collector credential on every collector endpoint. See
+  "Managed collector tokens" below; env keys remain the fallback.
+
+### Managed collector tokens (recommended)
+
+Alongside the env keys, every collector endpoint (`/api/collector/<plugin>`,
+`/pc`, `/apps`, `/heartbeat`, `/bulk`, `/status`) also accepts a **managed API
+token** (PAT) scoped to the `collector.ingest` permission. The env keys stay
+supported as a bootstrap/legacy fallback - nothing breaks - but a managed token
+is the preferred credential because it can be minted, rotated, and revoked from
+the UI (Settings > API Tokens) and its use shows up in `lastusedat` and the
+audit log.
+
+What makes a token a collector service token: it is scoped to ONLY
+`collector.ingest`. That scope authorizes the collector ingest API and NOTHING
+else. The existing scoped-token machinery contains it automatically - a scoped
+token passes `require_permission` only for its listed permissions and is denied
+on every role-gated (`require_role`) endpoint and on import mode, and
+`collector.ingest` gates no normal route. So a collector token that leaks cannot
+be used to read or write anything through the regular API; it can only submit
+collector payloads.
+
+Both wire transports are accepted (send whichever is convenient; GE-Enforce
+sends `X-API-Key` today, so that stays ergonomic):
+
+```
+POST /api/collector/computers
+X-API-Key: shopdb_pat_<40 hex>
+```
+or
+```
+POST /api/collector/computers
+Authorization: Bearer shopdb_pat_<40 hex>
+```
+
+An unscoped PAT, or a PAT scoped to some other permission, is NOT a collector
+token and is rejected (401) - only `collector.ingest` in the scope list counts.
+A revoked or expired token is rejected (401) on both transports.
+
+#### How to mint one (admin flow)
+
+The simplest contained flow: an **admin** mints the token, scoped to
+`collector.ingest`. Because the token is scoped, the admin-role bypass is
+suspended for it, so the token is contained to the collector API even though its
+owner is an admin - it cannot act with admin authority anywhere.
+
+1. Settings > API Tokens > New Token.
+2. Click the **Collector service token** preset (pre-selects only
+   `collector.ingest`), name it (e.g. `wj-fleet-collector`), optionally set an
+   expiry, Create.
+3. Copy the `shopdb_pat_...` secret (shown once) and deploy it to the fleet the
+   same way as the env key: the `collectorApiKey` field in per-site
+   `site-config.json` (see "Delivering the API key to clients" below). The
+   client sends it in `X-API-Key` exactly as it sends an env key today - no
+   client code change.
+
+Service identity (documented, not built): if you prefer a non-admin owner,
+create a dedicated low-privilege user (e.g. `svc-collector`) whose role holds
+only `collector.ingest`, plus `apitokens.create` if that user is to mint its own
+token. The scope ceiling then caps any token it mints at `collector.ingest`.
+The admin-minted route above is simpler and equally contained, so it is the
+recommended default.
+
+#### Rotation
+
+Managed tokens rotate without a fleet re-image:
+
+1. Mint a new collector token (steps above).
+2. Deploy it via `site-config.json` (`collectorApiKey`) - update the one per-site
+   value.
+3. Confirm the new token is in use: watch its `lastusedat` climb in Settings >
+   API Tokens (and the old token's `lastusedat` go stale).
+4. Revoke the old token once traffic has moved. Revocation is immediate.
 
 ### Generic endpoint contract: `POST /api/collector/<plugin>`
 
@@ -256,6 +330,13 @@ collector schema, and POSTs with the `X-API-Key` header over TLS 1.2. Every
 field name below was checked against `get_collector_schema` in
 `plugins/computers/plugin.py`.
 
+The `X-API-Key` value can be EITHER a `COLLECTOR_API_KEY[_COMPUTERS]` env key OR
+a managed token scoped to `collector.ingest` (a `shopdb_pat_...` secret; see
+"Managed collector tokens"). The script is identical for both - it just carries
+whatever `collectorApiKey` the site-config supplies - so switching a site from an
+env key to a managed token (and rotating it) is a config change, not a script
+change.
+
 ```powershell
 # Send-ShopdbCollectorReport.ps1
 # Reports this PC's identity to shopdb-flask via POST /api/collector/computers.
diff --git a/docs/CONFIG.md b/docs/CONFIG.md
index a1e0761..4de9723 100644
--- a/docs/CONFIG.md
+++ b/docs/CONFIG.md
@@ -59,9 +59,17 @@ limit is approximate across multiple gunicorn workers).
 
 | Variable | Required | Default | Notes |
 |----------|----------|---------|-------|
-| `COLLECTOR_API_KEY` | No | (empty) | Shared key for `/api/collector/*`. Endpoint fails closed (denies) when unset. Sent as the `X-API-Key` header. |
+| `COLLECTOR_API_KEY` | No | (empty) | Shared key for `/api/collector/*`. Endpoint fails closed (denies) when unset and no managed token is presented. Sent as the `X-API-Key` header. |
 | `COLLECTOR_API_KEY_<PLUGIN>` | No | (empty) | Per-plugin override, e.g. `COLLECTOR_API_KEY_COMPUTERS`. Checked before the shared key. |
 
+The collector endpoints ALSO accept a managed API token (PAT) scoped to the
+`collector.ingest` permission, sent in `X-API-Key` or as an
+`Authorization: Bearer` token. Env keys stay supported as a bootstrap/legacy
+fallback; a managed token is preferred because it is minted, rotated, and
+revoked from Settings > API Tokens with `lastusedat` visibility. A
+collector-scoped token is contained to the collector API and nothing else. See
+`docs/COLLECTOR-INTEGRATION.md` (Managed collector tokens).
+
 ### Zabbix (printer supply monitoring)
 
 | Variable | Required | Default | Notes |
diff --git a/frontend/src/views/settings/ApiTokensList.vue b/frontend/src/views/settings/ApiTokensList.vue
index 1227987..1557794 100644
--- a/frontend/src/views/settings/ApiTokensList.vue
+++ b/frontend/src/views/settings/ApiTokensList.vue
@@ -125,6 +125,18 @@
               <small class="form-hint">Leave blank for a token that never expires.</small>
             </div>
 
+            <div v-if="canCollectorPreset" class="form-group preset-group">
+              <button type="button" class="btn btn-secondary btn-sm"
+                      @click="applyCollectorPreset">
+                Collector service token
+              </button>
+              <small class="form-hint">
+                For GE-Enforce / fleet reporting. Pre-selects only
+                <code>collector.ingest</code>; the token works only on the
+                collector API and nothing else.
+              </small>
+            </div>
+
             <div class="form-group">
               <label class="checkbox-label">
                 <input type="checkbox" v-model="form.restrict" />
@@ -308,6 +320,17 @@ const availableGrouped = computed(() => {
   return result
 })
 
+// The collector preset is offered only when the caller can actually mint a
+// collector.ingest token (admins hold everything; others must hold the perm).
+const COLLECTOR_SCOPE = 'collector.ingest'
+const canCollectorPreset = computed(
+  () => isAdmin.value || myPermissions.value.includes(COLLECTOR_SCOPE))
+
+function applyCollectorPreset() {
+  form.value.restrict = true
+  form.value.scopes = [COLLECTOR_SCOPE]
+}
+
 onMounted(() => loadData())
 
 async function loadData() {
@@ -499,6 +522,17 @@ async function revokeToken() {
   color: var(--text-light);
   margin-top: 0.25rem;
 }
+.preset-group {
+  padding: 0.5rem;
+  border: 1px solid var(--border);
+  border-radius: 4px;
+  background: var(--bg);
+}
+.preset-group code {
+  background: var(--bg-card);
+  padding: 0.1rem 0.3rem;
+  border-radius: 3px;
+}
 .secret-warning {
   color: var(--danger);
   margin-bottom: 1rem;
diff --git a/shopdb/core/api/collector.py b/shopdb/core/api/collector.py
index c582b31..4c7ac57 100644
--- a/shopdb/core/api/collector.py
+++ b/shopdb/core/api/collector.py
@@ -8,15 +8,21 @@ API key (not JWT) for unattended scripts. Writes the asset/computer model
 
 from datetime import datetime, timezone
 from functools import wraps
-from flask import Blueprint, request, current_app
+from flask import Blueprint, request, current_app, g
 
 from shopdb.extensions import db
 from shopdb.core.models import Asset, Application
+from shopdb.core.models.apitoken import ApiToken, TOKEN_SECRET_PREFIX
 
 from shopdb.utils.responses import success_response, error_response, ErrorCodes
 
 collector_bp = Blueprint('collector', __name__)
 
+# A managed token scoped to this permission is a collector service token: it
+# authorizes the collector ingest API and nothing else (ADR-006, machine
+# identity - no require_permission on these routes).
+COLLECTOR_SCOPE = 'collector.ingest'
+
 
 def _computer_models():
     """Lazily import the computers plugin models.
@@ -33,24 +39,80 @@ def _computer_models():
         return None
 
 
+def _token_is_collector(token):
+    """True when this managed token is a collector service token: scoped for
+    collector.ingest AND its owner actually holds that permission (admins do)."""
+    from shopdb.core.models import User
+
+    scopelist = token.scopelist
+    if not scopelist or COLLECTOR_SCOPE not in scopelist:
+        return False
+    user = db.session.get(User, token.userid)
+    return (user is not None and user.isactive
+            and user.haspermission(COLLECTOR_SCOPE))
+
+
+def _collector_managed_token(api_key):
+    """Find the managed token backing this collector request, or None.
+
+    Two transports are accepted (GE-Enforce sends X-API-Key today; Bearer is
+    the standard PAT wire form):
+      X-API-Key: the before_request PAT shim ignores this header, so resolve the
+                 secret here.
+      Bearer:    the shim already resolved+swapped it (the raw Bearer secret is
+                 gone by now), leaving g.apitokenid. Reuse that.
+    """
+    from shopdb.utils.apitoken_auth import resolve_api_token
+
+    if api_key and api_key.startswith(TOKEN_SECRET_PREFIX):
+        resolved = resolve_api_token(api_key)
+        return resolved[0] if resolved else None
+    tokenid = getattr(g, 'apitokenid', None)
+    if tokenid is not None:
+        return db.session.get(ApiToken, tokenid)
+    return None
+
+
+def _check_collector_auth(expected_key):
+    """Authorize a collector request. Returns None when authorized, else the
+    error_response to return.
+
+    Accepts a collector-scoped managed token (Bearer or X-API-Key) OR the env
+    key via X-API-Key (legacy/bootstrap fallback). Env keys stay supported so
+    nothing breaks; a valid managed token works even with no env key set.
+    """
+    from shopdb.utils.apitoken_auth import touch_apitoken_lastused
+
+    api_key = request.headers.get('X-API-Key')
+
+    token = _collector_managed_token(api_key)
+    if token is not None and _token_is_collector(token):
+        touch_apitoken_lastused(token)
+        return None
+
+    if expected_key and api_key == expected_key:
+        return None
+
+    # Fail-closed only on pure server misconfiguration: no env key AND the
+    # caller presented no managed token at all. Otherwise it is a bad credential.
+    presented_pat = token is not None or (
+        api_key is not None and api_key.startswith(TOKEN_SECRET_PREFIX))
+    if not expected_key and not presented_pat:
+        return error_response(ErrorCodes.INTERNAL_ERROR,
+                              'Collector API key not configured', http_code=500)
+    return error_response(ErrorCodes.UNAUTHORIZED, 'Invalid API key',
+                          http_code=401)
+
+
 def require_api_key(f):
-    """Require API key authentication."""
+    """Require collector API-key OR collector-scoped managed-token auth."""
     @wraps(f)
     def decorated(*args, **kwargs):
         # Header only. Querystring api_key was dropped so keys do not land in
         # access logs / proxy history (breaking change, see COLLECTOR-INTEGRATION.md).
-        api_key = request.headers.get('X-API-Key')
-        expected_key = current_app.config.get('COLLECTOR_API_KEY')
-
-        if not expected_key:
-            return error_response(
-                ErrorCodes.INTERNAL_ERROR,
-                'Collector API key not configured',
-                http_code=500
-            )
-        if api_key != expected_key:
-            return error_response(ErrorCodes.UNAUTHORIZED, 'Invalid API key',
-                                  http_code=401)
+        error = _check_collector_auth(current_app.config.get('COLLECTOR_API_KEY'))
+        if error is not None:
+            return error
         return f(*args, **kwargs)
     return decorated
 
@@ -136,15 +198,11 @@ def generic_collect(pluginname):
 
     plugin, schema = plugins[pluginname]
 
-    expected_key = _plugin_api_key(pluginname)
-    if not expected_key:
-        return error_response(ErrorCodes.INTERNAL_ERROR,
-                              'Collector API key not configured', http_code=500)
-    # Header only (querystring fallback dropped, see require_api_key).
-    api_key = request.headers.get('X-API-Key')
-    if api_key != expected_key:
-        return error_response(ErrorCodes.UNAUTHORIZED, 'Invalid API key',
-                              http_code=401)
+    # Per-plugin env key (fallback) OR a collector-scoped managed token. Header
+    # only (querystring fallback dropped, see require_api_key).
+    error = _check_collector_auth(_plugin_api_key(pluginname))
+    if error is not None:
+        return error
 
     payload = request.get_json(silent=True)
     if not payload or not isinstance(payload, dict):
diff --git a/shopdb/core/models/user.py b/shopdb/core/models/user.py
index 2d564b5..2a40ae5 100644
--- a/shopdb/core/models/user.py
+++ b/shopdb/core/models/user.py
@@ -105,6 +105,12 @@ class Permission(db.Model):
         ('audit.view', 'View audit logs', 'admin'),
         # API tokens
         ('apitokens.create', 'Create and manage API tokens', 'apitokens'),
+        # Collector service tokens. A token scoped to ONLY this permission is a
+        # collector service token: it authorizes the collector ingest API and
+        # nothing else (scoped tokens pass require_permission only for listed
+        # perms, and this perm gates no other route). See collector.py.
+        ('collector.ingest', 'Submit collector payloads (fleet reporting)',
+         'collector'),
     ]
 
     def __repr__(self):
diff --git a/shopdb/utils/apitoken_auth.py b/shopdb/utils/apitoken_auth.py
index d9e22ab..f5b918f 100644
--- a/shopdb/utils/apitoken_auth.py
+++ b/shopdb/utils/apitoken_auth.py
@@ -47,8 +47,13 @@ def _extract_pat_secret():
     return None
 
 
-def _resolve_pat(secret):
-    """Validate a PAT secret. Return (token, user) or None."""
+def resolve_api_token(secret):
+    """Validate a PAT secret. Return (token, user) or None.
+
+    Shared validator: hash lookup, active token, unexpired, active owner. The
+    before_request shim and the collector API (which does not decode JWT) both
+    call this so the checks live in one place.
+    """
     from shopdb.core.models import User
 
     token = ApiToken.query.filter_by(
@@ -61,7 +66,7 @@ def _resolve_pat(secret):
     return token, user
 
 
-def _touch_lastused(token):
+def touch_apitoken_lastused(token):
     """Throttled lastusedat write. Independent commit; nothing else is pending
     this early in the request, so it cannot clobber route work."""
     now = _utcnow()
@@ -79,7 +84,7 @@ def install_apitoken_auth(app):
         secret = _extract_pat_secret()
         if secret is None:
             return
-        resolved = _resolve_pat(secret)
+        resolved = resolve_api_token(secret)
         if resolved is None:
             # The caller clearly meant to use a PAT (shopdb_pat_ prefix) but it
             # is unknown, revoked, or expired. Reject with a clear 401 instead
@@ -105,7 +110,7 @@ def install_apitoken_auth(app):
         g.apitokenid = token.tokenid
         g.apitokenuser = user
 
-        _touch_lastused(token)
+        touch_apitoken_lastused(token)
 
         # Mint a request-scoped JWT for the owner and swap it into the header
         # so the whole downstream auth stack authenticates as that user.
diff --git a/tests/test_core/test_apitokens.py b/tests/test_core/test_apitokens.py
index 6f19c41..1b0bdae 100644
--- a/tests/test_core/test_apitokens.py
+++ b/tests/test_core/test_apitokens.py
@@ -390,6 +390,22 @@ def test_scopes_update_round_trip(client, db, admin_user, auth_headers):
     assert cleared.get_json()['data']['scopes'] is None
 
 
+def test_collector_scoped_token_denied_on_normal_endpoints(client, db,
+                                                           admin_user,
+                                                           auth_headers):
+    """A collector-scoped token (collector.ingest only) is contained: the shim
+    still swaps it into a JWT, but collector.ingest gates no normal route, so it
+    is denied on a permission-gated create just like any other unlisted scope."""
+    secret = _create_token(
+        client, auth_headers,
+        scopes=['collector.ingest']).get_json()['data']['secret']
+
+    response = client.post('/api/applications', json={'appname': 'Nope'},
+                           headers=_pat_headers(secret))
+    assert response.status_code == 403
+    assert Application.query.filter_by(appname='Nope').first() 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',
diff --git a/tests/test_core/test_collector_contract.py b/tests/test_core/test_collector_contract.py
index 2a46e09..7fa2a62 100644
--- a/tests/test_core/test_collector_contract.py
+++ b/tests/test_core/test_collector_contract.py
@@ -208,3 +208,124 @@ def test_legacy_header_api_key_accepted(client, db, collector_key):
     """The X-API-Key header still authenticates the legacy endpoints."""
     resp = client.get('/api/collector/status', headers={'X-API-Key': KEY})
     assert resp.status_code == 200
+
+
+# --- Managed collector service tokens ------------------------------------
+
+def _mint_collector_token(client, auth_headers, scopes):
+    """Mint a managed token with the given scopes as the logged-in admin."""
+    resp = client.post('/api/apitokens',
+                       json={'name': 'collector svc', 'scopes': scopes},
+                       headers=auth_headers)
+    assert resp.status_code == 201, resp.get_json()
+    return resp.get_json()['data']['secret']
+
+
+def test_collector_token_via_x_api_key(client, db, auth_headers,
+                                       computer_assettype):
+    """A collector-scoped managed token authorizes ingest via X-API-Key, with
+    no env key set."""
+    secret = _mint_collector_token(client, auth_headers, ['collector.ingest'])
+    resp = client.post('/api/collector/computers',
+                       json={'hostname': 'WJTOK01'},
+                       headers={'X-API-Key': secret})
+    assert resp.status_code == 200, resp.get_json()
+    assert resp.get_json()['data']['action'] == 'created'
+
+
+def test_collector_token_via_bearer(client, db, auth_headers,
+                                    computer_assettype):
+    """The same token also works on the standard Bearer PAT transport."""
+    secret = _mint_collector_token(client, auth_headers, ['collector.ingest'])
+    resp = client.post('/api/collector/computers',
+                       json={'hostname': 'WJTOK02'},
+                       headers={'Authorization': f'Bearer {secret}'})
+    assert resp.status_code == 200, resp.get_json()
+    assert resp.get_json()['data']['action'] == 'created'
+
+
+def test_collector_token_stamps_lastusedat(client, db, auth_headers,
+                                           computer_assettype):
+    """Using a collector token stamps lastusedat."""
+    from shopdb.core.models import ApiToken
+
+    secret = _mint_collector_token(client, auth_headers, ['collector.ingest'])
+    token = ApiToken.query.filter_by(
+        tokenhash=ApiToken.hash_secret(secret)).first()
+    assert token.lastusedat is None
+
+    client.post('/api/collector/computers', json={'hostname': 'WJTOK03'},
+                headers={'X-API-Key': secret})
+    db.session.expire_all()
+    token = ApiToken.query.filter_by(
+        tokenhash=ApiToken.hash_secret(secret)).first()
+    assert token.lastusedat is not None
+
+
+def test_unscoped_token_rejected_by_collector(client, db, auth_headers,
+                                              collector_key, computer_assettype):
+    """An unscoped PAT is NOT a collector token: it lacks the collector.ingest
+    scope, so the collector API rejects it (env-key semantics unchanged)."""
+    secret = client.post('/api/apitokens', json={'name': 'full'},
+                         headers=auth_headers).get_json()['data']['secret']
+    resp = client.post('/api/collector/computers', json={'hostname': 'WJTOK04'},
+                       headers={'X-API-Key': secret})
+    assert resp.status_code == 401
+
+
+def test_wrong_scoped_token_rejected_by_collector(client, db, auth_headers,
+                                                  collector_key,
+                                                  computer_assettype):
+    """A PAT scoped to some OTHER permission is not a collector token."""
+    secret = _mint_collector_token(client, auth_headers, ['applications.create'])
+    resp = client.post('/api/collector/computers', json={'hostname': 'WJTOK05'},
+                       headers={'X-API-Key': secret})
+    assert resp.status_code == 401
+
+
+def test_revoked_collector_token_rejected(client, db, auth_headers,
+                                          collector_key, computer_assettype):
+    """A revoked collector token is rejected on both transports."""
+    from shopdb.core.models import ApiToken
+
+    secret = _mint_collector_token(client, auth_headers, ['collector.ingest'])
+    tokenid = ApiToken.query.filter_by(
+        tokenhash=ApiToken.hash_secret(secret)).first().tokenid
+    assert client.delete(f'/api/apitokens/{tokenid}',
+                        headers=auth_headers).status_code == 200
+
+    viakey = client.post('/api/collector/computers', json={'hostname': 'WJTOK06'},
+                        headers={'X-API-Key': secret})
+    assert viakey.status_code == 401
+    viabearer = client.post('/api/collector/computers',
+                           json={'hostname': 'WJTOK06'},
+                           headers={'Authorization': f'Bearer {secret}'})
+    assert viabearer.status_code == 401
+
+
+def test_collector_token_contained_to_collector_api(client, db, admin_user,
+                                                    auth_headers):
+    """Containment: a collector-scoped token is denied on a permission-gated
+    write (applications.create) AND a role-gated write (vendors, require_role)."""
+    from shopdb.core.models import Application, Vendor
+
+    secret = _mint_collector_token(client, auth_headers, ['collector.ingest'])
+
+    perm_gated = client.post('/api/applications', json={'appname': 'Nope'},
+                            headers={'Authorization': f'Bearer {secret}'})
+    assert perm_gated.status_code == 403
+    assert Application.query.filter_by(appname='Nope').first() is None
+
+    role_gated = client.post('/api/vendors', json={'vendor': 'Nope'},
+                            headers={'Authorization': f'Bearer {secret}'})
+    assert role_gated.status_code == 403
+    assert Vendor.query.filter_by(vendor='Nope').first() is None
+
+
+def test_env_key_still_works_alongside_tokens(client, db, collector_key,
+                                              computer_assettype):
+    """Regression: the env key path is unchanged when tokens are in play."""
+    resp = client.post('/api/collector/computers', json={'hostname': 'WJENV01'},
+                       headers={'X-API-Key': KEY})
+    assert resp.status_code == 200
+    assert resp.get_json()['data']['action'] == 'created'