"""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 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(): """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 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, name=name, tokenprefix=ApiToken.prefix_of(secret), tokenhash=ApiToken.hash_secret(secret), expiresat=expiresat, ) token.scopelist = scopelist token.resourcescopelist = resourcescopelist 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('/', 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 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') @apitokens_bp.route('/', 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')