Scott.Idem/OSIT-AE-API-FastAPI
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Working on user login, verification, and password change.

Browse Source
This commit is contained in:
Scott Idem
2021-10-06 17:34:29 -04:00
parent 7c919b513c
commit 455cc36a69
2 changed files with 127 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
app/models/user_models.py
View File

@@ -137,6 +137,7 @@ class User_New_Base(BaseModel):
class Config: class Config:
underscore_attrs_are_private = True underscore_attrs_are_private = True
allow_population_by_field_name = True
fields = base_fields fields = base_fields
# ### END ### API User Models ### User_New_Base() ### # ### END ### API User Models ### User_New_Base() ###
@@ -219,6 +220,7 @@ class User_Out_Base(BaseModel):
class Config: class Config:
underscore_attrs_are_private = True underscore_attrs_are_private = True
allow_population_by_field_name = True
fields = base_fields fields = base_fields
# ### END ### API User Models ### User_Out_Base() ### # ### END ### API User Models ### User_Out_Base() ###
@@ -359,5 +361,6 @@ class User_Base(BaseModel):
class Config: class Config:
underscore_attrs_are_private = True underscore_attrs_are_private = True
allow_population_by_field_name = True
fields = base_fields fields = base_fields
# ### END ### API User Models ### User_Base() ### # ### END ### API User Models ### User_Base() ###

139
app/routers/user.py
View File

