37 Commits
Beta ... 1.1.1

Author SHA1 Message Date
e49f2e00eb Made pip upgrade before copying requirements.txt in Dockerfile 2022-11-18 16:38:25 +00:00
fd002e3718 Added command to update pip setuptools and install wheel in Dockerfile 2022-11-18 11:34:04 +00:00
d68a7bb6e8 Moved check of generated password into separate function 2022-11-15 15:39:38 +03:00
d5f3708c50 Refactored utils.py
Created constants FORBIDDEN_CHARS, PUNCTUATION, PASSWORD_CHARS
Removed pipe from forbidden chars
Moved Message type alias to the top of the file
Optimized _base_check and made its parameter positional only
Changed gen_password to use said constants
2022-11-15 15:17:36 +03:00
cc13b35282 Made class Account private in utils.py 2022-11-14 16:58:34 +03:00
3802943225 Changed telebot.types.Message to Message in handlers for old functions 2022-11-14 16:58:34 +03:00
309fb2108b Removed unnessesary import from src.__init__ 2022-11-14 16:58:34 +03:00
9d5c52bebe optimized password generation function 2022-11-14 16:58:26 +03:00
a61deca6fa Removed unnecessary comments and type hints in src.__init__ 2022-11-13 19:49:21 +03:00
18abdecb74 Increased time untill deletion for /gen_password to 15 secs 2022-11-13 18:53:11 +03:00
46a1fe59b2 Documented /gen_password command 2022-11-13 18:51:23 +03:00
9ee51dc19f Added /gen_password command 2022-11-13 18:49:57 +03:00
2c07b3bed2 Added function for password generation is utils.py 2022-11-13 18:49:14 +03:00
b8113dc47b Sorted imports in utils.py 2022-11-13 18:16:03 +03:00
21bd01c3ed Changed and added comments in the database files 2022-11-10 18:45:10 +03:00
c2280e8bc2 Changed functions for encryption and decryption of accounts to use str of master_pass instead of bytes 2022-11-09 19:02:58 +03:00
f7f954ecd3 Sorted imports 2022-11-09 18:52:59 +03:00
559ee8f6d8 Renamed _memory_use in cryptography into MEMORY_USAGE 2022-11-07 21:02:13 +03:00
a1bed9014d 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
2022-11-07 16:30:45 +03:00
66ab13b45d Changed handlers.get_accounts to use a guard clause, added comment to clarify, why we are putting account in backticks 2022-11-07 16:29:37 +03:00
da42d7ad1d Changed comments in cryptograpy.other_accounts to be more precise 2022-11-07 16:29:31 +03:00
570f15001e switched mariadb connector to pymysql 2022-11-04 02:17:17 +03:00
b4bf9fbf41 Changed mariadb.Connection to Engine in __init__ of bot 2022-11-04 02:16:47 +03:00
042ca9312e Cleaned up code in handlers
Renamed variables in _set_master_pass2 for consistency
Added a few missing returns in guard clauses
Added file size limit for importing account to 100 MB
2022-11-04 01:20:25 +03:00
b10b1fe2b3 changed to the newer version of mariadb 2022-11-03 16:27:52 +00:00
03646b6156 Added checks to /import 2022-11-03 10:49:00 +03:00
b56ebd0b61 Added account name and account credential checking 2022-11-02 23:11:29 +03:00
adf9865fbe Added ability to copy account name from /get_accounts
That also fixed an issue of having wrong behaviour, if using any special chars in account name
2022-11-02 23:02:20 +03:00
4122e15308 Added data directory to .dockerignore 2022-11-02 18:29:54 +00:00
37a81ed22d Changed change_master_pass to be more readable 2022-11-02 18:34:02 +03:00
fb850e3737 Added ability to import accounts 2022-11-02 14:48:15 +03:00
dbf27d401e Added ability to export accounts to json 2022-11-02 14:48:15 +03:00
373623b0b4 Removed type_ from sqlmodel.Column, because it isn't needed 2022-11-02 14:42:26 +03:00
afc03a6c1f Removed comment which broke code, when uncommented 2022-11-02 14:42:26 +03:00
4ee7f0a609 Added installation of g++ to Dockerfile 2022-11-01 09:26:13 +00:00
ae2b214904 Fixed an error of getting wrong account
I'm now quite shure how it worked before
2022-10-31 21:09:12 +03:00
01ab461d28 Added message of failure to add account 2022-10-31 20:40:53 +03:00
17 changed files with 448 additions and 143 deletions

