modify_user function
@@ -2079,6 +2079,63 @@ class SQLAlchemySocket:
return (True, "Success")
+ def modify_user(self,
+ username: str,
+ password: Optional[str] = None,
+ reset_password: bool = False,
+ permissions: Optional[List[str]] = None) -> Tuple[bool, str]:
+ """
+ Alters a user's password, permissions, or both
+
+ Passwords are stored using bcrypt.
+
+ Parameters
+ ----------
+ username : str
+ The username
+ password : str, optional
+ The user's new password. If None, the password will not be updated. Excludes reset_password.
+ reset_password: bool, optional
+ Reset the user's password to a new autogenerated one.
+ permissions : list of str, optional
+ The associated permissions of a user ['read', 'write', 'compute', 'queue', 'admin']
+
+ Returns
+ -------
+ tuple
+ A tuple of (success flag, message)
+ """
+
+ if reset_password and password is not None:
+ return False, "only one of reset_password and password may be specified"
+
+ with self.session_scope() as session:
+ data = session.query(UserORM).filter_by(username=username).first()
+
+ if data is None:
+ return False, f"User {username} not found."
+
+ blob = {"username": username}
+
+ if permissions is not None:
+ # Make sure permissions are valid
+ if not SQLAlchemySocket.VALID_PERMISSIONS >= set(permissions):
+ return False, "Permissions not understood: {}".format(
+ set(permissions) - SQLAlchemySocket.VALID_PERMISSIONS)
+ blob["permissions"] = permissions
+ if reset_password:
+ password = SQLAlchemySocket._generate_password()
+ if password is not None:
+ blob["password"] = bcrypt.hashpw(password.encode("UTF-8"), bcrypt.gensalt(6))
+
+ count = session.query(UserORM).filter_by(username=username).update(blob)
+ success = count == 1
+
+ if success:
+ return True, None if password is None else f"New password is {password}"
+ else:
+ return False, f"Failed to modify user {username}"
+
def remove_user(self,
username: str) -> bool:
"""Removes a user
