diff --git a/.github/workflows/lock-delta.yml b/.github/workflows/lock-delta.yml new file mode 100644 index 00000000..91a64f01 --- /dev/null +++ b/.github/workflows/lock-delta.yml @@ -0,0 +1,31 @@ +name: Generated lock delta +on: + pull_request: + paths: + - 'requirements/**' + - 'tools/check_lock_delta.py' + - 'tools/check_changed_locks.py' + - 'tools/lock-delta-scope.json' + - 'tools/tests/test_lock_*.py' + - '.github/workflows/lock-delta.yml' +permissions: + contents: read +jobs: + lock-delta: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 + with: + fetch-depth: 0 + persist-credentials: false + - uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 + with: + python-version: '3.11' + - name: Guard regression controls + run: | + python -m unittest discover -s tools/tests -p 'test_lock_*.py' + python -O -m unittest discover -s tools/tests -p 'test_lock_*.py' + - name: Compare changed lock structure + env: + LOCK_BASE_SHA: ${{ github.event.pull_request.base.sha }} + run: python tools/check_changed_locks.py --base "$LOCK_BASE_SHA" diff --git a/CHANGELOG.md b/CHANGELOG.md index caaf5e60..e706403d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- Raise the A2A SDK and development extras' `pyasn1` security floor to 0.6.5, + alongside the development/CI hash lock, clearing four dependency-audit + findings for 0.6.4. + - Sealing to a low-order X25519 peer key raises `SealedChannelError`, consistent with other invalid channel keys, instead of leaking a backend `ValueError`. These keys were already refused; no ciphertext format or acceptance rule changes. diff --git a/pyproject.toml b/pyproject.toml index 4cf4691a..f6515ffe 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -49,6 +49,7 @@ tpm = [ a2a-sdk = [ "a2a-sdk>=1.1,<2", "aiohttp>=3.14.3", + "pyasn1>=0.6.5", ] dev = [ "pytest>=8.0", @@ -66,6 +67,7 @@ dev = [ "a2a-sdk[http-server]==1.2.2", "uvicorn==0.54.0", "aiohttp>=3.14.3", + "pyasn1>=0.6.5", ] [project.scripts] diff --git a/requirements/dev.in b/requirements/dev.in index ad636eff..36a848d0 100644 --- a/requirements/dev.in +++ b/requirements/dev.in @@ -25,3 +25,6 @@ a2a-sdk[http-server]==1.2.2 uvicorn==0.54.0 aiohttp>=3.14.3 setuptools>=83.0.0 + +# Security floor for google-auth's transitive ASN.1 dependency (four audit findings). +pyasn1>=0.6.5 diff --git a/requirements/dev.txt b/requirements/dev.txt index a57d254f..981e67dc 100644 --- a/requirements/dev.txt +++ b/requirements/dev.txt @@ -1710,10 +1710,12 @@ py-serializable==2.1.0 \ --hash=sha256:9d5db56154a867a9b897c0163b33a793c804c80cee984116d02d49e4578fc103 \ --hash=sha256:b56d5d686b5a03ba4f4db5e769dc32336e142fc3bd4d68a8c25579ebb0a67304 # via cyclonedx-python-lib -pyasn1==0.6.4 \ - --hash=sha256:9c447d8431c947fe4c8febc4ed9e760bc29011a5b01e5c74b67025bd9fb8ce81 \ - --hash=sha256:deda9277cfd454080ec40b207fb6df82206a3a2688735233cdcd8d3d565f088b - # via pyasn1-modules +pyasn1==0.6.5 \ + --hash=sha256:4017b0fce125096b8573b39c989fdb1a4b19cf761e683398184550fea0e66587 \ + --hash=sha256:6b11616a6bcf3a40e635f082bb232fa57b5e0dd7f517bd7ddd69ec3794508a47 + # via + # -r requirements/dev.in + # pyasn1-modules pyasn1-modules==0.4.2 \ --hash=sha256:29253a9207ce32b64c3ac6600edc75368f98473906e8fd1043bd6b5b1de2c14a \ --hash=sha256:677091de870a80aae844b1ca6134f54652fa2c8c5a52aa396440ac3106e941e6 diff --git a/tools/LOCK-VALIDATION.md b/tools/LOCK-VALIDATION.md new file mode 100644 index 00000000..f66345c3 --- /dev/null +++ b/tools/LOCK-VALIDATION.md @@ -0,0 +1,32 @@ +# Generated requirement lock review + +This draft port builds on agentrust-io/weight-custody-manifest#249. Source head `9fcf6bd6684dbd87473d1b0c670eaeb154749a75` still awaits independent acceptance; this port is not an accepted cross-project policy. + +The read-only PR job runs standard-library regression controls, then compares changed requirements/*.txt files against the PR base. It rejects removed package/extras/marker identities, malformed or absent SHA-256 hashes, duplicate identities and lock deletion. Renames are enumerated as a deletion and addition, so moving a lock (with or without dependency loss) requires explicit maintainer review instead of treating it only as a new baseline. Version and SHA-256 hash-set changes are reported for human review; hash ordering and duplicate identical hashes are ignored. New locks receive only a structural check because no prior dependency baseline exists. + +This recognizes the exact-pin/hash subset emitted by uv/pip-compile. It is not a general PEP 508 equivalence checker, dependency audit, package authentication or proof that an upgrade runs on Windows. Existing repository validation remains required. No SDK runtime or dependency pin changes. + +Manual cross-platform resolution is optional and separate: run tools/check_lock_delta.py --resolve --python-version 3.11 with a reviewed generated lock and uv available. It creates isolated Linux/Windows resolution contexts without installing their packages. Keep platform/version choices specific to each lock. + +Source: https://github.com/agentrust-io/weight-custody-manifest/pull/249 . + + +Explicitly excluded non-lock files: {"requirements/agt-overrides.txt": "compiler override constraints, not a generated hash lock"}. These remain under their existing compiler/composed-environment checks. The guard does not claim to validate them. + +Hash options must follow the marker (when present); every trailing token must be +a complete lowercase SHA-256 hash option. Quoted marker literals are retained. +Marker text is compared as an identity, not evaluated or fully validated as +PEP 508 syntax; supported-platform resolution is a separate check. + +The shared checker is byte-identical to WCM source head +`9fcf6bd6684dbd87473d1b0c670eaeb154749a75`. Its optional manual +`--all-locks` CLI checks the union of current and base top-level +`requirements/*.txt` paths; deleted/renamed paths and invalid bases fail, while +new locks are explicitly identified and structurally checked. This low-level +manual mode does not apply this port's exclusions. The PR workflow continues to +use `check_changed_locks.py`: all changed requirement locks are discovered with +`--no-renames`, and the existing exclusions above remain unchanged. The workflow +does not install packages or run platform resolvers. Tests exercise hash-only +changes through this real driver and valid/corrupted runtime, server and custom +lock filenames. Independent WCM source acceptance and repository-specific +review remain prerequisites; draft status is retained. diff --git a/tools/check_changed_locks.py b/tools/check_changed_locks.py new file mode 100644 index 00000000..7e5165b0 --- /dev/null +++ b/tools/check_changed_locks.py @@ -0,0 +1,51 @@ +"""Compare changed generated requirement locks without installing dependencies.""" +from __future__ import annotations + +import argparse +import json +import subprocess +from pathlib import Path + +from check_lock_delta import check_delta, parse_lock + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--base", required=True) + args = parser.parse_args() + root = Path(__file__).resolve().parents[1] + scope_file = root / "tools/lock-delta-scope.json" + excluded = json.loads(scope_file.read_text()) if scope_file.exists() else {} + changed = subprocess.check_output( + ["git", "diff", "--no-renames", "--name-only", "-z", args.base, "HEAD", "--", "requirements"], + cwd=root, + ).decode().split("\0") + for relative in changed: + if not relative.endswith(".txt"): + continue + if relative in excluded: + print(f"OUT OF SCOPE {relative}: {excluded[relative]}") + continue + path = root / relative + component = root + for part in Path(relative).parts: + component = component / part + if component.is_symlink(): + raise ValueError(f"lock path must not contain symlinks: {relative}") + if not path.is_file(): + raise ValueError(f"deleted lock requires explicit maintainer review: {relative}") + after = path.read_text(encoding="utf-8") + previous = subprocess.run( + ["git", "show", f"{args.base}:{relative}"], + cwd=root, capture_output=True, text=True, check=False, + ) + if previous.returncode: + parse_lock(after) + print(f"NEW {relative}: structure only; no prior dependency baseline") + else: + print(f"PASS {relative}: changed packages {check_delta(previous.stdout, after)}") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tools/check_lock_delta.py b/tools/check_lock_delta.py new file mode 100644 index 00000000..0b0961ce --- /dev/null +++ b/tools/check_lock_delta.py @@ -0,0 +1,165 @@ +"""Check universal hash-lock structure and optionally resolve two target platforms. + +This checks a declared subset of uv/pip-compile output, not arbitrary pip syntax. +Removal or marker changes require an explicit --allow-removed package review. +It does not authenticate package publishers or decide whether upgrades are safe. +""" +from __future__ import annotations + +import argparse +import re +import subprocess +import sys +import tempfile +import venv +from pathlib import Path + +PIN = re.compile(r"^([A-Za-z0-9_.-]+)(\[[^]]+\])?==([^\s;]+)(.*)$") +TOKEN = re.compile(r"\"[^\"]*\"|'[^']*'|[^\s'\"]+") +HASH = re.compile(r"--hash=sha256:[0-9a-f]{64}(?=\s|$)") + + +def parse_lock(text): + lines = [] + current = "" + for raw in text.splitlines(): + line = raw.strip() + if not line or line.startswith("#"): + continue + current += " " + line.removesuffix("\\").strip() + if not line.endswith("\\"): + lines.append(current.strip()) + current = "" + if current: + raise ValueError("unterminated lock entry") + result = {} + for line in lines: + match = PIN.fullmatch(line) + if not match: + raise ValueError("expected an exact version pin with SHA-256 hashes") + name, extras, version, tail = match.groups() + name = re.sub(r"[-_.]+", "-", name).lower() + extras = tuple(sorted((extras or "").strip("[]").lower().split(","))) + # Keep quoted marker literals intact, then require every trailing + # option to be a complete hash token. Removing valid hashes first can + # hide a malformed second hash (or another option) inside the marker. + tokens = list(TOKEN.finditer(tail)) + end = 0 + option = None + for token in tokens: + if tail[end:token.start()].strip(): + raise ValueError(f"{name}: unsupported lock option or malformed hash") + if option is None and token.group().startswith("--"): + option = token + end = token.end() + if tail[end:].strip(): + raise ValueError(f"{name}: unsupported lock option or malformed hash") + if option is None: + raise ValueError(f"{name}: missing SHA-256 hash") + marker = tail[:option.start()].strip() + if ((marker and not marker.startswith(";")) + or not all(HASH.fullmatch(value) + for value in tail[option.start():].split())): + raise ValueError(f"{name}: unsupported lock option or malformed hash") + # Collapse only unquoted whitespace: spaces inside a literal are part + # of marker identity, even when the resulting marker is never active. + marker = re.sub(r'''"[^"]*"|'[^']*'|\s+''', + lambda part: " " if part.group().isspace() else part.group(), marker) + key = (name, extras, marker) + if key in result: + raise ValueError(f"duplicate pin: {name} {marker}") + result[key] = (version, frozenset(tail[option.start():].split())) + if not result: + raise ValueError("empty lock") + return result + + +def check_delta(before, after, allowed=()): + old, new = parse_lock(before), parse_lock(after) + allowed = {re.sub(r"[-_.]+", "-", name).lower() for name in allowed} + missing = sorted(key for key in old.keys() - new.keys() if key[0] not in allowed) + if missing: + raise ValueError(f"removed pins or changed platform markers: {missing}") + return sorted({key[0] for key in new if old.get(key) != new[key]}) + + +def main(argv=None): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--base", help="existing git revision to compare") + parser.add_argument("--allow-removed", action="append", default=[]) + parser.add_argument("--resolve", action="store_true", help="requires uv and package network access") + parser.add_argument("--python-version", default="3.11") + parser.add_argument("--all-locks", action="store_true", + help="check current and base requirements/*.txt inventory") + parser.add_argument("locks", nargs="*", type=Path) + args = parser.parse_args(argv) + root = Path(__file__).resolve().parents[1] + if args.all_locks and args.locks: + parser.error("--all-locks cannot be combined with explicit paths") + if not args.all_locks and not args.locks: + parser.error("provide lock paths or --all-locks") + base_paths = set() + paths = args.locks + if args.all_locks: + paths = {path.relative_to(root) for path in (root / "requirements").glob("*.txt")} + if args.base: + listing = subprocess.run(["git", "ls-tree", "-r", "--name-only", "-z", + args.base, "--", "requirements/"], cwd=root, + capture_output=True, text=True, check=True) + base_paths = {Path(name) for name in listing.stdout.split("\0") + if Path(name).parent == Path("requirements") and name.endswith(".txt")} + paths |= base_paths + if not paths: + raise ValueError("no requirements locks found") + paths = sorted(paths) + for path in paths: + # Preserve the inventory's lexical name for the Git baseline. Resolving + # a symlink first can silently compare server.txt against dev.txt. + absolute = root / path + lexical = absolute.relative_to(root) + if ".." in lexical.parts: + raise ValueError(f"lock path must stay inside repository: {path}") + component = root + for part in lexical.parts: + component /= part + if component.is_symlink(): + raise ValueError(f"lock path must not contain symlinks: {path}") + if not absolute.is_file(): + raise ValueError(f"lock must be a regular file: {path}") + relative = lexical.as_posix() + after = absolute.read_text(encoding="utf-8") + parse_lock(after) + if args.base: + if args.all_locks and path not in base_paths: + # Only a successful base inventory can establish a new path. + # New locks still receive structural and requested resolver checks. + changed = sorted({key[0] for key in parse_lock(after)}) + print(f"{relative}: new lock; requires dependency review") + else: + previous = subprocess.run(["git", "show", f"{args.base}:{relative}"], cwd=root, + capture_output=True, text=True, check=False) + if previous.returncode: + raise ValueError(f"cannot read base lock {relative}; review new locks separately") + changed = check_delta(previous.stdout, after, args.allow_removed) + print(f"{relative}: changed packages {changed}") + if args.resolve: + # Each resolution uses an isolated interpreter context and ignores + # the reviewer's already-installed packages. No package is installed. + with tempfile.TemporaryDirectory(prefix="lock-resolution-") as directory: + venv.EnvBuilder(with_pip=False).create(directory) + python = Path(directory) / ("Scripts/python.exe" if sys.platform == "win32" else "bin/python") + for platform in ("x86_64-unknown-linux-gnu", "x86_64-pc-windows-msvc"): + subprocess.run(["uv", "pip", "install", "--python", str(python), "--dry-run", + "--require-hashes", "--python-version", + args.python_version, "--python-platform", platform, + "-r", str(absolute)], check=True, cwd=root) + print(f"PASS {relative}") + return 0 + + +if __name__ == "__main__": + try: + raise SystemExit(main()) + except (ValueError, OSError, subprocess.CalledProcessError) as exc: + print(f"FAIL {exc}", file=sys.stderr) + raise SystemExit(1) from None diff --git a/tools/lock-delta-scope.json b/tools/lock-delta-scope.json new file mode 100644 index 00000000..b34c0287 --- /dev/null +++ b/tools/lock-delta-scope.json @@ -0,0 +1,3 @@ +{ + "requirements/agt-overrides.txt": "compiler override constraints, not a generated hash lock" +} diff --git a/tools/tests/test_lock_delta.py b/tools/tests/test_lock_delta.py new file mode 100644 index 00000000..60d06410 --- /dev/null +++ b/tools/tests/test_lock_delta.py @@ -0,0 +1,102 @@ +"""Synthetic controls for the generated-lock subset; no package installation.""" +import importlib.util +import subprocess +import sys +import tempfile +import unittest +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[2] +SPEC = importlib.util.spec_from_file_location("check_lock_delta", ROOT / "tools/check_lock_delta.py") +guard = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(guard) +HASH = " --hash=sha256:" + "a" * 64 +BASE = "core==1" + HASH + "\ncolorama==1 ; sys_platform == 'win32'" + HASH + "\n" + + +class TestLockDelta(unittest.TestCase): + def test_version_upgrade_preserves_other_pins(self): + self.assertEqual(guard.check_delta(BASE, BASE.replace("core==1", "core==2")), ["core"]) + + def test_independent_corruptions(self): + cases = [ + "core==1" + HASH, + BASE.replace(" ; sys_platform == 'win32'", ""), + BASE.replace(HASH, ""), + BASE.replace("a" * 64, "a" * 63), + BASE + "core==2" + HASH, + "core>=1" + HASH, + "", + ] + for index, after in enumerate(cases): + with self.subTest(index=index), self.assertRaises(ValueError): + guard.check_delta(BASE, after) + + def test_extra_loss_and_explicit_scoped_removal(self): + with self.assertRaises(ValueError): + guard.check_delta("coverage[toml]==1" + HASH, "coverage==1" + HASH) + self.assertEqual(guard.check_delta(BASE, "core==1" + HASH, ["colorama"]), []) + with self.assertRaises(ValueError): + guard.check_delta(BASE, "colorama==1" + HASH, ["colorama"]) + + def test_optimized_cli_still_refuses_missing_hash(self): + with tempfile.NamedTemporaryFile(mode="w", suffix=".txt", dir=ROOT / "tools") as handle: + handle.write("unhashed==1\n") + handle.flush() + result = subprocess.run( + [sys.executable, "-O", str(ROOT / "tools/check_lock_delta.py"), handle.name], + capture_output=True, text=True, check=False, + ) + self.assertEqual(result.returncode, 1) + self.assertIn("missing SHA-256", result.stderr) + + + +class TestChangedLockCLI(unittest.TestCase): + def test_actual_driver_refuses_regeneration_loss_and_accepts_upgrade(self): + import shutil + + cases = [ + ("upgrade", BASE.replace("core==1", "core==2"), True), + ("removed-marker", BASE.replace(" ; sys_platform == 'win32'", ""), False), + ("deleted", None, False), + ("missing-hash", "core==2\n", False), + ] + for name, after, expected in cases: + with self.subTest(case=name), tempfile.TemporaryDirectory() as directory: + root = Path(directory) + (root / "tools").mkdir() + (root / "requirements").mkdir() + for source in ("check_lock_delta.py", "check_changed_locks.py"): + shutil.copy2(ROOT / "tools" / source, root / "tools" / source) + lock = root / "requirements/test.txt" + lock.write_text(BASE) + + def git(*args): + return subprocess.check_output( + ["git", "-c", "user.name=Lock test", "-c", + "user.email=lock-test@example.invalid", "-c", + "commit.gpgsign=false", *args], + cwd=root, stderr=subprocess.DEVNULL, text=True, + ).strip() + + git("init") + git("add", ".") + git("commit", "-m", "synthetic baseline") + base = git("rev-parse", "HEAD") + if after is None: + lock.unlink() + else: + lock.write_text(after) + git("add", "-A") + git("commit", "-m", "synthetic candidate") + result = subprocess.run( + [sys.executable, str(root / "tools/check_changed_locks.py"), + "--base", base], cwd=root, capture_output=True, text=True, + check=False, + ) + self.assertEqual(result.returncode == 0, expected, result.stderr) + + +if __name__ == "__main__": + unittest.main() diff --git a/tools/tests/test_lock_hashes.py b/tools/tests/test_lock_hashes.py new file mode 100644 index 00000000..35212adc --- /dev/null +++ b/tools/tests/test_lock_hashes.py @@ -0,0 +1,160 @@ +"""Hash-only changes and full lock discovery through real Git/CLI boundaries.""" +import importlib.util +import shutil +import subprocess +import sys +import tempfile +import unittest +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[2] +SPEC = importlib.util.spec_from_file_location("check_lock_hashes", ROOT / "tools/check_lock_delta.py") +guard = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(guard) +H1 = " --hash=sha256:" + "a" * 64 +H2 = " --hash=sha256:" + "b" * 64 + + +class TestHashDelta(unittest.TestCase): + def test_quoted_whitespace_is_preserved(self): + for quote in ('"', "'"): + before = f'demo==1 ; os_name == {quote}foo bar{quote}' + H1 + after = before.replace('foo bar', 'foo bar') + with self.subTest(quote=quote): + with self.assertRaisesRegex(ValueError, 'changed platform markers'): + guard.check_delta(before, after) + self.assertEqual(guard.check_delta(before, after, ['demo']), ['demo']) + self.assertEqual(guard.check_delta(before, before.replace( + '; os_name ==', '; os_name\t ==')), []) + + def test_same_version_hash_changes_are_reported(self): + for marker in ('', ' ; sys_platform == "win32"'): + before = 'demo==1' + marker + H1 + for after in ('demo==1' + marker + H2, before + H2): + for old, new in ((before, after), (after, before)): + with self.subTest(marker=marker, before=old, after=new): + self.assertEqual(guard.check_delta(old, new), ['demo']) + + def test_hash_order_and_duplicates_are_not_changes(self): + self.assertEqual(guard.check_delta('demo==1' + H1 + H2, + 'demo==1' + H2 + H1 + H1), []) + + +class TestLockInventoryCLI(unittest.TestCase): + def setUp(self): + self.directory = tempfile.TemporaryDirectory() + self.addCleanup(self.directory.cleanup) + self.root = Path(self.directory.name) + (self.root / 'tools').mkdir() + (self.root / 'requirements').mkdir() + for name in ('check_lock_delta.py', 'check_changed_locks.py'): + shutil.copy2(ROOT / 'tools' / name, self.root / 'tools' / name) + # An empty requirements inventory is intentional: later fixtures are new + # or existing locks according to a real successful Git base listing. + self.git('init', '-q') + self.git('add', '.') + self.git('commit', '-qm', 'baseline') + self.base = self.git('rev-parse', 'HEAD') + + def git(self, *args): + return subprocess.check_output( + ['git', '-c', 'user.name=Lock test', '-c', 'user.email=lock-test@example.invalid', + '-c', 'commit.gpgsign=false', *args], cwd=self.root, + stderr=subprocess.DEVNULL, text=True).strip() + + def commit(self): + self.git('add', '-A') + self.git('commit', '-qm', 'candidate') + + def run_cli(self, name, *args): + return subprocess.run([sys.executable, str(self.root / 'tools' / name), *args], + cwd=self.root, capture_output=True, text=True, check=False) + + def test_driver_reports_same_version_hash_changes(self): + lock = self.root / 'requirements/runtime.txt' + lock.write_text('demo==1' + H1) + self.commit() + base = self.git('rev-parse', 'HEAD') + lock.write_text('demo==1' + H2) + self.commit() + result = self.run_cli('check_changed_locks.py', '--base', base) + self.assertEqual(result.returncode, 0, result.stderr) + self.assertIn("PASS requirements/runtime.txt: changed packages ['demo']", result.stdout) + + def test_driver_discovers_each_requirement_lock_name(self): + for name in ('runtime', 'server', 'custom-generated'): + with self.subTest(name=name): + lock = self.root / f'requirements/{name}.txt' + lock.write_text('demo==1' + H1) + self.commit() + result = self.run_cli('check_changed_locks.py', '--base', self.base) + self.assertEqual(result.returncode, 0, result.stderr) + self.assertIn(f'NEW requirements/{name}.txt', result.stdout) + lock.write_text('demo==1') + self.commit() + result = self.run_cli('check_changed_locks.py', '--base', self.base) + self.assertNotEqual(result.returncode, 0) + self.assertIn('missing SHA-256 hash', result.stderr) + lock.write_text('demo==1' + H1) + self.commit() + + def test_manual_inventory_checks_existing_and_new_locks(self): + (self.root / 'requirements/runtime.txt').write_text('demo==1' + H1) + self.commit() + base = self.git('rev-parse', 'HEAD') + (self.root / 'requirements/server.txt').write_text('other==1' + H2) + result = self.run_cli('check_lock_delta.py', '--base', base, '--all-locks') + self.assertEqual(result.returncode, 0, result.stderr) + self.assertIn('PASS requirements/runtime.txt', result.stdout) + self.assertIn('PASS requirements/server.txt', result.stdout) + self.assertIn('new lock; requires dependency review', result.stdout) + (self.root / 'requirements/server.txt').write_text('other==1') + result = self.run_cli('check_lock_delta.py', '--base', base, '--all-locks') + self.assertNotEqual(result.returncode, 0) + self.assertIn('missing SHA-256 hash', result.stderr) + + def test_manual_inventory_cannot_hide_deleted_or_renamed_locks(self): + original = self.root / 'requirements/runtime.txt' + original.write_text('demo==1' + H1) + self.commit() + base = self.git('rev-parse', 'HEAD') + original.unlink() + for renamed in (False, True): + with self.subTest(renamed=renamed): + if renamed: + (self.root / 'requirements/renamed.txt').write_text('demo==1' + H1) + result = self.run_cli('check_lock_delta.py', '--base', base, '--all-locks') + self.assertNotEqual(result.returncode, 0) + self.assertIn('requirements/runtime.txt', result.stderr) + + def test_symlink_refused_by_inventory_explicit_and_changed_driver(self): + server = self.root / 'requirements/server.txt' + server.write_text('demo==1' + H1 + '\nserver-only==1' + H1) + (self.root / 'requirements/dev.txt').write_text('demo==1' + H1) + self.commit() + base = self.git('rev-parse', 'HEAD') + server.unlink() + try: + server.symlink_to('dev.txt') + except OSError as exc: + self.skipTest(f'symlink creation unavailable: {exc}') + self.commit() + commands = [('check_lock_delta.py', '--base', base, '--all-locks'), + ('check_lock_delta.py', '--base', base, 'requirements/server.txt'), + ('check_changed_locks.py', '--base', base)] + for command in commands: + with self.subTest(command=command): + result = self.run_cli(*command) + self.assertNotEqual(result.returncode, 0, result.stdout) + self.assertIn('symlink', result.stderr) + self.assertIn('requirements/server.txt', result.stderr) + + def test_bad_base_cannot_be_mistaken_for_new_inventory(self): + (self.root / 'requirements/runtime.txt').write_text('demo==1' + H1) + result = self.run_cli('check_lock_delta.py', '--base', 'missing-base', '--all-locks') + self.assertNotEqual(result.returncode, 0) + self.assertNotIn('new lock', result.stdout) + + +if __name__ == '__main__': + unittest.main() diff --git a/tools/tests/test_lock_parent_symlinks.py b/tools/tests/test_lock_parent_symlinks.py new file mode 100644 index 00000000..322c4df6 --- /dev/null +++ b/tools/tests/test_lock_parent_symlinks.py @@ -0,0 +1,65 @@ +"""Reject deleted Git locks substituted through a symlinked parent.""" +import shutil +import subprocess +import sys +import tempfile +import unittest +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[2] +LOCK = "demo==1 --hash=sha256:" + "a" * 64 + "\n" + + +class TestLockParentSymlinks(unittest.TestCase): + def test_deleted_lock_parent_substitution_and_ordinary_control(self): + for linked_parent in (False, True): + with self.subTest(linked_parent=linked_parent), tempfile.TemporaryDirectory() as directory: + root = Path(directory) + (root / "tools").mkdir() + nested = root / "requirements/nested" + nested.mkdir(parents=True) + elsewhere = root / "elsewhere" + elsewhere.mkdir() + (nested / "dev.txt").write_text(LOCK, encoding="utf-8") + (elsewhere / "dev.txt").write_text(LOCK, encoding="utf-8") + for source in ("check_lock_delta.py", "check_changed_locks.py"): + shutil.copy2(ROOT / "tools" / source, root / "tools" / source) + + def git(*args): + return subprocess.check_output( + ["git", "-c", "user.name=Lock test", "-c", + "user.email=lock-test@example.invalid", "-c", + "commit.gpgsign=false", *args], + cwd=root, stderr=subprocess.DEVNULL, text=True, + ).strip() + + git("init") + git("add", ".") + git("commit", "-m", "synthetic baseline") + base = git("rev-parse", "HEAD") + if linked_parent: + shutil.rmtree(nested) + try: + nested.symlink_to("../elsewhere", target_is_directory=True) + except OSError as exc: + self.skipTest(f"symlink creation unavailable: {exc}") + else: + (nested / "dev.txt").write_text("# ordinary directory control\n" + LOCK, encoding="utf-8") + git("add", "-A") + git("commit", "-m", "replace parent or edit ordinary lock") + status = git("diff", "--no-renames", "--name-status", base, "HEAD") + self.assertIn(("D" if linked_parent else "M") + "\trequirements/nested/dev.txt", status) + for optimized in (False, True): + with self.subTest(optimized=optimized): + result = subprocess.run( + [sys.executable, *(["-O"] if optimized else []), + str(root / "tools/check_changed_locks.py"), "--base", base], + cwd=root, capture_output=True, text=True, check=False, + ) + if linked_parent: + self.assertNotEqual(result.returncode, 0, result.stdout) + self.assertIn("lock path must not contain symlinks: requirements/nested/dev.txt", result.stderr) + self.assertNotIn("PASS requirements/nested/dev.txt", result.stdout) + else: + self.assertEqual(result.returncode, 0, result.stderr) + self.assertIn("PASS requirements/nested/dev.txt: changed packages []", result.stdout) diff --git a/tools/tests/test_lock_renames.py b/tools/tests/test_lock_renames.py new file mode 100644 index 00000000..9785d46f --- /dev/null +++ b/tools/tests/test_lock_renames.py @@ -0,0 +1,59 @@ +"""Exercise Git rename detection at the generated-lock driver boundary.""" +import shutil +import subprocess +import sys +import tempfile +import unittest +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[2] +LOCK = "".join(f"dependency-{i}==1 --hash=sha256:{'a' * 64}\n" for i in range(10)) + + +class TestRenamedLockCLI(unittest.TestCase): + def test_renamed_lock_requires_review_with_or_without_dependency_loss(self): + for drop_dependency in (False, True): + with self.subTest(drop_dependency=drop_dependency), tempfile.TemporaryDirectory() as directory: + root = Path(directory) + (root / "tools").mkdir() + (root / "requirements").mkdir() + for source in ("check_lock_delta.py", "check_changed_locks.py"): + shutil.copy2(ROOT / "tools" / source, root / "tools" / source) + original = root / "requirements/dev.txt" + original.write_text(LOCK, encoding="utf-8") + + def git(*args): + return subprocess.check_output( + ["git", "-c", "user.name=Lock test", "-c", + "user.email=lock-test@example.invalid", "-c", + "commit.gpgsign=false", *args], + cwd=root, stderr=subprocess.DEVNULL, text=True, + ).strip() + + git("init") + git("config", "diff.renames", "true") + git("add", ".") + git("commit", "-m", "synthetic baseline") + base = git("rev-parse", "HEAD") + renamed = original.with_name("renamed.txt") + original.rename(renamed) + if drop_dependency: + renamed.write_text("".join(LOCK.splitlines(keepends=True)[:-1]), encoding="utf-8") + git("add", "-A") + git("commit", "-m", "rename generated lock") + # Confirm this fixture really exercises Git's rename classification. + status = git("diff", "--name-status", "--find-renames", base, "HEAD") + self.assertRegex(status, r"^R\d+\s+requirements/dev\.txt\s+requirements/renamed\.txt$") + for optimized in (False, True): + with self.subTest(optimized=optimized): + result = subprocess.run( + [sys.executable, *(["-O"] if optimized else []), + str(root / "tools/check_changed_locks.py"), "--base", base], + cwd=root, capture_output=True, text=True, check=False, + ) + self.assertNotEqual(result.returncode, 0, result.stdout) + self.assertIn("deleted lock requires explicit maintainer review: requirements/dev.txt", result.stderr) + + +if __name__ == "__main__": + unittest.main() diff --git a/tools/tests/test_lock_suffix.py b/tools/tests/test_lock_suffix.py new file mode 100644 index 00000000..2fb0090f --- /dev/null +++ b/tools/tests/test_lock_suffix.py @@ -0,0 +1,93 @@ +"""Paired marker/hash controls through the parser and real Git driver.""" +import importlib.util +import shutil +import subprocess +import sys +import tempfile +import unittest +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[2] +SPEC = importlib.util.spec_from_file_location("check_lock_delta_suffix", ROOT / "tools/check_lock_delta.py") +guard = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(guard) + +SUFFIX_HASH = " --hash=sha256:" + "a" * 64 +SUFFIX_PIN = 'demo==1 ; python_version >= "3.11"' + + +class TestHashSuffix(unittest.TestCase): + def test_marker_and_multiple_hashes_remain_valid(self): + for marker in ('', ' ; python_version >= "3.11"', + ' ; os_name == "value --hash=sha256:BAD"'): + with self.subTest(marker=marker): + text = "demo==1" + marker + SUFFIX_HASH + SUFFIX_HASH.replace("a" * 64, "b" * 64) + self.assertEqual(guard.parse_lock(text), { + ("demo", ("",), marker.strip()): ( + "1", frozenset([SUFFIX_HASH.strip(), SUFFIX_HASH.replace("a" * 64, "b" * 64).strip()]))}) + + def test_entire_hash_suffix_is_validated(self): + for marker in ('', ' ; python_version >= "3.11"'): + for extra in (' --hash=sha256:BAD', ' --hash=sha256:' + 'b' * 63, + ' --hash=sha256:' + 'b' * 65, ' --hash=sha512:' + 'b' * 64, + ' --index-url=https://example.invalid', ' trailing-junk'): + with self.subTest(marker=marker, extra=extra): + with self.assertRaisesRegex(ValueError, "unsupported lock option or malformed hash"): + guard.parse_lock("demo==1" + marker + SUFFIX_HASH + extra) + + def test_unsupported_option_before_valid_hash_is_refused(self): + with self.assertRaisesRegex(ValueError, "unsupported lock option or malformed hash"): + guard.parse_lock(SUFFIX_PIN + " --index-url=https://example.invalid" + SUFFIX_HASH) + + def test_required_lock_cli_refuses_malformed_suffix(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + (root / "tools").mkdir() + shutil.copy2(ROOT / "tools/check_lock_delta.py", root / "tools/check_lock_delta.py") + lock = root / "test.txt" + for optimize in ([], ["-O"]): + for extra, expected in (('', 0), (' --hash=sha256:BAD', 1), + (' --index-url=https://example.invalid', 1)): + with self.subTest(optimize=optimize, extra=extra): + lock.write_text(SUFFIX_PIN + SUFFIX_HASH + extra) + result = subprocess.run( + [sys.executable, *optimize, str(root / "tools/check_lock_delta.py"), "test.txt"], + cwd=root, capture_output=True, text=True, check=False) + self.assertEqual(result.returncode, expected, result.stdout + result.stderr) + if expected: + self.assertIn("unsupported lock option or malformed hash", result.stderr) + + +class TestNewLockSuffix(unittest.TestCase): + def test_real_git_new_lock_checks_whole_suffix(self): + for extra, expected in (('', 0), (' --hash=sha256:BAD', 1), + (' --index-url=https://example.invalid', 1)): + with self.subTest(extra=extra), tempfile.TemporaryDirectory() as directory: + root = Path(directory) + (root / "tools").mkdir() + (root / "requirements").mkdir() + for source in ("check_lock_delta.py", "check_changed_locks.py"): + shutil.copy2(ROOT / "tools" / source, root / "tools" / source) + + def git(*args): + return subprocess.check_output( + ["git", "-c", "user.name=Lock test", "-c", "user.email=lock-test@example.invalid", + "-c", "commit.gpgsign=false", *args], + cwd=root, stderr=subprocess.DEVNULL, text=True).strip() + + git("init") + git("add", ".") + git("commit", "-m", "synthetic baseline") + base = git("rev-parse", "HEAD") + (root / "requirements/new.txt").write_text(SUFFIX_PIN + SUFFIX_HASH + extra) + git("add", ".") + git("commit", "-m", "synthetic new lock") + for optimize in ([], ["-O"]): + result = subprocess.run( + [sys.executable, *optimize, str(root / "tools/check_changed_locks.py"), "--base", base], + cwd=root, capture_output=True, text=True, check=False) + self.assertEqual(result.returncode, expected, result.stdout + result.stderr) + if expected: + self.assertIn("unsupported lock option or malformed hash", result.stderr) + else: + self.assertIn("NEW requirements/new.txt", result.stdout)