Changed the way the master password hashing works

Switched from Bcrypt to Scrypt for master password hashing
Changed models to use new sizes for hashes and salts, doubled the size of enc_login and enc_passwd for accounts
Created new function to check master password validity
Increased salt sizes for accounts and master passwords
Removed bcrypt from requirements

.dockerignore

@@ -24,3 +24,4 @@
 **/secrets.dev.yaml
 **/values.dev.yaml
 README.md
+data/

Dockerfile

@@ -1,4 +1,4 @@
-FROM python:3.10-slim
+FROM python:3.11-slim
 
 # Keeps Python from generating .pyc files in the container
 ENV PYTHONDONTWRITEBYTECODE=1
@@ -6,21 +6,20 @@ ENV PYTHONDONTWRITEBYTECODE=1
 # Turns off buffering for easier container logging
 ENV PYTHONUNBUFFERED=1
 
+WORKDIR /app
+
+# Creates new user
+RUN adduser -u 1000 --disabled-password --gecos "" appuser && chown -R appuser /app
+
 # Install deps
 RUN apt update && apt full-upgrade -y
-RUN apt install curl gcc -y
-RUN curl -sS https://downloads.mariadb.com/MariaDB/mariadb_repo_setup | bash
-RUN apt install libmariadb3 libmariadb-dev -y
 
 # Install pip requirements
 COPY requirements.txt .
 RUN python -m pip install -r requirements.txt
 
-
-WORKDIR /app
 COPY . /app
 
-RUN adduser -u 1000 --disabled-password --gecos "" appuser && chown -R appuser /app
 USER appuser
 
 CMD ["python", "main.py"]

README.md

@@ -12,17 +12,21 @@
 
 ### Команды и их синтаксис
 
-- /set_master_pass {Мастер пароль} - установить мастер пароль
+- /set_master_pass - установить мастер пароль
-- /add_account {Название} {Логин} {Пароль} {Мастер пароль} - создать аккаунт
+- /add_account - создать аккаунт
 - /get_accounts - получить список аккаунтов
-- /get_account {Название} {Мастер пароль} - получить логин и пароль аккаунта
+- /get_account - получить логин и пароль аккаунта
-- /delete_account {Название} - удалить аккаунт
+- /delete_account - удалить аккаунт
 - /delete_all - удалить все аккаунты и мастер пароль
-- /reset_master_pass {Новый мастер пароль} - удалить все аккаунты и изменить мастер пароль
+- /reset_master_pass- удалить все аккаунты и изменить мастер пароль
+- /cancel - отмена текущего действия
+- /help - помощь
+- /export - получить пароли в json формате
+- /import - импортировать пароли из json в файле в таком же формате, как из /export
 
 ### Настройка
 
-Настройка питизводится через переменные среды. Их можно прописать в файле .env, если не хотите задавать их каждый раз в случае, если вы не работаете с Docker
+Настройка производится через переменные среды. Их можно прописать в файле .env, если не хотите задавать их каждый раз в случае, если вы не используете Docker
 
 #### Переменные среды
 

compose.yaml

@@ -13,13 +13,14 @@ services:
       DB_USER: manager
       DB_PASS: passwd123!
       DB_NAME: passmanager
+      TG_TOKEN: ${TG_TOKEN}
     depends_on:
       - db
     networks:
       - password_manager
 
   db:
-    image: jc21/mariadb-aria
+    image: stnicolay/mariadb-aria
     restart: always
     environment:
       MYSQL_ROOT_PASSWORD: example123!

requirements.txt

@@ -1,6 +1,6 @@
-bcrypt
 cryptography
-mariadb
+pymysql
 python-dotenv
 pyTelegramBotAPI
 sqlmodel
+pydantic

src/bot/__init__.py

@@ -1,15 +1,14 @@
 import functools
 
-import mariadb
-import telebot
 from sqlalchemy.future import Engine
+import telebot
 
-from . import handlers
+from . import handlers, utils
 
