diff --git a/scripts/check-terraform-release-plan.py b/scripts/check-terraform-release-plan.py index 50ada68..6803643 100644 --- a/scripts/check-terraform-release-plan.py +++ b/scripts/check-terraform-release-plan.py @@ -32,10 +32,41 @@ PLAN_ID_RE = re.compile(r"^plan-[A-Za-z0-9]+$") VERSION_LABEL_RE = re.compile(r"^[0-9a-f]{40}-[0-9]+-[0-9]+$") IGNORED_ACTIONS = {"no-op", "read"} UNSAFE_ACTIONS = {"create", "delete"} +# Wholly unknown computed attributes may be ignored. Nested unknowns on any +# other attribute are treated as changes so the version-only guard fails closed. +COMPUTED_UNKNOWN_ATTRIBUTES = frozenset({"instances", "load_balancers"}) +REDIRECT_STATUSES = {301, 302, 303, 307, 308} UrlOpen = Callable[..., Any] +class _NoRedirectHandler(urllib.request.HTTPRedirectHandler): + """Return the redirect response instead of following it.""" + + def http_error_301(self, req, fp, code, msg, headers): + return self._capture(req, fp, code, headers) + + http_error_302 = http_error_303 = http_error_307 = http_error_308 = http_error_301 + + @staticmethod + def _capture(req, fp, code, headers): + response = urllib.response.addinfourl(fp, headers, req.full_url, code=code) + response.msg = "Redirect" + return response + + +def _urlopen_without_redirects( + *handlers: urllib.request.BaseHandler, +) -> UrlOpen: + context = ssl.create_default_context() + opener = urllib.request.build_opener( + urllib.request.HTTPSHandler(context=context), + _NoRedirectHandler, + *handlers, + ) + return opener.open + + def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser() source = parser.add_mutually_exclusive_group(required=True) @@ -67,13 +98,14 @@ def download_plan_json( token: str, *, urlopen: UrlOpen | None = None, + handlers: tuple[urllib.request.BaseHandler, ...] = (), ) -> dict[str, Any]: if not PLAN_ID_RE.fullmatch(plan_id): raise ValueError(f"plan id {plan_id!r} is not a valid HCP plan id") if not token: raise ValueError("TF_API_TOKEN is required to download plan JSON") - opener = urlopen or urllib.request.urlopen + opener = urlopen or _urlopen_without_redirects(*handlers) api_url = f"https://{API_HOST}/api/v2/plans/{plan_id}/json-output" request = urllib.request.Request( api_url, @@ -90,7 +122,7 @@ def download_plan_json( raise ValueError( "plan JSON is not ready; refusing to poll the plans endpoint" ) - if first.status not in {301, 302, 303, 307, 308}: + if first.status not in REDIRECT_STATUSES: raise ValueError( f"expected a redirect from {API_HOST}, got HTTP {first.status}" ) @@ -106,7 +138,7 @@ def download_plan_json( archive_request = urllib.request.Request(location, method="GET") second = _open_pinned(opener, archive_request, allowed_host=ARCHIVE_HOST) try: - if second.status in {301, 302, 303, 307, 308}: + if second.status in REDIRECT_STATUSES: raise ValueError( f"refusing a second redirect from {ARCHIVE_HOST}" ) @@ -141,6 +173,14 @@ def _open_pinned(urlopen: UrlOpen, request: urllib.request.Request, *, allowed_h return urlopen(request, timeout=30) +def _is_nested_unknown(value: Any) -> bool: + if isinstance(value, dict): + return any(item is True or _is_nested_unknown(item) for item in value.values()) + if isinstance(value, list): + return any(item is True or _is_nested_unknown(item) for item in value) + return False + + def changed_attributes(change: dict[str, Any]) -> set[str]: before = change.get("before") or {} after = change.get("after") or {} @@ -149,9 +189,13 @@ def changed_attributes(change: dict[str, Any]) -> set[str]: changed: set[str] = set() for key in keys: unknown_value = unknown.get(key) - if unknown_value is True or ( - isinstance(unknown_value, (dict, list)) and unknown_value - ): + if unknown_value is True: + if key in COMPUTED_UNKNOWN_ATTRIBUTES: + continue + changed.add(key) + continue + if _is_nested_unknown(unknown_value): + changed.add(key) continue if before.get(key) != after.get(key): changed.add(key) diff --git a/scripts/test-terraform-release-plan-check.py b/scripts/test-terraform-release-plan-check.py index 0dac5f4..c6ec333 100644 --- a/scripts/test-terraform-release-plan-check.py +++ b/scripts/test-terraform-release-plan-check.py @@ -4,8 +4,11 @@ from __future__ import annotations import importlib.util +import io import subprocess import sys +import urllib.request +from email.message import EmailMessage from pathlib import Path from urllib.request import Request @@ -157,11 +160,100 @@ def test_download_pinning() -> list[str]: return failures +def _scripted_https_handler(fixture: bytes, archive_url: str): + calls: list[str] = [] + api_prefix = "https://app.terraform.io/api/v2/plans/" + + class ScriptedHTTPSHandler(urllib.request.BaseHandler): + handler_order = 100 + + def https_open(self, req: Request): + url = req.full_url + calls.append(url) + headers = EmailMessage() + if url.startswith(api_prefix): + headers["Location"] = archive_url + body = b"" + status = 307 + msg = "Temporary Redirect" + elif url == archive_url: + body = fixture + status = 200 + msg = "OK" + else: + raise AssertionError(f"unexpected URL {url}") + response = urllib.response.addinfourl( + io.BytesIO(body), + headers, + url, + code=status, + ) + response.msg = msg + return response + + return ScriptedHTTPSHandler(), calls + + +def test_download_standard_opener_redirect() -> list[str]: + """urllib follows the HCP 307; the guard must still inspect that first hop.""" + module = load_check_module() + fixture = (FIXTURES / "version-only.json").read_bytes() + archive_url = "https://archivist.terraform.io/v1/object/example" + api_url = f"https://app.terraform.io/api/v2/plans/{PLAN_ID}/json-output" + failures: list[str] = [] + + following_handler, following_calls = _scripted_https_handler(fixture, archive_url) + followed = urllib.request.build_opener(following_handler).open(api_url) + try: + if followed.status != 200: + failures.append( + f"standard opener first status was {followed.status}, not 200" + ) + if following_calls != [api_url, archive_url]: + failures.append(f"standard opener URLs were {following_calls}") + finally: + followed.close() + + guard_handler, guard_calls = _scripted_https_handler(fixture, archive_url) + try: + plan = module.download_plan_json( + PLAN_ID, + "test-token", + handlers=(guard_handler,), + ) + except ValueError as exc: + failures.append(f"no-redirect download failed: {exc}") + return failures + + if plan["resource_changes"][1]["address"] != ( + "module.environment.aws_elastic_beanstalk_environment.this" + ): + failures.append("no-redirect download did not return the version-only fixture") + if guard_calls != [api_url, archive_url]: + failures.append(f"no-redirect download URLs were {guard_calls}") + + following_urlopen_handler, _ = _scripted_https_handler(fixture, archive_url) + following_urlopen = urllib.request.build_opener(following_urlopen_handler).open + try: + module.download_plan_json( + PLAN_ID, + "test-token", + urlopen=following_urlopen, + ) + failures.append("redirect-following urlopen was accepted as the first hop") + except ValueError as exc: + if "expected a redirect" not in str(exc): + failures.append(f"following urlopen error was {exc}") + + return failures + + def main() -> int: cases = [ ("version-only", run_case("version-only.json"), 0), ("wrong-label", run_case("wrong-label.json"), 1), ("eb-setting-change", run_case("eb-setting-change.json"), 1), + ("nested-unknown-tags", run_case("nested-unknown-tags.json"), 1), ("iam-update", run_case("iam-update.json"), 1), ("dns-update", run_case("dns-update.json"), 1), ("create", run_case("create.json"), 1), @@ -176,6 +268,8 @@ def main() -> int: if result.returncode != expected ] download_failures = test_download_pinning() + redirect_failures = test_download_standard_opener_redirect() + download_failures.extend(redirect_failures) if failures or download_failures: if failures: print( diff --git a/scripts/testdata/terraform-release-plans/nested-unknown-tags.json b/scripts/testdata/terraform-release-plans/nested-unknown-tags.json new file mode 100644 index 0000000..a9f2f43 --- /dev/null +++ b/scripts/testdata/terraform-release-plans/nested-unknown-tags.json @@ -0,0 +1,39 @@ +{ + "resource_changes": [ + { + "address": "module.environment.aws_elastic_beanstalk_environment.this", + "mode": "managed", + "type": "aws_elastic_beanstalk_environment", + "change": { + "actions": ["update"], + "before": { + "version_label": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa-1-1", + "setting": [ + { + "namespace": "aws:elasticbeanstalk:environment", + "name": "EnvironmentType", + "value": "LoadBalanced" + } + ], + "tags": { "env": "dev", "project": "shoc" } + }, + "after": { + "version_label": "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb-2-1", + "setting": [ + { + "namespace": "aws:elasticbeanstalk:environment", + "name": "EnvironmentType", + "value": "LoadBalanced" + } + ], + "tags": { "env": "prod", "project": "shoc" } + }, + "after_unknown": { + "instances": true, + "load_balancers": true, + "tags": { "env": true } + } + } + } + ] +}