dschick/git_brcal
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #46 from puls200/hotfix/20240909

Browse Source 
Hotfix/20240909
This commit is contained in:
Daniel Schick 2024-09-09 19:54:34 +02:00 committed by GitHub
commit 4f88e493d9
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
10 changed files with 149 additions and 131 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
src/server/BreCal/api/berths.py
View File

@ -1,8 +1,10 @@
import logging
from flask import Blueprint, request
from webargs.flaskparser import parser
from .. import impl
from ..services.auth_guard import auth_guard
import json
from BreCal.validators.validation_error import create_dynamic_exception_response
bp = Blueprint('berths', __name__)
@ -11,8 +13,12 @@ bp = Blueprint('berths', __name__)
@auth_guard() # no restriction by role
def GetBerths():
if 'Authorization' in request.headers:
token = request.headers.get('Authorization')
return impl.berths.GetBerths(token)
else:
return json.dumps("not authenticated"), 403
try:
if 'Authorization' in request.headers:
token = request.headers.get('Authorization')
return impl.berths.GetBerths(token)
else:
return create_dynamic_exception_response(ex=None, status_code=403, message="not authenticated")
except Exception as ex:
return create_dynamic_exception_response(ex=ex, status_code=400)

25
src/server/BreCal/api/history.py
View File

@ -1,21 +1,26 @@
import logging
from flask import Blueprint, request
from .. import impl
from ..services.auth_guard import auth_guard
import json
from BreCal.validators.validation_error import create_dynamic_exception_response
bp = Blueprint('history', __name__)
@bp.route('/history', methods=['get'])
@auth_guard() # no restriction by role
def GetParticipant():
def GetHistory():
if 'Authorization' in request.headers:
token = request.headers.get('Authorization')
options = {}
if not 'shipcall_id' in request.args:
return json.dumps("missing parameter"), 400
options["shipcall_id"] = request.args.get("shipcall_id")
return impl.history.GetHistory(options)
else:
return json.dumps("not authenticated"), 403
try:
if 'Authorization' in request.headers:
token = request.headers.get('Authorization')
options = {}
if not 'shipcall_id' in request.args:
return create_dynamic_exception_response(ex=None, status_code=400, message="missing parameter: shipcall_id")
options["shipcall_id"] = request.args.get("shipcall_id")
return impl.history.GetHistory(options)
else:
return create_dynamic_exception_response(ex=None, status_code=403, message="not authenticated")
except Exception as ex:
return create_dynamic_exception_response(ex=ex, status_code=400)

18
src/server/BreCal/api/notifications.py
View File

@ -3,6 +3,7 @@ from .. import impl
from ..services.auth_guard import auth_guard
import logging
import json
from BreCal.validators.validation_error import create_dynamic_exception_response
bp = Blueprint('notifications', __name__)
@ -10,10 +11,13 @@ bp = Blueprint('notifications', __name__)
@bp.route('/notifications', methods=['get'])
@auth_guard() # no restriction by role
def GetNotifications():
if 'shipcall_id' in request.args:
options = {}
options["shipcall_id"] = request.args.get("shipcall_id")
return impl.notifications.GetNotifications(options)
else:
logging.warning("attempt to load notifications without shipcall id")
return json.dumps("missing argument"), 400
try:
if 'shipcall_id' in request.args:
options = {}
options["shipcall_id"] = request.args.get("shipcall_id")
return impl.notifications.GetNotifications(options)
else:
return create_dynamic_exception_response(ex=None, status_code=400, message="missing argument: shipcall_id")
except Exception as ex:
return create_dynamic_exception_response(ex=ex, status_code=400)

19
src/server/BreCal/api/participant.py
View File

