shopdb-flask/shopdb/utils/apitoken_auth.py

"""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 authorized_service_token(scope):
    """Return the ApiToken authorizing this request for `scope`, or None.

    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

    api_key = request.headers.get('X-API-Key')
    token = None
    if api_key and api_key.startswith(TOKEN_SECRET_PREFIX):
        resolved = resolve_api_token(api_key)
        token = resolved[0] if resolved else None
    else:
        tokenid = getattr(g, 'apitokenid', None)
        if tokenid is not None:
            token = db.session.get(ApiToken, tokenid)
    if token is None:
        return None
    scopelist = token.scopelist
    if not scopelist or scope not in scopelist:
        return None
    user = db.session.get(User, token.userid)
    if user is None or not user.isactive or not user.haspermission(scope):
        return None
    touch_apitoken_lastused(token)
    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):
    """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}'