Apply PR #7 security changes directly to main: app/security/cli.py

app/security/cli.py

@@ -3,12 +3,16 @@ 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:
@@ -33,6 +37,58 @@ async def _create(arguments: argparse.Namespace) -> None:
         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")
@@ -49,6 +105,15 @@ def main() -> None:
         "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")
@@ -57,6 +122,8 @@ def main() -> 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))