Currently the function that allow you to update the user doesn't check if the new username already exist, this could cause authentication issue as the username is used to differentiates between user.
My solutions :
# UPDATE User profile
@router.put('/{username}/update', response_model=User)
async def update_user(attributes: dict, username: str):
# Add check to stop call if password is being changed
for k in attributes:
if k == 'hashed_password':
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail="Operation not permitted, cannot update password with this method.",
headers={"WWW-Authenticate": "Bearer"})
if k == 'username':
#print(f"\n update \n {username}\n-\n{attributes['username']}\n-\n")
query = 'MATCH (user:User) WHERE user.username = $username RETURN user'
name = attributes['username']
with neo4j_driver.session() as session:
user_in_db = session.run(query=query, parameters={'username': name})
data = user_in_db.data()
#print(f"data {data}\n")
if data:
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail="Username already exists.",
headers={"WWW-user-delete": "Bearer"})
if attributes:
unpacked_attributes = 'SET ' + ', '.join(f'user.{key}=\'{value}\'' for (key, value) in attributes.items())
else:
unpacked_attributes = ''
# Execute Cypher query to update the user attributes
cypher_update_user = ('MATCH (user: User) WHERE user.username = $user\n'
f'{unpacked_attributes}\n'
'RETURN user')
with neo4j_driver.session() as session:
updated_user = session.run(query=cypher_update_user,
parameters={'user': username})
user_data = updated_user.data()[0]['user']
return User(**user_data)
Currently the function that allow you to update the user doesn't check if the new username already exist, this could cause authentication issue as the username is used to differentiates between user.
My solutions :