""" control/provision.py -------------------- Tenant provisioning service (MT-3). Two entry points: create_tenant(...) — provision a brand-new tenant: 1. CREATE the per-tenant MySQL database + least-privilege user + grant (using a dedicated admin account from PROVISION_DB_URL) 2. insert the control Tenant row (Fernet-encrypted credentials) 3. bootstrap the schema (baseline + stamp head) [control.tenant_migrate] 4. seed the first tenant admin (set-password token + link) 5. register the subdomain (+ optional custom domain) On failure after the database is created, best-effort rollback drops the database/user and removes the tenant row so a retry starts clean. register_tenant_zero(...) — adopt an EXISTING database (e.g. LT) in place: no DB/user creation, no bootstrap, no admin seeding — just register the tenant row pointing at the live DB, read its current alembic head, and map its domains. Zero data movement, zero disruption. Environment: PROVISION_DB_URL admin connection that can CREATE DATABASE / CREATE USER / GRANT, e.g. mysql+pymysql://jqc_provisioner:pw@127.0.0.1/ TENANT_BASE_DOMAIN subdomain suffix for tenants, e.g. jqc.app CONTROL_DATABASE_URL, CONTROL_FERNET_KEY (as for the rest of the control plane) Required grants for the provisioner account (least privilege): GRANT CREATE, CREATE USER, GRANT OPTION, ... ON *.* (or scoped jqc_% dbs) """ import argparse import os import re import secrets import string import sys from sqlalchemy import create_engine, text from sqlalchemy.engine import make_url from werkzeug.security import generate_password_hash from control.base import control_session from control.models import Plan, Tenant, TenantDomain, ProvisioningJob from control.time_utils import now_eastern from control.tenant_migrate import bootstrap_tenant, current_revision SET_PASSWORD_PATH = '/customers/set-password' # role-agnostic; sets password_set=True _SLUG_RE = re.compile(r'^[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?$') # DNS label # ── helpers ───────────────────────────────────────────────────────────────── def _require(name): v = os.environ.get(name) if not v: raise RuntimeError(f"{name} is not set.") return v def _sanitize_ident(slug): """MySQL-safe identifier fragment from a slug (alnum + underscore).""" return re.sub(r'[^a-z0-9]', '_', slug.lower()) # Special chars chosen to be safe in SQL parameters, shell, and URLs # (no quotes, backslash, %, &, ;, space, /, :). _PW_SPECIAL = '!#^*+=-_@' def _gen_password(length=28): """Random password guaranteed to satisfy MySQL validate_password MEDIUM: at least one lowercase, uppercase, digit, and special character.""" alphabet = string.ascii_letters + string.digits + _PW_SPECIAL while True: chars = [ secrets.choice(string.ascii_lowercase), secrets.choice(string.ascii_uppercase), secrets.choice(string.digits), secrets.choice(_PW_SPECIAL), ] chars += [secrets.choice(alphabet) for _ in range(length - 4)] secrets.SystemRandom().shuffle(chars) pw = ''.join(chars) # guard against a leading '-' (awkward in some CLIs/URLs) if pw[0] != '-': return pw def db_name_for(slug): return f'jqc_{_sanitize_ident(slug)}'[:64] def db_user_for(slug): return f'jqc_{_sanitize_ident(slug)}_u'[:32] def _provision_engine(): return create_engine(_require('PROVISION_DB_URL'), future=True) def _default_db_host(): try: return make_url(_require('PROVISION_DB_URL')).host or '127.0.0.1' except Exception: return '127.0.0.1' def _base_domain(override=None): return override or _require('TENANT_BASE_DOMAIN') def setup_link(primary_domain, token, scheme='https'): return f'{scheme}://{primary_domain}{SET_PASSWORD_PATH}/{token}' # ── MySQL admin operations ─────────────────────────────────────────────────── def create_mysql_db_and_user(db_name, db_user, password, user_host='%'): """Idempotently create the tenant database, user, and scoped grant.""" eng = _provision_engine() try: with eng.connect() as c: c.execution_options(isolation_level='AUTOCOMMIT') c.execute(text( f"CREATE DATABASE IF NOT EXISTS `{db_name}` " f"CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci")) c.execute(text( f"CREATE USER IF NOT EXISTS '{db_user}'@'{user_host}' " f"IDENTIFIED BY :pw"), {'pw': password}) # Ensure the stored password matches (in case the user pre-existed). c.execute(text( f"ALTER USER '{db_user}'@'{user_host}' IDENTIFIED BY :pw"), {'pw': password}) c.execute(text( f"GRANT ALL PRIVILEGES ON `{db_name}`.* " f"TO '{db_user}'@'{user_host}'")) c.execute(text("FLUSH PRIVILEGES")) finally: eng.dispose() def drop_mysql_db_and_user(db_name, db_user, user_host='%'): """Best-effort teardown (used for rollback on failed provisioning).""" eng = _provision_engine() try: with eng.connect() as c: c.execution_options(isolation_level='AUTOCOMMIT') c.execute(text(f"DROP DATABASE IF EXISTS `{db_name}`")) c.execute(text(f"DROP USER IF EXISTS '{db_user}'@'{user_host}'")) c.execute(text("FLUSH PRIVILEGES")) finally: eng.dispose() # ── admin seeding (writes into the tenant DB) ──────────────────────────────── def seed_admin(db_uri, email, username=None, full_name=None, expires_hours=72): """Insert the first admin into a tenant DB with a set-password token. Returns (username, token). The account is created with password_set=0 and an unusable placeholder hash; the admin completes setup via the returned link. """ username = username or re.sub(r'[^a-zA-Z0-9_.-]', '', email.split('@')[0]) or 'admin' full_name = full_name or username token = secrets.token_hex(32) placeholder = generate_password_hash(secrets.token_urlsafe(32)) now = now_eastern() from datetime import timedelta expires = now + timedelta(hours=expires_hours) eng = create_engine(db_uri, future=True) try: with eng.begin() as c: c.execute(text(""" INSERT INTO users (username, full_name, email, password_hash, role, created_at, active, password_set, set_password_token, set_password_token_expires) VALUES (:u, :fn, :em, :ph, 'admin', :ca, 1, 0, :tok, :exp) """), {'u': username, 'fn': full_name, 'em': email, 'ph': placeholder, 'ca': now, 'tok': token, 'exp': expires}) finally: eng.dispose() return username, token # ── domain registration ────────────────────────────────────────────────────── def _upsert_domain(session, tenant_id, domain, kind, is_primary, verified, tls_status, verification_token=None): """Delete any stale row for this domain then insert fresh. Handles orphans from failed partial runs that left domain rows behind.""" existing = session.query(TenantDomain).filter_by(domain=domain).first() if existing is not None: session.delete(existing) session.flush() session.add(TenantDomain( tenant_id=tenant_id, domain=domain, kind=kind, is_primary=is_primary, verified=verified, verification_token=verification_token, tls_status=tls_status, created_at=now_eastern())) def _add_domains(session, tenant_id, slug, base_domain, custom_domain=None): primary = f'{slug}.{base_domain}'.lower() _upsert_domain(session, tenant_id, primary, 'subdomain', is_primary=True, verified=True, tls_status='pending') if custom_domain: _upsert_domain(session, tenant_id, custom_domain.lower(), 'custom', is_primary=False, verified=False, tls_status='pending', verification_token=secrets.token_hex(16)) return primary # ── public: create a brand-new tenant ──────────────────────────────────────── def create_tenant(slug, name, plan_code, admin_email, admin_username=None, admin_full_name=None, custom_domain=None, base_domain=None, db_host=None, user_host='%'): if not _SLUG_RE.match(slug or ''): raise ValueError(f"Invalid slug '{slug}' (must be a DNS label).") base = _base_domain(base_domain) host = db_host or _default_db_host() dbname = db_name_for(slug) dbuser = db_user_for(slug) password = _gen_password() with control_session() as s: if s.query(Tenant).filter_by(slug=slug).first(): raise ValueError(f"Tenant slug '{slug}' already exists.") plan = s.query(Plan).filter_by(code=plan_code).first() if plan is None: raise ValueError(f"Unknown plan '{plan_code}'.") plan_id = plan.id job_id = None tenant_id = None db_created = False try: with control_session() as s: job = ProvisioningJob(action='create_db', status='running', created_at=now_eastern()) s.add(job); s.flush(); job_id = job.id create_mysql_db_and_user(dbname, dbuser, password, user_host) db_created = True with control_session() as s: t = Tenant(slug=slug, name=name, plan_id=plan_id, status='provisioning', db_host=host, db_port=3306, db_name=dbname, db_user=dbuser, created_at=now_eastern()) t.set_db_password(password) s.add(t); s.flush() tenant_id = t.id db_uri = t.db_uri j = s.get(ProvisioningJob, job_id) if j: j.tenant_id = tenant_id # schema: baseline + stamp head (records its own migrate job + alembic_head) class _Ref: pass ref = _Ref(); ref.id = tenant_id; ref.db_uri = db_uri bootstrap_tenant(ref) # first admin + setup link username, token = seed_admin(db_uri, admin_email, admin_username, admin_full_name) with control_session() as s: primary = _add_domains(s, tenant_id, slug, base, custom_domain) t = s.get(Tenant, tenant_id) t.status = 'active' j = s.get(ProvisioningJob, job_id) if j: j.status = 'ok'; j.finished_at = now_eastern() j.log = f'provisioned tenant {slug} (db={dbname})' return { 'tenant_id': tenant_id, 'slug': slug, 'db_name': dbname, 'db_user': dbuser, 'primary_domain': primary, 'custom_domain': custom_domain, 'admin_username': username, 'setup_link': setup_link(primary, token), } except Exception as e: # best-effort rollback so a retry starts clean if job_id is not None: with control_session() as s: j = s.get(ProvisioningJob, job_id) if j: j.status = 'failed'; j.finished_at = now_eastern() j.log = f'{type(e).__name__}: {e}' if tenant_id is not None: with control_session() as s: s.query(TenantDomain).filter_by(tenant_id=tenant_id).delete() t = s.get(Tenant, tenant_id) if t: s.delete(t) if db_created: try: drop_mysql_db_and_user(dbname, dbuser, user_host) except Exception: pass raise # ── public: adopt an existing DB as tenant-zero ────────────────────────────── def register_tenant_zero(slug, name, plan_code, db_host, db_name, db_user, db_password, custom_domain=None, base_domain=None, db_port=3306): if not _SLUG_RE.match(slug or ''): raise ValueError(f"Invalid slug '{slug}' (must be a DNS label).") base = _base_domain(base_domain) with control_session() as s: if s.query(Tenant).filter_by(slug=slug).first(): raise ValueError(f"Tenant slug '{slug}' already exists.") plan = s.query(Plan).filter_by(code=plan_code).first() if plan is None: raise ValueError(f"Unknown plan '{plan_code}'.") plan_id = plan.id # build a temporary tenant to read the existing DB's current head with control_session() as s: t = Tenant(slug=slug, name=name, plan_id=plan_id, status='active', db_host=db_host, db_port=db_port, db_name=db_name, db_user=db_user, created_at=now_eastern()) t.set_db_password(db_password) s.add(t); s.flush() tenant_id = t.id db_uri = t.db_uri head = current_revision(db_uri) # read existing alembic_version (no changes) with control_session() as s: t = s.get(Tenant, tenant_id) t.alembic_head = head primary = _add_domains(s, tenant_id, slug, base, None) if custom_domain: # tenant-zero's existing domain is trusted/verified (already live) _upsert_domain(s, tenant_id, custom_domain.lower(), 'custom', is_primary=False, verified=True, tls_status='active') s.add(ProvisioningJob(tenant_id=tenant_id, action='create_db', status='ok', created_at=now_eastern(), finished_at=now_eastern(), log=f'registered tenant-zero {slug} in place (head={head})')) return {'tenant_id': tenant_id, 'slug': slug, 'db_name': db_name, 'primary_domain': primary, 'custom_domain': custom_domain, 'alembic_head': head} # ── public: delete / deregister a tenant ───────────────────────────────────── def delete_tenant(slug, drop_db=False, user_host='%'): """Remove a tenant's control records. Optionally drop its MySQL database. Safety: the MySQL drop only proceeds when the stored db_name matches the provisioner's naming convention (db_name_for(slug)). This refuses to drop an externally-named database such as tenant-zero's live LT database. Returns a summary dict. """ with control_session() as s: t = s.query(Tenant).filter_by(slug=slug).first() if t is None: raise ValueError(f"No tenant with slug '{slug}'.") tenant_id = t.id db_name = t.db_name db_user = t.db_user dropped = False if drop_db: if db_name != db_name_for(slug): raise ValueError( f"Refusing to drop database '{db_name}': it does not match the " f"provisioner naming ('{db_name_for(slug)}'). Drop it manually if " f"you are certain (this guard protects adopted DBs like tenant-zero).") drop_mysql_db_and_user(db_name, db_user, user_host) dropped = True with control_session() as s: # Delete by tenant_id (normal path) AND by domain string (catches orphans # from partial failed runs where the tenant row was rolled back but domain # rows were already flushed to disk). s.query(TenantDomain).filter_by(tenant_id=tenant_id).delete() s.query(ProvisioningJob).filter_by(tenant_id=tenant_id).delete() t = s.get(Tenant, tenant_id) if t: s.delete(t) # Purge any orphan domain rows for this slug's derived domains. for orphan_domain in [f'{slug}.{os.environ.get("TENANT_BASE_DOMAIN", "")}']: if orphan_domain.strip('.'): s.query(TenantDomain).filter_by(domain=orphan_domain.lower()).delete() return {'slug': slug, 'tenant_id': tenant_id, 'db_dropped': dropped, 'db_name': db_name} # ── CLI ────────────────────────────────────────────────────────────────────── def _cmd_create(args): info = create_tenant( slug=args.slug, name=args.name, plan_code=args.plan, admin_email=args.admin_email, admin_username=args.admin_username, admin_full_name=args.admin_name, custom_domain=args.custom_domain, base_domain=args.base_domain, db_host=args.db_host, user_host=args.user_host) print('Tenant provisioned:') for k in ('tenant_id', 'slug', 'db_name', 'db_user', 'primary_domain', 'custom_domain', 'admin_username'): print(f' {k}: {info[k]}') print(f'\n Admin setup link (send to {args.admin_email}):\n {info["setup_link"]}') return 0 def _cmd_register_zero(args): info = register_tenant_zero( slug=args.slug, name=args.name, plan_code=args.plan, db_host=args.db_host, db_name=args.db_name, db_user=args.db_user, db_password=args.db_password, custom_domain=args.custom_domain, base_domain=args.base_domain, db_port=args.db_port) print('Tenant-zero registered (no data moved):') for k, v in info.items(): print(f' {k}: {v}') return 0 def _cmd_delete(args): if not args.yes: print('Refusing to delete without --yes.') return 1 info = delete_tenant(args.slug, drop_db=args.drop_db, user_host=args.user_host) print('Tenant deleted:') for k, v in info.items(): print(f' {k}: {v}') return 0 def main(argv=None): p = argparse.ArgumentParser(prog='control.provision', description='Tenant provisioning') sub = p.add_subparsers(dest='command', required=True) c = sub.add_parser('create-tenant', help='Provision a brand-new tenant') c.add_argument('--slug', required=True) c.add_argument('--name', required=True) c.add_argument('--plan', required=True, help='plan code, e.g. pro') c.add_argument('--admin-email', required=True) c.add_argument('--admin-username', default=None) c.add_argument('--admin-name', default=None) c.add_argument('--custom-domain', default=None) c.add_argument('--base-domain', default=None, help='overrides TENANT_BASE_DOMAIN') c.add_argument('--db-host', default=None, help='host the app uses to reach the tenant DB') c.add_argument('--user-host', default='%') c.set_defaults(func=_cmd_create) z = sub.add_parser('register-tenant-zero', help='Adopt an existing DB in place') z.add_argument('--slug', required=True) z.add_argument('--name', required=True) z.add_argument('--plan', required=True) z.add_argument('--db-host', required=True) z.add_argument('--db-name', required=True) z.add_argument('--db-user', required=True) z.add_argument('--db-password', required=True) z.add_argument('--db-port', type=int, default=3306) z.add_argument('--custom-domain', default=None, help="tenant-zero's existing live domain") z.add_argument('--base-domain', default=None) z.set_defaults(func=_cmd_register_zero) d = sub.add_parser('delete-tenant', help='Remove a tenant (control records; optional DB drop)') d.add_argument('--slug', required=True) d.add_argument('--drop-db', action='store_true', help='also DROP the MySQL database + user (guarded to provisioner-named DBs)') d.add_argument('--user-host', default='%') d.add_argument('--yes', action='store_true', help='confirm deletion') d.set_defaults(func=_cmd_delete) args = p.parse_args(argv) sys.exit(args.func(args)) if __name__ == '__main__': main()