View File

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

View File

@ -13,13 +13,11 @@ RUN adduser -u 1000 --disabled-password --gecos "" appuser && chown -R appuser /
# 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
RUN pip install -U pip setuptools wheel
COPY requirements.txt .
RUN python -m pip install -r requirements.txt
RUN pip install -r requirements.txt
COPY . /app

View File

@ -21,6 +21,9 @@
- /reset_master_pass- удалить все аккаунты и изменить мастер пароль
- /cancel - отмена текущего действия
- /help - помощь
- /export - получить пароли в json формате
- /import - импортировать пароли из json в файле в таком же формате, как из /export
- /gen_password - создать 10 надёжных паролей
### Настройка

View File

@ -20,7 +20,7 @@ services:
- password_manager
db:
image: jc21/mariadb-aria
image: stnicolay/mariadb-aria
restart: always
environment:
MYSQL_ROOT_PASSWORD: example123!

View File

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

View File

@ -1,12 +1,10 @@
import os
from dotenv import load_dotenv
from sqlalchemy.future import Engine
from . import bot, cryptography, database
__all__ = ["bot", "cryptography", "database"]
engine: Engine
def main() -> None:
@ -16,7 +14,7 @@ def main() -> None:
user=os.getenv("DB_USER"),
passwd=os.getenv("DB_PASS"),
db=os.getenv("DB_NAME"),
) # type: ignore
)
database.prepare.prepare(engine)
bot_ = bot.create_bot(os.getenv("TG_TOKEN"), engine) # type: ignore
bot_ = bot.create_bot(os.getenv("TG_TOKEN"), engine)
bot_.infinity_polling()

View File

@ -1,14 +1,14 @@
import functools
import mariadb
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),
@ -40,4 +40,13 @@ def create_bot(token: str, engine: mariadb.Connection) -> telebot.TeleBot:
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"]
)
bot.register_message_handler(
functools.partial(handlers.gen_password, bot), commands=["gen_password"]
)
return bot

View File

