fix(cli): don't let bm cloud setup overwrite an existing rclone remote

Closes #922. `bm cloud setup [--workspace X]` re-minted credentials and
overwrote the target rclone remote with no warning — most dangerously, a
default-workspace setup silently repointing the shared `basic-memory-cloud`
remote from one tenant to another (surfaced while live-testing #920).

Setup now aborts if the target remote already exists unless `--force` is given.
The check runs before any credentials are minted, so an abort wastes nothing.

The credential-leak half (re-mint leaves the prior key unrevoked on the tenant
bucket) is a backend concern tracked in #922 for basic-memory-cloud.

Signed-off-by: phernandez <paul@basicmachines.co>

Author: phernandez

src/basic_memory/cli/commands/cloud/core_commands.py

@@ -28,6 +28,7 @@
 )
 from basic_memory.cli.commands.cloud.rclone_config import (
     configure_rclone_remote,
+    rclone_remote_exists,
     remote_name_for_workspace,
 )
 from basic_memory.cli.commands.cloud.rclone_installer import (
@@ -207,6 +208,11 @@ def setup(
         help="Set up sync for a specific workspace (slug, name, or tenant_id). "
         "Omit for your default workspace.",
     ),
+    force: bool = typer.Option(
+        False,
+        "--force",
+        help="Reconfigure an rclone remote that already exists (mints new credentials).",
+    ),
 ) -> None:
     """Set up cloud sync by installing rclone and configuring credentials.
 
@@ -239,6 +245,20 @@ def setup(
             workspace_id = None  # default tenant
             remote_name = remote_name_for_workspace(None, is_default=True)
 
+        # Trigger: the target rclone remote already exists.
+        # Why: re-running setup mints new credentials and overwrites the remote,
+        # silently repointing it (e.g. clobbering the shared basic-memory-cloud
+        # that served another tenant) — the #922 footgun. Checked BEFORE minting
+        # so an abort wastes no credentials.
+        # Outcome: stop unless the user explicitly opts in with --force.
+        if rclone_remote_exists(remote_name) and not force:
+            console.print(f"[red]rclone remote '{remote_name}' is already configured.[/red]")
+            console.print(
+                "Re-running setup mints new credentials and overwrites it. "
+                "Pass --force to reconfigure."
+            )
+            raise typer.Exit(1)
+
         # Step 2: Get tenant info (scoped to the target workspace when given)
         console.print("\n[blue]Step 2: Getting tenant information...[/blue]")
         tenant_info = run_with_cleanup(get_mount_info(workspace_id=workspace_id))

tests/cli/cloud/test_project_sync_command.py

@@ -625,42 +625,80 @@ def test_get_workspace_for_project_override_no_match_raises(monkeypatch, config_
     assert "No accessible workspace matches 'acme'" in str(exc_info.value)
 
 
-def test_cloud_setup_workspace_configures_named_remote(monkeypatch):
-    """`bm cloud setup --workspace acme` provisions the acme tenant's own remote."""
-    core = importlib.import_module("basic_memory.cli.commands.cloud.core_commands")
-    recorder: dict = {}
-
+def _stub_setup_env(monkeypatch, core, *, remote_exists=False, recorder=None):
+    """Stub the `bm cloud setup` dependency chain for the acme workspace."""
     monkeypatch.setattr(core, "install_rclone", lambda: None)
     monkeypatch.setattr(
         core,
         "get_available_workspaces",
         lambda: _async_value([_workspace("team-tenant", "organization", "acme")]),
     )
+    monkeypatch.setattr(core, "rclone_remote_exists", lambda _remote: remote_exists)
+
+    def _mint(_tenant_id):
+        if recorder is not None:
+            recorder["minted"] = True
+        return _async_value(SimpleNamespace(access_key="ak", secret_key="sk"))
+
     monkeypatch.setattr(
         core,
         "get_mount_info",
         lambda **_kwargs: _async_value(
             SimpleNamespace(tenant_id="team-tenant", bucket_name="acme-bucket")
         ),
     )
-    monkeypatch.setattr(
-        core,
-        "generate_mount_credentials",
-        lambda _tenant_id: _async_value(SimpleNamespace(access_key="ak", secret_key="sk")),
-    )
+    monkeypatch.setattr(core, "generate_mount_credentials", _mint)
 
     def _fake_configure(**kwargs):
-        recorder.update(kwargs)
+        if recorder is not None:
+            recorder.update(kwargs)
         return kwargs.get("remote_name")
 
     monkeypatch.setattr(core, "configure_rclone_remote", _fake_configure)
 
+
+def test_cloud_setup_workspace_configures_named_remote(monkeypatch):
+    """`bm cloud setup --workspace acme` provisions the acme tenant's own remote."""
+    core = importlib.import_module("basic_memory.cli.commands.cloud.core_commands")
+    recorder: dict = {}
+    _stub_setup_env(monkeypatch, core, remote_exists=False, recorder=recorder)
+
     result = runner.invoke(app, ["cloud", "setup", "--workspace", "acme"])
 
     assert result.exit_code == 0, result.output
     assert recorder["remote_name"] == "basic-memory-cloud-acme"
 
 
+def test_cloud_setup_aborts_when_remote_exists_without_force(monkeypatch):
+    """Setup refuses to overwrite an existing remote, and mints nothing (the #922 footgun)."""
+    core = importlib.import_module("basic_memory.cli.commands.cloud.core_commands")
+    recorder: dict = {}
+    _stub_setup_env(monkeypatch, core, remote_exists=True, recorder=recorder)
+
+    result = runner.invoke(app, ["cloud", "setup", "--workspace", "acme"])
+
+    assert result.exit_code == 1, result.output
+    output = " ".join(result.output.split())
+    assert "basic-memory-cloud-acme' is already configured" in output
+    assert "--force" in output
+    # Aborted before minting credentials or touching the remote.
+    assert "minted" not in recorder
+    assert "remote_name" not in recorder
+
+
+def test_cloud_setup_force_overwrites_existing_remote(monkeypatch):
+    """--force reconfigures an existing remote."""
+    core = importlib.import_module("basic_memory.cli.commands.cloud.core_commands")
+    recorder: dict = {}
+    _stub_setup_env(monkeypatch, core, remote_exists=True, recorder=recorder)
+
+    result = runner.invoke(app, ["cloud", "setup", "--workspace", "acme", "--force"])
+
+    assert result.exit_code == 0, result.output
+    assert recorder["remote_name"] == "basic-memory-cloud-acme"
+    assert recorder.get("minted") is True
+
+
 async def _async_value(value):
     return value