jgaunt/truenas-kmip-unlocker
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

updated for new api

Browse Source
This commit is contained in:
jgaunt
2025-10-24 01:41:46 +00:00
parent b6aa8121b8
commit eb226df901
1 changed files with 69 additions and 162 deletions
Download Patch File Download Diff File Expand all files Collapse all files

231
truenas_kmip_unlock.py
View File

@@ -9,7 +9,7 @@ import base64
import argparse import argparse
import configparser import configparser
import getpass import getpass
import requests from truenas_api_client import Client
import platform import platform
import subprocess import subprocess
import time import time
@@ -19,39 +19,12 @@ except ImportError:
import json import json
import hmac as pyhmac import hmac as pyhmac
import datetime import datetime
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
from kmip.core import enums from kmip.core import enums
from kmip.demos import utils from kmip.demos import utils
from kmip.pie import client from kmip.pie import client
def request(resource, api_key, method='GET', data=None):
if data is None:
data = ''
else:
data = json.dumps(data)
url = 'https://127.0.0.1/api/v2.0/{}'.format(resource)
logger.debug('Request URL: {}'.format(url))
logger.debug('Request Data: {}'.format(data))
r = requests.request(
method,
url,
data=data,
headers={'Content-Type': 'application/json', 'Authorization': 'Bearer {}'.format(api_key)},
verify=False
)
logger.debug('Request Status Code: {}'.format(r.status_code))
if r.ok:
try:
logger.debug('Request Returned JSON: {}'.format(r.json()))
return {'ok': r.ok, 'status_code': r.status_code, 'response': r.json()}
except:
logger.debug('Request Returned Text: {}'.format(r.text))
return {'ok': r.ok, 'status_code': r.status_code, 'response': r.text}
raise ValueError(r)
def build_logger(level): def build_logger(level):
logger = logging.getLogger() logger = logging.getLogger()
logger.setLevel(level) logger.setLevel(level)
@@ -296,7 +269,7 @@ def edit_pool_details(pools, pool_to_edit):
array[pool_name] = dict() array[pool_name] = dict()
array[pool_name]["pool_passphrase"] = pool_passphrase array[pool_name]["pool_passphrase"] = pool_passphrase
return array return array
def select_pool(pools, wording = "edit"): def select_pool(pools, wording = "edit"):
print("Which pool would you like to {}:".format(wording)) print("Which pool would you like to {}:".format(wording))
print(" ") print(" ")
@@ -324,13 +297,13 @@ if __name__ == '__main__':
dest="config", dest="config",
help="Edit pools configuration." help="Edit pools configuration."
) )
parser.add_argument( #parser.add_argument(
"-r", # "-r",
"--restartJails", # "--restartAppsVMs",
action="store_true", # action="store_true",
dest="restartJails", # dest="restartAppsVMs",
help="Restarts jails if running" # help="Restarts apps and VMs if running"
) #)
parser.add_argument ( parser.add_argument (
"-v", "-v",
"--verbose", "--verbose",
@@ -367,9 +340,13 @@ if __name__ == '__main__':
user_input = input("n/q> ") user_input = input("n/q> ")
if user_input.casefold() == "n": if user_input.casefold() == "n":
config = dict() config = dict()
config['URI'] = input("Please enter your hostname or IP address (default: localhost): ") or "localhost"
if ask_for_confirmation("Would you like to verify the TLS cert?"):
config['Verify TLS'] = True
else:
config['Verify TLS'] = False
config['API Key'] = input("Please enter your API key: ") config['API Key'] = input("Please enter your API key: ")
config['Pools'] = new_pool_details() config['Pools'] = new_pool_details()
config['jailStoragePool'] = select_pool(config['Pools'], "select for jail storage")
write_config_file(config, secrets_config_file) write_config_file(config, secrets_config_file)
break break
elif user_input.casefold() == "q": elif user_input.casefold() == "q":
@@ -383,20 +360,28 @@ if __name__ == '__main__':
for pool in config['Pools']: for pool in config['Pools']:
print(pool) print(pool)
print(" ") print(" ")
print("Jail storage pool: {}".format(config['jailStoragePool'])) print("URI: {}".format(config['URI']))
print("Verify TLS Cert: {}".format(config['Verify TLS']))
print(" ") print(" ")
print("a) Edit API Key") print("h) Edit Host, TLS, and API Key")
print("e) Edit pool") print("e) Edit pool")
print("j) Edit jail storage pool")
print("n) New pool") print("n) New pool")
print("d) Delete pool") print("d) Delete pool")
print("q) Quit config") print("q) Quit config")
user_input = input("a/e/j/n/d/q> ") user_input = input("h/e/n/d/q> ")
# Edit API Key # Edit Host, TLS, and API Key
if user_input.casefold() == "a": if user_input.casefold() == "h":
config['API Key'] = input("Please enter your API key: ") if ask_for_confirmation("Would you like to edit the hostname or IP address?\nCurrent Value: {}".format(config['URI'])):
config['URI'] = input("Please enter your hostname or IP address: ")
if ask_for_confirmation("Would you like to edit verify TLS cert?\nCurrent Value: {}".format(config['Verify TLS'])):
if ask_for_confirmation("Would you like to verify the TLS cert?"):
config['Verify TLS'] = True
else:
config['Verify TLS'] = False
if ask_for_confirmation("Would you like to edit the API key?"):
config['API Key'] = input("Please enter your API key: ")
write_config_file(config, secrets_config_file) write_config_file(config, secrets_config_file)
# Editing an account # Editing a pool
elif user_input.casefold() == "e": elif user_input.casefold() == "e":
pool_to_edit = select_pool(config['Pools']) pool_to_edit = select_pool(config['Pools'])
pool_details = edit_pool_details(config['Pools'], pool_to_edit) pool_details = edit_pool_details(config['Pools'], pool_to_edit)
@@ -404,27 +389,22 @@ if __name__ == '__main__':
config['Pools'].update(pool_details) config['Pools'].update(pool_details)
write_config_file(config, secrets_config_file) write_config_file(config, secrets_config_file)
break break
# Editing jail storage pool # Createing a new pool
elif user_input.casefold() == "j":
config['jailStoragePool'] = select_pool(config['Pools'], "select for jail storage")
write_config_file(config, secrets_config_file)
break
# Createing a new account
elif user_input.casefold() == "n": elif user_input.casefold() == "n":
pool_details = new_pool_details() pool_details = new_pool_details()
config['Pools'].update(pool_details) config['Pools'].update(pool_details)
write_config_file(config, secrets_config_file) write_config_file(config, secrets_config_file)
break break
# Deleting a pool
# Deleting an account
elif user_input.casefold() == "d": elif user_input.casefold() == "d":
pool_to_delete = select_pool(config['Pools'], "delete") pool_to_delete = select_pool(config['Pools'], "delete")
if not ask_for_confirmation("Are you sure you wish to delete {} pool? ".format(pool_to_delete)): if not ask_for_confirmation("Are you sure you wish to delete {} pool? ".format(pool_to_delete)):
break break
del config['Pools'][pool_to_delete] del config['Pools'][pool_to_delete]
if len(config['Pools']) == 0: if len(config['Pools']) == 0:
# no more accounts, remove secrets file if ask_for_confirmation("There are no pools left in the config, do you want to remove the secrets file?"):
os.remove(secrets_config_file) # no more accounts, remove secrets file if requests
os.remove(secrets_config_file)
else: else:
write_config_file(config, secrets_config_file) write_config_file(config, secrets_config_file)
break break
@@ -435,119 +415,46 @@ if __name__ == '__main__':
# Catch all for non-valid characters # Catch all for non-valid characters
else: else:
print("This value must be one of the following characters: a, e, j, n, d, q.") print("This value must be one of the following characters: h, e, n, d, q")
if not os.path.exists(secrets_config_file): if not os.path.exists(secrets_config_file):
print("No configuration file found. Please run {} -c for configuration.".format(script_name)) print("No configuration file found. Please run {} -c/--config for configuration".format(script_name))
sys.exit(-1) sys.exit(-1)
# run the decryption of the keys and unlock the pool # run the decryption of the keys and unlock the pool
config = read_config_file(secrets_config_file) config = read_config_file(secrets_config_file)
api_key = config['API Key'] URI = config['URI']
jail_storage_pool = config['jailStoragePool'] VERIFY_TLS = config['Verify TLS']
API_POOLS = request('pool', api_key)['response'] API_KEY = config['API Key']
API_DATASETS = request('pool/dataset', api_key)['response'] with Client(uri="wss://{}/api/current".format(URI), verify_ssl=VERIFY_TLS) as c:
for pool_dataset_name in config['Pools']: if c.call("auth.login_with_api_key", API_KEY):
if pool_dataset_name != 'DEFAULT': logger.debug('Successfully logged into {}'.format(URI))
name = pool_dataset_name else:
pool_passphrase = config['Pools'][name]['pool_passphrase'] logger.error('Unable to login to {}'.format(URI))
logger.info("Attempting to unlock {} pool.".format(name)) SYSTEM_INFORMATION = c.call("system.info")
for pool in API_POOLS: logger.debug('System Info: {}'.format(SYSTEM_INFORMATION))
if pool['name'] == name: POOL_DATASETS = c.call("pool.dataset.details")
# GELI encrypted pool logger.debug('Pool Datasets: {}'.format(POOL_DATASETS))
if 'encrypt' in pool: for config_pool in config['Pools']:
if pool['encrypt'] == 2: config_pool_name = config_pool
if pool['is_decrypted'] == False: pool_passphrase = config['Pools'][config_pool_name]['pool_passphrase']
logger.info('Pool {} is locked'.format(pool['name'])) logger.info("Attempting to unlock {} pool".format(config_pool_name))
services_to_restart = request('pool/unlock_services_restart_choices', api_key, "POST", pool['id'])['response'] for pool_dataset in POOL_DATASETS:
first = True pool_name = pool_dataset['name']
for service_to_restart in services_to_restart: if pool_name == config_pool_name:
if service_to_restart == "jails": pool_encrypted = pool_dataset['encrypted']
logger.debug('Skipping adding the {} service'.format(services_to_restart[service_to_restart])) pool_unlocked = pool_dataset['key_loaded']
continue if pool_encrypted == True and pool_unlocked == False:
if first: logger.info('Dataset {} is locked'.format(pool_name))
pool_services = "'{}'".format(service_to_restart) unlock_args = {"datasets": [{"name": pool_name, "passphrase": pool_passphrase}]}
first = False
else: unlock_status = c.call("pool.dataset.unlock", pool_name, unlock_args, job=True)
pool_services += ", '{}'".format(service_to_restart) # return will look like this: {'unlocked': ['test'], 'failed': {}}
logger.debug('Restarting {} services when unlocking pool.'.format(pool_services)) if pool_name in unlock_status.get('unlocked', []):
response = request('pool/id/{}/unlock'.format(pool['id']), api_key, 'POST', {'passphrase': '{}'.format(pool_passphrase), 'services_restart': "[" + pool_services + "]"}) logger.info('Dataset {} was unlocked successfully'.format(pool_name))
if response['ok']:
logger.info('Pool {} was unlocked and services restarted successfully'.format(pool['name']))
else:
logger.error('Pool {} was NOT unlocked and services NOT restarted'.format(pool['name']))
else:
logger.info('Pool {} is already unlocked'.format(pool['name']))
# start detection of locked zfs dataset
else:
# loop through the datasets to find the one we are looking for
for dataset in API_DATASETS:
if dataset['name'] == name:
if dataset['locked'] == True:
logger.info('Dataset {} is locked'.format(dataset['name']))
response = request('pool/dataset/unlock', api_key, 'POST', {'id': '{}'.format(dataset['id']), 'unlock_options': {'recursive': True, 'datasets': [{'name': '{}'.format(dataset['name']), 'passphrase': '{}'.format(pool_passphrase)}]}})
if response['ok']:
logger.info('Dataset {} was unlocked successfully'.format(dataset['name']))
# restart services since the dataset was unlocked
logger.info('Restarting services')
try:
services_to_restart = request('pool/unlock_services_restart_choices', api_key, "POST", pool['id'])['response']
except:
services_to_restart = request('pool/dataset/unlock_services_restart_choices', api_key, "POST", dataset['name'])['response']
for service_to_restart in services_to_restart:
if service_to_restart == "jails":
logger.debug('Skipping restarting the {} service'.format(services_to_restart[service_to_restart]))
continue
logger.debug('Restarting the {} service'.format(services_to_restart[service_to_restart]))
response = request('service/restart', api_key, 'POST', {'service': service_to_restart})
if response['ok']:
logger.debug('Service {} was restarted successfully'.format(services_to_restart[service_to_restart]))
else:
logger.error('Service {} was NOT restarted'.format(services_to_restart[service_to_restart]))
else:
logger.error('Dataset {} was NOT unlocked'.format(dataset['name']))
else:
logger.info('Dataset {} is already unlocked'.format(dataset['name']))
'''
# make sure jail storage pool is set
logger.info('Setting jail storage pool to {}'.format(jail_storage_pool))
response = request('jail/activate', api_key, 'POST', '{}'.format(jail_storage_pool))
if response['ok']:
logger.info('Successfully set jail storage pool to {}'.format(jail_storage_pool))
logger.debug("Sleeping for 120 seconds")
time.sleep(120)
# restart the jails service
#jails_service_name = 'jails'
#logger.debug('Restarting the {} service'.format(jails_service_name))
#response = request('service/restart', api_key, 'POST', {'service': jails_service_name})
#if response['ok']:
#logger.debug('Service {} was restarted successfully'.format(jails_service_name))
#else:
#logger.error('Service {} was NOT restarted'.format(jails_service_name))
# start any jails that are set to start on boot but are most likely on an encrypted pool
API_JAILS = request('jail', api_key)['response']
for jail in API_JAILS:
if jail['boot'] == 1:
logger.info('Jail {} is set to start on boot'.format(jail['id']))
if jail['state'] == 'down':
logger.info('Jail {} is stopped'.format(jail['id']))
logger.info('Starting jail {}'.format(jail['id']))
response = request('jail/start', api_key, 'POST', '{}'.format(jail['id']))
if response['ok']:
logger.info('Started jail {} successfully'.format(jail['id']))
else:
logger.error('Jail {} was NOT started'.format(jail['id']))
else:
if opts.restartJails:
logger.info('Jail {} is already started but --restartJails was passed in. Restarting jail.'.format(jail['id']))
response = request('jail/restart', api_key, 'POST', '{}'.format(jail['id']))
if response['ok']:
logger.info('Restarted jail {} successfully'.format(jail['id']))
else: else:
logger.error('Jail {} was NOT restarted'.format(jail['id'])) logger.error('Dataset {} was NOT unlocked'.format(pool_name))
else: else:
logger.info('Jail {} is already started'.format(jail['id'])) logger.info('Dataset {} is already unlocked'.format(pool_name))
else:
logger.error('Unable to set jail storage pool to {}'.format(jail_storage_pool)
'''
client.close() client.close()
sys.exit(0) sys.exit(0)