mirror of
https://github.com/GSA/notifications-api.git
synced 2026-07-09 10:53:55 -04:00
17d14f291e
- Created new endpoint user/<user_id>/sms-code to send the sms verification code to the user. - Create new endpoirtn user/<user_id>/email-code to send the email verifcation code to the user. - Marked the old methods, schema, tests with a TODO to be deleted when the admin app is no longer sending messages to /user/<user_id>/code - Added error handlers for DataError and NoResultFound. Data error catches invalid input errors. - Added error handler for SqlAlchemyError which catches any other database errors. - Removed the need for the try catches around the db calls in the user endpoints with the addition of the db error handlers. - We may want to wrap db excpetions in the dao, if we want the No results found message to be more specific and say no result found for user.
197 lines
7.3 KiB
Python
197 lines
7.3 KiB
Python
from datetime import datetime
|
|
from flask import (jsonify, request, abort, Blueprint, current_app)
|
|
|
|
from app import encryption
|
|
from app.dao.services_dao import get_model_services
|
|
from app.dao.users_dao import (
|
|
get_model_users,
|
|
save_model_user,
|
|
delete_model_user,
|
|
create_user_code,
|
|
get_user_code,
|
|
use_user_code,
|
|
increment_failed_login_count,
|
|
reset_failed_login_count
|
|
)
|
|
from app.schemas import (
|
|
user_schema, users_schema, service_schema, services_schema,
|
|
old_request_verify_code_schema, user_schema_load_json,
|
|
request_verify_code_schema)
|
|
from app.celery.tasks import (send_sms_code, send_email_code)
|
|
|
|
user = Blueprint('user', __name__)
|
|
|
|
from app.errors import register_errors
|
|
register_errors(user)
|
|
|
|
|
|
@user.route('', methods=['POST'])
|
|
def create_user():
|
|
user, errors = user_schema.load(request.get_json())
|
|
req_json = request.get_json()
|
|
# TODO password policy, what is valid password
|
|
if not req_json.get('password', None):
|
|
errors.update({'password': ['Missing data for required field.']})
|
|
return jsonify(result="error", message=errors), 400
|
|
if errors:
|
|
return jsonify(result="error", message=errors), 400
|
|
save_model_user(user, pwd=req_json.get('password'))
|
|
return jsonify(data=user_schema.dump(user).data), 201
|
|
|
|
|
|
@user.route('/<int:user_id>', methods=['PUT', 'DELETE'])
|
|
def update_user(user_id):
|
|
user_to_update = get_model_users(user_id=user_id)
|
|
|
|
if request.method == 'DELETE':
|
|
status_code = 202
|
|
delete_model_user(user_to_update)
|
|
else:
|
|
req_json = request.get_json()
|
|
update_dct, errors = user_schema_load_json.load(req_json)
|
|
pwd = req_json.get('password', None)
|
|
# TODO password validation, it is already done on the admin app
|
|
# but would be good to have the same validation here.
|
|
if pwd is not None and not pwd:
|
|
errors.update({'password': ['Invalid data for field']})
|
|
if errors:
|
|
return jsonify(result="error", message=errors), 400
|
|
status_code = 200
|
|
save_model_user(user_to_update, update_dict=update_dct, pwd=pwd)
|
|
return jsonify(data=user_schema.dump(user_to_update).data), status_code
|
|
|
|
|
|
@user.route('/<int:user_id>/verify/password', methods=['POST'])
|
|
def verify_user_password(user_id):
|
|
user_to_verify = get_model_users(user_id=user_id)
|
|
|
|
txt_pwd = None
|
|
try:
|
|
txt_pwd = request.get_json()['password']
|
|
except KeyError:
|
|
return jsonify(
|
|
result="error",
|
|
message={'password': ['Required field missing data']}), 400
|
|
if user_to_verify.check_password(txt_pwd):
|
|
reset_failed_login_count(user_to_verify)
|
|
return jsonify({}), 204
|
|
else:
|
|
increment_failed_login_count(user_to_verify)
|
|
return jsonify(result='error', message={'password': ['Incorrect password']}), 400
|
|
|
|
|
|
@user.route('/<int:user_id>/verify/code', methods=['POST'])
|
|
def verify_user_code(user_id):
|
|
user_to_verify = get_model_users(user_id=user_id)
|
|
|
|
txt_code = None
|
|
resp_json = request.get_json()
|
|
txt_type = None
|
|
errors = {}
|
|
try:
|
|
txt_code = resp_json['code']
|
|
except KeyError:
|
|
errors.update({'code': ['Required field missing data']})
|
|
try:
|
|
txt_type = resp_json['code_type']
|
|
except KeyError:
|
|
errors.update({'code_type': ['Required field missing data']})
|
|
if errors:
|
|
return jsonify(result="error", message=errors), 400
|
|
code = get_user_code(user_to_verify, txt_code, txt_type)
|
|
if not code:
|
|
return jsonify(result="error", message="Code not found"), 404
|
|
if datetime.now() > code.expiry_datetime or code.code_used:
|
|
return jsonify(result="error", message="Code has expired"), 400
|
|
use_user_code(code.id)
|
|
return jsonify({}), 204
|
|
|
|
|
|
@user.route('/<int:user_id>/sms-code', methods=['POST'])
|
|
def send_user_sms_code(user_id):
|
|
user_to_send_to = get_model_users(user_id=user_id)
|
|
|
|
verify_code, errors = request_verify_code_schema.load(request.get_json())
|
|
if errors:
|
|
return jsonify(result="error", message=errors), 400
|
|
|
|
from app.dao.users_dao import create_secret_code
|
|
secret_code = create_secret_code()
|
|
create_user_code(user_to_send_to, secret_code, 'sms')
|
|
|
|
mobile = user_to_send_to.mobile_number if verify_code.get('to', None) is None else verify_code.get('to')
|
|
verification_message = {'to': mobile, 'secret_code': secret_code}
|
|
|
|
send_sms_code.apply_async([encryption.encrypt(verification_message)], queue='sms-code')
|
|
|
|
return jsonify({}), 204
|
|
|
|
|
|
@user.route('/<int:user_id>/email-code', methods=['POST'])
|
|
def send_user_email_code(user_id):
|
|
user_to_send_to = get_model_users(user_id=user_id)
|
|
|
|
verify_code, errors = request_verify_code_schema.load(request.get_json())
|
|
if errors:
|
|
return jsonify(result="error", message=errors), 400
|
|
|
|
from app.dao.users_dao import create_secret_code
|
|
secret_code = create_secret_code()
|
|
create_user_code(user_to_send_to, secret_code, 'email')
|
|
|
|
email = user_to_send_to.email_address if verify_code.get('to', None) is None else verify_code.get('to')
|
|
verification_message = {'to': email, 'secret_code': secret_code}
|
|
|
|
send_email_code.apply_async([encryption.encrypt(verification_message)], queue='email-code')
|
|
|
|
return jsonify({}), 204
|
|
|
|
|
|
# TODO: Remove this method once the admin app has stopped using it.
|
|
@user.route('/<int:user_id>/code', methods=['POST'])
|
|
def send_user_code(user_id):
|
|
user_to_send_to = get_model_users(user_id=user_id)
|
|
|
|
verify_code, errors = old_request_verify_code_schema.load(request.get_json())
|
|
if errors:
|
|
return jsonify(result="error", message=errors), 400
|
|
|
|
from app.dao.users_dao import create_secret_code
|
|
secret_code = create_secret_code()
|
|
create_user_code(user_to_send_to, secret_code, verify_code.get('code_type'))
|
|
if verify_code.get('code_type') == 'sms':
|
|
mobile = user_to_send_to.mobile_number if verify_code.get('to', None) is None else verify_code.get('to')
|
|
verification_message = {'to': mobile, 'secret_code': secret_code}
|
|
send_sms_code.apply_async([encryption.encrypt(verification_message)], queue='sms-code')
|
|
elif verify_code.get('code_type') == 'email':
|
|
email = user_to_send_to.email_address if verify_code.get('to', None) is None else verify_code.get('to')
|
|
verification_message = {
|
|
'to_address': email,
|
|
'from_address': current_app.config['VERIFY_CODE_FROM_EMAIL_ADDRESS'],
|
|
'subject': 'Verification code',
|
|
'body': secret_code}
|
|
send_email_code.apply_async([encryption.encrypt(verification_message)], queue='email-code')
|
|
else:
|
|
abort(500)
|
|
return jsonify({}), 204
|
|
|
|
|
|
@user.route('/<int:user_id>', methods=['GET'])
|
|
@user.route('', methods=['GET'])
|
|
def get_user(user_id=None):
|
|
users = get_model_users(user_id=user_id)
|
|
|
|
result = users_schema.dump(users) if isinstance(users, list) else user_schema.dump(users)
|
|
return jsonify(data=result.data)
|
|
|
|
|
|
@user.route('/<int:user_id>/service', methods=['GET'])
|
|
@user.route('/<int:user_id>/service/<service_id>', methods=['GET'])
|
|
def get_service_by_user_id(user_id, service_id=None):
|
|
ret_user = get_model_users(user_id=user_id)
|
|
|
|
services = get_model_services(user_id=ret_user.id, service_id=service_id)
|
|
|
|
services, errors = services_schema.dump(services) if isinstance(services, list) else service_schema.dump(services)
|
|
return jsonify(data=services)
|