homeserver/plugins/lookup/keepass.py

129 lines
4.8 KiB
Python
Raw Normal View History

# -*- coding: utf-8 -*-
from __future__ import (absolute_import, division, print_function)
__metaclass__ = type
try:
from __main__ import display
except ImportError:
from ansible.utils.display import Display
display = Display()
import os
import json
import socket
import tempfile
from pykeepass import PyKeePass
from construct.core import ChecksumError
from ansible.errors import AnsibleError
from ansible.plugins.lookup import LookupBase
DOCUMENTATION = """
lookup: keepass
author: Victor Zemtsov <victor.zemtsov@gmail.com>
version_added: '0.2'
short_description: fetch data from KeePass file
description:
2021-01-27 08:41:58 +01:00
- This lookup returns a value of a property of a KeePass entry
- which fetched by given path
options:
_terms:
2021-01-27 08:41:58 +01:00
description:
- first is a path to KeePass entry
- second is a property name of the entry, e.g. username or password
2021-01-27 08:41:58 +01:00
- third (optional property) if true custem_field_property is return
required: True
notes:
- https://github.com/viczem/ansible-keepass
2021-01-27 08:41:58 +01:00
example:
- "{{ lookup('keepass', 'path/to/entry', 'password') }}"
"""
class LookupModule(LookupBase):
keepass = None
def run(self, terms, variables=None, **kwargs):
if not terms or len(terms) < 2 or len(terms) > 3:
raise AnsibleError('Wrong request format')
2021-01-27 08:41:58 +01:00
if variables is not None:
self._templar.available_variables = variables
variables_for_templating = getattr(self._templar, '_available_variables', {})
entry_path = terms[0].strip('/')
entry_attr = terms[1]
enable_custom_attr = False
2021-01-27 08:41:58 +01:00
if len(terms) == 3:
enable_custom_attr = terms[2]
2021-01-27 08:41:58 +01:00
kp_dbx = self._templar.template(variables_for_templating.get('keepass_dbx', ''), fail_on_undefined=True)
kp_dbx = os.path.realpath(os.path.expanduser(kp_dbx))
if os.path.isfile(kp_dbx):
display.v(u"Keepass: database file %s" % kp_dbx)
kp_soc = "%s/ansible-keepass.sock" % tempfile.gettempdir()
if os.path.exists(kp_soc):
display.v(u"Keepass: fetch from socket")
2021-01-27 08:41:58 +01:00
return self._fetch_socket(kp_soc, entry_path, entry_attr, enable_custom_attr)
2021-01-27 08:41:58 +01:00
kp_psw = self._templar.template(variables_for_templating.get('keepass_psw', ''), fail_on_undefined=True)
kp_key = self._templar.template(variables_for_templating.get('keepass_key', ''), fail_on_undefined=True)
display.v(u"Keepass: fetch from kdbx file")
return self._fetch_file(
kp_dbx, str(kp_psw), kp_key, entry_path, entry_attr, enable_custom_attr)
def _fetch_file(self, kp_dbx, kp_psw, kp_key, entry_path, entry_attr, enable_custom_attr):
if kp_key:
kp_key = os.path.realpath(os.path.expanduser(kp_key))
if os.path.isfile(kp_key):
display.vvv(u"Keepass: database keyfile: %s" % kp_key)
try:
if not LookupModule.keepass:
LookupModule.keepass = PyKeePass(kp_dbx, kp_psw, kp_key)
entry = LookupModule.keepass.\
find_entries_by_path(entry_path, first=True)
if entry is None:
raise AnsibleError(u"Entry '%s' is not found" % entry_path)
display.vv(
u"KeePass: attr: %s in path: %s" % (entry_attr, entry_path))
entry_val = None
if enable_custom_attr:
entry_val = entry.get_custom_property(entry_attr)
if entry_val is not None:
return [entry_val]
else:
raise AnsibleError(AttributeError(u"'No custom field property '%s'" % (entry_attr)))
else:
return [getattr(entry, entry_attr)]
except ChecksumError:
raise AnsibleError("Wrong password/keyfile {}".format(kp_dbx))
except (AttributeError, FileNotFoundError) as e:
raise AnsibleError(e)
2021-01-27 08:41:58 +01:00
def _fetch_socket(self, kp_soc, entry_path, entry_attr, enable_custom_attr):
display.vvvv(u"KeePass: try to socket connect")
sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
sock.connect(kp_soc)
display.vvvv(u"KeePass: connected")
2021-01-27 08:41:58 +01:00
data = {'attr': entry_attr, 'path': entry_path}
if enable_custom_attr:
data['enable_custom_attr'] = True
sock.send(json.dumps(data).encode())
display.vv(u"KeePass: attr: %s in path: %s" % (entry_attr, entry_path))
try:
msg = json.loads(sock.recv(1024).decode())
except json.JSONDecodeError as e:
raise AnsibleError(str(e))
finally:
sock.close()
display.vvvv(u"KeePass: disconnected")
if msg['status'] == 'error':
raise AnsibleError(msg['text'])
return [msg['text']]