forked from platypush/platypush
150 lines
4.6 KiB
Python
150 lines
4.6 KiB
Python
from platypush.plugins import Plugin, action
|
|
from platypush.user import UserManager
|
|
|
|
|
|
class UserPlugin(Plugin):
|
|
"""
|
|
Plugin to programmatically create and manage users and user sessions
|
|
"""
|
|
|
|
def __init__(self, **kwargs):
|
|
super().__init__(**kwargs)
|
|
self.user_manager = UserManager()
|
|
|
|
@action
|
|
def create_user(self, username, password, executing_user=None, executing_user_password=None, session_token=None,
|
|
**kwargs):
|
|
"""
|
|
Create a user. This action needs to be executed by an already existing user, who needs to authenticate with
|
|
their own credentials, unless this is the first user created on the system.
|
|
|
|
:return: dict.
|
|
|
|
Format::
|
|
|
|
{
|
|
"user_id": int,
|
|
"username": str,
|
|
"created_at": str (in ISO format)
|
|
}
|
|
|
|
"""
|
|
|
|
if self.user_manager.get_user_count() > 0 and not executing_user and not session_token:
|
|
return None, "You need to authenticate in order to create another user"
|
|
|
|
if not self.user_manager.authenticate_user(executing_user, executing_user_password):
|
|
user, session = self.user_manager.authenticate_user_session(session_token)
|
|
if not user:
|
|
return None, "Invalid credentials and/or session_token"
|
|
|
|
try:
|
|
user = self.user_manager.create_user(username, password, **kwargs)
|
|
except (NameError, ValueError) as e:
|
|
return None, str(e)
|
|
|
|
return {
|
|
'user_id': user.user_id,
|
|
'username': user.username,
|
|
'created_at': user.created_at.isoformat(),
|
|
}
|
|
|
|
@action
|
|
def authenticate_user(self, username, password):
|
|
"""
|
|
Authenticate a user
|
|
:return: True if the provided username and password are correct, False otherwise
|
|
"""
|
|
|
|
return self.user_manager.authenticate_user(username, password)
|
|
|
|
@action
|
|
def update_password(self, username, old_password, new_password):
|
|
"""
|
|
Update the password of a user
|
|
:return: True if the password was successfully updated, false otherwise
|
|
"""
|
|
|
|
return self.user_manager.update_password(username, old_password, new_password)
|
|
|
|
@action
|
|
def delete_user(self, username, executing_user=None, executing_user_password=None, session_token=None):
|
|
"""
|
|
Delete a user
|
|
"""
|
|
|
|
if not self.user_manager.authenticate_user(executing_user, executing_user_password):
|
|
user, session = self.user_manager.authenticate_user_session(session_token)
|
|
if not user:
|
|
return None, "Invalid credentials and/or session_token"
|
|
|
|
try:
|
|
return self.user_manager.delete_user(username)
|
|
except NameError:
|
|
return None, "No such user: {}".format(username)
|
|
|
|
@action
|
|
def create_session(self, username, password, expires_at=None):
|
|
"""
|
|
Create a user session
|
|
:return: dict::
|
|
|
|
{
|
|
"session_token": str,
|
|
"user_id": int,
|
|
"created_at": str (in ISO format),
|
|
"expires_at": str (in ISO format),
|
|
}
|
|
|
|
"""
|
|
|
|
session = self.user_manager.create_user_session(username=username,
|
|
password=password,
|
|
expires_at=expires_at)
|
|
|
|
if not session:
|
|
return None, "Invalid credentials"
|
|
|
|
return {
|
|
'session_token': session.session_token,
|
|
'user_id': session.user_id,
|
|
'created_at': session.created_at.isoformat(),
|
|
'expires_at': session.expires_at.isoformat() if session.expires_at else None,
|
|
}
|
|
|
|
@action
|
|
def authenticate_session(self, session_token):
|
|
"""
|
|
Authenticate a session by token and return the associated user
|
|
:return: dict.
|
|
|
|
Format::
|
|
|
|
{
|
|
"user_id": int,
|
|
"username": str,
|
|
"created_at": str (in ISO format)
|
|
}
|
|
|
|
"""
|
|
|
|
user, session = self.user_manager.authenticate_user_session(session_token=session_token)
|
|
if not user:
|
|
return None, 'Invalid session token'
|
|
|
|
return {
|
|
'user_id': user.user_id,
|
|
'username': user.username,
|
|
'created_at': user.created_at.isoformat(),
|
|
}
|
|
|
|
@action
|
|
def delete_session(self, session_token):
|
|
"""
|
|
Delete a user session
|
|
"""
|
|
|
|
return self.user_manager.delete_user_session(session_token)
|
|
|
|
|
|
# vim:sw=4:ts=4:et:
|