@ -1,7 +1,9 @@
import logging
from flask import Blueprint, request
from .. import impl
from ..services.auth_guard import auth_guard
import json
from BreCal.validators.validation_error import create_dynamic_exception_response
bp = Blueprint('participants', __name__)
@ -9,11 +11,14 @@ bp = Blueprint('participants', __name__)
@auth_guard() # no restriction by role
def GetParticipant():
if 'Authorization' in request.headers:
token = request.headers.get('Authorization')
options = {}
options["user_id"] = request.args.get("user_id")
return impl.participant.GetParticipant(options)
else:
return json.dumps("not authenticated"), 403
try:
if 'Authorization' in request.headers:
token = request.headers.get('Authorization')
options = {}
options["user_id"] = request.args.get("user_id")
return impl.participant.GetParticipant(options)
else:
return create_dynamic_exception_response(ex=None, status_code=403, message="not authenticated")
except Exception as ex:
return create_dynamic_exception_response(ex=ex, status_code=400)

56
src/server/BreCal/api/shipcalls.py
View File

@ -7,7 +7,7 @@ from ..services.auth_guard import auth_guard, check_jwt
from BreCal.validators.input_validation import validate_posted_shipcall_data, check_if_user_is_bsmd_type
from BreCal.validators.input_validation_shipcall import InputValidationShipcall
from BreCal.database.sql_handler import execute_sql_query_standalone
from BreCal.validators.validation_error import create_validation_error_response, create_werkzeug_error_response
from BreCal.validators.validation_error import create_validation_error_response, create_werkzeug_error_response, create_dynamic_exception_response
from . import verify_if_request_is_json
import logging
@ -20,24 +20,28 @@ bp = Blueprint('shipcalls', __name__)
@bp.route('/shipcalls', methods=['get'])
@auth_guard() # no restriction by role
def GetShipcalls():
if 'Authorization' in request.headers:
token = request.headers.get('Authorization') # see impl/login to see the token encoding, which is a JWT token.
try:
if 'Authorization' in request.headers:
token = request.headers.get('Authorization') # see impl/login to see the token encoding, which is a JWT token.
"""
from BreCal.services.jwt_handler import decode_jwt
jwt = token.split('Bearer ')[1] # string key
payload = decode_jwt(jwt) # dictionary, which includes 'id' (user id) and 'participant_id'
"""
from BreCal.services.jwt_handler import decode_jwt
jwt = token.split('Bearer ')[1] # string key
payload = decode_jwt(jwt) # dictionary, which includes 'id' (user id) and 'participant_id'
# oneline:
payload = decode_jwt(request.headers.get("Authorization").split("Bearer ")[-1])
"""
options = {}
options["participant_id"] = request.args.get("participant_id")
options["past_days"] = request.args.get("past_days", default=1, type=int)
# oneline:
payload = decode_jwt(request.headers.get("Authorization").split("Bearer ")[-1])
"""
options = {}
options["participant_id"] = request.args.get("participant_id")
options["past_days"] = request.args.get("past_days", default=1, type=int)
return impl.shipcalls.GetShipcalls(options)
else:
return json.dumps("not authenticated"), 403
return impl.shipcalls.GetShipcalls(options)
else:
return create_dynamic_exception_response(ex=None, status_code=403, message="not authenticated")
except Exception as ex:
return create_dynamic_exception_response(ex=ex, status_code=400)
@bp.route('/shipcalls', methods=['post'])
@ -55,19 +59,14 @@ def PostShipcalls():
# validate the posted shipcall data & the user's authority
InputValidationShipcall.evaluate_post_data(user_data, loadedModel, content)
return impl.shipcalls.PostShipcalls(loadedModel)
except ValidationError as ex:
logging.error(ex)
print(ex)
return create_validation_error_response(ex=ex, status_code=400)
except Exception as ex:
logging.error(ex)
logging.error(traceback.format_exc())
print(ex)
return json.dumps("bad format"), 400
return impl.shipcalls.PostShipcalls(loadedModel)
return create_dynamic_exception_response(ex=ex, status_code=400, message="bad format")
@bp.route('/shipcalls', methods=['put'])
@ -85,20 +84,15 @@ def PutShipcalls():
# validate the PUT shipcall data and the user's authority
InputValidationShipcall.evaluate_put_data(user_data, loadedModel, content)
return impl.shipcalls.PutShipcalls(loadedModel)
except ValidationError as ex:
logging.error(ex)
print(ex)
return create_validation_error_response(ex=ex, status_code=400)
except werkzeug.exceptions.Forbidden as ex:
logging.error(ex)
print(ex)
return create_werkzeug_error_response(ex=ex, status_code=403)
except Exception as ex:
logging.error(ex)
print(ex)
return json.dumps("bad format"), 400
logging.error(traceback.format_exc())
return create_dynamic_exception_response(ex=None, status_code=400, message="bad format")
return impl.shipcalls.PutShipcalls(loadedModel)

