manager

User and Group Management Module

This module provides two primary classes - UserManager and GroupManager - to handle authentication, authorization, and role‑based access control in a lightweight file‑based persistence system. User data, including hashed passwords, roles, and lock‑out state, is stored in a JSON file, while group membership and metadata are kept in a separate JSON file. The module is designed for small projects or prototyping, offering straightforward APIs for creating users, managing passwords, locking out accounts, and organizing users into groups with associated permissions.

Key Features

• Password hashing with Argon2 and secure verification.
• Account lock‑out after configurable failed login attempts.
• User activation/deactivation and role assignment.
• Group management: create, delete, add/remove users, and query memberships.
• Atomic JSON persistence to avoid data corruption.
• Minimal dependencies - relies only on the standard library plus argon2.

Usage Examples

from pathlib import Path from plantdb.commons.auth.manager import UserManager, GroupManager, Role

Initialize managers (will create JSON files if absent)

um = UserManager(users_file=Path('users.json')) gm = GroupManager(groups_file=Path('groups.json'))

Create a new user

um.create('alice', 'Alice Smith', 'secure123', roles={Role.ADMIN})

Validate credentials

assert um.validate('alice', 'secure123')

Create a group and add the user

group = gm.create_group('admins', creator='alice', users={'alice'}, description='Admin group')

Check membership

assert 'admins' in [g.name for g in gm.get_user_groups('alice')]

GroupExistsError

Bases: Exception

Raised when attempting to create a group that already exists.

GroupManager

GroupManager(groups_file='groups.json')

Bases: object

Manages groups for the RBAC system.

This class handles the creation, modification, and persistence of user groups. Groups are stored in a JSON file and loaded/saved as needed.

Attributes:

Name	Type	Description
groups_file	str | Path	The path to the JSON file where groups are stored.
groups	Dict[str, Group]	Dictionary mapping group names to Group objects.

Initialize the GroupManager.

Parameters:

Name	Type	Description	Default
groups_file	str | Path	The path to the JSON file for storing groups. Defaults to "groups.json".	'groups.json'

Source code in plantdb/commons/auth/manager.py

def __init__(self, groups_file: str | Path = "groups.json"):
    """
    Initialize the GroupManager.

    Parameters
    ----------
    groups_file : str | Path, optional
        The path to the JSON file for storing groups. Defaults to ``"groups.json"``.
    """
    self.groups_file = Path(groups_file)
    self.groups: Dict[str, Group] = {}
    self.logger = get_logger(__class__.__name__)
    self._load_groups()

add_user_to_group

add_user_to_group(group_name, username)

Add a user to a group.

Parameters:

Name	Type	Description	Default
group_name	str	The name of the group.	required
username	str	The username to add to the group.	required

Returns:

Type	Description
bool	True if the user was added, False if the group doesn't exist or the user was already in the group.

Source code in plantdb/commons/auth/manager.py

def add_user_to_group(self, group_name: str, username: str) -> bool:
    """Add a user to a group.

    Parameters
    ----------
    group_name : str
        The name of the group.
    username : str
        The username to add to the group.

    Returns
    -------
    bool
        ``True`` if the user was added, ``False`` if the group doesn't exist or the user was already in the group.
    """
    group = self.get_group(group_name)
    if not group:
        return False

    result = group.add_user(username)
    if result:
        self._save_groups()
    return result

create_group

create_group(name, creator, users=None, description=None)

Create a new group.

Parameters:

Name	Type	Description	Default
name	str	The unique name for the group.	required
creator	str	The username of the user creating the group.	required
users	Optional[Set[str]]	Initial set of users to add to the group. Creator is automatically added.	None
description	Optional[str]	Optional description of the group.	None

Returns:

Type	Description
Group	The created group object.

Raises:

Type	Description
ValueError	If a group with the same name already exists.

Source code in plantdb/commons/auth/manager.py

