I wrote a (very) simply Python3 password manager to familiarize myself with Sqlite3, and for fun. I am aware that there are already a variety of password managers out there, but I wanted something simple and secure(?) to replace the very insecure bashrc function that I use to generate random passwords and store them with a quick description on the fly. I tried writing a password manager in bash that used hashing and GPG, but decided it was easier to do in Python.
I am looking for advice on how to improve the security of my program. I have implemented features such as:
- Timeout if idle
- Remove key from memory before exit
- Master key is hashed with sha256 and salted
The entire program is available here on github, but the functionality I am looking for advise on is the hashing and crypto functions:
Hash Lib:
import hmac
import random
import string
from _hashlib import pbkdf2_hmac
from getpass import getpass
from os import urandom, mkdir
from typing import Tuple
from lib.clipw_conf import debug, config_dir, config_file
class HashPass(object):
"""
Wrapper Class for password hashing functions
"""
def __init__(self):
"""
Empty __init__ function
"""
pass
def hash_new_password(self, password: str) -> Tuple[bytes, bytes]:
"""
Hash the provided password with a randomly-generated salt and return the
salt and hash to store in the database.
:return salt, password hash
"""
salt = urandom(16)
pw_hash = pbkdf2_hmac('sha256', password.encode(), salt, 100000)
return salt, pw_hash
def check_pw_padding(self, pw):
"""
Check that password length is AES compliant, add padding if not
:param pw: password
:return: password with buffering
"""
pw_len = len(pw)
if pw_len <= 8:
while pw_len % 8 != 0:
pw += '0'
pw_len = (len(pw))
if pw_len <= 16:
while pw_len % 16 != 0:
pw += '0'
pw_len = (len(pw))
if pw_len <= 32:
while pw_len % 32 != 0:
pw += '0'
pw_len = (len(pw))
return pw
def is_correct_password(self, salt: bytes, pw_hash: bytes, password: str) -> bool:
"""
Given a previously-stored salt and hash, and a password provided by a user
trying to log in, check whether the password is correct.
"""
return hmac.compare_digest(
pw_hash,
pbkdf2_hmac('sha256', password.encode(), salt, 100000)
)
def generate_hash(self, pw):
"""
Generate a Hash and Salt from a password
:param self: password to hash
:return: salt and hash
"""
salt, pw_hash = self.hash_new_password(pw)
salt = salt.hex()
pw_hash = pw_hash.hex()
if self.is_correct_password(bytes.fromhex(salt), bytes.fromhex(pw_hash), pw):
if debug:
print('Test Succeeded!')
print("Salt: %s" % salt)
print("Hash: %s" % pw_hash)
return salt, pw_hash
else:
return False
def store_master_password(self):
"""
Upon init, store users master key to disc
:return: True on success
"""
try:
mkdir(config_dir)
except FileExistsError as err:
pass
else:
print('Created config directory %s' % config_dir)
# hash a password
while True:
pw = getpass('Password: ')
pw2 = getpass('Confirm: ')
try:
assert pw == pw2
except AssertionError:
print('Passwords do not match. Try again')
else:
pw_len = len(pw)
if pw_len > 32 or pw_len < 8:
print('Password must be at least 8 and no more than 32 characters.')
else:
s, p = self.generate_hash(pw)
hash_str = str(s + ":" + p)
with open(config_file, 'w+') as f:
f.write(hash_str)
return True
def get_master_password(self):
"""
Function to create an AES friendly Master Password
to encrypt all the passwords in the database...
Because they key needs to be either 8 or 16 characters,
we will add padding until it if an appropriate length.
:return: byte encoded password string
"""
with open(config_file, 'r') as ff:
master_hash = ff.read()
_salt = master_hash.split(':')[0]
_hash = master_hash.split(':')[1]
master_pw = getpass("Master Password: ")
if self.is_correct_password(bytes.fromhex(_salt), bytes.fromhex(_hash), master_pw):
print('Success!')
master_pw = self.check_pw_padding(master_pw)
return master_pw.encode()
else:
print('Incorrect Password')
exit(1)
def store_password(self):
"""
Function to get a password and confirm user enters it twice correctly.
:return: password
"""
while True:
pw = getpass('Password: ')
pw2 = getpass('Confirm: ')
try:
assert pw == pw2
except AssertionError:
print('Passwords do not match.')
else:
return pw
def random_password(self, n: int = 12) -> object:
"""
Generate a random password of length n
:param n: length of password to generate
:return: password string
"""
random_source = string.ascii_letters + string.digits + string.punctuation
password = random.choice(string.ascii_lowercase)
password += random.choice(string.ascii_uppercase)
password += random.choice(string.digits)
password += random.choice(string.punctuation)
for i in range(n):
password += random.choice(random_source)
password_list = list(password)
random.SystemRandom().shuffle(password_list)
password = ''.join(password_list)[:n]
return password
I am curious about the function
check_pw_padding, which adds padding to the key that it is of a compliant length for the AES functions. I want the user to be able to enter a key of arbitraty length, however I need that key to work with AES-CTR mode, which means it must be either 8, 16, or 32 bytes characters long. I believe I am following standard practice here, but I wanted to ask the community for advice on whether or not my implimentation follows best practice.
AES Library:
import pyaes
import base64
from .clipw_conf import *
class HandleAes(object):
"""
AES Encrypt/Decrypt Function
"""
def __init__(self, key):
self.key = key
def remove_from_mem(self):
self.key = None
def encrypt_data(self, data):
"""
ENCRYPTION
AES CRT mode - 256 bit (8, 16, or 32 byte) key
:param data: data to encrypt
:return: base64 wrapper AES encrypted data
"""
aes = pyaes.AESModeOfOperationCTR(self.key)
ciphertext = aes.encrypt(data)
encoded = base64.b64encode(ciphertext).decode('utf-8')
if debug:
print('Encrypted data:', encoded)
# show the encrypted data
return encoded
def decrypt_data(self, data):
"""
DECRYPTION
AES CRT mode decryption requires a new instance be created
:param data: base64 encoded ciphertext
:return: plaintext
"""
aes = pyaes.AESModeOfOperationCTR(self.key)
# decrypted data is always binary, need to decode to plaintext
decoded = base64.b64decode(data)
decrypted = aes.decrypt(decoded).decode('utf-8')
if debug:
print('Decrypted data:', decrypted)
return decrypted
Is AES-CTR sufficient for this use case? I want to use a stream cipher because the data I am storing is going to be of arbitrary length. Of course, I could always add padding to the data that it stores, but if I don't need to do that, and if this library is secure and I can reuse it, that would be great. This answer says that CTR is safe. Does my implementation follow best practice? Specifically, is it sufficient, or even necessary to ensure the key is cleared from memory by assisinging the variable None (in function
remove_from_mem)
SQL Lib:
import sqlite3
from .clipw_conf import *
class Sql:
"""
Wrapper object to handle sqlite functions
"""
def __init__(self):
self.database_file = database_file
def open(self):
"""
Connect to SQL database
"""
try:
self.sqlite_connection = sqlite3.connect(self.database_file)
except Exception as err:
print('Error connecting to database:', err)
else:
return self.sqlite_connection
def close(self):
"""
Close connection to SQL database
:return:
"""
try:
self.sqlite_connection.close()
except Exception as err:
print('Error closing database:', err)
return False
def init_database(self):
"""
Function to init database on first run.
:return:
"""
sqlite_connection = self.open()
try:
sqlite_create_table_query = '''CREATE TABLE Password_Store (
id INTEGER PRIMARY KEY,
desc text NOT NULL,
pass_hash text NOT NULL);'''
cursor = sqlite_connection.cursor()
print("Successfully Connected to SQLite")
cursor.execute(sqlite_create_table_query)
sqlite_connection.commit()
print("SQLite table created")
return True
except sqlite3.Error as error:
print("Error while creating a sqlite table:", error)
return False
finally:
self.close()
def edit_database(self, id, data, field):
"""
Broken function to update a field of a row in the table
:param id: primary key id for WHERE clause
:param data: edited field data to replace in current db entry
:param field: either desc (description) or pass_hash (password hash of entry)
:return: True om success, false om fail
"""
# sqlite_connection = self.sqlite_connection
sqlite_connection = self.open()
try:
cursor = sqlite_connection.cursor() # define our sqlite connection
print("Connected to SQLite")
if field == 'pass_hash': # updating password field
cursor.execute('''Update Password_Store SET pass_hash = ? WHERE id = ?''',
(data, id)) # 2md failed method
else:
cursor.execute('''UPDATE Password_Store SET desc = ? WHERE id = ?''', (data, id))
sqlite_connection.commit()
print("Record Updated successfully ")
return True
except sqlite3.Error as error:
print("Failed to update sqlite table", error)
return False
finally:
self.close()
def append_database(self, new_passwd, pw_description):
"""
:param new_passwd: entry's password to append
:param pw_description: entry's password description to append
:return:
"""
# sqlite_connection = self.sqlite_connection
sqlite_connection = self.open()
try:
cursor = sqlite_connection.cursor()
if debug:
print("Connected to SQLite")
data_copy = cursor.execute("select count(*) from Password_Store")
values = data_copy.fetchone()
id = int(values[0]) + 1
sqlite_insert_with_param = """INSERT INTO 'Password_Store'
('id', 'desc', 'pass_hash')
VALUES (?, ?, ?);"""
data_tuple = (id, pw_description, new_passwd)
cursor.execute(sqlite_insert_with_param, data_tuple)
sqlite_connection.commit()
if debug:
print("Python Variables inserted successfully into SqliteDb_developers table")
except sqlite3.Error as error:
print("Failed to insert Python variable into sqlite table:", error)
return False
else:
print('Stored password ok.')
finally:
self.close()
def open_database(self):
"""
Open the database
:return: list(entries in database)
"""
# sqlite_connection = self.sqlite_connection
sqlite_connection = self.open()
try:
cursor = sqlite_connection.cursor()
if debug:
print("Connected to SQLite")
sqlite_select_query = """SELECT * from Password_Store"""
cursor.execute(sqlite_select_query)
records = cursor.fetchall()
if debug:
print("Total rows are: ", len(records))
id_desc = []
for row in records:
_id = int(row[0])
_desc = str(row[1])
id_desc.append([_id, _desc])
return id_desc
except sqlite3.Error as error:
print("Failed to read data from sqlite table", error)
return False
except Exception as err:
print('Unknown error:', err)
finally:
self.close()
def select_from_db(self, id):
"""
Grab a certain password from the database
:param id: primary key of password to get
:return: aes encrypted password
"""
# sqlite_connection = self.sqlite_connection
sqlite_connection = self.open()
try:
cursor = sqlite_connection.cursor()
if debug:
print("Connected to SQLite")
sql_select_query = """select * from Password_Store where id = ?"""
cursor.execute(sql_select_query, (id,))
record = cursor.fetchone()
if debug:
print("Sql library: Retrieving ID:", id)
return record
except sqlite3.Error as error:
print("Failed to read data from sqlite table", error)
return False
finally:
self.close()
def delete_from_database(self, id):
sqlite_connection = self.open()
try:
cursor = sqlite_connection.cursor()
if debug:
print("Connected to SQLite")
# Deleting single record now
sql_delete_query = """DELETE from Password_Store where id = ?"""
cursor.execute(sql_delete_query, (id,))
sqlite_connection.commit()
return "Record deleted successfully "
except sqlite3.Error as error:
print("Failed to delete record from sqlite table", error)
return False
finally:
self.close()
Config Lib
debug = False
from os import getenv
config_dir = getenv('HOME') + '/.clipw'
config_file = config_dir + '/clipw.conf'
database_file = config_dir + '/clipw.db'
Main Logic:
#!/usr/bin/env python3
"""
CliPassWord Manager - A Very Simple Python3 Powered Command Line Password Manager
Author: Darkerego ~ November, 2019 <[email protected]>
"""
import argparse
import shutil
import signal
import threading
from os import mkdir, kill, getpid
from os.path import isfile
from sys import argv
from lib import aes_lib
from lib import passlib
from lib import sql_functions
from lib.clipw_conf import *
def get_input(_type: str or int, _prompt: str):
"""
Functions to get and validate input.
:param _type: input type: str or int
:param _prompt: input prompt
:return: input
"""
if _type == 'int':
while True:
try:
io = int(input(_prompt))
except TypeError:
print('Enter a valid integer.')
else:
return io
elif _type == 'str':
while True:
try:
io = str(input(_prompt))
except TypeError:
print('Enter a valid string.')
else:
return io
def get_selection(_prompt: str, opts: list, _type: int or str):
"""
Function to handle selecting input from a list of options
:param _type:
:param _prompt: input prompt
:param opts: list of choices
:return: selected option
"""
while True:
try:
selection = _type(input(_prompt))
except TypeError:
print('Invalid input type.')
else:
if selection in opts:
return selection
else:
print("%d is not a valid option" % selection)
def init_database():
"""
Initialize the database
:return:
"""
print('Initializing database ... %s' % database_file)
try:
mkdir(config_dir)
except FileExistsError:
pass
if isfile(database_file):
remove_db = input('[!] Database file found, would you like to delete it? (y/n) :')
if remove_db.lower() == 'y':
print('Making a backup of current database ...')
shutil.move(database_file, database_file + '.orig')
sql_ = sql_functions.Sql()
hash_pass_ = passlib.HashPass()
hash_pass_.store_master_password()
if sql_.init_database():
print('Successfully initialized the database.')
sql_.close()
aes.remove_from_mem()
def store_pass():
"""
Store a password from standard input
:return:
"""
pw_description = input('Password Description: ')
new_pass = hash_pass.store_password()
encrypted_pw = aes.encrypt_data(new_pass)
sql.append_database(encrypted_pw, pw_description)
return True
def generate_random_pw(pw_len=None):
pw_description = input('Password Description: ')
if not pw_len:
pw_len = [8]
new_pw = hash_pass.random_password(pw_len[0])
encrypted_pw = aes.encrypt_data(new_pw)
sql.append_database(encrypted_pw, pw_description)
# print(str('Password: %s' % new_pw))
return new_pw
def open_db():
"""
Function to open the database and retrieve an entry
:return:
"""
id_desc = sql.open_database()
print('--------------------------------------------------------------')
print(' Password Database: ')
print('--------------------------------------------------------------')
for i in id_desc:
print('ID: ', i[0], "Description", i[1])
print('--------------------------------------------------------------')
get_entry = get_input('int', 'Enter ID of password to decrypt: ')
ret = sql.select_from_db(id=get_entry)
if ret is None:
print('%d is not a valid entry. Try again.' % get_entry)
else:
print("Id: ", ret[0])
print("Description:", ret[1])
hashed = ret[2]
try:
decrypted = aes.decrypt_data(hashed)
except UnicodeDecodeError as err:
print('Error decrypting password:', err)
except Exception as err:
print('Error decrypting password:', err)
else:
print('Password:', decrypted)
def edit_db_entry():
"""
Function to edit a database entry
:return:
"""
id_desc = sql.open_database()
ids = []
for i in id_desc:
print('ID: ', i[0], "Description", i[1])
ids.append(i[0])
entry_edit = get_selection(_prompt='Enter ID of entry to edit: ', opts=ids, _type=int)
ret = sql.select_from_db(id=entry_edit)
print('Fields: ')
print('ID:', ret[0])
print('[1] Description', ret[1])
show_pass = input('Show password (y/n)? : ')
if show_pass == 'y':
hashed = ret[2]
decrypted = aes.decrypt_data(hashed)
print('[2] Password:', decrypted)
else:
print('[2] Password: x-x-x-x-x')
edit_field = get_selection('Enter field of entry to edit. (1: Description, 2: Password)',
opts=[1, 2], _type=int)
if edit_field == 1:
new_description = input('Description: ')
try:
sql.edit_database(id=entry_edit, field='desc', data=new_description)
except Exception as err:
print('Error editing entry:', err)
else:
return True
if edit_field == 2:
suggest_password = input('Generate new random password? (y/n): ')
if suggest_password == 'y':
new_pw = hash_pass.random_password(16)
else:
new_pw = input('Enter new password: ')
if new_pw is not None:
encrypted_pw = aes.encrypt_data(new_pw)
try:
sql.edit_database(id=entry_edit, field='pass_hash', data=encrypted_pw)
except Exception as err:
print('Error editing entry:', err)
else:
return True
else:
print('Error: password is None.')
return False
def delete_entry_from_db():
"""
Function to delete an entry from the database.
:return:
"""
id_desc = sql.open_database()
ids = []
for i in id_desc:
print('ID: ', i[0], "Description", i[1])
ids.append(i[0])
get = get_selection(_prompt='Entry to delete: ', opts=ids, _type=int)
if debug:
print('Retrieving: ', get)
ret = sql.delete_from_database(id=get)
if ret:
print(ret)
else:
print('Error deleting from database.')
return False
def get_args():
"""
Argument parser
:return: argparse generated object
"""
parser = argparse.ArgumentParser(description='Python Cli Password Manager')
parser.add_argument('--init', '--init_database', dest='init_db', action='store_true', help='Re|Init Database')
parser.add_argument('-i', '--interactive', dest='interactive', action='store_true', default=False,
help='Interactive mode')
parser.add_argument('-t,', '--timeout', default=600, type=int, dest='timeout', help='Override timeout - close after'
'this amount of seconds.')
parser.add_argument('-o', '--open', dest='open', help='Open the password database', action='store_true')
parser.add_argument('-s', '--store', dest='store', action='store_true', help='Enter and store a new password in'
' the database')
parser.add_argument('-e', '--edit', dest='edit', help='Edit an entry.', action='store_true')
parser.add_argument('-d', '--delete', dest='delete', help='Delete an entry', action='store_true')
parser.add_argument('-r', '--random', dest='gen_random', help='Generate and store a random password of n length',
type=int, default=None, nargs="*")
return parser.parse_args()
def cleanup():
"""
Cleanup function for security
:return:
"""
aes.remove_from_mem()
class TimeOutTime:
"""
Empty object to hold specified timeout seconds
"""
def __init__(self, secs):
"""
Init Function
:param secs: seconds before timeout
"""
self.secs = secs
def get(self):
"""
Return specified seconds
:return: secs
"""
return int(self.secs)
def app():
"""
Program main logic
:return: -- print to stdout
"""
args = get_args()
try:
program_name = argv[0]
except Exception:
program_name = 'clipw'
global sql, hash_pass, aes
if args.init_db:
init_database()
exit(0)
else:
if not (args.open or args.store or args.edit or args.delete or args.gen_random or args.interactive):
print("Run %s -i for interactive or --help for usage" % program_name)
exit(1)
sql = sql_functions.Sql()
hash_pass = passlib.HashPass()
aes = aes_lib.HandleAes(key=hash_pass.get_master_password())
if args.interactive:
timeout = args.timeout
to = TimeOutTime(timeout)
wd = Watchdog(to.get())
wd.start()
print('Running in interactive mode. ')
while True:
try:
action = get_selection("Enter action: store (s), generate: (g), open and select entry: (o), "
"edit entry: "
"(e), delete entry: (d), quit program (q) ",
opts=["s", "g", "o", "e", "d", "q"], _type=str)
except KeyboardInterrupt:
break
except Exception as err:
print(err)
else:
try:
sql = sql_functions.Sql()
except Exception as err:
if debug:
print('Error with SQL:', err)
pass
wd.refresh()
action = action.lower()
print('Selected:', action)
if action == 's':
store_pass()
elif action == 'g':
pw_len = get_input('int', 'Password Length: ')
ret = generate_random_pw([pw_len])
if ret:
print('Password:', ret)
elif action == 'o':
open_db()
elif action == 'e':
edit_db_entry()
elif action == 'd':
delete_entry_from_db()
elif action == 'q':
break
else:
raise ValueError('Invalid Selection')
print('Quitting...')
wd.do_expire()
if args.store:
store_pass()
if args.gen_random is not None:
pw_len = args.gen_random
ret = generate_random_pw(pw_len)
if ret:
print("Password", ret)
if args.open:
open_db()
if args.edit:
edit_db_entry()
if args.delete:
delete_entry_from_db()
# Remove the master key from memory on exit
cleanup()
class Watchdog:
def __init__(self, _timeout=600):
self.timeout = _timeout
self._t = None
self.wd_debug = False # debug messages
def do_expire(self):
cleanup()
kill(getpid(), signal.SIGKILL)
def _expire(self):
print("\nTimeout expired, exiting for security...")
self.do_expire()
def start(self):
if self._t is None:
self._t = threading.Timer(self.timeout, self._expire)
self._t.start()
def stop(self):
if self._t is not None:
self._t.cancel()
self._t = None
def refresh(self):
if self.wd_debug:
print('Debug message: watchdog refresh.')
if self._t is not None:
self.stop()
self.start()
def main():
try:
app()
except KeyboardInterrupt:
print('Caught Signal, exiting ...')
finally:
exit(0)
if __name__ == "__main__":
main()
Other Notes & Questions:
- Is there a "cleaner" way to kill the process if it's been idle for (in this case) ten minutes? It seems kind of of "hacky" to use
os.kill(), but I can't seem to find a way to simply useexit(0)while importing functions from multiple libaries. - I know this program is kind of rudimentary right now. I intend on adding additional fields to the database such as (timestamp) "created" and "last modified", as well as more details like "username" and "url", instead of just having an id and description for each corresponding password.
- I am trying to think of a good method for securely sharing a database between different devices. Personally I just use ssh in these cases, however a friend of mine said he would like an easier way to do it. I considered perhaps encrypting the database again with an "export password" and then uploading to a service like spruge.us or termbin.com, so that the user could retrieve and import the database easily on another device. (Thoughts on this?)
- I am mostly looking for critique on how the program handles password hashing and the encryption functionality, however am open to any other suggestions and critique. Thank you for taking the time to review this code!
