feat: add reflex cloud scan security review command

packages/reflex-hosting-cli/news/6632.feature.md

@@ -0,0 +1 @@
+Added `reflex cloud scan`, which uploads your app source for a Reflex-aware security review and reports security and logic flaws. Supports `--json` output and a `--fail-on` severity gate for CI.

packages/reflex-hosting-cli/src/reflex_cli/utils/hosting.py

@@ -1456,6 +1456,135 @@ def create_deployment(
     return response.json()
 
 
+class SecurityReviewError(ResponseError):
+    """Raised when a security review request fails."""
+
+
+_SECURITY_REVIEW_PREFIX = "/api/v1/agents/security-review"
+
+
+def _security_review_detail(response: Any) -> str:
+    """Extract a human-readable ``detail`` from a failed review response.
+
+    Args:
+        response: The error response from the security review API.
+
+    Returns:
+        The server-provided detail, or a generic fallback if the body is not
+        a JSON object with a ``detail`` field.
+
+    """
+    try:
+        return str(response.json()["detail"])
+    except (ValueError, TypeError, KeyError):
+        return "internal server error"
+
+
+def submit_security_review(zip_bytes: bytes, client: AuthenticatedClient) -> str:
+    """Submit a zipped app for security review.
+
+    Uploads the archive straight to object storage via a presigned URL, then
+    submits the stored object for review.
+
+    Args:
+        zip_bytes: The zipped app source to review.
+        client: The authenticated client.
+
+    Returns:
+        The id of the submitted job, to be polled with ``get_security_review``.
+
+    Raises:
+        NotAuthenticatedError: If the token is not valid.
+        SecurityReviewError: If any step of the submission fails.
+
+    """
+    import httpx
+
+    if not isinstance(client, AuthenticatedClient):
+        raise NotAuthenticatedError("not authenticated")
+
+    auth = authorization_header(client.token)
+
+    # 1. Ask the API for a presigned URL to upload the archive directly.
+    upload_url_response = httpx.post(
+        urljoin(
+            constants.Hosting.HOSTING_SERVICE,
+            f"{_SECURITY_REVIEW_PREFIX}/jobs/upload-url",
+        ),
+        json={"content_length": len(zip_bytes), "content_type": "application/zip"},
+        headers=auth,
+        timeout=constants.Hosting.TIMEOUT,
+    )
+    try:
+        upload_url_response.raise_for_status()
+    except httpx.HTTPStatusError as ex:
+        raise SecurityReviewError(_security_review_detail(ex.response)) from ex
+    upload = upload_url_response.json()
+
+    # 2. Upload the bytes to storage. The presigned URL pins the content length
+    #    and type, so send the returned headers verbatim and let httpx derive
+    #    Content-Length from the body — setting it manually breaks the signature.
+    put_response = httpx.put(
+        upload["url"],
+        content=zip_bytes,
+        headers=upload.get("headers", {}),
+        timeout=120,
+    )
+    try:
+        put_response.raise_for_status()
+    except httpx.HTTPStatusError as ex:
+        raise SecurityReviewError("failed to upload app source for review") from ex
+
+    # 3. Submit the uploaded object for review.
+    response = httpx.post(
+        urljoin(constants.Hosting.HOSTING_SERVICE, f"{_SECURITY_REVIEW_PREFIX}/jobs"),
+        json={"key": upload["key"]},
+        headers=auth,
+        timeout=constants.Hosting.TIMEOUT,
+    )
+    try:
+        response.raise_for_status()
+    except httpx.HTTPStatusError as ex:
+        raise SecurityReviewError(_security_review_detail(ex.response)) from ex
+    return response.json()["job_id"]
+
+
+def get_security_review(job_id: str, client: AuthenticatedClient) -> dict[str, Any]:
+    """Poll a previously submitted security review job.
+
+    Args:
+        job_id: The id returned by ``submit_security_review``.
+        client: The authenticated client.
+
+    Returns:
+        The job status payload: ``status`` is one of ``pending``, ``complete``
+        or ``error``; ``result`` holds the review once ``complete``.
+
+    Raises:
+        NotAuthenticatedError: If the token is not valid.
+        SecurityReviewError: If the server returns an error.
+
+    """
+    import httpx
+
+    if not isinstance(client, AuthenticatedClient):
+        raise NotAuthenticatedError("not authenticated")
+
+    response = httpx.get(
+        urljoin(
+            constants.Hosting.HOSTING_SERVICE,
+            f"{_SECURITY_REVIEW_PREFIX}/jobs/{job_id}",
+        ),
+        headers=authorization_header(client.token),
+        timeout=constants.Hosting.TIMEOUT,
+    )
+    try:
+        response.raise_for_status()
+    except httpx.HTTPStatusError as ex:
+        raise SecurityReviewError(_security_review_detail(ex.response)) from ex
+    return response.json()
+
+
 def stop_app(app_id: str, client: AuthenticatedClient):
     """Stop a running application.
 

packages/reflex-hosting-cli/src/reflex_cli/v2/deployments.py

@@ -14,6 +14,7 @@
 from reflex_cli.v2.apps import apps_cli
 from reflex_cli.v2.gcp import deploy_command as gcp_deploy_command
 from reflex_cli.v2.project import project_cli
+from reflex_cli.v2.scan import scan_command
 from reflex_cli.v2.secrets import secrets_cli
 from reflex_cli.v2.vmtypes_regions import vm_types_regions_cli
 
@@ -69,6 +70,10 @@ def hosting_cli(ctx: click.Context) -> None:
     gcp_deploy_command,
     name="deploy",
 )
+hosting_cli.add_command(
+    scan_command,
+    name="scan",
+)
 for name, command in vm_types_regions_cli.commands.items():
     # Add the command to the hosting CLI
     hosting_cli.add_command(command, name=name)