def create_group(self, name: str, creator: str, users: Optional[Set[str]] = None,
                 description: Optional[str] = None) -> Group:
    """Create a new group.

    Parameters
    ----------
    name : str
        The unique name for the group.
    creator : str
        The username of the user creating the group.
    users : Optional[Set[str]]
        Initial set of users to add to the group. Creator is automatically added.
    description : Optional[str]
        Optional description of the group.

    Returns
    -------
    Group
        The created group object.

    Raises
    ------
    ValueError
        If a group with the same name already exists.
    """
    if name in self.groups:
        raise ValueError(f"Group '{name}' already exists")

    # Initialize users set and ensure the creator is included
    if users is None:
        users = set()
    users.add(creator)

    group = Group(
        name=name,
        users=users,
        description=description,
        created_at=datetime.now(),
        created_by=creator
    )

    self.groups[name] = group
    self._save_groups()
    return group

delete_group

delete_group(name)

Delete a group.

Parameters:

Name	Type	Description	Default
name	str	The name of the group to delete.	required

Returns:

Type	Description
bool	True if the group was deleted, False if it didn't exist.

Source code in plantdb/commons/auth/manager.py

def delete_group(self, name: str) -> bool:
    """Delete a group.

    Parameters
    ----------
    name : str
        The name of the group to delete.

    Returns
    -------
    bool
        ``True`` if the group was deleted, ``False`` if it didn't exist.
    """
    if name not in self.groups:
        return False

    del self.groups[name]
    self._save_groups()
    return True

get_group

get_group(name)

Get a group by name.

Parameters:

Name	Type	Description	Default
name	str	The name of the group to retrieve.	required

Returns:

Type	Description
Optional[Group]	The group object if it exists, None otherwise.

Source code in plantdb/commons/auth/manager.py

def get_group(self, name: str) -> Optional[Group]:
    """Get a group by name.

    Parameters
    ----------
    name : str
        The name of the group to retrieve.

    Returns
    -------
    Optional[Group]
        The group object if it exists, ``None`` otherwise.
    """
    return self.groups.get(name)

get_user_groups

get_user_groups(username)

Get all groups that a user belongs to.

Parameters:

Name	Type	Description	Default
username	str	The username to search for.	required

Returns:

Type	Description
List[Group]	A list of Group objects that the user belongs to.

Source code in plantdb/commons/auth/manager.py

def get_user_groups(self, username: str) -> List[Group]:
    """Get all groups that a user belongs to.

    Parameters
    ----------
    username : str
        The username to search for.

    Returns
    -------
    List[Group]
        A list of Group objects that the user belongs to.
    """
    user_groups = []
    for group in self.groups.values():
        if group.has_user(username):
            user_groups.append(group)
    return user_groups

group_exists

group_exists(name)

Check if a group exists.

Parameters:

Name	Type	Description	Default
name	str	The name of the group to check.	required

Returns:

Type	Description
bool	True if the group exists, False otherwise.

Source code in plantdb/commons/auth/manager.py

def group_exists(self, name: str) -> bool:
    """Check if a group exists.

    Parameters
    ----------
    name : str
        The name of the group to check.

    Returns
    -------
    bool
        True if the group exists, False otherwise.
    """
    return name in self.groups

list_groups

list_groups()

Get a list of all groups.

Returns:

Type	Description
List[Group]	A list of all Group objects.

Source code in plantdb/commons/auth/manager.py

def list_groups(self) -> List[Group]:
    """Get a list of all groups.

    Returns
    -------
    List[Group]
        A list of all Group objects.
    """
    return list(self.groups.values())

remove_user_from_group

remove_user_from_group(group_name, username)

Remove a user from a group.

Parameters:

Name	Type	Description	Default
group_name	str	The name of the group.	required
username	str	The username to remove from the group.	required

Returns:

Type	Description
bool	True if the user was removed, False if the group doesn't exist or the user wasn't in the group.

Source code in plantdb/commons/auth/manager.py

def remove_user_from_group(self, group_name: str, username: str) -> bool:
    """Remove a user from a group.

    Parameters
    ----------
    group_name : str
        The name of the group.
    username : str
        The username to remove from the group.

    Returns
    -------
    bool
        ``True`` if the user was removed, ``False`` if the group doesn't exist or the user wasn't in the group.
    """
    group = self.get_group(group_name)
    if not group:
        return False

    result = group.remove_user(username)
    if result:
        self._save_groups()
    return result

