This repository has been archived on 2022-01-23. You can view files and clone it, but cannot push or open issues or pull requests.
vaultpass/vaultpass/mounts.py
2020-04-02 14:55:26 -04:00

219 lines
9.4 KiB
Python

import logging
import re
import warnings
##
import dpath.util # https://pypi.org/project/dpath/
import hvac.exceptions
_logger = logging.getLogger()
_mount_re = re.compile(r'^(?P<mount>.*)/$')
_subpath_re = re.compile(r'^/?(?P<path>.*)/$')
class CubbyHandler(object):
# There is no upstream support for directly reading cubby. So we do it ourselves.
# TODO: custom class/handler? https://hvac.readthedocs.io/en/stable/advanced_usage.html#custom-requests-http-adapter
def __init__(self, client):
self.client = client
def list_secrets(self, path, mount_point = 'cubbyhole'):
path = path.lstrip('/')
uri = '/v1/{0}/{1}'.format(mount_point, path)
resp = self.client._adapter.list(url = uri)
return(resp.json())
def read_secret(self, path, mount_point = 'cubbyhole'):
path = path.lstrip('/')
uri = '/v1/{0}/{1}'.format(mount_point, path)
resp = self.client._adapter.get(url = uri)
return(resp.json())
class MountHandler(object):
internal_mounts = ('identity', 'sys')
def __init__(self, client, mounts_xml = None):
self.client = client
self.cubbyhandler = CubbyHandler(self.client)
self.xml = mounts_xml
self.mounts = {}
self.paths = {}
self.getSysMounts()
def getMountType(self, mount):
if not self.mounts:
self.getSysMounts()
mtype = self.mounts.get(mount)
if not mtype:
_logger.error('Mount not found in defined mounts')
_logger.debug('The mount {0} was not found in the defined mounts.'.format(mount))
raise ValueError('Mount not found in defined mounts')
return(mtype)
def getSecret(self, path, mount, version = None):
pass
def getSecretNames(self, path, mount, version = None):
reader = None
mtype = self.getMountType(mount)
secrets_list = []
keypath = ['data']
args = {'path': path,
'mount_point': mount}
if mtype == 'cubbyhole':
reader = self.cubbyhandler.read_secret
elif mtype == 'kv1':
reader = self.client.secrets.kv.v1.read_secret
elif mtype == 'kv2':
if not any(((version is None), isinstance(version, int))):
_logger.error('version parameter must be an integer or None')
_logger.debug('The version parameter ({0}) must be an integer or None'.format(version))
raise ValueError('version parameter must be an integer or None')
reader = self.client.secrets.kv.v2.read_secret_version
args['version'] = version
keypath = ['data', 'data']
data = reader(**args)
try:
# secrets_list = list(data.get('data', {}).keys())
secrets_list = list(dpath.util.get(data, keypath, {}).keys())
except (KeyError, TypeError):
secrets_list = []
return(secrets_list)
def getSecretsTree(self, path = '/', mounts = None, version = None):
if not mounts:
mounts = self.mounts
if isinstance(mounts, dict):
mounts = list(mounts.keys())
if not isinstance(mounts, list):
mounts = [mounts]
for mount in mounts:
mtype = self.getMountType(mount)
handler = None
args = {'path': path,
'mount_point': mount}
relpath = path.replace('//', '/').lstrip('/')
fullpath = '/'.join((mount, relpath)).replace('//', '/').lstrip('/')
if mtype == 'cubbyhole':
handler = self.cubbyhandler
elif mtype == 'kv1':
handler = self.client.secrets.kv.v1
elif mtype == 'kv2':
if not any(((version is None), isinstance(version, int))):
_logger.error('version parameter must be an integer or None')
_logger.debug('The version parameter ({0}) must be an integer or None'.format(version))
raise ValueError('version parameter must be an integer or None')
handler = self.client.secrets.kv.v2
if mount not in self.paths.keys():
self.paths[mount] = {}
try:
_logger.debug('Fetching path {0} on mount {1}...'.format(path, mount))
paths = handler.list_secrets(**args)
except hvac.exceptions.InvalidPath:
# It's a secret name or doesn't exist.
_logger.debug('Path {0} on mount {1} is a secret, not a subdir.'.format(path, mount))
dpath.util.new(self.paths, fullpath, self.getSecretNames(path, mount, version = version))
continue
# if 'data' not in paths.keys() or 'keys' not in paths['data'].keys():
try:
paths_list = paths['data']['keys']
except (KeyError, TypeError):
_logger.warning('Mount has no secrets/subdirs')
_logger.debug('The mount {0} has no secrets or subdirectories'.format(mount))
warnings.warn('Mount has no secrets/subdirs')
continue
for p in paths_list:
p_relpath = '/'.join((relpath, p)).replace('//', '/').lstrip('/')
p_fullpath = '/'.join((fullpath, p)).replace('//', '/').lstrip('/')
_logger.debug(('Recursing getSecretsTree. '
'path={0} '
'fullpath={1} '
'relpath={2} '
'p={3} '
'p_relpath={4} '
'p_fullpath={5}').format(path,
fullpath,
relpath,
p,
p_relpath,
p_fullpath))
self.getSecretsTree(path = p_relpath, mounts = mount)
return(None)
def getSysMounts(self):
try:
for mount, mount_info in self.client.sys.list_mounted_secrets_engines()['data'].items():
r = _mount_re.search(mount)
if r:
mount = r.group('mount')
if mount in self.internal_mounts:
continue
# Get the mount type.
mtype = mount_info['type']
if mtype == 'kv':
mntopts = mount_info['options']
if mntopts and isinstance(mntopts, dict):
mver = mntopts.get('version')
if mver == '2':
mtype = 'kv2'
elif mver == '1':
mtype = 'kv1'
self.mounts[mount] = mtype
_logger.debug('Added mountpoint {0} to mounts list with type {1}'.format(mount, mtype))
except hvac.exceptions.Forbidden:
_logger.warning('Client does not have permission to read /sys/mounts.')
# TODO: should I blindly merge in instead?
if self.xml:
for mount in self.xml.findall('.//mount'):
mname = mount.text
mtype = mount.attrib.get('type', 'kv2')
if mname not in self.mounts.keys():
self.mounts[mname] = mtype
_logger.debug('Added mountpoint {0} to mounts list with type {1}'.format(mount, mtype))
return(None)
def printer(self, output = None, indent = 4):
def treePrint(obj, s = 'Password Store\n', level = 0):
prefix = '├──'
leading_prefix = ''
last_prefix = '└──'
pass
return(s)
if output:
output = output.lower()
if output and output not in ('pretty', 'yaml', 'json'):
# if output and output not in ('pretty', 'yaml', 'json', 'tree'):
_logger.error('Invalid output format')
_logger.debug('The output parameter ("{0}") must be one of: pretty, yaml, json, or None'.format(output))
# _logger.debug(('The output parameter ("{0}") must be one of: '
# 'pretty, yaml, json, tree, or None').format(output))
raise ValueError('Invalid output format')
if output in ('pretty', 'yaml', 'json'):
if not any(((indent is None), isinstance(indent, int))):
_logger.error('indent parameter must be an integer or None')
_logger.debug('The indent parameter ({0}) must be an integer or None'.format(indent))
raise ValueError('indent parameter must be an integer or None')
if not self.paths:
self.getSecretsTree()
if output == 'json':
import json
return(json.dumps(self.paths, indent = indent))
elif output == 'yaml':
import yaml # https://pypi.org/project/PyYAML/
# import pyaml # https://pypi.python.org/pypi/pyaml
return(yaml.dump(self.paths, indent = indent))
elif output == 'pretty':
import pprint
if indent is None:
indent = 1
return(pprint.pformat(self.paths, indent = indent))
# elif output == 'tree':
# import tree # TODO? Wayyy later.
elif not output:
return(str(self.paths))
return(None)
def search(self):
pass