auth

Authentication service layer.

This module contains pure‑Python functions that implement the business logic for user registration, login, logout, token validation and token refresh. The functions are deliberately small (atomic) and raise domain‑specific exceptions instead of returning HTTP responses. Flask‑RESTful resources in plantdb.server.api.auth act as thin adapters that translate these exceptions into proper HTTP status codes and JSON payloads.

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