44
src/server/BreCal/api/ships.py
View File

@ -6,7 +6,7 @@ from ..schemas import model
import json
import logging
from . import verify_if_request_is_json
from BreCal.validators.validation_error import create_validation_error_response
from BreCal.validators.validation_error import create_validation_error_response, create_dynamic_exception_response
from BreCal.validators.input_validation import check_if_user_is_bsmd_type
from BreCal.validators.input_validation_ship import InputValidationShip
@ -17,11 +17,15 @@ bp = Blueprint('ships', __name__)
@auth_guard() # no restriction by role
def GetShips():
if 'Authorization' in request.headers:
token = request.headers.get('Authorization')
return impl.ships.GetShips(token)
else:
return json.dumps("not authenticated"), 403
try:
if 'Authorization' in request.headers:
token = request.headers.get('Authorization')
return impl.ships.GetShips(token)
else:
return create_dynamic_exception_response(ex=None, status_code=403, message="not authenticated")
except Exception as ex:
return create_dynamic_exception_response(ex=ex, status_code=400)
@bp.route('/ships', methods=['post'])
@ -45,18 +49,13 @@ def PostShip():
# validate the request data & user permissions
InputValidationShip.evaluate_post_data(user_data, loadedModel, content)
return impl.ships.PostShip(loadedModel)
except ValidationError as ex:
logging.error(ex)
print(ex)
return create_validation_error_response(ex=ex, status_code=400)
except Exception as ex:
logging.error(ex)
print(ex)
return json.dumps(repr(ex)), 400
return impl.ships.PostShip(loadedModel)
return create_dynamic_exception_response(ex=ex, status_code=400, message=None)
@bp.route('/ships', methods=['put'])
@ -74,18 +73,13 @@ def PutShip():
# validate the request data & user permissions
InputValidationShip.evaluate_put_data(user_data, loadedModel, content)
return impl.ships.PutShip(loadedModel)
except ValidationError as ex:
logging.error(ex)
print(ex)
return create_validation_error_response(ex=ex, status_code=400)
except Exception as ex:
logging.error(ex)
print(ex)
return json.dumps(repr(ex)), 400
return impl.ships.PutShip(loadedModel)
return create_dynamic_exception_response(ex=ex, status_code=400)
@bp.route('/ships', methods=['delete'])
@ -103,19 +97,15 @@ def DeleteShip():
options = {}
options["id"] = request.args.get("id")
else:
return json.dumps("no id provided"), 400
return create_dynamic_exception_response(ex=ex, status_code=400, message="no id provided")
# validate the request data & user permissions
InputValidationShip.evaluate_delete_data(user_data, ship_id)
return impl.ships.DeleteShip(options)
except ValidationError as ex:
logging.error(ex)
print(ex)
return create_validation_error_response(ex=ex, status_code=400)
except Exception as ex:
logging.error(ex)
print(ex)
return json.dumps(repr(ex)), 400
return create_dynamic_exception_response(ex=ex, status_code=400)
return impl.ships.DeleteShip(options)

40
src/server/BreCal/api/times.py
View File

