auth

Authentication Service Layer

Provides pure-Python business-logic functions for user registration, login, logout, token validation, and token refresh. The functions are independent of the Flask-RESTful API layer and raise domain-specific exceptions, making them reusable across different interfaces and straightforward to unit-test.

Key Features

• Register new users with required-field validation.
• Check whether a username already exists.
• Authenticate credentials and obtain JWT access and refresh tokens.
• Invalidate a session (logout) and return the logged-out username.
• Validate JWT tokens and retrieve basic user information.
• Refresh access tokens using a refresh token.
• Consistent, domain-specific exception hierarchy (MissingFieldError, InvalidCredentialsError, TokenError, etc.).

Usage Examples

>>> from plantdb.server.services.auth import register_user, logout_user, authenticate_user
>>> from plantdb.commons.test_database import test_database
>>> # Initialize the database (creates base directory if needed)
>>> db = test_database(no_auth=True)
>>> db.connect()
>>> register_user(db, {"username": "alice", "fullname": "Alice", "password": "secret"})
>>> logout_user(db, db.session_manager._admin_token)
>>> access_token = authenticate_user(db, "alice", "secret")
>>> print(access_token)
phQk4Uez7Gim3SkY6oBzrCBvos2eHS_wU-_ZvvoPCQI

InvalidCredentialsError

Bases: ValueError

Raised when supplied credentials are not valid.

Raised by:

• Reference API server server services auth  authenticate_user

MissingFieldError

Bases: ValueError

Raised when a required field is missing from the request payload.

Raised by:

• Reference API server server services auth  register_user

TokenError

Bases: ValueError

Raised for generic token-related problems (validation / refresh).

authenticate_user

authenticate_user(db, username, password)

Authenticate credentials and return (access_token, refresh_token).

Raises:

Type	Description
InvalidCredentialsError	If the DB login method returns None.

Source code in plantdb/server/services/auth.py

def authenticate_user(db: Any, username: str, password: str) -> Tuple[str, str]:
    """Authenticate credentials and return ``(access_token, refresh_token)``.

    Raises
    ------
    InvalidCredentialsError
        If the DB ``login`` method returns ``None``.
    """
    tokens = db.login(username, password)
    if not tokens:
        raise InvalidCredentialsError("Invalid credentials")
    return tokens  # type: ignore[return-value]

check_username_exists

check_username_exists(db, username)

Return True if username exists in the database.

The underlying DB exposes rbac_manager.users.exists.

Source code in plantdb/server/services/auth.py

def check_username_exists(db: Any, username: str) -> bool:
    """Return ``True`` if ``username`` exists in the database.

    The underlying DB exposes ``rbac_manager.users.exists``.
    """
    return db.rbac_manager.users.exists(username)

logout_user

logout_user(db, token, logger=None)

Invalidate a session and return the username that was logged out.

Parameters:

Name	Type	Description	Default
db	FSDB	Database instance exposing logout.	required
token	str	JWT token extracted from the request (passed via **kwargs).	required
logger	Logger | None	Optional logger.	None

Returns:

Type	Description
str	Username of the user whose session was terminated.

Raises:

Type	Description
NoAuthUserError	If logout fails for any reason.

Source code in plantdb/server/services/auth.py

def logout_user(db: Any, token: str, logger: logging.Logger | None = None) -> str:
    """Invalidate a session and return the username that was logged out.

    Parameters
    ----------
    db : plantdb.commons.fsdb.core.FSDB
        Database instance exposing ``logout``.
    token: str
        JWT token extracted from the request (passed via ``**kwargs``).
    logger: logging.Logger | None
        Optional logger.

    Returns
    -------
    str
        Username of the user whose session was terminated.

    Raises
    ------
    NoAuthUserError
        If logout fails for any reason.
    """
    log = _get_logger(logger.name if logger else __name__)
    try:
        success, username = db.logout(token=token)
        if not success:
            raise NoAuthUserError("Logout failed")
        log.info("Logout successful for user %s", username)
        return username
    except SessionValidationError as exc:
        raise NoAuthUserError(str(exc)) from exc

refresh_token

refresh_token(db, refresh_token)

Refresh an access token using a valid refresh token.

