#!/usr/bin/env python3 """Host failure tests: real upload/auth/header admission and server reservation. The backend uses OTA/HTTP IO/task/database doubles; a separate contract test executes pinned SDK begin/abort with injected flash/allocation failures. SDK headers/getters come only from verified IDF 5.5.0. No build or device operations. """ import argparse import os from pathlib import Path import re import shutil import subprocess import tempfile HERE = Path(__file__).resolve().parent ROOT = HERE.parents[1] parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--idf-path", type=Path, help="Explicit ESP-IDF 5.5.0 source directory") args = parser.parse_args() if args.idf_path: IDF = args.idf_path.expanduser().resolve() else: # Only this project's active build tree; never candidate builds or the # unversioned PlatformIO package, which may now contain another SDK. paths = set() for cache in (ROOT / ".pio/build").glob("*/CMakeCache.txt"): match = re.search(r"^esp-idf_SOURCE_DIR:[^=]+=(.+)$", cache.read_text(), re.M) if match: paths.add(Path(match.group(1)).resolve()) if len(paths) != 1: parser.error("Cannot identify one active build SDK; pass --idf-path for ESP-IDF 5.5.0") IDF = paths.pop() try: version_header = (IDF / "components/esp_common/include/esp_idf_version.h").read_text() version = tuple(int(re.search(r"^#define ESP_IDF_VERSION_" + part + r"\s+(\d+)\s*$", version_header, re.M).group(1)) for part in ("MAJOR", "MINOR", "PATCH")) except (OSError, AttributeError) as error: parser.error(f"Cannot verify SDK version at {IDF}: {error}") if version != (5, 5, 0): parser.error(f"ESP-IDF 5.5.0 required, found {'.'.join(map(str, version))} at {IDF}") print(f"Using ESP-IDF 5.5.0: {IDF}", flush=True) def function(source, name): match = re.search(r"^(?:static )?(?:bool|void|size_t|esp_err_t|ota_ops_entry_t\s*\*)\s*" + name + r"\([^;{}]*\)\n\{.*?^\}", source, re.M | re.S) if not match: raise RuntimeError("Production function shape changed: " + name) return match.group() + "\n" with tempfile.TemporaryDirectory(prefix="web-firmware-update-") as directory: tmp = Path(directory) shutil.copy(HERE / "fakes.h", tmp / "fakes.h") shutil.copy(HERE / "test.c", tmp / "test.c") for name in ("esp_http_server.h", "esp_err.h", "esp_ota_ops.h", "esp_system.h", "esp_timer.h", "secure_random.h", "web_cookie_auth.h", "web_httpd_adapter.h", "web_security.h", "freertos/FreeRTOS.h", "freertos/task.h"): path = tmp / name path.parent.mkdir(parents=True, exist_ok=True) path.write_text('#pragma once\n#include "fakes.h"\n') (tmp / "esp_assert.h").write_text('#define ESP_STATIC_ASSERT(c,m) _Static_assert(c,m)\n') (tmp / "esp_attr.h").write_text('#define FORCE_INLINE_ATTR static inline\n') (tmp / "esp_flash_partitions.h").write_text('#pragma once\n#include \ntypedef struct { uint32_t offset, size; } esp_partition_pos_t;\n') for component, name in (("bootloader_support", "esp_app_format.h"), ("bootloader_support", "esp_image_format.h"), ("esp_app_format", "esp_app_desc.h")): shutil.copy(IDF / "components" / component / "include" / name, tmp / name) for name in ("web_firmware_update.c", "web_firmware_update.h", "web_auth_parse.c", "web_auth_parse.h"): shutil.copy(ROOT / "src" / name, tmp / name) auth = (ROOT / "src/web_cookie_auth.c").read_text() names = ("equal", "header", "origin", "cookie", "cookies_valid", "response", "failure", "require", "web_cookie_auth_require_body") constants = '\n'.join(line for line in auth.splitlines() if line.startswith(("#define SESSION_COOKIE", "#define PRELOGIN_COOKIE"))) (tmp / "auth_production.h").write_text(constants + "\n" + '\n'.join(function(auth, n) for n in names)) httpd = (IDF / "components/esp_http_server/src/httpd_parse.c").read_text() adapter = (ROOT / "src/web_httpd_adapter.c").read_text() (tmp / "httpd_production.h").write_text( '\n'.join(function(httpd, n) for n in ("httpd_req_get_hdr_value_len", "httpd_req_get_hdr_value_str")) + '\n#define web_httpd_headers_valid adapter_headers_valid\n' + function(adapter, "web_httpd_headers_valid") + '\n#undef web_httpd_headers_valid\n') sdk_ota = (IDF / "components/app_update/esp_ota_ops.c").read_text() # Extract the actual SDK registry/type and begin/abort implementations, not # a reimplementation of their ordering. Flash and allocation are injected. registry = sdk_ota[sdk_ota.index('typedef struct ota_ops_entry_'): sdk_ota.index('const static char *TAG')] (tmp / "sdk_ota_production.h").write_text(registry + '\n' + '\n'.join( function(sdk_ota, n) for n in ("is_ota_partition", "esp_ota_init_entry", "esp_ota_begin", "get_ota_ops_entry", "esp_ota_abort"))) shutil.copy(HERE / "sdk_contract.c", tmp / "sdk_contract.c") server = (ROOT / "src/web_server.c").read_text() (tmp / "server_production.h").write_text('\n'.join(function(server, n) for n in ("web_firmware_update_reserve", "web_firmware_update_release", "web_server_reboot_current"))) # Review guards used by the same reservation. Full lifecycle harness is owned # elsewhere; its hardcoded URI count must change from 39 to 40 upstream. assert "s_transitioning != reserved" in function(server, "stop_server") assert "s_transitioning != reserved" in function(server, "start_server") assert "if (s_transitioning ||" in function(server, "web_server_replace_identity") assert '.handler = web_firmware_update_handler' in server assert '&s_firmware_uri,' in server compiler = os.environ.get("CC", "cc") command = [compiler, "-std=c11", "-Wall", "-Wextra", "-Werror", "-g", "-fsanitize=undefined", "-fsanitize-undefined-trap-on-error", "-I", str(tmp), str(tmp / "test.c"), str(tmp / "web_auth_parse.c"), "-o", str(tmp / "test")] subprocess.run(command, check=True, timeout=30) subprocess.run([str(tmp / "test")], check=True, timeout=30) command = command[:command.index(str(tmp / "test.c"))] + [str(tmp / "sdk_contract.c"), "-o", str(tmp / "sdk_contract")] subprocess.run(command, check=True, timeout=30) subprocess.run([str(tmp / "sdk_contract")], check=True, timeout=30)