diff --git a/loopx/doctor_git.py b/loopx/doctor_git.py index ef18f4d055..ecc59d90ae 100644 --- a/loopx/doctor_git.py +++ b/loopx/doctor_git.py @@ -1,7 +1,8 @@ -"""Git source and revision diagnostics for the installation doctor.""" +"""Source-owned Git identity for installation manifests and diagnostics.""" from __future__ import annotations +import os import re import subprocess from enum import Enum @@ -17,6 +18,34 @@ class GitRevisionRelation(str, Enum): UNKNOWN = "unknown" +def _source_git_environment() -> dict[str, str]: + env = dict(os.environ) + # A caller (for example a Git hook) may target an unrelated checkout. + for key in ("GIT_DIR", "GIT_WORK_TREE", "GIT_COMMON_DIR", "GIT_INDEX_FILE"): + env.pop(key, None) + return env + + +def _is_git_worktree_root(source_root: Path) -> bool: + """Reject Git's upward discovery while accepting linked worktree gitfiles.""" + if not (source_root / ".git").exists(): + return False + try: + result = subprocess.run( + ["git", "-C", str(source_root), "rev-parse", "--show-toplevel"], + check=False, capture_output=True, text=True, + encoding="utf-8", errors="replace", + env=_source_git_environment(), + ) + return ( + result.returncode == 0 + and bool(result.stdout.strip()) + and Path(result.stdout.strip()).resolve() == source_root + ) + except OSError: + return False + + def git_metadata_for_root(root: Path | None) -> dict[str, Any]: if root is None: return { @@ -29,7 +58,7 @@ def git_metadata_for_root(root: Path | None) -> dict[str, Any]: source_root = root.expanduser().resolve() except OSError: source_root = root.expanduser() - if not source_root.exists(): + if not _is_git_worktree_root(source_root): return { "root": str(source_root), "git_commit": None, @@ -46,6 +75,7 @@ def _run(args: list[str]) -> str | None: text=True, encoding="utf-8", errors="replace", + env=_source_git_environment(), ) except OSError: return None @@ -92,7 +122,7 @@ def git_revision_relation( source_root = root.expanduser().resolve() except OSError: source_root = root.expanduser() - if not source_root.exists(): + if not _is_git_worktree_root(source_root): return GitRevisionRelation.UNKNOWN def _is_ancestor(ancestor: str, descendant: str) -> bool | None: @@ -112,6 +142,7 @@ def _is_ancestor(ancestor: str, descendant: str) -> bool | None: text=True, encoding="utf-8", errors="replace", + env=_source_git_environment(), ) except OSError: return None @@ -157,7 +188,7 @@ def trusted_release_ref_for_root( source_root = root.expanduser().resolve() except OSError: source_root = root.expanduser() - if not source_root.exists(): + if not _is_git_worktree_root(source_root): return None try: @@ -168,6 +199,7 @@ def trusted_release_ref_for_root( text=True, encoding="utf-8", errors="replace", + env=_source_git_environment(), ) except OSError: return None @@ -186,6 +218,7 @@ def trusted_release_ref_for_root( text=True, encoding="utf-8", errors="replace", + env=_source_git_environment(), ) except OSError: continue @@ -210,6 +243,7 @@ def trusted_release_ref_for_root( text=True, encoding="utf-8", errors="replace", + env=_source_git_environment(), ) commit = resolved.stdout.strip() if resolved.returncode == 0 else "" if commit: diff --git a/loopx/release_manifest.py b/loopx/release_manifest.py index b207f99d8d..bc912dd104 100644 --- a/loopx/release_manifest.py +++ b/loopx/release_manifest.py @@ -6,10 +6,10 @@ import json import os from pathlib import Path -import subprocess from typing import Any from . import __version__ +from .doctor_git import git_metadata_for_root RELEASE_MANIFEST_SCHEMA_VERSION = "loopx_release_manifest_v0" @@ -71,51 +71,6 @@ def _hash_tree(root: Path) -> dict[str, Any]: } -def _run_git(source_root: Path, args: list[str]) -> str | None: - try: - result = subprocess.run( - ["git", "-C", str(source_root), *args], - check=False, - capture_output=True, - text=True, encoding="utf-8", errors="replace", - ) - except OSError: - return None - if result.returncode != 0: - return None - value = result.stdout.strip() - return value or None - - -def _git_metadata(source_root: Path | None) -> dict[str, Any]: - if source_root is None: - return { - "git_commit": None, - "git_ref": None, - "git_dirty": None, - } - root = source_root.expanduser().resolve() - if not root.exists(): - return { - "git_commit": None, - "git_ref": None, - "git_dirty": None, - } - commit = _run_git(root, ["rev-parse", "HEAD"]) - branch = _run_git(root, ["symbolic-ref", "--quiet", "--short", "HEAD"]) - tag = _run_git(root, ["describe", "--tags", "--exact-match"]) - status = _run_git(root, ["status", "--porcelain"]) - if commit is None and branch is None and tag is None and status is None: - dirty: bool | None = None - else: - dirty = bool(status) - return { - "git_commit": commit, - "git_ref": branch or tag, - "git_dirty": dirty, - } - - def _manifest_source_metadata(source_root: Path | None) -> dict[str, Any]: if source_root is None: return { @@ -171,7 +126,8 @@ def _manifest_source_metadata(source_root: Path | None) -> dict[str, Any]: def _source_metadata(source_root: Path | None) -> dict[str, Any]: - git_metadata = _git_metadata(source_root) + git_metadata = git_metadata_for_root(source_root) + git_metadata.pop("root") if any(git_metadata.get(key) is not None for key in ("git_commit", "git_ref", "git_dirty")): return { **git_metadata, diff --git a/loopx/skill_install_readback.py b/loopx/skill_install_readback.py index ffb6724cc5..79ef7bdfd6 100644 --- a/loopx/skill_install_readback.py +++ b/loopx/skill_install_readback.py @@ -7,11 +7,11 @@ from email.parser import Parser from pathlib import Path import shutil -import subprocess import tempfile from typing import Any, Mapping, Sequence from . import __version__ +from .doctor_git import git_metadata_for_root SKILL_INSTALL_READBACK_SCHEMA_VERSION = "loopx_skill_install_readback_v1" SKILL_INSTALL_READBACK_FILENAME = ".loopx-skill-install.json" @@ -287,21 +287,6 @@ def hash_skill_tree( } -def _git_value(root: Path, *args: str) -> str | None: - try: - result = subprocess.run( - ["git", "-C", str(root), *args], - check=False, - capture_output=True, - text=True, encoding="utf-8", errors="replace", - ) - except OSError: - return None - if result.returncode != 0: - return None - return result.stdout.strip() or None - - def _source_readback( source_root: Path, *, @@ -363,13 +348,13 @@ def _source_readback( "git_dirty": False, } - commit = _git_value(source_root, "rev-parse", "HEAD") - status = _git_value(source_root, "status", "--porcelain") + metadata = git_metadata_for_root(source_root) + commit = metadata["git_commit"] return { "kind": "local_checkout", "revision": commit, "revision_kind": "git_commit" if commit else None, - "git_dirty": bool(status) if commit is not None else None, + "git_dirty": metadata["git_dirty"] if commit is not None else None, } diff --git a/scripts/install-local.sh b/scripts/install-local.sh index 416e3bc8d9..623bd86358 100755 --- a/scripts/install-local.sh +++ b/scripts/install-local.sh @@ -578,10 +578,22 @@ if os.environ.get("LOOPX_SKILL_ENTRY_HOST_SURFACE") and result["status"] in { PY } +read_source_git_commit() { + PYTHONSAFEPATH=1 PYTHONPATH="$repo_root${PYTHONPATH:+:$PYTHONPATH}" \ + "$LOOPX_PYTHON" - "$repo_root" <<'PY_SOURCE' +from pathlib import Path +import sys + +from loopx.doctor_git import git_metadata_for_root + +print(git_metadata_for_root(Path(sys.argv[1]))["git_commit"] or "") +PY_SOURCE +} + verify_default_promotion() { LOOPX_VERIFY_LINK="$bin_dir/loopx" \ LOOPX_VERIFY_RELEASE_DIR="$release_dir" \ - LOOPX_VERIFY_SOURCE_COMMIT="${LOOPX_RESOLVED_SOURCE_GIT_COMMIT:-$(git -C "$repo_root" rev-parse HEAD 2>/dev/null || true)}" \ + LOOPX_VERIFY_SOURCE_COMMIT="${LOOPX_RESOLVED_SOURCE_GIT_COMMIT:-$source_git_commit}" \ "${LOOPX_PYTHON:-python3}" - <<'PY' from pathlib import Path import json @@ -695,13 +707,14 @@ resolve_default_promotion() { ;; esac - if ! git -C "$repo_root" rev-parse --is-inside-work-tree >/dev/null 2>&1; then + source_git_commit="$(read_source_git_commit)" + if [[ -z "$source_git_commit" ]]; then promotion_mode="canary_only_unverified_source" return 1 fi local source_commit approved_commit source_status approved_ref - source_commit="$(git -C "$repo_root" rev-parse HEAD 2>/dev/null || true)" + source_commit="$source_git_commit" source_status="$(git -C "$repo_root" status --porcelain 2>/dev/null || true)" approved_ref="${LOOPX_APPROVED_DEFAULT_REF:-refs/remotes/origin/main}" approved_commit="$(git -C "$repo_root" rev-parse "$approved_ref" 2>/dev/null || true)" @@ -726,6 +739,9 @@ if [[ -z "$shell_profile" ]]; then fi configure_python_runtime +# Git hooks can pass selectors for their own repository into this installer. +unset GIT_DIR GIT_WORK_TREE GIT_COMMON_DIR GIT_INDEX_FILE +source_git_commit="" promote_default=0 if resolve_default_promotion; then promote_default=1 @@ -744,6 +760,11 @@ if [[ "$install_canary" != "0" ]]; then check_symlink_destination "$bin_dir/loopx-canary" fi if [[ "$promote_default" == "1" ]]; then + # Use the same source-root identity as the manifest and doctor, after rejecting + # unusable entry targets. A nested archive must not borrow its container's Git. + if [[ -z "$source_git_commit" ]]; then + source_git_commit="$(read_source_git_commit)" + fi # Preparing shared Chat assets is part of the guarded installation. mkdir -p "$releases_dir" run_under_install_guard "$@" diff --git a/tests/test_distribution_runtime_identity.py b/tests/test_distribution_runtime_identity.py index 0a46b0ed86..54b7131043 100644 --- a/tests/test_distribution_runtime_identity.py +++ b/tests/test_distribution_runtime_identity.py @@ -62,6 +62,7 @@ def test_real_pip_compiled_and_uncompiled_wheel_have_the_same_identity(tmp_path: contents = { "loopx/__init__.py": Path(manifest.__file__).with_name("__init__.py").read_text(), "loopx/release_manifest.py": Path(manifest.__file__).read_text(), + "loopx/doctor_git.py": Path(manifest.__file__).with_name("doctor_git.py").read_text(), f"{metadata}/METADATA": f"Metadata-Version: 2.1\nName: loopx\nVersion: {version}\n", f"{metadata}/WHEEL": "Wheel-Version: 1.0\nGenerator: test\nRoot-Is-Purelib: true\nTag: py3-none-any\n", } diff --git a/tests/test_doctor_git_encoding.py b/tests/test_doctor_git_encoding.py index aa7484db12..d8a20c4602 100644 --- a/tests/test_doctor_git_encoding.py +++ b/tests/test_doctor_git_encoding.py @@ -77,12 +77,15 @@ def test_trusted_release_preserves_unicode_remote_and_ref(git_repo, gbk_host): @pytest.fixture -def git_output(monkeypatch, gbk_host): +def git_output(monkeypatch, gbk_host, tmp_path): def install(*, returncode=0, fail_at=None): calls = [] + (tmp_path / ".git").mkdir(exist_ok=True) def run(command, **kwargs): args = command[3:] + if args == ["rev-parse", "--show-toplevel"]: + return SimpleNamespace(returncode=0, stdout=str(tmp_path) + "\n") calls.append(args) code = ( returncode[len(calls) - 1] diff --git a/tests/test_install_source_git_provenance.py b/tests/test_install_source_git_provenance.py new file mode 100644 index 0000000000..924cedb2d7 --- /dev/null +++ b/tests/test_install_source_git_provenance.py @@ -0,0 +1,196 @@ +"""Installation provenance belongs to the source root, never its container.""" + +from __future__ import annotations + +import json +import os +from pathlib import Path +import subprocess +import sys + +import pytest + +from loopx.doctor_git import ( + GitRevisionRelation, + git_metadata_for_root, + git_revision_relation, + trusted_release_ref_for_root, +) +from loopx.release_manifest import build_release_manifest +from loopx.skill_install_readback import build_skill_install_readback + + +def git(root: Path, *args: str) -> str: + return subprocess.run( + ["git", "-C", str(root), *args], check=True, + capture_output=True, encoding="utf-8", + ).stdout.strip() + + +@pytest.fixture +def repository(tmp_path, monkeypatch): + # Git hooks/CI may otherwise lend these disposable fixtures another identity. + for key in ("GIT_DIR", "GIT_WORK_TREE", "GIT_INDEX_FILE"): + monkeypatch.delenv(key, raising=False) + monkeypatch.delenv("LOOPX_RESOLVED_SOURCE_GIT_COMMIT", raising=False) + root = tmp_path / "仓库" + root.mkdir() + git(root, "init", "--initial-branch=source") + git(root, "config", "user.email", "fixture@example.invalid") + git(root, "config", "user.name", "Source Fixture") + (root / "tracked.txt").write_text("original\n", encoding="utf-8") + git(root, "add", "tracked.txt") + git(root, "commit", "-m", "source fixture") + return root + + +def source_observations(source: Path, tmp_path: Path): + manifest = build_release_manifest( + release_root=tmp_path / "release", release_id="fixture", source_root=source, + env={"LOOPX_PROMOTION_MODE": "canary"}, + )["source"] + skills = build_skill_install_readback( + skills_dir=tmp_path / "skills", skill_ids=[], source_root=source, + ) + return git_metadata_for_root(source), manifest, skills + + +def test_archive_inside_unrelated_repository_has_no_git_identity(repository, tmp_path): + archive = repository / "download" / "source" + archive.mkdir(parents=True) + (archive / "pyproject.toml").write_text("# archive\n", encoding="utf-8") + doctor, manifest, skills = source_observations(archive, tmp_path) + for observation in (doctor, manifest): + assert {key: observation[key] for key in ("git_commit", "git_ref", "git_dirty")} == { + "git_commit": None, "git_ref": None, "git_dirty": None, + } + assert skills["source"]["revision_kind"] == "skills_digest" + assert skills["source"]["revision"] == skills["skills"]["digest"] + assert skills["source"]["git_dirty"] is None + + parent_commit = git(repository, "rev-parse", "HEAD") + git(repository, "remote", "add", "origin", "https://github.com/example/source.git") + git(repository, "update-ref", "refs/remotes/origin/stable", parent_commit) + assert trusted_release_ref_for_root( + archive, repository="example/source", ref="stable", + ) is None + (repository / "tracked.txt").write_text("new\n", encoding="utf-8") + git(repository, "commit", "-am", "next fixture") + assert git_revision_relation( + archive, installed_commit=parent_commit, + comparison_commit=git(repository, "rev-parse", "HEAD"), + ) is GitRevisionRelation.UNKNOWN + + +@pytest.mark.parametrize("explicit", ["environment", "manifest"]) +def test_nested_archive_retains_explicit_source_identity( + repository, tmp_path, monkeypatch, explicit, +): + archive = repository / "archive" + archive.mkdir() + commit = "a" * 40 + if explicit == "environment": + monkeypatch.setenv("LOOPX_RESOLVED_SOURCE_GIT_COMMIT", commit) + env = {"LOOPX_RESOLVED_SOURCE_GIT_COMMIT": commit} + else: + (archive / "release.json").write_text(json.dumps({"source": { + "kind": "github_archive", "git_commit": commit, "git_ref": "v1.0.0", + "git_dirty": False, "ref": "v1.0.0", + }}), encoding="utf-8") + env = {"LOOPX_PROMOTION_MODE": "canary"} + manifest = build_release_manifest( + release_root=tmp_path / "release", release_id="fixture", + source_root=archive, env=env, + ) + skills = build_skill_install_readback( + skills_dir=tmp_path / "skills", skill_ids=[], source_root=archive, + ) + assert manifest["source"]["git_commit"] == commit + assert manifest["source"]["git_ref"] == ("v1.0.0" if explicit == "manifest" else None) + assert manifest["source"]["git_dirty"] is (False if explicit == "manifest" else None) + assert skills["source"]["revision"] == commit + assert skills["source"]["revision_kind"] == "git_commit" + assert skills["source"]["git_dirty"] is False + assert git_metadata_for_root(archive)["git_commit"] is None + + +@pytest.mark.parametrize("layout", ["checkout", "linked_worktree", "symlink"]) +@pytest.mark.parametrize("state", ["clean", "dirty", "tag"]) +def test_owned_git_sources_preserve_identity(repository, tmp_path, layout, state): + source = repository + if layout == "linked_worktree": + source = tmp_path / "linked" + git(repository, "worktree", "add", "-b", "linked-source", str(source)) + assert (source / ".git").is_file() + elif layout == "symlink": + source = tmp_path / "alias" + source.symlink_to(repository, target_is_directory=True) + expected_ref = "linked-source" if layout == "linked_worktree" else "source" + if state == "tag": + git(source, "tag", "v1.0.0") + git(source, "checkout", "--detach") + expected_ref = "v1.0.0" + elif state == "dirty": + (source / "tracked.txt").write_text("changed\n", encoding="utf-8") + expected_commit = git(source, "rev-parse", "HEAD") + doctor, manifest, skills = source_observations(source, tmp_path) + for observation in (doctor, manifest): + assert observation["git_commit"] == expected_commit + assert observation["git_ref"] == expected_ref + assert observation["git_dirty"] is (state == "dirty") + assert skills["source"]["revision"] == expected_commit + assert skills["source"]["git_dirty"] is (state == "dirty") + + +def test_missing_source_is_unavailable(tmp_path): + doctor, manifest, skills = source_observations(tmp_path / "missing", tmp_path) + assert doctor["git_commit"] is None + assert manifest["git_commit"] is None + assert skills["source"]["revision_kind"] == "skills_digest" + + +@pytest.mark.parametrize("marker", ["broken", "borrowed"]) +def test_git_marker_alone_cannot_lend_an_archive_identity( + repository, tmp_path, monkeypatch, marker, +): + archive = repository / "archive" + archive.mkdir() + (archive / ".git").write_text("gitdir: missing\n", encoding="utf-8") + if marker == "borrowed": + # A hook's ambient Git directory still names the enclosing checkout. + monkeypatch.setenv("GIT_DIR", str(repository / ".git")) + doctor, manifest, skills = source_observations(archive, tmp_path) + assert doctor["git_commit"] is None + assert manifest["git_commit"] is None + assert skills["source"]["revision_kind"] == "skills_digest" + + +@pytest.mark.skipif(os.name == "nt", reason="POSIX source installer") +@pytest.mark.parametrize("layout", ["checkout", "linked_worktree", "archive"]) +def test_installer_auto_promotion_uses_only_the_owned_source(repository, tmp_path, layout): + source = repository + if layout == "linked_worktree": + source = tmp_path / "linked" + git(repository, "worktree", "add", "-b", "linked-source", str(source)) + elif layout == "archive": + source = repository / "archive" + source.mkdir() + installer = (Path(__file__).parents[1] / "scripts/install-local.sh").read_text() + functions = "\n".join( + name + "() {" + installer.split(name + "() {", 1)[1].split("\n}\n", 1)[0] + "\n}" + for name in ("read_source_git_commit", "resolve_default_promotion") + ) + result = subprocess.run( + ["bash", "-c", 'set -eu\nrepo_root="$1"\npromote_default_request=auto\n' + 'source_git_commit=""\n' + functions + '\n' + 'if resolve_default_promotion; then printf "promote:%s" "$promotion_mode"; ' + 'else printf "hold:%s" "$promotion_mode"; fi', "source-test", str(source)], + env={**os.environ, "LOOPX_PYTHON": sys.executable, + "PYTHONPATH": str(Path(__file__).parents[1]), + "LOOPX_APPROVED_DEFAULT_REF": "HEAD"}, + capture_output=True, text=True, check=True, + ) + assert result.stdout == ( + "hold:canary_only_unverified_source" if layout == "archive" + else "promote:trusted_main_auto" + )