12175169e4
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>
120 lines
4.7 KiB
Python
120 lines
4.7 KiB
Python
"""Personal API token (PAT) authentication shim.
|
|
|
|
A request may send `Authorization: Bearer shopdb_pat_...`. This is recognized
|
|
BEFORE any JWT decode: a before_request hook validates the PAT (hash lookup,
|
|
active, not expired, active owner) and, on success, mints a short internal
|
|
request-scoped JWT for the token's user and swaps it into the request's
|
|
Authorization header.
|
|
|
|
Why mint a JWT instead of only stashing the user on g: every write route in
|
|
this app stacks a mandatory @jwt_required() ABOVE @require_permission /
|
|
@require_role. That mandatory decorator decodes the Authorization header
|
|
itself, so the ONLY way a PAT reaches the whole existing auth+authz stack
|
|
(jwt_required, require_permission, require_role, import_mode, current_user,
|
|
get_jwt_identity) unchanged is to present a genuine JWT downstream. The minted
|
|
token lives only in this request's environ and is never returned to the client.
|
|
|
|
Result: a PAT authenticates any route a login JWT would, acting as its owner,
|
|
with zero changes to the authz decorators or import-mode helpers.
|
|
"""
|
|
|
|
from datetime import datetime, timezone
|
|
|
|
from flask import g, request
|
|
from flask_jwt_extended import create_access_token
|
|
|
|
from shopdb.extensions import db
|
|
from shopdb.core.models.apitoken import ApiToken, TOKEN_SECRET_PREFIX
|
|
from shopdb.utils.responses import error_response, ErrorCodes
|
|
|
|
|
|
# Only rewrite lastusedat when it is older than this, to avoid a DB write on
|
|
# every single request a busy integration makes.
|
|
_LASTUSED_THROTTLE_SECONDS = 60
|
|
|
|
|
|
def _utcnow():
|
|
return datetime.now(timezone.utc).replace(tzinfo=None)
|
|
|
|
|
|
def _extract_pat_secret():
|
|
"""Return the PAT secret from the Authorization header, or None."""
|
|
header = request.headers.get('Authorization', '')
|
|
parts = header.split()
|
|
if len(parts) == 2 and parts[0] == 'Bearer' \
|
|
and parts[1].startswith(TOKEN_SECRET_PREFIX):
|
|
return parts[1]
|
|
return 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(
|
|
tokenhash=ApiToken.hash_secret(secret), isactive=True).first()
|
|
if token is None or token.is_expired:
|
|
return None
|
|
user = db.session.get(User, token.userid)
|
|
if user is None or not user.isactive:
|
|
return None
|
|
return token, user
|
|
|
|
|
|
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()
|
|
if token.lastusedat is None \
|
|
or (now - token.lastusedat).total_seconds() > _LASTUSED_THROTTLE_SECONDS:
|
|
token.lastusedat = now
|
|
db.session.commit()
|
|
|
|
|
|
def install_apitoken_auth(app):
|
|
"""Register the before_request PAT shim on the app."""
|
|
|
|
@app.before_request
|
|
def _apitoken_before_request():
|
|
secret = _extract_pat_secret()
|
|
if secret is None:
|
|
return
|
|
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
|
|
# of letting the JWT decoder emit a confusing 422 on the non-JWT.
|
|
return error_response(
|
|
ErrorCodes.UNAUTHORIZED,
|
|
'Invalid, revoked, or expired API token',
|
|
http_code=401)
|
|
token, user = resolved
|
|
|
|
# Read claim inputs before the (possible) commit expires the instance.
|
|
claims = {
|
|
'username': user.username,
|
|
'roles': [role.rolename for role in user.roles],
|
|
}
|
|
# A scoped token carries a patscopes claim; authz reads it to grant ONLY
|
|
# the listed permissions and to deny role gates + import mode. Unscoped
|
|
# tokens carry no such claim and mint exactly as a login JWT would.
|
|
scopelist = token.scopelist
|
|
if scopelist is not None:
|
|
claims['patscopes'] = scopelist
|
|
# Expose the token/user for audit and introspection if a handler wants it.
|
|
g.apitokenid = token.tokenid
|
|
g.apitokenuser = user
|
|
|
|
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.
|
|
access_token = create_access_token(
|
|
identity=str(user.userid), additional_claims=claims)
|
|
request.environ['HTTP_AUTHORIZATION'] = f'Bearer {access_token}'
|