.dockerignore

@@ -1,5 +1,5 @@
 **/__pycache__
-**/venv
+**/.venv
 **/.classpath
 **/.dockerignore
 **/.env
@@ -24,5 +24,4 @@
 **/secrets.dev.yaml
 **/values.dev.yaml
 README.md
-database/
+data/
-.flake8

.flake8

@@ -1,2 +0,0 @@
-[flake8]
-exclude=.git,__pycache__,venv

.gitignore

@@ -161,4 +161,4 @@ cython_debug/
 #.idea/
 
 # Database data
-database/data/
+data/

Dockerfile

@@ -15,9 +15,8 @@ RUN adduser -u 1000 --disabled-password --gecos "" appuser && chown -R appuser /
 RUN apt update && apt full-upgrade -y
 
 # Install pip requirements
-RUN pip install -U pip setuptools wheel
 COPY requirements.txt .
-RUN pip install -r requirements.txt
+RUN python -m pip install -r requirements.txt
 
 COPY . /app
 

compose.yaml

@@ -20,9 +20,7 @@ services:
       - password_manager
 
   db:
-    build:
+    image: stnicolay/mariadb-aria
-      context: ./database/
-      dockerfile: ./Dockerfile
     restart: always
     environment:
       MYSQL_ROOT_PASSWORD: example123!
@@ -30,6 +28,6 @@ services:
       MYSQL_USER: manager
       MYSQL_PASSWORD: passwd123!
     volumes:
-      - ./database/data:/var/lib/mysql
+      - ./data:/var/lib/mysql
     networks:
       - password_manager

database/Dockerfile

@@ -1,3 +0,0 @@
-FROM mariadb
-
-COPY mariadb.cnf /etc/mysql/mariadb.cnf

database/mariadb.cnf

@@ -1,38 +0,0 @@
-# The MariaDB configuration file
-#
-# The MariaDB/MySQL tools read configuration files in the following order:
-# 0. "/etc/mysql/my.cnf" symlinks to this file, reason why all the rest is read.
-# 1. "/etc/mysql/mariadb.cnf" (this file) to set global defaults,
-# 2. "/etc/mysql/conf.d/*.cnf" to set global options.
-# 3. "/etc/mysql/mariadb.conf.d/*.cnf" to set MariaDB-only options.
-# 4. "~/.my.cnf" to set user-specific options.
-#
-# If the same option is defined multiple times, the last one will apply.
-#
-# One can use all long options that the program supports.
-# Run program with --help to get a list of available options and with
-# --print-defaults to see which it would actually understand and use.
-#
-# If you are new to MariaDB, check out https://mariadb.com/kb/en/basic-mariadb-articles/
-
-#
-# This group is read both by the client and the server
-# use it for options that affect everything
-#
-[client-server]
-# Port or socket location where to connect
-# port = 3306
-socket = /run/mysqld/mysqld.sock
-
-# Import all .cnf files from configuration directory
-[mariadbd]
-skip-host-cache
-skip-name-resolve
-
-[mysqld]
-skip-innodb
-default-storage-engine=Aria
-default-tmp-storage-engine=Aria
-
-!includedir /etc/mysql/mariadb.conf.d/
-!includedir /etc/mysql/conf.d/

requirements-dev.txt

@@ -1,3 +1 @@
 black
-flake8
-isort

requirements.txt

@@ -4,4 +4,3 @@ python-dotenv
 pyTelegramBotAPI
 sqlmodel
 pydantic
-aiohttp

src/__init__.py

@@ -1,37 +1,22 @@
-import asyncio
 import os
 
 from dotenv import load_dotenv
+from sqlalchemy.future import Engine
 
-from . import (
+from . import bot, cryptography, database
-    account_checks,
-    account_parsing,
-    bot,
-    db,
-    decrypted_account,
-    encryption,
-    generate_password,
-)
 
-__all__ = [
+__all__ = ["bot", "cryptography", "database"]
-    "account_checks",
+engine: Engine
-    "account_parsing",
-    "bot",
-    "decrypted_account",
-    "encryption",
-    "db",
-    "generate_password",
-]
 
 
 def main() -> None:
     load_dotenv("./.env")
-    engine = db.prepare.get_engine(
+    engine = database.prepare.get_engine(
         host=os.getenv("DB_HOST"),
         user=os.getenv("DB_USER"),
         passwd=os.getenv("DB_PASS"),
         db=os.getenv("DB_NAME"),
-    )
+    )  # type: ignore
-    db.prepare.prepare(engine)
+    database.prepare.prepare(engine)
-    bot_ = bot.create_bot(os.getenv("TG_TOKEN"), engine)
+    bot_ = bot.create_bot(os.getenv("TG_TOKEN"), engine)  # type: ignore
-    asyncio.run(bot_.infinity_polling())
+    bot_.infinity_polling()

src/account_checks.py

@@ -1,50 +0,0 @@
-import string
-
-from .decrypted_account import DecryptedAccount
-
-FORBIDDEN_CHARS = frozenset("`\n\\")
-PUNCTUATION = frozenset(string.punctuation).difference(FORBIDDEN_CHARS)
-
-
-def _base_check(val: str, /) -> bool:
-    "Returns false if finds new lines or backtick (`)"
-    return not any(i in FORBIDDEN_CHARS for i in val)
-
-
-def check_account_name(name: str) -> bool:
-    "Returns true if account name is valid"
-    return _base_check(name)
-
-
-def check_login(login: str) -> bool:
-    "Returns true if login is valid"
-    return _base_check(login)
-
-
-def check_password(passwd: str) -> bool:
-    "Returns true if password is valid"
-    return _base_check(passwd)
-
-
-def check_account(account: DecryptedAccount) -> bool:
-    """Runs checks for account name, login and password"""
-    return all(
-        (
-            check_account_name(account.name),
-            check_login(account.login),
-            check_password(account.password),
-        )
-    )
-
-
-def check_gened_password(passwd: str, /) -> bool:
-    """Retuns true if generated password is valid,
-    false otherwise.
-    Password is valid if there is at least one lowercase character,
-    uppercase character and one punctuation character"""
-    return (
-        any(c.islower() for c in passwd)
-        and any(c.isupper() for c in passwd)
-        and any(c.isdigit() for c in passwd)
-        and any(c in PUNCTUATION for c in passwd)
-    )

src/account_parsing.py

@@ -1,50 +0,0 @@
-import io
-from typing import Iterable, Self
-
-import pydantic
-
-from .decrypted_account import DecryptedAccount
-
-
-class _Account(pydantic.BaseModel):
-    name: str
-    login: str
-    password: str
-
-    def to_usual_account(self, user_id: int) -> DecryptedAccount:
-        return DecryptedAccount(
-            user_id=user_id,
-            name=self.name,
-            login=self.login,
-            password=self.password,
-        )
-
-    @classmethod
-    def from_usual_account(cls, account: DecryptedAccount) -> Self:
-        return cls(
-            name=account.name,
-            login=account.login,
-            password=account.password,
-        )
-
-
-class _Accounts(pydantic.BaseModel):
-    accounts: list[_Account] = pydantic.Field(default_factory=list)
-
-
-def _accounts_list_to_json(accounts: Iterable[DecryptedAccount]) -> str:
-    result = _Accounts(
-        accounts=[_Account.from_usual_account(i) for i in accounts],
-    ).json(ensure_ascii=False, indent=2)
-    return result
-
-
-def json_to_accounts(json_: str, user_id: int) -> list[DecryptedAccount]:
-    accounts = _Accounts.parse_raw(json_)
-    return [account.to_usual_account(user_id) for account in accounts.accounts]
-
-
-def accounts_to_json(accounts: Iterable[DecryptedAccount]) -> io.StringIO:
-    file = io.StringIO(_accounts_list_to_json(accounts))
-    file.name = "passwords.json"
-    return file

src/bot/__init__.py

@@ -1,77 +1,52 @@
 import functools
 
 from sqlalchemy.future import Engine
-from telebot.async_telebot import AsyncTeleBot
+import telebot
 
-from . import callback_handlers, exception_handler, message_handlers
+from . import handlers, utils
 
-__all__ = ["callback_handlers", "exception_handler", "message_handlers"]
+__all__ = ["handlers", "utils"]
 
 
-def create_bot(token: str, engine: Engine) -> AsyncTeleBot:
+def create_bot(token: str, engine: Engine) -> telebot.TeleBot:
-    bot = AsyncTeleBot(token, exception_handler=exception_handler.Handler)
+    bot = telebot.TeleBot(token)
     bot.register_message_handler(
-        functools.partial(message_handlers.set_master_password, bot, engine),
+        functools.partial(handlers.set_master_password, bot, engine),
         commands=["set_master_pass"],
     )
     bot.register_message_handler(
-        functools.partial(message_handlers.get_account, bot, engine),
+        functools.partial(handlers.get_account, bot, engine), commands=["get_account"]
-        commands=["get_account"],
     )
     bot.register_message_handler(
-        functools.partial(message_handlers.get_accounts, bot, engine),
+        functools.partial(handlers.get_accounts, bot, engine), commands=["get_accounts"]
-        commands=["get_accounts"],
     )
     bot.register_message_handler(
-        functools.partial(message_handlers.add_account, bot, engine),
+        functools.partial(handlers.add_account, bot, engine), commands=["add_account"]
-        commands=["add_account"],
     )
     bot.register_message_handler(
-        functools.partial(message_handlers.delete_all, bot, engine),
+        functools.partial(handlers.delete_all, bot, engine), commands=["delete_all"]
-        commands=["delete_all"],
     )
     bot.register_message_handler(
-        functools.partial(message_handlers.reset_master_pass, bot, engine),
+        functools.partial(handlers.reset_master_pass, bot, engine),
         commands=["reset_master_pass"],
     )
     bot.register_message_handler(
-        functools.partial(message_handlers.delete_account, bot, engine),
+        functools.partial(handlers.delete_account, bot, engine),
         commands=["delete_account"],
     )
     bot.register_message_handler(
-        functools.partial(message_handlers.help_command, bot),
+        functools.partial(handlers.help, bot), commands=["help", "start"]
-        commands=["help", "start"],
     )
     bot.register_message_handler(
-        functools.partial(message_handlers.export, bot, engine),
+        functools.partial(handlers.cancel, bot), commands=["cancel"]
-        commands=["export"],
     )
     bot.register_message_handler(
-        functools.partial(message_handlers.import_accounts, bot, engine),
+        functools.partial(handlers.export, bot, engine), commands=["export"]
-        commands=["import"],
     )
     bot.register_message_handler(
-        functools.partial(message_handlers.gen_password, bot),
+        functools.partial(handlers.import_accounts, bot, engine), commands=["import"]
-        commands=["gen_password"],
     )
     bot.register_message_handler(
-        functools.partial(message_handlers.message_handler, bot),
+        functools.partial(handlers.gen_password, bot), commands=["gen_password"]
-        content_types=[
-            "text",
-            "audio",
-            "document",
-            "photo",
-            "sticker",
-            "video",
-            "video_note",
-            "voice",
-            "location",
-            "contact",
-        ],
-    )
-
-    bot.register_callback_query_handler(
-        functools.partial(callback_handlers.delete_message, bot),
-        lambda call: call.data == "DELETE",
     )
     return bot