-__all__ = ["handlers"]
+__all__ = ["handlers", "utils"]
 
 
-def create_bot(token: str, engine: mariadb.Connection) -> telebot.TeleBot:
+def create_bot(token: str, engine: Engine) -> telebot.TeleBot:
     bot = telebot.TeleBot(token)
     bot.register_message_handler(
         functools.partial(handlers.set_master_password, bot, engine),
@@ -22,7 +21,7 @@ def create_bot(token: str, engine: mariadb.Connection) -> telebot.TeleBot:
         functools.partial(handlers.get_accounts, bot, engine), commands=["get_accounts"]
     )
     bot.register_message_handler(
-        functools.partial(handlers.add_record, bot, engine), commands=["add_account"]
+        functools.partial(handlers.add_account, bot, engine), commands=["add_account"]
     )
     bot.register_message_handler(
         functools.partial(handlers.delete_all, bot, engine), commands=["delete_all"]
@@ -38,4 +37,13 @@ def create_bot(token: str, engine: mariadb.Connection) -> telebot.TeleBot:
     bot.register_message_handler(
         functools.partial(handlers.help, bot), commands=["help", "start"]
     )
+    bot.register_message_handler(
+        functools.partial(handlers.cancel, bot), commands=["cancel"]
+    )
+    bot.register_message_handler(
+        functools.partial(handlers.export, bot, engine), commands=["export"]
+    )
+    bot.register_message_handler(
+        functools.partial(handlers.import_accounts, bot, engine), commands=["import"]
+    )
     return bot

src/bot/handlers.py

@@ -1,177 +1,473 @@
+import functools
 import gc
-import shlex
 import time
 
 import telebot
 from sqlalchemy.future import Engine
 
 from .. import cryptography, database
-
+from .utils import (
-
+    accounts_to_json,
-def _send_tmp_message(
+    base_handler,
-    bot: telebot.TeleBot, chat_id: telebot.types.Message, text: str, timeout: int = 5
+    check_account_name,
-) -> None:
+    check_login,
-    bot_mes = bot.send_message(chat_id, text)
+    check_passwd,
-    time.sleep(timeout)
+    check_account,
-    bot.delete_message(chat_id, bot_mes.id)
+    get_all_accounts,
-
+    json_to_accounts,
-
+    send_tmp_message,
-def add_record(
-    bot: telebot.TeleBot, engine: Engine, mes: telebot.types.Message
-) -> None:
-    data = shlex.split(mes.text)
-    bot.delete_message(mes.chat.id, mes.id)
-    if len(data) != 5:
-        return _send_tmp_message(bot, mes.chat.id, "Неправильное кол-во аргументов")
-
-    master_password = data[4]
-    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, "Нет мастер пароля")
-
-    master_salt, hash_pass = master_password_from_db
-    if (
-        cryptography.master_pass.encrypt_master_pass(master_password, master_salt)
-        != hash_pass
-    ):
-        return _send_tmp_message(bot, mes.chat.id, "Не подходит главный пароль")
-
-    enc_login, enc_pass, salt = cryptography.other_accounts.encrypt_account_info(
-        data[2], data[3], master_password.encode("utf-8")
 )
-    result = database.add.add_account(
+
-        engine, mes.from_user.id, data[1], salt, enc_login, enc_pass
+Message = telebot.types.Message
-    )
-    _send_tmp_message(
-        bot,
-        mes.chat.id,
-        "Успех" if result else "Ошибка, вероятно аккаунт с таким же именем существует",
-    )
-    del data, master_password, mes
-    gc.collect()
 
 
 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)
-    bot.delete_message(mes.chat.id, mes.id)
+    if not accounts:
+        return send_tmp_message(bot, mes.chat.id, "У вас нет аккаунтов")
 
-    return _send_tmp_message(
+    # 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) if accounts else "У вас нет аккаунтов",
+        "Ваши аккаунты:\n"
+        + "\n".join(accounts)
+        + "\nНажмите на название, чтобы скопировать",
+        30,
     )
 
 
