kubestronaut-sim/sim/desktop.py

"""The exam's remote desktop: an XFCE session in a container, shown in the browser over noVNC.

It is the "base" machine of a session. It holds no kubeconfig - a task names the host to `ssh` to -
and its Firefox only opens the documentation the exam allows.
"""
import json

from sim import envs

NAME = "ksim-desktop"
PORT = 6080

# Documentation each exam lets you open, as host names. From the Linux Foundation's "Resources
# allowed" page (docs.linuxfoundation.org/tc-docs/certification/certification-resources-allowed,
# read 2026-10-12); Firefox in the desktop refuses everything else (KSIM_OPEN_WEB=1 lifts that).
# For CKS that page no longer lists Trivy or AppArmor: on the day, `trivy --help` is the manual.
DOCS = {
    "cka": {"Kubernetes docs": "https://kubernetes.io/docs/home/", "Kubernetes blog": "https://kubernetes.io/blog/",
            "Helm docs": "https://helm.sh/docs/", "Gateway API": "https://gateway-api.sigs.k8s.io/"},
    "ckad": {"Kubernetes docs": "https://kubernetes.io/docs/home/", "Kubernetes blog": "https://kubernetes.io/blog/",
             "Helm docs": "https://helm.sh/docs/"},
    "cks": {"Kubernetes docs": "https://kubernetes.io/docs/home/", "Kubernetes blog": "https://kubernetes.io/blog/",
            "Falco": "https://falco.org/docs/", "bom": "https://kubernetes-sigs.github.io/bom/cli-reference/",
            "etcd": "https://etcd.io/docs/",
            "ingress-nginx": "https://kubernetes.github.io/ingress-nginx/user-guide/nginx-configuration/",
            "Cilium": "https://docs.cilium.io/en/stable/", "Istio": "https://istio.io/latest/docs/"},
}


def url():
    return f"http://127.0.0.1:{PORT}/novnc/vnc.html?autoconnect=1&resize=remote&reconnect=1&reconnect_delay=1500&show_dot=1"


def running():
    rc, out = envs.sh([envs.RUNTIME, "inspect", "-f", "{{.State.Running}}", NAME])
    return rc == 0 and out.strip() == "true"


def published_port():
    rc, out = envs.sh([envs.RUNTIME, "port", NAME, "6080/tcp"])
    return int(out.rsplit(":", 1)[1]) if rc == 0 and ":" in out else None


def up(fresh=False):
    """Start the desktop on PORT; `fresh` replaces a running one, so every session begins on a
    clean one."""
    if running() and not fresh and published_port() == PORT:
        return
    if not envs.image_exists(envs.DESKTOP_IMAGE):
        raise envs.EnvError("the desktop image is not built yet - run:  ./ksim build")
    envs.ensure_network()
    envs.sh([envs.RUNTIME, "rm", "-f", NAME])
    # VNC has no password, so the port is published on loopback only: anyone who can reach it has
    # a shell that can ssh to every exam host.
    rc, out = envs.sh([envs.RUNTIME, "run", "-d", "--name", NAME, "--hostname", "base", "--network", envs.NETWORK,
                       "--shm-size=1g", "-p", f"127.0.0.1:{PORT}:6080", envs.DESKTOP_IMAGE])
    if rc != 0:
        raise envs.EnvError(f"could not start the desktop:\n{out[-600:]}")


def down():
    envs.sh([envs.RUNTIME, "rm", "-f", NAME])


def _exec(script, user="candidate"):
    return envs.sh([envs.RUNTIME, "exec", "-i", "-u", user, NAME, "bash", "-s"], stdin=script, timeout=60)


def _policy(track_id, open_web):
    docs = DOCS.get(track_id, DOCS["cka"])
    hosts = sorted({u.split("/")[2] for u in docs.values()})
    pol = {
        "DisableTelemetry": True, "DisableFirefoxStudies": True, "DisablePocket": True, "DisableAppUpdate": True,
        "DontCheckDefaultBrowser": True, "NoDefaultBookmarks": True, "DisplayBookmarksToolbar": "always",
        "OverrideFirstRunPage": "", "OverridePostUpdatePage": "",
        "UserMessaging": {"SkipOnboarding": True, "WhatsNew": False, "ExtensionRecommendations": False,
                          "FeatureRecommendations": False, "MoreFromMozilla": False},
        "Homepage": {"URL": next(iter(docs.values())), "StartPage": "homepage", "Locked": False},
        "ManagedBookmarks": [{"toplevel_name": "Allowed documentation"}] + [{"name": n, "url": u} for n, u in docs.items()],
    }
    if not open_web:
        pol["WebsiteFilter"] = {"Block": ["<all_urls>"],
                                "Exceptions": [p for h in hosts for p in (f"*://{h}/*", f"*://*.{h}/*")]}
    return json.dumps({"policies": pol}, indent=1)


def configure(track_id, hosts, open_web=False):
    """Point the desktop at a session: `ssh <host>` for each of its tasks, and the track's docs."""
    priv, _ = envs.ensure_keys()
    conf = "".join(f"Host {h}\n  HostName {h}-control-plane\n" for h in hosts) + (
        "Host *\n  User candidate\n  IdentityFile ~/.ssh/id_ed25519\n"
        "  StrictHostKeyChecking no\n  UserKnownHostsFile /dev/null\n  LogLevel ERROR\n")
    rc, out = _exec("set -e\ninstall -d -m700 ~/.ssh\n"
                    f"cat > ~/.ssh/id_ed25519 <<'KSIM'\n{priv}KSIM\n"
                    f"cat > ~/.ssh/config <<'KSIM'\n{conf}KSIM\n"
                    "chmod 600 ~/.ssh/id_ed25519 ~/.ssh/config\n")
    if rc != 0:
        raise envs.EnvError(f"could not configure ssh on the desktop:\n{out[-400:]}")
    _exec("install -d /usr/lib/firefox-esr/distribution\n"
          f"cat > /usr/lib/firefox-esr/distribution/policies.json <<'KSIM'\n{_policy(track_id, open_web)}\nKSIM\n", user="root")


def clip(text):
    """Put text on the desktop's clipboard - what clicking a highlighted value in a task does."""
    for sel in ("clipboard", "primary"):
        envs.sh([envs.RUNTIME, "exec", "-i", NAME, "sh", "-c",
                 f"DISPLAY=:1 xclip -selection {sel} -i >/dev/null 2>&1"], stdin=text, timeout=10)