206 lines
7.7 KiB
Python
206 lines
7.7 KiB
Python
"""
|
|
control/cli.py
|
|
--------------
|
|
Thin operator CLI for the control plane. Schema changes go through Alembic
|
|
(`alembic -c control/migrations/alembic.ini upgrade head`); this CLI covers
|
|
the bootstrap actions on top of an already-migrated control DB.
|
|
|
|
python -m control.cli seed
|
|
python -m control.cli create-superadmin --username admin --email you@example.com
|
|
python -m control.cli list-plans
|
|
python -m control.cli verify-keys
|
|
python -m control.cli rotate-keys --dry-run
|
|
"""
|
|
|
|
import argparse
|
|
import getpass
|
|
import sys
|
|
|
|
from control.base import control_session
|
|
from control.models import Plan, Superadmin, Tenant
|
|
from control.seed import seed_plans
|
|
|
|
|
|
def _cmd_seed(_args):
|
|
created, updated = seed_plans()
|
|
print(f'Plans seeded — created={created} updated={updated}')
|
|
|
|
|
|
def _cmd_list_plans(_args):
|
|
with control_session() as s:
|
|
plans = s.query(Plan).order_by(Plan.id).all()
|
|
if not plans:
|
|
print('No plans found. Run: python -m control.cli seed')
|
|
return
|
|
for p in plans:
|
|
price = p.stripe_price_id or '(not set)'
|
|
print(f' [{p.code}] {p.name} '
|
|
f'users={p.max_users} facilities={p.max_facilities} '
|
|
f'insp/mo={p.max_inspections_month} issues/mo={p.max_issues_month} '
|
|
f'custom_domain={p.allow_custom_domain} mobile={p.allow_mobile_api} '
|
|
f'sched_reports={p.allow_scheduled_reports} branding={p.allow_branding} '
|
|
f'stripe_price_id={price}')
|
|
|
|
|
|
def _cmd_create_superadmin(args):
|
|
with control_session() as s:
|
|
if s.query(Superadmin).filter_by(username=args.username).first():
|
|
print(f"Superadmin '{args.username}' already exists.")
|
|
sys.exit(1)
|
|
if s.query(Superadmin).filter_by(email=args.email).first():
|
|
print(f"Email '{args.email}' already in use.")
|
|
sys.exit(1)
|
|
password = args.password or getpass.getpass('Password: ')
|
|
if not password:
|
|
print('Password cannot be empty.')
|
|
sys.exit(1)
|
|
sa = Superadmin(username=args.username, email=args.email, active=True)
|
|
sa.set_password(password)
|
|
s.add(sa)
|
|
print(f"Superadmin '{args.username}' created.")
|
|
|
|
|
|
def _cmd_verify_keys(_args):
|
|
"""MT-23: confirm every stored credential is readable under the current key set.
|
|
|
|
Run this BEFORE retiring a key and again after rotating. It is the step
|
|
that makes rotation safe: it proves no row still depends on a key that is
|
|
about to be removed, instead of finding out at the next tenant request.
|
|
"""
|
|
from control import crypto
|
|
current = crypto.current_key_version()
|
|
print(f'Primary key fingerprint : {crypto.key_fingerprint()}')
|
|
print(f'Retired keys configured : {len(crypto.old_keys())}')
|
|
print(f'Current key version : {current}')
|
|
print('-' * 60)
|
|
|
|
ok = failed = empty = stale = 0
|
|
failures = []
|
|
with control_session() as s:
|
|
tenants = s.query(Tenant).order_by(Tenant.id).all()
|
|
for t in tenants:
|
|
if not t.db_password_enc:
|
|
empty += 1
|
|
continue
|
|
if crypto.can_decrypt(t.db_password_enc):
|
|
ok += 1
|
|
else:
|
|
failed += 1
|
|
failures.append((t.id, t.slug))
|
|
if (t.key_version or 1) < current:
|
|
stale += 1
|
|
|
|
print(f' readable : {ok}')
|
|
print(f' no credential set : {empty}')
|
|
print(f' UNREADABLE : {failed}')
|
|
print(f' awaiting rotation : {stale} (key_version < {current})')
|
|
|
|
if failures:
|
|
print('\nUnreadable rows — do NOT retire any key:')
|
|
for tid, slug in failures:
|
|
print(f' tenant {tid} ({slug})')
|
|
sys.exit(1)
|
|
|
|
if stale:
|
|
print(f'\n{stale} row(s) still on an older key. Run: '
|
|
'python -m control.cli rotate-keys')
|
|
else:
|
|
print('\nAll credentials current. Safe to drop retired keys from '
|
|
'CONTROL_FERNET_KEYS_OLD.')
|
|
|
|
|
|
def _cmd_rotate_keys(args):
|
|
"""MT-23: re-encrypt tenant credentials under the primary Fernet key.
|
|
|
|
Idempotent — rows already at the current version are skipped unless
|
|
--force is given. Each tenant is committed independently so a failure
|
|
part-way leaves a resumable state rather than an all-or-nothing rollback.
|
|
"""
|
|
from control import crypto
|
|
from cryptography.fernet import InvalidToken
|
|
|
|
current = crypto.current_key_version()
|
|
print(f'Primary key fingerprint : {crypto.key_fingerprint()}')
|
|
print(f'Target key version : {current}')
|
|
if args.dry_run:
|
|
print('MODE: dry run — nothing will be written.')
|
|
print('-' * 60)
|
|
|
|
rotated = skipped = empty = errors = 0
|
|
with control_session() as s:
|
|
tenants = s.query(Tenant).order_by(Tenant.id).all()
|
|
for t in tenants:
|
|
if (t.key_version or 1) >= current and not args.force:
|
|
skipped += 1
|
|
continue
|
|
if args.dry_run:
|
|
label = 'would rotate' if t.db_password_enc else 'would stamp (no credential)'
|
|
print(f' tenant {t.id} ({t.slug}): {label}')
|
|
rotated += 1
|
|
continue
|
|
try:
|
|
changed = t.rotate_db_password_key()
|
|
s.flush()
|
|
if changed:
|
|
rotated += 1
|
|
print(f' tenant {t.id} ({t.slug}): rotated -> v{t.key_version}')
|
|
else:
|
|
empty += 1
|
|
print(f' tenant {t.id} ({t.slug}): no credential, stamped '
|
|
f'-> v{t.key_version}')
|
|
except InvalidToken:
|
|
errors += 1
|
|
print(f' tenant {t.id} ({t.slug}): ERROR — no configured key opens '
|
|
f'this credential. Add the retiring key to '
|
|
f'CONTROL_FERNET_KEYS_OLD and re-run.')
|
|
except Exception as exc:
|
|
errors += 1
|
|
print(f' tenant {t.id} ({t.slug}): ERROR — {exc}')
|
|
|
|
print('-' * 60)
|
|
print(f'rotated={rotated} skipped={skipped} no_credential={empty} errors={errors}')
|
|
if errors:
|
|
print('\nRotation incomplete. Do NOT retire any key. Fix the errors above '
|
|
'and re-run — completed rows are skipped automatically.')
|
|
sys.exit(1)
|
|
if not args.dry_run:
|
|
print('\nNow run: python -m control.cli verify-keys')
|
|
|
|
|
|
def main(argv=None):
|
|
parser = argparse.ArgumentParser(prog='control.cli',
|
|
description='JQC control-plane CLI')
|
|
sub = parser.add_subparsers(dest='command', required=True)
|
|
|
|
sub.add_parser('seed', help='Seed/upsert baseline plans').set_defaults(func=_cmd_seed)
|
|
sub.add_parser('list-plans', help='List plans').set_defaults(func=_cmd_list_plans)
|
|
|
|
p_sa = sub.add_parser('create-superadmin', help='Create a superadmin account')
|
|
p_sa.add_argument('--username', required=True)
|
|
p_sa.add_argument('--email', required=True)
|
|
p_sa.add_argument('--password', default=None,
|
|
help='Omit to be prompted securely.')
|
|
p_sa.set_defaults(func=_cmd_create_superadmin)
|
|
|
|
sub.add_parser(
|
|
'verify-keys',
|
|
help='Check every tenant credential decrypts under the current key set',
|
|
).set_defaults(func=_cmd_verify_keys)
|
|
|
|
p_rot = sub.add_parser(
|
|
'rotate-keys',
|
|
help='Re-encrypt tenant credentials under the primary Fernet key',
|
|
)
|
|
p_rot.add_argument('--dry-run', action='store_true',
|
|
help='Report what would change; write nothing.')
|
|
p_rot.add_argument('--force', action='store_true',
|
|
help='Re-encrypt even rows already at the current version.')
|
|
p_rot.set_defaults(func=_cmd_rotate_keys)
|
|
|
|
args = parser.parse_args(argv)
|
|
args.func(args)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|