src/bot/callback_handlers.py

@@ -1,8 +0,0 @@
-from telebot.async_telebot import AsyncTeleBot
-from telebot.types import CallbackQuery as Callback
-
-from . import helper_functions
-
-
-async def delete_message(bot: AsyncTeleBot, call: Callback) -> None:
-    await helper_functions.delete_message(bot, call.message)

src/bot/exception_handler.py

@@ -1,8 +0,0 @@
-import traceback
-from typing import Type
-
-
-class Handler:
-    @staticmethod
-    def handle(exc: Type[BaseException]) -> None:
-        traceback.print_exception(exc)

src/bot/handlers.py

@@ -0,0 +1,487 @@
+import functools
+import gc
+import time
+
+import telebot
+from sqlalchemy.future import Engine
+
+from .. import cryptography, database
+from .utils import (
+    accounts_to_json,
+    base_handler,
+    check_account,
+    check_account_name,
+    check_login,
+    check_passwd,
+    gen_passwd,
+    get_all_accounts,
+    json_to_accounts,
+    send_tmp_message,
+)
+
+Message = telebot.types.Message
+
+
+def get_accounts(
+    bot: telebot.TeleBot, engine: Engine, mes: telebot.types.Message
+) -> None:
+    base_handler(bot, mes)
+    accounts = database.get.get_accounts(engine, mes.from_user.id)
+    if not accounts:
+        return send_tmp_message(bot, mes.chat.id, "У вас нет аккаунтов")
+
+    # Make accounts copyable and escape special chars
+    accounts = [f"`{account}`" for account in accounts]
+    send_tmp_message(
+        bot,
+        mes.chat.id,
+        "Ваши аккаунты:\n"
+        + "\n".join(accounts)
+        + "\nНажмите на название, чтобы скопировать",
+        30,
+    )
+
+
+def delete_all(
+    bot: telebot.TeleBot, engine: Engine, mes: telebot.types.Message
+) -> None:
+    base_handler(bot, mes)
+    bot_mes = bot.send_message(
+        mes.chat.id,
+        "Вы действительно хотите удалить все ваши аккаунты? Это действие нельзя отменить. Отправьте YES для подтверждения",
+    )
+    bot.register_next_step_handler(
+        mes, functools.partial(_delete_all, bot, engine, bot_mes)
+    )
+
+
+def _delete_all(
+    bot: telebot.TeleBot, engine: Engine, prev_mes: Message, mes: Message
+) -> None:
+    base_handler(bot, mes, prev_mes)
+    text = mes.text.strip()
+    if text == "YES":
+        database.delete.purge_accounts(engine, mes.from_user.id)
+        database.delete.delete_master_pass(engine, mes.from_user.id)
+        send_tmp_message(bot, mes.chat.id, "Всё успешно удалено", timeout=10)
+    else:
+        send_tmp_message(bot, mes.chat.id, "Вы отправили не YES, ничего не удалено")
+
+
+def set_master_password(bot: telebot.TeleBot, engine: Engine, mes: Message) -> None:
+    base_handler(bot, mes, None)
+    if database.get.get_master_pass(engine, mes.from_user.id) is not None:
+        return send_tmp_message(bot, mes.chat.id, "Мастер пароль уже существует")
+    bot_mes = bot.send_message(mes.chat.id, "Отправьте мастер пароль")
+    bot.register_next_step_handler(
+        mes, functools.partial(_set_master_pass2, bot, engine, bot_mes)
+    )
+
+
+def _set_master_pass2(
+    bot: telebot.TeleBot, engine: Engine, prev_mes: Message, mes: Message
+) -> None:
+    base_handler(bot, mes, prev_mes)
+    text = mes.text.strip()
+    if text == "/cancel":
+        return send_tmp_message(bot, mes.chat.id, "Успешная отмена")
+
+    hash_pass, master_salt = cryptography.master_pass.encrypt_master_pass(text)
+    database.add.add_master_pass(engine, mes.from_user.id, master_salt, hash_pass)
+    send_tmp_message(bot, mes.chat.id, "Успех")
+    del mes, text
+    gc.collect()
+
+
+def reset_master_pass(bot: telebot.TeleBot, engine: Engine, mes: Message) -> None:
+    base_handler(bot, mes)
+    if database.get.get_master_pass(engine, mes.from_user.id) is None:
+        return send_tmp_message(bot, mes.chat.id, "Мастер пароль не задан")
+    bot_mes = bot.send_message(
+        mes.chat.id,
+        "Отправьте новый мастер пароль, осторожно, все текущие аккаунты будут удалены навсегда",
+    )
+    bot.register_next_step_handler(
+        mes, functools.partial(_reset_master_pass2, bot, engine, bot_mes)
+    )
+
+
+def _reset_master_pass2(
+    bot: telebot.TeleBot, engine: Engine, prev_mes: Message, mes: Message
+) -> None:
+    base_handler(bot, mes, prev_mes)
+    text = mes.text.strip()
+    if text == "/cancel":
+        return send_tmp_message(bot, mes.chat.id, "Успешная отмена")
+
+    hash_, salt = cryptography.master_pass.encrypt_master_pass(text)
+    database.delete.purge_accounts(engine, mes.from_user.id)
+    database.change.change_master_pass(engine, mes.from_user.id, salt, hash_)
+    send_tmp_message(
+        bot, mes.chat.id, "Все ваши аккаунты удалены, а мастер пароль изменён"
+    )
+    del mes, text
+    gc.collect()
+
+
+def add_account(bot: telebot.TeleBot, engine: Engine, mes: Message) -> None:
+    base_handler(bot, mes)
+
+    master_password_from_db = database.get.get_master_pass(engine, mes.from_user.id)
+    if master_password_from_db is None:
+        return send_tmp_message(bot, mes.chat.id, "Нет мастер пароля")
+
+    bot_mes = bot.send_message(mes.chat.id, "Отправьте название аккаунта")
+
+    bot.register_next_step_handler(
+        mes, functools.partial(_add_account2, bot, engine, bot_mes)
+    )
+
+
+def _add_account2(
+    bot: telebot.TeleBot, engine: Engine, prev_mes: Message, mes: Message
+) -> None:
+    base_handler(bot, mes, prev_mes)
+    text = mes.text.strip()
+    if text == "/cancel":
+        return send_tmp_message(bot, mes.chat.id, "Успешная отмена")
+
+    if not check_account_name(text):
+        return send_tmp_message(bot, mes.chat.id, "Не корректное название аккаунта")
+    if text in database.get.get_accounts(engine, mes.from_user.id):
+        return send_tmp_message(
+            bot, mes.chat.id, "Аккаунт с таким именем уже существует"
+        )
+
+    bot_mes = bot.send_message(mes.chat.id, "Отправьте логин")
+    data = {"name": text}
+
+    bot.register_next_step_handler(
+        mes, functools.partial(_add_account3, bot, engine, bot_mes, data)
+    )
+
+
+def _add_account3(
+    bot: telebot.TeleBot,
+    engine: Engine,
+    prev_mes: Message,
+    data: dict[str, str],
+    mes: Message,
+) -> None:
+    base_handler(bot, mes, prev_mes)
+    text = mes.text.strip()
+    if text == "/cancel":
+        return send_tmp_message(bot, mes.chat.id, "Успешная отмена")
+    if not check_login(text):
+        return send_tmp_message(bot, mes.chat.id, "Не корректный логин")
+
+    data["login"] = text
+
+    bot_mes = bot.send_message(mes.chat.id, "Отправьте пароль от аккаунта")
+
+    bot.register_next_step_handler(
+        mes, functools.partial(_add_account4, bot, engine, bot_mes, data)
+    )
+
+
+def _add_account4(
+    bot: telebot.TeleBot,
+    engine: Engine,
+    prev_mes: Message,
+    data: dict[str, str],
+    mes: Message,
+) -> None:
+    base_handler(bot, mes, prev_mes)
+    text = mes.text.strip()
+    if text == "/cancel":
+        return send_tmp_message(bot, mes.chat.id, "Успешная отмена")
+    if not check_passwd(text):
+        return send_tmp_message(bot, mes.chat.id, "Не корректный пароль")
+
+    data["passwd"] = text
+
+    bot_mes = bot.send_message(mes.chat.id, "Отправьте мастер пароль")
+
+    bot.register_next_step_handler(
+        mes, functools.partial(_add_account5, bot, engine, bot_mes, data)
+    )
+
+
+def _add_account5(
+    bot: telebot.TeleBot,
+    engine: Engine,
+    prev_mes: Message,
+    data: dict[str, str],
+    mes: Message,
+) -> None:
+    base_handler(bot, mes, prev_mes)
+    text = mes.text.strip()
+    if text == "/cancel":
+        return send_tmp_message(bot, mes.chat.id, "Успешная отмена")
+
+    salt, hash_ = database.get.get_master_pass(engine, mes.from_user.id)
+    if not cryptography.master_pass.check_master_pass(text, hash_, salt):
+        return send_tmp_message(bot, mes.chat.id, "Не подходит главный пароль")
+
+    name, login, passwd = data["name"], data["login"], data["passwd"]
+
+    enc_login, enc_pass, salt = cryptography.other_accounts.encrypt_account_info(
+        login, passwd, text
+    )
+
+    result = database.add.add_account(
+        engine, mes.from_user.id, name, salt, enc_login, enc_pass
+    )
+
+    send_tmp_message(
+        bot, mes.chat.id, "Успех" if result else "Произошла не предвиденная ошибка"
+    )
+
+    del data, name, login, passwd, enc_login
+
+    gc.collect()
+
+
+def get_account(bot: telebot.TeleBot, engine: Engine, mes: Message) -> None:
+    base_handler(bot, mes)
+    bot_mes = bot.send_message(mes.chat.id, "Отправьте название аккаунта")
+
+    master_pass = database.get.get_master_pass(engine, mes.from_user.id)
+    if master_pass is None:
+        return send_tmp_message(bot, mes.chat.id, "Нет мастер пароля")
+
+    bot.register_next_step_handler(
+        mes, functools.partial(_get_account2, bot, engine, bot_mes)
+    )
+
+
+def _get_account2(
+    bot: telebot.TeleBot, engine: Engine, prev_mes: Message, mes: Message
+) -> None:
+    base_handler(bot, mes, prev_mes)
+    text = mes.text.strip()
+    if text == "/cancel":
+        return send_tmp_message(bot, mes.chat.id, "Успешная отмена")
+
+    if text not in database.get.get_accounts(engine, mes.from_user.id):
+        return send_tmp_message(bot, mes.chat.id, "Нет такого аккаунта")
+
+    bot_mes = bot.send_message(mes.chat.id, "Отправьте мастер пароль")
+    bot.register_next_step_handler(
+        mes, functools.partial(_get_account3, bot, engine, bot_mes, text)
+    )
+
+
+def _get_account3(
+    bot: telebot.TeleBot, engine: Engine, prev_mes: Message, name: str, mes: Message
+) -> None:
+    base_handler(bot, mes, prev_mes)
+    text = mes.text.strip()
+    if text == "/cancel":
+        return send_tmp_message(bot, mes.chat.id, "Успешная отмена")
+
+    master_salt, hash_pass = database.get.get_master_pass(engine, mes.from_user.id)
+
+    if not cryptography.master_pass.check_master_pass(text, hash_pass, master_salt):
+        return send_tmp_message(bot, mes.chat.id, "Не подходит мастер пароль")
+
+    salt, enc_login, enc_pass = database.get.get_account_info(
+        engine, mes.from_user.id, name
+    )
+    login, passwd = cryptography.other_accounts.decrypt_account_info(
+        enc_login, enc_pass, text, salt
+    )
+    send_tmp_message(
+        bot,
+        mes.chat.id,
+        f"Логин:\n`{login}`\nПароль:\n`{passwd}`\nНажмите на логин или пароль, чтобы скопировать",
+        30,
+    )
+
+    del text, mes, passwd, login
+    gc.collect()
+
+
+def delete_account(bot: telebot.TeleBot, engine: Engine, mes: Message) -> None:
+    base_handler(bot, mes)
+
+    master_pass = database.get.get_master_pass(engine, mes.from_user.id)
+    if master_pass is None:
+        return send_tmp_message(bot, mes.chat.id, "Нет мастер пароля")
+
+    bot_mes = bot.send_message(
+        mes.chat.id, "Отправьте название аккаунта, который вы хотите удалить"
+    )
+
+    bot.register_next_step_handler(
+        mes, functools.partial(_delete_account2, bot, engine, bot_mes)
+    )
+
+
+def _delete_account2(
+    bot: telebot.TeleBot, engine: Engine, prev_mes: Message, mes: Message
+) -> None:
+    base_handler(bot, mes, prev_mes)
+    text = mes.text.strip()
+    if text == "/cancel":
+        return send_tmp_message(bot, mes.chat.id, "Успешная отмена")
+
+    if text not in database.get.get_accounts(engine, mes.from_user.id):
+        return send_tmp_message(bot, mes.chat.id, "Нет такого аккаунта")
+
+    database.delete.delete_account(engine, mes.from_user.id, text)
+    send_tmp_message(bot, mes.chat.id, "Аккаунт удалён")
+
+
+def help(bot: telebot.TeleBot, mes: telebot.types.Message) -> None:
+    message = """Команды:
+/set_master_pass - установить мастер пароль
+/add_account - создать аккаунт
+/get_accounts - получить список аккаунтов
+/get_account - получить логин и пароль аккаунта
+/delete_account - удалить аккаунт
+/delete_all - удалить все аккаунты и мастер пароль
+/reset_master_pass - удалить все аккаунты и изменить мастер пароль
+/cancel - отмена текущего действия
+/help - помощь
+/export - получить пароли в json формате
+/import - импортировать пароли из json в файле в таком же формате, как из /export
+/gen_password - создать 10 надёжных паролей"""
+    bot.send_message(mes.chat.id, message)
+
+
+def cancel(bot: telebot.TeleBot, mes: Message) -> None:
+    send_tmp_message(bot, mes.chat.id, "Нет активного действия")
+
+
+def export(bot: telebot.TeleBot, engine: Engine, mes: Message) -> None:
+    base_handler(bot, mes)
+    master_password_from_db = database.get.get_master_pass(engine, mes.from_user.id)
+
+    if master_password_from_db is None:
+        return send_tmp_message(bot, mes.chat.id, "Нет мастер пароля")
+
+    if not database.get.get_accounts(engine, mes.from_user.id):
+        return send_tmp_message(bot, mes.chat.id, "Нет аккаунтов")
+
+    bot_mes = bot.send_message(mes.chat.id, "Отправьте мастер пароль")
+
+    bot.register_next_step_handler(
+        mes, functools.partial(_export2, bot, engine, bot_mes)
+    )
+
+
+def _export2(
+    bot: telebot.TeleBot, engine: Engine, prev_mes: Message, mes: Message
+) -> None:
+    base_handler(bot, mes, prev_mes)
+    text = mes.text.strip()
+    if text == "/cancel":
+        return send_tmp_message(bot, mes.chat.id, "Успешная отмена")
+
+    master_salt, hash_pass = database.get.get_master_pass(engine, mes.from_user.id)
+    if not cryptography.master_pass.check_master_pass(text, hash_pass, master_salt):
+        return send_tmp_message(bot, mes.chat.id, "Не подходит мастер пароль")
+
+    accounts = get_all_accounts(engine, mes.from_user.id, text)
+    json_io = accounts_to_json(accounts)
+    bot_mes = bot.send_document(mes.chat.id, json_io)
+
+    del text, accounts, json_io
+    gc.collect()
+    time.sleep(30)
+    bot.delete_message(bot_mes.chat.id, bot_mes.id)
+
+
+def import_accounts(bot: telebot.TeleBot, engine: Engine, mes: Message) -> None:
+    base_handler(bot, mes)
+    master_password_from_db = database.get.get_master_pass(engine, mes.from_user.id)
+
+    if master_password_from_db is None:
+        return send_tmp_message(bot, mes.chat.id, "Нет мастер пароля")
+
+    bot_mes = bot.send_message(mes.chat.id, "Отправьте json файл")
+
+    bot.register_next_step_handler(
+        mes, functools.partial(_import2, bot, engine, bot_mes)
+    )
+
+
+def _import2(
+    bot: telebot.TeleBot, engine: Engine, prev_mes: Message, mes: Message
+) -> None:
+    base_handler(bot, mes, prev_mes)
+    if mes.text is not None:
+        text = mes.text.strip()
+        if text == "/cancel":
+            return send_tmp_message(bot, mes.chat.id, "Успешная отмена")
+
+    if mes.document is None:
+        return send_tmp_message(bot, mes.chat.id, "Вы должны отправить документ")
+    if mes.document.file_size > 102_400:  # If file size is bigger that 100 MB
+        return send_tmp_message(bot, mes.chat.id, "Файл слишком большой")
+
+    file_info = bot.get_file(mes.document.file_id)
+    downloaded_file = bot.download_file(file_info.file_path)
+    try:
+        accounts = json_to_accounts(downloaded_file.decode("utf-8"))
+    except Exception:
+        return send_tmp_message(bot, mes.chat.id, "Ошибка во время работы с файлом")
+
+    bot_mes = bot.send_message(mes.chat.id, "Отправьте мастер пароль")
+    bot.register_next_step_handler(
+        mes, functools.partial(_import3, bot, engine, bot_mes, accounts)
+    )
+
+
+def _import3(
+    bot: telebot.TeleBot,
+    engine: Engine,
+    prev_mes: Message,
+    accounts: list[tuple[str, str, str]],
+    mes: Message,
+) -> None:
+    base_handler(bot, mes, prev_mes)
+    text = mes.text.strip()
+    if text == "/cancel":
+        return send_tmp_message(bot, mes.chat.id, "Успешная отмена")
+
+    master_salt, hash_pass = database.get.get_master_pass(engine, mes.from_user.id)
+    if not cryptography.master_pass.check_master_pass(text, hash_pass, master_salt):
+        return send_tmp_message(bot, mes.chat.id, "Не подходит мастер пароль")
+
+    # List of names of accounts, which failed to be added to the database or failed tests
+    failed: list[str] = []
+    for account in accounts:
+        name, login, passwd = account
+        if not check_account(name, login, passwd):
+            failed.append(name)
+            continue
+        enc_login, enc_passwd, salt = cryptography.other_accounts.encrypt_account_info(
+            login, passwd, text
+        )
+        result = database.add.add_account(
+            engine, mes.from_user.id, name, salt, enc_login, enc_passwd
+        )
+        if not result:
+            failed.append(name)
+
+    if failed:
+        mes_text = "Не удалось добавить:\n" + "\n".join(failed)
+    else:
+        mes_text = "Успех"
+    send_tmp_message(bot, mes.chat.id, mes_text, 10)
+    del text, mes, accounts
+    gc.collect()
+
+
+def gen_password(bot: telebot.TeleBot, mes: Message) -> None:
+    # Generate 10 passwords and put 'em in the backticks
+    base_handler(bot, mes)
+    passwords = (f"`{gen_passwd()}`" for _ in range(10))
+    text = (
+        "Пароли:\n"
+        + "\n".join(passwords)
+        + "\nНажмите на пароль, чтобы его скопировать"
+    )
+    send_tmp_message(bot, mes.chat.id, text, 15)

