| 1 | #!/usr/bin/env python3 |
| 2 | import argparse |
| 3 | import shlex |
| 4 | import subprocess |
| 5 | |
| 6 | |
| 7 | parser = argparse.ArgumentParser() |
| 8 | parser.add_argument("--host", default="root@zenith") |
| 9 | task = parser.add_mutually_exclusive_group() |
| 10 | task.add_argument("--prepare-task", action="store_true") |
| 11 | task.add_argument("--check-task", action="store_true") |
| 12 | args = parser.parse_args() |
| 13 | guest = ["ssh", "-i", "/srv/prod/openclaw/vm/ssh_key", "-p", "18892", |
| 14 | "-o", "BatchMode=yes", "-o", "IdentitiesOnly=yes", "-o", "ForwardAgent=no", |
| 15 | "-o", "StrictHostKeyChecking=yes", "-o", "UserKnownHostsFile=/srv/prod/openclaw/vm/known_hosts", |
| 16 | "ocadmin@127.0.0.1", "sudo docker exec -i openclaw python3 -"] |
| 17 | script = ''' |
| 18 | import json, os, pathlib, socket, subprocess, urllib.error, urllib.request |
| 19 | for name in ["NO_PROXY", "no_proxy"]: |
| 20 | os.environ.pop(name, None) |
| 21 | proxy = urllib.request.build_opener(urllib.request.ProxyHandler({"http": "http://10.0.2.100:3128", "https": "http://10.0.2.100:3128"})) |
| 22 | with proxy.open("https://example.com", timeout=20) as response: |
| 23 | assert response.status == 200 |
| 24 | for target in ["127.0.0.1", "10.0.0.1", "169.254.169.254", "192.168.1.1", "[::1]"]: |
| 25 | try: |
| 26 | proxy.open("http://" + target, timeout=5) |
| 27 | raise AssertionError("Private proxy target accepted: " + target) |
| 28 | except urllib.error.HTTPError as error: |
| 29 | assert error.code == 403, (target, error.code) |
| 30 | for target, port in [("10.0.2.2", 4646), ("10.0.0.1", 22), ("1.1.1.1", 443)]: |
| 31 | try: |
| 32 | connection = socket.create_connection((target, port), timeout=2) |
| 33 | except OSError: |
| 34 | continue |
| 35 | connection.close() |
| 36 | raise AssertionError("Direct egress accepted: " + target) |
| 37 | direct = urllib.request.build_opener(urllib.request.ProxyHandler({})) |
| 38 | with direct.open("http://10.0.2.101:8080/v1/models", timeout=20) as response: |
| 39 | assert json.load(response)["data"] |
| 40 | assert socket.getaddrinfo("example.com", 443) |
| 41 | assert not pathlib.Path("/var/run/docker.sock").exists() |
| 42 | assert not pathlib.Path("/srv/clover").exists() |
| 43 | print(json.dumps({"public_internet": True, "private_proxy_denied": 5, "direct_egress_denied": 3, "inference": True, "dns": True, "host_mounts": False}), flush=True) |
| 44 | ''' |
| 45 | if args.prepare_task or args.check_task: |
| 46 | script += r''' |
| 47 | directory = pathlib.Path("/home/node/.openclaw/workspace/openclaw-trial-check") |
| 48 | tests = "from median import median\nassert median([3,1,2]) == 2\nassert median([7,1,4,2]) == 3\na=[3,1,2]; median(a); assert a == [3,1,2]\nprint('MEDIAN_OK')\n" |
| 49 | ''' |
| 50 | if args.prepare_task: |
| 51 | script += r''' |
| 52 | directory.mkdir(exist_ok=True) |
| 53 | (directory / "median.py").write_text("def median(items):\n items.sort()\n return items[len(items) // 2]\n") |
| 54 | (directory / "check.py").write_text(tests) |
| 55 | print("Prepared coding fixture: " + str(directory)) |
| 56 | ''' |
| 57 | else: |
| 58 | script += """ |
| 59 | assert (directory / "check.py").read_text() == tests |
| 60 | subprocess.run(["python3", "check.py"], cwd=directory, check=True) |
| 61 | """ |
| 62 | subprocess.run(["ssh", "-o", "BatchMode=yes", args.host, shlex.join(guest)], |
| 63 | input=script, text=True, check=True) |