Private
Public Access
705 lines
35 KiB
Python
705 lines
35 KiB
Python
"""Disposable Bearer runtime for documentation examples. Requires Linux bubblewrap."""
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
import base64
|
|
import hashlib
|
|
import http.client
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import resource
|
|
import re
|
|
import selectors
|
|
import shutil
|
|
import signal
|
|
import socket
|
|
import struct
|
|
import subprocess
|
|
import tempfile
|
|
import threading
|
|
import time
|
|
import uuid
|
|
from urllib.parse import quote
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Result:
|
|
stdout: str = ""
|
|
stderr: str = ""
|
|
status: int = 0
|
|
http_status: int | None = None
|
|
close_code: int | None = None
|
|
capture_window_seconds: float | None = None
|
|
|
|
|
|
class _UnixHTTP(http.client.HTTPConnection):
|
|
def __init__(self, path: Path, timeout: float):
|
|
super().__init__("localhost", timeout=timeout)
|
|
self.path = path
|
|
|
|
def connect(self):
|
|
self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
|
|
self.sock.settimeout(self.timeout)
|
|
self.sock.connect(str(self.path))
|
|
self.transport_socket = self.sock
|
|
|
|
|
|
def _limits(cpu_seconds=30):
|
|
resource.setrlimit(resource.RLIMIT_CPU, (cpu_seconds, cpu_seconds))
|
|
resource.setrlimit(resource.RLIMIT_FSIZE, (64 * 1024 * 1024,) * 2)
|
|
resource.setrlimit(resource.RLIMIT_NOFILE, (256, 256))
|
|
resource.setrlimit(resource.RLIMIT_CORE, (0, 0))
|
|
|
|
|
|
def pi_package():
|
|
executable = shutil.which("pi")
|
|
if executable:
|
|
entry = Path(executable).resolve()
|
|
package = next(parent for parent in entry.parents if (parent / "package.json").is_file())
|
|
return entry, package
|
|
|
|
|
|
class IsolatedRuntime:
|
|
timeout = 20.0
|
|
output_limit = 1024 * 1024
|
|
|
|
lifetime = 15 * 60
|
|
service_capture_seconds = 3.0
|
|
|
|
def __init__(self, repo: Path):
|
|
self.repo = Path(repo).resolve()
|
|
self.process = None
|
|
self._temp = None
|
|
self._log = None
|
|
self._provenance = None
|
|
self._mysql = None
|
|
self._mysql_log = None
|
|
self._mysql_result = None
|
|
|
|
def __enter__(self):
|
|
try:
|
|
self._temp = tempfile.TemporaryDirectory(prefix="capy-doc-")
|
|
self.root = Path(self._temp.name)
|
|
self.site = self.root / "site"
|
|
self.site.mkdir()
|
|
fixtures = self.repo / "site/doc/examples"
|
|
if fixtures.is_dir():
|
|
# Never dereference a fixture link into the host filesystem.
|
|
shutil.copytree(fixtures, self.site / "doc/examples", symlinks=True)
|
|
runtime_fixtures = fixtures / "runtime"
|
|
self._fixture_roots = []
|
|
for fixture in sorted(runtime_fixtures.glob("*")):
|
|
if not fixture.resolve().is_relative_to(runtime_fixtures.resolve()):
|
|
raise RuntimeError("Runtime fixture must stay inside site/doc/examples/runtime")
|
|
overlay = fixture / "site"
|
|
if overlay.is_dir():
|
|
if not overlay.resolve().is_relative_to(runtime_fixtures.resolve()):
|
|
raise RuntimeError("Runtime site overlay must stay inside site/doc/examples/runtime")
|
|
shutil.copytree(overlay, self.site, dirs_exist_ok=True, symlinks=True)
|
|
for entry in sorted((fixture / "root").glob("*")):
|
|
if not entry.resolve().is_relative_to(runtime_fixtures.resolve()):
|
|
raise RuntimeError("Runtime fixture root must stay inside site/doc/examples/runtime")
|
|
self._fixture_roots.append(entry)
|
|
for name in ("cache", "uploads", "sessions", "tasks", "run/mysqld", "mysql", "home", "etc/bearer", "bin/wasm"):
|
|
(self.root / name).mkdir(parents=True, exist_ok=True)
|
|
settings = {
|
|
"SITE_DIRECTORY": "site", "BIN_DIRECTORY": "/work/cache",
|
|
"TMP_UPLOAD_PATH": "/work/uploads", "SESSION_PATH": "/work/sessions",
|
|
"TASK_DIRECTORY": "/work/tasks", "FCGI_SOCKET_PATH": "/work/run/fcgi.sock",
|
|
"CLI_SOCKET_PATH": "/work/run/cli.sock", "HTTP_SOCKET_PATH": "/work/run/http.sock",
|
|
"HTTP_DOCUMENT_ROOT": "/work/site", "WASM_CORE_PATH": "/work/bin/wasm/core.wasm",
|
|
"FCGI_PORT": "", "HTTP_PORT": 80, "HTTP_BIND_ADDRESS": "127.0.0.1", "PROACTIVE_COMPILE_ENABLED": 0,
|
|
"WORKER_COUNT": 2, "TASK_WORKERS": 1, "KV_CACHE_BYTES": 1048576,
|
|
"WASM_MEMORY_LIMIT_BYTES": 134217728, "WASM_INVOCATION_TIMEOUT_MS": 3000,
|
|
"TASK_EXECUTION_TIMEOUT_MS": 3000, "TRANSPORT_MAX_RESPONSE_BYTES": self.output_limit,
|
|
"TRANSPORT_HTTP_REQUEST_TIMEOUT_SECONDS": 5,
|
|
"TRANSPORT_MAX_WEBSOCKET_OUTPUT_BYTES": self.output_limit,
|
|
"CUSTOM_SERVER_UNIX_SOCKET_PREFIX": "/work/run/custom/",
|
|
}
|
|
(self.root / "etc/bearer/settings.cfg").write_text(
|
|
"".join(f"{key}={value}\n" for key, value in settings.items()))
|
|
(self.root / "etc/passwd").write_text("root:x:0:0:root:/work/home:/bin/sh\n")
|
|
(self.root / "etc/group").write_text("root:x:0:\n")
|
|
(self.root / "etc/hosts").write_text("127.0.0.1 localhost\n")
|
|
(self.root / "etc/nsswitch.conf").write_text("passwd: files\ngroup: files\nhosts: files\n")
|
|
self._log = tempfile.TemporaryFile()
|
|
self.process = self._spawn(["/usr/bin/timeout", str(self.lifetime), "/usr/bin/python3", "-c", self._launcher()],
|
|
self._log, self._log, self.lifetime)
|
|
deadline = time.monotonic() + self.timeout
|
|
while time.monotonic() < deadline:
|
|
if self.process.poll() is not None:
|
|
raise RuntimeError("Isolated Bearer failed: " + self._logs())
|
|
if (self.root / "run/cli.sock").exists() and (self.root / "run/http.sock").exists():
|
|
return self
|
|
time.sleep(0.05)
|
|
raise RuntimeError("Isolated Bearer startup timed out: " + self._logs())
|
|
except BaseException:
|
|
self.__exit__(None, None, None)
|
|
raise
|
|
|
|
def _launcher(self):
|
|
return '''import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import time
|
|
fixtures = []
|
|
for script in sorted(Path("/work/site/doc/examples/runtime").glob("*/start.sh")):
|
|
ready = script.with_name("ready.sh")
|
|
if not ready.is_file():
|
|
raise RuntimeError("Missing runtime fixture readiness script: " + str(ready))
|
|
fixtures.append((script, ready, subprocess.Popen(["/bin/sh", str(script)], stdin=subprocess.DEVNULL)))
|
|
for script, ready, process in fixtures:
|
|
deadline = time.monotonic() + 10
|
|
while True:
|
|
if process.poll() is not None:
|
|
raise RuntimeError("Runtime fixture exited before readiness: " + str(script) + " status=" + str(process.returncode))
|
|
remaining = deadline - time.monotonic()
|
|
if remaining <= 0:
|
|
raise RuntimeError("Runtime fixture readiness timed out: " + str(ready))
|
|
checked = subprocess.run(["/bin/sh", str(ready)], stdin=subprocess.DEVNULL,
|
|
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=remaining)
|
|
if checked.returncode == 0:
|
|
break
|
|
time.sleep(0.02)
|
|
os.execv("/work/bin/bearer_fastcgi.linux.bin", ["bearer_fastcgi"])
|
|
'''
|
|
|
|
def provenance(self):
|
|
def digest(path):
|
|
value = hashlib.sha256()
|
|
with path.open("rb") as stream:
|
|
for chunk in iter(lambda: stream.read(1024 * 1024), b""):
|
|
value.update(chunk)
|
|
return value.hexdigest()
|
|
|
|
if self._provenance is None:
|
|
self._provenance = {"compiler": digest(self.repo / "bin/capyc"),
|
|
"runtime": digest(self.repo / "bin/bearer_fastcgi.linux.bin"),
|
|
"core": digest(self.repo / "bin/wasm/core.wasm")}
|
|
pi = pi_package()
|
|
if pi:
|
|
entry, package = pi
|
|
fingerprint = hashlib.sha256()
|
|
for path in sorted(package.rglob("*")):
|
|
if path.is_file() or path.is_symlink():
|
|
value = str(path.readlink()) if path.is_symlink() else digest(path)
|
|
fingerprint.update((path.relative_to(package).as_posix() + "\0" + value + "\0").encode())
|
|
tools = {"pi": fingerprint.hexdigest(), "pi_version": json.loads((package / "package.json").read_text())["version"]}
|
|
node = shutil.which("node")
|
|
if node:
|
|
tools["node"] = digest(Path(node))
|
|
self._provenance["tools"] = tools
|
|
return dict(self._provenance)
|
|
|
|
def _spawn(self, argv, stdout, stderr, cpu_seconds=30):
|
|
info = tempfile.TemporaryFile()
|
|
try:
|
|
process = subprocess.Popen(self._sandbox(info.fileno()) + argv,
|
|
stdin=subprocess.DEVNULL, stdout=stdout, stderr=stderr, env={},
|
|
start_new_session=True, pass_fds=(info.fileno(),),
|
|
preexec_fn=lambda: _limits(cpu_seconds))
|
|
except BaseException:
|
|
info.close()
|
|
raise
|
|
process.namespace_fd = None
|
|
try:
|
|
deadline = time.monotonic() + self.timeout
|
|
while time.monotonic() < deadline:
|
|
data = os.pread(info.fileno(), 4096, 0)
|
|
if data:
|
|
try:
|
|
metadata = json.loads(data)
|
|
except json.JSONDecodeError:
|
|
time.sleep(0.01)
|
|
continue
|
|
try:
|
|
process.namespace_fd = os.pidfd_open(metadata["child-pid"])
|
|
except ProcessLookupError:
|
|
pass
|
|
return process
|
|
if process.poll() is not None:
|
|
return process
|
|
time.sleep(0.01)
|
|
raise RuntimeError("Isolated process did not report its namespace before the startup deadline")
|
|
except BaseException:
|
|
self._stop(process)
|
|
raise
|
|
finally:
|
|
info.close()
|
|
|
|
def _stop(self, process):
|
|
namespace = getattr(process, "namespace_fd", None)
|
|
try:
|
|
if namespace is not None:
|
|
try:
|
|
signal.pidfd_send_signal(namespace, signal.SIGKILL)
|
|
except ProcessLookupError:
|
|
pass
|
|
try:
|
|
os.killpg(process.pid, signal.SIGKILL)
|
|
except ProcessLookupError:
|
|
pass
|
|
code = process.wait(timeout=5)
|
|
if namespace is not None:
|
|
with selectors.DefaultSelector() as selector:
|
|
selector.register(namespace, selectors.EVENT_READ)
|
|
if not selector.select(5):
|
|
raise RuntimeError("Isolated PID namespace did not stop after SIGKILL")
|
|
return code
|
|
finally:
|
|
if namespace is not None:
|
|
os.close(namespace)
|
|
process.namespace_fd = None
|
|
|
|
def _sandbox(self, info_fd=None):
|
|
bwrap = shutil.which("bwrap")
|
|
if not bwrap:
|
|
raise RuntimeError("Documentation execution requires bubblewrap. No unsafe fallback is available.")
|
|
command = [bwrap, "--unshare-all", "--share-net", "--hostname", "localhost", "--die-with-parent", "--new-session", "--cap-drop", "ALL",
|
|
"--clearenv", "--setenv", "PATH", "/work/bin:/usr/bin:/bin", "--setenv", "HOME", "/work/home",
|
|
"--setenv", "TMPDIR", "/tmp", "--setenv", "LANG", "C.UTF-8",
|
|
"--ro-bind", "/usr/bin", "/usr/bin", "--ro-bind", "/usr/lib", "/usr/lib",
|
|
"--symlink", "usr/bin", "/bin", "--symlink", "usr/lib", "/lib",
|
|
"--symlink", "usr/lib64", "/lib64"]
|
|
if Path("/usr/lib64").exists():
|
|
command += ["--ro-bind", "/usr/lib64", "/usr/lib64"]
|
|
command += ["--proc", "/proc", "--dev", "/dev", "--size", "67108864", "--tmpfs", "/tmp",
|
|
"--bind", str(self.root), "/work", "--ro-bind", str(self.root / "etc"), "/etc",
|
|
"--bind", str(self.root / "run"), "/run", "--symlink", "/run", "/var/run",
|
|
"--bind", str(self.root / "mysql"), "/var/lib/mysql"]
|
|
for entry in self._fixture_roots:
|
|
command += ["--ro-bind", str(entry), "/" + entry.name]
|
|
for path in ("/usr/sbin", "/usr/share/mariadb", "/usr/share/zoneinfo", "/usr/share/nodejs"):
|
|
if Path(path).exists():
|
|
command += ["--ro-bind", path, path]
|
|
# Private aliases avoid host daemon profiles that assume the host PID namespace.
|
|
for path in [Path("/usr/sbin/mariadbd"), *Path("/usr/sbin").glob("php-fpm*")]:
|
|
if path.is_file():
|
|
command += ["--ro-bind", str(path), "/work/bin/" + path.name]
|
|
pi = pi_package()
|
|
if pi:
|
|
entry, package = pi
|
|
command += ["--ro-bind", str(package), str(package), "--symlink", str(entry), "/work/bin/pi"]
|
|
for name in ("bearer_fastcgi.linux.bin", "capyc", "wasm/core.wasm"):
|
|
command += ["--ro-bind", str(self.repo / "bin" / name), "/work/bin/" + name]
|
|
wasmtime = Path("/opt/wasmtime/lib")
|
|
if wasmtime.exists():
|
|
command += ["--ro-bind", str(wasmtime), str(wasmtime)]
|
|
if info_fd is not None:
|
|
command += ["--info-fd", str(info_fd)]
|
|
setup = '''import os, subprocess, sys
|
|
subprocess.run(["/usr/bin/mount", "--no-mtab", "-t", "proc", "proc", "/proc"], check=True, timeout=5)
|
|
with open("/proc/sys/net/ipv4/ip_unprivileged_port_start", "w") as stream:
|
|
stream.write("0")
|
|
subprocess.run(["/usr/sbin/ip", "link", "set", "lo", "up"], check=True, timeout=5)
|
|
os.execv(sys.argv[1], sys.argv[1:])
|
|
'''
|
|
# This trusted setup changes only the new network and private mount namespaces.
|
|
return ["/usr/bin/unshare", "--mount", "--net", "--propagation", "private", "/usr/bin/python3", "-c", setup] + command + ["--chdir", "/work"]
|
|
|
|
def _logs(self):
|
|
return os.pread(self._log.fileno(), self.output_limit, 0).decode("utf-8", "replace")
|
|
|
|
def __exit__(self, *args):
|
|
if self._mysql is not None:
|
|
self._stop(self._mysql)
|
|
self._mysql = None
|
|
if self._mysql_log is not None:
|
|
self._mysql_log.close()
|
|
self._mysql_log = None
|
|
if self.process is not None:
|
|
self._stop(self.process)
|
|
self.process = None
|
|
if self._log is not None:
|
|
self._log.close()
|
|
self._log = None
|
|
if self._temp is not None:
|
|
self._temp.cleanup()
|
|
self._temp = None
|
|
|
|
def enable_mysql(self):
|
|
"""Start the documentation database on a private Unix socket, without TCP."""
|
|
if self._mysql is not None and self._mysql.poll() is None:
|
|
return self._mysql_result or Result(stderr="Private MariaDB fixture did not finish initialization", status=1)
|
|
if not Path("/usr/sbin/mariadbd").exists() or not Path("/usr/bin/mariadb-install-db").exists():
|
|
return Result(stderr="MariaDB documentation fixtures require the installed mariadbd and mariadb-install-db tools", status=2)
|
|
seed = self.site / "doc/examples/mysql.sql"
|
|
if not seed.is_file():
|
|
return Result(stderr="Missing documentation database seed: site/doc/examples/mysql.sql", status=2)
|
|
initialized = self._command(["/usr/bin/env", "MYSQLD_BOOTSTRAP=/work/bin/mariadbd", "/usr/bin/mariadb-install-db", "--no-defaults", "--basedir=/usr",
|
|
"--datadir=/var/lib/mysql", "--user=root", "--skip-test-db",
|
|
"--auth-root-authentication-method=normal", "--innodb-log-file-size=5M",
|
|
"--innodb-buffer-pool-size=16M", "--aria-log-file-size=1M"])
|
|
if initialized.status:
|
|
return initialized
|
|
self._mysql_log = tempfile.TemporaryFile()
|
|
arguments = ["/work/bin/mariadbd", "--no-defaults", "--user=root", "--datadir=/var/lib/mysql",
|
|
"--socket=/run/mysqld/mysqld.sock", "--pid-file=/run/mysqld/mysqld.pid", "--skip-networking",
|
|
"--skip-log-bin", "--innodb-log-file-size=5M", "--innodb-buffer-pool-size=16M",
|
|
"--aria-log-file-size=1M", "--max-connections=8", "--performance-schema=OFF"]
|
|
self._mysql = self._spawn(["/usr/bin/timeout", str(self.lifetime)] + arguments,
|
|
self._mysql_log, self._mysql_log, self.lifetime)
|
|
deadline = time.monotonic() + self.timeout
|
|
while time.monotonic() < deadline:
|
|
if self._mysql.poll() is not None:
|
|
diagnostics = os.pread(self._mysql_log.fileno(), self.output_limit, 0).decode("utf-8", "replace")
|
|
return Result(stderr="Private MariaDB startup failed\n" + diagnostics, status=1)
|
|
if (self.root / "run/mysqld/mysqld.sock").exists():
|
|
seeded = self._command(["/usr/bin/mariadb", "--no-defaults", "--protocol=socket",
|
|
"--socket=/run/mysqld/mysqld.sock", "--user=root",
|
|
"--execute=source /work/site/doc/examples/mysql.sql"])
|
|
self._mysql_result = seeded
|
|
return seeded
|
|
time.sleep(0.05)
|
|
return Result(stderr="Private MariaDB startup timed out", status=124)
|
|
|
|
def request(self, unit, params=None, transport="cli", method=None):
|
|
if transport not in ("cli", "http"):
|
|
return Result(stderr="Unknown Unix transport: " + transport, status=2)
|
|
connection = _UnixHTTP(self.root / "run" / (transport + ".sock"), self.timeout)
|
|
body = None if params is None else json.dumps(params).encode()
|
|
if body is not None and len(body) > self.output_limit:
|
|
return Result(stderr="Request exceeds the byte limit", status=2)
|
|
deadline = time.monotonic() + self.timeout
|
|
def expire():
|
|
transport_socket = getattr(connection, "transport_socket", None)
|
|
if transport_socket:
|
|
try:
|
|
transport_socket.shutdown(socket.SHUT_RDWR)
|
|
except OSError:
|
|
pass
|
|
|
|
timer = threading.Timer(self.timeout, expire)
|
|
timer.daemon = True
|
|
timer.start()
|
|
try:
|
|
connection.request(method or ("GET" if body is None else "POST"),
|
|
quote("/" + str(unit).lstrip("/"), safe="/:?=&%"), body,
|
|
{"Content-Type": "application/json"} if body is not None else {})
|
|
response = connection.getresponse()
|
|
chunks = bytearray()
|
|
while len(chunks) <= self.output_limit:
|
|
remaining = deadline - time.monotonic()
|
|
if remaining <= 0:
|
|
raise TimeoutError("Request deadline exceeded")
|
|
connection.transport_socket.settimeout(remaining)
|
|
chunk = response.read1(min(65536, self.output_limit + 1 - len(chunks)))
|
|
if not chunk:
|
|
return Result(chunks.decode("utf-8", "replace"), status=0 if response.status < 400 else response.status,
|
|
http_status=response.status)
|
|
chunks.extend(chunk)
|
|
return Result(stderr="Response exceeds the byte limit", status=125)
|
|
except (OSError, http.client.HTTPException) as error:
|
|
return Result(stderr=str(error), status=124 if isinstance(error, TimeoutError) else 1)
|
|
finally:
|
|
timer.cancel()
|
|
timer.join(timeout=1)
|
|
connection.close()
|
|
|
|
def cli(self, unit, params=None):
|
|
return self.request(unit, params)
|
|
|
|
def render(self, unit, params=None):
|
|
return self.request(unit, params, transport="http")
|
|
|
|
def task(self, unit, handler, props=None):
|
|
return self._async_handler("TASK", unit, handler, props)
|
|
|
|
def service(self, unit, handler, props=None):
|
|
return self._async_handler("SERVICE", unit, handler, props)
|
|
|
|
def _site_file(self, relative, flags, mode="r"):
|
|
parts = Path(relative).parts
|
|
if not parts or any(part in ("..", "/") for part in parts):
|
|
raise ValueError("File must be inside the isolated site")
|
|
directory = os.open(self.root, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW)
|
|
try:
|
|
for part in ("site",) + parts[:-1]:
|
|
child = os.open(part, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=directory)
|
|
os.close(directory)
|
|
directory = child
|
|
descriptor = os.open(parts[-1], flags | os.O_NOFOLLOW | os.O_NONBLOCK, 0o600, dir_fd=directory)
|
|
return os.fdopen(descriptor, mode)
|
|
finally:
|
|
os.close(directory)
|
|
|
|
def _async_handler(self, kind, unit, handler, props):
|
|
route = str(unit).split(":", 1)[0]
|
|
name = str(handler).split(":")[-1]
|
|
if not re.fullmatch(r"[A-Za-z_][A-Za-z_0-9]*", name):
|
|
return Result(stderr="Invalid asynchronous handler name", status=2)
|
|
source_path = self.site / route.lstrip("/")
|
|
if not source_path.resolve().is_relative_to(self.site):
|
|
return Result(stderr="Unit must be inside the isolated site", status=2)
|
|
relative = source_path.relative_to(self.site)
|
|
with self._site_file(relative, os.O_RDONLY) as stream:
|
|
source = stream.read(self.output_limit + 1)
|
|
if len(source) > self.output_limit:
|
|
return Result(stderr="Asynchronous unit exceeds the byte limit", status=125)
|
|
pattern = re.compile(r"\bfunction\s+" + kind + r"\s*:\s*([A-Za-z_][A-Za-z_0-9]*)\b\s*(?:\([^)]*\))?")
|
|
masked = re.sub(r'"(?:\\.|[^"\\])*"|//[^\n]*|/\*.*?\*/', lambda match: " " * len(match.group()), source, flags=re.S)
|
|
matches = list(pattern.finditer(masked))
|
|
if name not in [match.group(1) for match in matches]:
|
|
return Result(stderr=f"Missing {kind}:{name} handler", status=2)
|
|
if kind == "TASK":
|
|
matches = [match for match in matches if match.group(1) == name]
|
|
token = uuid.uuid4().hex
|
|
controller = self.site / ("doc_example_control_" + token + ".capy")
|
|
output = self.site / (token + ".output")
|
|
done = self.site / (token + ".done")
|
|
output_name = "/work/site/" + output.name
|
|
done_name = "/work/site/" + done.name
|
|
def closing_brace(opening):
|
|
depth = 0
|
|
for index in range(opening, len(masked)):
|
|
if masked[index] == "{":
|
|
depth += 1
|
|
elif masked[index] == "}":
|
|
depth -= 1
|
|
if depth == 0:
|
|
return index
|
|
raise ValueError("Cannot locate the asynchronous handler body")
|
|
|
|
insertions = []
|
|
for match in matches:
|
|
event = match.group(1)
|
|
opening = masked.find("{", match.end())
|
|
closing = closing_brace(opening)
|
|
capture = f'''\n if file_append({json.dumps(output_name)}, ob_get_close())
|
|
{{
|
|
file_append({json.dumps(done_name)}, {json.dumps(event + chr(10))})
|
|
}}
|
|
'''
|
|
insertions.extend([(opening + 1, "\n ob_start()\n"), (closing, capture)])
|
|
nested_functions = []
|
|
for nested in re.finditer(r"\bfunction\b", masked[opening + 1:closing]):
|
|
nested_open = masked.find("{", opening + 1 + nested.end())
|
|
nested_functions.append((nested_open, closing_brace(nested_open)))
|
|
for returned in re.finditer(r"(?<![.\w])return\b", masked[opening + 1:closing]):
|
|
position = opening + 1 + returned.start()
|
|
if not any(start < position < end for start, end in nested_functions):
|
|
insertions.append((position, capture))
|
|
instrumented = source
|
|
for position, text in sorted(insertions, reverse=True):
|
|
instrumented = instrumented[:position] + text + instrumented[position:]
|
|
encoded_props = json.dumps(json.dumps({} if props is None else props))
|
|
target = json.dumps(route)
|
|
if kind == "TASK":
|
|
operation = f"task_execute({json.dumps(route + ':' + name)}, 3, input)"
|
|
statements = f"print(json_encode({operation}))"
|
|
else:
|
|
statements = f"print(service_start({target}, input))"
|
|
if name != "START":
|
|
statements += f"\n service_send({target}, {json.dumps(name)}, input)"
|
|
controller_source = f'''function CLI()
|
|
{{
|
|
var input := json_decode({encoded_props})
|
|
{statements}
|
|
}}
|
|
function CLI:STOP()
|
|
{{
|
|
service_stop({target})
|
|
}}
|
|
function CLI:STATUS()
|
|
{{
|
|
print(json_encode(service_status({target})))
|
|
}}
|
|
'''
|
|
with self._site_file(controller.name, os.O_WRONLY | os.O_CREAT | os.O_EXCL, "w") as stream:
|
|
stream.write(controller_source)
|
|
log_offset = os.fstat(self._log.fileno()).st_size
|
|
source_stream = None
|
|
try:
|
|
source_stream = self._site_file(relative, os.O_RDWR, "r+")
|
|
source_stream.write(instrumented)
|
|
source_stream.truncate()
|
|
source_stream.flush()
|
|
checked = self.compile(route)
|
|
if checked.status:
|
|
return checked
|
|
started = self.cli("/" + controller.name)
|
|
if started.status:
|
|
return started
|
|
try:
|
|
receipt = json.loads(started.stdout)
|
|
except ValueError:
|
|
return Result(stderr="Invalid native asynchronous start result: " + started.stdout, status=1)
|
|
if receipt is False:
|
|
return Result(stderr="Native asynchronous start was rejected", status=1)
|
|
if isinstance(receipt, dict) and receipt.get("error"):
|
|
return Result(stderr="Native asynchronous start failed: " + str(receipt["error"]), status=1)
|
|
deadline = time.monotonic() + self.timeout
|
|
completed = False
|
|
while time.monotonic() < deadline:
|
|
if done.exists():
|
|
with self._site_file(done.name, os.O_RDONLY) as stream:
|
|
events = stream.read(self.output_limit).splitlines()
|
|
if name in events:
|
|
completed = True
|
|
break
|
|
time.sleep(0.02)
|
|
if completed and kind == "SERVICE":
|
|
# START completion does not mean ports, timers, or delayed sends finished.
|
|
deadline = time.monotonic() + self.service_capture_seconds
|
|
while time.monotonic() < deadline:
|
|
time.sleep(min(0.02, max(0, deadline - time.monotonic())))
|
|
stopped = self.cli("/" + controller.name + ":STOP")
|
|
if stopped.status:
|
|
return stopped
|
|
deadline = time.monotonic() + self.timeout
|
|
completed = False
|
|
while time.monotonic() < deadline:
|
|
status = self.cli("/" + controller.name + ":STATUS")
|
|
if status.status:
|
|
return status
|
|
if json.loads(status.stdout).get("state") == "stopped":
|
|
completed = True
|
|
break
|
|
time.sleep(0.05)
|
|
diagnostics = os.pread(self._log.fileno(), self.output_limit, log_offset).decode("utf-8", "replace")
|
|
targets = ("/work/site/" + route.lstrip("/"), "/work/site/" + controller.name)
|
|
diagnostics = "\n".join(line for line in diagnostics.splitlines() if any(target in line for target in targets))
|
|
failed = re.search(r"(?m)^\(!\) (?:service|task) .*?(?:failed|crashed)", diagnostics)
|
|
failed = failed or re.search(r"(?m)^\(!\) service .* signal (?!DATA\b|EXIT\b|ACCEPT\b)\w+ dropped", diagnostics)
|
|
captured = b""
|
|
try:
|
|
with self._site_file(output.name, os.O_RDONLY, "rb") as stream:
|
|
captured = stream.read(self.output_limit + 1)
|
|
except FileNotFoundError:
|
|
pass
|
|
text = captured[:self.output_limit].decode("utf-8", "replace")
|
|
if len(captured) > self.output_limit:
|
|
return Result(text, stderr="Asynchronous output exceeds the byte limit", status=125)
|
|
if failed or not completed:
|
|
return Result(text, stderr="Native asynchronous lifecycle failed or did not complete output capture\n" + diagnostics,
|
|
status=1 if failed else 124)
|
|
return Result(text,
|
|
capture_window_seconds=self.service_capture_seconds if kind == "SERVICE" else None)
|
|
finally:
|
|
if kind == "SERVICE":
|
|
self.cli("/" + controller.name + ":STOP")
|
|
if source_stream is not None:
|
|
source_stream.seek(0)
|
|
source_stream.write(source)
|
|
source_stream.truncate()
|
|
source_stream.close()
|
|
controller.unlink(missing_ok=True)
|
|
output.unlink(missing_ok=True)
|
|
done.unlink(missing_ok=True)
|
|
|
|
def ws(self, unit, message="hello"):
|
|
"""Send one text frame and collect the first response message."""
|
|
payload = str(message).encode()
|
|
if len(payload) > self.output_limit:
|
|
return Result(stderr="WebSocket input exceeds the byte limit", status=2)
|
|
deadline = time.monotonic() + self.timeout
|
|
with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as connection:
|
|
def read_exact(size):
|
|
data = bytearray()
|
|
while len(data) < size:
|
|
remaining = deadline - time.monotonic()
|
|
if remaining <= 0:
|
|
raise TimeoutError("WebSocket deadline exceeded")
|
|
connection.settimeout(remaining)
|
|
part = connection.recv(size - len(data))
|
|
if not part:
|
|
raise OSError("WebSocket closed before a response")
|
|
data.extend(part)
|
|
return bytes(data)
|
|
|
|
try:
|
|
connection.settimeout(self.timeout)
|
|
connection.connect(str(self.root / "run/http.sock"))
|
|
key = base64.b64encode(os.urandom(16)).decode()
|
|
route = quote("/" + str(unit).lstrip("/"), safe="/:?=&%")
|
|
connection.sendall((f"GET {route} HTTP/1.1\r\nHost: localhost\r\n"
|
|
"Upgrade: websocket\r\nConnection: Upgrade\r\n"
|
|
f"Sec-WebSocket-Key: {key}\r\nSec-WebSocket-Version: 13\r\n\r\n").encode())
|
|
headers = bytearray()
|
|
while not headers.endswith(b"\r\n\r\n"):
|
|
if len(headers) >= 16384:
|
|
return Result(stderr="WebSocket headers exceed the byte limit", status=125)
|
|
headers.extend(read_exact(1))
|
|
accept = base64.b64encode(hashlib.sha1((key + "258EAFA5-E914-47DA-95CA-C5AB0DC85B11").encode()).digest())
|
|
if b" 101 " not in bytes(headers).split(b"\r\n", 1)[0] or accept not in headers:
|
|
return Result(stderr=headers.decode("utf-8", "replace"), status=1)
|
|
mask = os.urandom(4)
|
|
length = len(payload)
|
|
header = bytes([129, 128 | length]) if length < 126 else (
|
|
bytes([129, 254]) + struct.pack("!H", length) if length < 65536 else
|
|
bytes([129, 255]) + struct.pack("!Q", length))
|
|
connection.sendall(header + mask + bytes(value ^ mask[index % 4] for index, value in enumerate(payload)))
|
|
output = bytearray()
|
|
while True:
|
|
first, second = read_exact(2)
|
|
length = second & 127
|
|
if length == 126:
|
|
length = struct.unpack("!H", read_exact(2))[0]
|
|
elif length == 127:
|
|
length = struct.unpack("!Q", read_exact(8))[0]
|
|
if length + len(output) > self.output_limit:
|
|
return Result(stderr="WebSocket output exceeds the byte limit", status=125)
|
|
mask = read_exact(4) if second & 128 else None
|
|
data = read_exact(length)
|
|
if mask:
|
|
data = bytes(value ^ mask[index % 4] for index, value in enumerate(data))
|
|
opcode = first & 15
|
|
if opcode == 8:
|
|
code = struct.unpack("!H", data[:2])[0] if len(data) >= 2 else 1005
|
|
normal = code in (1000, 1001, 1005)
|
|
return Result(output.decode("utf-8", "replace"),
|
|
stderr="" if normal else "WebSocket close: " + data[2:].decode("utf-8", "replace"),
|
|
status=0 if normal else 1, close_code=code)
|
|
if opcode in (0, 1, 2):
|
|
output.extend(data)
|
|
if first & 128:
|
|
return Result(output.decode("utf-8", "replace"))
|
|
except OSError as error:
|
|
return Result(stderr=str(error), status=124 if isinstance(error, TimeoutError) else 1)
|
|
|
|
def compile(self, unit):
|
|
path = Path(str(unit).split(":", 1)[0])
|
|
if path.is_absolute() and path.is_relative_to(self.site):
|
|
path = path.relative_to(self.site)
|
|
path = Path(str(path).lstrip("/"))
|
|
if ".." in path.parts:
|
|
return Result(stderr="Unit must be inside the isolated site", status=2)
|
|
return self._command(["/work/bin/capyc", "--check", "--source-root", "/work/site", "/work/site/" + str(path)])
|
|
|
|
def _command(self, argv):
|
|
process = self._spawn(argv, subprocess.PIPE, subprocess.PIPE)
|
|
output = {process.stdout: bytearray(), process.stderr: bytearray()}
|
|
status = None
|
|
deadline = time.monotonic() + self.timeout
|
|
try:
|
|
with selectors.DefaultSelector() as selector:
|
|
for stream in output:
|
|
os.set_blocking(stream.fileno(), False)
|
|
selector.register(stream, selectors.EVENT_READ)
|
|
while selector.get_map() and status is None:
|
|
remaining = deadline - time.monotonic()
|
|
if remaining <= 0:
|
|
status = 124
|
|
break
|
|
for key, _ in selector.select(min(remaining, 0.1)):
|
|
chunk = os.read(key.fileobj.fileno(), 65536)
|
|
if not chunk:
|
|
selector.unregister(key.fileobj)
|
|
else:
|
|
output[key.fileobj].extend(chunk)
|
|
if sum(map(len, output.values())) > self.output_limit:
|
|
status = 125
|
|
break
|
|
if status is None:
|
|
try:
|
|
process.wait(timeout=max(0.01, deadline - time.monotonic()))
|
|
except subprocess.TimeoutExpired:
|
|
status = 124
|
|
finally:
|
|
code = self._stop(process)
|
|
for stream in output:
|
|
stream.close()
|
|
stdout, stderr = (bytes(value[:self.output_limit]).decode("utf-8", "replace") for value in output.values())
|
|
return Result(stdout, stderr, status if status is not None else code)
|