Spaces:
Running
Running
| from __future__ import annotations | |
| import argparse | |
| import asyncio | |
| import json | |
| from uuid import uuid4 | |
| from sqlalchemy import select | |
| from app.core.config import get_settings | |
| from app.security.database import SecurityDatabase | |
| from app.security.schemas import APIKeyCreate, APIKeyView | |
| from app.security.service import APIKeyService | |
| from app.security.tenancy import TenantService | |
| from app.security.models import User, Workspace, WorkspaceMembership | |
| async def _create(arguments: argparse.Namespace) -> None: | |
| settings = get_settings() | |
| database = SecurityDatabase(settings.database_url, auto_migrate=settings.security_auto_migrate) | |
| service = APIKeyService(database, settings, TenantService(database)) | |
| await database.initialize() | |
| try: | |
| record, secret = await service.create( | |
| APIKeyCreate( | |
| name=arguments.name, | |
| environment=arguments.environment, | |
| role=arguments.role, | |
| expires_in_seconds=arguments.expires_in, | |
| ), | |
| created_by="security-cli", | |
| ) | |
| payload = APIKeyView.model_validate(record).model_dump(mode="json") | |
| payload["api_key"] = secret | |
| print(json.dumps(payload, indent=2)) | |
| finally: | |
| await database.close() | |
| async def _provision_studio_user(arguments: argparse.Namespace) -> None: | |
| """Explicit operator-driven onboarding; never grant access from a browser request.""" | |
| settings = get_settings() | |
| database = SecurityDatabase(settings.database_url, auto_migrate=settings.security_auto_migrate) | |
| await database.initialize() | |
| try: | |
| async with database.session() as session: | |
| existing = await session.scalar(select(User).where(User.subject == arguments.subject)) | |
| if existing is not None: | |
| raise SystemExit("That identity subject is already provisioned; refusing to reassign it.") | |
| workspace = await session.scalar( | |
| select(Workspace).where(Workspace.slug == arguments.workspace_slug) | |
| ) | |
| if workspace is not None: | |
| raise SystemExit("Workspace slug already exists; refusing to attach an initial owner implicitly.") | |
| user_id = str(uuid4()) | |
| workspace_id = str(uuid4()) | |
| user = User( | |
| id=user_id, | |
| subject=arguments.subject, | |
| display_name=arguments.display_name, | |
| metadata_json={"provisioned_by": "security-cli"}, | |
| ) | |
| workspace = Workspace( | |
| id=workspace_id, | |
| slug=arguments.workspace_slug, | |
| name=arguments.workspace_name, | |
| metadata_json={"provisioned_by": "security-cli"}, | |
| ) | |
| membership = WorkspaceMembership( | |
| id=str(uuid4()), | |
| workspace_id=workspace_id, | |
| user_id=user_id, | |
| role=arguments.role, | |
| status="active", | |
| ) | |
| session.add_all([user, workspace, membership]) | |
| await session.commit() | |
| print(json.dumps({ | |
| "user_id": user_id, | |
| "subject": arguments.subject, | |
| "workspace_id": workspace_id, | |
| "workspace_slug": arguments.workspace_slug, | |
| "membership_id": membership.id, | |
| "role": arguments.role, | |
| "status": "active", | |
| }, indent=2)) | |
| finally: | |
| await database.close() | |
| def _bootstrap(arguments: argparse.Namespace) -> None: | |
| material = APIKeyService.generate_material(arguments.environment) | |
| print("Store the API key in your password manager. It will not be shown again.\n") | |
| print(f"API_KEY={material.api_key}") | |
| print(f"AUTH_BOOTSTRAP_KEY_HASH={material.key_hash}") | |
| print(f"AUTH_BOOTSTRAP_KEY_PREFIX={material.key_prefix}") | |
| print(f"AUTH_BOOTSTRAP_ENVIRONMENT={material.environment}") | |
| def main() -> None: | |
| parser = argparse.ArgumentParser(description="MediaRouter API-key administration") | |
| commands = parser.add_subparsers(dest="command", required=True) | |
| bootstrap = commands.add_parser( | |
| "generate-bootstrap", help="Generate a key and hash-only bootstrap settings" | |
| ) | |
| bootstrap.add_argument("--environment", choices=("live", "test"), default="live") | |
| provision = commands.add_parser( | |
| "provision-studio-user", | |
| help="Explicitly provision a verified Supabase subject and initial workspace membership", | |
| ) | |
| provision.add_argument("--subject", required=True, help="Exact Supabase auth user UUID/subject") | |
| provision.add_argument("--display-name", required=True) | |
| provision.add_argument("--workspace-slug", required=True) | |
| provision.add_argument("--workspace-name", required=True) | |
| provision.add_argument("--role", choices=("owner", "admin", "member"), default="owner") | |
| create = commands.add_parser("create", help="Create a key directly in the configured database") | |
| create.add_argument("--name", required=True) | |
| create.add_argument("--environment", choices=("live", "test"), default="live") | |
| create.add_argument("--role", default="admin") | |
| create.add_argument("--expires-in", type=int, default=None) | |
| arguments = parser.parse_args() | |
| if arguments.command == "generate-bootstrap": | |
| _bootstrap(arguments) | |
| elif arguments.command == "provision-studio-user": | |
| asyncio.run(_provision_studio_user(arguments)) | |
| else: | |
| asyncio.run(_create(arguments)) | |
| if __name__ == "__main__": | |
| main() | |