From c4bbea4a2b1b3d455b231d47550ce329c0240f7b Mon Sep 17 00:00:00 2001 From: Jens Tirsvad Nielsen Date: Sat, 10 Oct 2026 11:39:11 +0800 Subject: [PATCH] Make the wiki publisher a scoped workflow with an opt-in gate Embed the publish script in .gitea/scoped_workflows/publish-wiki.yml and remove the repository workflow and tools/publish_wiki.py. A repository publishes only when wiki/.publish exists; the word prune in it is the only thing that lets the workflow delete pages. The organisation's TOKEN_FOR_GITEA_WIKI is used with no fallback. This repository opts in with prune. Tests run the embedded script against a fake Gitea. Task: MIL-006#2 Task: MIL-006#3 --- .gitea/scoped_workflows/publish-wiki.yml | 288 +++++++++++++++++++++++ .gitea/workflows/publish-wiki.yml | 32 --- tests/test_publish_wiki.py | 274 +++++++++++++-------- tests/workflow_source.py | 26 ++ tools/publish_wiki.py | 256 -------------------- wiki/.publish | 1 + 6 files changed, 494 insertions(+), 383 deletions(-) create mode 100644 .gitea/scoped_workflows/publish-wiki.yml delete mode 100644 .gitea/workflows/publish-wiki.yml delete mode 100644 tools/publish_wiki.py create mode 100644 wiki/.publish diff --git a/.gitea/scoped_workflows/publish-wiki.yml b/.gitea/scoped_workflows/publish-wiki.yml new file mode 100644 index 0000000..4ba9f4e --- /dev/null +++ b/.gitea/scoped_workflows/publish-wiki.yml @@ -0,0 +1,288 @@ +name: Publish wiki + +# Shared through scoped delivery: it runs in every repository of an +# organization, but writes only where the repository has opted in with +# wiki/.publish (ADR-0004). TOKEN_FOR_GITEA_WIKI is read from the organization +# secrets; there is no fallback to the read-only sync token. +on: + push: + branches: [ main ] + paths: + - "wiki/**" + workflow_dispatch: + +concurrency: + group: publish-wiki + cancel-in-progress: false + +jobs: + publish-wiki: + permissions: + contents: read + runs-on: ubuntu-latest + steps: + - name: Check out the repository + uses: actions/checkout@v4 + + - name: Publish the pages to the wiki + env: + GITEA_API_URL: ${{ gitea.api_url }} + SOURCE_REPOSITORY: ${{ gitea.repository }} + WIKI_CREDENTIALS: ${{ secrets.TOKEN_FOR_GITEA_WIKI }} + run: | + python3 - <<'PY' + import base64 + import json + import os + import re + import sys + import urllib.error + import urllib.parse + import urllib.request + from pathlib import Path + + PAGE_LIMIT = 50 + REQUEST_TIMEOUT_SECONDS = 30 + COMMIT_MESSAGE = "Publish wiki from the repository" + MARKER_NAME = ".publish" + PRUNE_WORD = "prune" + + + class WorkflowError(RuntimeError): + """A failure with a category. + + configuration: a secret, a setting or the wiki source is missing or malformed. + credentials: Gitea rejected the token (HTTP 401). + permissions: the token is valid but not allowed (HTTP 403). + api: any other API or network failure. + """ + + def __init__(self, category, message): + super().__init__(f"[{category}] {message}") + self.category = category + self.detail = message + + + def request_json(url, method="GET", headers=None, body=None): + request = urllib.request.Request( + url, + data=json.dumps(body).encode("utf-8") if body is not None else None, + headers=headers or {}, + method=method, + ) + try: + with urllib.request.urlopen(request, timeout=REQUEST_TIMEOUT_SECONDS) as response: + content = response.read() + return json.loads(content) if content else None + except urllib.error.HTTPError as error: + detail = f"Gitea API request failed with HTTP {error.code} ({error.reason})" + if error.code == 401: + raise WorkflowError("credentials", f"{detail}. The token was rejected.") from None + if error.code == 403: + raise WorkflowError( + "permissions", f"{detail}. The token lacks a required permission." + ) from None + if error.code == 404: + raise WorkflowError( + "configuration", + f"{detail}. The repository or page was not found, or the token " + "cannot see it.", + ) from None + raise WorkflowError("api", detail) from None + except (urllib.error.URLError, TimeoutError) as error: + raise WorkflowError("api", f"Gitea API is unreachable ({error}).") from None + + + def parse_marker(text): + """Return True when the marker asks for pruning, False when it is empty.""" + word = (text or "").strip() + if not word: + return False + if word == PRUNE_WORD: + return True + raise WorkflowError( + "configuration", + f"wiki/{MARKER_NAME} may only be empty or contain the word '{PRUNE_WORD}'.", + ) + + + def parse_gitea_token(raw_token): + token = (raw_token or "").strip() + if not token: + raise WorkflowError( + "configuration", + "TOKEN_FOR_GITEA_WIKI is missing or empty. Store it as a secret of " + "the organization of this repository.", + ) + return token + + + def describe_token(token): + """Describe the shape of a token without revealing any of its characters.""" + facts = [f"{len(token)} characters"] + if re.fullmatch(r"[0-9a-f]{40}", token): + facts.append("looks like a Gitea token (40 hexadecimal characters)") + return ", ".join(facts) + facts.append("a Gitea token has 40 hexadecimal characters") + if set(token) <= {"*", "x", "X", "•"}: + facts.append("only mask characters: the masked text was stored") + if "=" in token: + facts.append("contains '=': the secret name may be part of the value") + if re.search(r"[\"'`]", token): + facts.append("contains a quote character") + if re.search(r"\s", token): + facts.append("contains whitespace") + return ", ".join(facts) + + + def split_repository(full_name): + owner, separator, repo = (full_name or "").partition("/") + if not separator or not owner or not repo: + raise WorkflowError("configuration", "Could not determine the Gitea repository.") + return owner, repo + + + def read_pages(wiki_dir): + """Return {page title: content} for every Markdown file in wiki_dir.""" + pages = {path.stem: path.read_bytes() for path in sorted(wiki_dir.glob("*.md"))} + if not pages: + # Publishing an empty directory would delete the whole wiki. + raise WorkflowError( + "configuration", f"The wiki source directory {wiki_dir} has no Markdown pages." + ) + return pages + + + def list_wiki_titles(wiki_url, headers, send): + """Return the titles of the pages that exist in the wiki. + + Gitea answers 404 for a wiki that has never been written to; that is an + empty wiki, not a failure. + """ + titles = set() + page = 1 + while True: + url = f"{wiki_url}/pages?page={page}&limit={PAGE_LIMIT}" + try: + batch = send(url, headers=headers) + except WorkflowError as error: + if error.category == "configuration" and page == 1: + return titles + raise + if not isinstance(batch, list): + raise WorkflowError("api", "Gitea returned an invalid wiki page list.") + titles.update(str(item.get("title", "")) for item in batch) + if len(batch) < PAGE_LIMIT: + return titles + page += 1 + + + def read_wiki_page(wiki_url, title, headers, send): + quoted = urllib.parse.quote(title, safe="") + page = send(f"{wiki_url}/page/{quoted}", headers=headers) + return base64.b64decode((page or {}).get("content_base64", "")) + + + def publish(pages, wiki_url, headers, send, prune): + """Create and update the pages; delete the others only when prune is set. + + Returns the titles created, updated and deleted, and the titles that were + left in the wiki because pruning is off. + """ + existing = list_wiki_titles(wiki_url, headers, send) + current = { + title: read_wiki_page(wiki_url, title, headers, send) + for title in sorted(existing & pages.keys()) + } + + result = {"created": [], "updated": [], "deleted": [], "kept": []} + for title, content in pages.items(): + body = { + "title": title, + "content_base64": base64.b64encode(content).decode("ascii"), + "message": COMMIT_MESSAGE, + } + if title not in existing: + send(f"{wiki_url}/new", method="POST", headers=headers, body=body) + result["created"].append(title) + elif current[title] != content: + quoted = urllib.parse.quote(title, safe="") + send(f"{wiki_url}/page/{quoted}", method="PATCH", headers=headers, body=body) + result["updated"].append(title) + + for title in sorted(existing - pages.keys()): + if not prune: + result["kept"].append(title) + continue + quoted = urllib.parse.quote(title, safe="") + send(f"{wiki_url}/page/{quoted}", method="DELETE", headers=headers) + result["deleted"].append(title) + return result + + + def main(): + wiki_dir = Path(os.environ.get("WIKI_DIR", "wiki")) + owner, repo = split_repository(os.environ.get("SOURCE_REPOSITORY")) + + marker = wiki_dir / MARKER_NAME + if not marker.is_file(): + print( + f"No {wiki_dir}/{MARKER_NAME} marker in {owner}/{repo}; " + "wiki publishing is not enabled for this repository." + ) + return 0 + prune = parse_marker(marker.read_text(encoding="utf-8")) + + pages = read_pages(wiki_dir) + token = parse_gitea_token(os.environ.get("WIKI_CREDENTIALS")) + + api_url = os.environ["GITEA_API_URL"].rstrip("/") + repository_url = ( + f"{api_url}/repos/{urllib.parse.quote(owner, safe='')}/" + f"{urllib.parse.quote(repo, safe='')}" + ) + headers = { + "Authorization": f"token {token}", + "Accept": "application/json", + "Content-Type": "application/json", + } + + try: + repository = request_json(repository_url, headers=headers) + if isinstance(repository, dict) and repository.get("has_wiki") is False: + raise WorkflowError( + "configuration", + f"The wiki is not enabled on {owner}/{repo}. Enable it in the " + "repository settings.", + ) + print("Preflight passed: Gitea accepted the token and the wiki is enabled.") + + result = publish(pages, f"{repository_url}/wiki", headers, request_json, prune) + except WorkflowError as error: + if error.category == "credentials": + raise WorkflowError( + "credentials", f"{error.detail} Token shape: {describe_token(token)}." + ) from None + raise + + print( + f"Published {len(pages)} page(s) to the wiki of {owner}/{repo}: " + f"{len(result['created'])} created, {len(result['updated'])} updated, " + f"{len(result['deleted'])} deleted." + ) + if result["kept"]: + print( + f"{len(result['kept'])} page(s) are in the wiki but not in " + f"{wiki_dir}/ and were kept; put the word '{PRUNE_WORD}' in " + f"{wiki_dir}/{MARKER_NAME} to delete them." + ) + return 0 + + + if __name__ == "__main__": + try: + sys.exit(main()) + except WorkflowError as error: + print(f"ERROR: {error}", file=sys.stderr) + sys.exit(1) + PY diff --git a/.gitea/workflows/publish-wiki.yml b/.gitea/workflows/publish-wiki.yml deleted file mode 100644 index 6a35c13..0000000 --- a/.gitea/workflows/publish-wiki.yml +++ /dev/null @@ -1,32 +0,0 @@ -name: Publish wiki - -on: - push: - branches: [ main ] - paths: - - "wiki/**" - workflow_dispatch: - -concurrency: - group: publish-wiki - cancel-in-progress: false - -jobs: - publish-wiki: - permissions: - contents: read - runs-on: ubuntu-latest - steps: - - name: Check out the repository - uses: actions/checkout@v4 - - - name: Check that the English pages are current - run: python3 tools/build_wiki.py --check - - - name: Publish the pages to the wiki - env: - GITEA_API_URL: ${{ gitea.api_url }} - SOURCE_REPOSITORY: ${{ gitea.repository }} - # TOKEN_FOR_GITEA_WIKI wins when it exists; otherwise TOKEN_FOR_GITEA is reused. - WIKI_CREDENTIALS: ${{ secrets.TOKEN_FOR_GITEA_WIKI || secrets.TOKEN_FOR_GITEA }} - run: python3 tools/publish_wiki.py diff --git a/tests/test_publish_wiki.py b/tests/test_publish_wiki.py index 559d8be..8172324 100644 --- a/tests/test_publish_wiki.py +++ b/tests/test_publish_wiki.py @@ -1,56 +1,68 @@ -"""Offline tests for tools/publish_wiki.py and its workflow.""" +"""Offline tests for the scoped publish-wiki workflow and its embedded script.""" import base64 -import importlib.util import io -import json import urllib.error -from pathlib import Path import pytest import yaml -REPO_ROOT = Path(__file__).resolve().parent.parent -WORKFLOW = REPO_ROOT / ".gitea" / "workflows" / "publish-wiki.yml" -_SPEC = importlib.util.spec_from_file_location( - "publish_wiki", REPO_ROOT / "tools" / "publish_wiki.py" -) -assert _SPEC is not None and _SPEC.loader is not None -publish_wiki = importlib.util.module_from_spec(_SPEC) -_SPEC.loader.exec_module(publish_wiki) +from workflow_source import PUBLISH_WORKFLOW, REPO_ROOT, load_publish_module API = "https://git.example.test/api/v1" REPO_URL = f"{API}/repos/TirSystem/repo" WIKI_URL = f"{REPO_URL}/wiki" +@pytest.fixture() +def publish_wiki(): + return load_publish_module() + + def b64(text: str) -> str: return base64.b64encode(text.encode()).decode() +class TestParseMarker: + @pytest.mark.parametrize( + ("text", "expected"), + [("", False), (" \n", False), (None, False), ("prune", True), (" prune\n", True)], + ) + def test_reads_the_empty_marker_and_the_prune_word(self, publish_wiki, text, expected): + assert publish_wiki.parse_marker(text) is expected + + @pytest.mark.parametrize("text", ["Prune", "prune: true", "delete", "prune\nmore"]) + def test_rejects_anything_else(self, publish_wiki, text): + with pytest.raises(publish_wiki.WorkflowError) as caught: + publish_wiki.parse_marker(text) + assert caught.value.category == "configuration" + assert "prune" in str(caught.value) + + class TestParseGiteaToken: - def test_returns_the_stripped_token(self): + def test_returns_the_stripped_token(self, publish_wiki): assert publish_wiki.parse_gitea_token(" abc \n") == "abc" @pytest.mark.parametrize("raw", [None, "", " "]) - def test_rejects_missing_or_empty_token(self, raw): + def test_rejects_missing_or_empty_token(self, publish_wiki, raw): with pytest.raises(publish_wiki.WorkflowError) as caught: publish_wiki.parse_gitea_token(raw) assert caught.value.category == "configuration" - assert "No token for the wiki" in str(caught.value) + assert "TOKEN_FOR_GITEA_WIKI is missing or empty" in str(caught.value) class TestReadPages: - def test_reads_markdown_files_by_stem(self, tmp_path): + def test_reads_markdown_files_by_stem_and_ignores_the_marker(self, publish_wiki, tmp_path): (tmp_path / "Home.md").write_bytes(b"home") + (tmp_path / "Home.da.md").write_bytes(b"hjem") + (tmp_path / ".publish").write_bytes(b"prune") (tmp_path / "Notes.txt").write_bytes(b"ignored") - assert publish_wiki.read_pages(tmp_path) == {"Home": b"home"} + assert publish_wiki.read_pages(tmp_path) == {"Home": b"home", "Home.da": b"hjem"} - def test_rejects_missing_directory(self, tmp_path): - with pytest.raises(publish_wiki.WorkflowError, match="missing"): - publish_wiki.read_pages(tmp_path / "nope") - - def test_rejects_empty_directory_so_the_wiki_is_never_wiped(self, tmp_path): + def test_rejects_a_directory_without_pages_so_the_wiki_is_never_wiped( + self, publish_wiki, tmp_path + ): + (tmp_path / ".publish").write_text("prune", encoding="utf-8") with pytest.raises(publish_wiki.WorkflowError, match="no Markdown pages"): publish_wiki.read_pages(tmp_path) @@ -58,19 +70,20 @@ class TestReadPages: class FakeGitea: """Answers like the Gitea repository and wiki endpoints.""" - def __init__(self, wiki=None, has_wiki=True, wiki_initialised=True): + def __init__(self, publish_wiki, wiki=None, has_wiki=True, wiki_initialised=True): + self.error = publish_wiki.WorkflowError self.wiki = dict(wiki or {}) self.has_wiki = has_wiki self.wiki_initialised = wiki_initialised self.calls = [] def __call__(self, url, method="GET", headers=None, body=None): - self.calls.append((method, url, body)) + self.calls.append((method, url, body, headers)) if url == REPO_URL: return {"has_wiki": self.has_wiki} if url.startswith(f"{WIKI_URL}/pages"): if not self.wiki_initialised: - raise publish_wiki.WorkflowError("configuration", "404") + raise self.error("configuration", "404") return [{"title": title} for title in self.wiki] if method == "POST" and url == f"{WIKI_URL}/new": self.wiki[body["title"]] = body["content_base64"] @@ -86,6 +99,10 @@ class FakeGitea: return None raise AssertionError(f"unexpected request {method} {url}") + @property + def writes(self): + return [call for call in self.calls if call[0] != "GET"] + @pytest.fixture() def environment(monkeypatch, tmp_path): @@ -93,6 +110,7 @@ def environment(monkeypatch, tmp_path): wiki_dir.mkdir() (wiki_dir / "Home.md").write_text("home v2", encoding="utf-8") (wiki_dir / "Credentials.md").write_text("credentials", encoding="utf-8") + (wiki_dir / ".publish").write_text("prune\n", encoding="utf-8") monkeypatch.setenv("WIKI_DIR", str(wiki_dir)) monkeypatch.setenv("GITEA_API_URL", f"{API}/") monkeypatch.setenv("SOURCE_REPOSITORY", "TirSystem/repo") @@ -100,20 +118,73 @@ def environment(monkeypatch, tmp_path): return wiki_dir -def run_main(monkeypatch, gitea): - monkeypatch.setattr(publish_wiki, "request_json", gitea) +def run_main(publish_wiki, gitea): + publish_wiki.request_json = gitea return publish_wiki.main() -class TestMain: - def test_creates_updates_and_deletes_to_match_the_source( - self, monkeypatch, environment, capsys +class TestOptIn: + def test_a_repository_without_the_marker_is_never_written_to( + self, publish_wiki, environment, capsys + ): + (environment / ".publish").unlink() + gitea = FakeGitea(publish_wiki, wiki={"Old": b64("hand written")}) + + assert run_main(publish_wiki, gitea) == 0 + + assert gitea.calls == [] + assert "wiki publishing is not enabled" in capsys.readouterr().out + assert gitea.wiki == {"Old": b64("hand written")} + + def test_a_repository_without_the_marker_does_not_need_the_secret( + self, publish_wiki, environment, monkeypatch + ): + (environment / ".publish").unlink() + monkeypatch.setenv("WIKI_CREDENTIALS", "") + assert run_main(publish_wiki, FakeGitea(publish_wiki)) == 0 + + def test_a_repository_without_a_wiki_directory_is_skipped( + self, publish_wiki, monkeypatch, tmp_path + ): + monkeypatch.setenv("WIKI_DIR", str(tmp_path / "missing")) + monkeypatch.setenv("SOURCE_REPOSITORY", "TirSystem/repo") + gitea = FakeGitea(publish_wiki) + assert run_main(publish_wiki, gitea) == 0 + assert gitea.calls == [] + + def test_an_opted_in_repository_without_the_secret_fails_before_any_request( + self, publish_wiki, environment, monkeypatch + ): + monkeypatch.setenv("WIKI_CREDENTIALS", "") + gitea = FakeGitea(publish_wiki) + + with pytest.raises(publish_wiki.WorkflowError, match="TOKEN_FOR_GITEA_WIKI"): + run_main(publish_wiki, gitea) + + assert gitea.calls == [] + + def test_an_unknown_marker_content_is_refused_before_any_request( + self, publish_wiki, environment + ): + (environment / ".publish").write_text("mirror everything", encoding="utf-8") + gitea = FakeGitea(publish_wiki) + + with pytest.raises(publish_wiki.WorkflowError, match="prune"): + run_main(publish_wiki, gitea) + + assert gitea.calls == [] + + +class TestPublish: + def test_prune_creates_updates_and_deletes_to_match_the_source( + self, publish_wiki, environment, capsys ): gitea = FakeGitea( - wiki={"Home": b64("home v1"), "Credentials": b64("credentials"), "Old": b64("x")} + publish_wiki, + wiki={"Home": b64("home v1"), "Credentials": b64("credentials"), "Old": b64("x")}, ) - assert run_main(monkeypatch, gitea) == 0 + assert run_main(publish_wiki, gitea) == 0 assert gitea.wiki == {"Home": b64("home v2"), "Credentials": b64("credentials")} methods = [(call[0], call[1].rsplit("/", 1)[-1]) for call in gitea.calls] @@ -122,73 +193,80 @@ class TestMain: assert ("PATCH", "Credentials") not in methods assert "0 created, 1 updated, 1 deleted" in capsys.readouterr().out - def test_creates_every_page_when_the_wiki_was_never_written( - self, monkeypatch, environment + def test_without_prune_pages_outside_the_source_are_kept( + self, publish_wiki, environment, capsys ): - gitea = FakeGitea(wiki_initialised=False) + (environment / ".publish").write_text("", encoding="utf-8") + gitea = FakeGitea( + publish_wiki, wiki={"Home": b64("home v1"), "Old": b64("hand written")} + ) - assert run_main(monkeypatch, gitea) == 0 + assert run_main(publish_wiki, gitea) == 0 + + assert gitea.wiki["Old"] == b64("hand written") + assert gitea.wiki["Home"] == b64("home v2") + assert "Credentials" in gitea.wiki + assert not [call for call in gitea.calls if call[0] == "DELETE"] + out = capsys.readouterr().out + assert "1 created, 1 updated, 0 deleted" in out + assert "1 page(s) are in the wiki but not in" in out + assert "prune" in out + + def test_creates_every_page_when_the_wiki_was_never_written( + self, publish_wiki, environment + ): + gitea = FakeGitea(publish_wiki, wiki_initialised=False) + + assert run_main(publish_wiki, gitea) == 0 assert set(gitea.wiki) == {"Home", "Credentials"} assert [call[0] for call in gitea.calls].count("POST") == 2 - def test_does_nothing_when_the_wiki_is_already_current(self, monkeypatch, environment): - gitea = FakeGitea(wiki={"Home": b64("home v2"), "Credentials": b64("credentials")}) + def test_does_nothing_when_the_wiki_is_already_current(self, publish_wiki, environment): + gitea = FakeGitea( + publish_wiki, wiki={"Home": b64("home v2"), "Credentials": b64("credentials")} + ) - assert run_main(monkeypatch, gitea) == 0 + assert run_main(publish_wiki, gitea) == 0 - assert all(call[0] == "GET" for call in gitea.calls) + assert gitea.writes == [] - def test_sends_the_token_and_never_prints_it(self, monkeypatch, environment, capsys): - seen = [] + def test_sends_the_token_and_never_prints_it(self, publish_wiki, environment, capsys): + gitea = FakeGitea(publish_wiki) - class Recording(FakeGitea): - def __call__(self, url, method="GET", headers=None, body=None): - seen.append(headers["Authorization"]) - return super().__call__(url, method, headers, body) + assert run_main(publish_wiki, gitea) == 0 - assert run_main(monkeypatch, Recording()) == 0 - - assert set(seen) == {"token wiki-token"} + assert {call[3]["Authorization"] for call in gitea.calls} == {"token wiki-token"} output = capsys.readouterr() assert "wiki-token" not in output.out + output.err - def test_fails_when_the_wiki_is_not_enabled(self, monkeypatch, environment): - gitea = FakeGitea(has_wiki=False) + def test_fails_when_the_wiki_is_not_enabled_and_writes_nothing( + self, publish_wiki, environment + ): + gitea = FakeGitea(publish_wiki, has_wiki=False) with pytest.raises(publish_wiki.WorkflowError) as caught: - run_main(monkeypatch, gitea) + run_main(publish_wiki, gitea) assert caught.value.category == "configuration" - assert all(call[0] == "GET" for call in gitea.calls) + assert gitea.writes == [] - def test_fails_before_any_request_when_the_secret_is_missing( - self, monkeypatch, environment - ): - monkeypatch.setenv("WIKI_CREDENTIALS", "") - gitea = FakeGitea() - - with pytest.raises(publish_wiki.WorkflowError, match="No token for the wiki"): - run_main(monkeypatch, gitea) - - assert gitea.calls == [] - - def test_fails_before_any_request_when_the_source_is_empty( - self, monkeypatch, environment + def test_an_empty_source_never_deletes_pages_even_with_prune( + self, publish_wiki, environment ): for path in environment.glob("*.md"): path.unlink() - gitea = FakeGitea(wiki={"Home": b64("keep me")}) + gitea = FakeGitea(publish_wiki, wiki={"Home": b64("keep me")}) with pytest.raises(publish_wiki.WorkflowError, match="no Markdown pages"): - run_main(monkeypatch, gitea) + run_main(publish_wiki, gitea) assert gitea.calls == [] assert gitea.wiki == {"Home": b64("keep me")} class TestDescribeToken: - def test_recognises_a_gitea_token_shape(self): + def test_recognises_a_gitea_token_shape(self, publish_wiki): description = publish_wiki.describe_token("a1b2c3d4e5" * 4) assert "40 characters" in description assert "looks like a Gitea token" in description @@ -202,20 +280,24 @@ class TestDescribeToken: ("abc def", "whitespace"), ], ) - def test_names_the_likely_mistake_without_echoing_the_token(self, token, expected): + def test_names_the_likely_mistake_without_echoing_the_token( + self, publish_wiki, token, expected + ): description = publish_wiki.describe_token(token) assert expected in description assert token not in description class TestRejectedToken: - def test_rejection_adds_the_token_shape_but_not_the_token(self, monkeypatch, environment): + def test_rejection_adds_the_token_shape_but_not_the_token( + self, publish_wiki, environment, monkeypatch + ): monkeypatch.setenv("WIKI_CREDENTIALS", "*******") def reject(url, method="GET", headers=None, body=None): raise publish_wiki.WorkflowError("credentials", "Gitea rejected the token.") - monkeypatch.setattr(publish_wiki, "request_json", reject) + publish_wiki.request_json = reject with pytest.raises(publish_wiki.WorkflowError) as caught: publish_wiki.main() @@ -226,11 +308,11 @@ class TestRejectedToken: assert "only mask characters" in message assert "*******" not in message.split("Token shape:")[0] - def test_other_categories_are_not_changed(self, monkeypatch, environment): + def test_other_categories_are_not_changed(self, publish_wiki, environment): def forbid(url, method="GET", headers=None, body=None): raise publish_wiki.WorkflowError("permissions", "Forbidden.") - monkeypatch.setattr(publish_wiki, "request_json", forbid) + publish_wiki.request_json = forbid with pytest.raises(publish_wiki.WorkflowError) as caught: publish_wiki.main() @@ -240,7 +322,7 @@ class TestRejectedToken: class TestRequestJson: @staticmethod - def raise_http_error(monkeypatch, code, reason): + def raise_http_error(publish_wiki, monkeypatch, code, reason): def fake_urlopen(request, timeout): raise urllib.error.HTTPError( request.full_url, code, reason, {}, io.BytesIO(b"secret detail") @@ -257,8 +339,8 @@ class TestRequestJson: (500, "Server Error", "api"), ], ) - def test_categorizes_http_errors(self, monkeypatch, code, reason, category): - self.raise_http_error(monkeypatch, code, reason) + def test_categorizes_http_errors(self, publish_wiki, monkeypatch, code, reason, category): + self.raise_http_error(publish_wiki, monkeypatch, code, reason) with pytest.raises(publish_wiki.WorkflowError) as caught: publish_wiki.request_json(REPO_URL) @@ -267,7 +349,7 @@ class TestRequestJson: assert f"HTTP {code}" in str(caught.value) assert "secret detail" not in str(caught.value) - def test_reports_unreachable_service_as_api_failure(self, monkeypatch): + def test_reports_unreachable_service_as_api_failure(self, publish_wiki, monkeypatch): def fake_urlopen(request, timeout): raise urllib.error.URLError("no route") @@ -282,7 +364,7 @@ class TestRequestJson: class TestWorkflow: @pytest.fixture() def document(self): - return yaml.safe_load(WORKFLOW.read_text(encoding="utf-8")) + return yaml.safe_load(PUBLISH_WORKFLOW.read_text(encoding="utf-8")) def test_runs_only_on_main_when_wiki_changes_or_by_hand(self, document): triggers = document[True] # PyYAML reads the key `on` as True @@ -291,21 +373,23 @@ class TestWorkflow: assert "workflow_dispatch" in triggers def test_job_has_read_only_permissions(self, document): - job = document["jobs"]["publish-wiki"] - assert job["permissions"] == {"contents": "read"} + assert document["jobs"]["publish-wiki"]["permissions"] == {"contents": "read"} - def test_checks_the_pages_before_publishing_them(self, document): - runs = [step.get("run", "") for step in document["jobs"]["publish-wiki"]["steps"]] - assert runs.index("python3 tools/build_wiki.py --check") < runs.index( - "python3 tools/publish_wiki.py" - ) - - def test_secret_is_passed_only_through_the_environment(self, document): - text = WORKFLOW.read_text(encoding="utf-8") - assert text.count("secrets.TOKEN_FOR_GITEA_WIKI") == 1 - steps = document["jobs"]["publish-wiki"]["steps"] - publish_step = steps[-1] - assert publish_step["env"]["WIKI_CREDENTIALS"] == ( - "${{ secrets.TOKEN_FOR_GITEA_WIKI || secrets.TOKEN_FOR_GITEA }}" - ) + def test_secret_is_passed_only_through_the_environment_and_has_no_fallback(self, document): + text = PUBLISH_WORKFLOW.read_text(encoding="utf-8") + assert "secrets.TOKEN_FOR_GITEA }}" not in text + publish_step = document["jobs"]["publish-wiki"]["steps"][-1] + assert publish_step["env"]["WIKI_CREDENTIALS"] == "${{ secrets.TOKEN_FOR_GITEA_WIKI }}" assert "secrets" not in publish_step["run"] + + def test_the_workflow_has_one_source_and_needs_nothing_from_tools(self): + assert not (REPO_ROOT / ".gitea" / "workflows" / "publish-wiki.yml").exists() + assert not (REPO_ROOT / "tools" / "publish_wiki.py").exists() + script = PUBLISH_WORKFLOW.read_text(encoding="utf-8").split("python3 - <<'PY'", 1)[1] + assert "tools/" not in script + + +def test_this_repository_opts_in_and_prunes(): + """The wiki of this repository must equal wiki/ (ADR-0001, ADR-0004).""" + marker = REPO_ROOT / "wiki" / ".publish" + assert marker.read_text(encoding="utf-8").strip() == "prune" diff --git a/tests/workflow_source.py b/tests/workflow_source.py index 5c3fbb8..3a1b771 100644 --- a/tests/workflow_source.py +++ b/tests/workflow_source.py @@ -11,6 +11,7 @@ WORKFLOW_DIRS = ( REPO_ROOT / ".gitea" / "workflows", ) SYNC_WORKFLOW = WORKFLOW_DIRS[0] / "sync-github-metadata.yml" +PUBLISH_WORKFLOW = WORKFLOW_DIRS[0] / "publish-wiki.yml" _HEREDOC = re.compile( r"^[ \t]*python3 - <<'PY'\r?\n(?P.*?)^[ \t]*PY[ \t]*\r?$", @@ -42,3 +43,28 @@ def load_sync_module(name: str = "embedded_sync") -> dict[str, Any]: namespace: dict[str, Any] = {"__name__": name} exec(compile(source, str(SYNC_WORKFLOW), "exec"), namespace) return namespace + + +class ScriptNamespace(dict): + """The globals of an embedded script, readable and writable as attributes. + + Functions of the script look names up in this dict, so a test that sets + `namespace.request_json = fake` changes what the script calls. + """ + + def __getattr__(self, name: str) -> Any: + try: + return self[name] + except KeyError: + raise AttributeError(name) from None + + def __setattr__(self, name: str, value: Any) -> None: + self[name] = value + + +def load_publish_module(name: str = "embedded_publish") -> ScriptNamespace: + """Execute the embedded publish script without running `main`.""" + (source,) = embedded_python(PUBLISH_WORKFLOW) + namespace = ScriptNamespace({"__name__": name}) + exec(compile(source, str(PUBLISH_WORKFLOW), "exec"), namespace) + return namespace diff --git a/tools/publish_wiki.py b/tools/publish_wiki.py deleted file mode 100644 index 1d780e1..0000000 --- a/tools/publish_wiki.py +++ /dev/null @@ -1,256 +0,0 @@ -"""Publish the pages in ``wiki/`` to the Gitea wiki of this repository. - -The pages are written through the Gitea wiki REST API. A page that is missing -in ``wiki/`` is removed from the wiki, so the wiki always equals the reviewed -source; edits made in the web UI are overwritten (MIL-005). - -Environment: - GITEA_API_URL Gitea API base, for example https://host/api/v1 - SOURCE_REPOSITORY owner/repo of the repository whose wiki is written - WIKI_CREDENTIALS the Gitea token, as clear text - WIKI_DIR directory with the pages (default: wiki) - -Exit codes: 0 published or nothing to do, 1 failed (message names a category). -""" - -import base64 -import json -import os -import re -import sys -import urllib.error -import urllib.parse -import urllib.request -from collections.abc import Callable -from pathlib import Path -from typing import Any - -PAGE_LIMIT = 50 -REQUEST_TIMEOUT_SECONDS = 30 -COMMIT_MESSAGE = "Publish wiki from the repository" - -JsonRequest = Callable[..., Any] - - -class WorkflowError(RuntimeError): - """A failure with a category. - - configuration: a secret, a setting or the wiki source is missing or malformed. - credentials: Gitea rejected the token (HTTP 401). - permissions: the token is valid but not allowed (HTTP 403). - api: any other API or network failure. - """ - - def __init__(self, category: str, message: str) -> None: - super().__init__(f"[{category}] {message}") - self.category = category - self.detail = message - - -def request_json( - url: str, - method: str = "GET", - headers: dict[str, str] | None = None, - body: dict[str, Any] | None = None, -) -> Any: - """Send one JSON request to Gitea and return the decoded response.""" - request = urllib.request.Request( - url, - data=json.dumps(body).encode("utf-8") if body is not None else None, - headers=headers or {}, - method=method, - ) - try: - with urllib.request.urlopen(request, timeout=REQUEST_TIMEOUT_SECONDS) as response: - content = response.read() - return json.loads(content) if content else None - except urllib.error.HTTPError as error: - detail = f"Gitea API request failed with HTTP {error.code} ({error.reason})" - if error.code == 401: - raise WorkflowError("credentials", f"{detail}. The token was rejected.") from None - if error.code == 403: - raise WorkflowError( - "permissions", f"{detail}. The token lacks a required permission." - ) from None - if error.code == 404: - raise WorkflowError( - "configuration", - f"{detail}. The repository or page was not found, or the token " - "cannot see it.", - ) from None - raise WorkflowError("api", detail) from None - except (urllib.error.URLError, TimeoutError) as error: - raise WorkflowError("api", f"Gitea API is unreachable ({error}).") from None - - -def parse_gitea_token(raw_token: str | None) -> str: - """Return the token, or fail when the secret is missing or empty.""" - token = (raw_token or "").strip() - if not token: - raise WorkflowError( - "configuration", - "No token for the wiki: set TOKEN_FOR_GITEA_WIKI, or TOKEN_FOR_GITEA.", - ) - return token - - -def describe_token(token: str) -> str: - """Describe the shape of a token without revealing any of its characters. - - Used when Gitea rejects the token, so a wrongly stored secret (the masked - text, ``NAME=value``, quotes) can be recognised from the log. - """ - facts = [f"{len(token)} characters"] - if re.fullmatch(r"[0-9a-f]{40}", token): - facts.append("looks like a Gitea token (40 hexadecimal characters)") - return ", ".join(facts) - facts.append("a Gitea token has 40 hexadecimal characters") - if set(token) <= {"*", "x", "X", "•"}: - facts.append("only mask characters: the masked text was stored") - if "=" in token: - facts.append("contains '=': the secret name may be part of the value") - if re.search(r"[\"'`]", token): - facts.append("contains a quote character") - if re.search(r"\s", token): - facts.append("contains whitespace") - return ", ".join(facts) - - -def split_repository(full_name: str | None) -> tuple[str, str]: - """Split ``owner/repo`` into its two parts.""" - owner, separator, repo = (full_name or "").partition("/") - if not separator or not owner or not repo: - raise WorkflowError("configuration", "Could not determine the Gitea repository.") - return owner, repo - - -def read_pages(wiki_dir: Path) -> dict[str, bytes]: - """Return ``{page title: content}`` for every Markdown file in ``wiki_dir``.""" - if not wiki_dir.is_dir(): - raise WorkflowError("configuration", f"The wiki source directory {wiki_dir} is missing.") - pages = {path.stem: path.read_bytes() for path in sorted(wiki_dir.glob("*.md"))} - if not pages: - # Publishing an empty directory would delete the whole wiki. - raise WorkflowError( - "configuration", f"The wiki source directory {wiki_dir} has no Markdown pages." - ) - return pages - - -def list_wiki_titles(wiki_url: str, headers: dict[str, str], send: JsonRequest) -> set[str]: - """Return the titles of the pages that exist in the wiki. - - Gitea answers 404 for a wiki that has never been written to; that is an - empty wiki, not a failure. - """ - titles: set[str] = set() - page = 1 - while True: - url = f"{wiki_url}/pages?page={page}&limit={PAGE_LIMIT}" - try: - batch = send(url, headers=headers) - except WorkflowError as error: - if error.category == "configuration" and page == 1: - return titles - raise - if not isinstance(batch, list): - raise WorkflowError("api", "Gitea returned an invalid wiki page list.") - titles.update(str(item.get("title", "")) for item in batch) - if len(batch) < PAGE_LIMIT: - return titles - page += 1 - - -def read_wiki_page( - wiki_url: str, title: str, headers: dict[str, str], send: JsonRequest -) -> bytes: - """Return the current content of one wiki page.""" - quoted = urllib.parse.quote(title, safe="") - page = send(f"{wiki_url}/page/{quoted}", headers=headers) - encoded = (page or {}).get("content_base64", "") - return base64.b64decode(encoded) - - -def publish( - pages: dict[str, bytes], - wiki_url: str, - headers: dict[str, str], - send: JsonRequest, -) -> dict[str, list[str]]: - """Make the wiki equal ``pages``; return the titles created, updated, deleted.""" - existing = list_wiki_titles(wiki_url, headers, send) - current = { - title: read_wiki_page(wiki_url, title, headers, send) - for title in sorted(existing & pages.keys()) - } - - result: dict[str, list[str]] = {"created": [], "updated": [], "deleted": []} - for title, content in pages.items(): - body = { - "title": title, - "content_base64": base64.b64encode(content).decode("ascii"), - "message": COMMIT_MESSAGE, - } - if title not in existing: - send(f"{wiki_url}/new", method="POST", headers=headers, body=body) - result["created"].append(title) - elif current[title] != content: - quoted = urllib.parse.quote(title, safe="") - send(f"{wiki_url}/page/{quoted}", method="PATCH", headers=headers, body=body) - result["updated"].append(title) - - for title in sorted(existing - pages.keys()): - quoted = urllib.parse.quote(title, safe="") - send(f"{wiki_url}/page/{quoted}", method="DELETE", headers=headers) - result["deleted"].append(title) - return result - - -def main() -> int: - """Run the publish; return the process exit code.""" - wiki_dir = Path(os.environ.get("WIKI_DIR", "wiki")) - pages = read_pages(wiki_dir) - token = parse_gitea_token(os.environ.get("WIKI_CREDENTIALS")) - owner, repo = split_repository(os.environ.get("SOURCE_REPOSITORY")) - - api_url = os.environ["GITEA_API_URL"].rstrip("/") - repository_url = ( - f"{api_url}/repos/{urllib.parse.quote(owner, safe='')}/" - f"{urllib.parse.quote(repo, safe='')}" - ) - headers = { - "Authorization": f"token {token}", - "Accept": "application/json", - "Content-Type": "application/json", - } - - try: - repository = request_json(repository_url, headers=headers) - if isinstance(repository, dict) and repository.get("has_wiki") is False: - raise WorkflowError( - "configuration", - f"The wiki is not enabled on {owner}/{repo}. Enable it in the repository settings.", - ) - print("Preflight passed: Gitea accepted the token and the wiki is enabled.") - - result = publish(pages, f"{repository_url}/wiki", headers, request_json) - except WorkflowError as error: - if error.category == "credentials": - raise WorkflowError( - "credentials", f"{error.detail} Token shape: {describe_token(token)}." - ) from None - raise - print( - f"Published {len(pages)} page(s) to the wiki of {owner}/{repo}: " - f"{len(result['created'])} created, {len(result['updated'])} updated, " - f"{len(result['deleted'])} deleted." - ) - return 0 - - -if __name__ == "__main__": - try: - sys.exit(main()) - except WorkflowError as error: - print(f"ERROR: {error}", file=sys.stderr) - sys.exit(1) diff --git a/wiki/.publish b/wiki/.publish new file mode 100644 index 0000000..9c920ae --- /dev/null +++ b/wiki/.publish @@ -0,0 +1 @@ +prune