@ -6,42 +6,42 @@ 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 _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 get_accounts(
bot: telebot.TeleBot, engine: Engine, mes: telebot.types.Message
) -> None:
def get_accounts(bot: telebot.TeleBot, engine: Engine, mes: 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 "У вас нет аккаунтов",
timeout=30,
"Ваши аккаунты:\n"
+ "\n".join(accounts)
+ "\nНажмите на название, чтобы скопировать",
30,
)
def _base(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 delete_all(
bot: telebot.TeleBot, engine: Engine, mes: telebot.types.Message
) -> None:
_base(bot, mes)
def delete_all(bot: telebot.TeleBot, engine: Engine, mes: Message) -> None:
base_handler(bot, mes)
bot_mes = bot.send_message(
mes.chat.id,
"Вы действительно хотите удалить все ваши аккаунты? Это действие нельзя отменить. Отправьте YES для подтверждения",
@ -54,20 +54,20 @@ def delete_all(
def _delete_all(
bot: telebot.TeleBot, engine: Engine, prev_mes: Message, mes: Message
) -> None:
_base(bot, mes, prev_mes)
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)
send_tmp_message(bot, mes.chat.id, "Всё успешно удалено", timeout=10)
else:
_send_tmp_message(bot, mes.chat.id, "Вы отправили не YES, ничего не удалено")
send_tmp_message(bot, mes.chat.id, "Вы отправили не YES, ничего не удалено")
def set_master_password(bot: telebot.TeleBot, engine: Engine, mes: Message) -> None:
_base(bot, mes, 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, "Мастер пароль уже существует")
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)
@ -77,22 +77,22 @@ def set_master_password(bot: telebot.TeleBot, engine: Engine, mes: Message) -> N
def _set_master_pass2(
bot: telebot.TeleBot, engine: Engine, prev_mes: Message, mes: Message
) -> None:
_base(bot, mes, prev_mes)
base_handler(bot, mes, prev_mes)
text = mes.text.strip()
if text == "/cancel":
return _send_tmp_message(bot, mes.chat.id, "Успешная отмена")
return send_tmp_message(bot, mes.chat.id, "Успешная отмена")
hash_, salt = cryptography.master_pass.encrypt_master_pass(text)
database.add.add_master_pass(engine, mes.from_user.id, salt, hash_)
_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(bot, mes)
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, "Мастер пароль не задан")
return send_tmp_message(bot, mes.chat.id, "Мастер пароль не задан")
bot_mes = bot.send_message(
mes.chat.id,
"Отправьте новый мастер пароль, осторожно, все текущие аккаунты будут удалены навсегда",
@ -105,15 +105,15 @@ def reset_master_pass(bot: telebot.TeleBot, engine: Engine, mes: Message) -> Non
def _reset_master_pass2(
bot: telebot.TeleBot, engine: Engine, prev_mes: Message, mes: Message
) -> None:
_base(bot, mes, prev_mes)
base_handler(bot, mes, prev_mes)
text = mes.text.strip()
if text == "/cancel":
return _send_tmp_message(bot, mes.chat.id, "Успешная отмена")
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(
send_tmp_message(
bot, mes.chat.id, "Все ваши аккаунты удалены, а мастер пароль изменён"
)
del mes, text
@ -121,11 +121,11 @@ def _reset_master_pass2(
def add_account(bot: telebot.TeleBot, engine: Engine, mes: Message) -> None:
_base(bot, mes)
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, "Нет мастер пароля")
return send_tmp_message(bot, mes.chat.id, "Нет мастер пароля")
bot_mes = bot.send_message(mes.chat.id, "Отправьте название аккаунта")
@ -137,13 +137,15 @@ def add_account(bot: telebot.TeleBot, engine: Engine, mes: Message) -> None:
def _add_account2(
bot: telebot.TeleBot, engine: Engine, prev_mes: Message, mes: Message
) -> None:
_base(bot, mes, prev_mes)
base_handler(bot, mes, prev_mes)
text = mes.text.strip()
if text == "/cancel":
return _send_tmp_message(bot, mes.chat.id, "Успешная отмена")
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(
return send_tmp_message(
bot, mes.chat.id, "Аккаунт с таким именем уже существует"
)
@ -162,10 +164,12 @@ def _add_account3(
data: dict[str, str],
mes: Message,
) -> None:
_base(bot, mes, prev_mes)
base_handler(bot, mes, prev_mes)
text = mes.text.strip()
if text == "/cancel":
return _send_tmp_message(bot, mes.chat.id, "Успешная отмена")
return send_tmp_message(bot, mes.chat.id, "Успешная отмена")
if not check_login(text):
return send_tmp_message(bot, mes.chat.id, "Не корректный логин")
data["login"] = text
@ -183,10 +187,12 @@ def _add_account4(
data: dict[str, str],
mes: Message,
) -> None:
_base(bot, mes, prev_mes)
base_handler(bot, mes, prev_mes)
text = mes.text.strip()
if text == "/cancel":
return _send_tmp_message(bot, mes.chat.id, "Успешная отмена")
return send_tmp_message(bot, mes.chat.id, "Успешная отмена")
if not check_passwd(text):
return send_tmp_message(bot, mes.chat.id, "Не корректный пароль")
data["passwd"] = text
@ -204,24 +210,28 @@ def _add_account5(
data: dict[str, str],
mes: Message,
) -> None:
_base(bot, mes, prev_mes)
base_handler(bot, mes, prev_mes)
text = mes.text.strip()
if text == "/cancel":
return _send_tmp_message(bot, mes.chat.id, "Успешная отмена")
return send_tmp_message(bot, mes.chat.id, "Успешная отмена")
salt, hash_ = database.get.get_master_pass(engine, mes.from_user.id)
if cryptography.master_pass.encrypt_master_pass(text, salt) != hash_:
return _send_tmp_message(bot, mes.chat.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")
login, passwd, text
)
database.add.add_account(engine, mes.from_user.id, name, salt, enc_login, enc_pass)
result = database.add.add_account(
engine, mes.from_user.id, name, salt, enc_login, enc_pass
)
_send_tmp_message(bot, mes.chat.id, "Успех")
send_tmp_message(
bot, mes.chat.id, "Успех" if result else "Произошла не предвиденная ошибка"
)
del data, name, login, passwd, enc_login
@ -229,8 +239,13 @@ def _add_account5(
def get_account(bot: telebot.TeleBot, engine: Engine, mes: Message) -> None:
_base(bot, mes)
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)
)
@ -239,13 +254,13 @@ def get_account(bot: telebot.TeleBot, engine: Engine, mes: Message) -> None:
def _get_account2(
bot: telebot.TeleBot, engine: Engine, prev_mes: Message, mes: Message
) -> None:
_base(bot, mes, prev_mes)
base_handler(bot, mes, prev_mes)
text = mes.text.strip()
if text == "/cancel":
return _send_tmp_message(bot, mes.chat.id, "Успешная отмена")
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, "Нет такого аккаунта")
return send_tmp_message(bot, mes.chat.id, "Нет такого аккаунта")
bot_mes = bot.send_message(mes.chat.id, "Отправьте мастер пароль")
bot.register_next_step_handler(
@ -256,27 +271,23 @@ def _get_account2(
def _get_account3(
bot: telebot.TeleBot, engine: Engine, prev_mes: Message, name: str, mes: Message
) -> None:
_base(bot, mes, prev_mes)
base_handler(bot, mes, prev_mes)
text = mes.text.strip()
if text == "/cancel":
return _send_tmp_message(bot, mes.chat.id, "Успешная отмена")
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 = database.get.get_master_pass(engine, mes.from_user.id)
master_salt, hash_pass = master_pass
if cryptography.master_pass.encrypt_master_pass(text, master_salt) != hash_pass:
return _send_tmp_message(bot, mes.chat.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
enc_login, enc_pass, text, salt
)
_send_tmp_message(
send_tmp_message(
bot,
mes.chat.id,
f"Логин:\n`{login}`\nПароль:\n`{passwd}`\nНажмите на логин или пароль, чтобы скопировать",
@ -288,7 +299,12 @@ def _get_account3(
def delete_account(bot: telebot.TeleBot, engine: Engine, mes: Message) -> None:
_base(bot, mes)
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, "Отправьте название аккаунта, который вы хотите удалить"
)
@ -301,19 +317,19 @@ def delete_account(bot: telebot.TeleBot, engine: Engine, mes: Message) -> None:
def _delete_account2(
bot: telebot.TeleBot, engine: Engine, prev_mes: Message, mes: Message
) -> None:
_base(bot, mes, prev_mes)
base_handler(bot, mes, prev_mes)
text = mes.text.strip()
if text == "/cancel":
return _send_tmp_message(bot, mes.chat.id, "Успешная отмена")
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, "Нет такого аккаунта")
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, "Аккаунт удалён")
send_tmp_message(bot, mes.chat.id, "Аккаунт удалён")
def help(bot: telebot.TeleBot, mes: telebot.types.Message) -> None:
def help(bot: telebot.TeleBot, mes: Message) -> None:
message = """Команды:
/set_master_pass - установить мастер пароль
/add_account - создать аккаунт
@ -323,9 +339,145 @@ def help(bot: telebot.TeleBot, mes: telebot.types.Message) -> None:
/delete_all - удалить все аккаунты и мастер пароль
/reset_master_pass - удалить все аккаунты и изменить мастер пароль
/cancel - отмена текущего действия
/help - помощь"""
/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, "Нет активного действия")
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)

131
src/bot/utils.py Normal file
View File

@ -0,0 +1,131 @@
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
FORBIDDEN_CHARS = frozenset("`\n")
PUNCTUATION = frozenset(string.punctuation).difference(FORBIDDEN_CHARS)
PASSWORD_CHARS = tuple(
frozenset(string.ascii_letters + string.digits).difference(FORBIDDEN_CHARS)
| PUNCTUATION
)
Message = telebot.types.Message
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]
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 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_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 _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)
)
def gen_passwd() -> 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

