diff --git a/nanobot/cli/commands.py b/nanobot/cli/commands.py index d5c6fbde7..790003760 100644 --- a/nanobot/cli/commands.py +++ b/nanobot/cli/commands.py @@ -72,6 +72,7 @@ from nanobot.bus.outbound_events import ( # noqa: E402 StreamEndEvent, outbound_event_from_message, ) +from nanobot.cli.extensions import create_extensions_app # noqa: E402 from nanobot.cli.gateway import create_gateway_app # noqa: E402 from nanobot.cli.stream import StreamRenderer, ThinkingSpinner # noqa: E402 from nanobot.config.paths import get_workspace_path, is_default_workspace # noqa: E402 @@ -2535,6 +2536,8 @@ def agent( # ============================================================================ +app.add_typer(create_extensions_app(console=console), name="extensions") + channels_app = typer.Typer(help="Manage channels") app.add_typer(channels_app, name="channels") diff --git a/nanobot/cli/extensions.py b/nanobot/cli/extensions.py new file mode 100644 index 000000000..d4624d767 --- /dev/null +++ b/nanobot/cli/extensions.py @@ -0,0 +1,213 @@ +"""Typer commands for installing and governing extensions.""" + +from __future__ import annotations + +import asyncio +from collections.abc import Callable +from typing import Any + +import typer +from rich.console import Console +from rich.table import Table + +from nanobot.extensions import ExtensionService + +ServiceFactory = Callable[[], ExtensionService] + + +def create_extensions_app( + *, + console: Console, + service_factory: ServiceFactory = ExtensionService, +) -> typer.Typer: + """Build the extension command group around the transport-neutral service.""" + app = typer.Typer(help="Discover, install, inspect, and govern extensions.") + + def service() -> ExtensionService: + return service_factory() + + def run(awaitable: Any) -> dict[str, Any]: + try: + return asyncio.run(awaitable) + except (KeyError, RuntimeError, ValueError) as exc: + console.print(f"[red]Error: {exc}[/red]") + raise typer.Exit(1) from exc + + @app.command("list") + def list_extensions() -> None: + """List installed extensions and their activation policy.""" + payload = run(service().status()) + table = Table(show_header=True, header_style="bold") + table.add_column("Extension") + table.add_column("Runtime") + table.add_column("State") + table.add_column("Trust") + table.add_column("Version") + for item in payload["extensions"]: + state = "active" if item["active"] else ("enabled" if item["enabled"] else "disabled") + table.add_row( + item["name"], + item["runtime"], + state, + "trusted" if item["trusted"] else "untrusted", + item["version"], + ) + console.print(table) + if not payload["extensions"]: + console.print("[dim]No extensions installed.[/dim]") + if payload["diagnostics"]: + console.print(f"[yellow]{len(payload['diagnostics'])} diagnostic(s)[/yellow]") + + @app.command("inspect") + def inspect_extension(extension_id: str = typer.Argument(..., help="Extension ID")) -> None: + """Show manifest, dependencies, permissions, and diagnostics.""" + payload = run(service().status()) + item = next( + (candidate for candidate in payload["extensions"] if candidate["id"] == extension_id), + None, + ) + if item is None: + console.print(f"[red]Extension not found: {extension_id}[/red]") + raise typer.Exit(1) + console.print(f"[bold]{item['name']}[/bold] [dim]{item['version']}[/dim]") + console.print(item["description"] or "[dim]No description.[/dim]") + console.print(f"Runtime: {item['runtime']} Scope: {item['scope']}") + console.print( + f"State: {'active' if item['active'] else 'inactive'} " + f"Trust: {'trusted' if item['trusted'] else 'untrusted'}" + ) + _print_named_rows(console, "Contributions", item["contributions"], "kind", "name") + _print_named_rows(console, "Dependencies", item["dependencies"], "kind", "name") + _print_permissions(console, item["permissions"], set(item["granted_permissions"])) + diagnostics = [ + diagnostic + for diagnostic in payload["diagnostics"] + if diagnostic["extension_id"] == extension_id + ] + if diagnostics: + console.print("\n[bold]Diagnostics[/bold]") + for diagnostic in diagnostics: + console.print( + f" [yellow]{diagnostic['code']}[/yellow] {diagnostic['message']}" + ) + + @app.command("search") + def search_extensions( + query: str = typer.Argument("", help="Package name or keyword"), + ecosystem: str = typer.Option( + "all", + "--ecosystem", + "-e", + help="all, nanobot, pi, or openclaw", + ), + limit: int = typer.Option(20, "--limit", min=1, max=100), + ) -> None: + """Search compatible extension packages on npm.""" + payload = run(service().search(query, ecosystem=ecosystem, limit=limit)) + table = Table(show_header=True, header_style="bold") + table.add_column("Package") + table.add_column("Ecosystem") + table.add_column("Version") + table.add_column("Description") + for package in payload["packages"]: + table.add_row( + package["name"], + package["ecosystem"], + package["version"], + package["description"], + ) + console.print(table) + if not payload["packages"]: + console.print("[dim]No compatible packages found.[/dim]") + + @app.command("install") + def install_extension( + source: str = typer.Argument(..., help="npm spec, Git URL, or local path"), + kind: str = typer.Option("npm", "--kind", help="npm, git, or local"), + ref: str = typer.Option("", "--ref", help="Git branch, tag, or commit"), + ) -> None: + """Install an extension without granting trust or permissions.""" + payload = run(service().install(source, kind=kind, ref=ref, trusted=False)) + record = payload["record"] + console.print( + f"[green]Installed {record['id']} {record['version']}[/green] " + "[yellow](untrusted)[/yellow]" + ) + console.print( + f"Review with [bold]nanobot extensions inspect {record['id']}[/bold], " + "then grant permissions and trust it explicitly." + ) + + def policy_command(name: str, value: bool, label: str, help_text: str) -> None: + @app.command(name, help=help_text) + def update(extension_id: str = typer.Argument(..., help="Extension ID")) -> None: + payload = run( + service().set_enabled(extension_id, value) + if name in {"enable", "disable"} + else service().set_trusted(extension_id, value) + ) + console.print(f"[green]{label}: {payload['record']['id']}[/green]") + + policy_command("enable", True, "Enabled", "Allow an installed extension to activate.") + policy_command("disable", False, "Disabled", "Prevent an installed extension from activating.") + policy_command("trust", True, "Trusted", "Trust an installed extension's executable code.") + policy_command("untrust", False, "Trust revoked", "Revoke trust and stop extension activation.") + + @app.command("permissions") + def set_permissions( + extension_id: str = typer.Argument(..., help="Extension ID"), + permissions: list[str] = typer.Argument( + None, + help="Exact permissions to grant; omit all to revoke every grant", + ), + ) -> None: + """Replace the extension's granted host permissions.""" + payload = run(service().set_permissions(extension_id, set(permissions or []))) + granted = payload["record"]["granted_permissions"] + console.print( + f"[green]Updated permissions for {extension_id}:[/green] " + + (", ".join(granted) if granted else "none") + ) + + @app.command("uninstall") + def uninstall_extension( + extension_id: str = typer.Argument(..., help="Extension ID"), + yes: bool = typer.Option(False, "--yes", "-y", help="Skip confirmation"), + ) -> None: + """Remove an installed extension.""" + if not yes and not typer.confirm(f"Uninstall extension '{extension_id}'?"): + raise typer.Abort() + run(service().uninstall(extension_id)) + console.print(f"[green]Uninstalled {extension_id}[/green]") + + return app + + +def _print_named_rows( + console: Console, + title: str, + rows: list[dict[str, Any]], + category_key: str, + name_key: str, +) -> None: + console.print(f"\n[bold]{title}[/bold]") + if not rows: + console.print(" [dim]None[/dim]") + return + for row in rows: + console.print(f" {row[category_key]}: {row[name_key]}") + + +def _print_permissions( + console: Console, + permissions: list[dict[str, str]], + granted: set[str], +) -> None: + console.print("\n[bold]Permissions[/bold]") + if not permissions: + console.print(" [dim]None requested[/dim]") + return + for permission in permissions: + status = "[green]granted[/green]" if permission["name"] in granted else "[yellow]pending[/yellow]" + reason = f" — {permission['reason']}" if permission["reason"] else "" + console.print(f" {permission['name']} ({status}){reason}") diff --git a/tests/cli/test_extensions.py b/tests/cli/test_extensions.py new file mode 100644 index 000000000..397769efa --- /dev/null +++ b/tests/cli/test_extensions.py @@ -0,0 +1,118 @@ +from __future__ import annotations + +from io import StringIO + +from rich.console import Console +from typer.testing import CliRunner + +from nanobot.cli.extensions import create_extensions_app + + +class _Service: + def __init__(self) -> None: + self.calls: list[tuple[str, object]] = [] + + async def status(self): + self.calls.append(("status", None)) + return { + "extensions": [ + { + "id": "sample", + "name": "Sample", + "version": "1.0.0", + "runtime": "pi", + "scope": "user", + "description": "Example extension", + "enabled": True, + "trusted": False, + "active": False, + "contributions": [], + "dependencies": [], + "permissions": [{"name": "network", "reason": "Fetch data"}], + "granted_permissions": [], + } + ], + "diagnostics": [], + } + + async def search(self, query, *, ecosystem, limit): + self.calls.append(("search", (query, ecosystem, limit))) + return {"packages": []} + + async def install(self, source, *, kind, ref, trusted): + self.calls.append(("install", (source, kind, ref, trusted))) + return { + "record": { + "id": "sample", + "version": "1.0.0", + "trusted": trusted, + } + } + + async def set_enabled(self, extension_id, enabled): + self.calls.append(("enabled", (extension_id, enabled))) + return {"record": {"id": extension_id}} + + async def set_trusted(self, extension_id, trusted): + self.calls.append(("trusted", (extension_id, trusted))) + return {"record": {"id": extension_id}} + + async def set_permissions(self, extension_id, permissions): + self.calls.append(("permissions", (extension_id, permissions))) + return { + "record": { + "id": extension_id, + "granted_permissions": sorted(permissions), + } + } + + async def uninstall(self, extension_id): + self.calls.append(("uninstall", extension_id)) + return {"removed": extension_id} + + +def _runner(service: _Service): + output = StringIO() + app = create_extensions_app( + console=Console(file=output, force_terminal=False), + service_factory=lambda: service, + ) + return CliRunner(), app, output + + +def test_extension_cli_inspects_and_searches() -> None: + service = _Service() + runner, app, output = _runner(service) + + inspected = runner.invoke(app, ["inspect", "sample"]) + searched = runner.invoke(app, ["search", "web", "--ecosystem", "pi", "--limit", "7"]) + + assert inspected.exit_code == 0 + assert searched.exit_code == 0 + assert "Sample" in output.getvalue() + assert service.calls == [ + ("status", None), + ("search", ("web", "pi", 7)), + ] + + +def test_extension_cli_install_and_policy_commands() -> None: + service = _Service() + runner, app, _output = _runner(service) + + assert runner.invoke(app, ["install", "pi-example"]).exit_code == 0 + assert runner.invoke(app, ["trust", "sample"]).exit_code == 0 + assert runner.invoke(app, ["disable", "sample"]).exit_code == 0 + assert runner.invoke( + app, + ["permissions", "sample", "network", "filesystem.read"], + ).exit_code == 0 + assert runner.invoke(app, ["uninstall", "sample", "--yes"]).exit_code == 0 + + assert service.calls == [ + ("install", ("pi-example", "npm", "", False)), + ("trusted", ("sample", True)), + ("enabled", ("sample", False)), + ("permissions", ("sample", {"network", "filesystem.read"})), + ("uninstall", "sample"), + ]