diff --git a/CHANGELOG.md b/CHANGELOG.md index f494fdf..ddc6da7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,22 @@ # Changelog +## Unreleased + +- `publish(signed, *, host, title, name=None, notebook=None, role="notebook", revises=None)` and + `publish_attestation(node, *, host, name)`: write a signed record, or a withdrawal or other + lifecycle attestation on one, to a GitHub Pages host made from the host template's publish mode, + as one commit through the Git Data API (typedstandards#141). `GitHubPagesHost(repository, *, + branch="main", token=None, ...)` names the repository; the token is `token=`, else + `TYPEDSTANDARDS_GITHUB_TOKEN`, a fine-grained token. The default name is the notebook's stem, + the record's date and the first eight hex of its `envelopeHash`. A record whose `envelopeHash` + any listed entry carries is not written again, under any name, and the receipt names that entry; + a listed name with another record needs `revises=`. What host-core's build would refuse is + refused before any write. Refusals raise + `PublishRefusedError` before any write; an API error, or a second non-fast-forward, raises + `PublishError`. The receipt is `{name, commit, bundle_url, verify_url, registry_url, written, + run}`, with `run` `None`. +- README: publishing, the token, and the seed in a hosted notebook and in GitHub Codespaces. + ## 0.1.1 — 2026-10-05 - Fixed: every mapping input to `sign`, `withdraw`, `attest` and `verify` reaches the CLI as a diff --git a/README.md b/README.md index 53a6b95..85c22cc 100644 --- a/README.md +++ b/README.md @@ -25,8 +25,8 @@ The wrapper looks for Node in this order: Only the calls that run the CLI need Node: `sign`, `withdraw`, `attest`, `view`, `verify`, `cli_version()`, and `show` without a precomputed result. With no Node, or one older than 20.19.0, each of those raises `typedstandards.NodeLocatorError`, whose message names the floor and -`TYPEDSTANDARDS_NODE`. `pin`, `badge_cell`, `comparison_cell`, `sidecar` and -`show(record, result)` run without Node. +`TYPEDSTANDARDS_NODE`. `pin`, `badge_cell`, `comparison_cell`, `sidecar`, `show(record, result)`, +`publish` and `publish_attestation` run without Node. Linux and macOS are tested. Windows is untested. @@ -190,10 +190,179 @@ ts.show(bundle) # in Jupyter; ts.show(bundle, marimo=True) in Marimo string or a list of strings); `show` labels it as the signer's and checks nothing about it. The notebook helpers edit the notebook as JSON, splicing the new cell into `cells` so every -other byte of the file stays as written. `httpx` is imported only inside `pin`, PyYAML only +other byte of the file stays as written. `httpx` is imported only inside `pin` and the publish +calls, PyYAML only inside `sidecar`, and `marimo` only inside a Marimo call; `IPython`, `marimo` and `nbformat` are not dependencies. +## Publishing to a GitHub Pages host + +`publish` writes a signed record to a GitHub repository made from the +[host template](https://github.com/npstorey/typedstandards-host-template) in its publish mode, +whose workflow builds the site from the repository's `host.json` and deploys it to GitHub Pages. +Each call is one commit, made through GitHub's Git Data API; the host's workflow then builds and +deploys it. `publish` does not wait for the deploy. The template's README sets a host up for this: +[Publishing from a notebook](https://github.com/npstorey/typedstandards-host-template#publishing-from-a-notebook). + +That setup names your key in `host-policy.json`'s `signer`, and `publish` refuses a record signed +by any other key. The CLI prints a `did:key` only in what it signs, so read yours from the first +record you sign, before its first publish: + +```python +signed = ts.sign(record, output_file="dog-licensing.ipynb") +signed["package"]["signer"]["identifier"] # did:key:z6Mk…: the policy's signer +``` + +A copy starts with no records; its first publish adds the first. + +```python +import typedstandards as ts + +host = ts.GitHubPagesHost("owner/repo") # branch="main"; the token from TYPEDSTANDARDS_GITHUB_TOKEN +signed = ts.sign(record, output_file="dog-licensing.ipynb") # record: an envelope input, as under Use +receipt = ts.publish(signed, host=host, notebook="dog-licensing.ipynb", title="Dog licensing by district") +receipt["verify_url"] # the verifier's link for the served bundle + +withdrawal = ts.withdraw( + {"targetNodeId": signed["envelopeHash"], "reason": "...", "signer": signed["package"]["signer"]} +) +ts.publish_attestation(withdrawal, host=host, name=receipt["name"]) +``` + +- **`GitHubPagesHost(repository, *, branch="main", token=None, ...)`**: the repository as + `owner/name`. Its `repr` shows the repository and the branch only. The HTTP client a call builds + ignores proxy and certificate environment variables; pass `client=` (an `httpx.Client`) for + those. +- **`publish(signed, *, host, title, name=None, notebook=None, role="notebook", revises=None)`** + writes what `sign` printed (or its path) to `records/.signed.json` and appends its entry to + `host.json`: `{name, signed, attestations: [], title, extensions: {role}}`. `host.json` is + rewritten with two-space indentation; no other field of it changes. +- **`publish_attestation(node, *, host, name)`** writes what `withdraw` or `attest` printed (or + its path) to `records/.-.json` and adds that path to the + record's `attestations`, in one commit. A node already listed there is not written again. + +### Names + +The **default name**, with `notebook=`, is `/-`: the notebook file's stem, +the date of the record's `createdAt` (UTC), and the first eight hex characters of its +`envelopeHash`, for example `dog-licensing/2026-10-04-ebb38315`. The signed document does not carry +the notebook's file name, so the stem comes from the argument. A rerun signs to another +`envelopeHash`, so it gets a new name. + +A record whose `envelopeHash` the host already lists is not written again, under any name: the +call writes nothing (`written: False`), and its receipt names the entry that lists it, with that +entry's `bundle_url` and `verify_url`. host-core's build refuses one `envelopeHash` listed twice, +so a second entry would stop every later deploy. To find a listed hash, each call makes one read of +each listed record's signed file. + +With an explicit `name=`, a name listed with another record is refused, unless `revises=` is +given. Then the +record is written under `-`, and the `revises=` node goes on the listed +record's entry, in the same commit. Sign the node first: + +```python +node = ts.attest( + { + "type": "attestation/revises/v1", + "targetNodeId": prior["envelopeHash"], # the listed record + "successorNodeId": signed["envelopeHash"], # this one + "signer": signed["package"]["signer"], + } +) +ts.publish(signed, host=host, name="dog-licensing", title="Dog licensing, rerun", revises=node) +``` + +`publish` compares the node's fields only: its type, its `successorNodeId` with this record's +`envelopeHash`, and its `targetNodeId` with the listed record's. Under the default name, a +`revises=` node goes on the entry of the listed record it targets. + +A name is `/`-separated segments of letters, digits, `.`, `_` and `-`, with no `.` or `..` +segment (host-core's rule), and no `records` or `evidence` segment, which the verifier reads as a +record page's URL rather than a bundle's. + +### Refusals + +Each raises `typedstandards.PublishRefusedError` before any write request: + +- a token that does not start `github_pat_`, or that holds whitespace, a quote or `op://` (the + message names where the token came from, never its value); +- a name that fails the rule above, or a bundle URL with a `records` or `evidence` segment; +- an empty title; +- a record whose `output` is a BlobRef (signed with `output_url=`): the host serves the signed + file only, so sign with `output_file=` alone; +- a record or node whose signer is not the `signer` in the host's `host-policy.json`, or whose + type the policy does not name; +- a role that no rule for `active` records in `host-policy.json` admits: such a record would fail + the host's build, and with it every later deploy; +- a listed name with another record and no `revises=`, or a `revises=` whose fields do not match; +- anything host-core's build would refuse once the commit lands: a signed file it cannot read as + UTF-8 JSON; a signature without its `signature` and `publicKey`, or with a `kid` other than the + signer; an empty `createdAt`; under a registry, a signer, display name or key other than the + host's first record's; with `registry: null`, a signer that is not a pseudonymous `did:key`; a + path another entry already names; a `host.json` that fails host-core's manifest rule; and a host + whose build already fails on a listed record; +- for `publish_attestation`, a withdrawal or supersession that would leave the record in a status + no rule of `host-policy.json` displays (the template's policy has no rule for `superseded`); and + a claim-to-claim node (`corroborates`, `contradicts`), a name the + host does not list, or a node whose `targetNodeId` is not that record's `envelopeHash`. + +A ref update GitHub does not accept re-reads the branch head first, since a write that errored may +have landed. If the commit did not land and the branch moved, `publish` plans again from the new +head once; a second failure raises `typedstandards.PublishError`, as does an API error, whose +message names the request and GitHub's own message. + +### The receipt + +| Key | Value | +|---|---| +| `name` | the name the record is listed under (the derived name for a revision) | +| `commit` | the new commit's id, or `None` when nothing was written | +| `bundle_url` | `/bundles/.bundle.json`, from `host.json`'s `origin` | +| `verify_url` | the verifier's link for `bundle_url`, the same link `badge_cell` writes | +| `registry_url` | `/.well-known/typed-publisher.json`, or `None` for a host with no registry | +| `written` | `True` for a new commit, `False` when the host already listed it | +| `run` | `None`: `publish` does not wait for the deploy | + +### The token + +A fine-grained personal access token, for the one publishing repository, with **Contents: read +and write** and nothing else (enough for a public repository; a private one is unmeasured). Give +it an expiry. It comes from `token=`, else from `TYPEDSTANDARDS_GITHUB_TOKEN`, read when a publish +runs; it is sent only as the `Authorization` header of the client the call builds and closes, and +it is in no message, log record, receipt or file. Never write it as a literal in a cell. Locally, +set it with the seed, from the secret store that starts the kernel (`op run --env-file=… -- +jupyter lab`); in a hosted notebook, from the hosting service's secret store, as for the seed +below. + +Commits made through the API are unsigned, so a branch rule that requires signed commits or pull +requests refuses them. The template's README gives the ruleset for a publishing branch: no +deletion and no force push. + +### The seed in a hosted notebook + +A hosted kernel has no launcher to set the seed's variable, so the author's own code sets it, in +one cell that prints nothing: + +```python +import os + +os.environ["TYPEDSTANDARDS_SIGNING_SEED_B64"] = read_secret("TYPEDSTANDARDS_SIGNING_SEED_B64") +``` + +where `read_secret` stands for the hosting service's own call that reads a stored secret. The +wrapper still never reads the variable; the CLI does. + +The seed must never be pasted into a cell, printed (by `print`, `%env`, or a cell whose last +expression is the value), or saved in the `.ipynb` in any other way: the notebook is the file that +is signed and published. In a hosted notebook, the hosting service's runtime holds the seed for as +long as the kernel runs. Make the seed once, outside any notebook (`openssl rand -base64 32`), and +keep it in a password manager as well as the service's secret store: a record signed by a key that +is lost can never be withdrawn or revised. + +In GitHub Codespaces, a Codespaces secret named `TYPEDSTANDARDS_SIGNING_SEED_B64` arrives in the +codespace as an environment variable, so no line is needed; neither a repository Actions variable +nor an Actions secret is a place for the seed. + ## Versions Each wrapper release pins one CLI version exactly. A CLI upgrade reaches users as a wrapper release diff --git a/scripts/github_stub.py b/scripts/github_stub.py new file mode 100644 index 0000000..91b969b --- /dev/null +++ b/scripts/github_stub.py @@ -0,0 +1,192 @@ +"""An in-memory stand-in for the GitHub REST API calls ``typedstandards.publish`` makes, served +through ``httpx.MockTransport``. No network, no file, no digest. + +It models one repository: a branch head, commits holding whole file maps, blobs, trees, the +contents API's raw read, and a fast-forward-only ``PATCH git/refs/heads/``. Every request +is recorded as ``(method, path)`` in ``requests``. A request whose ``Authorization`` is not +``Bearer `` answers 401. Object ids are counters, not hashes. + +Shared by ``tests/test_publish.py``, ``scripts/smoke_wheel.py`` and the test of +``scripts/smoke_publish.py``, after the stub of ``spike/test/stub-api.mjs`` in the measured +reference client. +""" + +from __future__ import annotations + +import base64 +import copy +import json +from collections.abc import Callable +from typing import Any +from urllib.parse import unquote + +import httpx + +API = "https://api.github.com" + +#: host-policy.json's `display` with `notebook` in the active rule, as P1 changes the template's. +TEMPLATE_DISPLAY = [ + { + "$comment": "An active note is shown as current.", + "status": "active", + "extensions": {"role": ["note", "notebook"]}, + "as": "current", + }, + {"$comment": "A withdrawn record stays listed, marked withdrawn.", "status": "withdrawn", "as": "withdrawn"}, +] + + +def manifest(origin: str = "https://publish.example.org", records: list[dict[str, Any]] | None = None) -> dict: + """A host.json shaped like the template's, with its own origin and records.""" + return { + "$comment": "The host manifest.", + "origin": origin, + "visibility": "public", + "registry": {"$comment": "This host's statement about its key."}, + "index": {"$comment": "The records this host serves."}, + "records": records if records is not None else [], + } + + +def policy(signer: str, display: list[dict[str, Any]] | None = None, **extra: Any) -> dict: + """A host-policy.json shaped like the template's, naming ``signer``.""" + return { + "$comment": "The display policy.", + "signer": signer, + "type": "content/analysis/v1", + "display": display if display is not None else copy.deepcopy(TEMPLATE_DISPLAY), + "unmatched": "refuse", + **extra, + } + + +def dumps(value: Any) -> bytes: + return (json.dumps(value, indent=2, ensure_ascii=False) + "\n").encode("utf-8") + + +class FakeGitHub: + """One repository's API. ``files`` maps a path to its bytes on the branch's head commit.""" + + def __init__( + self, + files: dict[str, bytes], + *, + token: str, + repository: str = "example-owner/example-host", + branch: str = "main", + ) -> None: + self.repository = repository + self.branch = branch + self.token = token + self.requests: list[tuple[str, str]] = [] + self.bodies: list[Any] = [] + self._count = 0 + self.blobs: dict[str, bytes] = {} + self.trees: dict[str, dict[str, bytes]] = {} + self.commits: dict[str, dict[str, Any]] = {} + first = self._id("c") + self.trees[self._id("t")] = dict(files) + self.commits[first] = {"tree": list(self.trees)[-1], "parents": [], "message": "initial"} + self.head = first + #: Before each of the next N ref updates, another writer's commit lands on the branch. + self.concurrent_writes = 0 + #: Statuses to answer the next ref updates with after applying them (a write that landed). + self.landed_but_failed: list[int] = [] + #: Called with each request before it is answered; may return a response to send instead. + self.hook: Callable[[httpx.Request], httpx.Response | None] | None = None + + # --- state ------------------------------------------------------------------------------- + + def _id(self, kind: str) -> str: + """A fresh 40-hex object id (a counter; ``kind`` is for the reader).""" + self._count += 1 + return f"{self._count:040x}" + + def files_at(self, commit: str | None = None) -> dict[str, bytes]: + return self.trees[self.commits[commit or self.head]["tree"]] + + def json_at(self, path: str, commit: str | None = None) -> Any: + return json.loads(self.files_at(commit)[path]) + + def writes(self) -> list[tuple[str, str]]: + return [r for r in self.requests if r[0] not in {"GET", "HEAD"}] + + def other_writer_commits(self) -> None: + """A concurrent writer's commit: an unrelated file added on the branch.""" + files = dict(self.files_at()) + files[f"other/{self._count}.txt"] = b"another writer\n" + tree = self._id("t") + self.trees[tree] = files + commit = self._id("c") + self.commits[commit] = {"tree": tree, "parents": [self.head], "message": "another writer"} + self.head = commit + + # --- the API ----------------------------------------------------------------------------- + + def transport(self) -> httpx.MockTransport: + return httpx.MockTransport(self.handle) + + def handle(self, request: httpx.Request) -> httpx.Response: + path = request.url.path + self.requests.append((request.method, path)) + body = json.loads(request.content) if request.content else None + self.bodies.append(body) + if self.hook is not None: + answer = self.hook(request) + if answer is not None: + return answer + if request.headers.get("authorization") != f"Bearer {self.token}": + return httpx.Response(401, json={"message": "Bad credentials"}) + repo = f"/repos/{self.repository}" + if not path.startswith(repo + "/"): + return httpx.Response(404, json={"message": "Not Found"}) + rest = path[len(repo) :] + method = request.method + + if method == "GET" and rest == f"/git/ref/heads/{self.branch}": + return httpx.Response(200, json={"ref": f"refs/heads/{self.branch}", "object": {"sha": self.head}}) + if method == "GET" and rest.startswith("/git/commits/"): + commit = self.commits.get(rest.rsplit("/", 1)[1]) + if commit is None: + return httpx.Response(404, json={"message": "Not Found"}) + return httpx.Response(200, json={"tree": {"sha": commit["tree"]}}) + if method == "GET" and rest.startswith("/contents/"): + ref = request.url.params.get("ref", self.head) + commit = self.commits.get(ref) + content = None if commit is None else self.trees[commit["tree"]].get(unquote(rest[len("/contents/") :])) + if content is None: + return httpx.Response(404, json={"message": "Not Found"}) + if "raw" in request.headers.get("accept", ""): + return httpx.Response(200, content=content) + encoded = base64.b64encode(content).decode() + return httpx.Response(200, json={"encoding": "base64", "content": encoded}) + if method == "POST" and rest == "/git/blobs": + sha = self._id("b") + self.blobs[sha] = base64.b64decode(body["content"]) + return httpx.Response(201, json={"sha": sha}) + if method == "POST" and rest == "/git/trees": + base = self.trees.get(body.get("base_tree")) + if base is None: + return httpx.Response(422, json={"message": "base_tree not found"}) + files = dict(base) + for item in body["tree"]: + files[item["path"]] = self.blobs[item["sha"]] + sha = self._id("t") + self.trees[sha] = files + return httpx.Response(201, json={"sha": sha}) + if method == "POST" and rest == "/git/commits": + sha = self._id("c") + self.commits[sha] = {"tree": body["tree"], "parents": body["parents"], "message": body["message"]} + return httpx.Response(201, json={"sha": sha, "verification": {"verified": False, "reason": "unsigned"}}) + if method == "PATCH" and rest == f"/git/refs/heads/{self.branch}": + if self.concurrent_writes > 0: + self.concurrent_writes -= 1 + self.other_writer_commits() + commit = self.commits.get(body["sha"]) + if commit is None or body.get("force") or commit["parents"] != [self.head]: + return httpx.Response(422, json={"message": "Update is not a fast forward"}) + self.head = body["sha"] + if self.landed_but_failed: + return httpx.Response(self.landed_but_failed.pop(0), json={"message": "Server Error"}) + return httpx.Response(200, json={"object": {"sha": self.head}}) + return httpx.Response(404, json={"message": f"stub: no route for {method} {path}"}) diff --git a/scripts/smoke_publish.py b/scripts/smoke_publish.py new file mode 100644 index 0000000..aae6769 --- /dev/null +++ b/scripts/smoke_publish.py @@ -0,0 +1,100 @@ +"""Live smoke check of ``publish``: sign one record and publish it to a host made from the host +template's publish mode (typedstandards#141, gate G3). It writes one commit to that repository. + +Run by the owner, with the installed wheel's Python, outside the source tree's import path, from a +terminal signed in to the secret store (``op signin``): + + op run --env-file=publish-check.env -- /bin/python scripts/smoke_publish.py + +where ``publish-check.env`` maps ``TYPEDSTANDARDS_GITHUB_TOKEN`` (a fine-grained token for that +repository alone, Contents read and write) and the CLI's signing-seed variable to secret +references. The script reads neither: ``publish`` reads the token's variable, and the CLI, run by +``sign``, reads the seed from the environment it inherits. + +It prints only the receipt's fields, none of which is secret, and exits non-zero on any refusal +or error: 2 a refusal before any write, 3 an API error, 4 a CLI error, 5 a receipt whose URLs are +not under ``--origin``. ``--repository``, ``--branch`` and ``--origin`` default to the scratch +copy the sprint's live checks use. +""" + +from __future__ import annotations + +import argparse +import json +import sys +import tempfile +from datetime import UTC, datetime +from pathlib import Path +from typing import Any + +import typedstandards as ts + +REPOSITORY = "npstorey/typedstandards-publish-check" +ORIGIN = "https://publish-check.typedstandards.org" + + +def notebook_text(when: str) -> str: + cells = [ + { + "cell_type": "markdown", + "id": "check", + "metadata": {}, + "source": [f"# Publish check\n\nA record signed and published by `scripts/smoke_publish.py` at {when}.\n"], + } + ] + document = {"cells": cells, "metadata": {}, "nbformat": 4, "nbformat_minor": 5} + return json.dumps(document, indent=1, sort_keys=True) + "\n" + + +def record() -> dict[str, Any]: + return { + "type": "content/analysis/v1", + "producerProfile": "scripted-recomputation/typedstandards-python-publish-check", + "captureMethod": "script-run", + "prompt": "Sign and publish one record from the installed wheel.", + "promptVisibility": "full_text", + "queries": [], + "dataSources": [], + "cost": {"model": "none"}, + "skillMetadata": {}, + "trace": {}, + "signer": {"bindingTier": "pseudonymous", "displayName": "typedstandards-python publish check"}, + } + + +def main(argv: list[str] | None = None, *, transport: Any = None) -> int: + parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + parser.add_argument("--repository", default=REPOSITORY) + parser.add_argument("--branch", default="main") + parser.add_argument("--origin", default=ORIGIN) + args = parser.parse_args(argv) + when = datetime.now(UTC).isoformat(timespec="seconds").replace("+00:00", "Z") + print(f"typedstandards {ts.__version__}; publishing one record to {args.repository} ({args.branch})") + try: + with tempfile.TemporaryDirectory() as directory: + notebook = Path(directory) / "publish-check.ipynb" + notebook.write_text(notebook_text(when), encoding="utf-8") + signed = ts.sign(record(), output_file=notebook) + print(f"signed {signed['envelopeHash']} by {signed['package']['signer']['identifier']}") + host = ts.GitHubPagesHost(args.repository, branch=args.branch, transport=transport) + receipt = ts.publish(signed, host=host, notebook=notebook, title=f"Publish check, {when}") + except ts.PublishRefusedError as error: + print(f"refused, nothing written: {error}") + return 2 + except ts.PublishError as error: + print(f"error: {error}") + return 3 + except ts.CliError as error: + print(f"the CLI exited {error.exit_code}: {error}") + return 4 + for key in ("name", "written", "commit", "bundle_url", "verify_url", "registry_url", "run"): + print(f"{key}: {receipt[key]}") + if not receipt["bundle_url"].startswith(f"{args.origin}/bundles/"): + print(f"the receipt's bundle_url is not under {args.origin}: host.json's origin differs") + return 5 + print("publish check passed: the host's workflow now builds and deploys the commit") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/smoke_wheel.py b/scripts/smoke_wheel.py index 50007c6..bacf752 100644 --- a/scripts/smoke_wheel.py +++ b/scripts/smoke_wheel.py @@ -6,9 +6,11 @@ TYPEDSTANDARDS_SIGNING_SEED_B64="$(openssl rand -base64 32)" /bin/python scripts/smoke_wheel.py It checks the import, CLI_VERSION, the vendored CLI's --version, the vendored tree's licences, -one sign-then-verify round trip (sign, view, verify) through the vendored CLI, and the five -helpers with the runtime dependencies the wheel declares (httpx for pin, PyYAML for sidecar), -offline: pin over httpx.MockTransport. It reads no seed. +one sign-then-verify round trip (sign, view, verify) through the vendored CLI, the five +helpers with the runtime dependencies the wheel declares (httpx for pin, PyYAML for sidecar), and +publish and publish_attestation, offline: pin and publish over httpx.MockTransport, publish against +the in-memory GitHub API in github_stub.py beside this script, with a visibly fake token. It reads +no seed. """ from __future__ import annotations @@ -65,6 +67,7 @@ def main() -> int: assert result["ok"] is True assert result["nodeId"] == signed["envelopeHash"] helpers(record) + publishing(signed) print("smoke check passed") return 0 @@ -105,5 +108,31 @@ def portal(request: httpx.Request) -> httpx.Response: print("helpers: pin, badge_cell, comparison_cell, sidecar and show ran from the installed wheel") +def publishing(signed: dict) -> None: + from github_stub import FakeGitHub, dumps, manifest, policy + + token = "github_pat_TESTONLY_wheel_smoke" + signer = signed["package"]["signer"] + files = {"host.json": dumps(manifest()), "host-policy.json": dumps(policy(signer["identifier"]))} + gh = FakeGitHub(files, token=token) + host = typedstandards.GitHubPagesHost(gh.repository, token=token, transport=gh.transport()) + receipt = typedstandards.publish(signed, host=host, name="smoke/record", title="Smoke check") + assert receipt["written"] is True and receipt["commit"] == gh.head, receipt + assert [m for m, _ in gh.requests] == ["GET"] * 4 + ["POST"] * 4 + ["PATCH"], gh.requests + assert typedstandards.publish(signed, host=host, name="smoke/record", title="Smoke check")["written"] is False + withdrawal = typedstandards.withdraw( + {"targetNodeId": signed["envelopeHash"], "reason": "The wheel's smoke check.", "signer": signer} + ) + assert typedstandards.publish_attestation(withdrawal, host=host, name="smoke/record")["written"] is True + assert repr(host) == f"GitHubPagesHost({gh.repository!r}, branch='main')" + try: + typedstandards.publish(signed, host=host, name="records/x", title="Smoke check") + except typedstandards.PublishRefusedError: + pass + else: + raise AssertionError("a records segment was not refused") + print(f"publish: {receipt['bundle_url']} in one commit, then a no-op, then a withdrawal, over MockTransport") + + if __name__ == "__main__": sys.exit(main()) diff --git a/src/typedstandards/__init__.py b/src/typedstandards/__init__.py index 26760da..be81cb2 100644 --- a/src/typedstandards/__init__.py +++ b/src/typedstandards/__init__.py @@ -13,6 +13,7 @@ from ._commands import attest, sign, verify, view, withdraw from ._comparison import comparison_cell from ._node import NODE_FLOOR, NODE_OVERRIDE, locate_node +from ._publish import TOKEN_VARIABLE, GitHubPagesHost, publish, publish_attestation from ._show import Shown, show from ._sidecar import sidecar from .errors import ( @@ -20,6 +21,8 @@ CliNotVendoredError, InternalError, NodeLocatorError, + PublishError, + PublishRefusedError, SeedError, UsageError, VerificationError, @@ -43,11 +46,15 @@ def cli_version() -> str: "NODE_OVERRIDE", "CliError", "CliNotVendoredError", + "GitHubPagesHost", "InternalError", "NodeLocatorError", "Pinned", + "PublishError", + "PublishRefusedError", "SeedError", "Shown", + "TOKEN_VARIABLE", "UsageError", "VerificationError", "__version__", @@ -58,6 +65,8 @@ def cli_version() -> str: "comparison_cell", "locate_node", "pin", + "publish", + "publish_attestation", "show", "sidecar", "sign", diff --git a/src/typedstandards/_publish.py b/src/typedstandards/_publish.py new file mode 100644 index 0000000..9a81b9e --- /dev/null +++ b/src/typedstandards/_publish.py @@ -0,0 +1,991 @@ +"""``publish`` and ``publish_attestation``: write a signed record, or an attestation on one, to a +GitHub Pages host made from the host template's publish mode, in one commit. + +The host's repository holds the template's inputs: ``host.json`` (the manifest host-core builds +from) and ``host-policy.json`` (the display policy). A publish reads both at the branch head, +checks the call against them, and writes the new files and the edited ``host.json`` through the +Git Data API: a blob per file, a tree on the head's tree, a commit whose parent is the head, and a +fast-forward of the branch. The host's workflow builds and deploys the site from that commit; +``publish`` does not wait for it. + +What it does not do, by the package's rules: it computes no hash (a blob's id comes back from +``POST git/blobs``; ``envelopeHash`` and the signer are read from what ``sign`` printed), it never +runs the CLI, and it reads no signing seed. It reads one environment variable of its own, the +token's, sends the token only as the ``Authorization`` header of a client it builds and closes, +and never writes it to a file, a log record, a message or a return value. +""" + +from __future__ import annotations + +import copy +import json +import logging +import os +import re +from collections.abc import Callable, Mapping +from dataclasses import dataclass, field +from pathlib import Path, PurePath +from typing import TYPE_CHECKING, Any, NoReturn +from urllib.parse import quote, urlsplit + +from ._badge import verify_href +from .errors import PublishError, PublishRefusedError + +if TYPE_CHECKING: # httpx is imported inside the calls, so importing the package does not load it + import httpx + +#: The environment variable the token is read from when no ``token=`` is given. +TOKEN_VARIABLE = "TYPEDSTANDARDS_GITHUB_TOKEN" + +#: A fine-grained personal access token's prefix. +FINE_GRAINED_PREFIX = "github_pat_" + +_log = logging.getLogger("typedstandards") + +_REPOSITORY = re.compile(r"^[A-Za-z0-9](?:[A-Za-z0-9-]*[A-Za-z0-9])?/[A-Za-z0-9._-]+$") +_BRANCH = re.compile(r"^[A-Za-z0-9._/-]+$") +_NAME_SEGMENT = re.compile(r"^[A-Za-z0-9._-]+$") +_HEX_64 = re.compile(r"^[0-9a-f]{64}$") +_OBJECT_ID = re.compile(r"^[0-9a-f]{40}(?:[0-9a-f]{24})?$") +_DATE = re.compile(r"^\d{4}-\d{2}-\d{2}") + +#: Path segments the verifier reads as a record page's URL anywhere in a bundle's path +#: (``apps/web/src/lib/verify-flow.ts:300,302,309`` in typedstandards). +_RECORD_PAGE_SEGMENTS = frozenset({"records", "evidence"}) + +#: The claim-to-claim sub-types a bundle does not carry (host-core ``records.ts``). +_CLAIM_TO_CLAIM = frozenset({"attestation/corroborates/v1", "attestation/contradicts/v1", "attestation/endorses/v1"}) +_REVISES = "attestation/revises/v1" + +#: Where the inputs a publish writes go, relative to host.json: the template keeps its signed +#: record under ``records/`` (``records/first-note.signed.json``), which is not served. +_INPUT_DIRECTORY = "records" + + +# --- the host -------------------------------------------------------------------------------- + + +class _Token: + """A token held for later, whose ``repr`` and ``str`` do not show it.""" + + __slots__ = ("_value",) + + def __init__(self, value: str) -> None: + self._value = value + + def __repr__(self) -> str: + return "" + + __str__ = __repr__ + + def __reduce__(self) -> NoReturn: + raise TypeError("a token is not pickled") + + def reveal(self) -> str: + return self._value + + +class GitHubPagesHost: + """A GitHub repository whose Pages site the host template's publish-mode workflow builds + from ``host.json`` on ``branch``. + + ``repository`` is ``owner/name``. The token is ``token=``, else the environment variable + ``TYPEDSTANDARDS_GITHUB_TOKEN`` read when a publish runs: a fine-grained personal access token + for this one repository, with Contents read and write. Its ``repr`` names the repository and + the branch only. ``client`` (used as given, not closed) or ``transport`` (for a client each + call builds and closes), ``api_url`` and ``timeout`` are for tests and for callers with their + own HTTP settings; redirects are never followed. The client a call builds ignores proxy and + certificate environment variables; pass ``client=`` for those. + """ + + __slots__ = ("repository", "branch", "api_url", "timeout", "_token", "_client", "_transport") + + def __init__( + self, + repository: str, + *, + branch: str = "main", + token: str | None = None, + api_url: str = "https://api.github.com", + client: httpx.Client | None = None, + transport: httpx.BaseTransport | None = None, + timeout: float = 30.0, + ) -> None: + if not isinstance(repository, str) or not _REPOSITORY.match(repository): + raise ValueError(f"repository must be owner/name, as in https://github.com/owner/name: {repository!r}") + if not isinstance(branch, str) or not _BRANCH.match(branch) or ".." in branch: + raise ValueError(f"branch must be a branch name: {branch!r}") + if not isinstance(api_url, str) or not api_url.startswith("https://") or api_url.endswith("/"): + raise ValueError("api_url must be an https:// URL with no trailing /") + if token is not None and not isinstance(token, str): + raise TypeError("token must be a string") + self.repository = repository + self.branch = branch + self.api_url = api_url + self.timeout = timeout + self._token = _Token(token) if token is not None else None + self._client = client + self._transport = transport + + def __repr__(self) -> str: + return f"GitHubPagesHost({self.repository!r}, branch={self.branch!r})" + + __str__ = __repr__ + + def __reduce__(self) -> NoReturn: + raise TypeError("a GitHubPagesHost is not pickled: it may hold a token") + + def _resolve_token(self) -> str: + """The token, checked before any request. A refusal holds no part of the value: not in its + message, and not as a local of any frame its traceback shows (a verbose notebook + traceback prints frames' locals).""" + value = self._token.reveal() if self._token is not None else os.environ.get(TOKEN_VARIABLE) + problem = _token_problem(value, "token=" if self._token is not None else TOKEN_VARIABLE) + if problem is not None: + del value + _refuse(problem) + return value # type: ignore[return-value] # _token_problem refuses None and "" + + +def _token_problem(value: str | None, source: str) -> str | None: + """Why ``value`` is not a usable token, in words that never quote it; ``None`` when it is.""" + if not value: + return ( + f"no GitHub token: pass token= or set {TOKEN_VARIABLE} (a fine-grained personal access token " + "for this one repository, with Contents read and write)" + ) + if "op://" in value: + return f"{source} holds an op:// secret reference: the secret store did not resolve it" + if re.search(r"[\s\"']", value): + return f"{source} holds whitespace or a quote: pass the token's characters only" + if not value.startswith(FINE_GRAINED_PREFIX) or value == FINE_GRAINED_PREFIX: + return ( + f"{source} is not a fine-grained personal access token (one starts {FINE_GRAINED_PREFIX}): " + "make one for this repository alone, with Contents read and write" + ) + return None + + +def _refuse(message: str) -> NoReturn: + raise PublishRefusedError(message) + + +# --- the API ----------------------------------------------------------------------------------- + + +class _Api: + """GitHub's REST API for one repository. The token is in the client's headers (or, for a + given client, each request's) and nowhere else; what is logged is a method, a path and a + status.""" + + def __init__(self, host: GitHubPagesHost) -> None: + import httpx + + self._httpx = httpx + self.host = host + self.base = f"{host.api_url}/repos/{host.repository}" + self._own = host._client is None + # The headers are built inside each call below, so no frame holds them as a local. + if host._client is None: + # trust_env=False: building the client does not iterate the environment (which holds the + # signing seed) for proxy settings, and reads no .netrc. A caller who needs a proxy or a + # certificate bundle passes client=. + self._http = httpx.Client( + headers=_headers(host), + transport=host._transport, + timeout=host.timeout, + follow_redirects=False, + trust_env=False, + ) + self._headers: dict[str, str] = {} + else: + self._http = host._client + self._headers = _headers(host) + + def close(self) -> None: + if self._own: + self._http.close() + + def _request_headers(self, accept: str | None) -> dict[str, str]: + return {**self._headers, **({"Accept": accept} if accept is not None else {})} + + def _send(self, method: str, path: str, body: Any = None, *, accept: str | None = None) -> httpx.Response: + try: + response = self._http.request( + method, self.base + path, json=body, headers=self._request_headers(accept), follow_redirects=False + ) + except self._httpx.TransportError as error: + _log.info("publish: %s %s -> no response (%s)", method, path, type(error).__name__) + raise _NoResponse(f"{method} {path}: no response ({type(error).__name__})") from None + _log.info("publish: %s %s -> %s", method, path, response.status_code) + return response + + @staticmethod + def _fail(method: str, path: str, response: httpx.Response) -> NoReturn: + try: + message = str(response.json().get("message", ""))[:200] + except (ValueError, AttributeError): + message = "" + raise _ApiError( + response.status_code, f"{method} {path} answered {response.status_code}: {message or '(no message)'}" + ) + + def get_json(self, path: str) -> Any: + response = self._send("GET", path) + if response.status_code != 200: + self._fail("GET", path, response) + return response.json() + + def read(self, path: str, ref: str) -> bytes | None: + """A file's bytes at ``ref`` (the contents API's raw media type), or ``None`` for a 404.""" + at = quote(path, safe="/") + response = self._send("GET", f"/contents/{at}?ref={ref}", accept="application/vnd.github.raw+json") + if response.status_code == 404: + return None + if response.status_code != 200: + self._fail("GET", f"/contents/{at}", response) + return response.content + + def write(self, method: str, path: str, body: Any, expect: int) -> Any: + """A Git Data API write: ``POST git/blobs|trees|commits`` or ``PATCH git/refs/heads/…``.""" + if method not in {"POST", "PATCH"} or not path.startswith("/git/"): + raise AssertionError(f"publish writes through the Git Data API only, not {method} {path}") + response = self._send(method, path, body) + if response.status_code != expect: + self._fail(method, path, response) + return response.json() + + def head(self) -> str: + sha = self.get_json(f"/git/ref/heads/{self.host.branch}").get("object", {}).get("sha") + return _object_id(sha, f"the head of {self.host.branch}") + + +def _headers(host: GitHubPagesHost) -> dict[str, str]: + return { + "Authorization": f"Bearer {host._resolve_token()}", + "Accept": "application/vnd.github+json", + "X-GitHub-Api-Version": "2022-11-28", + "User-Agent": "typedstandards-python", + } + + +class _ApiError(PublishError): + def __init__(self, status: int, message: str) -> None: + super().__init__(message) + self.status = status + + +class _NoResponse(PublishError): + pass + + +def _object_id(value: Any, what: str) -> str: + if not isinstance(value, str) or not _OBJECT_ID.match(value): + raise PublishError(f"GitHub's answer for {what} carries no object id") + return value + + +# --- JSON as host-core reads it ---------------------------------------------------------------- + + +def _no_constant(name: str) -> NoReturn: + raise ValueError(f"{name} is not JSON") + + +def _parse_host_json(content: bytes) -> Any: + """Parse bytes as host-core's ``parseJsonFile`` does (``json.ts:31-41``): strict UTF-8, with a + leading byte-order mark dropped as ``TextDecoder`` drops it, and ``JSON.parse``, which has no + ``NaN`` or ``Infinity``. Raises ``ValueError`` where host-core throws.""" + try: + text = content.decode("utf-8") + except UnicodeDecodeError as error: + raise ValueError("is not UTF-8") from error + return json.loads(text.removeprefix("\ufeff"), parse_constant=_no_constant) + + +# --- the host's files, read at the head ---------------------------------------------------------- + + +@dataclass +class _State: + head: str + tree: str + manifest: dict[str, Any] + policy: dict[str, Any] + api: _Api + signed_cache: dict[str, dict[str, Any] | None] = field(default_factory=dict) + + def entry(self, name: str) -> dict[str, Any] | None: + return next((e for e in self.manifest["records"] if e.get("name") == name), None) + + def signed_of(self, entry: Mapping[str, Any]) -> dict[str, Any] | None: + """What the entry's ``signed`` file holds at the head, parsed, or ``None``.""" + path = entry.get("signed") + if not isinstance(path, str): + return None + if path not in self.signed_cache: + content = self.api.read(path, self.head) + try: + value = _parse_host_json(content) if content is not None else None + except ValueError: + value = None + self.signed_cache[path] = value if isinstance(value, dict) else None + return self.signed_cache[path] + + def hash_of(self, entry: Mapping[str, Any]) -> str | None: + value = (self.signed_of(entry) or {}).get("envelopeHash") + return value if isinstance(value, str) else None + + +def _read_state(api: _Api, head: str | None = None) -> _State: + head = head or api.head() + tree = _object_id(api.get_json(f"/git/commits/{head}").get("tree", {}).get("sha"), f"the tree of {head}") + manifest = _read_json_file(api, "host.json", head) + policy = _read_json_file(api, "host-policy.json", head) + if not isinstance(manifest.get("records"), list) or not all(isinstance(e, dict) for e in manifest["records"]): + _refuse("host.json has no records list of objects: it is not the manifest publish edits") + origin = manifest.get("origin") + if not isinstance(origin, str) or not origin.startswith("https://") or origin.endswith("/"): + _refuse("host.json's origin is not an https:// origin with no trailing /") + return _State(head, tree, manifest, policy, api) + + +def _read_json_file(api: _Api, path: str, head: str) -> dict[str, Any]: + content = api.read(path, head) + if content is None: + _refuse(f"{path} is not on {api.host.branch}: publish writes to a host made from the host template") + try: + value = _parse_host_json(content) + except ValueError: + _refuse(f"{path} on {api.host.branch} is not JSON") + if not isinstance(value, dict): + _refuse(f"{path} on {api.host.branch} is not a JSON object") + return value + + +def _strings(value: Any) -> list[str] | None: + """A policy field that is a string or a list of strings, as a list; ``None`` when absent.""" + if value is None: + return None + if isinstance(value, str): + return [value] + return [v for v in value if isinstance(v, str)] if isinstance(value, list) else [] + + +def _check_signer(policy: Mapping[str, Any], signer: str, what: str) -> None: + signers = _strings(policy.get("signer")) + if not signers: + _refuse("host-policy.json names no signer: publish compares the record's signer with it") + if signer not in signers: + _refuse( + f"{what} is signed by {signer}, not by the signer host-policy.json names ({', '.join(signers)}): " + "a host serves one signer's records" + ) + + +def _check_admitted(policy: Mapping[str, Any], *, signer: str, type_: str, role: str) -> None: + """G0-3: an ``active`` record with this signer, type and role must match a display rule, + as host-core's ``displayOf`` reads the policy, or every later deploy would fail.""" + types = _strings(policy.get("type")) + if types is not None and type_ not in types: + _refuse(f"the record's type {type_} is not one host-policy.json names ({', '.join(types)})") + admitted: list[Any] = [] + for rule in policy.get("display") or []: + if not isinstance(rule, dict) or "active" not in (_strings(rule.get("status")) or []): + continue + if (s := _strings(rule.get("signer"))) is not None and signer not in s: + continue + if (t := _strings(rule.get("type"))) is not None and type_ not in t: + continue + extensions = rule.get("extensions") or {} + if not isinstance(extensions, dict) or set(extensions) - {"role"}: + continue + roles = extensions.get("role") + if roles is None or (isinstance(roles, list) and role in roles): + return + admitted += roles if isinstance(roles, list) else [] + known = ", ".join(str(r) for r in admitted) or "none" + _refuse(f"no active rule of host-policy.json admits the role {role} (roles active rules admit: {known})") + + +def _displayed(policy: Mapping[str, Any], *, status: str, signer: str, type_: str, extensions: Mapping) -> bool: + """Whether host-core's ``displayOf`` (``display.ts:118-145``) shows a record with this status, + signer, type and index ``extensions`` under the policy, as the template's display step runs it.""" + for key, value in (("signer", signer), ("type", type_)): + named = _strings(policy.get(key)) + if named is not None and value not in named: + return False + for rule in policy.get("display") or []: + if not isinstance(rule, dict) or status not in (_strings(rule.get("status")) or []): + continue + if any( + (named := _strings(rule.get(k))) is not None and v not in named + for k, v in (("signer", signer), ("type", type_)) + ): + continue + admitted = rule.get("extensions") or {} + if isinstance(admitted, dict) and all( + k in extensions and isinstance(values, list) and extensions[k] in values for k, values in admitted.items() + ): + return True + return False + + +#: The status a lifecycle node moves its record to, as verify-core reads the latest node +#: (``lifecycle.ts:192-198``); a ``revises`` node moves none. +_STATUS_AFTER = { + "attestation/withdraws/v1": "withdrawn", + "attestation/supersedes/v1": "superseded", + "attestation/reinstates/v1": "active", +} + +_MANIFEST_ORIGIN = re.compile(r"^https://[^/?#\s]+(/[^?#\s]*[^/?#\s])?$") +_WINDOWS_PATH = re.compile(r"^[A-Za-z]:[\\/]") + + +def _relative(path: Any) -> bool: + return isinstance(path, str) and path != "" and not path.startswith("/") and not _WINDOWS_PATH.match(path) + + +def _manifest_problem(m: Any) -> str | None: + """host-core's ``parseManifest`` (``manifest.ts:41-95``): why it would refuse ``m``, or ``None``.""" + if not isinstance(m, dict): + return "host.json must be a JSON object" + extra = sorted(set(m) - {"$comment", "origin", "visibility", "registry", "index", "records"}) + missing = sorted({"origin", "registry", "index", "records"} - set(m)) + if extra or missing: + return f"host.json: {', '.join(extra)} not defined" if extra else f"host.json is missing {', '.join(missing)}" + if not isinstance(m["origin"], str) or not _MANIFEST_ORIGIN.match(m["origin"]): + return "host.json: origin must be an https:// origin with no trailing /, query or fragment" + if not isinstance(m.get("visibility"), str) or not m["visibility"]: + return "host.json: visibility is required" + for key in ("registry", "index"): + value = m[key] + if key == "registry" and value is None: + continue + if not isinstance(value, dict) or set(value) - {"$comment"}: + return f"host.json: {key} must be an object holding at most $comment" + ( + " or null" if key == "registry" else "" + ) + if "$comment" in value and not isinstance(value["$comment"], str): + return f"host.json: {key}.$comment must be a string" + records = m["records"] + if not isinstance(records, list) or not records: + return "host.json: records must be a non-empty array" + seen: set[str] = set() + for i, r in enumerate(records): + at = f"host.json: records[{i}]" + if not isinstance(r, dict): + return f"{at} must be an object" + extra = sorted(set(r) - {"$comment", "name", "signed", "attestations", "title", "extensions"}) + missing = sorted({"name", "signed", "attestations", "title"} - set(r)) + if extra or missing: + return f"{at}: {', '.join(extra)} not defined" if extra else f"{at} is missing {', '.join(missing)}" + name = r["name"] + if not isinstance(name, str) or not all( + _NAME_SEGMENT.match(x) and x not in {".", ".."} for x in name.split("/") + ): + return f"{at}.name fails the record-name rule" + if name in seen: + return f"{at}.name: {name} is listed twice" + seen.add(name) + if not _relative(r["signed"]): + return f"{at}.signed must be a path relative to host.json" + if not isinstance(r["attestations"], list) or not all(_relative(a) for a in r["attestations"]): + return f"{at}.attestations must be an array of paths relative to host.json" + if not isinstance(r["title"], str) or not r["title"]: + return f"{at}.title must be a non-empty string" + if "extensions" in r and not isinstance(r["extensions"], dict): + return f"{at}.extensions must be an object" + return None + + +def _check_manifest(manifest: Any) -> None: + if (problem := _manifest_problem(manifest)) is not None: + _refuse(f"the host.json publish would write fails host-core's manifest rule: {problem}") + + +def _check_listed(state: _State) -> None: + """Read every listed record's signed file (once: ``signed_cache``) and refuse when host-core's + build already fails on one: unreadable, not what sign prints, or, under a registry, a signer, + display name or key unlike the first record's (``build.ts:49-66``, ``:93-97``).""" + first: dict[str, Any] | None = None + for entry in state.manifest["records"]: + document = state.signed_of(entry) + label = entry.get("name") + if document is None: + _refuse(f"the host's build already fails on {label}: its signed file cannot be read as JSON") + if (problem := _signed_problem(document)) is not None: + _refuse(f"the host's build already fails on {label}: {problem}") + if state.manifest.get("registry") is not None: + if first is None: + first = document + elif (problem := _one_signer_problem(document, first)) is not None: + _refuse(f"the host's build already fails on {label}: {problem}") + + +def _one_signer_problem(document: Mapping[str, Any], first: Mapping[str, Any]) -> str | None: + """host-core's registry rule (``build.ts:90-98``): one signer identifier, binding tier, display + name and key for every record under a registry, and a signature ``kid`` that is the signer.""" + a, b = _signer_of(document), _signer_of(first) + if any(a.get(k) != b.get(k) for k in ("identifier", "bindingTier", "displayName")) or ( + document["signature"].get("publicKey") != first["signature"].get("publicKey") + ): + return ( + f"its signer ({a.get('identifier')}, {a.get('bindingTier')}, {a.get('displayName')!r}) or key differs from " + f"the first record's ({b.get('identifier')}, {b.get('bindingTier')}, {b.get('displayName')!r}); a host " + "under a registry serves one signer" + ) + kid = document["signature"].get("kid") + if kid is not None and kid != a.get("identifier"): + return f"its signature's kid {kid} is not the signer's identifier {a.get('identifier')}" + return None + + +def _check_registry_rules(state: _State, document: Mapping[str, Any]) -> None: + """The rules host-core's build applies to a new record by the host's registry: under one, the + first listed record's signer and key (or, for a host's first record, the record's own kid); + with ``registry: null``, produce-core's view takes only a self-certifying signer, a + pseudonymous ``did:key`` (``build.ts:147-148``, ``commitment.ts:200-215``).""" + signer = _signer_of(document) + if state.manifest.get("registry") is None: + if signer.get("bindingTier") != "pseudonymous" or not str(signer.get("identifier", "")).startswith("did:key:"): + _refuse( + "host.json's registry is null, under which host-core builds a view only for a pseudonymous did:key " + f"signer; this record's is {signer.get('identifier')} at {signer.get('bindingTier')}" + ) + return + records = state.manifest["records"] + first = state.signed_of(records[0]) if records else None + if (problem := _one_signer_problem(document, first if first is not None else document)) is not None: + _refuse(f"the record: {problem}") + + +def _check_path_unnamed(manifest: Mapping[str, Any], path: str) -> None: + """A file another entry lists is not overwritten: the build would read this record's bytes + under that entry (``build.ts:65-66``, ``:70-71``).""" + for entry in manifest["records"]: + if entry.get("signed") == path or path in (entry.get("attestations") or []): + _refuse(f"host.json's entry {entry.get('name')} already names {path}: publish does not overwrite it") + + +# --- the call's own checks, before any request -------------------------------------------------- + + +def _load(value: Mapping[str, Any] | str | os.PathLike[str], what: str) -> tuple[Any, bytes | None]: + """A document given as a mapping (written as JSON) or a path (written as its bytes).""" + if isinstance(value, Mapping): + return value, None + if isinstance(value, (str, os.PathLike)): + content = Path(value).read_bytes() + try: + return _parse_host_json(content), content + except ValueError as error: + _refuse(f"{os.fspath(value)} {error}, as host-core reads a file (UTF-8, JSON.parse): {what}") + return value, None + + +def _serialize(value: Any) -> bytes: + """JSON with two-space indentation and a newline. A value JSON.parse would refuse (NaN, + Infinity) is refused.""" + try: + return (json.dumps(value, indent=2, ensure_ascii=False, allow_nan=False) + "\n").encode("utf-8") + except ValueError as error: + _refuse(f"a document holds a value JSON cannot carry ({error}): host-core could not read it") + + +def _signature_problem(value: Any) -> str | None: + """host-core's ``checkSignature`` (``records.ts:43-50``).""" + if not isinstance(value, Mapping) or not isinstance(value.get("signature"), str): + return "its signature must be an object with a signature and a publicKey string" + if not isinstance(value.get("publicKey"), str): + return "its signature must be an object with a signature and a publicKey string" + for key in ("algorithm", "kid"): + if key in value and not isinstance(value[key], str): + return f"its signature's {key} must be a string" + return None + + +def _signed_problem(value: Any) -> str | None: + """Why host-core's build would refuse a signed file: ``checkSignedDocument`` + (``records.ts:54-66``), then the package's ``metadata.createdAt`` and ``signer.identifier`` + (``build.ts:61-64``). ``None`` when it would not.""" + if not isinstance(value, Mapping) or set(value) != {"package", "envelopeHash", "signature"}: + return "it is not what sign prints: {package, envelopeHash, signature}" + if not isinstance(value["package"], Mapping): + return "its package is not an object" + if not isinstance(value["envelopeHash"], str) or not _HEX_64.match(value["envelopeHash"]): + return "its envelopeHash is not 64 lowercase hex characters" + if (problem := _signature_problem(value["signature"])) is not None: + return problem + metadata = value["package"].get("metadata") + created_at = metadata.get("createdAt") if isinstance(metadata, Mapping) else None + if not isinstance(created_at, str) or not created_at: + return "its package has no metadata.createdAt, which the index states" + if not _signer_of(value).get("identifier"): + return "its package has no signer.identifier, which the index states" + return None + + +def _signer_of(document: Mapping[str, Any]) -> dict[str, Any]: + signer = document["package"].get("signer") if isinstance(document.get("package"), Mapping) else None + return dict(signer) if isinstance(signer, Mapping) else {} + + +def _check_signed(value: Any) -> tuple[str, str, str, str]: + """``(envelopeHash, signer identifier, type, createdAt)`` from what ``sign`` printed.""" + if (problem := _signed_problem(value)) is not None: + _refuse(f"publish takes what sign prints, as host-core's build reads it: {problem}") + package = value["package"] + envelope_hash = value["envelopeHash"] + output = package.get("output") + if isinstance(output, Mapping): + _refuse( + "the record's output is a BlobRef (signed by reference, with output_url=): a host made from the " + "template serves the signed file only, so publish takes a record whose output is inline " + "(sign with output_file= and no output_url=)" + ) + signer = (package.get("signer") or {}).get("identifier") if isinstance(package.get("signer"), Mapping) else None + if not isinstance(signer, str) or not signer: + _refuse("the record's package names no signer.identifier") + type_ = package.get("type") + created_at = ( + (package.get("metadata") or {}).get("createdAt") if isinstance(package.get("metadata"), Mapping) else None + ) + if not isinstance(type_, str) or not isinstance(created_at, str): + _refuse("the record's package has no type or metadata.createdAt: publish takes what sign prints") + return envelope_hash, signer, type_, created_at + + +def _check_node(value: Any, what: str) -> tuple[str, str, str, dict[str, Any]]: + """``(nodeId, type, signer identifier, node)`` from what ``withdraw`` or ``attest`` printed.""" + if not isinstance(value, Mapping) or set(value) != {"node", "nodeId", "signature"}: + _refuse(f"{what} takes what withdraw or attest prints: {{node, nodeId, signature}}") + node, node_id = value["node"], value["nodeId"] + if not isinstance(node, Mapping) or not isinstance(node_id, str) or not _HEX_64.match(node_id): + _refuse(f"{what}: node must be an object and nodeId 64 lowercase hex characters") + type_ = node.get("type") + if not isinstance(type_, str) or not type_.startswith("attestation/"): + _refuse(f"{what}: node.type is not an attestation type") + if type_ in _CLAIM_TO_CLAIM: + _refuse(f"{what} is an {type_}, a claim-to-claim node: a bundle carries lifecycle attestations only") + signer = node.get("signer", {}).get("identifier") if isinstance(node.get("signer"), Mapping) else None + if not isinstance(signer, str): + _refuse(f"{what}: node names no signer.identifier") + if (problem := _signature_problem(value["signature"])) is not None: + _refuse(f"{what}: {problem}") + return node_id, type_, signer, dict(node) + + +def check_name(name: Any) -> str: + """host-core's record-name rule, and no ``records`` or ``evidence`` segment.""" + if not isinstance(name, str) or not all(_NAME_SEGMENT.match(s) and s not in {".", ".."} for s in name.split("/")): + _refuse( + f"the name {name!r} must be /-separated segments of letters, digits, '.', '_' and '-', with no " + "'.' or '..' segment (host-core's rule)" + ) + if _RECORD_PAGE_SEGMENTS & set(name.split("/")): + _refuse( + f"the name {name} has a records or evidence segment, which the verifier reads as a record page's " + "URL rather than a bundle's: choose another name" + ) + return name + + +def default_name(notebook: str | os.PathLike[str], envelope_hash: str, created_at: str) -> str: + """``/-``.""" + if not _DATE.match(created_at): + _refuse("the record's metadata.createdAt does not start with a date") + return f"{PurePath(os.fspath(notebook)).stem}/{created_at[:10]}-{envelope_hash[:8]}" + + +def _bundle_url(origin: str, name: str) -> str: + url = f"{origin}/bundles/{name}.bundle.json" + if _RECORD_PAGE_SEGMENTS & set(urlsplit(url).path.split("/")): + _refuse( + f"{url} has a records or evidence segment, which the verifier reads as a record page's URL: " + "host.json's origin or the name must change" + ) + return url + + +def _receipt(state: _State, name: str, commit: str | None) -> dict[str, Any]: + origin = state.manifest["origin"] + bundle_url = _bundle_url(origin, name) + registry = state.manifest.get("registry") + return { + "name": name, + "commit": commit, + "bundle_url": bundle_url, + "verify_url": verify_href(bundle_url), + "registry_url": f"{origin}/.well-known/typed-publisher.json" if registry is not None else None, + "written": commit is not None, + "run": None, + } + + +# --- the write --------------------------------------------------------------------------------- + + +@dataclass +class _Plan: + """What one commit writes: the files (path to bytes), the manifest, the message, the name.""" + + files: dict[str, bytes] + manifest: dict[str, Any] + message: str + name: str + + +def _commit(api: _Api, plan_for: Callable[[_State], _Plan | dict[str, Any]]) -> dict[str, Any]: + """Read the head, plan, write one commit, fast-forward the branch. On a ref update that did + not answer 200, re-read the head first: a write that errored may have landed. If it did not, + and the branch moved (not a fast forward) or no answer came, plan again from the new head + once; a second failure is an error.""" + head: str | None = None + for attempt in (1, 2): + state = _read_state(api, head) + plan = plan_for(state) + if isinstance(plan, dict): + return plan + entries = [] + for path, content in [*plan.files.items(), ("host.json", _serialize(plan.manifest))]: + blob = api.write("POST", "/git/blobs", {"content": _b64(content), "encoding": "base64"}, 201) + entries.append({"path": path, "mode": "100644", "type": "blob", "sha": _object_id(blob.get("sha"), path)}) + tree = api.write("POST", "/git/trees", {"base_tree": state.tree, "tree": entries}, 201) + made = api.write( + "POST", + "/git/commits", + {"message": plan.message, "tree": _object_id(tree.get("sha"), "the tree"), "parents": [state.head]}, + 201, + ) + commit = _object_id(made.get("sha"), "the commit") + ref = f"/git/refs/heads/{api.host.branch}" + try: + api.write("PATCH", ref, {"sha": commit, "force": False}, 200) + _log.info("publish: %s is %s on %s", plan.name, commit, api.host.branch) + return _receipt(state, plan.name, commit) + except (_ApiError, _NoResponse) as error: + head = api.head() + if head == commit: + _log.info("publish: the ref update errored but landed: %s", commit) + return _receipt(state, plan.name, commit) + retry = isinstance(error, _NoResponse) or error.status == 422 + if not retry or attempt == 2: + raise PublishError( + f"{error}; {api.host.branch} is at {head}, not the new commit {commit}" + + (" after one retry: not a fast forward twice" if retry else "") + ) from None + _log.info("publish: %s moved to %s; planning again from it", api.host.branch, head) + raise AssertionError("unreachable") + + +def _b64(content: bytes) -> str: + import base64 + + return base64.b64encode(content).decode("ascii") + + +def _run(host: GitHubPagesHost, plan_for: Callable[[_State], _Plan | dict[str, Any]]) -> dict[str, Any]: + if not isinstance(host, GitHubPagesHost): + raise TypeError("host must be a GitHubPagesHost") + api = _Api(host) + try: + return _commit(api, plan_for) + finally: + api.close() + + +# --- publish ------------------------------------------------------------------------------------ + + +def _node_path(record_name: str, type_: str, node_id: str) -> str: + return f"{_INPUT_DIRECTORY}/{record_name}.{type_.split('/')[1]}-{node_id[:8]}.json" + + +def publish( + signed: Mapping[str, Any] | str | os.PathLike[str], + *, + host: GitHubPagesHost, + title: str, + name: str | None = None, + notebook: str | os.PathLike[str] | None = None, + role: str = "notebook", + revises: Mapping[str, Any] | str | os.PathLike[str] | None = None, +) -> dict[str, Any]: + """Publish what :func:`sign` printed (or its path) to ``host`` in one commit. + + The record is listed in ``host.json`` under ``name``, or under the default name + ``/-`` from ``notebook`` (the file's stem), the date of the record's + ``createdAt`` and the first eight hex of its ``envelopeHash``. Its file is written at + ``records/.signed.json``, its entry gets ``title`` and ``extensions.role``. + + A record whose ``envelopeHash`` any listed entry carries is not written again, under any name + (``written: False``); the receipt names that entry. To find it, each call reads every listed + record's signed file once. A name listed with another record is refused, unless ``revises=`` + is what :func:`attest` printed for an ``attestation/revises/v1`` from the listed record to this + one: then the record is written under ``-``, and the node on the + listed record's entry, in the same commit. Under a name that is not listed, ``revises=`` goes + on the entry of the listed record it targets. + + Refused before any write: a token that is not a fine-grained one, a name that fails + host-core's rule or has a ``records`` or ``evidence`` segment, an empty title, a BlobRef + output, a signer other than ``host-policy.json``'s, a role no active rule admits, a + ``revises=`` whose type, ``successorNodeId`` or ``targetNodeId`` does not match, and what + host-core's build would refuse once the commit lands. + + Returns ``{name, commit, bundle_url, verify_url, registry_url, written, run}``; ``run`` is + ``None``: the host's workflow deploys the commit, and publish does not wait for it. + """ + document, original = _load(signed, "publish takes what sign prints") + envelope_hash, signer, type_, created_at = _check_signed(document) + if name is not None and notebook is not None: + raise TypeError("publish takes name= or notebook= (for the default name), not both") + if name is None: + if notebook is None: + raise TypeError("publish takes name= or notebook= (whose stem starts the default name)") + name = default_name(notebook, envelope_hash, created_at) + name = check_name(name) + if not isinstance(title, str) or not title.strip(): + _refuse("the title is empty: it becomes the view's subjectTitle") + if not isinstance(role, str) or not role: + _refuse("the role is empty") + node: dict[str, Any] | None = None + node_bytes: bytes | None = None + if revises is not None: + revises_document, revises_original = _load(revises, "revises= takes what attest prints") + node_id, node_type, node_signer, node = _check_node(revises_document, "revises=") + if node_type != _REVISES: + _refuse(f"revises= is an {node_type}, not an {_REVISES}") + if node.get("successorNodeId") != envelope_hash: + _refuse("revises= names another successorNodeId than this record's envelopeHash") + node_bytes = revises_original if revises_original is not None else _serialize(revises_document) + content = original if original is not None else _serialize(document) + host._resolve_token() # refuse a token before any request + + def plan_for(state: _State) -> _Plan | dict[str, Any]: + _bundle_url(state.manifest["origin"], name) + # Every listed record's signed file, read once: host-core lists a hash once (build.ts:65-66). + _check_listed(state) + carried = next((e for e in state.manifest["records"] if state.hash_of(e) == envelope_hash), None) + if carried is not None: + return _receipt(state, carried["name"], None) + _check_signer(state.policy, signer, "the record") + if node is not None: + _check_signer(state.policy, node_signer, "revises=") + _check_admitted(state.policy, signer=signer, type_=type_, role=role) + _check_registry_rules(state, document) + record_name = name + target: dict[str, Any] | None = None + listed = state.entry(name) + if listed is not None: + listed_hash = state.hash_of(listed) + if node is None: + _refuse( + f"{name} is listed with another record (envelopeHash {listed_hash or 'unread'}): pass " + "revises= (what attest printed for attestation/revises/v1 from it to this record) or another name" + ) + if node.get("targetNodeId") != listed_hash: + _refuse(f"revises= targets {node.get('targetNodeId')}, not the record listed as {name}") + target = listed + record_name = check_name(f"{name}-{envelope_hash[:8]}") + if state.entry(record_name) is not None: + _refuse(f"{record_name}, the name a revision of {name} takes, is listed with another record") + _bundle_url(state.manifest["origin"], record_name) + elif node is not None: + target = next( + (e for e in reversed(state.manifest["records"]) if state.hash_of(e) == node.get("targetNodeId")), + None, + ) + if target is None: + _refuse(f"revises= targets {node.get('targetNodeId')}, and no record this host lists has that hash") + manifest = copy.deepcopy(state.manifest) + signed_path = f"{_INPUT_DIRECTORY}/{record_name}.signed.json" + _check_path_unnamed(manifest, signed_path) + files = {signed_path: content} + manifest["records"].append( + { + "name": record_name, + "signed": signed_path, + "attestations": [], + "title": title, + "extensions": {"role": role}, + } + ) + message = f"Publish {record_name}" + if target is not None and node is not None and node_bytes is not None: + node_path = _node_path(target["name"], _REVISES, node_id) + entry = next(e for e in manifest["records"] if e.get("name") == target["name"]) + if node_path not in entry.setdefault("attestations", []): # a listed node file stays as it is + _check_path_unnamed(manifest, node_path) + entry["attestations"].append(node_path) + files[node_path] = node_bytes + message = f"Publish {record_name}, a revision of {target['name']}" + _check_manifest(manifest) + return _Plan(files, manifest, message, record_name) + + return _run(host, plan_for) + + +def publish_attestation( + node: Mapping[str, Any] | str | os.PathLike[str], *, host: GitHubPagesHost, name: str +) -> dict[str, Any]: + """Add what :func:`withdraw` or :func:`attest` printed (or its path) to the record listed as + ``name``, in one commit: the node's file at ``records/.-.json`` and its + path in the entry's ``attestations``. + + Refused before any write: a token that is not a fine-grained one, a claim-to-claim node + (``corroborates``, ``contradicts``), a name the host does not list, a node aimed at another + record (its ``targetNodeId`` is not the listed record's ``envelopeHash``), a signer other + than ``host-policy.json``'s, and a withdrawal or supersession that would leave the record in a + status no policy rule displays. A node already listed on the entry is not written again + (``written: False``). Returns the receipt :func:`publish` returns, for the record. + """ + document, original = _load(node, "publish_attestation takes what withdraw or attest prints") + node_id, type_, signer, body = _check_node(document, "the attestation") + check_name(name) + content = original if original is not None else _serialize(document) + host._resolve_token() + + def plan_for(state: _State) -> _Plan | dict[str, Any]: + _bundle_url(state.manifest["origin"], name) + _check_signer(state.policy, signer, "the attestation") + listed = state.entry(name) + if listed is None: + _refuse(f"host.json lists no record named {name}") + listed_hash = state.hash_of(listed) + if body.get("targetNodeId") != listed_hash: + _refuse(f"the attestation's targetNodeId is not the envelopeHash of the record listed as {name}") + path = _node_path(name, type_, node_id) + if path in (listed.get("attestations") or []): + existing = state.api.read(path, state.head) + try: + same = existing is not None and _parse_host_json(existing).get("nodeId") == node_id + except (ValueError, AttributeError): + same = False + if same: + return _receipt(state, name, None) + _refuse(f"{path} is listed on {name} with another node") + status = _STATUS_AFTER.get(type_) + record = state.signed_of(listed) or {} + if status is not None and not _displayed( + state.policy, + status=status, + signer=str(_signer_of(record).get("identifier", "")) if record else "", + type_=str((record.get("package") or {}).get("type") or "content/analysis/v1"), + extensions=listed.get("extensions") or {}, + ): + _refuse( + f"no rule of host-policy.json displays {name} once it is {status}: the host's display step would " + f"refuse it, and with it every later deploy. Add a rule for {status} records first" + ) + manifest = copy.deepcopy(state.manifest) + _check_path_unnamed(manifest, path) + entry = next(e for e in manifest["records"] if e.get("name") == name) + entry.setdefault("attestations", []).append(path) + _check_manifest(manifest) + return _Plan( + {path: content}, manifest, f"Add the {type_.split('/')[1]} attestation {node_id[:8]} to {name}", name + ) + + return _run(host, plan_for) diff --git a/src/typedstandards/errors.py b/src/typedstandards/errors.py index 4946046..f46ad49 100644 --- a/src/typedstandards/errors.py +++ b/src/typedstandards/errors.py @@ -3,6 +3,8 @@ The CLI's exit codes 1 to 4 (``packages/cli/src/errors.ts`` in typedstandards) map to the four subclasses of :class:`CliError`, each carrying the exit code and the CLI's stderr text. A missing or too-old Node is a :class:`NodeLocatorError`, raised before the CLI runs. +``publish`` and ``publish_attestation`` raise :class:`PublishError`, and +:class:`PublishRefusedError` for a refusal made before any write. """ from __future__ import annotations @@ -56,3 +58,14 @@ class SeedError(CliError): class InternalError(CliError): """Exit 4: an internal error in the CLI.""" + + +class PublishError(RuntimeError): + """``publish`` or ``publish_attestation`` did not complete: the GitHub API answered with an + error, or the branch moved again after the one retry. The message names the request and + GitHub's own message, never a credential.""" + + +class PublishRefusedError(PublishError): + """``publish`` or ``publish_attestation`` refused the call before any write request: the + token, the name, the title, the record, the role or the host's files did not pass a check.""" diff --git a/tests/conftest.py b/tests/conftest.py index 1131288..22cd056 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -9,6 +9,7 @@ from typing import Any import pytest +from publish_support import Docs, make_docs from support import SEED_VARIABLE, NetworkBlocked, fresh_seed_b64 @@ -89,3 +90,9 @@ def communicate(self, input: Any = None, timeout: Any = None) -> Any: monkeypatch.setattr(subprocess, "Popen", RecordingPopen) return calls + + +@pytest.fixture(scope="session") +def docs(tmp_path_factory: pytest.TempPathFactory) -> Docs: + """The publish tests' records and attestations, signed once by the vendored CLI.""" + return make_docs(tmp_path_factory.mktemp("publish")) diff --git a/tests/fixtures/README.md b/tests/fixtures/README.md index 8d46096..a0718af 100644 --- a/tests/fixtures/README.md +++ b/tests/fixtures/README.md @@ -8,14 +8,21 @@ provenance is recorded here. |---|---|---|---|---|---| | `reference-golden.json` | `npstorey/typedstandards`, `packages/produce-core/src/__fixtures__/reference-golden.json` | `116882a` | `ea75a1d` | `d2bcfc2bc017b07502b3b00c3aa16de402df134128a374b4582650b79fb501c1` | `tests/test_golden.py` | | `first-note.bundle.json` | `npstorey/typedstandards-host-template`, `docs/bundles/first-note.bundle.json` | `70bfd18` | `26dff9b` | `cb11d2a229c9695db6c7f4d6c9349ccee14f14af6c39844886480699ba2401ba` | `tests/test_d9.py` | +| `template-host.json` | `npstorey/typedstandards-host-template`, `host.json` | `70bfd18` | `26dff9b` | `8a80c9ea344196ac2b27b3a91eb3439f9217e72c5491b18540d253925b28a6fe` | `tests/test_publish.py` | +| `template-host-policy.json` | `npstorey/typedstandards-host-template`, `host-policy.json` | `70bfd18` | `8b88a98` | `9150db40fee1f2e17bf09d128c080f0bee9a3ac0c8bda9dbd0467e11a144c0f8` | `tests/test_publish.py` | -Re-derive either copy with `git -C show : > tests/fixtures/`. +Re-derive any copy with `git -C show : > tests/fixtures/`. - **`reference-golden.json`** holds 9 envelope cases and 6 attestation cases, captured from the reference implementation as its `_meta` records. No npm tarball ships it, so the tests carry this copy. `tests/test_golden.py` replays the 9 envelope cases through `sign` and the `withdraws` case through `withdraw`. It carries the reference platform's own identifiers, as it does in its source repository. +- **`template-host.json`** and **`template-host-policy.json`** are the files a host made from the + template starts with. `tests/test_publish.py` starts its fake host's `main` from them: the + manifest as is, and the policy with its `signer` set to the test key and `notebook` added to the + active rule's roles, as the template's publish mode does (typedstandards#141 G0-3). One test + keeps the policy as shipped and shows `publish`'s default role refused under it. - **`first-note.bundle.json`** is the bundle the host template serves, written by `@typedstandards/host-core` 0.1.1 with a top-level `trustRegistry`. `tests/test_d9.py` verifies it through the wrapper, which drops that key before the CLI sees it (typedstandards#136). diff --git a/tests/fixtures/template-host-policy.json b/tests/fixtures/template-host-policy.json new file mode 100644 index 0000000..3fa9732 --- /dev/null +++ b/tests/fixtures/template-host-policy.json @@ -0,0 +1,23 @@ +{ + "$comment": "The display policy: which records a page shows, and as what. It is this host's own rule, not signed and not verified. Every rule names its statuses, so a status no rule names is refused. A copy of the template changes signer to its own did:key.", + "signer": "did:key:z6Mks7BK2kyVhoPY3ayt6ALKeZY5eCbu64XyQuje9gTUxiUB", + "type": "content/analysis/v1", + "display": [ + { + "$comment": "An active note is shown as current.", + "status": "active", + "extensions": { + "role": [ + "note" + ] + }, + "as": "current" + }, + { + "$comment": "A withdrawn record stays listed, marked withdrawn.", + "status": "withdrawn", + "as": "withdrawn" + } + ], + "unmatched": "refuse" +} diff --git a/tests/fixtures/template-host.json b/tests/fixtures/template-host.json new file mode 100644 index 0000000..bb63417 --- /dev/null +++ b/tests/fixtures/template-host.json @@ -0,0 +1,20 @@ +{ + "$comment": "The host manifest. typedstandards-host build writes docs/ from it; every path is relative to this file. A copy of the template changes origin to the URL its Pages site is served at, and replaces the record.", + "origin": "https://host-template.typedstandards.org", + "visibility": "public", + "registry": { + "$comment": "This template's own statement about the signing key of its example record, served at https://host-template.typedstandards.org/.well-known/typed-publisher.json. The key is a pseudonymous did:key, generated for one signature and deleted after it. This registry is not a Typed Standards record, and not an endorsement by the Typed Standards specification or by typedstandards.org, although this host is a subdomain of it. activatedAt is the earliest createdAt among the records this host serves, as the signer states it." + }, + "index": { + "$comment": "The records this template serves: one example record." + }, + "records": [ + { + "name": "first-note", + "signed": "records/first-note.signed.json", + "attestations": [], + "title": "A first signed note", + "extensions": { "role": "note" } + } + ] +} diff --git a/tests/publish_support.py b/tests/publish_support.py new file mode 100644 index 0000000..28cd389 --- /dev/null +++ b/tests/publish_support.py @@ -0,0 +1,146 @@ +"""Shared by the publish tests: a visibly fake token, the documents the vendored CLI signs for +them (``docs``, a session fixture in ``conftest.py``), and a fake host whose files start as the +template's ``host.json`` and ``host-policy.json`` at ``70bfd18`` (``fixtures/template-host*.json``). +""" + +from __future__ import annotations + +import json +import sys +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +import pytest +from support import FIXTURES, SEED_VARIABLE, analysis_input, fresh_seed_b64, synthetic_notebook + +import typedstandards as ts + +sys.path.insert(0, str(Path(__file__).parent.parent / "scripts")) +from github_stub import FakeGitHub, dumps # noqa: E402 + +#: A visibly fake token: the fine-grained prefix, then text no real token has. +TOKEN = "github_pat_TESTONLY_fake_value_for_tests" +ORIGIN = "https://host-template.typedstandards.org" + + +@dataclass +class Docs: + first: dict[str, Any] # the record a host lists as "dog-licensing" + second: dict[str, Any] # a rerun: another envelopeHash + blobref: dict[str, Any] # output signed by reference + foreign: dict[str, Any] # signed under another key + revises: dict[str, Any] # attestation/revises/v1: target first, successor second + withdrawal: dict[str, Any] # attestation/withdraws/v1 of first + corroboration: dict[str, Any] # a claim-to-claim node on first + signer: str + note: dict[str, Any] # the template's example entry, first-note, re-signed by the test key + renamed: dict[str, Any] # the test key, another signer.displayName + supersedes: dict[str, Any] # attestation/supersedes/v1: target first, successor second + third: dict[str, Any] # another record by the test key + + +def make_docs(directory: Path) -> Docs: + """Sign the publish tests' documents through the vendored CLI under fresh seeds.""" + notebook = directory / "dog-licensing.ipynb" + notebook.write_text(synthetic_notebook(), encoding="utf-8") + with pytest.MonkeyPatch.context() as mp: + mp.setenv(SEED_VARIABLE, fresh_seed_b64()) + first = ts.sign(analysis_input(), output_file=notebook) + second = ts.sign(analysis_input(), output_file=notebook) + blobref = ts.sign( + analysis_input(), + output_file=notebook, + output_url="https://files.example.org/dog-licensing.ipynb", + content_type="application/x-ipynb+json", + ) + signer = first["package"]["signer"] + revises = ts.attest( + { + "type": "attestation/revises/v1", + "targetNodeId": first["envelopeHash"], + "successorNodeId": second["envelopeHash"], + "signer": signer, + } + ) + withdrawal = ts.withdraw( + {"targetNodeId": first["envelopeHash"], "reason": "A test withdrawal.", "signer": signer} + ) + corroboration = ts.attest( + { + "type": "attestation/corroborates/v1", + "targetNodeId": first["envelopeHash"], + "scope": "the whole record", + "signer": signer, + } + ) + supersedes = ts.attest( + { + "type": "attestation/supersedes/v1", + "targetNodeId": first["envelopeHash"], + "successorNodeId": second["envelopeHash"], + "signer": signer, + } + ) + note = ts.sign(analysis_input(output="A first signed note.")) + third = ts.sign(analysis_input(output="A third record.")) + renamed_signer = {"bindingTier": "pseudonymous", "displayName": "Another display name"} + renamed = ts.sign(analysis_input(signer=renamed_signer), output_file=notebook) + mp.setenv(SEED_VARIABLE, fresh_seed_b64()) + foreign = ts.sign(analysis_input(), output_file=notebook) + return Docs( + first, + second, + blobref, + foreign, + revises, + withdrawal, + corroboration, + signer["identifier"], + note, + renamed, + supersedes, + third, + ) + + +def template_manifest() -> dict[str, Any]: + return json.loads((FIXTURES / "template-host.json").read_text(encoding="utf-8")) + + +def template_policy(signer: str) -> dict[str, Any]: + """The template's policy with this test's signer, and ``notebook`` in the active rule (P1).""" + value = json.loads((FIXTURES / "template-host-policy.json").read_text(encoding="utf-8")) + value["signer"] = signer + value["display"][0]["extensions"]["role"].append("notebook") + return value + + +def github( + docs: Docs, + *, + listed: dict[str, dict[str, Any]] | None = None, + policy: dict[str, Any] | None = None, + manifest: dict[str, Any] | None = None, +) -> FakeGitHub: + """A host with the template's files, plus each of ``listed`` (name to signed document).""" + files = { + "host.json": (FIXTURES / "template-host.json").read_bytes(), + "host-policy.json": dumps(policy if policy is not None else template_policy(docs.signer)), + # The template's example entry, its file re-signed by this test's key: a host serves one signer. + "records/first-note.signed.json": dumps(docs.note), + } + value = manifest if manifest is not None else template_manifest() + for name, signed in (listed or {}).items(): + path = f"records/{name}.signed.json" + files[path] = dumps(signed) + value["records"].append( + {"name": name, "signed": path, "attestations": [], "title": name, "extensions": {"role": "notebook"}} + ) + if listed or manifest is not None: + files["host.json"] = dumps(value) + return FakeGitHub(files, token=TOKEN) + + +def host(gh: FakeGitHub, **kwargs: Any) -> ts.GitHubPagesHost: + return ts.GitHubPagesHost(gh.repository, token=TOKEN, transport=gh.transport(), **kwargs) diff --git a/tests/test_publish.py b/tests/test_publish.py new file mode 100644 index 0000000..289d7f0 --- /dev/null +++ b/tests/test_publish.py @@ -0,0 +1,763 @@ +"""``publish`` and ``publish_attestation`` (typedstandards#141 P2, acceptance 1 and 3). + +Every test drives a fake GitHub API (``scripts/github_stub.py``) through ``httpx.MockTransport``; +the autouse guard fails any real connection. The host's files start as the template's +``host.json`` and ``host-policy.json`` at ``70bfd18`` (``fixtures/template-host*.json``), with the +policy's signer set to the test key and ``notebook`` added to the active rule, as P1 changes the +template. The records are signed by the vendored CLI under fresh seeds (``docs``, ``conftest.py``). + +Every refusal is asserted to happen before any write request: no POST, PATCH, PUT or DELETE +reaches the transport. +""" + +from __future__ import annotations + +import copy +import json +from pathlib import Path +from typing import Any + +import pytest +from github_stub import FakeGitHub, dumps # publish_support puts scripts/ on the path +from publish_support import ORIGIN, TOKEN, Docs, github, host, template_manifest, template_policy +from support import FIXTURES + +import typedstandards as ts + +WRITE_METHODS = {"POST", "PATCH", "PUT", "DELETE"} + + +def paths(gh: FakeGitHub, head: str, reads: tuple[str, ...] = ("records/first-note.signed.json",)) -> list: + """The reads a publish makes from the branch head ``head``: the ref, the commit, host.json, + host-policy.json, then each listed record's signed file (``reads``), once each, in the + manifest's order. The template's ``first-note`` is listed unless a test says otherwise.""" + r = f"/repos/{gh.repository}" + return [ + ("GET", f"{r}/git/ref/heads/main"), + ("GET", f"{r}/git/commits/{head}"), + ("GET", f"{r}/contents/host.json"), + ("GET", f"{r}/contents/host-policy.json"), + ] + [("GET", f"{r}/contents/{path}") for path in reads] + + +def writes(gh: FakeGitHub, blobs: int) -> list[tuple[str, str]]: + r = f"/repos/{gh.repository}" + return [("POST", f"{r}/git/blobs")] * blobs + [ + ("POST", f"{r}/git/trees"), + ("POST", f"{r}/git/commits"), + ("PATCH", f"{r}/git/refs/heads/main"), + ] + + +def assert_nothing_written(gh: FakeGitHub) -> None: + assert [r for r in gh.requests if r[0] in WRITE_METHODS] == [] + + +def assert_git_data_only(gh: FakeGitHub) -> None: + """Acceptance 3: no PUT or DELETE, and no write to the contents API.""" + for method, path in gh.requests: + assert method in {"GET", "POST", "PATCH"}, (method, path) + if method != "GET": + assert "/contents/" not in path, (method, path) + assert "/git/" in path, (method, path) + + +def default_name(signed: dict[str, Any], stem: str = "dog-licensing") -> str: + return f"{stem}/{signed['package']['metadata']['createdAt'][:10]}-{signed['envelopeHash'][:8]}" + + +# --- a new record --------------------------------------------------------------------------------- + + +def test_a_new_record_is_two_blobs_one_tree_one_commit_and_one_fast_forward(docs: Docs) -> None: + gh = github(docs) + start = gh.head + receipt = ts.publish(docs.first, host=host(gh), name="dog-licensing", title="Dog licensing") + + assert gh.requests == paths(gh, start) + writes(gh, blobs=2) + assert_git_data_only(gh) + commit = gh.commits[gh.head] + assert commit["parents"] == [start] + files = gh.files_at() + assert json.loads(files["records/dog-licensing.signed.json"]) == docs.first + manifest = json.loads(files["host.json"]) + before = template_manifest() + assert {k: v for k, v in manifest.items() if k != "records"} == {k: v for k, v in before.items() if k != "records"} + assert manifest["records"] == before["records"] + [ + { + "name": "dog-licensing", + "signed": "records/dog-licensing.signed.json", + "attestations": [], + "title": "Dog licensing", + "extensions": {"role": "notebook"}, + } + ] + assert files["host.json"].endswith(b"}\n") and b'\n "origin"' in files["host.json"] + bundle_url = f"{ORIGIN}/bundles/dog-licensing.bundle.json" + assert receipt == { + "name": "dog-licensing", + "commit": gh.head, + "bundle_url": bundle_url, + "verify_url": "https://typedstandards.org/verify?url=" + bundle_url.replace(":", "%3A").replace("/", "%2F"), + "registry_url": f"{ORIGIN}/.well-known/typed-publisher.json", + "written": True, + "run": None, + } + + +def test_the_receipts_verify_url_is_the_badges_link(docs: Docs) -> None: + from typedstandards._badge import verify_href + + gh = github(docs) + receipt = ts.publish(docs.first, host=host(gh), name="dog-licensing", title="Dog licensing") + assert receipt["verify_url"] == verify_href(receipt["bundle_url"]) + + +def test_the_default_name_is_the_stem_the_date_and_eight_hex(docs: Docs) -> None: + gh = github(docs) + receipt = ts.publish(docs.first, host=host(gh), notebook="work/dog-licensing.ipynb", title="Dog licensing") + assert receipt["name"] == default_name(docs.first) + assert f"records/{default_name(docs.first)}.signed.json" in gh.files_at() + + +def test_a_rerun_under_the_default_name_gets_a_new_name(docs: Docs) -> None: + gh = github(docs) + first = ts.publish(docs.first, host=host(gh), notebook="dog-licensing.ipynb", title="Dog licensing") + second = ts.publish(docs.second, host=host(gh), notebook="dog-licensing.ipynb", title="Dog licensing") + assert first["name"] != second["name"] and second["written"] is True + assert [r["name"] for r in gh.json_at("host.json")["records"]][-2:] == [first["name"], second["name"]] + + +def test_a_signed_file_path_is_published_as_its_bytes(docs: Docs, tmp_path: Path) -> None: + path = tmp_path / "first.signed.json" + path.write_bytes(json.dumps(docs.first).encode()) + gh = github(docs) + ts.publish(path, host=host(gh), name="dog-licensing", title="Dog licensing") + assert gh.files_at()["records/dog-licensing.signed.json"] == path.read_bytes() + + +def test_name_and_notebook_are_one_or_the_other(docs: Docs) -> None: + gh = github(docs) + with pytest.raises(TypeError, match="name= or notebook="): + ts.publish(docs.first, host=host(gh), title="Dog licensing") + with pytest.raises(TypeError, match="not both"): + ts.publish(docs.first, host=host(gh), name="a", notebook="a.ipynb", title="Dog licensing") + assert gh.requests == [] + + +# --- a listed hash, a listed name ------------------------------------------------------------- + + +def test_a_listed_hash_writes_nothing(docs: Docs) -> None: + gh = github(docs, listed={"dog-licensing": docs.first}) + start = gh.head + receipt = ts.publish(docs.first, host=host(gh), name="dog-licensing", title="Dog licensing") + assert receipt["written"] is False and receipt["commit"] is None and receipt["name"] == "dog-licensing" + assert gh.head == start + assert_nothing_written(gh) + assert gh.requests[-1] == ("GET", f"/repos/{gh.repository}/contents/records/dog-licensing.signed.json") + + +def test_a_repeated_publish_writes_once(docs: Docs) -> None: + gh = github(docs) + assert ts.publish(docs.first, host=host(gh), notebook="dog-licensing.ipynb", title="T")["written"] is True + count = len(gh.writes()) + assert ts.publish(docs.first, host=host(gh), notebook="dog-licensing.ipynb", title="T")["written"] is False + assert len(gh.writes()) == count + + +def test_a_listed_name_with_another_hash_is_refused_without_revises(docs: Docs) -> None: + gh = github(docs, listed={"dog-licensing": docs.first}) + with pytest.raises(ts.PublishRefusedError, match="dog-licensing is listed with another record"): + ts.publish(docs.second, host=host(gh), name="dog-licensing", title="Dog licensing") + assert_nothing_written(gh) + + +def test_a_listed_name_with_another_hash_is_written_with_revises(docs: Docs) -> None: + gh = github(docs, listed={"dog-licensing": docs.first}) + start = gh.head + receipt = ts.publish( + docs.second, host=host(gh), name="dog-licensing", title="Dog licensing, rerun", revises=docs.revises + ) + + derived = f"dog-licensing-{docs.second['envelopeHash'][:8]}" + assert receipt["name"] == derived and receipt["written"] is True + r = f"/repos/{gh.repository}" + assert gh.requests == paths(gh, start) + [("GET", f"{r}/contents/records/dog-licensing.signed.json")] + writes( + gh, blobs=3 + ) + assert_git_data_only(gh) + assert gh.commits[gh.head]["parents"] == [start] + files = gh.files_at() + node_path = f"records/dog-licensing.revises-{docs.revises['nodeId'][:8]}.json" + assert json.loads(files[node_path]) == docs.revises + assert json.loads(files[f"records/{derived}.signed.json"]) == docs.second + entries = {e["name"]: e for e in json.loads(files["host.json"])["records"]} + assert entries["dog-licensing"]["attestations"] == [node_path] + assert entries[derived]["attestations"] == [] + assert entries[derived]["title"] == "Dog licensing, rerun" + + +def test_a_repeated_revise_writes_nothing(docs: Docs) -> None: + gh = github(docs, listed={"dog-licensing": docs.first}) + kwargs = {"name": "dog-licensing", "title": "Rerun", "revises": docs.revises} + ts.publish(docs.second, host=host(gh), **kwargs) + count = len(gh.writes()) + assert ts.publish(docs.second, host=host(gh), **kwargs)["written"] is False + assert len(gh.writes()) == count + + +def test_revises_under_the_default_name_goes_on_its_targets_entry(docs: Docs) -> None: + gh = github(docs, listed={default_name(docs.first): docs.first}) + receipt = ts.publish( + docs.second, host=host(gh), notebook="dog-licensing.ipynb", title="Rerun", revises=docs.revises + ) + assert receipt["name"] == default_name(docs.second) + entries = {e["name"]: e for e in gh.json_at("host.json")["records"]} + node_path = f"records/{default_name(docs.first)}.revises-{docs.revises['nodeId'][:8]}.json" + assert entries[default_name(docs.first)]["attestations"] == [node_path] + assert len([r for r in gh.requests if r[0] == "PATCH"]) == 1 + + +@pytest.mark.parametrize("case", ["not a revises node", "another successor", "another target"]) +def test_revises_whose_fields_do_not_match_is_refused(docs: Docs, case: str) -> None: + node = { + "not a revises node": docs.withdrawal, + "another successor": docs.revises, + "another target": docs.revises, + }[case] + signed = docs.first if case == "another successor" else docs.second + listed = {"dog-licensing": docs.third if case == "another target" else docs.first} + gh = github(docs, listed=listed) + with pytest.raises(ts.PublishRefusedError, match="revises"): + ts.publish(signed, host=host(gh), name="dog-licensing", title="Rerun", revises=node) + assert_nothing_written(gh) + + +def test_revises_with_no_listed_target_is_refused(docs: Docs) -> None: + gh = github(docs) + with pytest.raises(ts.PublishRefusedError, match="no record this host lists"): + ts.publish(docs.second, host=host(gh), notebook="dog-licensing.ipynb", title="Rerun", revises=docs.revises) + assert_nothing_written(gh) + + +# --- names -------------------------------------------------------------------------------------- + + +@pytest.mark.parametrize("name", ["records", "records/dog-licensing", "notes/evidence/x", "a/b/records"]) +def test_a_name_with_a_records_or_evidence_segment_is_refused(docs: Docs, name: str) -> None: + gh = github(docs) + with pytest.raises(ts.PublishRefusedError, match="records|evidence"): + ts.publish(docs.first, host=host(gh), name=name, title="Dog licensing") + assert gh.requests == [] + + +@pytest.mark.parametrize("name", ["", "dog licensing", "../x", "x/../y", "x/./y", "a//b", "x/", "/x", "café", "a:b"]) +def test_a_name_failing_host_cores_rule_is_refused(docs: Docs, name: str) -> None: + gh = github(docs) + with pytest.raises(ts.PublishRefusedError, match="segments of letters, digits"): + ts.publish(docs.first, host=host(gh), name=name, title="Dog licensing") + assert gh.requests == [] + + +def test_a_notebook_stem_failing_the_rule_is_refused(docs: Docs) -> None: + gh = github(docs) + with pytest.raises(ts.PublishRefusedError, match="segments of letters, digits"): + ts.publish(docs.first, host=host(gh), notebook="Dog licensing.ipynb", title="Dog licensing") + assert gh.requests == [] + + +def test_an_origin_with_a_records_segment_is_refused(docs: Docs) -> None: + manifest = template_manifest() + manifest["origin"] = "https://example-owner.github.io/records" + gh = github(docs, manifest=manifest) + with pytest.raises(ts.PublishRefusedError, match="records"): + ts.publish(docs.first, host=host(gh), name="dog-licensing", title="Dog licensing") + assert_nothing_written(gh) + + +# --- the record, the signer, the role, the title --------------------------------------------------- + + +def test_a_blobref_output_is_refused(docs: Docs) -> None: + gh = github(docs) + with pytest.raises(ts.PublishRefusedError, match="BlobRef"): + ts.publish(docs.blobref, host=host(gh), name="dog-licensing", title="Dog licensing") + assert gh.requests == [] + + +@pytest.mark.parametrize( + "value", + [{"package": {}, "envelopeHash": "0" * 64}, {"package": {}, "envelopeHash": "XYZ", "signature": {}}, ["x"]], +) +def test_a_document_sign_did_not_print_is_refused(docs: Docs, value: Any) -> None: + gh = github(docs) + with pytest.raises(ts.PublishRefusedError, match="what sign prints|envelopeHash"): + ts.publish(value, host=host(gh), name="dog-licensing", title="Dog licensing") + assert gh.requests == [] + + +def test_a_signer_other_than_the_policys_is_refused(docs: Docs) -> None: + gh = github(docs) + with pytest.raises(ts.PublishRefusedError, match="signer"): + ts.publish(docs.foreign, host=host(gh), name="dog-licensing", title="Dog licensing") + assert_nothing_written(gh) + + +@pytest.mark.parametrize("role", ["claim", "note-book"]) +def test_a_role_no_active_rule_admits_is_refused(docs: Docs, role: str) -> None: + gh = github(docs) + with pytest.raises(ts.PublishRefusedError, match=f"no active rule .* admits the role {role}"): + ts.publish(docs.first, host=host(gh), name="dog-licensing", title="Dog licensing", role=role) + assert_nothing_written(gh) + + +def test_a_role_only_a_withdrawn_rule_admits_is_refused(docs: Docs) -> None: + value = template_policy(docs.signer) + value["display"] = [ + {"status": "active", "extensions": {"role": ["note"]}, "as": "current"}, + {"status": "withdrawn", "extensions": {"role": ["notebook"]}, "as": "withdrawn"}, + ] + gh = github(docs, policy=value) + with pytest.raises(ts.PublishRefusedError, match="admits the role notebook"): + ts.publish(docs.first, host=host(gh), name="dog-licensing", title="Dog licensing") + assert_nothing_written(gh) + + +def test_the_template_policy_as_shipped_refuses_the_default_role(docs: Docs) -> None: + """At 70bfd18 the active rule admits note only; P1 adds notebook (G0-3).""" + value = json.loads((FIXTURES / "template-host-policy.json").read_text(encoding="utf-8")) + value["signer"] = docs.signer + gh = github(docs, policy=value) + with pytest.raises(ts.PublishRefusedError, match="admits the role notebook"): + ts.publish(docs.first, host=host(gh), name="dog-licensing", title="Dog licensing") + receipt = ts.publish(docs.first, host=host(gh), name="dog-licensing", title="Dog licensing", role="note") + assert receipt["written"] is True + + +def test_a_type_the_policy_does_not_name_is_refused(docs: Docs) -> None: + value = template_policy(docs.signer) + value["type"] = "content/claim/v1" + gh = github(docs, policy=value) + with pytest.raises(ts.PublishRefusedError, match="type"): + ts.publish(docs.first, host=host(gh), name="dog-licensing", title="Dog licensing") + assert_nothing_written(gh) + + +@pytest.mark.parametrize("title", ["", " "]) +def test_an_empty_title_is_refused(docs: Docs, title: str) -> None: + gh = github(docs) + with pytest.raises(ts.PublishRefusedError, match="title"): + ts.publish(docs.first, host=host(gh), name="dog-licensing", title=title) + assert gh.requests == [] + + +# --- the ref update -------------------------------------------------------------------------------- + + +def test_a_non_fast_forward_is_retried_once_after_rereading_the_head(docs: Docs) -> None: + gh = github(docs) + start = gh.head + gh.concurrent_writes = 1 + receipt = ts.publish(docs.first, host=host(gh), name="dog-licensing", title="Dog licensing") + moved = gh.commits[gh.head]["parents"][0] + r = f"/repos/{gh.repository}" + assert gh.requests == ( + paths(gh, start) + + writes(gh, blobs=2) + + [("GET", f"{r}/git/ref/heads/main")] + + paths(gh, moved)[1:] + + writes(gh, blobs=2) + ) + assert receipt["written"] is True and receipt["commit"] == gh.head + assert gh.commits[moved]["message"] == "another writer" + assert "other/" in "".join(gh.files_at()) # the other writer's file is kept + + +def test_a_second_non_fast_forward_is_an_error(docs: Docs) -> None: + gh = github(docs) + gh.concurrent_writes = 2 + with pytest.raises(ts.PublishError, match="not a fast forward") as caught: + ts.publish(docs.first, host=host(gh), name="dog-licensing", title="Dog licensing") + assert not isinstance(caught.value, ts.PublishRefusedError) + assert [m for m, _ in gh.requests].count("PATCH") == 2 + assert gh.requests[-1] == ("GET", f"/repos/{gh.repository}/git/ref/heads/main") + assert "dog-licensing" not in json.dumps([e["name"] for e in gh.json_at("host.json")["records"]]) + + +def test_a_ref_update_that_errored_but_landed_is_not_retried(docs: Docs) -> None: + gh = github(docs) + gh.landed_but_failed = [502] + receipt = ts.publish(docs.first, host=host(gh), name="dog-licensing", title="Dog licensing") + assert [m for m, _ in gh.requests].count("PATCH") == 1 + assert gh.requests[-1] == ("GET", f"/repos/{gh.repository}/git/ref/heads/main") + assert receipt["written"] is True and receipt["commit"] == gh.head + + +def test_a_non_fast_forward_retry_sees_a_record_another_writer_listed(docs: Docs) -> None: + gh = github(docs) + + def other_writer_publishes_the_same(request: Any) -> None: + if request.method == "PATCH" and gh.hook is not None: + gh.hook = None + files = dict(gh.files_at()) + manifest = json.loads(files["host.json"]) + manifest["records"].append( + { + "name": "dog-licensing", + "signed": "records/dog-licensing.signed.json", + "attestations": [], + "title": "T", + } + ) + files["host.json"] = dumps(manifest) + files["records/dog-licensing.signed.json"] = dumps(docs.first) + tree, commit = gh._id("t"), gh._id("c") + gh.trees[tree] = files + gh.commits[commit] = {"tree": tree, "parents": [gh.head], "message": "another writer"} + gh.head = commit + return None + + gh.hook = other_writer_publishes_the_same + receipt = ts.publish(docs.first, host=host(gh), name="dog-licensing", title="Dog licensing") + assert receipt["written"] is False and gh.commits[gh.head]["message"] == "another writer" + assert [m for m, _ in gh.requests].count("PATCH") == 1 + + +def test_an_api_error_names_the_request_and_githubs_message(docs: Docs) -> None: + gh = github(docs) + bad = ts.GitHubPagesHost(gh.repository, token=TOKEN + "x", transport=gh.transport()) + with pytest.raises(ts.PublishError, match=r"GET .*/git/ref/heads/main answered 401: Bad credentials"): + ts.publish(docs.first, host=bad, name="dog-licensing", title="Dog licensing") + assert_nothing_written(gh) + + +# --- publish_attestation ----------------------------------------------------------------------- + + +def test_a_withdrawal_is_one_commit_on_its_records_entry(docs: Docs) -> None: + gh = github(docs, listed={"dog-licensing": docs.first}) + start = gh.head + receipt = ts.publish_attestation(docs.withdrawal, host=host(gh), name="dog-licensing") + r = f"/repos/{gh.repository}" + assert gh.requests == paths(gh, start, reads=()) + [ + ("GET", f"{r}/contents/records/dog-licensing.signed.json") + ] + writes(gh, blobs=2) + assert_git_data_only(gh) + node_path = f"records/dog-licensing.withdraws-{docs.withdrawal['nodeId'][:8]}.json" + assert json.loads(gh.files_at()[node_path]) == docs.withdrawal + entry = next(e for e in gh.json_at("host.json")["records"] if e["name"] == "dog-licensing") + assert entry["attestations"] == [node_path] + assert receipt["name"] == "dog-licensing" and receipt["written"] is True and receipt["commit"] == gh.head + assert receipt["bundle_url"] == f"{ORIGIN}/bundles/dog-licensing.bundle.json" + + +def test_a_listed_attestation_writes_nothing(docs: Docs) -> None: + gh = github(docs, listed={"dog-licensing": docs.first}) + ts.publish_attestation(docs.withdrawal, host=host(gh), name="dog-licensing") + count = len(gh.writes()) + assert ts.publish_attestation(docs.withdrawal, host=host(gh), name="dog-licensing")["written"] is False + assert len(gh.writes()) == count + + +def test_a_claim_to_claim_node_is_refused(docs: Docs) -> None: + gh = github(docs, listed={"dog-licensing": docs.first}) + with pytest.raises(ts.PublishRefusedError, match="claim-to-claim"): + ts.publish_attestation(docs.corroboration, host=host(gh), name="dog-licensing") + assert gh.requests == [] + + +def test_an_attestation_aimed_at_another_record_is_refused(docs: Docs) -> None: + gh = github(docs, listed={"dog-licensing": docs.second}) + with pytest.raises(ts.PublishRefusedError, match="targetNodeId"): + ts.publish_attestation(docs.withdrawal, host=host(gh), name="dog-licensing") + assert_nothing_written(gh) + + +def test_an_attestation_for_an_unlisted_name_is_refused(docs: Docs) -> None: + gh = github(docs) + with pytest.raises(ts.PublishRefusedError, match="lists no record named dog-licensing"): + ts.publish_attestation(docs.withdrawal, host=host(gh), name="dog-licensing") + assert_nothing_written(gh) + + +def test_an_attestation_by_another_signer_is_refused(docs: Docs) -> None: + gh = github( + docs, + listed={"dog-licensing": docs.first}, + policy=template_policy(docs.foreign["package"]["signer"]["identifier"]), + ) + with pytest.raises(ts.PublishRefusedError, match="signer"): + ts.publish_attestation(docs.withdrawal, host=host(gh), name="dog-licensing") + assert_nothing_written(gh) + + +# --- acceptance 3 --------------------------------------------------------------------------------- + + +def test_every_write_is_a_git_data_call(docs: Docs) -> None: + gh = github(docs, listed={"dog-licensing": docs.first}) + gh.concurrent_writes = 1 + ts.publish(docs.second, host=host(gh), name="dog-licensing", title="Rerun", revises=docs.revises) + ts.publish_attestation(docs.withdrawal, host=host(gh), name="dog-licensing") + ts.publish(docs.third, host=host(gh), notebook="dog-licensing.ipynb", title="Another") + assert_git_data_only(gh) + assert {m for m, _ in gh.requests} == {"GET", "POST", "PATCH"} + patches = [m for m, _ in gh.requests].count("PATCH") + commits = gh.requests.count(("POST", f"/repos/{gh.repository}/git/commits")) + assert commits == patches == 4 # one commit per ref update: three calls, one of them retried + + +def test_the_source_writes_through_the_git_data_api_only() -> None: + """No module sends a PUT or a DELETE, and the contents API is only read (a GET).""" + for path in sorted((Path(ts.__file__).parent).glob("*.py")): + source = path.read_text(encoding="utf-8") + for word in ('"PUT"', "'PUT'", '"DELETE"', "'DELETE'", ".put(", ".delete("): + assert word not in source, (path.name, word) + for number, line in enumerate(source.splitlines(), 1): + if "/contents/" in line and not line.lstrip().startswith("#"): + assert '"GET"' in line, f"{path.name}:{number}: {line.strip()}" + + +def test_a_policy_document_copy_is_not_changed(docs: Docs) -> None: + gh = github(docs) + before = copy.deepcopy(gh.files_at()["host-policy.json"]) + ts.publish(docs.first, host=host(gh), name="dog-licensing", title="Dog licensing") + assert gh.files_at()["host-policy.json"] == before + + +# --- fixture provenance --------------------------------------------------------------------------- + +TEMPLATE_PINNED = { + "template-host.json": "8a80c9ea344196ac2b27b3a91eb3439f9217e72c5491b18540d253925b28a6fe", + "template-host-policy.json": "9150db40fee1f2e17bf09d128c080f0bee9a3ac0c8bda9dbd0467e11a144c0f8", +} + + +@pytest.mark.parametrize("name", TEMPLATE_PINNED) +def test_template_fixture_is_the_verbatim_copy(name: str) -> None: + import hashlib + + assert hashlib.sha256((FIXTURES / name).read_bytes()).hexdigest() == TEMPLATE_PINNED[name] + + +def test_publishing_needs_no_node(docs: Docs, monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + monkeypatch.setenv(ts.NODE_OVERRIDE, str(tmp_path / "no-node-here")) + with pytest.raises(ts.NodeLocatorError): + ts.locate_node() + gh = github(docs) + assert ts.publish(docs.first, host=host(gh), name="dog-licensing", title="T")["written"] is True + assert ts.publish_attestation(docs.withdrawal, host=host(gh), name="dog-licensing")["written"] is True + + +def test_the_first_publish_to_a_copy_in_its_starting_state(docs: Docs) -> None: + """A publish-mode copy starts with ``"records": []`` and no ``records/`` directory (the template's + README, setup step 5, at P1's 09fb6a5): the first publish appends the first entry, and the tree + on ``base_tree`` creates ``records/.signed.json``.""" + empty = template_manifest() + empty["records"] = [] + files = {"host.json": dumps(empty), "host-policy.json": dumps(template_policy(docs.signer))} + gh = FakeGitHub(files, token=TOKEN) + assert not any(path.startswith("records/") for path in gh.files_at()) + start = gh.head + receipt = ts.publish(docs.first, host=host(gh), notebook="dog-licensing.ipynb", title="Dog licensing") + assert gh.requests == paths(gh, start, reads=()) + writes(gh, blobs=2) + name = receipt["name"] + assert gh.json_at("host.json")["records"] == [ + { + "name": name, + "signed": f"records/{name}.signed.json", + "attestations": [], + "title": "Dog licensing", + "extensions": {"role": "notebook"}, + } + ] + assert json.loads(gh.files_at()[f"records/{name}.signed.json"]) == docs.first + tree_body = next(b for (m, p), b in zip(gh.requests, gh.bodies, strict=True) if p.endswith("/git/trees")) + assert {item["path"] for item in tree_body["tree"]} == {f"records/{name}.signed.json", "host.json"} + + +# --- a hash any listed entry carries (D8 A; host-core build.ts:65-66 refuses a hash listed twice) --- + + +def _assert_names_the_listed_entry(gh: FakeGitHub, receipt: dict[str, Any], listed: str) -> None: + assert_nothing_written(gh) + assert receipt["written"] is False and receipt["commit"] is None + assert receipt["name"] == listed + assert receipt["bundle_url"] == f"{ORIGIN}/bundles/{listed}.bundle.json" + from typedstandards._badge import verify_href + + assert receipt["verify_url"] == verify_href(receipt["bundle_url"]) + + +def test_a_hash_listed_under_another_name_writes_nothing(docs: Docs) -> None: + gh = github(docs, listed={"a": docs.first}) + receipt = ts.publish(docs.first, host=host(gh), name="b", title="T") + _assert_names_the_listed_entry(gh, receipt, "a") + + +def test_a_hash_listed_under_another_default_name_writes_nothing(docs: Docs) -> None: + listed = default_name(docs.first, stem="another-stem") + gh = github(docs, listed={listed: docs.first}) + receipt = ts.publish(docs.first, host=host(gh), notebook="dog-licensing.ipynb", title="T") + _assert_names_the_listed_entry(gh, receipt, listed) + + +def test_a_revision_whose_hash_is_listed_writes_nothing(docs: Docs) -> None: + gh = github(docs, listed={"dog-licensing": docs.first, "rerun": docs.second}) + receipt = ts.publish(docs.second, host=host(gh), name="dog-licensing", title="T", revises=docs.revises) + _assert_names_the_listed_entry(gh, receipt, "rerun") + + +def test_each_listed_file_is_read_once_per_call(docs: Docs) -> None: + gh = github(docs, listed={"dog-licensing": docs.first, "rerun": docs.foreign}) + with pytest.raises(ts.PublishRefusedError): + ts.publish(docs.second, host=host(gh), notebook="x.ipynb", title="T", revises=docs.revises) + reads = [p for m, p in gh.requests if "/contents/records/" in p] + assert len(reads) == len(set(reads)) == 3 + + +# --- the other throws host-core's build reaches (typedstandards 116882a) -------------------------- + + +def _edited(document: dict[str, Any], edit: Any) -> dict[str, Any]: + value = copy.deepcopy(document) + edit(value) + return value + + +def test_a_signer_unlike_the_listed_records_is_refused(docs: Docs) -> None: + """build.ts:93-94: every record under a registry has one signer, display name and key.""" + gh = github(docs) + with pytest.raises(ts.PublishRefusedError, match="one signer"): + ts.publish(docs.renamed, host=host(gh), name="renamed", title="T") + assert_nothing_written(gh) + + +def test_a_signature_kid_other_than_the_signer_is_refused(docs: Docs) -> None: + """build.ts:96-97.""" + edited = _edited(docs.first, lambda d: d["signature"].update(kid="did:key:z6MkOther")) + gh = github(docs) + with pytest.raises(ts.PublishRefusedError, match="kid"): + ts.publish(edited, host=host(gh), name="dog-licensing", title="T") + assert_nothing_written(gh) + + +@pytest.mark.parametrize("field", ["publicKey", "signature"]) +def test_a_signature_host_core_refuses_is_refused(docs: Docs, field: str) -> None: + """records.ts:44-48.""" + edited = _edited(docs.first, lambda d: d["signature"].pop(field)) + gh = github(docs) + with pytest.raises(ts.PublishRefusedError, match="signature"): + ts.publish(edited, host=host(gh), name="dog-licensing", title="T") + assert gh.requests == [] + + +def test_an_empty_created_at_is_refused(docs: Docs) -> None: + """build.ts:61-62.""" + edited = _edited(docs.first, lambda d: d["package"]["metadata"].update(createdAt="")) + gh = github(docs) + with pytest.raises(ts.PublishRefusedError, match="createdAt"): + ts.publish(edited, host=host(gh), name="dog-licensing", title="T") + assert gh.requests == [] + + +def test_no_registry_refuses_a_signer_that_is_not_self_certifying(docs: Docs) -> None: + """build.ts:147-148, produce-core's view: with registry null, only a pseudonymous did:key.""" + edited = _edited(docs.first, lambda d: d["package"]["signer"].update(bindingTier="verified")) + manifest = template_manifest() + manifest["registry"] = None + gh = github(docs, manifest=manifest) + with pytest.raises(ts.PublishRefusedError, match="registry"): + ts.publish(edited, host=host(gh), name="dog-licensing", title="T") + assert_nothing_written(gh) + + +def test_a_signed_path_another_entry_names_is_refused(docs: Docs) -> None: + """Overwriting a file another entry lists would list one hash twice (build.ts:65-66).""" + manifest = template_manifest() + manifest["records"].append( + {"name": "x", "signed": "records/dog-licensing.signed.json", "attestations": [], "title": "x"} + ) + gh = github(docs, manifest=manifest) + gh.trees[gh.commits[gh.head]["tree"]]["records/dog-licensing.signed.json"] = dumps(docs.second) + with pytest.raises(ts.PublishRefusedError, match="already names"): + ts.publish(docs.first, host=host(gh), name="dog-licensing", title="T") + assert_nothing_written(gh) + + +def test_an_attestation_path_another_entry_names_is_refused(docs: Docs) -> None: + """Overwriting a node file another entry lists would aim it at the wrong record (build.ts:70-71).""" + gh = github(docs, listed={"dog-licensing": docs.first}) + path = f"records/dog-licensing.withdraws-{docs.withdrawal['nodeId'][:8]}.json" + manifest = gh.json_at("host.json") + manifest["records"][0]["attestations"].append(path) + gh.trees[gh.commits[gh.head]["tree"]]["host.json"] = dumps(manifest) + with pytest.raises(ts.PublishRefusedError, match="already names"): + ts.publish_attestation(docs.withdrawal, host=host(gh), name="dog-licensing") + assert_nothing_written(gh) + + +@pytest.mark.parametrize( + "edit", + [ + lambda m: m["records"][0].update(note="not a field"), + lambda m: m.pop("visibility"), + lambda m: m["records"][0].update(title=""), + lambda m: m.update(index={"$comment": 1}), + ], + ids=["a record key host-core does not define", "no visibility", "an empty title", "a $comment not a string"], +) +def test_a_host_json_host_core_refuses_is_refused(docs: Docs, edit: Any) -> None: + """manifest.ts: the manifest publish writes must pass parseManifest.""" + manifest = template_manifest() + edit(manifest) + gh = github(docs, manifest=manifest) + with pytest.raises(ts.PublishRefusedError, match="host-core"): + ts.publish(docs.first, host=host(gh), name="dog-licensing", title="T") + assert_nothing_written(gh) + + +def test_a_listed_record_that_cannot_be_read_is_refused(docs: Docs) -> None: + """build.ts:49-50 and records.ts:54-59: the build already fails on it.""" + manifest = template_manifest() + manifest["records"].append({"name": "gone", "signed": "records/gone.signed.json", "attestations": [], "title": "g"}) + gh = github(docs, manifest=manifest) + with pytest.raises(ts.PublishRefusedError, match="gone"): + ts.publish(docs.first, host=host(gh), name="dog-licensing", title="T") + assert_nothing_written(gh) + + +@pytest.mark.parametrize("encoding", ["NaN", "utf-16"]) +def test_a_signed_file_json_parse_refuses_is_refused(docs: Docs, tmp_path: Path, encoding: str) -> None: + """json.ts:31-41: host-core reads strict UTF-8 and JSON.parse, which has no NaN.""" + path = tmp_path / "signed.json" + if encoding == "NaN": + edited = _edited(docs.first, lambda d: d["package"].update(extra=float("nan"))) + path.write_text(json.dumps(edited), encoding="utf-8") + else: + path.write_text(json.dumps(docs.first), encoding="utf-16") + gh = github(docs) + with pytest.raises(ts.PublishRefusedError, match="UTF-8|JSON"): + ts.publish(path, host=host(gh), name="dog-licensing", title="T") + assert gh.requests == [] + + +# --- the display step after a lifecycle node (the template's display.mjs; G0-3's reason) ---------- + + +def test_a_withdrawal_no_rule_displays_is_refused(docs: Docs) -> None: + value = template_policy(docs.signer) + value["display"] = [value["display"][0]] # no rule for withdrawn records + gh = github(docs, listed={"dog-licensing": docs.first}, policy=value) + with pytest.raises(ts.PublishRefusedError, match="withdrawn"): + ts.publish_attestation(docs.withdrawal, host=host(gh), name="dog-licensing") + assert_nothing_written(gh) + + +def test_a_supersession_the_template_policy_cannot_display_is_refused(docs: Docs) -> None: + """The template's policy names active and withdrawn only: a superseded record would be refused.""" + gh = github(docs, listed={"dog-licensing": docs.first}) + with pytest.raises(ts.PublishRefusedError, match="superseded"): + ts.publish_attestation(docs.supersedes, host=host(gh), name="dog-licensing") + assert_nothing_written(gh) diff --git a/tests/test_publish_token.py b/tests/test_publish_token.py new file mode 100644 index 0000000..14c6784 --- /dev/null +++ b/tests/test_publish_token.py @@ -0,0 +1,338 @@ +"""The GitHub token (typedstandards#141 P2, acceptance 1's token line, 2 and 4). + +- A token comes from ``token=``, else ``TYPEDSTANDARDS_GITHUB_TOKEN``; one that does not start + ``github_pat_``, or that holds whitespace, a quote or ``op://``, is refused before any request, + and the refusal's text does not hold the value. +- No captured stdout, stderr, log record, warning or exception text holds the token's value on + any publish path, and the scanner is itself driven over offending paths to show it fails there. +- No file is opened or written while publishing, and the host's ``repr`` names only the + repository and the branch. + +The token is visibly fake. Every test drives the fake API through ``httpx.MockTransport``. +""" + +from __future__ import annotations + +import builtins +import contextlib +import io +import json +import logging +import os +import pickle +import sys +import traceback +import warnings +from collections.abc import Callable +from pathlib import Path +from typing import Any + +import httpx +import pytest +from github_stub import FakeGitHub # publish_support puts scripts/ on the path +from publish_support import TOKEN, Docs, github, host + +import typedstandards as ts + +#: The part of a fine-grained token after its public prefix: the secret itself. +SECRET_TAIL = TOKEN[len("github_pat_") :] + + +# --- the scanner ------------------------------------------------------------------------------ + + +class _Records(logging.Handler): + def __init__(self) -> None: + super().__init__(level=logging.DEBUG) + self.lines: list[str] = [] + + def emit(self, record: logging.LogRecord) -> None: + self.lines.append(self.format(record)) + self.lines.append(repr(record.args)) + + +def captured(call: Callable[[], Any]) -> list[tuple[str, str]]: + """Run ``call`` with stdout, stderr, every logger at DEBUG and warnings captured. Returns + ``(where, text)`` for each: the outputs, each log record (formatted, and its arguments), each + warning, the returned value's ``repr``, and an exception's ``str``, ``repr`` and traceback, + also with each frame's locals.""" + out, err = io.StringIO(), io.StringIO() + handler = _Records() + handler.setFormatter(logging.Formatter("%(name)s %(levelname)s %(message)s")) + root = logging.getLogger() + level = root.level + root.addHandler(handler) + root.setLevel(logging.DEBUG) + texts: list[tuple[str, str]] = [] + try: + with ( + contextlib.redirect_stdout(out), + contextlib.redirect_stderr(err), + warnings.catch_warnings(record=True) as caught, + ): + warnings.simplefilter("always") + try: + texts.append(("return value", repr(call()))) + except Exception as error: # every exception's text is scanned + texts.append(("exception str", str(error))) + texts.append(("exception repr", repr(error))) + texts.append(("traceback", "".join(traceback.format_exception(error)))) + # What a verbose notebook traceback (IPython's %xmode Verbose) shows: each frame's locals. + with_locals = traceback.TracebackException.from_exception(error, capture_locals=True) + texts.append(("traceback with locals", "".join(with_locals.format()))) + texts += [("warning", str(w.message)) for w in caught] + finally: + root.removeHandler(handler) + root.setLevel(level) + texts += [("stdout", out.getvalue()), ("stderr", err.getvalue())] + texts += [("log record", line) for line in handler.lines] + return texts + + +def leaks(token: str, texts: list[tuple[str, str]]) -> list[str]: + """Where the token's value, or its secret part, appears in captured text.""" + tail = token[len("github_pat_") :] if token.startswith("github_pat_") else token + return sorted({where for where, text in texts if token in text or (tail and tail in text)}) + + +# --- every publish path, scanned ------------------------------------------------------------- + + +def publish_paths(docs: Docs) -> dict[str, Callable[[], Any]]: + """Each path of acceptance 1 as a call, on a fresh fake host per call.""" + + def run(listed: dict | None = None, setup: Callable[[FakeGitHub], None] | None = None, **kwargs: Any): + def call() -> Any: + gh = github(docs, listed=listed) + if setup: + setup(gh) + target = kwargs.pop("_target", "publish") + if target == "attestation": + return ts.publish_attestation(docs.withdrawal, host=host(gh), name="dog-licensing") + return ts.publish(kwargs.pop("_signed", docs.first), host=host(gh), **kwargs) + + return call + + first = {"dog-licensing": docs.first} + return { + "a new record": run(name="dog-licensing", title="T"), + "the default name": run(notebook="dog-licensing.ipynb", title="T"), + "a listed hash": run(first, name="dog-licensing", title="T"), + "a listed name, refused": run(first, _signed=docs.second, name="dog-licensing", title="T"), + "a listed name, revises": run( + first, _signed=docs.second, name="dog-licensing", title="T", revises=docs.revises + ), + "a records segment": run(name="records/x", title="T"), + "a name failing the rule": run(name="a b", title="T"), + "a BlobRef output": run(_signed=docs.blobref, name="x", title="T"), + "another signer": run(_signed=docs.foreign, name="x", title="T"), + "a role no rule admits": run(name="x", title="T", role="claim"), + "an empty title": run(name="x", title=""), + "a non-fast-forward, retried": run(setup=lambda gh: setattr(gh, "concurrent_writes", 1), name="x", title="T"), + "a non-fast-forward, twice": run(setup=lambda gh: setattr(gh, "concurrent_writes", 2), name="x", title="T"), + "a ref update that landed": run(setup=lambda gh: setattr(gh, "landed_but_failed", [502]), name="x", title="T"), + "an attestation": run(first, _target="attestation"), + "an API error": run(setup=lambda gh: setattr(gh, "token", "github_pat_other"), name="x", title="T"), + } + + +def test_no_output_holds_the_token_on_any_publish_path(docs: Docs) -> None: + found = {} + for label, call in publish_paths(docs).items(): + texts = captured(call) + assert texts, label + if leaks(TOKEN, texts): + found[label] = leaks(TOKEN, texts) + assert found == {} + + +def test_the_paths_raise_and_log_what_they_should(docs: Docs) -> None: + """The scan above reads real text: refusals raise, and a write logs its requests.""" + texts = captured(publish_paths(docs)["a records segment"]) + assert any(where == "exception str" and "records" in text for where, text in texts) + texts = captured(publish_paths(docs)["a new record"]) + assert any(where == "log record" and "git/refs/heads/main" in text for where, text in texts) + + +@pytest.mark.parametrize( + "value", + [ + "github_pat_TESTONLY with a space", + "github_pat_TESTONLY_trailing_newline\n", + "\tgithub_pat_TESTONLY_leading_tab", + '"github_pat_TESTONLY_double_quoted"', + "'github_pat_TESTONLY_single_quoted'", + "op://Example Vault/example item/credential", + "github_pat_TESTONLY_op://unresolved", + "ghp_TESTONLYclassicshape", + "TESTONLYnoprefixatall", + "github_pat_", + ], +) +def test_a_token_of_the_wrong_shape_is_refused_without_its_value(docs: Docs, value: str) -> None: + gh = github(docs) + bad = ts.GitHubPagesHost(gh.repository, token=value, transport=gh.transport()) + texts = captured(lambda: ts.publish(docs.first, host=bad, name="dog-licensing", title="T")) + error = next(text for where, text in texts if where == "exception repr") + assert error.startswith("PublishRefusedError(") + if value != "github_pat_": # the bare prefix is in the refusal's own words + assert leaks(value, texts) == [] and leaks(value.strip("\"' \t\n"), texts) == [] + assert gh.requests == [] + + +def test_a_wrong_token_from_the_environment_is_refused_without_its_value( + docs: Docs, monkeypatch: pytest.MonkeyPatch +) -> None: + value = "github_pat_TESTONLY from the environment" + monkeypatch.setenv(ts.TOKEN_VARIABLE, value) + gh = github(docs) + texts = captured( + lambda: ts.publish( + docs.first, host=ts.GitHubPagesHost(gh.repository, transport=gh.transport()), name="x", title="T" + ) + ) + assert any(where == "exception str" and ts.TOKEN_VARIABLE in text for where, text in texts) + assert leaks(value, texts) == [] + assert gh.requests == [] + + +def test_no_token_is_refused(docs: Docs, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv(ts.TOKEN_VARIABLE, raising=False) + gh = github(docs) + with pytest.raises(ts.PublishRefusedError, match="TYPEDSTANDARDS_GITHUB_TOKEN"): + ts.publish(docs.first, host=ts.GitHubPagesHost(gh.repository, transport=gh.transport()), name="x", title="T") + assert gh.requests == [] + + +# --- the scanner fails on an offender ------------------------------------------------------------ + + +def _offender(docs: Docs, how: str) -> Callable[[], Any]: + """A publish whose transport leaks the Authorization header one way.""" + + def call() -> Any: + gh = github(docs) + + def leak(request: httpx.Request) -> None: + value = request.headers["authorization"] + if how == "stdout": + print(value) + elif how == "stderr": + print(value, file=sys.stderr) + elif how == "log record": + logging.getLogger("typedstandards").debug("sent %s", value) + elif how == "warning": + warnings.warn(f"sent {value}", stacklevel=1) + elif how == "exception": + raise RuntimeError(f"sent {value}") + elif how == "frame local": + raise RuntimeError("a frame holding the header as a local raised") + return None + + gh.hook = leak + return ts.publish(docs.first, host=host(gh), name="dog-licensing", title="T") + + return call + + +@pytest.mark.parametrize("how", ["stdout", "stderr", "log record", "warning", "exception", "frame local"]) +def test_the_scanner_fails_on_an_offender(docs: Docs, how: str) -> None: + found = leaks(TOKEN, captured(_offender(docs, how))) + expected = {"exception": "exception str", "frame local": "traceback with locals"}.get(how, how) + assert expected in found + + +def test_the_scanner_finds_the_secret_part_alone() -> None: + assert leaks(TOKEN, [("stdout", f"…{SECRET_TAIL}")]) == ["stdout"] + assert leaks(TOKEN, [("stdout", "github_pat_ and nothing else")]) == [] + + +# --- acceptance 4 ------------------------------------------------------------------------------- + + +def test_the_token_comes_from_the_argument_else_the_environment(docs: Docs, monkeypatch: pytest.MonkeyPatch) -> None: + gh = github(docs) + monkeypatch.setenv(ts.TOKEN_VARIABLE, TOKEN) + from_env = ts.GitHubPagesHost(gh.repository, transport=gh.transport()) + assert ts.publish(docs.first, host=from_env, name="a", title="T")["written"] is True + + monkeypatch.setenv(ts.TOKEN_VARIABLE, "github_pat_TESTONLY_the_environments") + assert ts.publish(docs.second, host=host(gh), name="b", title="T")["written"] is True # token= wins + with pytest.raises(ts.PublishError, match="401"): + ts.publish(docs.third, host=ts.GitHubPagesHost(gh.repository, transport=gh.transport()), name="c", title="T") + + +def test_the_environment_is_read_when_publishing(docs: Docs, monkeypatch: pytest.MonkeyPatch) -> None: + """A host made before the variable is set (a notebook's first cell) still finds it.""" + monkeypatch.delenv(ts.TOKEN_VARIABLE, raising=False) + gh = github(docs) + made_first = ts.GitHubPagesHost(gh.repository, transport=gh.transport()) + monkeypatch.setenv(ts.TOKEN_VARIABLE, TOKEN) + assert ts.publish(docs.first, host=made_first, name="a", title="T")["written"] is True + + +def test_publishing_opens_and_writes_no_file(docs: Docs, monkeypatch: pytest.MonkeyPatch) -> None: + opened: list[str] = [] + real_open, real_os_open = builtins.open, os.open + + def recording_open(file: Any, *args: Any, **kwargs: Any) -> Any: + opened.append(f"open {file}") + return real_open(file, *args, **kwargs) + + def recording_os_open(path: Any, *args: Any, **kwargs: Any) -> Any: + opened.append(f"os.open {path}") + return real_os_open(path, *args, **kwargs) + + gh = github(docs, listed={"dog-licensing": docs.first}) + target = host(gh) + monkeypatch.setattr(builtins, "open", recording_open) + monkeypatch.setattr(os, "open", recording_os_open) + for method in ("write_text", "write_bytes", "touch", "open"): + monkeypatch.setattr(Path, method, lambda self, *a, _m=method, **k: opened.append(f"Path.{_m} {self}")) + ts.publish(docs.second, host=target, name="dog-licensing", title="T", revises=docs.revises) + ts.publish_attestation(docs.withdrawal, host=target, name="dog-licensing") + assert opened == [] + + +def test_the_hosts_repr_shows_the_repository_and_branch_only() -> None: + made = ts.GitHubPagesHost("example-owner/example-host", token=TOKEN) + assert repr(made) == "GitHubPagesHost('example-owner/example-host', branch='main')" + assert str(made) == repr(made) + other = ts.GitHubPagesHost("example-owner/example-host", branch="pages", token=TOKEN) + assert repr(other) == "GitHubPagesHost('example-owner/example-host', branch='pages')" + + +def test_the_host_does_not_expose_the_token() -> None: + made = ts.GitHubPagesHost("example-owner/example-host", token=TOKEN) + with pytest.raises(TypeError): + vars(made) + with pytest.raises(TypeError): + pickle.dumps(made) + texts = [repr(getattr(made, name, None)) for name in dir(made)] + assert leaks(TOKEN, [("attribute", text) for text in texts]) == [] + assert TOKEN not in json.dumps(texts) + + +@pytest.mark.parametrize("repository", ["", "example-owner", "a/b/c", "https://github.com/a/b", "a b/c"]) +def test_a_repository_that_is_not_owner_slash_name_is_refused(repository: str) -> None: + with pytest.raises(ValueError, match="owner/name"): + ts.GitHubPagesHost(repository, token=TOKEN) + + +def test_publishing_reads_the_token_variable_and_never_the_whole_environment( + docs: Docs, monkeypatch: pytest.MonkeyPatch +) -> None: + """The client a call builds (no transport given) does not iterate os.environ, which holds the + seed, for proxy settings. The autouse guard stops the request at the socket.""" + from support import SEED_VARIABLE + from test_guards import RecordingEnviron + + monkeypatch.setenv(ts.TOKEN_VARIABLE, TOKEN) + monkeypatch.setenv(SEED_VARIABLE, "not-a-seed") + recorder = RecordingEnviron(os.environ) + monkeypatch.setattr(os, "environ", recorder) + with pytest.raises(Exception, match="network connection"): + ts.publish(docs.first, host=ts.GitHubPagesHost("example-owner/example-host"), name="x", title="T") + assert ts.TOKEN_VARIABLE in recorder.keys_read + assert SEED_VARIABLE not in recorder.keys_read + assert recorder.read_all is False diff --git a/tests/test_readme.py b/tests/test_readme.py index fce9e53..d8e7714 100644 --- a/tests/test_readme.py +++ b/tests/test_readme.py @@ -69,3 +69,75 @@ def test_the_helpers_need_no_node(monkeypatch: pytest.MonkeyPatch, tmp_path: Pat ): with pytest.raises(typedstandards.NodeLocatorError, match="20.19"): call() + + +# --- publishing (typedstandards#141 P2, acceptance 5) --------------------------------------------- + +CUSTODY = "In a hosted notebook, the hosting service's runtime holds the seed for as long as the kernel runs." + + +def _publishing_section() -> str: + text = README.read_text(encoding="utf-8") + start = text.index("## Publishing to a GitHub Pages host") + return text[start : text.index("\n## ", start + 1)] + + +def test_readme_states_the_seeds_custody_in_a_hosted_notebook() -> None: + assert CUSTODY in " ".join(_publishing_section().split()) + + +def test_readme_sets_the_seed_from_a_secret_store_in_one_line() -> None: + section = _publishing_section() + lines = [line for line in section.splitlines() if 'os.environ["TYPEDSTANDARDS_SIGNING_SEED_B64"] = ' in line] + assert len(lines) == 1, lines + + +def test_readme_names_the_unsafe_forms_and_codespaces() -> None: + section = " ".join(_publishing_section().split()) + for phrase in ("pasted into a cell", "printed", "saved in the `.ipynb`", "Codespaces secret", "Actions secret"): + assert phrase in section, phrase + + +def test_readme_documents_the_calls_receipt_and_token() -> None: + section = " ".join(_publishing_section().split()) + for phrase in ( + "ts.GitHubPagesHost(", + "ts.publish(", + "ts.publish_attestation(", + "TYPEDSTANDARDS_GITHUB_TOKEN", + "github_pat_", + "Contents", + "revises=", + ): + assert phrase in section, phrase + for key in ("name", "commit", "bundle_url", "verify_url", "registry_url", "written", "run"): + assert f"`{key}`" in section, key + # The seat's note on G0-4: the default name leads; the derived name is the explicit-name case. + assert section.index("The **default name**") < section.index("`-`") + + +def test_the_seed_scanner_reads_python_modules_only(tmp_path: Path) -> None: + """The README's seed line is documentation for the author, not package code: the guard's seed + scanner covers the package's ``.py`` modules, so it neither sees nor needs to allow it.""" + from guards import seed_references + + (tmp_path / "README.md").write_text('os.environ["TYPEDSTANDARDS_SIGNING_SEED_B64"] = secret\n', encoding="utf-8") + assert seed_references(tmp_path) == [] + (tmp_path / "module.py").write_text("# TYPEDSTANDARDS_SIGNING_SEED_B64\n", encoding="utf-8") + assert seed_references(tmp_path) == ["module.py:1: names TYPEDSTANDARDS_SIGNING_SEED_B64"] + + +def test_readme_shows_how_to_read_the_did_key_before_the_first_publish() -> None: + """The template's setup sets host-policy.json's signer to the author's did:key, which the CLI + prints only in what it signs: the README shows reading it from a signed record.""" + section = " ".join(_publishing_section().split()) + assert 'signed["package"]["signer"]["identifier"]' in section + assert "typedstandards-host-template#publishing-from-a-notebook" in section + + +def test_readme_states_that_a_listed_hash_is_written_once() -> None: + """D8 A: a hash any listed entry carries is not written again, under any name.""" + section = " ".join(_publishing_section().split()) + assert "becomes two entries" not in section + assert "under any name" in section + assert "one read of each listed record's signed file" in section diff --git a/tests/test_smoke_publish.py b/tests/test_smoke_publish.py new file mode 100644 index 0000000..0f536c9 --- /dev/null +++ b/tests/test_smoke_publish.py @@ -0,0 +1,69 @@ +"""``scripts/smoke_publish.py``, the live publish check for gate G3, run here against the fake +GitHub API over ``httpx.MockTransport``: never live. It signs through the vendored CLI with the +test seed in the environment, publishes under the default name, prints the receipt's fields and +nothing secret, and exits non-zero on a refusal.""" + +from __future__ import annotations + +import sys +from pathlib import Path + +import pytest +from github_stub import FakeGitHub, dumps, manifest, policy +from publish_support import TOKEN +from support import analysis_input + +import typedstandards as ts + +sys.path.insert(0, str(Path(__file__).parent.parent / "scripts")) +import smoke_publish # noqa: E402 + + +def fake_host(signer: str, origin: str = smoke_publish.ORIGIN) -> FakeGitHub: + files = {"host.json": dumps(manifest(origin)), "host-policy.json": dumps(policy(signer))} + return FakeGitHub(files, token=TOKEN, repository=smoke_publish.REPOSITORY) + + +@pytest.fixture +def signer(seed: str) -> str: + """The test seed's did:key, as the CLI derives it.""" + return ts.sign(analysis_input(output="x"))["package"]["signer"]["identifier"] + + +def test_the_publish_check_publishes_one_record( + signer: str, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + monkeypatch.setenv(ts.TOKEN_VARIABLE, TOKEN) + gh = fake_host(signer) + assert smoke_publish.main([], transport=gh.transport()) == 0 + out = capsys.readouterr().out + print(out) + assert "written: True" in out and "publish check passed" in out + assert f"bundle_url: {smoke_publish.ORIGIN}/bundles/publish-check/" in out + assert "run: None" in out + assert TOKEN not in out and TOKEN[len("github_pat_") :] not in out + assert [m for m, _ in gh.requests].count("PATCH") == 1 + assert any(p.startswith("records/publish-check/") for p in gh.files_at()) + + +def test_the_publish_check_exits_non_zero_on_a_refusal( + signer: str, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + monkeypatch.setenv(ts.TOKEN_VARIABLE, TOKEN) + gh = fake_host("did:key:z6MkhaXgBZDvotDkL5257faiztiGiC2QtKLGpbnnEGta2doK") + assert smoke_publish.main([], transport=gh.transport()) == 2 + assert "refused, nothing written" in capsys.readouterr().out + assert gh.writes() == [] + + monkeypatch.setenv(ts.TOKEN_VARIABLE, "op://Example Vault/example item/credential") + assert smoke_publish.main([], transport=fake_host(signer).transport()) == 2 + assert "op://" in capsys.readouterr().out + + +def test_the_publish_check_names_another_origin( + signer: str, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + monkeypatch.setenv(ts.TOKEN_VARIABLE, TOKEN) + gh = fake_host(signer, origin="https://other.example.org") + assert smoke_publish.main([], transport=gh.transport()) == 5 + assert "is not under" in capsys.readouterr().out