UserExistsError

Bases: Exception

Raised when attempting to create a user that already exists.

UserManager

UserManager(users_file='users.json', max_login_attempts=3, lockout_duration=900)

Bases: object

UserManager class for managing user data.

The UserManager class provides methods to create, load, save, and manage users. It uses a JSON file to persist user data. It includes functionalities like password hashing and verification, user account lockout management, and handling of guest accounts.

Attributes:

Name	Type	Description
users_file	str	Path to the JSON file where user data is stored.
users	Dict[str, User]	Dictionary containing all users, indexed by username.
GUEST_USERNAME	str	Default username for guest accounts.
GUEST_PASSWORD	str	Default password for guest accounts.

Examples:

>>> from pathlib import Path
>>> from tempfile import gettempdir
>>> from plantdb.commons.auth.manager import UserManager
>>> manager = UserManager(users_file=Path(gettempdir())/'users.json')
>>> manager.validate_user_password(manager.GUEST_USERNAME, manager.GUEST_PASSWORD)
True
>>> manager.validate(manager.GUEST_USERNAME, manager.GUEST_PASSWORD)
True
>>> manager.validate('gest', manager.GUEST_PASSWORD)
False
>>> manager.deactivate(manager.get_user(manager.GUEST_USERNAME))
>>> manager.validate(manager.GUEST_USERNAME, manager.GUEST_PASSWORD)
WARNING  [UserManager] Account guest is not active.
False

User manager constructor.

Parameters:

Name	Type	Description	Default
users_file	str or Path	The path to the JSON file where user data is stored. Defaults to "users.json".	'users.json'
max_login_attempts	int	Maximum number of failed login attempts before account lockout. Defaults to 3.	3
lockout_duration	int or timedelta	Account lockout duration in seconds. Defaults to 900 (15 minutes).	900

Source code in plantdb/commons/auth/manager.py

def __init__(self, users_file: str | Path = 'users.json', max_login_attempts=3, lockout_duration=900):
    """User manager constructor.

    Parameters
    ----------
    users_file : str or pathlib.Path, optional
        The path to the JSON file where user data is stored. Defaults to ``"users.json"``.
    max_login_attempts : int, optional
        Maximum number of failed login attempts before account lockout. Defaults to ``3``.
    lockout_duration : int or timedelta, optional
        Account lockout duration in seconds. Defaults to ``900`` (15 minutes).
    """
    self.ADMIN_USERNAME = "admin"
    self.GUEST_USERNAME = "guest"
    self.GUEST_PASSWORD = "guest"
    self.users_file = Path(users_file)
    self.users: Dict[str, User] = {}
    self.logger = get_logger(__class__.__name__)
    self._load_users()
    self._ensure_guest_user()
    self._ensure_admin_user()

    self.max_login_attempts = max_login_attempts
    self.lockout_duration = timedelta(seconds=lockout_duration) if isinstance(lockout_duration,
                                                                              int) else lockout_duration

activate

activate(user)

Activates a user.

Parameters:

Name	Type	Description	Default
user	User	The user to activate.	required

Source code in plantdb/commons/auth/manager.py

def activate(self, user: User) -> None:
    """Activates a user.

    Parameters
    ----------
    user : plantdb.commons.auth.User
        The user to activate.
    """
    # Verify if the User exists
    try:
        assert self.exists(user.username)
    except AssertionError:
        self.logger.error(f"User '{user.username}' does not exists!")
        return

    user.is_active = True
    self._save_users()
    return

create

create(username, fullname, password, roles=None)

Create a new user and store the user information in a file.

Parameters:

Name	Type	Description	Default
username	str	The username of the user to create. This will be converted to lowercase.	required
fullname	str	The full name of the user to create.	required
password	str	The password of the user to create.	required
roles	Role or Set[Role]	The roles to associate with the user. If None (default), use the Role.READER role.	None

Examples:

