Project Description

This Python module is designed to simplify user management in Telegram bots, providing the necessary functionality for tasks such as user authentication, authorization and compliance with speed limits, ensuring effective management of user attributes and access rights.

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

This module contains constant values

The Users class provides authentication, authorization and management of user attributes for Telegram bots. You can initialize it using various options

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

  • (VaultClient): an already initialized instance for interacting with the Vault API.
  • (dict): configuration for initializing a VaultClient instance in this class.

• rate_limits (bool): Enable rate limit functionality.

• Examples:

  • Initialize with VaultClient and without rate_limits:

    users_without_ratelimits = Users(vault=vault_client, rate_limits=False)

  • Initialize with VaultClient and with rate_limits:

    users_with_ratelimits = Users(vault=vault_client)

  • Initialize with Vault configuration dictionary:

    vault_config = {
      "name": "my_project",
      "url": "https://vault.example.com",
      "approle": {
          "id": "my_approle",
          "secret-id": "my_secret"
      }
    }
    users_with_dict_vault = Users(vault=vault_config)

For more details and examples, see the class docstring.

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.

• Examples:

  user_access_check(user_id='user1', role_id='admin_role')

• 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': {
          'end_time': '2023-08-06 11:47:09.440933'
      }
    }

For more details, see the method docstring.

The authentication method checks if the specified user ID has access to the bot.

• Arguments:

  • user_id (str): Required user ID.

• Example:

  authentication(user_id='user1')

• Returns:

  • self.user_status_allow if access is granted.
  • self.user_status_deny if access is denied.

The authorization method checks whether the user has the specified role.

• Arguments:

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

• Example:

  authorization(user_id='user1', role_id='admin_role')

• Returns:

  • self.user_status_allow if the user has the role.
  • self.user_status_deny if the user does not have the role.

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

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

  • (VaultClient): an already initialized instance for interacting with the Vault API.
  • (dict): configuration for initializing a VaultClient instance in this class.

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

• Examples:

  • Initialize with a VaultClient instance:

    limiter = RateLimiter(vault=vault_client, user_id='User1')

  • Initialize with a Vault configuration dictionary:

    vault_config = {
      "name": "my_project",
      "url": "https://vault.example.com",
      "approle": {
          "id": "my_approle",
          "secret-id": "my_secret"
      }
    }
    limiter = RateLimiter(vault=vault_config, user_id='User1')

For more details and examples, see the class docstring.

The determine_rate_limit method is the main entry point for checking restrictions on requests to the bot for the specified user. It returns information about whether the request rate limits are active.

• Arguments:

  • None

• Examples:

  determine_rate_limit()

• Returns:

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

    (dict | None)
    {"end_time": "2023-08-07 10:39:00.000000"}

For more details, see the method docstring.

The _active_rate_limit method is used for a situation when restrictions on requests are already applied and the user ID has a timestamp of the end of restrictions. The method checks whether it is time to reset the request rate limit.

• Arguments:

  • None

• Example:

  _active_rate_limit()

• Returns:

  • A dictionary with a timestamp - if it has not expired yet and the restrictions on requests continue, or None - if the time for limiting requests has expired and you need to reset the restrictions.

      (dict | None)
      {"end_time": "2023-08-07 10:39:00.000000"}

The _apply_rate_limit method is used if the request limit counters is full and it is necessary to apply request limits for the specified user.

• Arguments:

  • None

• Example:

  _apply_rate_limit()

• Returns:

  • Dictionary with the end time of restrictions on requests for the specified user.

      (dict | None)
      {"end_time": "2023-08-07 10:39:00.000000"}

The _update_requests_counters method recalculates the request counters based on the historical user data (per_hour and per_day).

• Arguments:

  • None

• Example:

  _update_requests_counters()

• Returns:

  • None

The _update_requests_history method added current request timestamp to the requests history list in the Vault.

• Arguments:

  • None

• Example:

  _update_requests_history()

• Returns:

  • None

This project uses a Vault server with the KV2 engine to store and retrieve configuration data. It supports user configurations to define system access rights, roles, and request restrictions.

• 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

    {
      "requests_per_day": 10,
      "requests_per_hour": 1,
      "random_shift_minutes": 15
    }

• 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
  }
}

