15 Commits

Author SHA1 Message Date
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
14 changed files with 137 additions and 77 deletions

View File

@ -13,9 +13,6 @@ RUN adduser -u 1000 --disabled-password --gecos "" appuser && chown -R appuser /
# Install deps # Install deps
RUN apt update && apt full-upgrade -y RUN apt update && apt full-upgrade -y
RUN apt install curl gcc g++ -y
RUN curl -sS https://downloads.mariadb.com/MariaDB/mariadb_repo_setup | bash
RUN apt install libmariadb3 libmariadb-dev -y
# Install pip requirements # Install pip requirements
COPY requirements.txt . COPY requirements.txt .

View File

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

View File

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

View File

@ -1,6 +1,6 @@
import functools import functools
import mariadb from sqlalchemy.future import Engine
import telebot import telebot
from . import handlers, utils from . import handlers, utils
@ -8,7 +8,7 @@ from . import handlers, utils
__all__ = ["handlers", "utils"] __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 = telebot.TeleBot(token)
bot.register_message_handler( bot.register_message_handler(
functools.partial(handlers.set_master_password, bot, engine), functools.partial(handlers.set_master_password, bot, engine),
@ -46,4 +46,7 @@ def create_bot(token: str, engine: mariadb.Connection) -> telebot.TeleBot:
bot.register_message_handler( bot.register_message_handler(
functools.partial(handlers.import_accounts, bot, engine), commands=["import"] functools.partial(handlers.import_accounts, bot, engine), commands=["import"]
) )
bot.register_message_handler(
functools.partial(handlers.gen_password, bot), commands=["gen_password"]
)
return bot return bot

View File

@ -9,10 +9,11 @@ from .. import cryptography, database
from .utils import ( from .utils import (
accounts_to_json, accounts_to_json,
base_handler, base_handler,
check_account,
check_account_name, check_account_name,
check_login, check_login,
check_passwd, check_passwd,
check_account, gen_passwd,
get_all_accounts, get_all_accounts,
json_to_accounts, json_to_accounts,
send_tmp_message, send_tmp_message,
@ -26,17 +27,19 @@ def get_accounts(
) -> None: ) -> None:
base_handler(bot, mes) base_handler(bot, mes)
accounts = database.get.get_accounts(engine, mes.from_user.id) accounts = database.get.get_accounts(engine, mes.from_user.id)
if accounts: if not accounts:
accounts = [f"`{account}`" for account in accounts] return send_tmp_message(bot, mes.chat.id, "У вас нет аккаунтов")
return send_tmp_message(
bot, # Make accounts copyable and escape special chars
mes.chat.id, accounts = [f"`{account}`" for account in accounts]
"Ваши аккаунты:\n" send_tmp_message(
+ "\n".join(accounts) bot,
+ "\nНажмите на название, чтобы скопировать", mes.chat.id,
30, "Ваши аккаунты:\n"
) + "\n".join(accounts)
send_tmp_message(bot, mes.chat.id, "У вас нет аккаунтов") + "\nНажмите на название, чтобы скопировать",
30,
)
def delete_all( def delete_all(
@ -83,8 +86,8 @@ def _set_master_pass2(
if text == "/cancel": 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) hash_pass, master_salt = cryptography.master_pass.encrypt_master_pass(text)
database.add.add_master_pass(engine, mes.from_user.id, salt, hash_) database.add.add_master_pass(engine, mes.from_user.id, master_salt, hash_pass)
send_tmp_message(bot, mes.chat.id, "Успех") send_tmp_message(bot, mes.chat.id, "Успех")
del mes, text del mes, text
gc.collect() gc.collect()
@ -217,13 +220,13 @@ def _add_account5(
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) salt, hash_ = database.get.get_master_pass(engine, mes.from_user.id)
if cryptography.master_pass.encrypt_master_pass(text, salt) != hash_: if not cryptography.master_pass.check_master_pass(text, hash_, salt):
return send_tmp_message(bot, mes.chat.id, "Не подходит главный пароль") return send_tmp_message(bot, mes.chat.id, "Не подходит главный пароль")
name, login, passwd = data["name"], data["login"], data["passwd"] name, login, passwd = data["name"], data["login"], data["passwd"]
enc_login, enc_pass, salt = cryptography.other_accounts.encrypt_account_info( enc_login, enc_pass, salt = cryptography.other_accounts.encrypt_account_info(
login, passwd, text.encode("utf-8") login, passwd, text
) )
result = database.add.add_account( result = database.add.add_account(
@ -242,6 +245,11 @@ def _add_account5(
def get_account(bot: telebot.TeleBot, engine: Engine, mes: Message) -> None: def get_account(bot: telebot.TeleBot, engine: Engine, mes: Message) -> None:
base_handler(bot, mes) base_handler(bot, mes)
bot_mes = bot.send_message(mes.chat.id, "Отправьте название аккаунта") 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( bot.register_next_step_handler(
mes, functools.partial(_get_account2, bot, engine, bot_mes) mes, functools.partial(_get_account2, bot, engine, bot_mes)
) )
@ -272,20 +280,16 @@ def _get_account3(
if text == "/cancel": 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) master_salt, hash_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 not cryptography.master_pass.check_master_pass(text, hash_pass, master_salt):
if cryptography.master_pass.encrypt_master_pass(text, master_salt) != hash_pass:
return send_tmp_message(bot, mes.chat.id, "Не подходит мастер пароль") return send_tmp_message(bot, mes.chat.id, "Не подходит мастер пароль")
salt, enc_login, enc_pass = database.get.get_account_info( salt, enc_login, enc_pass = database.get.get_account_info(
engine, mes.from_user.id, name engine, mes.from_user.id, name
) )
login, passwd = cryptography.other_accounts.decrypt_account_info( 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, bot,
@ -300,6 +304,11 @@ def _get_account3(
def delete_account(bot: telebot.TeleBot, engine: Engine, mes: Message) -> None: def delete_account(bot: telebot.TeleBot, engine: Engine, mes: Message) -> None:
base_handler(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( bot_mes = bot.send_message(
mes.chat.id, "Отправьте название аккаунта, который вы хотите удалить" mes.chat.id, "Отправьте название аккаунта, который вы хотите удалить"
) )
@ -336,7 +345,8 @@ def help(bot: telebot.TeleBot, mes: telebot.types.Message) -> None:
/cancel - отмена текущего действия /cancel - отмена текущего действия
/help - помощь /help - помощь
/export - получить пароли в json формате /export - получить пароли в json формате
/import - импортировать пароли из json в файле в таком же формате, как из /export""" /import - импортировать пароли из json в файле в таком же формате, как из /export
/gen_password - создать 10 надёжных паролей"""
bot.send_message(mes.chat.id, message) bot.send_message(mes.chat.id, message)
@ -370,7 +380,7 @@ def _export2(
return send_tmp_message(bot, mes.chat.id, "Успешная отмена") 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 = database.get.get_master_pass(engine, mes.from_user.id)
if cryptography.master_pass.encrypt_master_pass(text, master_salt) != hash_pass: if not cryptography.master_pass.check_master_pass(text, hash_pass, master_salt):
return send_tmp_message(bot, mes.chat.id, "Не подходит мастер пароль") return send_tmp_message(bot, mes.chat.id, "Не подходит мастер пароль")
accounts = get_all_accounts(engine, mes.from_user.id, text) accounts = get_all_accounts(engine, mes.from_user.id, text)
@ -407,14 +417,16 @@ def _import2(
return send_tmp_message(bot, mes.chat.id, "Успешная отмена") return send_tmp_message(bot, mes.chat.id, "Успешная отмена")
if mes.document is None: if mes.document is None:
send_tmp_message(bot, mes.chat.id, "Вы должны отправить документ") 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) file_info = bot.get_file(mes.document.file_id)
downloaded_file = bot.download_file(file_info.file_path) downloaded_file = bot.download_file(file_info.file_path)
try: try:
accounts = json_to_accounts(downloaded_file.decode("utf-8")) accounts = json_to_accounts(downloaded_file.decode("utf-8"))
except Exception: except Exception:
send_tmp_message(bot, mes.chat.id, "Ошибка во время работы с файлом") return send_tmp_message(bot, mes.chat.id, "Ошибка во время работы с файлом")
bot_mes = bot.send_message(mes.chat.id, "Отправьте мастер пароль") bot_mes = bot.send_message(mes.chat.id, "Отправьте мастер пароль")
bot.register_next_step_handler( bot.register_next_step_handler(
@ -435,7 +447,7 @@ def _import3(
return send_tmp_message(bot, mes.chat.id, "Успешная отмена") 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 = database.get.get_master_pass(engine, mes.from_user.id)
if cryptography.master_pass.encrypt_master_pass(text, master_salt) != hash_pass: if not cryptography.master_pass.check_master_pass(text, hash_pass, master_salt):
return send_tmp_message(bot, mes.chat.id, "Не подходит мастер пароль") return send_tmp_message(bot, mes.chat.id, "Не подходит мастер пароль")
# List of names of accounts, which failed to be added to the database or failed tests # List of names of accounts, which failed to be added to the database or failed tests
@ -446,7 +458,7 @@ def _import3(
failed.append(name) failed.append(name)
continue continue
enc_login, enc_passwd, salt = cryptography.other_accounts.encrypt_account_info( enc_login, enc_passwd, salt = cryptography.other_accounts.encrypt_account_info(
login, passwd, text.encode("utf-8") login, passwd, text
) )
result = database.add.add_account( result = database.add.add_account(
engine, mes.from_user.id, name, salt, enc_login, enc_passwd engine, mes.from_user.id, name, salt, enc_login, enc_passwd
@ -461,3 +473,15 @@ def _import3(
send_tmp_message(bot, mes.chat.id, mes_text, 10) send_tmp_message(bot, mes.chat.id, mes_text, 10)
del text, mes, accounts del text, mes, accounts
gc.collect() 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)

View File

@ -1,12 +1,14 @@
import io import io
import string
import time import time
from random import SystemRandom
from typing import Self, Type from typing import Self, Type
import pydantic import pydantic
import telebot import telebot
from sqlalchemy.future import Engine from sqlalchemy.future import Engine
from .. import database, cryptography from .. import cryptography, database
class Account(pydantic.BaseModel): class Account(pydantic.BaseModel):
@ -59,7 +61,6 @@ def get_all_accounts(
engine: Engine, user_id: int, master_pass: str engine: Engine, user_id: int, master_pass: str
) -> list[tuple[str, str, str]]: ) -> list[tuple[str, str, str]]:
accounts: 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): for account_name in database.get.get_accounts(engine, user_id):
salt, enc_login, enc_passwd = database.get.get_account_info( salt, enc_login, enc_passwd = database.get.get_account_info(
engine, user_id, account_name engine, user_id, account_name
@ -100,3 +101,22 @@ def check_passwd(passwd: str) -> bool:
def check_account(name: str, login: str, passwd: str) -> bool: def check_account(name: str, login: str, passwd: str) -> bool:
"""Runs checks for account name, login and password""" """Runs checks for account name, login and password"""
return check_account_name(name) and check_login(login) and check_passwd(passwd) 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

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 _get_kdf(salt: bytes) -> Scrypt:
def encrypt_master_pass(passwd: str, salt: bytes) -> bytes: 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) -> 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""" """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: else:
gened_salt = False return True
hashed = bcrypt.hashpw(passwd.encode("utf-8"), salt)
return (hashed, salt) if gened_salt else hashed

View File

@ -1,6 +1,5 @@
import base64 import base64
import os
import bcrypt
from cryptography.fernet import Fernet from cryptography.fernet import Fernet
from cryptography.hazmat.backends import default_backend from cryptography.hazmat.backends import default_backend
@ -21,12 +20,12 @@ def _generate_key(salt: bytes, master_pass: bytes) -> bytes:
def encrypt_account_info( def encrypt_account_info(
login: str, passwd: str, master_pass: bytes login: str, passwd: str, master_pass: str
) -> tuple[bytes, bytes, 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) key = _generate_key(salt, master_pass.encode("utf-8"))
f = Fernet(key) f = Fernet(key)
enc_login = f.encrypt(login.encode("utf-8")) enc_login = f.encrypt(login.encode("utf-8"))
enc_passwd = f.encrypt(passwd.encode("utf-8")) enc_passwd = f.encrypt(passwd.encode("utf-8"))
@ -34,9 +33,11 @@ def encrypt_account_info(
def decrypt_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]: ) -> 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) f = Fernet(key)
login_bytes = f.decrypt(enc_login) login_bytes = f.decrypt(enc_login)
pass_bytes = f.decrypt(enc_pass) pass_bytes = f.decrypt(enc_pass)

View File

@ -13,7 +13,7 @@ def add_account(
enc_login: bytes, enc_login: bytes,
enc_pass: bytes, enc_pass: bytes,
) -> bool: ) -> bool:
"""Adds account to db. Returns true, if on success""" """Adds account to the database. Returns true on success, false otherwise"""
account = models.Account( account = models.Account(
user_id=user_id, name=name, salt=salt, enc_login=enc_login, enc_pass=enc_pass 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: 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) master_pass = models.MasterPass(user_id=user_id, salt=salt, passwd=passwd)
try: try:
with sqlmodel.Session(engine) as session: with sqlmodel.Session(engine) as session:

View File

@ -7,6 +7,7 @@ from . import models
def change_master_pass( def change_master_pass(
engine: Engine, user_id: int, salt: bytes, passwd: bytes engine: Engine, user_id: int, salt: bytes, passwd: bytes
) -> None: ) -> None:
"""Changes master password and salt in the database"""
statement = ( statement = (
sqlmodel.update(models.MasterPass) sqlmodel.update(models.MasterPass)
.where(models.MasterPass.user_id == user_id) .where(models.MasterPass.user_id == user_id)

View File

@ -5,6 +5,7 @@ from . import models
def purge_accounts(engine: Engine, user_id: int) -> None: 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) statement = sqlmodel.delete(models.Account).where(models.Account.user_id == user_id)
with sqlmodel.Session(engine) as session: with sqlmodel.Session(engine) as session:
session.exec(statement) 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: def delete_master_pass(engine: Engine, user_id: int) -> None:
"""Delets master password of the user"""
statement = sqlmodel.delete(models.MasterPass).where( statement = sqlmodel.delete(models.MasterPass).where(
models.MasterPass.user_id == user_id 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: def delete_account(engine: Engine, user_id: int, name: str) -> None:
"""Deletes specific user account"""
statement = sqlmodel.delete(models.Account).where( statement = sqlmodel.delete(models.Account).where(
models.Account.user_id == user_id, models.Account.name == name 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: 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( statement = sqlmodel.select(models.MasterPass).where(
models.MasterPass.user_id == user_id models.MasterPass.user_id == user_id
) )
@ -27,7 +28,8 @@ def get_accounts(engine: Engine, user_id: int) -> list[str]:
def get_account_info( def get_account_info(
engine: Engine, user_id: int, name: str engine: Engine, user_id: int, name: str
) -> tuple[bytes, bytes, bytes]: ) -> 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( statement = sqlmodel.select(models.Account).where(
models.Account.user_id == user_id, models.Account.name == name models.Account.user_id == user_id, models.Account.name == name
) )

View File

@ -8,10 +8,10 @@ class MasterPass(sqlmodel.SQLModel, table=True):
id: Optional[int] = sqlmodel.Field(primary_key=True) id: Optional[int] = sqlmodel.Field(primary_key=True)
user_id: int = sqlmodel.Field(nullable=False, index=True, unique=True) user_id: int = sqlmodel.Field(nullable=False, index=True, unique=True)
salt: bytes = sqlmodel.Field( salt: bytes = sqlmodel.Field(
sa_column=sqlmodel.Column(sqlmodel.VARBINARY(255), nullable=False) sa_column=sqlmodel.Column(sqlmodel.BINARY(64), nullable=False)
) )
passwd: bytes = sqlmodel.Field( passwd: bytes = sqlmodel.Field(
sa_column=sqlmodel.Column(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) user_id: int = sqlmodel.Field(nullable=False, index=True)
name: str = sqlmodel.Field(nullable=False, index=True, max_length=255) name: str = sqlmodel.Field(nullable=False, index=True, max_length=255)
salt: bytes = sqlmodel.Field( salt: bytes = sqlmodel.Field(
sa_column=sqlmodel.Column(sqlmodel.VARBINARY(255), nullable=False) sa_column=sqlmodel.Column(sqlmodel.BINARY(64), nullable=False)
) )
enc_login: bytes = sqlmodel.Field( enc_login: bytes = sqlmodel.Field(
sa_column=sqlmodel.Column(sqlmodel.VARBINARY(255), nullable=False) sa_column=sqlmodel.Column(sqlmodel.VARBINARY(500), nullable=False)
) )
enc_pass: bytes = sqlmodel.Field( enc_pass: bytes = sqlmodel.Field(
sa_column=sqlmodel.Column(sqlmodel.VARBINARY(255), nullable=False) sa_column=sqlmodel.Column(sqlmodel.VARBINARY(500), nullable=False)
) )

View File

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