Users Package

About this project

Project Description

This Python module is designed to simplify user management in Telegram Bots by providing necessary functionality such as: authentication, authorization and request limitation, providing efficient management of user attributes and access rights.

Interaction Model 1: Using a Unified Entrypoint (Method: user_access_check())

sequenceDiagram
    participant Bot
    participant Users-Module
    Bot->>Users-Module: Create Users-Module instance with rate limits
    Bot->>Users-Module: Call user_access_check(user_id, role_id)
    Users-Module-->>Bot: Return access, permissions, and rate limits

Loading

Key Features

• This module is designed primarily for Telegram bots but can be adapted for various projects that require user management, role-based access control, and request rate limiting.

• This module requires certain dependencies related to

  • Vault
    • Vault Server for storing user configurations and historical data
    • Additional Module to interact with the Vault API
    • Vault Policy with access rights to the Vault Server
  • PostgreSQL
    • PostgreSQL Server for storing user data and historical records
    • PostgreSQL Schema for creating tables in the database

Table of Contents

• Description of module Constants
• Description of module Exceptions
• Users class
• RateLimiter class
• Storage class
• Token Authentication
• Structure of configuration in Vault
• Structure of historical data in PostgreSQL
• Additional usage example
• Installing

Description of module Constants

This module contains constant values

Constant Name	Description	Default Value
USERS_VAULT_CONFIG_PATH	Path for configuration data in Vault.	"configuration/users"
USER_STATUS_ALLOW	User access status for allowed access.	"allowed"
USER_STATUS_DENY	User access status for denied access.	"denied"

Description of module Exceptions

Exception	Describe	Tips
WrongUserConfiguration	Raised when user configuration is wrong.	Please, see the configuration example
VaultInstanceNotSet	Raised when the Vault instance is not set.	Please, see documentation
FailedDeterminateRateLimit	Raised when the rate limit cannot be determined.	Please, check misconfiguration between configuration and users requests in PostgreSQL
StorageInstanceNotSet	Raised when the storage instance (PostgreSQL) is not set.	Please, see documentation
FailedStorageConnection	Raised when the connection to the storage (PostgreSQL) failed.	Please, check the connection to the PostgreSQL server

Users class

Class Initialization

The Users class provides authentication, authorization, user attribute management and user request logging for Telegram bots. You can initialize it with different options

• vault (any): Configuration for initializing the Vault client.

  • (object): an already initialized instance of VaultClient for interacting with the Vault API.
  • (dict): extended configuration for VaultClient (for database engine).
    • instance (VaultClient): An already initialized instance for interacting with the Vault API.
    • role (str): The role name for the Vault database engine.

• rate_limits (bool): Enable rate limit functionality.

• storage_connection (any): Connection object to connect to the storage. Do not use if you are using Vault database engine.

• Examples:

  • Initialize with VaultClient and without rate_limits:

    users_without_ratelimits = Users(vault=<VaultClient>, rate_limits=False, storage_connection=psycopg2.connect(**db_config))

  • Initialize with VaultClient and with rate_limits:

    users_with_ratelimits = Users(vault=<VaultClient>, storage_connection=psycopg2.connect(**db_config))

  • Initialize with Vault configuration dictionary (for using the vault database engine):

    vault_config = {'instance': <VaultClient>, 'role': 'my_db_role'}
    users_with_dict_vault = Users(vault=vault_config)

decorator: Access Control

The access_control() decorator is used to control access to specific functions based on user roles and permissions.
Required the pyTelegramBotAPI objects: telegram.telegram_types.Message or telegram.telegram_types.CallbackQuery

• Arguments:

  • role_id (str): Required role ID for the specified user ID.
  • flow (str): The flow of the function, which can be either.
    • auth for authentication. Default value.
    • authz for authorization.

• Examples:
  Role-based access control

    @telegram.message_handler(commands=['start'])
    @access_control(role_id='admin_role', flow='authz')
    # Decorator returns user information about access, permissions, and rate limits into access_result argument
    def my_function(message: telegram.telegram_types.Message, access_result: dict = None):
        print(f"User permissions: {access_result}")
        pass

  Just authentication

    @telegram.message_handler(commands=['start'])
    @access_control()
    # Decorator returns user information about access, permissions, and rate limits into access_result argument
    def my_function(message: telegram.telegram_types.Message, access_result: dict = None):
        print(f"User permissions: {access_result}")
        pass

• Returns:

  • Breaks the function and returns an error message if the user does not have the required role or permission.

method: User Access Check

The user_access_check() method is the main entry point for authentication, authorization, and request rate limit verification. It is used to control the request rate (limits) for a specific user.