src/bot/helper_functions.py

@@ -1,87 +0,0 @@
-import asyncio
-from typing import Any, Awaitable, Callable
-
-import telebot
-from telebot.async_telebot import AsyncTeleBot
-
-from . import markups
-
-Message = telebot.types.Message
-Handler = Callable[[Message], Awaitable[Any]]
-Markups = (
-    telebot.types.ReplyKeyboardMarkup
-    | telebot.types.InlineKeyboardMarkup
-    | telebot.types.ReplyKeyboardRemove
-)
-states: dict[tuple[int, int], Handler] = {}
-
-
-def register_state(
-    message: Message,
-    handler: Handler,
-) -> None:
-    states[(message.chat.id, message.from_user.id)] = handler
-
-
-def reset_state(message: Message) -> None:
-    try:
-        del states[(message.chat.id, message.from_user.id)]
-    except KeyError:
-        pass
-
-
-def get_state(message: Message) -> Handler | None:
-    return states.get((message.chat.id, message.from_user.id))
-
-
-async def delete_message(
-    bot: AsyncTeleBot,
-    mes: Message,
-    *,
-    sleep_time: int = 0,
-) -> bool:
-    await asyncio.sleep(sleep_time)
-    try:
-        await bot.delete_message(mes.chat.id, mes.id)
-    except telebot.apihelper.ApiException:
-        return False
-    else:
-        return True
-
-
-async def send_tmp_message(
-    bot: AsyncTeleBot,
-    chat_id: telebot.types.Message,
-    text: str,
-    *,
-    sleep_time: int = 5,
-    markup: Markups | None = None,
-) -> None:
-    bot_mes = await bot.send_message(
-        chat_id, text, parse_mode="MarkdownV2", reply_markup=markup
-    )
-    await delete_message(bot, bot_mes, sleep_time=sleep_time)
-
-
-async def base_handler(
-    bot: AsyncTeleBot, mes: Message, prev_mes: Message | None = None
-) -> None:
-    reset_state(mes)
-    await delete_message(bot, mes)
-    if prev_mes is not None:
-        await delete_message(bot, prev_mes)
-
-
-async def send_deleteable_message(
-    bot: AsyncTeleBot,
-    chat_id: int,
-    text: str,
-) -> None:
-    """Sends a message with a delete button"""
-    markup = markups.deletion_markup()
-    await bot.send_message(
-        chat_id,
-        text,
-        parse_mode="MarkdownV2",
-        reply_markup=markup,
-    )