View File

@ -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_USAGE = 2**14
@overload
def encrypt_master_pass(passwd: str, salt: bytes) -> bytes:
...
def _get_kdf(salt: bytes) -> Scrypt:
kdf = Scrypt(
salt=salt,
length=128,
n=MEMORY_USAGE,
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 = bcrypt.gensalt()
gened_salt = True
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:
gened_salt = False
hashed = bcrypt.hashpw(passwd.encode("utf-8"), salt)
return (hashed, salt) if gened_salt else hashed
return True

View File

@ -1,6 +1,5 @@
import base64
import bcrypt
import os
from cryptography.fernet import Fernet
from cryptography.hazmat.backends import default_backend
@ -21,12 +20,12 @@ def _generate_key(salt: bytes, master_pass: bytes) -> bytes:
def encrypt_account_info(
login: str, passwd: str, master_pass: bytes
login: str, passwd: str, master_pass: str
) -> tuple[bytes, bytes, bytes]:
"""Encrypts login and password of a user using hash of their master password as a key.
Returns a tuple of encrypted login password and salt"""
salt = bcrypt.gensalt()
key = _generate_key(salt, master_pass)
"""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"))
@ -34,9 +33,11 @@ def encrypt_account_info(
def decrypt_account_info(
enc_login: bytes, enc_pass: bytes, master_pass: bytes, salt: bytes
enc_login: bytes, enc_pass: bytes, master_pass: str, salt: bytes
) -> tuple[str, str]:
key = _generate_key(salt, master_pass)
"""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)

