bca04f4ca2
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.
522 lines
23 KiB
Python
522 lines
23 KiB
Python
"""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()
|