@@ -85,16 +85,17 @@ async def post_user_obj_new(
# ### END ### API User ### post_user_obj_new() ### # ### END ### API User ### post_user_obj_new() ###
@router.patch('/change_password/{user_id}', response_model=Resp_Body_Base) # ### BEGIN ### API User ### user_obj_change_password() ###
async def change_user_obj_password( @router.patch('/{user_id}/change_password', response_model=Resp_Body_Base)
async def user_obj_change_password(
user_id: Union[int,str], user_id: Union[int,str],
password: Optional[str] = Query(None, min_length=6, max_length=50), user_obj: User_Base,
x_account_id: Optional[str] = Header(..., ), x_account_id: Optional[str] = Header(..., ),
return_obj: bool = False, return_obj: bool = False,
inc_user_role_list: bool = False, inc_user_role_list: bool = False,
inc_contact: bool = False, # inc_contact: bool = False,
inc_organization: bool = False, # inc_organization: bool = False,
inc_person: bool = False, # inc_person: bool = False,
by_alias: bool = True, by_alias: bool = True,
exclude_unset: bool = True, exclude_unset: bool = True,
response: Response = Response, response: Response = Response,
@@ -102,10 +103,19 @@ async def change_user_obj_password(
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals()) log.debug(locals())
if password := user_obj.password: pass
else: return mk_resp(data=False, status_code=400, status_message='The new password is required.', response=response) # Bad Request
generated_password = None
if password and len(password) >= 10: pass if password and len(password) >= 10: pass
elif password and len(password) < 10:
log.warning(f'The password given must be at least 10 characters. User ID: {user_id}')
return mk_resp(data=False, status_code=400, response=response, status_message=f'The password given must be at least 10 characters. User ID: {user_id}') # Bad Request
else: else:
log.warning('The password given must be at least 10 characters. Generating a new random password.') log.warning('No password was given. Generating a new random password.')
password = secrets.token_urlsafe(default_num_bytes) generated_password = secrets.token_urlsafe(default_num_bytes)
password = generated_password
if user_id := redis_lookup_id_random(record_id_random=user_id, table_name='user'): pass if user_id := redis_lookup_id_random(record_id_random=user_id, table_name='user'): pass
else: return mk_resp(data=False, status_code=404, response=response) # Not Found else: return mk_resp(data=False, status_code=404, response=response) # Not Found
@@ -116,20 +126,26 @@ async def change_user_obj_password(
user_data['password'] = secure_hash_string(string=password) user_data['password'] = secure_hash_string(string=password)
table_name = 'user' table_name = 'user'
user_rec_update_result = sql_update(data=user_data, table_name=table_name, record_id=user_id, id_random_length=None) if user_rec_update_result := sql_update(data=user_data, table_name=table_name, record_id=user_id, id_random_length=None): pass
else: mk_resp(data=False, status_code=500, status_message='Something went wrong while trying to update the password record.', response=response)
if return_obj: if return_obj:
user_obj = load_user_obj( user_obj = load_user_obj(
user_id=user_id, user_id = user_id,
inc_contact=inc_contact, inc_user_role_list = inc_user_role_list
inc_organization=inc_organization, # inc_contact = inc_contact,
inc_person=inc_person # inc_organization = inc_organization,
# inc_person = inc_person
).dict(by_alias=by_alias, exclude_unset=exclude_unset) ).dict(by_alias=by_alias, exclude_unset=exclude_unset)
data = user_obj data = user_obj
else: else:
data = True data = True
return mk_resp(data=data, response=response) if generated_password:
return mk_resp(data=data, status_message='Generated password: fake-testing-12345', response=response)
else:
return mk_resp(data=data, status_message='The password has been changed.', response=response)
#return mk_resp(data=None, status_code=501, response=response) # Not Implemented #return mk_resp(data=None, status_code=501, response=response) # Not Implemented
# ### END ### API User ### user_obj_change_password() ###
@router.patch('/{obj_id}', response_model=Resp_Body_Base) @router.patch('/{obj_id}', response_model=Resp_Body_Base)
@@ -212,7 +228,7 @@ async def user_authenticate(
account_id: Optional[Union[int,str]] = None, account_id: Optional[Union[int,str]] = None,
user_id: Optional[str] = Query(None, min_length=11, max_length=22), user_id: Optional[str] = Query(None, min_length=11, max_length=22),
username: Optional[str] = Query(None, min_length=3, max_length=50), username: Optional[str] = Query(None, min_length=3, max_length=50),
password: Optional[str] = Query(None, min_length=6, max_length=100), password: Optional[str] = Query(None, min_length=8, max_length=100),
auth_key: Optional[str] = Query(None, min_length=11, max_length=22), auth_key: Optional[str] = Query(None, min_length=11, max_length=22),
x_account_id: str = Header(...), x_account_id: str = Header(...),
inc_user_role_list: bool = False, inc_user_role_list: bool = False,
@@ -362,6 +378,99 @@ async def user_authenticate(
# ### END ### API User Routers ### user_authenticate() ### # ### END ### API User Routers ### user_authenticate() ###
# ### BEGIN ### API User ### user_verify_password() ###
# @router.post('/{user_id}/verify_password', response_model=Resp_Body_Base)
@router.post('/verify_password', response_model=Resp_Body_Base)
async def user_verify_password(
user_obj: User_Base,
# user_id: Optional[str] = Query(None, min_length=11, max_length=22),
# username: Optional[str] = Query(None, min_length=3, max_length=50),
# password: Optional[str] = Query(None, min_length=8, max_length=50),
x_account_id: Optional[str] = Header(..., ),
return_obj: bool = False,
by_alias: bool = True,
exclude_unset: bool = True,
response: Response = Response,
):
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
if account_id := redis_lookup_id_random(record_id_random=x_account_id, table_name='account'): pass
else: return mk_resp(data=False, status_code=404, response=response) # Not Found
if password := user_obj.password: pass
else: return mk_resp(data=False, status_code=400, status_message='The password to verify is required.', response=response) # Bad Request
if user_id_random := user_obj.id_random: # Use id_random instead of user_id_random when getting from User model.
log.info(f'Using the user ID to look up the user. User ID: {user_id_random}')
# NOTE: Not doing a redis lookup since we have to look up the record again. Redis lookup may save or add an insignificant amount of time.
user_data = {}
user_data['user_id_random'] = user_id_random
sql = f"""
SELECT `user`.id AS 'user_id', `user`.id_random AS 'user_id_random', `user`.username, `user`.password, `user`.enable, `user`.enable_from, `user`.enable_to
FROM `user` AS `user`
WHERE `user`.id_random = :user_id_random
LIMIT 1
"""
if user_rec_result := sql_select(data=user_data, sql=sql):
user_id = user_rec_result.get('user_id', None)
if password_hash := user_rec_result.get('password', None):
username = user_rec_result.get('username', None)
if verify_secure_hash_string(string=password, string_hash=password_hash):
log.info(f'The username was found, and the password matched. Log in allowed if the account is enabled. Account ID: {account_id}, Username: {username}')
return mk_resp(data=True, response=response)
else:
log.info(f'The username was found, but the password did not match. Not allowed to log in. Account ID: {account_id}, Username: {username}')
# NOTE: Returning a 404 instead of 200 even though the actual user record was found.
return mk_resp(data=False, status_code=404, status_message=f'The username was found, but the password did not match. Not allowed to log in. Account ID: {account_id}, Username: {username}', response=response) # Not Found
else:
log.warning(f'The password hash has was not found. Not allowed to log in. Account ID: {account_id}, Username: {username}')
return mk_resp(data=False, status_code=400, status_message=f'The password hash has was not found. Not allowed to log in. Account ID: {account_id}, Username: {username}', response=response)
else:
log.info(f'A user account was not found with the account and username given. Not allowed to log in. Account ID: {account_id}, Username: {username}')
return mk_resp(data=None, status_code=404, status_message=f'A user account was not found with the account and username given. Not allowed to log in. Account ID: {account_id}, Username: {username}', response=response) # Not Found
elif username := user_obj.username:
log.info(f'Using the username to look up the user. User ID: {username}')
user_data = {}
user_data['account_id'] = account_id
user_data['username'] = username
sql = f"""
SELECT `user`.id AS 'user_id', `user`.id_random AS 'user_id_random', `user`.username, `user`.password, `user`.enable, `user`.enable_from, `user`.enable_to
FROM `user` AS `user`
WHERE `user`.account_id = :account_id AND `user`.username = :username
LIMIT 1
"""
if user_rec_result := sql_select(data=user_data, sql=sql):
user_id = user_rec_result.get('user_id', None)
if password_hash := user_rec_result.get('password', None):
if verify_secure_hash_string(string=password, string_hash=password_hash):
log.info(f'The username was found, and the password matched. Log in allowed if the account is enabled. Account ID: {account_id}, Username: {username}')
return mk_resp(data=True, response=response)
else:
log.info(f'The username was found, but the password did not match. Not allowed to log in. Account ID: {account_id}, Username: {username}')
return mk_resp(data=False, status_message=f'The username was found, but the password did not match. Not allowed to log in. Account ID: {account_id}, Username: {username}', response=response)
else:
log.warning(f'The password hash has was not found. Not allowed to log in. Account ID: {account_id}, Username: {username}')
return mk_resp(data=False, status_code=400, status_message=f'The password hash has was not found. Not allowed to log in. Account ID: {account_id}, Username: {username}', response=response)
else:
log.info(f'A user account was not found with the account and username given. Not allowed to log in. Account ID: {account_id}, Username: {username}')
return mk_resp(data=None, status_code=404, status_message=f'A user account was not found with the account and username given. Not allowed to log in. Account ID: {account_id}, Username: {username}', response=response) # Not Found
else:
log.warning(f'A user ID or username is required. Can not verify password.')
return mk_resp(data=False, status_code=400, status_message=f'A user ID or username is required. Can not verify password.', response=response)
# ### END ### API User ### user_verify_password() ###
@router.get('/list', response_model=Resp_Body_Base) @router.get('/list', response_model=Resp_Body_Base)
async def get_user_obj_li( async def get_user_obj_li(
for_obj_type: Optional[str] = Query(None, min_length=2, max_length=50), for_obj_type: Optional[str] = Query(None, min_length=2, max_length=50),