Returns a tuple (new_access_token, new_refresh_token). Raises: TokenError: If the refresh token is invalid or the refresh operation fails.

Source code in plantdb/server/services/auth.py

def refresh_token(db: Any, refresh_token: str) -> Tuple[str, str]:
    """Refresh an access token using a valid refresh token.

    Returns a tuple ``(new_access_token, new_refresh_token)``.
    Raises:
        TokenError: If the refresh token is invalid or the refresh operation fails.
    """
    try:
        tokens = db.session_manager.refresh_session(refresh_token)
        if not tokens:
            raise TokenError("Invalid or expired refresh token")
        return tokens  # type: ignore[return-value]
    except Exception as exc:
        raise TokenError(str(exc)) from exc

register_user

register_user(db, payload, **kwargs)

Create a new user record in the supplied database.

Parameters:

Name	Type	Description	Default
db	FSDB	An FSDB instance that implements a create_user method.	required
payload	Dict[str, Any]	Mapping containing the keys username, fullname and password. The values are expected to be strings, but the function does not enforce a specific type beyond the presence of the keys.	required

Other Parameters:

Name	Type	Description
token	str	The token for authentication.
logger	Logger	Optional logger used for diagnostic messages. If None (default), a module-level logger is obtained via _get_logger.

Raises:

Type	Description
MissingFieldError	If any of the required keys (username, fullname, password) are absent from payload.
UserAlreadyExistsError	If the underlying db.create_user raises an exception that indicates a duplicate user or any other database-level problem.
SessionValidationError	Propagated unchanged when the database signals that the operation is invalid for the current session (e.g., unauthorized).

Source code in plantdb/server/services/auth.py

def register_user(db: Any, payload: Dict[str, Any], **kwargs) -> None:
    """Create a new user record in the supplied database.

    Parameters
    ----------
    db : plantdb.commons.fsdb.core.FSDB
        An FSDB instance that implements a ``create_user`` method.
    payload : Dict[str, Any]
        Mapping containing the keys ``username``, ``fullname`` and ``password``.
        The values are expected to be strings, but the function does not enforce
        a specific type beyond the presence of the keys.

    Other Parameters
    ----------------
    token : str
        The token for authentication.
    logger : logging.Logger
        Optional logger used for diagnostic messages.
        If ``None`` (default), a module-level logger is obtained via ``_get_logger``.

    Raises
    ------
    MissingFieldError
        If any of the required keys (``username``, ``fullname``, ``password``) are absent from ``payload``.
    UserAlreadyExistsError
        If the underlying ``db.create_user`` raises an exception that indicates
        a duplicate user or any other database-level problem.
    SessionValidationError
        Propagated unchanged when the database signals that the operation is
        invalid for the current session (e.g., unauthorized).
    """
    log = _get_logger(kwargs.get('logger'))
    required = {"username", "fullname", "password"}
    missing = required - set(payload.keys())
    if missing:
        raise MissingFieldError(f"Missing required fields: {', '.join(sorted(missing))}")

    try:
        db.create_user(
            new_username=payload["username"],
            fullname=payload["fullname"],
            password=payload["password"],
            token=kwargs.get("token"),
        )
        log.info("User '%s' successfully created", payload["username"])
    except UserExistsError as exc:
        # Keep original semantics - caller will translate to 401
        raise exc
    except SessionValidationError as exc:
        # Keep original semantics - caller will translate to 401
        raise exc
    except Exception as exc:
        # Log the unexpected error and propagate it unchanged
        log.error("Unexpected error while creating user: %s", exc)
        raise  # propagates the original exception (preserves traceback)

validate_token

validate_token(db, token)

Validate a JWT and return basic user information.

Returns a mapping with username and fullname. Raises: TokenError: If validation fails.

Source code in plantdb/server/services/auth.py

def validate_token(db: Any, token: str) -> Mapping[str, Any]:
    """Validate a JWT and return basic user information.

    Returns a mapping with ``username`` and ``fullname``.
    Raises:
        TokenError: If validation fails.
    """
    try:
        user = db.get_user_data(token=token)
        return {"username": user.username, "fullname": user.fullname}
    except Exception as exc:
        raise TokenError(str(exc)) from exc