some modifications - VaultPass GPG-encrypted creds are almost working.
This commit is contained in:
@@ -16,7 +16,6 @@ class PassMan(object):
|
||||
|
||||
def __init__(self, cfg = '~/.config/vaultpass.xml'):
|
||||
self.cfg = config.getConfig(cfg)
|
||||
self.cfg.main()
|
||||
self._getURI()
|
||||
self.getClient()
|
||||
|
||||
@@ -29,6 +28,8 @@ class PassMan(object):
|
||||
def getClient(self):
|
||||
# This may need to be re-tooled in the future.
|
||||
auth_xml = self.cfg.xml.find('auth')
|
||||
if auth_xml is None:
|
||||
raise RuntimeError('Could not find authentication')
|
||||
authmethod_xml = auth_xml.getchildren()[0]
|
||||
for a in dir(auth):
|
||||
if a.startswith('_'):
|
||||
|
||||
@@ -128,6 +128,7 @@ class Token(_AuthBase):
|
||||
self.token = self._getEnv(e)
|
||||
else:
|
||||
self.token = self._getFile(a)
|
||||
self.client = hvac.Client(url = self.uri)
|
||||
self.client.token = self.token
|
||||
self.authCheck()
|
||||
return(None)
|
||||
|
||||
@@ -3,6 +3,7 @@ import os
|
||||
import logging
|
||||
import re
|
||||
##
|
||||
from . import gpg_handler
|
||||
import requests
|
||||
from lxml import etree
|
||||
|
||||
@@ -12,6 +13,8 @@ _logger = logging.getLogger()
|
||||
|
||||
|
||||
class Config(object):
|
||||
gpg = None
|
||||
gpg_elems = ('authGpg', 'unsealGpg')
|
||||
xsd_path = None
|
||||
tree = None
|
||||
namespaced_tree = None
|
||||
@@ -31,6 +34,31 @@ class Config(object):
|
||||
self.populateDefaults()
|
||||
if validate:
|
||||
self.validate()
|
||||
g = self.parseGpg()
|
||||
if g:
|
||||
# And do it again.
|
||||
if populate_defaults:
|
||||
self.populateDefaults()
|
||||
if validate:
|
||||
self.validate()
|
||||
return(None)
|
||||
|
||||
def decryptGpg(self, gpg_xml):
|
||||
home = gpg_xml.attrib.get('gpgHome')
|
||||
tag = gpg_xml.tag
|
||||
ns_xml = self.xml.find(tag)
|
||||
xml = self.stripNS(obj = ns_xml).tag
|
||||
fpath = gpg_xml.text
|
||||
if not self.gpg:
|
||||
self.gpg = gpg_handler.GPG(home = home)
|
||||
else:
|
||||
self.gpg.gpg.home = home
|
||||
self.gpg.initHome()
|
||||
ns_dcrpt_xml = etree.fromstring(self.gpg.decrypt(fpath))
|
||||
dcrpt_xml = self.stripNS(obj = ns_dcrpt_xml)
|
||||
ns_xml.getparent().replace(ns_xml, ns_dcrpt_xml)
|
||||
xml.getparent().replace(xml, dcrpt_xml)
|
||||
self.parse()
|
||||
return(None)
|
||||
|
||||
def fetch(self): # Just a fail-safe; this is overridden by specific subclasses.
|
||||
@@ -87,11 +115,8 @@ class Config(object):
|
||||
_logger.info('Rendered XSD.')
|
||||
return(None)
|
||||
|
||||
def parseRaw(self, parser = None):
|
||||
self.xml = etree.fromstring(self.raw, parser = parser)
|
||||
_logger.debug('Generated xml.')
|
||||
self.namespaced_xml = etree.fromstring(self.raw, parser = parser)
|
||||
_logger.debug('Generated namespaced xml.')
|
||||
def parse(self):
|
||||
# This can used to "re-parse" the self.xml and self.namespaced_xml.
|
||||
self.tree = self.xml.getroottree()
|
||||
_logger.debug('Generated tree.')
|
||||
self.namespaced_tree = self.namespaced_xml.getroottree()
|
||||
@@ -103,6 +128,26 @@ class Config(object):
|
||||
self.stripNS()
|
||||
return(None)
|
||||
|
||||
def parseGpg(self):
|
||||
gpg_elem_found = False # Change to True if we find any GPG-encrypted elems
|
||||
search = []
|
||||
for x in self.gpg_elems:
|
||||
search.append("local-name()='{0}'".format(x))
|
||||
search = '[{0}]'.format(' or '.join(search))
|
||||
print(search)
|
||||
gpg_elems = self.namespaced_xml.findall('|'.join(search))
|
||||
for e in gpg_elems:
|
||||
print(e)
|
||||
return(gpg_elem_found)
|
||||
|
||||
def parseRaw(self, parser = None):
|
||||
self.xml = etree.fromstring(self.raw, parser = parser)
|
||||
_logger.debug('Generated xml.')
|
||||
self.namespaced_xml = etree.fromstring(self.raw, parser = parser)
|
||||
_logger.debug('Generated namespaced xml.')
|
||||
self.parse()
|
||||
return(None)
|
||||
|
||||
def populateDefaults(self):
|
||||
_logger.info('Populating missing values with defaults from XSD.')
|
||||
if not self.xsd:
|
||||
|
||||
39
vaultpass/gpg_handler.py
Normal file
39
vaultpass/gpg_handler.py
Normal file
@@ -0,0 +1,39 @@
|
||||
import io
|
||||
import logging
|
||||
import os
|
||||
##
|
||||
import gpg # https://pypi.org/project/gpg/
|
||||
|
||||
|
||||
_logger = logging.getLogger()
|
||||
|
||||
|
||||
class GPG(object):
|
||||
home = None
|
||||
gpg = None
|
||||
|
||||
def __init__(self, home = None):
|
||||
if home:
|
||||
self.home = home
|
||||
self.initHome()
|
||||
|
||||
def decrypt(self, fpath):
|
||||
fpath = os.path.abspath(os.path.expanduser(fpath))
|
||||
with open(fpath, 'rb') as fh:
|
||||
iobuf = io.BytesIO(fh.read())
|
||||
iobuf.seek(0, 0)
|
||||
rslt = self.gpg.decrypt(iobuf)
|
||||
decrypted = rslt[0]
|
||||
return(decrypted)
|
||||
|
||||
def initHome(self):
|
||||
if not self.home:
|
||||
h = os.environ.get('GNUPGHOME')
|
||||
if h:
|
||||
self.home = h
|
||||
if self.home:
|
||||
self.home = os.path.abspath(os.path.expanduser(self.home))
|
||||
if not os.path.isdir(self.home):
|
||||
raise ValueError('GPG home does not exist')
|
||||
_logger.debug('Set GPG home to explicitly specified value {0}'.format(self.home))
|
||||
return(None)
|
||||
@@ -1,16 +0,0 @@
|
||||
import os
|
||||
import logging
|
||||
##
|
||||
import gpg
|
||||
|
||||
|
||||
# Special shoutout to Jthan for ruining my life.
|
||||
|
||||
|
||||
_logger = logging.getLogger()
|
||||
|
||||
|
||||
class GPGAuth(object):
|
||||
def __init__(self, gpgauth_xml):
|
||||
pass
|
||||
|
||||
Reference in New Issue
Block a user