@ -7,7 +7,7 @@ import logging
from marshmallow import ValidationError
from BreCal.validators.input_validation_times import InputValidationTimes
from . import verify_if_request_is_json
from BreCal.validators.validation_error import create_validation_error_response, create_werkzeug_error_response
from BreCal.validators.validation_error import create_validation_error_response, create_werkzeug_error_response, create_dynamic_exception_response
bp = Blueprint('times', __name__)
@ -16,10 +16,13 @@ bp = Blueprint('times', __name__)
@auth_guard() # no restriction by role
def GetTimes():
options = {}
options["shipcall_id"] = request.args.get("shipcall_id")
return impl.times.GetTimes(options)
try:
options = {}
options["shipcall_id"] = request.args.get("shipcall_id")
return impl.times.GetTimes(options)
except Exception as ex:
return create_dynamic_exception_response(ex=ex, status_code=400)
@bp.route('/times', methods=['post'])
@ -41,18 +44,14 @@ def PostTimes():
# validate the request
InputValidationTimes.evaluate_post_data(user_data, loadedModel, content)
return impl.times.PostTimes(loadedModel)
except ValidationError as ex:
logging.error(ex)
print(ex)
return create_validation_error_response(ex=ex, status_code=400)
except Exception as ex:
logging.error(ex)
print(ex)
return json.dumps("bad format"), 400
return create_dynamic_exception_response(ex=ex, status_code=400, message="bad format")
return impl.times.PostTimes(loadedModel)
@bp.route('/times', methods=['put'])
@ -70,18 +69,14 @@ def PutTimes():
# validate the request
InputValidationTimes.evaluate_put_data(user_data, loadedModel, content)
return impl.times.PutTimes(loadedModel)
except ValidationError as ex:
logging.error(ex)
print(ex)
return create_validation_error_response(ex=ex, status_code=400)
except Exception as ex:
logging.error(ex)
print(ex)
return json.dumps("bad format"), 400
return create_dynamic_exception_response(ex=ex, status_code=400, message="bad format")
return impl.times.PutTimes(loadedModel)
@bp.route('/times', methods=['delete'])
@ -101,15 +96,10 @@ def DeleteTimes():
return impl.times.DeleteTimes(options)
else:
logging.warning("Times delete missing id argument")
return json.dumps("missing argument"), 400
return create_dynamic_exception_response(ex=None, status_code=400, message="Times delete missing argument: id")
except ValidationError as ex:
logging.error(ex)
print(ex)
return create_validation_error_response(ex=ex, status_code=400)
except Exception as ex:
logging.error(ex)
print(ex)
return json.dumps("bad format"), 400
return create_dynamic_exception_response(ex=ex, status_code=400, message="bad format")

11
src/server/BreCal/api/user.py
View File

@ -6,6 +6,7 @@ import json
import logging
from marshmallow import ValidationError
from . import verify_if_request_is_json
from BreCal.validators.validation_error import create_dynamic_exception_response, create_validation_error_response
bp = Blueprint('user', __name__)
@ -18,16 +19,12 @@ def PutUser():
content = request.get_json(force=True)
loadedModel = model.UserSchema().load(data=content, many=False, partial=True)
return impl.user.PutUser(loadedModel)
except ValidationError as ex:
logging.error(ex)
print(ex)
return json.dumps(f"bad format. \nError Messages: {ex.messages}. \nValid Data: {ex.valid_data}"), 400
return create_validation_error_response(ex=ex, status_code=400)
except Exception as ex:
logging.error(ex)
print(ex)
return json.dumps("bad format"), 400
return create_dynamic_exception_response(ex=None, status_code=400, message="bad format")
return impl.user.PutUser(loadedModel)

8
src/server/BreCal/validators/input_validation_times.py
View File

