65 lines
2.8 KiB
Python
65 lines
2.8 KiB
Python
"""Emit shell exports for independent sources; never print provider errors or values to stderr."""
|
|
import argparse
|
|
import concurrent.futures
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import shlex
|
|
import subprocess
|
|
import sys
|
|
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument('manifest', type=Path)
|
|
parser.add_argument('--tools-only', action='store_true')
|
|
args = parser.parse_args()
|
|
groups = {
|
|
'company': ['email', 'password'],
|
|
'live': ['YOOKR_MYSQL_LIVE_DATABASE_URL', 'YOOKR_MONGODB_LIVE_DATABASE_URL'],
|
|
'test': ['YOOKR_MYSQL_TEST_DATABASE_URL', 'YOOKR_MONGODB_TEST_DATABASE_URL',
|
|
'RABBITMQADMIN_USERNAME', 'RABBITMQADMIN_PASSWORD'],
|
|
}
|
|
|
|
def load(group):
|
|
env = os.environ.copy()
|
|
# Explicit manifests/profiles/providers must not inherit a user's override.
|
|
for name in ('SECRETSPEC_PROVIDER', 'SECRETSPEC_PROFILE', 'SECRETSPEC_SCOPE'):
|
|
env.pop(name, None)
|
|
if group == 'test':
|
|
# A dedicated, explicitly provisioned test kubeconfig is mandatory.
|
|
kubeconfig = Path(os.environ.get('XDG_RUNTIME_DIR', '/nonexistent')) / 'workbench/test-kubeconfig'
|
|
manifest = Path.home() / '.config/workbench/test-secrets.toml'
|
|
if not kubeconfig.is_file() or not manifest.is_file():
|
|
return group, None
|
|
env['KUBECONFIG'] = str(kubeconfig)
|
|
else:
|
|
manifest = args.manifest
|
|
try:
|
|
result = subprocess.run(
|
|
['secretspec', '--file', str(manifest), 'export', '--profile', group, '--format', 'json'],
|
|
env=env, stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL,
|
|
timeout=20, check=True, text=True)
|
|
values = json.loads(result.stdout)
|
|
if not all(isinstance(values.get(key), str) and values[key] for key in groups[group]):
|
|
return group, None
|
|
return group, {key: values[key] for key in groups[group]}
|
|
except (OSError, ValueError, subprocess.SubprocessError):
|
|
return group, None
|
|
|
|
# Clear inherited target values before attempting a reload, including on failure.
|
|
for keys in groups.values():
|
|
for key in keys:
|
|
print('unset ' + key)
|
|
if args.tools_only:
|
|
for group in groups:
|
|
print('export WORKBENCH_' + group.upper() + '_STATUS=disabled')
|
|
else:
|
|
with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
|
|
for group, values in executor.map(load, groups):
|
|
status = 'ready' if values is not None else 'unavailable'
|
|
print('export WORKBENCH_' + group.upper() + '_STATUS=' + status)
|
|
if values:
|
|
for key, value in values.items():
|
|
print('export ' + key + '=' + shlex.quote(value))
|
|
print(f'workbench: {group} {status}', file=sys.stderr)
|
|
print('Retry after provisioning/login: workbench-reload. Clean tools: workbench-reload --tools-only.', file=sys.stderr)
|