From 7b39ea4765c58ea9ee0f0b552f184ed22a8045dd Mon Sep 17 00:00:00 2001 From: Maxmilian Ottomansky Date: Wed, 22 Apr 2026 14:32:49 +0200 Subject: [PATCH 1/9] feat(0.22.0): project pin + firewall flags (FIIA P0-4) Adds an explicit project-pin UX + session-only firewall sugar that together close the FIIA P0-4 gap. New surface: - `kbagent project use ` -- pins default_project in config.json. The field already existed (auto-set on first add, auto-rotated on remove) but had no explicit CLI verb. Now it does. - `kbagent project current` -- prints effective alias + source (env / pin / none). Reports both the env override AND the persisted pin so misconfigurations surface loudly instead of silently. - Top-level `--deny-writes` / `--deny-destructive` flags -- synthesize a session-only PermissionPolicy and merge with any persisted policy. Never written to config.json. `--deny-writes` covers the wide net (write+destructive+admin); `--deny-destructive` narrowly covers data destruction so callers can keep build-up capabilities while blocking tear-down. - `KBAGENT_PROJECT` env var overrides the persisted pin for a single shell. Resolution precedence for single-project ops: `--project` > `KBAGENT_PROJECT` > pin > sole-project > fail-hard with a repair-friendly CONFIG_ERROR. - `ProjectService.resolve_pinned_alias()` and the thin `commands._helpers.resolve_project_alias()` helper -- public contract for future single-project write commands to adopt without retrofitting every read-path fan-out. Wire notes: - `project.use` registered as 'write', `project.current` as 'read' in the permission engine's OPERATION_REGISTRY so persisted policies gate them correctly. - `_apply_firewall_flags()` builds a fresh PermissionPolicy and never mutates the persisted deny list. - Stale pin (default_project pointing at a deleted alias) now raises a repair-friendly CONFIG_ERROR instead of silently fanning out. --- CLAUDE.md | 4 +- plugins/kbagent/.claude-plugin/plugin.json | 2 +- plugins/kbagent/skills/kbagent/SKILL.md | 2 + .../kbagent/references/commands-reference.md | 7 + pyproject.toml | 2 +- src/keboola_agent_cli/changelog.py | 8 + src/keboola_agent_cli/cli.py | 78 +++- src/keboola_agent_cli/commands/_helpers.py | 37 ++ src/keboola_agent_cli/commands/context.py | 22 +- src/keboola_agent_cli/commands/project.py | 81 ++++ src/keboola_agent_cli/constants.py | 4 + src/keboola_agent_cli/permissions.py | 2 + .../services/project_service.py | 140 +++++++ tests/test_cli.py | 384 ++++++++++++++++++ tests/test_e2e.py | 348 ++++++++++++++++ tests/test_helpers.py | 146 +++++++ tests/test_services.py | 175 ++++++++ 17 files changed, 1431 insertions(+), 11 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index 5db8defb..e355b962 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -233,7 +233,7 @@ Note: `SKILL.md` instructs Claude to run `kbagent context` as its first step, wh ## All CLI Commands ``` -# Global options: --json, --verbose, --no-color, --config-dir, --hint client|service +# Global options: --json, --verbose, --no-color, --config-dir, --hint client|service, --deny-writes, --deny-destructive kbagent project add --project NAME --url URL --token TOKEN kbagent project list @@ -244,6 +244,8 @@ kbagent project refresh --project ALIAS [--dry-run] [--force] [--yes] [--token-d kbagent project refresh --all [--dry-run] [--force] [--yes] [--token-description DESC] [--token-expires-in N] kbagent project description-get --project NAME kbagent project description-set --project NAME [--text STR | --file PATH | --stdin] +kbagent project use ALIAS +kbagent project current kbagent config list [--project NAME] [--component-type TYPE] [--component-id ID] [--branch ID] kbagent config detail --project NAME --component-id ID --config-id ID [--branch ID] diff --git a/plugins/kbagent/.claude-plugin/plugin.json b/plugins/kbagent/.claude-plugin/plugin.json index ef33fbf1..d4cd469c 100644 --- a/plugins/kbagent/.claude-plugin/plugin.json +++ b/plugins/kbagent/.claude-plugin/plugin.json @@ -1,6 +1,6 @@ { "name": "kbagent", - "version": "0.21.2", + "version": "0.22.0", "description": "AI-friendly interface to Keboola Connection projects — explore configs, jobs, lineage, call MCP tools, manage dev branches, and debug SQL in workspaces", "author": { "name": "Keboola", diff --git a/plugins/kbagent/skills/kbagent/SKILL.md b/plugins/kbagent/skills/kbagent/SKILL.md index 06540e47..f2569306 100644 --- a/plugins/kbagent/skills/kbagent/SKILL.md +++ b/plugins/kbagent/skills/kbagent/SKILL.md @@ -81,6 +81,8 @@ When working inside a git repository or project directory, run `kbagent init` (o | Edit an existing Keboola project connection | `kbagent project edit --project ALIAS` | | Test connectivity to connected Keboola projects | `kbagent project status` | | Refresh expired or invalid Storage API tokens | `kbagent project refresh` | +| Pin as the default project for subsequent commands | `kbagent project use ` | +| Show the effective default project | `kbagent project current` | | Get the Keboola dashboard project description | `kbagent project description-get --project PROJECT` | | Set the Keboola dashboard project description (markdown) | `kbagent project description-set --project PROJECT` | | Set up projects and register them in the kbagent config | `kbagent org setup --url URL` | diff --git a/plugins/kbagent/skills/kbagent/references/commands-reference.md b/plugins/kbagent/skills/kbagent/references/commands-reference.md index 7217f944..3f3f2aae 100644 --- a/plugins/kbagent/skills/kbagent/references/commands-reference.md +++ b/plugins/kbagent/skills/kbagent/references/commands-reference.md @@ -18,6 +18,13 @@ All commands support `--json` for structured output. Multi-project flags (`--pro - `project status [--project NAME]` -- test connectivity and response time - `project description-get --project NAME` -- read the dashboard project description (KBC.projectDescription on the default branch). Returns `{"description": ""}` if not set, not an error - `project description-set --project NAME [--text STR | --file PATH | --stdin]` -- set the dashboard project description (markdown). Pass exactly one of `--text`, `--file`, or `--stdin`. Writes to `KBC.projectDescription` on the default branch -- always the main branch, regardless of any active dev branch +- `project use ALIAS` -- pin `ALIAS` as the persistent default project. Stored as `default_project` in config.json. Overridden at runtime by `KBAGENT_PROJECT=ALIAS` (env, beats pin) and by `--project ALIAS` (CLI flag, beats both) +- `project current` -- print the effective default project and its source (`env` / `pin` / `none`). Reports both the env override AND the persisted pin so misconfigurations are visible. Returns `{"alias": null, "source": "none"}` when neither is set + +## Permission flags (top-level, session-only) +- `--deny-writes` -- block all write/destructive/admin operations for this single invocation. Merges with any persisted permission policy; never written to config.json. Exit code 6 (PERMISSION_DENIED) on blocked operations +- `--deny-destructive` -- block only destructive operations (delete-table, delete-bucket, terminate-job, etc.) for this invocation. Pure-write ops like create-table stay allowed. Use this when you want to keep build-up capabilities but lock out tear-downs +- Both flags compose: `kbagent --deny-writes --deny-destructive ...` is the safest read-only run ## Organization - `org setup --org-id ID --url URL [--dry-run] [--yes]` -- bulk-onboard all projects from an org (org admin, needs `KBC_MANAGE_API_TOKEN`) diff --git a/pyproject.toml b/pyproject.toml index e37cb260..d622abf9 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "keboola-agent-cli" -version = "0.21.2" +version = "0.22.0" description = "AI-friendly CLI for managing Keboola projects" readme = "README.md" requires-python = ">=3.12" diff --git a/src/keboola_agent_cli/changelog.py b/src/keboola_agent_cli/changelog.py index 1514b530..8c3cdb75 100644 --- a/src/keboola_agent_cli/changelog.py +++ b/src/keboola_agent_cli/changelog.py @@ -8,6 +8,14 @@ # Ordered newest-first. Each value is a list of brief one-line descriptions. CHANGELOG: dict[str, list[str]] = { + "0.22.0": [ + "New: `kbagent project use ` -- pin a project as the default for subsequent commands. Persists `default_project` in config.json (the field already existed; now there is an explicit CLI verb to set it).", + "New: `kbagent project current` -- print the effective default project and its source (env / pin / none). Reports both the env override and the persisted pin so misconfigurations are visible, not silent.", + "New: `KBAGENT_PROJECT` env var overrides the persisted pin for a single shell/session. Resolution precedence for single-project ops: explicit `--project` > `KBAGENT_PROJECT` > pin > sole-project fallback > fail-hard with CONFIG_ERROR.", + "New: top-level `--deny-writes` / `--deny-destructive` flags synthesize a session-only firewall that merges with any persisted permission policy. Never written to config.json. `--deny-writes` blocks the wide net (write+destructive+admin); `--deny-destructive` is narrower and blocks only data destruction.", + "New: `ProjectService.resolve_pinned_alias()` plus `commands._helpers.resolve_project_alias()` -- single-project alias resolution contract for write/destructive commands. Public API for future PRs to adopt; FIIA P0-4 acceptance criterion.", + "Fix: stale pin (default_project pointing at a deleted alias) now raises a repair-friendly CONFIG_ERROR with `kbagent project use ` guidance instead of silently fanning out.", + ], "0.21.2": [ "Fix: `kbagent config search` now scans `rows[].configuration` in addition to the top-level configuration body (#196) -- queries like `--query '\"incremental\": false'` previously returned zero matches for row-based components (Snowflake/MySQL/BigQuery writers, DB extractors, Google Sheets) because the service only fetched `include=configuration`; match paths are now reported as `rows[N].configuration.parameters.`", "Fix: `kbagent storage tables` now accepts zero-or-more `--project` flags and queries all connected projects in parallel (#198) -- matches the multi-project behaviour of `storage buckets`, `config list`, `job list`; JSON envelope now returns `{tables: [...], errors: [...]}` with per-row `project_alias`; `--branch` still requires exactly one `--project`", diff --git a/src/keboola_agent_cli/cli.py b/src/keboola_agent_cli/cli.py index aeff431e..e851b54c 100644 --- a/src/keboola_agent_cli/cli.py +++ b/src/keboola_agent_cli/cli.py @@ -30,6 +30,7 @@ from .config_store import ConfigStore, resolve_config_dir from .constants import EXIT_PERMISSION_DENIED from .errors import PermissionDeniedError +from .models import PermissionPolicy from .output import OutputFormatter from .permissions import PermissionEngine from .services.branch_service import BranchService @@ -92,6 +93,58 @@ app.add_typer(encrypt_app, name="encrypt", rich_help_panel=_DEV) +def _apply_firewall_flags( + persisted: PermissionPolicy | None, + *, + deny_writes: bool, + deny_destructive: bool, +) -> PermissionPolicy | None: + """Merge --deny-writes / --deny-destructive into the active policy for this invocation. + + Session-only: does NOT touch config.json. If neither flag is set, the + persisted policy is returned unchanged (possibly None). + + Merge semantics: + - A fresh session policy synthesized from the flags uses mode='allow' + so everything is allowed unless matched by the deny list. + - When a persisted policy already exists, the flag-implied deny patterns + are appended to its deny list (dedup); the mode is preserved. This is + strictly additive -- adding a flag never relaxes the persisted policy. + """ + if not deny_writes and not deny_destructive: + return persisted + + extra_deny: list[str] = [] + if deny_writes: + # cli:write pattern intentionally spans write+destructive+admin + # (permissions._matches_pattern lines 175-178). tool:write spans + # tool write+destructive. Wide net: --deny-writes blocks anything + # that mutates state. + extra_deny.extend(["cli:write", "tool:write"]) + if deny_destructive: + # cli:destructive narrowly matches only ops categorized 'destructive' + # (data destruction). Admin and pure-write are left allowed by design: + # the two flags exist precisely so callers can opt into the narrower + # block without forfeiting writes (e.g. allow create-bucket, block + # delete-bucket). + extra_deny.extend(["cli:destructive", "tool:destructive"]) + + if persisted is None: + return PermissionPolicy(mode="allow", allow=[], deny=extra_deny) + + # Preserve persisted mode, allow list; extend deny list without duplicates. + merged_deny = list(persisted.deny) + for pattern in extra_deny: + if pattern not in merged_deny: + merged_deny.append(pattern) + + return PermissionPolicy( + mode=persisted.mode, + allow=list(persisted.allow), + deny=merged_deny, + ) + + @app.callback() def main( ctx: typer.Context, @@ -123,6 +176,18 @@ def main( help="Show equivalent Python code instead of executing. " "Values: 'client' (direct API usage, default) or 'service' (uses CLI config).", ), + deny_writes: bool = typer.Option( + False, + "--deny-writes", + help="Session-only firewall: block every write (and destructive/admin) operation " + "for this invocation. Merges with any persisted permission policy.", + ), + deny_destructive: bool = typer.Option( + False, + "--deny-destructive", + help="Session-only firewall: block destructive operations (delete/kill/reset) " + "for this invocation. Merges with any persisted permission policy.", + ), ) -> None: """Global options applied to all commands.""" from .auto_update import maybe_auto_update, show_post_update_changelog @@ -199,10 +264,17 @@ def main( try: config = config_store.load() - permission_engine = PermissionEngine(config.permissions) + persisted_policy = config.permissions except Exception: - # Config may be invalid (e.g. corrupted JSON) -- skip permission check - permission_engine = PermissionEngine(None) + # Config may be invalid (e.g. corrupted JSON) -- skip persisted policy + persisted_policy = None + + session_policy = _apply_firewall_flags( + persisted_policy, + deny_writes=deny_writes, + deny_destructive=deny_destructive, + ) + permission_engine = PermissionEngine(session_policy) # Resolve hint mode hint_mode = None diff --git a/src/keboola_agent_cli/commands/_helpers.py b/src/keboola_agent_cli/commands/_helpers.py index dc082a0e..39e7e575 100644 --- a/src/keboola_agent_cli/commands/_helpers.py +++ b/src/keboola_agent_cli/commands/_helpers.py @@ -153,6 +153,43 @@ def check_cli_permission(ctx: typer.Context, group_name: str) -> None: raise typer.Exit(code=EXIT_PERMISSION_DENIED) from None +def resolve_project_alias( + ctx: typer.Context, + formatter: OutputFormatter, + explicit: str | None, +) -> str: + """Resolve the effective project alias for a single-project operation. + + Precedence (first match wins): + 1. ``explicit`` (typically the CLI ``--project`` flag) + 2. ``KBAGENT_PROJECT`` env var + 3. Persisted pin (``config.default_project`` set by ``kbagent project use``) + 4. Sole registered project when exactly one exists (convenience) + 5. Exit code 5 with a CONFIG_ERROR if none of the above resolves + + Use this from write/destructive command paths where implicit fan-out + (``resolve_projects(None)`` returning every project) would be surprising + or unsafe. Read paths should keep their existing fan-out behavior. + + Args: + ctx: Typer context (must contain ``project_service``). + formatter: Output formatter for structured error emission. + explicit: The value of the CLI --project flag, or None. + + Returns: + The resolved project alias (guaranteed to be registered). + """ + from ..errors import ConfigError as _ConfigError + + service = get_service(ctx, "project_service") + try: + alias, _source = service.resolve_pinned_alias(explicit=explicit) + except _ConfigError as exc: + formatter.error(message=exc.message, error_code="CONFIG_ERROR") + raise typer.Exit(code=5) from None + return alias + + def validate_branch_requires_project( formatter: OutputFormatter, branch: int | None, diff --git a/src/keboola_agent_cli/commands/context.py b/src/keboola_agent_cli/commands/context.py index 1b1eea7a..6c8b8c64 100644 --- a/src/keboola_agent_cli/commands/context.py +++ b/src/keboola_agent_cli/commands/context.py @@ -43,11 +43,13 @@ ## Global Flags - --json / -j JSON output (always use for programmatic parsing) - --verbose / -v Verbose output - --no-color Disable colors (auto-disabled in non-TTY) - --config-dir Override config directory path - --hint MODE Generate Python code instead of executing (MODE: client or service) + --json / -j JSON output (always use for programmatic parsing) + --verbose / -v Verbose output + --no-color Disable colors (auto-disabled in non-TTY) + --config-dir Override config directory path + --hint MODE Generate Python code instead of executing (MODE: client or service) + --deny-writes Session-only firewall: block all write operations for this invocation + --deny-destructive Session-only firewall: block destructive (delete/kill) operations for this invocation ## All Commands @@ -83,6 +85,15 @@ Set the dashboard project description. Pass exactly one of --text, --file, or --stdin. Writes KBC.projectDescription to the default branch. + kbagent project use ALIAS + Pin ALIAS as the default project. Persists to config.json. + Env var KBAGENT_PROJECT=ALIAS overrides the pin for a single shell/session; + an explicit --project flag overrides both. + + kbagent project current + Print the effective default project and its source (env / pin / none). + Resolution order for single-project operations: --project > KBAGENT_PROJECT > pin. + ### Component Discovery kbagent component list [--project NAME] [--type TYPE] [--query "search"] @@ -500,6 +511,7 @@ KBC_MASTER_TOKEN Master token for sharing ops (global fallback) KBC_MASTER_TOKEN_* Per-project master token (e.g. KBC_MASTER_TOKEN_PROD) KBAGENT_CONFIG_DIR Override config directory + KBAGENT_PROJECT Override the pinned default project for this shell/session (beats pin, loses to --project) KBAGENT_MAX_PARALLEL_WORKERS Max concurrent threads for multi-project ops (default 10, max 100) KBAGENT_AUTO_UPDATE Set to "false" to disable automatic update on startup KBAGENT_UPDATED_FROM Set to an older version to trigger "What's new" display on next run diff --git a/src/keboola_agent_cli/commands/project.py b/src/keboola_agent_cli/commands/project.py index 0f537dc4..9191336f 100644 --- a/src/keboola_agent_cli/commands/project.py +++ b/src/keboola_agent_cli/commands/project.py @@ -473,6 +473,87 @@ def project_refresh( formatter.output(result, _format_refresh_result) +# ── Project pin (default project) ───────────────────────────────────── + + +@project_app.command("use") +def project_use( + ctx: typer.Context, + alias: str = typer.Argument(..., help="Project alias to pin as default"), +) -> None: + """Pin as the default project for subsequent commands. + + The pin persists in config.json. ``KBAGENT_PROJECT`` overrides it for a + single invocation; an explicit ``--project`` flag overrides both. + """ + formatter = get_formatter(ctx) + service = get_service(ctx, "project_service") + + try: + result = service.use_project(alias=alias) + except ConfigError as exc: + formatter.error(message=exc.message, error_code="CONFIG_ERROR") + raise typer.Exit(code=5) from None + + def _human(c: Console, d: dict[str, Any]) -> None: + previous = d.get("previous") + if previous and previous != d["alias"]: + c.print( + f"[bold green]Pinned:[/bold green] default project is now " + f"[bold]{d['alias']}[/bold] (was [dim]{previous}[/dim])" + ) + else: + c.print( + f"[bold green]Pinned:[/bold green] default project is [bold]{d['alias']}[/bold]" + ) + env_override = d.get("env_override") + if env_override and env_override != d["alias"]: + c.print( + f"[yellow]Note:[/yellow] KBAGENT_PROJECT='{env_override}' is set " + "and overrides this pin for the current shell." + ) + + formatter.output(result, _human) + + +@project_app.command("current") +def project_current(ctx: typer.Context) -> None: + """Show the effective default project. + + Reports whether the value comes from the ``KBAGENT_PROJECT`` env var + (``env``) or the persisted pin (``pin``). Prints nothing but a hint if + neither is set. + """ + formatter = get_formatter(ctx) + service = get_service(ctx, "project_service") + + result = service.current_project() + + def _human(c: Console, d: dict[str, Any]) -> None: + alias = d.get("alias") + source = d.get("source") + if alias is None: + c.print( + "[dim](no default project set)[/dim] -- pass --project, set " + "KBAGENT_PROJECT, or run 'kbagent project use '" + ) + return + if source == "env": + c.print(f"[bold cyan]{alias}[/bold cyan] [dim](source: KBAGENT_PROJECT env var)[/dim]") + if d.get("env_points_to_configured_project") is False: + c.print( + f"[yellow]Warning:[/yellow] '{alias}' is NOT in your " + "configured projects. Commands that use this pin will fail." + ) + pinned = d.get("pinned") + if pinned: + c.print(f"[dim] (pinned in config: {pinned}, overridden)[/dim]") + else: + c.print(f"[bold cyan]{alias}[/bold cyan] [dim](source: pinned default)[/dim]") + + formatter.output(result, _human) + + # ── Project description (dashboard KBC.projectDescription) ──────────── diff --git a/src/keboola_agent_cli/constants.py b/src/keboola_agent_cli/constants.py index 32118767..22210e4b 100644 --- a/src/keboola_agent_cli/constants.py +++ b/src/keboola_agent_cli/constants.py @@ -91,6 +91,10 @@ ENV_CONFIG_DIR: str = "KBAGENT_CONFIG_DIR" LOCAL_CONFIG_DIR_NAME: str = ".kbagent" +# --- Project Pin --- +# Overrides the persisted `default_project` pin for a single invocation/session. +ENV_KBAGENT_PROJECT: str = "KBAGENT_PROJECT" + # --- Environment Variable Names --- ENV_MAX_PARALLEL_WORKERS: str = "KBAGENT_MAX_PARALLEL_WORKERS" ENV_KBC_TOKEN: str = "KBC_TOKEN" diff --git a/src/keboola_agent_cli/permissions.py b/src/keboola_agent_cli/permissions.py index 1955f61c..6f63e47d 100644 --- a/src/keboola_agent_cli/permissions.py +++ b/src/keboola_agent_cli/permissions.py @@ -22,6 +22,8 @@ "project.refresh": "admin", "project.description-get": "read", "project.description-set": "write", + "project.use": "write", + "project.current": "read", # Config browsing & management "config.list": "read", "config.detail": "read", diff --git a/src/keboola_agent_cli/services/project_service.py b/src/keboola_agent_cli/services/project_service.py index 1b43bc7a..73019fa1 100644 --- a/src/keboola_agent_cli/services/project_service.py +++ b/src/keboola_agent_cli/services/project_service.py @@ -3,9 +3,11 @@ Orchestrates config persistence and API calls without knowing about CLI or HTTP details. """ +import os import time from typing import Any +from ..constants import ENV_KBAGENT_PROJECT from ..errors import ConfigError, KeboolaApiError, mask_token from ..models import ProjectConfig from .base import BaseService @@ -249,3 +251,141 @@ def get_status(self, aliases: list[str] | None = None) -> list[dict[str, Any]]: results.sort(key=lambda r: r.get("alias", "")) return results + + def use_project(self, alias: str) -> dict[str, Any]: + """Pin an alias as the persistent default project. + + The pin is stored as ``config.default_project`` in config.json. + It is overridden at runtime by the ``KBAGENT_PROJECT`` env var and by + explicit ``--project`` flags. + + Args: + alias: The project alias to pin. + + Returns: + Dict with the new pin, previous pin, and source. + + Raises: + ConfigError: If the alias does not exist. + """ + config = self._config_store.load() + if alias not in config.projects: + raise ConfigError(f"Project '{alias}' not found.") + + previous = config.default_project or None + config.default_project = alias + self._config_store.save(config) + + env_override = os.environ.get(ENV_KBAGENT_PROJECT) + return { + "alias": alias, + "previous": previous, + "source": "pin", + "env_override": env_override or None, + } + + def current_project(self) -> dict[str, Any]: + """Report the effective default project and its source. + + Resolution: + - If ``KBAGENT_PROJECT`` is set, it wins (source=env). + - Otherwise the persisted pin wins (source=pin). + - If neither is set, ``alias`` is ``None``. + + The env override is reported even when it points at a project that is + not (yet) registered in config.json -- callers get the true effective + alias plus an ``env_points_to_configured_project`` flag to reason about + it. This avoids silently masking misconfigurations. + + Returns: + Dict with keys: alias, source ('env' | 'pin' | 'none'), pinned, + env_override, env_points_to_configured_project. + """ + config = self._config_store.load() + pinned = config.default_project or None + # Treat KBAGENT_PROJECT="" the same as unset (Unix shell convention: + # empty env is commonly produced by `unset` substitutes / blank + # exports). Strict rejection would surprise CI users who export it + # conditionally. Apply consistently in resolve_pinned_alias(). + env_value = os.environ.get(ENV_KBAGENT_PROJECT) + env_override = env_value if env_value else None + + if env_override is not None: + return { + "alias": env_override, + "source": "env", + "pinned": pinned, + "env_override": env_override, + "env_points_to_configured_project": env_override in config.projects, + } + + return { + "alias": pinned, + "source": "pin" if pinned else "none", + "pinned": pinned, + "env_override": None, + "env_points_to_configured_project": None, + } + + def resolve_pinned_alias(self, explicit: str | None = None) -> tuple[str, str]: + """Resolve the effective project alias for a single-project operation. + + Precedence (first match wins): + 1. ``explicit`` argument (typically the CLI ``--project`` flag) + 2. ``KBAGENT_PROJECT`` env var + 3. Persisted ``default_project`` pin + 4. If exactly one project is registered, fall back to it (source=sole) + 5. Fail hard with ConfigError + + This is the single-project analog of ``resolve_projects()`` (which + fans out to all projects). Use this from write/destructive command + paths where fan-out would be surprising or unsafe. + + Args: + explicit: Explicit alias from a CLI flag, or None. + + Returns: + Tuple of (alias, source). + + Raises: + ConfigError: If the resolved alias is not registered, or if none + can be resolved. + """ + config = self._config_store.load() + + if explicit: + if explicit not in config.projects: + raise ConfigError(f"Project '{explicit}' not found.") + return explicit, "explicit" + + env_value = os.environ.get(ENV_KBAGENT_PROJECT) + if env_value: + if env_value not in config.projects: + raise ConfigError( + f"{ENV_KBAGENT_PROJECT}='{env_value}' points to a project " + "that is not registered. Use 'kbagent project add' or " + "unset the env var." + ) + return env_value, "env" + + pinned = config.default_project + if pinned: + if pinned not in config.projects: + raise ConfigError( + f"Pinned default project '{pinned}' is not registered. " + "Run 'kbagent project use ' to repair." + ) + return pinned, "pin" + + if len(config.projects) == 1: + (sole,) = config.projects.keys() + return sole, "sole" + + if not config.projects: + raise ConfigError("No projects configured. Run 'kbagent project add' first.") + + raise ConfigError( + "Multiple projects configured and no default pinned. " + "Pass --project , set KBAGENT_PROJECT, or run " + "'kbagent project use '." + ) diff --git a/tests/test_cli.py b/tests/test_cli.py index 86055d81..facae65e 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -470,6 +470,390 @@ def test_project_status_human(self, tmp_path: Path) -> None: assert "Project Status" in result.output +class TestProjectUse: + """Tests for `kbagent project use ` (pin default project).""" + + def _seed(self, config_dir: Path, *aliases: str) -> None: + """Seed a ConfigStore with one or more projects via the live add path.""" + mock_client = make_mock_client() + with ( + patch("keboola_agent_cli.cli.ConfigStore") as MockStore, + patch("keboola_agent_cli.cli.ProjectService") as MockService, + patch.dict(os.environ, {"KBC_TOKEN": TEST_TOKEN}), + ): + store = ConfigStore(config_dir=config_dir) + MockStore.return_value = store + MockService.return_value = ProjectService( + config_store=store, + client_factory=lambda url, token: mock_client, + ) + for alias in aliases: + runner.invoke(app, ["project", "add", "--project", alias]) + + def test_project_use_pins_alias(self, tmp_path: Path) -> None: + """project use ALIAS persists default_project to config.json.""" + config_dir = tmp_path / "config" + config_dir.mkdir() + self._seed(config_dir, "prod", "stage") + + # default_project should be the first-added (prod); now pin stage. + with patch("keboola_agent_cli.cli.ConfigStore") as MockStore: + store = ConfigStore(config_dir=config_dir) + MockStore.return_value = store + result = runner.invoke(app, ["--json", "project", "use", "stage"]) + + assert result.exit_code == 0, result.output + data = json.loads(result.output)["data"] + assert data["alias"] == "stage" + assert data["previous"] == "prod" + # Verify persistence: re-load the store and check default_project. + persisted = ConfigStore(config_dir=config_dir).load() + assert persisted.default_project == "stage" + + def test_project_use_unknown_alias_exit_5(self, tmp_path: Path) -> None: + """project use on an unregistered alias returns exit code 5.""" + config_dir = tmp_path / "config" + config_dir.mkdir() + self._seed(config_dir, "prod") + + with patch("keboola_agent_cli.cli.ConfigStore") as MockStore: + MockStore.return_value = ConfigStore(config_dir=config_dir) + result = runner.invoke(app, ["--json", "project", "use", "does-not-exist"]) + + assert result.exit_code == 5 + data = json.loads(result.output) + assert data["status"] == "error" + assert data["error"]["code"] == "CONFIG_ERROR" + + def test_project_use_human_mode_confirms_pin(self, tmp_path: Path) -> None: + """project use in human mode prints the new pin.""" + config_dir = tmp_path / "config" + config_dir.mkdir() + self._seed(config_dir, "prod") + + with patch("keboola_agent_cli.cli.ConfigStore") as MockStore: + MockStore.return_value = ConfigStore(config_dir=config_dir) + result = runner.invoke(app, ["project", "use", "prod"]) + + assert result.exit_code == 0 + assert "prod" in result.output + assert "Pinned" in result.output or "pinned" in result.output + + def test_project_current_with_pin(self, tmp_path: Path) -> None: + """project current reports the persisted pin when no env is set.""" + config_dir = tmp_path / "config" + config_dir.mkdir() + self._seed(config_dir, "prod") + + with ( + patch("keboola_agent_cli.cli.ConfigStore") as MockStore, + patch.dict(os.environ, {}, clear=False), + ): + os.environ.pop("KBAGENT_PROJECT", None) + MockStore.return_value = ConfigStore(config_dir=config_dir) + result = runner.invoke(app, ["--json", "project", "current"]) + + assert result.exit_code == 0 + data = json.loads(result.output)["data"] + assert data["alias"] == "prod" + assert data["source"] == "pin" + assert data["env_override"] is None + + def test_project_current_env_overrides_pin(self, tmp_path: Path) -> None: + """KBAGENT_PROJECT wins over persisted pin; env presence is reported.""" + config_dir = tmp_path / "config" + config_dir.mkdir() + self._seed(config_dir, "prod", "stage") + + with ( + patch("keboola_agent_cli.cli.ConfigStore") as MockStore, + patch.dict(os.environ, {"KBAGENT_PROJECT": "stage"}), + ): + MockStore.return_value = ConfigStore(config_dir=config_dir) + result = runner.invoke(app, ["--json", "project", "current"]) + + assert result.exit_code == 0 + data = json.loads(result.output)["data"] + assert data["alias"] == "stage" + assert data["source"] == "env" + assert data["pinned"] == "prod" + assert data["env_points_to_configured_project"] is True + + def test_project_current_env_points_to_unknown(self, tmp_path: Path) -> None: + """Unregistered KBAGENT_PROJECT value is still reported, with a flag.""" + config_dir = tmp_path / "config" + config_dir.mkdir() + self._seed(config_dir, "prod") + + with ( + patch("keboola_agent_cli.cli.ConfigStore") as MockStore, + patch.dict(os.environ, {"KBAGENT_PROJECT": "mystery"}), + ): + MockStore.return_value = ConfigStore(config_dir=config_dir) + result = runner.invoke(app, ["--json", "project", "current"]) + + assert result.exit_code == 0 + data = json.loads(result.output)["data"] + assert data["alias"] == "mystery" + assert data["source"] == "env" + assert data["env_points_to_configured_project"] is False + + def test_project_current_human_mode_with_pin(self, tmp_path: Path) -> None: + """project current in human mode prints alias + source label.""" + config_dir = tmp_path / "config" + config_dir.mkdir() + self._seed(config_dir, "prod") + + with ( + patch("keboola_agent_cli.cli.ConfigStore") as MockStore, + patch.dict(os.environ, {}, clear=False), + ): + os.environ.pop("KBAGENT_PROJECT", None) + MockStore.return_value = ConfigStore(config_dir=config_dir) + result = runner.invoke(app, ["project", "current"]) + + assert result.exit_code == 0 + assert "prod" in result.output + # Rich output should mention the source. + assert "pin" in result.output.lower() + + def test_project_current_human_mode_env_warns_unknown(self, tmp_path: Path) -> None: + """Human-mode project current warns when env points to unregistered alias.""" + config_dir = tmp_path / "config" + config_dir.mkdir() + self._seed(config_dir, "prod") + + with ( + patch("keboola_agent_cli.cli.ConfigStore") as MockStore, + patch.dict(os.environ, {"KBAGENT_PROJECT": "ghost"}), + ): + MockStore.return_value = ConfigStore(config_dir=config_dir) + result = runner.invoke(app, ["project", "current"]) + + assert result.exit_code == 0 + assert "ghost" in result.output + # Output should warn the alias is not registered. + assert "Warning" in result.output or "NOT" in result.output + + def test_project_use_blocked_by_persisted_deny_writes(self, tmp_path: Path) -> None: + """A persisted policy denying cli:write must block project use (it's a write op).""" + from keboola_agent_cli.models import PermissionPolicy + + config_dir = tmp_path / "config" + config_dir.mkdir() + self._seed(config_dir, "prod", "stage") + + # Persist a default-allow policy that denies cli:write. + store = ConfigStore(config_dir=config_dir) + cfg = store.load() + cfg.permissions = PermissionPolicy(mode="allow", allow=[], deny=["cli:write"]) + store.save(cfg) + + with patch("keboola_agent_cli.cli.ConfigStore") as MockStore: + MockStore.return_value = ConfigStore(config_dir=config_dir) + result = runner.invoke(app, ["--json", "project", "use", "stage"]) + + assert result.exit_code == 6, result.output + data = json.loads(result.output) + assert data["error"]["code"] == "PERMISSION_DENIED" + + def test_project_current_allowed_under_deny_writes(self, tmp_path: Path) -> None: + """project current is classified read, so cli:write deny must NOT block it.""" + from keboola_agent_cli.models import PermissionPolicy + + config_dir = tmp_path / "config" + config_dir.mkdir() + self._seed(config_dir, "prod") + + store = ConfigStore(config_dir=config_dir) + cfg = store.load() + cfg.permissions = PermissionPolicy(mode="allow", allow=[], deny=["cli:write"]) + store.save(cfg) + + with patch("keboola_agent_cli.cli.ConfigStore") as MockStore: + MockStore.return_value = ConfigStore(config_dir=config_dir) + result = runner.invoke(app, ["--json", "project", "current"]) + + assert result.exit_code == 0, result.output + data = json.loads(result.output) + assert data["status"] == "ok" + + def test_project_current_none_set(self, tmp_path: Path) -> None: + """With no projects and no env, current reports alias=None source=none.""" + config_dir = tmp_path / "config" + config_dir.mkdir() + + with ( + patch("keboola_agent_cli.cli.ConfigStore") as MockStore, + patch.dict(os.environ, {}, clear=False), + ): + os.environ.pop("KBAGENT_PROJECT", None) + MockStore.return_value = ConfigStore(config_dir=config_dir) + result = runner.invoke(app, ["--json", "project", "current"]) + + assert result.exit_code == 0 + data = json.loads(result.output)["data"] + assert data["alias"] is None + assert data["source"] == "none" + + +class TestFirewallFlags: + """Tests for top-level --deny-writes / --deny-destructive session flags.""" + + def _seed(self, config_dir: Path) -> None: + mock_client = make_mock_client() + with ( + patch("keboola_agent_cli.cli.ConfigStore") as MockStore, + patch("keboola_agent_cli.cli.ProjectService") as MockService, + patch.dict(os.environ, {"KBC_TOKEN": TEST_TOKEN}), + ): + store = ConfigStore(config_dir=config_dir) + MockStore.return_value = store + MockService.return_value = ProjectService( + config_store=store, + client_factory=lambda url, token: mock_client, + ) + runner.invoke(app, ["project", "add", "--project", "prod"]) + + def test_deny_writes_blocks_project_add(self, tmp_path: Path) -> None: + """--deny-writes blocks project.add (admin is a superset of cli:write).""" + config_dir = tmp_path / "config" + config_dir.mkdir() + + with ( + patch("keboola_agent_cli.cli.ConfigStore") as MockStore, + patch.dict(os.environ, {"KBC_TOKEN": TEST_TOKEN}), + ): + MockStore.return_value = ConfigStore(config_dir=config_dir) + result = runner.invoke( + app, + [ + "--deny-writes", + "--json", + "project", + "add", + "--project", + "foo", + "--url", + "https://connection.keboola.com", + "--token", + TEST_TOKEN, + ], + ) + + assert result.exit_code == 6 + data = json.loads(result.output) + assert data["error"]["code"] == "PERMISSION_DENIED" + + def test_deny_writes_allows_read(self, tmp_path: Path) -> None: + """--deny-writes must not block read operations.""" + config_dir = tmp_path / "config" + config_dir.mkdir() + self._seed(config_dir) + + with patch("keboola_agent_cli.cli.ConfigStore") as MockStore: + MockStore.return_value = ConfigStore(config_dir=config_dir) + result = runner.invoke(app, ["--deny-writes", "--json", "project", "list"]) + + assert result.exit_code == 0, result.output + data = json.loads(result.output) + assert data["status"] == "ok" + + def test_deny_destructive_blocks_delete_table(self, tmp_path: Path) -> None: + """--deny-destructive blocks storage.delete-table at the permission callback.""" + config_dir = tmp_path / "config" + config_dir.mkdir() + self._seed(config_dir) + + with patch("keboola_agent_cli.cli.ConfigStore") as MockStore: + MockStore.return_value = ConfigStore(config_dir=config_dir) + result = runner.invoke( + app, + [ + "--deny-destructive", + "--json", + "storage", + "delete-table", + "--project", + "prod", + "--table-id", + "in.c-x.y", + "--yes", + ], + ) + + assert result.exit_code == 6 + data = json.loads(result.output) + assert data["error"]["code"] == "PERMISSION_DENIED" + + def test_deny_destructive_allows_write(self, tmp_path: Path) -> None: + """--deny-destructive must NOT block pure 'write' (non-destructive) ops. + + Uses 'permissions check' which evaluates against the PERSISTED policy, + so the session-only --deny-destructive does not apply there. We instead + attempt a write op and assert it is not blocked by the permission gate + (any later error must not be PERMISSION_DENIED). + """ + config_dir = tmp_path / "config" + config_dir.mkdir() + self._seed(config_dir) + + # storage.create-bucket is classified 'write', so --deny-destructive alone + # does not block it. The command will fail for other reasons (mock client), + # but the failure code must not be PERMISSION_DENIED. + with patch("keboola_agent_cli.cli.ConfigStore") as MockStore: + MockStore.return_value = ConfigStore(config_dir=config_dir) + result = runner.invoke( + app, + [ + "--deny-destructive", + "--json", + "storage", + "create-bucket", + "--project", + "prod", + "--stage", + "in", + "--name", + "b", + ], + ) + + # The permission check must not fire. + assert result.exit_code != 6 or "PERMISSION_DENIED" not in result.output + + def test_deny_writes_and_destructive_merge_with_persisted(self, tmp_path: Path) -> None: + """Flags merge with persisted policy; never persist to disk.""" + from keboola_agent_cli.models import AppConfig, PermissionPolicy + + config_dir = tmp_path / "config" + config_dir.mkdir() + self._seed(config_dir) + + # Persist a policy that only denies branch.delete + store = ConfigStore(config_dir=config_dir) + cfg = store.load() + cfg.permissions = PermissionPolicy(mode="allow", allow=[], deny=["branch.delete"]) + store.save(cfg) + + # Run once with --deny-writes: project.add must be blocked + with ( + patch("keboola_agent_cli.cli.ConfigStore") as MockStore, + patch.dict(os.environ, {"KBC_TOKEN": TEST_TOKEN}), + ): + MockStore.return_value = ConfigStore(config_dir=config_dir) + result = runner.invoke( + app, + ["--deny-writes", "--json", "project", "add", "--project", "new"], + ) + assert result.exit_code == 6 + + # The persisted policy on disk must NOT have been mutated. + reloaded: AppConfig = ConfigStore(config_dir=config_dir).load() + assert reloaded.permissions is not None + assert reloaded.permissions.deny == ["branch.delete"] + + class TestProjectEdit: """Tests for `kbagent project edit` command.""" diff --git a/tests/test_e2e.py b/tests/test_e2e.py index c18e0336..b44885d9 100644 --- a/tests/test_e2e.py +++ b/tests/test_e2e.py @@ -558,6 +558,9 @@ def test_full_cli_e2e(self) -> None: _step(42, "storage delete-table + delete-bucket", "CLI-driven cleanup") self._test_storage_cleanup(bucket_id, table_id) + _step("42.5", "project use / current + firewall flags") + self._test_project_pin_and_firewall() + _step(43, "project edit + remove", "final cleanup") self._test_project_edit_and_remove() @@ -2382,6 +2385,129 @@ def _test_storage_cleanup(self, bucket_id: str, table_id: str) -> None: assert bucket_id in data["data"]["deleted"] self._created_buckets.remove(bucket_id) + def _test_project_pin_and_firewall(self) -> None: + """End-to-end coverage for `project use`, `project current`, and --deny-* flags.""" + # --- Pin lifecycle ------------------------------------------------- + + # Pre-condition: first-added is already the default. Verify via current. + data = self._run_ok("project", "current") + assert data["data"]["alias"] == self.alias + assert data["data"]["source"] == "pin" + + # Explicit `project use` is a no-op in value but confirms it persists. + data = self._run_ok("project", "use", self.alias) + assert data["data"]["alias"] == self.alias + # source is always "pin" on use (the field describes where the new + # pin ended up, not how it arrived). + assert data["data"]["source"] == "pin" + + # `project use nonexistent` fails with exit 5 (CONFIG_ERROR). + result = self._run("project", "use", "does-not-exist-alias") + assert result.exit_code == 5 + + # --- KBAGENT_PROJECT env override --------------------------------- + # Set the env var to a bogus value and confirm `current` reports env + # as the source + flags the unknown alias. + with patch.dict(os.environ, {"KBAGENT_PROJECT": "mystery-alias"}): + data = self._run_ok("project", "current") + assert data["data"]["alias"] == "mystery-alias" + assert data["data"]["source"] == "env" + assert data["data"]["env_points_to_configured_project"] is False + assert data["data"]["pinned"] == self.alias + + # After unsetting, pin is restored as the effective alias. + data = self._run_ok("project", "current") + assert data["data"]["source"] == "pin" + assert data["data"]["alias"] == self.alias + + # --- --deny-writes blocks writes, allows reads -------------------- + # Read still succeeds. + data = self._run_ok( + "--deny-writes", # top-level flag must come before subcommand + "project", + "list", + ) + assert any(p["alias"] == self.alias for p in data["data"]) + + # Attempting a write under --deny-writes must exit 6 PERMISSION_DENIED. + # create-bucket is a safe write to try: if the firewall fails to + # block it we'd create a real bucket, so track it for cleanup just + # in case the block logic regresses. + guard_bucket_name = f"{RUN_ID.replace('-', '_')}_firewall_guard" + result = self._run( + "--deny-writes", + "storage", + "create-bucket", + "--project", + self.alias, + "--stage", + "in", + "--name", + guard_bucket_name, + ) + assert result.exit_code == 6, ( + f"--deny-writes should block storage.create-bucket (exit 6), " + f"got {result.exit_code}: {result.output}" + ) + data = json.loads(result.output) + assert data["error"]["code"] == "PERMISSION_DENIED" + + # Safety: if the block failed silently and a bucket was actually + # created, schedule cleanup. We don't fail louder because the + # exit_code assert above already did. + try: + buckets = self.api.list_buckets() + for bucket in buckets: + if bucket.get("name") == guard_bucket_name: + self._created_buckets.append(bucket["id"]) + except Exception: + pass # Best-effort cleanup tracking only. + + # --- --deny-destructive blocks destructive ops -------------------- + # delete-bucket is destructive; must exit 6 even on a bucket that + # does not exist (permission check fires before the API call). + result = self._run( + "--deny-destructive", + "storage", + "delete-bucket", + "--project", + self.alias, + "--bucket-id", + "in.c-never-existed", + "--yes", + ) + assert result.exit_code == 6, ( + f"--deny-destructive should block storage.delete-bucket (exit 6), " + f"got {result.exit_code}: {result.output}" + ) + data = json.loads(result.output) + assert data["error"]["code"] == "PERMISSION_DENIED" + + # --- --deny-destructive allows non-destructive writes ------------- + # project.description-set is classified 'write' (not destructive), + # so --deny-destructive must NOT block it. We pass an empty string + # write -- this goes to the API, but the permission gate is the + # only thing under test here, so any non-6 exit is acceptable. + result = self._run( + "--deny-destructive", + "project", + "description-get", + "--project", + self.alias, + ) + assert result.exit_code != 6, ( + "--deny-destructive must not block read op project.description-get" + ) + + # --- Persistence check -------------------------------------------- + # None of the --deny-* flags may have written to config.json. + store = ConfigStore(config_dir=self.config_dir) + persisted = store.load() + assert persisted.permissions is None, ( + "--deny-writes / --deny-destructive must be session-only; " + f"found persisted policy: {persisted.permissions}" + ) + def _test_project_edit_and_remove(self) -> None: """Edit project URL, then remove it.""" # project edit -- change URL back to same (just verify command works) @@ -3413,3 +3539,225 @@ def test_resolver_prefers_explicit_values_id_over_first_row(self) -> None: ) print(f" {_DIM}resolved={resolved} pinned={pinned_row_id} first={first_row_id}{_RESET}") assert resolved == pinned_row_id + + +# --------------------------------------------------------------------------- +# Project pin + firewall flag E2E (PR5) +# --------------------------------------------------------------------------- + + +@skip_without_credentials +@pytest.mark.e2e +class TestPinAndFirewallE2E: + """Focused E2E for `project use`, `project current`, and --deny-* flags. + + Exercises the real API to confirm: + - `project use` persists the pin to config.json. + - `project current` reports the effective alias + source correctly. + - `KBAGENT_PROJECT` env var overrides the pin at runtime. + - `--deny-writes` blocks the permission gate on a real write op (exit 6). + - `--deny-destructive` blocks a real destructive op (exit 6). + - Neither flag persists to config.json. + """ + + @pytest.fixture(autouse=True) + def setup(self, tmp_path: Path) -> None: + self.token = os.environ[ENV_TOKEN] + raw_url = os.environ.get(ENV_URL, "connection.keboola.com") + self.url = raw_url if raw_url.startswith("https://") else f"https://{raw_url}" + self.config_dir = tmp_path / "config" + self.config_dir.mkdir() + self.alias_a = f"{RUN_ID}-pin-a" + self.alias_b = f"{RUN_ID}-pin-b" + + def _run(self, *args: str) -> Any: + return _invoke(self.config_dir, ["--json", *args]) + + def test_pin_lifecycle_against_real_project(self) -> None: + """End-to-end: add, use, current, env override.""" + # Register two aliases pointing at the SAME real project. We only + # need distinct aliases to observe the pin switching. + self._run( + "project", + "add", + "--project", + self.alias_a, + "--url", + self.url, + "--token", + self.token, + ) + self._run( + "project", + "add", + "--project", + self.alias_b, + "--url", + self.url, + "--token", + self.token, + ) + + # First-added becomes default. + current = self._run("project", "current") + assert current.exit_code == 0 + data = _json_ok(current) + assert data["data"]["alias"] == self.alias_a + assert data["data"]["source"] == "pin" + + # `project use` switches the pin; persistence survives next invocation. + use_result = self._run("project", "use", self.alias_b) + use_data = _json_ok(use_result) + assert use_data["data"]["alias"] == self.alias_b + assert use_data["data"]["previous"] == self.alias_a + + after = _json_ok(self._run("project", "current")) + assert after["data"]["alias"] == self.alias_b + assert after["data"]["source"] == "pin" + + # Unknown alias -> exit 5. + bad = self._run("project", "use", "does-not-exist") + assert bad.exit_code == 5 + bad_data = json.loads(bad.output) + assert bad_data["error"]["code"] == "CONFIG_ERROR" + + # KBAGENT_PROJECT overrides the pin. + with patch.dict(os.environ, {"KBAGENT_PROJECT": self.alias_a}): + env_view = _json_ok(self._run("project", "current")) + assert env_view["data"]["alias"] == self.alias_a + assert env_view["data"]["source"] == "env" + assert env_view["data"]["pinned"] == self.alias_b + assert env_view["data"]["env_points_to_configured_project"] is True + + def test_deny_writes_blocks_real_write_op(self) -> None: + """--deny-writes must exit 6 on a real create-bucket attempt.""" + self._run( + "project", + "add", + "--project", + self.alias_a, + "--url", + self.url, + "--token", + self.token, + ) + + # Use a name that won't collide; the permission gate fires before + # the API call so the bucket must never appear. + bucket_name = f"{RUN_ID.replace('-', '_')}_fw_w" + result = self._run( + "--deny-writes", + "storage", + "create-bucket", + "--project", + self.alias_a, + "--stage", + "in", + "--name", + bucket_name, + ) + assert result.exit_code == 6, ( + f"--deny-writes should block storage.create-bucket; got exit " + f"{result.exit_code}: {result.output}" + ) + data = json.loads(result.output) + assert data["error"]["code"] == "PERMISSION_DENIED" + + # Defensive: if the block leaked and a bucket was actually created, + # clean it up and fail the assertion above (already failed) more loudly. + import contextlib + + api = KeboolaClient(self.url, self.token) + try: + for bucket in api.list_buckets(): + if bucket.get("name") == bucket_name: + with contextlib.suppress(Exception): + api.delete_bucket(bucket["id"], force=True) + raise AssertionError( + f"--deny-writes failed to block: bucket {bucket['id']} was created" + ) + finally: + api.close() + + def test_deny_destructive_blocks_real_destructive_op(self) -> None: + """--deny-destructive must exit 6 on storage.delete-bucket.""" + self._run( + "project", + "add", + "--project", + self.alias_a, + "--url", + self.url, + "--token", + self.token, + ) + + result = self._run( + "--deny-destructive", + "storage", + "delete-bucket", + "--project", + self.alias_a, + "--bucket-id", + "in.c-does-not-exist-for-sure", + "--yes", + ) + assert result.exit_code == 6, ( + f"--deny-destructive should block delete-bucket; got exit " + f"{result.exit_code}: {result.output}" + ) + data = json.loads(result.output) + assert data["error"]["code"] == "PERMISSION_DENIED" + + def test_deny_destructive_allows_read_op(self) -> None: + """--deny-destructive must NOT block read ops (regression guard).""" + self._run( + "project", + "add", + "--project", + self.alias_a, + "--url", + self.url, + "--token", + self.token, + ) + result = self._run( + "--deny-destructive", + "storage", + "buckets", + "--project", + self.alias_a, + ) + # Read must succeed (exit 0) OR fail for non-permission reasons. + assert result.exit_code != 6, f"--deny-destructive blocked a read op: {result.output}" + + def test_firewall_flags_never_persist(self) -> None: + """Neither --deny-writes nor --deny-destructive may write to config.json.""" + self._run( + "project", + "add", + "--project", + self.alias_a, + "--url", + self.url, + "--token", + self.token, + ) + # Run a blocked op under both flags. + self._run( + "--deny-writes", + "--deny-destructive", + "storage", + "create-bucket", + "--project", + self.alias_a, + "--stage", + "in", + "--name", + "never_created", + ) + # Persisted policy must still be None. + persisted = ConfigStore(config_dir=self.config_dir).load() + assert persisted.permissions is None, ( + f"--deny-* flags leaked to config.json: {persisted.permissions}" + ) diff --git a/tests/test_helpers.py b/tests/test_helpers.py index d076f9dc..71baba7c 100644 --- a/tests/test_helpers.py +++ b/tests/test_helpers.py @@ -372,3 +372,149 @@ def test_ignore_active_branch_single_project_inferred(self, tmp_config_dir) -> N assert project == "prod" assert branch_id is None formatter.err_console.print.assert_called_once() + + +class TestResolveProjectAlias: + """Tests for resolve_project_alias() (write-op precedence).""" + + def _build(self, tmp_config_dir) -> tuple: + from unittest.mock import MagicMock + + from keboola_agent_cli.config_store import ConfigStore + from keboola_agent_cli.models import ProjectConfig + from keboola_agent_cli.services.project_service import ProjectService + + store = ConfigStore(config_dir=tmp_config_dir) + for alias, pid in (("prod", 1), ("stage", 2)): + store.add_project( + alias, + ProjectConfig( + stack_url="https://connection.keboola.com", + token=f"t-{alias}", + project_name=alias, + project_id=pid, + ), + ) + service = ProjectService(config_store=store) + ctx = MagicMock() + ctx.obj = {"project_service": service} + formatter = MagicMock(json_mode=False) + return ctx, formatter, store + + def test_explicit_wins_over_env_and_pin(self, tmp_config_dir, monkeypatch) -> None: + from keboola_agent_cli.commands._helpers import resolve_project_alias + + monkeypatch.setenv("KBAGENT_PROJECT", "stage") + ctx, formatter, _ = self._build(tmp_config_dir) + assert resolve_project_alias(ctx, formatter, explicit="prod") == "prod" + + def test_env_beats_pin(self, tmp_config_dir, monkeypatch) -> None: + from keboola_agent_cli.commands._helpers import resolve_project_alias + + monkeypatch.setenv("KBAGENT_PROJECT", "stage") + ctx, formatter, _ = self._build(tmp_config_dir) + # pin is prod (first-added); env overrides to stage + assert resolve_project_alias(ctx, formatter, explicit=None) == "stage" + + def test_pin_used_when_no_env(self, tmp_config_dir, monkeypatch) -> None: + from keboola_agent_cli.commands._helpers import resolve_project_alias + + monkeypatch.delenv("KBAGENT_PROJECT", raising=False) + ctx, formatter, _ = self._build(tmp_config_dir) + assert resolve_project_alias(ctx, formatter, explicit=None) == "prod" + + def test_fail_hard_multi_no_pin(self, tmp_config_dir, monkeypatch) -> None: + import typer + + from keboola_agent_cli.commands._helpers import resolve_project_alias + + monkeypatch.delenv("KBAGENT_PROJECT", raising=False) + ctx, formatter, store = self._build(tmp_config_dir) + cfg = store.load() + cfg.default_project = "" + store.save(cfg) + + with pytest.raises(typer.Exit) as exc_info: + resolve_project_alias(ctx, formatter, explicit=None) + assert exc_info.value.exit_code == 5 + formatter.error.assert_called_once() + + +class TestApplyFirewallFlags: + """Tests for cli._apply_firewall_flags (session-only policy merge).""" + + def test_no_flags_returns_persisted_as_is(self) -> None: + from keboola_agent_cli.cli import _apply_firewall_flags + from keboola_agent_cli.models import PermissionPolicy + + persisted = PermissionPolicy(mode="allow", allow=[], deny=["branch.delete"]) + result = _apply_firewall_flags(persisted, deny_writes=False, deny_destructive=False) + assert result is persisted + + def test_no_flags_no_persisted_returns_none(self) -> None: + from keboola_agent_cli.cli import _apply_firewall_flags + + result = _apply_firewall_flags(None, deny_writes=False, deny_destructive=False) + assert result is None + + def test_deny_writes_synthesizes_fresh_policy(self) -> None: + from keboola_agent_cli.cli import _apply_firewall_flags + + result = _apply_firewall_flags(None, deny_writes=True, deny_destructive=False) + assert result is not None + assert result.mode == "allow" + assert "cli:write" in result.deny + assert "tool:write" in result.deny + + def test_deny_destructive_synthesizes_fresh_policy(self) -> None: + from keboola_agent_cli.cli import _apply_firewall_flags + + result = _apply_firewall_flags(None, deny_writes=False, deny_destructive=True) + assert result is not None + assert "cli:destructive" in result.deny + assert "tool:destructive" in result.deny + assert "cli:write" not in result.deny + + def test_flags_merge_with_persisted_deny_no_duplicates(self) -> None: + from keboola_agent_cli.cli import _apply_firewall_flags + from keboola_agent_cli.models import PermissionPolicy + + persisted = PermissionPolicy(mode="allow", allow=[], deny=["branch.delete", "cli:write"]) + result = _apply_firewall_flags(persisted, deny_writes=True, deny_destructive=False) + assert result is not None + # Existing cli:write preserved (no dup); tool:write appended; custom entry kept. + assert result.deny.count("cli:write") == 1 + assert "tool:write" in result.deny + assert "branch.delete" in result.deny + # Mode preserved. + assert result.mode == persisted.mode + + def test_flags_preserve_persisted_mode_deny_mode(self) -> None: + from keboola_agent_cli.cli import _apply_firewall_flags + from keboola_agent_cli.models import PermissionPolicy + + persisted = PermissionPolicy(mode="deny", allow=["cli:read"], deny=[]) + result = _apply_firewall_flags(persisted, deny_writes=True, deny_destructive=False) + assert result is not None + assert result.mode == "deny" + assert result.allow == ["cli:read"] + assert "cli:write" in result.deny + + def test_both_flags_combine(self) -> None: + from keboola_agent_cli.cli import _apply_firewall_flags + + result = _apply_firewall_flags(None, deny_writes=True, deny_destructive=True) + assert result is not None + # Both prefixes present. + assert {"cli:write", "tool:write", "cli:destructive", "tool:destructive"} <= set( + result.deny + ) + + def test_flags_do_not_mutate_persisted(self) -> None: + from keboola_agent_cli.cli import _apply_firewall_flags + from keboola_agent_cli.models import PermissionPolicy + + persisted = PermissionPolicy(mode="allow", allow=[], deny=["branch.delete"]) + before = list(persisted.deny) + _apply_firewall_flags(persisted, deny_writes=True, deny_destructive=True) + assert persisted.deny == before, "persisted.deny was mutated in place" diff --git a/tests/test_services.py b/tests/test_services.py index 33f64572..abc310a0 100644 --- a/tests/test_services.py +++ b/tests/test_services.py @@ -519,6 +519,181 @@ def _make_list_components_client( ] +class TestUseAndCurrentProject: + """Tests for ProjectService.use_project() / current_project() / resolve_pinned_alias().""" + + def _seed_two(self, tmp_config_dir: Path) -> ConfigStore: + store = ConfigStore(config_dir=tmp_config_dir) + for alias, pid in (("prod", 1), ("stage", 2)): + store.add_project( + alias, + ProjectConfig( + stack_url="https://connection.keboola.com", + token=f"901-x-{alias}", + project_name=alias.title(), + project_id=pid, + ), + ) + return store + + def test_use_project_pins_and_persists(self, tmp_config_dir: Path) -> None: + store = self._seed_two(tmp_config_dir) + service = ProjectService(config_store=store) + + result = service.use_project(alias="stage") + + assert result["alias"] == "stage" + assert result["previous"] == "prod" + assert result["source"] == "pin" + # Persistence check + assert ConfigStore(config_dir=tmp_config_dir).load().default_project == "stage" + + def test_use_project_unknown_raises(self, tmp_config_dir: Path) -> None: + store = self._seed_two(tmp_config_dir) + service = ProjectService(config_store=store) + with pytest.raises(ConfigError, match="not found"): + service.use_project(alias="does-not-exist") + + def test_current_project_pin_only(self, tmp_config_dir: Path, monkeypatch) -> None: + monkeypatch.delenv("KBAGENT_PROJECT", raising=False) + store = self._seed_two(tmp_config_dir) + service = ProjectService(config_store=store) + + result = service.current_project() + assert result["alias"] == "prod" + assert result["source"] == "pin" + assert result["env_override"] is None + + def test_current_project_env_override(self, tmp_config_dir: Path, monkeypatch) -> None: + monkeypatch.setenv("KBAGENT_PROJECT", "stage") + store = self._seed_two(tmp_config_dir) + service = ProjectService(config_store=store) + + result = service.current_project() + assert result["alias"] == "stage" + assert result["source"] == "env" + assert result["pinned"] == "prod" + assert result["env_points_to_configured_project"] is True + + def test_current_project_env_unknown(self, tmp_config_dir: Path, monkeypatch) -> None: + monkeypatch.setenv("KBAGENT_PROJECT", "mystery") + store = self._seed_two(tmp_config_dir) + service = ProjectService(config_store=store) + + result = service.current_project() + assert result["alias"] == "mystery" + assert result["env_points_to_configured_project"] is False + + def test_current_project_no_pin_no_env(self, tmp_config_dir: Path, monkeypatch) -> None: + monkeypatch.delenv("KBAGENT_PROJECT", raising=False) + # Empty store -- no pin possible + store = ConfigStore(config_dir=tmp_config_dir) + service = ProjectService(config_store=store) + + result = service.current_project() + assert result["alias"] is None + assert result["source"] == "none" + + # ── resolve_pinned_alias precedence ──────────────────────────────── + + def test_resolve_explicit_wins(self, tmp_config_dir: Path, monkeypatch) -> None: + monkeypatch.setenv("KBAGENT_PROJECT", "stage") + store = self._seed_two(tmp_config_dir) + service = ProjectService(config_store=store) + + alias, source = service.resolve_pinned_alias(explicit="prod") + assert alias == "prod" + assert source == "explicit" + + def test_resolve_env_beats_pin(self, tmp_config_dir: Path, monkeypatch) -> None: + monkeypatch.setenv("KBAGENT_PROJECT", "stage") + store = self._seed_two(tmp_config_dir) + # default is prod (first added) + service = ProjectService(config_store=store) + + alias, source = service.resolve_pinned_alias() + assert alias == "stage" + assert source == "env" + + def test_resolve_pin_used(self, tmp_config_dir: Path, monkeypatch) -> None: + monkeypatch.delenv("KBAGENT_PROJECT", raising=False) + store = self._seed_two(tmp_config_dir) + service = ProjectService(config_store=store) + + alias, source = service.resolve_pinned_alias() + assert alias == "prod" + assert source == "pin" + + def test_resolve_sole_project_fallback(self, tmp_config_dir: Path, monkeypatch) -> None: + monkeypatch.delenv("KBAGENT_PROJECT", raising=False) + store = ConfigStore(config_dir=tmp_config_dir) + store.add_project( + "only", + ProjectConfig( + stack_url="https://connection.keboola.com", + token="t", + project_name="Only", + project_id=7, + ), + ) + # Clear the pin to exercise the sole-project fallback. + cfg = store.load() + cfg.default_project = "" + store.save(cfg) + + service = ProjectService(config_store=store) + alias, source = service.resolve_pinned_alias() + assert alias == "only" + assert source == "sole" + + def test_resolve_fail_hard_multi_no_pin(self, tmp_config_dir: Path, monkeypatch) -> None: + monkeypatch.delenv("KBAGENT_PROJECT", raising=False) + store = self._seed_two(tmp_config_dir) + cfg = store.load() + cfg.default_project = "" + store.save(cfg) + + service = ProjectService(config_store=store) + with pytest.raises(ConfigError, match="Multiple projects"): + service.resolve_pinned_alias() + + def test_resolve_explicit_unknown_raises(self, tmp_config_dir: Path, monkeypatch) -> None: + monkeypatch.delenv("KBAGENT_PROJECT", raising=False) + store = self._seed_two(tmp_config_dir) + service = ProjectService(config_store=store) + with pytest.raises(ConfigError, match="not found"): + service.resolve_pinned_alias(explicit="ghost") + + def test_resolve_env_unknown_raises(self, tmp_config_dir: Path, monkeypatch) -> None: + monkeypatch.setenv("KBAGENT_PROJECT", "mystery") + store = self._seed_two(tmp_config_dir) + service = ProjectService(config_store=store) + with pytest.raises(ConfigError, match="not registered"): + service.resolve_pinned_alias() + + def test_resolve_no_projects_raises(self, tmp_config_dir: Path, monkeypatch) -> None: + monkeypatch.delenv("KBAGENT_PROJECT", raising=False) + store = ConfigStore(config_dir=tmp_config_dir) + service = ProjectService(config_store=store) + with pytest.raises(ConfigError, match="No projects configured"): + service.resolve_pinned_alias() + + def test_resolve_pinned_alias_points_to_unregistered( + self, tmp_config_dir: Path, monkeypatch + ) -> None: + """Stale pin (pointing at deleted project) raises a repair-friendly ConfigError.""" + monkeypatch.delenv("KBAGENT_PROJECT", raising=False) + store = self._seed_two(tmp_config_dir) + # Hand-edit default_project to a value that isn't in projects. + cfg = store.load() + cfg.default_project = "ghost" + store.save(cfg) + + service = ProjectService(config_store=store) + with pytest.raises(ConfigError, match="not registered"): + service.resolve_pinned_alias() + + class TestConfigServiceListConfigs: """Tests for ConfigService.list_configs().""" From 5ba4f87263b480c11452a3ba6f1795f1b728d016 Mon Sep 17 00:00:00 2001 From: Maxmilian Ottomansky Date: Wed, 22 Apr 2026 15:25:06 +0200 Subject: [PATCH 2/9] fix(repl): forward --deny-writes / --deny-destructive into REPL subcommands Review caught a silent-elevation bug: starting the CLI with `kbagent --deny-writes` and then entering the REPL (or typing `repl`) let inner commands re-parse WITHOUT the firewall flags, restoring full write/destructive access. The same PR5 firewall flag that a user opted into was silently dropped the moment they landed at the prompt. Root cause: `_run_repl()` only forwarded four flags (--json, --verbose, --no-color, --config-dir) when rebuilding argv for each inner command; the two new session-firewall flags were never threaded through. Fix: - Extend `_run_repl(deny_writes, deny_destructive)` and re-append both flags on every inner invocation unless the user already typed them. - Thread the flags from `cli.py::main()` into both REPL launch paths (implicit no-subcommand launch + explicit `kbagent repl`) via `ctx.obj["deny_writes"]` / `ctx.obj["deny_destructive"]`. - Banner advertises the active firewall so users can see what's on. Tests (`TestReplFirewallPropagation`): - `test_deny_writes_appended_to_inner_argv` -- guard against the regression by asserting every rebuilt argv carries --deny-writes. - `test_deny_destructive_appended_to_inner_argv` -- same for destructive. - `test_no_duplicate_when_user_retypes_flag` -- idempotence guard. Also closes other review-loop gaps (non-REPL): - `test_project_help` now asserts `use` and `current` subcommands. - `test_root_help_lists_firewall_flags` guards the global flags. - `test_list_includes_project_use_and_current` guards the registry. - `test_hint_on_non_api_project_use/current` cover --hint graceful exit. - `test_deny_writes_composes_with_persisted_deny_mode` validates the runtime composition under default-deny mode. --- src/keboola_agent_cli/cli.py | 4 + src/keboola_agent_cli/commands/repl.py | 36 ++++++- tests/test_cli.py | 77 +++++++++++++++ tests/test_permissions_cli.py | 21 ++++ tests/test_repl.py | 129 +++++++++++++++++++++++++ 5 files changed, 264 insertions(+), 3 deletions(-) diff --git a/src/keboola_agent_cli/cli.py b/src/keboola_agent_cli/cli.py index e851b54c..814778fc 100644 --- a/src/keboola_agent_cli/cli.py +++ b/src/keboola_agent_cli/cli.py @@ -290,6 +290,8 @@ def main( ctx.obj["permission_engine"] = permission_engine ctx.obj["verbose"] = verbose ctx.obj["no_color"] = effective_no_color + ctx.obj["deny_writes"] = deny_writes + ctx.obj["deny_destructive"] = deny_destructive ctx.obj["config_store"] = config_store ctx.obj["project_service"] = project_service ctx.obj["component_service"] = component_service @@ -361,5 +363,7 @@ def main( verbose=verbose, no_color=effective_no_color, config_dir=config_dir, + deny_writes=deny_writes, + deny_destructive=deny_destructive, ) raise typer.Exit() diff --git a/src/keboola_agent_cli/commands/repl.py b/src/keboola_agent_cli/commands/repl.py index bfbb0894..3770af74 100644 --- a/src/keboola_agent_cli/commands/repl.py +++ b/src/keboola_agent_cli/commands/repl.py @@ -67,8 +67,21 @@ def _get_history_path() -> Path: return config_dir / "repl_history" -def _run_repl(json_mode: bool, verbose: bool, no_color: bool, config_dir: str | None) -> None: - """Main REPL loop.""" +def _run_repl( + json_mode: bool, + verbose: bool, + no_color: bool, + config_dir: str | None, + deny_writes: bool = False, + deny_destructive: bool = False, +) -> None: + """Main REPL loop. + + Global flags from the outer invocation are re-applied on every command + executed inside the REPL. This includes the session-only firewall flags + ``--deny-writes`` / ``--deny-destructive`` -- dropping them here would + silently elevate the REPL above the policy the user started it with. + """ from ..cli import app as typer_app # Build command tree for completion @@ -94,7 +107,15 @@ def _run_repl(json_mode: bool, verbose: bool, no_color: bool, config_dir: str | # Show banner sys.stderr.write(f"\nkbagent v{__version__} -- interactive mode\n") sys.stderr.write("Type 'help' for commands, 'exit' to quit.\n") - sys.stderr.write(f"Global flags: --json={json_mode}, --verbose={verbose}\n\n") + sys.stderr.write(f"Global flags: --json={json_mode}, --verbose={verbose}\n") + if deny_writes or deny_destructive: + active = [] + if deny_writes: + active.append("--deny-writes") + if deny_destructive: + active.append("--deny-destructive") + sys.stderr.write(f"Session firewall: {' '.join(active)} (active for all commands)\n") + sys.stderr.write("\n") while True: try: @@ -141,6 +162,13 @@ def _run_repl(json_mode: bool, verbose: bool, no_color: bool, config_dir: str | full_argv.append("--no-color") if config_dir and "--config-dir" not in argv: full_argv.extend(["--config-dir", config_dir]) + # Session firewall flags: re-applied on every REPL invocation so the + # user's opt-in policy survives across prompts. Forgetting to forward + # these would silently restore write/destructive access inside the REPL. + if deny_writes and "--deny-writes" not in argv: + full_argv.append("--deny-writes") + if deny_destructive and "--deny-destructive" not in argv: + full_argv.append("--deny-destructive") full_argv.extend(argv) # Prevent recursive REPL @@ -176,4 +204,6 @@ def repl_command(ctx: typer.Context) -> None: verbose=ctx.obj.get("verbose", False), no_color=ctx.obj.get("no_color", False), config_dir=None, # Already resolved in ctx + deny_writes=ctx.obj.get("deny_writes", False), + deny_destructive=ctx.obj.get("deny_destructive", False), ) diff --git a/tests/test_cli.py b/tests/test_cli.py index facae65e..40251fba 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -657,6 +657,29 @@ def test_project_use_blocked_by_persisted_deny_writes(self, tmp_path: Path) -> N data = json.loads(result.output) assert data["error"]["code"] == "PERMISSION_DENIED" + def test_hint_on_non_api_project_use(self, tmp_path: Path) -> None: + """project use is purely local -- --hint must exit cleanly, not crash.""" + config_dir = tmp_path / "config" + config_dir.mkdir() + self._seed(config_dir, "prod") + + with patch("keboola_agent_cli.cli.ConfigStore") as MockStore: + MockStore.return_value = ConfigStore(config_dir=config_dir) + result = runner.invoke(app, ["--hint", "client", "project", "use", "prod"]) + # Should exit 0 with a clear message -- no hint available for local ops. + assert result.exit_code == 0 + + def test_hint_on_non_api_project_current(self, tmp_path: Path) -> None: + """project current is purely local -- --hint must exit cleanly, not crash.""" + config_dir = tmp_path / "config" + config_dir.mkdir() + self._seed(config_dir, "prod") + + with patch("keboola_agent_cli.cli.ConfigStore") as MockStore: + MockStore.return_value = ConfigStore(config_dir=config_dir) + result = runner.invoke(app, ["--hint", "service", "project", "current"]) + assert result.exit_code == 0 + def test_project_current_allowed_under_deny_writes(self, tmp_path: Path) -> None: """project current is classified read, so cli:write deny must NOT block it.""" from keboola_agent_cli.models import PermissionPolicy @@ -822,6 +845,50 @@ def test_deny_destructive_allows_write(self, tmp_path: Path) -> None: # The permission check must not fire. assert result.exit_code != 6 or "PERMISSION_DENIED" not in result.output + def test_deny_writes_composes_with_persisted_deny_mode(self, tmp_path: Path) -> None: + """End-to-end: default-deny policy + --deny-writes still blocks writes. + + A persisted default-deny policy that allows cli:write (unusual but + syntactically valid) composed with --deny-writes must resolve to + "write denied" because deny takes precedence over allow in the engine + (permissions.py: default-deny rule is 'allowed and not denied'). + """ + from keboola_agent_cli.models import PermissionPolicy + + config_dir = tmp_path / "config" + config_dir.mkdir() + self._seed(config_dir) + + # Persist a default-deny policy that explicitly allows cli:write. + store = ConfigStore(config_dir=config_dir) + cfg = store.load() + cfg.permissions = PermissionPolicy(mode="deny", allow=["cli:write", "cli:read"], deny=[]) + store.save(cfg) + + with ( + patch("keboola_agent_cli.cli.ConfigStore") as MockStore, + patch.dict(os.environ, {"KBC_TOKEN": TEST_TOKEN}), + ): + MockStore.return_value = ConfigStore(config_dir=config_dir) + result = runner.invoke( + app, + [ + "--deny-writes", + "--json", + "project", + "add", + "--project", + "newproj", + ], + ) + + # Merged policy: mode=deny, allow=[cli:write,cli:read], deny=[cli:write,tool:write]. + # Default-deny rule: allowed AND not denied. cli:write matches both + # allow and deny -- deny wins. + assert result.exit_code == 6, result.output + data = json.loads(result.output) + assert data["error"]["code"] == "PERMISSION_DENIED" + def test_deny_writes_and_destructive_merge_with_persisted(self, tmp_path: Path) -> None: """Flags merge with persisted policy; never persist to disk.""" from keboola_agent_cli.models import AppConfig, PermissionPolicy @@ -4004,6 +4071,16 @@ def test_project_help(self) -> None: assert "remove" in result.output assert "edit" in result.output assert "status" in result.output + # PR5 additions -- guard against accidental removal. + assert "use" in result.output + assert "current" in result.output + + def test_root_help_lists_firewall_flags(self) -> None: + """Root --help must advertise the session firewall flags.""" + result = runner.invoke(app, ["--help"]) + assert result.exit_code == 0 + assert "--deny-writes" in result.output + assert "--deny-destructive" in result.output def test_config_help(self) -> None: """config --help shows subcommands.""" diff --git a/tests/test_permissions_cli.py b/tests/test_permissions_cli.py index deb7f4a9..20d1455e 100644 --- a/tests/test_permissions_cli.py +++ b/tests/test_permissions_cli.py @@ -76,6 +76,27 @@ def test_list_filter_by_category(self, tmp_path: Path) -> None: for op in data: assert op["category"] == "destructive" + def test_list_includes_project_use_and_current(self, tmp_path: Path) -> None: + """Registry entries for the PR5 commands are visible to the engine. + + Guards against a future refactor that accidentally drops them from + OPERATION_REGISTRY -- unregistered commands would silently default + to 'write' for category matching, which is fail-closed but invisible. + """ + store = _make_store(tmp_path) + with patch("keboola_agent_cli.cli.ConfigStore") as MockStore: + MockStore.return_value = store + result = runner.invoke(app, ["--json", "permissions", "list"]) + assert result.exit_code == 0 + data = json.loads(result.output)["data"] + by_name = {op["name"]: op for op in data} + + assert "project.use" in by_name, "project.use missing from permissions list" + assert by_name["project.use"]["category"] == "write" + + assert "project.current" in by_name, "project.current missing from permissions list" + assert by_name["project.current"]["category"] == "read" + class TestPermissionsShow: """Tests for `kbagent permissions show`.""" diff --git a/tests/test_repl.py b/tests/test_repl.py index 753a4ee9..e259609c 100644 --- a/tests/test_repl.py +++ b/tests/test_repl.py @@ -120,3 +120,132 @@ def test_no_args_non_tty_shows_help(self) -> None: assert result.exit_code == 0 assert "Commands" in result.output or "kbagent" in result.output + + +class TestReplFirewallPropagation: + """REPL must forward --deny-writes / --deny-destructive to every inner command. + + Dropping the flags when re-invoking inside the REPL would silently + restore write access for a user who started the session with a firewall. + """ + + def test_deny_writes_appended_to_inner_argv(self, monkeypatch) -> None: + """When deny_writes=True, the REPL rebuilds argv with --deny-writes.""" + from keboola_agent_cli.commands import repl as repl_module + + captured: list[list[str]] = [] + + class _FakeClickApp: + def __call__(self, argv, standalone_mode=False): + captured.append(list(argv)) + raise SystemExit(0) + + monkeypatch.setattr(repl_module.typer.main, "get_command", lambda _app: _FakeClickApp()) + + class _FakeSession: + def __init__(self, **_kwargs): + self._replies = iter(["project list", ""]) + + def prompt(self, _text): + try: + return next(self._replies) + except StopIteration as exc: + raise EOFError from exc + + monkeypatch.setattr(repl_module, "PromptSession", _FakeSession) + + repl_module._run_repl( + json_mode=True, + verbose=False, + no_color=True, + config_dir=None, + deny_writes=True, + deny_destructive=False, + ) + + # Every captured invocation must carry --deny-writes in the rebuilt argv. + assert captured, "REPL did not dispatch any commands" + for argv in captured: + assert "--deny-writes" in argv, ( + f"REPL dropped --deny-writes when rebuilding argv: {argv}" + ) + + def test_deny_destructive_appended_to_inner_argv(self, monkeypatch) -> None: + """When deny_destructive=True, the REPL rebuilds argv with --deny-destructive.""" + from keboola_agent_cli.commands import repl as repl_module + + captured: list[list[str]] = [] + + class _FakeClickApp: + def __call__(self, argv, standalone_mode=False): + captured.append(list(argv)) + raise SystemExit(0) + + monkeypatch.setattr(repl_module.typer.main, "get_command", lambda _app: _FakeClickApp()) + + class _FakeSession: + def __init__(self, **_kwargs): + self._replies = iter(["storage delete-table --project p --table-id t"]) + + def prompt(self, _text): + try: + return next(self._replies) + except StopIteration as exc: + raise EOFError from exc + + monkeypatch.setattr(repl_module, "PromptSession", _FakeSession) + + repl_module._run_repl( + json_mode=False, + verbose=False, + no_color=True, + config_dir=None, + deny_writes=False, + deny_destructive=True, + ) + + assert captured + for argv in captured: + assert "--deny-destructive" in argv, ( + f"REPL dropped --deny-destructive when rebuilding argv: {argv}" + ) + + def test_no_duplicate_when_user_retypes_flag(self, monkeypatch) -> None: + """If user types the flag inside the REPL, we must not double it.""" + from keboola_agent_cli.commands import repl as repl_module + + captured: list[list[str]] = [] + + class _FakeClickApp: + def __call__(self, argv, standalone_mode=False): + captured.append(list(argv)) + raise SystemExit(0) + + monkeypatch.setattr(repl_module.typer.main, "get_command", lambda _app: _FakeClickApp()) + + class _FakeSession: + def __init__(self, **_kwargs): + self._replies = iter(["--deny-writes project list"]) + + def prompt(self, _text): + try: + return next(self._replies) + except StopIteration as exc: + raise EOFError from exc + + monkeypatch.setattr(repl_module, "PromptSession", _FakeSession) + + repl_module._run_repl( + json_mode=False, + verbose=False, + no_color=True, + config_dir=None, + deny_writes=True, + deny_destructive=False, + ) + + assert captured + argv = captured[0] + assert argv.count("--deny-writes") == 1, ( + f"REPL duplicated --deny-writes when user also typed it: {argv}" + ) From 149fb6193fcdaac0959a30b34f3971d0095215ca Mon Sep 17 00:00:00 2001 From: Maxmilian Ottomansky Date: Wed, 22 Apr 2026 15:31:48 +0200 Subject: [PATCH 3/9] fix(test): use callback signature for firewall-flag help assertion The original `test_root_help_lists_firewall_flags` rendered `kbagent --help` via CliRunner and string-matched `--deny-writes` / `--deny-destructive`. CI's narrow default terminal width made Rich's typer help panel truncate the long option names into `...`, failing the assertion even though the flags are registered correctly. Switch to inspecting `cli.main()` function signature. The signature is the source of truth for registered options and is independent of terminal width. Still catches the regression (accidental flag removal). --- tests/test_cli.py | 26 ++++++++++++++++++++------ 1 file changed, 20 insertions(+), 6 deletions(-) diff --git a/tests/test_cli.py b/tests/test_cli.py index 40251fba..5bd00dc1 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -4075,12 +4075,26 @@ def test_project_help(self) -> None: assert "use" in result.output assert "current" in result.output - def test_root_help_lists_firewall_flags(self) -> None: - """Root --help must advertise the session firewall flags.""" - result = runner.invoke(app, ["--help"]) - assert result.exit_code == 0 - assert "--deny-writes" in result.output - assert "--deny-destructive" in result.output + def test_root_callback_registers_firewall_flags(self) -> None: + """App callback signature declares --deny-writes and --deny-destructive. + + Tests the callback signature rather than rendered --help output: Rich + truncates options in narrow terminals (CI's default width collapses + long flag names into '...'), which makes a string-match on the help + text flaky. The signature is what users see once the terminal has + room and is a strict superset guarantee. + """ + import inspect + + from keboola_agent_cli.cli import main as cli_main + + sig = inspect.signature(cli_main) + assert "deny_writes" in sig.parameters, ( + "cli.main() must accept deny_writes (top-level --deny-writes flag)" + ) + assert "deny_destructive" in sig.parameters, ( + "cli.main() must accept deny_destructive (top-level --deny-destructive flag)" + ) def test_config_help(self) -> None: """config --help shows subcommands.""" From 4ede2b8e22a6452d3a66e20400e7b8ed26455378 Mon Sep 17 00:00:00 2001 From: Maxmilian Ottomansky Date: Wed, 22 Apr 2026 19:09:31 +0200 Subject: [PATCH 4/9] fix(permissions): show/list now reflect session firewall flags UX gap surfaced in the review loop: `permissions show` / `permissions list` previously inspected ONLY the persisted policy. A user running `kbagent --deny-writes permissions show` (or inside a REPL launched with those flags) saw misleading output: "No permission policy configured" even though their writes were in fact blocked for the invocation. Fix: - `permissions list` now builds the engine from the MERGED policy (`_apply_firewall_flags(persisted, deny_writes=..., deny_destructive=...)`). The allowed/denied column matches what the command will actually do. Human-mode footer advertises active session flags. - `permissions show` reports both layers in the JSON envelope: { "active": bool, # true if persisted OR session present "persisted": {mode, allow, deny} | null, "session_flags": [...], # legacy top-level keys preserved when persisted != null "mode": ..., "allow": ..., "deny": ... } Human-mode prints a "Session firewall:" line whenever flags are active. Legacy JSON consumers that read `.mode` / `.allow` / `.deny` keep working. - `permissions show` with no persisted policy + no session flags still prints the legacy "No permission policy configured" line (empty-state contract preserved for existing tests). Tests added: - `test_list_reflects_session_deny_writes` -- project.add flips to denied when --deny-writes is active. - `test_show_reports_session_flags_without_persisted_policy` -- active=true even without a persisted policy. - `test_show_reports_session_flags_alongside_persisted` -- both layers reported + legacy keys still present. - `test_show_human_mode_no_policy_no_session` -- empty-state regression. --- src/keboola_agent_cli/commands/permissions.py | 105 +++++++++++++++--- tests/test_permissions_cli.py | 64 +++++++++++ 2 files changed, 153 insertions(+), 16 deletions(-) diff --git a/src/keboola_agent_cli/commands/permissions.py b/src/keboola_agent_cli/commands/permissions.py index b281b65f..c72a6c4d 100644 --- a/src/keboola_agent_cli/commands/permissions.py +++ b/src/keboola_agent_cli/commands/permissions.py @@ -98,12 +98,28 @@ def permissions_list( help="Filter by risk category: read, write, destructive, admin", ), ) -> None: - """List all operations with their risk category and current allowed/denied status.""" + """List all operations with their risk category and current allowed/denied status. + + The allowed/denied column reflects the EFFECTIVE policy for this + invocation -- i.e. the persisted policy merged with any top-level + session flags like ``--deny-writes`` / ``--deny-destructive``. This + matches what a command will actually do right now. + """ + from ..cli import _apply_firewall_flags + formatter = get_formatter(ctx) config_store: ConfigStore = get_service(ctx, "config_store") config = config_store.load() - engine = PermissionEngine(config.permissions) + deny_writes = bool(ctx.obj.get("deny_writes")) if ctx.obj else False + deny_destructive = bool(ctx.obj.get("deny_destructive")) if ctx.obj else False + effective_policy = _apply_firewall_flags( + config.permissions, + deny_writes=deny_writes, + deny_destructive=deny_destructive, + ) + + engine = PermissionEngine(effective_policy) ops = engine.list_operations() if formatter.json_mode: @@ -116,44 +132,101 @@ def permissions_list( formatter.err_console.print( "\n[dim]No permission policy active. All operations are allowed.[/dim]" ) + elif deny_writes or deny_destructive: + active_flags = [] + if deny_writes: + active_flags.append("--deny-writes") + if deny_destructive: + active_flags.append("--deny-destructive") + formatter.err_console.print( + f"\n[dim]Session firewall active: {' '.join(active_flags)} (not persisted).[/dim]" + ) @permissions_app.command("show") def permissions_show( ctx: typer.Context, ) -> None: - """Show the current active permission policy.""" + """Show the current active permission policy. + + Reports both the PERSISTED policy (from config.json) and any SESSION + firewall layered on top via top-level ``--deny-writes`` / + ``--deny-destructive`` flags. Session flags are shown but are never + written to config.json -- they apply to this invocation only. + """ formatter = get_formatter(ctx) config_store: ConfigStore = get_service(ctx, "config_store") config = config_store.load() - if config.permissions is None: + deny_writes = bool(ctx.obj.get("deny_writes")) if ctx.obj else False + deny_destructive = bool(ctx.obj.get("deny_destructive")) if ctx.obj else False + session_flags: list[str] = [] + if deny_writes: + session_flags.append("--deny-writes") + if deny_destructive: + session_flags.append("--deny-destructive") + + persisted = config.permissions + + if persisted is None and not session_flags: if formatter.json_mode: - formatter.output({"active": False, "message": "No permission policy configured"}) + formatter.output( + { + "active": False, + "message": "No permission policy configured", + "session_flags": [], + } + ) else: formatter.console.print("No permission policy configured. All operations are allowed.") return - policy_data = { - "active": True, - "mode": config.permissions.mode, - "allow": config.permissions.allow, - "deny": config.permissions.deny, + policy_data: dict[str, Any] = { + "active": persisted is not None or bool(session_flags), + "persisted": ( + None + if persisted is None + else { + "mode": persisted.mode, + "allow": persisted.allow, + "deny": persisted.deny, + } + ), + "session_flags": session_flags, } + # Keep legacy top-level keys when a persisted policy exists so existing + # JSON consumers that read policy_data["mode"] / ["allow"] / ["deny"] + # remain compatible. Clients that need the new session-layer view read + # ``session_flags`` and ``persisted``. + if persisted is not None: + policy_data["mode"] = persisted.mode + policy_data["allow"] = persisted.allow + policy_data["deny"] = persisted.deny + if formatter.json_mode: formatter.output(policy_data) - else: + return + + if persisted is not None: mode_desc = ( "default-allow (everything allowed unless denied)" - if config.permissions.mode == "allow" + if persisted.mode == "allow" else "default-deny (everything denied unless allowed)" ) formatter.console.print(f"[bold]Mode:[/bold] {mode_desc}") - if config.permissions.allow: - formatter.console.print(f"[bold]Allow:[/bold] {', '.join(config.permissions.allow)}") - if config.permissions.deny: - formatter.console.print(f"[bold]Deny:[/bold] {', '.join(config.permissions.deny)}") + if persisted.allow: + formatter.console.print(f"[bold]Allow:[/bold] {', '.join(persisted.allow)}") + if persisted.deny: + formatter.console.print(f"[bold]Deny:[/bold] {', '.join(persisted.deny)}") + else: + formatter.console.print("[dim]No persisted permission policy (config.json is clean).[/dim]") + + if session_flags: + formatter.console.print( + f"[bold yellow]Session firewall:[/bold yellow] {' '.join(session_flags)} " + "[dim](active for this invocation only; not persisted)[/dim]" + ) @permissions_app.command("set") diff --git a/tests/test_permissions_cli.py b/tests/test_permissions_cli.py index 20d1455e..ebe12634 100644 --- a/tests/test_permissions_cli.py +++ b/tests/test_permissions_cli.py @@ -76,6 +76,23 @@ def test_list_filter_by_category(self, tmp_path: Path) -> None: for op in data: assert op["category"] == "destructive" + def test_list_reflects_session_deny_writes(self, tmp_path: Path) -> None: + """--deny-writes must flip 'allowed' -> 'denied' on write ops in the table.""" + store = _make_store(tmp_path) + with patch("keboola_agent_cli.cli.ConfigStore") as MockStore: + MockStore.return_value = store + result = runner.invoke(app, ["--deny-writes", "--json", "permissions", "list"]) + assert result.exit_code == 0 + data = json.loads(result.output)["data"] + by_name = {op["name"]: op for op in data} + + # Write-classified op should show denied under session --deny-writes. + assert by_name["project.add"]["status"] == "denied", ( + "permissions list ignored session --deny-writes" + ) + # Read-classified op should still be allowed. + assert by_name["project.list"]["status"] == "allowed" + def test_list_includes_project_use_and_current(self, tmp_path: Path) -> None: """Registry entries for the PR5 commands are visible to the engine. @@ -123,6 +140,53 @@ def test_show_with_policy(self, tmp_path: Path) -> None: assert "cli:write" in data["deny"] assert "tool:write" in data["deny"] + def test_show_reports_session_flags_without_persisted_policy(self, tmp_path: Path) -> None: + """--deny-writes with no persisted policy must still report 'active'.""" + store = _make_store(tmp_path) + with patch("keboola_agent_cli.cli.ConfigStore") as MockStore: + MockStore.return_value = store + result = runner.invoke(app, ["--deny-writes", "--json", "permissions", "show"]) + assert result.exit_code == 0 + data = json.loads(result.output)["data"] + assert data["active"] is True, ( + "permissions show reported inactive despite --deny-writes being set" + ) + assert data["persisted"] is None + assert "--deny-writes" in data["session_flags"] + + def test_show_reports_session_flags_alongside_persisted(self, tmp_path: Path) -> None: + """Session flags are reported in addition to persisted policy.""" + policy = PermissionPolicy(mode="allow", deny=["branch.delete"]) + store = _make_store(tmp_path, policy) + with patch("keboola_agent_cli.cli.ConfigStore") as MockStore: + MockStore.return_value = store + result = runner.invoke( + app, + [ + "--deny-writes", + "--deny-destructive", + "--json", + "permissions", + "show", + ], + ) + assert result.exit_code == 0 + data = json.loads(result.output)["data"] + # Persisted policy still present + legacy top-level keys still there. + assert data["persisted"]["deny"] == ["branch.delete"] + assert data["deny"] == ["branch.delete"] # legacy consumer compatibility + # Session flags both reported. + assert data["session_flags"] == ["--deny-writes", "--deny-destructive"] + + def test_show_human_mode_no_policy_no_session(self, tmp_path: Path) -> None: + """Bare 'permissions show' still prints the legacy empty-state message.""" + store = _make_store(tmp_path) + with patch("keboola_agent_cli.cli.ConfigStore") as MockStore: + MockStore.return_value = store + result = runner.invoke(app, ["permissions", "show"]) + assert result.exit_code == 0 + assert "No permission policy configured" in result.output + class TestPermissionsSet: """Tests for `kbagent permissions set`.""" From 6ee65b0e32a5787c046135be990c0f7e2e96bdfb Mon Sep 17 00:00:00 2001 From: Maxmilian Ottomansky Date: Wed, 22 Apr 2026 19:40:46 +0200 Subject: [PATCH 5/9] docs(cli): clarify --deny-destructive is NARROW; admin ops need --deny-writes Iter-4 review flagged a doc/flag-help ambiguity: the `--deny-destructive` help string said "destructive (delete/kill/reset) operations" which a user could reasonably read as covering 'project remove' and 'org setup'. The actual behavior (correctly) leaves admin-classified ops allowed, because 'admin' is a separate risk category from 'destructive' in the OPERATION registry. Only --deny-writes catches the wide net. Fixes: - `cli.py`: spell out WIDE vs NARROW in each flag's help string. Explicitly call out 'project remove' / 'org setup' as NOT blocked by --deny-destructive. - `context.py`: mirror the same WIDE vs NARROW contract in the AI-agent reference text (global flags section). Tests added (`TestFirewallFlags`): - `test_deny_destructive_allows_admin_ops` -- contract lock: running `kbagent --deny-destructive project remove prod` succeeds (exit != 6). - `test_deny_writes_blocks_admin_ops` -- complement: --deny-writes IS the wide net and DOES block admin ops (exit 6). Test hygiene (`TestPermissionsShow`): - `test_show_reports_session_flags_without_persisted_policy` now also asserts the legacy `mode`/`allow`/`deny` top-level keys are ABSENT when there is no persisted policy. Documents the new shape contract for downstream JSON consumers. --- src/keboola_agent_cli/cli.py | 11 ++-- src/keboola_agent_cli/commands/context.py | 4 +- tests/test_cli.py | 64 +++++++++++++++++++++++ tests/test_permissions_cli.py | 7 +++ 4 files changed, 80 insertions(+), 6 deletions(-) diff --git a/src/keboola_agent_cli/cli.py b/src/keboola_agent_cli/cli.py index 814778fc..d6d8bfbc 100644 --- a/src/keboola_agent_cli/cli.py +++ b/src/keboola_agent_cli/cli.py @@ -179,14 +179,17 @@ def main( deny_writes: bool = typer.Option( False, "--deny-writes", - help="Session-only firewall: block every write (and destructive/admin) operation " - "for this invocation. Merges with any persisted permission policy.", + help="Session-only firewall: block write, destructive, AND admin " + "operations (the wide net -- project add/remove/edit, org setup, " + "storage writes and deletes, etc.). Merges with any persisted policy.", ), deny_destructive: bool = typer.Option( False, "--deny-destructive", - help="Session-only firewall: block destructive operations (delete/kill/reset) " - "for this invocation. Merges with any persisted permission policy.", + help="Session-only firewall: block ONLY data-destructive operations " + "(storage delete-table/delete-bucket/delete-column, job terminate, " + "branch delete, etc.). Admin ops like 'project remove' and 'org setup' " + "are NOT blocked -- use --deny-writes for the wide net.", ), ) -> None: """Global options applied to all commands.""" diff --git a/src/keboola_agent_cli/commands/context.py b/src/keboola_agent_cli/commands/context.py index 6c8b8c64..ad492c46 100644 --- a/src/keboola_agent_cli/commands/context.py +++ b/src/keboola_agent_cli/commands/context.py @@ -48,8 +48,8 @@ --no-color Disable colors (auto-disabled in non-TTY) --config-dir Override config directory path --hint MODE Generate Python code instead of executing (MODE: client or service) - --deny-writes Session-only firewall: block all write operations for this invocation - --deny-destructive Session-only firewall: block destructive (delete/kill) operations for this invocation + --deny-writes Session-only firewall: block the WIDE NET -- every write, destructive, AND admin op (project add/remove/edit, org setup, all storage mutations) + --deny-destructive Session-only firewall: NARROW -- block only data-destructive ops in Keboola (delete-table/bucket/column, terminate-job, branch delete). Admin ops (project remove, org setup) stay allowed -- use --deny-writes for those ## All Commands diff --git a/tests/test_cli.py b/tests/test_cli.py index 5bd00dc1..0f4eab6f 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -809,6 +809,70 @@ def test_deny_destructive_blocks_delete_table(self, tmp_path: Path) -> None: data = json.loads(result.output) assert data["error"]["code"] == "PERMISSION_DENIED" + def test_deny_destructive_allows_admin_ops(self, tmp_path: Path) -> None: + """--deny-destructive must NOT block admin-tier ops (project.remove, org.setup). + + Documented semantics: --deny-destructive is NARROW (data destruction + only). Admin operations fall through to --deny-writes, which is the + wide net. This test locks the contract so a future registry change + can't widen --deny-destructive's scope silently. + """ + config_dir = tmp_path / "config" + config_dir.mkdir() + self._seed(config_dir) + + with patch("keboola_agent_cli.cli.ConfigStore") as MockStore: + MockStore.return_value = ConfigStore(config_dir=config_dir) + # project.remove is classified 'admin'; --deny-destructive must + # NOT block it (permission gate exits 6 if blocked). The command + # will succeed in removing 'prod' since the seed registers it. + result = runner.invoke( + app, + [ + "--deny-destructive", + "--json", + "project", + "remove", + "--project", + "prod", + ], + ) + + assert result.exit_code != 6, ( + f"--deny-destructive incorrectly blocked admin op project.remove " + f"(exit {result.exit_code}): {result.output}" + ) + # Verify the actual operation also succeeded (not just the perm gate). + data = json.loads(result.output) + assert data["status"] == "ok" + + def test_deny_writes_blocks_admin_ops(self, tmp_path: Path) -> None: + """Complement of the above: --deny-writes IS the wide net and DOES block admin.""" + config_dir = tmp_path / "config" + config_dir.mkdir() + self._seed(config_dir) + + with patch("keboola_agent_cli.cli.ConfigStore") as MockStore: + MockStore.return_value = ConfigStore(config_dir=config_dir) + result = runner.invoke( + app, + [ + "--deny-writes", + "--json", + "project", + "remove", + "--project", + "prod", + ], + ) + + assert result.exit_code == 6, ( + f"--deny-writes must block admin op project.remove (exit 6); " + f"got {result.exit_code}: {result.output}" + ) + data = json.loads(result.output) + assert data["error"]["code"] == "PERMISSION_DENIED" + def test_deny_destructive_allows_write(self, tmp_path: Path) -> None: """--deny-destructive must NOT block pure 'write' (non-destructive) ops. diff --git a/tests/test_permissions_cli.py b/tests/test_permissions_cli.py index ebe12634..8f422108 100644 --- a/tests/test_permissions_cli.py +++ b/tests/test_permissions_cli.py @@ -153,6 +153,13 @@ def test_show_reports_session_flags_without_persisted_policy(self, tmp_path: Pat ) assert data["persisted"] is None assert "--deny-writes" in data["session_flags"] + # Defensive: legacy top-level keys (mode/allow/deny) must NOT be + # present when there is no persisted policy. Downstream consumers + # that historically used `data["mode"]` should KeyError here and + # be forced to read the new `session_flags` / `persisted` shape. + assert "mode" not in data + assert "allow" not in data + assert "deny" not in data def test_show_reports_session_flags_alongside_persisted(self, tmp_path: Path) -> None: """Session flags are reported in addition to persisted policy.""" From 9c901eebe526be9a9ee05223b59bc5f0a42b3fdf Mon Sep 17 00:00:00 2001 From: ottomansky Date: Thu, 23 Apr 2026 09:13:54 +0200 Subject: [PATCH 6/9] docs(references): add Session firewall flags section to permissions-workflow.md Document --deny-writes (WIDE) vs --deny-destructive (NARROW) semantics, REPL forwarding behavior, and the additive-merge relationship with any persisted policy (session flags never relax what is persisted, never land in config.json). Cross-references the project-pin UX. --- .../references/permissions-workflow.md | 71 +++++++++++++++++++ 1 file changed, 71 insertions(+) diff --git a/plugins/kbagent/skills/kbagent/references/permissions-workflow.md b/plugins/kbagent/skills/kbagent/references/permissions-workflow.md index fd430e08..70763fab 100644 --- a/plugins/kbagent/skills/kbagent/references/permissions-workflow.md +++ b/plugins/kbagent/skills/kbagent/references/permissions-workflow.md @@ -102,6 +102,77 @@ kbagent --json permissions list | `sync.*` | All sync subcommands (glob) | | `tool:create_*` | MCP tools matching glob pattern | +## Session firewall flags + +Two top-level flags let an operator harden a single invocation WITHOUT editing +the persisted policy in `config.json`. They are session-only, additive, and +evaluated alongside any persisted policy. + +```bash +# Wide net: blocks writes + destructive + admin +kbagent --deny-writes + +# Narrow net: blocks only data-destructive ops (pure writes still allowed) +kbagent --deny-destructive + +# Both (equivalent to --deny-writes here, since wide subsumes narrow) +kbagent --deny-writes --deny-destructive +``` + +### `--deny-writes` (WIDE) + +Appends `cli:write` and `tool:write` to the deny list. The `cli:write` pattern +intentionally spans the **write + destructive + admin** categories, so this one +flag blocks everything that mutates state -- config create/update/delete, branch +delete, project add/remove/edit, org setup, storage writes, sync push, and every +MCP write tool (`create_*`, `update_*`, `delete_*`, `add_*`, `set_*`, `remove_*`). + +Use this when you want a strict read-only session without touching the persisted +policy. Blocked operations exit with code 6 (`PERMISSION_DENIED`). + +### `--deny-destructive` (NARROW) + +Appends `cli:destructive` and `tool:destructive` to the deny list. This pattern +matches **only** operations categorized as destructive (data destruction) -- +`branch.delete`, `workspace.delete`, `config.delete`, `storage.delete-table`, +`storage.delete-bucket`, `storage.delete-column`, `job.terminate`, and MCP +`delete_*` / `remove_*` tools. + +Pure-write operations (e.g. `storage create-bucket`, `config update`) and admin +operations (e.g. `project remove`, `org setup`) are **still allowed**. Use this +when an agent needs to create/modify resources but must not be able to destroy +existing data. + +### REPL forwarding + +When invoked as `kbagent --deny-writes repl` (or `--deny-destructive`), the +flags propagate into every subcommand run inside the REPL session, so each +inner invocation picks them up automatically. A duplicate-append guard prevents +the flag from being injected twice if a user also types it explicitly on a REPL +line. + +### Relationship to persisted policy + +The session flags merge **additively** with the persisted policy for the +duration of the invocation: + +- The persisted `mode`, `allow` list, and existing `deny` entries are preserved + unchanged. Only the flag-implied deny patterns are appended (deduped). +- Session flags **can only add more deny entries** -- they NEVER relax the + persisted policy. Running `kbagent --deny-writes` against a policy that + already denies everything does not re-open anything. +- The merged policy lives in memory for this process only. It is never written + to `config.json`, so subsequent invocations without the flag revert to the + persisted policy alone. +- `kbagent permissions list` and `kbagent permissions show` render the + **effective** policy (persisted merged with session flags) so you can verify + what is actually active right now. The `session_flags` field in the JSON + output of `permissions show` surfaces which flags are in play. + +For the complementary project-pin UX (`kbagent project use `, which +persists a default project so you can drop `--project` from subsequent +commands), see the project management section of the skill. + ## Defense in depth (`--read-only`) `kbagent init --read-only` applies three layers of protection: From ed81d0751f6ca7eb048cee03ba3b0beb3cfa7e5c Mon Sep 17 00:00:00 2001 From: ottomansky Date: Thu, 23 Apr 2026 09:14:01 +0200 Subject: [PATCH 7/9] docs(gotchas): document KBAGENT_PROJECT env var semantics and precedence Record the three non-obvious rules: empty string = unset (Unix convention), pointing to an unregistered alias fails hard with CONFIG_ERROR (exit 5), and the full precedence chain --project > KBAGENT_PROJECT > persisted pin > sole-project fallback > fail. --- .../skills/kbagent/references/gotchas.md | 26 +++++++++++++++++++ 1 file changed, 26 insertions(+) diff --git a/plugins/kbagent/skills/kbagent/references/gotchas.md b/plugins/kbagent/skills/kbagent/references/gotchas.md index e8ab84f4..08db2036 100644 --- a/plugins/kbagent/skills/kbagent/references/gotchas.md +++ b/plugins/kbagent/skills/kbagent/references/gotchas.md @@ -169,6 +169,32 @@ kbagent looks for configuration in this order: Use `kbagent init` to create a local `.kbagent/` workspace for per-directory isolation. +## `KBAGENT_PROJECT` environment variable + +Lets callers override the default project for one shell/session without editing +`config.json`. A few non-obvious rules: + +- **Empty string counts as unset.** `KBAGENT_PROJECT=""` (or a value consisting + only of whitespace) is treated exactly like the variable not being set at + all. This follows the standard Unix shell convention and prevents a stray + `export KBAGENT_PROJECT=` from silently breaking every subsequent command. +- **Points to an unregistered alias -> hard fail.** If the env var names an + alias that is NOT in your configured projects, write-ops (the ones that + consult the pin) fail with `CONFIG_ERROR` and exit code 5. Repair either by + running `kbagent project use ` and unsetting the env var, or by + `unset KBAGENT_PROJECT`. The CLI will not fall back silently to the persisted + pin -- that would mask a misconfiguration. +- **Precedence for resolving the target project** (highest wins): + 1. `--project ` CLI flag (explicit per-command) + 2. `KBAGENT_PROJECT` env var + 3. Persisted pin (`default_project` in `config.json`, set via + `kbagent project use `) + 4. Sole-project fallback (if exactly one project is configured) + 5. Hard fail with `CONFIG_ERROR` (no ambiguous defaulting) +- `kbagent project current` reports which of (2) or (3) is active and flags + when the env var points to an unregistered alias, so you can diagnose + precedence issues without reading the source. + ## config update vs MCP update_config For updating configuration content, prefer `kbagent config update` over MCP's `update_config` tool: From 510e23a13f8d44331e723703cc82dfd409eb374c Mon Sep 17 00:00:00 2001 From: ottomansky Date: Thu, 23 Apr 2026 09:14:10 +0200 Subject: [PATCH 8/9] chore(hints): clarify project use/current are local-only (no --hint) Both commands operate purely on the local ConfigStore (no HTTP or service call to render), so registering HintRegistry entries would be meaningless. Add a single-line comment at each command so future contributors don't rediscover this by accident. --- src/keboola_agent_cli/commands/project.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/src/keboola_agent_cli/commands/project.py b/src/keboola_agent_cli/commands/project.py index 9191336f..210c3800 100644 --- a/src/keboola_agent_cli/commands/project.py +++ b/src/keboola_agent_cli/commands/project.py @@ -486,6 +486,7 @@ def project_use( The pin persists in config.json. ``KBAGENT_PROJECT`` overrides it for a single invocation; an explicit ``--project`` flag overrides both. """ + # No --hint: local-only ConfigStore mutation; no client or service call to render. formatter = get_formatter(ctx) service = get_service(ctx, "project_service") @@ -524,6 +525,7 @@ def project_current(ctx: typer.Context) -> None: (``env``) or the persisted pin (``pin``). Prints nothing but a hint if neither is set. """ + # No --hint: local-only ConfigStore read; no client or service call to render. formatter = get_formatter(ctx) service = get_service(ctx, "project_service") From 6a84df653c5fbc89c09c74f84d07320bf47ad277 Mon Sep 17 00:00:00 2001 From: ottomansky Date: Thu, 23 Apr 2026 09:14:16 +0200 Subject: [PATCH 9/9] refactor(cli): promote _apply_firewall_flags to public apply_firewall_flags The function is imported from commands/permissions.py, which made the underscore prefix misleading (it looked private but wasn't). Drop the leading underscore at the definition, both call sites (cli.py, commands/permissions.py), and the test suite so the API boundary is honest. --- src/keboola_agent_cli/cli.py | 4 +-- src/keboola_agent_cli/commands/permissions.py | 4 +-- tests/test_helpers.py | 34 +++++++++---------- 3 files changed, 21 insertions(+), 21 deletions(-) diff --git a/src/keboola_agent_cli/cli.py b/src/keboola_agent_cli/cli.py index d6d8bfbc..8a685a37 100644 --- a/src/keboola_agent_cli/cli.py +++ b/src/keboola_agent_cli/cli.py @@ -93,7 +93,7 @@ app.add_typer(encrypt_app, name="encrypt", rich_help_panel=_DEV) -def _apply_firewall_flags( +def apply_firewall_flags( persisted: PermissionPolicy | None, *, deny_writes: bool, @@ -272,7 +272,7 @@ def main( # Config may be invalid (e.g. corrupted JSON) -- skip persisted policy persisted_policy = None - session_policy = _apply_firewall_flags( + session_policy = apply_firewall_flags( persisted_policy, deny_writes=deny_writes, deny_destructive=deny_destructive, diff --git a/src/keboola_agent_cli/commands/permissions.py b/src/keboola_agent_cli/commands/permissions.py index c72a6c4d..29b81d66 100644 --- a/src/keboola_agent_cli/commands/permissions.py +++ b/src/keboola_agent_cli/commands/permissions.py @@ -105,7 +105,7 @@ def permissions_list( session flags like ``--deny-writes`` / ``--deny-destructive``. This matches what a command will actually do right now. """ - from ..cli import _apply_firewall_flags + from ..cli import apply_firewall_flags formatter = get_formatter(ctx) config_store: ConfigStore = get_service(ctx, "config_store") @@ -113,7 +113,7 @@ def permissions_list( deny_writes = bool(ctx.obj.get("deny_writes")) if ctx.obj else False deny_destructive = bool(ctx.obj.get("deny_destructive")) if ctx.obj else False - effective_policy = _apply_firewall_flags( + effective_policy = apply_firewall_flags( config.permissions, deny_writes=deny_writes, deny_destructive=deny_destructive, diff --git a/tests/test_helpers.py b/tests/test_helpers.py index 71baba7c..ca1dbeab 100644 --- a/tests/test_helpers.py +++ b/tests/test_helpers.py @@ -441,46 +441,46 @@ def test_fail_hard_multi_no_pin(self, tmp_config_dir, monkeypatch) -> None: class TestApplyFirewallFlags: - """Tests for cli._apply_firewall_flags (session-only policy merge).""" + """Tests for cli.apply_firewall_flags (session-only policy merge).""" def test_no_flags_returns_persisted_as_is(self) -> None: - from keboola_agent_cli.cli import _apply_firewall_flags + from keboola_agent_cli.cli import apply_firewall_flags from keboola_agent_cli.models import PermissionPolicy persisted = PermissionPolicy(mode="allow", allow=[], deny=["branch.delete"]) - result = _apply_firewall_flags(persisted, deny_writes=False, deny_destructive=False) + result = apply_firewall_flags(persisted, deny_writes=False, deny_destructive=False) assert result is persisted def test_no_flags_no_persisted_returns_none(self) -> None: - from keboola_agent_cli.cli import _apply_firewall_flags + from keboola_agent_cli.cli import apply_firewall_flags - result = _apply_firewall_flags(None, deny_writes=False, deny_destructive=False) + result = apply_firewall_flags(None, deny_writes=False, deny_destructive=False) assert result is None def test_deny_writes_synthesizes_fresh_policy(self) -> None: - from keboola_agent_cli.cli import _apply_firewall_flags + from keboola_agent_cli.cli import apply_firewall_flags - result = _apply_firewall_flags(None, deny_writes=True, deny_destructive=False) + result = apply_firewall_flags(None, deny_writes=True, deny_destructive=False) assert result is not None assert result.mode == "allow" assert "cli:write" in result.deny assert "tool:write" in result.deny def test_deny_destructive_synthesizes_fresh_policy(self) -> None: - from keboola_agent_cli.cli import _apply_firewall_flags + from keboola_agent_cli.cli import apply_firewall_flags - result = _apply_firewall_flags(None, deny_writes=False, deny_destructive=True) + result = apply_firewall_flags(None, deny_writes=False, deny_destructive=True) assert result is not None assert "cli:destructive" in result.deny assert "tool:destructive" in result.deny assert "cli:write" not in result.deny def test_flags_merge_with_persisted_deny_no_duplicates(self) -> None: - from keboola_agent_cli.cli import _apply_firewall_flags + from keboola_agent_cli.cli import apply_firewall_flags from keboola_agent_cli.models import PermissionPolicy persisted = PermissionPolicy(mode="allow", allow=[], deny=["branch.delete", "cli:write"]) - result = _apply_firewall_flags(persisted, deny_writes=True, deny_destructive=False) + result = apply_firewall_flags(persisted, deny_writes=True, deny_destructive=False) assert result is not None # Existing cli:write preserved (no dup); tool:write appended; custom entry kept. assert result.deny.count("cli:write") == 1 @@ -490,20 +490,20 @@ def test_flags_merge_with_persisted_deny_no_duplicates(self) -> None: assert result.mode == persisted.mode def test_flags_preserve_persisted_mode_deny_mode(self) -> None: - from keboola_agent_cli.cli import _apply_firewall_flags + from keboola_agent_cli.cli import apply_firewall_flags from keboola_agent_cli.models import PermissionPolicy persisted = PermissionPolicy(mode="deny", allow=["cli:read"], deny=[]) - result = _apply_firewall_flags(persisted, deny_writes=True, deny_destructive=False) + result = apply_firewall_flags(persisted, deny_writes=True, deny_destructive=False) assert result is not None assert result.mode == "deny" assert result.allow == ["cli:read"] assert "cli:write" in result.deny def test_both_flags_combine(self) -> None: - from keboola_agent_cli.cli import _apply_firewall_flags + from keboola_agent_cli.cli import apply_firewall_flags - result = _apply_firewall_flags(None, deny_writes=True, deny_destructive=True) + result = apply_firewall_flags(None, deny_writes=True, deny_destructive=True) assert result is not None # Both prefixes present. assert {"cli:write", "tool:write", "cli:destructive", "tool:destructive"} <= set( @@ -511,10 +511,10 @@ def test_both_flags_combine(self) -> None: ) def test_flags_do_not_mutate_persisted(self) -> None: - from keboola_agent_cli.cli import _apply_firewall_flags + from keboola_agent_cli.cli import apply_firewall_flags from keboola_agent_cli.models import PermissionPolicy persisted = PermissionPolicy(mode="allow", allow=[], deny=["branch.delete"]) before = list(persisted.deny) - _apply_firewall_flags(persisted, deny_writes=True, deny_destructive=True) + apply_firewall_flags(persisted, deny_writes=True, deny_destructive=True) assert persisted.deny == before, "persisted.deny was mutated in place"