Phase 1: Security Hardening

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()

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)

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)

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}",

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."""

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,