-def set_master_password(
-    bot: telebot.TeleBot, engine: Engine, mes: telebot.types.Message
-) -> None:
-    data = shlex.split(mes.text)
-    bot.delete_message(mes.chat.id, mes.id)
-
-    if database.get.get_master_pass(engine, mes.from_user.id) is not None:
-        return _send_tmp_message(bot, mes.chat.id, "У вас уже установлен мастер пароль")
-
-    if len(data) != 2:
-        return _send_tmp_message(bot, mes.chat.id, "Неправильное количество аргументов")
-
-    pass_, salt = cryptography.master_pass.encrypt_master_pass(data[1])
-    result = database.add.add_master_pass(engine, mes.from_user.id, salt, pass_)
-    _send_tmp_message(
-        bot,
-        mes.chat.id,
-        "Успех" if result else "Ошибка, вероятно аккаунт с таким же именем существует",
-    )
-
-    del data, mes
-    gc.collect()
-
-
-def get_account(
-    bot: telebot.TeleBot, engine: Engine, mes: telebot.types.Message
-) -> None:
-    data = shlex.split(mes.text)
-    bot.delete_message(mes.chat.id, mes.id)
-
-    if len(data) != 3:
-        return _send_tmp_message(bot, 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, "Нет мастер пароля")
-    master_salt, hash_pass = master_pass
-
-    if data[1] not in database.get.get_accounts(engine, mes.from_user.id):
-        return _send_tmp_message(bot, mes.chat.id, "Нет такого аккаунта")
-
-    if cryptography.master_pass.encrypt_master_pass(data[2], master_salt) != hash_pass:
-        return _send_tmp_message(bot, mes.chat.id, "Не подходит главный пароль")
-
-    salt, enc_login, enc_pass = database.get.get_account_info(
-        engine, mes.from_user.id, data[1]
-    )
-    login, passwd = cryptography.other_accounts.decrypt_account_info(
-        enc_login, enc_pass, data[2].encode("utf-8"), salt
-    )
-    _send_tmp_message(bot, mes.chat.id, f"Логин:\n{login}\nПароль:\n{passwd}", 30)
-
-    del data, mes, passwd, login
-    gc.collect()
-
-
 def delete_all(
     bot: telebot.TeleBot, engine: Engine, mes: telebot.types.Message
 ) -> None:
-    database.delete.purge_accounts(engine, mes.from_user.id)
+    base_handler(bot, mes)
-    database.delete.delete_master_pass(engine, mes.from_user.id)
+    bot_mes = bot.send_message(
-    bot.send_message(mes.chat.id, "Все ваши данные удалены из базы данных")
+        mes.chat.id,
-
+        "Вы действительно хотите удалить все ваши аккаунты? Это действие нельзя отменить. Отправьте YES для подтверждения",
-
+    )
-def reset_master_pass(
+    bot.register_next_step_handler(
-    bot: telebot.TeleBot, engine: Engine, mes: telebot.types.Message
+        mes, functools.partial(_delete_all, bot, engine, bot_mes)
-) -> None:
-    bot.delete_message(mes.chat.id, mes.id)
-    data = shlex.split(mes.text)
-    if len(data) != 2:
-        return _send_tmp_message(bot, mes.chat.id, "Неправильное количество аргументов")
-
-    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, "Нет мастер пароля")
-
-    master_password = data[1]
-    enc_pass, salt = cryptography.master_pass.encrypt_master_pass(master_password)
-    database.delete.purge_accounts(engine, mes.from_user.id)
-    database.change.change_master_pass(engine, mes.from_user.id, salt, enc_pass)
-    _send_tmp_message(
-        bot, mes.chat.id, "Все ваши аккаунты удалены, а мастер пароль изменён"
     )
 
 
-def delete_account(
+def _delete_all(
-    bot: telebot.TeleBot, engine: Engine, mes: telebot.types.Message
+    bot: telebot.TeleBot, engine: Engine, prev_mes: Message, mes: Message
 ) -> None:
-    data = shlex.split(mes.text)
+    base_handler(bot, mes, prev_mes)
-    bot.delete_message(mes.chat.id, mes.id)
+    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, ничего не удалено")
 
-    if len(data) != 2:
-        return _send_tmp_message(bot, mes.chat.id, "Неправильное количество аргументов")
 
-    if data[1] not in database.get.get_accounts(engine, mes.from_user.id):
+def set_master_password(bot: telebot.TeleBot, engine: Engine, mes: Message) -> None:
-        return _send_tmp_message(bot, mes.chat.id, "Нет такого аккаунта")
+    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)
+    )
 