>>> from plantdb.commons.auth.manager import Role, UserManager
>>> manager = UserManager()
>>> manager.create('batman', "Bruce Wayne", "joker", Role.ADMIN)
>>> print(manager.users)
batman

Source code in plantdb/commons/auth/manager.py

def create(self, username: str, fullname: str, password: str, roles: Union[Role, Set[Role]] = None) -> None:
    """Create a new user and store the user information in a file.

    Parameters
    ----------
    username : str
        The username of the user to create.
        This will be converted to lowercase.
    fullname : str
        The full name of the user to create.
    password : str
        The password of the user to create.
    roles : plantdb.commons.auth.models.Role or Set[plantdb.commons.auth.models.Role], optional
        The roles to associate with the user.
        If ``None`` (default), use the `Role.READER` role.

    Examples
    --------
    >>> from plantdb.commons.auth.manager import Role, UserManager
    >>> manager = UserManager()
    >>> manager.create('batman', "Bruce Wayne", "joker", Role.ADMIN)
    >>> print(manager.users)
    batman
    """
    username = username.lower()  # Convert the username to lowercase to maintain uniformity.
    timestamp = datetime.now()  # Get the current timestamp for tracking user creation time.

    # Verify if the username is available
    try:
        assert not self.exists(username)
    except AssertionError:
        self.logger.error(f"User '{username}' already exists!")
        raise UserExistsError(f"User '{username}' already exists")

    # Convert to a list:
    if isinstance(roles, Role):
        roles = {roles}

    # Add the new user's data to the `self.users` dictionary.
    self.users[username] = User(
        username=username,
        fullname=fullname,
        password_hash=self._hash_password(password),
        roles=roles or {Role.READER},
        created_at=timestamp,
        password_last_change=timestamp,
    )
    # Save all user data (including the newly created user) to the 'users.json' file.
    self._save_users()
    self.logger.debug(f"Created user '{username}' with fullname '{fullname}'.")
    if not username == self.GUEST_USERNAME:
        self.logger.info(f"Welcome {fullname}, please log in...'")
    return

deactivate

deactivate(user)

Deactivates a user.

Parameters:

Name	Type	Description	Default
user	User	The user to deactivate.	required

Source code in plantdb/commons/auth/manager.py

def deactivate(self, user: User) -> None:
    """Deactivates a user.

    Parameters
    ----------
    user : plantdb.commons.auth.User
        The user to deactivate.
    """
    # Verify if the User exists
    try:
        assert self.exists(user.username)
    except AssertionError:
        self.logger.error(f"User '{user.username}' does not exists!")
        return

    user.is_active = False
    self._save_users()
    return

exists

exists(username)

Check if a user exists.

Parameters:

Name	Type	Description	Default
username	str	The name of the user to check for existence.	required

Returns:

Type	Description
bool	True if the user exists, otherwise False.

Notes

This function is case-sensitive. For case-insensitive checks, convert both the username and the keys to the lower or upper case before comparison.

Source code in plantdb/commons/auth/manager.py

def exists(self, username: str) -> bool:
    """Check if a user exists.

    Parameters
    ----------
    username : str
        The name of the user to check for existence.

    Returns
    -------
    bool
        `True` if the user exists, otherwise `False`.

    Notes
    -----
    This function is case-sensitive. For case-insensitive checks, convert both
    the username and the keys to the lower or upper case before comparison.
    """
    return username in self.users

get_token_user

get_token_user(token_payload)

Retrieve a User object based on the provided username.

Parameters:

Name	Type	Description	Default
token_payload	dict	The token containing the identifier for the user to be retrieved and special permissions. Must be a token with the type 'api'.	required

Returns:

Type	Description
TokenUser | None	An instance of the User class representing the requested user. If it does not exist, None is returned.

Raises:

Type	Description
TypeError	If the provided token payload does not correspond to an api token.

Source code in plantdb/commons/auth/manager.py

