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 mint_service_token(scope, name, owner_userid, revoke_existing=True):
    """Mint a managed service token scoped to exactly ONE permission.

    The contract-surface way for a plugin to provision an unattended service
    credential (an endpoint agent key, a collector key) without importing core
    token internals. The returned secret is the ONLY time the full secret is
    available - callers reveal it once and store only the prefix afterwards.

    scope           the single permission the token authorizes (e.g.
                    collector.ingest.endpoints). The owner must hold it.
    name            human label shown in the token list UI.
    owner_userid    the acting admin who owns (and can be audited for) the key.
    revoke_existing when True (default), every currently-active token scoped to
                    EXACTLY this one permission is deactivated first, so
                    regenerating is a hard cutover: the old agent key stops
                    working the instant the new one is minted.

    Returns (token, secret). Commits.
    """
    from shopdb.core.models import User

    owner = db.session.get(User, owner_userid)
    if owner is None or not owner.isactive or not owner.haspermission(scope):
        raise PermissionError(
            'Token owner must be an active user holding ' + scope)

    if revoke_existing:
        for existing in list_service_tokens(scope):
            existing.isactive = False

    secret = ApiToken.generate_secret()
    token = ApiToken(
        userid=owner_userid,
        name=name,
        tokenprefix=ApiToken.prefix_of(secret),
        tokenhash=ApiToken.hash_secret(secret),
    )
    token.scopelist = [scope]
    db.session.add(token)
    db.session.commit()
    return token, secret


def list_service_tokens(scope, active_only=True):
    """Return managed tokens scoped to EXACTLY the single permission `scope`.

    Used by a plugin settings page to render the current agent-key state
    (prefix, created, last-used) without exposing token internals. Filters in
    Python because scopes are stored as a JSON blob, not a queryable column.
    """
    query = ApiToken.query
    if active_only:
        query = query.filter_by(isactive=True)
    matches = []
    for token in query.order_by(ApiToken.createddate.desc()).all():
        scopelist = token.scopelist
        if scopelist and len(scopelist) == 1 and scopelist[0] == scope:
            matches.append(token)
    return matches


def revoke_service_tokens(scope):
    """Deactivate every active managed token scoped to EXACTLY `scope`.

    The kill switch: revoking the endpoint agent key stops all endpoint-agent
    traffic authorized by a managed token. Returns the count revoked. Commits.
    """
    revoked = 0
    for token in list_service_tokens(scope):
        token.isactive = False
        revoked += 1
    if revoked:
        db.session.commit()
    return revoked


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 and any permission a plugin declares for a machine caller
    of its own), 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}'