#!/usr/bin/env python3 """K-Echo 安装准备工具（Python 3.10+）。只准备文件，不代表插件已安装。""" import argparse import base64 import hashlib import io import json import os from pathlib import Path, PurePosixPath import platform import re import shutil import stat import sys import tempfile import urllib.parse import urllib.request import zipfile OFFICIAL_HOSTS = {'k-echo.com', 'download.k-echo.com'} MANIFEST_URL = 'https://k-echo.com/release-manifest.json' MAX_DOWNLOAD = 300 * 1024 * 1024 MAX_EXPANDED = 1024 * 1024 * 1024 MARKER = '.k-echo-prepared.json' def detect_system(value=None): value = value or platform.system() if value == 'Windows': return 'Windows' if value == 'Darwin': return 'macOS' raise ValueError('当前系统不在支持范围内；仅支持 Windows 和 macOS。') def trusted_url(url): u = urllib.parse.urlsplit(url) if u.scheme != 'https' or u.hostname not in OFFICIAL_HOSTS or u.username or u.password or u.port not in (None, 443) or u.fragment: raise ValueError('下载地址不是允许的官方 HTTPS 来源。') return url class OfficialRedirects(urllib.request.HTTPRedirectHandler): def redirect_request(self, req, fp, code, msg, headers, newurl): trusted_url(newurl) return super().redirect_request(req, fp, code, msg, headers, newurl) def official_open(url): trusted_url(url) response = urllib.request.build_opener(OfficialRedirects()).open(url, timeout=45) trusted_url(response.geturl()) return response def digest_file(path): h = hashlib.sha256() with path.open('rb') as source: for block in iter(lambda: source.read(1024 * 1024), b''): h.update(block) return h.hexdigest() def safe_name(info): name = info.filename parts = PurePosixPath(name).parts if not parts or name.startswith('/') or '\\' in name or any(p in ('.','..') or ':' in p or p.endswith(('.', ' ')) or re.match(r'^(CON|PRN|AUX|NUL|COM[1-9]|LPT[1-9])(?:\.|$)', p, re.I) for p in parts): raise ValueError('ZIP包含不安全路径。') if stat.S_ISLNK(info.external_attr >> 16): raise ValueError('ZIP包含符号链接。') return parts def result(target, status, release): return {'status': status, 'system': release['system'], 'version': release['version'], 'directory': str(target), 'extensionDirectory': str(target/'K-Echo/浏览器插件'), 'guide': str(target/'K-Echo/先看这里.html'), 'extensionLoaded': False, 'helperConnected': False, 'userLoggedIn': False, 'next': '检查已有K-Echo；确认要使用的浏览器，按官方指南加载。Windows再按授权安装配套程序。登录由用户完成。'} def prepare(release, target, opener=official_open): target = Path(target).expanduser().absolute() trusted_url(release['url']) size = release['sizeBytes']; expected = release['sha256'] if type(size) is not int or not 0 < size <= MAX_DOWNLOAD or not re.fullmatch('[a-f0-9]{64}', expected): raise ValueError('版本清单的大小或校验值无效。') if release.get('system') not in ('Windows', 'macOS') or not re.fullmatch(r'\d+\.\d+\.\d+', release.get('version','')): raise ValueError('版本清单的系统或版本无效。') if any(p.is_symlink() for p in (target, *target.parents)): raise ValueError('安装路径不能包含符号链接。') if target.exists(): marker = target/MARKER if not target.is_dir() or marker.is_symlink() or not marker.is_file(): raise ValueError('目标已存在且不是本工具准备的目录；保留原数据，停止操作。') old = json.loads(marker.read_text('utf-8')) if old.get('sha256') != expected or old.get('system') != release['system']: raise ValueError('已有版本与目标不同；不自动覆盖，请按官方更新流程操作。') for name, value in old.get('files', {}).items(): safe_name(zipfile.ZipInfo(name)) f = target/name if any(x.is_symlink() for x in (f, *f.parents)) or not f.is_file() or digest_file(f) != value: raise ValueError('已有安装目录发生变化；不覆盖，停止操作。') if not old.get('files'): raise ValueError('已有准备记录不完整。') return result(target, 'already_prepared', release) target.parent.mkdir(parents=True, exist_ok=True) lock = target.parent/(target.name+'.prepare-lock') try: lock.mkdir() except FileExistsError: raise ValueError('同一目标已有准备任务或遗留锁；请检查，停止操作。') try: with tempfile.TemporaryDirectory(prefix='.k-echo-stage-', dir=target.parent) as tmp: stage = Path(tmp); archive = stage/'download.zip'; total=0; h=hashlib.sha256() with opener(release['url']) as response, archive.open('wb') as output: while True: block = response.read(1024 * 1024) if not block: break total += len(block) if total > size: raise ValueError('实际下载大小超出版本清单。') h.update(block); output.write(block) if total != size or h.hexdigest() != expected: raise ValueError('下载大小或SHA-256不匹配；未解压。') unpacked = stage/'files'; unpacked.mkdir(); files={} with zipfile.ZipFile(archive) as z: entries=z.infolist() if len(entries)>5000 or sum(x.file_size for x in entries)>MAX_EXPANDED: raise ValueError('ZIP解压规模超限。') names=set() for info in entries: parts=safe_name(info); name='/'.join(parts) if parts[0] != 'K-Echo': raise ValueError('ZIP包含安装目录之外的文件。') if name.casefold() in names: raise ValueError('ZIP包含重复路径。') names.add(name.casefold()) for info in entries: name='/'.join(safe_name(info)); path=unpacked/name if info.is_dir(): path.mkdir(parents=True, exist_ok=True);continue path.parent.mkdir(parents=True, exist_ok=True) with z.open(info) as source, path.open('xb') as output: shutil.copyfileobj(source, output) files[name]=digest_file(path) if not (unpacked/'K-Echo/浏览器插件/manifest.json').is_file() or not (unpacked/'K-Echo/先看这里.html').is_file(): raise ValueError('安装包缺少插件或指南。') extension=json.loads((unpacked/'K-Echo/浏览器插件/manifest.json').read_text('utf-8')) if extension.get('version') != release['version']: raise ValueError('插件版本与清单不匹配。') if release.get('extensionId'): key=base64.b64decode(extension.get('key',''),validate=True) extension_id=''.join(chr(97+int(x,16)) for x in hashlib.sha256(key).hexdigest()[:32]) if extension_id != release['extensionId']: raise ValueError('插件身份与清单不匹配。') executables=[p for p in unpacked.rglob('*') if p.is_file() and p.suffix.lower()=='.exe'] if release['system']=='macOS' and executables: raise ValueError('macOS安装包不能包含Windows程序。') if release['system']=='Windows': helper=release.get('helper') if not helper: raise ValueError('Windows清单缺少配套程序信息。') safe_name(zipfile.ZipInfo(helper['path'])) executable=unpacked/helper['path'] if not executable.is_file() or digest_file(executable)!=helper['sha256']: raise ValueError('Windows配套程序与清单不匹配。') (unpacked/MARKER).write_text(json.dumps({'sha256':expected,'system':release['system'],'files':files},ensure_ascii=False,indent=2),encoding='utf-8') if target.exists(): raise ValueError('准备期间目标目录已出现；不覆盖。') os.rename(unpacked, target) return result(target, 'prepared', release) finally: lock.rmdir() def main(): parser=argparse.ArgumentParser(description=__doc__) parser.add_argument('--destination', help='新建的固定目录；已有未知目录不会被覆盖') args=parser.parse_args() try: system=detect_system() with official_open(MANIFEST_URL) as response: raw=response.read(1024*1024+1) if len(raw)>1024*1024: raise ValueError('版本清单过大。') manifest=json.loads(raw) if manifest.get('schemaVersion') != 1: raise ValueError('版本清单格式不支持。') release=manifest['releases'][system] if release['system']!=system:raise ValueError('版本清单系统不匹配。') parent=Path(os.environ.get('LOCALAPPDATA',str(Path.home()/'AppData/Local'))) if system=='Windows' else Path.home()/'Library/Application Support' target=Path(args.destination) if args.destination else parent/'K-Echo'/('manual-'+release['version']) print(json.dumps(prepare(release,target),ensure_ascii=False,indent=2)) return 0 except Exception as error: print(json.dumps({'status':'failed','error':str(error),'extensionLoaded':False,'helperConnected':False},ensure_ascii=False),file=sys.stderr) return 1 if __name__=='__main__': sys.exit(main())