# -*- coding: utf-8 -*-
#
# Copyright (C) 2021-2022 Red Dove Consultants Limited
#
"""
This module supports key management, encryption/decryption and signing/verification
using age and minisign.
"""
import base64
import functools
import hashlib
import json
import logging
import os
from pathlib import Path
import re
import shutil
import subprocess
import sys
import tempfile
import threading
__version__ = '0.1.1.dev0'
__author__ = 'Vinay Sajip'
__date__ = "$05-Dec-2021 12:39:53$"
if sys.version_info[:2] < (3, 6): # pragma: no cover
raise ImportError('This module requires Python >= 3.6 to run.')
logger = logging.getLogger(__name__)
__all__ = [
'Identity', 'CryptException', 'remove_identities', 'clear_identities',
'list_identities', 'encrypt', 'decrypt', 'encrypt_mem', 'decrypt_mem',
'sign', 'verify', 'encrypt_and_sign', 'verify_and_decrypt'
]
class CryptException(Exception):
"""
Base class of all exceptions defined in this module.
"""
pass
if os.name == 'nt':
PAGESIGN_DIR = os.path.join(os.environ['LOCALAPPDATA'], 'pagesign')
else:
PAGESIGN_DIR = os.path.expanduser('~/.pagesign')
CREATED_PATTERN = re.compile('# created: (.*)', re.I)
APK_PATTERN = re.compile('# public key: (.*)', re.I)
ASK_PATTERN = re.compile(r'AGE-SECRET-KEY-.*')
MPI_PATTERN = re.compile(r'minisign public key (\S+)')
if not os.path.exists(PAGESIGN_DIR): # pragma: no cover
os.makedirs(PAGESIGN_DIR)
if not os.path.isdir(PAGESIGN_DIR): # pragma: no cover
raise ValueError('%s exists but is not a directory.' % PAGESIGN_DIR)
os.chmod(PAGESIGN_DIR, 0o700)
def _load_keys():
result = {}
p = os.path.join(PAGESIGN_DIR, 'keys')
if os.path.exists(p): # pragma: no branch
with open(p, encoding='utf-8') as f:
result = json.load(f)
return result
def _save_keys(keys):
p = os.path.join(PAGESIGN_DIR, 'keys')
with open(p, 'w', encoding='utf-8') as f:
json.dump(keys, f, indent=2, sort_keys=True)
os.chmod(p, 0o600)
KEYS = _load_keys()
PUBLIC_ATTRS = ('created', 'crypt_public', 'sign_public', 'sign_id')
ATTRS = PUBLIC_ATTRS + ('crypt_secret', 'sign_secret', 'sign_pass')
def clear_identities():
"""
Clear all identities saved locally.
"""
if len(KEYS):
KEYS.clear()
_save_keys(KEYS)
def remove_identities(*args):
"""
Remove the identities stored locally whose names are in *args*. Names are
case-sensitive.
Args:
args (list[str]): The list of identities to remove.
"""
changed = False
for name in args:
if name in KEYS:
del KEYS[name]
changed = True
if changed:
_save_keys(KEYS)
def list_identities():
"""
Return an iterator over the locally stored identities, as name-value 2-tuples.
"""
return KEYS.items()
def _make_password(length):
return base64.b64encode(os.urandom(length)).decode('ascii')
def _read_out(stream, result, key='stdout'):
data = b''
while True:
c = stream.read1(100)
if not c:
break
data += c
result[key] = data
def _run_command(cmd, wd, err_reader=None, decode=True):
# print('Running: %s' % (cmd if isinstance(cmd, str) else ' '.join(cmd)))
# if cmd[0] == 'age': import pdb; pdb.set_trace()
if not isinstance(cmd, list):
cmd = cmd.split()
logger.debug('Running: %s' % cmd)
kwargs = {'cwd': wd, 'stdout': subprocess.PIPE, 'stderr': subprocess.PIPE}
if err_reader:
kwargs['stdin'] = subprocess.PIPE
p = subprocess.Popen(cmd, **kwargs)
if err_reader is None:
stdout, stderr = p.communicate()
else:
data = {}
rout = threading.Thread(target=_read_out, args=(p.stdout, data))
rout.daemon = True
rout.start()
rerr = threading.Thread(target=err_reader,
args=(p.stderr, p.stdin, data))
rerr.daemon = True
rerr.start()
rout.join()
rerr.join()
p.wait()
stdout = data['stdout']
stderr = data['stderr']
p.stdout.close()
p.stderr.close()
if p.returncode == 0:
if decode:
stdout = stdout.decode('utf-8')
stderr = stderr.decode('utf-8')
return stdout, stderr
else: # pragma: no cover
raise subprocess.CalledProcessError(p.returncode,
p.args,
output=stdout,
stderr=stderr)
def _get_work_file(**kwargs):
fd, result = tempfile.mkstemp(**kwargs)
os.close(fd)
return result
def _shred(path, delete=True):
size = os.stat(path).st_size
passes = 2
with open(path, 'wb') as f:
for i in range(passes):
if i > 0:
f.seek(0)
f.write(os.urandom(size))
if delete:
os.remove(path)
[docs]
class Identity:
"""
This class represents both remote identities (used for encryption and verification
only) and local identities (used for all functions - encryption, decryption,
signing and verification).
"""
encoding = 'utf-8'
[docs]
def __init__(self, name=None):
"""
Either retrieve an existing identity named *name*, or, if not specified, create
a new local identity which can later be named using its :meth:`save` method.
Names are case-sensitive.
"""
if name:
if name in KEYS:
self.__dict__.update(KEYS[name])
else: # pragma: no cover
raise ValueError('No such identity: %r' % name)
else:
# Generate a new identity
wd = tempfile.mkdtemp(dir=PAGESIGN_DIR, prefix='work-')
try:
p = os.path.join(wd, 'age-key')
cmd = 'age-keygen -o %s' % p
try:
_run_command(cmd, wd)
self._parse_age_file(p)
for name in ('created', 'crypt_public', 'crypt_secret'):
getattr(self, name) # ensure the attribute is there
except Exception as e: # pragma: no cover
raise CryptException(
'Identity creation failed (crypt)') from e
finally:
# the whole working directory will get removed, so pass False
# to _shred as we don't need to delete the file now
# (same logic applies to _shred calls below)
_shred(p, False)
sfn = _get_work_file(prefix='msk-', dir=wd)
pfn = _get_work_file(prefix='mpk-', dir=wd)
self.sign_pass = _make_password(12)
cmd = 'minisign -fG -p %s -s %s' % (pfn, sfn)
try:
_run_command(cmd, wd, self._read_minisign_gen_err)
self._parse_minisign_file(pfn)
for name in ('sign_id', 'sign_public'):
getattr(self, name) # ensure the attribute is there
self.sign_secret = Path(sfn).read_text(self.encoding)
except Exception as e: # pragma: no cover
raise CryptException(
'Identity creation failed (sign)') from e
finally:
_shred(pfn, False)
_shred(sfn, False)
finally:
shutil.rmtree(wd)
for attr in ATTRS:
assert hasattr(self, attr)
def _parse_age_file(self, fn):
with open(fn, encoding=self.encoding) as f:
lines = f.read().splitlines()
for line in lines:
m = CREATED_PATTERN.match(line)
if m:
self.created = m.groups()[0]
continue
m = APK_PATTERN.match(line)
if m:
self.crypt_public = m.groups()[0]
continue
m = ASK_PATTERN.match(line)
assert m, 'Secret key line not seen'
self.crypt_secret = line
def _parse_minisign_file(self, fn):
with open(fn, encoding=self.encoding) as f:
lines = f.read().splitlines()
for line in lines:
m = MPI_PATTERN.search(line)
if m:
self.sign_id = m.groups()[0]
else:
self.sign_public = line
[docs]
def save(self, name):
"""
Save this instance with the specified *name*, which cannot be blank or
``None``. Names are case-sensitive.
"""
if not name or not isinstance(name, str): # pragma: no cover
raise ValueError('Invalid name: %r' % name)
d = dict(self.__dict__)
# might need to remove some attrs from d here ...
KEYS[name] = d
_save_keys(KEYS)
def _read_minisign_gen_err(self, stream, stdin, result):
data = b''
pwd = (self.sign_pass + os.linesep).encode('ascii')
pwd_written = 0
sep = os.linesep.encode('ascii')
prompt1 = b'Password: '
prompt2 = prompt1 + sep + b'Password (one more time): '
prompts = (prompt1, prompt2)
while True:
c = stream.read1(100)
data += c
# print('err: %s' % data)
if data in prompts:
stdin.write(pwd)
stdin.flush()
pwd_written += 1
# print('Wrote pwd')
if pwd_written == 2:
stdin.close()
break
result['stderr'] = data
def _read_minisign_sign_err(self, stream, stdin, result):
data = b''
pwd = (self.sign_pass + os.linesep).encode('ascii')
while True:
c = stream.read(1)
data += c
# print('err: %s' % data)
if data == b'Password: ':
stdin.write(pwd)
stdin.close()
break
result['stderr'] = data
[docs]
def export(self):
"""
Export this instance. Only public attributes are preserved in the export
--- it is meant for sending to someone securely.
Returns:
dict: A dictionary containing the exportable items of the instance.
"""
d = dict(self.__dict__)
for k in self.__dict__:
if '_secret' in k or '_pass' in k:
del d[k]
return d
[docs]
@classmethod
def imported(cls, d, name):
"""
Return a remote identity instance created from *d* and with local name *name*.
Args:
d (dict): A dictionary from some external source. It must contain the public
attributes *created*, *crypt_public*, *sign_public* and
*sign_id* (which will be present in dictionaries created
using the :meth:`export` method)
name (str): A name against which to save the imported information.
Note that names are case-sensitive.
Returns:
Identity: The saved identity constructed from *d*.
"""
result = object.__new__(cls)
for k in PUBLIC_ATTRS:
try:
setattr(result, k, d[k])
except KeyError: # pragma: no cover
logger.warning('Attribute absent: %s', k)
result.save(name)
return result
def _get_encryption_command(recipients, armor):
if not recipients: # pragma: no cover
raise ValueError('At least one recipient needs to be specified.')
result = ['age', '-e']
if armor:
result.append('-a')
if isinstance(recipients, str):
recipients = [recipients]
if not isinstance(recipients, (list, tuple)): # pragma: no cover
raise ValueError('invalid recipients: %s' % recipients)
for r in recipients:
if r not in KEYS: # pragma: no cover
raise ValueError('No such recipient: %s' % r)
info = KEYS[r]
result.extend(['-r', info['crypt_public']])
return result
[docs]
def encrypt(path, recipients, outpath=None, armor=False):
"""
Encrypt the file at *path* for identities whose names are in *recipients* and
save the encrypted data in *outpath*. The output data is ASCII-armored if *armor*
is true, else it is binary.
Args:
path (str): The path to the data to be encrypted.
recipients (str|list[str]): The name(s) of the identities of the recipient(s)
of the data.
outpath (str): The path to which the encrypted data should be written.
If not specified, it will be set to *path* with
``'.age'`` appended.
armor (bool): Whether the output is to be ASCII-armored.
Returns:
str: The value of *outpath* is returned.
"""
if not os.path.isfile(path): # pragma: no cover
raise ValueError('No such file: %s' % path)
if outpath is None:
outpath = '%s.age' % path
else:
d = os.path.dirname(outpath)
if not os.path.exists(d): # pragma: no cover
os.makedirs(d)
elif not os.path.isdir(d): # pragma: no cover
raise ValueError('Not a directory: %s' % d)
# if dir, assume writeable, for now
cmd = _get_encryption_command(recipients, armor)
cmd.extend(['-o', outpath])
cmd.append(path)
try:
_run_command(cmd, os.getcwd())
return outpath
except subprocess.CalledProcessError as e: # pragma: no cover
raise CryptException('Encryption failed') from e
def _data_writer(data, stream, stdin, result):
stdin.write(data)
stdin.close()
_read_out(stream, result, 'stderr')
[docs]
def encrypt_mem(data, recipients, armor=False):
"""
Encrypt the in-memory *data* for identities whose names are in *recipients*. The
output data is ASCII-armored if *armor* is true, else it is binary. The encrypted
data is returned as bytes.
Args:
data (str|bytes): The data to be encrypted.
recipients (str|list[str]): The name(s) of the identities of the
recipient(s) of the data.
armor (bool): Whether the output is to be ASCII-armored.
Returns:
bytes: The encrypted data.
"""
cmd = _get_encryption_command(recipients, armor)
if isinstance(data, str):
data = data.encode('utf-8')
if not isinstance(data, bytes): # pragma: no cover
raise TypeError('invalid data: %s' % data)
err_reader = functools.partial(_data_writer, data)
try:
stdout, stderr = _run_command(cmd, os.getcwd(), err_reader, False)
return stdout
except subprocess.CalledProcessError as e: # pragma: no cover
raise CryptException('Encryption failed') from e
def _get_decryption_command(identities):
if not identities: # pragma: no cover
raise ValueError('At least one identity needs to be specified.')
cmd = ['age', '-d']
if isinstance(identities, str):
identities = [identities]
if not isinstance(identities, (list, tuple)): # pragma: no cover
raise ValueError('invalid identities: %s' % identities)
fn = _get_work_file(dir=PAGESIGN_DIR, prefix='ident-')
ident_values = []
for ident in identities:
if ident not in KEYS: # pragma: no cover
raise ValueError('No such identity: %s' % ident)
ident_values.append(KEYS[ident]['crypt_secret'])
with open(fn, 'w', encoding='utf-8') as f:
f.write('\n'.join(ident_values))
cmd.extend(['-i', fn])
return cmd, fn
[docs]
def decrypt(path, identities, outpath=None):
"""
Decrypt the data at *path* which is intended for recipients named in *identities*
and save the decrypted data at *outpath*.
Args:
path (str): The path to the data to be decrypted.
identities (str|list[str]): The name(s) of the recipient(s) of the data.
outpath (str): The path to which the decrypted data should be written.
If not specified and *path* ends with ``'.age'``, then
*outpath* will be set to *path* with that suffix
stripped. Otherwise, it will be set to *path* with
``'.dec'`` appended.
Returns:
str: The value of *outpath* is returned.
"""
if not os.path.isfile(path): # pragma: no cover
raise ValueError('No such file: %s' % path)
if outpath is None:
if path.endswith('.age'):
outpath = path[:-4]
else:
outpath = '%s.dec' % path
else:
d = os.path.dirname(outpath)
if not os.path.exists(d): # pragma: no cover
os.makedirs(d)
elif not os.path.isdir(d): # pragma: no cover
raise ValueError('Not a directory: %s' % d)
# if dir, assume writeable, for now
cmd, fn = _get_decryption_command(identities)
# import pdb; pdb.set_trace()
try:
cmd.extend(['-o', outpath])
cmd.append(path)
_run_command(cmd, os.getcwd())
return outpath
except subprocess.CalledProcessError as e: # pragma: no cover
raise CryptException('Decryption failed') from e
finally:
_shred(fn)
[docs]
def decrypt_mem(data, identities):
"""
Decrypt the in-memory *data* for recipients whose names are in *identities*.
Args:
data (str|bytes): The data to decrypt.
identities (str|list[str]): The name(s) of the identities of the
recipient(s) of the data.
Returns:
bytes: The decrypted data.
"""
cmd, fn = _get_decryption_command(identities)
if isinstance(data, str): # pragma: no cover
data = data.encode('utf-8')
if not isinstance(data, bytes): # pragma: no cover
raise TypeError('invalid data: %s' % data)
err_reader = functools.partial(_data_writer, data)
try:
stdout, stderr = _run_command(cmd, os.getcwd(), err_reader, False)
return stdout
except subprocess.CalledProcessError as e: # pragma: no cover
raise CryptException('Decryption failed') from e
finally:
_shred(fn)
[docs]
def sign(path, identity, outpath=None):
"""
Sign the data at *path* with the named *identity* and save the signature in
*outpath*.
Args:
path (str): The path to the data to be signed.
identity (str): The name of the signer's identity.
outpath (str): The path to which the signature is to be written. If not
specified, *outpath* is set to *path* with ``'.sig'``
appended.
Returns:
str: The value of *outpath* is returned.
"""
if not identity: # pragma: no cover
raise ValueError('An identity needs to be specified.')
if identity not in KEYS: # pragma: no cover
raise ValueError('No such identity: %s' % identity)
ident = Identity(identity)
if not os.path.isfile(path): # pragma: no cover
raise ValueError('No such file: %s' % path)
if outpath is None:
outpath = '%s.sig' % path
else:
d = os.path.dirname(outpath)
if not os.path.exists(d): # pragma: no cover
os.makedirs(d)
elif not os.path.isdir(d): # pragma: no cover
raise ValueError('Not a directory: %s' % d)
# if dir, assume writeable, for now
fd, fn = tempfile.mkstemp(dir=PAGESIGN_DIR, prefix='seckey-')
os.write(fd, (KEYS[identity]['sign_secret'] + os.linesep).encode('ascii'))
os.close(fd)
try:
cmd = ['minisign', '-S', '-x', outpath, '-s', fn, '-m', path]
_run_command(cmd, os.getcwd(), ident._read_minisign_sign_err)
except Exception as e: # pragma: no cover
raise CryptException('Signing failed') from e
finally:
_shred(fn)
return outpath
[docs]
def verify(path, identity, sigpath=None):
"""
Verify that the data at *path* was signed with the identity named *identity*,
where the signature is at *sigpath*. If verification fails, an exception is
raised, otherwise this function returns `None`.
Args:
path (str): The path to the data to be verified.
identity (str): The name of the signer's identity.
sigpath (str): The path where the signature is stored. If not specified,
*sigpath* is set to *path* with `'.sig'` appended.
"""
if not identity: # pragma: no cover
raise ValueError('An identity needs to be specified.')
if identity not in KEYS: # pragma: no cover
raise ValueError('No such identity: %s' % identity)
ident = Identity(identity)
if not os.path.isfile(path): # pragma: no cover
raise ValueError('No such file: %s' % path)
if sigpath is None:
sigpath = '%s.sig' % path
if not os.path.isfile(sigpath): # pragma: no cover
raise ValueError('No such file: %s' % sigpath)
cmd = [
'minisign', '-V', '-x', sigpath, '-P', ident.sign_public, '-m', path
]
# import pdb; pdb.set_trace()
try:
_run_command(cmd, os.getcwd())
except subprocess.CalledProcessError as e: # pragma: no cover
raise CryptException('Verification failed') from e
def _get_b64(path):
with open(path, 'rb') as f:
return base64.b64encode(f.read()).decode('ascii')
[docs]
def encrypt_and_sign(path,
recipients,
signer,
armor=False,
outpath=None,
sigpath=None):
"""
Encrypt the data at *path* for identities named in *recipients* and sign it with
the identity named by *signer*. Write the encrypted data to *outpath* and
the signature to *sigpath*.
Note that you'll need to call :func:`verify_and_decrypt` to reverse this process.
Args:
path (str): The path to the data to be decrypted.
recipients (str|list[str]): The name(s) of the identities of the
recipient(s) of the encrypted data.
signer (str): The name of the signer identity.
armor (bool): If `True`, use ASCII armor for the encrypted data, else
save it as binary.
outpath (str): The output path to which the encrypted data should be
written, If not specified, it will be set to *path*
with ``'.age'`` appended.
sigpath (str): The path to which the signature should be written. If not
specified, *sigpath* is set to *outpath* with ``'.sig'``
appended.
Returns:
tuple(str, str): A tuple of *outpath* and *sigpath* is returned.
"""
if not recipients or not signer: # pragma: no cover
raise ValueError(
'At least one recipient (and one signer) needs to be specified.')
if not os.path.isfile(path): # pragma: no cover
raise ValueError('No such file: %s' % path)
naive = False
if naive: # pragma: no cover
outpath = encrypt(path, recipients, outpath=outpath, armor=armor)
sigpath = sign(outpath, signer, outpath=sigpath)
return outpath, sigpath
else:
# Use a sign/encrypt/sign strategy:
# 1. Sign the plaintext.
# 2. Construct a JSON of the base64-encoded plaintext and signature.
# 3. Encrypt that.
# 4. Hash all the recipient public keys into a list.
# 5. Construct a JSON of the encrypted data and recipient hashes.
# 6. Sign that.
fn = _get_work_file(dir=PAGESIGN_DIR, prefix='sig-')
sigpath = sign(path, signer, fn)
inner = {'plaintext': _get_b64(path), 'signature': _get_b64(sigpath)}
os.remove(sigpath)
data = json.dumps(inner).encode('ascii')
encrypted = encrypt_mem(data, recipients, armor)
if not armor:
encrypted = base64.b64encode(encrypted)
if isinstance(recipients, str):
recipients = [recipients]
# if we encrypted OK, there can't have been problems with the recipients
hashes = []
for r in recipients:
info = KEYS[r]
pk = info['crypt_public'].encode('ascii')
hashes.append(hashlib.sha256(pk).hexdigest())
outer = {
'encrypted': encrypted.decode('ascii'),
'armored': armor,
'recipients': hashes
}
data = json.dumps(outer).encode('ascii')
outpath = _get_work_file(dir=PAGESIGN_DIR, prefix='message-')
Path(outpath).write_bytes(data)
sigpath = sign(outpath, signer)
return outpath, sigpath
[docs]
def verify_and_decrypt(path, recipients, signer, outpath=None, sigpath=None):
"""
Verify the encrypted and signed data at *path* as having been signed by the
identity named by *signer* and intended for identities named in *recipients*.
The signature for *path* is in *sigpath*. If not specified, it will be set to
*path* with ``'.sig'`` appended. If verification or decryption fails, an exception
will be raised. Otherwise, the decrypted data will be stored at *outpath*. If
not specified, it will be set to *path* with the suffix stripped (if it ends in
``'.age'``) or with ``'.dec'`` appended.
The function returns *outpath*.
Note that the file inputs to this function should have been created using
:func:`encrypt_and_sign`.
Args:
path (str): The path to the encrypted and signed data.
recipients (str|list[str]): The name(s) of the recipient(s) of the encrypted data.
signer (str): The name of the signer identity.
outpath (str): The output path to which the decrypted data should be written,
sigpath (str): The path in which the signature is to be found.
Returns:
str: The value of outpath is returned.
"""
if not signer or not recipients: # pragma: no cover
raise ValueError(
'At least one recipient (and one signer) needs to be specified.')
if not os.path.isfile(path): # pragma: no cover
raise ValueError('No such file: %s' % path)
if sigpath is None: # pragma: no cover
sigpath = path + '.sig'
if not os.path.exists(sigpath): # pragma: no cover
raise ValueError('no such file: %s' % sigpath)
verify(path, signer, sigpath)
naive = False
if naive: # pragma: no cover
return decrypt(path, recipients, outpath)
else:
with open(path, 'r', encoding='ascii') as f:
outer = json.load(f)
encrypted = outer['encrypted'].encode('ascii')
if not outer['armored']:
encrypted = base64.b64decode(encrypted)
hashes = set(outer['recipients'])
if isinstance(recipients, str):
recipients = [recipients]
for r in recipients:
if r not in KEYS: # pragma: no cover
raise ValueError('No such recipient: %s' % r)
info = KEYS[r]
pk = info['crypt_public'].encode('ascii')
h = hashlib.sha256(pk).hexdigest()
if h not in hashes: # pragma: no cover
raise ValueError('Not a valid recipient: %s' % r)
decrypted = decrypt_mem(encrypted, recipients).decode('ascii')
inner = json.loads(decrypted)
fd, outpath = tempfile.mkstemp(dir=PAGESIGN_DIR, prefix='msg-')
os.write(fd, base64.b64decode(inner['plaintext'].encode('ascii')))
os.close(fd)
sigpath = outpath + '.sig'
Path(sigpath).write_bytes(
base64.b64decode(inner['signature'].encode('ascii')))
verify(outpath, signer, sigpath)
os.remove(sigpath)
return outpath