before cleanup
This commit is contained in:
@@ -1,63 +1,191 @@
|
||||
# Copy this to ".ansible/plugins/lookup"
|
||||
# Copy this file to ~/.ansible/plugins/lookup/keepass.py
|
||||
#
|
||||
# Prerequisites:
|
||||
# pip install keepassxc-proxy-client
|
||||
#
|
||||
# KeePassXC setup:
|
||||
# Tools > Settings > Browser Integration > Enable browser integration
|
||||
# The plugin will prompt you to approve the association in KeePassXC on first run.
|
||||
# Association credentials are saved to ~/.keepassxc_ansible_assoc (override with
|
||||
# env var KEEPASSXC_ASSOC_FILE).
|
||||
#
|
||||
# Usage:
|
||||
# {{ lookup('keepass', 'https://example.com', 'password') }}
|
||||
# {{ lookup('keepass', 'https://example.com', 'username') }}
|
||||
# {{ lookup('keepass', 'https://example.com') }} # defaults to password
|
||||
#
|
||||
# Terms:
|
||||
# terms[0] URL (or title) to search for — matched against KeePassXC entries
|
||||
# terms[1] Attribute to return: username / user / password / pass / passwd
|
||||
# Defaults to 'password' if omitted.
|
||||
|
||||
from __future__ import absolute_import, division, print_function
|
||||
__metaclass__ = type
|
||||
|
||||
import json
|
||||
import os
|
||||
|
||||
from ansible.errors import AnsibleError
|
||||
from ansible.plugins.lookup import LookupBase
|
||||
from keepasshttplib import keepasshttplib, encrypter
|
||||
import requests
|
||||
|
||||
DOCUMENTATION = """
|
||||
DOCUMENTATION = r"""
|
||||
lookup: keepass
|
||||
author: Martin Bauer <bauer_martin@gmx.de>
|
||||
version_added: '0.2'
|
||||
short_description: fetch data from KeePass over KeePassHTTP
|
||||
short_description: Fetch credentials from KeePassXC via the browser proxy protocol
|
||||
description:
|
||||
- This lookup returns a username or password queried by the URL of the keepass entry
|
||||
- Connects to a running KeePassXC instance over its browser-integration Unix socket
|
||||
(or Windows named pipe) and retrieves credentials for a given URL.
|
||||
- On first use the plugin associates itself with KeePassXC; you will see an
|
||||
approval dialog in KeePassXC. The association is saved to disk so subsequent
|
||||
runs are silent.
|
||||
options:
|
||||
_terms:
|
||||
description:
|
||||
- first is the URL to search for
|
||||
- second is a property name of the entry, e.g. username or password
|
||||
required: True
|
||||
description:
|
||||
- First term is the URL (or entry title URL) to look up.
|
||||
- Second term (optional) is the attribute to return.
|
||||
Accepted values are username, user, password, pass, passwd.
|
||||
Defaults to password.
|
||||
required: true
|
||||
notes:
|
||||
- https://github.com/viczem/ansible-keepass
|
||||
|
||||
example:
|
||||
- "{{ lookup('keepass', 'urlOfEntry', 'password') }}"
|
||||
- KeePassXC must be running with Browser Integration enabled.
|
||||
- Requires the Python package keepassxc-proxy-client.
|
||||
"""
|
||||
|
||||
EXAMPLES = r"""
|
||||
- name: Get password for an entry
|
||||
debug:
|
||||
msg: "{{ lookup('keepass', 'https://example.com') }}"
|
||||
|
||||
- name: Get username for an entry
|
||||
debug:
|
||||
msg: "{{ lookup('keepass', 'https://example.com', 'username') }}"
|
||||
|
||||
- name: Set pi account password from KeePassXC
|
||||
user:
|
||||
name: pi
|
||||
password: "{{ lookup('keepass', 'default_rpi_password') | password_hash('sha512') }}"
|
||||
"""
|
||||
|
||||
ASSOC_FILE_ENV = "KEEPASSXC_ASSOC_FILE"
|
||||
ASSOC_FILE_DEFAULT = os.path.expanduser("~/.keepassxc_ansible_assoc")
|
||||
ASSOC_NAME = "ansible-keepassxc"
|
||||
|
||||
USERNAME_ATTRS = {"username", "user"}
|
||||
PASSWORD_ATTRS = {"password", "pass", "passwd"}
|
||||
|
||||
|
||||
def _get_assoc_file():
|
||||
return os.environ.get(ASSOC_FILE_ENV, ASSOC_FILE_DEFAULT)
|
||||
|
||||
|
||||
def _load_assoc(connection):
|
||||
"""Load a previously saved association from disk into *connection*. Returns True on success."""
|
||||
path = _get_assoc_file()
|
||||
if not os.path.exists(path):
|
||||
return False
|
||||
try:
|
||||
with open(path, "r") as fh:
|
||||
data = json.load(fh)
|
||||
connection.load_associate(data["name"], bytes(data["public_key"]))
|
||||
return True
|
||||
except (KeyError, ValueError, OSError):
|
||||
return False
|
||||
|
||||
|
||||
def _save_assoc(connection):
|
||||
"""Persist the current association to disk."""
|
||||
path = _get_assoc_file()
|
||||
assoc_id, public_key = connection.dump_associate()
|
||||
data = {"name": assoc_id, "public_key": list(public_key)}
|
||||
with open(path, "w") as fh:
|
||||
json.dump(data, fh)
|
||||
os.chmod(path, 0o600)
|
||||
|
||||
|
||||
def _connect():
|
||||
"""Return an authenticated, test-verified Connection to KeePassXC."""
|
||||
try:
|
||||
from keepassxc_proxy_client.protocol import Connection, ResponseUnsuccesfulException
|
||||
except ImportError:
|
||||
raise AnsibleError(
|
||||
"keepassxc-proxy-client is not installed. "
|
||||
"Run: pip install keepassxc-proxy-client"
|
||||
)
|
||||
|
||||
conn = Connection()
|
||||
|
||||
try:
|
||||
conn.connect()
|
||||
except Exception as exc:
|
||||
raise AnsibleError(
|
||||
"Cannot connect to KeePassXC. Is it running with Browser Integration enabled? "
|
||||
"Error: {}".format(exc)
|
||||
)
|
||||
|
||||
conn.change_public_keys()
|
||||
|
||||
# Try to reuse an existing association
|
||||
if _load_assoc(conn):
|
||||
try:
|
||||
conn.test_associate(trigger_unlock=True)
|
||||
return conn # existing association is still valid
|
||||
except Exception:
|
||||
pass # fall through and re-associate
|
||||
|
||||
# First run (or stale association) — ask KeePassXC to authorise us
|
||||
try:
|
||||
conn.associate()
|
||||
except Exception as exc:
|
||||
raise AnsibleError(
|
||||
"KeePassXC association failed. "
|
||||
"Please approve the request in the KeePassXC dialog. "
|
||||
"Error: {}".format(exc)
|
||||
)
|
||||
|
||||
try:
|
||||
conn.test_associate(trigger_unlock=True)
|
||||
except Exception as exc:
|
||||
raise AnsibleError("KeePassXC association test failed: {}".format(exc))
|
||||
|
||||
_save_assoc(conn)
|
||||
return conn
|
||||
|
||||
|
||||
class LookupModule(LookupBase):
|
||||
|
||||
def __init__(self, *args, **kwargs):
|
||||
super(LookupModule, self).__init__(*args, **kwargs)
|
||||
self.k = keepasshttplib.Keepasshttplib()
|
||||
|
||||
|
||||
def run(self, terms, variables=None, **kwargs):
|
||||
if not terms or len(terms) > 2:
|
||||
raise AnsibleError('Keepass wrong request format')
|
||||
if len(terms) == 1:
|
||||
entry_path, entry_attr = terms[0], 'password'
|
||||
else:
|
||||
entry_path, entry_attr = terms[0], terms[1]
|
||||
|
||||
#if not self._test_connection():
|
||||
# raise AnsibleError('Keepass is closed!')
|
||||
try:
|
||||
auth = self.k.get_credentials(entry_path)
|
||||
except Exception as e:
|
||||
raise AnsibleError('Keepass error obtaining entry {}: {}'.format(entry_path, e))
|
||||
if auth:
|
||||
if entry_attr not in ('username', 'user', 'pass', 'passwd', 'password'):
|
||||
raise AnsibleError("Keepass wrong entry")
|
||||
|
||||
ret = auth[0] if entry_attr.startswith('user') else auth[1]
|
||||
return [ret]
|
||||
raise AnsibleError(
|
||||
"keepass lookup expects 1 or 2 arguments: "
|
||||
"lookup('keepass', '<url>', '<attribute>')"
|
||||
)
|
||||
|
||||
url = terms[0]
|
||||
attr = terms[1].lower() if len(terms) == 2 else "password"
|
||||
|
||||
if attr not in USERNAME_ATTRS | PASSWORD_ATTRS:
|
||||
raise AnsibleError(
|
||||
"keepass: unsupported attribute '{}'. "
|
||||
"Use one of: username, user, password, pass, passwd".format(attr)
|
||||
)
|
||||
|
||||
conn = _connect()
|
||||
|
||||
def _test_connection(self):
|
||||
key = self.k.get_key_from_keyring()
|
||||
if key is None:
|
||||
key = encrypter.generate_key()
|
||||
id_ = self.k.get_id_from_keyring()
|
||||
try:
|
||||
return self.k.test_associate(key, id_)
|
||||
except requests.exceptions.ConnectionError as e:
|
||||
raise AnsibleError('Keepass Connection Error: {}'.format(e))
|
||||
entries = conn.get_logins(url)
|
||||
except Exception as exc:
|
||||
raise AnsibleError(
|
||||
"keepass: failed to retrieve logins for '{}': {}".format(url, exc)
|
||||
)
|
||||
|
||||
if not entries:
|
||||
raise AnsibleError(
|
||||
"keepass: no entries found in KeePassXC for URL '{}'".format(url)
|
||||
)
|
||||
|
||||
# Return the first matching entry (KeePassXC already filters by URL)
|
||||
entry = entries[0]
|
||||
|
||||
if attr in USERNAME_ATTRS:
|
||||
return [entry.get("login", "")]
|
||||
else:
|
||||
return [entry.get("password", "")]
|
||||
|
||||
Reference in New Issue
Block a user