src/bot/markups.py

@@ -1,18 +0,0 @@
-from telebot.types import (
-    InlineKeyboardButton,
-    InlineKeyboardMarkup,
-    ReplyKeyboardMarkup,
-)
-
-
-def deletion_markup() -> InlineKeyboardMarkup:
-    markup = InlineKeyboardMarkup(row_width=5)
-    button = InlineKeyboardButton("Удалить сообщение", callback_data="DELETE")
-    markup.add(button)
-    return markup
-
-
-def account_markup(account_names: list[str]) -> ReplyKeyboardMarkup:
-    markup = ReplyKeyboardMarkup(resize_keyboard=True, row_width=2)
-    markup.add(*account_names)
-    return markup

src/bot/message_handlers.py

@@ -1,761 +0,0 @@
-import asyncio
-import functools
-import gc
-import itertools
-from concurrent.futures import ProcessPoolExecutor
-
-import telebot
-from sqlalchemy.future import Engine
-from telebot.async_telebot import AsyncTeleBot
-
-from .. import db, encryption, generate_password
-from ..account_checks import (
-    check_account,
-    check_account_name,
-    check_login,
-    check_password,
-)
-from ..account_parsing import accounts_to_json, json_to_accounts
-from ..decrypted_account import DecryptedAccount
-from . import markups
-from .helper_functions import (
-    base_handler,
-    delete_message,
-    get_state,
-    register_state,
-    send_deleteable_message,
-    send_tmp_message,
-)
-
-Message = telebot.types.Message
-
-
-async def get_accounts(
-    bot: AsyncTeleBot,
-    engine: Engine,
-    mes: Message,
-) -> None:
-    await base_handler(bot, mes)
-    accounts = db.get.get_account_names(
-        engine,
-        mes.from_user.id,
-        to_sort=True,
-    )
-    if not accounts:
-        return await send_tmp_message(bot, mes.chat.id, "У вас нет аккаунтов")
-
-    # Make accounts copyable and escape special chars
-    accounts = [f"`{account}`" for account in accounts]
-    await send_deleteable_message(
-        bot,
-        mes.chat.id,
-        "Ваши аккаунты:\n"
-        + "\n".join(accounts)
-        + "\nНажмите на название, чтобы скопировать\n"
-        f"Всего {len(accounts)}",
-    )
-
-
-async def delete_all(bot: AsyncTeleBot, engine: Engine, mes: Message) -> None:
-    await base_handler(bot, mes)
-    master_pass = db.get.get_master_pass(engine, mes.from_user.id)
-    if master_pass is None:
-        await send_tmp_message(bot, mes.chat.id, "У вас нет мастер пароля")
-        return
-    bot_mes = await bot.send_message(
-        mes.chat.id,
-        "Вы действительно хотите удалить все ваши аккаунты? Это действие "
-        "нельзя отменить. "
-        "Отправьте мастер пароль для подтверждения",
-    )
-    register_state(
-        mes, functools.partial(_delete_all2, bot, engine, master_pass, bot_mes)
-    )
-
-
-async def _delete_all2(
-    bot: AsyncTeleBot,
-    engine: Engine,
-    master_pass: db.models.MasterPass,
-    prev_mes: Message,
-    mes: Message,
-) -> None:
-    await base_handler(bot, mes, prev_mes)
-    text = mes.text.strip()
-    if encryption.master_pass.check_master_pass(text, master_pass):
-        db.delete.purge_accounts(engine, mes.from_user.id)
-        db.delete.delete_master_pass(engine, mes.from_user.id)
-        await send_tmp_message(
-            bot,
-            mes.chat.id,
-            "Всё успешно удалено",
-            sleep_time=10,
-        )
-    else:
-        await send_tmp_message(
-            bot,
-            mes.chat.id,
-            "Вы отправили не верный мастер пароль, ничего не удалено",
-        )
-
-
-async def set_master_password(
-    bot: AsyncTeleBot,
-    engine: Engine,
-    mes: Message,
-) -> None:
-    await base_handler(bot, mes, None)
-    if db.get.get_master_pass(engine, mes.from_user.id) is not None:
-        return await send_tmp_message(
-            bot,
-            mes.chat.id,
-            "Мастер пароль уже существует",
-        )
-    bot_mes = await bot.send_message(mes.chat.id, "Отправьте мастер пароль")
-    register_state(
-        mes,
-        functools.partial(_set_master_pass2, bot, engine, bot_mes),
-    )
-
-
-async def _set_master_pass2(
-    bot: AsyncTeleBot, engine: Engine, prev_mes: Message, mes: Message
-) -> None:
-    await base_handler(bot, mes, prev_mes)
-    text = mes.text.strip()
-    if text == "/cancel":
-        return await send_tmp_message(bot, mes.chat.id, "Успешная отмена")
-
-    master_password = encryption.master_pass.encrypt_master_pass(
-        mes.from_user.id,
-        text,
-    )
-    db.add.add_master_pass(engine, master_password)
-
-    await send_tmp_message(bot, mes.chat.id, "Успех")
-    del mes, text
-    gc.collect()
-
-
-async def reset_master_pass(
-    bot: AsyncTeleBot,
-    engine: Engine,
-    mes: Message,
-) -> None:
-    await base_handler(bot, mes)
-
-    master_pass = db.get.get_master_pass(engine, mes.from_user.id)
-    if master_pass is None:
-        return await send_tmp_message(
-            bot,
-            mes.chat.id,
-            "Мастер пароль не задан",
-        )
-
-    bot_mes = await bot.send_message(
-        mes.chat.id,
-        "Отправьте текущий мастер пароль",
-    )
-
-    register_state(
-        mes,
-        functools.partial(
-            _reset_master_pass2,
-            bot,
-            engine,
-            master_pass,
-            bot_mes,
-        ),
-    )
-
-
-async def _reset_master_pass2(
-    bot: AsyncTeleBot,
-    engine: Engine,
-    master_pass: db.models.MasterPass,
-    prev_mes: Message,
-    mes: Message,
-) -> None:
-    await base_handler(bot, mes, prev_mes)
-    text = mes.text.strip()
-    if text == "/cancel":
-        await send_tmp_message(bot, mes.chat.id, "Успешная отмена")
-
-    if not encryption.master_pass.check_master_pass(text, master_pass):
-        await send_tmp_message(bot, mes.chat.id, "Неверный мастер пароль")
-        return
-
-    bot_mes = await bot.send_message(
-        mes.chat.id,
-        "Отправьте новый мастер пароль. Осторожно, все аккаунты будут удалены",
-    )
-    register_state(
-        mes,
-        functools.partial(_reset_master_pass3, bot, engine, bot_mes),
-    )
-
-
-async def _reset_master_pass3(
-    bot: AsyncTeleBot, engine: Engine, prev_mes: Message, mes: Message
-) -> None:
-    await base_handler(bot, mes, prev_mes)
-    text = mes.text.strip()
-    if text == "/cancel":
-        return await send_tmp_message(bot, mes.chat.id, "Успешная отмена")
-
-    master_password = encryption.master_pass.encrypt_master_pass(
-        mes.from_user.id,
-        text,
-    )
-    db.delete.purge_accounts(engine, mes.from_user.id)
-    db.change.change_master_pass(engine, master_password)
-
-    await send_tmp_message(
-        bot, mes.chat.id, "Все ваши аккаунты удалены, а мастер пароль изменён"
-    )
-    del mes, text
-    gc.collect()
-
-
-async def add_account(bot: AsyncTeleBot, engine: Engine, mes: Message) -> None:
-    await base_handler(bot, mes)
-
-    master_password_from_db = db.get.get_master_pass(
-        engine,
-        mes.from_user.id,
-    )
-    if master_password_from_db is None:
-        return await send_tmp_message(bot, mes.chat.id, "Нет мастер пароля")
-
-    bot_mes = await bot.send_message(
-        mes.chat.id,
-        "Отправьте название аккаунта",
-    )
-
-    register_state(
-        mes,
-        functools.partial(_add_account2, bot, engine, bot_mes),
-    )
-
-
-async def _add_account2(
-    bot: AsyncTeleBot, engine: Engine, prev_mes: Message, mes: Message
-) -> None:
-    await base_handler(bot, mes, prev_mes)
-    text = mes.text.strip()
-    if text == "/cancel":
-        return await send_tmp_message(bot, mes.chat.id, "Успешная отмена")
-
-    if not check_account_name(text):
-        return await send_tmp_message(
-            bot,
-            mes.chat.id,
-            "Не корректное название аккаунта",
-        )
-    if text in db.get.get_account_names(engine, mes.from_user.id):
-        return await send_tmp_message(
-            bot, mes.chat.id, "Аккаунт с таким именем уже существует"
-        )
-
-    bot_mes = await bot.send_message(mes.chat.id, "Отправьте логин")
-    data = {"name": text}
-
-    register_state(
-        mes,
-        functools.partial(_add_account3, bot, engine, bot_mes, data),
-    )
-
-
-async def _add_account3(
-    bot: AsyncTeleBot,
-    engine: Engine,
-    prev_mes: Message,
-    data: dict[str, str],
-    mes: Message,
-) -> None:
-    await base_handler(bot, mes, prev_mes)
-    text = mes.text.strip()
-    if text == "/cancel":
-        return await send_tmp_message(bot, mes.chat.id, "Успешная отмена")
-    if not check_login(text):
-        return await send_tmp_message(bot, mes.chat.id, "Не корректный логин")
-
-    data["login"] = text
-
-    bot_mes = await bot.send_message(
-        mes.chat.id,
-        "Отправьте пароль от аккаунта",
-    )
-
-    register_state(
-        mes,
-        functools.partial(_add_account4, bot, engine, bot_mes, data),
-    )
-
-
-async def _add_account4(
-    bot: AsyncTeleBot,
-    engine: Engine,
-    prev_mes: Message,
-    data: dict[str, str],
-    mes: Message,
-) -> None:
-    await base_handler(bot, mes, prev_mes)
-    text = mes.text.strip()
-    if text == "/cancel":
-        return await send_tmp_message(bot, mes.chat.id, "Успешная отмена")
-    if not check_password(text):
-        return await send_tmp_message(
-            bot,
-            mes.chat.id,
-            "Не корректный пароль",
-        )
-
-    data["passwd"] = text
-
-    bot_mes = await bot.send_message(mes.chat.id, "Отправьте мастер пароль")
-
-    register_state(
-        mes,
-        functools.partial(_add_account5, bot, engine, bot_mes, data),
-    )
-
-
-async def _add_account5(
-    bot: AsyncTeleBot,
-    engine: Engine,
-    prev_mes: Message,
-    data: dict[str, str],
-    mes: Message,
-) -> None:
-    await base_handler(bot, mes, prev_mes)
-    text = mes.text.strip()
-    if text == "/cancel":
-        return await send_tmp_message(bot, mes.chat.id, "Успешная отмена")
-
-    master_password = db.get.get_master_pass(engine, mes.from_user.id)
-    if not encryption.master_pass.check_master_pass(text, master_password):
-        return await send_tmp_message(
-            bot,
-            mes.chat.id,
-            "Не подходит главный пароль",
-        )
-
-    # name, login, passwd = data["name"], data["login"], data["passwd"]
-    account = DecryptedAccount(
-        user_id=mes.from_user.id,
-        name=data["name"],
-        login=data["login"],
-        password=data["passwd"],
-    )
-
-    encrypted_account = encryption.accounts.encrypt(account, text)
-
-    result = db.add.add_account(
-        engine,
-        encrypted_account,
-    )
-
-    await send_tmp_message(
-        bot,
-        mes.chat.id,
-        "Успех" if result else "Произошла не предвиденная ошибка",
-    )
-
-    del data, account
-
-    gc.collect()
-
-
-async def get_account(bot: AsyncTeleBot, engine: Engine, mes: Message) -> None:
-    await base_handler(bot, mes)
-
-    master_pass = db.get.get_master_pass(engine, mes.from_user.id)
-    if master_pass is None:
-        return await send_tmp_message(bot, mes.chat.id, "Нет мастер пароля")
-
-    accounts = db.get.get_account_names(
-        engine,
-        mes.from_user.id,
-        to_sort=True,
-    )
-    markup = markups.account_markup(accounts)
-    bot_mes = await bot.send_message(
-        mes.chat.id, "Отправьте название аккаунта", reply_markup=markup
-    )
-    register_state(
-        mes,
-        functools.partial(_get_account2, bot, engine, bot_mes),
-    )
-
-
-async def _get_account2(
-    bot: AsyncTeleBot, engine: Engine, prev_mes: Message, mes: Message
-) -> None:
-    await base_handler(bot, mes, prev_mes)
-    text = mes.text.strip()
-    if text == "/cancel":
-        return await send_tmp_message(bot, mes.chat.id, "Успешная отмена")
-
-    if text not in db.get.get_account_names(engine, mes.from_user.id):
-        return await send_tmp_message(bot, mes.chat.id, "Нет такого аккаунта")
-
-    bot_mes = await bot.send_message(mes.chat.id, "Отправьте мастер пароль")
-    register_state(
-        mes,
-        functools.partial(_get_account3, bot, engine, bot_mes, text),
-    )
-
-
-async def _get_account3(
-    bot: AsyncTeleBot,
-    engine: Engine,
-    prev_mes: Message,
-    name: str,
-    mes: Message,
-) -> None:
-    await base_handler(bot, mes, prev_mes)
-    text = mes.text.strip()
-    if text == "/cancel":
-        return await send_tmp_message(bot, mes.chat.id, "Успешная отмена")
-
-    master_password = db.get.get_master_pass(
-        engine,
-        mes.from_user.id,
-    )
-
-    if not encryption.master_pass.check_master_pass(text, master_password):
-        return await send_tmp_message(
-            bot,
-            mes.chat.id,
-            "Не подходит мастер пароль",
-        )
-
-    account = db.get.get_account_info(engine, mes.from_user.id, name)
-    account = encryption.accounts.decrypt(account, text)
-    await send_deleteable_message(
-        bot,
-        mes.chat.id,
-        f"Название:\n`{account.name}`\n"
-        f"Логин:\n`{account.login}`\nПароль:\n`{account.password}`\nНажмите "
-        "на название, логин или пароль, чтобы скопировать",
-    )
-
-    del text, mes
-    gc.collect()
-
-
-async def delete_account(
-    bot: AsyncTeleBot,
-    engine: Engine,
-    mes: Message,
-) -> None:
-    await base_handler(bot, mes)
-
-    master_pass = db.get.get_master_pass(engine, mes.from_user.id)
-    if master_pass is None:
-        return await send_tmp_message(bot, mes.chat.id, "Нет мастер пароля")
-
-    accounts = db.get.get_account_names(
-        engine,
-        mes.from_user.id,
-        to_sort=True,
-    )
-    markup = markups.account_markup(accounts)
-    bot_mes = await bot.send_message(
-        mes.chat.id,
-        "Отправьте название аккаунта, который вы хотите удалить",
-        reply_markup=markup,
-    )
-
-    register_state(
-        mes,
-        functools.partial(_delete_account2, bot, engine, master_pass, bot_mes),
-    )
-
-
-async def _delete_account2(
-    bot: AsyncTeleBot,
-    engine: Engine,
-    master_pass: db.models.MasterPass,
-    prev_mes: Message,
-    mes: Message,
-):
-    await base_handler(bot, mes, prev_mes)
-    text = mes.text.strip()
-    if text == "/cancel":
-        return await send_tmp_message(bot, mes.chat.id, "Успешная отмена")
-
-    if text not in db.get.get_account_names(engine, mes.from_user.id):
-        return await send_tmp_message(bot, mes.chat.id, "Нет такого аккаунта")
-
-    bot_mes = await bot.send_message(
-        mes.from_user.id,
-        f'Вы уверены, что хотите удалить аккаунт "{text}"?\nОтправьте мастер '
-        "пароль для подтверждения",
-    )
-
-    register_state(
-        mes,
-        functools.partial(
-            _delete_account3,
-            bot,
-            engine,
-            master_pass,
-            bot_mes,
-            text,
-        ),
-    )
-
-
-async def _delete_account3(
-    bot: AsyncTeleBot,
-    engine: Engine,
-    master_pass: db.models.MasterPass,
-    prev_mes: Message,
-    account_name: str,
-    mes: Message,
-) -> None:
-    await base_handler(bot, mes, prev_mes)
-    text = mes.text.strip()
-    if not encryption.master_pass.check_master_pass(text, master_pass):
-        await send_tmp_message(bot, mes.chat.id, "Неверный пароль")
-        return
-
-    db.delete.delete_account(engine, mes.from_user.id, account_name)
-    await send_tmp_message(bot, mes.chat.id, "Аккаунт удалён")
-
-
-async def help_command(bot: AsyncTeleBot, mes: Message) -> None:
-    message = """Команды:
-/set_master_pass - установить мастер пароль
-/add_account - создать аккаунт
-/get_accounts - получить список аккаунтов
-/get_account - получить логин и пароль аккаунта
-/delete_account - удалить аккаунт
-/delete_all - удалить все аккаунты и мастер пароль
-/reset_master_pass - удалить все аккаунты и изменить мастер пароль
-/cancel - отмена текущего действия
-/help - помощь
-/export - получить пароли в json формате
-/import - импортировать пароли из json в файле в таком же формате, \
-как из /export
-/gen_password - создать 10 надёжных паролей"""
-    await bot.send_message(mes.chat.id, message)
-
-
-async def export(bot: AsyncTeleBot, engine: Engine, mes: Message) -> None:
-    await base_handler(bot, mes)
-    master_password_from_db = db.get.get_master_pass(
-        engine,
-        mes.from_user.id,
-    )
-
-    if master_password_from_db is None:
-        return await send_tmp_message(bot, mes.chat.id, "Нет мастер пароля")
-
-    if not db.get.get_account_names(engine, mes.from_user.id):
-        return await send_tmp_message(bot, mes.chat.id, "Нет аккаунтов")
-
-    bot_mes = await bot.send_message(mes.chat.id, "Отправьте мастер пароль")
-
-    register_state(mes, functools.partial(_export2, bot, engine, bot_mes))
-
-
-async def _export2(
-    bot: AsyncTeleBot, engine: Engine, prev_mes: Message, mes: Message
-) -> None:
-    await base_handler(bot, mes, prev_mes)
-    text = mes.text.strip()
-    if text == "/cancel":
-        return await send_tmp_message(bot, mes.chat.id, "Успешная отмена")
-
-    master_password = db.get.get_master_pass(
-        engine,
-        mes.from_user.id,
-    )
-    if not encryption.master_pass.check_master_pass(text, master_password):
-        return await send_tmp_message(
-            bot,
-            mes.chat.id,
-            "Не подходит мастер пароль",
-        )
-
-    accounts = db.get.get_accounts(engine, mes.from_user.id)
-    with ProcessPoolExecutor() as pool:
-        loop = asyncio.get_running_loop()
-        tasks = []
-        for account in accounts:
-            function = functools.partial(
-                encryption.accounts.decrypt,
-                account,
-                text,
-            )
-            tasks.append(loop.run_in_executor(pool, function))
-        accounts = await asyncio.gather(*tasks)
-    json_io = accounts_to_json(accounts)
-    await bot.send_document(
-        mes.chat.id,
-        json_io,
-        reply_markup=markups.deletion_markup(),
-    )
-
-    del text, accounts, json_io
-    gc.collect()
-
-
-async def import_accounts(
-    bot: AsyncTeleBot,
-    engine: Engine,
-    mes: Message,
-) -> None:
-    await base_handler(bot, mes)
-    master_password_from_db = db.get.get_master_pass(
-        engine,
-        mes.from_user.id,
-    )
-
-    if master_password_from_db is None:
-        return await send_tmp_message(bot, mes.chat.id, "Нет мастер пароля")
-
-    bot_mes = await bot.send_message(mes.chat.id, "Отправьте json файл")
-
-    register_state(mes, functools.partial(_import2, bot, engine, bot_mes))
-
-
-async def _import2(
-    bot: AsyncTeleBot, engine: Engine, prev_mes: Message, mes: Message
-) -> None:
-    await base_handler(bot, mes, prev_mes)
-    if mes.text is not None:
-        text = mes.text.strip()
-        if text == "/cancel":
-            return await send_tmp_message(bot, mes.chat.id, "Успешная отмена")
-
-    if mes.document is None:
-        return await send_tmp_message(
-            bot,
-            mes.chat.id,
-            "Вы должны отправить документ",
-        )
-    if mes.document.file_size > 102_400:  # If file size is bigger than 100 MB
-        return await send_tmp_message(
-            bot,
-            mes.chat.id,
-            "Файл слишком большой",
-        )
-
-    file_info = await bot.get_file(mes.document.file_id)
-    downloaded_file = await bot.download_file(file_info.file_path)
-    try:
-        accounts = json_to_accounts(
-            downloaded_file.decode("utf-8"),
-            mes.from_user.id,
-        )
-    except Exception:
-        return await send_tmp_message(
-            bot,
-            mes.chat.id,
-            "Ошибка во время работы с файлом",
-        )
-
-    bot_mes = await bot.send_message(mes.chat.id, "Отправьте мастер пароль")
-    register_state(
-        mes,
-        functools.partial(_import3, bot, engine, bot_mes, accounts),
-    )
-
-
-async def _import3(
-    bot: AsyncTeleBot,
-    engine: Engine,
-    prev_mes: Message,
-    accounts: list[DecryptedAccount],
-    mes: Message,
-) -> None:
-    await base_handler(bot, mes, prev_mes)
-    text = mes.text.strip()
-    if text == "/cancel":
-        return await send_tmp_message(bot, mes.chat.id, "Успешная отмена")
-
-    master_password = db.get.get_master_pass(
-        engine,
-        mes.from_user.id,
-    )
-    if not encryption.master_pass.check_master_pass(text, master_password):
-        return await send_tmp_message(
-            bot,
-            mes.chat.id,
-            "Не подходит мастер пароль",
-        )
-
-    # List of names of accounts, which failed to be added to the database
-    # or failed the tests
-    failed: list[str] = []
-    tasks: list[asyncio.Future[db.models.Account]] = []
-    loop = asyncio.get_running_loop()
-    with ProcessPoolExecutor() as pool:
-        for account in accounts:
-            if not check_account(account):
-                failed.append(account.name)
-                continue
-            function = functools.partial(
-                encryption.accounts.encrypt,
-                account,
-                text,
-            )
-            tasks.append(loop.run_in_executor(pool, function))
-        enc_accounts: list[db.models.Account] = await asyncio.gather(*tasks)
-    results = db.add.add_accounts(engine, enc_accounts)
-
-    failed_accounts = itertools.compress(
-        enc_accounts, (not result for result in results)
-    )
-    failed.extend((account.name for account in failed_accounts))
-
-    if failed:
-        await send_deleteable_message(
-            bot, mes.chat.id, "Не удалось добавить:\n" + "\n".join(failed)
-        )
-    else:
-        await send_tmp_message(bot, mes.chat.id, "Успех")
-
-    del text, mes, accounts, function, tasks, failed_accounts
-    gc.collect()
-
-
-async def gen_password(bot: AsyncTeleBot, mes: Message) -> None:
-    await base_handler(bot, mes)
-    # Generate 10 passwords and put them in the backticks
-    passwords = (f"`{generate_password.gen_password()}`" for _ in range(10))
-    text = (
-        "Пароли:\n"
-        + "\n".join(passwords)
-        + "\nНажмите на пароль, чтобы его скопировать"
-    )
-    await send_deleteable_message(bot, mes.chat.id, text)
-
-
-async def message_handler(bot: AsyncTeleBot, mes: Message) -> None:
-    handler = get_state(mes)
-    if handler is None:
-        await delete_message(bot, mes)
-        if mes.text.strip() == "/cancel":
-            await send_tmp_message(bot, mes.chat.id, "Нет активного действия")
-            return
-        await send_tmp_message(
-            bot,
-            mes.chat.id,
-            "Вы отправили не корректное сообщение",
-        )
-        return
-
-    try:
-        await handler(mes)
-    except Exception:
-        await send_tmp_message(
-            bot,
-            mes.chat.id,
-            "Произошла непредвиденная ошибка",
-        )
-        raise

