Alexander/FitTrackee
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

API - authentication decorators return user directly instead of user id

Browse Source
This commit is contained in:
Sam
2021-12-01 19:22:47 +01:00
parent d9048032e4
commit 00b6e05805
10 changed files with 88 additions and 133 deletions
Download Patch File Download Diff File Expand all files Collapse all files

65
fittrackee/users/auth.py
View File

@ -224,7 +224,7 @@ def login_user() -> Union[Dict, HttpResponse]:
@auth_blueprint.route('/auth/logout', methods=['GET'])
@authenticate
def logout_user(auth_user_id: int) -> Union[Dict, HttpResponse]:
def logout_user(auth_user: User) -> Union[Dict, HttpResponse]:
"""
user logout
@ -274,7 +274,7 @@ def logout_user(auth_user_id: int) -> Union[Dict, HttpResponse]:
auth_token = auth_header.split(' ')[1]
resp = User.decode_auth_token(auth_token)
if isinstance(auth_user_id, str):
if isinstance(resp, str):
return UnauthorizedErrorResponse(resp)
return {
@ -286,7 +286,7 @@ def logout_user(auth_user_id: int) -> Union[Dict, HttpResponse]:
@auth_blueprint.route('/auth/profile', methods=['GET'])
@authenticate
def get_authenticated_user_profile(
auth_user_id: int,
auth_user: User,
) -> Union[Dict, HttpResponse]:
"""
get authenticated user info
@ -381,13 +381,12 @@ def get_authenticated_user_profile(
- invalid token, please log in again
"""
user = User.query.filter_by(id=auth_user_id).first()
return {'status': 'success', 'data': user.serialize()}
return {'status': 'success', 'data': auth_user.serialize()}
@auth_blueprint.route('/auth/profile/edit', methods=['POST'])
@authenticate
def edit_user(auth_user_id: int) -> Union[Dict, HttpResponse]:
def edit_user(auth_user: User) -> Union[Dict, HttpResponse]:
"""
edit authenticated user
@ -523,24 +522,23 @@ def edit_user(auth_user_id: int) -> Union[Dict, HttpResponse]:
).decode()
try:
user = User.query.filter_by(id=auth_user_id).first()
user.first_name = first_name
user.last_name = last_name
user.bio = bio
user.location = location
user.birth_date = (
auth_user.first_name = first_name
auth_user.last_name = last_name
auth_user.bio = bio
auth_user.location = location
auth_user.birth_date = (
datetime.datetime.strptime(birth_date, '%Y-%m-%d')
if birth_date
else None
)
if password is not None and password != '':
user.password = password
auth_user.password = password
db.session.commit()
return {
'status': 'success',
'message': 'user profile updated',
'data': user.serialize(),
'data': auth_user.serialize(),
}
# handler errors
@ -550,7 +548,7 @@ def edit_user(auth_user_id: int) -> Union[Dict, HttpResponse]:
@auth_blueprint.route('/auth/profile/edit/preferences', methods=['POST'])
@authenticate
def edit_user_preferences(auth_user_id: int) -> Union[Dict, HttpResponse]:
def edit_user_preferences(auth_user: User) -> Union[Dict, HttpResponse]:
"""
edit authenticated user preferences
@ -670,17 +668,16 @@ def edit_user_preferences(auth_user_id: int) -> Union[Dict, HttpResponse]:
weekm = post_data.get('weekm')
try:
user = User.query.filter_by(id=auth_user_id).first()
user.imperial_units = imperial_units
user.language = language
user.timezone = timezone
user.weekm = weekm
auth_user.imperial_units = imperial_units
auth_user.language = language
auth_user.timezone = timezone
auth_user.weekm = weekm
db.session.commit()
return {
'status': 'success',
'message': 'user preferences updated',
'data': user.serialize(),
'data': auth_user.serialize(),
}
# handler errors
@ -691,7 +688,7 @@ def edit_user_preferences(auth_user_id: int) -> Union[Dict, HttpResponse]:
@auth_blueprint.route('/auth/profile/edit/sports', methods=['POST'])
@authenticate
def edit_user_sport_preferences(
auth_user_id: int,
auth_user: User,
) -> Union[Dict, HttpResponse]:
"""
edit authenticated user sport preferences
@ -758,12 +755,12 @@ def edit_user_sport_preferences(
try:
user_sport = UserSportPreference.query.filter_by(
user_id=auth_user_id,
user_id=auth_user.id,
sport_id=sport_id,
).first()
if not user_sport:
user_sport = UserSportPreference(
user_id=auth_user_id,
user_id=auth_user.id,
sport_id=sport_id,
stopped_speed_threshold=sport.stopped_speed_threshold,
)
@ -792,7 +789,7 @@ def edit_user_sport_preferences(
@auth_blueprint.route('/auth/picture', methods=['POST'])
@authenticate
def edit_picture(auth_user_id: int) -> Union[Dict, HttpResponse]:
def edit_picture(auth_user: User) -> Union[Dict, HttpResponse]:
"""
update authenticated user picture
@ -848,23 +845,22 @@ def edit_picture(auth_user_id: int) -> Union[Dict, HttpResponse]:
file = request.files['file']
filename = secure_filename(file.filename) # type: ignore
dirpath = os.path.join(
current_app.config['UPLOAD_FOLDER'], 'pictures', str(auth_user_id)
current_app.config['UPLOAD_FOLDER'], 'pictures', str(auth_user.id)
)
if not os.path.exists(dirpath):
os.makedirs(dirpath)
absolute_picture_path = os.path.join(dirpath, filename)
relative_picture_path = os.path.join(
'pictures', str(auth_user_id), filename
'pictures', str(auth_user.id), filename
)
try:
user = User.query.filter_by(id=auth_user_id).first()
if user.picture is not None:
old_picture_path = get_absolute_file_path(user.picture)
if auth_user.picture is not None:
old_picture_path = get_absolute_file_path(auth_user.picture)
if os.path.isfile(get_absolute_file_path(old_picture_path)):
os.remove(old_picture_path)
file.save(absolute_picture_path)
user.picture = relative_picture_path
auth_user.picture = relative_picture_path
db.session.commit()
return {
'status': 'success',
@ -879,7 +875,7 @@ def edit_picture(auth_user_id: int) -> Union[Dict, HttpResponse]:
@auth_blueprint.route('/auth/picture', methods=['DELETE'])
@authenticate
def del_picture(auth_user_id: int) -> Union[Tuple[Dict, int], HttpResponse]:
def del_picture(auth_user: User) -> Union[Tuple[Dict, int], HttpResponse]:
"""
delete authenticated user picture
@ -908,11 +904,10 @@ def del_picture(auth_user_id: int) -> Union[Tuple[Dict, int], HttpResponse]:
"""
try:
user = User.query.filter_by(id=auth_user_id).first()
picture_path = get_absolute_file_path(user.picture)
picture_path = get_absolute_file_path(auth_user.picture)
if os.path.isfile(picture_path):
os.remove(picture_path)
user.picture = None
auth_user.picture = None
db.session.commit()
return {'status': 'no content'}, 204
except (exc.IntegrityError, ValueError) as e:

10
fittrackee/users/decorators.py
View File

@ -13,11 +13,10 @@ def authenticate(f: Callable) -> Callable:
def decorated_function(
*args: Any, **kwargs: Any
) -> Union[Callable, HttpResponse]:
verify_admin = False
response_object, resp = verify_user(request, verify_admin)
response_object, user = verify_user(request, verify_admin=False)
if response_object:
return response_object
return f(resp, *args, **kwargs)
return f(user, *args, **kwargs)
return decorated_function
@ -27,10 +26,9 @@ def authenticate_as_admin(f: Callable) -> Callable:
def decorated_function(
*args: Any, **kwargs: Any
) -> Union[Callable, HttpResponse]:
verify_admin = True
response_object, resp = verify_user(request, verify_admin)
response_object, user = verify_user(request, verify_admin=True)
if response_object:
return response_object
return f(resp, *args, **kwargs)
return f(user, *args, **kwargs)
return decorated_function

18
fittrackee/users/users.py
View File

@ -27,7 +27,7 @@ USER_PER_PAGE = 10
@users_blueprint.route('/users', methods=['GET'])
@authenticate
def get_users(auth_user_id: int) -> Dict:
def get_users(auth_user: User) -> Dict:
"""
Get all users
@ -144,8 +144,6 @@ def get_users(auth_user_id: int) -> Dict:
"status": "success"
}
:param integer auth_user_id: authenticate user id (from JSON Web Token)
:query integer page: page if using pagination (default: 1)
:query integer per_page: number of users per page (default: 10, max: 50)
:query string q: query on user name
@ -219,7 +217,7 @@ def get_users(auth_user_id: int) -> Dict:
@users_blueprint.route('/users/<user_name>', methods=['GET'])
@authenticate
def get_single_user(
auth_user_id: int, user_name: str
auth_user: User, user_name: str
) -> Union[Dict, HttpResponse]:
"""
Get single user details
@ -306,7 +304,6 @@ def get_single_user(
"status": "success"
}
:param integer auth_user_id: authenticate user id (from JSON Web Token)
:param integer user_name: user name
:reqheader Authorization: OAuth 2.0 Bearer Token
@ -371,9 +368,7 @@ def get_picture(user_name: str) -> Any:
@users_blueprint.route('/users/<user_name>', methods=['PATCH'])
@authenticate_as_admin
def update_user(
auth_user_id: int, user_name: str
) -> Union[Dict, HttpResponse]:
def update_user(auth_user: User, user_name: str) -> Union[Dict, HttpResponse]:
"""
Update user to add admin rights
@ -461,7 +456,6 @@ def update_user(
"status": "success"
}
:param integer auth_user_id: authenticate user id (from JSON Web Token)
:param string user_name: user name
:<json boolean admin: does the user have administrator rights
@ -500,7 +494,7 @@ def update_user(
@users_blueprint.route('/users/<user_name>', methods=['DELETE'])
@authenticate
def delete_user(
auth_user_id: int, user_name: str
auth_user: User, user_name: str
) -> Union[Tuple[Dict, int], HttpResponse]:
"""
Delete a user account
@ -524,7 +518,6 @@ def delete_user(
HTTP/1.1 204 NO CONTENT
Content-Type: application/json
:param integer auth_user_id: authenticate user id (from JSON Web Token)
:param string user_name: user name
:reqheader Authorization: OAuth 2.0 Bearer Token
@ -543,12 +536,11 @@ def delete_user(
"""
try:
auth_user = User.query.filter_by(id=auth_user_id).first()
user = User.query.filter_by(username=user_name).first()
if not user:
return UserNotFoundErrorResponse()
if user.id != auth_user_id and not auth_user.admin:
if user.id != auth_user.id and not auth_user.admin:
return ForbiddenErrorResponse()
if (
user.admin is True

6
fittrackee/users/utils.py
View File

@ -62,7 +62,7 @@ def register_controls(
def verify_user(
current_request: Request, verify_admin: bool
) -> Tuple[Optional[HttpResponse], Optional[int]]:
) -> Tuple[Optional[HttpResponse], Optional[User]]:
"""
Return user id, if the provided token is valid and if user has admin
rights if 'verify_admin' is True
@ -78,9 +78,9 @@ def verify_user(
user = User.query.filter_by(id=resp).first()
if not user:
return UnauthorizedErrorResponse(default_message), None
if verify_admin and not is_admin(resp):
if verify_admin and not user.admin:
return ForbiddenErrorResponse(), None
return None, resp
return None, user
def can_view_workout(