feat(encrypt): --json plaintext_written field across secret-write paths (GHSA-7jrf follow-up, code-only)

src/keboola_agent_cli/commands/_data_app_runtime.py

@@ -209,6 +209,19 @@ def data_app_secrets_set(
                 style="yellow",
             )
 
+    # Plaintext-on-encrypt-failure fallback -- name the secret key-paths written
+    # in PLAINTEXT (keys only) so the leak is visible. JSON mode carries the same
+    # list on the envelope via plaintext_written, so warn only in human mode.
+    plaintext_written = result.get("plaintext_written")
+    if plaintext_written and not formatter.json_mode:
+        formatter.warning(
+            f"{len(plaintext_written)} secret(s) were written in PLAINTEXT (encryption "
+            f"failed and --allow-plaintext-on-encrypt-failure was set): "
+            f"{', '.join(plaintext_written)}. Rotate these credentials and re-encrypt "
+            f"once the Encryption API is reachable -- config version history retains the "
+            f"plaintext copy."
+        )
+
     if no_hint_next and isinstance(result, dict):
         result.pop("next_step", None)
 

src/keboola_agent_cli/commands/config.py

@@ -711,6 +711,7 @@ def config_update(
             f"{branch_info}"
         )
         _emit_normalizations_warning(formatter, normalizations)
+        _emit_plaintext_written_warning(formatter, result)
 
 
 def _emit_normalizations_warning(formatter: Any, normalizations: list[dict[str, Any]]) -> None:
@@ -735,6 +736,26 @@ def _emit_normalizations_warning(formatter: Any, normalizations: list[dict[str,
         )
 
 
+def _emit_plaintext_written_warning(formatter: Any, result: dict[str, Any]) -> None:
+    """Surface a plaintext-on-encrypt-failure fallback in human mode.
+
+    When ``--allow-plaintext-on-encrypt-failure`` lets a write proceed despite a
+    failed encryption, the service result carries ``plaintext_written`` -- the
+    secret key-paths now stored in PLAINTEXT (key-paths only, never the values).
+    Name them and the remediation so the leak is visible and actionable. JSON
+    mode already exposes the same list on the envelope, so emit only here.
+    """
+    leaked = result.get("plaintext_written")
+    if not leaked:
+        return
+    formatter.warning(
+        f"{len(leaked)} secret(s) were written in PLAINTEXT (encryption failed and "
+        f"--allow-plaintext-on-encrypt-failure was set): {', '.join(leaked)}. "
+        f"Rotate these credentials and re-encrypt once the Encryption API is reachable "
+        f"-- config version history retains the plaintext copy."
+    )
+
+
 @config_app.command("set-default-bucket", rich_help_panel="Storage")
 def config_set_default_bucket(
     ctx: typer.Context,
@@ -1417,6 +1438,7 @@ def _render_push_result_human(
             "[dim]Note: schema validation was skipped "
             "(empty shell, no schema available, or --no-validate).[/dim]"
         )
+    _emit_plaintext_written_warning(formatter, result)
 
 
 # ── Config metadata commands ───────────────────────────────────────────
@@ -1771,6 +1793,7 @@ def config_variables_set(
         formatter.output(result)
     else:
         _format_variables_set(formatter, result)
+        _emit_plaintext_written_warning(formatter, result)
 
 
 @config_app.command("variables-get", rich_help_panel="Variables")
@@ -2095,6 +2118,7 @@ def config_row_create(
             f"Created row '{escape(row_name)}' [{row_id}] "
             f"in {escape(component_id)}/{escape(config_id)}{branch_info}"
         )
+        _emit_plaintext_written_warning(formatter, result)
 
 
 # ── config row-update ──────────────────────────────────────────────────────────
@@ -2298,6 +2322,7 @@ def config_row_update(
             f"Updated row '{escape(updated_name)}' [{row_id}] "
             f"in {escape(component_id)}/{escape(config_id)}{branch_info}"
         )
+        _emit_plaintext_written_warning(formatter, result)
 
 
 # ── config row-delete ──────────────────────────────────────────────────────────

src/keboola_agent_cli/services/config_service.py

@@ -22,7 +22,7 @@
 from ..sync.code_extraction import normalize_blocks_codes_script
 from ..sync.manifest import Manifest, load_manifest, save_manifest
 from ..sync.naming import sanitize_name
-from ._encryption import collect_secrets, encrypt_secrets_in_config
+from ._encryption import collect_secrets, encrypt_secrets_in_config, find_plaintext_secret_keys
 from .base import BaseService, ClientFactory, sanitize_unexpected_error
 from .workspace_service import find_storage_workspace_for_sandbox_config
 
@@ -622,11 +622,16 @@ def _encrypt_secrets_before_write(
             # Secrets present but the Encryption API call cannot be scoped.
             # Fail closed rather than silently write plaintext.
             if allow_plaintext_fallback:
+                # GHSA-7jrf: name the exact secret key-paths written in PLAINTEXT
+                # (keys only -- `secrets` maps flattened path -> value -- never the
+                # values), consistent with the encryption-failure warning in
+                # `encrypt_secrets_in_config`. No plaintext-write path stays silent.
                 logger.warning(
-                    "Cannot resolve project_id for %s; writing %d secret(s) as "
-                    "plaintext (allow_plaintext_fallback=True)",
+                    "Cannot resolve project_id for %s; --allow-plaintext-on-encrypt-failure "
+                    "is set, so %d secret value(s) are being written in PLAINTEXT: %s.",
                     component_id,
                     len(secrets),
+                    ", ".join(sorted(secrets)) or "(unable to enumerate)",
                 )
                 return configuration
             raise KeboolaApiError(
@@ -785,6 +790,12 @@ def update_config(
         result["project_alias"] = alias
         result["branch_id"] = effective_branch_id
         result["normalizations"] = normalizations
+        # Surface a plaintext-on-encrypt-failure fallback structurally (not just
+        # via the stderr warning) so --json consumers see the leaked key-paths.
+        # find_plaintext_secret_keys returns [] when encryption succeeded.
+        result["plaintext_written"] = (
+            find_plaintext_secret_keys(final_config) if final_config else []
+        )
         return result
 
     def _resolve_configuration(
@@ -1601,6 +1612,8 @@ def create_config_row(
 
         result["project_alias"] = alias
         result["branch_id"] = effective_branch_id
+        # Structurally surface any plaintext-fallback leak (empty when encrypted).
+        result["plaintext_written"] = find_plaintext_secret_keys(row_config) if row_config else []
         return result
 
     # ── config create (one-shot remote create via `config new --push`) ─────────
@@ -1726,6 +1739,10 @@ def create_config(
         # empty -- but we annotate it anyway so JSON consumers can rely on
         # the key being present.
         result["validation_errors"] = validation_errors
+        # Structurally surface any plaintext-fallback leak (empty when encrypted).
+        result["plaintext_written"] = (
+            find_plaintext_secret_keys(encrypted_config) if encrypted_config else []
+        )
         return result
 
     def _validate_config_body(
@@ -1931,6 +1948,10 @@ def update_config_row(
 
         result["project_alias"] = alias
         result["branch_id"] = effective_branch_id
+        # Structurally surface any plaintext-fallback leak (empty when encrypted).
+        result["plaintext_written"] = (
+            find_plaintext_secret_keys(final_config) if final_config else []
+        )
         return result
 
     def _resolve_row_configuration(

src/keboola_agent_cli/services/data_app_service.py

@@ -1359,6 +1359,11 @@ def set_data_app_secrets(
                 "secrets_set": secrets_set,
                 "secrets_unchanged": unchanged,
                 "shadowed_by_runtime": shadowed,
+                # Keys whose ciphertext check failed but were written anyway under
+                # --allow-plaintext-on-encrypt-failure (empty on full encryption).
+                # Surfaces the plaintext leak structurally in --json, not just the
+                # stderr warning above.
+                "plaintext_written": sorted(problems),
                 "config_version_before": old_version,
                 "config_version_after": new_version,
                 "deploy_required": True,

src/keboola_agent_cli/services/variables_service.py

@@ -15,17 +15,46 @@
 
 import copy
 import logging
+from dataclasses import dataclass
 from typing import Any
 
 from ..errors import ConfigError, KeboolaApiError
-from ._encryption import encrypt_secrets_in_config
+from ._encryption import encrypt_secrets_in_config, find_plaintext_secret_keys
 from .base import BaseService
 
 logger = logging.getLogger(__name__)
 
 VARIABLES_COMPONENT_ID = "keboola.variables"
 
 
+@dataclass(frozen=True)
+class _CreatedLinkedVariables:
+    """Result of the auto-create path -- a new ``keboola.variables`` config + row.
+
+    Named fields replace a positional tuple (CONTRIBUTING.md: multi-value returns
+    use a dataclass, never a bare tuple beyond two values).
+    """
+
+    variables_id: str
+    values_id: str
+    values: dict[str, str]
+    plaintext_written: list[str]
+
+
+@dataclass(frozen=True)
+class _UpdatedLinkedVariables:
+    """Result of the update path -- an existing config's default values row.
+
+    ``plaintext_written`` holds the secret key-paths left unencrypted by an
+    allowed plaintext-on-encrypt-failure fallback (``[]`` when encryption
+    succeeded, never the values).
+    """
+
+    values_id: str
+    values: dict[str, str]
+    plaintext_written: list[str]
+
+
 class VariablesService(BaseService):
     """Assign, read, and detach variables on any Keboola config.
 
@@ -144,11 +173,7 @@ def set_variables(
             action = "updated"
 
             if not linked_vars_id:
-                (
-                    linked_vars_id,
-                    linked_values_id,
-                    final_values,
-                ) = self._create_linked_variables(
+                created = self._create_linked_variables(
                     client=client,
                     project_id=project_id,
                     parent_name=parent_name,
@@ -158,9 +183,13 @@ def set_variables(
                     branch_id=branch_id,
                     allow_plaintext_fallback=allow_plaintext_fallback,
                 )
+                linked_vars_id = created.variables_id
+                linked_values_id = created.values_id
+                final_values = created.values
+                plaintext_written = created.plaintext_written
                 action = "created"
             else:
-                linked_values_id, final_values = self._update_linked_variables(
+                updated = self._update_linked_variables(
                     client=client,
                     project_id=project_id,
                     variables_id=linked_vars_id,
@@ -170,6 +199,9 @@ def set_variables(
                     branch_id=branch_id,
                     allow_plaintext_fallback=allow_plaintext_fallback,
                 )
+                linked_values_id = updated.values_id
+                final_values = updated.values
+                plaintext_written = updated.plaintext_written
 
             # Ensure the parent config carries the link. Existing-linked path
             # may no-op; auto-create path always writes.
@@ -196,6 +228,9 @@ def set_variables(
                 "action": action,
                 "values": final_values,
                 "encrypted_keys": encrypted_keys,
+                # Empty unless an allowed plaintext-on-encrypt-failure fallback
+                # left secret key-paths unencrypted in the row that was written.
+                "plaintext_written": plaintext_written,
             }
         finally:
             client.close()
@@ -258,7 +293,7 @@ def _create_linked_variables(
         variables: dict[str, str],
         branch_id: int | None,
         allow_plaintext_fallback: bool,
-    ) -> tuple[str, str, dict[str, str]]:
+    ) -> _CreatedLinkedVariables:
         """Auto-create path: new variables config + default row, parent not yet linked."""
         var_name = (parent_name or parent_config_id) + "-vars"
         schema = [{"name": k, "type": "string"} for k in variables]
@@ -278,6 +313,7 @@ def _create_linked_variables(
             variables=variables,
             allow_plaintext_fallback=allow_plaintext_fallback,
         )
+        plaintext_written = find_plaintext_secret_keys(row_config)
 
         new_row = client.create_config_row(
             component_id=VARIABLES_COMPONENT_ID,
@@ -287,7 +323,12 @@ def _create_linked_variables(
             description="Auto-created default row by kbagent",
             branch_id=branch_id,
         )
-        return variables_id, new_row["id"], dict(variables)
+        return _CreatedLinkedVariables(
+            variables_id=variables_id,
+            values_id=new_row["id"],
+            values=dict(variables),
+            plaintext_written=plaintext_written,
+        )
 
     def _update_linked_variables(
         self,
@@ -300,7 +341,7 @@ def _update_linked_variables(
         replace: bool,
         branch_id: int | None,
         allow_plaintext_fallback: bool,
-    ) -> tuple[str, dict[str, str]]:
+    ) -> _UpdatedLinkedVariables:
         """Update path: parent already linked (or explicit --variables-id). Merge or replace."""
         vars_cfg = client.get_config_detail(
             VARIABLES_COMPONENT_ID, variables_id, branch_id=branch_id
@@ -315,6 +356,7 @@ def _update_linked_variables(
                 variables=variables,
                 allow_plaintext_fallback=allow_plaintext_fallback,
             )
+            plaintext_written = find_plaintext_secret_keys(row_config)
             new_row = client.create_config_row(
                 component_id=VARIABLES_COMPONENT_ID,
                 config_id=variables_id,
@@ -329,7 +371,11 @@ def _update_linked_variables(
                 variables=variables,
                 branch_id=branch_id,
             )
-            return new_row["id"], dict(variables)
+            return _UpdatedLinkedVariables(
+                values_id=new_row["id"],
+                values=dict(variables),
+                plaintext_written=plaintext_written,
+            )
 
         existing_values = target_row.get("configuration", {}).get("values", [])
         existing_dict = {v["name"]: v["value"] for v in existing_values}
@@ -348,6 +394,7 @@ def _update_linked_variables(
             variables=final_values,
             allow_plaintext_fallback=allow_plaintext_fallback,
         )
+        plaintext_written = find_plaintext_secret_keys(row_config)
 
         client.update_config_row(
             component_id=VARIABLES_COMPONENT_ID,
@@ -364,7 +411,11 @@ def _update_linked_variables(
             variables=final_values,
             branch_id=branch_id,
         )
-        return target_row["id"], final_values
+        return _UpdatedLinkedVariables(
+            values_id=target_row["id"],
+            values=final_values,
+            plaintext_written=plaintext_written,
+        )
 
     @staticmethod
     def _build_encrypted_row_configuration(
@@ -380,6 +431,10 @@ def _build_encrypted_row_configuration(
         transformation runner both key off it). :func:`encrypt_secrets_in_config`
         recognizes the ``{name, value}`` list shape directly, so no pre-flatten
         dance is needed.
+
+        Returns the encrypted row config. Callers surface any plaintext-fallback
+        leak by passing the returned config to :func:`find_plaintext_secret_keys`
+        (key-paths only, never the values).
         """
         row_config: dict[str, Any] = {
             "values": [{"name": k, "value": v} for k, v in variables.items()],