ElebobaandClaude Opus 5.5
c443190bf4
CI / ruff · mypy · pytest (pull_request) Successful in 4m3s
- infra/runner/tgit-runner: deploy/restart/remove/list по SSH forced command; аргументы не исполняются и не глобятся; архив — только файлы и папки, без «..», ≤ 20 МБ, есть main.py; health_url — только localhost - песочница: uid 10001, read-only, tmpfs noexec, cap-drop ALL, no-new-privileges, 128 МБ без swap, 0.25 CPU, 64 PID, --network none (сеть — только по egress-allow админа), лимит 10 сервисов, 2 образа - новая версия стартует рядом (-next) и заменяет старую, только став healthy; иначе удаляется, старая работает, exit 3 - бот: после push git archive изменённых services/<name>/ → runner; автору — «🚀 работает» или «⛔ не развёрнут» с логом, команде — алерт ⛔; удалённая папка → remove; «♻️ Перезапустить» в «Статусе» для админа (S-5) - examples/escape_attempts: все попытки выбраться — BLOCKED на живом Docker - install.sh, authorized_keys.example; CD бота обновляет runner на хосте Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
98 lines
2.6 KiB
Python
98 lines
2.6 KiB
Python
"""Tries to break out of the tgit-runner sandbox and reports each attempt. Every line should say BLOCKED.
|
|
|
|
Deploy it like any service (put it into services/escape/ of the uploads repo) and read its log in «📜 Логи».
|
|
Memory is not probed here: past 128 MB the kernel OOM-kills the process — the limit works, but the report would die.
|
|
"""
|
|
|
|
import os
|
|
import socket
|
|
import subprocess
|
|
import time
|
|
|
|
|
|
def attempt(name: str, action) -> None: # type: ignore[no-untyped-def]
|
|
try:
|
|
detail = action()
|
|
except Exception as e: # each failure is the expected outcome
|
|
print(f"BLOCKED {name}: {type(e).__name__}: {e}", flush=True)
|
|
return
|
|
print(f"ALLOWED {name}: {detail}", flush=True)
|
|
|
|
|
|
def write_code() -> str:
|
|
with open("/app/pwned.txt", "w") as f:
|
|
f.write("x")
|
|
return "wrote into /app"
|
|
|
|
|
|
def write_root() -> str:
|
|
with open("/etc/pwned", "w") as f:
|
|
f.write("x")
|
|
return "wrote into /etc"
|
|
|
|
|
|
def internet() -> str:
|
|
socket.create_connection(("1.1.1.1", 443), timeout=3).close()
|
|
return "reached 1.1.1.1:443"
|
|
|
|
|
|
def dns() -> str:
|
|
return socket.gethostbyname("example.com")
|
|
|
|
|
|
def docker_socket() -> str:
|
|
s = socket.socket(socket.AF_UNIX)
|
|
s.connect("/var/run/docker.sock")
|
|
return "connected to docker.sock"
|
|
|
|
|
|
def become_root() -> str:
|
|
os.setuid(0)
|
|
return "setuid(0) worked"
|
|
|
|
|
|
def raw_socket() -> str:
|
|
socket.socket(socket.AF_INET, socket.SOCK_RAW, socket.IPPROTO_ICMP)
|
|
return "opened a raw socket"
|
|
|
|
|
|
def fork_bomb() -> str:
|
|
procs = []
|
|
try:
|
|
for _ in range(200):
|
|
procs.append(subprocess.Popen(["sleep", "30"]))
|
|
finally:
|
|
started = len(procs)
|
|
for p in procs:
|
|
p.kill()
|
|
if started >= 200:
|
|
return "started 200 processes"
|
|
raise RuntimeError(f"stopped after {started} processes")
|
|
|
|
|
|
def exec_tmp() -> str:
|
|
path = "/tmp/run.sh"
|
|
with open(path, "w") as f:
|
|
f.write("#!/bin/sh\necho hi\n")
|
|
os.chmod(path, 0o755)
|
|
return subprocess.run([path], capture_output=True, check=True, text=True).stdout.strip()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
print(f"uid={os.getuid()} gid={os.getgid()}", flush=True)
|
|
for name, fn in [
|
|
("write into the code dir", write_code),
|
|
("write into /etc", write_root),
|
|
("connect to the internet", internet),
|
|
("resolve DNS", dns),
|
|
("talk to docker.sock", docker_socket),
|
|
("setuid(0)", become_root),
|
|
("raw socket", raw_socket),
|
|
("200 processes", fork_bomb),
|
|
("execute from /tmp", exec_tmp),
|
|
]:
|
|
attempt(name, fn)
|
|
print("done", flush=True)
|
|
while True:
|
|
time.sleep(3600)
|