src/bot/utils.py

@@ -0,0 +1,122 @@
+import io
+import string
+import time
+from random import SystemRandom
+from typing import Self, Type
+
+import pydantic
+import telebot
+from sqlalchemy.future import Engine
+
+from .. import cryptography, database
+
+
+class Account(pydantic.BaseModel):
+    name: str
+    login: str
+    passwd: str
+
+    @classmethod
+    def from_tuple(cls: Type[Self], tuple_: tuple[str, str, str]) -> Self:
+        return cls(name=tuple_[0], login=tuple_[1], passwd=tuple_[2])
+
+    def as_tuple(self: Self) -> tuple[str, str, str]:
+        return (self.name, self.login, self.passwd)
+
+
+class _Accounts(pydantic.BaseModel):
+    accounts: list[Account] = pydantic.Field(default_factory=list)
+
+
+def _accounts_list_to_json(accounts: list[tuple[str, str, str]]) -> str:
+    accounts = _Accounts(accounts=[Account.from_tuple(i) for i in accounts])
+    return accounts.json()
+
+
+def json_to_accounts(json_: str) -> list[tuple[str, str, str]]:
+    accounts = _Accounts.parse_raw(json_)
+    return [i.as_tuple() for i in accounts.accounts]
+
+
+Message = telebot.types.Message
+
+
+def send_tmp_message(
+    bot: telebot.TeleBot, chat_id: telebot.types.Message, text: str, timeout: int = 5
+) -> None:
+    bot_mes = bot.send_message(chat_id, text, parse_mode="MarkdownV2")
+    time.sleep(timeout)
+    bot.delete_message(chat_id, bot_mes.id)
+
+
+def base_handler(
+    bot: telebot.TeleBot, mes: Message, prev_mes: Message | None = None
+) -> None:
+    bot.delete_message(mes.chat.id, mes.id)
+    if prev_mes is not None:
+        bot.delete_message(prev_mes.chat.id, prev_mes.id)
+
+
+def get_all_accounts(
+    engine: Engine, user_id: int, master_pass: str
+) -> list[tuple[str, str, str]]:
+    accounts: list[tuple[str, str, str]] = []
+    for account_name in database.get.get_accounts(engine, user_id):
+        salt, enc_login, enc_passwd = database.get.get_account_info(
+            engine, user_id, account_name
+        )
+        login, passwd = cryptography.other_accounts.decrypt_account_info(
+            enc_login, enc_passwd, master_pass, salt
+        )
+        accounts.append((account_name, login, passwd))
+    return accounts
+
+
+def accounts_to_json(accounts: list[tuple[str, str, str]]) -> io.StringIO:
+    file = io.StringIO(_accounts_list_to_json(accounts))
+    file.name = "passwords.json"
+    return file
+
+
+def _base_check(val: str) -> bool:
+    "Returns false if finds new lines or backtick (`)"
+    return not ("\n" in val or "`" in val)
+
+
+def check_account_name(name: str) -> bool:
+    "Returns true if account name is valid"
+    return _base_check(name)
+
+
+def check_login(login: str) -> bool:
+    "Returns true if login is valid"
+    return _base_check(login)
+
+
+def check_passwd(passwd: str) -> bool:
+    "Returns true if password is valid"
+    return _base_check(passwd)
+
+
+def check_account(name: str, login: str, passwd: str) -> bool:
+    """Runs checks for account name, login and password"""
+    return check_account_name(name) and check_login(login) and check_passwd(passwd)
+
+
+def gen_passwd() -> str:
+    """Generates password of length 32"""
+    choices = SystemRandom().choices
+    chars = frozenset(string.ascii_letters + string.digits + string.punctuation)
+    # Remove backtick and pipe characters and convert into tuple
+    chars = tuple(chars.difference("`|"))
+    while True:
+        passwd = "".join(choices(chars, k=32))
+        passwd_chars = frozenset(passwd)
+        # If there is at least one lowercase character, uppercase character
+        # and one punctuation character
+        if (
+            passwd_chars.intersection(string.ascii_lowercase)
+            and passwd_chars.intersection(string.ascii_uppercase)
+            and passwd_chars.intersection(string.punctuation)
+        ):
+            return passwd