• Arguments:

  • user_id (str): Required user ID.
  • role_id (str): Required role ID for the specified user ID.

• Keyword Arguments:

  • chat_id (str): Required chat ID for the specified user ID. Additional context for logging.
  • message_id (str): Required message ID for the specified user ID. Additional context for logging.

• Examples:

  user_access_check(user_id='user1', role_id='admin_role', chat_id='chat1', message_id='msg1')

• Returns:

  • A dictionary with access status, permissions, and rate limit information.

    {
      'access': self.user_status_allow / self.user_status_deny,
      'permissions': self.user_status_allow / self.user_status_deny,
      'rate_limits': '2023-08-06 11:47:09.440933' / None
    }

Description of class attributes

Data Type	Attribute	Purpose	Default Value
object	vault	Vault instance for interacting with the Vault API.	None
dict	storage	Configuration for initializing the storage client.	None
bool	rate_limits	Enable request rate limit functionality.	True
str	user_status_allow	User access status: allowed.	"allowed"
str	user_status_deny	User access status: denied.	"denied"
str	vault_config_path	The prefix of the configuration path in the Vault.	"configuration/users"

RateLimiter class

Class Initialization

The RateLimiter class provides restriction functionality for user requests to the Telegram bot in the context of a specific user.

• vault (VaultClient): An already initialized instance for interacting with the Vault API or a configuration dictionary for initializing a VaultClient instance in this class.

• storage (Storage): An already initialized instance for interacting with the storage (PostgreSQL) or a configuration dictionary for initializing a Storage instance in this class.

• user_id (str): User ID for checking speed limits.

• Examples:

  limiter = RateLimiter(vault=<VaultClient>, storage=storage_client, user_id='User1')

method: Rate Limit Determination

The determine_rate_limit() method is the main entry point for checking bot request limits for the specified user. It returns information about whether the request rate limits are active and when they expire

• Examples:

  determine_rate_limit()

• Returns:

  • String with a timestamp of the end of restrictions on requests or None if rate limit is not applied.

    ("2023-08-07 10:39:00.000000" | None)

method: Get User Requests Counters

The get_user_request_counters() method calculates the number of requests made by the user and returns the number of requests per day and per hour.

• Examples:

  get_user_request_counters()

• Returns:

  • A dictionary with the number of requests per day and per hour.

    {
      'requests_per_day': 9,
      'requests_per_hour': 1
    }

Description of Class Attributes

Data Type	Attribute	Purpose	Default Value
VaultClient	vault	Vault instance for interacting with the Vault API.	None
Storage	storage	Storage instance for interacting with the storage (PostgreSQL).	None
str	user_id	User ID for checking speed limits.	None
str	vault_config_path	The prefix of the configuration path in the Vault.	"configuration/users"
dict	requests_configuration	User request limits configuration.	None
dict	requests_counters	Counters for the number of requests per day and per hour.	None

Storage class

Class Initialization

The storage class for the storage of user data: requests, access logs, etc in the PostgreSQL database.
Only one of the parameters is required for initialization: db_connection or vault.

• db_connection (object): The database connection object for interacting with the PostgreSQL database (psycopg2).

• vault (dict): Configuration for initializing the Vault client.

  • instance (VaultClient): An already initialized instance for interacting with the Vault API.
  • role (str): The role name for the Vault database engine.

• Examples:

  storage = Storage(db_connection=psycopg2.connect(**db_config))

  storage = Storage(vault={'instance': <VaultClient>, 'role': 'my_db_role'})

method: Create connection to the PostgreSQL database

The create_connection() method creates a connection to the PostgreSQL database.

• Examples:

create_connection()

• Returns:
  • Connection object to the PostgreSQL database.

method: Register of User

The register_user() method registers a new user in the database.

• Arguments:

  • user_id (str): User ID for registration.
  • chat_id (str): Chat ID for registration.
  • status (str): The user state in the system. Values: self.user_status_allow or self.user_status_deny.

• Examples:

  register_user(user_id='user1', chat_id='chat1', status='allowed')

method: Log user request

The log_user_request() method logs the user request in the database.

• Arguments:

  • user_id (str): User ID for logging.
  • request (dict): The user request details.

• Examples:

  log_user_request(user_id='user1', request={'chat_id': 'chat1', 'message_id': 'msg1'})

method: Get user requests

The get_user_requests() method retrieves the user's requests from the database.

• Arguments:

  • user_id (str): User ID for retrieving requests.
  • limit (int): The number of requests to retrieve.
  • order (str): The order of the requests. Values: asc or desc.

