fix(deploy): inspect HCP plan redirect and nested unknowns

Stop auto-following the json-output 307 so the guard can pin archivist.terraform.io, and fail closed when tags or settings change under nested after_unknown values.

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)

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(

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 }
+        }
+      }
+    }
+  ]
+}