src/cryptography/__init__.py

@@ -0,0 +1,3 @@
+from . import master_pass, other_accounts
+
+__all__ = ["master_pass", "other_accounts"]

src/cryptography/master_pass.py

@@ -0,0 +1,35 @@
+import os
+
+from cryptography.exceptions import InvalidKey
+from cryptography.hazmat.primitives.kdf.scrypt import Scrypt
+
+MEMORY_USAGE = 2**14
+
+
+def _get_kdf(salt: bytes) -> Scrypt:
+    kdf = Scrypt(
+        salt=salt,
+        length=128,
+        n=MEMORY_USAGE,
+        r=8,
+        p=1,
+    )
+    return kdf
+
+
+def encrypt_master_pass(passwd: str) -> tuple[bytes, bytes]:
+    """Hashes master password and return tuple of hashed password and salt"""
+    salt = os.urandom(64)
+    kdf = _get_kdf(salt)
+    return kdf.derive(passwd.encode("utf-8")), salt
+
+
+def check_master_pass(passwd: str, enc_pass: bytes, salt: bytes) -> bool:
+    """Checks if the master password is correct"""
+    kdf = _get_kdf(salt)
+    try:
+        kdf.verify(passwd.encode("utf-8"), enc_pass)
+    except InvalidKey:
+        return False
+    else:
+        return True

