Enable true one-click panel update from Settings (v2.5.1).

Download release ZIP from git.evilfox.cc when git is unavailable, add upgrade_panel API, and show a primary Update panel button.
2026-07-28 09:41:51 +03:00
parent 77c6f0dab7
commit 599e0a487c
9 changed files with 350 additions and 83 deletions
@@ -1,14 +1,19 @@
-"""Panel self-update from the EvilFox Gitea repository."""
+"""Panel self-update from the EvilFox Gitea repository (git or release archive)."""

 from __future__ import annotations

 import logging
 import os
 import re
+import shutil
 import subprocess
 import sys
+import tempfile
 import threading
 import time
+import urllib.error
+import urllib.request
+import zipfile
 from typing import Optional

 logger = logging.getLogger(__name__)
@@ -18,6 +23,13 @@ DEFAULT_GIT_URL = 'https://git.evilfox.cc/test2/Amnezia-Web-Panel-main.git'
 DEFAULT_API_LATEST = 'https://git.evilfox.cc/api/v1/repos/test2/Amnezia-Web-Panel-main/releases/latest'
 UPDATE_REMOTE = 'panel-update'

+ARCHIVE_SKIP_DIRS = frozenset({
+    '__pycache__', '.git', 'bin', 'data', 'node_modules', '.venv', 'venv', '.cursor',
+})
+ARCHIVE_SKIP_FILES = frozenset({
+    '.env', 'data.json', 'tunnels_state.json',
+})
+

 def _env(name: str, default: str) -> str:
    return (os.environ.get(name) or default).strip()
@@ -51,7 +63,6 @@ def parse_version(value: str) -> tuple:
            parts.append(int(m.group(1)) if m else 0)
    while len(parts) < 3:
        parts.append(0)
-    # release > prerelease: empty pre sorts higher
    pre_rank = 0 if not pre else 1
    return (parts[0], parts[1], parts[2], pre_rank, pre)

@@ -60,6 +71,13 @@ def version_gt(a: str, b: str) -> bool:
    return parse_version(a) > parse_version(b)


