Fix GCS bucket-wide permissions vulnerability using Signed URLs

kinetic/backend/execution.py

@@ -77,6 +77,7 @@ class JobContext:
   image_uri: Optional[str] = None
   payload_sha256: Optional[str] = None
   context_sha256: Optional[str] = None
+  signed_urls: Optional[dict[str, str]] = None
 
   def __post_init__(self):
     self.bucket_name = build_bucket_name(self.project, self.cluster_name)
@@ -203,6 +204,7 @@ def submit_job(self, ctx: JobContext) -> Any:
       debug=ctx.debug,
       payload_sha256=ctx.payload_sha256,
       context_sha256=ctx.context_sha256,
+      signed_urls=ctx.signed_urls,
     )
 
   def wait_for_job(self, job: Any, ctx: JobContext) -> None:
@@ -261,6 +263,7 @@ def submit_job(self, ctx: JobContext) -> Any:
       debug=ctx.debug,
       payload_sha256=ctx.payload_sha256,
       context_sha256=ctx.context_sha256,
+      signed_urls=ctx.signed_urls,
     )
 
   def wait_for_job(self, job: Any, ctx: JobContext) -> None:
@@ -575,6 +578,27 @@ def _upload_artifacts(ctx: JobContext) -> bool:
     project=ctx.project,
     requirements_content=requirements_content,
   )
+
+  # Generate signed URLs for the job (with fallback to legacy GCS access)
+  signer_email = f"kn-{ctx.cluster_name}-signer@{ctx.project}.iam.gserviceaccount.com"
+  try:
+    ctx.signed_urls = storage.generate_job_signed_urls(
+        bucket_name=ctx.bucket_name,
+        job_id=ctx.job_id,
+        project=ctx.project,
+        signer_sa_email=signer_email,
+        has_requirements=has_requirements,
+        debug=ctx.debug,
+    )
+    logging.info("Successfully generated Signed URLs for secure job isolation.")
+  except Exception as e:
+    logging.warning(
+        "Failed to generate Signed URLs (likely due to missing 'signer' GSA or lack of "
+        "impersonation permissions). Falling back to legacy GCS access. "
+        "Error: %s", e
+    )
+    ctx.signed_urls = None
+
   return has_requirements
 
 

kinetic/backend/gke_client.py

@@ -30,6 +30,7 @@ def submit_k8s_job(
   debug=False,
   payload_sha256=None,
   context_sha256=None,
+  signed_urls=None,
 ):
   """Submit a Kubernetes Job to GKE cluster.
 
@@ -45,6 +46,7 @@ def submit_k8s_job(
           install (prebuilt image mode).
       payload_sha256: Optional SHA-256 hash of payload.pkl for verification.
       context_sha256: Optional SHA-256 hash of context.zip for verification.
+      signed_urls: Optional dict of GCS Signed URLs for job isolation.
 
   Returns:
       kubernetes.client.V1Job object
@@ -66,6 +68,7 @@ def submit_k8s_job(
     debug=debug,
     payload_sha256=payload_sha256,
     context_sha256=context_sha256,
+    signed_urls=signed_urls,
   )
 
   # Submit job
@@ -322,6 +325,7 @@ def _create_job_spec(
   debug=False,
   payload_sha256=None,
   context_sha256=None,
+  signed_urls=None,
 ):
   """Create Kubernetes Job specification.
 
