certbot-ci: unzip pebble assets

certbot-ci/certbot_integration_tests/utils/pebble_artifacts.py

@@ -1,9 +1,11 @@
 # pylint: disable=missing-module-docstring
 import atexit
+import io
 import json
 import os
 import stat
 import sys
+import zipfile
 from contextlib import ExitStack
 from typing import Tuple
 
@@ -22,34 +24,43 @@
 
 def fetch(workspace: str, http_01_port: int = DEFAULT_HTTP_01_PORT) -> Tuple[str, str, str]:
     # pylint: disable=missing-function-docstring
-    suffix = 'linux-amd64' if os.name != 'nt' else 'windows-amd64.exe'
-
     file_manager = ExitStack()
     atexit.register(file_manager.close)
     pebble_path_ref = importlib_resources.files('certbot_integration_tests') / 'assets'
     assets_path = str(file_manager.enter_context(importlib_resources.as_file(pebble_path_ref)))
 
-    pebble_path = _fetch_asset('pebble', suffix, assets_path)
-    challtestsrv_path = _fetch_asset('pebble-challtestsrv', suffix, assets_path)
+    pebble_path = _fetch_asset('pebble', assets_path)
+    challtestsrv_path = _fetch_asset('pebble-challtestsrv', assets_path)
     pebble_config_path = _build_pebble_config(workspace, http_01_port, assets_path)
 
     return pebble_path, challtestsrv_path, pebble_config_path
 
 
-def _fetch_asset(asset: str, suffix: str, assets_path: str) -> str:
-    asset_path = os.path.join(assets_path, '{0}_{1}_{2}'.format(asset, PEBBLE_VERSION, suffix))
+def _fetch_asset(asset: str, assets_path: str) -> str:
+    asset_path = os.path.join(assets_path, '{0}_{1}_linux-amd64'.format(asset, PEBBLE_VERSION))
     if not os.path.exists(asset_path):
-        asset_url = ('https://github.com/letsencrypt/pebble/releases/download/{0}/{1}-{2}'
-                     .format(PEBBLE_VERSION, asset, suffix))
+        asset_url = ('https://github.com/letsencrypt/pebble/releases/download/{0}/{1}-linux-amd64.zip'
+                     .format(PEBBLE_VERSION, asset))
         response = requests.get(asset_url, timeout=30)
         response.raise_for_status()
+        asset_data = _unzip_asset(response.content, asset)
+        if asset_data is None:
+            raise ValueError(f"zipfile {asset_url} didn't contain file {asset}")
         with open(asset_path, 'wb') as file_h:
-            file_h.write(response.content)
+            file_h.write(asset_data)
     os.chmod(asset_path, os.stat(asset_path).st_mode | stat.S_IEXEC)
 
     return asset_path
 
 
+def _unzip_asset(zipped_data: bytes, asset_name: str) -> bytes | None:
+    zip_file = zipfile.ZipFile(io.BytesIO(zipped_data))
+    for entry in zip_file.filelist:
+        if not entry.is_dir() and entry.filename.endswith(asset_name):
+            return zip_file.read(entry)
+    return None
+
+
 def _build_pebble_config(workspace: str, http_01_port: int, assets_path: str) -> str:
     config_path = os.path.join(workspace, 'pebble-config.json')
     with open(config_path, 'w') as file_h: