From dcfd374c00828fa69eecf5f2e50707da7d27d6de Mon Sep 17 00:00:00 2001 From: Pedro Werneck Date: Mon, 14 Sep 2026 21:30:29 -0300 Subject: [PATCH 1/3] Improve log and traceback frames visibility for enclave introspection - Updated StderrViewer and StdoutViewer to use artifact_path for accessing stderr and stdout files. - Introduced staging directory for job outputs, separating them from review directory. - Enhanced SyftJobRunner to manage stdout and stderr in staging, with options for sharing logs with submitters. - Implemented traceback capture functionality to sanitize and record job errors securely. - Added tests to verify the behavior of staged logs and their release process. - Adjusted job processing methods to ensure logs are only shared when explicitly released. --- .../src/syft_bg/notify/handlers/job.py | 2 +- .../syft-enclave/src/syft_enclaves/client.py | 230 +++++++++- .../src/syft_enclaves/enclave_job_client.py | 8 + .../src/syft_enclaves/enclave_job_info.py | 101 ++++- .../syft-enclave/tests/test_disclosures.py | 414 ++++++++++++++++++ .../syft-job/src/syft_job/_trace_runner.py | 67 +++ packages/syft-job/src/syft_job/client.py | 12 +- packages/syft-job/src/syft_job/config.py | 29 ++ packages/syft-job/src/syft_job/job.py | 62 ++- packages/syft-job/src/syft_job/job_repr.py | 4 +- packages/syft-job/src/syft_job/job_runner.py | 93 +++- packages/syft-job/src/syft_job/job_stdout.py | 4 +- packages/syft-job/src/syft_job/job_storage.py | 3 + .../src/syft_job/protocolcodecs/base.py | 3 + .../src/syft_job/protocolcodecs/v0.py | 5 + .../src/syft_job/protocolcodecs/v1.py | 5 + .../src/syft_job/traceback_capture.py | 248 +++++++++++ packages/syft-job/tests/test_job_flow.py | 32 +- packages/syft-job/tests/test_staging.py | 172 ++++++++ .../syft-job/tests/test_traceback_capture.py | 278 ++++++++++++ packages/syft-rds/src/syft_rds/client.py | 2 +- tests/unit/test_truncated_logs.py | 5 +- 22 files changed, 1714 insertions(+), 65 deletions(-) create mode 100644 packages/syft-enclave/tests/test_disclosures.py create mode 100644 packages/syft-job/src/syft_job/_trace_runner.py create mode 100644 packages/syft-job/src/syft_job/traceback_capture.py create mode 100644 packages/syft-job/tests/test_staging.py create mode 100644 packages/syft-job/tests/test_traceback_capture.py diff --git a/packages/syft-bg/src/syft_bg/notify/handlers/job.py b/packages/syft-bg/src/syft_bg/notify/handlers/job.py index a7358766b2b..a65f39281ed 100644 --- a/packages/syft-bg/src/syft_bg/notify/handlers/job.py +++ b/packages/syft-bg/src/syft_bg/notify/handlers/job.py @@ -69,7 +69,7 @@ def _read_job_stderr( return None, None stderr_text = None - stderr_file = job.job_review_path / "stderr.txt" + stderr_file = job.artifact_path("stderr.txt") if stderr_file.exists(): try: file_size = stderr_file.stat().st_size diff --git a/packages/syft-enclave/src/syft_enclaves/client.py b/packages/syft-enclave/src/syft_enclaves/client.py index 8dce7d1e4ff..ba7dddd90e6 100644 --- a/packages/syft-enclave/src/syft_enclaves/client.py +++ b/packages/syft-enclave/src/syft_enclaves/client.py @@ -1,6 +1,8 @@ from datetime import datetime, timezone from pathlib import Path -from typing import Optional +from typing import Iterable, Optional +import hashlib +import json import os from syft_rds import SyftRDSClient, SyftRDSClientConfig @@ -12,9 +14,13 @@ from syft_job.job_storage import JobRef from syft_job.models import JobState, JobStatus +from syft_job.traceback_capture import FRAMES_FILENAME, write_no_failure_record + from syft_enclaves.enclave_job_info import ( + DisclosureItem, EnclaveJobInfo, PartyApprovalStatus, + approved_disclosures, enclave_approval_file_name, ) from syft_enclaves.attestation import ( @@ -37,6 +43,22 @@ ) +# The artifacts that move only under the agreement of every party. Everything +# else in the review folder stays readable by the submitter. +FORWARDED_RECORD = "forwarded_artifacts.json" +RESULTS_SHARED_MARKER = "results_shared" + +GATED_ARTIFACTS = { + DisclosureItem.LOGS.value: ("stdout.txt", "stderr.txt"), + DisclosureItem.TRACEBACK_FRAMES.value: (FRAMES_FILENAME,), +} + + +def pre_sync_enabled() -> bool: + """Whether the client syncs on its own. ``PRE_SYNC=false`` turns it off.""" + return os.environ.get("PRE_SYNC", "true").lower() == "true" + + class SyftEnclaveClient: def __init__( self, @@ -162,6 +184,7 @@ def submit_python_job( job_name: Optional[str] = "", datasets: Optional[dict[str, list[str]]] = None, share_results_with_do: bool = False, + request_disclosures: Optional[list[str]] = None, force_submission: bool = False, ignore_peer_version: bool = False, **kwargs, @@ -187,6 +210,7 @@ def submit_python_job( job_name, datasets=datasets, share_results_with_do=share_results_with_do, + request_disclosures=request_disclosures, **kwargs, ) self._rds.sync_engine.push_job_files(job_dir) @@ -203,36 +227,173 @@ def run_jobs(self) -> None: state.status = JobStatus.APPROVED state.save(job.job_review_path / "state.yaml") + # share_logs_with_submitter=False keeps the logs in staging. Only a + # grant from every party releases them. self._rds.process_approved_jobs( force_execution=True, share_outputs_with_submitter=True, - share_logs_with_submitter=True, + share_logs_with_submitter=False, ) + self._apply_disclosure_policy() + + def _requested_disclosures(self, job: JobInfo) -> list[str]: + """The items the submitter asked for, as recorded at submission.""" + requested = job.job_headers.get("requested_disclosures") + return list(requested) if isinstance(requested, list) else [] + + def granted_disclosures(self, job: JobInfo) -> set[str]: + """The items that every party released for this job.""" + return approved_disclosures( + job.job_review_path, self._requested_disclosures(job) + ) + + def _local_jobs(self) -> JobsList: + """The job list read from disk, with no sync. + + ``self.jobs`` syncs first, therefore a caller that must act before a + push reads the list through this method. + """ + return self._rds.job_client.jobs + + def _apply_disclosure_policy(self) -> None: + """Move each granted artifact from the staging folder into review. + + An artifact that no party released stays in the staging folder, where no + submitter holds a grant. + """ + for job in self._local_jobs(): + if job.status not in ("done", "failed"): + continue + granted = self.granted_disclosures(job) + released = [ + name + for item, names in GATED_ARTIFACTS.items() + if item in granted + for name in names + ] + job.release_artifacts(released) def distribute_results(self) -> None: """Distribute job results to DS (always) and optionally to DOs.""" for job in self.jobs: - if job.status != "done": + if job.status not in ("done", "failed"): continue - results_shared_marker = job.job_review_path / "results_shared" - if results_shared_marker.exists(): + if self._results_already_shared(job): continue # Always share results with the DS (submitter) self._forward_results_to_recipients(job, [job.submitted_by]) # Optionally share with DOs - if job.job_headers.get("share_results_with_do"): - datasets = job.job_metadata.datasets - if datasets: - do_emails = list(datasets.keys()) - job.share_outputs(do_emails) - self._forward_results_to_recipients(job, do_emails) + datasets = job.job_metadata.datasets + if job.job_headers.get("share_results_with_do") and datasets: + do_emails = list(datasets.keys()) + job.share_outputs(do_emails) + self._forward_results_to_recipients(job, do_emails) - results_shared_marker.write_text("shared") + self._mark_results_shared(job) + + # A grant can arrive after the results went out, so released artifacts + # travel on their own schedule. + for job in self.jobs: + if job.status in ("done", "failed"): + self._forward_new_releases(job) self._rds.sync() + def _state_digest(self, job: JobInfo) -> str: + """A digest of the job state, which changes with every run.""" + state_file = Path(job.job_review_path) / "state.yaml" + if not state_file.is_file(): + return "" + return hashlib.sha256(state_file.read_bytes()).hexdigest() + + def _results_already_shared(self, job: JobInfo) -> bool: + """Whether the results of the current run already went out. + + A rerun writes a new state, so the marker names the run it covers. A + marker that names an earlier run does not stop the new results. + """ + marker = Path(job.job_review_path) / RESULTS_SHARED_MARKER + if not marker.exists(): + return False + try: + recorded = json.loads(marker.read_text()) + except (OSError, json.JSONDecodeError): + recorded = None + if not isinstance(recorded, dict): + # A marker from an older client holds plain text. Record the + # current run and keep the results as sent. + self._mark_results_shared(job) + return True + return recorded.get("state") == self._state_digest(job) + + def _mark_results_shared(self, job: JobInfo) -> None: + marker = Path(job.job_review_path) / RESULTS_SHARED_MARKER + marker.write_text(json.dumps({"state": self._state_digest(job)})) + + def _forward_new_releases(self, job: JobInfo) -> list[str]: + """Send each released artifact that the parties do not hold yet. + + A party that releases an item also receives it, so the data owners get + the same file as the submitter. + + The record holds a digest for each name, not the name alone. A rerun + writes a new record under the same name, and a digest tells the two + apart, so the parties never keep the artifact of an earlier run. + """ + review_dir = Path(job.job_review_path) + sent_path = review_dir / FORWARDED_RECORD + try: + sent = json.loads(sent_path.read_text()) + except (OSError, json.JSONDecodeError): + sent = {} + if not isinstance(sent, dict): + sent = {} + + granted = self.granted_disclosures(job) + + # A later run that did not fail leaves no record, so the parties would + # keep the crash of an earlier run. Replace it instead. + if ( + DisclosureItem.TRACEBACK_FRAMES.value in granted + and FRAMES_FILENAME in sent + and not (review_dir / FRAMES_FILENAME).is_file() + ): + write_no_failure_record(review_dir) + + released = [ + name + for item, names in GATED_ARTIFACTS.items() + if item in granted + for name in names + if (review_dir / name).is_file() + ] + + pending = {} + for name in released: + content = (review_dir / name).read_bytes() + digest = hashlib.sha256(content).hexdigest() + if sent.get(name) != digest: + pending[name] = (content, digest) + if not pending: + return [] + + datasite_dir = self._rds.syftbox_folder / self._rds.email + files = { + (review_dir / name).relative_to(datasite_dir): content + for name, (content, _) in pending.items() + } + datasets = job.job_metadata.datasets or {} + recipients = list( + dict.fromkeys([job.submitted_by, *(e for e in datasets if e != self.email)]) + ) + self._push_files_to_recipients(files, recipients) + + sent.update({name: digest for name, (_, digest) in pending.items()}) + sent_path.write_text(json.dumps(sent, sort_keys=True)) + return sorted(pending) + def _read_state_file(self, job: JobInfo) -> dict[Path, bytes]: """Read the job state.yaml as a {path_in_datasite: bytes} dict.""" state_file = job.job_review_path / "state.yaml" @@ -245,10 +406,16 @@ def _read_state_file(self, job: JobInfo) -> dict[Path, bytes]: def _forward_results_to_recipients(self, job: JobInfo, recipients: list[str]): """Forward job output files and state to recipients via event outbox.""" outputs_dir = job.job_review_path / "outputs" - if not outputs_dir.exists(): - return - files_by_datasite_path = self._get_files_in_dir(outputs_dir) + files_by_datasite_path = ( + self._get_files_in_dir(outputs_dir) if outputs_dir.exists() else {} + ) files_by_datasite_path.update(self._read_state_file(job)) + self._push_files_to_recipients(files_by_datasite_path, recipients) + + def _push_files_to_recipients( + self, files_by_datasite_path: dict, recipients: list[str] + ) -> None: + """Queue the files for the recipients through the event outbox.""" if not files_by_datasite_path: return syncer = self._rds.sync_engine.datasite_owner_syncer @@ -261,12 +428,19 @@ def _forward_results_to_recipients(self, job: JobInfo, recipients: list[str]): ) syncer.process_syftbox_events_queue() - def approve_job(self, job: JobInfo) -> None: - """Approve an enclave job and push the approval state file to the enclave.""" - if os.environ.get("PRE_SYNC", "true").lower() == "true": + def approve_job( + self, job: JobInfo, disclosures: Optional[Iterable[str]] = None + ) -> None: + """Approve an enclave job and push the approval state file to the enclave. + + ``disclosures`` names the items in ``DisclosureItem`` that this data + owner releases. The enclave releases an item only when every data owner + released it. Omit the argument to release nothing. + """ + if pre_sync_enabled(): self._rds.sync() - job.approve() + job.approve(disclosures) file_name = enclave_approval_file_name(self.email) approval_file = job.job_review_path / file_name if not approval_file.exists(): @@ -278,6 +452,24 @@ def approve_job(self, job: JobInfo) -> None: relative_path, process_now=True ) + def update_disclosures( + self, job: JobInfo, disclosures: Optional[Iterable[str]] + ) -> dict: + """Change the items this data owner releases, and push the new set. + + The enclave applies the new set on its next cycle. + """ + if pre_sync_enabled(): + self._rds.sync() + + result = job.update_disclosures(disclosures) + approval_file = job.job_review_path / enclave_approval_file_name(self.email) + relative_path = approval_file.relative_to(self._rds.syftbox_folder) + self._rds.sync_engine.datasite_watcher_syncer.on_file_change( + relative_path, process_now=True + ) + return result + def receive_jobs(self): """Receive and distribute enclave jobs to relevant DOs. diff --git a/packages/syft-enclave/src/syft_enclaves/enclave_job_client.py b/packages/syft-enclave/src/syft_enclaves/enclave_job_client.py index 342bbada5a9..5fe54d91c24 100644 --- a/packages/syft-enclave/src/syft_enclaves/enclave_job_client.py +++ b/packages/syft-enclave/src/syft_enclaves/enclave_job_client.py @@ -5,6 +5,8 @@ from syft_job.job import JobsList from syft_job.models import JobSubmissionMetadata +from syft_enclaves.enclave_job_info import normalize_disclosures + class EnclaveJobClient(BaseJobClient): """Wraps a JobClient to add enclave-specific job submission behavior. @@ -52,6 +54,7 @@ def submit_python_job( job_name: Optional[str] = "", datasets: Optional[dict[str, list[str]]] = None, share_results_with_do: bool = False, + request_disclosures: Optional[list[str]] = None, **kwargs, ) -> Path: """Submit a Python job with enclave metadata. @@ -69,6 +72,11 @@ def submit_python_job( config.headers = { "job_type": "enclave", "share_results_with_do": share_results_with_do, + # The items the submitter asks for. Each data owner sees this list + # next to the code, then releases none, some, or all of it. + "requested_disclosures": sorted( + normalize_disclosures(request_disclosures) + ), } config.save(job_dir / "config.yaml") diff --git a/packages/syft-enclave/src/syft_enclaves/enclave_job_info.py b/packages/syft-enclave/src/syft_enclaves/enclave_job_info.py index 287487a8e84..b96b37c035f 100644 --- a/packages/syft-enclave/src/syft_enclaves/enclave_job_info.py +++ b/packages/syft-enclave/src/syft_enclaves/enclave_job_info.py @@ -2,14 +2,46 @@ import json from datetime import datetime, timezone +from enum import Enum from pathlib import Path -from typing import Optional +from typing import Iterable, Optional, Union -from pydantic import BaseModel +from pydantic import BaseModel, Field from syft_job.job import JobInfo from syft_job.models import JobStatus +class DisclosureItem(str, Enum): + """A class of job data that a party can release to the other parties. + + - ``TRACEBACK_FRAMES``: the failure position in the approved code, and the + builtin exception type. Bounded. + - ``LOGS``: stdout and stderr. Unbounded, and the job chooses every byte. + """ + + TRACEBACK_FRAMES = "traceback_frames" + LOGS = "logs" + + +DISCLOSURE_ITEMS = frozenset(item.value for item in DisclosureItem) + + +def normalize_disclosures( + items: Union[str, DisclosureItem, Iterable[str], None], +) -> dict[str, bool]: + """Return the known items in ``items`` as a map. Unknown names are dropped. + + A single name is accepted on its own, because iterating a string would + produce its characters and grant nothing. + """ + if not items: + return {} + if isinstance(items, (str, DisclosureItem)): + items = [items] + names = {str(getattr(i, "value", i)) for i in items} + return {name: True for name in sorted(names & DISCLOSURE_ITEMS)} + + class PartyApprovalStatus(BaseModel): """Tracks approval from a single party in a multi-party (enclave) job.""" @@ -17,6 +49,9 @@ class PartyApprovalStatus(BaseModel): dataset: Optional[str] = None status: JobStatus = JobStatus.PENDING approved_at: Optional[datetime] = None + # The items this party releases. Absent means the party released nothing, + # so an approval file from an older client grants nothing. + disclosures: dict[str, bool] = Field(default_factory=dict) def save_json(self, path: Path) -> None: path.parent.mkdir(parents=True, exist_ok=True) @@ -66,8 +101,13 @@ def status(self) -> str: return JobStatus.APPROVED.value return JobStatus.PENDING.value - def approve(self) -> None: - """Write approval to the DO's individual approval state file.""" + def approve(self, disclosures: Optional[Iterable[str]] = None) -> None: + """Write approval to the DO's individual approval state file. + + ``disclosures`` names the items in ``DisclosureItem`` that this party + releases to the other parties. The enclave releases an item only when + every party released it. Omit the argument to release nothing. + """ file_name = enclave_approval_file_name(self.current_user_email) approval_file = self.job_review_path / file_name if not approval_file.exists(): @@ -80,5 +120,58 @@ def approve(self) -> None: raise ValueError(f"Already in status: {approval.status.value}") approval.status = JobStatus.APPROVED approval.approved_at = datetime.now(timezone.utc) + approval.disclosures = normalize_disclosures(disclosures) approval.save_json(approval_file) print(f"Job '{self.name}' approved by {self.current_user_email}!") + + def update_disclosures(self, disclosures: Optional[Iterable[str]]) -> dict: + """Replace the items this party releases, and return the new map. + + A copy that already reached another party does not come back. + """ + file_name = enclave_approval_file_name(self.current_user_email) + approval_file = self.job_review_path / file_name + if not approval_file.exists(): + raise PermissionError( + f"No approval file found for {self.current_user_email}. " + f"You may not be a designated party for this job." + ) + approval = PartyApprovalStatus.load_json(approval_file) + if approval.status != JobStatus.APPROVED: + raise ValueError( + f"Approve the job first (current: {approval.status.value})." + ) + approval.disclosures = normalize_disclosures(disclosures) + approval.save_json(approval_file) + return approval.disclosures + + +def approved_disclosures( + review_dir: Path, requested: Optional[Iterable[str]] = None +) -> set[str]: + """Return the items that every party released. + + An item needs the agreement of all parties, therefore the result is the + intersection of the parties' grants. The result is empty if a party did not + approve the job yet, because a party grants nothing before it approves. + When ``requested`` is given, the result holds only the requested items. + """ + approvals = load_enclave_approval_files(review_dir) + if not approvals: + return set() + + granted: Optional[set[str]] = None + for approval in approvals: + if approval.status != JobStatus.APPROVED: + return set() + party_grant = { + name + for name, allowed in approval.disclosures.items() + if allowed and name in DISCLOSURE_ITEMS + } + granted = party_grant if granted is None else granted & party_grant + + result = granted or set() + if requested is not None: + result = result & set(normalize_disclosures(requested)) + return result diff --git a/packages/syft-enclave/tests/test_disclosures.py b/packages/syft-enclave/tests/test_disclosures.py new file mode 100644 index 00000000000..d737eb0fec3 --- /dev/null +++ b/packages/syft-enclave/tests/test_disclosures.py @@ -0,0 +1,414 @@ +"""Per-item disclosure: the enclave releases an item only under full agreement.""" + +import json +import os +import random +import shutil +import tempfile +from pathlib import Path + +import pytest + +os.environ["PRE_SYNC"] = "false" + +from syft_job.models import JobStatus # noqa: E402 +from syft_job.traceback_capture import FRAMES_FILENAME # noqa: E402 + +from syft_enclaves import SyftEnclaveClient # noqa: E402 +from syft_enclaves.enclave_job_info import ( # noqa: E402 + DisclosureItem, + PartyApprovalStatus, + approved_disclosures, + enclave_approval_file_name, + normalize_disclosures, +) + +LOGS = DisclosureItem.LOGS.value +FRAMES = DisclosureItem.TRACEBACK_FRAMES.value + + +def write_approval(review_dir, party, status, disclosures): + approval = PartyApprovalStatus( + party=party, status=status, disclosures=disclosures or {} + ) + approval.save_json(review_dir / enclave_approval_file_name(party)) + + +# -- the resolution rule -------------------------------------------------------- + + +def test_an_item_needs_every_party(tmp_path): + write_approval(tmp_path, "do1@x.com", JobStatus.APPROVED, {LOGS: True}) + write_approval(tmp_path, "do2@x.com", JobStatus.APPROVED, {LOGS: True}) + assert approved_disclosures(tmp_path) == {LOGS} + + +def test_one_party_withholding_blocks_the_item(tmp_path): + write_approval(tmp_path, "do1@x.com", JobStatus.APPROVED, {LOGS: True}) + write_approval(tmp_path, "do2@x.com", JobStatus.APPROVED, {FRAMES: True}) + assert approved_disclosures(tmp_path) == set() + + +def test_a_party_that_has_not_approved_blocks_everything(tmp_path): + write_approval(tmp_path, "do1@x.com", JobStatus.APPROVED, {LOGS: True}) + write_approval(tmp_path, "do2@x.com", JobStatus.PENDING, {LOGS: True}) + assert approved_disclosures(tmp_path) == set() + + +def test_an_approval_file_without_the_field_grants_nothing(tmp_path): + """An older client writes no disclosures key, so it releases nothing.""" + path = tmp_path / enclave_approval_file_name("do1@x.com") + path.write_text(json.dumps({"party": "do1@x.com", "status": "approved"})) + assert PartyApprovalStatus.load_json(path).disclosures == {} + assert approved_disclosures(tmp_path) == set() + + +def test_the_requested_set_narrows_the_result(tmp_path): + write_approval(tmp_path, "do1@x.com", JobStatus.APPROVED, {LOGS: True, FRAMES: True}) + assert approved_disclosures(tmp_path, [FRAMES]) == {FRAMES} + + +def test_no_approval_file_grants_nothing(tmp_path): + assert approved_disclosures(tmp_path) == set() + + +def test_an_unknown_item_never_survives(tmp_path): + write_approval(tmp_path, "do1@x.com", JobStatus.APPROVED, {"everything": True}) + assert approved_disclosures(tmp_path) == set() + assert normalize_disclosures(["everything", LOGS]) == {LOGS: True} + + +# -- end to end ----------------------------------------------------------------- + + +def build_quad(): + enclave, do1, do2, ds = SyftEnclaveClient.quad_with_mock_drive_service_connection( + use_in_memory_cache=False + ) + for do, name in ((do1, "dataset1"), (do2, "dataset2")): + d = Path(tempfile.mkdtemp()) / f"d{random.randint(1, 10**6)}" + d.mkdir(parents=True) + (d / "m.txt").write_text("m") + (d / "p.txt").write_text("p") + do.create_dataset( + name=name, + mock_path=d / "m.txt", + private_path=d / "p.txt", + summary=name, + users=[ds.email, enclave.email], + upload_private=True, + sync=False, + ) + do.share_private_dataset(name, enclave.email) + do.sync() + ds.sync() + return enclave, do1, do2, ds + + +def submit(ds, enclave, do1, do2, code, requested): + path = Path(tempfile.mkdtemp()) / "main.py" + path.write_text(code) + ds.submit_python_job( + enclave.email, + str(path), + "j", + datasets={do1.email: ["dataset1"], do2.email: ["dataset2"]}, + request_disclosures=requested, + ) + + +OK_CODE = "import os, json\nos.makedirs('outputs', exist_ok=True)\nopen('outputs/r.json','w').write('{}')\n" +CRASH_CODE = "x = 1\ny = 2\nraise ValueError('patient 4171 is positive')\n" + + +def run_to_completion(enclave, do1, do2, ds, grants): + enclave.sync() + enclave.receive_jobs() + do1.sync() + do2.sync() + do1.approve_job(do1.jobs["j"], grants) + do2.approve_job(do2.jobs["j"], grants) + enclave.sync() + enclave.run_jobs() + enclave.distribute_results() + ds.sync() + + +def test_without_a_grant_the_submitter_gets_no_logs(): + enclave, do1, do2, ds = build_quad() + submit(ds, enclave, do1, do2, OK_CODE, [LOGS]) + run_to_completion(enclave, do1, do2, ds, None) + + review = Path(ds.jobs["j"].job_review_path) + assert not (review / "stdout.txt").exists() + assert not (review / "stderr.txt").exists() + + +def test_a_full_grant_sends_the_logs_to_the_submitter(): + enclave, do1, do2, ds = build_quad() + submit(ds, enclave, do1, do2, OK_CODE, [LOGS]) + run_to_completion(enclave, do1, do2, ds, [LOGS]) + + assert enclave.granted_disclosures(enclave.jobs["j"]) == {LOGS} + assert (Path(ds.jobs["j"].job_review_path) / "stdout.txt").exists() + + +def test_granted_frames_carry_the_position_but_not_the_message(): + enclave, do1, do2, ds = build_quad() + submit(ds, enclave, do1, do2, CRASH_CODE, [FRAMES]) + run_to_completion(enclave, do1, do2, ds, [FRAMES]) + + record = json.loads( + (Path(ds.jobs["j"].job_review_path) / FRAMES_FILENAME).read_text() + ) + entry = record["chain"][0] + assert entry["type"] == "ValueError" + assert {"file": "main.py", "line": 3} in entry["frames"] + assert "positive" not in json.dumps(record) + + +def test_frames_reach_the_data_owners_too(): + """A party that releases an item also receives it.""" + enclave, do1, do2, ds = build_quad() + submit(ds, enclave, do1, do2, CRASH_CODE, [FRAMES]) + run_to_completion(enclave, do1, do2, ds, [FRAMES]) + do1.sync() + + assert (Path(do1.jobs["j"].job_review_path) / FRAMES_FILENAME).exists() + + +def test_one_owner_withholding_blocks_the_frames(): + enclave, do1, do2, ds = build_quad() + submit(ds, enclave, do1, do2, CRASH_CODE, [FRAMES]) + enclave.sync() + enclave.receive_jobs() + do1.sync() + do2.sync() + do1.approve_job(do1.jobs["j"], [FRAMES]) + do2.approve_job(do2.jobs["j"], None) + enclave.sync() + enclave.run_jobs() + enclave.distribute_results() + ds.sync() + + assert not (Path(ds.jobs["j"].job_review_path) / FRAMES_FILENAME).exists() + + +def test_a_later_grant_releases_a_withheld_artifact(): + """A party can release an item after the run, without a new submission.""" + enclave, do1, do2, ds = build_quad() + submit(ds, enclave, do1, do2, OK_CODE, [LOGS]) + run_to_completion(enclave, do1, do2, ds, None) + assert not (Path(ds.jobs["j"].job_review_path) / "stdout.txt").exists() + + # Both owners amend their approval, then the enclave applies the grants. + for do in (do1, do2): + assert do.update_disclosures(do.jobs["j"], [LOGS]) == {LOGS: True} + enclave.sync() + enclave.run_jobs() + enclave.distribute_results() + ds.sync() + + assert (Path(ds.jobs["j"].job_review_path) / "stdout.txt").exists() + + +def test_no_sync_runs_while_an_ungranted_artifact_sits_in_review(monkeypatch): + """An ungranted artifact is never readable, whenever a sync runs. + + The job runner writes the logs into staging, so no ordering rule protects + them. This test guards that property: it fails if a later change writes a + gated artifact into the review folder before a release. + """ + enclave, do1, do2, ds = build_quad() + submit(ds, enclave, do1, do2, OK_CODE, [LOGS]) + enclave.sync() + enclave.receive_jobs() + do1.sync() + do2.sync() + do1.approve_job(do1.jobs["j"], None) + do2.approve_job(do2.jobs["j"], None) + enclave.sync() + + # Read the review path without a sync, so the spy below cannot recurse. + review = Path(enclave._rds.job_client.jobs["j"].job_review_path) + exposed = [] + engine = enclave._rds.sync_engine + real_sync = type(engine).sync + + def spy(self, *args, **kwargs): + if self is engine: + exposed.append( + sorted( + name + for name in ("stdout.txt", "stderr.txt") + if (review / name).exists() + ) + ) + return real_sync(self, *args, **kwargs) + + monkeypatch.setenv("PRE_SYNC", "true") + monkeypatch.setattr(type(engine), "sync", spy) + enclave.run_jobs() + + assert not any(exposed), f"ungranted logs were readable during a sync: {exposed}" + + +def test_an_amendment_needs_an_approval_first(): + enclave, do1, do2, ds = build_quad() + submit(ds, enclave, do1, do2, OK_CODE, [LOGS]) + enclave.sync() + enclave.receive_jobs() + do1.sync() + + with pytest.raises(ValueError, match="Approve the job first"): + do1.update_disclosures(do1.jobs["j"], [LOGS]) + + +def test_a_late_grant_reaches_the_data_owners_too(): + """distribute_results runs once, so a later release needs its own path.""" + enclave, do1, do2, ds = build_quad() + submit(ds, enclave, do1, do2, CRASH_CODE, [FRAMES]) + run_to_completion(enclave, do1, do2, ds, None) + do1.sync() + assert not (Path(do1.jobs["j"].job_review_path) / FRAMES_FILENAME).exists() + + for do in (do1, do2): + do.update_disclosures(do.jobs["j"], [FRAMES]) + enclave.sync() + enclave.run_jobs() + enclave.distribute_results() + ds.sync() + do1.sync() + do2.sync() + + assert (Path(ds.jobs["j"].job_review_path) / FRAMES_FILENAME).exists() + assert (Path(do1.jobs["j"].job_review_path) / FRAMES_FILENAME).exists() + assert (Path(do2.jobs["j"].job_review_path) / FRAMES_FILENAME).exists() + + +def test_a_released_artifact_is_not_sent_twice(): + enclave, do1, do2, ds = build_quad() + submit(ds, enclave, do1, do2, CRASH_CODE, [FRAMES]) + run_to_completion(enclave, do1, do2, ds, [FRAMES]) + + job = enclave.jobs["j"] + assert enclave._forward_new_releases(job) == [] + + +def test_a_changed_record_reaches_the_parties_again(): + """The forwarded record holds a digest, so new content goes out again.""" + enclave, do1, do2, ds = build_quad() + submit(ds, enclave, do1, do2, CRASH_CODE, [FRAMES]) + run_to_completion(enclave, do1, do2, ds, [FRAMES]) + + review = Path(enclave.jobs["j"].job_review_path) + first = json.loads((review / FRAMES_FILENAME).read_text()) + assert enclave._forward_new_releases(enclave.jobs["j"]) == [] + + # A later run writes a different record under the same name. + (review / FRAMES_FILENAME).write_text( + json.dumps({"chain": [{"type": "KeyError", "frames": []}]}) + ) + assert enclave._forward_new_releases(enclave.jobs["j"]) == [FRAMES_FILENAME] + + do1.sync() + delivered = json.loads( + (Path(do1.jobs["j"].job_review_path) / FRAMES_FILENAME).read_text() + ) + assert delivered["chain"][0]["type"] == "KeyError" + assert delivered != first + + +def test_a_new_run_is_not_skipped_as_already_shared(): + """distribute_results marks a failed job, so a later run must not be skipped. + + The marker names the run it covers. An end-to-end rerun cannot drive this: + `rerun()` leaves code/.venv and `uv venv` then fails, for every python job. + """ + enclave, do1, do2, ds = build_quad() + submit(ds, enclave, do1, do2, CRASH_CODE, [FRAMES]) + run_to_completion(enclave, do1, do2, ds, [FRAMES]) + + job = enclave.jobs["j"] + assert job.status == "failed" + assert enclave._results_already_shared(job) + + # A later run writes a new state, which is what rerun() prepares for. + state_file = Path(job.job_review_path) / "state.yaml" + state_file.write_text(state_file.read_text() + "\n# next run\n") + assert not enclave._results_already_shared(job) + + +def test_a_marker_from_an_older_client_still_counts_as_shared(): + enclave, do1, do2, ds = build_quad() + submit(ds, enclave, do1, do2, OK_CODE, [LOGS]) + run_to_completion(enclave, do1, do2, ds, [LOGS]) + + job = enclave.jobs["j"] + marker = Path(job.job_review_path) / "results_shared" + marker.write_text("shared") + + assert enclave._results_already_shared(job) + # The marker now names the current run, so the next run redistributes. + assert json.loads(marker.read_text())["state"] == enclave._state_digest(job) + + +# A job that fails once, then succeeds, with no edit between the runs. +FAIL_THEN_PASS_CODE = """\ +import json, os + +flag = {flag!r} +if not os.path.exists(flag): + open(flag, "w").write("x") + raise ValueError("patient 4171 is positive") + +os.makedirs("outputs", exist_ok=True) +open("outputs/r.json", "w").write(json.dumps({{"ok": 1}})) +""" + + +def rerun_on_the_enclave(enclave): + """Prepare a rerun, and clear what blocks one. + + `rerun()` leaves code/.venv, and `run.sh` then runs `uv venv`, which stops + under `set -euo pipefail`. That defect predates the staging work and no + caller hits it, so the test clears the directory itself. + """ + job = enclave.jobs["j"] + job.rerun() + shutil.rmtree(Path(job.job_submission_path) / "code" / ".venv", ignore_errors=True) + + +def test_a_run_that_does_not_fail_replaces_the_crash_record(): + enclave, do1, do2, ds = build_quad() + flag = str(Path(tempfile.mkdtemp()) / "once") + submit(ds, enclave, do1, do2, FAIL_THEN_PASS_CODE.format(flag=flag), [FRAMES]) + run_to_completion(enclave, do1, do2, ds, [FRAMES]) + do1.sync() + + assert enclave.jobs["j"].status == "failed" + do1_record = Path(do1.jobs["j"].job_review_path) / FRAMES_FILENAME + assert json.loads(do1_record.read_text())["chain"][0]["type"] == "ValueError" + + rerun_on_the_enclave(enclave) + enclave.run_jobs() + enclave.distribute_results() + ds.sync() + do1.sync() + + assert enclave.jobs["j"].status == "done" + # No party keeps a crash description for a run that did not fail. + for holder in (ds, do1): + record = Path(holder.jobs["j"].job_review_path) / FRAMES_FILENAME + assert json.loads(record.read_text()) == {"chain": []} + + +def test_no_replacement_when_the_parties_hold_no_record(): + """A job that never failed sends nothing, so there is nothing to replace.""" + enclave, do1, do2, ds = build_quad() + submit(ds, enclave, do1, do2, OK_CODE, [FRAMES]) + run_to_completion(enclave, do1, do2, ds, [FRAMES]) + + job = enclave.jobs["j"] + assert not (Path(job.job_review_path) / FRAMES_FILENAME).exists() + assert enclave._forward_new_releases(job) == [] diff --git a/packages/syft-job/src/syft_job/_trace_runner.py b/packages/syft-job/src/syft_job/_trace_runner.py new file mode 100644 index 00000000000..c3c85231959 --- /dev/null +++ b/packages/syft-job/src/syft_job/_trace_runner.py @@ -0,0 +1,67 @@ +"""Run a job entrypoint and record an uncaught exception as plain data. + +This file runs inside the job process. It imports the standard library only, +because the job virtual environment does not hold ``syft_job``. + +The record is untrusted. The job controls this process, so the job can write any +string into the record or replace the file. ``syft_job.traceback_capture`` +checks every field before a party reads it. +""" + +import json +import os +import runpy +import sys +import traceback + +RAW_TRACE_PATH_ENV = "SYFT_JOB_RAW_TRACE_PATH" +MAX_CHAIN = 5 +MAX_FRAMES = 64 + + +def raw_record(exc: BaseException) -> dict: + """Return the exception chain as plain data. + + The record holds the class names and the frame positions. It never holds the + exception message, the source text, or the local variables. + """ + chain = [] + seen = set() + while exc is not None and id(exc) not in seen and len(chain) < MAX_CHAIN: + seen.add(id(exc)) + frames = traceback.extract_tb(exc.__traceback__)[:MAX_FRAMES] + chain.append( + { + "mro": [cls.__name__ for cls in type(exc).__mro__], + "frames": [ + {"filename": f.filename, "lineno": f.lineno} for f in frames + ], + } + ) + exc = exc.__cause__ or exc.__context__ + return {"chain": chain} + + +def main(argv: list) -> None: + entrypoint = argv[1] + sys.argv = argv[1:] + try: + runpy.run_path(entrypoint, run_name="__main__") + except SystemExit: + # A deliberate exit carries no failure position. + raise + except BaseException as exc: + # Observe the exception, then let it propagate unchanged. The process + # must keep the exit code and the stderr traceback it would have had. + path = os.environ.get(RAW_TRACE_PATH_ENV) + if path: + try: + with open(path, "w") as f: + json.dump(raw_record(exc), f) + except OSError: + pass + raise + + +if __name__ == "__main__": + main(sys.argv) diff --git a/packages/syft-job/src/syft_job/client.py b/packages/syft-job/src/syft_job/client.py index 1a7d7acaa9d..b7bc1e85e0f 100644 --- a/packages/syft-job/src/syft_job/client.py +++ b/packages/syft-job/src/syft_job/client.py @@ -354,7 +354,11 @@ def _generate_python_run_script( uv sync --python {RUN_SCRIPT_PYTHON_VERSION} {install_deps_cmd} export PYTHONPATH=.:${{PYTHONPATH:-}} -python {entrypoint_path} +if [ -n "${{SYFT_JOB_TRACE_RUNNER:-}}" ]; then + python "$SYFT_JOB_TRACE_RUNNER" {entrypoint_path} +else + python {entrypoint_path} +fi """ else: # entrypoint_path is just the filename (e.g. "main.py") @@ -370,7 +374,11 @@ def _generate_python_run_script( source .venv/bin/activate uv pip install {deps_str} export PYTHONPATH=.:${{PYTHONPATH:-}} -python {entrypoint_path} +if [ -n "${{SYFT_JOB_TRACE_RUNNER:-}}" ]; then + python "$SYFT_JOB_TRACE_RUNNER" {entrypoint_path} +else + python {entrypoint_path} +fi """ def submit_python_job( diff --git a/packages/syft-job/src/syft_job/config.py b/packages/syft-job/src/syft_job/config.py index 4a8ea4b3fea..2a62d20a741 100644 --- a/packages/syft-job/src/syft_job/config.py +++ b/packages/syft-job/src/syft_job/config.py @@ -97,6 +97,18 @@ def get_review_dir(self, datasite_email: str) -> Path: """ return self.get_job_dir(datasite_email) / "review" + def get_staging_dir(self, datasite_email: str) -> Path: + """ + Get the staging directory for artifacts that no party released yet. + + Path: SyftBox//app_data/job/staging/ + + The submitter holds a read grant on the review directory, and a nested + file inherits that grant. An artifact therefore waits here until a + release moves it into the review directory. + """ + return self.get_job_dir(datasite_email) / "staging" + def get_job_submission_dir( self, datasite_email: str, @@ -131,6 +143,23 @@ def get_review_job_dir( segment = protocol_dir_name(protocol_version) return base / segment / job_name if segment else base / job_name + def get_staging_job_dir( + self, + datasite_email: str, + ds_email: str, + job_name: str, + protocol_version: str = JOB_PROTOCOL_VERSION, + ) -> Path: + """ + Get the staging path for a specific job. + + Path: SyftBox//app_data/job/staging//v// + (no v segment for protocol 0) + """ + base = self.get_staging_dir(datasite_email) / ds_email + segment = protocol_dir_name(protocol_version) + return base / segment / job_name if segment else base / job_name + def _get_job_submission_dir_for_me( self, target_datasite_owner_email: str, diff --git a/packages/syft-job/src/syft_job/job.py b/packages/syft-job/src/syft_job/job.py index 0eed2bbf7ad..8fa53145379 100644 --- a/packages/syft-job/src/syft_job/job.py +++ b/packages/syft-job/src/syft_job/job.py @@ -3,7 +3,7 @@ import shutil from datetime import datetime, timezone from pathlib import Path -from typing import TYPE_CHECKING, List, Optional +from typing import TYPE_CHECKING, Iterable, List, Optional from syft_permissions.spec.ruleset import PERMISSION_FILE_NAME @@ -16,6 +16,11 @@ from .job_stdout import StdoutViewer from .job_storage import JobRef from .models import JobState, JobStatus, JobSubmissionMetadata +from .traceback_capture import FRAMES_FILENAME + +# The artifacts that wait in staging until a party releases them. returncode.txt +# stays in review: it holds one integer, which state.yaml already carries. +STAGED_LOG_FILES = ("stdout.txt", "stderr.txt") if TYPE_CHECKING: from .client import JobClient @@ -53,6 +58,48 @@ def job_submission_path(self) -> Path: def job_review_path(self) -> Path: return self._client.manager.review_dir(self._ref) + @property + def job_staging_path(self) -> Path: + """Where an artifact waits until a party releases it.""" + return self._client.manager.staging_dir(self._ref) + + def artifact_path(self, filename: str) -> Path: + """Return where ``filename`` is now: released, or still staged. + + A released artifact sits in the review directory. Every other artifact + sits in the staging directory. The datasite owner reads both, so a + viewer on the owner side finds the file either way. + """ + released = self.job_review_path / filename + if released.exists(): + return released + return self.job_staging_path / filename + + def release_logs(self) -> list[str]: + """Move the staged logs into the review directory, and return the names. + + The submitter reads the review directory, so this move discloses the + logs. A copy that already reached another party does not come back. + """ + return self.release_artifacts(STAGED_LOG_FILES) + + def release_artifacts(self, filenames: Iterable[str]) -> list[str]: + """Move the named staged artifacts into the review directory. + + Returns the names it moved. A name that is absent from staging, or that + a release already moved, is skipped. + """ + moved = [] + for filename in filenames: + staged = self.job_staging_path / filename + if not staged.exists(): + continue + target = self.job_review_path / filename + target.parent.mkdir(parents=True, exist_ok=True) + shutil.move(str(staged), str(target)) + moved.append(filename) + return moved + # ────────────────────────────────────────────── # Properties from config (inbox/) # ────────────────────────────────────────────── @@ -324,12 +371,13 @@ def rerun(self) -> None: changes_made = [] - # Clean up review/ artifacts - for filename in ("stdout.txt", "stderr.txt", "returncode.txt"): - f = self.job_review_path / filename - if f.exists(): - f.unlink() - changes_made.append(filename) + # Clean up the artifacts of the previous run. A staged copy must go + # too, or a later release sends a log that belongs to the old run. + for filename in ("stdout.txt", "stderr.txt", "returncode.txt", FRAMES_FILENAME): + for f in (self.job_review_path / filename, self.job_staging_path / filename): + if f.exists(): + f.unlink() + changes_made.append(filename) outputs_dir = self.job_review_path / "outputs" if outputs_dir.exists() and outputs_dir.is_dir(): diff --git a/packages/syft-job/src/syft_job/job_repr.py b/packages/syft-job/src/syft_job/job_repr.py index a12507aaab7..56e3d797089 100644 --- a/packages/syft-job/src/syft_job/job_repr.py +++ b/packages/syft-job/src/syft_job/job_repr.py @@ -60,7 +60,7 @@ def __str__(self) -> str: if self.job_info.status not in ("done", "failed"): return "No stderr available - job not completed yet" - stderr_file = self.job_info.job_review_path / "stderr.txt" + stderr_file = self.job_info.artifact_path("stderr.txt") if not stderr_file.exists(): return "No stderr file found" @@ -89,7 +89,7 @@ def _repr_html_(self) -> str: if self.job_info.status not in ("done", "failed"): error_msg = "No stderr available - job not completed yet" else: - stderr_file = self.job_info.job_review_path / "stderr.txt" + stderr_file = self.job_info.artifact_path("stderr.txt") if not stderr_file.exists(): error_msg = "No stderr file found" diff --git a/packages/syft-job/src/syft_job/job_runner.py b/packages/syft-job/src/syft_job/job_runner.py index 471eca5d811..c6260262f1a 100644 --- a/packages/syft-job/src/syft_job/job_runner.py +++ b/packages/syft-job/src/syft_job/job_runner.py @@ -1,6 +1,7 @@ import os import shutil import subprocess +import tempfile import time from datetime import datetime, timezone from pathlib import Path @@ -14,6 +15,15 @@ from .config import SyftJobConfig from .job_storage import JobRef, JobStorage, JobStateNotFoundError from .models import JobState, JobStatus, JobSubmissionMetadata +from .traceback_capture import ( + BUNDLE_FILENAME, + RAW_TRACE_FILENAME, + load_or_create_bundle, + RAW_TRACE_PATH_ENV, + TRACE_RUNNER_ENV, + trace_runner_path, + write_frames_record, +) # Default timeout for job execution (10 minutes) DEFAULT_JOB_TIMEOUT_SECONDS = 600 @@ -32,6 +42,18 @@ def get_job_timeout_seconds() -> int: IS_IN_JOB_ENV_VAR = "SYFT_IS_IN_JOB" +def _add_trace_env(env: dict, trace_dir: "Path | None") -> None: + """Point the job process at the traceback wrapper and its output file. + + Without these variables run.sh runs the entrypoint directly, so an older + run.sh and a newer run.sh both work. + """ + if trace_dir is None: + return + env[TRACE_RUNNER_ENV] = str(trace_runner_path()) + env[RAW_TRACE_PATH_ENV] = str(trace_dir / RAW_TRACE_FILENAME) + + def _kill_process_tree(pid: int, timeout: float = 2.0) -> None: """Kill `pid` and every descendant. Cross-platform via psutil.""" try: @@ -51,7 +73,7 @@ def _kill_process_tree(pid: int, timeout: float = 2.0) -> None: class SyftJobRunner: """Job runner that monitors and executes approved jobs. - Reads run.sh from inbox/, writes all output artifacts to review/. + Reads run.sh from inbox/. Outputs and state go to review/, logs to staging/. """ def __init__(self, config: SyftJobConfig, poll_interval: int = 5): @@ -218,13 +240,14 @@ def _find_jobref_from_name(self, job_name: str, user: str | None = None) -> JobR self.config.current_user_email, job_name, ds_email=user ) - def _execute_job_streaming(self, ref: JobRef, timeout: int) -> int: + def _execute_job_streaming( + self, ref: JobRef, timeout: int, trace_dir: Path | None = None + ) -> int: """Execute job with real-time streaming output. - Reads run.sh from inbox/, writes stdout/stderr to review/. + Reads run.sh from inbox/, writes stdout/stderr to staging/. """ submission_dir = self.manager.submission_dir(ref) - review_dir = self.manager.review_dir(ref) run_script = submission_dir / "run.sh" job_name = ref.job_name @@ -240,10 +263,12 @@ def _execute_job_streaming(self, ref: JobRef, timeout: int) -> int: env["SYFTBOX_EMAIL"] = self.config.current_user_email env[IS_IN_JOB_ENV_VAR] = "true" env["PYTHONUNBUFFERED"] = "1" + _add_trace_env(env, trace_dir) - # stdout/stderr go to review/ - stdout_file = review_dir / "stdout.txt" - stderr_file = review_dir / "stderr.txt" + staging_dir = self.manager.staging_dir(ref) + staging_dir.mkdir(parents=True, exist_ok=True) + stdout_file = staging_dir / "stdout.txt" + stderr_file = staging_dir / "stderr.txt" import selectors @@ -308,13 +333,14 @@ def _execute_job_streaming(self, ref: JobRef, timeout: int) -> int: return returncode - def _execute_job_captured(self, ref: JobRef, timeout: int) -> int: + def _execute_job_captured( + self, ref: JobRef, timeout: int, trace_dir: Path | None = None + ) -> int: """Execute job with captured output (non-streaming). - Reads run.sh from inbox/, writes stdout/stderr to review/. + Reads run.sh from inbox/, writes stdout/stderr to staging/. """ submission_dir = self.manager.submission_dir(ref) - review_dir = self.manager.review_dir(ref) run_script = submission_dir / "run.sh" job_name = ref.job_name @@ -327,6 +353,7 @@ def _execute_job_captured(self, ref: JobRef, timeout: int) -> int: env["SYFTBOX_EMAIL"] = self.config.current_user_email env[IS_IN_JOB_ENV_VAR] = "true" env["PYTHONUNBUFFERED"] = "1" + _add_trace_env(env, trace_dir) process = subprocess.Popen( ["bash", str(run_script)], @@ -348,11 +375,14 @@ def _execute_job_captured(self, ref: JobRef, timeout: int) -> int: stderr = (stderr or "") + "\n--- PROCESS TIMED OUT ---\n" print(f" Job {job_name} timed out after {timeout // 60} minutes") - stdout_file = review_dir / "stdout.txt" + staging_dir = self.manager.staging_dir(ref) + staging_dir.mkdir(parents=True, exist_ok=True) + + stdout_file = staging_dir / "stdout.txt" with open(stdout_file, "w") as f: f.write(stdout) - stderr_file = review_dir / "stderr.txt" + stderr_file = staging_dir / "stderr.txt" with open(stderr_file, "w") as f: f.write(stderr) @@ -367,7 +397,7 @@ def _execute_job( """ Execute run.sh for an approved job. - Reads run.sh from inbox/, writes all output to review/. + Reads run.sh from inbox/. Outputs and state go to review/, logs to staging/. Args: ref: Ref of the job to execute. @@ -399,11 +429,27 @@ def _execute_job( state.status = JobStatus.RUNNING self.manager.write_state(ref, state) + trace_dir = Path(tempfile.mkdtemp(prefix="syft-job-trace-")) + # The job writes into code/ while it runs, so the record must come from + # the tree as it stood before the first run. + code_bundle = load_or_create_bundle( + self.manager.staging_dir(ref) / BUNDLE_FILENAME, + submission_dir / "code", + ) try: if stream_output: - returncode = self._execute_job_streaming(ref, timeout) + returncode = self._execute_job_streaming(ref, timeout, trace_dir) else: - returncode = self._execute_job_captured(ref, timeout) + returncode = self._execute_job_captured(ref, timeout, trace_dir) + + # The job writes the raw record, so check it before any party reads + # it. A job that writes nothing produces no file. + write_frames_record( + trace_dir / RAW_TRACE_FILENAME, + self.manager.staging_dir(ref), + submission_dir / "code", + code_bundle, + ) # Move outputs from inbox/ to review/ self._move_outputs_to_review(submission_dir, review_dir) @@ -416,8 +462,9 @@ def _execute_job( # Update state to DONE or FAILED self._set_finalized_job_state(ref, returncode) - stdout_file = review_dir / "stdout.txt" - stderr_file = review_dir / "stderr.txt" + staging_dir = self.manager.staging_dir(ref) + stdout_file = staging_dir / "stdout.txt" + stderr_file = staging_dir / "stderr.txt" if returncode == 0: print(f" Job {job_name} completed successfully") @@ -442,6 +489,9 @@ def _execute_job( self._set_finalized_job_state(ref, -1) return False + finally: + shutil.rmtree(trace_dir, ignore_errors=True) + def _set_finalized_job_state(self, ref: JobRef, returncode: int) -> None: state = self.manager.read_state(ref) state.status = JobStatus.DONE if returncode == 0 else JobStatus.FAILED @@ -521,7 +571,7 @@ def process_approved_jobs( timeout: int | None = None, skip_job_names: list[str] | None = None, share_outputs_with_submitter: bool = False, - share_logs_with_submitter: bool = False, + share_logs_with_submitter: bool = True, ) -> None: """Process all jobs in approved status. @@ -530,7 +580,9 @@ def process_approved_jobs( timeout: Timeout in seconds per job. Defaults to 300 (5 minutes). skip_job_names: Optional list of job names to skip. share_outputs_with_submitter: If True, grant read access on outputs to submitter. - share_logs_with_submitter: If True, grant read access on logs to submitter. + share_logs_with_submitter: If True (default), release the logs to the + submitter. False keeps them in staging, where only the datasite + owner reads them. """ approved_jobs = self._get_jobs_in_approved() @@ -579,6 +631,9 @@ def _share_job_results( if share_outputs: job_info.share_outputs([ref.ds_email]) if share_logs: + # The move into the review directory is what discloses the logs. + # The grant covers a reader that holds no folder grant. + job_info.release_logs() job_info.share_logs([ref.ds_email]) def run(self) -> None: diff --git a/packages/syft-job/src/syft_job/job_stdout.py b/packages/syft-job/src/syft_job/job_stdout.py index d5b189555e7..64ca7bffbf8 100644 --- a/packages/syft-job/src/syft_job/job_stdout.py +++ b/packages/syft-job/src/syft_job/job_stdout.py @@ -67,7 +67,7 @@ def __str__(self) -> str: if self.job_info.status != "done": return "No stdout available - job not completed yet" - stdout_file = self.job_info.job_review_path / "stdout.txt" + stdout_file = self.job_info.artifact_path("stdout.txt") if not stdout_file.exists(): return "No stdout file found" @@ -97,7 +97,7 @@ def _repr_html_(self) -> str: if self.job_info.status != "done": error_msg = "No stdout available - job not completed yet" else: - stdout_file = self.job_info.job_review_path / "stdout.txt" + stdout_file = self.job_info.artifact_path("stdout.txt") if not stdout_file.exists(): error_msg = "No stdout file found" diff --git a/packages/syft-job/src/syft_job/job_storage.py b/packages/syft-job/src/syft_job/job_storage.py index c0b5c403f96..1052247219e 100644 --- a/packages/syft-job/src/syft_job/job_storage.py +++ b/packages/syft-job/src/syft_job/job_storage.py @@ -121,6 +121,9 @@ def submission_dir(self, ref: JobRef) -> Path: def review_dir(self, ref: JobRef) -> Path: return self._codec_for(ref.protocol_version).review_dir(ref) + def staging_dir(self, ref: JobRef) -> Path: + return self._codec_for(ref.protocol_version).staging_dir(ref) + def new_submission_ref(self, do_email: str, job_name: str) -> JobRef: """A ref for submitting a new job to ``do_email``.""" return JobRef( diff --git a/packages/syft-job/src/syft_job/protocolcodecs/base.py b/packages/syft-job/src/syft_job/protocolcodecs/base.py index 9649eea8541..ccf1e357a76 100644 --- a/packages/syft-job/src/syft_job/protocolcodecs/base.py +++ b/packages/syft-job/src/syft_job/protocolcodecs/base.py @@ -32,6 +32,9 @@ def submission_dir(self, ref: JobRef) -> Path: ... @abstractmethod def review_dir(self, ref: JobRef) -> Path: ... + @abstractmethod + def staging_dir(self, ref: JobRef) -> Path: ... + @abstractmethod def submission_metadata_path(self, ref: JobRef) -> Path: ... diff --git a/packages/syft-job/src/syft_job/protocolcodecs/v0.py b/packages/syft-job/src/syft_job/protocolcodecs/v0.py index 8fd8dbe468c..804b554ae8b 100644 --- a/packages/syft-job/src/syft_job/protocolcodecs/v0.py +++ b/packages/syft-job/src/syft_job/protocolcodecs/v0.py @@ -25,6 +25,11 @@ def review_dir(self, ref: JobRef) -> Path: ref.datasite_email, ref.ds_email, ref.job_name, ref.protocol_version ) + def staging_dir(self, ref: JobRef) -> Path: + return self.config.get_staging_job_dir( + ref.datasite_email, ref.ds_email, ref.job_name, ref.protocol_version + ) + def submission_metadata_path(self, ref: JobRef) -> Path: return self.submission_dir(ref) / self.submission_marker diff --git a/packages/syft-job/src/syft_job/protocolcodecs/v1.py b/packages/syft-job/src/syft_job/protocolcodecs/v1.py index af00bf6f701..8b3d54f17bc 100644 --- a/packages/syft-job/src/syft_job/protocolcodecs/v1.py +++ b/packages/syft-job/src/syft_job/protocolcodecs/v1.py @@ -25,6 +25,11 @@ def review_dir(self, ref: JobRef) -> Path: ref.datasite_email, ref.ds_email, ref.job_name, ref.protocol_version ) + def staging_dir(self, ref: JobRef) -> Path: + return self.config.get_staging_job_dir( + ref.datasite_email, ref.ds_email, ref.job_name, ref.protocol_version + ) + def submission_metadata_path(self, ref: JobRef) -> Path: return self.submission_dir(ref) / self.submission_marker diff --git a/packages/syft-job/src/syft_job/traceback_capture.py b/packages/syft-job/src/syft_job/traceback_capture.py new file mode 100644 index 00000000000..c0455ffa24d --- /dev/null +++ b/packages/syft-job/src/syft_job/traceback_capture.py @@ -0,0 +1,248 @@ +"""Turn an untrusted traceback record into a bounded one. + +The job process writes the raw record, so a job chooses every string in it. A +job can put private data in a file name, in a function name, or in an exception +message. Therefore only these fields reach a party: + +- a path that resolves to a file in the approved code bundle, or a fixed label; +- a line number inside that file; +- an exception type that is a builtin. + +The function names, the exception messages, and the local variables never reach +a party. +""" + +from __future__ import annotations + +import builtins +import json +import os +from pathlib import Path +from typing import Optional + +from syft_job._trace_runner import RAW_TRACE_PATH_ENV # noqa: F401 (re-export) + +RAW_TRACE_FILENAME = "_raw_traceback.json" +FRAMES_FILENAME = "traceback_frames.json" +TRACE_RUNNER_ENV = "SYFT_JOB_TRACE_RUNNER" + +EXTERNAL_LABEL = "" +FALLBACK_TYPE = "Exception" +MAX_CHAIN = 5 +MAX_FRAMES = 64 +BUNDLE_FILENAME = "code_bundle.json" +RUNNER_DIRS = frozenset({".venv", "__pycache__"}) +MAX_BUNDLE_FILES = 2000 +MAX_BUNDLE_FILE_BYTES = 5 * 1024 * 1024 + + +def trace_runner_path() -> Path: + """Return the path of the wrapper script that the job process runs.""" + return Path(__file__).with_name("_trace_runner.py") + + +def _builtin_exception_name(mro: object) -> str: + """Return the first name in ``mro`` that names a builtin exception. + + The class names come from the job, so each name must match a builtin before + it reaches a party. A custom class therefore reports its nearest builtin + base, such as ``ValueError`` for ``class Secret(ValueError)``. + """ + if not isinstance(mro, list): + return FALLBACK_TYPE + for name in mro: + if not isinstance(name, str): + continue + cls = getattr(builtins, name, None) + if isinstance(cls, type) and issubclass(cls, BaseException): + return name + return FALLBACK_TYPE + + +def _line_count(path: Path) -> int: + with open(path, "rb") as f: + return sum(1 for _ in f) + + +def snapshot_code_bundle(code_root: Path) -> dict[str, int]: + """Record the approved files and their lengths, before the job runs. + + Returns a map of a path relative to ``code_root`` to a line count. + + The job writes into its own code directory: ``run.sh`` builds a virtual + environment there. A job can therefore plant a file whose name carries + private data, then raise from it. The record must come from the tree the + parties approved, so this runs before execution. + """ + bundle: dict[str, int] = {} + if not code_root.is_dir(): + return bundle + try: + root = code_root.resolve() + except OSError: + return bundle + + for path in sorted(root.rglob("*")): + if len(bundle) >= MAX_BUNDLE_FILES: + break + if not path.is_file() or path.is_symlink(): + continue + name = path.relative_to(root).as_posix() + # run.sh builds the virtual environment here, so these paths belong to + # the runner and never to the approved bundle. + if any(part in RUNNER_DIRS for part in name.split("/")[:-1]): + continue + try: + if path.stat().st_size > MAX_BUNDLE_FILE_BYTES: + continue + bundle[name] = _line_count(path) + except OSError: + continue + return bundle + + +def _safe_frame(frame: object, code_root: Path, bundle: dict) -> dict: + """Return one frame, or ```` when ``bundle`` does not hold it.""" + external = {"file": EXTERNAL_LABEL, "line": None} + if not isinstance(frame, dict): + return external + + filename = frame.get("filename") + lineno = frame.get("lineno") + if not isinstance(filename, str) or not filename: + return external + + try: + if os.path.isabs(filename): + candidate = Path(filename).resolve() + else: + candidate = (code_root / filename).resolve() + except OSError: + return external + + if not candidate.is_relative_to(code_root): + return external + + name = candidate.relative_to(code_root).as_posix() + total = bundle.get(name) + if total is None: + # The job planted this file after the parties approved the bundle. + return external + if not isinstance(lineno, int) or isinstance(lineno, bool): + return external + if lineno < 1 or lineno > total: + # A line outside the file means the job forged the position. + return external + + return {"file": name, "line": lineno} + + +def sanitize_raw_trace( + raw: object, code_root: Path, bundle: dict[str, int] +) -> Optional[dict]: + """Return the bounded record, or None when ``raw`` holds no usable chain. + + ``code_root`` is the directory of the approved code bundle, and ``bundle`` + is the snapshot that ``snapshot_code_bundle`` took before the job ran. A + frame counts as safe only when the snapshot holds its file and its line. + """ + if not isinstance(raw, dict): + return None + chain = raw.get("chain") + if not isinstance(chain, list) or not chain: + return None + + try: + root = code_root.resolve() + except OSError: + return None + + safe_chain = [] + for entry in chain[:MAX_CHAIN]: + if not isinstance(entry, dict): + continue + frames = entry.get("frames") + frames = frames[:MAX_FRAMES] if isinstance(frames, list) else [] + safe_chain.append( + { + "type": _builtin_exception_name(entry.get("mro")), + "frames": [_safe_frame(f, root, bundle) for f in frames], + } + ) + + if not safe_chain: + return None + return {"chain": safe_chain} + + +def write_frames_record( + raw_path: Path, out_dir: Path, code_root: Path, bundle: dict[str, int] +) -> Optional[Path]: + """Read the raw record, check it, and write ``traceback_frames.json``. + + ``out_dir`` is the staging directory, therefore the record waits there until + a party releases it. + + Returns the path of the written file, or None when the job wrote no usable + record. A job that writes nothing, or writes invalid JSON, produces no file. + """ + if not raw_path.is_file(): + return None + try: + with open(raw_path, "r") as f: + raw = json.load(f) + except (OSError, json.JSONDecodeError): + return None + + record = sanitize_raw_trace(raw, code_root, bundle) + if record is None: + return None + + out_dir.mkdir(parents=True, exist_ok=True) + out_path = out_dir / FRAMES_FILENAME + with open(out_path, "w") as f: + json.dump(record, f, indent=2) + return out_path + + +def load_or_create_bundle(bundle_path: Path, code_root: Path) -> dict[str, int]: + """Return the approved tree, recorded once before the first run. + + A rerun executes the same bundle, and the first run leaves files behind. + Reading a stored record therefore keeps a later run from approving a file + that the parties never saw. + """ + if bundle_path.is_file(): + try: + stored = json.loads(bundle_path.read_text()) + except (OSError, json.JSONDecodeError): + stored = None + if isinstance(stored, dict): + return { + name: count + for name, count in stored.items() + if isinstance(name, str) + and isinstance(count, int) + and not isinstance(count, bool) + } + + bundle = snapshot_code_bundle(code_root) + try: + bundle_path.parent.mkdir(parents=True, exist_ok=True) + bundle_path.write_text(json.dumps(bundle)) + except OSError: + pass + return bundle + + +def write_no_failure_record(out_dir: Path) -> Path: + """Write a record that says the run did not fail. + + An empty chain never comes from a crash, because a record with no chain is + dropped. It therefore marks a run that produced no failure. + """ + out_dir.mkdir(parents=True, exist_ok=True) + out_path = out_dir / FRAMES_FILENAME + with open(out_path, "w") as f: + json.dump({"chain": []}, f, indent=2) + return out_path diff --git a/packages/syft-job/tests/test_job_flow.py b/packages/syft-job/tests/test_job_flow.py index d5022e76417..f09b34a41fa 100644 --- a/packages/syft-job/tests/test_job_flow.py +++ b/packages/syft-job/tests/test_job_flow.py @@ -70,8 +70,12 @@ def test_full_job_lifecycle(tmp_path: Path): job.approve() assert job.status == "approved" - # --- DO runs approved jobs --- - do_runner.process_approved_jobs(stream_output=False, timeout=120) + # --- DO runs approved jobs, holding the logs back --- + # share_logs_with_submitter=False keeps the logs in staging/, so the + # assertions below see the state before any release. + do_runner.process_approved_jobs( + stream_output=False, timeout=120, share_logs_with_submitter=False + ) # Re-fetch to get updated status job = do_client.jobs[0] @@ -84,12 +88,18 @@ def test_full_job_lifecycle(tmp_path: Path): result_file = next(p for p in job.output_paths if p.name == "result.txt") assert result_file.read_text().strip() == "done" - # --- Check stdout / stderr (now in review/) --- - stdout_path = review_path / "stdout.txt" - stderr_path = review_path / "stderr.txt" + # --- Check stdout / stderr (staged until released) --- + staging_path = do_config.get_staging_job_dir(DO_EMAIL, DS_EMAIL, "test.job") + stdout_path = staging_path / "stdout.txt" + stderr_path = staging_path / "stderr.txt" assert stdout_path.exists() assert stderr_path.exists() assert "hello from job" in stdout_path.read_text() + # Nothing readable sits in review/ before the release. + assert not (review_path / "stdout.txt").exists() + assert not (review_path / "stderr.txt").exists() + # The viewer finds the staged file, because the DO reads both directories. + assert "hello from job" in str(job.stdout) # --- Check returncode (now in review/) --- returncode_path = review_path / "returncode.txt" @@ -97,6 +107,10 @@ def test_full_job_lifecycle(tmp_path: Path): assert returncode_path.read_text().strip() == "0" # --- Before sharing, DS should NOT have read access --- + # This test never calls setup_ds_job_folder_as_do, so the DS holds no grant + # on review//. A real datasite always calls it, and that folder grant + # covers every nested file. share_logs() therefore grants access the folder + # grant has already given, and these assertions hold only here. ctx = SyftPermContext(datasite=syftbox / DO_EMAIL) assert not ctx.open( f"app_data/job/review/{DS_EMAIL}/v1/test.job/outputs/" @@ -111,10 +125,16 @@ def test_full_job_lifecycle(tmp_path: Path): f"app_data/job/review/{DS_EMAIL}/v1/test.job/returncode.txt" ).has_read_access(DS_EMAIL) - # --- Share outputs and logs with DS --- + # --- Release the logs, then share outputs and logs with DS --- job.share_outputs([DS_EMAIL]) + assert sorted(job.release_logs()) == ["stderr.txt", "stdout.txt"] job.share_logs([DS_EMAIL]) + # The release moved the files into review/, which is the disclosure. + assert (review_path / "stdout.txt").exists() + assert (review_path / "stderr.txt").exists() + assert not stdout_path.exists() + # --- Verify DS has read access via SyftPermContext --- ctx = SyftPermContext(datasite=syftbox / DO_EMAIL) diff --git a/packages/syft-job/tests/test_staging.py b/packages/syft-job/tests/test_staging.py new file mode 100644 index 00000000000..706f1db60ab --- /dev/null +++ b/packages/syft-job/tests/test_staging.py @@ -0,0 +1,172 @@ +"""Staged artifacts reach the submitter only when a release moves them.""" + +import json +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.traceback_capture import ( + BUNDLE_FILENAME, + FRAMES_FILENAME, + load_or_create_bundle, +) + +DO_EMAIL = "do@test.org" +DS_EMAIL = "ds@test.org" + +OK_PY = """\ +import os + +print("hello from job") +os.makedirs("outputs", exist_ok=True) +with open("outputs/result.txt", "w") as f: + f.write("done") +""" + +CRASH_PY = """\ +x = 1 +raise ValueError("patient 4171 is positive") +""" + + +def run_job(tmp_path: Path, code: str, share_logs: bool): + """Submit and run one job. Returns (job, review_dir, staging_dir).""" + syftbox = tmp_path / "SyftBox" + syftbox.mkdir(exist_ok=True) + code_file = tmp_path / "main.py" + code_file.write_text(code) + + do_config = SyftJobConfig(syftbox_folder=syftbox, current_user_email=DO_EMAIL) + ds_config = SyftJobConfig(syftbox_folder=syftbox, current_user_email=DS_EMAIL) + ds_client = JobClient(config=ds_config) + do_client = JobClient(config=do_config) + do_runner = SyftJobRunner(config=do_config) + + ds_client.submit_python_job( + user=DO_EMAIL, code_path=str(code_file), job_name="test.job" + ) + job = do_client.jobs[0] + job.approve() + do_runner.process_approved_jobs( + stream_output=False, timeout=180, share_logs_with_submitter=share_logs + ) + job = do_client.jobs[0] + return ( + job, + do_config.get_review_job_dir(DO_EMAIL, DS_EMAIL, "test.job"), + do_config.get_staging_job_dir(DO_EMAIL, DS_EMAIL, "test.job"), + ) + + +def test_logs_wait_in_staging_when_nothing_released_them(tmp_path): + job, review, staging = run_job(tmp_path, OK_PY, share_logs=False) + assert (staging / "stdout.txt").exists() + assert (staging / "stderr.txt").exists() + assert not (review / "stdout.txt").exists() + assert not (review / "stderr.txt").exists() + + +def test_returncode_stays_in_review(tmp_path): + """returncode.txt holds one integer, which state.yaml already carries.""" + job, review, staging = run_job(tmp_path, OK_PY, share_logs=False) + assert (review / "returncode.txt").exists() + assert not (staging / "returncode.txt").exists() + + +def test_a_release_moves_the_logs_into_review(tmp_path): + job, review, staging = run_job(tmp_path, OK_PY, share_logs=False) + assert sorted(job.release_logs()) == ["stderr.txt", "stdout.txt"] + assert (review / "stdout.txt").exists() + assert not (staging / "stdout.txt").exists() + + +def test_the_default_releases_the_logs(tmp_path): + """A caller that passes nothing keeps the behaviour it had before staging.""" + job, review, staging = run_job(tmp_path, OK_PY, share_logs=True) + assert (review / "stdout.txt").exists() + assert not (staging / "stdout.txt").exists() + + +def test_a_second_release_is_harmless(tmp_path): + job, review, staging = run_job(tmp_path, OK_PY, share_logs=False) + job.release_logs() + assert job.release_logs() == [] + assert (review / "stdout.txt").exists() + + +def test_the_owner_reads_a_staged_log_through_the_viewer(tmp_path): + job, review, staging = run_job(tmp_path, OK_PY, share_logs=False) + assert "hello from job" in str(job.stdout) + assert job.artifact_path("stdout.txt") == staging / "stdout.txt" + job.release_logs() + assert job.artifact_path("stdout.txt") == review / "stdout.txt" + + +def test_the_traceback_record_is_staged(tmp_path): + job, review, staging = run_job(tmp_path, CRASH_PY, share_logs=False) + record_path = staging / FRAMES_FILENAME + assert record_path.exists() + assert not (review / FRAMES_FILENAME).exists() + + record = json.loads(record_path.read_text()) + assert record["chain"][0]["type"] == "ValueError" + assert "positive" not in record_path.read_text() + + assert job.release_artifacts([FRAMES_FILENAME]) == [FRAMES_FILENAME] + assert (review / FRAMES_FILENAME).exists() + + +def test_rerun_clears_both_locations(tmp_path): + """A staged log from the old run must not survive into a later release.""" + job, review, staging = run_job(tmp_path, OK_PY, share_logs=False) + (review / "returncode.txt").write_text("0") + assert (staging / "stdout.txt").exists() + + job.rerun() + + assert not (staging / "stdout.txt").exists() + assert not (staging / "stderr.txt").exists() + assert not (review / "stdout.txt").exists() + + +def test_rerun_clears_the_crash_record(tmp_path): + """A crash record from the old run must not survive a later release. + + A rerun that succeeds writes no record, so a leftover record would describe + a failure that the new run never had. + """ + job, review, staging = run_job(tmp_path, CRASH_PY, share_logs=False) + assert (staging / FRAMES_FILENAME).exists() + + job.rerun() + + assert not (staging / FRAMES_FILENAME).exists() + assert not (review / FRAMES_FILENAME).exists() + + +def test_a_rerun_keeps_the_bundle_the_parties_approved(tmp_path): + """The first run leaves .venv and any file the job wrote in code/. + + A rerun must judge frames against the tree as it stood before the first + run, or a planted file counts as approved and the real sources drop out. + """ + job, review, staging = run_job(tmp_path, OK_PY, share_logs=False) + bundle_path = staging / BUNDLE_FILENAME + assert bundle_path.exists() + + recorded = json.loads(bundle_path.read_text()) + assert "main.py" in recorded + assert not any(name.startswith(".venv") for name in recorded) + + # The first run really did leave a virtual environment behind. + code_dir = job.job_submission_path / "code" + assert (code_dir / ".venv").is_dir() + + # A job that plants a file gains nothing on the next run. + (code_dir / "DO2_row_4171_POSITIVE.py").write_text("\n" * 50) + reloaded = load_or_create_bundle(bundle_path, code_dir) + assert "DO2_row_4171_POSITIVE.py" not in reloaded + assert reloaded == recorded diff --git a/packages/syft-job/tests/test_traceback_capture.py b/packages/syft-job/tests/test_traceback_capture.py new file mode 100644 index 00000000000..968b1360b4e --- /dev/null +++ b/packages/syft-job/tests/test_traceback_capture.py @@ -0,0 +1,278 @@ +"""The sanitizer treats the raw record as attacker-controlled input.""" + +import json +import subprocess +import sys +from pathlib import Path + +import pytest + +from syft_job._trace_runner import raw_record +from syft_job.traceback_capture import ( + EXTERNAL_LABEL, + FALLBACK_TYPE, + FRAMES_FILENAME, + MAX_CHAIN, + MAX_FRAMES, + sanitize_raw_trace, + snapshot_code_bundle, + trace_runner_path, + write_frames_record, +) + + +def sanitize(raw, code_root): + """Sanitize against a snapshot taken from the current tree.""" + return sanitize_raw_trace(raw, code_root, snapshot_code_bundle(code_root)) + + +@pytest.fixture +def code_root(tmp_path): + root = tmp_path / "code" + root.mkdir() + (root / "main.py").write_text("\n".join(f"line {i}" for i in range(1, 21))) + return root + + +def frame(filename, lineno): + return {"filename": filename, "lineno": lineno} + + +def chain(mro, frames): + return {"chain": [{"mro": mro, "frames": frames}]} + + +# -- what a reader is allowed to see ------------------------------------------ + + +def test_frame_in_the_bundle_keeps_its_file_and_line(code_root): + out = sanitize(chain(["ValueError"], [frame("main.py", 12)]), code_root) + assert out["chain"][0]["frames"] == [{"file": "main.py", "line": 12}] + assert out["chain"][0]["type"] == "ValueError" + + +def test_absolute_path_in_the_bundle_is_relative_in_the_record(code_root): + raw = chain(["KeyError"], [frame(str(code_root / "main.py"), 3)]) + out = sanitize(raw, code_root) + assert out["chain"][0]["frames"] == [{"file": "main.py", "line": 3}] + + +def test_the_record_never_holds_a_message_or_a_function_name(code_root): + raw = chain(["ValueError"], [frame("main.py", 1)]) + raw["chain"][0]["frames"][0]["name"] = "leak_do2_row_4171" + raw["chain"][0]["message"] = "patient 4171 is positive" + out = sanitize(raw, code_root) + assert set(out["chain"][0]) == {"type", "frames"} + assert set(out["chain"][0]["frames"][0]) == {"file", "line"} + + +# -- smuggling attempts -------------------------------------------------------- + + +def test_a_forged_file_name_never_reaches_the_record(code_root): + """A job picks the file name of a frame through compile().""" + raw = chain(["ValueError"], [frame("DO2_row_4171_diagnosis_POSITIVE.py", 1)]) + out = sanitize(raw, code_root) + assert out["chain"][0]["frames"] == [{"file": EXTERNAL_LABEL, "line": None}] + + +def test_a_path_outside_the_bundle_is_labelled(code_root, tmp_path): + outside = tmp_path / "secret.py" + outside.write_text("x = 1\n") + out = sanitize(chain(["OSError"], [frame(str(outside), 1)]), code_root) + assert out["chain"][0]["frames"] == [{"file": EXTERNAL_LABEL, "line": None}] + + +def test_a_traversal_path_stays_outside(code_root): + raw = chain(["OSError"], [frame("../../../etc/passwd", 1)]) + out = sanitize(raw, code_root) + assert out["chain"][0]["frames"][0]["file"] == EXTERNAL_LABEL + + +def test_a_line_past_the_end_of_the_file_is_dropped(code_root): + """main.py holds 20 lines, so line 99999 is a forged position.""" + out = sanitize(chain(["ValueError"], [frame("main.py", 99999)]), code_root) + assert out["chain"][0]["frames"] == [{"file": EXTERNAL_LABEL, "line": None}] + + +def test_a_custom_exception_reports_its_builtin_base(code_root): + raw = chain(["DO2DataWasPositive", "ValueError", "Exception"], []) + out = sanitize(raw, code_root) + assert out["chain"][0]["type"] == "ValueError" + + +def test_an_all_custom_mro_falls_back(code_root): + out = sanitize(chain(["SecretName", "AlsoSecret"], []), code_root) + assert out["chain"][0]["type"] == FALLBACK_TYPE + + +def test_the_chain_and_the_frame_count_are_bounded(code_root): + raw = { + "chain": [ + {"mro": ["ValueError"], "frames": [frame("main.py", 1)] * 500} + for _ in range(50) + ] + } + out = sanitize(raw, code_root) + assert len(out["chain"]) == MAX_CHAIN + assert len(out["chain"][0]["frames"]) == MAX_FRAMES + + +@pytest.mark.parametrize( + "raw", + [None, {}, {"chain": []}, {"chain": "nope"}, [1, 2, 3], {"chain": [None]}], +) +def test_malformed_input_yields_no_record(raw, code_root): + assert sanitize(raw, code_root) in (None, {"chain": []}) + + +def test_a_non_integer_line_is_dropped(code_root): + out = sanitize(chain(["ValueError"], [frame("main.py", "12")]), code_root) + assert out["chain"][0]["frames"][0]["line"] is None + + +# -- the file the parent writes ------------------------------------------------- + + +def test_write_frames_record_writes_only_checked_fields(tmp_path, code_root): + raw_path = tmp_path / "raw.json" + review = tmp_path / "review" + raw_path.write_text( + json.dumps(chain(["ValueError"], [frame("main.py", 2), frame("/etc/shadow", 1)])) + ) + out = write_frames_record( + raw_path, review, code_root, snapshot_code_bundle(code_root) + ) + assert out == review / FRAMES_FILENAME + record = json.loads(out.read_text()) + assert record["chain"][0]["frames"] == [ + {"file": "main.py", "line": 2}, + {"file": EXTERNAL_LABEL, "line": None}, + ] + + +def test_no_raw_file_means_no_record(tmp_path, code_root): + review = tmp_path / "review" + bundle = snapshot_code_bundle(code_root) + assert ( + write_frames_record(tmp_path / "absent.json", review, code_root, bundle) is None + ) + assert not (review / FRAMES_FILENAME).exists() + + +def test_invalid_json_means_no_record(tmp_path, code_root): + raw_path = tmp_path / "raw.json" + raw_path.write_text("{not json") + bundle = snapshot_code_bundle(code_root) + assert ( + write_frames_record(raw_path, tmp_path / "review", code_root, bundle) is None + ) + + +# -- the wrapper that runs inside the job --------------------------------------- + + +def test_raw_record_holds_no_message(): + try: + raise ValueError("patient 4171 is positive") + except ValueError as exc: + record = raw_record(exc) + blob = json.dumps(record) + assert "positive" not in blob + assert record["chain"][0]["mro"][0] == "ValueError" + + +def test_raw_record_follows_a_chained_exception(): + try: + try: + raise KeyError("inner") + except KeyError as inner: + raise RuntimeError("outer") from inner + except RuntimeError as exc: + record = raw_record(exc) + assert [c["mro"][0] for c in record["chain"]] == ["RuntimeError", "KeyError"] + assert "inner" not in json.dumps(record) + + +def test_the_wrapper_records_a_crash_and_keeps_the_exit_code(tmp_path): + entry = tmp_path / "main.py" + entry.write_text("raise ValueError('secret value')\n") + raw_path = tmp_path / "raw.json" + + proc = subprocess.run( + [sys.executable, str(trace_runner_path()), str(entry)], + capture_output=True, + text=True, + env={"PATH": "/usr/bin:/bin", "SYFT_JOB_RAW_TRACE_PATH": str(raw_path)}, + ) + + assert proc.returncode != 0 + assert "ValueError" in proc.stderr # the real traceback still reaches stderr + record = json.loads(raw_path.read_text()) + assert record["chain"][0]["mro"][0] == "ValueError" + assert "secret value" not in raw_path.read_text() + + out = sanitize_raw_trace(record, tmp_path, snapshot_code_bundle(tmp_path)) + assert out["chain"][0]["frames"][-1] == {"file": "main.py", "line": 1} + + +def test_the_wrapper_leaves_a_clean_run_alone(tmp_path): + entry = tmp_path / "main.py" + entry.write_text("print('ok')\n") + raw_path = tmp_path / "raw.json" + proc = subprocess.run( + [sys.executable, str(trace_runner_path()), str(entry)], + capture_output=True, + text=True, + env={"PATH": "/usr/bin:/bin", "SYFT_JOB_RAW_TRACE_PATH": str(raw_path)}, + ) + assert proc.returncode == 0 + assert "ok" in proc.stdout + assert not raw_path.exists() + + +def test_a_deliberate_exit_writes_no_record(tmp_path): + entry = tmp_path / "main.py" + entry.write_text("import sys; sys.exit(3)\n") + raw_path = tmp_path / "raw.json" + proc = subprocess.run( + [sys.executable, str(trace_runner_path()), str(entry)], + capture_output=True, + text=True, + env={"PATH": "/usr/bin:/bin", "SYFT_JOB_RAW_TRACE_PATH": str(raw_path)}, + ) + assert proc.returncode == 3 + assert not raw_path.exists() + + +def test_a_file_the_job_plants_after_approval_is_external(code_root): + """run.sh builds a venv inside code/, so the job can write there.""" + bundle = snapshot_code_bundle(code_root) + + planted = code_root / "DO2_row_4171_diagnosis_POSITIVE.py" + planted.write_text("\n" * 50) + + raw = chain(["ValueError"], [frame(planted.name, 42)]) + out = sanitize_raw_trace(raw, code_root, bundle) + assert out["chain"][0]["frames"] == [{"file": EXTERNAL_LABEL, "line": None}] + + +def test_a_file_the_job_grows_keeps_its_approved_length(code_root): + """A longer file must not widen the range of accepted line numbers.""" + bundle = snapshot_code_bundle(code_root) + (code_root / "main.py").write_text("\n" * 5000) + + out = sanitize_raw_trace(chain(["ValueError"], [frame("main.py", 900)]), code_root, bundle) + assert out["chain"][0]["frames"] == [{"file": EXTERNAL_LABEL, "line": None}] + + +def test_a_venv_file_is_external(code_root): + bundle = snapshot_code_bundle(code_root) + venv_file = code_root / ".venv" / "lib" / "evil.py" + venv_file.parent.mkdir(parents=True) + venv_file.write_text("x = 1\n") + + raw = chain(["ValueError"], [frame(".venv/lib/evil.py", 1)]) + assert sanitize_raw_trace(raw, code_root, bundle)["chain"][0]["frames"] == [ + {"file": EXTERNAL_LABEL, "line": None} + ] diff --git a/packages/syft-rds/src/syft_rds/client.py b/packages/syft-rds/src/syft_rds/client.py index 3af8076ccd6..783c66b3731 100644 --- a/packages/syft-rds/src/syft_rds/client.py +++ b/packages/syft-rds/src/syft_rds/client.py @@ -362,7 +362,7 @@ def process_approved_jobs( timeout: int | None = None, force_execution: bool = False, share_outputs_with_submitter: bool = False, - share_logs_with_submitter: bool = False, + share_logs_with_submitter: bool = True, ignore_peer_version: bool = False, ) -> None: """Process approved jobs (DO only). Auto-syncs after unless PRE_SYNC=false.""" diff --git a/tests/unit/test_truncated_logs.py b/tests/unit/test_truncated_logs.py index 990c0724e69..5eefe489b80 100644 --- a/tests/unit/test_truncated_logs.py +++ b/tests/unit/test_truncated_logs.py @@ -62,8 +62,9 @@ def func_d(): ref = runner._find_jobref_from_name(job_name, user=ds_email) runner._execute_job(ref, stream_output=True, timeout=30) - stdout_content = (review_dir / "stdout.txt").read_text() - stderr_content = (review_dir / "stderr.txt").read_text() + staging_dir = config.get_staging_job_dir(email, ds_email, job_name) + stdout_content = (staging_dir / "stdout.txt").read_text() + stderr_content = (staging_dir / "stderr.txt").read_text() # Check stdout has all print statements assert "Starting job..." in stdout_content From c877f37e1fc16a2fdf3920759a0bdf7e46fdd64d Mon Sep 17 00:00:00 2001 From: Pedro Werneck Date: Mon, 14 Sep 2026 22:23:37 -0300 Subject: [PATCH 2/3] Simplify traceback handling and improve artifact release logic --- .../syft-enclave/src/syft_enclaves/client.py | 147 +++----- .../src/syft_enclaves/enclave_job_client.py | 4 +- .../src/syft_enclaves/enclave_job_info.py | 6 +- .../syft-enclave/tests/test_disclosures.py | 124 +------ .../syft-job/src/syft_job/_trace_runner.py | 67 ---- packages/syft-job/src/syft_job/client.py | 12 +- packages/syft-job/src/syft_job/job.py | 5 +- packages/syft-job/src/syft_job/job_runner.py | 70 ++-- .../src/syft_job/traceback_capture.py | 260 ++++---------- packages/syft-job/tests/test_staging.py | 32 +- .../syft-job/tests/test_traceback_capture.py | 325 ++++++------------ 11 files changed, 253 insertions(+), 799 deletions(-) delete mode 100644 packages/syft-job/src/syft_job/_trace_runner.py diff --git a/packages/syft-enclave/src/syft_enclaves/client.py b/packages/syft-enclave/src/syft_enclaves/client.py index ba7dddd90e6..dbba3770ecb 100644 --- a/packages/syft-enclave/src/syft_enclaves/client.py +++ b/packages/syft-enclave/src/syft_enclaves/client.py @@ -1,7 +1,6 @@ from datetime import datetime, timezone from pathlib import Path from typing import Iterable, Optional -import hashlib import json import os @@ -14,7 +13,7 @@ from syft_job.job_storage import JobRef from syft_job.models import JobState, JobStatus -from syft_job.traceback_capture import FRAMES_FILENAME, write_no_failure_record +from syft_job.traceback_capture import FRAMES_FILENAME from syft_enclaves.enclave_job_info import ( DisclosureItem, @@ -54,6 +53,16 @@ } +def gated_names(granted: Iterable[str]) -> list[str]: + """The filenames that the granted disclosure items cover.""" + return [ + name + for item, names in GATED_ARTIFACTS.items() + if item in granted + for name in names + ] + + def pre_sync_enabled() -> bool: """Whether the client syncs on its own. ``PRE_SYNC=false`` turns it off.""" return os.environ.get("PRE_SYNC", "true").lower() == "true" @@ -264,125 +273,66 @@ def _apply_disclosure_policy(self) -> None: for job in self._local_jobs(): if job.status not in ("done", "failed"): continue - granted = self.granted_disclosures(job) - released = [ - name - for item, names in GATED_ARTIFACTS.items() - if item in granted - for name in names - ] - job.release_artifacts(released) + job.release_artifacts(gated_names(self.granted_disclosures(job))) def distribute_results(self) -> None: """Distribute job results to DS (always) and optionally to DOs.""" for job in self.jobs: if job.status not in ("done", "failed"): continue - if self._results_already_shared(job): - continue - - # Always share results with the DS (submitter) - self._forward_results_to_recipients(job, [job.submitted_by]) - - # Optionally share with DOs - datasets = job.job_metadata.datasets - if job.job_headers.get("share_results_with_do") and datasets: - do_emails = list(datasets.keys()) - job.share_outputs(do_emails) - self._forward_results_to_recipients(job, do_emails) - - self._mark_results_shared(job) - - # A grant can arrive after the results went out, so released artifacts - # travel on their own schedule. - for job in self.jobs: - if job.status in ("done", "failed"): - self._forward_new_releases(job) + self._share_results_once(job) + # A grant can arrive after the results went out, so a released + # artifact travels on its own schedule. + self._forward_new_releases(job) self._rds.sync() - def _state_digest(self, job: JobInfo) -> str: - """A digest of the job state, which changes with every run.""" - state_file = Path(job.job_review_path) / "state.yaml" - if not state_file.is_file(): - return "" - return hashlib.sha256(state_file.read_bytes()).hexdigest() + def _share_results_once(self, job: JobInfo) -> None: + """Send the outputs and the state, the first time the job finishes.""" + marker = job.job_review_path / RESULTS_SHARED_MARKER + if marker.exists(): + return - def _results_already_shared(self, job: JobInfo) -> bool: - """Whether the results of the current run already went out. + # Always share results with the DS (submitter) + self._forward_results_to_recipients(job, [job.submitted_by]) - A rerun writes a new state, so the marker names the run it covers. A - marker that names an earlier run does not stop the new results. - """ - marker = Path(job.job_review_path) / RESULTS_SHARED_MARKER - if not marker.exists(): - return False - try: - recorded = json.loads(marker.read_text()) - except (OSError, json.JSONDecodeError): - recorded = None - if not isinstance(recorded, dict): - # A marker from an older client holds plain text. Record the - # current run and keep the results as sent. - self._mark_results_shared(job) - return True - return recorded.get("state") == self._state_digest(job) - - def _mark_results_shared(self, job: JobInfo) -> None: - marker = Path(job.job_review_path) / RESULTS_SHARED_MARKER - marker.write_text(json.dumps({"state": self._state_digest(job)})) + # Optionally share with DOs + datasets = job.job_metadata.datasets + if job.job_headers.get("share_results_with_do") and datasets: + do_emails = list(datasets.keys()) + job.share_outputs(do_emails) + self._forward_results_to_recipients(job, do_emails) + + marker.write_text("shared") def _forward_new_releases(self, job: JobInfo) -> list[str]: """Send each released artifact that the parties do not hold yet. A party that releases an item also receives it, so the data owners get - the same file as the submitter. - - The record holds a digest for each name, not the name alone. A rerun - writes a new record under the same name, and a digest tells the two - apart, so the parties never keep the artifact of an earlier run. + the same file as the submitter. The record names what already went out, + because a grant can arrive long after the results did. """ - review_dir = Path(job.job_review_path) + review_dir = job.job_review_path sent_path = review_dir / FORWARDED_RECORD try: - sent = json.loads(sent_path.read_text()) - except (OSError, json.JSONDecodeError): - sent = {} - if not isinstance(sent, dict): - sent = {} - - granted = self.granted_disclosures(job) - - # A later run that did not fail leaves no record, so the parties would - # keep the crash of an earlier run. Replace it instead. - if ( - DisclosureItem.TRACEBACK_FRAMES.value in granted - and FRAMES_FILENAME in sent - and not (review_dir / FRAMES_FILENAME).is_file() - ): - write_no_failure_record(review_dir) - - released = [ - name - for item, names in GATED_ARTIFACTS.items() - if item in granted - for name in names - if (review_dir / name).is_file() - ] + sent = set(json.loads(sent_path.read_text())) + except (OSError, TypeError, json.JSONDecodeError): + sent = set() - pending = {} - for name in released: - content = (review_dir / name).read_bytes() - digest = hashlib.sha256(content).hexdigest() - if sent.get(name) != digest: - pending[name] = (content, digest) + pending = sorted( + name + for name in gated_names(self.granted_disclosures(job)) + if name not in sent and (review_dir / name).is_file() + ) if not pending: return [] datasite_dir = self._rds.syftbox_folder / self._rds.email files = { - (review_dir / name).relative_to(datasite_dir): content - for name, (content, _) in pending.items() + (review_dir / name).relative_to(datasite_dir): ( + review_dir / name + ).read_bytes() + for name in pending } datasets = job.job_metadata.datasets or {} recipients = list( @@ -390,9 +340,8 @@ def _forward_new_releases(self, job: JobInfo) -> list[str]: ) self._push_files_to_recipients(files, recipients) - sent.update({name: digest for name, (_, digest) in pending.items()}) - sent_path.write_text(json.dumps(sent, sort_keys=True)) - return sorted(pending) + sent_path.write_text(json.dumps(sorted(sent.union(pending)))) + return pending def _read_state_file(self, job: JobInfo) -> dict[Path, bytes]: """Read the job state.yaml as a {path_in_datasite: bytes} dict.""" diff --git a/packages/syft-enclave/src/syft_enclaves/enclave_job_client.py b/packages/syft-enclave/src/syft_enclaves/enclave_job_client.py index 5fe54d91c24..df4789c5ec9 100644 --- a/packages/syft-enclave/src/syft_enclaves/enclave_job_client.py +++ b/packages/syft-enclave/src/syft_enclaves/enclave_job_client.py @@ -74,9 +74,7 @@ def submit_python_job( "share_results_with_do": share_results_with_do, # The items the submitter asks for. Each data owner sees this list # next to the code, then releases none, some, or all of it. - "requested_disclosures": sorted( - normalize_disclosures(request_disclosures) - ), + "requested_disclosures": sorted(normalize_disclosures(request_disclosures)), } config.save(job_dir / "config.yaml") diff --git a/packages/syft-enclave/src/syft_enclaves/enclave_job_info.py b/packages/syft-enclave/src/syft_enclaves/enclave_job_info.py index b96b37c035f..4ec2caa34e0 100644 --- a/packages/syft-enclave/src/syft_enclaves/enclave_job_info.py +++ b/packages/syft-enclave/src/syft_enclaves/enclave_job_info.py @@ -14,9 +14,9 @@ class DisclosureItem(str, Enum): """A class of job data that a party can release to the other parties. - - ``TRACEBACK_FRAMES``: the failure position in the approved code, and the - builtin exception type. Bounded. - - ``LOGS``: stdout and stderr. Unbounded, and the job chooses every byte. + - ``TRACEBACK_FRAMES``: the file and the line of each frame, and the + exception type. Never the exception message. + - ``LOGS``: stdout and stderr, as the job wrote them. """ TRACEBACK_FRAMES = "traceback_frames" diff --git a/packages/syft-enclave/tests/test_disclosures.py b/packages/syft-enclave/tests/test_disclosures.py index d737eb0fec3..38c9c3d405f 100644 --- a/packages/syft-enclave/tests/test_disclosures.py +++ b/packages/syft-enclave/tests/test_disclosures.py @@ -3,7 +3,6 @@ import json import os import random -import shutil import tempfile from pathlib import Path @@ -64,7 +63,9 @@ def test_an_approval_file_without_the_field_grants_nothing(tmp_path): def test_the_requested_set_narrows_the_result(tmp_path): - write_approval(tmp_path, "do1@x.com", JobStatus.APPROVED, {LOGS: True, FRAMES: True}) + write_approval( + tmp_path, "do1@x.com", JobStatus.APPROVED, {LOGS: True, FRAMES: True} + ) assert approved_disclosures(tmp_path, [FRAMES]) == {FRAMES} @@ -293,122 +294,3 @@ def test_a_released_artifact_is_not_sent_twice(): job = enclave.jobs["j"] assert enclave._forward_new_releases(job) == [] - - -def test_a_changed_record_reaches_the_parties_again(): - """The forwarded record holds a digest, so new content goes out again.""" - enclave, do1, do2, ds = build_quad() - submit(ds, enclave, do1, do2, CRASH_CODE, [FRAMES]) - run_to_completion(enclave, do1, do2, ds, [FRAMES]) - - review = Path(enclave.jobs["j"].job_review_path) - first = json.loads((review / FRAMES_FILENAME).read_text()) - assert enclave._forward_new_releases(enclave.jobs["j"]) == [] - - # A later run writes a different record under the same name. - (review / FRAMES_FILENAME).write_text( - json.dumps({"chain": [{"type": "KeyError", "frames": []}]}) - ) - assert enclave._forward_new_releases(enclave.jobs["j"]) == [FRAMES_FILENAME] - - do1.sync() - delivered = json.loads( - (Path(do1.jobs["j"].job_review_path) / FRAMES_FILENAME).read_text() - ) - assert delivered["chain"][0]["type"] == "KeyError" - assert delivered != first - - -def test_a_new_run_is_not_skipped_as_already_shared(): - """distribute_results marks a failed job, so a later run must not be skipped. - - The marker names the run it covers. An end-to-end rerun cannot drive this: - `rerun()` leaves code/.venv and `uv venv` then fails, for every python job. - """ - enclave, do1, do2, ds = build_quad() - submit(ds, enclave, do1, do2, CRASH_CODE, [FRAMES]) - run_to_completion(enclave, do1, do2, ds, [FRAMES]) - - job = enclave.jobs["j"] - assert job.status == "failed" - assert enclave._results_already_shared(job) - - # A later run writes a new state, which is what rerun() prepares for. - state_file = Path(job.job_review_path) / "state.yaml" - state_file.write_text(state_file.read_text() + "\n# next run\n") - assert not enclave._results_already_shared(job) - - -def test_a_marker_from_an_older_client_still_counts_as_shared(): - enclave, do1, do2, ds = build_quad() - submit(ds, enclave, do1, do2, OK_CODE, [LOGS]) - run_to_completion(enclave, do1, do2, ds, [LOGS]) - - job = enclave.jobs["j"] - marker = Path(job.job_review_path) / "results_shared" - marker.write_text("shared") - - assert enclave._results_already_shared(job) - # The marker now names the current run, so the next run redistributes. - assert json.loads(marker.read_text())["state"] == enclave._state_digest(job) - - -# A job that fails once, then succeeds, with no edit between the runs. -FAIL_THEN_PASS_CODE = """\ -import json, os - -flag = {flag!r} -if not os.path.exists(flag): - open(flag, "w").write("x") - raise ValueError("patient 4171 is positive") - -os.makedirs("outputs", exist_ok=True) -open("outputs/r.json", "w").write(json.dumps({{"ok": 1}})) -""" - - -def rerun_on_the_enclave(enclave): - """Prepare a rerun, and clear what blocks one. - - `rerun()` leaves code/.venv, and `run.sh` then runs `uv venv`, which stops - under `set -euo pipefail`. That defect predates the staging work and no - caller hits it, so the test clears the directory itself. - """ - job = enclave.jobs["j"] - job.rerun() - shutil.rmtree(Path(job.job_submission_path) / "code" / ".venv", ignore_errors=True) - - -def test_a_run_that_does_not_fail_replaces_the_crash_record(): - enclave, do1, do2, ds = build_quad() - flag = str(Path(tempfile.mkdtemp()) / "once") - submit(ds, enclave, do1, do2, FAIL_THEN_PASS_CODE.format(flag=flag), [FRAMES]) - run_to_completion(enclave, do1, do2, ds, [FRAMES]) - do1.sync() - - assert enclave.jobs["j"].status == "failed" - do1_record = Path(do1.jobs["j"].job_review_path) / FRAMES_FILENAME - assert json.loads(do1_record.read_text())["chain"][0]["type"] == "ValueError" - - rerun_on_the_enclave(enclave) - enclave.run_jobs() - enclave.distribute_results() - ds.sync() - do1.sync() - - assert enclave.jobs["j"].status == "done" - # No party keeps a crash description for a run that did not fail. - for holder in (ds, do1): - record = Path(holder.jobs["j"].job_review_path) / FRAMES_FILENAME - assert json.loads(record.read_text()) == {"chain": []} - - -def test_no_replacement_when_the_parties_hold_no_record(): - """A job that never failed sends nothing, so there is nothing to replace.""" - enclave, do1, do2, ds = build_quad() - submit(ds, enclave, do1, do2, OK_CODE, [FRAMES]) - run_to_completion(enclave, do1, do2, ds, [FRAMES]) - - job = enclave.jobs["j"] - assert not (Path(job.job_review_path) / FRAMES_FILENAME).exists() - assert enclave._forward_new_releases(job) == [] diff --git a/packages/syft-job/src/syft_job/_trace_runner.py b/packages/syft-job/src/syft_job/_trace_runner.py deleted file mode 100644 index c3c85231959..00000000000 --- a/packages/syft-job/src/syft_job/_trace_runner.py +++ /dev/null @@ -1,67 +0,0 @@ -"""Run a job entrypoint and record an uncaught exception as plain data. - -This file runs inside the job process. It imports the standard library only, -because the job virtual environment does not hold ``syft_job``. - -The record is untrusted. The job controls this process, so the job can write any -string into the record or replace the file. ``syft_job.traceback_capture`` -checks every field before a party reads it. -""" - -import json -import os -import runpy -import sys -import traceback - -RAW_TRACE_PATH_ENV = "SYFT_JOB_RAW_TRACE_PATH" -MAX_CHAIN = 5 -MAX_FRAMES = 64 - - -def raw_record(exc: BaseException) -> dict: - """Return the exception chain as plain data. - - The record holds the class names and the frame positions. It never holds the - exception message, the source text, or the local variables. - """ - chain = [] - seen = set() - while exc is not None and id(exc) not in seen and len(chain) < MAX_CHAIN: - seen.add(id(exc)) - frames = traceback.extract_tb(exc.__traceback__)[:MAX_FRAMES] - chain.append( - { - "mro": [cls.__name__ for cls in type(exc).__mro__], - "frames": [ - {"filename": f.filename, "lineno": f.lineno} for f in frames - ], - } - ) - exc = exc.__cause__ or exc.__context__ - return {"chain": chain} - - -def main(argv: list) -> None: - entrypoint = argv[1] - sys.argv = argv[1:] - try: - runpy.run_path(entrypoint, run_name="__main__") - except SystemExit: - # A deliberate exit carries no failure position. - raise - except BaseException as exc: - # Observe the exception, then let it propagate unchanged. The process - # must keep the exit code and the stderr traceback it would have had. - path = os.environ.get(RAW_TRACE_PATH_ENV) - if path: - try: - with open(path, "w") as f: - json.dump(raw_record(exc), f) - except OSError: - pass - raise - - -if __name__ == "__main__": - main(sys.argv) diff --git a/packages/syft-job/src/syft_job/client.py b/packages/syft-job/src/syft_job/client.py index b7bc1e85e0f..1a7d7acaa9d 100644 --- a/packages/syft-job/src/syft_job/client.py +++ b/packages/syft-job/src/syft_job/client.py @@ -354,11 +354,7 @@ def _generate_python_run_script( uv sync --python {RUN_SCRIPT_PYTHON_VERSION} {install_deps_cmd} export PYTHONPATH=.:${{PYTHONPATH:-}} -if [ -n "${{SYFT_JOB_TRACE_RUNNER:-}}" ]; then - python "$SYFT_JOB_TRACE_RUNNER" {entrypoint_path} -else - python {entrypoint_path} -fi +python {entrypoint_path} """ else: # entrypoint_path is just the filename (e.g. "main.py") @@ -374,11 +370,7 @@ def _generate_python_run_script( source .venv/bin/activate uv pip install {deps_str} export PYTHONPATH=.:${{PYTHONPATH:-}} -if [ -n "${{SYFT_JOB_TRACE_RUNNER:-}}" ]; then - python "$SYFT_JOB_TRACE_RUNNER" {entrypoint_path} -else - python {entrypoint_path} -fi +python {entrypoint_path} """ def submit_python_job( diff --git a/packages/syft-job/src/syft_job/job.py b/packages/syft-job/src/syft_job/job.py index 8fa53145379..4a375fd7186 100644 --- a/packages/syft-job/src/syft_job/job.py +++ b/packages/syft-job/src/syft_job/job.py @@ -374,7 +374,10 @@ def rerun(self) -> None: # Clean up the artifacts of the previous run. A staged copy must go # too, or a later release sends a log that belongs to the old run. for filename in ("stdout.txt", "stderr.txt", "returncode.txt", FRAMES_FILENAME): - for f in (self.job_review_path / filename, self.job_staging_path / filename): + for f in ( + self.job_review_path / filename, + self.job_staging_path / filename, + ): if f.exists(): f.unlink() changes_made.append(filename) diff --git a/packages/syft-job/src/syft_job/job_runner.py b/packages/syft-job/src/syft_job/job_runner.py index c6260262f1a..23446b49cf4 100644 --- a/packages/syft-job/src/syft_job/job_runner.py +++ b/packages/syft-job/src/syft_job/job_runner.py @@ -1,7 +1,6 @@ import os import shutil import subprocess -import tempfile import time from datetime import datetime, timezone from pathlib import Path @@ -15,15 +14,7 @@ from .config import SyftJobConfig from .job_storage import JobRef, JobStorage, JobStateNotFoundError from .models import JobState, JobStatus, JobSubmissionMetadata -from .traceback_capture import ( - BUNDLE_FILENAME, - RAW_TRACE_FILENAME, - load_or_create_bundle, - RAW_TRACE_PATH_ENV, - TRACE_RUNNER_ENV, - trace_runner_path, - write_frames_record, -) +from .traceback_capture import write_frames_record # Default timeout for job execution (10 minutes) DEFAULT_JOB_TIMEOUT_SECONDS = 600 @@ -42,18 +33,6 @@ def get_job_timeout_seconds() -> int: IS_IN_JOB_ENV_VAR = "SYFT_IS_IN_JOB" -def _add_trace_env(env: dict, trace_dir: "Path | None") -> None: - """Point the job process at the traceback wrapper and its output file. - - Without these variables run.sh runs the entrypoint directly, so an older - run.sh and a newer run.sh both work. - """ - if trace_dir is None: - return - env[TRACE_RUNNER_ENV] = str(trace_runner_path()) - env[RAW_TRACE_PATH_ENV] = str(trace_dir / RAW_TRACE_FILENAME) - - def _kill_process_tree(pid: int, timeout: float = 2.0) -> None: """Kill `pid` and every descendant. Cross-platform via psutil.""" try: @@ -240,9 +219,7 @@ def _find_jobref_from_name(self, job_name: str, user: str | None = None) -> JobR self.config.current_user_email, job_name, ds_email=user ) - def _execute_job_streaming( - self, ref: JobRef, timeout: int, trace_dir: Path | None = None - ) -> int: + def _execute_job_streaming(self, ref: JobRef, timeout: int) -> int: """Execute job with real-time streaming output. Reads run.sh from inbox/, writes stdout/stderr to staging/. @@ -263,7 +240,6 @@ def _execute_job_streaming( env["SYFTBOX_EMAIL"] = self.config.current_user_email env[IS_IN_JOB_ENV_VAR] = "true" env["PYTHONUNBUFFERED"] = "1" - _add_trace_env(env, trace_dir) staging_dir = self.manager.staging_dir(ref) staging_dir.mkdir(parents=True, exist_ok=True) @@ -333,9 +309,7 @@ def _execute_job_streaming( return returncode - def _execute_job_captured( - self, ref: JobRef, timeout: int, trace_dir: Path | None = None - ) -> int: + def _execute_job_captured(self, ref: JobRef, timeout: int) -> int: """Execute job with captured output (non-streaming). Reads run.sh from inbox/, writes stdout/stderr to staging/. @@ -353,7 +327,6 @@ def _execute_job_captured( env["SYFTBOX_EMAIL"] = self.config.current_user_email env[IS_IN_JOB_ENV_VAR] = "true" env["PYTHONUNBUFFERED"] = "1" - _add_trace_env(env, trace_dir) process = subprocess.Popen( ["bash", str(run_script)], @@ -429,27 +402,13 @@ def _execute_job( state.status = JobStatus.RUNNING self.manager.write_state(ref, state) - trace_dir = Path(tempfile.mkdtemp(prefix="syft-job-trace-")) - # The job writes into code/ while it runs, so the record must come from - # the tree as it stood before the first run. - code_bundle = load_or_create_bundle( - self.manager.staging_dir(ref) / BUNDLE_FILENAME, - submission_dir / "code", - ) try: if stream_output: - returncode = self._execute_job_streaming(ref, timeout, trace_dir) + returncode = self._execute_job_streaming(ref, timeout) else: - returncode = self._execute_job_captured(ref, timeout, trace_dir) - - # The job writes the raw record, so check it before any party reads - # it. A job that writes nothing produces no file. - write_frames_record( - trace_dir / RAW_TRACE_FILENAME, - self.manager.staging_dir(ref), - submission_dir / "code", - code_bundle, - ) + returncode = self._execute_job_captured(ref, timeout) + + self._capture_traceback(ref, returncode) # Move outputs from inbox/ to review/ self._move_outputs_to_review(submission_dir, review_dir) @@ -489,8 +448,19 @@ def _execute_job( self._set_finalized_job_state(ref, -1) return False - finally: - shutil.rmtree(trace_dir, ignore_errors=True) + def _capture_traceback(self, ref: JobRef, returncode: int) -> None: + """Stage where the job failed, from the traceback it printed. + + A run that succeeds prints no traceback, so it leaves no record. + """ + if returncode == 0: + return + staging_dir = self.manager.staging_dir(ref) + write_frames_record( + staging_dir / "stderr.txt", + staging_dir, + self.manager.submission_dir(ref) / "code", + ) def _set_finalized_job_state(self, ref: JobRef, returncode: int) -> None: state = self.manager.read_state(ref) diff --git a/packages/syft-job/src/syft_job/traceback_capture.py b/packages/syft-job/src/syft_job/traceback_capture.py index c0455ffa24d..52f99b3a34a 100644 --- a/packages/syft-job/src/syft_job/traceback_capture.py +++ b/packages/syft-job/src/syft_job/traceback_capture.py @@ -1,248 +1,104 @@ -"""Turn an untrusted traceback record into a bounded one. +"""Record where a job failed, from the traceback the job printed to stderr. -The job process writes the raw record, so a job chooses every string in it. A -job can put private data in a file name, in a function name, or in an exception -message. Therefore only these fields reach a party: - -- a path that resolves to a file in the approved code bundle, or a fixed label; -- a line number inside that file; -- an exception type that is a builtin. - -The function names, the exception messages, and the local variables never reach -a party. +The record holds the file and the line of each frame, and the exception type. +It never holds the exception message, because ``traceback_frames`` discloses +where a job failed and ``logs`` discloses what the job wrote. A party that +needs the message asks for ``logs`` as well. """ from __future__ import annotations -import builtins import json -import os +import re from pathlib import Path from typing import Optional -from syft_job._trace_runner import RAW_TRACE_PATH_ENV # noqa: F401 (re-export) - -RAW_TRACE_FILENAME = "_raw_traceback.json" FRAMES_FILENAME = "traceback_frames.json" -TRACE_RUNNER_ENV = "SYFT_JOB_TRACE_RUNNER" - -EXTERNAL_LABEL = "" FALLBACK_TYPE = "Exception" -MAX_CHAIN = 5 -MAX_FRAMES = 64 -BUNDLE_FILENAME = "code_bundle.json" -RUNNER_DIRS = frozenset({".venv", "__pycache__"}) -MAX_BUNDLE_FILES = 2000 -MAX_BUNDLE_FILE_BYTES = 5 * 1024 * 1024 +# Size limits, not disclosure limits. Deep recursion prints thousands of +# frames, and the record is a summary of where the job stopped. +MAX_CHAIN = 5 +MAX_FRAMES = 50 -def trace_runner_path() -> Path: - """Return the path of the wrapper script that the job process runs.""" - return Path(__file__).with_name("_trace_runner.py") - - -def _builtin_exception_name(mro: object) -> str: - """Return the first name in ``mro`` that names a builtin exception. - - The class names come from the job, so each name must match a builtin before - it reaches a party. A custom class therefore reports its nearest builtin - base, such as ``ValueError`` for ``class Secret(ValueError)``. - """ - if not isinstance(mro, list): - return FALLBACK_TYPE - for name in mro: - if not isinstance(name, str): - continue - cls = getattr(builtins, name, None) - if isinstance(cls, type) and issubclass(cls, BaseException): - return name - return FALLBACK_TYPE - - -def _line_count(path: Path) -> int: - with open(path, "rb") as f: - return sum(1 for _ in f) - +TRACEBACK_HEADER = "Traceback (most recent call last):" +# Python 3.13 and later colour the traceback when the environment asks for it. +_ANSI = re.compile(r"\x1b\[[0-9;]*[A-Za-z]") +_FRAME = re.compile(r'^ File "(?P.+)", line (?P\d+)', re.M) +_TYPE = re.compile(r"^(?P[A-Za-z_][A-Za-z0-9_.]*)(?::|$)", re.M) -def snapshot_code_bundle(code_root: Path) -> dict[str, int]: - """Record the approved files and their lengths, before the job runs. - Returns a map of a path relative to ``code_root`` to a line count. +def _frame_file(raw: str, code_root: Path) -> str: + """The frame path, relative to ``code_root`` when it points inside it. - The job writes into its own code directory: ``run.sh`` builds a virtual - environment there. A job can therefore plant a file whose name carries - private data, then raise from it. The record must come from the tree the - parties approved, so this runs before execution. + The job runs with ``code/`` as its working directory, so a frame in the + submitted code is already relative. A frame in a library is not, and it + keeps the path the interpreter printed. """ - bundle: dict[str, int] = {} - if not code_root.is_dir(): - return bundle + path = Path(raw) + if not path.is_absolute(): + path = code_root / raw try: - root = code_root.resolve() - except OSError: - return bundle - - for path in sorted(root.rglob("*")): - if len(bundle) >= MAX_BUNDLE_FILES: - break - if not path.is_file() or path.is_symlink(): - continue - name = path.relative_to(root).as_posix() - # run.sh builds the virtual environment here, so these paths belong to - # the runner and never to the approved bundle. - if any(part in RUNNER_DIRS for part in name.split("/")[:-1]): - continue - try: - if path.stat().st_size > MAX_BUNDLE_FILE_BYTES: - continue - bundle[name] = _line_count(path) - except OSError: - continue - return bundle - - -def _safe_frame(frame: object, code_root: Path, bundle: dict) -> dict: - """Return one frame, or ```` when ``bundle`` does not hold it.""" - external = {"file": EXTERNAL_LABEL, "line": None} - if not isinstance(frame, dict): - return external - - filename = frame.get("filename") - lineno = frame.get("lineno") - if not isinstance(filename, str) or not filename: - return external + return path.resolve().relative_to(code_root).as_posix() + except (OSError, ValueError): + return raw - try: - if os.path.isabs(filename): - candidate = Path(filename).resolve() - else: - candidate = (code_root / filename).resolve() - except OSError: - return external - if not candidate.is_relative_to(code_root): - return external +def _frames(block: str, code_root: Path) -> list[dict]: + return [ + {"file": _frame_file(m["file"], code_root), "line": int(m["line"])} + for m in list(_FRAME.finditer(block))[:MAX_FRAMES] + ] - name = candidate.relative_to(code_root).as_posix() - total = bundle.get(name) - if total is None: - # The job planted this file after the parties approved the bundle. - return external - if not isinstance(lineno, int) or isinstance(lineno, bool): - return external - if lineno < 1 or lineno > total: - # A line outside the file means the job forged the position. - return external - return {"file": name, "line": lineno} +def _exception_type(block: str) -> str: + """The type name on the line that closes a traceback block.""" + tail = block[max((m.end() for m in _FRAME.finditer(block)), default=0) :] + match = _TYPE.search(tail) + return match["type"] if match else FALLBACK_TYPE -def sanitize_raw_trace( - raw: object, code_root: Path, bundle: dict[str, int] -) -> Optional[dict]: - """Return the bounded record, or None when ``raw`` holds no usable chain. +def frames_from_stderr(stderr: str, code_root: Path) -> Optional[dict]: + """Return the failure positions in ``stderr``, or None when it holds none. - ``code_root`` is the directory of the approved code bundle, and ``bundle`` - is the snapshot that ``snapshot_code_bundle`` took before the job ran. A - frame counts as safe only when the snapshot holds its file and its line. + Python prints a chained exception oldest first, therefore the chain is + reversed: ``chain[0]`` is the exception that stopped the job. """ - if not isinstance(raw, dict): + blocks = _ANSI.sub("", stderr).split(TRACEBACK_HEADER)[1:] + if not blocks: return None - chain = raw.get("chain") - if not isinstance(chain, list) or not chain: - return None - try: root = code_root.resolve() except OSError: - return None - - safe_chain = [] - for entry in chain[:MAX_CHAIN]: - if not isinstance(entry, dict): - continue - frames = entry.get("frames") - frames = frames[:MAX_FRAMES] if isinstance(frames, list) else [] - safe_chain.append( - { - "type": _builtin_exception_name(entry.get("mro")), - "frames": [_safe_frame(f, root, bundle) for f in frames], - } - ) - - if not safe_chain: - return None - return {"chain": safe_chain} + root = code_root + return { + "chain": [ + {"type": _exception_type(b), "frames": _frames(b, root)} + for b in reversed(blocks[-MAX_CHAIN:]) + ] + } def write_frames_record( - raw_path: Path, out_dir: Path, code_root: Path, bundle: dict[str, int] + stderr_path: Path, out_dir: Path, code_root: Path ) -> Optional[Path]: - """Read the raw record, check it, and write ``traceback_frames.json``. + """Read the staged stderr and write ``traceback_frames.json`` beside it. - ``out_dir`` is the staging directory, therefore the record waits there until - a party releases it. + ``out_dir`` is the staging directory, therefore the record waits there + until a party releases it. - Returns the path of the written file, or None when the job wrote no usable - record. A job that writes nothing, or writes invalid JSON, produces no file. + Returns the path written, or None when the job printed no traceback. """ - if not raw_path.is_file(): - return None try: - with open(raw_path, "r") as f: - raw = json.load(f) - except (OSError, json.JSONDecodeError): + stderr = stderr_path.read_text(errors="replace") + except OSError: return None - record = sanitize_raw_trace(raw, code_root, bundle) + record = frames_from_stderr(stderr, code_root) if record is None: return None out_dir.mkdir(parents=True, exist_ok=True) out_path = out_dir / FRAMES_FILENAME - with open(out_path, "w") as f: - json.dump(record, f, indent=2) - return out_path - - -def load_or_create_bundle(bundle_path: Path, code_root: Path) -> dict[str, int]: - """Return the approved tree, recorded once before the first run. - - A rerun executes the same bundle, and the first run leaves files behind. - Reading a stored record therefore keeps a later run from approving a file - that the parties never saw. - """ - if bundle_path.is_file(): - try: - stored = json.loads(bundle_path.read_text()) - except (OSError, json.JSONDecodeError): - stored = None - if isinstance(stored, dict): - return { - name: count - for name, count in stored.items() - if isinstance(name, str) - and isinstance(count, int) - and not isinstance(count, bool) - } - - bundle = snapshot_code_bundle(code_root) - try: - bundle_path.parent.mkdir(parents=True, exist_ok=True) - bundle_path.write_text(json.dumps(bundle)) - except OSError: - pass - return bundle - - -def write_no_failure_record(out_dir: Path) -> Path: - """Write a record that says the run did not fail. - - An empty chain never comes from a crash, because a record with no chain is - dropped. It therefore marks a run that produced no failure. - """ - out_dir.mkdir(parents=True, exist_ok=True) - out_path = out_dir / FRAMES_FILENAME - with open(out_path, "w") as f: - json.dump({"chain": []}, f, indent=2) + out_path.write_text(json.dumps(record, indent=2)) return out_path diff --git a/packages/syft-job/tests/test_staging.py b/packages/syft-job/tests/test_staging.py index 706f1db60ab..59216689526 100644 --- a/packages/syft-job/tests/test_staging.py +++ b/packages/syft-job/tests/test_staging.py @@ -3,16 +3,11 @@ import json 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.traceback_capture import ( - BUNDLE_FILENAME, - FRAMES_FILENAME, - load_or_create_bundle, -) +from syft_job.traceback_capture import FRAMES_FILENAME DO_EMAIL = "do@test.org" DS_EMAIL = "ds@test.org" @@ -145,28 +140,3 @@ def test_rerun_clears_the_crash_record(tmp_path): assert not (staging / FRAMES_FILENAME).exists() assert not (review / FRAMES_FILENAME).exists() - - -def test_a_rerun_keeps_the_bundle_the_parties_approved(tmp_path): - """The first run leaves .venv and any file the job wrote in code/. - - A rerun must judge frames against the tree as it stood before the first - run, or a planted file counts as approved and the real sources drop out. - """ - job, review, staging = run_job(tmp_path, OK_PY, share_logs=False) - bundle_path = staging / BUNDLE_FILENAME - assert bundle_path.exists() - - recorded = json.loads(bundle_path.read_text()) - assert "main.py" in recorded - assert not any(name.startswith(".venv") for name in recorded) - - # The first run really did leave a virtual environment behind. - code_dir = job.job_submission_path / "code" - assert (code_dir / ".venv").is_dir() - - # A job that plants a file gains nothing on the next run. - (code_dir / "DO2_row_4171_POSITIVE.py").write_text("\n" * 50) - reloaded = load_or_create_bundle(bundle_path, code_dir) - assert "DO2_row_4171_POSITIVE.py" not in reloaded - assert reloaded == recorded diff --git a/packages/syft-job/tests/test_traceback_capture.py b/packages/syft-job/tests/test_traceback_capture.py index 968b1360b4e..e846900db9a 100644 --- a/packages/syft-job/tests/test_traceback_capture.py +++ b/packages/syft-job/tests/test_traceback_capture.py @@ -1,278 +1,179 @@ -"""The sanitizer treats the raw record as attacker-controlled input.""" +"""``traceback_frames`` reports where a job failed, and never the message.""" import json import subprocess import sys -from pathlib import Path import pytest -from syft_job._trace_runner import raw_record from syft_job.traceback_capture import ( - EXTERNAL_LABEL, FALLBACK_TYPE, FRAMES_FILENAME, MAX_CHAIN, MAX_FRAMES, - sanitize_raw_trace, - snapshot_code_bundle, - trace_runner_path, + frames_from_stderr, write_frames_record, ) +CRASH = """\ +def inner(): + raise ValueError("patient 4171 is positive") -def sanitize(raw, code_root): - """Sanitize against a snapshot taken from the current tree.""" - return sanitize_raw_trace(raw, code_root, snapshot_code_bundle(code_root)) + +def outer(): + inner() + + +outer() +""" + +CHAINED = """\ +try: + raise KeyError("patient 4171") +except KeyError as exc: + raise RuntimeError("wrapped") from exc +""" @pytest.fixture def code_root(tmp_path): root = tmp_path / "code" root.mkdir() - (root / "main.py").write_text("\n".join(f"line {i}" for i in range(1, 21))) return root -def frame(filename, lineno): - return {"filename": filename, "lineno": lineno} - - -def chain(mro, frames): - return {"chain": [{"mro": mro, "frames": frames}]} - - -# -- what a reader is allowed to see ------------------------------------------ - - -def test_frame_in_the_bundle_keeps_its_file_and_line(code_root): - out = sanitize(chain(["ValueError"], [frame("main.py", 12)]), code_root) - assert out["chain"][0]["frames"] == [{"file": "main.py", "line": 12}] - assert out["chain"][0]["type"] == "ValueError" - - -def test_absolute_path_in_the_bundle_is_relative_in_the_record(code_root): - raw = chain(["KeyError"], [frame(str(code_root / "main.py"), 3)]) - out = sanitize(raw, code_root) - assert out["chain"][0]["frames"] == [{"file": "main.py", "line": 3}] - - -def test_the_record_never_holds_a_message_or_a_function_name(code_root): - raw = chain(["ValueError"], [frame("main.py", 1)]) - raw["chain"][0]["frames"][0]["name"] = "leak_do2_row_4171" - raw["chain"][0]["message"] = "patient 4171 is positive" - out = sanitize(raw, code_root) - assert set(out["chain"][0]) == {"type", "frames"} - assert set(out["chain"][0]["frames"][0]) == {"file", "line"} - - -# -- smuggling attempts -------------------------------------------------------- - - -def test_a_forged_file_name_never_reaches_the_record(code_root): - """A job picks the file name of a frame through compile().""" - raw = chain(["ValueError"], [frame("DO2_row_4171_diagnosis_POSITIVE.py", 1)]) - out = sanitize(raw, code_root) - assert out["chain"][0]["frames"] == [{"file": EXTERNAL_LABEL, "line": None}] - - -def test_a_path_outside_the_bundle_is_labelled(code_root, tmp_path): - outside = tmp_path / "secret.py" - outside.write_text("x = 1\n") - out = sanitize(chain(["OSError"], [frame(str(outside), 1)]), code_root) - assert out["chain"][0]["frames"] == [{"file": EXTERNAL_LABEL, "line": None}] - - -def test_a_traversal_path_stays_outside(code_root): - raw = chain(["OSError"], [frame("../../../etc/passwd", 1)]) - out = sanitize(raw, code_root) - assert out["chain"][0]["frames"][0]["file"] == EXTERNAL_LABEL - - -def test_a_line_past_the_end_of_the_file_is_dropped(code_root): - """main.py holds 20 lines, so line 99999 is a forged position.""" - out = sanitize(chain(["ValueError"], [frame("main.py", 99999)]), code_root) - assert out["chain"][0]["frames"] == [{"file": EXTERNAL_LABEL, "line": None}] +def run(code_root, source): + """Run a script the way run.sh does, from inside code/, and return stderr.""" + (code_root / "main.py").write_text(source) + proc = subprocess.run( + [sys.executable, "main.py"], + cwd=code_root, + capture_output=True, + text=True, + ) + assert proc.returncode != 0 + return proc.stderr -def test_a_custom_exception_reports_its_builtin_base(code_root): - raw = chain(["DO2DataWasPositive", "ValueError", "Exception"], []) - out = sanitize(raw, code_root) - assert out["chain"][0]["type"] == "ValueError" +# -- what the record holds ------------------------------------------------------ -def test_an_all_custom_mro_falls_back(code_root): - out = sanitize(chain(["SecretName", "AlsoSecret"], []), code_root) - assert out["chain"][0]["type"] == FALLBACK_TYPE +def test_record_holds_every_frame_of_the_crash(code_root): + record = frames_from_stderr(run(code_root, CRASH), code_root) + assert record["chain"][0]["type"] == "ValueError" + assert record["chain"][0]["frames"] == [ + {"file": "main.py", "line": 9}, + {"file": "main.py", "line": 6}, + {"file": "main.py", "line": 2}, + ] -def test_the_chain_and_the_frame_count_are_bounded(code_root): - raw = { - "chain": [ - {"mro": ["ValueError"], "frames": [frame("main.py", 1)] * 500} - for _ in range(50) - ] - } - out = sanitize(raw, code_root) - assert len(out["chain"]) == MAX_CHAIN - assert len(out["chain"][0]["frames"]) == MAX_FRAMES +def test_record_never_holds_the_exception_message(code_root): + record = frames_from_stderr(run(code_root, CRASH), code_root) + assert "positive" not in json.dumps(record) + assert "4171" not in json.dumps(record) -@pytest.mark.parametrize( - "raw", - [None, {}, {"chain": []}, {"chain": "nope"}, [1, 2, 3], {"chain": [None]}], -) -def test_malformed_input_yields_no_record(raw, code_root): - assert sanitize(raw, code_root) in (None, {"chain": []}) +def test_frame_holds_only_a_file_and_a_line(code_root): + record = frames_from_stderr(run(code_root, CRASH), code_root) + assert set(record["chain"][0]) == {"type", "frames"} + assert set(record["chain"][0]["frames"][0]) == {"file", "line"} -def test_a_non_integer_line_is_dropped(code_root): - out = sanitize(chain(["ValueError"], [frame("main.py", "12")]), code_root) - assert out["chain"][0]["frames"][0]["line"] is None +def test_chained_exception_puts_the_last_failure_first(code_root): + record = frames_from_stderr(run(code_root, CHAINED), code_root) + assert [entry["type"] for entry in record["chain"]] == ["RuntimeError", "KeyError"] + assert "patient" not in json.dumps(record) -# -- the file the parent writes ------------------------------------------------- +def test_library_frame_keeps_the_path_the_interpreter_printed(code_root): + source = "import json\njson.loads('{')\n" + record = frames_from_stderr(run(code_root, source), code_root) + files = [f["file"] for f in record["chain"][0]["frames"]] + assert files[0] == "main.py" + assert any(f.endswith("json/decoder.py") for f in files) -def test_write_frames_record_writes_only_checked_fields(tmp_path, code_root): - raw_path = tmp_path / "raw.json" - review = tmp_path / "review" - raw_path.write_text( - json.dumps(chain(["ValueError"], [frame("main.py", 2), frame("/etc/shadow", 1)])) - ) - out = write_frames_record( - raw_path, review, code_root, snapshot_code_bundle(code_root) - ) - assert out == review / FRAMES_FILENAME - record = json.loads(out.read_text()) - assert record["chain"][0]["frames"] == [ - {"file": "main.py", "line": 2}, - {"file": EXTERNAL_LABEL, "line": None}, - ] +# -- input that carries no failure ---------------------------------------------- -def test_no_raw_file_means_no_record(tmp_path, code_root): - review = tmp_path / "review" - bundle = snapshot_code_bundle(code_root) - assert ( - write_frames_record(tmp_path / "absent.json", review, code_root, bundle) is None +def test_clean_run_leaves_no_record(code_root): + (code_root / "main.py").write_text("print('ok')\n") + proc = subprocess.run( + [sys.executable, "main.py"], cwd=code_root, capture_output=True, text=True ) - assert not (review / FRAMES_FILENAME).exists() + assert proc.returncode == 0 + assert frames_from_stderr(proc.stderr, code_root) is None -def test_invalid_json_means_no_record(tmp_path, code_root): - raw_path = tmp_path / "raw.json" - raw_path.write_text("{not json") - bundle = snapshot_code_bundle(code_root) - assert ( - write_frames_record(raw_path, tmp_path / "review", code_root, bundle) is None +def test_deliberate_exit_leaves_no_record(code_root): + source = "import sys\nsys.exit(3)\n" + (code_root / "main.py").write_text(source) + proc = subprocess.run( + [sys.executable, "main.py"], cwd=code_root, capture_output=True, text=True ) + assert proc.returncode == 3 + assert frames_from_stderr(proc.stderr, code_root) is None -# -- the wrapper that runs inside the job --------------------------------------- - +@pytest.mark.parametrize("stderr", ["", "some log output\n", "Traceback: not really\n"]) +def test_stderr_without_a_traceback_yields_no_record(stderr, code_root): + assert frames_from_stderr(stderr, code_root) is None -def test_raw_record_holds_no_message(): - try: - raise ValueError("patient 4171 is positive") - except ValueError as exc: - record = raw_record(exc) - blob = json.dumps(record) - assert "positive" not in blob - assert record["chain"][0]["mro"][0] == "ValueError" +def test_install_output_before_the_traceback_is_ignored(code_root): + noisy = "+ pandas==2.0.0\n+ numpy==1.26\n" + run(code_root, CRASH) + record = frames_from_stderr(noisy, code_root) + assert record["chain"][0]["type"] == "ValueError" -def test_raw_record_follows_a_chained_exception(): - try: - try: - raise KeyError("inner") - except KeyError as inner: - raise RuntimeError("outer") from inner - except RuntimeError as exc: - record = raw_record(exc) - assert [c["mro"][0] for c in record["chain"]] == ["RuntimeError", "KeyError"] - assert "inner" not in json.dumps(record) +# -- size limits ---------------------------------------------------------------- -def test_the_wrapper_records_a_crash_and_keeps_the_exit_code(tmp_path): - entry = tmp_path / "main.py" - entry.write_text("raise ValueError('secret value')\n") - raw_path = tmp_path / "raw.json" - proc = subprocess.run( - [sys.executable, str(trace_runner_path()), str(entry)], - capture_output=True, - text=True, - env={"PATH": "/usr/bin:/bin", "SYFT_JOB_RAW_TRACE_PATH": str(raw_path)}, +def test_chain_and_frame_count_are_capped(code_root): + block = ( + "Traceback (most recent call last):\n" + + "".join(f' File "main.py", line {i}, in f\n' for i in range(1, 200)) + + "ValueError\n" ) + record = frames_from_stderr(block * 20, code_root) + assert len(record["chain"]) == MAX_CHAIN + assert len(record["chain"][0]["frames"]) == MAX_FRAMES - assert proc.returncode != 0 - assert "ValueError" in proc.stderr # the real traceback still reaches stderr - record = json.loads(raw_path.read_text()) - assert record["chain"][0]["mro"][0] == "ValueError" - assert "secret value" not in raw_path.read_text() - out = sanitize_raw_trace(record, tmp_path, snapshot_code_bundle(tmp_path)) - assert out["chain"][0]["frames"][-1] == {"file": "main.py", "line": 1} - - -def test_the_wrapper_leaves_a_clean_run_alone(tmp_path): - entry = tmp_path / "main.py" - entry.write_text("print('ok')\n") - raw_path = tmp_path / "raw.json" - proc = subprocess.run( - [sys.executable, str(trace_runner_path()), str(entry)], - capture_output=True, - text=True, - env={"PATH": "/usr/bin:/bin", "SYFT_JOB_RAW_TRACE_PATH": str(raw_path)}, +def test_a_coloured_traceback_parses(code_root): + """Python 3.13 and later colour the traceback when the environment asks.""" + coloured = ( + "Traceback (most recent call last):\n" + ' File \x1b[35m"main.py"\x1b[0m, line \x1b[35m2\x1b[0m, in \x1b[35mf\x1b[0m\n' + "\x1b[1;35mValueError\x1b[0m: \x1b[35mpatient 4171\x1b[0m\n" ) - assert proc.returncode == 0 - assert "ok" in proc.stdout - assert not raw_path.exists() + record = frames_from_stderr(coloured, code_root) + assert record["chain"][0]["type"] == "ValueError" + assert record["chain"][0]["frames"] == [{"file": "main.py", "line": 2}] + assert "4171" not in json.dumps(record) -def test_a_deliberate_exit_writes_no_record(tmp_path): - entry = tmp_path / "main.py" - entry.write_text("import sys; sys.exit(3)\n") - raw_path = tmp_path / "raw.json" - proc = subprocess.run( - [sys.executable, str(trace_runner_path()), str(entry)], - capture_output=True, - text=True, - env={"PATH": "/usr/bin:/bin", "SYFT_JOB_RAW_TRACE_PATH": str(raw_path)}, - ) - assert proc.returncode == 3 - assert not raw_path.exists() - +def test_a_block_without_a_type_line_falls_back(code_root): + block = 'Traceback (most recent call last):\n File "main.py", line 1, in f\n' + record = frames_from_stderr(block, code_root) + assert record["chain"][0]["type"] == FALLBACK_TYPE -def test_a_file_the_job_plants_after_approval_is_external(code_root): - """run.sh builds a venv inside code/, so the job can write there.""" - bundle = snapshot_code_bundle(code_root) - planted = code_root / "DO2_row_4171_diagnosis_POSITIVE.py" - planted.write_text("\n" * 50) +# -- the file the runner writes ------------------------------------------------- - raw = chain(["ValueError"], [frame(planted.name, 42)]) - out = sanitize_raw_trace(raw, code_root, bundle) - assert out["chain"][0]["frames"] == [{"file": EXTERNAL_LABEL, "line": None}] +def test_write_frames_record_writes_the_staged_file(tmp_path, code_root): + stderr_path = tmp_path / "stderr.txt" + stderr_path.write_text(run(code_root, CRASH)) + staging = tmp_path / "staging" -def test_a_file_the_job_grows_keeps_its_approved_length(code_root): - """A longer file must not widen the range of accepted line numbers.""" - bundle = snapshot_code_bundle(code_root) - (code_root / "main.py").write_text("\n" * 5000) + out = write_frames_record(stderr_path, staging, code_root) + assert out == staging / FRAMES_FILENAME + assert json.loads(out.read_text())["chain"][0]["type"] == "ValueError" - out = sanitize_raw_trace(chain(["ValueError"], [frame("main.py", 900)]), code_root, bundle) - assert out["chain"][0]["frames"] == [{"file": EXTERNAL_LABEL, "line": None}] - -def test_a_venv_file_is_external(code_root): - bundle = snapshot_code_bundle(code_root) - venv_file = code_root / ".venv" / "lib" / "evil.py" - venv_file.parent.mkdir(parents=True) - venv_file.write_text("x = 1\n") - - raw = chain(["ValueError"], [frame(".venv/lib/evil.py", 1)]) - assert sanitize_raw_trace(raw, code_root, bundle)["chain"][0]["frames"] == [ - {"file": EXTERNAL_LABEL, "line": None} - ] +def test_no_stderr_file_means_no_record(tmp_path, code_root): + staging = tmp_path / "staging" + assert write_frames_record(tmp_path / "absent.txt", staging, code_root) is None + assert not (staging / FRAMES_FILENAME).exists() From 5aa81285f563cf93df7442c6345fd5fe923b54f4 Mon Sep 17 00:00:00 2001 From: Pedro Werneck Date: Fri, 18 Sep 2026 17:19:26 -0300 Subject: [PATCH 3/3] fix: traceback record and disclosure grant defects - Drop a traceback block with no frames and read the type only after the last frame, so job output cannot land in the record as the exception type. - Match frames behind an ExceptionGroup gutter. - Keep the innermost frames when the frame cap truncates. - Reject a non-enclave job in approve_job, which took disclosures as approve's reason and granted nothing. - Honour a False value in a map passed to normalize_disclosures. --- .../syft-enclave/src/syft_enclaves/client.py | 7 +- .../src/syft_enclaves/enclave_job_info.py | 9 +- .../syft-enclave/tests/test_disclosures.py | 39 ++++++++- .../src/syft_job/traceback_capture.py | 57 ++++++++++--- packages/syft-job/tests/test_staging.py | 4 +- .../syft-job/tests/test_traceback_capture.py | 85 +++++++++++++++++-- 6 files changed, 175 insertions(+), 26 deletions(-) diff --git a/packages/syft-enclave/src/syft_enclaves/client.py b/packages/syft-enclave/src/syft_enclaves/client.py index dbba3770ecb..e598f21a8c1 100644 --- a/packages/syft-enclave/src/syft_enclaves/client.py +++ b/packages/syft-enclave/src/syft_enclaves/client.py @@ -386,10 +386,15 @@ def approve_job( owner releases. The enclave releases an item only when every data owner released it. Omit the argument to release nothing. """ + if not isinstance(job, EnclaveJobInfo): + raise TypeError( + f"Job '{job.name}' is not an enclave job, so it carries no " + f"disclosures. Approve it through the datasite client." + ) if pre_sync_enabled(): self._rds.sync() - job.approve(disclosures) + job.approve(disclosures=disclosures) file_name = enclave_approval_file_name(self.email) approval_file = job.job_review_path / file_name if not approval_file.exists(): diff --git a/packages/syft-enclave/src/syft_enclaves/enclave_job_info.py b/packages/syft-enclave/src/syft_enclaves/enclave_job_info.py index 4ec2caa34e0..0b0feac0cc4 100644 --- a/packages/syft-enclave/src/syft_enclaves/enclave_job_info.py +++ b/packages/syft-enclave/src/syft_enclaves/enclave_job_info.py @@ -2,6 +2,7 @@ import json from datetime import datetime, timezone +from collections.abc import Mapping from enum import Enum from pathlib import Path from typing import Iterable, Optional, Union @@ -27,17 +28,23 @@ class DisclosureItem(str, Enum): def normalize_disclosures( - items: Union[str, DisclosureItem, Iterable[str], None], + items: Union[str, DisclosureItem, Iterable[str], Mapping[str, bool], None], ) -> dict[str, bool]: """Return the known items in ``items`` as a map. Unknown names are dropped. A single name is accepted on its own, because iterating a string would produce its characters and grant nothing. + + A map is accepted in the form that this function returns, therefore a + caller can read the current grant, set an item to False, and send it back + to drop that item. """ if not items: return {} if isinstance(items, (str, DisclosureItem)): items = [items] + elif isinstance(items, Mapping): + items = [name for name, allowed in items.items() if allowed] names = {str(getattr(i, "value", i)) for i in items} return {name: True for name in sorted(names & DISCLOSURE_ITEMS)} diff --git a/packages/syft-enclave/tests/test_disclosures.py b/packages/syft-enclave/tests/test_disclosures.py index 38c9c3d405f..5f392652757 100644 --- a/packages/syft-enclave/tests/test_disclosures.py +++ b/packages/syft-enclave/tests/test_disclosures.py @@ -16,6 +16,7 @@ from syft_enclaves import SyftEnclaveClient # noqa: E402 from syft_enclaves.enclave_job_info import ( # noqa: E402 DisclosureItem, + EnclaveJobInfo, PartyApprovalStatus, approved_disclosures, enclave_approval_file_name, @@ -79,6 +80,18 @@ def test_an_unknown_item_never_survives(tmp_path): assert normalize_disclosures(["everything", LOGS]) == {LOGS: True} +def test_a_mapping_with_a_false_value_drops_the_item(tmp_path): + """update_disclosures returns a map, so a caller can send one back. + + Every element used to count as a grant, therefore a map carrying False + re-granted the item it was meant to drop. + """ + assert normalize_disclosures({LOGS: True, FRAMES: False}) == {LOGS: True} + + granted = normalize_disclosures([LOGS]) + assert normalize_disclosures({**granted, LOGS: False}) == {} + + # -- end to end ----------------------------------------------------------------- @@ -119,7 +132,7 @@ def submit(ds, enclave, do1, do2, code, requested): OK_CODE = "import os, json\nos.makedirs('outputs', exist_ok=True)\nopen('outputs/r.json','w').write('{}')\n" -CRASH_CODE = "x = 1\ny = 2\nraise ValueError('patient 4171 is positive')\n" +CRASH_CODE = "x = 1\ny = 2\nraise ValueError('account 88213 holds 4120550')\n" def run_to_completion(enclave, do1, do2, ds, grants): @@ -165,7 +178,7 @@ def test_granted_frames_carry_the_position_but_not_the_message(): entry = record["chain"][0] assert entry["type"] == "ValueError" assert {"file": "main.py", "line": 3} in entry["frames"] - assert "positive" not in json.dumps(record) + assert "4120550" not in json.dumps(record) def test_frames_reach_the_data_owners_too(): @@ -294,3 +307,25 @@ def test_a_released_artifact_is_not_sent_twice(): job = enclave.jobs["j"] assert enclave._forward_new_releases(job) == [] + + +def test_approve_job_rejects_a_job_that_is_not_an_enclave_job(): + """JobInfo.approve takes a reason first, so disclosures would land there. + + SyftEnclaveClient.jobs wraps only a job whose job_type header says enclave. + An unwrapped job used to record the disclosures as its approval reason and + grant nothing, with no error. + """ + enclave, do1, do2, ds = build_quad() + submit(ds, enclave, do1, do2, OK_CODE, [LOGS]) + enclave.sync() + enclave.receive_jobs() + do1.sync() + + # The same job, read through the plain client, is a JobInfo and not an + # EnclaveJobInfo. + plain = do1._rds.job_client.jobs["j"] + assert not isinstance(plain, EnclaveJobInfo) + + with pytest.raises(TypeError, match="not an enclave job"): + do1.approve_job(plain, [LOGS]) diff --git a/packages/syft-job/src/syft_job/traceback_capture.py b/packages/syft-job/src/syft_job/traceback_capture.py index 52f99b3a34a..830eeadf73f 100644 --- a/packages/syft-job/src/syft_job/traceback_capture.py +++ b/packages/syft-job/src/syft_job/traceback_capture.py @@ -24,8 +24,10 @@ TRACEBACK_HEADER = "Traceback (most recent call last):" # Python 3.13 and later colour the traceback when the environment asks for it. _ANSI = re.compile(r"\x1b\[[0-9;]*[A-Za-z]") -_FRAME = re.compile(r'^ File "(?P.+)", line (?P\d+)', re.M) -_TYPE = re.compile(r"^(?P[A-Za-z_][A-Za-z0-9_.]*)(?::|$)", re.M) +# An ExceptionGroup prints each frame behind a "|" gutter, one level per nesting. +_FRAME = re.compile(r'^[ |]+File "(?P.+)", line (?P\d+)') +_GUTTER = re.compile(r"^[ |]*") +_TYPE = re.compile(r"^(?P[A-Za-z_][A-Za-z0-9_.]*)(?::|$)") def _frame_file(raw: str, code_root: Path) -> str: @@ -44,18 +46,45 @@ def _frame_file(raw: str, code_root: Path) -> str: return raw +def _frame_matches(block: str) -> list: + """Every frame line in ``block``, in the order the interpreter printed it.""" + return [m for m in (_FRAME.match(line) for line in block.splitlines()) if m] + + def _frames(block: str, code_root: Path) -> list[dict]: return [ {"file": _frame_file(m["file"], code_root), "line": int(m["line"])} - for m in list(_FRAME.finditer(block))[:MAX_FRAMES] + for m in _frame_matches(block)[-MAX_FRAMES:] ] def _exception_type(block: str) -> str: - """The type name on the line that closes a traceback block.""" - tail = block[max((m.end() for m in _FRAME.finditer(block)), default=0) :] - match = _TYPE.search(tail) - return match["type"] if match else FALLBACK_TYPE + """The type name on the line that closes a traceback block. + + Only the lines after the last frame are read, and only until one of them + names a type. Anything the job printed after the traceback therefore stays + out of the record. + """ + lines = block.splitlines() + positions = [i for i, line in enumerate(lines) if _FRAME.match(line)] + if not positions: + return FALLBACK_TYPE + + last = positions[-1] + depth = _GUTTER.match(lines[last]).end() + for line in lines[last + 1 :]: + prefix = _GUTTER.match(line).end() + # A source line and an annotation sit deeper than the frame they follow. + # The line that closes the block sits at the depth of the frame or less, + # which holds for an ExceptionGroup behind its gutter too. + if prefix > depth: + continue + text = line[prefix:] + if not text: + continue + match = _TYPE.match(text) + return match["type"] if match else FALLBACK_TYPE + return FALLBACK_TYPE def frames_from_stderr(stderr: str, code_root: Path) -> Optional[dict]: @@ -71,12 +100,14 @@ def frames_from_stderr(stderr: str, code_root: Path) -> Optional[dict]: root = code_root.resolve() except OSError: root = code_root - return { - "chain": [ - {"type": _exception_type(b), "frames": _frames(b, root)} - for b in reversed(blocks[-MAX_CHAIN:]) - ] - } + chain = [] + for block in reversed(blocks[-MAX_CHAIN:]): + frames = _frames(block, root) + if not frames: + # No frame means the job printed the header itself, not a traceback. + continue + chain.append({"type": _exception_type(block), "frames": frames}) + return {"chain": chain} if chain else None def write_frames_record( diff --git a/packages/syft-job/tests/test_staging.py b/packages/syft-job/tests/test_staging.py index 59216689526..c8165fcb43c 100644 --- a/packages/syft-job/tests/test_staging.py +++ b/packages/syft-job/tests/test_staging.py @@ -23,7 +23,7 @@ CRASH_PY = """\ x = 1 -raise ValueError("patient 4171 is positive") +raise ValueError("account 88213 holds 4120550") """ @@ -108,7 +108,7 @@ def test_the_traceback_record_is_staged(tmp_path): record = json.loads(record_path.read_text()) assert record["chain"][0]["type"] == "ValueError" - assert "positive" not in record_path.read_text() + assert "4120550" not in record_path.read_text() assert job.release_artifacts([FRAMES_FILENAME]) == [FRAMES_FILENAME] assert (review / FRAMES_FILENAME).exists() diff --git a/packages/syft-job/tests/test_traceback_capture.py b/packages/syft-job/tests/test_traceback_capture.py index e846900db9a..0ded718c9b7 100644 --- a/packages/syft-job/tests/test_traceback_capture.py +++ b/packages/syft-job/tests/test_traceback_capture.py @@ -17,7 +17,7 @@ CRASH = """\ def inner(): - raise ValueError("patient 4171 is positive") + raise ValueError("account 88213 holds 4120550") def outer(): @@ -27,9 +27,20 @@ def outer(): outer() """ +EXCEPTION_GROUP = """\ +def boom(): + raise ValueError("account 88213 holds 4120550") + + +try: + boom() +except ValueError as exc: + raise ExceptionGroup("eg", [exc]) +""" + CHAINED = """\ try: - raise KeyError("patient 4171") + raise KeyError("account 88213") except KeyError as exc: raise RuntimeError("wrapped") from exc """ @@ -70,8 +81,8 @@ def test_record_holds_every_frame_of_the_crash(code_root): def test_record_never_holds_the_exception_message(code_root): record = frames_from_stderr(run(code_root, CRASH), code_root) - assert "positive" not in json.dumps(record) - assert "4171" not in json.dumps(record) + assert "4120550" not in json.dumps(record) + assert "88213" not in json.dumps(record) def test_frame_holds_only_a_file_and_a_line(code_root): @@ -83,7 +94,7 @@ def test_frame_holds_only_a_file_and_a_line(code_root): def test_chained_exception_puts_the_last_failure_first(code_root): record = frames_from_stderr(run(code_root, CHAINED), code_root) assert [entry["type"] for entry in record["chain"]] == ["RuntimeError", "KeyError"] - assert "patient" not in json.dumps(record) + assert "account" not in json.dumps(record) def test_library_frame_keeps_the_path_the_interpreter_printed(code_root): @@ -94,6 +105,49 @@ def test_library_frame_keeps_the_path_the_interpreter_printed(code_root): assert any(f.endswith("json/decoder.py") for f in files) +def test_header_without_frames_yields_no_record(code_root): + """A job that prints the header itself must not reach the record. + + The type was previously read from anywhere in the block, so a line of + ordinary stderr became the exception type. + """ + stderr = "Traceback (most recent call last):\naccount_88213_balance: 4120550\n" + assert frames_from_stderr(stderr, code_root) is None + + +def test_output_after_the_traceback_is_not_read_as_the_type(code_root): + stderr = run(code_root, CRASH) + "account_88213_balance: 4120550\n" + record = frames_from_stderr(stderr, code_root) + assert record["chain"][0]["type"] == "ValueError" + assert "account_88213_balance" not in json.dumps(record) + + +@pytest.mark.skipif( + sys.version_info < (3, 11), reason="ExceptionGroup needs Python 3.11" +) +def test_exception_group_keeps_its_frames_and_leaks_no_output(code_root): + """An ExceptionGroup prints each frame behind a '|' gutter. + + No frame matched before, so the whole block was scanned for a type and a + later line of job output supplied it. + """ + stderr = run(code_root, EXCEPTION_GROUP) + "account_88213_balance: 4120550\n" + record = frames_from_stderr(stderr, code_root) + + blob = json.dumps(record) + assert "account_88213_balance" not in blob + assert "88213" not in blob + assert any( + frame["file"] == "main.py" + for entry in record["chain"] + for frame in entry["frames"] + ) + assert {entry["type"] for entry in record["chain"]} <= { + "ExceptionGroup", + "ValueError", + } + + # -- input that carries no failure ---------------------------------------------- @@ -146,12 +200,29 @@ def test_a_coloured_traceback_parses(code_root): coloured = ( "Traceback (most recent call last):\n" ' File \x1b[35m"main.py"\x1b[0m, line \x1b[35m2\x1b[0m, in \x1b[35mf\x1b[0m\n' - "\x1b[1;35mValueError\x1b[0m: \x1b[35mpatient 4171\x1b[0m\n" + "\x1b[1;35mValueError\x1b[0m: \x1b[35maccount 88213\x1b[0m\n" ) record = frames_from_stderr(coloured, code_root) assert record["chain"][0]["type"] == "ValueError" assert record["chain"][0]["frames"] == [{"file": "main.py", "line": 2}] - assert "4171" not in json.dumps(record) + assert "88213" not in json.dumps(record) + + +def test_frame_cap_keeps_the_innermost_frames(code_root): + """Python prints the outermost frame first, so the job stopped at the last. + + The cap previously kept the first frames, which drops the failure site on + exactly the deep-recursion case that motivates the cap. + """ + block = ( + "Traceback (most recent call last):\n" + + "".join(f' File "main.py", line {i}, in f\n' for i in range(1, 200)) + + "ValueError\n" + ) + frames = frames_from_stderr(block, code_root)["chain"][0]["frames"] + assert len(frames) == MAX_FRAMES + assert frames[-1]["line"] == 199 + assert frames[0]["line"] == 199 - MAX_FRAMES + 1 def test_a_block_without_a_type_line_falls_back(code_root):