@@ -336,6 +340,7 @@ def _create_job_spec(
           install (prebuilt image mode).
       payload_sha256: Optional SHA-256 hash of payload.pkl
       context_sha256: Optional SHA-256 hash of context.zip
+      signed_urls: Optional dict of GCS Signed URLs for job isolation.
 
   Returns:
       V1Job object ready for creation
@@ -364,16 +369,30 @@ def _create_job_spec(
     )
 
   # Container arguments: context, payload, result, [requirements]
-  container_args = [
-    "--context-gcs",
-    f"gs://{bucket_name}/{job_id}/context.zip",
-    "--payload-gcs",
-    f"gs://{bucket_name}/{job_id}/payload.pkl",
-    "--result-gcs",
-    f"gs://{bucket_name}/{job_id}/result.pkl",
-  ]
-  if requirements_uri:
-    container_args.extend(["--requirements-gcs", requirements_uri])
+  if signed_urls:
+    container_args = [
+      "--context-gcs", signed_urls["context_download"],
+      "--payload-gcs", signed_urls["payload_download"],
+      "--result-gcs", signed_urls["result_upload"],
+    ]
+    if "requirements_download" in signed_urls:
+      container_args.extend(["--requirements-gcs", signed_urls["requirements_download"]])
+    if "debug_ready_upload" in signed_urls:
+      container_args.extend(["--debug-ready-url", signed_urls["debug_ready_upload"]])
+    if "leader_ready_upload" in signed_urls:
+      container_args.extend(["--leader-ready-upload-url", signed_urls["leader_ready_upload"]])
+    if "leader_ready_download" in signed_urls:
+      container_args.extend(["--leader-ready-download-url", signed_urls["leader_ready_download"]])
+  else:
+    # Legacy GCS URIs
+    container_args = [
+      "--context-gcs", f"gs://{bucket_name}/{job_id}/context.zip",
+      "--payload-gcs", f"gs://{bucket_name}/{job_id}/payload.pkl",
+      "--result-gcs", f"gs://{bucket_name}/{job_id}/result.pkl",
+    ]
+    if requirements_uri:
+      container_args.extend(["--requirements-gcs", requirements_uri])
+
   if payload_sha256:
     container_args.extend(["--payload-sha256", payload_sha256])
   if context_sha256:

kinetic/backend/pathways_client.py

@@ -73,6 +73,7 @@ def submit_pathways_job(
   debug=False,
   payload_sha256=None,
   context_sha256=None,
+  signed_urls=None,
 ):
   """Submit a LeaderWorkerSet to GKE cluster.
 
@@ -88,6 +89,7 @@ def submit_pathways_job(
           install (prebuilt image mode).
       payload_sha256: Optional SHA-256 hash of payload.pkl for verification.
       context_sha256: Optional SHA-256 hash of context.zip for verification.
+      signed_urls: Optional dict of GCS Signed URLs for job isolation.
 
   Returns:
       dict: The created LeaderWorkerSet object
@@ -120,6 +122,7 @@ def submit_pathways_job(
     debug=debug,
     payload_sha256=payload_sha256,
     context_sha256=context_sha256,
+    signed_urls=signed_urls,
   )
 
   custom_api = _custom_api()
@@ -440,6 +443,7 @@ def _create_lws_spec(
   debug=False,
   payload_sha256=None,
   context_sha256=None,
+  signed_urls=None,
 ):
   """Create a LeaderWorkerSet manifest."""
 
@@ -466,16 +470,31 @@ def _create_lws_spec(
       entry["value"] = t["value"]
     tolerations.append(entry)
 
-  container_args = [
-    "--context-gcs",
-    f"gs://{bucket_name}/{job_id}/context.zip",
-    "--payload-gcs",
-    f"gs://{bucket_name}/{job_id}/payload.pkl",
-    "--result-gcs",
-    f"gs://{bucket_name}/{job_id}/result.pkl",
-  ]
-  if requirements_uri:
-    container_args.extend(["--requirements-gcs", requirements_uri])
+  # Container arguments: context, payload, result, [requirements]
+  if signed_urls:
+    container_args = [
+      "--context-gcs", signed_urls["context_download"],
+      "--payload-gcs", signed_urls["payload_download"],
+      "--result-gcs", signed_urls["result_upload"],
+    ]
+    if "requirements_download" in signed_urls:
+      container_args.extend(["--requirements-gcs", signed_urls["requirements_download"]])
+    if "debug_ready_upload" in signed_urls:
+      container_args.extend(["--debug-ready-url", signed_urls["debug_ready_upload"]])
+    if "leader_ready_upload" in signed_urls:
+      container_args.extend(["--leader-ready-upload-url", signed_urls["leader_ready_upload"]])
+    if "leader_ready_download" in signed_urls:
+      container_args.extend(["--leader-ready-download-url", signed_urls["leader_ready_download"]])
+  else:
+    # Legacy GCS URIs
+    container_args = [
+      "--context-gcs", f"gs://{bucket_name}/{job_id}/context.zip",
+      "--payload-gcs", f"gs://{bucket_name}/{job_id}/payload.pkl",
+      "--result-gcs", f"gs://{bucket_name}/{job_id}/result.pkl",
+    ]
+    if requirements_uri:
+      container_args.extend(["--requirements-gcs", requirements_uri])
+
   if payload_sha256:
     container_args.extend(["--payload-sha256", payload_sha256])
   if context_sha256: