shopdb-flask/shopdb/core/api/apitokens.py

"""Personal API token management endpoints.

Any authenticated user may manage their OWN tokens; an admin may list or revoke
anyone's. Endpoints are jwt_required (a token must be bootstrapped from a real
login or an existing token). The full secret is returned ONCE, on create.
"""

from flask import Blueprint, request
from flask_jwt_extended import jwt_required, current_user

from shopdb.extensions import db
from shopdb.core.models import ApiToken, AuditLog
from shopdb.utils.responses import success_response, error_response, ErrorCodes
from shopdb.utils.authz import require_permission
from shopdb.utils.import_mode import parse_import_datetime

apitokens_bp = Blueprint('apitokens', __name__)


def _validate_scopes(scopes, owner):
    """Validate a scopes payload against the token OWNER. Return
    (scopelist_or_none, error_response); the second is None when valid.

    scopes may be None/absent (unscoped) or a list of permission names. Every
    name must exist in the catalog AND be held by the owner (the scope ceiling):
    a token can never grant more than its owner holds. Admins hold everything.
    """
    if scopes is None:
        return None, None
    if not isinstance(scopes, list) or not all(isinstance(s, str) for s in scopes):
        return None, error_response(
            ErrorCodes.VALIDATION_ERROR,
            'scopes must be a list of permission names')
    unknown = ApiToken.unknown_scope_names(scopes)
    if unknown:
        return None, error_response(
            ErrorCodes.VALIDATION_ERROR,
            'Unknown permission names: ' + ', '.join(unknown))
    # Scope ceiling: the owner must actually hold each scoped permission.
    disallowed = [n for n in scopes if not owner.haspermission(n)]
    if disallowed:
        return None, error_response(
            ErrorCodes.VALIDATION_ERROR,
            'Permissions not held by the token owner: ' + ', '.join(disallowed))
    return scopes, None


@apitokens_bp.route('', methods=['GET'])
@jwt_required()
def list_apitokens():
    """List the caller's own tokens. Admins may pass ?all=true for everyone's.

    Never returns hashes or secrets.
    """
    wants_all = request.args.get('all', 'false').lower() == 'true'
    is_admin = current_user.hasrole('admin')

    query = ApiToken.query
    if wants_all and is_admin:
        include_owner = True
    else:
        query = query.filter(ApiToken.userid == current_user.userid)
        include_owner = False

    query = query.order_by(ApiToken.createddate.desc())
    tokens = [t.to_dict(include_owner=include_owner) for t in query.all()]
    return success_response(tokens)


@apitokens_bp.route('', methods=['POST'])
@jwt_required()
@require_permission('apitokens.create')
def create_apitoken():
    """Create a token for the caller. Returns the full secret ONCE.

    Optional scopes limit the token to a subset of the caller's own
    permissions; absent/null means an unscoped token acting as the caller.
    """
    data = request.get_json() or {}

    name = (data.get('name') or '').strip()
    if not name:
        return error_response(ErrorCodes.VALIDATION_ERROR, 'name is required')

    expiresat = None
    if data.get('expiresat'):
        expiresat = parse_import_datetime(data.get('expiresat'))
        if expiresat is None:
            return error_response(ErrorCodes.VALIDATION_ERROR,
                                  'expiresat is not a valid date/datetime')

    scopelist, scope_error = _validate_scopes(data.get('scopes'), current_user)
    if scope_error is not None:
        return scope_error

    secret = ApiToken.generate_secret()
    token = ApiToken(
        userid=current_user.userid,
        name=name,
        tokenprefix=ApiToken.prefix_of(secret),
        tokenhash=ApiToken.hash_secret(secret),
        expiresat=expiresat,
    )
    token.scopelist = scopelist
    db.session.add(token)
    db.session.flush()

    AuditLog.log('created', 'ApiToken', entityid=token.tokenid, entityname=name)
    db.session.commit()

    result = token.to_dict()
    # The secret appears here and NOWHERE else, ever. Not stored, not logged.
    result['secret'] = secret
    result['warning'] = ('Save this token now. It will not be shown again. '
                         'Store it somewhere safe.')
    return success_response(result, message='Token created', http_code=201)


@apitokens_bp.route('/<int:tokenid>', methods=['PUT'])
@jwt_required()
@require_permission('apitokens.create')
def update_apitoken(tokenid: int):
    """Rename, rescope, or deactivate a token. Own token, or any if admin.

    Scopes are validated against the token OWNER's permissions (the ceiling),
    not the editor's - so an admin rescoping someone else's token still cannot
    grant that owner more than the owner holds.
    """
    token = db.session.get(ApiToken, tokenid)
    if token is None:
        return error_response(ErrorCodes.NOT_FOUND, 'Token not found', http_code=404)

    if token.userid != current_user.userid and not current_user.hasrole('admin'):
        return error_response(ErrorCodes.FORBIDDEN,
                              'You may only manage your own tokens', http_code=403)

    data = request.get_json() or {}
    if 'name' in data:
        newname = (data.get('name') or '').strip()
        if not newname:
            return error_response(ErrorCodes.VALIDATION_ERROR, 'name cannot be empty')
        token.name = newname
    if 'isactive' in data:
        token.isactive = bool(data['isactive'])
    if 'scopes' in data:
        scopelist, scope_error = _validate_scopes(data.get('scopes'), token.user)
        if scope_error is not None:
            return scope_error
        token.scopelist = scopelist

    db.session.commit()
    return success_response(token.to_dict(), message='Token updated')


@apitokens_bp.route('/<int:tokenid>', methods=['DELETE'])
@jwt_required()
@require_permission('apitokens.create')
def revoke_apitoken(tokenid: int):
    """Revoke (deactivate) a token. Own token, or any if admin."""
    token = db.session.get(ApiToken, tokenid)
    if token is None:
        return error_response(ErrorCodes.NOT_FOUND, 'Token not found', http_code=404)

    if token.userid != current_user.userid and not current_user.hasrole('admin'):
        return error_response(ErrorCodes.FORBIDDEN,
                              'You may only manage your own tokens', http_code=403)

    token.isactive = False
    AuditLog.log('deleted', 'ApiToken', entityid=token.tokenid, entityname=token.name)
    db.session.commit()
    return success_response(message='Token revoked')