Verified MCL uploads: hash every file, repair over HTTP, gate the import
Small Control JSONs have arrived empty or wrong more than once, and the old check compared sizes only and skipped missing files. A copy of Deploy/Control on the dev PC still holds two 0-byte JSONs. Upload-Image.ps1 (tested on Windows PowerShell 5.1 in the win11 VM): - refuses an unfinished MCL cache: exactly one <OS>-Production.tag (MCL writes it last), no Control file newer than it, selected OS matches, every Control/Tools JSON non-empty and valid - checks payload against the SHA-256 MCL's own Control JSONs publish. MCL resumes downloads onto old files (FileMode.Append) and then trusts any file of the right size: the September junk-tail corruption. Offers to delete the bad cache files so MCL downloads them again - hashes every file into a manifest, robocopies, POSTs the manifest - re-sends bad small files over HTTP (no proxy), large ones by deleting the bad copy and robocopy /J (robocopy skips a same-size, same-time file as "modified" even with /IS /IT) Webapp: - services/upload_verify.py hashes the upload on the server's own disk in a background thread; reports missing, differing, bad-in-MCL-cache and stray files; PUT /api/upload/file writes temp, checks hash, renames - import of the upload folder only after a passed verification, with every file unchanged (size, mtime, ctime, inode), no stray files, and the target the OS allows (LTSC: Shop Floor only; else Standard or Engineer; unknown OS: none). Import, verify and PUT share one lock - paths matched case-insensitively against the disk, as Samba does - import page shows the upload, MCL run, verification and stray files with a remove button; record of each import kept on the image Reviewed adversarially; all 11 confirmed findings fixed with tests. 42 new unit tests. VM test: clean upload; same-size same-mtime damage on 3 files repaired in one round; stray file blocks import until removed; Shop Floor refused for a Pro upload; import records and clears the folder; missing tag, 0-byte JSON, junk-head WIM and late Control file each refused with nothing uploaded.
This commit is contained in:
cproudlock
521
webapp/tests/test_upload_verify.py
Normal file
521
webapp/tests/test_upload_verify.py
Normal file
@@ -0,0 +1,521 @@
|
||||
"""Verified MCL uploads: manifest checks, server-side hashing, the HTTP
|
||||
second channel, and the import gate.
|
||||
|
||||
Run from the webapp folder:
|
||||
PYTHONPATH=.:<deps> python -m unittest tests.test_upload_verify -v
|
||||
|
||||
Every config path points into a temp folder, so nothing on the host is
|
||||
touched. The planted failures are the real ones: a 0-byte Control JSON,
|
||||
a file with the right size and the wrong content (the September MCL
|
||||
cache corruption), a file SMB never delivered, and a symlink escape.
|
||||
"""
|
||||
import hashlib
|
||||
import io
|
||||
import json
|
||||
import os
|
||||
import shutil
|
||||
import sys
|
||||
import tempfile
|
||||
import time
|
||||
import unittest
|
||||
from unittest import mock
|
||||
|
||||
_T = tempfile.mkdtemp(prefix="pxe-upv-test-")
|
||||
for _k, _sub in {"SAMBA_SHARE": "samba", "CLONEZILLA_SHARE": "cz", "BLANCCO_REPORTS": "br",
|
||||
"ENROLLMENT_SHARE": "enr", "WEB_ROOT": "www", "IMAGING_DIR": "img", "UPDATES_DIR": "upd",
|
||||
"TFTP_ROOT": "tftp", "BIOS_DIR": "bios", "UPLOAD_DIR": "upl", "SAMBA_LOG_DIR": "smblog"}.items():
|
||||
os.environ.setdefault(_k, os.path.join(_T, _sub))
|
||||
os.makedirs(os.environ[_k], exist_ok=True)
|
||||
for _k in ("AUDIT_LOG", "HARDWARE_STATE", "BLANCCO_STATE", "DNSMASQ_LEASES", "APACHE_ACCESS_LOG",
|
||||
"DNSMASQ_SYSLOG", "NIC_STATUS_FILE", "NIC_CONF", "DNSMASQ_NIC_CONF", "DNSMASQ_CONF", "CLONEZILLA_STATE"):
|
||||
os.environ.setdefault(_k, os.path.join(_T, _k.lower()))
|
||||
os.makedirs(os.path.join(os.environ["ENROLLMENT_SHARE"], "ppkgs"), exist_ok=True)
|
||||
|
||||
WEBAPP = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||||
sys.path.insert(0, WEBAPP)
|
||||
os.chdir(WEBAPP)
|
||||
|
||||
import config # noqa: E402
|
||||
import app as A # noqa: E402
|
||||
from services import upload_verify as UV # noqa: E402
|
||||
|
||||
WIM = b"W" * 4096
|
||||
WIM_DIR = "Deploy/Operating Systems/Windows 11.0.26100.0 Pro x64/September 9, 2025"
|
||||
|
||||
|
||||
def sha(b):
|
||||
return hashlib.sha256(b).hexdigest()
|
||||
|
||||
|
||||
def os_json(product="Windows 11 Professional", wim=WIM):
|
||||
return json.dumps([{"operatingSystemVersion": {
|
||||
"productName": product, "versionNumber": "24H2", "buildNumber": "26100", "id": 21,
|
||||
"wim": {"hash": sha(wim).upper(), "size": len(wim),
|
||||
"destinationDir": "*destinationdir*\\" + WIM_DIR.replace("/", "\\")}}}]).encode()
|
||||
|
||||
|
||||
class Base(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.tmp = tempfile.mkdtemp(prefix="upv-", dir=_T)
|
||||
self.up = os.path.join(self.tmp, "upload")
|
||||
self.share = os.path.join(self.tmp, "samba")
|
||||
os.makedirs(self.up)
|
||||
os.makedirs(self.share)
|
||||
self.patches = [mock.patch.object(config, "UPLOAD_DIR", self.up),
|
||||
mock.patch.object(config, "SAMBA_SHARE", self.share),
|
||||
mock.patch.object(config, "SHARED_DIR", os.path.join(self.share, "_shared"))]
|
||||
for p in self.patches:
|
||||
p.start()
|
||||
self.files = {}
|
||||
|
||||
def tearDown(self):
|
||||
for p in self.patches:
|
||||
p.stop()
|
||||
shutil.rmtree(self.tmp, ignore_errors=True)
|
||||
|
||||
def put(self, rel, data, on_server=True):
|
||||
"""data = what the PC has; it goes into the manifest, and (unless
|
||||
on_server is False) onto the server too."""
|
||||
self.files[rel] = data
|
||||
if on_server:
|
||||
p = os.path.join(self.up, *rel.split("/"))
|
||||
os.makedirs(os.path.dirname(p), exist_ok=True)
|
||||
with open(p, "wb") as f:
|
||||
f.write(data)
|
||||
|
||||
def standard_upload(self, product="Windows 11 Professional"):
|
||||
self.put("Deploy/Control/OperatingSystem.json", os_json(product))
|
||||
self.put("Deploy/Control/packages.json", b"[]")
|
||||
self.put("Tools/user_selections.json", b'[{"OperatingSystemSelection": "21"}]')
|
||||
self.put(f"{WIM_DIR}/install.wim", WIM)
|
||||
|
||||
def manifest(self, uid="20261016-1200-PC1", **extra):
|
||||
m = {"manifest_version": 1, "upload_id": uid, "source_pc": "PC1",
|
||||
"mcl": {"os_name": "Windows 11 Professional", "finished": "2026-10-16T11:00:00"},
|
||||
"files": [{"path": r, "size": len(d), "sha256": sha(d)} for r, d in self.files.items()]}
|
||||
m.update(extra)
|
||||
return m
|
||||
|
||||
def verify(self, m=None):
|
||||
"""Synchronous verify the way the POST stores it."""
|
||||
m = m or self.manifest()
|
||||
raw = json.dumps(m, indent=1).encode()
|
||||
with open(os.path.join(self.up, UV.MANIFEST), "wb") as f:
|
||||
f.write(raw)
|
||||
return UV.run_verify(UV.parse_manifest(m), hashlib.sha256(raw).hexdigest())
|
||||
|
||||
|
||||
class ManifestTests(Base):
|
||||
def test_bad_paths_rejected(self):
|
||||
for p in ["../x", "a/../../x", "/etc/passwd", "C:/Windows/x", "C:\\x", "a//b", "./a",
|
||||
"a\\..\\..\\b", UV.MANIFEST, UV.STATE, "a/.x.json.part-0123456789abcdef", "", "a\x00b"]:
|
||||
with self.assertRaises(UV.UploadError, msg=p):
|
||||
UV.norm_path(p)
|
||||
|
||||
def test_backslashes_normalised(self):
|
||||
self.assertEqual(UV.norm_path("Deploy\\Control\\os.json"), "Deploy/Control/os.json")
|
||||
|
||||
def test_bad_manifests_rejected(self):
|
||||
ok = {"path": "a.json", "size": 1, "sha256": "0" * 64}
|
||||
for m in [None, [], {"manifest_version": 2, "upload_id": "x", "files": [ok]},
|
||||
{"manifest_version": 1, "upload_id": "../x", "files": [ok]},
|
||||
{"manifest_version": 1, "upload_id": "x", "files": []},
|
||||
{"manifest_version": 1, "upload_id": "x", "files": [dict(ok, sha256="zz")]},
|
||||
{"manifest_version": 1, "upload_id": "x", "files": [dict(ok, size=-1)]},
|
||||
{"manifest_version": 1, "upload_id": "x", "files": [dict(ok, size=True)]},
|
||||
{"manifest_version": 1, "upload_id": "x", "files": [ok, dict(ok, path="A.JSON")]}]:
|
||||
with self.assertRaises(UV.UploadError, msg=repr(m)):
|
||||
UV.parse_manifest(m)
|
||||
|
||||
def test_mcl_rel_mapping(self):
|
||||
self.assertEqual(UV._mcl_rel("*destinationdir*\\Deploy\\Packages\\X", "a.msu"), "Deploy/Packages/X/a.msu")
|
||||
self.assertEqual(UV._mcl_rel("C:\\Deploy\\Languages\\Y", "b.cab"), "Deploy/Languages/Y/b.cab")
|
||||
|
||||
|
||||
class VerifyTests(Base):
|
||||
def test_good_upload_passes(self):
|
||||
self.standard_upload()
|
||||
r = self.verify()
|
||||
self.assertTrue(r["ok"], r)
|
||||
self.assertTrue(r["importable"])
|
||||
self.assertEqual(r["allowed_targets"], ["gea-standard", "gea-engineer"])
|
||||
self.assertEqual(len(UV.load_detail()["snapshot"]), 4)
|
||||
|
||||
def test_ltsc_is_shopfloor_only(self):
|
||||
self.standard_upload(product="Windows 11 LTSC")
|
||||
r = self.verify()
|
||||
self.assertEqual(r["allowed_targets"], ["gea-shopfloor"])
|
||||
|
||||
def test_missing_file_caught(self):
|
||||
self.standard_upload()
|
||||
self.put("Deploy/Control/HardwareDriver.json", b"[]", on_server=False)
|
||||
r = self.verify()
|
||||
self.assertFalse(r["ok"])
|
||||
self.assertEqual(r["missing"], ["Deploy/Control/HardwareDriver.json"])
|
||||
|
||||
def test_zero_byte_json_on_server_caught(self):
|
||||
# SMB left the file at 0 bytes: size differs, so it is a mismatch.
|
||||
self.standard_upload()
|
||||
open(os.path.join(self.up, "Deploy", "Control", "packages.json"), "wb").close()
|
||||
r = self.verify()
|
||||
self.assertEqual([m["path"] for m in r["mismatch"]], ["Deploy/Control/packages.json"])
|
||||
|
||||
def test_same_size_wrong_content_caught(self):
|
||||
# The case the old size-only check could never see.
|
||||
self.standard_upload()
|
||||
with open(os.path.join(self.up, "Deploy", "Control", "packages.json"), "wb") as f:
|
||||
f.write(b"{}")
|
||||
r = self.verify()
|
||||
self.assertEqual(r["mismatch"], [{"path": "Deploy/Control/packages.json", "reason": "content differs"}])
|
||||
|
||||
def test_zero_byte_json_in_cache_caught(self):
|
||||
# Broken at the source: the PC's own copy is empty, so the hashes
|
||||
# agree, but an empty Control JSON is never right.
|
||||
self.standard_upload()
|
||||
self.put("Deploy/Control/boot.json", b"")
|
||||
r = self.verify()
|
||||
self.assertFalse(r["ok"])
|
||||
self.assertEqual(r["bad_in_cache"], [{"path": "Deploy/Control/boot.json", "reason": "empty Control JSON"}])
|
||||
|
||||
def test_september_pattern_caught(self):
|
||||
# MCL resumed onto an older file: right size, junk content, on the
|
||||
# PC. Manifest and server agree; MCL's published hash does not.
|
||||
self.standard_upload()
|
||||
self.put(f"{WIM_DIR}/install.wim", b"X" * 100 + WIM[100:])
|
||||
r = self.verify()
|
||||
self.assertFalse(r["ok"])
|
||||
self.assertEqual([b["path"] for b in r["bad_in_cache"]], [f"{WIM_DIR}/install.wim"])
|
||||
self.assertEqual(r["mismatch"], [])
|
||||
|
||||
def test_mcl_path_case_insensitive(self):
|
||||
# Windows paths ignore case; MCL's JSON and the folder may differ.
|
||||
self.standard_upload()
|
||||
del self.files[f"{WIM_DIR}/install.wim"]
|
||||
os.remove(os.path.join(self.up, *WIM_DIR.split("/"), "install.wim"))
|
||||
self.put(f"{WIM_DIR}/INSTALL.WIM", b"Z" * len(WIM))
|
||||
r = self.verify()
|
||||
self.assertEqual([b["path"] for b in r["bad_in_cache"]], [f"{WIM_DIR}/INSTALL.WIM"])
|
||||
|
||||
def test_extras_block_import_not_verify(self):
|
||||
self.standard_upload()
|
||||
os.makedirs(os.path.join(self.up, "Deploy", "Old"))
|
||||
with open(os.path.join(self.up, "Deploy", "Old", "stale.json"), "w") as f:
|
||||
f.write("{}")
|
||||
r = self.verify()
|
||||
self.assertTrue(r["ok"])
|
||||
self.assertFalse(r["importable"])
|
||||
self.assertEqual(r["extra"], ["Deploy/Old/stale.json"])
|
||||
|
||||
def test_symlink_escape_refused(self):
|
||||
self.standard_upload()
|
||||
outside = os.path.join(self.tmp, "outside")
|
||||
os.makedirs(outside)
|
||||
with open(os.path.join(outside, "x.json"), "wb") as f:
|
||||
f.write(b"[]")
|
||||
os.symlink(outside, os.path.join(self.up, "Evil"))
|
||||
self.put("Evil/x.json", b"[]", on_server=False)
|
||||
r = self.verify()
|
||||
self.assertIn("Evil/x.json", r["missing"])
|
||||
self.assertIn("Evil", r["extra"]) # the link itself blocks the import
|
||||
|
||||
def test_second_round_reuses_hashes(self):
|
||||
self.standard_upload()
|
||||
self.verify()
|
||||
with mock.patch.object(UV, "_sha256", side_effect=AssertionError("rehashed")):
|
||||
r = self.verify()
|
||||
self.assertTrue(r["ok"])
|
||||
|
||||
def rewrite_keep_mtime(self, rel, data):
|
||||
"""What robocopy does: new content, source mtime kept."""
|
||||
p = os.path.join(self.up, *rel.split("/"))
|
||||
st = os.stat(p)
|
||||
time.sleep(0.01)
|
||||
with open(p, "wb") as f:
|
||||
f.write(data)
|
||||
os.utime(p, ns=(st.st_atime_ns, st.st_mtime_ns))
|
||||
|
||||
def test_bad_copy_then_good_recopy_same_mtime(self):
|
||||
# Review finding 2: a re-copied large file must be hashed again even
|
||||
# though robocopy gives it the same size and mtime as the bad copy.
|
||||
self.standard_upload()
|
||||
wim_rel = f"{WIM_DIR}/install.wim"
|
||||
self.rewrite_keep_mtime(wim_rel, b"Q" * len(WIM))
|
||||
self.assertEqual([m["path"] for m in self.verify()["mismatch"]], [wim_rel])
|
||||
self.rewrite_keep_mtime(wim_rel, WIM)
|
||||
self.assertTrue(self.verify()["ok"])
|
||||
|
||||
def test_good_file_corrupted_in_place_same_mtime(self):
|
||||
self.standard_upload()
|
||||
self.assertTrue(self.verify()["ok"])
|
||||
self.rewrite_keep_mtime("Deploy/Control/packages.json", b"{}")
|
||||
r = self.verify()
|
||||
self.assertEqual([m["path"] for m in r["mismatch"]], ["Deploy/Control/packages.json"])
|
||||
|
||||
def test_os_picked_by_selected_id(self):
|
||||
# Review finding 3: LTSC listed first, Pro (21) selected.
|
||||
self.standard_upload()
|
||||
ltsc = json.loads(os_json("Windows 11 LTSC"))[0]
|
||||
ltsc["operatingSystemVersion"]["id"] = 20
|
||||
pro = json.loads(os_json())[0]
|
||||
self.put("Deploy/Control/OperatingSystem.json", json.dumps([ltsc, pro]).encode())
|
||||
r = self.verify(self.manifest(mcl={"os_id": 21}))
|
||||
self.assertEqual(r["allowed_targets"], ["gea-standard", "gea-engineer"])
|
||||
r = self.verify(self.manifest(mcl={"os_id": 20}))
|
||||
self.assertEqual(r["allowed_targets"], ["gea-shopfloor"])
|
||||
|
||||
def test_unknown_os_fails_closed(self):
|
||||
# Review finding 4: valid JSON, wrong shape.
|
||||
self.standard_upload()
|
||||
self.put("Deploy/Control/OperatingSystem.json", b'{"x": 1}')
|
||||
r = self.verify()
|
||||
self.assertTrue(r["ok"])
|
||||
self.assertEqual(r["allowed_targets"], [])
|
||||
self.assertFalse(r["importable"])
|
||||
ok, reason, _ = UV.gate("gea-standard")
|
||||
self.assertFalse(ok)
|
||||
self.assertIn("Cannot tell which OS", reason)
|
||||
|
||||
def test_nested_part_file_is_stray(self):
|
||||
# Review finding 5: debris from a killed PUT must not ride along.
|
||||
self.standard_upload()
|
||||
with open(os.path.join(self.up, "Deploy", "Control", ".x.msu.part-0123456789abcdef"), "wb") as f:
|
||||
f.write(b"junk")
|
||||
r = self.verify()
|
||||
self.assertEqual(r["extra"], ["Deploy/Control/.x.msu.part-0123456789abcdef"])
|
||||
self.assertEqual(UV.remove_extras(), 1)
|
||||
|
||||
def test_case_differs_from_manifest(self):
|
||||
# Review finding 8: manifest says packages.json, disk has Packages.json.
|
||||
self.standard_upload()
|
||||
os.rename(os.path.join(self.up, "Deploy", "Control", "packages.json"),
|
||||
os.path.join(self.up, "Deploy", "Control", "Packages.json"))
|
||||
r = self.verify()
|
||||
self.assertTrue(r["ok"], r)
|
||||
self.assertEqual(r["extra"], [])
|
||||
self.assertTrue(UV.gate("gea-standard")[0])
|
||||
|
||||
def test_two_cases_on_disk(self):
|
||||
self.standard_upload()
|
||||
with open(os.path.join(self.up, "Deploy", "Control", "PACKAGES.json"), "wb") as f:
|
||||
f.write(b"[]")
|
||||
r = self.verify()
|
||||
self.assertFalse(r["ok"])
|
||||
self.assertIn("differ only in case", r["mismatch"][0]["reason"])
|
||||
|
||||
|
||||
class ReceiveTests(Base):
|
||||
def setUp(self):
|
||||
super().setUp()
|
||||
self.standard_upload()
|
||||
self.put("Deploy/Control/HardwareDriver.json", b'[{"x": 1}]', on_server=False)
|
||||
self.m = self.manifest()
|
||||
self.verify(self.m)
|
||||
self.uid = self.m["upload_id"]
|
||||
|
||||
def send(self, rel, data, uid=None, length=None):
|
||||
return UV.receive_file(uid or self.uid, rel, io.BytesIO(data), len(data) if length is None else length)
|
||||
|
||||
def test_resend_repairs_and_verify_passes(self):
|
||||
r = self.send("Deploy/Control/HardwareDriver.json", b'[{"x": 1}]')
|
||||
self.assertEqual(r["size"], 10)
|
||||
self.assertTrue(self.verify(self.m)["ok"])
|
||||
|
||||
def test_wrong_hash_leaves_file_alone(self):
|
||||
p = os.path.join(self.up, "Deploy", "Control", "packages.json")
|
||||
with self.assertRaises(UV.UploadError) as cm:
|
||||
self.send("Deploy/Control/packages.json", b"{}")
|
||||
self.assertEqual(cm.exception.status, 422)
|
||||
with open(p, "rb") as f:
|
||||
self.assertEqual(f.read(), b"[]")
|
||||
self.assertEqual([n for n in os.listdir(os.path.dirname(p)) if ".part-" in n], [])
|
||||
|
||||
def test_wrong_upload_id(self):
|
||||
with self.assertRaises(UV.UploadError) as cm:
|
||||
self.send("Deploy/Control/packages.json", b"[]", uid="other")
|
||||
self.assertEqual(cm.exception.status, 409)
|
||||
|
||||
def test_not_in_manifest(self):
|
||||
with self.assertRaises(UV.UploadError) as cm:
|
||||
self.send("Deploy/Control/new.json", b"[]")
|
||||
self.assertEqual(cm.exception.status, 404)
|
||||
|
||||
def test_length_must_match(self):
|
||||
with self.assertRaises(UV.UploadError):
|
||||
self.send("Deploy/Control/packages.json", b"[]", length=99)
|
||||
|
||||
def test_traversal_refused(self):
|
||||
with self.assertRaises(UV.UploadError):
|
||||
self.send("../../etc/x", b"[]")
|
||||
|
||||
def test_big_file_goes_back_to_robocopy(self):
|
||||
with mock.patch.object(UV, "SMALL_MAX", 100):
|
||||
with self.assertRaises(UV.UploadError) as cm:
|
||||
self.send(f"{WIM_DIR}/install.wim", WIM)
|
||||
self.assertEqual(cm.exception.status, 413)
|
||||
|
||||
def test_put_reuses_existing_case(self):
|
||||
# The folder on disk is "deploy/control" (lower case): the PUT must
|
||||
# write into it, not create a second "Deploy/Control".
|
||||
os.rename(os.path.join(self.up, "Deploy"), os.path.join(self.up, "deploy"))
|
||||
os.rename(os.path.join(self.up, "deploy", "Control"), os.path.join(self.up, "deploy", "control"))
|
||||
r = self.send("Deploy/Control/HardwareDriver.json", b'[{"x": 1}]')
|
||||
self.assertEqual(r["written_as"], "deploy/control/HardwareDriver.json")
|
||||
self.assertFalse(os.path.exists(os.path.join(self.up, "Deploy")))
|
||||
self.assertTrue(self.verify(self.m)["ok"])
|
||||
|
||||
def test_put_refused_while_locked(self):
|
||||
self.assertTrue(UV.try_lock())
|
||||
try:
|
||||
with self.assertRaises(UV.UploadError) as cm:
|
||||
self.send("Deploy/Control/HardwareDriver.json", b'[{"x": 1}]')
|
||||
self.assertEqual(cm.exception.status, 409)
|
||||
finally:
|
||||
UV.unlock()
|
||||
|
||||
def test_link_in_parent_refused(self):
|
||||
outside = os.path.join(self.tmp, "outside2")
|
||||
os.makedirs(outside)
|
||||
shutil.rmtree(os.path.join(self.up, "Tools"))
|
||||
os.symlink(outside, os.path.join(self.up, "Tools"))
|
||||
with self.assertRaises(UV.UploadError):
|
||||
self.send("Tools/user_selections.json", self.files["Tools/user_selections.json"])
|
||||
self.assertEqual(os.listdir(outside), [])
|
||||
|
||||
|
||||
class GateTests(Base):
|
||||
def test_no_manifest(self):
|
||||
ok, reason, _ = UV.gate("gea-standard")
|
||||
self.assertFalse(ok)
|
||||
self.assertIn("Upload-Image.ps1", reason)
|
||||
|
||||
def test_unverified(self):
|
||||
self.standard_upload()
|
||||
with open(os.path.join(self.up, UV.MANIFEST), "w") as f:
|
||||
json.dump(self.manifest(), f)
|
||||
ok, reason, _ = UV.gate("gea-standard")
|
||||
self.assertFalse(ok)
|
||||
self.assertIn("not been verified", reason)
|
||||
|
||||
def test_pass_then_changed_file(self):
|
||||
self.standard_upload()
|
||||
self.verify()
|
||||
self.assertTrue(UV.gate("gea-standard")[0])
|
||||
self.assertFalse(UV.gate("gea-shopfloor")[0])
|
||||
p = os.path.join(self.up, "Deploy", "Control", "packages.json")
|
||||
time.sleep(0.01)
|
||||
with open(p, "wb") as f:
|
||||
f.write(b"[]")
|
||||
ok, reason, _ = UV.gate("gea-standard")
|
||||
self.assertFalse(ok)
|
||||
self.assertIn("changed since verification", reason)
|
||||
|
||||
def test_manifest_swapped_after_verify(self):
|
||||
self.standard_upload()
|
||||
self.verify()
|
||||
with open(os.path.join(self.up, UV.MANIFEST), "w") as f:
|
||||
json.dump(self.manifest(), f) # same content, different bytes
|
||||
self.assertFalse(UV.gate("gea-standard")[0])
|
||||
|
||||
def test_extras_then_removed(self):
|
||||
self.standard_upload()
|
||||
with open(os.path.join(self.up, "stale.txt"), "w") as f:
|
||||
f.write("old")
|
||||
self.verify()
|
||||
ok, reason, info = UV.gate("gea-standard")
|
||||
self.assertFalse(ok)
|
||||
self.assertEqual(info["extra_count"], 1)
|
||||
self.assertEqual(UV.remove_extras(), 1)
|
||||
self.assertTrue(UV.gate("gea-standard")[0])
|
||||
|
||||
|
||||
class RouteTests(Base):
|
||||
def setUp(self):
|
||||
super().setUp()
|
||||
A.app.config["TESTING"] = True
|
||||
self.c = A.app.test_client()
|
||||
|
||||
def wait(self):
|
||||
for _ in range(200):
|
||||
s = self.c.get("/api/upload/verify").get_json()
|
||||
if s.get("state") in ("done", "error"):
|
||||
return s
|
||||
time.sleep(0.02)
|
||||
self.fail("verify never finished")
|
||||
|
||||
def test_full_flow_through_http(self):
|
||||
self.standard_upload()
|
||||
self.put("Deploy/Control/HardwareDriver.json", b'[{"x": 1}]', on_server=False)
|
||||
m = self.manifest()
|
||||
r = self.c.post("/api/upload/verify", json=m)
|
||||
self.assertEqual(r.status_code, 202, r.get_data(as_text=True))
|
||||
s = self.wait()
|
||||
self.assertEqual(s["missing"], ["Deploy/Control/HardwareDriver.json"])
|
||||
r = self.c.put("/api/upload/file", query_string={"upload_id": m["upload_id"],
|
||||
"path": "Deploy/Control/HardwareDriver.json"}, data=b'[{"x": 1}]')
|
||||
self.assertEqual(r.status_code, 200, r.get_data(as_text=True))
|
||||
self.assertEqual(self.c.post("/api/upload/verify", json=m).status_code, 202)
|
||||
s = self.wait()
|
||||
self.assertTrue(s["ok"] and s["importable"], s)
|
||||
|
||||
# Import: refused into the wrong image, accepted into the right one.
|
||||
with self.c.session_transaction() as sess:
|
||||
sess["_csrf_token"] = "t"
|
||||
r = self.c.post("/images/import", data={"_csrf_token": "t", "source": self.up, "target": "gea-shopfloor"})
|
||||
self.assertFalse(os.path.isdir(os.path.join(self.share, "gea-shopfloor", "Deploy", "Control")))
|
||||
r = self.c.post("/images/import", data={"_csrf_token": "t", "source": self.up, "target": "gea-standard"})
|
||||
self.assertEqual(r.status_code, 302)
|
||||
img = os.path.join(self.share, "gea-standard")
|
||||
self.assertTrue(os.path.isfile(os.path.join(img, "Deploy", "Control", "HardwareDriver.json")))
|
||||
with open(os.path.join(img, ".mcl-import.json")) as f:
|
||||
rec = json.load(f)
|
||||
self.assertEqual(rec["upload_id"], m["upload_id"])
|
||||
self.assertFalse(os.path.exists(os.path.join(img, UV.MANIFEST)))
|
||||
self.assertFalse(os.path.exists(os.path.join(img, UV.STATE)))
|
||||
self.assertFalse(os.path.exists(os.path.join(self.up, UV.MANIFEST)))
|
||||
self.assertFalse(os.path.exists(os.path.join(img, UV.DETAIL)))
|
||||
|
||||
def test_bad_manifest_400(self):
|
||||
r = self.c.post("/api/upload/verify", json={"manifest_version": 1, "upload_id": "x",
|
||||
"files": [{"path": "../x", "size": 1, "sha256": "0" * 64}]})
|
||||
self.assertEqual(r.status_code, 400)
|
||||
|
||||
def test_unverified_upload_refused_by_import(self):
|
||||
self.standard_upload()
|
||||
with self.c.session_transaction() as sess:
|
||||
sess["_csrf_token"] = "t"
|
||||
self.c.post("/images/import", data={"_csrf_token": "t", "source": self.up, "target": "gea-standard"})
|
||||
self.assertFalse(os.path.exists(os.path.join(self.share, "gea-standard", "Deploy", "Control")))
|
||||
|
||||
def test_import_refused_while_locked(self):
|
||||
# Review finding 7: no import while a verify or PUT holds the lock.
|
||||
self.standard_upload()
|
||||
self.verify()
|
||||
with self.c.session_transaction() as sess:
|
||||
sess["_csrf_token"] = "t"
|
||||
self.assertTrue(UV.try_lock())
|
||||
try:
|
||||
self.c.post("/images/import", data={"_csrf_token": "t", "source": self.up, "target": "gea-standard"})
|
||||
finally:
|
||||
UV.unlock()
|
||||
self.assertFalse(os.path.exists(os.path.join(self.share, "gea-standard", "Deploy", "Control")))
|
||||
self.assertTrue(os.path.exists(os.path.join(self.up, UV.MANIFEST)))
|
||||
self.assertTrue(UV.try_lock()) # released again
|
||||
UV.unlock()
|
||||
|
||||
def test_import_page_renders(self):
|
||||
self.standard_upload()
|
||||
with open(os.path.join(self.up, "stale.txt"), "w") as f:
|
||||
f.write("old")
|
||||
self.verify()
|
||||
r = self.c.get("/images/import")
|
||||
self.assertEqual(r.status_code, 200)
|
||||
body = r.get_data(as_text=True)
|
||||
self.assertIn("not part of this upload", body)
|
||||
|
||||
|
||||
import atexit # noqa: E402
|
||||
atexit.register(shutil.rmtree, _T, True)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user