Compare commits
4 Commits
d79b57b1f0
...
671286dc39
Author | SHA1 | Date | |
---|---|---|---|
671286dc39 | |||
5dbf93013a | |||
c051c14f1f | |||
3686195396 |
@ -58,21 +58,31 @@ async def get_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,
|
||||
"Вы действительно хотите удалить все ваши аккаунты? Это действие "
|
||||
"нельзя отменить. "
|
||||
"Отправьте YES для подтверждения",
|
||||
"Отправьте мастер пароль для подтверждения",
|
||||
)
|
||||
register_state(
|
||||
mes, functools.partial(_delete_all2, bot, engine, master_pass, bot_mes)
|
||||
)
|
||||
register_state(mes, functools.partial(_delete_all2, bot, engine, bot_mes))
|
||||
|
||||
|
||||
async def _delete_all2(
|
||||
bot: AsyncTeleBot, engine: Engine, prev_mes: Message, mes: Message
|
||||
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 == "YES":
|
||||
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(
|
||||
@ -85,7 +95,7 @@ async def _delete_all2(
|
||||
await send_tmp_message(
|
||||
bot,
|
||||
mes.chat.id,
|
||||
"Вы отправили не YES, ничего не удалено",
|
||||
"Вы отправили не верный мастер пароль, ничего не удалено",
|
||||
)
|
||||
|
||||
|
||||
@ -134,7 +144,8 @@ async def reset_master_pass(
|
||||
) -> None:
|
||||
await base_handler(bot, mes)
|
||||
|
||||
if db.get.get_master_pass(engine, mes.from_user.id) is None:
|
||||
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,
|
||||
@ -143,17 +154,48 @@ async def reset_master_pass(
|
||||
|
||||
bot_mes = await bot.send_message(
|
||||
mes.chat.id,
|
||||
"Отправьте новый мастер пароль, осторожно, все текущие аккаунты "
|
||||
"будут удалены навсегда",
|
||||
"Отправьте текущий мастер пароль",
|
||||
)
|
||||
|
||||
register_state(
|
||||
mes,
|
||||
functools.partial(_reset_master_pass2, bot, engine, bot_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)
|
||||
@ -428,13 +470,14 @@ async def delete_account(
|
||||
|
||||
register_state(
|
||||
mes,
|
||||
functools.partial(_delete_account2, bot, engine, bot_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,
|
||||
):
|
||||
@ -448,27 +491,36 @@ async def _delete_account2(
|
||||
|
||||
bot_mes = await bot.send_message(
|
||||
mes.from_user.id,
|
||||
f'Вы уверены, что хотите удалить аккаунт "{text}"?\nОтправьте YES для '
|
||||
"подтверждения",
|
||||
f'Вы уверены, что хотите удалить аккаунт "{text}"?\nОтправьте мастер '
|
||||
"пароль для подтверждения",
|
||||
)
|
||||
|
||||
register_state(
|
||||
mes,
|
||||
functools.partial(_delete_account3, bot, engine, bot_mes, text),
|
||||
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 text != "YES":
|
||||
return 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
|
||||
|
||||
db.delete.delete_account(engine, mes.from_user.id, account_name)
|
||||
await send_tmp_message(bot, mes.chat.id, "Аккаунт удалён")
|
||||
@ -697,4 +749,13 @@ async def message_handler(bot: AsyncTeleBot, mes: Message) -> None:
|
||||
"Вы отправили не корректное сообщение",
|
||||
)
|
||||
return
|
||||
await handler(mes)
|
||||
|
||||
try:
|
||||
await handler(mes)
|
||||
except Exception:
|
||||
await send_tmp_message(
|
||||
bot,
|
||||
mes.chat.id,
|
||||
"Произошла непредвиденная ошибка",
|
||||
)
|
||||
raise
|
||||
|
@ -1,26 +1,43 @@
|
||||
import base64
|
||||
import os
|
||||
from typing import Self
|
||||
|
||||
from cryptography.fernet import Fernet
|
||||
from cryptography.hazmat.backends import default_backend
|
||||
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
|
||||
|
||||
|
||||
def _generate_key(salt: bytes, master_pass: bytes) -> bytes:
|
||||
"""Generates key for fernet encryption"""
|
||||
kdf = PBKDF2HMAC(
|
||||
algorithm=hashes.SHA256(),
|
||||
length=32,
|
||||
salt=salt,
|
||||
iterations=100000,
|
||||
backend=default_backend(),
|
||||
)
|
||||
key = base64.urlsafe_b64encode(kdf.derive(master_pass))
|
||||
return key
|
||||
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(
|
||||
@ -29,15 +46,10 @@ def encrypt(
|
||||
) -> Account:
|
||||
"""Encrypts account using master password and returns Account object"""
|
||||
salt = os.urandom(64)
|
||||
key = _generate_key(salt, master_pass.encode("utf-8"))
|
||||
f = Fernet(key)
|
||||
cipher = Cipher.generate_cipher(salt, master_pass.encode("utf-8"))
|
||||
|
||||
enc_login = base64.urlsafe_b64decode(
|
||||
f.encrypt(account.login.encode("utf-8")),
|
||||
)
|
||||
enc_password = base64.urlsafe_b64decode(
|
||||
f.encrypt(account.password.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,
|
||||
@ -54,15 +66,10 @@ def decrypt(
|
||||
) -> DecryptedAccount:
|
||||
"""Decrypts account using master password and returns
|
||||
DecryptedAccount object"""
|
||||
key = _generate_key(account.salt, master_pass.encode("utf-8"))
|
||||
f = Fernet(key)
|
||||
cipher = Cipher.generate_cipher(account.salt, master_pass.encode("utf-8"))
|
||||
|
||||
login = f.decrypt(
|
||||
base64.urlsafe_b64encode(account.enc_login),
|
||||
).decode("utf-8")
|
||||
password = f.decrypt(
|
||||
base64.urlsafe_b64encode(account.enc_password),
|
||||
).decode("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,
|
||||
|
Reference in New Issue
Block a user