diff --git a/.changeset/current-clippy-baseline.md b/.changeset/current-clippy-baseline.md new file mode 100644 index 000000000..a14e86242 --- /dev/null +++ b/.changeset/current-clippy-baseline.md @@ -0,0 +1,5 @@ +--- +"@googleworkspace/cli": patch +--- + +Keep Apps Script file selection compatible with the current Clippy checks. diff --git a/.changeset/docs-review-workflow.md b/.changeset/docs-review-workflow.md new file mode 100644 index 000000000..eee647198 --- /dev/null +++ b/.changeset/docs-review-workflow.md @@ -0,0 +1,9 @@ +--- +"@googleworkspace/cli": minor +--- + +Add a standalone Python Docs review example that plans one literal text replacement, +binds it to a source revision and tab, applies it through existing gws commands, +and verifies the result without retrying ambiguous writes. Validate structures +and text ranges across all tabs before normalization, and preserve attempted or +confirmed mutation outcomes through final output failures and interruptions. diff --git a/.github/workflows/docs-review.yml b/.github/workflows/docs-review.yml new file mode 100644 index 000000000..a530b13f8 --- /dev/null +++ b/.github/workflows/docs-review.yml @@ -0,0 +1,46 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +name: Docs Review Example + +on: + push: + branches: [main] + pull_request: + branches: [main] + +permissions: + contents: read + +concurrency: + group: ${{ github.workflow }}-${{ github.ref }} + cancel-in-progress: ${{ github.ref != 'refs/heads/main' }} + +jobs: + docs-review: + name: Docs Review Python Example + runs-on: ${{ matrix.os }} + strategy: + matrix: + os: [ubuntu-latest, macos-latest] + env: + PYTHONDONTWRITEBYTECODE: "1" + steps: + - uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5 # v4.3.1 + with: + persist-credentials: false + - name: Test with synthetic fixtures and stub gws + run: | + python3 --version + python3 -m unittest discover -s examples/docs-review -p 'test_*.py' -v diff --git a/crates/google-workspace-cli/src/helpers/script.rs b/crates/google-workspace-cli/src/helpers/script.rs index 11bcdebec..4b31db62d 100644 --- a/crates/google-workspace-cli/src/helpers/script.rs +++ b/crates/google-workspace-cli/src/helpers/script.rs @@ -169,13 +169,7 @@ fn process_file(path: &Path) -> Result, GwsError> { filename.trim_end_matches(".js").trim_end_matches(".gs"), ), "html" => ("HTML", filename.trim_end_matches(".html")), - "json" => { - if filename == "appsscript.json" { - ("JSON", "appsscript") - } else { - return Ok(None); - } - } + "json" if filename == "appsscript.json" => ("JSON", "appsscript"), _ => return Ok(None), }; diff --git a/examples/docs-review/README.md b/examples/docs-review/README.md new file mode 100644 index 000000000..404b80f6f --- /dev/null +++ b/examples/docs-review/README.md @@ -0,0 +1,174 @@ +# Review one Google Docs text patch + +`docs_review.py` is a standalone Python standard-library companion to `gws`. +It creates a reviewable plan, checks that the source has not changed, applies +one literal replacement, and verifies the result. It adds no authentication +implementation or Rust commands and needs no other examples. + +Requirements: Python 3.10+ on a POSIX system supporting `O_NOFOLLOW` and +directory-relative file access (Linux/macOS), with `gws` on `PATH` for live +reads/apply. Existing `gws` authentication and Model Armor environment settings +are inherited. Offline preview and tests need neither credentials nor `gws`. + +## Workflow + +All file arguments must be relative to your current working directory, with +existing parent directories. Absolute paths, `..`, symlinks (including inward +symlinks), devices and overwriting an existing plan are rejected. Plan files +are created with mode `0600`. Keep both the plan and its digest from review. + +These commands use synthetic placeholder IDs/text. The `plan` command performs +one document read, including every tab and inline suggestions. It never sends +a document mutation. Replace the placeholder IDs only when working on a +document you are authorized to edit. + +```sh +# From the repository root. printf deliberately does not append a newline. +printf '%s' 'world' > find.txt +printf '%s' 'reader' > replacement.txt + +python3 examples/docs-review/docs_review.py plan \ + --document SYNTHETIC_DOCUMENT_ID --tab t.synthetic \ + --find find.txt --replacement replacement.txt --out reviewed-plan.json + +# Inspect the full JSON plan: IDs, revision, literal text, diff, target, digest. +cat reviewed-plan.json + +# Local request/diff preview: validates the plan, calls no gws, writes no files. +python3 examples/docs-review/docs_review.py apply \ + --plan reviewed-plan.json --dry-run + +# After review: reread, compare, submit once, reread and verify. +python3 examples/docs-review/docs_review.py apply --plan reviewed-plan.json +``` + +Omit `--tab` only for a document containing exactly one tab. Child tabs count; +titles are not selectors. Use a new `--out` path when regenerating a plan. +An empty replacement file deletes the matched text. Empty finds and no-op +replacements are rejected. Do not use `echo` to prepare text files: it commonly +adds a newline, which this workflow deliberately rejects. + +The versioned plan contains the document/tab IDs, exact source revision, source +fingerprint, find/replacement text, expected count `1`, UTF-16 target offsets, +diff, and deterministic SHA-256 digest. It does not contain document titles, +the complete source, surrounding text, credentials, or command strings. +The diff is the exact removed/inserted text, not a full-document preview. +Plans still contain sensitive text if your inputs do; treat them accordingly. + +The digest detects accidental edits and binds the review fields together; it +is **not a signature or an authorization mechanism**. Anyone who can replace +the entire plan can recompute it. Protect the reviewed file and compare its +digest with your separately retained review record before applying. + +## Supported text and verification + +- Exactly one case-sensitive literal occurrence in one tab. Regex is disabled. + Overlapping occurrences also count as ambiguous. +- The match must fit inside one top-level body paragraph. It may cross text + style runs. Unicode offsets use UTF-16 code units, including emoji. +- Tables, images, headers, footers, footnotes, other paragraphs and other tabs + remain in the document. Text in non-body regions still counts toward + uniqueness: a duplicate in a header or table causes refusal. +- Apply reconstructs the plan from the fresh source and compares every field. + A changed revision, source, target or digest stops before submission. +- Exactly one `replaceAllText` request is sent with + `writeControl.requiredRevisionId` and `tabsCriteria.tabIds`. No block deletion, + full-document reupload, or arbitrary request from a plan is allowed. +- Success requires `occurrencesChanged == 1`, a new returned revision, and a + reread at that revision. The resulting text, paragraph/structure metadata, + shifted body indices, image metadata, other tabs, and untouched character + formatting must match the expected result. + +Google controls formatting inheritance **inside the replaced span**. This +example does not set or promise the replacement's character styles. It checks +the replacement text and all formatting outside that span, accepting changes +in text-run splitting that leave character formatting unchanged. Temporary +image `contentUri` values and gws `_sanitization` annotations are excluded from +fingerprints; other image/style metadata remains checked. The verification is +an API snapshot, not a visual rendering or a guarantee against later edits. + +## Deliberate limitations + +This is a text patch workflow, not a structural document editor. Paragraph +breaks, tabs/control characters, private-use characters and object markers +are rejected in find/replacement files. Moving blocks, inserting tables/images, +editing across images, and targets inside tables, tables of contents, headers, +footers or footnotes are unsupported. Structural requests cannot be supplied +through the plan. + +The selected tab must have no unresolved suggestions, named ranges or +bookmarks. Unknown tab regions, unknown structural blocks, equations, +automatic text, rich links and smart chips in the selected tab are refused. +These strict limits avoid interpreting inaccessible text, unstable anchors, +or unsupported element boundaries as a safe replacement. Other tabs are +fingerprinted and verified, not edited. Before normalization, every tab's +paragraph elements must have valid UTF-16 lengths and contiguous ranges. +Recognized regions and structural blocks must have valid object/list shapes, +including table rows and cell content. Malformed snapshots in any tab are +refused before submission or reported as ambiguous after submission. +Unknown metadata is retained for comparison; unknown structural blocks and +extra text-element fields that would be discarded are refused. New API +structures may require an explicit compatibility update. + +Each text input is limited to 16 KiB, the plan to 1 MiB, and each gws response +to 16 MiB. Source verification additionally limits total text to 250,000 +characters and its estimated expanded style representation to 16 MiB. +Oversized or malformed inputs fail closed. `--timeout SECONDS` bounds each +gws subprocess (default 60, maximum 600); there are no mutation retries. + +## Outcomes and recovery + +Commands emit JSON. `plan`, offline preview, and a verified apply exit `0` +with status `planned`, `preview`, or `applied`. Preview only validates local +data: it cannot establish that a live revision is still current. + +Exit `2` / `refused` means the companion did not submit a document write. +Correct the input or reread the source and review a **new** plan. + +Exit `3` / `ambiguous` means a write **may have applied**, or its result could +not be verified or reported successfully. This includes subprocess timeouts, +nonzero write exits (including a concurrent API 400), malformed responses, missing revisions, +unexpected reply counts, failed/mismatching rereads, and interruptions or +output failures after submission. The diagnostic JSON on stderr includes +`mutation_state: "attempted"` or `"confirmed"`. A confirmed mutation followed +by a closed stdout consumer or failed final reporting still exits `3`; it +does not claim that the write was refused or never submitted. Stdout is +flushed before success is returned, so buffered output failures are handled +inside the same outcome check. + +The original plan stays unchanged. Inspect the document and revision through +your normal tools; do not blindly rerun apply. Keep the failed plan as your review record and +create a new plan if further work is required. The companion does not persist +a separate attempt journal or prevent an operator from manually rerunning it. + +All subprocesses use argument arrays, checked exit codes and timeouts. +Raw gws output is not echoed into errors. Diagnostics produce a fixed, +redacted notice; inspect your gws/Model Armor configuration when it appears. +Model Armor block outcomes stop the workflow; settings are never disabled. +This example uses raw `gws docs documents` methods, whose executor at the +accompanying source revision does not retry requests. + +## Tests + +```sh +python3 -m unittest discover -s examples/docs-review -p 'test_*.py' -v +``` + +The suite executes the real companion CLI with a temporary stub `gws` and +synthetic documents. It passes an isolated environment, never reads real +credentials, and never contacts Google. CI runs it independently of Rust +changes on Linux and macOS. Tests cover the request contract, no-write +planning/preview, revision/source checks, Unicode/tabs, paths and plan +tampering, structures/styles, reply counts, ambiguity and plan retention. + +## API references + +- [Documents.get](https://developers.google.com/workspace/docs/api/reference/rest/v1/documents/get): + `includeTabsContent` and `suggestionsViewMode`. +- [Document structure](https://developers.google.com/workspace/docs/api/reference/rest/v1/documents): + indices are measured in UTF-16 code units; revisions are opaque. +- [ReplaceAllTextRequest](https://developers.google.com/workspace/docs/api/reference/rest/v1/documents/request#ReplaceAllTextRequest): + literal matching and tab criteria. +- [BatchUpdate / WriteControl](https://developers.google.com/workspace/docs/api/reference/rest/v1/documents/batchUpdate): + stale `requiredRevisionId` requests are rejected with HTTP 400. A required + revision in the response identifies the revision after application. diff --git a/examples/docs-review/docs_review.py b/examples/docs-review/docs_review.py new file mode 100755 index 000000000..1f122daaf --- /dev/null +++ b/examples/docs-review/docs_review.py @@ -0,0 +1,710 @@ +#!/usr/bin/env python3 +# Copyright 2026 Google LLC +# SPDX-License-Identifier: Apache-2.0 +"""Review and apply one revision-bound Google Docs text replacement (stdlib). + +Only the gws executable communicates with Google. Plans contain data, never +commands. See README.md for the intentionally limited structural support. +""" + +import argparse +from contextlib import contextmanager +import difflib +import hashlib +import hmac +import json +import math +import os +import re +import stat +import subprocess +import sys +import tempfile + + +MAX_PLAN = 1024 * 1024 +MAX_TEXT = 16 * 1024 +MAX_DOCUMENT = 16 * 1024 * 1024 +PLAN_KEYS = { + "version", "document_id", "tab_id", "revision_id", "source_sha256", + "find", "replacement", "expected_occurrences", "target", "diff", "digest", +} + + +class Refusal(Exception): + """A fixed, safe message; never construct one from subprocess output.""" + + +class Ambiguous(Refusal): + """A submission may have applied; never retry automatically.""" + + +def require(condition, message): + if not condition: + raise Refusal(message) + + +def canonical(value): + return json.dumps(value, sort_keys=True, separators=(",", ":"), + ensure_ascii=True, allow_nan=False).encode("utf-8") + + +def sha256(value): + return hashlib.sha256(canonical(value)).hexdigest() + + +def utf16(text): + return len(text.encode("utf-16-le")) // 2 + + +def unique_object(pairs): + result = {} + for key, value in pairs: + require(key not in result, "Duplicate JSON keys are not supported.") + result[key] = value + return result + + +def parse_json(data): + try: + return json.loads(data, object_pairs_hook=unique_object, + parse_constant=lambda _: invalid_json()) + except (ValueError, UnicodeError, RecursionError): + raise Refusal("Invalid UTF-8 JSON; regenerate the plan or check gws.") from None + + +def invalid_json(): + raise Refusal("Non-finite JSON numbers are not supported.") + + +def identifier(value, tab=False): + pattern = r"[A-Za-z0-9_.-]{1,200}" if tab else r"[A-Za-z0-9_-]{1,200}" + require(isinstance(value, str) and re.fullmatch(pattern, value) is not None + and ".." not in value, "Invalid document or tab ID; supply an ID, not a URL.") + return value + + +def revision(value): + require(isinstance(value, str) and 0 < len(value) <= 4096 + and not any(ord(c) < 32 or 127 <= ord(c) <= 159 for c in value), + "Missing or invalid revision; read an editable document again.") + return value + + +def text_input(value, allow_empty=False): + require(isinstance(value, str), "Find and replacement must be UTF-8 text.") + require(allow_empty or bool(value), "Find text must not be empty.") + # Docs can strip these or treat them as structural edits. Reject rather + # than silently submit a replacement different from the reviewed text. + require(not any( + ord(c) < 32 or 127 <= ord(c) <= 159 + or 0xD800 <= ord(c) <= 0xF8FF or 0xFFF9 <= ord(c) <= 0xFFFF + or ord(c) in (0x2028, 0x2029) + or 0xF0000 <= ord(c) <= 0xFFFFD + or 0x100000 <= ord(c) <= 0x10FFFD + for c in value + ), "Unsupported structural/control character; use single-paragraph plain text.") + require(len(value.encode("utf-8")) <= MAX_TEXT, "Text input exceeds 16 KiB.") + return value + + +@contextmanager +def confined_parent(path): + """Hold directory descriptors so symlink swaps cannot redirect file I/O. + + All symlinks (even inward ones) are rejected. POSIX openat/O_NOFOLLOW is + required; fail closed on platforms without it. + """ + require(isinstance(path, str) and path and not path.startswith("/") + and "\\" not in path and ":" not in path + and not any(ord(c) < 32 or 127 <= ord(c) <= 159 for c in path), + "Use a relative path within the current directory.") + parts = path.split("/") + require(all(p not in ("", ".", "..") for p in parts), + "Path traversal and empty path components are not allowed.") + require(hasattr(os, "O_NOFOLLOW") and os.open in os.supports_dir_fd, + "Safe file access requires a POSIX platform with O_NOFOLLOW.") + descriptors = [] + try: + flags = os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW + descriptors.append(os.open(".", flags)) + for part in parts[:-1]: + descriptors.append(os.open(part, flags, dir_fd=descriptors[-1])) + yield descriptors[-1], parts[-1] + except OSError: + raise Refusal( + "Cannot access path safely; check parents, symlinks and permissions." + ) from None + finally: + for descriptor in reversed(descriptors): + os.close(descriptor) + + +def read_file(path, limit): + with confined_parent(path) as (parent, name): + descriptor = os.open(name, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK, + dir_fd=parent) + with os.fdopen(descriptor, "rb") as stream: + require(stat.S_ISREG(os.fstat(stream.fileno()).st_mode), + "Input must be a regular file.") + data = stream.read(limit + 1) + require(len(data) <= limit, "Input file exceeds the supported size limit.") + try: + return data.decode("utf-8") + except UnicodeError: + raise Refusal("Input file must be valid UTF-8.") from None + + +def unused_output(parent, name): + try: + os.stat(name, dir_fd=parent, follow_symlinks=False) + except FileNotFoundError: + return + raise Refusal("Output already exists; choose a new plan path.") + + +def write_plan(parent, name, plan): + data = json.dumps(plan, ensure_ascii=True, sort_keys=True, indent=2) + "\n" + require(len(data.encode()) <= MAX_PLAN, "Plan exceeds 1 MiB; use a smaller patch.") + # Exclusive creation prevents overwrites/hardlink attacks; plans are private. + descriptor = os.open(name, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, + 0o600, dir_fd=parent) + with os.fdopen(descriptor, "w", encoding="utf-8") as stream: + stream.write(data) + stream.flush() + os.fsync(stream.fileno()) + + +class Gws: + def __init__(self, timeout): + self.timeout = timeout + self.diagnostics = False + self.mutation_state = "not_attempted" + + def call(self, method, document_id, request=None): + params = {"documentId": document_id} + if method == "get": + params.update(includeTabsContent=True, suggestionsViewMode="SUGGESTIONS_INLINE") + args = ["gws", "docs", "documents", method, + "--params", canonical(params).decode(), "--format", "json"] + if request is not None: + args += ["--json", canonical(request).decode()] + try: + # Inherit gws auth/Model Armor settings. Raw diagnostics never reach + # the terminal (they may contain document text or credentials). + with tempfile.TemporaryFile() as output, tempfile.TemporaryFile() as diagnostics: + result = subprocess.run( + args, stdin=subprocess.DEVNULL, stdout=output, + stderr=diagnostics, timeout=self.timeout, check=False, + ) + self.diagnostics |= diagnostics.tell() > 0 + require(result.returncode == 0, + "gws failed; check access, revision and Model Armor settings.") + output.seek(0) + data = output.read(MAX_DOCUMENT + 1) + require(len(data) <= MAX_DOCUMENT, "gws response exceeds 16 MiB.") + value = parse_json(data) + require(isinstance(value, dict) and "error" not in value, + "gws did not return a successful JSON object.") + return value + except subprocess.TimeoutExpired: + raise Refusal("gws timed out; check connectivity and timeout settings.") from None + except OSError: + raise Refusal("Cannot execute gws; check installation and PATH.") from None + + def get(self, document_id): + return self.call("get", document_id) + + +def walk(value, path=()): + if isinstance(value, dict): + yield path, value + for key, item in value.items(): + yield from walk(item, path + (key,)) + elif isinstance(value, list): + for index, item in enumerate(value): + yield from walk(item, path + (index,)) + + +def at(value, path): + for key in path: + value = value[key] + return value + + +def select_tab(document, document_id, tab_id): + require(isinstance(document, dict) and document.get("documentId") == document_id, + "Document identity mismatch.") + revision(document.get("revisionId")) + require(document.get("suggestionsViewMode") == "SUGGESTIONS_INLINE", + "Expected suggestions-inline source; refusing an incomplete view.") + require(isinstance(document.get("tabs"), list) and document["tabs"], + "Missing all-tabs content; check gws Docs discovery support.") + tabs = [] + + def visit(items, path): + require(isinstance(items, list), "Malformed document tabs.") + for i, item in enumerate(items): + require(isinstance(item, dict) + and isinstance(item.get("tabProperties"), dict) + and isinstance(item.get("documentTab"), dict), "Unsupported tab shape.") + identity = identifier(item["tabProperties"].get("tabId"), tab=True) + tabs.append((identity, path + (i, "documentTab"))) + if "childTabs" in item: + visit(item["childTabs"], path + (i, "childTabs")) + + visit(document["tabs"], ("tabs",)) + require(len({identity for identity, _ in tabs}) == len(tabs), "Duplicate tab IDs.") + if tab_id is None: + require(len(tabs) == 1, "Multiple tabs; select exactly one with --tab ID.") + tab_id = tabs[0][0] + matches = [path for identity, path in tabs if identity == tab_id] + require(len(matches) == 1, "Selected tab does not exist.") + return tab_id, matches[0] + + +def check_supported(tab): + require(isinstance(tab.get("body"), dict) + and isinstance(tab["body"].get("content"), list), "Tab body is missing.") + require(set(tab) <= { + "body", "headers", "footers", "footnotes", "documentStyle", "namedStyles", + "lists", "namedRanges", "inlineObjects", "positionedObjects", "bookmarks", + "suggestedDocumentStyleChanges", "suggestedNamedStylesChanges", + }, "Unsupported tab region; this example requires a known document structure.") + for _, obj in walk(tab): + require(not any(k.startswith("suggested") and v for k, v in obj.items()), + "Suggested content in selected tab is unsupported; resolve suggestions first.") + require(not obj.get("namedRanges") and not obj.get("bookmarks"), + "Named ranges and bookmarks in the selected tab are unsupported.") + if "paragraph" not in obj: + continue + for element in obj["paragraph"]["elements"]: + kinds = set(element) - {"startIndex", "endIndex"} + require(len(kinds) == 1 and kinds <= { + "textRun", "inlineObjectElement", "footnoteReference", + "horizontalRule", "pageBreak", "columnBreak"}, + "Unsupported paragraph element; rich links, equations and chips are excluded.") + + +def index_range(value): + start, end = value.get("startIndex", 0), value.get("endIndex") + require(type(start) is int and type(end) is int and 0 <= start < end, + "Invalid structural or paragraph element indices.") + return start, end + + +def check_region(region): + require(isinstance(region, dict) and isinstance(region.get("content"), list), + "Malformed document region; expected structural content.") + + +def check_content(content): + for block in content: + require(isinstance(block, dict), "Malformed structural block.") + kinds = set(block) - {"startIndex", "endIndex"} + require(len(kinds) == 1 and kinds <= { + "paragraph", "sectionBreak", "table", "tableOfContents"}, + "Unsupported document structure.") + require(isinstance(block[next(iter(kinds))], dict), "Malformed structural block value.") + index_range(block) + + +def check_table(table): + require(isinstance(table, dict), "Malformed table.") + require(type(table.get("rows")) is int and table["rows"] > 0 + and type(table.get("columns")) is int and table["columns"] > 0, + "Malformed table dimensions.") + rows = table.get("tableRows") + require(isinstance(rows, list) and len(rows) == table["rows"], "Malformed table rows.") + for row in rows: + require(isinstance(row, dict) and isinstance(row.get("tableCells"), list) + and 0 < len(row["tableCells"]) <= table["columns"], "Malformed table row.") + for cell in row["tableCells"]: + check_region(cell) + + +def check_paragraph(block): + paragraph = block["paragraph"] + require(isinstance(paragraph, dict) + and isinstance(paragraph.get("elements"), list) + and paragraph["elements"], "Malformed paragraph.") + require(isinstance(paragraph.get("paragraphStyle", {}), dict), + "Malformed paragraph style.") + cursor, paragraph_end = index_range(block) + for element in paragraph["elements"]: + require(isinstance(element, dict), "Malformed paragraph element.") + start, end = index_range(element) + require(start == cursor, "Non-contiguous paragraph indices.") + kinds = set(element) - {"startIndex", "endIndex"} + # Extra siblings of textRun would otherwise disappear in normalization. + # Unrecognized non-text unions are retained whole in unselected tabs. + require(len(kinds) == 1 and isinstance(element[next(iter(kinds))], dict), + "Malformed or unsupported paragraph element fields.") + if "textRun" in element: + run = element["textRun"] + require(isinstance(run.get("content"), str) + and isinstance(run.get("textStyle", {}), dict), "Malformed text run.") + require(utf16(run["content"]) == end - start, + "Text run does not match its UTF-16 indices.") + cursor = end + require(paragraph_end == cursor, "Paragraph end index mismatch.") + + +def check_document_structure(document): + """Validate every region/range the normalizer interprets, in every tab. + + Unknown metadata is retained unchanged. Unknown structural blocks or + text-element siblings that cannot be retained safely are refused. + """ + for _, obj in walk(document): + if "documentTab" in obj: + require(isinstance(obj["documentTab"], dict) and "body" in obj["documentTab"], + "Missing document tab body.") + if "body" in obj: + check_region(obj["body"]) + for group in ("headers", "footers", "footnotes"): + if group in obj: + require(isinstance(obj[group], dict), "Malformed document region map.") + for region in obj[group].values(): + check_region(region) + if isinstance(obj.get("content"), list): + check_content(obj["content"]) + if "paragraph" in obj: + check_paragraph(obj) + if "table" in obj: + check_table(obj["table"]) + if "tableOfContents" in obj: + check_region(obj["tableOfContents"]) + if "sectionBreak" in obj: + require(isinstance(obj["sectionBreak"], dict) + and isinstance(obj["sectionBreak"].get("sectionStyle", {}), dict), + "Malformed section break.") + + +def check_document_budget(document): + """Bound character/style expansion before constructing canonical atoms.""" + expanded_bytes = 0 + characters = 0 + for _, obj in walk(document): + if "textRun" not in obj: + continue + run = obj["textRun"] + require(isinstance(run, dict) and isinstance(run.get("content"), str), + "Malformed text run.") + length = len(run["content"]) + characters += length + metadata = {k: v for k, v in run.items() if k != "content"} + expanded_bytes += length * (len(canonical(metadata)) + 64) + require(characters <= 250000 and expanded_bytes <= MAX_DOCUMENT, + "Document is too large for safe text/style verification.") + + +def normalized(document): + # This gate is part of normalization itself, so no caller can accidentally + # compare discarded text-run indices before validating the complete source. + check_document_structure(document) + return _normalized(document) + + +def _normalized(value, path=()): + """Canonical semantic shape; text-run splitting is not a style change.""" + if isinstance(value, list): + return [_normalized(v, path + (i,)) for i, v in enumerate(value)] + if not isinstance(value, dict): + return value + result = {} + for key, item in value.items(): + if not path and key in ("revisionId", "_sanitization"): + continue + # Google refreshes this temporary URL on reads; keep all other image + # metadata, including sourceUri, object IDs, dimensions and crop data. + if key == "contentUri" and path and path[-1] == "imageProperties": + continue + if key == "elements" and path and path[-1] == "paragraph": + elements = [] + for element in item: + if "textRun" in element: + run = element["textRun"] + metadata = {k: v for k, v in run.items() if k != "content"} + metadata.setdefault("textStyle", {}) + for char in run["content"]: + elements.append({"text": char, "format": metadata}) + else: + elements.append(_normalized(element, path + (key,))) + result[key] = elements + else: + result[key] = _normalized(item, path + (key,)) + return result + + +def locate(document, tab_path, find): + """Count overlapping occurrences in every text segment of the chosen tab.""" + tab = at(document, tab_path) + matches = [] + for path, block in walk(tab): + if "paragraph" not in block: + continue + # Only top-level body paragraphs are editable. Tables, TOCs, headers, + # footers and footnotes still participate in ambiguity detection. + supported = (len(path) == 3 and path[:2] == ("body", "content") + and isinstance(path[2], int)) + cursor = block.get("startIndex", 0) + tokens = [] + indices = [] + for element in block["paragraph"]["elements"]: + if "textRun" in element: + for char in element["textRun"]["content"]: + tokens.append(char) + indices.append(cursor) + cursor += utf16(char) + else: + tokens.append("\ufffc") + indices.append(cursor) + cursor = element["endIndex"] + text = "".join(tokens) + offset = text.find(find) + while offset != -1: + matches.append((supported, tab_path + path + ("paragraph", "elements"), + offset, indices[offset])) + offset = text.find(find, offset + 1) + require(len(matches) == 1, + "Expected exactly one match in the selected tab; found zero or multiple.") + supported, path, offset, start = matches[0] + require(supported, "Target is outside a supported body paragraph.") + return path, offset, start + + +def review_diff(find, replacement): + return "".join(difflib.unified_diff( + [find + "\n"], [replacement + "\n"], fromfile="before", tofile="after", + )) + + +def build_plan(document, document_id, tab_id, find, replacement): + identifier(document_id) + text_input(find) + text_input(replacement, allow_empty=True) + require(find != replacement, "No-op replacement; choose different text.") + tab_id, tab_path = select_tab(document, document_id, tab_id) + check_document_budget(document) + source = normalized(document) + check_supported(at(document, tab_path)) + _, _, start = locate(document, tab_path, find) + result = { + "version": 1, "document_id": document_id, "tab_id": tab_id, + "revision_id": document["revisionId"], "source_sha256": sha256(source), + "find": find, "replacement": replacement, "expected_occurrences": 1, + "target": {"start_index": start, "end_index": start + utf16(find)}, + "diff": review_diff(find, replacement), + } + result["digest"] = sha256(result) + return result + + +def validate_plan(plan): + require(isinstance(plan, dict) and set(plan) == PLAN_KEYS, "Unsupported plan schema.") + require(type(plan["version"]) is int and plan["version"] == 1, + "Unsupported plan version.") + require(type(plan["expected_occurrences"]) is int and plan["expected_occurrences"] == 1, + "Plan must specify exactly one replacement.") + identifier(plan["document_id"]) + identifier(plan["tab_id"], tab=True) + revision(plan["revision_id"]) + text_input(plan["find"]) + text_input(plan["replacement"], allow_empty=True) + require(plan["find"] != plan["replacement"], "No-op replacement.") + target = plan["target"] + require(isinstance(target, dict) and set(target) == {"start_index", "end_index"} + and all(type(v) is int and v >= 1 for v in target.values()) + and target["end_index"] - target["start_index"] == utf16(plan["find"]), + "Invalid UTF-16 target range.") + require(plan["diff"] == review_diff(plan["find"], plan["replacement"]), + "Plan diff does not match its replacement.") + for field in ("digest", "source_sha256"): + require(isinstance(plan[field], str) and re.fullmatch(r"[0-9a-f]{64}", plan[field]), + "Invalid SHA-256 field.") + payload = {k: v for k, v in plan.items() if k != "digest"} + require(hmac.compare_digest(plan["digest"], sha256(payload)), + "Plan digest mismatch; regenerate and review a fresh plan.") + return plan + + +def request_body(plan): + return { + "writeControl": {"requiredRevisionId": plan["revision_id"]}, + "requests": [{"replaceAllText": { + "containsText": {"text": plan["find"], "matchCase": True, "searchByRegex": False}, + "replaceText": plan["replacement"], + "tabsCriteria": {"tabIds": [plan["tab_id"]]}, + }}], + } + + +def verify(before, after, plan, write_revision): + _, tab_path = select_tab(before, plan["document_id"], plan["tab_id"]) + select_tab(after, plan["document_id"], plan["tab_id"]) + require(after["revisionId"] == write_revision, "Post-write revision mismatch.") + check_document_budget(after) + expected, actual = normalized(before), normalized(after) + check_supported(at(after, tab_path)) + path, offset, _ = locate(before, tab_path, plan["find"]) + end = plan["target"]["end_index"] + delta = utf16(plan["replacement"]) - utf16(plan["find"]) + # Indices in the body shift; headers/footers/footnotes have separate indices. + for _, obj in walk(at(expected, tab_path)["body"]): + for key in ("startIndex", "endIndex"): + if key in obj and obj[key] >= end: + obj[key] += delta + replacement = [{"text": c} for c in plan["replacement"]] + expected_elements = at(expected, path) + expected_elements[offset:offset + len(plan["find"])] = replacement + actual_elements = at(actual, path) + # Formatting within newly inserted text is Google-controlled. Verify its + # exact characters but compare formatting of every untouched character. + for i in range(offset, offset + len(replacement)): + require(i < len(actual_elements) and "text" in actual_elements[i], + "Replacement missing from expected location.") + actual_elements[i] = {"text": actual_elements[i]["text"]} + require(actual == expected, "Post-write text, structure or untouched style mismatch.") + + +def apply_plan(plan, gws): + before = gws.get(plan["document_id"]) + rebuilt = build_plan(before, plan["document_id"], plan["tab_id"], + plan["find"], plan["replacement"]) + require(rebuilt == plan, + "Source revision or content changed; regenerate and review a new plan.") + try: + gws.mutation_state = "attempted" + reply = gws.call("batchUpdate", plan["document_id"], request_body(plan)) + require(reply.get("documentId") == plan["document_id"], "Write identity mismatch.") + replies = reply.get("replies") + require(isinstance(replies, list) and len(replies) == 1 + and isinstance(replies[0], dict), "Missing replacement reply.") + change = replies[0].get("replaceAllText") + require(isinstance(change, dict) and type(change.get("occurrencesChanged")) is int + and change["occurrencesChanged"] == 1, "Replacement count was not exactly one.") + control = reply.get("writeControl") + require(isinstance(control, dict), "Missing write revision.") + write_revision = revision(control.get("requiredRevisionId")) + require(write_revision != plan["revision_id"], "Write revision did not advance.") + after = gws.get(plan["document_id"]) + verify(before, after, plan, write_revision) + gws.mutation_state = "confirmed" + except (Refusal, OSError, ValueError, KeyError, TypeError, IndexError, RecursionError, + KeyboardInterrupt): + raise Ambiguous( + "Write may have applied; verification did not establish success. " + "Keep the original plan, inspect the document and revision, and do not " + "blindly retry. A concurrent revision rejection requires a newly reviewed plan." + ) from None + return {"status": "applied", "digest": plan["digest"], "revision_id": write_revision} + + +class Parser(argparse.ArgumentParser): + def error(self, message): + raise Refusal("Invalid arguments; use --help for supported options.") + + +def silence_failed_stdout(): + """Prevent a failed buffered write from being retried at interpreter exit.""" + try: + with open(os.devnull, "w") as sink: + os.dup2(sink.fileno(), sys.stdout.fileno()) + except (OSError, ValueError, AttributeError): + pass + + +def report_failure(error, mutation_state): + if mutation_state != "not_attempted": + message = ( + "Write was confirmed, but final reporting failed. " + if mutation_state == "confirmed" else + "Write may have applied; verification did not establish success. " + ) + outcome = { + "status": "ambiguous", "mutation_state": mutation_state, + "message": message + "Keep the original plan, inspect the document and revision, " + "and do not blindly retry. Further changes require a newly reviewed plan.", + } + code = 3 + else: + if isinstance(error, KeyboardInterrupt): + message = "Interrupted before submission." + elif isinstance(error, Refusal): + message = str(error) + else: + message = "Malformed source or inaccessible file; check inputs and regenerate." + outcome = {"status": "refused", "message": message} + code = 2 + try: + print(json.dumps(outcome), file=sys.stderr, flush=True) + except (OSError, KeyboardInterrupt): + # Neither an unavailable diagnostic channel nor another interruption + # changes whether a mutation was attempted. + pass + return code + + +def main(argv=None): + parser = Parser(description=__doc__) + commands = parser.add_subparsers(dest="command", required=True) + plan_parser = commands.add_parser("plan", help="Read a document and create a reviewable plan") + plan_parser.add_argument("--document", required=True) + plan_parser.add_argument( + "--tab", help="Exact tab ID (required when there is more than one tab)") + plan_parser.add_argument("--find", required=True, help="Relative UTF-8 input file") + plan_parser.add_argument("--replacement", required=True, help="Relative UTF-8 input file") + plan_parser.add_argument("--out", required=True, help="New relative plan file") + apply_parser = commands.add_parser("apply", help="Validate, apply once, and verify") + apply_parser.add_argument("--plan", required=True) + apply_parser.add_argument("--dry-run", action="store_true", + help="Offline plan validation and request preview; no gws calls") + for command in (plan_parser, apply_parser): + command.add_argument("--timeout", type=float, default=60, + help="Per-gws-call timeout in seconds (default: 60)") + gws = None + reporting = False + try: + args = parser.parse_args(argv) + require(math.isfinite(args.timeout) and 0 < args.timeout <= 600, + "Timeout must be greater than zero and at most 600 seconds.") + gws = Gws(args.timeout) + if args.command == "plan": + identifier(args.document) + if args.tab is not None: + identifier(args.tab, tab=True) + find = text_input(read_file(args.find, MAX_TEXT)) + replacement = text_input(read_file(args.replacement, MAX_TEXT), allow_empty=True) + with confined_parent(args.out) as (parent, name): + unused_output(parent, name) + plan = build_plan( + gws.get(args.document), args.document, args.tab, find, replacement) + write_plan(parent, name, plan) + result = {"status": "planned", "digest": plan["digest"], + "message": "Review the plan and digest before applying."} + else: + plan = validate_plan(parse_json(read_file(args.plan, MAX_PLAN))) + if args.dry_run: + result = {"status": "preview", "digest": plan["digest"], + "diff": plan["diff"], "request": request_body(plan)} + else: + result = apply_plan(plan, gws) + if gws.diagnostics: + result["gws_diagnostics"] = ( + "gws reported diagnostics; check gws and Model Armor settings. " + "Raw output was suppressed to protect document content." + ) + reporting = True + print(json.dumps(result, ensure_ascii=True, sort_keys=True), flush=True) + return 0 + except (Refusal, OSError, ValueError, KeyError, TypeError, IndexError, + RecursionError, KeyboardInterrupt) as error: + if reporting: + silence_failed_stdout() + mutation_state = gws.mutation_state if gws is not None else "not_attempted" + return report_failure(error, mutation_state) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/examples/docs-review/test_docs_review.py b/examples/docs-review/test_docs_review.py new file mode 100644 index 000000000..03f5af589 --- /dev/null +++ b/examples/docs-review/test_docs_review.py @@ -0,0 +1,771 @@ +#!/usr/bin/env python3 +# Copyright 2026 Google LLC +# SPDX-License-Identifier: Apache-2.0 +"""Behavior tests: all gws calls go to an isolated executable, never Google.""" + +import copy +import hashlib +import json +import os +from pathlib import Path +import subprocess +import sys +import tempfile +import unittest + + +SCRIPT = Path(__file__).with_name("docs_review.py") + + +def paragraph(parts, start=1): + elements = [] + cursor = start + for text, style in parts: + end = cursor + len(text.encode("utf-16-le")) // 2 + elements.append({ + "startIndex": cursor, "endIndex": end, + "textRun": {"content": text, "textStyle": style}, + }) + cursor = end + return { + "startIndex": start, "endIndex": cursor, + "paragraph": { + "elements": elements, + "paragraphStyle": {"namedStyleType": "NORMAL_TEXT"}, + }, + } + + +def document(text="Hello world.\n", revision="rev-1"): + return { + "documentId": "synthetic-doc", "revisionId": revision, + "title": "Synthetic review fixture", + "suggestionsViewMode": "SUGGESTIONS_INLINE", + "tabs": [{ + "tabProperties": {"tabId": "t.main", "title": "Main", "index": 0}, + "documentTab": { + "body": {"content": [ + {"endIndex": 1, "sectionBreak": { + "sectionStyle": {"sectionType": "CONTINUOUS"}}}, + paragraph([(text, {})]), + ]}, + }, + }], + } + + +def body(doc): + return doc["tabs"][0]["documentTab"]["body"]["content"] + + +def resign(plan): + payload = {k: v for k, v in plan.items() if k != "digest"} + plan["digest"] = hashlib.sha256(json.dumps( + payload, sort_keys=True, separators=(",", ":"), ensure_ascii=True, + allow_nan=False, + ).encode()).hexdigest() + return plan + + +# The executable checks the actual argv contract and simulates the remote +# boundary only. Its output fixtures are independent of production helpers. +STUB = r''' +import json, os, pathlib, sys, time +root = pathlib.Path(os.environ["STUB_ROOT"]) +args = sys.argv[1:] +with (root / "calls.jsonl").open("a") as f: + f.write(json.dumps(args) + "\n") +assert args[:2] == ["docs", "documents"], args +assert args[args.index("--format") + 1] == "json", args +params = json.loads(args[args.index("--params") + 1]) +assert params["documentId"] == "synthetic-doc", params +method = args[2] +mode = (root / "mode").read_text() +if method == "get": + assert params["includeTabsContent"] is True, params + assert params["suggestionsViewMode"] == "SUGGESTIONS_INLINE", params + name = "after.json" if (root / "submitted").exists() else "before.json" + if mode == "read-error" or (mode == "verify-error" and name == "after.json"): + print("secret-token \x1b[31m remote private content", file=sys.stderr) + sys.exit(1) + if mode == "armor-block": + print(json.dumps({"error": "Content blocked by Model Armor"})) + sys.exit(1) + if mode == "armor-warn": + assert os.environ["GOOGLE_WORKSPACE_CLI_SANITIZE_TEMPLATE"] == "synthetic-template" + print("secret-token \x1b[31m Model Armor warning", file=sys.stderr) + print((root / name).read_text()) +elif method == "batchUpdate": + request = json.loads(args[args.index("--json") + 1]) + (root / "request.json").write_text(json.dumps(request)) + (root / "submitted").touch() + if mode == "timeout": + time.sleep(5) + if mode in ("conflict", "write-error"): + print(json.dumps({"error": {"code": 400 if mode == "conflict" else 503, + "message": "secret-token \x1b[31m private response"}})) + sys.exit(1) + if mode == "bad-response": + print("secret-token malformed") + else: + result = {"documentId": "synthetic-doc", + "replies": [{"replaceAllText": {"occurrencesChanged": 1}}], + "writeControl": {"requiredRevisionId": "rev-2"}} + if mode == "zero": + result["replies"][0]["replaceAllText"]["occurrencesChanged"] = 0 + if mode == "two": + result["replies"][0]["replaceAllText"]["occurrencesChanged"] = 2 + if mode == "missing-reply": + result["replies"] = [] + if mode == "missing-write-revision": + result.pop("writeControl") + print(json.dumps(result)) +else: + raise AssertionError(args) +''' + + +class ReviewCliTests(unittest.TestCase): + def setUp(self): + self.temp = tempfile.TemporaryDirectory() + self.addCleanup(self.temp.cleanup) + self.root = Path(self.temp.name) + self.bin = self.root / "bin" + self.bin.mkdir() + stub = self.bin / "gws" + stub.write_text("#!" + sys.executable + "\n" + STUB) + stub.chmod(0o700) + # Do not pass real credentials/configuration into any child process. + self.env = { + "PATH": str(self.bin) + os.pathsep + os.defpath, + "STUB_ROOT": str(self.root), + "PYTHONDONTWRITEBYTECODE": "1", + "GOOGLE_WORKSPACE_CLI_CONFIG_DIR": str(self.root / "config"), + } + self.put("mode", "ok") + self.put("find.txt", "world") + self.put("replacement.txt", "reader") + self.fixture(document(), document("Hello reader.\n", "rev-2")) + + def put(self, name, value): + (self.root / name).write_text(value, encoding="utf-8") + + def fixture(self, before, after=None): + self.put("before.json", json.dumps(before)) + self.put("after.json", json.dumps(after or before)) + + def cli(self, *args): + return subprocess.run( + [sys.executable, str(SCRIPT), *args], cwd=self.root, + env=self.env, text=True, capture_output=True, timeout=10, + ) + + def plan(self, *extra): + return self.cli("plan", "--document", "synthetic-doc", + "--find", "find.txt", "--replacement", "replacement.txt", + "--out", "plan.json", *extra) + + def apply(self, *extra): + return self.cli("apply", "--plan", "plan.json", *extra) + + def load_plan(self): + return json.loads((self.root / "plan.json").read_text()) + + def calls(self): + path = self.root / "calls.jsonl" + return [json.loads(x) for x in path.read_text().splitlines()] if path.exists() else [] + + def success(self, result): + self.assertEqual(result.returncode, 0, result.stderr) + return json.loads(result.stdout) + + def refused(self, result, status="refused"): + self.assertNotEqual(result.returncode, 0) + self.assertNotIn("secret-token", result.stderr + result.stdout) + self.assertNotIn("\x1b", result.stderr + result.stdout) + try: + payload = json.loads(result.stderr) + except ValueError: + self.fail("Expected a structured refusal, got: " + result.stderr[:200]) + self.assertEqual(payload["status"], status) + + def test_plan_is_deterministic_reviewable_and_reads_only_once(self): + self.success(self.plan()) + original = (self.root / "plan.json").read_bytes() + plan = self.load_plan() + self.assertEqual(plan["version"], 1) + self.assertEqual(plan["revision_id"], "rev-1") + self.assertEqual(plan["tab_id"], "t.main") + self.assertEqual(plan["expected_occurrences"], 1) + self.assertEqual(plan["target"], {"start_index": 7, "end_index": 12}) + self.assertIn("-world", plan["diff"]) + self.assertIn("+reader", plan["diff"]) + self.assertNotIn("Synthetic review fixture", original.decode()) + self.assertNotIn("Hello", original.decode()) + self.assertEqual(resign(copy.deepcopy(plan)), plan) + self.assertEqual([c[2] for c in self.calls()], ["get"]) + (self.root / "plan.json").unlink() + self.success(self.plan()) + self.assertEqual((self.root / "plan.json").read_bytes(), original) + + def test_apply_submits_only_reviewed_revision_and_tab_then_verifies(self): + self.success(self.plan()) + original = (self.root / "plan.json").read_bytes() + result = self.success(self.apply()) + self.assertEqual(result["status"], "applied") + self.assertEqual(json.loads((self.root / "request.json").read_text()), { + "writeControl": {"requiredRevisionId": "rev-1"}, + "requests": [{"replaceAllText": { + "containsText": {"text": "world", "matchCase": True, + "searchByRegex": False}, + "replaceText": "reader", "tabsCriteria": {"tabIds": ["t.main"]}, + }}], + }) + self.assertEqual([c[2] for c in self.calls()], + ["get", "get", "batchUpdate", "get"]) + self.assertEqual((self.root / "plan.json").read_bytes(), original) + + def test_closed_stdout_after_apply_keeps_confirmed_mutation_state(self): + self.success(self.plan()) + original = (self.root / "plan.json").read_bytes() + for python_flags in [[], ["-u"]]: + with self.subTest(python_flags=python_flags): + (self.root / "submitted").unlink(missing_ok=True) + previous_calls = len(self.calls()) + read_fd, write_fd = os.pipe() + os.close(read_fd) + try: + result = subprocess.run( + [sys.executable, *python_flags, str(SCRIPT), + "apply", "--plan", "plan.json"], + cwd=self.root, env=self.env, stdout=write_fd, + stderr=subprocess.PIPE, text=True, timeout=10, + ) + finally: + os.close(write_fd) + self.assertEqual(result.returncode, 3, result.stderr) + outcome = json.loads(result.stderr) + self.assertEqual(outcome["status"], "ambiguous") + self.assertEqual(outcome["mutation_state"], "confirmed") + self.assertIn("inspect", outcome["message"]) + self.assertIn("do not blindly retry", outcome["message"]) + self.assertEqual([c[2] for c in self.calls()[previous_calls:]], + ["get", "batchUpdate", "get"]) + self.assertEqual((self.root / "plan.json").read_bytes(), original) + + def test_interruption_after_apply_returns_keeps_confirmed_mutation_state(self): + self.success(self.plan()) + original = (self.root / "plan.json").read_bytes() + # Inject only the interruption at the return boundary. The actual + # apply implementation still performs every read/write/verification. + wrapper = """ +import runpy, sys +namespace = runpy.run_path(sys.argv[1]) +real_apply = namespace["apply_plan"] +def interrupted_return(*args, **kwargs): + real_apply(*args, **kwargs) + raise KeyboardInterrupt +namespace["main"].__globals__["apply_plan"] = interrupted_return +sys.exit(namespace["main"](["apply", "--plan", "plan.json"])) +""" + result = subprocess.run( + [sys.executable, "-c", wrapper, str(SCRIPT)], + cwd=self.root, env=self.env, capture_output=True, text=True, timeout=10, + ) + self.assertEqual(result.returncode, 3, result.stderr) + self.refused(result, "ambiguous") + outcome = json.loads(result.stderr) + self.assertEqual(outcome["mutation_state"], "confirmed") + self.assertIn("do not blindly retry", outcome["message"]) + self.assertNotIn("before submission", outcome["message"]) + self.assertEqual([c[2] for c in self.calls()], + ["get", "get", "batchUpdate", "get"]) + self.assertEqual((self.root / "plan.json").read_bytes(), original) + + def test_final_serialization_failure_keeps_confirmed_mutation_state(self): + self.success(self.plan()) + original = (self.root / "plan.json").read_bytes() + wrapper = """ +import json, runpy, sys +namespace = runpy.run_path(sys.argv[1]) +real_dumps = json.dumps +def failed_result(value, *args, **kwargs): + if isinstance(value, dict) and value.get("status") == "applied": + raise ValueError("secret-token final output failure") + return real_dumps(value, *args, **kwargs) +json.dumps = failed_result +sys.exit(namespace["main"](["apply", "--plan", "plan.json"])) +""" + result = subprocess.run( + [sys.executable, "-c", wrapper, str(SCRIPT)], + cwd=self.root, env=self.env, capture_output=True, text=True, timeout=10, + ) + self.assertEqual(result.returncode, 3, result.stderr) + self.refused(result, "ambiguous") + self.assertEqual(json.loads(result.stderr)["mutation_state"], "confirmed") + self.assertEqual((self.root / "plan.json").read_bytes(), original) + + def test_zero_multiple_and_overlapping_matches_refuse_without_write(self): + for text, find in [("Nothing.\n", "world"), + ("world world\n", "world"), ("aaa\n", "aa")]: + with self.subTest(text=text): + self.fixture(document(text)) + self.put("find.txt", find) + self.refused(self.plan()) + self.assertFalse((self.root / "plan.json").exists()) + self.assertTrue(all(c[2] == "get" for c in self.calls())) + + def test_unicode_utf16_and_nested_tab_scope(self): + before = document("😀 café world.\n") + before["tabs"][0]["childTabs"] = [{ + "tabProperties": {"tabId": "t.child", "title": "Main", "index": 0}, + "documentTab": {"body": {"content": [paragraph([("world\n", {})])]}} + }] + after = copy.deepcopy(before) + after["revisionId"] = "rev-2" + body(after)[1] = paragraph([("😀 café reader.\n", {})]) + self.fixture(before, after) + self.refused(self.plan()) # No silent first-tab selection. + self.success(self.plan("--tab", "t.main")) + self.assertEqual(self.load_plan()["target"], + {"start_index": 9, "end_index": 14}) + self.success(self.apply()) + request = json.loads((self.root / "request.json").read_text()) + self.assertEqual(request["requests"][0]["replaceAllText"]["tabsCriteria"], + {"tabIds": ["t.main"]}) + + def test_replaces_unicode_in_selected_child_without_touching_parent(self): + before = document("😀targetZ\n") + child = copy.deepcopy(before["tabs"][0]) + child["tabProperties"]["tabId"] = "t.child" + before["tabs"][0]["childTabs"] = [child] + after = copy.deepcopy(before) + after["revisionId"] = "rev-2" + # Preserve Z/newline styles while allowing Google to style inserted text. + after["tabs"][0]["childTabs"][0]["documentTab"]["body"]["content"][1] = ( + paragraph([("🛰️", {"italic": True}), ("Z\n", {})])) + self.fixture(before, after) + self.put("find.txt", "😀target") + self.put("replacement.txt", "🛰️") + self.success(self.plan("--tab", "t.child")) + self.assertEqual(self.load_plan()["target"], + {"start_index": 1, "end_index": 9}) + self.success(self.apply()) + + def test_literal_matching_does_not_enable_regex_or_shell(self): + literal = "$(touch injected);.*" + self.fixture(document(literal + "\n"), document("done\n", "rev-2")) + self.put("find.txt", literal) + self.put("replacement.txt", "done") + self.success(self.plan()) + self.success(self.apply()) + self.assertFalse((self.root / "injected").exists()) + + def test_preview_is_offline_and_does_not_submit(self): + self.success(self.plan()) + calls = self.calls() + before = (self.root / "plan.json").read_bytes() + self.success(self.apply("--dry-run")) + self.assertEqual(self.calls(), calls) + self.assertEqual((self.root / "plan.json").read_bytes(), before) + + def test_source_revision_missing_or_changed_refuses(self): + self.success(self.plan()) + original = (self.root / "plan.json").read_bytes() + for doc in [document(revision="rev-new"), document("Different world.\n"), + {k: v for k, v in document().items() if k != "revisionId"}]: + with self.subTest(doc=doc): + self.fixture(doc) + self.refused(self.apply()) + self.assertFalse((self.root / "submitted").exists()) + self.assertEqual((self.root / "plan.json").read_bytes(), original) + + def test_plan_without_revision_refuses(self): + doc = document() + doc.pop("revisionId") + self.fixture(doc) + self.refused(self.plan()) + + def test_malformed_tampered_and_oversized_plan_refuse_offline(self): + self.success(self.plan()) + original = self.load_plan() + tampered = copy.deepcopy(original) + tampered["replacement"] = "unreviewed" + unknown = resign(dict(original, command="touch injected")) + wrong_target = copy.deepcopy(original) + wrong_target["target"]["start_index"] = -1 + wrong_version = resign(dict(original, version=True)) + for value in ["{", "[]", '{"version":1,"version":2}', + json.dumps(tampered), json.dumps(unknown), + json.dumps(resign(wrong_target)), json.dumps(wrong_version), + " " * (1024 * 1024 + 1)]: + with self.subTest(value=value[:90]): + self.put("plan.json", value) + before = (self.root / "plan.json").read_bytes() + calls = self.calls() + self.refused(self.apply()) + self.assertEqual(self.calls(), calls) + self.assertEqual((self.root / "plan.json").read_bytes(), before) + + def test_resigned_semantic_tampering_is_reconstructed_before_write(self): + self.success(self.plan()) + plan = self.load_plan() + altered_target = copy.deepcopy(plan) + altered_target["target"] = {"start_index": 8, "end_index": 13} + altered_fingerprint = dict(plan, source_sha256="0" * 64) + for altered in [altered_target, altered_fingerprint]: + with self.subTest(altered=altered): + self.put("plan.json", json.dumps(resign(altered))) + original = (self.root / "plan.json").read_bytes() + previous_calls = len(self.calls()) + self.refused(self.apply()) + self.assertEqual([c[2] for c in self.calls()[previous_calls:]], ["get"]) + self.assertFalse((self.root / "submitted").exists()) + self.assertEqual((self.root / "plan.json").read_bytes(), original) + + def test_unsupported_structural_edits_and_noops_refuse(self): + for find, replacement in [("", "new"), ("world", "world"), + ("world", "one\ntwo"), ("world", "\ufffc"), + ("world.\n", "new"), ("world", "\x00")]: + with self.subTest(find=find, replacement=replacement): + self.put("find.txt", find) + self.put("replacement.txt", replacement) + self.refused(self.plan()) + self.assertFalse((self.root / "plan.json").exists()) + + def test_table_header_suggestion_and_image_crossing_targets_refuse(self): + table = document() + body(table)[1] = {"startIndex": 1, "endIndex": 14, "table": { + "rows": 1, "columns": 1, "tableRows": [{"tableCells": [{ + "content": [paragraph([("world\n", {})], 3)]}]}]}} + header = document("Other.\n") + header["tabs"][0]["documentTab"]["headers"] = { + "h.1": {"content": [paragraph([("world\n", {})])]}} + suggested = document() + body(suggested)[1]["paragraph"]["elements"][0]["textRun"][ + "suggestedInsertionIds"] = ["suggestion-1"] + image = document() + body(image)[1] = paragraph([("wor", {}), ("ld\n", {})]) + elements = body(image)[1]["paragraph"]["elements"] + elements.insert(1, {"startIndex": 4, "endIndex": 5, + "inlineObjectElement": {"inlineObjectId": "img-1"}}) + elements[2]["startIndex"] += 1 + elements[2]["endIndex"] += 1 + body(image)[1]["endIndex"] += 1 + for doc in [table, header, suggested, image]: + with self.subTest(doc=doc): + self.fixture(doc) + self.refused(self.plan()) + self.assertFalse((self.root / "submitted").exists()) + + def test_duplicate_in_table_or_header_also_refuses(self): + doc = document() + doc["tabs"][0]["documentTab"]["footers"] = { + "f.1": {"content": [paragraph([("world\n", {})])]}} + self.fixture(doc) + self.refused(self.plan()) + + def test_style_run_splitting_and_deletion_are_supported(self): + before = document() + body(before)[1] = paragraph([ + ("Hello wo", {"bold": True}), ("rld", {"italic": True}), (".\n", {})]) + after = document(revision="rev-2") + body(after)[1] = paragraph([("Hello ", {"bold": True}), (".\n", {})]) + self.fixture(before, after) + self.put("replacement.txt", "") + self.success(self.plan()) + self.success(self.apply()) + + def test_preserves_tables_images_styles_and_checks_untouched_content(self): + before = document() + body(before).append({"startIndex": 14, "endIndex": 15, "paragraph": { + "elements": [{"startIndex": 14, "endIndex": 15, + "inlineObjectElement": {"inlineObjectId": "img-1"}}]}}) + before["tabs"][0]["documentTab"]["inlineObjects"] = { + "img-1": {"inlineObjectProperties": {"embeddedObject": { + "imageProperties": {"contentUri": "https://example.invalid/temporary"}, + "size": {"width": {"magnitude": 50, "unit": "PT"}}}}}} + body(before).append({"startIndex": 15, "endIndex": 25, "table": { + "rows": 1, "columns": 1, "tableRows": [{"tableCells": [{ + "content": [paragraph([("cell\n", {"bold": True})], 18)]}]}]}}) + after = copy.deepcopy(before) + after["revisionId"] = "rev-2" + body(after)[1] = paragraph([("Hello reader.\n", {})]) + # One extra UTF-16 unit shifts following structures, not their content. + def shift(value): + if isinstance(value, dict): + for k, v in value.items(): + if k in ("startIndex", "endIndex"): + value[k] = v + 1 + else: + shift(v) + elif isinstance(value, list): + for v in value: + shift(v) + shift(body(after)[2:]) + after["tabs"][0]["documentTab"]["inlineObjects"]["img-1"][ + "inlineObjectProperties"]["embeddedObject"]["imageProperties"][ + "contentUri"] = "https://example.invalid/refreshed" + self.fixture(before, after) + self.success(self.plan()) + self.success(self.apply()) + for kind in ["table", "image-id", "image-size"]: + with self.subTest(kind=kind): + changed = copy.deepcopy(after) + if kind == "table": + table_paragraph = body(changed)[3]["table"]["tableRows"][0][ + "tableCells"][0]["content"][0] + table_paragraph["paragraph"]["elements"][0]["textRun"]["content"] = "sell\n" + elif kind == "image-id": + body(changed)[2]["paragraph"]["elements"][0][ + "inlineObjectElement"]["inlineObjectId"] = "different-image" + else: + changed["tabs"][0]["documentTab"]["inlineObjects"]["img-1"][ + "inlineObjectProperties"]["embeddedObject"]["size"]["width"][ + "magnitude"] = 51 + (self.root / "submitted").unlink() + self.fixture(before, changed) + self.refused(self.apply(), "ambiguous") + + def test_post_write_mismatch_and_missing_revision_are_not_success(self): + self.success(self.plan()) + original = (self.root / "plan.json").read_bytes() + missing = document("Hello reader.\n") + missing.pop("revisionId") + changed_style = document("Hello reader.\n", "rev-2") + body(changed_style)[1]["paragraph"]["elements"][0]["textRun"][ + "textStyle"] = {"bold": True} + for doc in [document("Hello incorrect.\n", "rev-2"), missing, + changed_style, document("Hello reader.\n", "rev-unexpected")]: + with self.subTest(doc=doc): + (self.root / "submitted").unlink(missing_ok=True) + self.fixture(document(), doc) + self.refused(self.apply(), "ambiguous") + self.assertEqual((self.root / "plan.json").read_bytes(), original) + + def test_failed_write_or_verification_never_retries_and_keeps_plan(self): + self.success(self.plan()) + original = (self.root / "plan.json").read_bytes() + for mode in ["conflict", "write-error", "zero", "two", "missing-reply", + "bad-response", "missing-write-revision", "verify-error"]: + with self.subTest(mode=mode): + (self.root / "submitted").unlink(missing_ok=True) + self.put("mode", mode) + calls = len(self.calls()) + self.refused(self.apply(), "ambiguous") + writes = [c for c in self.calls()[calls:] if c[2] == "batchUpdate"] + self.assertEqual(len(writes), 1) + self.assertEqual((self.root / "plan.json").read_bytes(), original) + + def test_timeout_after_submission_reports_possible_application(self): + self.success(self.plan()) + original = (self.root / "plan.json").read_bytes() + self.put("mode", "timeout") + result = self.apply("--timeout", "0.3") + self.refused(result, "ambiguous") + self.assertIn("may have", json.loads(result.stderr)["message"]) + self.assertEqual(len([c for c in self.calls() if c[2] == "batchUpdate"]), 1) + self.assertEqual((self.root / "plan.json").read_bytes(), original) + + def test_paths_are_relative_confined_and_never_overwrite(self): + self.success(self.plan()) + original = (self.root / "plan.json").read_bytes() + self.refused(self.plan()) + self.assertEqual((self.root / "plan.json").read_bytes(), original) + for path in ["../escape", "/tmp/escape", "sub/../../escape", "bad\nname"]: + with self.subTest(path=path): + self.refused(self.cli("apply", "--plan", path)) + self.refused(self.cli("plan", "--document", "synthetic-doc", + "--find", path, "--replacement", "replacement.txt", + "--out", "new-plan.json")) + self.refused(self.cli("plan", "--document", "synthetic-doc", + "--find", "find.txt", "--replacement", "replacement.txt", + "--out", path)) + with tempfile.TemporaryDirectory() as outside: + (self.root / "escape").symlink_to(outside, target_is_directory=True) + Path(outside, "plan.json").write_bytes(original) + self.refused(self.cli("apply", "--plan", "escape/plan.json")) + self.refused(self.cli("plan", "--document", "synthetic-doc", + "--find", "find.txt", "--replacement", "replacement.txt", + "--out", "escape/new.json")) + self.assertFalse(Path(outside, "new.json").exists()) + + def test_invalid_document_id_and_gws_failure_are_redacted(self): + result = self.cli("plan", "--document", "../secret?token", + "--find", "find.txt", "--replacement", "replacement.txt", + "--out", "plan.json") + self.refused(result) + self.assertEqual(self.calls(), []) + self.put("mode", "read-error") + self.refused(self.plan()) + + def test_model_armor_diagnostics_remain_visible_without_leaking_content(self): + self.env["GOOGLE_WORKSPACE_CLI_SANITIZE_TEMPLATE"] = "synthetic-template" + self.put("mode", "armor-block") + self.refused(self.plan()) + self.assertFalse((self.root / "plan.json").exists()) + self.put("mode", "armor-warn") + result = self.success(self.plan()) + self.assertTrue(result.get("gws_diagnostics")) + self.assertNotIn("secret-token", json.dumps(result)) + self.assertNotIn("\x1b", json.dumps(result)) + + def test_unknown_regions_and_malformed_structures_fail_before_plan(self): + unknown = document() + unknown["tabs"][0]["documentTab"]["futureRegion"] = {"text": "world"} + malformed = document() + body(malformed)[0]["futureBlock"] = {} + for doc in [unknown, malformed]: + with self.subTest(doc=doc): + self.fixture(doc) + self.refused(self.plan()) + self.assertFalse((self.root / "plan.json").exists()) + + def test_null_scalar_and_incomplete_structures_refuse_before_plan(self): + cases = [] + for value in [None, 3, {}, {"rows": 1, "columns": 1, "tableRows": None}, + {"rows": 1, "columns": 1, "tableRows": [None]}, + {"rows": 1, "columns": 1, "tableRows": [{"tableCells": [None]}]}, + {"rows": 1, "columns": 1, "tableRows": [ + {"tableCells": [{"content": "not structural content"}]}]}]: + doc = document() + body(doc).append({"startIndex": 14, "endIndex": 15, "table": value}) + cases.append(doc) + for region in ["headers", "footers", "footnotes"]: + for value in [None, [], {"segment-1": None}, + {"segment-1": {"content": "not structural content"}}]: + doc = document() + doc["tabs"][0]["documentTab"][region] = value + cases.append(doc) + for kind in ["sectionBreak", "tableOfContents"]: + doc = document() + body(doc).append({"startIndex": 14, "endIndex": 15, kind: None}) + cases.append(doc) + for doc in cases: + with self.subTest(document=doc): + (self.root / "plan.json").unlink(missing_ok=True) + self.fixture(doc) + self.refused(self.plan()) + self.assertFalse((self.root / "plan.json").exists()) + self.assertFalse((self.root / "submitted").exists()) + + def test_unselected_tab_malformed_ranges_refuse_preflight_and_postwrite(self): + before = document() + child = copy.deepcopy(document("😀 untouched\n")["tabs"][0]) + child["tabProperties"]["tabId"] = "t.other" + before["tabs"].append(child) + after = copy.deepcopy(before) + after["revisionId"] = "rev-2" + body(after)[1] = paragraph([("Hello reader.\n", {})]) + self.fixture(before, after) + self.success(self.plan("--tab", "t.main")) + original = (self.root / "plan.json").read_bytes() + for phase in ["preflight", "postwrite"]: + for field, value in [("endIndex", 999), ("startIndex", 2), + ("endIndex", True)]: + with self.subTest(phase=phase, field=field, value=value): + (self.root / "submitted").unlink(missing_ok=True) + bad = copy.deepcopy(before if phase == "preflight" else after) + bad["tabs"][1]["documentTab"]["body"]["content"][1][ + "paragraph"]["elements"][0][field] = value + self.fixture(bad if phase == "preflight" else before, + bad if phase == "postwrite" else after) + previous_calls = len(self.calls()) + self.refused(self.apply(), + "refused" if phase == "preflight" else "ambiguous") + expected_calls = (["get"] if phase == "preflight" else + ["get", "batchUpdate", "get"]) + self.assertEqual([c[2] for c in self.calls()[previous_calls:]], + expected_calls) + self.assertEqual((self.root / "plan.json").read_bytes(), original) + + def test_unselected_text_run_shape_is_checked_before_discarding_fields(self): + for extra in [{"textRun": None}, {"textRun": {"content": "extra\n", + "textStyle": "invalid"}}, + {"futureElementMetadata": {"value": "must not disappear"}}]: + with self.subTest(extra=extra): + (self.root / "plan.json").unlink(missing_ok=True) + doc = document() + other = copy.deepcopy(document("extra\n")["tabs"][0]) + other["tabProperties"]["tabId"] = "t.other" + other["documentTab"]["body"]["content"][1]["paragraph"]["elements"][0].update( + extra) + doc["tabs"].append(other) + self.fixture(doc) + self.refused(self.plan("--tab", "t.main")) + self.assertFalse((self.root / "plan.json").exists()) + self.assertFalse((self.root / "submitted").exists()) + + def test_omitted_empty_text_style_does_not_fail_verification(self): + before = document() + after = document("Hello reader.\n", "rev-2") + body(after)[1]["paragraph"]["elements"][0]["textRun"].pop("textStyle") + self.fixture(before, after) + self.success(self.plan()) + self.success(self.apply()) + + def test_c1_control_characters_in_paths_are_rejected_before_read(self): + self.put("unsafe\u0085.txt", "world") + self.refused(self.cli( + "plan", "--document", "synthetic-doc", "--find", "unsafe\u0085.txt", + "--replacement", "replacement.txt", "--out", "plan.json", + )) + self.assertEqual(self.calls(), []) + + def test_nonregular_files_and_leaf_symlinks_refuse(self): + os.mkfifo(self.root / "pipe") + (self.root / "link.txt").symlink_to(self.root / "find.txt") + for path in ["pipe", "link.txt"]: + with self.subTest(path=path): + self.refused(self.cli( + "plan", "--document", "synthetic-doc", "--find", path, + "--replacement", "replacement.txt", "--out", "plan.json", + )) + self.assertEqual(self.calls(), []) + + def test_invalid_utf8_oversized_text_and_timeout_values_refuse(self): + for data in [b"\xff", b"x" * (16 * 1024 + 1)]: + with self.subTest(size=len(data)): + (self.root / "find.txt").write_bytes(data) + self.refused(self.plan()) + for timeout in ["nan", "inf", "0", "-1", "601"]: + with self.subTest(timeout=timeout): + self.refused(self.plan("--timeout", timeout)) + self.assertEqual(self.calls(), []) + + def test_malformed_source_and_unsupported_anchors_refuse(self): + for value in ["[]", "{", '{"documentId":NaN}']: + with self.subTest(value=value): + self.put("before.json", value) + self.refused(self.plan()) + for key in ["namedRanges", "bookmarks"]: + doc = document() + doc["tabs"][0]["documentTab"][key] = {"synthetic-anchor": {}} + self.fixture(doc) + self.refused(self.plan()) + self.assertFalse((self.root / "submitted").exists()) + + def test_large_expanded_style_payload_refuses_before_plan(self): + doc = document("world" + "a" * 20000 + "\n") + body(doc)[1]["paragraph"]["elements"][0]["textRun"]["textStyle"] = { + "link": {"url": "https://example.invalid/" + "x" * 2000}} + self.fixture(doc) + self.refused(self.plan()) + self.assertFalse((self.root / "plan.json").exists()) + + def test_default_zero_indices_in_untouched_header_are_supported(self): + before = document() + header = paragraph([("Header\n", {})], start=0) + del header["startIndex"] + del header["paragraph"]["elements"][0]["startIndex"] + before["tabs"][0]["documentTab"]["headers"] = {"h.1": {"content": [header]}} + after = copy.deepcopy(before) + after["revisionId"] = "rev-2" + body(after)[1] = paragraph([("Hello reader.\n", {})]) + self.fixture(before, after) + self.success(self.plan()) + self.success(self.apply()) + + +if __name__ == "__main__": + unittest.main()