352 lines
16 KiB
Bash
Executable File
352 lines
16 KiB
Bash
Executable File
#!/usr/bin/env bash
|
||
# Canonical installation source; the embedded Python shares this entry's lifecycle.
|
||
# The complete body must parse before any installation command can execute.
|
||
dever_install() {
|
||
if [[ ! -x /usr/bin/python3 ]]; then
|
||
printf '%s\n' 'Dever 安装需要 /usr/bin/python3 3.9 或更新版本' >&2
|
||
return 1
|
||
fi
|
||
/usr/bin/python3 -I -c 'import sys; sys.exit(0 if sys.version_info >= (3, 9) else "Dever 安装需要 Python 3.9 或更新版本")' || return "$?"
|
||
local argument help=0
|
||
for argument in "$@"; do
|
||
case "$argument" in -h|--help) help=1 ;; esac
|
||
done
|
||
local -a command=(/usr/bin/python3 -I -)
|
||
if (( EUID != 0 && help == 0 )); then
|
||
if [[ ! -x /usr/bin/sudo ]]; then
|
||
printf '%s\n' 'Dever 机器安装需要管理员权限；请安装 sudo 或由管理员执行此脚本' >&2
|
||
return 1
|
||
fi
|
||
command=(/usr/bin/sudo -- "${command[@]}")
|
||
fi
|
||
"${command[@]}" "$@" <<'DEVER_INSTALL_PYTHON'
|
||
"""Install an authenticated Dever release; no application configuration is read."""
|
||
|
||
import argparse
|
||
import hashlib
|
||
import json
|
||
import os
|
||
from pathlib import Path, PurePosixPath
|
||
import re
|
||
import ssl
|
||
import subprocess
|
||
import sys
|
||
import tempfile
|
||
import urllib.error
|
||
import urllib.parse
|
||
import urllib.request
|
||
|
||
RELEASES = "https://git.dever.cc/api/v1/repos/shemic/dever/releases"
|
||
MAX_METADATA = 2 * 1024 * 1024
|
||
MAX_RELEASE = 2 * 1024 * 1024 * 1024
|
||
ASSET_HOSTS = {"git.dever.cc"}
|
||
DEFAULT_PUBLIC_KEY = "7889e0edaf5fbf1060bef4ad6e27106128485c8412a6f6cd4457dc67e112986e"
|
||
|
||
|
||
def official_address(address):
|
||
target = urllib.parse.urlsplit(address)
|
||
if (target.scheme != "https" or target.hostname not in ASSET_HOSTS
|
||
or target.username or target.password or target.port not in (None, 443)
|
||
or target.fragment):
|
||
raise ValueError("发行下载地址必须位于官方 HTTPS 站点")
|
||
return address
|
||
|
||
|
||
class OfficialRedirect(urllib.request.HTTPRedirectHandler):
|
||
def redirect_request(self, request, stream, code, message, headers, address):
|
||
official_address(address)
|
||
return super().redirect_request(request, stream, code, message, headers, address)
|
||
|
||
|
||
def download(address):
|
||
official_address(address)
|
||
# Installation does not inherit HTTP proxy/environment configuration.
|
||
client = urllib.request.build_opener(
|
||
urllib.request.ProxyHandler({}), OfficialRedirect(),
|
||
urllib.request.HTTPSHandler(context=ssl.create_default_context()),
|
||
)
|
||
try:
|
||
return client.open(address, timeout=60)
|
||
except urllib.error.HTTPError as error:
|
||
if error.code == 404:
|
||
raise ValueError("该版本/平台尚无官方发行包，未修改已安装的 Dever") from error
|
||
raise ValueError(f"官方发行下载失败：HTTP {error.code}") from error
|
||
|
||
|
||
def platform_name():
|
||
host = os.uname()
|
||
names = {"x86_64": "linux-x86_64", "aarch64": "linux-aarch64"}
|
||
if host.sysname != "Linux" or host.machine not in names:
|
||
raise ValueError("安装器当前只支持 Linux x86_64/aarch64，且需对应平台已发布工具包")
|
||
return names[host.machine]
|
||
|
||
|
||
def prerequisites(platform):
|
||
if sys.version_info < (3, 9):
|
||
raise ValueError("安装需要 /usr/bin/python3 3.9 或更新版本")
|
||
if not os.access("/usr/bin/openssl", os.X_OK):
|
||
raise ValueError("安装需要 /usr/bin/openssl（支持 Ed25519）")
|
||
try:
|
||
libc = os.confstr("CS_GNU_LIBC_VERSION") or ""
|
||
except (OSError, ValueError):
|
||
libc = ""
|
||
match = re.fullmatch(r"glibc ([0-9]+)\.([0-9]+)", libc)
|
||
if not match:
|
||
raise ValueError("官方 Linux 工具包需要 glibc，当前系统 ABI 不支持")
|
||
if platform == "linux-x86_64" and tuple(map(int, match.groups())) < (2, 39):
|
||
raise ValueError("当前 Linux x86_64 工具包需要 glibc 2.39 或更新版本")
|
||
|
||
|
||
def read_bounded(path, limit):
|
||
with path.open("rb") as stream:
|
||
contents = stream.read(limit + 1)
|
||
if len(contents) > limit:
|
||
raise ValueError("发行元数据超过大小限制")
|
||
return contents
|
||
|
||
|
||
def resolve_release(selection):
|
||
endpoint = "latest" if selection == "latest" else f"tags/v{selection}"
|
||
with download(f"{RELEASES}/{endpoint}") as stream:
|
||
contents = stream.read(MAX_METADATA + 1)
|
||
if len(contents) > MAX_METADATA:
|
||
raise ValueError("发行元数据超过大小限制")
|
||
record = json.loads(contents)
|
||
if (not isinstance(record, dict) or record.get("draft") is not False
|
||
or record.get("prerelease") is not False):
|
||
raise ValueError("需要已发布的稳定发行版本")
|
||
tag = record.get("tag_name")
|
||
if not isinstance(tag, str) or not re.fullmatch(
|
||
r"v(?:0|[1-9][0-9]*)\.(?:0|[1-9][0-9]*)\.(?:0|[1-9][0-9]*)", tag):
|
||
raise ValueError("发行标签不是准确的 v<major.minor.patch>")
|
||
version = tag[1:]
|
||
if selection != "latest" and version != selection:
|
||
raise ValueError("发行版本与请求不匹配")
|
||
entries = record.get("assets")
|
||
if not isinstance(entries, list) or not 1 <= len(entries) <= 512:
|
||
raise ValueError("发行附件列表缺失或超限")
|
||
attachments = {}
|
||
for entry in entries:
|
||
if not isinstance(entry, dict):
|
||
raise ValueError("发行附件元数据错误")
|
||
name, size, address = (entry.get(key) for key in ("name", "size", "browser_download_url"))
|
||
if (not isinstance(name, str) or not name or name in attachments
|
||
or type(size) is not int or size <= 0 or size > MAX_RELEASE
|
||
or not isinstance(address, str)):
|
||
raise ValueError("发行附件名称重复、长度或地址错误")
|
||
attachments[name] = {"address": official_address(address), "bytes": size}
|
||
return version, attachments
|
||
|
||
|
||
def asset(attachments, platform, suffix, limit=MAX_RELEASE):
|
||
name = f"dever-{platform}.{suffix}"
|
||
entry = attachments.get(name)
|
||
if entry is None:
|
||
raise ValueError(f"该发行版本缺少附件：{name}")
|
||
if entry["bytes"] > limit:
|
||
raise ValueError("发行附件超过大小限制")
|
||
return entry
|
||
|
||
|
||
def metadata(release, source, attachments, platform):
|
||
for filename in ("manifest.json", "manifest.sig"):
|
||
if source:
|
||
contents = read_bounded(source / filename, MAX_METADATA)
|
||
else:
|
||
entry = asset(attachments, platform, filename, MAX_METADATA)
|
||
with download(entry["address"]) as stream:
|
||
contents = stream.read(MAX_METADATA + 1)
|
||
if len(contents) > MAX_METADATA:
|
||
raise ValueError("发行元数据超过大小限制")
|
||
if len(contents) != entry["bytes"]:
|
||
raise ValueError("发行元数据长度不匹配")
|
||
(release / filename).write_bytes(contents)
|
||
|
||
|
||
def authenticate(release, key_path, scratch, platform, requested):
|
||
key = bytes.fromhex(read_bounded(key_path, 256).decode().strip())
|
||
signature = bytes.fromhex(read_bounded(release / "manifest.sig", 256).decode().strip())
|
||
if len(key) != 32 or len(signature) != 64:
|
||
raise ValueError("发行公钥或签名长度错误")
|
||
public_der = scratch / "public.der"
|
||
public_der.write_bytes(bytes.fromhex("302a300506032b6570032100") + key)
|
||
signature_file = scratch / "signature.bin"
|
||
signature_file.write_bytes(signature)
|
||
verified = subprocess.run(
|
||
["/usr/bin/openssl", "pkeyutl", "-verify", "-pubin", "-keyform", "DER",
|
||
"-inkey", str(public_der), "-rawin", "-in", str(release / "manifest.json"),
|
||
"-sigfile", str(signature_file)],
|
||
env={}, capture_output=True, timeout=15, check=False,
|
||
)
|
||
if verified.returncode:
|
||
raise ValueError("发行签名校验失败，未执行安装程序")
|
||
manifest = json.loads((release / "manifest.json").read_text())
|
||
if (set(manifest) != {"format", "version", "platform", "artifacts", "extensions"}
|
||
or manifest.get("format") != "dever-release-v2" or manifest.get("platform") != platform
|
||
or not isinstance(manifest["extensions"], list) or len(manifest["extensions"]) > 12):
|
||
raise ValueError("发行格式或平台不匹配")
|
||
version = manifest.get("version", "")
|
||
if not re.fullmatch(r"(?:0|[1-9][0-9]*)\.(?:0|[1-9][0-9]*)\.(?:0|[1-9][0-9]*)", version):
|
||
raise ValueError("发行版本不是准确的 major.minor.patch")
|
||
if requested != "latest" and version != requested:
|
||
raise ValueError("发行版本与请求不匹配")
|
||
artifacts = manifest.get("artifacts", [])
|
||
if not 1 <= len(artifacts) <= 4096:
|
||
raise ValueError("发行文件数超限")
|
||
catalog = {}
|
||
total = 0
|
||
for artifact in artifacts:
|
||
name, size, digest = artifact["path"], artifact["bytes"], artifact["sha256"]
|
||
if (set(artifact) != {"path", "bytes", "sha256"}
|
||
or not name or name.startswith("/")
|
||
or any(part in ("", ".", "..") for part in name.split("/"))
|
||
or any(character in name for character in ("\\", ":", "\0"))
|
||
or str(PurePosixPath(name)) != name
|
||
or name in catalog or name in ("manifest.json", "manifest.sig")):
|
||
raise ValueError("发行包含不安全或重复路径")
|
||
if type(size) is not int or size < 0 or not re.fullmatch(r"[0-9a-f]{64}", digest):
|
||
raise ValueError("发行文件长度或摘要错误")
|
||
total += size
|
||
if total > MAX_RELEASE:
|
||
raise ValueError("发行解压长度超过 2 GiB")
|
||
catalog[name] = artifact
|
||
if not {"dever-core", "bootstrap/dever", "skills/dever-language/SKILL.md"} <= catalog.keys():
|
||
raise ValueError("首装包缺少核心、bootstrap 启动器或配套 skill")
|
||
return version, catalog, key
|
||
|
||
|
||
def copy_artifact(stream, destination, artifact):
|
||
destination.parent.mkdir(parents=True, exist_ok=True)
|
||
remaining = artifact["bytes"]
|
||
digest = hashlib.sha256()
|
||
with destination.open("xb") as output:
|
||
while remaining:
|
||
contents = stream.read(min(remaining, 1024 * 1024))
|
||
if not contents:
|
||
raise ValueError("发行文件被截断")
|
||
output.write(contents)
|
||
digest.update(contents)
|
||
remaining -= len(contents)
|
||
if digest.hexdigest() != artifact["sha256"]:
|
||
raise ValueError("发行文件摘要校验失败")
|
||
executable = artifact["path"] in {"dever-core", "bootstrap/dever", "bootstrap/deverd"}
|
||
destination.chmod(0o755 if executable else 0o644)
|
||
|
||
|
||
def payload(release, source, platform, catalog, trusted_key=None, attachments=None):
|
||
if source:
|
||
for name, artifact in catalog.items():
|
||
path = source / name
|
||
for ancestor in (path, *path.parents):
|
||
if ancestor.is_symlink():
|
||
raise ValueError("本地发行文件不能经过符号链接")
|
||
if ancestor == source:
|
||
break
|
||
if not path.is_file() or path.stat().st_size != artifact["bytes"]:
|
||
raise ValueError("本地发行文件类型或长度错误")
|
||
with path.open("rb") as stream:
|
||
copy_artifact(stream, release / name, artifact)
|
||
return
|
||
if trusted_key is None:
|
||
raise ValueError("网络发行解压需要独立信任公钥")
|
||
# Only signed bytes are executable. Keep the helper separate so Rust can
|
||
# create_new every archive member and enforce one complete codec/TAR contract.
|
||
with tempfile.TemporaryDirectory(prefix=".extract-", dir=release.parent) as directory:
|
||
scratch = Path(directory)
|
||
helper = scratch / "helper"
|
||
for name, artifact in catalog.items():
|
||
if name == "bootstrap/dever" or name.startswith("bootstrap/lib/"):
|
||
entry = asset(attachments, platform, f"blob-{artifact['sha256']}")
|
||
if entry["bytes"] != artifact["bytes"]:
|
||
raise ValueError("bootstrap 附件长度与签名不匹配")
|
||
with download(entry["address"]) as stream:
|
||
copy_artifact(stream, helper / name, artifact)
|
||
if stream.read(1):
|
||
raise ValueError("bootstrap 下载超过签名长度")
|
||
compressed = scratch / "base.tar.zst"
|
||
entry = asset(attachments, platform, "tar.zst")
|
||
download_archive(entry["address"], compressed)
|
||
if compressed.stat().st_size != entry["bytes"]:
|
||
raise ValueError("发行附件长度不匹配")
|
||
configuration = scratch / "extract.json"
|
||
configuration.write_text(json.dumps({
|
||
"release": str(release), "archive": str(compressed),
|
||
"trusted_key": str(trusted_key),
|
||
}))
|
||
subprocess.run([str(helper / "bootstrap/dever"), "--dever-extract", str(configuration)],
|
||
env={}, check=True, timeout=180)
|
||
|
||
|
||
def download_archive(address, destination):
|
||
remaining = MAX_RELEASE
|
||
with download(address) as stream, destination.open("xb") as output:
|
||
while contents := stream.read(min(remaining + 1, 1024 * 1024)):
|
||
remaining -= len(contents)
|
||
if remaining < 0:
|
||
raise ValueError("发行下载超过 2 GiB")
|
||
output.write(contents)
|
||
|
||
|
||
def install(arguments):
|
||
if os.geteuid() != 0:
|
||
raise ValueError("机器安装需要管理员权限；请通过根 install 入口执行")
|
||
platform = platform_name()
|
||
prerequisites(platform)
|
||
if arguments.version != "latest" and not re.fullmatch(
|
||
r"(?:0|[1-9][0-9]*)\.(?:0|[1-9][0-9]*)\.(?:0|[1-9][0-9]*)", arguments.version):
|
||
raise ValueError("--version 需要 latest 或准确版本")
|
||
system_root = arguments.system_root.absolute()
|
||
if not system_root.is_dir() or system_root.is_symlink():
|
||
raise ValueError("--system-root 必须是已存在的真实目录")
|
||
source = arguments.release.absolute() if arguments.release else None
|
||
if source and (not source.is_dir() or source.is_symlink()):
|
||
raise ValueError("--release 必须是已存在的真实发行目录")
|
||
# The root-owned staging also keeps the pinned key outside the downloaded release.
|
||
with tempfile.TemporaryDirectory(prefix=".dever-install-", dir=system_root) as directory:
|
||
scratch = Path(directory)
|
||
release = scratch / "release"
|
||
release.mkdir()
|
||
selection, attachments = ((arguments.version, None) if source
|
||
else resolve_release(arguments.version))
|
||
metadata(release, source, attachments, platform)
|
||
key_path = arguments.trusted_key
|
||
if key_path is None:
|
||
key_path = scratch / "independent-release.pub"
|
||
key_path.write_text(DEFAULT_PUBLIC_KEY)
|
||
version, catalog, key = authenticate(release, key_path, scratch, platform, selection)
|
||
trusted = scratch / "release.pub"
|
||
trusted.write_text(key.hex() + "\n")
|
||
payload(release, source, platform, catalog, trusted, attachments)
|
||
configuration = scratch / "config"
|
||
configuration.mkdir()
|
||
(configuration / "setting.json").write_text(json.dumps({
|
||
"release": str(release), "system_root": str(system_root),
|
||
"trusted_key": str(trusted),
|
||
"activate_service": system_root == Path("/") and not arguments.no_start,
|
||
}))
|
||
subprocess.run([str(release / "bootstrap/dever"), "--dever-install", str(scratch)],
|
||
env={}, check=True, timeout=180)
|
||
print(f"Dever {version} 已安装到 {system_root / 'opt/dever'}")
|
||
|
||
|
||
def main():
|
||
parser = argparse.ArgumentParser(prog="dever install", description=__doc__)
|
||
parser.add_argument("--version", default="latest")
|
||
parser.add_argument("--release", type=Path, help="使用本地已签名发行目录，不下载")
|
||
parser.add_argument("--trusted-key", type=Path)
|
||
parser.add_argument("--system-root", type=Path, default=Path("/"), help="已有的自有安装镜像目录；默认安装当前机器")
|
||
parser.add_argument("--no-start", action="store_true", help="不启动当前机器 systemd 服务")
|
||
arguments = parser.parse_args()
|
||
try:
|
||
install(arguments)
|
||
except (OSError, ValueError, KeyError, TypeError, subprocess.SubprocessError) as error:
|
||
parser.exit(1, f"Dever 安装失败：{error}\n")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|
||
|
||
DEVER_INSTALL_PYTHON
|
||
}
|
||
dever_install "$@"
|