diff --git a/src/keboola_agent_cli/client.py b/src/keboola_agent_cli/client.py index 3a9a73cc..2bd8fe4b 100644 --- a/src/keboola_agent_cli/client.py +++ b/src/keboola_agent_cli/client.py @@ -7,7 +7,7 @@ import time from typing import Any -from urllib.parse import urlparse, urlunparse +from urllib.parse import quote, urlparse, urlunparse import httpx @@ -158,6 +158,11 @@ def _raise_api_error(self, response: httpx.Response, base_url: str | None = None except Exception: api_message = response.text + # Truncate to prevent Rich markup injection and excessive output + max_api_error_length = 500 + if isinstance(api_message, str) and len(api_message) > max_api_error_length: + api_message = api_message[:max_api_error_length] + "..." + if status == 401: raise KeboolaApiError( message=f"Invalid or expired token (token: {self._masked_token}): {api_message}", @@ -236,9 +241,11 @@ def get_config_detail(self, component_id: str, config_id: str) -> dict[str, Any] Returns: Configuration detail dict from the API. """ + safe_component_id = quote(component_id, safe="") + safe_config_id = quote(config_id, safe="") response = self._request( "GET", - f"/v2/storage/components/{component_id}/configs/{config_id}", + f"/v2/storage/components/{safe_component_id}/configs/{safe_config_id}", ) return response.json() @@ -297,5 +304,6 @@ def get_job_detail(self, job_id: str) -> dict[str, Any]: Returns: Job detail dict from the Queue API. """ - response = self._queue_request("GET", f"/jobs/{job_id}") + safe_job_id = quote(job_id, safe="") + response = self._queue_request("GET", f"/jobs/{safe_job_id}") return response.json() diff --git a/src/keboola_agent_cli/commands/project.py b/src/keboola_agent_cli/commands/project.py index e80ce357..6f1519bb 100644 --- a/src/keboola_agent_cli/commands/project.py +++ b/src/keboola_agent_cli/commands/project.py @@ -4,6 +4,8 @@ No business logic belongs here. """ +import os +import sys from typing import Any import typer @@ -85,6 +87,36 @@ def _format_status_table(console: Console, statuses: list[dict[str, Any]]) -> No console.print(table) +def _resolve_token() -> str: + """Resolve the Storage API token from env var or interactive prompt. + + Token resolution order: + 1. KBC_TOKEN env var (for CI/CD and automation) + 2. Interactive prompt with hidden input (if TTY) + 3. Error if neither available + + Returns: + The Storage API token. + + Raises: + typer.Exit: If no token can be resolved. + """ + env_token = os.environ.get("KBC_TOKEN") + if env_token: + return env_token + + is_tty = hasattr(sys.stdin, "isatty") and sys.stdin.isatty() + if is_tty: + return typer.prompt("Storage API token", hide_input=True) + + typer.echo( + "Error: No token available. Set KBC_TOKEN env var " + "or run interactively.", + err=True, + ) + raise typer.Exit(code=2) + + @project_app.command("add") def project_add( ctx: typer.Context, @@ -94,15 +126,15 @@ def project_add( help="Keboola stack URL", envvar="KBC_STORAGE_API_URL", ), - token: str = typer.Option( - ..., - help="Storage API token", - envvar="KBC_TOKEN", - ), ) -> None: - """Add a new Keboola project connection.""" + """Add a new Keboola project connection. + + The Storage API token is read from KBC_TOKEN env var or prompted + interactively (never passed as a CLI argument for security). + """ formatter = _get_formatter(ctx) service = _get_service(ctx) + token = _resolve_token() try: result = service.add_project(alias=alias, stack_url=url, token=token) @@ -164,11 +196,23 @@ def project_edit( ctx: typer.Context, alias: str = typer.Option(..., help="Alias of the project to edit"), url: str | None = typer.Option(None, help="New Keboola stack URL"), - token: str | None = typer.Option(None, help="New Storage API token"), + new_token: bool = typer.Option( + False, + "--new-token", + help="Provide a new Storage API token (from KBC_TOKEN env var or interactive prompt)", + ), ) -> None: - """Edit an existing Keboola project connection.""" + """Edit an existing Keboola project connection. + + To change the token, use --new-token flag. The token is read from + KBC_TOKEN env var or prompted interactively (never passed as a CLI + argument for security). + """ formatter = _get_formatter(ctx) service = _get_service(ctx) + token: str | None = None + if new_token: + token = _resolve_token() try: result = service.edit_project(alias=alias, stack_url=url, token=token) diff --git a/src/keboola_agent_cli/config_store.py b/src/keboola_agent_cli/config_store.py index 4e284222..e98d995e 100644 --- a/src/keboola_agent_cli/config_store.py +++ b/src/keboola_agent_cli/config_store.py @@ -86,7 +86,7 @@ def save(self, config: AppConfig) -> None: ConfigError: If the file cannot be written. """ try: - self._config_dir.mkdir(parents=True, exist_ok=True) + self._config_dir.mkdir(parents=True, exist_ok=True, mode=0o700) json_str = config.model_dump_json(indent=2) self._config_path.write_text(json_str + "\n", encoding="utf-8") self._config_path.chmod(0o600) diff --git a/src/keboola_agent_cli/manage_client.py b/src/keboola_agent_cli/manage_client.py index 76bde599..d95ded83 100644 --- a/src/keboola_agent_cli/manage_client.py +++ b/src/keboola_agent_cli/manage_client.py @@ -120,6 +120,11 @@ def _raise_api_error(self, response: httpx.Response) -> None: except Exception: api_message = response.text + # Truncate to prevent Rich markup injection and excessive output + max_api_error_length = 500 + if isinstance(api_message, str) and len(api_message) > max_api_error_length: + api_message = api_message[:max_api_error_length] + "..." + if status == 401: raise KeboolaApiError( message=f"Invalid or expired manage token (token: {self._masked_token}): {api_message}", diff --git a/src/keboola_agent_cli/models.py b/src/keboola_agent_cli/models.py index 66bac4c0..83c25d6e 100644 --- a/src/keboola_agent_cli/models.py +++ b/src/keboola_agent_cli/models.py @@ -2,7 +2,7 @@ from typing import Any -from pydantic import BaseModel, Field +from pydantic import BaseModel, Field, field_validator class ProjectConfig(BaseModel): @@ -15,6 +15,17 @@ class ProjectConfig(BaseModel): ) project_id: int = Field(default=0, description="Keboola project ID (populated on add)") + @field_validator("stack_url") + @classmethod + def validate_stack_url_scheme(cls, v: str) -> str: + """Enforce HTTPS scheme on stack URL to prevent SSRF and protocol abuse.""" + if not v.startswith("https://"): + raise ValueError( + f"Stack URL must use https:// scheme, got: {v!r}. " + "Plain HTTP, file://, and other protocols are not allowed." + ) + return v + class AppConfig(BaseModel): """Top-level application configuration persisted to config.json.""" diff --git a/src/keboola_agent_cli/services/project_service.py b/src/keboola_agent_cli/services/project_service.py index 4cf44c4c..86d2e8f5 100644 --- a/src/keboola_agent_cli/services/project_service.py +++ b/src/keboola_agent_cli/services/project_service.py @@ -122,7 +122,11 @@ def edit_project( self._config_store.edit_project(alias, **updates) updated = self._config_store.get_project(alias) - assert updated is not None # we just edited it + if updated is None: + raise ConfigError( + f"Project '{alias}' could not be retrieved after editing. " + "Config store may be in an inconsistent state." + ) return { "alias": alias, diff --git a/tests/test_cli.py b/tests/test_cli.py index 055f5866..1cb3ae38 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -1,6 +1,7 @@ """Tests for CLI commands via CliRunner - project, config, context, doctor commands.""" import json +import os from pathlib import Path from unittest.mock import MagicMock, patch @@ -15,6 +16,8 @@ from keboola_agent_cli.services.lineage_service import LineageService from keboola_agent_cli.services.project_service import ProjectService +TEST_TOKEN = "901-10493007-VDtlEDWDF6Tx5V8jjE8FshFlqM0Hl0c08KHqpt0k" + runner = CliRunner() @@ -46,6 +49,7 @@ def test_project_add_success_json(self, tmp_path: Path) -> None: 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_instance = ConfigStore(config_dir=config_dir) MockStore.return_value = store_instance @@ -66,8 +70,6 @@ def test_project_add_success_json(self, tmp_path: Path) -> None: "prod", "--url", "https://connection.keboola.com", - "--token", - "901-10493007-VDtlEDWDF6Tx5V8jjE8FshFlqM0Hl0c08KHqpt0k", ], ) @@ -89,6 +91,7 @@ def test_project_add_success_human(self, tmp_path: Path) -> None: 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_instance = ConfigStore(config_dir=config_dir) MockStore.return_value = store_instance @@ -108,8 +111,6 @@ def test_project_add_success_human(self, tmp_path: Path) -> None: "test", "--url", "https://connection.keboola.com", - "--token", - "901-10493007-VDtlEDWDF6Tx5V8jjE8FshFlqM0Hl0c08KHqpt0k", ], ) @@ -133,6 +134,7 @@ def test_project_add_invalid_token_exit_code_3(self, tmp_path: Path) -> None: with ( patch("keboola_agent_cli.cli.ConfigStore") as MockStore, patch("keboola_agent_cli.cli.ProjectService") as MockService, + patch.dict(os.environ, {"KBC_TOKEN": "invalid-token-abcdefgh"}), ): store_instance = ConfigStore(config_dir=config_dir) MockStore.return_value = store_instance @@ -151,8 +153,6 @@ def test_project_add_invalid_token_exit_code_3(self, tmp_path: Path) -> None: "add", "--alias", "bad", - "--token", - "invalid-token-abcdefgh", ], ) @@ -177,6 +177,7 @@ def test_project_add_timeout_exit_code_4(self, tmp_path: Path) -> None: 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_instance = ConfigStore(config_dir=config_dir) MockStore.return_value = store_instance @@ -195,8 +196,6 @@ def test_project_add_timeout_exit_code_4(self, tmp_path: Path) -> None: "add", "--alias", "timeout", - "--token", - "901-10493007-VDtlEDWDF6Tx5V8jjE8FshFlqM0Hl0c08KHqpt0k", ], ) @@ -235,6 +234,7 @@ def test_project_list_json_with_projects(self, tmp_path: Path) -> None: 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_instance = ConfigStore(config_dir=config_dir) MockStore.return_value = store_instance @@ -255,8 +255,6 @@ def test_project_list_json_with_projects(self, tmp_path: Path) -> None: "test", "--url", "https://connection.keboola.com", - "--token", - "901-10493007-VDtlEDWDF6Tx5V8jjE8FshFlqM0Hl0c08KHqpt0k", ], ) @@ -268,8 +266,7 @@ def test_project_list_json_with_projects(self, tmp_path: Path) -> None: assert len(output["data"]) == 1 assert output["data"][0]["alias"] == "test" # Token must be masked - full_token = "901-10493007-VDtlEDWDF6Tx5V8jjE8FshFlqM0Hl0c08KHqpt0k" - assert output["data"][0]["token"] != full_token + assert output["data"][0]["token"] != TEST_TOKEN def test_project_list_human_mode(self, tmp_path: Path) -> None: """project list in human mode outputs a Rich table.""" @@ -280,6 +277,7 @@ def test_project_list_human_mode(self, tmp_path: Path) -> None: 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_instance = ConfigStore(config_dir=config_dir) MockStore.return_value = store_instance @@ -299,8 +297,6 @@ def test_project_list_human_mode(self, tmp_path: Path) -> None: "prod", "--url", "https://connection.keboola.com", - "--token", - "901-10493007-VDtlEDWDF6Tx5V8jjE8FshFlqM0Hl0c08KHqpt0k", ], ) @@ -341,6 +337,7 @@ def test_project_remove_success_json(self, tmp_path: Path) -> None: 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_instance = ConfigStore(config_dir=config_dir) MockStore.return_value = store_instance @@ -358,8 +355,6 @@ def test_project_remove_success_json(self, tmp_path: Path) -> None: "add", "--alias", "test", - "--token", - "901-10493007-VDtlEDWDF6Tx5V8jjE8FshFlqM0Hl0c08KHqpt0k", ], ) @@ -421,6 +416,7 @@ def test_project_status_json(self, tmp_path: Path) -> None: 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_instance = ConfigStore(config_dir=config_dir) MockStore.return_value = store_instance @@ -438,8 +434,6 @@ def test_project_status_json(self, tmp_path: Path) -> None: "add", "--alias", "prod", - "--token", - "901-10493007-VDtlEDWDF6Tx5V8jjE8FshFlqM0Hl0c08KHqpt0k", ], ) @@ -462,6 +456,7 @@ def test_project_status_human(self, tmp_path: Path) -> None: 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_instance = ConfigStore(config_dir=config_dir) MockStore.return_value = store_instance @@ -479,8 +474,6 @@ def test_project_status_human(self, tmp_path: Path) -> None: "add", "--alias", "test", - "--token", - "901-10493007-VDtlEDWDF6Tx5V8jjE8FshFlqM0Hl0c08KHqpt0k", ], ) @@ -502,6 +495,7 @@ def test_project_edit_url_json(self, tmp_path: Path) -> None: 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_instance = ConfigStore(config_dir=config_dir) MockStore.return_value = store_instance @@ -521,8 +515,6 @@ def test_project_edit_url_json(self, tmp_path: Path) -> None: "test", "--url", "https://old.keboola.com", - "--token", - "901-10493007-VDtlEDWDF6Tx5V8jjE8FshFlqM0Hl0c08KHqpt0k", ], ) @@ -553,6 +545,7 @@ def test_project_edit_config_error_exit_code_5(self, tmp_path: Path) -> None: 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_instance = ConfigStore(config_dir=config_dir) MockStore.return_value = store_instance @@ -570,8 +563,6 @@ def test_project_edit_config_error_exit_code_5(self, tmp_path: Path) -> None: "add", "--alias", "test", - "--token", - "901-10493007-VDtlEDWDF6Tx5V8jjE8FshFlqM0Hl0c08KHqpt0k", ], ) @@ -2306,6 +2297,7 @@ def test_auth_error_exit_code_3(self, tmp_path: Path) -> None: with ( patch("keboola_agent_cli.cli.ConfigStore") as MockStore, patch("keboola_agent_cli.cli.ProjectService") as MockService, + patch.dict(os.environ, {"KBC_TOKEN": "invalid-token-abcdefgh"}), ): store_instance = ConfigStore(config_dir=config_dir) MockStore.return_value = store_instance @@ -2322,8 +2314,6 @@ def test_auth_error_exit_code_3(self, tmp_path: Path) -> None: "add", "--alias", "bad", - "--token", - "invalid-token-abcdefgh", ], ) @@ -2348,6 +2338,7 @@ def test_network_error_exit_code_4(self, tmp_path: Path) -> None: 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_instance = ConfigStore(config_dir=config_dir) MockStore.return_value = store_instance @@ -2364,8 +2355,6 @@ def test_network_error_exit_code_4(self, tmp_path: Path) -> None: "add", "--alias", "unreachable", - "--token", - "901-10493007-VDtlEDWDF6Tx5V8jjE8FshFlqM0Hl0c08KHqpt0k", ], ) @@ -2570,12 +2559,13 @@ def test_config_help(self) -> None: assert "detail" in result.output def test_project_add_help(self) -> None: - """project add --help shows options.""" + """project add --help shows options but NOT --token (security).""" result = runner.invoke(app, ["project", "add", "--help"]) assert result.exit_code == 0 assert "--alias" in result.output - assert "--token" in result.output assert "--url" in result.output + # --token must NOT appear as a CLI option (S1 security fix) + assert "--token" not in result.output def test_config_list_help(self) -> None: """config list --help shows options.""" @@ -2650,6 +2640,7 @@ def test_project_add_missing_alias(self, tmp_path: Path) -> None: 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}), ): MockStore.return_value = ConfigStore(config_dir=config_dir) MockService.return_value = ProjectService(config_store=MockStore.return_value) @@ -2659,22 +2650,23 @@ def test_project_add_missing_alias(self, tmp_path: Path) -> None: [ "project", "add", - "--token", - "901-10493007-VDtlEDWDF6Tx5V8jjE8FshFlqM0Hl0c08KHqpt0k", ], ) assert result.exit_code != 0 - def test_project_add_missing_token(self, tmp_path: Path) -> None: - """project add without --token shows error.""" + def test_project_add_missing_token_non_tty(self, tmp_path: Path) -> None: + """project add without KBC_TOKEN in non-TTY exits with code 2.""" config_dir = tmp_path / "config" config_dir.mkdir() with ( patch("keboola_agent_cli.cli.ConfigStore") as MockStore, patch("keboola_agent_cli.cli.ProjectService") as MockService, + patch.dict(os.environ, {}, clear=False), ): + # Ensure KBC_TOKEN is not set + os.environ.pop("KBC_TOKEN", None) MockStore.return_value = ConfigStore(config_dir=config_dir) MockService.return_value = ProjectService(config_store=MockStore.return_value) @@ -2739,7 +2731,7 @@ class TestProjectEditTokenReverify: """Tests for project edit with token changes.""" def test_project_edit_token_reverify_json(self, tmp_path: Path) -> None: - """project edit --token triggers re-verification and returns updated info.""" + """project edit --new-token triggers re-verification and returns updated info.""" config_dir = tmp_path / "config" config_dir.mkdir() mock_client = _make_mock_client(project_name="Original", project_id=100) @@ -2757,6 +2749,7 @@ def client_factory(url, token): 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_instance = ConfigStore(config_dir=config_dir) MockStore.return_value = store_instance @@ -2775,12 +2768,10 @@ def client_factory(url, token): "add", "--alias", "test", - "--token", - "901-10493007-VDtlEDWDF6Tx5V8jjE8FshFlqM0Hl0c08KHqpt0k", ], ) - # Edit with new token + # Edit with new token via --new-token flag + env var result = runner.invoke( app, [ @@ -2789,14 +2780,209 @@ def client_factory(url, token): "edit", "--alias", "test", - "--token", - "902-newtoken-abcdefghijklmn", + "--new-token", + ], + ) + + assert result.exit_code == 0, f"Exit code {result.exit_code}: {result.output}" + output = json.loads(result.output) + assert output["status"] == "ok" + + +class TestProjectAddTokenSecurity: + """Tests for S1: Token input security (env var and interactive prompt).""" + + def test_project_add_token_from_env(self, tmp_path: Path) -> None: + """Token from KBC_TOKEN env var works for project add.""" + mock_client = _make_mock_client(project_name="EnvProject", project_id=999) + config_dir = tmp_path / "config" + config_dir.mkdir() + + 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_instance = ConfigStore(config_dir=config_dir) + MockStore.return_value = store_instance + + service_instance = ProjectService( + config_store=store_instance, + client_factory=lambda url, token: mock_client, + ) + MockService.return_value = service_instance + + result = runner.invoke( + app, + [ + "--json", + "project", + "add", + "--alias", + "envtest", + "--url", + "https://connection.keboola.com", + ], + ) + + assert result.exit_code == 0, f"Exit code {result.exit_code}: {result.output}" + output = json.loads(result.output) + assert output["status"] == "ok" + assert output["data"]["alias"] == "envtest" + assert output["data"]["project_name"] == "EnvProject" + + def test_project_add_token_interactive(self, tmp_path: Path) -> None: + """Interactive hidden prompt works for project add when no env var. + + We mock _resolve_token to simulate the interactive prompt returning a token, + since CliRunner does not have a real TTY and sys.stdin.isatty() is False. + """ + mock_client = _make_mock_client(project_name="PromptProject", project_id=888) + config_dir = tmp_path / "config" + config_dir.mkdir() + + with ( + patch("keboola_agent_cli.cli.ConfigStore") as MockStore, + patch("keboola_agent_cli.cli.ProjectService") as MockService, + patch( + "keboola_agent_cli.commands.project._resolve_token", + return_value=TEST_TOKEN, + ), + ): + store_instance = ConfigStore(config_dir=config_dir) + MockStore.return_value = store_instance + + service_instance = ProjectService( + config_store=store_instance, + client_factory=lambda url, token: mock_client, + ) + MockService.return_value = service_instance + + result = runner.invoke( + app, + [ + "--json", + "project", + "add", + "--alias", + "prompttest", + "--url", + "https://connection.keboola.com", + ], + ) + + assert result.exit_code == 0, f"Exit code {result.exit_code}: {result.output}" + output = json.loads(result.output) + assert output["status"] == "ok" + assert output["data"]["alias"] == "prompttest" + + def test_project_add_rejects_http_url(self, tmp_path: Path) -> None: + """http:// URL rejected with error at project add time.""" + mock_client = _make_mock_client() + config_dir = tmp_path / "config" + config_dir.mkdir() + + 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_instance = ConfigStore(config_dir=config_dir) + MockStore.return_value = store_instance + + service_instance = ProjectService( + config_store=store_instance, + client_factory=lambda url, token: mock_client, + ) + MockService.return_value = service_instance + + result = runner.invoke( + app, + [ + "--json", + "project", + "add", + "--alias", + "insecure", + "--url", + "http://connection.keboola.com", + ], + ) + + assert result.exit_code != 0, f"Expected failure but got: {result.output}" + + def test_project_add_rejects_file_url(self, tmp_path: Path) -> None: + """file:// URL rejected with error at project add time.""" + mock_client = _make_mock_client() + config_dir = tmp_path / "config" + config_dir.mkdir() + + 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_instance = ConfigStore(config_dir=config_dir) + MockStore.return_value = store_instance + + service_instance = ProjectService( + config_store=store_instance, + client_factory=lambda url, token: mock_client, + ) + MockService.return_value = service_instance + + result = runner.invoke( + app, + [ + "--json", + "project", + "add", + "--alias", + "fileurl", + "--url", + "file:///etc/passwd", + ], + ) + + assert result.exit_code != 0, f"Expected failure but got: {result.output}" + + def test_project_add_accepts_https_url(self, tmp_path: Path) -> None: + """https:// URL is accepted at project add time.""" + mock_client = _make_mock_client(project_name="SecureProject", project_id=777) + config_dir = tmp_path / "config" + config_dir.mkdir() + + 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_instance = ConfigStore(config_dir=config_dir) + MockStore.return_value = store_instance + + service_instance = ProjectService( + config_store=store_instance, + client_factory=lambda url, token: mock_client, + ) + MockService.return_value = service_instance + + result = runner.invoke( + app, + [ + "--json", + "project", + "add", + "--alias", + "secure", + "--url", + "https://connection.keboola.com", ], ) assert result.exit_code == 0, f"Exit code {result.exit_code}: {result.output}" output = json.loads(result.output) assert output["status"] == "ok" + assert output["data"]["alias"] == "secure" # --------------------------------------------------------------------------- diff --git a/tests/test_client.py b/tests/test_client.py index 48d08778..7e2712ed 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -1,5 +1,7 @@ """Tests for KeboolaClient - verify_token, retries, timeouts, error handling.""" +from urllib.parse import quote + import httpx import pytest @@ -988,3 +990,160 @@ def test_list_buckets_empty(self, httpx_mock) -> None: ) as client: result = client.list_buckets() assert result == [] + + +class TestApiErrorMessageTruncation: + """Tests for S4: API error message truncation to 500 characters.""" + + def test_api_error_message_truncation(self, httpx_mock) -> None: + """Long server response is truncated to 500 characters in error message.""" + long_message = "A" * 1000 + httpx_mock.add_response( + url="https://connection.keboola.com/v2/storage/tokens/verify", + json={"error": long_message}, + status_code=400, + ) + + with KeboolaClient( + stack_url="https://connection.keboola.com", + token="901-10493007-VDtlEDWDF6Tx5V8jjE8FshFlqM0Hl0c08KHqpt0k", + ) as client: + with pytest.raises(KeboolaApiError) as exc_info: + client.verify_token() + + # The full 1000 char message should NOT appear + assert long_message not in exc_info.value.message + # The truncated message (500 chars + "...") should be present + assert "A" * 500 + "..." in exc_info.value.message + + def test_short_error_message_not_truncated(self, httpx_mock) -> None: + """Short server response is not truncated.""" + short_message = "Bad request" + httpx_mock.add_response( + url="https://connection.keboola.com/v2/storage/tokens/verify", + json={"error": short_message}, + status_code=400, + ) + + with KeboolaClient( + stack_url="https://connection.keboola.com", + token="901-10493007-VDtlEDWDF6Tx5V8jjE8FshFlqM0Hl0c08KHqpt0k", + ) as client: + with pytest.raises(KeboolaApiError) as exc_info: + client.verify_token() + + assert short_message in exc_info.value.message + # Should not have the truncation indicator + assert "..." not in exc_info.value.message or short_message in exc_info.value.message + + def test_exactly_500_chars_not_truncated(self, httpx_mock) -> None: + """Error message of exactly 500 characters is not truncated.""" + exact_message = "B" * 500 + httpx_mock.add_response( + url="https://connection.keboola.com/v2/storage/tokens/verify", + json={"error": exact_message}, + status_code=400, + ) + + with KeboolaClient( + stack_url="https://connection.keboola.com", + token="901-10493007-VDtlEDWDF6Tx5V8jjE8FshFlqM0Hl0c08KHqpt0k", + ) as client: + with pytest.raises(KeboolaApiError) as exc_info: + client.verify_token() + + # Exactly 500 chars should not be truncated + assert exact_message in exc_info.value.message + + def test_rich_markup_in_error_truncated(self, httpx_mock) -> None: + """Rich markup brackets in error messages are contained by truncation.""" + malicious_msg = "[bold red]" + "X" * 600 + "[/bold red]" + httpx_mock.add_response( + url="https://connection.keboola.com/v2/storage/tokens/verify", + json={"error": malicious_msg}, + status_code=400, + ) + + with KeboolaClient( + stack_url="https://connection.keboola.com", + token="901-10493007-VDtlEDWDF6Tx5V8jjE8FshFlqM0Hl0c08KHqpt0k", + ) as client: + with pytest.raises(KeboolaApiError) as exc_info: + client.verify_token() + + # Full malicious markup should not appear + assert malicious_msg not in exc_info.value.message + # Should be truncated + assert "..." in exc_info.value.message + + +class TestUrlPathEncoding: + """Tests for S5: URL-encode path parameters to prevent path traversal.""" + + def test_url_path_encoding_component_id(self, httpx_mock) -> None: + """Special characters in component_id are URL-encoded.""" + encoded_component = quote("keboola.ex-db/../admin", safe="") + encoded_config = quote("42", safe="") + + httpx_mock.add_response( + url=f"https://connection.keboola.com/v2/storage/components/{encoded_component}/configs/{encoded_config}", + json={"id": "42", "name": "Config"}, + status_code=200, + ) + + with KeboolaClient( + stack_url="https://connection.keboola.com", + token="901-10493007-VDtlEDWDF6Tx5V8jjE8FshFlqM0Hl0c08KHqpt0k", + ) as client: + result = client.get_config_detail("keboola.ex-db/../admin", "42") + assert result["id"] == "42" + + def test_url_path_encoding_config_id(self, httpx_mock) -> None: + """Special characters in config_id are URL-encoded.""" + encoded_component = quote("keboola.ex-db-snowflake", safe="") + encoded_config = quote("42/../secret", safe="") + + httpx_mock.add_response( + url=f"https://connection.keboola.com/v2/storage/components/{encoded_component}/configs/{encoded_config}", + json={"id": "42", "name": "Config"}, + status_code=200, + ) + + with KeboolaClient( + stack_url="https://connection.keboola.com", + token="901-10493007-VDtlEDWDF6Tx5V8jjE8FshFlqM0Hl0c08KHqpt0k", + ) as client: + result = client.get_config_detail("keboola.ex-db-snowflake", "42/../secret") + assert result["id"] == "42" + + def test_url_path_encoding_job_id(self, httpx_mock) -> None: + """Special characters in job_id are URL-encoded.""" + encoded_job = quote("1001/../admin", safe="") + + httpx_mock.add_response( + url=f"https://queue.keboola.com/jobs/{encoded_job}", + json={"id": "1001", "status": "success"}, + status_code=200, + ) + + with KeboolaClient( + stack_url="https://connection.keboola.com", + token="901-10493007-VDtlEDWDF6Tx5V8jjE8FshFlqM0Hl0c08KHqpt0k", + ) as client: + result = client.get_job_detail("1001/../admin") + assert result["id"] == "1001" + + def test_normal_ids_not_affected(self, httpx_mock) -> None: + """Normal IDs without special chars work correctly with encoding.""" + httpx_mock.add_response( + url="https://connection.keboola.com/v2/storage/components/keboola.ex-db-snowflake/configs/42", + json={"id": "42", "name": "Config"}, + status_code=200, + ) + + with KeboolaClient( + stack_url="https://connection.keboola.com", + token="901-10493007-VDtlEDWDF6Tx5V8jjE8FshFlqM0Hl0c08KHqpt0k", + ) as client: + result = client.get_config_detail("keboola.ex-db-snowflake", "42") + assert result["id"] == "42" diff --git a/tests/test_config_store.py b/tests/test_config_store.py index 1bc6a76c..9a34afe0 100644 --- a/tests/test_config_store.py +++ b/tests/test_config_store.py @@ -535,6 +535,39 @@ def test_load_unreadable_config_file(self, tmp_config_dir: Path) -> None: config_file.chmod(0o644) +class TestDirectoryPermissions: + """Tests for S3: config directory permissions.""" + + def test_config_dir_permissions(self, tmp_path: Path) -> None: + """Config directory is created with 0o700 permissions (owner only).""" + nested_dir = tmp_path / "secure" / "config" + store = ConfigStore(config_dir=nested_dir) + store.save(AppConfig()) + + dir_stat = os.stat(nested_dir) + mode = stat.S_IMODE(dir_stat.st_mode) + + # Directory should be owner-only accessible (0o700) + assert mode == 0o700, f"Expected 0o700, got {oct(mode)}" + + def test_config_dir_permissions_on_add_project(self, tmp_path: Path) -> None: + """Config directory created via add_project also has 0o700 permissions.""" + new_dir = tmp_path / "fresh" / "secure_config" + store = ConfigStore(config_dir=new_dir) + store.add_project( + "test", + ProjectConfig( + stack_url="https://connection.keboola.com", + token="901-abcdef-12345678", + ), + ) + + dir_stat = os.stat(new_dir) + mode = stat.S_IMODE(dir_stat.st_mode) + + assert mode == 0o700, f"Expected 0o700, got {oct(mode)}" + + class TestConfigPath: """Tests for config_path property.""" diff --git a/tests/test_models.py b/tests/test_models.py index 4adc416a..b9058608 100644 --- a/tests/test_models.py +++ b/tests/test_models.py @@ -2,6 +2,9 @@ import json +import pytest +from pydantic import ValidationError + from keboola_agent_cli.models import AppConfig, ErrorResponse, ProjectConfig, SuccessResponse @@ -190,3 +193,63 @@ def test_json_round_trip(self) -> None: json_str = original.model_dump_json() restored = SuccessResponse.model_validate_json(json_str) assert restored == original + + +class TestStackUrlValidation: + """Tests for S2: URL validation on ProjectConfig.stack_url.""" + + def test_project_add_rejects_http_url(self) -> None: + """http:// URL is rejected with a ValidationError.""" + with pytest.raises(ValidationError, match="https://"): + ProjectConfig( + stack_url="http://connection.keboola.com", + token="901-token", + ) + + def test_project_add_rejects_file_url(self) -> None: + """file:// URL is rejected with a ValidationError.""" + with pytest.raises(ValidationError, match="https://"): + ProjectConfig( + stack_url="file:///etc/passwd", + token="901-token", + ) + + def test_project_add_rejects_ftp_url(self) -> None: + """ftp:// URL is rejected with a ValidationError.""" + with pytest.raises(ValidationError, match="https://"): + ProjectConfig( + stack_url="ftp://connection.keboola.com", + token="901-token", + ) + + def test_project_add_rejects_no_scheme(self) -> None: + """URL without scheme is rejected with a ValidationError.""" + with pytest.raises(ValidationError, match="https://"): + ProjectConfig( + stack_url="connection.keboola.com", + token="901-token", + ) + + def test_project_add_accepts_https_url(self) -> None: + """https:// URL is accepted without error.""" + config = ProjectConfig( + stack_url="https://connection.keboola.com", + token="901-token", + ) + assert config.stack_url == "https://connection.keboola.com" + + def test_project_add_accepts_https_azure(self) -> None: + """https:// Azure stack URL is accepted.""" + config = ProjectConfig( + stack_url="https://connection.north-europe.azure.keboola.com", + token="901-token", + ) + assert config.stack_url == "https://connection.north-europe.azure.keboola.com" + + def test_project_add_accepts_https_gcp(self) -> None: + """https:// GCP stack URL is accepted.""" + config = ProjectConfig( + stack_url="https://connection.europe-west3.gcp.keboola.com", + token="901-token", + ) + assert config.stack_url == "https://connection.europe-west3.gcp.keboola.com"