@ -85,13 +85,10 @@ class InputValidationTimes():
# 2.) datasets may only be created, if the respective participant type did not already create one.
InputValidationTimes.check_if_entry_already_exists_for_participant_type(user_data, loadedModel, content)
# 3.) only users who are *not* of type BSMD may post times datasets.
InputValidationTimes.check_user_is_not_bsmd_type(user_data)
# 4.) Reference checking
# 3.) Reference checking
InputValidationTimes.check_dataset_references(content)
# 5.) Value checking
# 4.) Value checking
InputValidationTimes.check_dataset_values(user_data, loadedModel, content)
return
@ -143,6 +140,7 @@ class InputValidationTimes():
@staticmethod
def check_user_is_not_bsmd_type(user_data:dict):
"""a new dataset may only be created by a user who is *not* belonging to participant group BSMD"""
# this method is deprecated for /times POST requests. As the function may be used elsewhere, it will, for now, not be removed
is_bsmd = check_if_user_is_bsmd_type(user_data)
if is_bsmd:
raise ValidationError({"participant_type":f"current user belongs to BSMD. Cannot post 'times' datasets. Found user data: {user_data}"})

43
src/server/BreCal/validators/validation_error.py
View File

@ -6,8 +6,16 @@ import werkzeug
from werkzeug.exceptions import Forbidden
def create_default_error_format(error_description):
json_response = {
"message":f"{error_description}",
"errors":[],
"valid_data":{}
}
return json_response
def create_validation_error_response(ex:ValidationError, status_code:int=400)->typing.Tuple[str,int]:
def create_validation_error_response(ex:ValidationError, status_code:int=400, create_log:bool=True)->typing.Tuple[str,int]:
# generate an overview the errors
#example:
# {'lock_time': ['The provided value must be in the future. Current Time: 2024-09-02 08:23:32.600791, Value: 2024-09-01 08:20:41.853000']}
@ -25,6 +33,7 @@ def create_validation_error_response(ex:ValidationError, status_code:int=400)->t
# the following conversion snipped ensures a dictionary output
if isinstance(errors, (str,list)):
errors = {"undefined_schema":errors}
errors = {k:v if isinstance(v,list) else [v] for k,v in errors.items()}
# hence, errors always has the following type: dict[str, list[str]]
@ -37,21 +46,41 @@ def create_validation_error_response(ex:ValidationError, status_code:int=400)->t
# }
valid_data = ex.valid_data
json_response = {
"message":"ValidationError",
message = "ValidationError"
json_response = create_default_error_format(error_description=message)
json_response.update({
"errors":errors,
"valid_data":valid_data
}
})
# json.dumps with default=str automatically converts non-serializable values to strings. Hence, datetime objects (which are not)
# natively serializable are properly serialized.
serialized_response = json.dumps(json_response, default=str)
if create_log:
logging.warning(ex) if ex is not None else logging.warning(message)
print(ex) if ex is not None else print(message)
return (serialized_response, status_code)
def create_werkzeug_error_response(ex:Forbidden, status_code:int=403)->typing.Tuple[str,int]:
def create_werkzeug_error_response(ex:Forbidden, status_code:int=403, create_log:bool=True)->typing.Tuple[str,int]:
# json.dumps with default=str automatically converts non-serializable values to strings. Hence, datetime objects (which are not)
# natively serializable are properly serialized.
serialized_response = json.dumps({"message":ex.description}, default=str)
message = ex.description
json_response = create_default_error_format(error_description=message)
serialized_response = json.dumps(json_response, default=str)
if create_log:
logging.warning(ex) if ex is not None else logging.warning(message)
print(ex) if ex is not None else print(message)
return serialized_response, status_code
def create_dynamic_exception_response(ex, status_code:int=400, message:typing.Optional[str]=None, create_log:bool=True):
message = repr(ex) if message is None else message
json_response = create_default_error_format(error_description=message)
serialized_response = json.dumps(json_response, default=str)
if create_log:
logging.warning(ex) if ex is not None else logging.warning(message)
print(ex) if ex is not None else print(message)
return (serialized_response, status_code)