-    database.delete.delete_account(engine, mes.from_user.id, data[1])
+
-    _send_tmp_message(bot, mes.chat.id, "Аккаунт удалён")
+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.encode("utf-8")
+    )
+
+    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.encode("utf-8"), 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 {Мастер пароль} - установить мастер пароль
+/set_master_pass - установить мастер пароль
-/add_account {Название} {Логин} {Пароль} {Мастер пароль} - создать аккаунт
+/add_account - создать аккаунт
 /get_accounts - получить список аккаунтов
-/get_account {Название} {Мастер пароль} - получить логин и пароль аккаунта
+/get_account - получить логин и пароль аккаунта
-/delete_account {Название} - удалить аккаунт
+/delete_account - удалить аккаунт
 /delete_all - удалить все аккаунты и мастер пароль
-/reset_master_pass {Новый мастер пароль} - удалить все аккаунты и изменить мастер пароль"""
+/reset_master_pass - удалить все аккаунты и изменить мастер пароль
+/cancel - отмена текущего действия
+/help - помощь
+/export - получить пароли в json формате
+/import - импортировать пароли из json в файле в таком же формате, как из /export"""
     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.encode("utf-8")
+        )
+        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()

src/bot/utils.py

@@ -0,0 +1,102 @@
+import io
+import time
+from typing import Self, Type
+
+import pydantic
+import telebot
+from sqlalchemy.future import Engine
+
+from .. import database, cryptography
+
+
+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]] = []
+    master_pass = master_pass.encode("utf-8")
+    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)

src/cryptography/master_pass.py

@@ -1,26 +1,35 @@
-from typing import overload
+import os
 
-import bcrypt
+from cryptography.exceptions import InvalidKey
+from cryptography.hazmat.primitives.kdf.scrypt import Scrypt
+
+_memory_use = 2**14
 
 
-@overload
+def _get_kdf(salt: bytes) -> Scrypt:
-def encrypt_master_pass(passwd: str, salt: bytes) -> bytes:
+    kdf = Scrypt(
-    ...
+        salt=salt,
+        length=128,
+        n=_memory_use,
+        r=8,
+        p=1,
+    )
+    return kdf
 
 
-@overload
 def encrypt_master_pass(passwd: str) -> tuple[bytes, bytes]:
-    ...
-
-
-def encrypt_master_pass(
-    passwd: str, salt: bytes | None = None
-) -> tuple[bytes, bytes] | bytes:
     """Hashes master password and return tuple of hashed password and salt"""
-    if salt is None:
+    salt = os.urandom(64)
-        salt = bcrypt.gensalt()
+    kdf = _get_kdf(salt)
-        gened_salt = True
+    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:
-        gened_salt = False
+        return True
-    hashed = bcrypt.hashpw(passwd.encode("utf-8"), salt)
-    return (hashed, salt) if gened_salt else hashed

src/cryptography/other_accounts.py

@@ -1,6 +1,5 @@
 import base64
-
+import os
-import bcrypt
 
 from cryptography.fernet import Fernet
 from cryptography.hazmat.backends import default_backend
@@ -23,9 +22,9 @@ def _generate_key(salt: bytes, master_pass: bytes) -> bytes:
 def encrypt_account_info(
     login: str, passwd: str, master_pass: bytes
 ) -> tuple[bytes, bytes, bytes]:
-    """Encrypts login and password of a user using hash of their master password as a key.
+    """Encrypts login and password of a user using their master password as a key.
-    Returns a tuple of encrypted login password and salt"""
+    Returns a tuple of encrypted login, password and salt"""
-    salt = bcrypt.gensalt()
+    salt = os.urandom(64)
     key = _generate_key(salt, master_pass)
     f = Fernet(key)
     enc_login = f.encrypt(login.encode("utf-8"))
@@ -36,6 +35,8 @@ def encrypt_account_info(
 def decrypt_account_info(
     enc_login: bytes, enc_pass: bytes, master_pass: bytes, 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)
     f = Fernet(key)
     login_bytes = f.decrypt(enc_login)

src/database/change.py

@@ -7,10 +7,11 @@ from . import models
 def change_master_pass(
     engine: Engine, user_id: int, salt: bytes, passwd: bytes
 ) -> None:
-    statement = sqlmodel.update(
+    statement = (
-        models.MasterPass,
+        sqlmodel.update(models.MasterPass)
-        models.MasterPass.user_id == user_id,
+        .where(models.MasterPass.user_id == user_id)
-        {"salt": salt, "passwd": passwd},
+        .values(salt=salt, passwd=passwd)
     )
     with sqlmodel.Session(engine) as session:
         session.exec(statement)
+        session.commit()

src/database/delete.py

@@ -8,6 +8,7 @@ def purge_accounts(engine: Engine, user_id: int) -> None:
     statement = sqlmodel.delete(models.Account).where(models.Account.user_id == user_id)
     with sqlmodel.Session(engine) as session:
         session.exec(statement)
+        session.commit()
 
 
 def delete_master_pass(engine: Engine, user_id: int) -> None:
@@ -16,11 +17,13 @@ def delete_master_pass(engine: Engine, user_id: int) -> None:
     )
     with sqlmodel.Session(engine) as session:
         session.exec(statement)
+        session.commit()
 
 
 def delete_account(engine: Engine, user_id: int, name: str) -> None:
     statement = sqlmodel.delete(models.Account).where(
-        models.Account.user_id == user_id and models.Account.name == name
+        models.Account.user_id == user_id, models.Account.name == name
     )
     with sqlmodel.Session(engine) as session:
         session.exec(statement)
+        session.commit()

src/database/get.py

@@ -29,7 +29,7 @@ def get_account_info(
 ) -> tuple[bytes, bytes, bytes]:
     """Gets account info. Returns tuple of salt, login and password"""
     statement = sqlmodel.select(models.Account).where(
-        models.Account.user_id == user_id and models.Account.name == name
+        models.Account.user_id == user_id, models.Account.name == name
     )
     with sqlmodel.Session(engine) as session:
         result = session.exec(statement).first()

src/database/models.py

@@ -8,10 +8,10 @@ class MasterPass(sqlmodel.SQLModel, table=True):
     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(type_=sqlmodel.VARBINARY(255), nullable=False)
+        sa_column=sqlmodel.Column(sqlmodel.BINARY(64), nullable=False)
     )
     passwd: bytes = sqlmodel.Field(
-        sa_column=sqlmodel.Column(type_=sqlmodel.VARBINARY(255), nullable=False)
+        sa_column=sqlmodel.Column(sqlmodel.BINARY(128), nullable=False)
     )
 
 
@@ -22,11 +22,11 @@ class Account(sqlmodel.SQLModel, table=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(type_=sqlmodel.VARBINARY(255), nullable=False)
+        sa_column=sqlmodel.Column(sqlmodel.BINARY(64), nullable=False)
     )
     enc_login: bytes = sqlmodel.Field(
-        sa_column=sqlmodel.Column(type_=sqlmodel.VARBINARY(255), nullable=False)
+        sa_column=sqlmodel.Column(sqlmodel.VARBINARY(500), nullable=False)
     )
     enc_pass: bytes = sqlmodel.Field(
-        sa_column=sqlmodel.Column(type_=sqlmodel.VARBINARY(255), nullable=False)
+        sa_column=sqlmodel.Column(sqlmodel.VARBINARY(500), nullable=False)
     )

src/database/prepare.py

@@ -5,14 +5,9 @@ from . import models
 
 
 def get_engine(host: str, user: str, passwd: str, db: str) -> Engine:
-    engine = sqlmodel.create_engine(
+    engine = sqlmodel.create_engine(f"mariadb+pymysql://{user}:{passwd}@{host}/{db}")
-        f"mariadb+mariadbconnector://{user}:{passwd}@{host}/{db}"
-    )
     return engine
 
 
 def prepare(engine: Engine) -> None:
-    sqlmodel.SQLModel.metadata.create_all(
+    sqlmodel.SQLModel.metadata.create_all(engine)
-        engine,
-        # [models.Account, models.MasterPass]
-    )