pikram/user-microservice
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Added blocking not authorized access

Browse Source
This commit is contained in:
Marcin-Ramotowski 2025-03-15 15:29:10 +00:00
parent c0401a2eb0
commit d7c089aa83
1 changed files with 6 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
api/user_views.py
View File

@ -1,25 +1,24 @@
from flask import Blueprint, jsonify, request, abort from flask import Blueprint, jsonify, request, abort, g
from flask_jwt_extended import create_access_token, set_access_cookies, jwt_required, get_jwt_identity, unset_jwt_cookies from flask_jwt_extended import create_access_token, set_access_cookies, jwt_required, get_jwt_identity, unset_jwt_cookies
from models import User, db from models import User, db
from werkzeug.security import check_password_hash, generate_password_hash from werkzeug.security import check_password_hash, generate_password_hash
user_bp = Blueprint('user_bp', __name__) user_bp = Blueprint('user_bp', __name__)
def admin_required(): def admin_required(username):
current_user = User.query.filter_by(username=get_jwt_identity()).first() user = User.query.filter_by(username=username).first()
if not current_user or current_user.role != 'admin': if user is None or user.role != "Administrator":
abort(403, {'error': 'Access denied.'}) abort(403, {'error': f'Access denied.'})
@user_bp.route('/users', methods=['GET']) @user_bp.route('/users', methods=['GET'])
@jwt_required() @jwt_required()
@admin_required()
def get_all_users(): def get_all_users():
admin_required(get_jwt_identity())
users = User.query.all() users = User.query.all()
return jsonify([user.to_dict() for user in users]) return jsonify([user.to_dict() for user in users])
@user_bp.route('/users/<int:user_id>', methods=['GET']) @user_bp.route('/users/<int:user_id>', methods=['GET'])
@jwt_required() @jwt_required()
@admin_required()
def get_user(user_id): def get_user(user_id):
user = User.query.get_or_404(user_id) user = User.query.get_or_404(user_id)
return jsonify(user.to_dict()) return jsonify(user.to_dict())
@ -35,7 +34,6 @@ def create_user():
@user_bp.route('/users/<int:user_id>', methods=['PUT']) @user_bp.route('/users/<int:user_id>', methods=['PUT'])
@jwt_required() @jwt_required()
def edit_user(user_id): def edit_user(user_id):
current_user = User.query.filter_by(username=get_jwt_identity()).first()
request_data = request.get_json() request_data = request.get_json()
user_to_update = User.query.get_or_404(user_id) user_to_update = User.query.get_or_404(user_id)
request_username = request_data.get('username') request_username = request_data.get('username')