Harden Kubernetes discovery and security setting output

Preserve explicit security settings and bound Kubernetes discovery calls.

linPEAS/builder/linpeas_parts/2_container/6_Kubernetes_surfaces.sh

@@ -364,8 +364,9 @@ k8s_direct_api_ready() {
 }
 
 k8s_sa_api_get() {
+  # Ignore curlrc: inherited tracing or credentials must not affect discovery.
   printf 'header = "Authorization: Bearer %s"\n' "$k8s_direct_token" |
-    curl --config - -fsS --connect-timeout 2 --max-time 5 \
+    curl -q --config - -fsS --connect-timeout 2 --max-time 5 \
       --cacert "$k8s_direct_ca" -H 'Accept: application/json' \
       "$k8s_direct_base$1" 2>/dev/null
 }
@@ -378,7 +379,7 @@ k8s_scan_sa_api() {
   k8s_sa_api_get "/api/v1/namespaces/$k8s_namespace/pods?limit=100" |
     jq -r '.items[]? | (
       "\(.metadata.name) sa=\(.spec.serviceAccountName // "default") hostPID=\(.spec.hostPID // false) hostIPC=\(.spec.hostIPC // false) hostNetwork=\(.spec.hostNetwork // false)",
-      (.spec.containers[]? | "  container \(.name) privileged=\(.securityContext.privileged // false) allowPE=\(.securityContext.allowPrivilegeEscalation // "default") caps=\((.securityContext.capabilities.add // []) | join(","))"),
+      (.spec.containers[]? | "  container \(.name) privileged=\(.securityContext.privileged // false) allowPE=\((.securityContext.allowPrivilegeEscalation | if . == null then "default" else . end)) caps=\((.securityContext.capabilities.add // []) | join(","))"),
       (.spec.volumes[]? | select(.hostPath.path != null) | "  hostPath \(.hostPath.path)")
     )' 2>/dev/null | head -n 100 | sed 's/^/    pod: /'
   for k8s_file in services serviceaccounts secrets; do
@@ -391,59 +392,66 @@ k8s_scan_sa_api() {
   k8s_direct_token=''
 }
 
+k8s_kubectl() {
+  # Request timeouts do not bound credential plugins or all paginated requests.
+  # Skip kubectl when a process-wide deadline cannot be enforced.
+  command -v timeout >/dev/null 2>&1 || return 1
+  timeout -s KILL 5 kubectl "$@"
+}
+
 k8s_scan_kubectl() {
   if ! command -v kubectl >/dev/null 2>&1; then
     [ "$EXTRA_CHECKS" ] && k8s_scan_sa_api
     return 0
   fi
   echo '  Local kubeconfig contexts:'
-  kubectl config get-contexts -o name 2>/dev/null | head -n 30 | sed 's/^/    /'
+  k8s_kubectl config get-contexts -o name 2>/dev/null | head -n 30 | sed 's/^/    /'
   [ "$EXTRA_CHECKS" ] || return 0
 
-  k8s_current_context="$(kubectl config current-context 2>/dev/null | head -n 1)"
+  k8s_current_context="$(k8s_kubectl config current-context 2>/dev/null | head -n 1)"
   if [ ! "$k8s_current_context" ] && k8s_direct_api_ready; then
     k8s_direct_token=''
     k8s_scan_sa_api
     return 0
   fi
-  k8s_context_name="$(kubectl config view --minify -o 'jsonpath={..namespace}' 2>/dev/null | head -c 200)"
+  k8s_context_name="$(k8s_kubectl config view --minify -o 'jsonpath={..namespace}' 2>/dev/null | head -c 200)"
   [ "$k8s_context_name" ] && k8s_namespace="$k8s_context_name"
   [ "$k8s_namespace" ] || k8s_namespace=default
 
   echo '  Current context authorization rules:'
-  kubectl --request-timeout=5s auth can-i --list -n "$k8s_namespace" \
+  k8s_kubectl --request-timeout=5s auth can-i --list -n "$k8s_namespace" \
     2>/dev/null | head -n 80
   echo '  Other configured context authorization rules (up to five):'
-  kubectl config get-contexts -o name 2>/dev/null | head -n 5 |
+  k8s_kubectl config get-contexts -o name 2>/dev/null | head -n 5 |
   while IFS= read -r k8s_context_name; do
     [ "$k8s_context_name" = "$k8s_current_context" ] && continue
     printf '    Context: %s\n' "$k8s_context_name"
-    kubectl --context="$k8s_context_name" --request-timeout=5s auth can-i --list \
+    k8s_kubectl --context="$k8s_context_name" --request-timeout=5s auth can-i --list \
       2>/dev/null | head -n 40 | sed 's/^/      /'
   done
   echo '  Accessible namespace names:'
-  kubectl --request-timeout=5s get namespaces -o name 2>/dev/null | head -n 40
+  k8s_kubectl --request-timeout=5s get namespaces -o name 2>/dev/null | head -n 40
   printf '  Accessible object names in namespace %s:\n' "$k8s_namespace"
-  kubectl --request-timeout=5s get pods,services,serviceaccounts,secrets \
+  k8s_kubectl --request-timeout=5s get pods,services,serviceaccounts,secrets \
     -n "$k8s_namespace" -o name 2>/dev/null | head -n 80
   echo '  Accessible node names:'
-  kubectl --request-timeout=5s get nodes -o name 2>/dev/null | head -n 40
+  k8s_kubectl --request-timeout=5s get nodes -o name 2>/dev/null | head -n 40
   echo '  Node-specific get nodes/proxy authorization (up to eight nodes):'
-  kubectl --request-timeout=5s get nodes -o name 2>/dev/null | head -n 8 |
+  k8s_kubectl --request-timeout=5s get nodes -o name 2>/dev/null | head -n 8 |
   while IFS= read -r k8s_node_name; do
     case "$k8s_node_name" in node/*|nodes/*) k8s_node_name="${k8s_node_name#*/}" ;; *) continue ;; esac
