199 lines
9.6 KiB
Python
199 lines
9.6 KiB
Python
import base64
|
|
import hashlib
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import shutil
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
import uuid
|
|
import zipfile
|
|
import zlib
|
|
|
|
REPO = Path(__file__).resolve().parents[2]
|
|
spec = importlib.util.spec_from_file_location("publish_cdn", REPO / "Tools/publish_cdn.py")
|
|
publish = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(publish)
|
|
|
|
|
|
class PublishTests(unittest.TestCase):
|
|
def setUp(self):
|
|
(REPO / "Temp").mkdir(exist_ok=True)
|
|
self.temp = tempfile.TemporaryDirectory(prefix="cdn-publish-tests-", dir=REPO / "Temp")
|
|
self.directory = Path(self.temp.name)
|
|
self.stage = self.directory / "stage"
|
|
self.root = self.directory / "CDN website with spaces"
|
|
self.local = self.directory / "localappdata"
|
|
self.local.mkdir()
|
|
self.env = dict(os.environ, LOCALAPPDATA=str(self.local))
|
|
self.old = self.create_stage(b"old index", b"old bundle")
|
|
self.root.mkdir()
|
|
for item in self.old["files"]:
|
|
target = self.root / item["path"]
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
shutil.copyfile(item["source"], target)
|
|
self.old_version = (self.root / "cdn/version.json").read_bytes()
|
|
self.plan = self.create_stage(b"new index", b"new bundle")
|
|
self.request = publish.request_for({"remote_root": str(self.root)}, self.plan, uuid.uuid4().hex)
|
|
self.request["expectedVersionSha256"] = hashlib.sha256(self.old_version).hexdigest()
|
|
|
|
def tearDown(self):
|
|
self.temp.cleanup()
|
|
|
|
def create_stage(self, index, bundle):
|
|
directory = self.stage / "1.0.0/DefaultPackage"
|
|
directory.mkdir(parents=True, exist_ok=True)
|
|
for file in directory.iterdir():
|
|
file.unlink()
|
|
md5 = hashlib.md5(index).hexdigest()
|
|
(directory / f"DefaultPackage_{md5}.bytes").write_bytes(index)
|
|
(directory / f"DefaultPackage_{md5}.hash").write_text(zlib.crc32(index).to_bytes(4, "little").hex())
|
|
(directory / "DefaultPackage.version").write_text(md5)
|
|
(directory / f"payload_{hashlib.md5(bundle).hexdigest()}.bundle").write_bytes(bundle)
|
|
version = {"appVersionMin": "1.0.0", "appVersionMax": "1.0.0", "packVersion": md5,
|
|
"resVersion": "", "cdnRoot": "https://www.xworld.ren/game/Friends/webgl/yoo/DefaultPackage",
|
|
"packages": ["DefaultPackage"], "notice": "", "forceUpdate": False}
|
|
publish.write_json(self.stage / "version.prod.json", version)
|
|
return publish.make_plan(self.stage, self.directory / "plan.json")
|
|
|
|
def invoke(self, action, request=None, success=True):
|
|
if not shutil.which("powershell.exe"):
|
|
self.skipTest("Windows PowerShell required for remote transaction tests")
|
|
payload = base64.b64encode(json.dumps(request or self.request).encode()).decode()
|
|
result = subprocess.run(["powershell.exe", "-NoProfile", "-NonInteractive", "-File", str(publish.REMOTE_SCRIPT),
|
|
"-Action", action, "-RequestBase64", payload, "-StateRoot", str(self.local / "AIGameCdnPublish")], env=self.env,
|
|
capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=30)
|
|
if success:
|
|
self.assertEqual(result.returncode, 0, result.stderr + result.stdout)
|
|
return json.loads(result.stdout)
|
|
self.assertNotEqual(result.returncode, 0)
|
|
return result.stderr
|
|
|
|
def prepare(self):
|
|
comparison = self.invoke("Plan")
|
|
prepared = self.invoke("Prepare")
|
|
publish.make_archive(self.plan, comparison["changed"], Path(prepared["archive"]))
|
|
return Path(prepared["archive"])
|
|
|
|
def test_local_manifest_crc_and_md5(self):
|
|
directory = self.stage / "1.0.0/DefaultPackage"
|
|
next(directory.glob("*.bytes")).write_bytes(b"corrupt")
|
|
with self.assertRaisesRegex(ValueError, "MD5"):
|
|
publish.make_plan(self.stage, self.directory / "bad-plan.json")
|
|
|
|
def test_local_crc_failure(self):
|
|
next((self.stage / "1.0.0/DefaultPackage").glob("*.hash")).write_text("00000000")
|
|
with self.assertRaisesRegex(ValueError, "hash marker"):
|
|
publish.make_plan(self.stage, self.directory / "bad-plan.json")
|
|
|
|
def test_archive_rejects_changed_local_payload(self):
|
|
entry = next(f for f in self.plan["files"] if f["path"].endswith(".bundle"))
|
|
Path(entry["source"]).write_bytes(b"changed after planning")
|
|
with self.assertRaisesRegex(ValueError, "changed since planning"):
|
|
publish.make_archive(self.plan, [entry["path"]], self.directory / "bad.zip")
|
|
|
|
def test_reject_traversal_and_unplanned_file(self):
|
|
for path in ("../version.json", "yoo/Art/..", "yoo/Art/../../evil", "cdn/secret.txt"):
|
|
with self.assertRaises(ValueError):
|
|
publish.validate_path(path)
|
|
with self.assertRaises(ValueError):
|
|
publish.make_archive(self.plan, ["unknown"], self.directory / "bad.zip")
|
|
|
|
def test_plan_is_read_only(self):
|
|
result = self.invoke("Plan")
|
|
self.assertIn("cdn/version.json", result["changed"])
|
|
self.assertFalse((self.local / "AIGameCdnPublish").exists())
|
|
self.assertEqual((self.root / "cdn/version.json").read_bytes(), self.old_version)
|
|
|
|
def startup_export(self):
|
|
export = self.directory / "export"
|
|
(export / "minigame").mkdir(parents=True)
|
|
(export / "webgl").mkdir()
|
|
data = b"new Unity shell startup data"
|
|
tag = hashlib.md5(data).hexdigest()[8:24]
|
|
name = tag + ".webgl.data.unityweb.bin.txt"
|
|
(export / "webgl" / name).write_bytes(data)
|
|
(export / "minigame/game.js").write_text(
|
|
f"const config={{DATA_FILE_MD5:'{tag}',DATA_FILE_SIZE:'{len(data)}',"
|
|
"DATA_CDN:'https://www.xworld.ren/game/Friends/webgl/'};", encoding="utf-8")
|
|
return export, name, data
|
|
|
|
def test_startup_export_transaction(self):
|
|
export, name, data = self.startup_export()
|
|
self.plan = publish.make_plan(self.stage, self.directory / "startup-plan.json", export)
|
|
self.request = publish.request_for({"remote_root": str(self.root)}, self.plan, uuid.uuid4().hex)
|
|
self.request["expectedVersionSha256"] = hashlib.sha256(self.old_version).hexdigest()
|
|
self.prepare()
|
|
self.assertTrue(self.invoke("Publish")["published"])
|
|
self.assertEqual((self.root / name).read_bytes(), data)
|
|
self.assertTrue(self.invoke("Rollback")["rolledBack"])
|
|
self.assertEqual((self.root / name).read_bytes(), data)
|
|
self.assertEqual((self.root / "cdn/version.json").read_bytes(), self.old_version)
|
|
|
|
def test_startup_export_rejects_corruption(self):
|
|
export, name, data = self.startup_export()
|
|
(export / "webgl" / name).write_bytes(b"x" * len(data))
|
|
with self.assertRaisesRegex(ValueError, "size or MD5"):
|
|
publish.make_plan(self.stage, self.directory / "bad-startup.json", export)
|
|
|
|
def test_startup_export_rejects_unknown_and_unsafe_files(self):
|
|
for name in ("webgl.data", "game.js", "../1234567890abcdef.webgl.data.unityweb.bin.txt"):
|
|
with self.assertRaises(ValueError):
|
|
publish.validate_path(name)
|
|
request = dict(self.request)
|
|
request["files"] = [*self.request["files"], {"path": name,"size": 1,"sha256": "0" * 64}]
|
|
self.invoke("Plan", request=request, success=False)
|
|
|
|
def test_publish_noop_and_rollback(self):
|
|
self.prepare()
|
|
receipt = self.invoke("Publish")
|
|
self.assertTrue(receipt["published"])
|
|
self.assertEqual(json.loads((self.root / "cdn/version.json").read_bytes()), self.plan["version"])
|
|
self.assertEqual(self.invoke("Plan")["changed"], [])
|
|
old_index = self.old["version"]["packVersion"]
|
|
self.assertTrue((self.root / f"yoo/DefaultPackage/DefaultPackage_{old_index}.bytes").exists())
|
|
self.assertTrue(self.invoke("Rollback")["rolledBack"])
|
|
self.assertEqual((self.root / "cdn/version.json").read_bytes(), self.old_version)
|
|
|
|
def test_corrupt_upload_keeps_live_version(self):
|
|
archive = self.prepare()
|
|
with zipfile.ZipFile(archive, "w") as stream:
|
|
item = self.plan["files"][0]
|
|
stream.writestr(item["path"], b"x" * item["size"])
|
|
self.invoke("Publish", success=False)
|
|
self.assertEqual((self.root / "cdn/version.json").read_bytes(), self.old_version)
|
|
|
|
def test_incomplete_upload_keeps_live_version(self):
|
|
archive = self.prepare()
|
|
with zipfile.ZipFile(archive, "w"):
|
|
pass
|
|
self.invoke("Publish", success=False)
|
|
self.assertEqual((self.root / "cdn/version.json").read_bytes(), self.old_version)
|
|
|
|
def test_stale_version_keeps_live_version(self):
|
|
self.prepare()
|
|
self.request["expectedVersionSha256"] = "0" * 64
|
|
self.invoke("Publish", success=False)
|
|
self.assertEqual((self.root / "cdn/version.json").read_bytes(), self.old_version)
|
|
|
|
def test_zip_traversal_keeps_live_version(self):
|
|
archive = self.prepare()
|
|
with zipfile.ZipFile(archive, "w") as stream:
|
|
stream.writestr("../evil", b"bad")
|
|
self.invoke("Publish", success=False)
|
|
self.assertFalse((archive.parent / "evil").exists())
|
|
self.assertEqual((self.root / "cdn/version.json").read_bytes(), self.old_version)
|
|
|
|
def test_rollback_refuses_later_release(self):
|
|
self.prepare()
|
|
self.invoke("Publish")
|
|
(self.root / "cdn/version.json").write_bytes(b"later release")
|
|
self.invoke("Rollback", success=False)
|
|
self.assertEqual((self.root / "cdn/version.json").read_bytes(), b"later release")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|