fix(agent-team): close Confluence macro-loss + harden write authz (security-review)

Resolve the security-review BLOCKER and the confirmed authz/info findings on
the Confluence-writer node:

- BLOCKER (data-loss): _page_has_macros fail-OPENed (real ConfluenceClient has
  no page_has_macros/ADF methods) so every live write fell through to a
  wholesale storage-body PUT that drops Mermaid macros — the page-1540098
  diagram-loss class. Add count_storage_macros + a real ConfluenceClient
  .page_has_macros (storage-body detection); make _page_has_macros FAIL CLOSED;
  and add _assert_macros_preserved as the final backstop: refuse any storage
  write whose body carries fewer macros than the live page.
- AUTHZ-CONF-01: add an opt-in AGENT_TEAM_CONFLUENCE_ALLOWED_PAGE_IDS allowlist
  enforced server-side before a live update (model-derived page_id).
- AUTHZ-CONF-02: stop silently picking the first accessible Confluence site;
  require CONFLUENCE_CLOUD_ID when multiple resolve.
- INFO: 'applied' now fail-honest (defaults False, not True) on a missing attr.

+10 regression tests; full suite 1612 passed; ruff clean.

agent-team/agent_team/confluence/client.py

@@ -42,6 +42,7 @@ import base64
 import difflib
 import json
 import os
+import re
 from dataclasses import dataclass, field
 from typing import Any, Protocol
 from urllib import error as _urlerror
@@ -59,9 +60,19 @@ __all__ = [
     "PlannedPageUpdate",
     "UrllibHttpClient",
     "body_diff",
+    "count_storage_macros",
     "init_auth",
 ]
 
+# Confluence storage-format macros (Mermaid diagrams, info panels, etc.) are
+# serialised as ``<ac:structured-macro>`` / ``<ac:adf-extension>`` elements whose
+# payload does NOT survive a wholesale body replacement. Counting them lets a
+# writer refuse a storage update that would silently drop diagram macros — the
+# exact failure that erased every diagram on page 1540098. Case-insensitive.
+_STORAGE_MACRO_RE = re.compile(
+    r"<ac:(?:structured-macro|adf-extension)\b", re.IGNORECASE
+)
+
 # Default 2LO token endpoint (overridable via CONFLUENCE_OAUTH_TOKEN_URL).
 DEFAULT_OAUTH_TOKEN_URL = "https://auth.atlassian.com/oauth/token"
 # OAuth-native gateway base for per-cloud Confluence access.
