test: cover raw bytes across the job approval lifecycle

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)

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"