Harden pwnshop challenge validation and runtimes

Author: zardus

runtime/workspace/agent/cmd/workspace-entrypoint/main.go

@@ -26,6 +26,8 @@ const workspaceProfileBin = "/run/workspace/profile/bin"
 const workspaceProfileScript = "/run/workspace/profile/etc/profile.d/99-pwn-workspace.sh"
 const workspaceUserRunDir = "/run/workspace/user"
 const workspaceServicesDir = "/run/workspace/user/services"
+const legacyDojoWorkspaceDir = "/run/dojo/sys/workspace"
+const legacyDojoPrivilegedFile = "/run/dojo/sys/workspace/privileged"
 const systemProfileScript = "/etc/profile.d/99-pwn-workspace.sh"
 const userShell = "/run/workspace/profile/bin/bash"
 
@@ -141,6 +143,9 @@ func prepareWorkspace(config workspaceConfig) error {
 	if err := setupRunDirectories(); err != nil {
 		return err
 	}
+	if err := setupLegacyDojoWorkspace(); err != nil {
+		return err
+	}
 	if err := setupSystemEnvironment(); err != nil {
 		return err
 	}
@@ -182,6 +187,20 @@ func setupRunDirectories() error {
 	return linkServiceDefinitions()
 }
 
+func setupLegacyDojoWorkspace() error {
+	privileged := "0\n"
+	if os.Getenv("PWN_WORKSPACE_PRIVILEGED") == "1" {
+		privileged = "1\n"
+	}
+	if err := os.MkdirAll(legacyDojoWorkspaceDir, 0755); err != nil {
+		return err
+	}
+	if err := os.WriteFile(legacyDojoPrivilegedFile, []byte(privileged), 0644); err != nil {
+		return err
+	}
+	return os.Chmod(legacyDojoPrivilegedFile, 0644)
+}
+
 func linkChallengeBin() error {
 	if _, err := os.Stat(challengeBin); errors.Is(err, os.ErrNotExist) {
 		return nil

tools/dojo/parse-dojo-yml

@@ -22,6 +22,7 @@ from pydantic import BaseModel, ConfigDict, Field, ValidationError, field_valida
 ID_REGEX = re.compile(r"^[a-z0-9-]{1,32}$")
 
 INHERIT_KEYS = ["privileged", "interfaces"]
+LOCAL_CHALLENGE_CONFIG_KEYS = {"allow_unsupported_tests"}
 
 
 class Visibility(BaseModel):
@@ -278,6 +279,11 @@ def main() -> int:
                     if resource.get("type") == "challenge":
                         challenge_id = resource.get("id")
                         challenge_data, challenge_description_md = challenge_reads.get(challenge_id, ({}, None))
+                        challenge_data = {
+                            key: value
+                            for key, value in challenge_data.items()
+                            if key not in LOCAL_CHALLENGE_CONFIG_KEYS
+                        }
 
                         resource = {**challenge_data, **resource, "type": "challenge", "id": challenge_id}
                         resource.setdefault(
@@ -300,6 +306,11 @@ def main() -> int:
 
                     challenge_id = challenge.get("id")
                     challenge_data, challenge_description_md = challenge_reads.get(challenge_id, ({}, None))
+                    challenge_data = {
+                        key: value
+                        for key, value in challenge_data.items()
+                        if key not in LOCAL_CHALLENGE_CONFIG_KEYS
+                    }
 
                     resource = {**challenge_data, **challenge, "type": "challenge", "id": challenge_id}
                     resource.setdefault(

tools/pwnshop/src/pwnshop/commands/test.py

@@ -9,6 +9,7 @@
 
 import click
 import requests
+import yaml
 from rich.progress import (
     BarColumn,
     MofNCompleteColumn,
@@ -23,6 +24,33 @@
 
 logger = logging.getLogger(__name__)
 
+UNSUPPORTED_TEST_EXIT_CODE = 77
+
+
+def _read_metadata(path: pathlib.Path) -> dict:
+    if not path.is_file():
+        return {}
+    metadata = yaml.safe_load(path.read_text()) or {}
+    return metadata if isinstance(metadata, dict) else {}
+
+
+def _allows_unsupported_tests(challenge_path: pathlib.Path) -> bool:
+    return (
+        _read_metadata(challenge_path / "challenge.yml").get("allow_unsupported_tests")
+        is True
+    )
+
+
+def _challenge_requires_solve(challenge_path: pathlib.Path) -> bool:
+    module = _read_metadata(challenge_path.parent / "module.yml")
+    for resource in module.get("resources", []):
+        if (
+            resource.get("type") == "challenge"
+            and resource.get("id") == challenge_path.name
+        ):
+            return resource.get("required", True) is not False
+    return True
+
 
 def run_workspace_command(
     workspace_url: str,
@@ -73,7 +101,7 @@ def run_workspace_command(
     default=None,
     help="Timeout in seconds for each individual test.",
 )
-@click.option("--require-solved", is_flag=True, help="Fail if any challenge is unsolved.")
+@click.option("--require-solved", is_flag=True, help="Fail if any required challenge is unsolved.")
 @click.option(
     "--log-failures",
     metavar="DIR",
@@ -110,22 +138,25 @@ def test_challenge(challenge_path):
         challenge_path = pathlib.Path(challenge_path)
         rendered = None
         try:
+            allow_unsupported_tests = _allows_unsupported_tests(challenge_path)
+            requires_solve = _challenge_requires_solve(challenge_path)
             rendered = lib.render_challenge(challenge_path)
             image_id = lib.build_challenge(challenge_path)
             tests = sorted(rendered.rglob("test*/test_*"))
             if not tests:
                 logger.warning("no tests found for %s", challenge_path)
-                return {"path": challenge_path, "tests": [], "solved": False}
+                return {"path": challenge_path, "tests": [], "solved": not requires_solve}
             logger.info("running %d test(s) for %s", len(tests), challenge_path)
             results = []
-            solved = False
+            solved = not requires_solve
             for test in tests:
                 test_name = test.relative_to(rendered)
                 logger.debug("running test %s in %s", test_name, challenge_path)
                 passed = False
                 last_output = ""
                 failed_attempt_outputs = []
                 for attempt in range(1, attempts + 1):
+                    test_unsupported = False
                     logger.debug("running test %s in %s (attempt %d/%d)", test_name, challenge_path, attempt, attempts)
                     with lib.run_challenge(challenge_path, image_id, volumes=[test]) as (
                         _container,
@@ -148,17 +179,21 @@ def test_challenge(challenge_path):
                                 last_output += e.stderr
                             passed = False
                         else:
-                            passed = run.returncode == 0
+                            test_unsupported = (
+                                allow_unsupported_tests
+                                and run.returncode == UNSUPPORTED_TEST_EXIT_CODE
+                            )
                             last_output = (run.stdout or "") + (run.stderr or "")
+                            passed = run.returncode == 0 or test_unsupported
                             logger.debug(
                                 "test %s %s (rc=%d, attempt %d/%d)",
                                 test_name,
-                                "PASSED" if passed else "FAILED",
+                                "UNSUPPORTED" if test_unsupported else "PASSED" if passed else "FAILED",
                                 run.returncode,
                                 attempt,
                                 attempts,
                             )
-                    solved = solved or flag in last_output
+                    solved = solved or test_unsupported or flag in last_output
                     if passed:
                         if attempt > 1:
                             logger.info(
@@ -217,7 +252,8 @@ def test_challenge(challenge_path):
                 else:
                     console.print(f"[red]FAIL[/] {challenge}: {error}")
             elif not tests:
-                unsolved.add(challenge)
+                if not solved:
+                    unsolved.add(challenge)
                 passed_count += 1
             else:
                 for test_path, passed, output in tests:

tools/pwnshop/src/pwnshop/lib/__init__.py

@@ -1,5 +1,6 @@
 import base64
 import contextlib
+import errno
 import json
 import logging
 import os
@@ -8,8 +9,10 @@
 import random
 import re
 import shutil
+import struct
 import subprocess
 import tempfile
+import threading
 import time
 from typing import Iterable, Iterator, List, Optional, Sequence
 
@@ -22,6 +25,20 @@
 
 CHALLENGE_SEED = int(os.environ.get("CHALLENGE_SEED", "0"))
 
+_KATA_TRANSITION_LOCK = threading.Lock()
+KATA_KVM_ACL_TIMEOUT = 10
+KATA_KVM_ACL_STABLE_TIME = 0.25
+KATA_KVM_POLL_INTERVAL = 0.05
+KVM_DEVICE_PATH = pathlib.Path("/dev/kvm")
+KVM_ACL_XATTR = "system.posix_acl_access"
+
+_POSIX_ACL_XATTR_VERSION = 2
+_POSIX_ACL_USER = 2
+_POSIX_ACL_MASK = 16
+_POSIX_ACL_READ_WRITE = 6
+_POSIX_ACL_HEADER = struct.Struct("<I")
+_POSIX_ACL_ENTRY = struct.Struct("<HHI")
+
 clang_format = shutil.which("clang-format")
 if not clang_format:
     logger.warning("clang-format not found; C templates will not be formatted")
@@ -100,6 +117,79 @@ def image_path(challenge_image: str) -> str:
     return ""
 
 
+def _mapped_root_kvm_acl_is_healthy() -> Optional[bool]:
+    """Check whether the outer user namespace grants mapped root KVM access."""
+    try:
+        acl = os.getxattr(KVM_DEVICE_PATH, KVM_ACL_XATTR)
+    except OSError as error:
+        if error.errno in {
+            errno.EACCES,
+            errno.ENODATA,
+            errno.ENOENT,
+            errno.ENOTSUP,
+            errno.EOPNOTSUPP,
+            errno.EPERM,
+        }:
+            return None
+        raise RuntimeError(f"Could not inspect {KVM_DEVICE_PATH} ACL: {error}") from error
+
+    if len(acl) < _POSIX_ACL_HEADER.size:
+        return False
+    (version,) = _POSIX_ACL_HEADER.unpack_from(acl)
+    entries = acl[_POSIX_ACL_HEADER.size :]
+    if version != _POSIX_ACL_XATTR_VERSION or len(entries) % _POSIX_ACL_ENTRY.size:
+        return False
+
+    parsed_entries = list(_POSIX_ACL_ENTRY.iter_unpack(entries))
+    mapped_root_permissions = next(
+        (permissions for tag, permissions, user_id in parsed_entries if tag == _POSIX_ACL_USER and user_id == 0),
+        None,
+    )
+    mask_permissions = next(
+        (permissions for tag, permissions, _ in parsed_entries if tag == _POSIX_ACL_MASK),
+        None,
+    )
+    if mapped_root_permissions is None or mask_permissions is None:
+        return False
+    return mapped_root_permissions & mask_permissions & _POSIX_ACL_READ_WRITE == _POSIX_ACL_READ_WRITE
+
+
+def _wait_for_stable_mapped_root_kvm_acl() -> None:
+    deadline = time.monotonic() + KATA_KVM_ACL_TIMEOUT
+    healthy_since = None
+
+    while True:
+        now = time.monotonic()
+        if _mapped_root_kvm_acl_is_healthy() is True:
+            if healthy_since is None:
+                healthy_since = now
+            elif now - healthy_since >= KATA_KVM_ACL_STABLE_TIME:
+                return
+        else:
+            healthy_since = None
+
+        if now >= deadline:
+            raise RuntimeError(
+                f"Timed out after {KATA_KVM_ACL_TIMEOUT}s waiting for {KVM_DEVICE_PATH} "
+                "to retain its mapped-root read/write ACL"
+            )
+        time.sleep(min(KATA_KVM_POLL_INTERVAL, max(0, deadline - now)))
+
+
+def _run_with_transition_lock(runtime: str, function, *args, wait_before: bool = True, **kwargs):
+    if runtime != "kata":
+        return function(*args, **kwargs)
+    with _KATA_TRANSITION_LOCK:
+        acl_is_observable = _mapped_root_kvm_acl_is_healthy() is not None
+        if acl_is_observable and wait_before:
+            _wait_for_stable_mapped_root_kvm_acl()
+        try:
+            return function(*args, **kwargs)
+        finally:
+            if acl_is_observable:
+                _wait_for_stable_mapped_root_kvm_acl()
+
+
 @contextlib.contextmanager
 def run_challenge(
     challenge_path: pathlib.Path,
@@ -135,7 +225,9 @@ def run_challenge(
     )
     container = None
     try:
-        container = subprocess.check_output(
+        container = _run_with_transition_lock(
+            runtime,
+            subprocess.check_output,
             [
                 "docker",
                 "run",
@@ -203,8 +295,11 @@ def run_challenge(
     finally:
         if container:
             logger.debug("removing container %s", container[:12])
-            subprocess.run(
+            _run_with_transition_lock(
+                runtime,
+                subprocess.run,
                 ["docker", "rm", "--force", container],
+                wait_before=False,
                 stdout=subprocess.DEVNULL,
                 stderr=subprocess.DEVNULL,
             )