• path to the secret: data/users/{user_id}
• keys and values:

  • requests_counters: Historical data with statistics on user requests. It includes counters for the number of requests

    • requests_per_day
    • requests_per_hour

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

  • requests_ratelimits: Information about rate limits, including the

    • end_time of the rate limit

    {
      "end_time": "2023-08-07 10:39:00.000000"
    }

    or

    {
      "end_time": None
    }

  • requests_history: Historical data with timestamps of user requests

    • list of timestamps of user requests

    [
      "2023-08-07 10:39:00.000000",
      "2023-08-07 10:40:00.000000",
      "2023-08-06 10:00:00.000000"
    ]

  • authorization: Details about the authorization process, including the time, status

    • timestamp
    • self.user_status_allow or self.user_status_deny
    • role ID

    {
      "time": "2023-08-07 10:39:00.000000",
      "status": "allowed",
      "role": "role1"
    }

  • authentication: Records of the authentication process, indicating the time and status

    • timestamp
    • self.user_status_allow or self.user_status_deny

    {
      "time": "2023-08-07 10:39:00.000000",
      "status": "allowed"
    }

• example of a secret with historical data:

  "requests_counters": {
    "requests_per_day": 9,
    "requests_per_hour": 1
  }

  "requests_ratelimits": {"end_time": "2023-08-07 10:39:00.000000"}

  "requests_history": [
    "2023-08-07 10:39:00.000000",
    "2023-08-07 10:40:00.000000",
    "2023-08-06 10:00:00.000000"
  ]

  "authorization": {
    "time": "2023-08-07 10:39:00.000000",
    "status": "allowed",
    "role": "role1"
  }

  "authentication": {
    "time": "2023-08-07 10:39:00.000000",
    "status": "allowed"
  }

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

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

Interaction Model 2: Using Separate Methods for Authentication, Authorization, and Rate Limits

sequenceDiagram
    participant User
    participant Users
    User->>Users: Create Users instance with rate limits
    User->>Users: Call authentication(user_id)
    Users-->>User: Return access status
    User->>Users: Call authorization(user_id, role_id)
    Users-->>User: Return permissions status

sequenceDiagram
    participant User
    participant RateLimiter
    User->>RateLimiter: Create RateLimiter instance with user_id
    User->>RateLimiter: Call determine_rate_limit()
    RateLimiter-->>User: Return rate limit timestamp for user_id (or None)

Example 1 - With Entrypoint and 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',
  name='mybot1',
  approle={
      'id': id,
      'secret-id': secret-id
  }
)

# create the Users instance of the class with rate limits
users = Users(vault=vault_client)

# use the main entrypoint
user_info = users.user_access_check(
  user_id=message.chat.id,
  role_id="admin_role"
)
# 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"]["end_time"]:
            print(f"You have sent too many requests, the limit is applied until {user_info['rate_limits']['end_time']}")
        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 - With Entrypoint and Without 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',
  name='mybot1',
  approle={
      'id': id,
      'secret-id': secret-id
  }
)

# create the Users instance of the class without rate limits
users = Users(
  vault=vault_client,
  rate_limits=False
)

# use the main entrypoint
user_info = users.user_access_check(
  user_id=message.chat.id,
  role_id="admin_role"
)
# 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:
        print("You have admin's rights")
    else:
        print("You do not have access rights to this function")

else:
    print("Access denied, goodbye!")

Example 3 - Without Entrypoint:

# import modules
from vault import VaultClient
from users import Users

# create the vault client
vault_client = VaultClient(
  url='http://0.0.0.0:8200',
  name='mybot1',
  approle={
      'id': id,
      'secret-id': secret-id
  }
)

# create the Users instance of the class
users = Users(vault=vault_client)

# check access to the bot
if users.authentication(user_id='user1') == users.user_status_allow:
    print("You can use this bot")

# check user role
if users.authorization(
  user_id='user1',
  role_id='admin_role'
) == users.user_status_allow:
    print("You have admin's permissions")

# check rate limit for user_id
limiter = RateLimiter(
  vault=vault_client,
  user_id='user1'
)
if limiter.determine_rate_limit()['end_time']:
    print(f"You have sent too many requests, the limit is applied until {user_info['rate_limits']['end_time']}")

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

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

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

poetry install