Add personal API tokens; wire measuring tools into remaining surfaces
API tokens: any user mints named, optionally-expiring tokens (shopdb_pat_..., sha256-stored, secret shown once) at Settings > API Tokens; a before-request shim swaps a valid PAT for a request-scoped JWT of its owner, so the entire existing auth/authz/import-mode stack works unchanged and revoked/expired tokens 401 cleanly. Built for long-running scripts - the legacy import no longer dies when a login JWT expires. Migration 7d21_apitokens; create/revoke audit-logged. Audited integration gaps fixed: Asset.to_dict serializes measuring tools (typedata + pluginid - relationship links to tools resolve); map subtype filter/colors and MapEditor include them; dashboard totals count them; warranty links use a new by-asset route; the measuringtools ADR-010 hooks are real (corrected presentation token, implemented map-overlay endpoint); the login avatar resolves through the employee-photo helper. 737 tests pass; naming green; frontend builds; both features verified live end-to-end. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
120
shopdb/core/api/apitokens.py
Normal file
120
shopdb/core/api/apitokens.py
Normal file
@@ -0,0 +1,120 @@
|
||||
"""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.import_mode import parse_import_datetime
|
||||
|
||||
apitokens_bp = Blueprint('apitokens', __name__)
|
||||
|
||||
|
||||
@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()
|
||||
def create_apitoken():
|
||||
"""Create a token for the caller. Returns the full secret ONCE."""
|
||||
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')
|
||||
|
||||
secret = ApiToken.generate_secret()
|
||||
token = ApiToken(
|
||||
userid=current_user.userid,
|
||||
name=name,
|
||||
tokenprefix=ApiToken.prefix_of(secret),
|
||||
tokenhash=ApiToken.hash_secret(secret),
|
||||
expiresat=expiresat,
|
||||
)
|
||||
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()
|
||||
def update_apitoken(tokenid: int):
|
||||
"""Rename or 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)
|
||||
|
||||
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'])
|
||||
|
||||
db.session.commit()
|
||||
return success_response(token.to_dict(), message='Token updated')
|
||||
|
||||
|
||||
@apitokens_bp.route('/<int:tokenid>', methods=['DELETE'])
|
||||
@jwt_required()
|
||||
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')
|
||||
Reference in New Issue
Block a user