fix updater

Author: oxGorou

UXTU4Linux/Assets/Modules/updater.py

@@ -1,3 +1,4 @@
+import hashlib
 import json
 import os
 import shutil
@@ -14,7 +15,14 @@
 from .service import restart_service, service_running
 
 _STABLE_URL = "https://github.com/HorizonUnix/UXTU4Linux/releases/latest/download/UXTU4Linux.zip"
-_BETA_URL   = "https://github.com/HorizonUnix/UXTU4Linux/releases/download/U4L-Beta/UXTU4Linux.zip"
+_BETA_URL = "https://github.com/HorizonUnix/UXTU4Linux/releases/download/U4L-Beta/UXTU4Linux.zip"
+_STABLE_SHA256_URL = "https://github.com/HorizonUnix/UXTU4Linux/releases/latest/download/UXTU4Linux.zip.sha256"
+_BETA_SHA256_URL = "https://github.com/HorizonUnix/UXTU4Linux/releases/download/U4L-Beta/UXTU4Linux.zip.sha256"
+
+_SHA256_URL: dict[str, str] = {
+    _STABLE_URL: _STABLE_SHA256_URL,
+    _BETA_URL: _BETA_SHA256_URL,
+}
 
 
 def _ver_tuple(v: str) -> tuple:
@@ -27,12 +35,29 @@ def _ver_tuple(v: str) -> tuple:
         return (0,)
 
 
-def _release_url(url: str) -> str:
+def _get_release_tag(url: str) -> str:
     return url.rstrip("/").split("/")[-1]
 
 
+def _sha256_file(path: str) -> str:
+    h = hashlib.sha256()
+    with open(path, "rb") as f:
+        for chunk in iter(lambda: f.read(1024 * 1024), b""):
+            h.update(chunk)
+    return h.hexdigest()
+
+
+def _fetch_expected_sha256(sha_url: str) -> str:
+    with urllib.request.urlopen(sha_url, timeout=10) as resp:
+        raw = resp.read().decode("utf-8", errors="replace").strip()
+    expected = raw.split()[0].lower()
+    if len(expected) != 64 or not all(c in "0123456789abcdef" for c in expected):
+        raise RuntimeError(f"Invalid checksum format from {sha_url}")
+    return expected
+
+
 def get_latest_version() -> str:
-    tag = _release_url(urllib.request.urlopen(cfg.LATEST_VER_URL, timeout=5).geturl())
+    tag = _get_release_tag(urllib.request.urlopen(cfg.LATEST_VER_URL, timeout=5).geturl())
     if not tag or not tag.lstrip("v")[:1].isdigit():
         raise ValueError(f"Unexpected version tag: {tag!r}")
     return tag
@@ -44,7 +69,7 @@ def get_changelog() -> str:
     return data.get("body", "No changelog available.")
 
 
-def get_beta_commit() -> str | None:
+def get_beta_tag_sha() -> str | None:
     req = urllib.request.Request(
         "https://api.github.com/repos/HorizonUnix/UXTU4Linux/git/ref/tags/U4L-Beta",
         headers={"Accept": "application/vnd.github+json"},
@@ -54,11 +79,15 @@ def get_beta_commit() -> str | None:
     return sha[:7] if sha else None
 
 
+def get_beta_commit() -> str | None:
+    return get_beta_tag_sha()
+
+
 def _do_update(url: str = _STABLE_URL) -> None:
 
     module_file = Path(__file__).resolve()
-    assets_dir  = str(module_file.parent.parent)
-    src_dir     = str(module_file.parent.parent.parent)
+    assets_dir = str(module_file.parent.parent)
+    src_dir = str(module_file.parent.parent.parent)
     install_dir = str(module_file.parent.parent.parent.parent)
 
     zip_path = os.path.join(install_dir, "UXTU4Linux.zip")
@@ -69,7 +98,15 @@ def _do_update(url: str = _STABLE_URL) -> None:
     presets_src = os.path.join(assets_dir, "custom.json")
 
     def _sudo(*args: str) -> int:
-        return subprocess.run(["sudo", *args]).returncode
+        result = subprocess.run(["sudo", *args], capture_output=True, text=True)
+        if result.returncode != 0:
+            cmd = " ".join(["sudo", *args])
+            print(f"  Command failed ({result.returncode}): {cmd}")
+            if result.stderr and result.stderr.strip():
+                print(f"  stderr: {result.stderr.strip()}")
+            if result.stdout and result.stdout.strip():
+                print(f"  stdout: {result.stdout.strip()}")
+        return result.returncode
 
     try:
         if os.path.exists(config_src):
@@ -80,6 +117,19 @@ def _sudo(*args: str) -> int:
         print("  Downloading update...")
         urllib.request.urlretrieve(url, zip_path)
 
+        sha_url = _SHA256_URL.get(url)
+        if sha_url:
+            try:
+                print("  Verifying integrity...")
+                expected = _fetch_expected_sha256(sha_url)
+                actual = _sha256_file(zip_path)
+                if actual != expected:
+                    raise RuntimeError(
+                        f"Integrity check failed (expected {expected}, got {actual})."
+                    )
+            except urllib.error.URLError:
+                print("  Warning: checksum file not available — skipping integrity check.")
+
         print("  Extracting...")
         with zipfile.ZipFile(zip_path, "r") as zf:
             zf.extractall(new_folder)
@@ -200,4 +250,4 @@ def check_updates() -> None:
             "It may be unstable and is intended for testing only."
         )
         if not confirm("Continue?", subtitle=subtitle):
-            sys.exit("Quitting.")
+            sys.exit("Quitting.")