-    k8s_probe_status="$(kubectl --request-timeout=5s auth can-i get \
+    k8s_probe_status="$(k8s_kubectl --request-timeout=5s auth can-i get \
       "nodes/$k8s_node_name" --subresource=proxy 2>/dev/null)"
     case "$k8s_probe_status" in
       yes|no) printf '    %s: %s\n' "$k8s_node_name" "$k8s_probe_status" ;;
     esac
   done
   echo '  Pod service accounts, host namespaces and container security settings:'
-  kubectl --request-timeout=5s get pods -n "$k8s_namespace" \
+  k8s_kubectl --request-timeout=5s get pods -n "$k8s_namespace" \
     -o 'jsonpath={range .items[*]}{.metadata.name}{" sa="}{.spec.serviceAccountName}{" hostPID="}{.spec.hostPID}{" hostIPC="}{.spec.hostIPC}{" hostNetwork="}{.spec.hostNetwork}{" containers="}{range .spec.containers[*]}{.name}{"(privileged="}{.securityContext.privileged}{",allowPE="}{.securityContext.allowPrivilegeEscalation}{",caps="}{.securityContext.capabilities.add}{") "}{end}{"\n"}{end}' \
     2>/dev/null | head -n 60
   echo '  Pod hostPath sources:'
-  kubectl --request-timeout=5s get pods -n "$k8s_namespace" \
+  k8s_kubectl --request-timeout=5s get pods -n "$k8s_namespace" \
     -o 'jsonpath={range .items[*]}{.metadata.name}{": "}{range .spec.volumes[*]}{.hostPath.path}{" "}{end}{"\n"}{end}' \
     2>/dev/null | grep -E ': /' | head -n 60
 }