View File

@ -13,7 +13,7 @@ def add_account(
enc_login: bytes,
enc_pass: bytes,
) -> bool:
"""Adds account to db. Returns true, if on success"""
"""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
)
@ -28,7 +28,7 @@ def add_account(
def add_master_pass(engine: Engine, user_id: int, salt: bytes, passwd: bytes) -> bool:
"""Adds master password to db. Returns true, if on success"""
"""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:

View File

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

View File

@ -5,6 +5,7 @@ from . import models
def purge_accounts(engine: Engine, user_id: int) -> None:
"""Deletes all user's accounts"""
statement = sqlmodel.delete(models.Account).where(models.Account.user_id == user_id)
with sqlmodel.Session(engine) as session:
session.exec(statement)
@ -12,6 +13,7 @@ def purge_accounts(engine: Engine, user_id: int) -> None:
def delete_master_pass(engine: Engine, user_id: int) -> None:
"""Delets master password of the user"""
statement = sqlmodel.delete(models.MasterPass).where(
models.MasterPass.user_id == user_id
)
@ -21,6 +23,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.name == name
)

View File

@ -5,7 +5,8 @@ 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"""
"""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
)
@ -27,9 +28,10 @@ def get_accounts(engine: Engine, user_id: int) -> list[str]:
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"""
"""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 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()

View File

@ -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)
)

View File

@ -5,14 +5,11 @@ from . import models
def get_engine(host: str, user: str, passwd: str, db: str) -> Engine:
engine = sqlmodel.create_engine(
f"mariadb+mariadbconnector://{user}:{passwd}@{host}/{db}"
)
"""Creates an engine for mariadb with pymysql as connector"""
engine = sqlmodel.create_engine(f"mariadb+pymysql://{user}:{passwd}@{host}/{db}")
return engine
def prepare(engine: Engine) -> None:
sqlmodel.SQLModel.metadata.create_all(
engine,
# [models.Account, models.MasterPass]
)
"""Creates all tables, indexes and constrains in the database"""
sqlmodel.SQLModel.metadata.create_all(engine)