@@ -238,11 +249,20 @@ def _resolve_cloud_id(
             if isinstance(resource, dict) and resource.get("url") == site_url:
                 chosen = resource.get("id")
                 break
-    if not chosen:
+    if not chosen and not site_url:
-        for resource in body:
+        # No site_url to disambiguate: auto-resolve ONLY when there is exactly
-            if isinstance(resource, dict) and resource.get("id"):
+        # one accessible resource. Silently picking the first of several could
-                chosen = resource["id"]
+        # target the WRONG Confluence site (a confused-deputy / wrong-blast-radius
-                break
+        # hazard) — require an explicit CONFLUENCE_CLOUD_ID instead.
+        candidates = [r["id"] for r in body if isinstance(r, dict) and r.get("id")]
+        if len(candidates) == 1:
+            chosen = candidates[0]
+        elif len(candidates) > 1:
+            raise ConfluenceError(
+                0,
+                "OAuth: multiple accessible Confluence sites; set "
+                "CONFLUENCE_CLOUD_ID (or CONFLUENCE_BASE_URL) to disambiguate",
+            )
     if not chosen:
         raise ConfluenceError(
             0, "OAuth: could not resolve cloudId (set CONFLUENCE_CLOUD_ID)"
@@ -430,6 +450,19 @@ class ConfluenceClient:
             raise ConfluenceError(status, _stringify(body))
         return self._as_dict(body)
 
+    def page_has_macros(self, page_id: str) -> bool:
+        """Whether the page's CURRENT storage body carries Confluence macros.
+
+        Reads the page (storage format) and counts ``<ac:structured-macro>`` /
+        ``<ac:adf-extension>`` elements (e.g. Mermaid diagram extensions). The
+        writer node uses this to refuse a wholesale storage overwrite that would
+        drop diagram macros. Raises :class:`ConfluenceError` if the page cannot
+        be read — the caller treats an unreadable page as "assume macros" (fail
+        closed) rather than overwriting blindly.
+        """
+        page = self.get_page(str(page_id))
+        return count_storage_macros(_extract_storage_body(page)) > 0
+
     def update_page(
         self,
         page_id: str,
@@ -527,6 +560,20 @@ class ConfluenceClient:
 # ---------------------------------------------------------------------------
 
 
+def count_storage_macros(body: str) -> int:
+    """Count Confluence macro elements in a storage-format body.
+
+    Mermaid diagrams and other extensions live in storage XHTML as
+    ``<ac:structured-macro>`` / ``<ac:adf-extension>`` elements whose payload is
+    lost if the body is wholesale-replaced. Counting them lets a writer refuse a
+    storage update that would drop macros (the page-1540098 diagram-loss class).
+    Returns ``0`` for an empty/None body.
+    """
+    if not body:
+        return 0
+    return len(_STORAGE_MACRO_RE.findall(body))
+
+
 def _extract_storage_body(page: dict[str, Any]) -> str:
     """Pull ``body.storage.value`` from a v2 page object, defaulting to ``""``."""
     body = page.get("body")

agent-team/agent_team/nodes/confluence_writer.py

@@ -445,6 +445,26 @@ def _default_confluence_client() -> ConfluenceClient:
     return ConfluenceClient()
 
 
+def _assert_page_allowed(page_id: Any) -> None:
+    """Refuse a live write to a page outside the configured allowlist (opt-in).
+
+    When ``AGENT_TEAM_CONFLUENCE_ALLOWED_PAGE_IDS`` (comma-separated page ids) is
+    set, a live update may target only those pages. Unset/empty leaves the write
+    unconstrained (preserving current behaviour) — set it in the box env to pin
+    the agent to e.g. the IT architecture-map page. The env is read at call time.
+    """
+    raw = os.environ.get("AGENT_TEAM_CONFLUENCE_ALLOWED_PAGE_IDS")
+    if not raw or not raw.strip():
+        return
+    allowed = {part.strip() for part in raw.split(",") if part.strip()}
+    if str(page_id) not in allowed:
+        raise ConfluenceWriteError(
+            f"refusing Confluence write to page {page_id!r}: not in the "
+            "AGENT_TEAM_CONFLUENCE_ALLOWED_PAGE_IDS allowlist "
+            f"({sorted(allowed)})."
+        )
+
+
 def _apply_enabled(state: PipelineState, config: dict[str, Any] | None) -> bool:
     """Decide whether a LIVE write may be issued (dry-run is the default).
 
@@ -525,6 +545,12 @@ def conf_write_node(
 
     # Live write path (only with an explicit apply flag AND prior gate approval).
     conf = client if client is not None else _default_confluence_client()
+    # Defense-in-depth: the write target page_id is model-derived. When an
+    # allowlist is configured, refuse a live update to any page not on it so a
+    # hallucinated/injected id cannot redirect an approved write to an arbitrary
+    # page the service-account token can reach.
+    if page_id:
+        _assert_page_allowed(page_id)
     try:
         if mermaid_edits and _page_has_macros(conf, page_id):
             outcome, applied = _apply_mermaid_edits(conf, page_id, mermaid_edits)
@@ -554,35 +580,23 @@ def conf_write_node(
     )
 
 
-def _current_page_version(client: ConfluenceClient, page_id: Any) -> int:
-    """Read the target page's CURRENT version number (Confluence concurrency).
-
-    ``ConfluenceClient.update_page`` requires the CURRENT version (it derives the
-    new version as ``version_number + 1`` itself). A page object missing a usable
-    ``version.number`` is treated as version 0 so the update still issues against
-    a sane baseline rather than crashing.
-    """
-    page = client.get_page(str(page_id))
-    version = page.get("version") if isinstance(page, dict) else None
-    if isinstance(version, dict):
-        number = version.get("number")
-        if isinstance(number, int):
-            return number
-    return 0
-
-
 def _apply_storage_update(
     client: ConfluenceClient, page_id: Any, draft: dict[str, Any]
 ) -> tuple[Any, bool]:
     """Issue the live storage-format page update; return ``(outcome, applied)``.
 
-    Fetches the page's CURRENT version (``update_page`` expects the current
+    Fetches the page's CURRENT version + body in one read (``update_page`` expects
-    number and bumps it internally per Confluence's optimistic-concurrency
+    the current number and bumps it internally per Confluence's optimistic-
-    contract) and calls ``update_page(..., apply=True)``. ``applied`` is read off
+    concurrency contract), enforces the macro-preservation guard so a wholesale
-    the returned :class:`PlannedPageUpdate` (``outcome.applied``) rather than
+    body replace can never DROP diagram macros (the page-1540098 data-loss class),
-    hardcoded, so a client that declines to apply is reported honestly.
+    then calls ``update_page(..., apply=True)``. ``applied`` is read off the
+    returned :class:`PlannedPageUpdate` (``outcome.applied``); a missing attribute
+    defaults to ``False`` (fail-honest — never claim a success without evidence).
     """
-    version_number = _current_page_version(client, page_id) if page_id else 0
+    version_number = 0
+    if page_id:
+        version_number, current_body = _read_version_and_body(client, page_id)
+        _assert_macros_preserved(page_id, current_body, draft.get("body_storage"))
     outcome = client.update_page(
         page_id=page_id,
         title=draft.get("title"),
@@ -590,10 +604,55 @@ def _apply_storage_update(
         version_number=version_number,
         apply=True,
     )
-    applied = bool(getattr(outcome, "applied", True))
+    applied = bool(getattr(outcome, "applied", False))
     return outcome, applied
 
 
+def _read_version_and_body(client: ConfluenceClient, page_id: Any) -> tuple[int, str]:
+    """Read the target page's CURRENT version number and storage body in one GET.
+
+    A page object missing a usable ``version.number`` is treated as version 0 so
+    the update still issues against a sane baseline rather than crashing; a
+    missing body is ``""``.
+    """
+    page = client.get_page(str(page_id))
+    version = 0
+    current_body = ""
+    if isinstance(page, dict):
+        version_obj = page.get("version")
+        if isinstance(version_obj, dict) and isinstance(version_obj.get("number"), int):
+            version = version_obj["number"]
+        body = page.get("body")
+        if isinstance(body, dict):
+            storage = body.get("storage")
+            if isinstance(storage, dict) and isinstance(storage.get("value"), str):
+                current_body = storage["value"]
+    return version, current_body
+
+
+def _assert_macros_preserved(page_id: Any, current_body: str, new_body: Any) -> None:
+    """Refuse a storage write that would DROP Confluence macros (fail closed).
+
+    A wholesale storage-format body replacement silently destroys ``<ac:...>``
+    macro extensions (Mermaid diagrams) the model-authored body does not
+    reproduce — the exact failure that erased every diagram on page 1540098. If
+    the current page carries more macros than the proposed body, refuse rather
+    than overwrite. Counting is delegated to the client's storage-macro counter.
+    """
+    from agent_team.confluence.client import count_storage_macros
+
+    current = count_storage_macros(current_body or "")
+    proposed = count_storage_macros(str(new_body or ""))
+    if current > proposed:
+        raise ConfluenceWriteError(
+            f"refusing storage write to page {page_id}: the current page carries "
+            f"{current} macro(s) but the proposed body has {proposed} — a "
+            "wholesale storage update would DROP diagram/extension macros (e.g. "
+            "Mermaid). Edit via the ADF path (get_page_adf/update_page_adf) or "
+            "preserve the existing macros in body_storage."
+        )
+
+
 def _apply_mermaid_edits(
     client: ConfluenceClient, page_id: Any, mermaid_edits: list[dict[str, Any]]
 ) -> tuple[Any, bool]:
@@ -640,26 +699,36 @@ def _apply_mermaid_edits(
             "Mermaid live apply found no Mermaid macros on the page (skip_mermaid)."
         )
     outcome = put_adf(str(page_id), plan.new_adf)
-    applied = bool(getattr(outcome, "applied", True))
+    applied = bool(getattr(outcome, "applied", False))
     return outcome, applied
 
 
 def _page_has_macros(client: ConfluenceClient, page_id: Any) -> bool:
-    """Best-effort check that the target page carries diagram macros.
+    """Check whether the target page carries diagram macros — FAIL CLOSED.
 
-    Delegates to the injected client's ``page_has_macros`` when available; a
+    Delegates to the injected client's ``page_has_macros`` when available, else
-    missing capability (older client / no page id) degrades to ``False`` so the
+    reads the page's storage body and counts ``<ac:...>`` macro elements. When
-    update falls back to a plain storage-format body update rather than crashing.
+    macro presence cannot be determined (probe raises / page unreadable), returns
+    ``True`` (assume macros) so a macro page is NEVER mistaken for a plain page and
+    routed into a destructive wholesale storage overwrite. The storage path's own
+    :func:`_assert_macros_preserved` guard is the final backstop.
     """
     if not page_id:
         return False
     checker = getattr(client, "page_has_macros", None)
-    if checker is None:
+    if callable(checker):
-        return False
+        try:
+            return bool(checker(page_id))
+        except Exception:  # noqa: BLE001 - undetermined => fail closed
+            return True
+    # No explicit capability: detect from the storage body (fail closed on error).
     try:
-        return bool(checker(page_id))
+        from agent_team.confluence.client import count_storage_macros
-    except Exception:  # noqa: BLE001 - capability probe must never crash the write
+
-        return False
+        _version, current_body = _read_version_and_body(client, page_id)
+        return count_storage_macros(current_body) > 0
+    except Exception:  # noqa: BLE001 - undetermined => fail closed
+        return True
 
 
 def _compose_revert(state: PipelineState, draft: dict[str, Any]) -> dict[str, Any]:

agent-team/tests/test_confluence_client.py

@@ -28,6 +28,7 @@ from agent_team.confluence.client import (
     ConfluenceError,
     PlannedPageUpdate,
     body_diff,
+    count_storage_macros,
     init_auth,
 )
 
@@ -477,3 +478,53 @@ def test_body_diff_shows_added_and_removed() -> None:
     assert "page/99@planned" in delta
     assert "-line b" in delta
     assert "+line c" in delta
+
+
+# --------------------------------------------------------------------------- #
+# count_storage_macros + page_has_macros (macro-preservation support)
+# --------------------------------------------------------------------------- #
+
+
+def test_count_storage_macros_counts_structured_and_adf_extensions() -> None:
+    body = (
+        '<ac:structured-macro ac:name="mermaid-cloud"/>'
+        "<p>text</p>"
+        '<AC:STRUCTURED-MACRO ac:name="info"/>'  # case-insensitive
+        "<ac:adf-extension>x</ac:adf-extension>"
+    )
+    assert count_storage_macros(body) == 3
+    assert count_storage_macros("") == 0
+    assert count_storage_macros("<p>no macros here</p>") == 0
+
+
+def test_page_has_macros_true_when_body_carries_macro() -> None:
+    page = _storage_page("1540098", 7, '<ac:structured-macro ac:name="mermaid"/>')
+    http = FakeHttp(get=(200, page))
+    auth = ConfluenceAuth(mode="basic", base="https://x.atlassian.net")
+    client = ConfluenceClient(http=http, auth=auth)
+    assert client.page_has_macros("1540098") is True
+
+
+def test_page_has_macros_false_for_plain_body() -> None:
+    page = _storage_page("100", 1, "<p>plain</p>")
+    http = FakeHttp(get=(200, page))
+    auth = ConfluenceAuth(mode="basic", base="https://x.atlassian.net")
+    client = ConfluenceClient(http=http, auth=auth)
+    assert client.page_has_macros("100") is False
+
+
+def test_oauth_cloud_id_ambiguous_multiple_resources_raises(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    # AUTHZ-CONF-02: with no site_url to disambiguate and MULTIPLE accessible
+    # sites, refuse to silently pick the first — require CONFLUENCE_CLOUD_ID.
+    _clear_confluence_env(monkeypatch)
+    monkeypatch.setenv("CONFLUENCE_OAUTH_CLIENT_ID", "cid")
+    monkeypatch.setenv("CONFLUENCE_OAUTH_CLIENT_SECRET", "secret")
+    resources = [
+        {"id": "site-a", "url": "https://a.atlassian.net"},
+        {"id": "site-b", "url": "https://b.atlassian.net"},
+    ]
+    http = FakeHttp(post=(200, {"access_token": "t"}), get=(200, resources))
+    with pytest.raises(ConfluenceError, match="multiple accessible Confluence sites"):
+        init_auth(http)

agent-team/tests/test_confluence_writer.py

@@ -65,8 +65,9 @@ def _restore_invoker():
 
 @pytest.fixture(autouse=True)
 def _clear_apply_env(monkeypatch):
-    """Ensure the apply env flag never leaks in from the host environment."""
+    """Ensure the apply/allowlist env never leaks in from the host environment."""
     monkeypatch.delenv("AGENT_TEAM_CONFLUENCE_APPLY", raising=False)
+    monkeypatch.delenv("AGENT_TEAM_CONFLUENCE_ALLOWED_PAGE_IDS", raising=False)
 
 
 def _bind_invoker(reply: str) -> list[dict[str, Any]]:
@@ -533,3 +534,126 @@ def test_conf_write_client_failure_normalized_to_write_error() -> None:
             config={"confluence_apply": True},
             client=_BoomClient(),
         )
+
+
+# --------------------------------------------------------------------------- #
+# Macro-preservation guard — a storage write must NEVER drop diagram macros
+# (the page-1540098 data-loss class; security-review BLOCKER fix)
+# --------------------------------------------------------------------------- #
+
+
+class _MacroBodyClient(FakeConfluenceClient):
+    """Fake whose current page body carries Confluence macros (storage XHTML)."""
+
+    def __init__(self, *, current_macros: int = 1, **kw: Any) -> None:
+        super().__init__(**kw)
+        self._body = "".join(
+            f'<ac:structured-macro ac:name="mermaid-cloud" id="m{i}">'
+            f'<ac:parameter ac:name="code">graph TD; A-->B{i}</ac:parameter>'
+            "</ac:structured-macro>"
+            for i in range(current_macros)
+        )
+
+    def get_page(self, page_id) -> dict[str, Any]:
+        self.get_page_calls.append(page_id)
+        return {
+            "id": page_id,
+            "version": {"number": self._current_version},
+            "body": {"storage": {"value": self._body}},
+        }
+
+
+def test_conf_write_storage_refuses_to_drop_macros() -> None:
+    # BLOCKER fix: the live page has a Mermaid macro; the model-authored body
+    # (plain <p>) has none. A wholesale storage PUT would erase the diagram, so
+    # the write must REFUSE rather than overwrite (no update_page issued).
+    client = _MacroBodyClient(current_macros=1)
+    with pytest.raises(ConfluenceWriteError, match="would DROP diagram/extension"):
+        conf_write_node(
+            _state(confluence_draft=dict(_VALID_DRAFT)),
+            config={"confluence_apply": True},
+            client=client,
+        )
+    assert client.update_calls == []  # nothing was written
+
+
+def test_conf_write_storage_allows_when_macros_preserved() -> None:
+    # When the new body reproduces the macro count, the write proceeds.
+    client = _MacroBodyClient(current_macros=1)
+    draft = dict(_VALID_DRAFT)
+    draft["body_storage"] = (
+        '<ac:structured-macro ac:name="mermaid-cloud" id="m0">'
+        '<ac:parameter ac:name="code">graph TD; A-->B0_edited</ac:parameter>'
+        "</ac:structured-macro>"
+    )
+    conf_write_node(
+        _state(confluence_draft=draft),
+        config={"confluence_apply": True},
+        client=client,
+    )
+    assert len(client.update_calls) == 1
+
+
+def test_page_has_macros_fails_closed_without_capability() -> None:
+    # A client with NO page_has_macros method must be probed via the storage body
+    # and detect macros there (fail-closed), routing a macro page with edits into
+    # the ADF path (which raises here, since this client has no ADF seam) rather
+    # than the destructive storage overwrite.
+    class _NoCapMacroClient(_MacroBodyClient):
+        page_has_macros = None  # type: ignore[assignment]
+
+    draft = dict(_VALID_DRAFT)
+    draft["mermaid_edits"] = [{"macro_id": "m0", "mermaid": "graph TD; A-->B"}]
+    with pytest.raises(ConfluenceWriteError, match="ADF persistence"):
+        conf_write_node(
+            _state(confluence_draft=draft),
+            config={"confluence_apply": True},
+            client=_NoCapMacroClient(),
+        )
+
+
+# --------------------------------------------------------------------------- #
+# Page allowlist (AUTHZ-CONF-01 defense-in-depth) + fail-honest applied default
+# --------------------------------------------------------------------------- #
+
+
+def test_conf_write_refuses_page_outside_allowlist(monkeypatch) -> None:
+    monkeypatch.setenv("AGENT_TEAM_CONFLUENCE_ALLOWED_PAGE_IDS", "999,1234")
+    client = FakeConfluenceClient()
+    with pytest.raises(ConfluenceWriteError, match="not in the .*allowlist"):
+        conf_write_node(
+            _state(confluence_draft=dict(_VALID_DRAFT)),  # page_id 1540098
+            config={"confluence_apply": True},
+            client=client,
+        )
+    assert client.update_calls == []
+
+
+def test_conf_write_allows_page_on_allowlist(monkeypatch) -> None:
+    monkeypatch.setenv("AGENT_TEAM_CONFLUENCE_ALLOWED_PAGE_IDS", "1540098, 999")
+    client = FakeConfluenceClient()
+    conf_write_node(
+        _state(confluence_draft=dict(_VALID_DRAFT)),
+        config={"confluence_apply": True},
+        client=client,
+    )
+    assert len(client.update_calls) == 1
+
+
+def test_conf_write_applied_defaults_false_when_attr_missing() -> None:
+    # Fail-honest: an outcome object lacking ``.applied`` must NOT be reported as
+    # a successful write (was: defaulted True).
+    class _AttrlessOutcome:
+        page_id = "1540098"
+
+    class _AttrlessClient(FakeConfluenceClient):
+        def update_page(self, **kw):
+            self.update_calls.append(kw)
+            return _AttrlessOutcome()
+
+    out = conf_write_node(
+        _state(confluence_draft=dict(_VALID_DRAFT)),
+        config={"confluence_apply": True},
+        client=_AttrlessClient(),
+    )
+    assert out["confluence_result"]["applied"] is False