Spaces:
Running
Running
File size: 5,534 Bytes
c91c7db 878ab1e c91c7db 3493993 878ab1e c91c7db 3493993 c91c7db 878ab1e c91c7db 878ab1e c91c7db 878ab1e c91c7db | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 | 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()
|