advancedsecurityGoogle styleNetflix styleUber styleStripe style

Google SRE Python code review: How would you safely run kubectl and cloud CLI commands from a Python automation script without creating shell injection, hung-process, or unreadable-error risks?

⚡

Quick Answer

Use subprocess with an argument list, shell=False, explicit timeouts, captured output, check=True, and structured error handling. Avoid string-concatenated shell commands, validate user-controlled arguments, and return a typed result that preserves stdout, stderr, exit code, and duration.

Detailed Answer

Think of an automation script like a dispatcher sending technicians to data centers. If the dispatcher writes a vague instruction on a sticky note, the technician may interpret it differently, take too long, or do something dangerous. A production Python script calling kubectl, aws, helm, or terraform needs precise instructions, a deadline, and a clean incident report when something fails.

In Python, the safe default is subprocess.run with a list of arguments and shell=False. The Python documentation recommends passing arguments as a sequence because Python can handle escaping and quoting without involving a shell. That matters because shell=True turns one string into shell syntax, so characters such as semicolons, pipes, dollar expansion, and backticks become meaningful. In interview terms, this is not just style; it is a security boundary.

A strong implementation wraps subprocess calls in one function. The wrapper sets timeout, text=True, capture_output=True, check=True, cwd only when needed, and a minimal env when credentials or PATH should be controlled. It catches subprocess.TimeoutExpired and subprocess.CalledProcessError separately because a timeout is an operational failure while a non-zero exit is a command-level failure. The wrapper should log the command safely, redacting tokens and kubeconfigs, and it should never dump secrets from stderr into chat tools or tickets.

At scale, the tricky behavior is that automation failures are often partial. A kubectl apply may fail on one manifest after mutating another. An aws command may succeed but return paginated or eventually consistent data. A hung process can hold a deployment lock and block the next run. That is why serious automation includes idempotency, dry-run mode, bounded retries around known transient errors, and audit output that lets an on-call engineer prove exactly what was attempted.

The non-obvious gotcha is that avoiding shell=True does not automatically make input safe. If a user can pass --kubeconfig, --context, --namespace, or a file path, they can still redirect the automation into the wrong cluster or sensitive files. Senior candidates explain both layers: command invocation safety and business-rule validation of the arguments being passed.

from __future__ import annotations # Enable modern typing syntax on supported Python versions. import subprocess # Use the standard library process API instead of os.system. from dataclasses import dataclass # Return structured command results to callers. from time import monotonic # Measure elapsed time without wall-clock jumps. @dataclass(frozen=True) # Make command results immutable after creation. class CmdResult: # Store the operational evidence from a command run. argv: list[str] # Keep the exact argv list without shell interpolation. stdout: str # Capture standard output for parsing or reporting. stderr: str # Capture standard error for diagnostics. seconds: float # Record duration for slow-command alerts. ALLOWED_NAMESPACES = {"payments", "orders", "platform"} # Restrict blast radius. def run_checked(argv: list[str], timeout: int = 30) -> CmdResult: # Central safe subprocess wrapper. if not argv or argv[0] not in {"kubectl", "aws", "helm"}: # Allow only approved CLIs. raise ValueError("unsupported command") # Fail closed for unknown executables. start = monotonic() # Start a monotonic timer before launching the process. completed = subprocess.run( # Run the command without invoking a shell. argv, # Pass a list so arguments are not interpreted as shell syntax. shell=False, # Keep shell metacharacters inert. text=True, # Decode stdout and stderr as strings. capture_output=True, # Capture both streams for structured error handling. timeout=timeout, # Kill commands that hang due to network or API issues. check=True, # Raise CalledProcessError on non-zero exit codes. ) # subprocess.run waits for completion or timeout. return CmdResult(argv, completed.stdout, completed.stderr, monotonic() - start) # Return evidence. def restart_deployment(namespace: str, deployment: str) -> CmdResult: # Safe operational action. if namespace not in ALLOWED_NAMESPACES: # Validate business-level scope. raise ValueError(f"namespace {namespace!r} is not approved") # Prevent wrong-cluster style mistakes. if not deployment.replace("-", "").isalnum(): # Keep resource names simple and expected. raise ValueError("deployment name contains unexpected characters") # Reject suspicious input. return run_checked(["kubectl", "rollout", "restart", f"deployment/{deployment}", "-n", namespace]) # Execute safely.

Google SRE Python code review: How would you safely run kubectl and cloud CLI commands from a Python automation script without creating shell injection, hung-process, or unreadable-error risks?

More Python DevOps Automation questions