src/cryptography/other_accounts.py

@@ -0,0 +1,44 @@
+import base64
+import os
+
+from cryptography.fernet import Fernet
+from cryptography.hazmat.backends import default_backend
+from cryptography.hazmat.primitives import hashes
+from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
+
+
+def _generate_key(salt: bytes, master_pass: bytes) -> bytes:
+    kdf = PBKDF2HMAC(
+        algorithm=hashes.SHA256(),
+        length=32,
+        salt=salt,
+        iterations=100000,
+        backend=default_backend(),
+    )
+    key = base64.urlsafe_b64encode(kdf.derive(master_pass))
+    return key
+
+
+def encrypt_account_info(
+    login: str, passwd: str, master_pass: str
+) -> tuple[bytes, bytes, bytes]:
+    """Encrypts login and password of a user using their master password as a key.
+    Returns a tuple of encrypted login, password and salt"""
+    salt = os.urandom(64)
+    key = _generate_key(salt, master_pass.encode("utf-8"))
+    f = Fernet(key)
+    enc_login = f.encrypt(login.encode("utf-8"))
+    enc_passwd = f.encrypt(passwd.encode("utf-8"))
+    return (enc_login, enc_passwd, salt)
+
+
+def decrypt_account_info(
+    enc_login: bytes, enc_pass: bytes, master_pass: str, salt: bytes
+) -> tuple[str, str]:
+    """Decrypts login and password using their master password as a key.
+    Returns a tuple of decrypted login and password"""
+    key = _generate_key(salt, master_pass.encode("utf-8"))
+    f = Fernet(key)
+    login_bytes = f.decrypt(enc_login)
+    pass_bytes = f.decrypt(enc_pass)
+    return (login_bytes.decode("utf-8"), pass_bytes.decode("utf-8"))

src/database/__init__.py

@@ -1,3 +1,3 @@
-from . import add, change, delete, get, models, prepare
+from . import add, delete, get, models, prepare, change
 
 __all__ = ["add", "delete", "get", "models", "prepare", "change"]

src/database/add.py

@@ -0,0 +1,40 @@
+import sqlmodel
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.future import Engine
+
+from . import models
+
+
+def add_account(
+    engine: Engine,
+    user_id: int,
+    name: str,
+    salt: bytes,
+    enc_login: bytes,
+    enc_pass: bytes,
+) -> bool:
+    """Adds account to the database. Returns true on success, false otherwise"""
+    account = models.Account(
+        user_id=user_id, name=name, salt=salt, enc_login=enc_login, enc_pass=enc_pass
+    )
+    try:
+        with sqlmodel.Session(engine) as session:
+            session.add(account)
+            session.commit()
+    except IntegrityError:
+        return False
+    else:
+        return True
+
+
+def add_master_pass(engine: Engine, user_id: int, salt: bytes, passwd: bytes) -> bool:
+    """Adds master password the database. Returns true on success, false otherwise"""
+    master_pass = models.MasterPass(user_id=user_id, salt=salt, passwd=passwd)
+    try:
+        with sqlmodel.Session(engine) as session:
+            session.add(master_pass)
+            session.commit()
+    except IntegrityError:
+        return False
+    else:
+        return True

src/database/change.py

@@ -5,17 +5,13 @@ from . import models
 
 
 def change_master_pass(
-    engine: Engine,
+    engine: Engine, user_id: int, salt: bytes, passwd: bytes
-    master_password: models.MasterPass,
 ) -> None:
     """Changes master password and salt in the database"""
     statement = (
         sqlmodel.update(models.MasterPass)
-        .where(models.MasterPass.user_id == master_password.user_id)
+        .where(models.MasterPass.user_id == user_id)
-        .values(
+        .values(salt=salt, passwd=passwd)
-            salt=master_password.salt,
-            password_hash=master_password.password_hash,
-        )
     )
     with sqlmodel.Session(engine) as session:
         session.exec(statement)

src/database/delete.py

@@ -6,9 +6,7 @@ from . import models
 
 def purge_accounts(engine: Engine, user_id: int) -> None:
     """Deletes all user's accounts"""
-    statement = sqlmodel.delete(models.Account).where(
+    statement = sqlmodel.delete(models.Account).where(models.Account.user_id == user_id)
-        models.Account.user_id == user_id,
-    )
     with sqlmodel.Session(engine) as session:
         session.exec(statement)
         session.commit()
@@ -27,8 +25,7 @@ def delete_master_pass(engine: Engine, user_id: int) -> None:
 def delete_account(engine: Engine, user_id: int, name: str) -> None:
     """Deletes specific user account"""
     statement = sqlmodel.delete(models.Account).where(
-        models.Account.user_id == user_id,
+        models.Account.user_id == user_id, models.Account.name == name
-        models.Account.name == name,
     )
     with sqlmodel.Session(engine) as session:
         session.exec(statement)

src/database/get.py

@@ -0,0 +1,40 @@
+import sqlmodel
+from sqlalchemy.future import Engine
+
+from . import models
+
+
+def get_master_pass(engine: Engine, user_id: int) -> tuple[bytes, bytes] | None:
+    """Gets master pass. Returns tuple of salt and password
+    or None if it wasn't found"""
+    statement = sqlmodel.select(models.MasterPass).where(
+        models.MasterPass.user_id == user_id
+    )
+    with sqlmodel.Session(engine) as session:
+        result = session.exec(statement).first()
+    if result is None:
+        return
+    return (result.salt, result.passwd)
+
+
+def get_accounts(engine: Engine, user_id: int) -> list[str]:
+    """Gets list of account names"""
+    statement = sqlmodel.select(models.Account).where(models.Account.user_id == user_id)
+    with sqlmodel.Session(engine) as session:
+        result = session.exec(statement)
+        return [account.name for account in result]
+
+
+def get_account_info(
+    engine: Engine, user_id: int, name: str
+) -> tuple[bytes, bytes, bytes]:
+    """Gets account info. Returns tuple of salt, login and password
+    or None if it wasn't found"""
+    statement = sqlmodel.select(models.Account).where(
+        models.Account.user_id == user_id, models.Account.name == name
+    )
+    with sqlmodel.Session(engine) as session:
+        result = session.exec(statement).first()
+        if result is None:
+            return
+        return (result.salt, result.enc_login, result.enc_pass)

src/database/models.py

@@ -0,0 +1,32 @@
+from typing import Optional
+
+import sqlmodel
+
+
+class MasterPass(sqlmodel.SQLModel, table=True):
+    __tablename__ = "master_passwords"
+    id: Optional[int] = sqlmodel.Field(primary_key=True)
+    user_id: int = sqlmodel.Field(nullable=False, index=True, unique=True)
+    salt: bytes = sqlmodel.Field(
+        sa_column=sqlmodel.Column(sqlmodel.BINARY(64), nullable=False)
+    )
+    passwd: bytes = sqlmodel.Field(
+        sa_column=sqlmodel.Column(sqlmodel.BINARY(128), nullable=False)
+    )
+
+
+class Account(sqlmodel.SQLModel, table=True):
+    __tablename__ = "accounts"
+    __table_args__ = (sqlmodel.UniqueConstraint("user_id", "name"),)
+    id: Optional[int] = sqlmodel.Field(primary_key=True)
+    user_id: int = sqlmodel.Field(nullable=False, index=True)
+    name: str = sqlmodel.Field(nullable=False, index=True, max_length=255)
+    salt: bytes = sqlmodel.Field(
+        sa_column=sqlmodel.Column(sqlmodel.BINARY(64), nullable=False)
+    )
+    enc_login: bytes = sqlmodel.Field(
+        sa_column=sqlmodel.Column(sqlmodel.VARBINARY(500), nullable=False)
+    )
+    enc_pass: bytes = sqlmodel.Field(
+        sa_column=sqlmodel.Column(sqlmodel.VARBINARY(500), nullable=False)
+    )

src/database/prepare.py

@@ -1,15 +1,12 @@
 import sqlmodel
 from sqlalchemy.future import Engine
 
-from . import models  # noqa
+from . import models
-
-HOUR_IN_SECONDS = 3600
 
 
 def get_engine(host: str, user: str, passwd: str, db: str) -> Engine:
     """Creates an engine for mariadb with pymysql as connector"""
-    uri = f"mariadb+pymysql://{user}:{passwd}@{host}/{db}"
+    engine = sqlmodel.create_engine(f"mariadb+pymysql://{user}:{passwd}@{host}/{db}")
-    engine = sqlmodel.create_engine(uri, pool_recycle=HOUR_IN_SECONDS)
     return engine
 
 

src/db/add.py

@@ -1,46 +0,0 @@
-import sqlmodel
-from sqlalchemy.exc import IntegrityError
-from sqlalchemy.future import Engine
-
-from . import models
-
-
-def _add_model(
-    session: sqlmodel.Session, model: models.Account | models.MasterPass
-) -> bool:
-    """Adds model to the session. Returns true on success,
-    false otherwise"""
-    try:
-        session.add(model)
-    except IntegrityError:
-        return False
-    else:
-        return True
-
-
-def add_account(engine: Engine, account: models.Account) -> bool:
-    """Adds account to the database. Returns true on success,
-    false otherwise"""
-    with sqlmodel.Session(engine) as session:
-        result = _add_model(session, account)
-        session.commit()
-    return result
-
-
-def add_master_pass(engine: Engine, master_pass: models.MasterPass) -> bool:
-    """Adds master password the database. Returns true on success,
-    false otherwise"""
-    with sqlmodel.Session(engine) as session:
-        result = _add_model(session, master_pass)
-        session.commit()
-    return result
-
-
-def add_accounts(
-    engine: Engine,
-    accounts: list[models.Account],
-) -> list[bool]:
-    with sqlmodel.Session(engine) as session:
-        result = [_add_model(session, account) for account in accounts]
-        session.commit()
-    return result

src/db/get.py

@@ -1,63 +0,0 @@
-import sqlmodel
-from sqlalchemy.future import Engine
-
-from . import models
-
-
-def get_master_pass(
-    engine: Engine,
-    user_id: int,
-) -> models.MasterPass | None:
-    """Gets master password of a user"""
-    statement = sqlmodel.select(models.MasterPass).where(
-        models.MasterPass.user_id == user_id,
-    )
-    with sqlmodel.Session(engine) as session:
-        result = session.exec(statement).first()
-    return result
-
-
-def get_account_names(
-    engine: Engine,
-    user_id: int,
-    *,
-    to_sort: bool = False,
-) -> list[str]:
-    """Gets a list of account names of a user"""
-    statement = sqlmodel.select(models.Account.name).where(
-        models.Account.user_id == user_id,
-    )
-    if to_sort:
-        statement = statement.order_by(models.Account.name)
-    with sqlmodel.Session(engine) as session:
-        result = session.exec(statement).fetchall()
-    return result
-
-
-def get_accounts(engine: Engine, user_id: int) -> list[models.Account]:
-    """Returns a list of accounts of a user"""
-    statement = (
-        sqlmodel.select(models.Account)
-        .where(
-            models.Account.user_id == user_id,
-        )
-        .order_by(models.Account.name)
-    )
-    with sqlmodel.Session(engine) as session:
-        result = session.exec(statement).fetchall()
-    return result
-
-
-def get_account_info(
-    engine: Engine,
-    user_id: int,
-    name: str,
-) -> models.Account:
-    """Gets account info"""
-    statement = sqlmodel.select(models.Account).where(
-        models.Account.user_id == user_id,
-        models.Account.name == name,
-    )
-    with sqlmodel.Session(engine) as session:
-        result = session.exec(statement).first()
-    return result

src/db/models.py

@@ -1,42 +0,0 @@
-import sqlmodel
-
-
-class MasterPass(sqlmodel.SQLModel, table=True):
-    __tablename__ = "master_passwords"
-    user_id: int = sqlmodel.Field(
-        sa_column=sqlmodel.Column(
-            sqlmodel.INT(),
-            primary_key=True,
-            autoincrement=False,
-        )
-    )
-    salt: bytes = sqlmodel.Field(
-        sa_column=sqlmodel.Column(sqlmodel.BINARY(64), nullable=False),
-        max_length=64,
-        min_length=64,
-    )
-    password_hash: bytes = sqlmodel.Field(
-        sa_column=sqlmodel.Column(sqlmodel.BINARY(128), nullable=False),
-        max_length=128,
-        min_length=128,
-    )
-
-
-class Account(sqlmodel.SQLModel, table=True):
-    __tablename__ = "accounts"
-    __table_args__ = (sqlmodel.PrimaryKeyConstraint("user_id", "name"),)
-    user_id: int = sqlmodel.Field()
-    name: str = sqlmodel.Field(max_length=256)
-    salt: bytes = sqlmodel.Field(
-        sa_column=sqlmodel.Column(sqlmodel.BINARY(64), nullable=False),
-        max_length=64,
-        min_length=64,
-    )
-    enc_login: bytes = sqlmodel.Field(
-        sa_column=sqlmodel.Column(sqlmodel.VARBINARY(256), nullable=False),
-        max_length=256,
-    )
-    enc_password: bytes = sqlmodel.Field(
-        sa_column=sqlmodel.Column(sqlmodel.VARBINARY(256), nullable=False),
-        max_length=256,
-    )

src/decrypted_account.py

@@ -1,8 +0,0 @@
-import pydantic
-
-
-class DecryptedAccount(pydantic.BaseModel):
-    user_id: int
-    name: str
-    login: str
-    password: str

src/encryption/__init__.py

@@ -1,3 +0,0 @@
-from . import accounts, master_pass
-
-__all__ = ["master_pass", "accounts"]

src/encryption/accounts.py

@@ -1,79 +0,0 @@
-import os
-from typing import Self
-
-from cryptography.hazmat.primitives import hashes
-from cryptography.hazmat.primitives.ciphers.aead import ChaCha20Poly1305
-from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
-
-from ..db.models import Account
-from ..decrypted_account import DecryptedAccount
-
-
-class Cipher:
-    def __init__(self, key: bytes) -> None:
-        self._chacha = ChaCha20Poly1305(key)
-
-    @classmethod
-    def generate_cipher(cls, salt: bytes, password: bytes) -> Self:
-        """Generates cipher which uses key derived from a given password"""
-        kdf = PBKDF2HMAC(
-            algorithm=hashes.SHA256(),
-            length=32,
-            salt=salt,
-            iterations=100000,
-        )
-        return cls(kdf.derive(password))
-
-    def encrypt(self, data: bytes) -> bytes:
-        nonce = os.urandom(12)
-        return nonce + self._chacha.encrypt(
-            nonce,
-            data,
-            associated_data=None,
-        )
-
-    def decrypt(self, data: bytes) -> bytes:
-        return self._chacha.decrypt(
-            nonce=data[:12],
-            data=data[12:],
-            associated_data=None,
-        )
-
-
-def encrypt(
-    account: DecryptedAccount,
-    master_pass: str,
-) -> Account:
-    """Encrypts account using master password and returns Account object"""
-    salt = os.urandom(64)
-    cipher = Cipher.generate_cipher(salt, master_pass.encode("utf-8"))
-
-    enc_login = cipher.encrypt(account.login.encode("utf-8"))
-    enc_password = cipher.encrypt(account.password.encode("utf-8"))
-
-    return Account(
-        user_id=account.user_id,
-        name=account.name,
-        salt=salt,
-        enc_login=enc_login,
-        enc_password=enc_password,
-    )
-
-
-def decrypt(
-    account: Account,
-    master_pass: str,
-) -> DecryptedAccount:
-    """Decrypts account using master password and returns
-    DecryptedAccount object"""
-    cipher = Cipher.generate_cipher(account.salt, master_pass.encode("utf-8"))
-
-    login = cipher.decrypt(account.enc_login).decode("utf-8")
-    password = cipher.decrypt(account.enc_password).decode("utf-8")
-
-    return DecryptedAccount(
-        user_id=account.user_id,
-        name=account.name,
-        login=login,
-        password=password,
-    )

src/encryption/master_pass.py

@@ -1,42 +0,0 @@
-import os
-
-from cryptography.exceptions import InvalidKey
-from cryptography.hazmat.primitives.kdf.scrypt import Scrypt
-
-from ..db.models import MasterPass
-
-MEMORY_USAGE = 2**14
-
-
-def _get_kdf(salt: bytes) -> Scrypt:
-    kdf = Scrypt(
-        salt=salt,
-        length=128,
-        n=MEMORY_USAGE,
-        r=8,
-        p=1,
-    )
-    return kdf
-
-
-def encrypt_master_pass(user_id: int, password: str) -> MasterPass:
-    """Hashes master password and returns MasterPass object"""
-    salt = os.urandom(64)
-    kdf = _get_kdf(salt)
-    password_hash = kdf.derive(password.encode("utf-8"))
-    return MasterPass(
-        user_id=user_id,
-        password_hash=password_hash,
-        salt=salt,
-    )
-
-
-def check_master_pass(password: str, master_password: MasterPass) -> bool:
-    """Checks if the master password is correct"""
-    kdf = _get_kdf(master_password.salt)
-    try:
-        kdf.verify(password.encode("utf-8"), master_password.password_hash)
-    except InvalidKey:
-        return False
-    else:
-        return True

src/generate_password.py

@@ -1,18 +0,0 @@
-import string
-from random import SystemRandom
-
-from .account_checks import FORBIDDEN_CHARS, PUNCTUATION, check_gened_password
-
-PASSWORD_CHARS = tuple(
-    frozenset(string.ascii_letters + string.digits).difference(FORBIDDEN_CHARS)
-    | PUNCTUATION
-)
-
-
-def gen_password() -> str:
-    """Generates password of length 32"""
-    choices = SystemRandom().choices
-    while True:
-        passwd = "".join(choices(PASSWORD_CHARS, k=32))
-        if check_gened_password(passwd):
-            return passwd