#!/usr/bin/env python3 """Build actual ssh_memory.c with a guarded heap double; no device/network work.""" import argparse import os from pathlib import Path import re import shlex import subprocess import tempfile ROOT = Path(__file__).resolve().parents[2] ENV = dict(os.environ, CCACHE_DISABLE="1") def function(source, name): match = re.search(r"\b" + re.escape(name) + r"\s*\([^;{}]*\)\s*\{", source) if not match: raise AssertionError(f"Missing function {name}") start = source.index("{", match.start()) depth = 1 end = start + 1 while depth: depth += (source[end] == "{") - (source[end] == "}") end += 1 return source[match.start():end] def normalized(text): text = re.sub(r"/\*.*?\*/|//[^\n]*", "", text, flags=re.S) return re.sub(r"\s+", "", text) def sdk_contract(sdk): heap = sdk / "components/heap" contracts = [ ("heap_caps.c", "heap_caps_get_allocated_size", """ heap_caps_get_allocated_size(void *ptr) { ptr = MULTI_HEAP_REMOVE_BLOCK_OWNER_OFFSET(ptr); heap_t *heap = find_containing_heap(ptr); assert(heap); size_t size = multi_heap_get_allocated_size(heap->heap, ptr); return MULTI_HEAP_REMOVE_BLOCK_OWNER_SIZE(size); }"""), ("multi_heap.c", "multi_heap_get_allocated_size_impl", """ multi_heap_get_allocated_size_impl(multi_heap_handle_t heap, void *p) { return tlsf_block_size(p); }"""), ("tlsf/tlsf.c", "tlsf_block_size", """ tlsf_block_size(void* ptr) { size_t size = 0; if (ptr) { const block_header_t* block = block_from_ptr(ptr); size = block_size(block); } return size; }"""), ] for path, name, expected in contracts: actual = function((heap / path).read_text(), name) assert normalized(actual) == normalized(expected), f"Reaudit {path}:{name}" multi = normalized((heap / "multi_heap.c").read_text()) assert normalized('size_t multi_heap_get_allocated_size(multi_heap_handle_t heap, void *p) ' '__attribute__((alias("multi_heap_get_allocated_size_impl")));') in multi header = (heap / "include/esp_heap_caps.h").read_text() assert "size_t heap_caps_get_allocated_size(void *ptr);" in header print("PASS installed SDK source contract: extent query, multi_heap alias, TLSF size accessor") def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--idf-path", type=Path, help="also check installed IDF extent source and compile with its version header") args = parser.parse_args() cc = shlex.split(ENV.get("CC", "cc")) with tempfile.TemporaryDirectory(prefix="ssh-memory-") as temporary: directory = Path(temporary) (directory / "esp_err.h").write_text("typedef int esp_err_t;\n") (directory / "esp_heap_caps.h").write_text(""" #pragma once #include #define MALLOC_CAP_SPIRAM (1U << 10) #define MALLOC_CAP_INTERNAL (1U << 11) #define MALLOC_CAP_8BIT (1U << 2) void *heap_caps_malloc_prefer(size_t size, size_t count, ...); size_t heap_caps_get_allocated_size(void *pointer); void heap_caps_free(void *pointer); """) version = """ #define ESP_IDF_VERSION_VAL(a,b,c) (((a) << 16) | ((b) << 8) | (c)) #define ESP_IDF_VERSION ESP_IDF_VERSION_VAL(5,5,0) """ version_header = directory / "esp_idf_version.h" version_header.write_text(version) config = directory / "sdkconfig.h" config.write_text("#define CONFIG_HEAP_POISONING_DISABLED 1\n") # Use the real volatile wipe body, without pulling in unrelated DRBG/IDF. wipe = function((ROOT / "src/secure_random.c").read_text(), "secure_wipe") (directory / "wipe.c").write_text("#include \n#include \nvoid " + wipe + "\n") common = cc + ["-std=c11", "-Wall", "-Wextra", "-Werror", "-pedantic", *shlex.split(ENV.get("CFLAGS", "-O2")), "-I", str(directory), "-I", str(ROOT / "src")] source = str(ROOT / "src/ssh_memory.c") binary = directory / "test" subprocess.run(common + [source, str(ROOT / "tests/ssh_memory/test.c"), str(directory / "wipe.c"), "-o", str(binary)], env=ENV, check=True) subprocess.run([str(binary)], env=ENV, check=True) def compile_only(expected_error=None): result = subprocess.run(common + [source, "-c", "-o", str(directory / "memory.o")], env=ENV, text=True, capture_output=True) if expected_error is None: assert result.returncode == 0, result.stderr else: assert result.returncode != 0 and expected_error in result.stderr, result.stderr for flags in ("", "#define CONFIG_HEAP_POISONING_DISABLED 0\n", "#define CONFIG_HEAP_POISONING_LIGHT 1\n", "#define CONFIG_HEAP_POISONING_COMPREHENSIVE 1\n", "#define CONFIG_HEAP_POISONING_DISABLED 1\n#define CONFIG_HEAP_POISONING_LIGHT 1\n", "#define CONFIG_HEAP_POISONING_DISABLED 1\n#define CONFIG_HEAP_POISONING_COMPREHENSIVE 1\n"): config.write_text(flags) compile_only("SSH memory requires heap poisoning disabled") config.write_text("#define CONFIG_HEAP_POISONING_DISABLED 1\n" "#define CONFIG_HEAP_POISONING_LIGHT 0\n" "#define CONFIG_HEAP_POISONING_COMPREHENSIVE 0\n") compile_only() for unsupported in ("5,4,0", "5,5,1", "5,6,0", "6,0,0"): version_header.write_text(version.replace("5,5,0", unsupported)) compile_only("Reaudit SSH memory usable extent contract for this IDF") version_header.write_text(version) print("PASS compile guards: 6 invalid poisoning profiles, explicit disabled profile, 4 unsupported IDF versions") if args.idf_path: sdk_contract(args.idf_path) version_header.write_text((args.idf_path / "components/esp_common/include/esp_idf_version.h").read_text()) compile_only() print("PASS actual module compile with installed IDF version header") else: print("SKIP installed SDK source contract (provide --idf-path)") if __name__ == "__main__": main()