geenforce: resource-scope binding for fetch tokens (0.15.0)

A geenforce.fetch token can now be pinned to specific manifest scopes so a
fleet-wide key (a display's, delivered by DSC or baked into the image) is not a
skeleton key for the whole content store. NULL binding = unrestricted, so every
existing service token keeps working.

Core:
- ApiToken.resourcescopes column + resourcescopelist property (migration
  7d30_apitoken_resourcescopes; NULL = unrestricted).
- apitokens API create/update accept + persist an optional resourcescopes list
  (a resource-name allowlist; not permission-catalog names).
- New contract helper authorized_service_token(scope): same check as
  service_token_authorized but returns the ApiToken so a plugin can read its
  binding. Contract 0.14.0 -> 0.15.0; also export SupportTeam.

GE-Enforce enforcement:
- get_manifest: a bound token requesting a scope outside its allowlist -> 403.
- get_payload: a bound token may only pull a blob its own scope(s) reference
  (service.blob_referenced_by_scopes); anything else -> 404 (no hash probing).
- Decorator stashes the authorized token on g for the route to read.

Also fixes a pre-existing contract-surface violation: the printers/printedparts
alert helpers imported shopdb.core.models / shopdb.extensions directly; now
via shopdb.api (SupportTeam newly exported). Docs: GE-ENFORCE-DISPLAY.md
provisioning note, PLUGIN-HOOKS.md, CLAUDE.md.

9 new resource-binding tests; full suite 1131 passing.
This commit is contained in:
cproudlock
2026-07-23 09:02:42 -04:00
parent d0bf37ced7
commit 75386d2f51
15 changed files with 344 additions and 25 deletions

View File

@@ -36,7 +36,7 @@ from .plugins import plugin_manager
# unattended endpoints (e.g. the GE-Enforce fetch API) can authorize a scoped
# managed service token without importing core token internals. Additive name
# on the import surface, minor bump.
__contract_version__ = '0.14.0'
__contract_version__ = '0.15.0'
# Product release version (see ADR-007). The product version and the
# plugin-contract version above are distinct series with independent

View File

@@ -46,6 +46,7 @@ from shopdb.core.models import (
RelationshipType,
User,
Role,
SupportTeam,
)
# Response + pagination helpers for plugin API blueprints
@@ -63,7 +64,9 @@ from shopdb.utils.authz import require_permission, require_role
# Service-token authorization for unattended plugin endpoints (collector,
# GE-Enforce fetch, ...): checks a scoped managed token without exposing token
# internals.
from shopdb.utils.apitoken_auth import service_token_authorized
from shopdb.utils.apitoken_auth import (
service_token_authorized, authorized_service_token,
)
# Import-mode helpers: preserve legacy timestamps during a bulk data import
from shopdb.utils.import_mode import (
@@ -263,6 +266,8 @@ __all__ = [
'require_permission',
'require_role',
'service_token_authorized',
'authorized_service_token',
'SupportTeam',
# Import-mode helpers
'apply_import_timestamps',
'import_mode_active',

View File

@@ -45,6 +45,24 @@ def _validate_scopes(scopes, owner):
return scopes, None
def _validate_resourcescopes(resourcescopes):
"""Validate a resourcescopes payload. Return (list_or_none, error_response).
resourcescopes is a DIFFERENT axis from scopes: it pins the token to a set
of resource names (today geenforce manifest scope names, e.g.
gea-shopfloor-display), not permission names, so there is no catalog to
check against - a plugin owns the meaning. None/absent = unrestricted.
"""
if resourcescopes is None:
return None, None
if not isinstance(resourcescopes, list) \
or not all(isinstance(s, str) for s in resourcescopes):
return None, error_response(
ErrorCodes.VALIDATION_ERROR,
'resourcescopes must be a list of resource-scope names')
return [s.strip() for s in resourcescopes if s.strip()], None
@apitokens_bp.route('', methods=['GET'])
@jwt_required()
def list_apitokens():
@@ -93,6 +111,11 @@ def create_apitoken():
if scope_error is not None:
return scope_error
resourcescopelist, resource_error = _validate_resourcescopes(
data.get('resourcescopes'))
if resource_error is not None:
return resource_error
secret = ApiToken.generate_secret()
token = ApiToken(
userid=current_user.userid,
@@ -102,6 +125,7 @@ def create_apitoken():
expiresat=expiresat,
)
token.scopelist = scopelist
token.resourcescopelist = resourcescopelist
db.session.add(token)
db.session.flush()
@@ -147,6 +171,12 @@ def update_apitoken(tokenid: int):
if scope_error is not None:
return scope_error
token.scopelist = scopelist
if 'resourcescopes' in data:
resourcescopelist, resource_error = _validate_resourcescopes(
data.get('resourcescopes'))
if resource_error is not None:
return resource_error
token.resourcescopelist = resourcescopelist
db.session.commit()
return success_response(token.to_dict(), message='Token updated')

View File

@@ -53,6 +53,13 @@ class ApiToken(BaseModel):
# authority). A scoped token grants ONLY these, intersected with what the
# owner holds, and suspends the admin bypass. See scopelist below.
scopes = db.Column(db.Text, nullable=True)
# JSON array of RESOURCE-scope names (a different axis from `scopes`, which
# is what the token may DO). Today these are geenforce manifest scope names
# (e.g. gea-shopfloor-display): a resource-bound geenforce.fetch token may
# only pull those scopes' manifests and only blobs those manifests
# reference. NULL = unrestricted (any resource), for back-compat. See
# resourcescopelist below.
resourcescopes = db.Column(db.Text, nullable=True)
user = db.relationship('User', backref=db.backref('apitokens', lazy='dynamic'))
@@ -98,6 +105,25 @@ class ApiToken(BaseModel):
else:
self.scopes = json.dumps(list(names))
@property
def resourcescopelist(self):
"""Parsed resource-scope names, or None when the token is unrestricted."""
if self.resourcescopes is None:
return None
try:
value = json.loads(self.resourcescopes)
except (ValueError, TypeError):
return None
return value if isinstance(value, list) else None
@resourcescopelist.setter
def resourcescopelist(self, names):
"""Store a resource-scope list, or None to clear the restriction."""
if names is None:
self.resourcescopes = None
else:
self.resourcescopes = json.dumps(list(names))
@staticmethod
def unknown_scope_names(names) -> list:
"""Return the subset of names that are not in the permission catalog.
@@ -119,6 +145,7 @@ class ApiToken(BaseModel):
'expiresat': self.expiresat.isoformat() + 'Z' if self.expiresat else None,
'lastusedat': self.lastusedat.isoformat() + 'Z' if self.lastusedat else None,
'scopes': self.scopelist,
'resourcescopes': self.resourcescopelist,
'isactive': self.isactive,
'isexpired': self.is_expired,
'createddate': self.createddate.isoformat() + 'Z' if self.createddate else None,

View File

@@ -76,16 +76,13 @@ def touch_apitoken_lastused(token):
db.session.commit()
def service_token_authorized(scope):
"""True when the current request carries a managed token scoped for `scope`
whose owner is active and holds that permission. Accepts X-API-Key or a
Bearer PAT (the before_request shim resolves Bearer into g.apitokenid).
Touches lastusedat on success.
def authorized_service_token(scope):
"""Return the ApiToken authorizing this request for `scope`, or None.
The single contract-surface entry point for unattended SERVICE tokens
(collector.ingest, geenforce.fetch, ...), so plugins authorize a service
token without reaching into core token internals. Returns False on any
miss; the caller returns its own 401.
Same checks as service_token_authorized (managed token scoped for `scope`,
active owner holding the permission), but hands back the token itself so a
caller can read its resource binding (token.resourcescopelist) without
reaching into core token internals. Touches lastusedat on success.
"""
from shopdb.core.models import User
@@ -99,15 +96,29 @@ def service_token_authorized(scope):
if tokenid is not None:
token = db.session.get(ApiToken, tokenid)
if token is None:
return False
return None
scopelist = token.scopelist
if not scopelist or scope not in scopelist:
return False
return None
user = db.session.get(User, token.userid)
if user is None or not user.isactive or not user.haspermission(scope):
return False
return None
touch_apitoken_lastused(token)
return True
return token
def service_token_authorized(scope):
"""True when the current request carries a managed token scoped for `scope`
whose owner is active and holds that permission. Accepts X-API-Key or a
Bearer PAT (the before_request shim resolves Bearer into g.apitokenid).
Touches lastusedat on success.
The single contract-surface entry point for unattended SERVICE tokens
(collector.ingest, geenforce.fetch, ...), so plugins authorize a service
token without reaching into core token internals. Returns False on any
miss; the caller returns its own 401.
"""
return authorized_service_token(scope) is not None
def install_apitoken_auth(app):