"""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()