Coverage for scripts / live_release_validation / ownership / transaction_search.py: 100.00%
143 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-10-28 16:19 +0000
1"""Transaction Search baseline: what the account looked like before the run touched it.
3GCO's regional stack turns on CloudWatch Transaction Search so its services'
4OpenTelemetry spans can be searched: a custom resource points the Region's
5X-Ray trace segment destination at ``CloudWatchLogs`` and writes the
6``gco-transaction-search-xray-access`` CloudWatch Logs resource policy, and
7X-Ray then creates the ``aws/spans`` (and Application Signals the
8``/aws/application-signals/data``) log group. All three are account- and
9Region-scoped, and none of them is a CloudFormation resource: the enabler's
10Delete is a deliberate no-op because other workloads in an account may depend
11on Transaction Search. Stack teardown therefore leaves them behind, and the
12harness has to put the account back itself.
14This module is the authority for that restoration. Before anything deploys,
15``baseline`` records per regional Region the destination and its status,
16whether the GCO policy exists (only a SHA-256 of its document, never the
17document: it names the account), and whether each span log group exists (its
18creation time, so a later same-name generation is never mistaken for the
19original). Nothing here mutates AWS state; ``cleanup/transaction_search.py``
20acts on this record and ``final-inventory`` re-observes the account and
21compares it with the record through :func:`_transaction_search_differences`.
23A log group that did not exist at baseline is only ever the run's when the
24Region's destination was ``XRay`` at baseline: had Transaction Search already
25been on, every span producer in the account writes to the same group, so it
26is retained and reported as accepted instead of deleted.
27"""
29from __future__ import annotations
31import copy
32import hashlib
33import json
34from collections.abc import Mapping
35from typing import Any, cast
37from ..models import RunContext, utc_now
38from ..ownership.log_groups import _observe_log_group_stability
40#: The account-level CloudWatch Logs resource policy GCO's enabler writes.
41TRANSACTION_SEARCH_POLICY_NAME = "gco-transaction-search-xray-access"
42#: The span log groups Transaction Search (and Application Signals) write to.
43TRANSACTION_SEARCH_LOG_GROUPS = ("aws/spans", "/aws/application-signals/data")
44#: Checkpoint state key. Kept out of ``checkpoint.baseline`` on purpose: that
45#: object stays exactly the protected-stack/ECR capture ``compare_baseline``
46#: re-captures against after teardown.
47TRANSACTION_SEARCH_BASELINE_STATE_KEY = "transaction_search_baseline"
49_SCHEMA_VERSION = 1
50_DESTINATIONS = frozenset({"XRay", "CloudWatchLogs"})
51_STATUSES = frozenset({"PENDING", "ACTIVE"})
52#: Consecutive agreeing reads before a log group's presence counts as settled.
53_STABLE_OBSERVATIONS = 2
54_XRAY_SERVICE_PRINCIPAL = "xray.amazonaws.com"
57def _read_trace_segment_destination(xray: Any) -> dict[str, str]:
58 """Return ``{"destination", "status"}``, failing closed on an unknown shape."""
59 response = xray.get_trace_segment_destination()
60 destination = str(response.get("Destination") or "")
61 status = str(response.get("Status") or "")
62 if destination not in _DESTINATIONS or status not in _STATUSES:
63 raise RuntimeError(
64 "X-Ray returned an unexpected trace segment destination "
65 f"{destination!r} with status {status!r}"
66 )
67 return {"destination": destination, "status": status}
70def _gco_resource_policy(logs: Any) -> dict[str, Any] | None:
71 """Return the account-level GCO Transaction Search policy record, or ``None``."""
72 found: dict[str, Any] | None = None
73 kwargs: dict[str, Any] = {}
74 while True:
75 response = logs.describe_resource_policies(**kwargs)
76 for raw in response.get("resourcePolicies") or []:
77 if not isinstance(raw, Mapping):
78 raise RuntimeError("CloudWatch Logs returned a non-object resource policy record")
79 policy = cast(Mapping[str, Any], raw)
80 if policy.get("policyName") != TRANSACTION_SEARCH_POLICY_NAME:
81 continue
82 if found is not None:
83 raise RuntimeError(
84 f"CloudWatch Logs listed resource policy {TRANSACTION_SEARCH_POLICY_NAME} twice"
85 )
86 found = {str(key): value for key, value in policy.items()}
87 token = response.get("nextToken")
88 if not token:
89 return found
90 kwargs["nextToken"] = token
93def _statement_list(document: Any) -> list[Mapping[str, Any]]:
94 statements = document.get("Statement") if isinstance(document, Mapping) else None
95 if isinstance(statements, Mapping):
96 statements = [statements]
97 if not isinstance(statements, list):
98 return []
99 return [item for item in statements if isinstance(item, Mapping)]
102def _as_list(value: Any) -> list[Any]:
103 return value if isinstance(value, list) else [value]
106def _grants_xray_span_delivery(document_text: str) -> bool:
107 """Whether a policy document lets X-Ray deliver spans (the enabler's shape)."""
108 try:
109 document = json.loads(document_text)
110 except json.JSONDecodeError:
111 return False
112 for statement in _statement_list(document):
113 principal = statement.get("Principal")
114 services = principal.get("Service") if isinstance(principal, Mapping) else None
115 if (
116 statement.get("Effect") == "Allow"
117 and _XRAY_SERVICE_PRINCIPAL in _as_list(services)
118 and "logs:PutLogEvents" in _as_list(statement.get("Action"))
119 ):
120 return True
121 return False
124def _policy_summary(policy: Mapping[str, Any] | None) -> dict[str, Any]:
125 """Sanitized evidence for the GCO policy: presence, document hash, write time."""
126 if policy is None:
127 return {
128 "present": False,
129 "document_sha256": None,
130 "last_updated_time": None,
131 "grants_xray_span_delivery": False,
132 }
133 document = str(policy.get("policyDocument") or "")
134 last_updated = policy.get("lastUpdatedTime")
135 return {
136 "present": True,
137 "document_sha256": hashlib.sha256(document.encode("utf-8")).hexdigest(),
138 "last_updated_time": (
139 last_updated
140 if isinstance(last_updated, int) and not isinstance(last_updated, bool)
141 else None
142 ),
143 "grants_xray_span_delivery": _grants_xray_span_delivery(document),
144 }
147def _log_group_presence(logs: Any, region: str, name: str) -> dict[str, Any]:
148 """Settled presence of one span log group, identified by its creation time."""
149 outcome = _observe_log_group_stability(
150 logs,
151 region,
152 name,
153 required_present=_STABLE_OBSERVATIONS,
154 required_absent=_STABLE_OBSERVATIONS,
155 )
156 if outcome["status"] == "absent":
157 return {"present": False, "creation_time": None}
158 identity = outcome.get("identity")
159 if outcome["status"] != "present" or not isinstance(identity, dict):
160 raise RuntimeError(
161 f"CloudWatch log group {region}:{name} did not settle: {outcome['status']}"
162 )
163 return {"present": True, "creation_time": int(identity["creation_time"])}
166def _observe_transaction_search_state(ctx: RunContext, region: str) -> dict[str, Any]:
167 """Read one Region's destination, GCO policy, and span log groups (read-only)."""
168 xray = ctx.session.client("xray", region_name=region)
169 logs = ctx.session.client("logs", region_name=region)
170 return {
171 "observed_at": utc_now(),
172 **_read_trace_segment_destination(xray),
173 "resource_policy": _policy_summary(_gco_resource_policy(logs)),
174 "log_groups": {
175 name: _log_group_presence(logs, region, name) for name in TRANSACTION_SEARCH_LOG_GROUPS
176 },
177 }
180def _capture_transaction_search_baseline(ctx: RunContext) -> dict[str, Any]:
181 """Observe every regional Region before the run changes anything."""
182 return {
183 "schema_version": _SCHEMA_VERSION,
184 "policy_name": TRANSACTION_SEARCH_POLICY_NAME,
185 "regions": {
186 region: _observe_transaction_search_state(ctx, region)
187 for region in ctx.deployment_regions
188 },
189 }
192def _validate_region_state(region: str, state: Any) -> None:
193 """Fail closed on a malformed checkpointed Region record."""
194 problem = f"Transaction Search baseline for {region} is malformed"
195 if not isinstance(state, dict):
196 raise RuntimeError(problem)
197 if state.get("destination") not in _DESTINATIONS or state.get("status") not in _STATUSES:
198 raise RuntimeError(f"{problem}: destination")
199 policy = state.get("resource_policy")
200 if (
201 not isinstance(policy, dict)
202 or not isinstance(policy.get("present"), bool)
203 or isinstance(policy.get("document_sha256"), str) is not policy["present"]
204 ):
205 raise RuntimeError(f"{problem}: resource policy")
206 groups = state.get("log_groups")
207 if not isinstance(groups, dict) or set(groups) != set(TRANSACTION_SEARCH_LOG_GROUPS):
208 raise RuntimeError(f"{problem}: log groups")
209 for name, group in groups.items():
210 creation_time = group.get("creation_time") if isinstance(group, dict) else None
211 if (
212 not isinstance(group, dict)
213 or not isinstance(group.get("present"), bool)
214 or (isinstance(creation_time, int) and not isinstance(creation_time, bool))
215 is not group["present"]
216 ):
217 raise RuntimeError(f"{problem}: log group {name}")
220def _validated_transaction_search_baseline(ctx: RunContext) -> dict[str, Any]:
221 """Return the checkpointed baseline, requiring exactly the deployed Regions."""
222 raw = ctx.checkpoint.state.get(TRANSACTION_SEARCH_BASELINE_STATE_KEY)
223 if not isinstance(raw, dict) or raw.get("schema_version") != _SCHEMA_VERSION:
224 raise RuntimeError(
225 "Checkpoint has no Transaction Search baseline; the account's X-Ray trace "
226 "segment destination cannot be proven or restored without it"
227 )
228 regions = raw.get("regions")
229 if not isinstance(regions, dict) or set(regions) != set(ctx.deployment_regions):
230 raise RuntimeError(
231 "Transaction Search baseline does not cover exactly the deployed Regions"
232 )
233 for region, state in regions.items():
234 _validate_region_state(str(region), state)
235 return raw
238def _transaction_search_differences(
239 baseline: Mapping[str, Any],
240 current: Mapping[str, Any],
241) -> tuple[list[str], list[dict[str, str]]]:
242 """Compare one Region's current state with its baseline.
244 Returns ``(differences, accepted)``. A span log group that appeared while
245 Transaction Search was already enabled at baseline, and still is, is
246 accepted rather than a difference: the run cannot tell its spans from
247 anybody else's in a shared service log group.
248 """
249 differences: list[str] = []
250 accepted: list[dict[str, str]] = []
251 if current["destination"] != baseline["destination"]:
252 differences.append(
253 f"trace segment destination is {current['destination']}, "
254 f"baseline {baseline['destination']}"
255 )
256 before = baseline["resource_policy"]
257 after = current["resource_policy"]
258 if before["present"] != after["present"]:
259 differences.append(
260 f"resource policy {TRANSACTION_SEARCH_POLICY_NAME} is "
261 f"{'present' if after['present'] else 'absent'}, baseline "
262 f"{'present' if before['present'] else 'absent'}"
263 )
264 elif before["present"] and before["document_sha256"] != after["document_sha256"]:
265 differences.append(
266 f"resource policy {TRANSACTION_SEARCH_POLICY_NAME} document changed since baseline"
267 )
268 shared = baseline["destination"] == "CloudWatchLogs" == current["destination"]
269 for name in TRANSACTION_SEARCH_LOG_GROUPS:
270 was = baseline["log_groups"][name]
271 now = current["log_groups"][name]
272 if was["present"]:
273 if not now["present"]:
274 differences.append(f"log group {name} existed at baseline and is gone")
275 elif now["creation_time"] != was["creation_time"]:
276 differences.append(f"log group {name} was replaced since baseline")
277 elif now["present"]:
278 if shared:
279 accepted.append(
280 {
281 "log_group": name,
282 "reason": (
283 "Transaction Search was already enabled at baseline; the span "
284 "log group is shared by every span producer in the account"
285 ),
286 }
287 )
288 else:
289 differences.append(f"log group {name} did not exist at baseline and remains")
290 return differences, accepted
293def _verify_transaction_search_restored(ctx: RunContext) -> dict[str, Any]:
294 """Re-observe every regional Region and compare it with the baseline."""
295 baseline = _validated_transaction_search_baseline(ctx)
296 regions: dict[str, Any] = {}
297 differences: list[str] = []
298 for region in sorted(baseline["regions"]):
299 before = baseline["regions"][region]
300 current = _observe_transaction_search_state(ctx, region)
301 region_differences, accepted = _transaction_search_differences(before, current)
302 regions[region] = {
303 "baseline": copy.deepcopy(before),
304 "current": current,
305 "differences": region_differences,
306 "accepted_retained": accepted,
307 }
308 differences.extend(f"{region}: {item}" for item in region_differences)
309 return {"regions": regions, "differences": differences}