admin/basicmachines-co-basic-memory
1
0
Fork 0
mirror of https://github.com/basicmachines-co/basic-memory synced 2026-06-21 13:47:35 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(cli): route share commands to the configured workspace and encode project filter

Browse Source 
The `bm cloud share` commands authenticated but never sent the
X-Workspace-ID header that sibling cloud calls (cloud_utils._workspace_headers)
use for workspace scoping. The cloud /api/shares endpoints resolve the target
tenant from that header (resolve_workspace in basic-memory-cloud deps.py), so
share create/list/update/revoke for a team-workspace project were evaluated
against the caller's default tenant.

Resolve the workspace via resolve_configured_workspace (explicit --workspace >
project's configured workspace_id > global default) and pass X-Workspace-ID on
all four commands, mirroring the established cloud command pattern. Adds a
--workspace option to each, matching `bm cloud pull/push`.

Also URL-encode the list `--project` filter with urllib.parse.urlencode so
project names containing query-reserved characters (&, +, #, spaces) reach the
server as a single faithful value instead of splitting into stray query params.

Extends the mocked tests to assert the header is built/routed and that a
project name with special characters is percent-encoded.

Co-Authored-By: Claude <noreply@anthropic.com>
Signed-off-by: Drew Cain <groksrc@gmail.com>
This commit is contained in:
Drew Cain
2026-06-11 09:39:32 -05:00
parent 22901c366d
commit a8e452d9b4
2 changed files with 219 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/basic_memory/cli/commands/cloud/shares.py
+59 -1
View File
@@ -15,6 +15,7 @@ helper, matching the `snapshot.py` command group.
import asyncio
from datetime import datetime
from typing import Optional
from urllib.parse import urlencode
import typer
from rich.console import Console
@@ -26,10 +27,36 @@ from basic_memory.cli.commands.cloud.api_client import (
make_api_request,
)
from basic_memory.config import ConfigManager
from basic_memory.mcp.async_client import resolve_configured_workspace
console = Console()
share_app = typer.Typer(help="Manage public share links for notes")
# Header the cloud uses to route a request to a specific tenant's workspace.
# Mirrors basic_memory.cli.commands.cloud.cloud_utils._workspace_headers: the
# cloud /api/shares endpoints resolve the workspace from X-Workspace-ID (see
# resolve_workspace in basic-memory-cloud deps.py), so without it a team
# workspace project would be evaluated against the caller's default tenant.
WORKSPACE_ID_HEADER = "X-Workspace-ID"
def _workspace_headers(
*,
project_name: Optional[str] = None,
workspace: Optional[str] = None,
) -> dict[str, str]:
"""Resolve the target workspace and build the routing header, if any.
Resolution chain (see resolve_configured_workspace): explicit --workspace,
then the project's configured workspace_id, then the global default. Returns
an empty dict when nothing resolves so the request falls back to the
caller's default tenant exactly as before.
"""
resolved = resolve_configured_workspace(project_name=project_name, workspace=workspace)
if resolved is None:
return {}
return {WORKSPACE_ID_HEADER: resolved}
def _format_timestamp(iso_timestamp: Optional[str]) -> str:
"""Format an ISO timestamp to a human-readable form, or '-' when absent."""
@@ -88,12 +115,18 @@ def create(
"-e",
help="Optional expiration date/time (ISO 8601, e.g. 2025-12-31)",
),
workspace: Optional[str] = typer.Option(
None,
"--workspace",
help="Workspace (slug, name, or tenant_id) when the project name is ambiguous",
),
) -> None:
"""Create a public share link for a note.
Examples:
bm cloud share create my-project notes/my-idea
bm cloud share create my-project notes/my-idea --expires-at 2025-12-31
bm cloud share create my-project notes/my-idea --workspace acme
"""
# Validate --expires-at before any async/API work so a parse error surfaces
@@ -118,6 +151,7 @@ def create(
method="POST",
url=f"{host_url}/api/shares",
json_data=payload,
headers=_workspace_headers(project_name=project, workspace=workspace),
)
data = response.json()
@@ -153,12 +187,18 @@ def list_shares(
"-p",
help="Filter shares by project name",
),
workspace: Optional[str] = typer.Option(
None,
"--workspace",
help="Workspace (slug, name, or tenant_id) when the project name is ambiguous",
),
) -> None:
"""List public share links.
Examples:
bm cloud share list
bm cloud share list --project my-project
bm cloud share list --project my-project --workspace acme
"""
async def _list():
@@ -169,13 +209,17 @@ def list_shares(
url = f"{host_url}/api/shares"
if project:
url += f"?project_name={project}"
# Encode the filter so project names with query-reserved
# characters (&, +, #, spaces) reach the server intact rather
# than being parsed as extra query parameters.
url += f"?{urlencode({'project_name': project})}"
console.print("[blue]Fetching share links...[/blue]")
response = await make_api_request(
method="GET",
url=url,
headers=_workspace_headers(project_name=project, workspace=workspace),
)
data = response.json()
@@ -248,6 +292,11 @@ def update(
"-e",
help="New expiration date/time (ISO 8601). Use 'none' to clear it.",
),
workspace: Optional[str] = typer.Option(
None,
"--workspace",
help="Workspace (slug, name, or tenant_id) the share belongs to",
),
) -> None:
"""Update a share link: enable/disable it or change its expiration.
@@ -256,6 +305,7 @@ def update(
bm cloud share update abc123 --enable
bm cloud share update abc123 --expires-at 2026-01-01
bm cloud share update abc123 --expires-at none
bm cloud share update abc123 --disable --workspace acme
"""
async def _update():
@@ -294,6 +344,7 @@ def update(
method="PATCH",
url=f"{host_url}/api/shares/{token}",
json_data=payload,
headers=_workspace_headers(workspace=workspace),
)
data = response.json()
@@ -333,12 +384,18 @@ def revoke(
"-f",
help="Skip confirmation prompt",
),
workspace: Optional[str] = typer.Option(
None,
"--workspace",
help="Workspace (slug, name, or tenant_id) the share belongs to",
),
) -> None:
"""Revoke (delete) a public share link.
Examples:
bm cloud share revoke abc123
bm cloud share revoke abc123 --force
bm cloud share revoke abc123 --force --workspace acme
"""
async def _revoke():
@@ -358,6 +415,7 @@ def revoke(
await make_api_request(
method="DELETE",
url=f"{host_url}/api/shares/{token}",
headers=_workspace_headers(workspace=workspace),
)
console.print(f"[green]Share {token} revoked successfully[/green]")
tests/cli/test_share_commands.py
+160 -6
View File
@@ -38,6 +38,19 @@ def _mock_config_manager():
return mock_config_manager
def _patch_workspace(resolved):
"""Patch the workspace resolver used by the share commands.
Returns whatever ``resolved`` is for every lookup, so tests can assert the
X-Workspace-ID header is built (and routed) the way the cloud expects
without depending on real config files.
"""
return patch(
"basic_memory.cli.commands.cloud.shares.resolve_configured_workspace",
return_value=resolved,
)
class TestShareCreateCommand:
"""Tests for 'bm cloud share create' command."""
@@ -52,6 +65,7 @@ class TestShareCreateCommand:
async def mock_make_api_request(*args, **kwargs):
captured["json_data"] = kwargs.get("json_data")
captured["headers"] = kwargs.get("headers")
return mock_response
with patch(
@@ -62,9 +76,10 @@ class TestShareCreateCommand:
"basic_memory.cli.commands.cloud.shares.ConfigManager",
return_value=_mock_config_manager(),
):
result = runner.invoke(
app, ["cloud", "share", "create", "my-project", "notes/my-idea"]
)
with _patch_workspace("tenant-xyz"):
result = runner.invoke(
app, ["cloud", "share", "create", "my-project", "notes/my-idea"]
)
assert result.exit_code == 0
assert "Share link created successfully" in result.stdout
@@ -75,6 +90,90 @@ class TestShareCreateCommand:
"project_name": "my-project",
"note_permalink": "notes/my-idea",
}
# Workspace routing: the resolved tenant must travel as the
# X-Workspace-ID header so team-workspace projects aren't
# evaluated against the caller's default tenant.
assert captured["headers"] == {"X-Workspace-ID": "tenant-xyz"}
def test_create_share_explicit_workspace_routes_header(self):
"""--workspace overrides resolution and is passed through as the header."""
runner = CliRunner()
mock_response = Mock(spec=httpx.Response)
mock_response.status_code = 200
mock_response.json.return_value = SHARE_RESPONSE
captured = {}
async def mock_make_api_request(*args, **kwargs):
captured["headers"] = kwargs.get("headers")
return mock_response
seen = {}
def fake_resolve(*, project_name=None, workspace=None):
seen["project_name"] = project_name
seen["workspace"] = workspace
return workspace
with patch(
"basic_memory.cli.commands.cloud.shares.make_api_request",
side_effect=mock_make_api_request,
):
with patch(
"basic_memory.cli.commands.cloud.shares.ConfigManager",
return_value=_mock_config_manager(),
):
with patch(
"basic_memory.cli.commands.cloud.shares.resolve_configured_workspace",
side_effect=fake_resolve,
):
result = runner.invoke(
app,
[
"cloud",
"share",
"create",
"my-project",
"notes/my-idea",
"--workspace",
"acme",
],
)
assert result.exit_code == 0
assert seen == {"project_name": "my-project", "workspace": "acme"}
assert captured["headers"] == {"X-Workspace-ID": "acme"}
def test_create_share_no_workspace_sends_no_header(self):
"""When nothing resolves, no routing header is added (default tenant)."""
runner = CliRunner()
mock_response = Mock(spec=httpx.Response)
mock_response.status_code = 200
mock_response.json.return_value = SHARE_RESPONSE
captured = {}
async def mock_make_api_request(*args, **kwargs):
captured["headers"] = kwargs.get("headers")
return mock_response
with patch(
"basic_memory.cli.commands.cloud.shares.make_api_request",
side_effect=mock_make_api_request,
):
with patch(
"basic_memory.cli.commands.cloud.shares.ConfigManager",
return_value=_mock_config_manager(),
):
with _patch_workspace(None):
result = runner.invoke(
app, ["cloud", "share", "create", "my-project", "notes/my-idea"]
)
assert result.exit_code == 0
assert captured["headers"] == {}
def test_create_share_with_expires_at(self):
runner = CliRunner()
@@ -290,6 +389,7 @@ class TestShareListCommand:
async def mock_make_api_request(*args, **kwargs):
captured["url"] = kwargs.get("url", args[1] if len(args) > 1 else "")
captured["headers"] = kwargs.get("headers")
return mock_response
with patch(
@@ -300,10 +400,52 @@ class TestShareListCommand:
"basic_memory.cli.commands.cloud.shares.ConfigManager",
return_value=_mock_config_manager(),
):
result = runner.invoke(app, ["cloud", "share", "list", "--project", "my-project"])
with _patch_workspace("tenant-xyz"):
result = runner.invoke(
app, ["cloud", "share", "list", "--project", "my-project"]
)
assert result.exit_code == 0
assert "project_name=my-project" in captured["url"]
assert captured["headers"] == {"X-Workspace-ID": "tenant-xyz"}
def test_list_shares_project_filter_url_encoded(self):
"""Project names with query-reserved chars must be percent-encoded.
A name like "R&D+notes #1" interpolated raw would split into bogus
query params (project_name=R, plus a stray "D+notes #1" key); encoding
keeps it a single faithful project_name value.
"""
runner = CliRunner()
mock_response = Mock(spec=httpx.Response)
mock_response.status_code = 200
mock_response.json.return_value = {"shares": [SHARE_RESPONSE], "total": 1}
captured = {}
async def mock_make_api_request(*args, **kwargs):
captured["url"] = kwargs.get("url", args[1] if len(args) > 1 else "")
return mock_response
with patch(
"basic_memory.cli.commands.cloud.shares.make_api_request",
side_effect=mock_make_api_request,
):
with patch(
"basic_memory.cli.commands.cloud.shares.ConfigManager",
return_value=_mock_config_manager(),
):
with _patch_workspace(None):
result = runner.invoke(
app, ["cloud", "share", "list", "--project", "R&D+notes #1"]
)
assert result.exit_code == 0
# Reserved characters are percent-encoded into a single value.
assert "project_name=R%26D%2Bnotes+%231" in captured["url"]
# And the raw, ambiguous form never reaches the wire.
assert "project_name=R&D" not in captured["url"]
def test_list_shares_api_error(self):
runner = CliRunner()
@@ -340,6 +482,7 @@ class TestShareUpdateCommand:
async def mock_make_api_request(*args, **kwargs):
captured["json_data"] = kwargs.get("json_data")
captured["method"] = kwargs.get("method")
captured["headers"] = kwargs.get("headers")
return mock_response
with patch(
@@ -350,12 +493,17 @@ class TestShareUpdateCommand:
"basic_memory.cli.commands.cloud.shares.ConfigManager",
return_value=_mock_config_manager(),
):
result = runner.invoke(app, ["cloud", "share", "update", "abc123", "--disable"])
with _patch_workspace("tenant-xyz"):
result = runner.invoke(
app,
["cloud", "share", "update", "abc123", "--disable", "--workspace", "acme"],
)
assert result.exit_code == 0
assert "updated successfully" in result.stdout
assert captured["method"] == "PATCH"
assert captured["json_data"] == {"enabled": False}
assert captured["headers"] == {"X-Workspace-ID": "tenant-xyz"}
def test_update_enable(self):
runner = CliRunner()
@@ -516,6 +664,7 @@ class TestShareRevokeCommand:
async def mock_make_api_request(*args, **kwargs):
captured["method"] = kwargs.get("method")
captured["url"] = kwargs.get("url")
captured["headers"] = kwargs.get("headers")
return mock_response
with patch(
@@ -526,12 +675,17 @@ class TestShareRevokeCommand:
"basic_memory.cli.commands.cloud.shares.ConfigManager",
return_value=_mock_config_manager(),
):
result = runner.invoke(app, ["cloud", "share", "revoke", "abc123", "--force"])
with _patch_workspace("tenant-xyz"):
result = runner.invoke(
app,
["cloud", "share", "revoke", "abc123", "--force", "--workspace", "acme"],
)
assert result.exit_code == 0
assert "revoked successfully" in result.stdout
assert captured["method"] == "DELETE"
assert captured["url"].endswith("/api/shares/abc123")
assert captured["headers"] == {"X-Workspace-ID": "tenant-xyz"}
def test_revoke_cancelled(self):
runner = CliRunner()