2020-04-13 10:51:37 +01:00
|
|
|
#!/usr/bin/env python3
|
|
|
|
"""
|
|
|
|
Extracts SSH keys from Bitwarden vault
|
|
|
|
"""
|
|
|
|
|
|
|
|
import argparse
|
|
|
|
import getpass
|
|
|
|
import json
|
|
|
|
import logging
|
|
|
|
import os
|
|
|
|
import subprocess
|
|
|
|
import sys
|
|
|
|
|
2020-04-16 12:11:18 +01:00
|
|
|
from pkg_resources import parse_version
|
|
|
|
|
|
|
|
|
|
|
|
def memoize(func):
|
|
|
|
"""
|
|
|
|
Decorator function to cache the results of another function call
|
|
|
|
"""
|
|
|
|
cache = dict()
|
|
|
|
|
|
|
|
def memoized_func(*args):
|
|
|
|
if args in cache:
|
|
|
|
return cache[args]
|
|
|
|
result = func(*args)
|
|
|
|
cache[args] = result
|
|
|
|
return result
|
|
|
|
|
|
|
|
return memoized_func
|
|
|
|
|
|
|
|
|
|
|
|
@memoize
|
|
|
|
def bwcli_version():
|
|
|
|
"""
|
|
|
|
Function to return the version of the Bitwarden CLI
|
|
|
|
"""
|
2021-01-26 19:29:22 +00:00
|
|
|
proc_version = subprocess.run(
|
|
|
|
['bw', '--version'],
|
|
|
|
stdout=subprocess.PIPE,
|
2021-01-26 21:17:01 +00:00
|
|
|
text=True,
|
|
|
|
check=True,
|
2020-04-16 12:11:18 +01:00
|
|
|
)
|
2021-01-26 19:29:22 +00:00
|
|
|
return proc_version.stdout
|
2020-04-16 12:11:18 +01:00
|
|
|
|
|
|
|
|
|
|
|
@memoize
|
|
|
|
def cli_supports(feature):
|
|
|
|
"""
|
|
|
|
Function to return whether the current Bitwarden CLI supports a particular
|
|
|
|
feature
|
|
|
|
"""
|
|
|
|
version = parse_version(bwcli_version())
|
|
|
|
|
|
|
|
if feature == 'nointeraction' and version >= parse_version('1.9.0'):
|
|
|
|
return True
|
|
|
|
return False
|
|
|
|
|
2020-04-13 10:51:37 +01:00
|
|
|
|
|
|
|
def get_session():
|
|
|
|
"""
|
|
|
|
Function to return a valid Bitwarden session
|
|
|
|
"""
|
|
|
|
# Check for an existing, user-supplied Bitwarden session
|
2021-04-16 09:59:05 +01:00
|
|
|
session = os.environ.get('BW_SESSION')
|
|
|
|
if session is not None:
|
2021-01-26 19:29:22 +00:00
|
|
|
logging.debug('Existing Bitwarden session found')
|
|
|
|
return session
|
2020-04-13 10:51:37 +01:00
|
|
|
|
|
|
|
# Check if we're already logged in
|
2021-01-26 19:29:22 +00:00
|
|
|
proc_logged = subprocess.run(['bw', 'login', '--check', '--quiet'])
|
2020-04-13 10:51:37 +01:00
|
|
|
|
2021-01-26 19:29:22 +00:00
|
|
|
if proc_logged.returncode:
|
2020-04-13 10:51:37 +01:00
|
|
|
logging.debug('Not logged into Bitwarden')
|
|
|
|
operation = 'login'
|
|
|
|
else:
|
|
|
|
logging.debug('Bitwarden vault is locked')
|
|
|
|
operation = 'unlock'
|
|
|
|
|
2021-01-26 19:29:22 +00:00
|
|
|
proc_session = subprocess.run(
|
|
|
|
['bw', '--raw', operation],
|
2020-04-13 10:51:37 +01:00
|
|
|
stdout=subprocess.PIPE,
|
2021-01-26 19:29:22 +00:00
|
|
|
text=True,
|
2021-01-26 21:17:01 +00:00
|
|
|
check=True,
|
2020-04-13 10:51:37 +01:00
|
|
|
)
|
2021-01-26 19:29:22 +00:00
|
|
|
return proc_session.stdout
|
2020-04-13 10:51:37 +01:00
|
|
|
|
|
|
|
|
|
|
|
def get_folders(session, foldername):
|
|
|
|
"""
|
|
|
|
Function to return the ID of the folder that matches the provided name
|
|
|
|
"""
|
|
|
|
logging.debug('Folder name: %s', foldername)
|
|
|
|
|
2021-01-26 19:29:22 +00:00
|
|
|
proc_folders = subprocess.run(
|
|
|
|
['bw', 'list', 'folders', '--search', foldername, '--session', session],
|
2020-04-13 10:51:37 +01:00
|
|
|
stdout=subprocess.PIPE,
|
2021-01-26 19:29:22 +00:00
|
|
|
text=True,
|
2021-01-26 21:17:01 +00:00
|
|
|
check=True,
|
2020-04-13 10:51:37 +01:00
|
|
|
)
|
|
|
|
|
2021-01-26 19:29:22 +00:00
|
|
|
folders = json.loads(proc_folders.stdout)
|
2020-04-13 10:51:37 +01:00
|
|
|
|
|
|
|
if not folders:
|
|
|
|
logging.error('"%s" folder not found', foldername)
|
|
|
|
return None
|
|
|
|
|
|
|
|
# Do we have any folders
|
|
|
|
if len(folders) != 1:
|
|
|
|
logging.error('%d folders with the name "%s" found', len(folders), foldername)
|
|
|
|
return None
|
|
|
|
|
|
|
|
return folders[0]['id']
|
|
|
|
|
|
|
|
|
|
|
|
def folder_items(session, folder_id):
|
|
|
|
"""
|
|
|
|
Function to return items from a folder
|
|
|
|
"""
|
|
|
|
logging.debug('Folder ID: %s', folder_id)
|
|
|
|
|
2021-01-26 19:29:22 +00:00
|
|
|
proc_items = subprocess.run(
|
|
|
|
[ 'bw', 'list', 'items', '--folderid', folder_id, '--session', session],
|
2020-04-13 10:51:37 +01:00
|
|
|
stdout=subprocess.PIPE,
|
2021-01-26 19:29:22 +00:00
|
|
|
text=True,
|
2021-01-26 21:17:01 +00:00
|
|
|
check=True,
|
2020-04-13 10:51:37 +01:00
|
|
|
)
|
2021-01-26 19:29:22 +00:00
|
|
|
return json.loads(proc_items.stdout)
|
2020-04-13 10:51:37 +01:00
|
|
|
|
|
|
|
|
|
|
|
def add_ssh_keys(session, items, keyname):
|
|
|
|
"""
|
|
|
|
Function to attempt to get keys from a vault item
|
|
|
|
"""
|
|
|
|
for item in items:
|
|
|
|
try:
|
|
|
|
private_key_file = [k['value'] for k in item['fields']
|
|
|
|
if k['name'] == keyname and k['type'] == 0][0]
|
|
|
|
except IndexError:
|
|
|
|
logging.warning('No "%s" field found for item %s', keyname, item['name'])
|
|
|
|
continue
|
|
|
|
logging.debug('Private key file declared')
|
|
|
|
|
|
|
|
try:
|
|
|
|
private_key_id = [k['id'] for k in item['attachments']
|
|
|
|
if k['fileName'] == private_key_file][0]
|
|
|
|
except IndexError:
|
|
|
|
logging.warning(
|
|
|
|
'No attachment called "%s" found for item %s',
|
|
|
|
private_key_file,
|
|
|
|
item['name']
|
|
|
|
)
|
|
|
|
continue
|
|
|
|
logging.debug('Private key ID found')
|
|
|
|
|
2021-05-11 09:13:04 +01:00
|
|
|
try:
|
|
|
|
ssh_add(session, item['id'], private_key_id)
|
|
|
|
except subprocess.SubprocessError:
|
2021-01-16 10:26:30 +00:00
|
|
|
logging.warning('Could not add key to the SSH agent')
|
2020-04-13 10:51:37 +01:00
|
|
|
|
|
|
|
|
|
|
|
def ssh_add(session, item_id, key_id):
|
|
|
|
"""
|
|
|
|
Function to get the key contents from the Bitwarden vault
|
|
|
|
"""
|
|
|
|
logging.debug('Item ID: %s', item_id)
|
|
|
|
logging.debug('Key ID: %s', key_id)
|
|
|
|
|
2021-01-26 19:29:22 +00:00
|
|
|
proc_attachment = subprocess.run([
|
|
|
|
'bw',
|
|
|
|
'get',
|
|
|
|
'attachment', key_id,
|
|
|
|
'--itemid', item_id,
|
|
|
|
'--raw',
|
|
|
|
'--session', session
|
|
|
|
],
|
|
|
|
stdout=subprocess.PIPE,
|
|
|
|
text=True,
|
2021-01-26 21:17:01 +00:00
|
|
|
check=True,
|
2021-01-26 19:29:22 +00:00
|
|
|
)
|
|
|
|
ssh_key = proc_attachment.stdout
|
2020-04-13 10:51:37 +01:00
|
|
|
|
2021-01-26 19:29:22 +00:00
|
|
|
logging.debug("Running ssh-add")
|
2020-04-13 10:51:37 +01:00
|
|
|
|
2021-01-26 19:29:22 +00:00
|
|
|
# CAVEAT: `ssh-add` provides no useful output, even with maximum verbosity
|
2021-05-11 09:13:04 +01:00
|
|
|
subprocess.run(
|
2021-01-26 21:17:01 +00:00
|
|
|
['ssh-add', '-'],
|
|
|
|
input=ssh_key,
|
|
|
|
text=True,
|
|
|
|
check=True,
|
|
|
|
)
|
2020-04-13 10:51:37 +01:00
|
|
|
|
|
|
|
|
|
|
|
if __name__ == '__main__':
|
|
|
|
def parse_args():
|
|
|
|
"""
|
|
|
|
Function to parse command line arguments
|
|
|
|
"""
|
|
|
|
parser = argparse.ArgumentParser()
|
|
|
|
parser.add_argument(
|
|
|
|
'-d', '--debug',
|
|
|
|
action='store_true',
|
|
|
|
help='show debug output',
|
|
|
|
)
|
|
|
|
parser.add_argument(
|
|
|
|
'-f', '--foldername',
|
|
|
|
default='ssh-agent',
|
|
|
|
help='folder name to use to search for SSH keys',
|
|
|
|
)
|
|
|
|
parser.add_argument(
|
|
|
|
'-c', '--customfield',
|
|
|
|
default='private',
|
|
|
|
help='custom field name where private key filename is stored',
|
|
|
|
)
|
|
|
|
|
|
|
|
return parser.parse_args()
|
|
|
|
|
|
|
|
|
|
|
|
def main():
|
|
|
|
"""
|
|
|
|
Main program logic
|
|
|
|
"""
|
|
|
|
|
|
|
|
args = parse_args()
|
|
|
|
|
|
|
|
if args.debug:
|
|
|
|
loglevel = logging.DEBUG
|
|
|
|
else:
|
|
|
|
loglevel = logging.INFO
|
|
|
|
|
|
|
|
logging.basicConfig(level=loglevel)
|
|
|
|
|
2021-01-26 19:29:22 +00:00
|
|
|
try:
|
|
|
|
logging.info('Getting Bitwarden session')
|
|
|
|
session = get_session()
|
|
|
|
logging.debug('Session = %s', session)
|
|
|
|
|
|
|
|
logging.info('Getting folder list')
|
|
|
|
folder_id = get_folders(session, args.foldername)
|
|
|
|
|
|
|
|
logging.info('Getting folder items')
|
|
|
|
items = folder_items(session, folder_id)
|
|
|
|
|
|
|
|
logging.info('Attempting to add keys to ssh-agent')
|
|
|
|
add_ssh_keys(session, items, args.customfield)
|
|
|
|
except subprocess.CalledProcessError as e:
|
|
|
|
if e.stderr:
|
|
|
|
logging.error('`%s` error: %s', e.cmd[0], e.stderr)
|
|
|
|
logging.debug('Error running %s', e.cmd)
|
2020-04-13 10:51:37 +01:00
|
|
|
|
|
|
|
main()
|