• Examples:

  get_user_requests(user_id='user1', limit=10, order='asc')

• Returns:

  • A list of user requests [(id, timestamp, rate_limits), ...].

method: Get users

The get_users() method retrieves all users from the database.

• Arguments:

  • only_allowed (bool): Retrieve only allowed users.

• Examples:

  get_users()

• Returns:

  • A list of users [{'user_id': '12345', 'chat_id': '67890', 'status': 'denied'}, ...].

Token Authentication

Overview

Starting from v4.2.0, the Users package supports generic token-based authentication for frontend integration (web UIs, mobile apps, CLI tools). This feature enables temporary access without storing user credentials.

Use Cases

• Database Bridge Pattern: Telegram bot issues tokens, frontend validates them through Users module using shared PostgreSQL database
• Restricted Bot Access: Alternative authentication when Telegram auth widget is unavailable (e.g., bot with closed external access)
• Web Authentication: Secure temporary access for web interfaces
• API Access: Short-lived tokens for API clients
• Mobile Apps: Authenticate mobile app users without password storage

Token Format

Tokens follow the format user_id.token_id:

• Example: "123456.a8f3kjs9dfjkl23jrlksjdf..."
• Total length: ~45-50 characters
• Cryptographically secure using PBKDF2 with salt

Available Methods

issue_token(user_id, ttl_minutes=10)

Generate a temporary access token for a user.

Arguments:

• user_id (str): User ID to issue token for
• ttl_minutes (int): Token validity period in minutes (default: 10)

Returns: Token string in format "user_id.token_id"

Example:

token = users.issue_token(user_id='user1', ttl_minutes=15)
print(token)  # "user1.a8f3kjs9dfjkl23jrlksjdf..."

validate_token(token)

Validate a token and return user information.

Arguments:

• token (str): Token string in format "user_id.token_id"

Returns:

• dict: User info {'user_id': str, 'status': str, 'roles': list} if valid
• None: If token is invalid, expired, or already used

Example:

user_info = users.validate_token(token=token)
if user_info:
    print(f"User: {user_info['user_id']}")
    print(f"Status: {user_info['status']}")
    print(f"Roles: {user_info['roles']}")

Note: Tokens are single-use and automatically marked as used after validation.

revoke_token(user_id)

Revoke all existing tokens for a user.

Arguments:

• user_id (str): User ID to revoke tokens for

Example:

users.revoke_token(user_id='user1')

Security Features

• PBKDF2 Hashing: Tokens hashed with 100,000 iterations
• Salt: Unique 32-byte salt per token
• Single-Use: Tokens automatically invalidated after first use
• Auto-Revocation: New token issuance revokes previous tokens
• Expiration: Configurable TTL with automatic cleanup

Database Schema

Tokens stored in users_tokens table. See full schema in tables.sql.

Backward Compatibility

All token methods gracefully handle missing users_tokens table, logging warnings without raising exceptions. Existing deployments continue working without schema updates.

Structure of configuration in Vault

This project uses a Vault server with the KV2 engine and Database Engine for storing user configurations and database connection data. It supports user configurations to define system access rights, roles, and request restrictions.

Users Configuration

• path to the secret: configuration/users/{user_id}

• keys and Values:

  • status: The status of user access, which can be either
    • self.user_status_allow
    • self.user_status_deny
  • roles: A list of roles associated with the user ID, e.g., ["role1", "role2"].
  • requests: Limits on the number of requests
    • requests_per_day
    • requests_per_hour
    • random_shift_time (additional, random shift in minutes from 0 to the specified number) in minutes

• example of a secret with configuration:

{
  "status": "allowed",
  "roles": ["admin_role", "additional_role"],
  "requests": {
    "requests_per_day": 10,
    "requests_per_hour": 1,
    "random_shift_minutes": 15
  }
}

Database Configuration

• path to the secret: configuration/database

• keys and values with simple database connection:

  • host: The host of the PostgreSQL server.
  • port: The port of the PostgreSQL server.
  • database: The name of the PostgreSQL database.
  • user: The username for the PostgreSQL database.
  • password: The password for the PostgreSQL database.

  {
    "host": "localhost",
    "port": 5432,
    "dbname": "mydatabase",
    "user": "myuser",
    "password": "mypassword",
  }

• keys and values with Vault Database Engine:

  • role: The role name for the Vault database engine.
  • instance: The instance of the VaultClient for interacting with the Vault API.

  {
    "host": "localhost",
    "port": 5432,
    "dbname": "mydatabase",
  }

Structure of historical data in PostgreSQL