def get_token_user(self, token_payload: dict) -> TokenUser | None:
    """Retrieve a User object based on the provided username.

    Parameters
    ----------
    token_payload : dict
        The token containing the identifier for the user to be retrieved and special permissions.
        Must be a token with the type `'api'`.

    Returns
    -------
    TokenUser | None
        An instance of the ``User`` class representing the requested user.
        If it does not exist, ``None`` is returned.

    Raises
    ------
    TypeError
        If the provided token payload does not correspond to an api token.
    """
    if token_payload["type"] != "api":
        self.logger.error("Provided token is not an api token!")
        raise TypeError("Provided token is not an api token!")
    username = token_payload.get('username')
    if username not in self.users:
        self.logger.error(f"User '{username}' does not exist!")
        return None
    user = self.users[username]

    self.logger.debug(f"Creating TokenUser for user '{username}'...")
    self.logger.debug(f"Dataset permission requested: {token_payload.get('datasets')}")
    return TokenUser(**user.to_dict(), dataset_permissions=token_payload.get("datasets"))

get_user

get_user(username)

Retrieve a User object based on the provided username.

Parameters:

Name	Type	Description	Default
username	str	The unique identifier for the user to be retrieved.	required

Returns:

Type	Description
User | None	An instance of the User class representing the requested user. If it does not exist, None is returned.

Source code in plantdb/commons/auth/manager.py

def get_user(self, username: str) -> User | None:
    """Retrieve a User object based on the provided username.

    Parameters
    ----------
    username : str
        The unique identifier for the user to be retrieved.

    Returns
    -------
    plantdb.commons.auth.models.User | None
        An instance of the ``User`` class representing the requested user.
        If it does not exist, ``None`` is returned.
    """
    if not self.exists(username):
        self.logger.error(f"User '{username}' does not exist!")
        return None
    return self.users[username]

get_user_from_decoded_token

get_user_from_decoded_token(token_payload)

Extract a user representation from a decoded JWT payload.

Parameters:

Name	Type	Description	Default
token_payload	dict	The decoded JWT payload as a dict. It must contain a "type" entry and, for "access" or "refresh" tokens, a "username" entry.	required

Returns:

Type	Description
User | TokenUser | None	User: when token_type is "access" or "refresh". TokenUser: when token_type is "api". None: when token_type is not recognized; an error is logged.

Source code in plantdb/commons/auth/manager.py

def get_user_from_decoded_token(self, token_payload: dict) -> User | TokenUser | None:
    """Extract a user representation from a decoded JWT payload.

    Parameters
    ----------
    token_payload
        The decoded JWT payload as a ``dict``. It must contain a ``"type"`` entry and,
        for ``"access"`` or ``"refresh"`` tokens, a ``"username"`` entry.

    Returns
    -------
    User | TokenUser | None
        * ``User``: when ``token_type`` is ``"access"`` or ``"refresh"``.
        * ``TokenUser``: when ``token_type`` is ``"api"``.
        * ``None``: when ``token_type`` is not recognized; an error is logged.
    """
    token_type = token_payload["type"]
    if token_type == "api":
        return self.get_token_user(token_payload)
    elif token_type in ["access", "refresh"]:
        return self.get_user(token_payload["username"])
    else:
        self.logger.error(f"Token type '{token_type}' does not exist!")
        return None

is_active

is_active(username)

Check whether a user account is active.

Parameters:

Name	Type	Description	Default
username	str	The name of the user account to check for activity status.	required

Returns:

Type	Description
bool	True if the user account is active, False otherwise.

Source code in plantdb/commons/auth/manager.py

def is_active(self, username: str) -> bool:
    """Check whether a user account is active.

    Parameters
    ----------
    username : str
        The name of the user account to check for activity status.

    Returns
    -------
    bool
        ``True`` if the user account is active, ``False`` otherwise.
    """
    user = self.get_user(username)
    if not user.is_active:
        self.logger.warning(f"Account {username} is not active.")
    return user.is_active

is_locked_out

is_locked_out(username)

Check if an account is locked.

Parameters:

Name	Type	Description	Default
username	str	The username of the account to check.	required

Returns:

Type	Description
bool	True if the account is locked, False otherwise.

Source code in plantdb/commons/auth/manager.py