@@ -458,7 +466,7 @@ k8s_probe_kubelet() {
     [ "$k8s_probe_port" = 10255 ] && k8s_probe_scheme=http
     # No credentials are sent. An HTTP status only establishes reachability;
     # it does not prove kubelet authentication or authorization.
-    k8s_probe_status="$(curl -ksS --noproxy '*' --connect-timeout 1 --max-time 2 \
+    k8s_probe_status="$(curl -q -ksS --noproxy '*' --connect-timeout 1 --max-time 2 \
       -o /dev/null -w '%{http_code}' \
       "$k8s_probe_scheme://$k8s_probe_host:$k8s_probe_port/healthz" 2>/dev/null)"
     case "$k8s_probe_status" in
@@ -474,7 +482,7 @@ k8s_scan_kubelet_network() {
   echo '  Reachability of node kubelet ports (up to eight addresses; status only):'
   k8s_node_addresses=''
   if command -v kubectl >/dev/null 2>&1; then
-    k8s_node_addresses="$(kubectl --request-timeout=5s get nodes \
+    k8s_node_addresses="$(k8s_kubectl --request-timeout=5s get nodes \
       -o 'jsonpath={range .items[*]}{range .status.addresses[*]}{.address}{"\n"}{end}{end}' \
       2>/dev/null)"
   fi
@@ -500,14 +508,14 @@ k8s_scan_kops_storage() {
   command -v timeout >/dev/null 2>&1 || return
   if command -v aws >/dev/null 2>&1; then
     echo '  Accessible S3 buckets (up to 10) and candidate kOps Secret keys (first 100 objects each):'
-    AWS_MAX_ATTEMPTS=1 timeout 12 aws s3api list-buckets \
+    AWS_MAX_ATTEMPTS=1 timeout -s KILL 12 aws s3api list-buckets \
       --max-items 10 --query 'Buckets[].Name' --output text \
       --cli-connect-timeout 2 --cli-read-timeout 5 \
       2>/dev/null | tr '\t' '\n' | head -n 10 |
     while IFS= read -r k8s_aws_bucket; do
       case "$k8s_aws_bucket" in ''|*[!a-z0-9.-]*) continue ;; esac
       printf '    s3://%s\n' "$k8s_aws_bucket"
-      AWS_MAX_ATTEMPTS=1 timeout 12 aws s3api list-objects-v2 \
+      AWS_MAX_ATTEMPTS=1 timeout -s KILL 12 aws s3api list-objects-v2 \
         --bucket "$k8s_aws_bucket" --max-items 100 \
         --query 'Contents[].Key' --output text \
         --cli-connect-timeout 2 --cli-read-timeout 5 \
@@ -517,13 +525,13 @@ k8s_scan_kops_storage() {
   fi
   if command -v gcloud >/dev/null 2>&1; then
     echo '  Accessible GCS buckets (up to 10) and candidate kOps Secret objects (first 100 each):'
-    timeout 12 gcloud --quiet storage buckets list --limit=10 \
+    timeout -s KILL 12 gcloud --quiet storage buckets list --limit=10 \
       --format='value(name)' 2>/dev/null | head -n 10 |
     while IFS= read -r k8s_gcs_bucket; do
       k8s_gcs_bucket="${k8s_gcs_bucket#gs://}"
       case "$k8s_gcs_bucket" in ''|*[!a-z0-9._-]*) continue ;; esac
       printf '    gs://%s\n' "$k8s_gcs_bucket"
-      timeout 12 gcloud --quiet storage objects list \
+      timeout -s KILL 12 gcloud --quiet storage objects list \
         "gs://$k8s_gcs_bucket/**" --limit=100 --format='value(name)' \
         2>/dev/null | grep -E '(^|/)secrets/' | head -n 20 |
         sed 's/^/      candidate object: /'

linPEAS/tests/test_kubernetes_surfaces.py

@@ -1,6 +1,9 @@
+import json
 import os
+import shutil
 import subprocess
 import tempfile
+import time
 import unittest
 from pathlib import Path
 
@@ -25,6 +28,7 @@ class KubernetesSurfacesTests(unittest.TestCase):
             capture_output=True,
             text=True,
             env=env,
+            timeout=15,
         ).stdout
 
     def test_projected_and_legacy_volumes_report_paths_without_values(self):
@@ -237,6 +241,54 @@ class KubernetesSurfacesTests(unittest.TestCase):
             self.assertNotIn("SENSITIVE_SECRET_VALUE", output)
             self.assertNotIn("SENSITIVE_TOKEN_VALUE", calls.read_text())
             self.assertTrue(all("--max-time 5" in call for call in calls.read_text().splitlines()))
+            self.assertTrue(all(call.startswith("-q ") for call in calls.read_text().splitlines()))
+
+    def test_service_account_api_preserves_explicit_false_security_setting(self):
+        with tempfile.TemporaryDirectory() as root:
+            fixture = Path(root) / "pods.json"
+            fixture.write_text(json.dumps({"items": [{
+                "metadata": {"name": "example"},
+                "spec": {"containers": [
+                    {"name": "restricted", "securityContext": {"allowPrivilegeEscalation": False}},
+                    {"name": "allowed", "securityContext": {"allowPrivilegeEscalation": True}},
+                    {"name": "unspecified"},
+                ]},
+            }]}))
+            output = self.run_shell(
+                'fixture="$1"; '
+                'k8s_direct_api_ready() { return 0; }; '
+                'k8s_sa_api_get() { case "$1" in '
+                '*/pods\\?*) cat "$fixture" ;; '
+                '*) echo \'{"items":[]}\' ;; esac; }; '
+                'k8s_scan_sa_api',
+                fixture,
+            )
+            self.assertIn("container restricted privileged=false allowPE=false", output)
+            self.assertIn("container allowed privileged=false allowPE=true", output)
+            self.assertIn("container unspecified privileged=false allowPE=default", output)
+
+    @unittest.skipUnless(shutil.which("timeout"), "timeout is required")
+    def test_kubectl_has_process_deadline_not_only_request_timeout(self):
+        with tempfile.TemporaryDirectory() as root:
+            kubectl = Path(root) / "kubectl"
+            kubectl.write_text("#!/bin/sh\nsleep 30\n")
+            kubectl.chmod(0o755)
+            env = os.environ.copy()
+            env["PATH"] = f"{root}:{env['PATH']}"
+            started = time.monotonic()
+            output = self.run_shell(
+                "k8s_kubectl config current-context || echo bounded", env=env
+            )
+            self.assertEqual("bounded\n", output)
+            self.assertLess(time.monotonic() - started, 10)
+
+    def test_kubectl_is_skipped_without_deadline_tool(self):
+        output = self.run_shell(
+            'command() { return 1; }; '
+            'kubectl() { echo SHOULD_NOT_RUN; }; '
+            'k8s_kubectl config current-context || echo skipped'
+        )
+        self.assertEqual("skipped\n", output)
 
     def test_escape_capability_bits_are_decoded(self):
         result = self.run_shell(
@@ -296,12 +348,17 @@ class KubernetesSurfacesTests(unittest.TestCase):
             self.assertEqual(2, len(calls))
             self.assertTrue(all("--max-time 2" in call for call in calls))
             self.assertTrue(all("Authorization" not in call for call in calls))
+            self.assertTrue(all(call.startswith("-q ") for call in calls))
 
     def test_kops_storage_lists_candidate_names_without_reading_objects(self):
         with tempfile.TemporaryDirectory() as root:
             calls = Path(root) / "cloud-calls"
             timeout = Path(root) / "timeout"
-            timeout.write_text('#!/bin/sh\nshift\n"$@"\n')
+            timeout.write_text(
+                '#!/bin/sh\n'
+                'printf "%s\\n" "$*" >> "$TIMEOUT_CALLS"\n'
+                '[ "$1" = -s ] && shift 2\nshift\n"$@"\n'
+            )
             timeout.chmod(0o755)
             aws = Path(root) / "aws"
             aws.write_text(
@@ -326,6 +383,8 @@ class KubernetesSurfacesTests(unittest.TestCase):
             env = os.environ.copy()
             env["PATH"] = f"{root}:{env['PATH']}"
             env["CLOUD_CALLS"] = str(calls)
+            deadlines = Path(root) / "timeout-calls"
+            env["TIMEOUT_CALLS"] = str(deadlines)
 
             output = self.run_shell("k8s_scan_kops_storage", env=env)
             self.assertIn("s3://kops-state", output)
@@ -334,6 +393,7 @@ class KubernetesSurfacesTests(unittest.TestCase):
             self.assertIn("candidate object: cluster/secrets/sa-token", output)
             self.assertNotIn("config.yaml", output)
             self.assertTrue(all("list" in call for call in calls.read_text().splitlines()))
+            self.assertTrue(all(call.startswith("-s KILL 12 ") for call in deadlines.read_text().splitlines()))
 
 
 if __name__ == "__main__":