diff --git a/docs/jobs.md b/docs/jobs.md index 0494513c96b..fe6d585de55 100644 --- a/docs/jobs.md +++ b/docs/jobs.md @@ -171,9 +171,27 @@ The runner gives each policy these environment variables: The module docstring of `syft_job/policies/env.py` documents every variable and convention in one place. +### Describing a policy + +The DS sees the full source of an API's policies when they look at the API, for example in `ds_client.api["word_count"]`. Start every policy with a docstring of one or two lines that says in plain words what the policy allows and what it refuses, so that is the first thing the DS reads: + +```python +"""Allows each caller up to 10 calls every 24 hours, and refuses any call over that limit.""" +``` + +A bash policy can start with a comment instead. + ### Denying a call -A policy denies a call by exiting with a non-zero code. In Python, any uncaught exception does that. The job is then marked `failed`. The last line of the policy's error output is written to `policy_error.txt` and shared with the DS. The full error output stays with the DO, because a traceback can show private paths or data. +A policy denies a call by raising an exception, or by exiting with a non-zero code. The job is then marked `failed`. The last line of the policy's error output is written to `policy_error.txt` and shared with the DS, who reads it in `job.policy_error`. The full error output stays with the DO, because a traceback can show private paths or data. + +For an uncaught exception, that last line is the exception's name and message. Keep the message to one line that says what was refused and why: + +```python +raise ValueError("Rate limit reached: 10 calls per 24 hours") +``` + +The DS then sees `ValueError: Rate limit reached: 10 calls per 24 hours`. ### State and locking @@ -182,6 +200,8 @@ The state folder belongs to the API, and only the DO can read it. The policy wri ### Example: rate limit ```python +"""Allows each caller up to 10 calls every 24 hours, and refuses any call over that limit.""" + import json import os from datetime import datetime, timedelta, timezone @@ -246,6 +266,9 @@ The DS always sees the job's status. When the output policy denies, the job is ` ### Example: block a word ```python +"""Shares the outputs only when no output file contains the word SECRET. +Refuses to share any of them when one does.""" + import os from pathlib import Path @@ -253,7 +276,7 @@ outputs = Path(os.environ["SYFT_POLICY_OUTPUTS_DIR"]) for f in outputs.rglob("*"): if f.is_file() and "SECRET" in f.read_text(errors="ignore"): - raise ValueError(f"Output {f.name} contains a blocked word") + raise ValueError(f"Outputs not shared: {f.name} contains a blocked word") ``` ### Example: redact CSV rows @@ -261,6 +284,9 @@ for f in outputs.rglob("*"): Only `outputs/` is shared, so a job and its output policy can agree on another folder for staging. In this example the job writes `results.csv` to `output_candidates/`, next to `outputs/`. The policy replaces every row that mentions the secret, writes the cleaned file to `outputs/`, and logs the original rows in the state folder. `output_candidates/` is a convention of this example; the runner does not know about it. ```python +"""Shares results.csv with every row that mentions SECRET replaced by [REDACTED]. +The original rows stay with the data owner.""" + import csv import os from pathlib import Path @@ -295,6 +321,8 @@ result.outputs_dir # the outputs after the run ### Example: add an auditor ```python +"""Shares the outputs with the caller and with auditor@example.com.""" + import json import os from pathlib import Path diff --git a/notebooks/ai_audit/external/DS_Tutorial_v2.ipynb b/notebooks/ai_audit/external/DS_Tutorial_v2.ipynb index 5050b2c4f82..e3256d61999 100644 --- a/notebooks/ai_audit/external/DS_Tutorial_v2.ipynb +++ b/notebooks/ai_audit/external/DS_Tutorial_v2.ipynb @@ -1048,7 +1048,9 @@ "source": [ "### 11d — Call the api\n", "\n", - "Once the DO has created an api from the job (they name it after the job by default), you can call it. The call sends a request with only your parameters, no code. The argument names come from `params.json`. With `block=False` the call returns the request right away (`call.job` is the job once the DO accepts it); by default it waits and returns the finished job." + "Once the DO has created an api from the job (they name it after the job by default), you can call it. The call sends a request with only your parameters, no code. The argument names come from `params.json`. With `block=False` the call returns the request right away (`call.job` is the job once the DO accepts it); by default it waits and returns the finished job.\n", + "\n", + "The DO can also attach policies to the api: one can refuse your call before it runs, or hold its outputs back afterwards. If a policy refuses, the job is `failed` and the DO's reason is in `job.policy_error`." ] }, { diff --git a/notebooks/ai_audit/internal/DO_Tutorial_V3.ipynb b/notebooks/ai_audit/internal/DO_Tutorial_V3.ipynb index 85a19a2343c..bfb019d4fd9 100644 --- a/notebooks/ai_audit/internal/DO_Tutorial_V3.ipynb +++ b/notebooks/ai_audit/internal/DO_Tutorial_V3.ipynb @@ -35,7 +35,7 @@ "| 8 | Review and approve (or reject) a job |\n", "| 9 | Run approved jobs and share results |\n", "| 10 | Run unattended with `syft_bg` background services |\n", - "| 11 | Turn a reviewed job into an api |\n", + "| 11 | Turn a reviewed job into an api, and add a policy |\n", "| 12 | Inspect service logs |\n", "| 13 | Reset (only if needed) |\n", "\n", @@ -46,7 +46,7 @@ "2. **OAuth client credentials** (`credentials.json` from Google Cloud Console). One-time setup; see Step 3.\n", "3. **The Data Scientist's email** — they'll send you a peer request, and you decide whether to approve it.\n", "\n", - "> 💡 This is a **Jupyter notebook running on your local machine** (laptop, desktop, or always-on server). Data Owners need a persistent host because the background services in Step 11 keep running between sessions. If you're running Jupyter on a *headless* remote server (no browser/display), see the heads-up in Step 3 about the OAuth flow.\n" + "> 💡 This is a **Jupyter notebook running on your local machine** (laptop, desktop, or always-on server). Data Owners need a persistent host because the background services in Step 10 keep running between sessions. If you're running Jupyter on a *headless* remote server (no browser/display), see the heads-up in Step 3 about the OAuth flow.\n" ] }, { @@ -825,7 +825,7 @@ "|---------|--------------|\n", "| 🔄 **`sync`** | Keeps your SyftBox in sync with Drive — new submissions appear without manual `client.sync()` calls. |\n", "| 🔔 **`notify`** | Sends you an email when a new job arrives or a peer requests connection. |\n", - "| 🤖 **`approve`** | Approves and runs calls of your apis (Step 12) — and, optionally, auto-approves peer requests from whitelisted domains. |\n", + "| 🤖 **`approve`** | Approves and runs calls of your apis (Step 11) — and, optionally, auto-approves peer requests from whitelisted domains. |\n", "| 📧 **`email_approve`** | Lets you approve/reject jobs by replying to the notification emails. Requires a Google Cloud Pub/Sub topic. |\n", "\n", "### Three-stage setup\n", @@ -923,9 +923,9 @@ }, "source": [ "---\n", - "## Step 10 — Apis for trusted job patterns\n", + "## Step 11 — Apis for trusted job patterns\n", "\n", - "Reviewing every job by hand gets old fast for repeat workflows. An **api** is a job you approved in advance: the DS *calls* it with their own parameters, and each call runs without you reviewing it again. The api pins the code (`run.sh` and the Python file under `code/`, matched by content hash), and takes `config.yaml` and `code/params.json` by name, so the DS can change the parameters between calls.\n", + "Reviewing every job by hand gets old fast for repeat workflows. An **api** is a job you approved in advance: the DS *calls* it with their own parameters, and each call runs without you reviewing it again. The api pins the code (`run.sh` and the Python file under `code/`, matched by content hash), and takes `code/params.json` by name, so the DS can change the parameters between calls.\n", "\n", "> 💡 Calls of your apis need the `approve` service from Step 10 to run on their own. Without it, call `client.process_approved_jobs(...)` yourself.\n", "\n", @@ -934,7 +934,9 @@ "| Match type | When to use it |\n", "|------------|----------------|\n", "| **By content (hash)** | The file shouldn't change. Used for `run.sh` and `main.py` — a call can't change the code. |\n", - "| **By name only** | The file's name is fixed but its contents change between calls. Used for `config.yaml` and `code/params.json`. |\n", + "| **By name only** | The file's name is fixed but its contents change between calls. Used for `code/params.json`. |\n", + "\n", + "`config.yaml` is matched neither way: the api leaves it out, because your client writes it for each call.\n", "\n", "### The cleanest pattern\n", "\n", @@ -943,8 +945,8 @@ "```python\n", "syft_bg.create_api_from_job(\n", " job,\n", - " # code/params.json and config.yaml are name-matched by default;\n", - " # all other files are content-matched\n", + " # code/params.json is name-matched by default, config.yaml is left out\n", + " # (your client writes it for each call), all other files are content-matched\n", " # peers=[ds_email] # optional: restrict to a specific DS\n", ")\n", "```\n", @@ -986,7 +988,8 @@ }, "outputs": [], "source": [ - "# Create the api. run.sh and main.py are content-hashed; config.yaml and code/params.json match by name.\n", + "# Create the api. run.sh and main.py are content-hashed; code/params.json matches by name.\n", + "# config.yaml is left out: your client writes it for each call.\n", "# The api is named after the job. Pass name=\"...\" to choose another name.\n", "\n", "result = syft_bg.create_api_from_job(candidate)\n", @@ -1000,6 +1003,124 @@ { "cell_type": "markdown", "id": "45", + "metadata": { + "id": "api-policy-md" + }, + "source": [ + "### Add an input policy\n", + "\n", + "A **policy** is a short script that you attach to an api. An **input policy** runs before each call's job and decides whether the call may run. An **output policy** runs after the job and checks the outputs before they're shared ([`docs/jobs.md`](https://github.com/OpenMined/PySyft/blob/dev/docs/jobs.md#output-policies) has examples).\n", + "\n", + "The input policy below allows calls on your ocean dataset only, up to 3 per DS every 24 hours. Two conventions make it easy for the DS to read:\n", + "\n", + "- **Start with a one- or two-line docstring** that says in plain words what the policy allows and refuses. The DS sees the policy's full source when they look at your api in `client.api`, so this is the first thing they read.\n", + "- **Refuse with `raise ValueError(\"\")`.** The DS sees the exception's name and message in `job.policy_error`, as in `ValueError: Rate limit reached: 3 calls per 24 hours`, so keep the message to one line that says what was refused and why.\n", + "\n", + "Set `ALLOWED_DATASETS` to the dataset name from Step 6a. If you change it or `MAX_CALLS`, change the docstring too." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "46", + "metadata": { + "id": "api-policy-write-cell" + }, + "outputs": [], + "source": [ + "%%writefile input_policy.py\n", + "\"\"\"Allows calls on the ocean_water_quality2 dataset only, up to 3 per person every 24 hours.\n", + "Refuses calls on any other dataset, and any call over that limit.\"\"\"\n", + "\n", + "import json\n", + "import os\n", + "from datetime import datetime, timedelta, timezone\n", + "from pathlib import Path\n", + "\n", + "ALLOWED_DATASETS = {\"ocean_water_quality2\"}\n", + "MAX_CALLS = 3\n", + "\n", + "dataset = json.loads(Path(os.environ[\"SYFT_POLICY_PARAMS_PATH\"]).read_text()).get(\"dataset\")\n", + "if dataset not in ALLOWED_DATASETS:\n", + " raise ValueError(f\"Dataset {dataset!r} is not allowed. Allowed: {', '.join(sorted(ALLOWED_DATASETS))}\")\n", + "\n", + "state_path = Path(os.environ[\"SYFT_POLICY_DEFAULT_STATE_PATH\"])\n", + "submitter = os.environ[\"SYFT_POLICY_SUBMITTER\"]\n", + "state = json.loads(state_path.read_text())\n", + "\n", + "now = datetime.now(timezone.utc)\n", + "calls = [t for t in state.get(submitter, []) if now - datetime.fromisoformat(t) < timedelta(days=1)]\n", + "if len(calls) >= MAX_CALLS:\n", + " raise ValueError(f\"Rate limit reached: {MAX_CALLS} calls per 24 hours\")\n", + "\n", + "state[submitter] = calls + [now.isoformat()]\n", + "state_path.write_text(json.dumps(state))" + ] + }, + { + "cell_type": "markdown", + "id": "47", + "metadata": { + "id": "api-policy-test-md" + }, + "source": [ + "Test it before you attach it. `syft_bg.run_policy` runs a policy the way the runner does, with no job, peer or sync. `error` is what the DS would see." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "48", + "metadata": { + "id": "api-policy-test-cell" + }, + "outputs": [], + "source": [ + "import tempfile\n", + "\n", + "ok = syft_bg.run_policy(\"input_policy.py\", params={\"dataset\": \"ocean_water_quality2\"})\n", + "refused = syft_bg.run_policy(\"input_policy.py\", params={\"dataset\": \"lab_reports3\"})\n", + "print(f\"ocean_water_quality2: passed={ok.passed}\")\n", + "print(f\"lab_reports3: passed={refused.passed}, the DS sees {refused.error!r}\")\n", + "\n", + "# Four calls against one state folder: the fourth is over the limit.\n", + "state_dir = tempfile.mkdtemp()\n", + "for n in range(1, 5):\n", + " r = syft_bg.run_policy(\"input_policy.py\", params={\"dataset\": \"ocean_water_quality2\"}, state_dir=state_dir)\n", + " print(f\"call {n}: passed={r.passed}\", r.error or \"\")" + ] + }, + { + "cell_type": "markdown", + "id": "49", + "metadata": { + "id": "api-policy-attach-md" + }, + "source": [ + "Attach it to the api you created above. From the next call on, a refused call is `failed`, and the DS reads your message in `job.policy_error`. `set_policies` finds your apis through the `syft_bg.init` config from Step 10.\n", + "\n", + "> 💡 When you create an api, you can attach a policy straight away: `syft_bg.create_api_from_job(job, input_policy=\"input_policy.py\")`." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "50", + "metadata": { + "id": "api-policy-attach-cell" + }, + "outputs": [], + "source": [ + "attached = syft_bg.set_policies(result.name, input_policy=\"input_policy.py\")\n", + "if attached.success:\n", + " print(f\"✅ Input policy attached to {result.name!r}\")\n", + "else:\n", + " print(f\"❌ Failed: {attached.error}\")" + ] + }, + { + "cell_type": "markdown", + "id": "51", "metadata": { "id": "step12-md" }, @@ -1017,7 +1138,7 @@ { "cell_type": "code", "execution_count": null, - "id": "46", + "id": "52", "metadata": { "id": "bg-logs-cell" }, @@ -1036,7 +1157,7 @@ }, { "cell_type": "markdown", - "id": "47", + "id": "53", "metadata": { "id": "step13-md" }, @@ -1054,7 +1175,7 @@ { "cell_type": "code", "execution_count": null, - "id": "48", + "id": "54", "metadata": { "id": "reset-bg-cell" }, @@ -1063,13 +1184,13 @@ "# Uncomment ONLY if you want to wipe syft-bg state and start fresh:\n", "\n", "# syft_bg.reset()\n", - "# print(\"✅ All services stopped and uninstalled. Re-run Step 11 to start again.\")\n" + "# print(\"✅ All services stopped and uninstalled. Re-run Step 10 to start again.\")\n" ] }, { "cell_type": "code", "execution_count": null, - "id": "49", + "id": "55", "metadata": { "id": "delete-dataset-cell" }, @@ -1082,7 +1203,7 @@ }, { "cell_type": "markdown", - "id": "50", + "id": "56", "metadata": { "id": "done-md" }, @@ -1098,6 +1219,7 @@ "- ✅ Ran the job and shared the results back\n", "- ✅ Started background services for unattended operation\n", "- ✅ Created an api from a reviewed job, so the DS can call it without a new review\n", + "- ✅ Added an input policy that checks each call before it runs\n", "\n", "## Day-to-day workflow\n", "\n", @@ -1110,7 +1232,7 @@ "# Or, in the notebook: client.process_approved_jobs(share_outputs_with_submitter=True)\n", "```\n", "\n", - "If something gets stuck, `syft_bg.reset()` + re-run Step 11 brings up a clean state.\n", + "If something gets stuck, `syft_bg.reset()` + re-run Step 10 brings up a clean state.\n", "\n", "## Resources\n", "\n", diff --git a/packages/syft-job/src/syft_job/policies/env.py b/packages/syft-job/src/syft_job/policies/env.py index 4a769f541d0..e66b44d2299 100644 --- a/packages/syft-job/src/syft_job/policies/env.py +++ b/packages/syft-job/src/syft_job/policies/env.py @@ -36,10 +36,16 @@ file in SYFT_POLICY_STATE_DIR. Nothing changes unless the policy writes. The runner holds a lock on the state for the whole run, so two calls of one api never run their policies at once. - Deny Exit with a non-zero code. In Python, any uncaught exception - does this. The job is marked failed. + Describe Start the policy with a one- or two-line docstring (a comment + in bash) that says in plain words what it allows and refuses. + The data scientist sees the policy's full source in client.api. + Deny Raise an exception, or exit with a non-zero code. The job is + marked failed. Reason The last line of stderr is what the data scientist sees, in - policy_error.txt. The full stderr stays with the data owner. + policy_error.txt. For an uncaught exception that is its name + and message, e.g. "ValueError: Rate limit reached", so keep + the message to one line. The full stderr stays with the data + owner. Call files A call sends exactly the files the api lists by name. The input policy can read them in SYFT_POLICY_CALL_DIR and deny the call because of what they hold.