+def normalize_tag(value: str) -> str:
+    tag = (value or '').strip()
+    if not tag:
+        return ''
+    return tag if tag.startswith('v') else f'v{tag}'
+
+
 def _run(cmd: list, cwd: str, timeout: int = 120) -> tuple[int, str, str]:
    try:
        proc = subprocess.run(
@@ -83,23 +101,62 @@ class UpdateManager:
        self.app_root = os.path.abspath(app_root)
        self.current_version = current_version

+    def _in_docker(self) -> bool:
+        return os.path.exists('/.dockerenv') or os.environ.get('PANEL_IN_DOCKER') == '1'
+
+    def _docker_update_allowed(self) -> bool:
+        return os.environ.get('PANEL_UPDATE_ALLOW_DOCKER', '').strip().lower() in ('1', 'true', 'yes')
+
+    def _app_root_writable(self) -> bool:
+        probe = os.path.join(self.app_root, '.panel_update_write_probe')
+        try:
+            with open(probe, 'w', encoding='utf-8') as fh:
+                fh.write('ok')
+            os.remove(probe)
+            return True
+        except OSError:
+            return False
+
    def detect_mode(self) -> dict:
        frozen = bool(getattr(sys, 'frozen', False))
-        in_docker = os.path.exists('/.dockerenv') or os.environ.get('PANEL_IN_DOCKER') == '1'
+        in_docker = self._in_docker()
        git_dir = os.path.join(self.app_root, '.git')
        is_git = os.path.isdir(git_dir)
        git_ok = False
        if is_git and not frozen:
            code, _, _ = _run(['git', '--version'], self.app_root, timeout=10)
            git_ok = code == 0
-        can_auto = bool(git_ok and not frozen)
-        mode = 'git' if can_auto else ('docker' if in_docker else ('binary' if frozen else 'manual'))
+
+        writable = self._app_root_writable()
+        can_git = bool(git_ok and is_git and not frozen)
+        can_archive = bool(
+            writable
+            and not frozen
+            and (not in_docker or self._docker_update_allowed())
+        )
+        can_auto = can_git or can_archive
+
+        if can_git:
+            method = 'git'
+        elif can_archive:
+            method = 'archive'
+        elif in_docker:
+            method = 'docker'
+        elif frozen:
+            method = 'binary'
+        else:
+            method = 'manual'
+
        return {
-            'mode': mode,
+            'mode': method,
+            'update_method': method,
            'can_auto_update': can_auto,
+            'can_git_update': can_git,
+            'can_archive_update': can_archive,
            'is_git': is_git,
            'frozen': frozen,
            'in_docker': in_docker,
+            'writable': writable,
            'app_root': self.app_root,
            'repo_url': repo_url(),
            'git_url': git_url(),
@@ -131,31 +188,27 @@ class UpdateManager:
        code, out, _ = _run(['git', 'status', '--porcelain'], self.app_root, timeout=30)
        return code == 0 and bool(out.strip())

-    def apply_update(self, target_version: str, allow_dirty: bool = False) -> dict:
-        mode = self.detect_mode()
-        if not mode['can_auto_update']:
-            return {
-                'status': 'error',
-                'message': (
-                    'Auto-update requires a git checkout of the panel sources. '
-                    'Docker/binary installs: pull a new image or download a release manually.'
-                ),
-                'mode': mode,
-            }
-
-        target = (target_version or '').strip()
-        if not re.match(r'^v?\d+(\.\d+){0,3}([.-][\w.]+)?$', target):
-            return {'status': 'error', 'message': f'Invalid target version: {target_version}'}
-        if not target.startswith('v'):
-            target = 'v' + target
+    def _pip_install(self) -> tuple[bool, str]:
+        req = os.path.join(self.app_root, 'requirements.txt')
+        if not os.path.isfile(req):
+            return True, ''
+        code_p, out_p, err_p = _run(
+            [sys.executable, '-m', 'pip', 'install', '-r', req],
+            self.app_root,
+            timeout=300,
+        )
+        log = (out_p or err_p or '')[:1500]
+        return code_p == 0, log

+    def _apply_git_update(self, target: str, allow_dirty: bool) -> dict:
        if not allow_dirty and self.working_tree_dirty():
            return {
                'status': 'error',
                'message': (
-                    'Working tree has local changes. Commit/stash them or set allow_dirty, '
-                    'then retry auto-update.'
+                    'Working tree has local changes. Confirm update anyway to force checkout, '
+                    'or commit/stash changes first.'
                ),
+                'needs_confirm_dirty': True,
            }

        err = self._ensure_update_remote()
@@ -168,35 +221,28 @@ class UpdateManager:
            timeout=180,
        )
        if code != 0:
-            return {
-                'status': 'error',
-                'message': f'git fetch failed: {stderr or out or code}',
-            }
+            return {'status': 'error', 'message': f'git fetch failed: {stderr or out or code}'}

-        # Prefer annotated/lightweight tag; fall back to remote main tip if tag missing
+        # Ensure tag is present locally
        code, _, _ = _run(
            ['git', 'rev-parse', '-q', '--verify', f'refs/tags/{target}'],
            self.app_root,
            timeout=15,
        )
-        if code == 0:
-            checkout_ref = target
-        else:
-            # tag might only exist as FETCH_HEAD / remote tag
-            code2, _, _ = _run(
-                ['git', 'rev-parse', '-q', '--verify', f'refs/tags/{target}^{{}}'],
+        if code != 0:
+            _run(
+                ['git', 'fetch', '--depth', '1', UPDATE_REMOTE, f'refs/tags/{target}:refs/tags/{target}'],
                self.app_root,
-                timeout=15,
+                timeout=120,
            )
-            checkout_ref = target if code2 == 0 else f'tags/{target}'

+        checkout_ref = f'tags/{target}'
        code, out, stderr = _run(
            ['git', 'checkout', '--force', checkout_ref],
            self.app_root,
            timeout=60,
        )
        if code != 0:
-            # last resort: checkout remote main
            code_m, out_m, err_m = _run(
                ['git', 'checkout', '--force', '-B', 'main', f'{UPDATE_REMOTE}/main'],
                self.app_root,
@@ -209,31 +255,130 @@ class UpdateManager:
                }
            checkout_ref = f'{UPDATE_REMOTE}/main'

-        req = os.path.join(self.app_root, 'requirements.txt')
-        pip_log = ''
-        if os.path.isfile(req):
-            code_p, out_p, err_p = _run(
-                [sys.executable, '-m', 'pip', 'install', '-r', req],
-                self.app_root,
-                timeout=300,
-            )
-            pip_log = (out_p or err_p or '')[:1500]
-            if code_p != 0:
-                return {
-                    'status': 'error',
-                    'message': f'pip install failed after checkout: {err_p or out_p or code_p}',
-                    'checkout': checkout_ref,
-                }
+        ok, pip_log = self._pip_install()
+        if not ok:
+            return {
+                'status': 'error',
+                'message': f'pip install failed after checkout: {pip_log[:500]}',
+                'checkout': checkout_ref,
+                'method': 'git',
+            }

        return {
            'status': 'success',
-            'message': f'Updated to {target}. Panel will restart.',
+            'message': f'Updated to {target} via git. Panel will restart.',
            'target_version': target,
            'checkout': checkout_ref,
            'pip': pip_log[:500],
+            'method': 'git',
            'restart': True,
        }

+    def _archive_urls(self, tag: str) -> list[str]:
+        base = repo_url()
+        return [
+            f'{base}/archive/{tag}.zip',
+            f'{base}/archive/{tag.lstrip("v")}.zip',
+            f'{api_latest_url().rsplit("/releases/latest", 1)[0]}/archive/{tag}.zip',
+        ]
+
+    def _download_archive(self, tag: str, dest_path: str) -> None:
+        last_err = 'unknown error'
+        for url in self._archive_urls(tag):
+            try:
+                req = urllib.request.Request(url, headers={'User-Agent': 'Amnezia-Web-Panel-Updater'})
+                with urllib.request.urlopen(req, timeout=180) as resp:
+                    with open(dest_path, 'wb') as out:
+                        shutil.copyfileobj(resp, out)
+                if os.path.getsize(dest_path) > 1024:
+                    return
+                last_err = f'empty archive from {url}'
+            except urllib.error.HTTPError as e:
+                last_err = f'HTTP {e.code} for {url}'
+            except Exception as e:
+                last_err = str(e)
+        raise RuntimeError(f'Failed to download release archive: {last_err}')
+
+    @staticmethod
+    def _extract_zip_root(extract_dir: str) -> str:
+        entries = [e for e in os.listdir(extract_dir) if e not in ('.', '..')]
+        if len(entries) == 1:
+            only = os.path.join(extract_dir, entries[0])
+            if os.path.isdir(only):
+                return only
+        return extract_dir
+
+    def _sync_tree(self, src_root: str) -> None:
+        for root, dirs, files in os.walk(src_root):
+            dirs[:] = [d for d in dirs if d not in ARCHIVE_SKIP_DIRS]
+            rel = os.path.relpath(root, src_root)
+            dest_root = self.app_root if rel in ('.', '') else os.path.join(self.app_root, rel)
+            os.makedirs(dest_root, exist_ok=True)
+            for fname in files:
+                if fname in ARCHIVE_SKIP_FILES:
+                    continue
+                src_file = os.path.join(root, fname)
+                dest_file = os.path.join(dest_root, fname)
+                shutil.copy2(src_file, dest_file)
+
+    def _apply_archive_update(self, target: str) -> dict:
+        tmpdir = tempfile.mkdtemp(prefix='panel-update-')
+        zip_path = os.path.join(tmpdir, f'{target}.zip')
+        extract_dir = os.path.join(tmpdir, 'extract')
+        try:
+            self._download_archive(target, zip_path)
+            os.makedirs(extract_dir, exist_ok=True)
+            with zipfile.ZipFile(zip_path, 'r') as zf:
+                zf.extractall(extract_dir)
+            src_root = self._extract_zip_root(extract_dir)
+            if not os.path.isdir(src_root):
+                return {'status': 'error', 'message': 'Invalid release archive layout'}
+            self._sync_tree(src_root)
+        except Exception as e:
+            return {'status': 'error', 'message': str(e), 'method': 'archive'}
+        finally:
+            shutil.rmtree(tmpdir, ignore_errors=True)
+
+        ok, pip_log = self._pip_install()
+        if not ok:
+            return {
+                'status': 'error',
+                'message': f'Files updated but pip install failed: {pip_log[:500]}',
+                'method': 'archive',
+            }
+
+        return {
+            'status': 'success',
+            'message': f'Updated to {target} from release archive. Panel will restart.',
+            'target_version': target,
+            'pip': pip_log[:500],
+            'method': 'archive',
+            'restart': True,
+        }
+
+    def apply_update(self, target_version: str, allow_dirty: bool = False) -> dict:
+        mode = self.detect_mode()
+        target = normalize_tag(target_version)
+        if not target or not re.match(r'^v\d+(\.\d+){0,3}([.-][\w.]+)?$', target):
+            return {'status': 'error', 'message': f'Invalid target version: {target_version}'}
+
+        if mode.get('can_git_update'):
+            return self._apply_git_update(target, allow_dirty)
+        if mode.get('can_archive_update'):
+            return self._apply_archive_update(target)
+
+        hint = (
+            'Cannot update from inside this install. '
+            'Use a git clone or writable source directory, or rebuild the Docker image on the host.'
+        )
+        if mode.get('in_docker'):
+            hint = (
+                'Docker image install: on the host run '
+                '`git pull && docker compose up -d --build` '
+                '(or set PANEL_UPDATE_ALLOW_DOCKER=1 if /app is bind-mounted).'
+            )
+        return {'status': 'error', 'message': hint, 'mode': mode}
+
    @staticmethod
    def schedule_restart(delay_sec: float = 1.5) -> None:
        def _restart():