diff --git a/packages/syft-job/tests/test_submission_bytes.py b/packages/syft-job/tests/test_submission_bytes.py new file mode 100644 index 00000000000..0320f834a0b --- /dev/null +++ b/packages/syft-job/tests/test_submission_bytes.py @@ -0,0 +1,153 @@ +"""Receipt, approval and execution must agree on the submitted bytes.""" + +from pathlib import Path + +import pytest +from syft_job.client import JobClient +from syft_job.config import SyftJobConfig +from syft_job.job_runner import SyftJobRunner +from syft_job.policies import JobPolicies, ResolvedApi +from syft_job.submission import request_hash, submission_hash + +import syft + +DO_EMAIL = "do@test.org" +DS_EMAIL = "ds@test.org" +ClientPair = tuple[JobClient, JobClient] + + +@pytest.fixture +def clients(tmp_path: Path) -> ClientPair: + syftbox = tmp_path / "SyftBox" + syftbox.mkdir() + do = JobClient(SyftJobConfig(syftbox_folder=syftbox, current_user_email=DO_EMAIL)) + ds = JobClient(SyftJobConfig(syftbox_folder=syftbox, current_user_email=DS_EMAIL)) + ds.peer_install_sources[DO_EMAIL] = str(Path(syft.__file__).resolve().parents[1]) + do.setup_ds_job_folder_as_do(DS_EMAIL) + return do, ds + + +@pytest.mark.parametrize("newline", [b"\n", b"\r\n"], ids=["lf", "crlf"]) +def test_raw_bytes_bind_receipt_approval_and_execution( + clients: ClientPair, tmp_path: Path, newline: bytes +) -> None: + do, ds = clients + source = tmp_path / "source" + source.mkdir() + code = ( + ( + "from pathlib import Path\n" + "assert Path(__file__).with_name('payload.bin').read_bytes() == " + "b'\\xff\\x00\\r\\n\\x80'\n" + "print('données approuvées')\n" + ) + .encode("utf-8") + .replace(b"\n", newline) + ) + (source / "main.py").write_bytes(code) + (source / "payload.bin").write_bytes(b"\xff\x00\r\n\x80") + inbox = ds.submit_python_job(DO_EMAIL, str(source), "bytes.job") + + [job] = do.jobs + assert job.status == "pending" + record = do.manager.read_owner_record(job._ref) + assert record is not None + received = submission_hash(inbox) + assert record.received_hash == received + job.approve() + record = do.manager.read_owner_record(job._ref) + assert record is not None + assert record.approved_hash == record.received_hash == received + + SyftJobRunner(do.config).process_approved_jobs(stream_output=False, timeout=60) + + [job] = do.jobs + assert job.status == "done" + assert (job.job_staging_path / "stdout.txt").read_text().strip() == ( + "données approuvées" + ) + assert (inbox / "code/main.py").read_bytes() == code + assert (inbox / "code/payload.bin").read_bytes() == b"\xff\x00\r\n\x80" + + +@pytest.mark.parametrize("change", ["newline", "binary"]) +@pytest.mark.parametrize( + "when", ["before_receipt", "before_approval", "after_approval"] +) +def test_changed_bytes_never_execute( + clients: ClientPair, tmp_path: Path, when: str, change: str +) -> None: + do, ds = clients + source = tmp_path / "source" + source.mkdir() + marker = tmp_path / "executed.txt" + (source / "main.py").write_text( + f"from pathlib import Path\nPath({str(marker)!r}).write_text('executed')\n" + ) + (source / "payload.bin").write_bytes(b"\xff\x00\r\n\x80") + inbox = ds.submit_python_job(DO_EMAIL, str(source), "changed.job") + if when != "before_receipt": + [job] = do.jobs + assert job.status == "pending" + if when == "after_approval": + job.approve() + + if change == "newline": + target = inbox / "code/main.py" + target.write_bytes(target.read_bytes().replace(b"\n", b"\r\n")) + else: + (inbox / "code/payload.bin").write_bytes(b"\xff\x00\n\x80") + + if when == "before_receipt": + [received_job] = do.jobs + assert received_job.status == "received" + elif when == "before_approval": + with pytest.raises(ValueError, match="changed after it was received"): + job.approve() + else: + SyftJobRunner(do.config).process_approved_jobs(stream_output=False, timeout=10) + [failed_job] = do.jobs + assert failed_job.status == "failed" + assert not marker.exists() + + +@pytest.mark.parametrize( + "payload", + [b'{"word":"caf\xc3\xa9"}\r\n', b"\xff\x00\r\n\x80"], + ids=["utf8-crlf", "binary"], +) +def test_request_bytes_survive_conversion_to_approved_job( + clients: ClientPair, tmp_path: Path, payload: bytes +) -> None: + do, ds = clients + files = tmp_path / "api" + files.mkdir() + (files / "run.sh").write_text("echo approved\n") + policies = JobPolicies(api_name="bytes", state_dir=tmp_path / "policy") + + class ByteApi: + def __call__(self, api_name: str) -> ResolvedApi: + assert api_name == "bytes" + return ResolvedApi(api_name, files, policies) + + def check_request( + self, api_name: str, request_dir: Path, submitter: str + ) -> None: + self(api_name) + + do.api_resolver = ByteApi() + request_dir = ds.submit_api_request( + DO_EMAIL, "bytes.job", "bytes", {"code/payload.bin": payload} + ) + [ref] = list(ds.manager.iter_request_refs(DO_EMAIL)) + assert ds.manager.read_request(ref).files_hash == request_hash(request_dir) + + jobs = do.jobs + assert len(jobs) == 1 + [job] = jobs + assert job.status == "approved" + assert (job.job_submission_path / "code/payload.bin").read_bytes() == payload + record = do.manager.read_owner_record(job._ref) + assert record is not None + assert record.approved_hash == record.received_hash + assert record.approved_hash == submission_hash(job.job_submission_path) diff --git a/tests/unit/test_job_byte_transport.py b/tests/unit/test_job_byte_transport.py new file mode 100644 index 00000000000..f2ef315bb92 --- /dev/null +++ b/tests/unit/test_job_byte_transport.py @@ -0,0 +1,81 @@ +"""Keep the receipt hash contract through filesystem-backed mock Drive sync.""" + +from pathlib import Path + +import pytest +from syft_job.client import JobClient +from syft_job.config import SyftJobConfig +from syft_job.submission import SUBMISSION_HASH_HEADER, code_hash + +from syft.sync.syftbox_manager import SyftboxManager + +TransportPair = tuple[JobClient, JobClient, SyftboxManager, SyftboxManager] + + +@pytest.fixture +def transport(tmp_path: Path) -> TransportPair: + ds_sync, do_sync = SyftboxManager.pair_with_mock_drive_service_connection( + email1="do@test.org", + email2="ds@test.org", + base_path1=str(tmp_path / "owner"), + base_path2=str(tmp_path / "submitter"), + use_in_memory_cache=False, + persist_owner_state=False, + sync_automatically=False, + ) + do = JobClient( + SyftJobConfig( + syftbox_folder=do_sync.syftbox_folder, current_user_email=do_sync.email + ) + ) + ds = JobClient( + SyftJobConfig( + syftbox_folder=ds_sync.syftbox_folder, current_user_email=ds_sync.email + ) + ) + do.setup_ds_job_folder_as_do(ds_sync.email) + return do, ds, do_sync, ds_sync + + +@pytest.mark.parametrize( + "payload", [b"caf\xc3\xa9\r\n", b"\xff\x00\r\n\x80"], ids=["utf8-crlf", "binary"] +) +def test_job_transport_preserves_raw_declared_bytes( + transport: TransportPair, tmp_path: Path, payload: bytes +) -> None: + do, ds, do_sync, ds_sync = transport + source = tmp_path / "source" + source.mkdir() + code = b"# caf\xc3\xa9\r\nprint('approved')\r\n" + (source / "main.py").write_bytes(code) + (source / "payload.bin").write_bytes(payload) + inbox = ds.submit_python_job(do_sync.email, str(source), "transport.job") + [ref] = list(ds.manager.iter_submission_refs(do_sync.email)) + declared = ds.manager.read_submission(ref).headers[SUBMISSION_HASH_HEADER] + + ds_sync.push_job_files(inbox) + do_sync.sync() + [job] = do.jobs + + assert job.status == "pending" + assert (job.job_submission_path / "code/main.py").read_bytes() == code + assert (job.job_submission_path / "code/payload.bin").read_bytes() == payload + assert code_hash(job.job_submission_path) == declared + assert (inbox / "code/main.py").read_bytes() == code + + +def test_job_transport_refuses_newline_change_after_declaration( + transport: TransportPair, tmp_path: Path +) -> None: + do, ds, do_sync, ds_sync = transport + source = tmp_path / "main.py" + source.write_bytes(b"print('approved')\n") + inbox = ds.submit_python_job(do_sync.email, str(source), "changed.job") + changed = inbox / "code/main.py" + changed.write_bytes(changed.read_bytes().replace(b"\n", b"\r\n")) + + ds_sync.push_job_files(inbox) + do_sync.sync() + + [job] = do.jobs + assert job.status == "received"