def is_locked_out(self, username: str) -> bool:
    """Check if an account is locked.

    Parameters
    ----------
    username : str
        The username of the account to check.

    Returns
    -------
    bool
        ``True`` if the account is locked, ``False`` otherwise.
    """
    user = self.get_user(username)
    is_locked = user._is_locked_out()
    if is_locked:
        self.logger.info(f"Account {username} is locked, try logging in after {user.locked_until}.")
    return is_locked

unlock_user

unlock_user(user)

Unlock a specified user.

Parameters:

Name	Type	Description	Default
user	User	The user to unlock.	required

Source code in plantdb/commons/auth/manager.py

def unlock_user(self, user: User) -> None:
    """Unlock a specified user.

    Parameters
    ----------
    user : plantdb.commons.auth.User
        The user to unlock.
    """
    user.locked_until = None
    self._save_users()
    return

update_password

update_password(username, password, new_password)

Update the password of an existing user.

Parameters:

Name	Type	Description	Default
username	str	The name of the user whose password is to be updated.	required
password	str	The current password of the user to verify their identity.	required
new_password	str	The new password to set for the user. If None or empty, no change will occur.	required

Source code in plantdb/commons/auth/manager.py

def update_password(self, username: str, password: str, new_password: str) -> None:
    """Update the password of an existing user.

    Parameters
    ----------
    username : str
        The name of the user whose password is to be updated.
    password : str
        The current password of the user to verify their identity.
    new_password : str
        The new password to set for the user. If ``None`` or empty, no change will occur.
    """
    # Verify if the login exists
    try:
        assert self.exists(username)
    except AssertionError:
        self.logger.error(f"User '{username}' does not exists!")
        return

    if not self.validate(username, password):
        return

    user = self.get_user(username)
    timestamp = datetime.now()  # Get the current timestamp for tracking user creation time.
    if new_password:
        user.password_hash = self._hash_password(new_password)
        user.password_last_change = timestamp

    self._save_users()
    self.logger.info(f"Password updated for user '{username}'...")
    return

validate

validate(username, password)

Validate the user credentials.

Parameters:

Name	Type	Description	Default
username	str	The username provided by the user attempting to log in.	required
password	str	The password provided by the user attempting to log in.	required

Returns:

Type	Description
bool	True if the login attempt is successful, False otherwise.

Source code in plantdb/commons/auth/manager.py

def validate(self, username: str, password: str) -> bool:
    """Validate the user credentials.

    Parameters
    ----------
    username : str
        The username provided by the user attempting to log in.
    password : str
        The password provided by the user attempting to log in.

    Returns
    -------
    bool
        ``True`` if the login attempt is successful, ``False`` otherwise.
    """
    if not self.exists(username):
        return False
    if not self.is_active(username):
        return False
    if self.is_locked_out(username):
        return False

    user = self.get_user(username)
    if self.validate_user_password(username, password):
        # Reset failed attempts on successful login
        user.failed_attempts = 0
        user.last_login = datetime.now()
        self._save_users()
        return True
    else:
        # Record a failed attemp
        self._record_failed_attempt(username, self.max_login_attempts, self.lockout_duration)
        self.logger.error(f"Invalid credentials for user '{username}'")
        return False

validate_user_password

validate_user_password(username, password)

Validate a user's password.

This function checks if the provided plaintext password matches the hashed password stored for the given username.

Parameters:

Name	Type	Description	Default
username	str	The username of the user.	required
password	str	The plain-text password to validate.	required

Returns:

Type	Description
bool	True if the password is valid, False otherwise.

Source code in plantdb/commons/auth/manager.py

def validate_user_password(self, username: str, password: str) -> bool:
    """Validate a user's password.

    This function checks if the provided plaintext password matches the hashed password stored for the given username.

    Parameters
    ----------
    username : str
        The username of the user.
    password : str
        The plain-text password to validate.

    Returns
    -------
    bool
        ``True`` if the password is valid, ``False`` otherwise.
    """
    pw_hash = self.get_user(username).password_hash
    try:
        # Verify password, raises an exception if wrong.
        ph.verify(pw_hash, password)
    except Exception as e:
        self.logger.error(f"Failed to verify password for {username}: {e}")
        return False
    else:
        return True