This project uses a PostgreSQL database to store historical data about user requests and access events. It supports user request logging to track user activity and access rights. The detailed table schema can be found in this sql file.

Users Requests Table

Contains records of user requests, access permission, access level, and apply limits on the number of requests.

Users Table

Contains records of user metadata for the Telegram bot, such as user ID, chat ID, and message ID.

Additional usage example

Example 1 - With Rate Limits

# import modules
from vault import VaultClient
from users import Users

# create the vault client
vault_client = VaultClient(
  url='http://0.0.0.0:8200',
  namespace='my_project',
  auth={
      'type': 'approle',
      'role_id': 'my_role',
      'secret_id': 'my_secret_id'
  }
)

# create the Users instance of the class with rate limits and get user information
users = Users(vault=<VaultClient>, rate_limits=True, storage_connection=psycopg2.connect(**db_config))
user_info = users.user_access_check(user_id=message.chat.id, role_id="admin_role", chat_id=message.chat.id, message_id=message.message_id)

# check permissions, roles, and rate limits
if user_info["access"] == users.user_status_allow:
    print("Hi, you can use the bot!")
    if user_info["permissions"] == users.user_status_allow:
        if user_info["rate_limits"]:
            print(f"You have sent too many requests, the limit is applied until {user_info['rate_limits']}")
        else:
            print("You have admin's rights")
    else:
        print("You do not have access rights to this function")
else:
    print("Access denied, goodbye!")

Example 2 - Without Rate Limits

# import modules
from vault import VaultClient
from users import Users

# create the vault client
vault_client = VaultClient(
  url='http://vault.example.com',
  namespace='my_project',
  auth={
      'type': 'approle',
      'role_id': 'my_role',
      'secret_id': 'my_secret_id'
  }
)

# create the Users instance of the class without rate limits and get user information
users = Users(vault=<VaultClient>, storage_connection=psycopg2.connect(**db_config))
user_info = users.user_access_check(user_id=message.chat.id, role_id="admin_role", chat_id=message.chat.id, message_id=message.message_id)

# check permissions and roles
if user_info["access"] == users.user_status_allow:
    print("Hi, you can use the bot!")
    if user_info["permissions"] == users.user_status_allow:
        print("You have admin's rights")
    else:
        print("You do not have access rights to this function")
else:
    print("Access denied, goodbye!")

Example 3 - Decorator Usage

# import modules
from vault import VaultClient
from users import Users

# create the vault client
vault_client = VaultClient(
  url='http://vault.example.com',
  namespace='my_project',
  auth={
      'type': 'approle',
      'role_id': 'my_role',
      'secret_id':
  }
)

# create the Users instance of the class with rate limits
users = Users(vault=<VaultClient>, rate_limits=True, storage_connection=psycopg2.connect(**db_config))

# create a function with the access_control decorator
@telegram.message_handler(commands=['start'])
@access_control()
# Decorator returns user information about access, permissions, and rate limits into access_result argument
def my_function(message: telegram.telegram_types.Message, access_result: dict = None):
    print(f"User permissions: {access_result}")
    pass

# call the function
my_function(message)

Example 4 - Token Authentication (v4.2.0+)

# import modules
from vault import VaultClient
from users import Users
import psycopg2

# create the vault client and users instance
vault_client = VaultClient(
  url='http://vault.example.com',
  namespace='my_project',
  auth={
      'type': 'approle',
      'role_id': 'my_role',
      'secret_id': 'my_secret_id'
  }
)

users = Users(vault=vault_client, storage_connection=psycopg2.connect(**db_config))

# Issue a temporary access token for a user (e.g., from Telegram bot)
token = users.issue_token(user_id='user1', ttl_minutes=15)
print(f"Token: {token}")  # Returns: "user1.a8f3kjs9dfjkl23jrlksjdf..."

# Validate the token (e.g., in web frontend or API)
user_info = users.validate_token(token=token)
if user_info:
    print(f"User authenticated: {user_info['user_id']}")
    print(f"Status: {user_info['status']}")
    print(f"Roles: {user_info['roles']}")
else:
    print("Invalid or expired token")

# Revoke all tokens for a user
users.revoke_token(user_id='user1')

Installing

tee -a pyproject.toml <<EOF
[tool.poetry]
name = myproject"
version = "1.0.0"
description = ""

[tool.poetry.dependencies]
python = "^3.12"
users = { git = "https://github.com/obervinov/users-package.git", tag = "v4.1.3" }

[build-system]
requires = ["poetry-core"]
build-backend = "poetry.core.masonry.api"
EOF

poetry install

GitHub Actions

Name	Version
GitHub Actions Templates	v2.1.1