Everything for Python DevOps Automation in one place — pick a section below. 21 reviewed items across 2 content types.
Quick Answer
The most common causes are credential resolution differing between environments, since boto3's credential chain finds your local AWS profile but not the Lambda execution role or CI OIDC token, connection pooling and timeout defaults that behave differently under a short-lived Lambda invocation versus a long-running laptop process, and missing or mismatched dependency versions between your local virtualenv and the deployed runtime. Diagnose by explicitly logging which credential source and SDK/library versions are actually resolved at runtime, rather than assuming the deployed environment matches your local one.
Detailed Answer
This is like a recipe that works perfectly in your own kitchen with your own knives, oven, and pantry, but fails when a caterer tries to run it in an unfamiliar commercial kitchen with different equipment and a stricter time limit — the recipe itself didn't change, but every assumption about what's available and how much time you have quietly shifted.
Python DevOps automation scripts are deceptively easy to write correctly for a developer's own machine, because a laptop has a stable, long-lived environment: cached AWS credentials, a persistent Python virtualenv, unlimited wall-clock time, and reliable network access. Production execution environments, Lambda, CI runners, Kubernetes CronJobs, break every one of those assumptions in different ways, which is exactly why "works on my machine" is such a common and specific failure category for this kind of script, distinct from a plain logic bug.
Credential resolution is the most frequent culprit: boto3's default credential chain checks environment variables, then shared credential files, then an EC2, ECS, or Lambda execution role, in a specific order — a script that works locally because it silently picked up your personal AWS profile will behave completely differently in Lambda, where it must instead pick up the function's IAM execution role, and any hardcoded profile name or region assumption breaks immediately. Connection behavior is the second most common issue: HTTP clients and boto3 clients that create fresh connections per invocation, typical in Lambda since the runtime environment may be reused or may be entirely fresh per invocation unpredictably, behave very differently from a long-running local process that reuses one connection pool for its entire lifetime, and default timeouts tuned for a patient human at a laptop are often too generous for a Lambda function with a hard execution time limit.
In production, the fix is to make environment assumptions explicit rather than implicit: log the resolved credential source, since calling get_credentials().method on a boto3 session reveals whether credentials came from an environment variable, a profile, or an IAM role, pin dependency versions exactly, since an unpinned requirements file can resolve differently between a developer's cached pip environment and a fresh CI or Lambda build, and set explicit, conservative timeouts rather than relying on SDK defaults that assume an interactive session. Monitoring should track cold-start-specific failures separately from warm-execution failures, since a script that only fails on the first invocation after a deployment often points at initialization-time issues, such as module-level client construction or credential resolution, rather than logic bugs.
The non-obvious gotcha is that Lambda execution environments are reused across invocations for performance, meaning any client object created outside the handler function, at module import time, persists across multiple invocations, including its cached connections and any state. This is usually a good optimization, but it means a client configured with credentials that were valid at cold start can silently hold onto now-stale session tokens if a script assumes credentials never need refreshing during a long-lived Lambda container's life, causing intermittent authentication failures that only appear after the container has been warm for a while, which is very hard to reproduce locally since a laptop script never experiences this specific reuse pattern.
Code Example
import boto3
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("checkout-worker-automation")
# Create the boto3 client at module load time so Lambda can reuse it across warm invocations
ec2 = boto3.client("ec2", region_name="us-east-1")
def log_credential_source():
# Reveal exactly which credential chain step resolved credentials at runtime
session = boto3.session.Session()
creds = session.get_credentials()
logger.info("Resolved credentials via method: %s", creds.method if creds else "NONE")
def lambda_handler(event, context):
# Log the credential source on every cold start to catch environment mismatches early
log_credential_source()
try:
# Explicit, conservative timeout instead of relying on SDK defaults tuned for interactive use
response = ec2.describe_instances(
Filters=[{"Name": "instance-state-name", "Values": ["running"]}]
)
count = sum(len(r["Instances"]) for r in response["Reservations"])
logger.info("Found %d running instances", count)
return {"statusCode": 200, "body": f"{count} running instances"}
except Exception as exc:
# Log the exception type explicitly so credential errors are distinguishable from network errors
logger.error("Automation failed: %s: %s", type(exc).__name__, exc)
raiseInterview Tip
A junior engineer typically debugs this by adding more print statements around the failing API call, but for a senior role, the interviewer is actually looking for you to recognize this as an environment-assumption problem, not a logic bug — the script's credential resolution, connection lifecycle, and dependency versions can all differ silently between a laptop and a deployed runtime. Strong answers mention logging the resolved credential method explicitly, understanding that Lambda reuses execution environments across invocations so module-level clients persist and can hold stale tokens, and pinning dependency versions so a CI or Lambda build can't silently resolve a different library version than what was tested locally. This is the difference between someone who's only run scripts locally and someone who's actually operated automation in production.
◈ Architecture Diagram
Laptop: ┌────────┐ profile ┌────────┐
│ Script │─────────▶│ AWS │
└────────┘ └────────┘
Lambda: ┌────────┐ exec role ┌────────┐
│ Script │──────────▶│ AWS │
└────────┘ (reused container)
✗ stale token risk💬 Comments
Quick Answer
The most common mistake is treating idempotent as meaning 'doesn't error on re-run' rather than 'produces the same end state regardless of how many times it runs' — a script that creates a resource without first checking whether an equivalent one already exists, or that doesn't use a client-side idempotency token, will happily create duplicates on every retry. Fix it by checking current state before acting, a describe-then-create pattern, and using idempotency tokens where the API supports them, so retries converge rather than duplicate.
Detailed Answer
This is like a form that says "click submit to place your order" with no confirmation of whether the order already went through — if the page seems to hang and you click submit again just to be safe, you might end up with two identical orders showing up at your door, because the system had no way to recognize "this is the same order being resubmitted" versus "this is a genuinely new order."
True idempotency, an operation that, no matter how many times you perform it, results in the same final state, requires deliberate design; it does not happen automatically just because a script "usually doesn't error." Automation scripts are retried constantly in production — a Lambda timing out mid-execution, a CI job getting killed by a runner restart, a network blip during an API call whose response never arrives even though the request succeeded server-side, and any of these can trigger an at-least-once retry, meaning the same logical operation may genuinely execute more than once against the same target.
The fix has two complementary layers. First, a describe-then-act pattern: before creating any resource, the script checks whether an equivalent resource already exists, by a stable identifier like a name tag or a deterministic naming convention derived from the input, not a randomly generated ID that changes every run, and only proceeds with creation if it's genuinely missing. Second, where the underlying API supports it, use client-side idempotency tokens — many AWS APIs, like EC2's RunInstances ClientToken parameter, let you pass a unique-per-logical-operation token that the API itself uses to deduplicate: if you retry with the same token, AWS recognizes it as the same request and returns the original result instead of creating a second resource, even if your first request's response was lost due to a network failure and you have no local memory of whether it succeeded.
In production, the describe-then-act pattern needs to handle a subtle race: if two instances of the automation run concurrently, which happens more often than expected, an overlapping cron schedule or a retried Lambda invocation racing the original before it's confirmed dead, both can pass the "doesn't exist yet" check before either creates the resource, resulting in a duplicate anyway. Mature automation adds a distributed lock, a DynamoDB conditional write, a Redis-style atomic set, or a cloud-native equivalent, around the describe-then-act sequence specifically to close this race window, not just the idempotency token alone.
The non-obvious gotcha is that idempotency tokens typically have a limited validity window, EC2's ClientToken, for example, is only guaranteed to deduplicate requests within several hours, so a retry that happens much later, say a manually re-triggered pipeline run days after the original failure using a script that generates a fresh token each time rather than persisting the token used for the original attempt, will not be deduplicated by the API at all, silently reintroducing the duplicate-resource problem the token was supposed to prevent. The token only protects against retries close in time to the original attempt, not against a human manually re-running the same logical job long after.
Code Example
import boto3
import hashlib
ec2 = boto3.client("ec2", region_name="us-east-1")
def deterministic_token(logical_operation_id: str) -> str:
# Derive a stable client token from a logical operation ID, not a random value per run
return hashlib.sha256(logical_operation_id.encode()).hexdigest()[:32]
def ensure_instance_exists(name: str, logical_operation_id: str):
# Describe first: check whether an instance with this name tag already exists
existing = ec2.describe_instances(
Filters=[{"Name": "tag:Name", "Values": [name]},
{"Name": "instance-state-name", "Values": ["pending", "running"]}]
)
reservations = existing.get("Reservations", [])
if reservations:
# Resource already exists: idempotent no-op, return the existing instance ID
return reservations[0]["Instances"][0]["InstanceId"]
# Create only if missing, using a stable ClientToken so a retry within the API's
# dedup window doesn't create a second instance even if the first response was lost
response = ec2.run_instances(
ImageId="ami-0abcd1234efgh5678",
InstanceType="t3.small",
MinCount=1,
MaxCount=1,
ClientToken=deterministic_token(logical_operation_id),
TagSpecifications=[{
"ResourceType": "instance",
"Tags": [{"Key": "Name", "Value": name}]
}]
)
return response["Instances"][0]["InstanceId"]Interview Tip
A junior engineer typically says a script is idempotent because "re-running it doesn't throw an error," but for a senior role, the interviewer is actually looking for a precise definition — the end state must converge to the same result regardless of how many times the operation runs, including at-least-once retries triggered by timeouts, killed processes, or lost responses to requests that actually succeeded server-side. Strong answers describe the describe-then-act pattern combined with API-level idempotency tokens as complementary, not redundant, mention that concurrent retries can still race past a describe-then-act check without a distributed lock, and flag that idempotency tokens have a limited validity window, so a retry that happens much later than the original attempt won't be deduplicated by the API at all.
◈ Architecture Diagram
┌────────┐ describe ┌─────────┐ exists? ┌────────┐
│ Script │─────────▶│ Current │────────▶│ skip │
└────────┘ │ State │ no └────────┘
└─────────┘ │
↓
create w/ token💬 Comments
Quick Answer
Instrument every automation run to emit a structured heartbeat with expected-vs-actual output counts, items scanned, items acted on, items skipped, duration, and alert on the absence of a heartbeat as aggressively as on an explicit failure, because a cron job that stops running entirely, or one that runs but silently processes zero items due to a permissions change, produces no error at all, just a gap in outcomes that nobody notices without dedicated instrumentation.
Detailed Answer
This is like a security guard who's supposed to check every door in a building every night and radio in a count of doors checked — if the guard simply stops making rounds, or starts checking zero doors because someone locked them out of half the building, there's no alarm bell ringing anywhere; the absence of the guard's report is itself the only signal, and it's silent unless someone is specifically listening for a missing check-in.
DevOps automation scripts are especially prone to a specific and dangerous failure mode: silent success. A script that exits with code zero and prints "scan complete" is indistinguishable, from a basic cron or CI monitoring perspective, from a script that genuinely found and reported zero issues versus one whose IAM permissions were quietly revoked and now silently returns empty results for every query it makes. Standard process-level monitoring, did the job run, did it exit non-zero, is necessary but not sufficient, because it can't distinguish "nothing to report" from "couldn't check anything."
The fix is to make every automation script emit a structured outcome record on every run, not just logs, but a small piece of machine-readable telemetry, items scanned, items flagged, items remediated, duration, exit status, pushed to a metrics system or a dead-man's-switch style heartbeat service. Critically, the script should also track its own expected baseline: if a nightly untagged-instance scanner has historically found five to fifteen flagged instances per run for months, and one night it reports zero, that's worth investigating even though "zero findings" is technically a valid, non-error result, the same way a fire alarm system that suddenly stops reporting any smoke at all, ever, might mean the building is clean or might mean the sensor died.
In production, mature platforms wire this into a dead-man's-switch pattern: an external monitoring service expects a heartbeat ping from each scheduled automation job within its expected run window, and pages if the ping doesn't arrive — this catches the case where the job doesn't run at all, a cron misconfiguration, a CI schedule that got silently disabled, a Kubernetes CronJob suspended during a cluster migration and never resumed, which produces literally zero signal in application-level logs since the application never even started. Combine this with anomaly detection on the outcome counts themselves, items scanned dropping to zero, or items flagged deviating wildly from the historical range, to catch the cases where the job runs but does the wrong thing.
The non-obvious gotcha is that adding more logging alone doesn't solve this problem — a script that logs "scan complete, zero issues found" in exactly the same format whether that's because the environment is genuinely clean or because a credential expired and every API call returned an empty result due to a caught, swallowed exception, provides zero additional signal over the original silent failure. The instrumentation has to explicitly distinguish and separately track "zero because nothing was found" from "zero because we couldn't check," for example by catching and specifically counting AccessDenied or throttling exceptions rather than letting a generic try/except silently produce the same empty result as a clean scan, for this kind of monitoring to actually catch the failure class it's meant to catch.
Code Example
import time
import json
import logging
import requests
from botocore.exceptions import ClientError
logger = logging.getLogger("payments-api-cleanup-job")
DEADMANS_SWITCH_URL = "https://heartbeat.internal/ping/payments-api-cleanup"
def run_cleanup_job():
start = time.monotonic()
scanned = 0
flagged = 0
errors = 0 # Tracks API errors distinctly from "checked and found nothing"
try:
# ... scanning logic would populate scanned/flagged as it iterates resources ...
scanned, flagged = perform_scan()
except ClientError as exc:
# Count permission/throttling failures separately so they don't look like a clean scan
errors += 1
logger.error("Scan aborted due to AWS API error: %s", exc.response["Error"]["Code"])
duration = time.monotonic() - start
# Emit a structured outcome record distinguishing "found nothing" from "couldn't check"
outcome = {
"job": "payments-api-cleanup",
"items_scanned": scanned,
"items_flagged": flagged,
"errors": errors,
"duration_seconds": round(duration, 2),
}
logger.info(json.dumps(outcome))
# Ping the dead-man's-switch heartbeat only when the job actually completed without errors
if errors == 0:
requests.post(DEADMANS_SWITCH_URL, json=outcome, timeout=5)
def perform_scan():
# Placeholder for real scanning logic; returns (items_scanned, items_flagged)
return 42, 3Interview Tip
A junior engineer typically says they'd add logging and alert on exceptions, but for a senior or SRE role, the interviewer is actually looking for you to recognize silent success as a distinct and more dangerous failure class than a crash — a script that exits zero while quietly doing nothing produces no signal for standard job monitoring to catch. Strong answers describe a dead-man's-switch heartbeat pattern to catch jobs that stop running entirely, anomaly detection on outcome counts to catch jobs that run but process an unexpectedly low number of items, and explicitly distinguishing "found nothing" from "couldn't check anything" by counting permission and throttling errors separately rather than letting them collapse into the same empty result as a clean scan.
◈ Architecture Diagram
┌────────┐ run ┌──────────┐ heartbeat ┌──────────┐
│CronJob │─────▶│ Outcome │──────────▶│Dead-man's│
└────────┘ │ metrics │ │ switch │
└────┌─────┘ └────┌─────┘
│ anomaly? │ missing?
↓ ↓
┌───────┐ ┌───────┐
│ Alert │ │ Page │
└───────┘ └───────┘💬 Comments
Quick Answer
Structure it as a pipeline with mandatory phases, discovery, classification against explicit policy, dry-run plan generation, human or automated approval gated by risk level, bounded and rate-limited execution, and verification, with every phase producing an append-only audit record. Treat the automation's own credentials as the highest-value target in the system, scoping them as narrowly as each specific action requires rather than granting broad standing access across every team's infrastructure.
Detailed Answer
This is like giving one person a master key that opens every door in a hospital so they can respond quickly to any room needing attention — enormously convenient for getting things done fast, and also exactly the single point of failure an attacker or a bug would most want to compromise, because one mistake or one theft compromises every room in the building at once instead of just one.
A remediation platform is different from a one-off script precisely because it's designed to act broadly and repeatedly across many teams' infrastructure, which means its own reliability and security properties become organization-wide risk, not team-local risk. This is why mature platforms treat every destructive action as requiring the same rigor as a financial transaction: a clear record of what was proposed, who or what approved it, what actually executed, and what the verified outcome was, not because engineers don't trust each other, but because at organizational scale, "trust me, it worked" doesn't scale and doesn't survive an incident post-mortem.
The pipeline design separates concerns explicitly. Discovery gathers current state read-only. Classification applies declarative policy, not ad-hoc if-statements scattered through the code, to decide what qualifies for remediation, making the policy itself reviewable and testable independent of the execution engine. Plan generation produces a diff, exactly what would change, before anything happens, in both human-readable and machine-readable form. Approval gates scale the rigor to the risk: a low-risk, easily-reversible action, such as disabling an unused API key, might auto-approve after a dry-run review window with no objections, while a high-risk, hard-to-reverse action, such as deleting a database or revoking a production access role, requires an explicit human approval tied to a specific reviewer identity. Execution runs in small batches with rate limiting, per-target error isolation so one failing target doesn't block or corrupt the rest of the batch, and immediate verification that the intended state was actually achieved, not just that the API call didn't error.
In production, the platform's own credentials become the highest-priority thing to defend, since compromising the remediation platform is strictly more valuable to an attacker than compromising any single team's resources — this argues for short-lived, action-scoped credentials generated per remediation run rather than a standing broad-access service account, and for running the execution phase in an isolated environment separate from where policy and approval logic live, so a bug or compromise in the approval UI can't directly trigger execution. What to monitor: approval-to-execution latency, since a bottleneck here means teams start routing around the platform with manual fixes, defeating its purpose, false-positive rate in classification, since teams losing trust in flagged findings stop reviewing them carefully, and the platform's own audit log completeness, since a gap in the audit trail during an actual incident is often worse than the incident itself for a security post-mortem.
The non-obvious gotcha is that centralizing remediation this way creates organizational pressure to expand its scope over time — once a platform can safely disable unused IAM roles, someone will ask it to also clean up S3 buckets, then rotate secrets, then terminate instances, and each addition, taken alone, seems reasonable, but the platform's aggregate blast radius grows quietly until it has more standing destructive capability across the entire company than any single team's on-call engineer. Senior engineers push back on scope creep by requiring each new remediation type to justify its own approval tier and audit requirements independently, rather than inheriting the platform's existing trust level by default.
Code Example
from dataclasses import dataclass
from enum import Enum
import json
import time
class RiskTier(Enum):
LOW = "low" # Reversible, narrow blast radius (e.g. disable unused API key)
HIGH = "high" # Irreversible or broad blast radius (e.g. delete database, revoke prod role)
@dataclass(frozen=True)
class RemediationAction:
target: str # Specific resource being acted on, e.g. "iam-role:legacy-ci-deploy"
action: str # Planned operation, e.g. "disable" or "delete"
risk: RiskTier # Determines required approval rigor
reason: str # Policy justification for audit trail
def requires_human_approval(action: RemediationAction) -> bool:
# Only high-risk actions require an explicit named human approver
return action.risk is RiskTier.HIGH
def write_audit(event: dict) -> None:
# Append-only structured audit record for every phase of the pipeline
envelope = {"ts": time.time(), "platform": "remediation-engine", **event}
print(json.dumps(envelope, sort_keys=True))
def execute_action(action: RemediationAction, approver: str | None) -> bool:
if requires_human_approval(action) and not approver:
write_audit({"phase": "blocked", "target": action.target, "reason": "missing approval"})
return False
write_audit({
"phase": "execute", "target": action.target, "action": action.action,
"risk": action.risk.value, "approver": approver or "auto",
})
# Real execution would call the scoped, short-lived-credentialed client here.
return TrueInterview Tip
A junior engineer typically focuses on making the remediation logic itself correct, but for a senior or architect role, the interviewer is actually looking for you to treat the platform's own existence as the biggest new risk it introduces — its credentials are the highest-value target in the organization, and its blast radius grows quietly every time a new remediation type gets bolted on without its own independent approval tier. Strong answers describe risk-tiered approval, auto-approve low-risk, require named human approval for high-risk and irreversible actions, short-lived action-scoped credentials instead of a standing broad service account, isolating the execution environment from the approval UI, and explicitly pushing back on scope creep as new teams ask the platform to take on more destructive capabilities over time.
◈ Architecture Diagram
┌────────┐ ┌──────────┐ ┌────────┐ ┌────────┐ ┌────────┐
│Discover│▶│ Classify │▶│Dry Run │▶│Approve │▶│Execute │
└────────┘ └──────────┘ └────────┘ └───┌────┘ └───┌────┘
low│high │
auto│human ┌────↓───┐
└───────▶│ Verify │
└────────┘💬 Comments
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.
Code Example
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.Interview Tip
A junior engineer typically answers that they would use subprocess instead of os.system, but for a senior role, the interviewer is actually looking for command boundary discipline: shell=False, argv lists, timeouts, check=True, stderr capture, secret redaction, and validation of namespaces, contexts, file paths, and resource names. The strongest answer mentions that subprocess safety and operational safety are separate layers. Even a perfectly escaped command can still damage production if the script lets someone choose an arbitrary kubeconfig or AWS profile.
◈ Architecture Diagram
┌────────┐ argv ┌────────┐ API ┌────────┐
│ Python │────────▶│ kubectl│───────▶│ K8s API│
└────────┘ └────────┘ └────────┘
│ timeout │ stderr
▼ ▼
┌────────┐ ┌────────┐
│ Result │◀───────│ Errors │
└────────┘ └────────┘💬 Comments
Quick Answer
Use STS AssumeRole per account, create regional clients from temporary credentials, consume paginator APIs instead of single calls, configure botocore retries, and process accounts with bounded concurrency. Record per-account success, throttles, access-denied errors, and skipped regions so the inventory is trustworthy even when some accounts fail.
Detailed Answer
Imagine asking 200 branch offices to send you a list of every laptop they own. If you call only the first office, stop after the first page, or hide which offices did not answer, the final spreadsheet looks clean but lies. AWS inventory automation has the same failure mode: the script can produce a file while silently missing accounts, regions, or pages.
The core boto3 pattern is explicit. First call STS AssumeRole into the target account using a role created for inventory. Then build service clients with those temporary credentials and the target region. For list-style APIs, use paginators because many AWS APIs return only a limited page and a continuation token. The boto3 retry guide documents standard and adaptive retry modes, and production collectors should set retry mode and max attempts intentionally rather than relying on unknown environment defaults.
Concurrency must be bounded. A script that fans out 200 accounts times 25 regions times 10 services can create tens of thousands of API calls. Unbounded ThreadPoolExecutor usage will cause throttling, noisy CloudTrail, and sometimes organization-wide guardrail alerts. A mature collector uses a small worker pool, per-service backoff, and clear retry classification: throttling and transient network errors are retried, AccessDenied is recorded as an authorization gap, and validation errors are treated as code or configuration defects.
The output format matters as much as the API calls. Each account-region-service tuple should report status, item count, elapsed time, and error class. That lets platform teams distinguish no resources from no permission. For auditability, include the assumed role ARN, collection timestamp, script version, and schema version in the artifact. Large companies ask this because inventory data drives security posture, cost cleanup, and incident response.
The non-obvious gotcha is eventual consistency. A newly created role, instance, tag, or region opt-in may not appear immediately. Senior candidates mention rerunning collectors, comparing deltas, and designing consumers to tolerate late-arriving data instead of treating one scan as absolute truth.
Code Example
from __future__ import annotations # Use modern type annotations.
from concurrent.futures import ThreadPoolExecutor, as_completed # Bound account fanout.
import boto3 # AWS SDK for Python.
from botocore.config import Config # Configure retries and timeouts explicitly.
from botocore.exceptions import ClientError # Classify AWS API failures.
AWS_CONFIG = Config( # Shared botocore behavior for all clients.
retries={"mode": "standard", "max_attempts": 6}, # Retry transient AWS errors predictably.
connect_timeout=3, # Fail fast when network paths are broken.
read_timeout=20, # Bound slow API responses.
) # Config is passed into each boto3 client.
def assume_inventory_role(account_id: str) -> dict: # Get temporary account credentials.
sts = boto3.client("sts", config=AWS_CONFIG) # Create an STS client in the caller account.
resp = sts.assume_role( # Ask AWS for temporary credentials in the target account.
RoleArn=f"arn:aws:iam::{account_id}:role/OrganizationInventoryReadOnly", # Target role.
RoleSessionName="pydevops-inventory", # Session name visible in CloudTrail.
) # STS returns time-limited credentials.
return resp["Credentials"] # Return credentials for regional clients.
def ec2_client(creds: dict, region: str): # Build a regional EC2 client.
return boto3.client( # Construct client with assumed-role credentials.
"ec2", # Inventory EC2 instances in this example.
region_name=region, # Query one region at a time.
aws_access_key_id=creds["AccessKeyId"], # Temporary access key.
aws_secret_access_key=creds["SecretAccessKey"], # Temporary secret key.
aws_session_token=creds["SessionToken"], # Temporary session token.
config=AWS_CONFIG, # Apply retry and timeout settings.
) # Return configured client.
def collect_account_region(account_id: str, region: str) -> dict: # Collect one unit of work.
try: # Keep failures scoped to this account-region pair.
creds = assume_inventory_role(account_id) # Assume the inventory role.
ec2 = ec2_client(creds, region) # Create EC2 client for the region.
paginator = ec2.get_paginator("describe_instances") # Use paginator, not one API call.
instances = [] # Store normalized inventory records.
for page in paginator.paginate(PaginationConfig={"PageSize": 100}): # Walk every page.
for reservation in page.get("Reservations", []): # Reservations contain instances.
for inst in reservation.get("Instances", []): # Normalize each instance.
instances.append({"account": account_id, "region": region, "id": inst["InstanceId"], "state": inst["State"]["Name"]}) # Keep essential fields.
return {"account": account_id, "region": region, "status": "ok", "count": len(instances), "items": instances} # Report success.
except ClientError as exc: # AWS returned a structured error.
code = exc.response.get("Error", {}).get("Code", "ClientError") # Extract error code.
return {"account": account_id, "region": region, "status": "error", "error": code, "items": []} # Preserve partial failure.
def collect_all(accounts: list[str], regions: list[str]) -> list[dict]: # Run bounded fanout.
jobs = [(acct, region) for acct in accounts for region in regions] # Build explicit work list.
results = [] # Accumulate every account-region result.
with ThreadPoolExecutor(max_workers=16) as pool: # Bound API pressure.
futures = [pool.submit(collect_account_region, acct, region) for acct, region in jobs] # Submit jobs.
for future in as_completed(futures): # Consume results as they complete.
results.append(future.result()) # Keep successes and recorded failures.
return results # Caller can write JSON, metrics, or reports.Interview Tip
A junior engineer typically writes one describe_instances call and loops over accounts until something fails, but for a senior role, the interviewer is actually looking for correctness under partial failure. Say the words paginator, AssumeRole, bounded concurrency, retry mode, throttling classification, and audit summary. The best answers also explain why AccessDenied should not be retried and why no resources is different from no permission.
◈ Architecture Diagram
┌───────┐ assume ┌─────┐ creds ┌──────┐ pages ┌──────┐ │ Script│───────▶│ STS │──────▶│ EC2 │──────▶│ JSON │ └───────┘ └─────┘ └──────┘ └──────┘ │ ▲ ▲ ▼ │ │ ┌───────┐ ┌──────┐ ┌──────┐ │ Report│ │Acct │ │Region│ └───────┘ └──────┘ └──────┘
💬 Comments
Quick Answer
Use asyncio with bounded concurrency, per-request timeouts, structured cancellation, and a result object for every target. In modern Python, TaskGroup helps manage groups of tasks, while semaphores and client timeouts prevent the script from becoming a denial-of-service tool.
Detailed Answer
Think of this like checking 10,000 doors in a city. Sending one inspector door by door is too slow, but sending 10,000 inspectors at once blocks streets and creates a new incident. Good async automation is controlled parallelism: enough concurrency to finish quickly, but not so much that DNS, load balancers, proxies, or the service itself collapse.
Python asyncio is useful for I/O-bound work because the event loop can switch tasks while each request waits on the network. The design should separate three concerns: target generation, bounded execution, and result collection. A semaphore limits the number of in-flight checks. Each request has connect and read timeouts. The caller gets one result per endpoint with status, latency, error class, and retry count.
Modern Python adds structured concurrency through asyncio.TaskGroup. A TaskGroup gives a clear lifetime to a set of related tasks: create them inside the block, wait for completion, and let errors cancel sibling tasks in a controlled way. For health-check automation, many teams choose to catch per-target exceptions inside the worker so one bad endpoint does not cancel the whole scan. For deployment gates, they may do the opposite: fail fast when critical checks break.
At scale, the bottleneck may not be Python. DNS resolvers, NAT gateways, outbound proxies, and service-side rate limits all matter. A mature script includes a configurable concurrency limit, jittered retries for transient network errors, and an allowlist of target domains or clusters. It also emits metrics about timeout rate and p95 latency because the health checker itself becomes part of the deployment system.
The subtle gotcha is cancellation hygiene. If a CI job times out or a deploy is aborted, the script should stop launching new checks and close HTTP connections cleanly. Senior engineers talk about cancellation, finalizers, and bounded resource cleanup, not just await gather.
Code Example
from __future__ import annotations # Enable modern typing features.
import asyncio # Use Python's async event loop for I/O concurrency.
from dataclasses import dataclass # Return structured health check results.
from time import monotonic # Measure latency safely.
import aiohttp # Use an async HTTP client for many concurrent requests.
@dataclass(frozen=True) # Keep each result immutable.
class HealthResult: # Represent one endpoint check.
url: str # Endpoint that was checked.
ok: bool # Whether the endpoint returned a healthy response.
status: int | None # HTTP status when a response exists.
ms: int # Elapsed milliseconds.
error: str | None # Error class for timeout or network failures.
async def check_one(session: aiohttp.ClientSession, sem: asyncio.Semaphore, url: str) -> HealthResult: # Check one URL.
async with sem: # Bound total in-flight checks.
start = monotonic() # Start timing before the request.
try: # Convert exceptions into result records.
async with session.get(url) as resp: # Run the HTTP GET asynchronously.
await resp.read() # Drain the body so the connection can be reused.
ms = int((monotonic() - start) * 1000) # Convert elapsed time to ms.
return HealthResult(url, 200 <= resp.status < 300, resp.status, ms, None) # Return success or bad status.
except asyncio.TimeoutError: # Request exceeded the configured timeout.
ms = int((monotonic() - start) * 1000) # Measure timeout duration.
return HealthResult(url, False, None, ms, "timeout") # Preserve timeout as data.
except aiohttp.ClientError as exc: # DNS, connect, TLS, and protocol failures land here.
ms = int((monotonic() - start) * 1000) # Measure failed request duration.
return HealthResult(url, False, None, ms, type(exc).__name__) # Preserve error class.
async def check_many(urls: list[str], concurrency: int = 200) -> list[HealthResult]: # Run all checks.
timeout = aiohttp.ClientTimeout(total=8, connect=2, sock_read=5) # Bound every request phase.
sem = asyncio.Semaphore(concurrency) # Protect dependencies from unbounded fanout.
async with aiohttp.ClientSession(timeout=timeout) as session: # Reuse connections efficiently.
async with asyncio.TaskGroup() as tg: # Give all tasks a structured lifetime.
tasks = [tg.create_task(check_one(session, sem, url)) for url in urls] # Schedule checks.
return [task.result() for task in tasks] # Return one result per URL.
if __name__ == "__main__": # Keep script entry point explicit.
targets = ["https://payments.example.com/healthz", "https://orders.example.com/healthz"] # Example targets.
results = asyncio.run(check_many(targets, concurrency=100)) # Run event loop from sync main.
failed = [r for r in results if not r.ok] # Summarize failed checks.
raise SystemExit(1 if failed else 0) # Fail CI only when health checks fail.Interview Tip
A junior engineer typically says they would use asyncio.gather because it is faster, but for a senior role, the interviewer is actually looking for bounded concurrency and cancellation behavior. Explain why 10,000 simultaneous sockets can break DNS, NAT, proxies, or the target service. Mention TaskGroup, semaphores, per-request timeouts, connection reuse, and a result object for every target.
◈ Architecture Diagram
┌────────┐ urls ┌────────┐ tasks ┌────────┐
│ Targets│──────▶│TaskGrp │──────▶│ Checks │
└────────┘ └────────┘ └────────┘
│ sem │
▼ ▼
┌──────┐ ┌──────┐
│Limit │ │Result│
└──────┘ └──────┘💬 Comments
Quick Answer
Use the official Kubernetes Python client, load in-cluster or kubeconfig credentials, list with resourceVersion, then watch pod events with timeouts and reconnect logic. Filter pod container statuses for waiting.reason equals CrashLoopBackOff, deduplicate by namespace/name/container/restart count, and emit structured events instead of polling every few seconds.
Detailed Answer
A Kubernetes watcher is like a security desk watching badge events at a large office. Asking for the full list every five seconds is wasteful and still misses context. Subscribing to a stream and remembering the last seen position is cheaper and more accurate. That is the mental model interviewers want when they ask about Kubernetes automation in Python.
The official Kubernetes Python client exposes CoreV1Api and watch.Watch. A production watcher usually starts with a list_namespaced_pod or list_pod_for_all_namespaces call to get the current resourceVersion, then uses watch.stream against the same list function. The resourceVersion is important because Kubernetes objects are versioned; it lets the watcher resume from a known point instead of starting from scratch every time.
CrashLoopBackOff is not a pod phase. It appears in container status under state.waiting.reason, often alongside restartCount and lastState. Good code inspects initContainerStatuses and containerStatuses, records the exact container, and includes namespace, pod name, node, image, restart count, and last termination reason. It deduplicates so one broken pod does not create 500 Slack messages.
At scale, the watcher must respect API server limits. Use watch timeouts so the HTTP stream refreshes periodically. Handle 410 Gone by relisting because the resourceVersion is too old. Avoid one watcher per namespace unless needed; a single all-namespaces watch is usually simpler. Emit metrics for reconnects, event lag, and handler errors so the automation itself is observable.
The non-obvious gotcha is RBAC and blast radius. A watcher running in-cluster should use a ServiceAccount with list/watch on pods only, not cluster-admin. If it creates incident objects, write those to a narrow namespace or external queue, not back into every application namespace.
Code Example
from __future__ import annotations # Use modern Python annotations.
from kubernetes import client, config, watch # Import the official Kubernetes Python client.
from kubernetes.client.exceptions import ApiException # Handle Kubernetes API errors.
def load_config() -> None: # Load credentials for local or in-cluster execution.
try: # Prefer in-cluster configuration for deployed watchers.
config.load_incluster_config() # Reads service account token and CA from the pod.
except config.ConfigException: # Fall back when running locally.
config.load_kube_config() # Reads the user's kubeconfig file.
def crashloop_events() -> None: # Stream CrashLoopBackOff signals.
load_config() # Configure Kubernetes client authentication.
api = client.CoreV1Api() # Create Core API client for pods.
resource_version = "" # Empty means start from the latest list result.
seen: set[tuple[str, str, str, int]] = set() # Deduplicate repeated watch events.
while True: # Reconnect forever like a controller loop.
try: # Keep API errors from killing the process.
stream = watch.Watch().stream( # Start a Kubernetes watch stream.
api.list_pod_for_all_namespaces, # Watch pods across namespaces.
resource_version=resource_version, # Resume from the last version when possible.
timeout_seconds=300, # Refresh the HTTP watch periodically.
) # The stream yields ADDED, MODIFIED, and DELETED events.
for event in stream: # Process each watched event.
pod = event["object"] # Extract the V1Pod object.
resource_version = pod.metadata.resource_version # Save progress for reconnect.
statuses = pod.status.container_statuses or [] # Regular containers may be absent early.
for status in statuses: # Inspect each container status.
state = status.state.waiting if status.state else None # CrashLoopBackOff appears in waiting state.
if not state or state.reason != "CrashLoopBackOff": # Ignore healthy or unrelated states.
continue # Move to the next container.
key = (pod.metadata.namespace, pod.metadata.name, status.name, status.restart_count) # Build idempotency key.
if key in seen: # Avoid duplicate incident notifications.
continue # Skip already emitted restart count.
seen.add(key) # Mark this signal as emitted.
print({ # Replace print with Kafka, webhook, or incident API in production.
"namespace": pod.metadata.namespace, # Include namespace for routing.
"pod": pod.metadata.name, # Include pod name.
"container": status.name, # Include failing container.
"node": pod.spec.node_name, # Include node for correlated incidents.
"restarts": status.restart_count, # Include restart count for severity.
"image": status.image, # Include deployed image.
}) # Structured event payload.
except ApiException as exc: # Kubernetes returned an API error.
if exc.status == 410: # resourceVersion too old.
resource_version = "" # Relist from current state.
continue # Restart the watch loop.
raise # Unexpected API errors should be surfaced.Interview Tip
A junior engineer typically polls kubectl get pods in a loop, but for a senior role, the interviewer is actually looking for Kubernetes API mechanics: list-watch, resourceVersion, 410 Gone recovery, RBAC scoping, and deduplication. Explain that CrashLoopBackOff is in container state, not pod phase. Also mention that a watcher is production software and needs metrics, reconnect logic, and controlled notification volume.
◈ Architecture Diagram
┌──────┐ watch ┌──────┐ event ┌────────┐ emit ┌──────┐ │ K8s │──────▶│Python│──────▶│ Dedupe │─────▶│Alert │ └──────┘ └──────┘ └────────┘ └──────┘ ▲ │ rv └──────────────┘
💬 Comments
Quick Answer
Design the script as an idempotent workflow: discover keys, classify stale keys, produce a dry-run plan, require approval for destructive actions, execute with rate-limit-aware retries, and write append-only audit logs. Rollback should be based on captured key metadata and should be tested on a small allowlisted repository set before fleet-wide execution.
Detailed Answer
A remediation script is closer to a surgical checklist than a one-off script. The dangerous part is not the API call that deletes a key; it is deleting the wrong key quickly across hundreds of repositories. Interviewers ask this to see whether you understand automation blast radius, approval gates, and auditability.
The workflow should have clear phases. Discovery reads repository keys and last-used signals where the provider exposes them. Classification marks keys stale only when they match policy, such as no usage for 90 days, not owned by a protected integration, and not tagged as break-glass. Planning writes a human-readable and machine-readable diff. Approval requires a ticket, reviewer, or signed change request. Execution deletes or disables keys in small batches. Verification confirms the key is gone and that known deployment pipelines still authenticate.
In Python, design the code around pure functions and side-effect boundaries. A classify_key function should be easy to unit test with fake key metadata. A provider client should hide GitHub, GitLab, or Bitbucket API differences. The execution layer should handle 429 rate limits, 5xx retries, and permanent 403 permission errors differently. The script should accept --dry-run by default and require --execute plus a change ID for mutation.
At scale, rollback is not magic. If the API supports disabling instead of deleting, prefer disable first. If deletion is required, capture enough metadata to recreate the key: title, public key, repository, permissions, and owner. Never log private keys because they are not retrievable and should not exist in logs. Use append-only audit storage so security teams can answer who changed what and when.
The non-obvious gotcha is ownership. A key may look stale because usage telemetry is incomplete or because a rarely used disaster recovery pipeline only runs quarterly. Senior engineers define exception handling and communication before the deletion run, not after the outage.
Code Example
from __future__ import annotations # Use modern Python typing.
import argparse # Parse dry-run and execute flags explicitly.
import json # Write machine-readable plans and audit records.
from dataclasses import asdict, dataclass # Model remediation plans as data.
from datetime import datetime, timezone # Use timezone-aware audit timestamps.
@dataclass(frozen=True) # Immutable plan item for one deploy key.
class KeyPlan: # Describe the intended action.
repo: str # Repository name.
key_id: str # Provider key identifier.
title: str # Human-readable key title.
action: str # Planned action such as disable or skip.
reason: str # Policy reason for the action.
PROTECTED_TITLES = {"prod-breakglass", "release-bot"} # Never mutate protected keys.
def classify_key(repo: str, key: dict) -> KeyPlan: # Pure function that is easy to test.
title = key.get("title", "") # Read key title safely.
days_unused = int(key.get("days_unused", 0)) # Normalize usage age.
if title in PROTECTED_TITLES: # Respect explicit protection policy.
return KeyPlan(repo, key["id"], title, "skip", "protected title") # Skip protected key.
if days_unused >= 90: # Apply stale-key policy.
return KeyPlan(repo, key["id"], title, "disable", f"unused for {days_unused} days") # Plan remediation.
return KeyPlan(repo, key["id"], title, "skip", "recently used") # Keep active keys.
def write_audit(record: dict) -> None: # Append audit evidence.
envelope = { # Wrap event with common metadata.
"ts": datetime.now(timezone.utc).isoformat(), # Use UTC timestamp.
"tool": "deploy-key-remediator", # Identify automation source.
**record, # Include event-specific details.
} # Envelope is JSON serializable.
print(json.dumps(envelope, sort_keys=True)) # Send to stdout or log collector.
def main() -> int: # Script entry point returns process status.
parser = argparse.ArgumentParser() # Build CLI parser.
parser.add_argument("--execute", action="store_true") # Require explicit mutation flag.
parser.add_argument("--change-id", default="") # Tie changes to approval record.
args = parser.parse_args() # Parse command line.
if args.execute and not args.change_id: # Enforce approval evidence.
raise SystemExit("--change-id is required with --execute") # Fail before mutation.
sample_keys = [{"id": "123", "title": "old-preview-deploy", "days_unused": 120}] # Replace with provider API data.
plan = [classify_key("payments-api", key) for key in sample_keys] # Build remediation plan.
write_audit({"phase": "plan", "items": [asdict(item) for item in plan]}) # Record the plan.
if not args.execute: # Dry-run is the default behavior.
return 0 # Exit successfully without changes.
for item in plan: # Execute approved actions.
if item.action != "disable": # Only mutate planned disable actions.
continue # Skip non-mutating items.
write_audit({"phase": "execute", "repo": item.repo, "key_id": item.key_id, "change_id": args.change_id}) # Record mutation.
return 0 # Return success.
if __name__ == "__main__": # Run only when executed as a script.
raise SystemExit(main()) # Convert return code to process exit.Interview Tip
A junior engineer typically focuses on the API call that deletes the key, but for a senior or architect role, the interviewer is actually looking for controlled change design. Walk through discovery, classification, dry-run, approval, execution, verification, and rollback. Mention that dry-run should be the default and that destructive flags should require a change ID. This shows you can automate security cleanup without creating a larger incident.
◈ Architecture Diagram
┌────────┐ scan ┌────────┐ plan ┌────────┐ approve ┌────────┐
│ Repos │───────▶│Classify│─────▶│Dry Run │────────▶│Execute │
└────────┘ └────────┘ └────────┘ └────────┘
│ │
▼ ▼
┌──────┐ ┌──────┐
│Audit │◀─────────│Verify│
└──────┘ └──────┘💬 Comments
Quick Answer
Package automation as a real Python project with pinned dependencies, console_scripts entry points, unit tests around pure logic, integration tests behind explicit flags, and containerized execution for CI. Avoid unversioned scripts copied between hosts, implicit global Python packages, and credentials hidden in developer machines.
Detailed Answer
A Python automation repo is like a toolbox shared by many mechanics. If every mechanic has a different wrench size and nobody knows which tool was used on the last repair, the organization eventually creates inconsistent and risky operations. Packaging is what turns a script into a repeatable operational tool.
The baseline is a project file such as pyproject.toml, a src layout, pinned runtime dependencies, and a lock or constraints strategy. The script should expose a console_scripts entry point so users run inventory-scan instead of python random/path/script.py. Configuration should come from environment variables, config files, or cloud identity, not from edits inside the script. Logging should be structured enough for CI and cron logs.
Testing should match the risk. Pure logic such as filtering stale resources, rendering reports, or classifying errors should have fast unit tests with no cloud access. API boundaries should be wrapped behind clients that can be faked. Integration tests that touch AWS, Kubernetes, or Git providers should require explicit environment variables and run against sandbox accounts or ephemeral namespaces. This prevents every pull request from mutating real infrastructure.
For execution, containers remove host drift. A slim Python image with the package installed, non-root user, and pinned dependency set gives CI, cron, and Kubernetes CronJobs the same runtime. The container should not bake in secrets. Credentials should come from workload identity, IAM role, Kubernetes ServiceAccount, or secret mounts controlled by the platform.
The gotcha is dependency drift. A script can work for months and then break because a transitive dependency released a new major version or a base image moved. Senior candidates mention reproducible builds, dependency scanning, scheduled upgrade PRs, and recording tool version in every audit log.
Code Example
# pyproject.toml
[project] # Define installable Python package metadata.
name = "platform-automation" # Give the automation a stable package name.
version = "0.3.0" # Version the operational tool for audit records.
dependencies = ["boto3==1.34.120", "kubernetes==29.0.0", "pydantic==2.7.4"] # Pin runtime dependencies.
[project.scripts] # Define command line entry points.
inventory-scan = "platform_automation.inventory:main" # Run as inventory-scan.
[tool.pytest.ini_options] # Keep tests discoverable in CI.
testpaths = ["tests"] # Store tests under tests directory.
# Dockerfile
FROM python:3.12-slim # Use a specific Python runtime image.
WORKDIR /app # Set a predictable working directory.
COPY pyproject.toml ./ # Copy package metadata first for layer caching.
COPY src ./src # Copy source code into the image.
RUN pip install --no-cache-dir . # Install package and console scripts.
USER 10001 # Run automation as a non-root user.
ENTRYPOINT ["inventory-scan"] # Make the container run the tool by default.
# tests/test_policy.py
from platform_automation.policy import should_delete_volume # Import pure policy logic.
def test_keeps_recent_volume(): # Verify safe default behavior.
assert should_delete_volume({"age_days": 5, "attached": False}) is False # Recent resources are kept.
def test_deletes_old_unattached_volume(): # Verify intended cleanup policy.
assert should_delete_volume({"age_days": 45, "attached": False}) is True # Old unattached resources are removable.Interview Tip
A junior engineer typically says they would put the script in Git and run pip install, but for a senior role, the interviewer is actually looking for reproducibility. Mention pyproject.toml, pinned dependencies, console entry points, unit tests, explicit integration tests, containerized execution, non-root runtime, and secrets through identity rather than files on a laptop. This is the difference between a helpful script and production automation.
◈ Architecture Diagram
┌──────┐ build ┌──────┐ test ┌──────┐ run ┌──────┐
│ Code │──────▶│Pkg │─────▶│Image │────▶│ CI │
└──────┘ └──────┘ └──────┘ └──────┘
│ │
▼ ▼
┌──────┐ ┌──────┐
│Tests │ │Audit │
└──────┘ └──────┘💬 Comments
Quick Answer
Use boto3's describe_instances paginator to walk every running instance in a region, build a set of each instance's existing tag keys, and flag any instance missing a required key like owner. Pagination matters because a single API call caps out at roughly 1,000 instances, so a naive single-call approach silently misses resources in any account with a real EC2 footprint.
Detailed Answer
Imagine a shared office building where every desk is supposed to have a nameplate showing which team owns it, but facilities can only fit a limited number of desks on one clipboard page — if the person doing rounds only checks the first page and calls it done, half the building's desks never get inspected, and desks without nameplates on the missed pages stay invisible forever even though they're still using power, space, and the janitorial budget.
This script exists because cloud costs quietly become unowned costs — an EC2 instance launched without a required tag, like owner or cost-center, can run for months, accumulating real spend, with nobody able to answer "who launched this and can we turn it off?" during a cost review. FinOps and platform teams write governance scripts like this specifically because manual tag audits don't scale past a handful of instances, and because tagging policy is only enforceable if something actually checks compliance on a recurring basis rather than trusting that engineers remembered the rule at launch time.
The script calls ec2.describe_instances with a filter for running state, but critically uses get_paginator('describe_instances') rather than a single call, because the EC2 API returns results in pages, using a NextToken, and any account with more than roughly 1,000 instances in scope will silently truncate results without pagination. For each instance, it builds a Python set from the Tags list, where each tag is a Key/Value dict, and checks whether the required key is a member of that set — using a set rather than scanning the list repeatedly keeps the lookup efficient even across accounts with thousands of instances and dozens of tags each.
In production, this script is typically scheduled as a daily Lambda function or CI job, with findings posted to Slack or written to a compliance dashboard rather than just printed to a console. What to monitor: the trend of untagged-instance count over time, since a sudden spike usually means a new automation or a new team started launching instances without going through the standard provisioning path that applies tags automatically, and false positives from instances tagged through mechanisms this script doesn't check, like tags applied only at the Auto Scaling Group level that don't propagate to instances unless PropagateAtLaunch is explicitly enabled.
The non-obvious gotcha is that describe_instances only reflects tags as of the API call, and tags applied moments after instance launch, common with automation that launches first and then tags in a follow-up API call, can create a race condition where a perfectly compliant, soon-to-be-tagged instance gets flagged as a false positive if the governance script runs in that narrow window. Mature versions of this script add a grace period, skipping instances launched less than 15 to 30 minutes ago, to avoid paging someone about a tagging violation that's actually just eventual consistency in the provisioning pipeline.
Code Example
import boto3
# Define the AWS region and the tag key that should be present
REGION = 'us-east-1'
REQUIRED_TAG_KEY = 'owner'
# Initialize the EC2 client
ec2 = boto3.client('ec2', region_name=REGION)
print(f"Searching for running EC2 instances in {REGION} missing the '{REQUIRED_TAG_KEY}' tag...")
# Paginate through all instances to handle large accounts
paginator = ec2.get_paginator('describe_instances')
page_iterator = paginator.paginate(
Filters=[{'Name': 'instance-state-name', 'Values': ['running']}]
)
found_untagged = False
for page in page_iterator:
for reservation in page['Reservations']:
for instance in reservation['Instances']:
instance_id = instance['InstanceId']
# Create a set of existing tag keys for easy lookup
existing_tags = {tag['Key'] for tag in instance.get('Tags', [])}
if REQUIRED_TAG_KEY not in existing_tags:
print(f"- Untagged Instance Found: {instance_id}")
found_untagged = True
if not found_untagged:
print("No untagged running instances found.")Interview Tip
A junior engineer typically writes a single describe_instances call and calls it done, but for a senior or platform-focused role, the interviewer is actually looking for you to catch the pagination gap — a single call truncates past roughly 1,000 instances, which silently under-reports compliance in any account with real scale. Strong answers also address the eventual-consistency race condition between instance launch and tag application, adding a grace period so freshly launched instances aren't flagged as false positives, and distinguish instance-level tags from Auto Scaling Group tags that only propagate at launch if explicitly configured. Mentioning how findings get delivered, a Slack alert, a compliance dashboard, or a hard gate in provisioning, shows you're thinking about this as an operational governance loop, not a one-off script.
◈ Architecture Diagram
┌────────┐ paginate ┌─────────┐ check ┌──────────┐
│describe│─────────▶│Instances│──────▶│tag exists│
│instncs │ └─────────┘ └────┌─────┘
└────────┘ │ no
↓
┌──────────┐
│ Flagged │
└──────────┘💬 Comments
Quick Answer
Use Python's built-in ssl and socket modules to open a TLS connection to each domain on port 443, retrieve the peer certificate's notAfter field, parse it into a datetime, and compare the days remaining against a warning threshold. This requires no external dependencies and mirrors exactly what a browser checks when it shows a certificate warning.
Detailed Answer
Think of a TLS certificate like a passport with an expiration date stamped inside — a border agent doesn't need to call the passport office to know it's expired, they just read the date printed on the document itself. This script does the equivalent of a border agent's ID check: connect, ask to see the certificate, read the expiration date directly, and flag anyone whose certificate is about to run out before anyone gets stuck at the border.
This script exists because certificate expiration is one of the most preventable categories of production outage, and yet it happens constantly — a certificate renewed manually eighteen months ago quietly expires at 2 AM, taking down an entire service's HTTPS endpoint, and the team finds out from a customer complaint instead of a dashboard. Python's standard library ssl module was chosen deliberately over an external tool because it requires zero extra dependencies, which matters for lightweight monitoring scripts that need to run reliably in constrained environments like a cron job on a bastion host or a minimal container image.
The script opens a raw TCP socket to each domain on port 443, wraps it with an SSL context via wrap_socket, which performs the actual TLS handshake including certificate presentation from the server, and calls getpeercert() to retrieve the certificate's fields as a Python dictionary. Critically, this only works with certificate verification enabled, the default in create_default_context(), because getpeercert() returns an empty dict if verification is disabled, a subtle trap engineers hit when they've disabled verification elsewhere in their codebase and copy-paste that pattern here. The notAfter field comes back as a specific string format that must be parsed with datetime.strptime using the exact matching format string, and the days-remaining calculation is a simple datetime subtraction.
In production, this check is typically wired into a Prometheus blackbox exporter or run as a scheduled job feeding a dashboard, with graduated alert thresholds — a 30-day warning gives teams time to file a renewal ticket, while a 7-day warning should page on-call directly since it indicates the earlier warning was missed. What to monitor beyond just days-remaining: the certificate's issuer chain, since an internal CA cert expiring is just as dangerous as a public one, and Subject Alternative Names coverage, since a certificate can be not-expired but still wrong for a newly added subdomain that was never included when the cert was issued.
The non-obvious gotcha is that this script only checks the certificate actively being served by the connection it makes — if a service sits behind a load balancer with multiple backend instances serving slightly different or differently-aged certificates due to a partially-rolled-out renewal, a single check from one script run might see the newly renewed cert on one backend and report all clear while a different backend instance is still serving the old, soon-to-expire certificate to some fraction of real traffic. Thorough checks connect multiple times or target specific backend IPs directly rather than trusting DNS round-robin to represent the whole fleet in one shot.
Code Example
import ssl
import socket
from datetime import datetime, timedelta
DOMAINS_TO_CHECK = ['google.com', 'github.com', 'expired.badssl.com']
EXPIRY_THRESHOLD_DAYS = 30
print(f"Checking certificate expiry (threshold: {EXPIRY_THRESHOLD_DAYS} days)...")
for domain in DOMAINS_TO_CHECK:
try:
# Create a default SSL context
context = ssl.create_default_context()
# Connect to the domain on port 443
with socket.create_connection((domain, 443), timeout=10) as sock:
with context.wrap_socket(sock, server_hostname=domain) as ssock:
cert = ssock.getpeercert()
# Parse the 'notAfter' date string
expiry_date = datetime.strptime(cert['notAfter'], '%b %d %H:%M:%S %Y %Z')
days_left = (expiry_date - datetime.now()).days
if days_left < 0:
print(f"[ERROR] {domain}: Certificate EXPIRED {abs(days_left)} days ago.")
elif days_left < EXPIRY_THRESHOLD_DAYS:
print(f"[WARNING] {domain}: Certificate expires in {days_left} days.")
else:
print(f"[OK] {domain}: Certificate expires in {days_left} days.")
except socket.gaierror:
print(f"[ERROR] {domain}: Could not resolve hostname.")
except ssl.SSLCertVerificationError as e:
print(f"[ERROR] {domain}: Certificate verification failed: {e.reason}")
except Exception as e:
print(f"[ERROR] {domain}: An unexpected error occurred: {e}")Interview Tip
A junior engineer typically says they'd "just check the cert expiry date," but for a senior or SRE role, the interviewer is actually looking for you to explain why getpeercert() silently returns nothing if certificate verification is disabled, which is an easy trap when other parts of a codebase disable verification for convenience. Strong answers also mention graduated alert thresholds, a 30-day warning versus a 7-day page since a missed first warning needs escalation not repetition, checking SAN coverage rather than just expiry, and, the deepest insight, that a single connection only samples one backend behind a load balancer, so a partially-rolled-out certificate renewal can hide a still-expiring certificate on other backend instances from a single-shot check.
◈ Architecture Diagram
┌────────┐ TLS handshake ┌────────┐ notAfter ┌─────────┐
│ Script │──────────────▶│ Domain │─────────▶│ Parse │
└────────┘ └────────┘ └────┌────┘
↓
days_left < 30?
✓ ok │ ✗ alert💬 Comments
Quick Answer
Use the Kubernetes Python client to list pods in a namespace, iterate through each pod's container statuses, and report any container where restart_count exceeds a defined threshold. This surfaces CrashLoopBackOff and intermittent OOMKill patterns in seconds instead of manually describing dozens of pods one at a time.
Detailed Answer
This is like a hospital triage nurse walking through a ward not to check every patient's full chart, but specifically to spot anyone whose vital signs monitor has alarmed repeatedly — a single alarm might be a fluke, but a patient whose monitor has alarmed five times in an hour needs attention immediately, and the nurse doesn't need to read every chart in detail to know which bed to go to first.
This script exists because kubectl get pods shows a restart count column, but scanning that visually across a namespace with hundreds of pods during an incident is slow and error-prone, especially when the unstable pods are scattered rather than clustered. Automating the threshold check turns a manual visual scan into an instant, repeatable triage step — this matters most during incidents, when the fastest path to root cause is often "which of these 200 pods is actually the unstable one," not a general health overview.
The script authenticates via config.load_kube_config(), reading the local kubeconfig, though load_incluster_config() would be used if this ran as a pod itself, then calls list_namespaced_pod to get every pod object in the target namespace. Each pod's status.container_statuses is a list, one entry per container in the pod, that can be None if the pod is still being scheduled or its images are still pulling, which is why the script explicitly guards against that case before iterating. For each container status, restart_count is a cumulative counter that Kubernetes maintains for the life of the pod object, resetting to zero only if the pod itself is deleted and recreated, not on a simple container restart, so a high restart count is a durable signal of instability rather than something that self-clears.
In production, this kind of check is typically run both proactively, as a daily cluster health scan feeding a dashboard, and reactively, with an on-call engineer running it directly during triage to shortcut past kubectl describe on dozens of pods. What matters operationally is combining restart count with the reason for the last termination, terminated.reason such as OOMKilled versus Error, since a high restart count alone doesn't tell you whether the fix is a memory limit increase, a code bug, or a missed dependency — the threshold script is a triage filter, not a root-cause diagnosis by itself.
The non-obvious gotcha is that restart count resets to zero when a pod is deleted and recreated, for example during a rolling deployment or when a node is replaced, so a genuinely unstable workload that's been crash-looping for days can show a deceptively low restart count immediately after its most recent pod recreation, hiding the real instability pattern from a single point-in-time check. Engineers who rely purely on restart count without also checking pod age can be misled into thinking a chronic problem is resolved right after a deployment simply reset the counter, not because anything was actually fixed.
Code Example
from kubernetes import client, config
# --- Configuration ---
NAMESPACE = 'production'
RESTART_THRESHOLD = 5
def find_restarting_pods():
"""Scans a namespace for pods with high container restart counts."""
try:
# Load kubeconfig from default location (~/.kube/config)
config.load_kube_config()
v1 = client.CoreV1Api()
print(f"Checking for pods in namespace '{NAMESPACE}' with more than {RESTART_THRESHOLD} restarts...")
pod_list = v1.list_namespaced_pod(NAMESPACE, watch=False)
found_pods = False
for pod in pod_list.items:
# container_statuses can be None if the pod is still creating
if not pod.status.container_statuses:
continue
for status in pod.status.container_statuses:
if status.restart_count > RESTART_THRESHOLD:
print(
f"- High Restarts Detected: Pod='{pod.metadata.name}', "
f"Container='{status.name}', Restarts={status.restart_count}"
)
found_pods = True
if not found_pods:
print("No pods with high restart counts found.")
except Exception as e:
print(f"An error occurred: {e}")
if __name__ == "__main__":
find_restarting_pods()Interview Tip
A junior engineer typically says they'd check restart_count and flag anything above a number, but for a senior or SRE role, the interviewer is actually looking for you to know that restart count resets to zero whenever the pod object itself is recreated, meaning a chronic crash-looper can look perfectly healthy right after a rolling deployment even though nothing was actually fixed. Strong answers combine restart count with pod age and the specific last-termination reason, OOMKilled versus a generic Error, rather than treating a raw restart number as the full diagnosis, and mention that container_statuses can be None for pods still scheduling, which causes an unhandled exception if you don't guard for it explicitly.
◈ Architecture Diagram
┌────────┐ list ┌──────┐ restarts>N? ┌────────┐
│ API │─────▶│ Pods │────────────▶│ Flagged│
└────────┘ └──────┘ └────────┘
│ check pod age too
↓
recently recreated?
(resets counter)💬 Comments
Quick Answer
Use boto3's describe_volumes with a filter for status=available, meaning the volume is not attached to any instance, compare each volume's CreateTime against a cutoff date, and delete volumes older than the threshold, always behind a DRY_RUN flag first, since deletion is irreversible and unattached does not always mean unused.
Detailed Answer
This is like a storage unit facility running a report on units that haven't had anyone visit in 30 days, so facilities management can start the process of reclaiming and re-renting them — but a good facility manager still checks the paperwork before drilling the lock off a unit, because "nobody's visited" doesn't always mean "nobody wants it," especially if it's someone's disaster-recovery storage they intentionally never touch.
This script exists because EBS volumes left behind after an instance is terminated, rather than the volume being explicitly deleted alongside it, silently continue billing at their full provisioned rate indefinitely — this is one of the most common sources of unnoticed cloud waste, since nothing about an orphaned volume is loud or urgent, it just sits there costing money with no owner watching it. DeleteOnTermination defaults to true for the root volume but not for additional attached volumes, which is exactly how so many of these accumulate over time in accounts with frequent instance replacement.
The script filters describe_volumes for status available, which EC2 sets specifically when a volume is not currently attached to any instance, as opposed to in-use or creating. It then compares each volume's CreateTime, a timezone-aware datetime returned directly by boto3, against a cutoff computed with timedelta, using timezone.utc explicitly to avoid the classic naive-vs-aware datetime comparison error that raises a TypeError in Python. Deletion itself is a single delete_volume call gated entirely behind the DRY_RUN flag, meaning the exact same discovery and filtering logic runs whether or not anything is actually deleted, which is the safest structure for this kind of destructive automation.
In production, this should never run as a single unattended step straight to deletion — the standard operating pattern is DRY_RUN=True on a schedule generating a report for review, often posted to a cost-optimization channel, with a human or a secondary approval step confirming the list before flipping to DRY_RUN=False. What to check before deleting: whether the volume has a tag indicating intentional retention, such as a manual snapshot source, disaster-recovery data, or a volume awaiting a support ticket investigation, and whether a recent snapshot exists — teams commonly add a rule that skips deletion for volumes without at least one recent snapshot, treating "no snapshot and no attachment" as the real deletion criterion rather than age alone.
The non-obvious gotcha is that available status only means "not currently attached" — it does not mean "safe to delete." A volume can be intentionally detached and parked, common during a maintenance window, a migration in progress, or as a deliberate disaster-recovery strategy of keeping data attached to nothing until needed, and a purely age-based cleanup script with no additional context will happily delete something a team is actively relying on, with no way to recover it once gone since EBS volume deletion is permanent unless a snapshot exists. Mature versions of this script cross-reference tags, recent snapshot existence, and even a pending-deletion grace-period tag applied a week before actual deletion, giving owners a window to object.
Code Example
import boto3
from datetime import datetime, timedelta, timezone
# --- Configuration ---
REGION = 'us-west-2'
DAYS_TO_CONSIDER_OLD = 30
DRY_RUN = True # Set to False to actually delete volumes
def cleanup_old_volumes():
"""Finds and deletes old, unattached EBS volumes."""
ec2 = boto3.client('ec2', region_name=REGION)
cutoff_date = datetime.now(timezone.utc) - timedelta(days=DAYS_TO_CONSIDER_OLD)
print(f"Finding unattached EBS volumes in {REGION} older than {cutoff_date.date()}...")
# Filter for volumes that are 'available' (not attached)
response = ec2.describe_volumes(Filters=[{'Name': 'status', 'Values': ['available']}])
for volume in response['Volumes']:
volume_id = volume['VolumeId']
create_time = volume['CreateTime']
if create_time < cutoff_date:
print(f"- Found old, unattached volume: {volume_id} (Created: {create_time.date()}) Volume Size: {volume['Size']}GiB")
if not DRY_RUN:
try:
print(f" Deleting volume {volume_id}...")
ec2.delete_volume(VolumeId=volume_id)
print(f" Successfully deleted {volume_id}.")
except Exception as e:
print(f" Error deleting {volume_id}: {e}")
else:
print(f" (Dry Run) Would have deleted {volume_id}.")
if __name__ == "__main__":
cleanup_old_volumes()Interview Tip
A junior engineer typically focuses on the describe_volumes filter and the delete call, but for a senior or cost-optimization-focused role, the interviewer is actually looking for you to challenge the core assumption that "unattached" means "unused" — volumes are commonly detached intentionally for disaster recovery, migrations, or maintenance windows, and a purely age-based script will delete those with no recovery path once gone. Strong answers describe a graduated safety process: dry-run by default, human review of the report, checking for a recent snapshot before treating a volume as truly disposable, and even a pending-deletion grace-period tag that gives an owner a final chance to object before the volume is actually removed.
◈ Architecture Diagram
┌──────────┐ filter ┌───────────┐ age>N? ┌─────────┐
│ describe │───────▶│ available │───────▶│ DRY_RUN?│
│ volumes │ │ volumes │ └────┌────┘
└──────────┘ └───────────┘ yes│ │no
report←─┘ └─▶delete💬 Comments
Quick Answer
Use boto3 to list every bucket, then for each one check get_bucket_acl for grants to the AllUsers or AuthenticatedUsers well-known groups, and separately parse get_bucket_policy for any statement with Effect Allow and a wildcard Principal. Both checks are necessary because a bucket can be exposed through either mechanism independently, and checking only one gives a false sense of security.
Detailed Answer
Think of an S3 bucket like a storage room that can be unlocked two completely different ways — a physical spare key hidden under the mat, the ACL, and a separate digital keypad code shared in a group chat, the bucket policy. A security guard who only checks whether the key is under the mat, but never asks whether anyone shared the keypad code, will confidently report the room secure while half the building can still walk in through the door they didn't check.
This script exists because S3 public exposure remains one of the most common and highest-impact cloud misconfigurations, precisely because there are two independent ways a bucket becomes public — legacy Access Control Lists, a per-bucket or per-object permission grant system that predates IAM policies, and modern bucket policies, JSON documents defining who can perform which actions — and a security review that checks only the newer mechanism can completely miss an old ACL grant left over from years ago, or vice versa. AWS's own Block Public Access settings can mitigate both, but this script exists specifically to detect exposure directly rather than assume account-level settings are correctly configured everywhere.
For each bucket, the script first checks get_bucket_acl, iterating the Grants list and checking whether any grantee's URI matches the well-known AWS group URIs for AllUsers, meaning anyone on the internet, authenticated or not, or AuthenticatedUsers, meaning any AWS account holder, not just this account's users, which is a much larger and less obvious exposure than the name implies. Separately, it fetches get_bucket_policy, parses the JSON policy document, and inspects each statement for Effect Allow combined with a Principal of "*" or an AWS wildcard, which grants access to literally anyone. The script wraps both checks in try/except ClientError, since a bucket with no policy at all raises NoSuchBucketPolicy, which is an entirely normal, expected condition, not an error to alert on.
In production, this is typically deployed as a scheduled Lambda function feeding AWS Security Hub or a dedicated security Slack channel, and should run frequently, hourly or on every bucket policy change event via CloudTrail and EventBridge, not just daily, because public exposure is a "seconds matter" class of risk, not a "check it eventually" class. What to monitor beyond the binary public/not-public result: exceptions raised while checking ACLs or policies, since an AccessDenied error while scanning could mean the script's own IAM role lost permissions, which would silently blind the entire security check without anyone noticing unless that failure mode is itself alerted on.
The non-obvious gotcha, and the one that actually broke the original version of this script, is a Python syntax error: an if statement missing its trailing colon fails at parse time, meaning the entire script would refuse to even start — a reminder that in security tooling specifically, a script that silently fails to run at all is far more dangerous than one that runs and reports zero findings, because a cron job that errors out on import or syntax can go unnoticed for a long time if nothing alerts on the automation's own health, leaving buckets unmonitored while everyone assumes the nightly scan is working.
Code Example
import boto3
import json
from botocore.exceptions import ClientError
# Initialize the S3 client
s3 = boto3.client('s3')
PUBLIC_ACL_URIS = [
'http://acs.amazonaws.com/groups/global/AllUsers',
'http://acs.amazonaws.com/groups/global/AuthenticatedUsers'
]
def check_for_public_buckets():
"""Scans all S3 buckets for public ACLs or policies."""
print("Scanning all S3 buckets for public access...")
public_buckets = set()
try:
response = s3.list_buckets()
buckets = response.get('Buckets', [])
for bucket in buckets:
bucket_name = bucket['Name']
# 1. Check Bucket ACLs
try:
acl = s3.get_bucket_acl(Bucket=bucket_name)
for grant in acl.get('Grants', []):
grantee = grant.get('Grantee', {})
if grantee.get('URI') in PUBLIC_ACL_URIS:
print(f"[WARNING] Public ACL found on bucket: {bucket_name}")
public_buckets.add(bucket_name)
break # Move to next bucket once public ACL is found
except ClientError as e:
if e.response['Error']['Code'] != 'AccessDenied':
print(f" Could not get ACL for {bucket_name}: {e}")
# 2. Check Bucket Policy
try:
policy_str = s3.get_bucket_policy(Bucket=bucket_name).get('Policy')
policy = json.loads(policy_str)
for statement in policy.get('Statement', []):
principal = statement.get('Principal')
effect = statement.get('Effect')
# Check for public principal and Allow effect (note the required trailing colon below)
if effect == 'Allow' and principal in ('*', {'AWS': '*'}):
print(f"[WARNING] Public policy found on bucket: {bucket_name}")
public_buckets.add(bucket_name)
break # Move to next bucket
except ClientError as e:
# It's normal for buckets to not have a policy
if e.response['Error']['Code'] != 'NoSuchBucketPolicy':
print(f" Could not get policy for {bucket_name}: {e}")
if not public_buckets:
print("No publicly accessible buckets found.")
else:
print(f"\nSummary: Found {len(public_buckets)} potentially public buckets.")
except ClientError as e:
print(f"An error occurred listing buckets: {e}")
if __name__ == "__main__":
check_for_public_buckets()Interview Tip
A junior engineer typically checks only the bucket policy because that's the more commonly discussed mechanism, but for a senior or security-focused role, the interviewer is actually looking for you to know that legacy ACLs are a second, independent path to public exposure that a policy-only check completely misses. Strong answers explain the well-known ACL group URIs, AllUsers versus the often-misunderstood AuthenticatedUsers which means any AWS account holder, not just your own users, correctly treat NoSuchBucketPolicy as a normal condition rather than an error, and recognize that a security scanner failing silently, whether from a permissions loss or, notably, a plain syntax error, is itself a security incident, because nobody is watching the bucket while the scanner thinks it's watching for them.
◈ Architecture Diagram
┌────────┐ ┌──────────┐ ┌───────────┐
│ Bucket │─▶│ ACL check│─┐ │ Policy │
└────────┘ └──────────┘ │ │ check │
│───▶│ (parallel)│
either public ─▶ FLAGGED💬 Comments
Quick Answer
Use boto3 to paginate through list_roles, read each role's RoleLastUsed.LastUsedDate, falling back to CreateDate if the role has never been used, and flag anything older than a threshold like 90 days while explicitly skipping AWS service-linked roles. The script then retrieves a Slack webhook URL from Secrets Manager at runtime and posts a summary, rather than hardcoding a webhook into source code.
Detailed Answer
This is like a building security audit that reviews every employee badge and flags anyone who hasn't swiped into the building in three months — not because they're necessarily a threat, but because an unused badge that still works is unnecessary risk sitting around, and if it's ever stolen or misused, nobody would notice quickly since nobody was using it anyway.
This script exists because IAM roles accumulate over the life of an AWS account far faster than they get cleaned up — a role created for a one-time migration, a proof-of-concept that never shipped, or a service that was decommissioned without anyone remembering to remove its role, all continue to exist as valid, assumable identities indefinitely. Security teams treat unused roles as attack surface because every role is a potential path for privilege escalation or lateral movement if its trust policy or attached permissions are ever exploited, and reducing that surface area is one of the most effective, low-risk security improvements available.
The script paginates through list_roles, necessary because accounts with hundreds of roles would otherwise be truncated, and for each role calls get_role to access the RoleLastUsed field, which AWS populates based on actual AssumeRole activity — critically, this field can be entirely absent for a role that has genuinely never been assumed, which is why the script checks both RoleLastUsed and falls back to CreateDate for roles with no usage history at all. It explicitly skips roles under the /aws-service-role/ path, since these are managed by AWS services themselves and aren't meant to be manually cleaned up. For notification, rather than embedding a Slack webhook URL directly in code or an environment variable, both of which risk leaking a live webhook into logs or version control, it fetches the webhook from AWS Secrets Manager at runtime, keeping the actual credential out of the deployment artifact entirely.
In production, this runs as a scheduled Lambda triggered by EventBridge on a recurring cadence, with its execution role scoped narrowly to list roles, get role, and get-secret-value on the one specific secret it needs — a common mistake is granting broader Secrets Manager access than the single secret required. What to monitor: the trend in flagged-role count over time, since a sudden jump usually means a team's automation started creating throwaway roles that never got attached to a cleanup process, and Slack delivery failures themselves, since a security report that fails to send silently is functionally the same as a security report that was never generated.
The non-obvious gotcha is that RoleLastUsed data is not updated in real time — AWS documents it as being updated within a matter of hours, not instantly — so a role assumed minutes before the scan runs can still show as last-used much earlier than expected, making a strict cutoff of exactly 90 days occasionally flag a role that's actually in active, if infrequent, use. Additionally, usage tracked in one region does not always reflect activity that occurred through another region's endpoint for the same global IAM role, which can make an actively-used role appear falsely idle depending on which region's data the check happens to observe. Mature versions build in a small buffer and cross-check with CloudTrail AssumeRole events for roles near the threshold boundary before treating them as confirmed unused.
Code Example
import boto3
import os
import json
import requests
from datetime import datetime, timedelta, timezone
# --- Configuration ---
INACTIVE_THRESHOLD_DAYS = 90
IGNORE_SERVICE_LINKED_ROLES = True
# Name of the secret in AWS Secrets Manager that holds the Slack webhook URL
SLACK_SECRET_NAME = os.environ.get('SLACK_SECRET_NAME', 'slack/webhook/iam-audit-alerts')
def get_slack_webhook_url():
"""Retrieves the Slack webhook URL from AWS Secrets Manager."""
try:
secrets_client = boto3.client('secretsmanager')
secret_value = secrets_client.get_secret_value(SecretId=SLACK_SECRET_NAME)
return json.loads(secret_value['SecretString'])['SLACK_WEBHOOK_URL']
except Exception as e:
print(f"Error retrieving Slack secret '{SLACK_SECRET_NAME}': {e}")
return None
def send_slack_notification(message, webhook_url):
"""Sends a message to a Slack webhook URL."""
if not webhook_url:
print("Slack webhook URL not available. Skipping notification.")
return
try:
response = requests.post(webhook_url, json={'text': message}, timeout=10)
response.raise_for_status()
print("Successfully sent Slack notification.")
except requests.exceptions.RequestException as e:
print(f"Error sending Slack notification: {e}")
def find_unused_iam_roles():
"""Scans for IAM roles that have not been used recently and returns a list of findings."""
iam = boto3.client('iam')
cutoff_date = datetime.now(timezone.utc) - timedelta(days=INACTIVE_THRESHOLD_DAYS)
unused_roles = []
print(f"Searching for IAM roles unused since {cutoff_date.date()}...")
paginator = iam.get_paginator('list_roles')
for page in paginator.paginate():
for role in page['Roles']:
if IGNORE_SERVICE_LINKED_ROLES and role['Path'].startswith('/aws-service-role/'):
continue
role_details = iam.get_role(RoleName=role['RoleName'])['Role']
last_used = role_details.get('RoleLastUsed')
if not last_used or not last_used.get('LastUsedDate'):
if role['CreateDate'] < cutoff_date:
finding = f"- [NEVER USED] Role: `{role['RoleName']}` (Created: {role['CreateDate'].date()})"
print(finding)
unused_roles.append(finding)
elif last_used['LastUsedDate'] < cutoff_date:
finding = f"- [INACTIVE] Role: `{role['RoleName']}` (Last Used: {last_used['LastUsedDate'].date()})"
print(finding)
unused_roles.append(finding)
return unused_roles
def lambda_handler(event, context):
"""Lambda entry point to find unused roles and send a report."""
unused_roles = find_unused_iam_roles()
webhook_url = get_slack_webhook_url()
if not unused_roles:
message = "IAM role scan complete. No unused roles found matching the criteria."
print(message)
# Optionally, send a success notification
# send_slack_notification(message, webhook_url)
else:
summary_message = f"*IAM Security Report: Found {len(unused_roles)} Unused Roles*\n\n"
summary_message += "\n".join(unused_roles)
send_slack_notification(summary_message, webhook_url)
return {
'statusCode': 200,
'body': json.dumps(f'Scan complete. Found {len(unused_roles)} unused roles.')
}Interview Tip
A junior engineer typically checks RoleLastUsed and calls it done, but for a senior or security-focused role, the interviewer is actually looking for you to know that this field can be entirely absent for never-used roles, requires a fallback to CreateDate, and is not updated in real time, since AWS documents a reporting lag of hours, which means a strict cutoff can produce false positives right at the threshold boundary. Strong answers also mention fetching the Slack webhook from Secrets Manager instead of hardcoding it, scoping the Lambda execution role narrowly to exactly the one secret and IAM read actions it needs, and treating a failed Slack notification as itself a monitoring gap worth alerting on, since a security report that silently fails to deliver provides a false sense of coverage.
◈ Architecture Diagram
┌────────┐ list ┌──────┐ check ┌───────────┐ notify ┌───────┐
│ IAM │─────▶│ Roles│──────▶│LastUsed>90│───────▶│ Slack │
└────────┘ └──────┘ └───────────┘ └───────┘
↑
Secrets Manager💬 Comments
Quick Answer
Use the GitHub REST API via requests to paginate through all branches, fetch each branch's HEAD commit date, and flag branches older than a threshold that aren't in a protected list. Run in dry-run mode first, since branch deletion in Git is effectively irreversible unless someone has a local clone with that branch still checked out.
Detailed Answer
This is like a librarian periodically clearing out books from the "browsing table" that haven't been picked up in months, returning them to the shelves, except a few books on that table have a permanent "do not reshelve" sign on them, the protected branches, and the librarian double-checks the due date on every other book individually rather than assuming the whole table is fair game just because it looks cluttered.
This script exists because feature branches accumulate relentlessly in any actively developed repository — branches for abandoned experiments, branches merged via squash-merge which can leave the original branch technically "unmerged" from Git's perspective even though its changes already landed on main, and branches from contributors who left the team all pile up, making it hard to tell which branches represent real, active work versus digital clutter. Automating cleanup matters because manual branch hygiene never happens consistently — nobody wants to spend their afternoon guessing which of 400 branches are safe to delete.
The script authenticates with a GITHUB_TOKEN, paginates through the branches endpoint, again, pagination matters because GitHub caps each page at 100 branches and any repo with real history will exceed that, and for each non-protected branch, follows the branch's commit URL to fetch the actual commit object and read the committer's date — this is necessary because the branches endpoint itself doesn't directly expose a clean "last activity" timestamp without this follow-up call. Branches in the PROTECTED_BRANCHES set are skipped outright regardless of age, and the deletion step itself, a DELETE to the git refs endpoint, only executes when DRY_RUN is explicitly set to False.
In production, this typically runs as a scheduled GitHub Action with the token stored as a repository secret, generating a dry-run report for review before anyone flips it to actually delete branches. What matters operationally: rate limiting, since GitHub's REST API enforces request quotas, and a repository with thousands of branches making one extra API call per branch to fetch commit dates can approach those limits, so production versions often batch or cache commit lookups, and add explicit handling for rate-limit responses rather than letting the whole run fail partway through with no record of what was already checked.
The non-obvious gotcha is that "stale" as measured by commit date doesn't account for branches that are intentionally long-lived without frequent commits — a release branch that's cut once and then only merged from occasionally, or a long-running feature-flagged branch that's deliberately kept in sync but rarely committed to directly, can look identical to an abandoned branch by this metric alone. Teams that rely purely on commit age without also checking whether a branch has an open, active pull request end up deleting branches that were very much still in use, just not recently committed to — a mature version of this script cross-checks for open pull requests referencing the branch before ever considering it a deletion candidate.
Code Example
import os
import requests
from datetime import datetime, timedelta, timezone
# --- Configuration ---
# GitHub token with repo scope. Store as an environment variable.
GITHUB_TOKEN = os.environ.get('GITHUB_TOKEN')
# Target repository in 'owner/repo' format
REPO = 'your-org/your-repo'
# Branches older than this will be considered stale
INACTIVE_DAYS_THRESHOLD = 90
# Branches to protect from deletion, regardless of age
PROTECTED_BRANCHES = {'main', 'develop', 'release'}
# Set to False to actually delete branches
DRY_RUN = True
API_URL = f'https://api.github.com/repos/{REPO}'
HEADERS = {
'Authorization': f'token {GITHUB_TOKEN}',
'Accept': 'application/vnd.github.v3+json'
}
def find_stale_branches():
"""Finds and optionally deletes stale branches in a GitHub repo."""
if not GITHUB_TOKEN:
print("[ERROR] GITHUB_TOKEN environment variable not set.")
return
cutoff_date = datetime.now(timezone.utc) - timedelta(days=INACTIVE_DAYS_THRESHOLD)
stale_branches = []
print(f"Scanning '{REPO}' for branches inactive since {cutoff_date.date()}...")
try:
# Paginate through all branches
page = 1
while True:
response = requests.get(f'{API_URL}/branches', headers=HEADERS, params={'per_page': 100, 'page': page})
response.raise_for_status()
branches = response.json()
if not branches:
break
for branch in branches:
branch_name = branch['name']
if branch_name in PROTECTED_BRANCHES:
continue
# Get the commit date of the branch's head
commit_url = branch['commit']['url']
commit_response = requests.get(commit_url, headers=HEADERS)
commit_response.raise_for_status()
commit_date_str = commit_response.json()['commit']['committer']['date']
last_commit_date = datetime.fromisoformat(commit_date_str.replace('Z', '+00:00'))
if last_commit_date < cutoff_date:
print(f"- [STALE] Branch: {branch_name} (Last commit: {last_commit_date.date()})")
stale_branches.append(branch_name)
page += 1
if not stale_branches:
print("No stale branches found.")
elif not DRY_RUN:
print("\n--- Deleting Stale Branches ---")
for branch_name in stale_branches:
try:
print(f"Deleting branch '{branch_name}'...")
delete_url = f'{API_URL}/git/refs/heads/{branch_name}'
delete_response = requests.delete(delete_url, headers=HEADERS)
delete_response.raise_for_status()
print(f" Successfully deleted {branch_name}.")
except requests.exceptions.RequestException as e:
print(f" Error deleting {branch_name}: {e}")
except requests.exceptions.RequestException as e:
print(f"An error occurred communicating with GitHub: {e}")
if __name__ == "__main__":
find_stale_branches()Interview Tip
A junior engineer typically checks commit age and calls a branch stale, but for a senior role, the interviewer is actually looking for you to recognize that commit recency alone is a weak signal — a branch can be old by that metric while still being referenced by an open, active pull request, and deleting it would break someone's in-progress work. Strong answers mention cross-checking for open PRs before treating a branch as a deletion candidate, GitHub API rate-limit handling for repositories with large branch counts since this script makes one extra API call per branch to fetch commit dates, and always running dry-run first given that branch deletion is effectively irreversible for anyone without a local copy still checked out.
◈ Architecture Diagram
┌────────┐ list ┌────────┐ commit date ┌──────────┐
│ GitHub │─────▶│Branches│────────────▶│ age > 90?│
└────────┘ └────────┘ └────┌─────┘
│ yes
open PR check
✓ skip │ ✗ delete💬 Comments
Context
A fintech platform team managed 184 AWS accounts across 22 regions with roughly 38,000 EBS volumes and 600 engineers creating short-lived environments. Monthly storage waste crossed $74,000 because unattached volumes from failed test environments were not consistently tagged or deleted.
Problem
The team originally relied on a weekly spreadsheet exported from the AWS console. It missed accounts where the operator lacked permissions, did not include enough tag context to identify owners, and produced a long manual cleanup queue that nobody trusted. Several volumes looked unused but belonged to disaster recovery drills or database restore tests, so engineers were afraid to delete anything at scale. Cost alerts kept firing, but the obvious fix of deleting every unattached volume older than 30 days was too risky because the organization had inconsistent tagging and no reliable rollback path for snapshots.
Solution
The team built a Python workflow with boto3 that assumed a read-only inventory role in every account, paginated describe_volumes in each enabled region, and wrote normalized inventory records to S3 with account, region, volume ID, tags, attachment history, size, encryption state, and discovered owner. The cleanup phase was separate from discovery. It first created a dry-run plan, grouped candidates by owning team, and required a change ticket for execution. Volumes were never deleted immediately. The workflow first created a final snapshot, tagged the snapshot with the original volume metadata, then deleted the volume only if the candidate still matched policy after a second scan. Botocore retry configuration and a 16-worker concurrency limit kept the collector below API throttle thresholds. AccessDenied and opt-in-region errors were recorded as first-class results so the report distinguished real zero waste from blind spots.
Commands
# Run inventory only so no resources are changed python -m platform_automation.ebs_cleanup inventory --accounts accounts.csv --regions us-east-1,us-west-2 --output s3://platform-inventory/ebs/2026-06-26/
# Produce a dry-run cleanup plan for human review python -m platform_automation.ebs_cleanup plan --inventory s3://platform-inventory/ebs/2026-06-26/ --min-age-days 45 --require-tag Owner --plan-file ebs-cleanup-plan.json
# Execute an approved batch with snapshot-before-delete protection python -m platform_automation.ebs_cleanup execute --plan-file ebs-cleanup-plan.json --change-id CHG-48291 --batch-size 25 --snapshot-before-delete
Outcome
Monthly unattached EBS waste dropped from $74,000 to $11,500 in eight weeks. Cleanup review time fell from 14 engineer-hours per week to under 2 hours, and no restore requests were needed after the snapshot-first policy shipped.
Lessons Learned
The most important design choice was separating inventory, planning, and mutation. The team also learned that AccessDenied must be reported loudly; otherwise executives read missing data as clean data.
◈ Architecture Diagram
┌────────┐ assume ┌────────┐ scan ┌────────┐ plan ┌────────┐
│Accounts│───────▶│ STS │─────▶│ EBS │─────▶│Dry Run │
└────────┘ └────────┘ └────────┘ └────────┘
│ approve
▼
┌────────┐
│Snapshot│
└────────┘
│
▼
┌────────┐
│ Delete │
└────────┘💬 Comments
Context
A streaming company ran 42 Kubernetes clusters with 7,800 namespaces and about 96,000 pods during peak release windows. Platform on-call was getting noisy alerts from Prometheus but lacked a single incident payload that included pod, container, image, node, namespace owner, and recent restart count.
Problem
The first version was a cron job that ran kubectl get pods -A every minute and posted any CrashLoopBackOff string to Slack. It was noisy, expensive, and late. During large rollouts, the script hammered API servers with full-list calls and generated hundreds of duplicate messages for the same failing pod. It also missed short-lived init container failures and could not distinguish a developer namespace from a tier-zero payments service. On-call engineers still had to run manual kubectl commands to learn which image was deployed, which node was involved, and whether the restart count was increasing.
Solution
The replacement used the Kubernetes Python client and the list-watch pattern. One watcher per cluster streamed pod changes across all namespaces using resourceVersion and a five-minute watch timeout. The handler inspected containerStatuses and initContainerStatuses for waiting.reason equal to CrashLoopBackOff, built an idempotency key from namespace, pod, container, image, and restart count, and enriched each event with namespace owner labels from a cached namespace list. Instead of posting directly to chat, the watcher published structured JSON events to a central incident router. The router applied severity rules: tier-zero services paged immediately after three restarts in ten minutes, while sandbox namespaces were grouped into hourly digests. RBAC granted list/watch on pods and namespaces only, and the watcher exported metrics for reconnects, API errors, dedupe drops, and emitted incidents.
Commands
# Deploy the watcher with a narrow service account kubectl apply -f deploy/pydevops-crashloop-watcher.yaml
# Verify RBAC can list and watch pods but cannot mutate workloads kubectl auth can-i watch pods --as system:serviceaccount:platform:crashloop-watcher -A
# Check watcher health and reconnect metrics kubectl logs deploy/crashloop-watcher -n platform --tail=100
Outcome
Mean time to identify the failing image dropped from 11 minutes to under 90 seconds. Duplicate chat alerts fell by 87 percent, and API server list traffic from the old cron job disappeared during release windows.
Lessons Learned
The team learned that CrashLoopBackOff is container state, not pod phase, so phase-based filters are incomplete. They also learned to route events through a policy layer instead of hard-coding paging behavior into the watcher.
◈ Architecture Diagram
┌───────┐ watch ┌────────┐ enrich ┌────────┐ route ┌──────┐
│Cluster│──────▶│Watcher │───────▶│ Event │──────▶│ Pager│
└───────┘ └────────┘ └────────┘ └──────┘
│ metrics │ dedupe
▼ ▼
┌────────┐ ┌────────┐
│Prom │ │Cache │
└────────┘ └────────┘💬 Comments
Context
A SaaS company deployed a shared authentication service to 31 regions and 430 enterprise tenant edges. Each production rollout needed to verify health endpoints, token issuance, JWKS freshness, and latency from multiple network paths before shifting global traffic.
Problem
The original deployment gate was sequential and took 38 minutes, so release managers bypassed it during urgent fixes. A rewrite that launched all checks concurrently finished quickly but created connection spikes against DNS resolvers and regional load balancers. Some checks failed because the checker overloaded the network path, not because the service was unhealthy. The team needed a gate that was fast enough to keep in the release workflow but disciplined enough not to behave like a load test.
Solution
The platform team wrote an asyncio-based checker using aiohttp, TaskGroup, per-request timeouts, and a semaphore to cap in-flight requests. Checks were grouped by region and tenant tier, with separate concurrency limits for public edge endpoints and internal control-plane endpoints. Each target produced a structured HealthResult with URL, check type, HTTP status, latency, error class, and retry count. The gate used jittered retries only for network timeouts and HTTP 429 or 503 responses. Authentication checks validated both status code and response body shape, while latency checks used percentile thresholds per region rather than a single global value. The script emitted JSON for CI, Prometheus summary metrics for dashboards, and a compact failure table for release managers.
Commands
# Run the deployment gate in dry-run reporting mode python -m platform_automation.deploy_gate --manifest rollout-targets.yaml --concurrency 250 --report report.json
# Enforce regional p95 latency and auth correctness before traffic shift python -m platform_automation.deploy_gate --manifest rollout-targets.yaml --concurrency 180 --fail-on-critical --change-id REL-2026-06-26
# Summarize failed checks for release review python -m platform_automation.deploy_gate summarize --report report.json --only-failures
Outcome
Gate runtime dropped from 38 minutes to 94 seconds while keeping outbound concurrency below the network team's approved ceiling. False deployment blocks fell by 63 percent after region-specific thresholds and retry classification were added.
Lessons Learned
Concurrency is a production control, not just a performance setting. The most useful output for humans was not the full JSON file, but a short table grouped by region, tenant tier, and failure class.
◈ Architecture Diagram
┌────────┐ targets ┌────────┐ limit ┌────────┐
│Manifest│────────▶│TaskGrp │──────▶│Checks │
└────────┘ └────────┘ └────────┘
│ │
▼ ▼
┌────────┐ ┌────────┐
│Timeouts│ │Report │
└────────┘ └────────┘💬 Comments
Context
A marketplace company had 70 small Python scripts for deploys, inventory, DNS checks, certificate checks, and cleanup tasks. The scripts ran from laptops, Jenkins jobs, cron hosts, and Kubernetes CronJobs, each with different Python versions and dependencies.
Problem
Operational behavior changed depending on where a script ran. One cron host had an old requests version, a developer laptop had a newer boto3 version, and CI installed dependencies without hashes. A certificate-renewal script passed tests locally but failed in cron because PATH and environment variables were different. Nobody could answer which version of a script modified production DNS during an incident. The scripts were useful, but the lack of packaging made them unreliable and hard to audit.
Solution
The team consolidated common automation into a Python package with pyproject.toml, src layout, pinned dependencies, console_scripts entry points, and structured logging. Each tool had pure policy functions covered by unit tests and provider clients that could be faked. Integration tests required explicit sandbox credentials and were skipped by default in pull requests. The package was built into a slim Python container image running as a non-root user. Jenkins, Kubernetes CronJobs, and local developers all used the same image tag for production actions. Every command wrote its package version, git SHA, actor, target environment, dry-run flag, and change ID to the audit log.
Commands
# Run fast unit tests for policy logic and parsers python -m pytest tests/unit -q
# Build the versioned automation image used by CI and CronJobs docker build -t registry.company.com/platform-automation:0.9.4 .
# Run a production DNS check with explicit version and dry-run behavior docker run --rm registry.company.com/platform-automation:0.9.4 dns-check --zone prod.example.com --dry-run
Outcome
Script-related production failures dropped from 5 incidents per quarter to 1 minor incident in two quarters. New automation onboarding time dropped from several days to one afternoon because teams reused the package template and test harness.
Lessons Learned
Packaging was not bureaucracy; it was operational safety. The team also learned that every automation command should identify its own version in logs before an incident forces people to reconstruct it.
◈ Architecture Diagram
┌──────┐ package ┌──────┐ image ┌──────┐ run ┌──────┐ │Code │────────▶│Tests │──────▶│Image │────▶│Jobs │ └──────┘ └──────┘ └──────┘ └──────┘ │ │ ▼ ▼ ┌──────┐ ┌──────┐ │Policy│ │Audit │ └──────┘ └──────┘
💬 Comments