Add AES CTR with key as iv

This commit is contained in:
lash 2022-01-21 10:59:27 +00:00
parent 36b4fcab93
commit d302c5754c
Signed by: lash
GPG Key ID: 21D2E7BB88C2A746
8 changed files with 112 additions and 48 deletions

View File

@ -22,6 +22,7 @@ from clicada.cli.http import (
HTTPSession, HTTPSession,
PGPClientSession, PGPClientSession,
) )
from clicada.crypt.aes import AESCTREncrypt
logg = logging.getLogger() logg = logging.getLogger()
@ -150,6 +151,7 @@ class CmdCtrl:
auth_db_path = self.get('AUTH_DB_PATH', default_auth_db_path) auth_db_path = self.get('AUTH_DB_PATH', default_auth_db_path)
self.__auth = PGPAuthCrypt(auth_db_path, self.get('AUTH_KEY'), self.get('AUTH_KEYRING_PATH')) self.__auth = PGPAuthCrypt(auth_db_path, self.get('AUTH_KEY'), self.get('AUTH_KEYRING_PATH'))
self.__auth.get_secret(self.get('AUTH_PASSPHRASE')) self.__auth.get_secret(self.get('AUTH_PASSPHRASE'))
self.encrypter = AESCTREncrypt(auth_db_path, self.__auth.secret)
def get(self, k, default=None): def get(self, k, default=None):

View File

@ -26,6 +26,7 @@ class PGPAuthCrypt:
raise AuthError('invalid key {}'.format(auth_key)) raise AuthError('invalid key {}'.format(auth_key))
self.auth_key = auth_key self.auth_key = auth_key
self.gpg = gnupg.GPG(gnupghome=pgp_dir) self.gpg = gnupg.GPG(gnupghome=pgp_dir)
self.secret = None
def get_secret(self, passphrase=''): def get_secret(self, passphrase=''):
@ -49,10 +50,11 @@ class PGPAuthCrypt:
f.write(secret.data) f.write(secret.data)
f.close() f.close()
f = open(p, 'rb') f = open(p, 'rb')
self.secret = self.gpg.decrypt_file(f, passphrase=passphrase) secret = self.gpg.decrypt_file(f, passphrase=passphrase)
if not self.secret.ok: if not secret.ok:
raise AuthError('could not decrypt encryption secret. wrong password?') raise AuthError('could not decrypt encryption secret. wrong password?')
f.close() f.close()
self.secret = secret.data
self.__passphrase = passphrase self.__passphrase = passphrase

32
clicada/cli/notify.py Normal file
View File

@ -0,0 +1,32 @@
# standard imports
import os
import sys
class NotifyWriter:
def __init__(self, writer=sys.stdout):
(c, r) = os.get_terminal_size()
self.cols = c
self.fmt = "\r{:" + "<{}".format(c) + "}"
self.w = writer
self.notify_max = self.cols - 4
def notify(self, v):
if len(v) > self.notify_max:
v = v[:self.notify_max]
self.write('\x1b[0;36m... ' + v + '\x1b[0;39m')
def ouch(self, v):
if len(v) > self.notify_max:
v = v[:self.notify_max]
self.write('\x1b[0;91m!!! ' + v + '\x1b[0;39m')
def write(self, v):
s = str(v)
if len(s) > self.cols:
s = s[:self.cols]
self.w.write(self.fmt.format(s))

View File

@ -56,7 +56,7 @@ def execute(ctrl):
store_path = '.clicada' store_path = '.clicada'
user_phone_file_label = 'phone' user_phone_file_label = 'phone'
user_phone_store = FileUserStore(ctrl.opener('meta'), ctrl.chain(), user_phone_file_label, store_path, int(ctrl.get('FILESTORE_TTL'))) user_phone_store = FileUserStore(ctrl.opener('meta'), ctrl.chain(), user_phone_file_label, store_path, int(ctrl.get('FILESTORE_TTL')), encrypter=ctrl.encrypter)
ctrl.notify('resolving identifier {} to wallet address'.format(ctrl.get('_IDENTIFIER'))) ctrl.notify('resolving identifier {} to wallet address'.format(ctrl.get('_IDENTIFIER')))
user_address = user_phone_store.by_phone(ctrl.get('_IDENTIFIER'), update=ctrl.get('_FORCE')) user_address = user_phone_store.by_phone(ctrl.get('_IDENTIFIER'), update=ctrl.get('_FORCE'))
@ -78,7 +78,7 @@ def execute(ctrl):
token_store = FileTokenStore(ctrl.chain(), ctrl.conn(), 'token', store_path) token_store = FileTokenStore(ctrl.chain(), ctrl.conn(), 'token', store_path)
user_address_file_label = 'address' user_address_file_label = 'address'
user_address_store = FileUserStore(ctrl.opener('meta'), ctrl.chain(), user_address_file_label, store_path, int(ctrl.get('FILESTORE_TTL'))) user_address_store = FileUserStore(ctrl.opener('meta'), ctrl.chain(), user_address_file_label, store_path, int(ctrl.get('FILESTORE_TTL')), encrypter=ctrl.encrypter)
ctrl.notify('resolving metadata for address {}'.format(user_address_normal)) ctrl.notify('resolving metadata for address {}'.format(user_address_normal))
try: try:

View File

@ -1,38 +0,0 @@
# standard imports
import os
import logging
from Crypto.Cipher import AES
from Crypto.Util import Counter
logg = logging.getLogger(__name__)
class Encrypt:
aesBlockSize = 1 << 7
def __init__(self, secret, db_dir):
fp = os.path.join(db_dir, '.aes_ctr_iv')
try:
f = open(fp, 'rb')
self.iv = f.read()
except FileNotFoundError:
logg.debug('generating new iv for aes-ctr')
self.iv = os.urandom(8)
f = open(fp, 'wb')
f.write(self.iv)
f.close()
iv_num = int.from_bytes(self.iv, 'big')
self.ctr = Counter.new(aesBlockSize, initial_value=iv_num)
self.cipher = AES.new(secret, AES.MODE_CTR, counter=self.ctr)
def encrypt(self, v):
return self.cipher.encrypt(v)
def decrypt(self, v):
return self.cipher.decrypt(v)

42
clicada/crypt/aes.py Normal file
View File

@ -0,0 +1,42 @@
# standard imports
import os
import logging
import hashlib
from Crypto.Cipher import AES
from Crypto.Util import Counter
from .base import Encrypter
logg = logging.getLogger(__name__)
class AESCTREncrypt(Encrypter):
aes_block_size = 1 << 7
counter_bytes = int(128 / 8)
def __init__(self, db_dir, secret):
self.secret = secret
def key_to_iv(self, k):
h = hashlib.sha256()
h.update(k.encode('utf-8'))
h.update(self.secret)
z = h.digest()
return int.from_bytes(z[:self.counter_bytes], 'big')
def encrypt(self, k, v):
iv = self.key_to_iv(k)
ctr = Counter.new(self.aes_block_size, initial_value=iv)
cipher = AES.new(self.secret, AES.MODE_CTR, counter=ctr)
return cipher.encrypt(v)
def decrypt(self, k, v):
iv = self.key_to_iv(k)
ctr = Counter.new(self.aes_block_size, initial_value=iv)
cipher = AES.new(self.secret, AES.MODE_CTR, counter=ctr)
return cipher.decrypt(v)

8
clicada/crypt/base.py Normal file
View File

@ -0,0 +1,8 @@
class Encrypter:
def encrypt(self, v):
raise NotImplementedError()
def decrypt(self, v):
raise NotImplementedError()

View File

@ -65,7 +65,7 @@ class Account(Person):
class FileUserStore: class FileUserStore:
def __init__(self, metadata_opener, chain_spec, label, store_base_path, ttl): def __init__(self, metadata_opener, chain_spec, label, store_base_path, ttl, encrypter=None):
invalidate_before = datetime.datetime.now() - datetime.timedelta(seconds=ttl) invalidate_before = datetime.datetime.now() - datetime.timedelta(seconds=ttl)
self.invalidate_before = int(invalidate_before.timestamp()) self.invalidate_before = int(invalidate_before.timestamp())
self.have_xattr = False self.have_xattr = False
@ -82,6 +82,7 @@ class FileUserStore:
self.__validate_dir() self.__validate_dir()
self.metadata_opener = metadata_opener self.metadata_opener = metadata_opener
self.failed_entities = {} self.failed_entities = {}
self.encrypter = encrypter
def __validate_dir(self): def __validate_dir(self):
@ -108,8 +109,14 @@ class FileUserStore:
if have_file and not its_time and not force: if have_file and not its_time and not force:
raise FileExistsError('user resolution already exists for {}'.format(k)) raise FileExistsError('user resolution already exists for {}'.format(k))
ve = v
f = None
if self.encrypter != None:
ve = self.encrypter.encrypt(k, ve.encode('utf-8'))
f = open(p, 'wb')
else:
f = open(p, 'w') f = open(p, 'w')
f.write(v) f.write(ve)
f.close() f.close()
logg.info('added user store {} record {} -> {}'.format(self.label, k, v)) logg.info('added user store {} record {} -> {}'.format(self.label, k, v))
@ -174,12 +181,21 @@ class FileUserStore:
self.__unstick(p) self.__unstick(p)
self.check_expiry(p) self.check_expiry(p)
f = None
if self.encrypter != None:
f = open(p, 'rb')
else:
f = open(p, 'r') f = open(p, 'r')
r = f.read() v = f.read()
f.close() f.close()
if self.encrypter != None:
v = self.encrypter.decrypt(k, v)
logg.debug('>>>>>>>>>>>>< v decoded {}'.format(v))
v = v.decode('utf-8')
logg.debug('retrieved {} from {}'.format(k, p)) logg.debug('retrieved {} from {}'.format(k, p))
return r.strip() return v.strip()
def by_phone(self, phone, update=False): def by_phone(self, phone, update=False):