253 lines
8.7 KiB
Python
253 lines
8.7 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import datetime, timezone
|
|
import json
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
import zipfile
|
|
|
|
import pytest
|
|
|
|
ROOT=Path(__file__).resolve().parents[1]
|
|
APP=ROOT/"active-runtime/app"
|
|
sys.path.insert(0,str(APP))
|
|
|
|
import aios_all_modified_files_packager as packager
|
|
|
|
|
|
def init_repo(tmp_path: Path) -> tuple[Path, Path, str]:
|
|
bare=tmp_path/"repo.git"
|
|
work=tmp_path/"work"
|
|
subprocess.run(["git","init","--bare",str(bare)],check=True,capture_output=True)
|
|
subprocess.run(["git","clone",str(bare),str(work)],check=True,capture_output=True)
|
|
subprocess.run(["git","-C",str(work),"config","user.email","test@example.invalid"],check=True)
|
|
subprocess.run(["git","-C",str(work),"config","user.name","AIOS Test"],check=True)
|
|
(work/"README.md").write_text("base\n",encoding="utf-8")
|
|
subprocess.run(["git","-C",str(work),"add","README.md"],check=True)
|
|
subprocess.run(["git","-C",str(work),"commit","-m","base"],check=True,capture_output=True)
|
|
subprocess.run(["git","-C",str(work),"push","origin","HEAD:refs/heads/main"],check=True,capture_output=True)
|
|
base=subprocess.check_output(["git","-C",str(work),"rev-parse","HEAD"],text=True).strip()
|
|
return bare,work,base
|
|
|
|
|
|
def commit(work: Path, message: str) -> str:
|
|
subprocess.run(["git","-C",str(work),"add","-A"],check=True)
|
|
subprocess.run(["git","-C",str(work),"commit","-m",message],check=True,capture_output=True)
|
|
candidate=subprocess.check_output(["git","-C",str(work),"rev-parse","HEAD"],text=True).strip()
|
|
subprocess.run(["git","-C",str(work),"push","origin","HEAD:refs/heads/main"],check=True,capture_output=True)
|
|
return candidate
|
|
|
|
|
|
def real_pem_header() -> bytes:
|
|
return b"-----" + b"BEGIN " + b"PRIVATE KEY" + b"-----"
|
|
|
|
|
|
def bearer_value() -> bytes:
|
|
return b"Authorization" + b": Bearer " + (b"a"*40)
|
|
|
|
|
|
def key_value_secret() -> bytes:
|
|
return b"api_key" + b"=" + (b"x"*24)
|
|
|
|
|
|
def test_plain_private_key_words_are_not_secret_findings():
|
|
for value in (
|
|
b"BEGIN PRIVATE KEY",
|
|
b"PRIVATE KEY",
|
|
b"PEM private-key header",
|
|
b"assert 'BEGIN PRIVATE KEY' in message",
|
|
):
|
|
assert packager.secret_finding(value) is None
|
|
|
|
|
|
def test_complete_pem_private_key_header_is_rejected():
|
|
assert packager.secret_finding(real_pem_header())=="PEM_PRIVATE_KEY"
|
|
assert packager.secret_finding(
|
|
b"-----BEGIN OPENSSH " + b"PRIVATE KEY-----"
|
|
)=="PEM_PRIVATE_KEY"
|
|
|
|
|
|
def test_bearer_and_key_value_secret_detection_remains_fail_closed():
|
|
assert packager.secret_finding(bearer_value())=="AUTHORIZATION_BEARER"
|
|
assert packager.secret_finding(key_value_secret())=="KEY_VALUE_SECRET"
|
|
|
|
|
|
def test_short_or_descriptive_secret_words_do_not_false_positive():
|
|
assert packager.secret_finding(b"secret=short") is None
|
|
assert packager.secret_finding(b"password field is validated") is None
|
|
assert packager.secret_finding(b"api_key example placeholder") is None
|
|
|
|
|
|
def test_end_to_end_bundle_accepts_literal_begin_private_key_text(tmp_path):
|
|
bare,work,base=init_repo(tmp_path)
|
|
(work/"tests").mkdir()
|
|
(work/"tests"/"test_text.py").write_text(
|
|
'ERROR = "BEGIN PRIVATE KEY is not a complete PEM header"\n',
|
|
encoding="utf-8",
|
|
)
|
|
(work/"docs").mkdir()
|
|
(work/"docs"/"note.md").write_text("Scanner regression.\n",encoding="utf-8")
|
|
candidate=commit(work,"candidate")
|
|
|
|
extra=tmp_path/"execution"
|
|
extra.mkdir()
|
|
(extra/"evidence.json").write_text('{"status":"PASS"}\n',encoding="utf-8")
|
|
out=tmp_path/"out"
|
|
|
|
summary=packager.build_bundle(
|
|
git_dir=bare,
|
|
base=base,
|
|
candidate=candidate,
|
|
task_id="r42-packager-test",
|
|
extra_dir=extra,
|
|
output_root=out,
|
|
now=datetime(2026,8,19,tzinfo=timezone.utc),
|
|
)
|
|
|
|
assert summary["changed_file_count"]==2
|
|
assert summary["execution_file_count"]==1
|
|
assert summary["repository_document_count"]==1
|
|
assert summary["execution_document_count"]==0
|
|
bundle_path=Path(summary["bundle_path"])
|
|
assert bundle_path.is_file()
|
|
assert packager.sha256(bundle_path.read_bytes())==summary["bundle_sha256"]
|
|
|
|
with zipfile.ZipFile(bundle_path) as archive:
|
|
names=set(archive.namelist())
|
|
assert "repo/after/tests/test_text.py" in names
|
|
assert "repo/after/docs/note.md" in names
|
|
assert "execution/evidence.json" in names
|
|
assert "documents-index.json" in names
|
|
assert "all-modified-files-manifest.json" in names
|
|
assert "checksums.sha256" in names
|
|
manifest=json.loads(archive.read("all-modified-files-manifest.json"))
|
|
assert manifest["schema"]=="aios.all-modified-files.bundle.v1"
|
|
assert manifest["candidate_oid"]==candidate
|
|
|
|
packager.verify_checksums(out/"bundle")
|
|
|
|
|
|
def test_end_to_end_bundle_rejects_real_pem_header(tmp_path):
|
|
bare,work,base=init_repo(tmp_path)
|
|
(work/"credential.txt").write_bytes(
|
|
real_pem_header()+b"\nnot-a-real-key-body\n"
|
|
)
|
|
candidate=commit(work,"credential-like candidate")
|
|
|
|
with pytest.raises(
|
|
packager.PackagerError,
|
|
match="PEM_PRIVATE_KEY:after:credential.txt",
|
|
):
|
|
packager.build_bundle(
|
|
git_dir=bare,
|
|
base=base,
|
|
candidate=candidate,
|
|
task_id="r42-packager-pem-reject",
|
|
output_root=tmp_path/"out",
|
|
)
|
|
|
|
|
|
def test_execution_evidence_bearer_is_rejected(tmp_path):
|
|
bare,work,base=init_repo(tmp_path)
|
|
(work/"normal.txt").write_text("normal\n",encoding="utf-8")
|
|
candidate=commit(work,"normal")
|
|
extra=tmp_path/"execution"
|
|
extra.mkdir()
|
|
(extra/"unsafe.txt").write_bytes(bearer_value()+b"\n")
|
|
|
|
with pytest.raises(
|
|
packager.PackagerError,
|
|
match="AUTHORIZATION_BEARER:execution:unsafe.txt",
|
|
):
|
|
packager.build_bundle(
|
|
git_dir=bare,
|
|
base=base,
|
|
candidate=candidate,
|
|
task_id="r42-packager-bearer-reject",
|
|
extra_dir=extra,
|
|
output_root=tmp_path/"out",
|
|
)
|
|
|
|
|
|
def test_execution_symlink_is_rejected(tmp_path):
|
|
bare,work,base=init_repo(tmp_path)
|
|
(work/"normal.txt").write_text("normal\n",encoding="utf-8")
|
|
candidate=commit(work,"normal")
|
|
extra=tmp_path/"execution"
|
|
extra.mkdir()
|
|
target=extra/"target.txt"
|
|
target.write_text("safe\n",encoding="utf-8")
|
|
(extra/"link.txt").symlink_to(target)
|
|
|
|
with pytest.raises(
|
|
packager.PackagerError,
|
|
match="EXTRA_FILE_SYMLINK_REJECTED",
|
|
):
|
|
packager.build_bundle(
|
|
git_dir=bare,
|
|
base=base,
|
|
candidate=candidate,
|
|
task_id="r42-packager-symlink-reject",
|
|
extra_dir=extra,
|
|
output_root=tmp_path/"out",
|
|
)
|
|
|
|
|
|
def test_non_ancestor_lineage_is_rejected(tmp_path):
|
|
bare,work,base=init_repo(tmp_path)
|
|
(work/"one.txt").write_text("one\n",encoding="utf-8")
|
|
first=commit(work,"first")
|
|
|
|
subprocess.run(["git","-C",str(work),"checkout","--detach",base],check=True,capture_output=True)
|
|
(work/"other.txt").write_text("other\n",encoding="utf-8")
|
|
subprocess.run(["git","-C",str(work),"add","other.txt"],check=True)
|
|
subprocess.run(["git","-C",str(work),"commit","-m","other"],check=True,capture_output=True)
|
|
other=subprocess.check_output(["git","-C",str(work),"rev-parse","HEAD"],text=True).strip()
|
|
subprocess.run(
|
|
["git","-C",str(work),"push","origin","HEAD:refs/heads/other"],
|
|
check=True,
|
|
capture_output=True,
|
|
)
|
|
|
|
with pytest.raises(packager.PackagerError,match="LINEAGE_INVALID"):
|
|
packager.validate_lineage(bare,first,other)
|
|
|
|
|
|
def test_bundle_checksum_closure_detects_tamper(tmp_path):
|
|
bare,work,base=init_repo(tmp_path)
|
|
(work/"normal.txt").write_text("normal\n",encoding="utf-8")
|
|
candidate=commit(work,"normal")
|
|
out=tmp_path/"out"
|
|
packager.build_bundle(
|
|
git_dir=bare,
|
|
base=base,
|
|
candidate=candidate,
|
|
task_id="r42-packager-checksum",
|
|
output_root=out,
|
|
)
|
|
(out/"bundle"/"repo"/"after"/"normal.txt").write_text(
|
|
"tampered\n",
|
|
encoding="utf-8",
|
|
)
|
|
with pytest.raises(
|
|
packager.PackagerError,
|
|
match="CHECKSUM_CLOSURE_INVALID",
|
|
):
|
|
packager.verify_checksums(out/"bundle")
|
|
|
|
|
|
def test_source_does_not_contain_complete_pem_header_literal():
|
|
source=(ROOT/"active-runtime/app/aios_all_modified_files_packager.py").read_bytes()
|
|
tests=(ROOT/"tests/test_all_modified_files_packager.py").read_bytes()
|
|
complete=real_pem_header()
|
|
assert complete not in source
|
|
assert complete not in tests
|
|
|
|
|
|
def test_document_path_contract():
|
|
assert packager.is_document_path("docs/spec.md")
|
|
assert packager.is_document_path("README")
|
|
assert packager.is_document_path("CHANGELOG.md")
|
|
assert not packager.is_document_path("active-runtime/app/main.py")
|