AdamWen95
158 lines
7.7 KiB
Python
158 lines
7.7 KiB
Python
"""Console contracts: immutable uploads, isolation and durable queue semantics."""
|
|
import json
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
from fastapi.testclient import TestClient
|
|
from console.app import create_app
|
|
from console.settings import Settings
|
|
from console.store import Store
|
|
|
|
|
|
class ConsoleTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.root = Path(self.temp.name)
|
|
self.source = self.root / "original" / "scripts"
|
|
self.source.mkdir(parents=True)
|
|
self.settings = Settings(self.root / "runtime", [
|
|
{"id": "trial", "path": str(self.source)}], allowed_hosts=["testserver"],
|
|
allowed_networks=["127.0.0.0/8"], test_mode=True)
|
|
self.settings.test_definitions = [{"key": name, "source_group": "trial", "entry": name,
|
|
"name": name, "kind": "test"} for name in ("sample.py", "test.py", "stop.sh")]
|
|
self.app = create_app(self.settings)
|
|
self.client = TestClient(self.app, client=("127.0.0.1", 5555))
|
|
self.headers = {"X-XAS-Console": "1"}
|
|
|
|
def tearDown(self):
|
|
self.client.close()
|
|
self.temp.cleanup()
|
|
|
|
def upload(self, content=b"print('hello')\n", name="test.py"):
|
|
r = self.client.post("/api/scripts/upload", files={"file": (name, content)},
|
|
data={"name": "Sample"}, headers=self.headers)
|
|
self.assertEqual(r.status_code, 200, r.text)
|
|
return r.json()
|
|
|
|
def queue(self, script, key="9ad0745c-0d5e-4d1e-88e8-000000000001"):
|
|
return self.client.post(f"/api/scripts/{script['id']}/runs",
|
|
json={"request_id": key}, headers=self.headers)
|
|
|
|
def test_upload_versions_are_immutable_and_runs_pin_version(self):
|
|
script = self.upload()
|
|
run = self.queue(script).json()
|
|
new = self.client.post(f"/api/scripts/{script['id']}/versions",
|
|
files={"file": ("test.py", b"print('v2')\n")}, headers=self.headers)
|
|
self.assertEqual(new.status_code, 200, new.text)
|
|
old = self.client.get(f"/api/scripts/{script['id']}/code?version=1").json()
|
|
self.assertEqual(old["code"], "print('hello')\n")
|
|
self.assertEqual(self.client.get(f"/api/runs/{run['id']}").json()["version"], 1)
|
|
self.assertEqual(new.json()["version"], 2)
|
|
|
|
def test_code_cannot_be_patched_and_path_cannot_be_replaced(self):
|
|
s = self.upload()
|
|
for patch in ({"code": "bad"}, {"source_path": "/etc/passwd"},
|
|
{"config": {"profile": "legacy_smoke"}}):
|
|
r = self.client.patch(f"/api/scripts/{s['id']}", json=patch, headers=self.headers)
|
|
self.assertEqual(r.status_code, 422, r.text)
|
|
|
|
def test_upload_rejects_traversal_secrets_and_oversize(self):
|
|
for name, content in [("../x.py", b"pass"), ("x.env", b"pass"),
|
|
("x.py", b"API_KEY = 'abc-secret-key-value-01234'"),
|
|
("x.py", b"x" * (1024 * 1024 + 1))]:
|
|
r = self.client.post("/api/scripts/upload", files={"file": (name, content)},
|
|
headers=self.headers)
|
|
self.assertIn(r.status_code, (400, 413, 422), r.text)
|
|
|
|
def test_cross_site_mutations_and_host_spoofing_rejected(self):
|
|
r = self.client.post("/api/discover", headers={**self.headers, "Origin": "https://evil.test"})
|
|
self.assertEqual(r.status_code, 403)
|
|
self.assertEqual(self.client.post("/api/discover").status_code, 403)
|
|
self.assertEqual(self.client.get("/api/summary", headers={"Host": "evil.test"}).status_code, 403)
|
|
|
|
def test_duplicate_submission_and_single_claim(self):
|
|
s = self.upload()
|
|
a, b = self.queue(s), self.queue(s)
|
|
self.assertEqual(a.json()["id"], b.json()["id"])
|
|
store = Store(self.settings)
|
|
self.assertEqual(store.claim()["id"], a.json()["id"])
|
|
self.assertIsNone(store.claim())
|
|
self.queue(s, "9ad0745c-0d5e-4d1e-88e8-000000000002")
|
|
self.assertIsNone(store.claim())
|
|
|
|
def test_pending_cancellation_does_not_execute(self):
|
|
run = self.queue(self.upload()).json()
|
|
r = self.client.post(f"/api/runs/{run['id']}/cancel", headers=self.headers)
|
|
self.assertEqual(r.json()["state"], "cancelled")
|
|
self.assertIsNone(Store(self.settings).claim())
|
|
|
|
def test_external_inventory_readonly_and_missing_detection(self):
|
|
p = self.source / "sample.py"
|
|
p.write_text("print('original')", encoding="utf-8")
|
|
self.client.post("/api/discover", headers=self.headers)
|
|
s = self.client.get("/api/scripts").json()["items"][0]
|
|
self.assertEqual(s["kind"], "external")
|
|
self.assertFalse(s["enabled"])
|
|
self.assertEqual(p.read_text(), "print('original')")
|
|
p.unlink()
|
|
self.client.post("/api/discover", headers=self.headers)
|
|
self.assertEqual(self.client.get(f"/api/scripts/{s['id']}").json()["status"], "missing")
|
|
|
|
def test_service_management_script_cannot_be_enabled(self):
|
|
(self.source / "stop.sh").write_text("docker stop other-service\n")
|
|
self.client.post("/api/discover", headers=self.headers)
|
|
s = self.client.get("/api/scripts").json()["items"][0]
|
|
r = self.client.patch(f"/api/scripts/{s['id']}", json={"enabled": True}, headers=self.headers)
|
|
self.assertEqual(r.status_code, 400, r.text)
|
|
|
|
def test_external_output_path_cannot_escape_source_family(self):
|
|
(self.source / "test.py").write_text("print(1)\n")
|
|
self.client.post("/api/discover", headers=self.headers)
|
|
s = self.client.get("/api/scripts").json()["items"][0]
|
|
for path in ["/etc", str(self.source), str(self.source.parent / "configs")]:
|
|
r = self.client.patch(f"/api/scripts/{s['id']}",
|
|
json={"config": {"output_paths": [path]}}, headers=self.headers)
|
|
self.assertEqual(r.status_code, 400, r.text)
|
|
|
|
def test_log_offset_unicode_and_artifact_path_boundary(self):
|
|
run = self.queue(self.upload()).json()
|
|
d = self.settings.runs / run["id"]
|
|
d.mkdir(exist_ok=True)
|
|
(d / "console.log").write_bytes("你好\n".encode("utf-8"))
|
|
a = self.client.get(f"/api/runs/{run['id']}/log?offset=0").json()
|
|
self.assertEqual(a["text"], "你好\n")
|
|
b = self.client.get(f"/api/runs/{run['id']}/log?offset={a['next_offset']}").json()
|
|
self.assertEqual(b["text"], "")
|
|
self.assertEqual(self.client.get(f"/api/runs/{run['id']}/artifacts/nope").status_code, 404)
|
|
|
|
def test_finished_log_utf8_character_across_chunk_boundary(self):
|
|
run = self.queue(self.upload()).json()
|
|
directory = self.settings.runs / run["id"]
|
|
directory.mkdir()
|
|
content = "x" * 65535 + "你" + "tail"
|
|
(directory / "console.log").write_bytes(content.encode("utf-8"))
|
|
Store(self.settings).finish(run["id"], "succeeded")
|
|
first = self.client.get(f"/api/runs/{run['id']}/log").json()
|
|
second = self.client.get(f"/api/runs/{run['id']}/log?offset={first['next_offset']}").json()
|
|
self.assertEqual(first["text"] + second["text"], content)
|
|
|
|
def test_runtime_parent_link_rejected_before_writes(self):
|
|
from deploy.dedicated import validate_runtime
|
|
outside = self.root / "outside"
|
|
outside.mkdir()
|
|
project = self.root / "project"
|
|
project.mkdir()
|
|
try:
|
|
(project / "runtime").symlink_to(outside, target_is_directory=True)
|
|
except OSError:
|
|
self.skipTest("OS account cannot create directory symlinks")
|
|
with self.assertRaises(ValueError):
|
|
validate_runtime(project)
|
|
with self.assertRaises(ValueError):
|
|
Settings(project / "runtime", []).prepare()
|
|
self.assertEqual(list(outside.iterdir()), [])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|