Merge PR #13 with gzip-only WebConfig assets
Keep the graceful final-ACK TCP close and optional FEM control guards from pixitha's browser compatibility fixes. Replace Brotli negotiation and duplicate page assets with one gzip representation compressed by Google Zopfli 0.4.3 using 1,000 iterations. Compress the bootstrap, captive success, compact OTA, and static WiFi setup pages with the same validated build cache. Honor gzip quality values and wildcard precedence. Validate compressed OTA page evidence in the firmware capability gate, and add native and Python regressions plus CI coverage. Remove the unused Brotli build dependency. Validation: 54 native tests, 56 Python/browser checks, full V4.3 build and verified application flash. Actual V4 WiFi responses matched the generated gzip assets; 30 repeat downloads completed without reboot. Chromium loaded the setup and settings pages without script or transfer errors. Physical Android testing is pending because the Pi detects no phone over USB or mDNS.
This commit is contained in:
commit
02b8935794
23 files changed
+2674
-2084
No files matched your search
@@ -9,6 +9,7 @@ import unittest
|
||||
import struct
|
||||
import io
|
||||
import zipfile
|
||||
import gzip
|
||||
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
@@ -75,6 +76,21 @@ class FirmwareCapabilityCheckerTest(unittest.TestCase):
|
||||
self.assertFalse(manifest["verified"])
|
||||
self.assertIn("promised", result.stderr)
|
||||
|
||||
def test_gzip_asset_requires_complete_validated_stream(self):
|
||||
page = gzip.compress(b"<title>MeshCore firmware update</title>", mtime=0)
|
||||
for image, accepted in [(b"ELF" + page + b"symbols", True),
|
||||
(page[:-1], False),
|
||||
(page[:-8] + b"\x00" * 8, False),
|
||||
(b"MeshCore firmware update", False),
|
||||
(gzip.compress(b"different page"), False),
|
||||
(gzip.compress(b"MeshCore firmware update" + b"x" * 1048576), False)]:
|
||||
with self.subTest(accepted=accepted, size=len(image)):
|
||||
result, manifest = self.run_checker(
|
||||
image, "--expect-gzip", "ota.update.wifi=MeshCore firmware update")
|
||||
self.assertEqual(result.returncode, 0 if accepted else 1, result.stderr)
|
||||
self.assertEqual(manifest["verified"], accepted)
|
||||
self.assertEqual(manifest["verification"][0]["source"], "linked gzip asset")
|
||||
|
||||
def test_application_gate_rejects_elf_only_logging_evidence(self):
|
||||
# Debug/symbol strings in the ELF must not qualify an image whose
|
||||
# uploadable binary has had the implementation compiled or linked out.
|
||||
|
||||
@@ -0,0 +1,34 @@
|
||||
#include <gtest/gtest.h>
|
||||
#include <helpers/HttpContentEncoding.h>
|
||||
|
||||
TEST(HttpContentEncoding, DefaultAndIdentitySemantics) {
|
||||
EXPECT_TRUE(mesh::http::acceptsGzip(nullptr));
|
||||
for (const char* header : {"", " ", "identity", "br", "deflate, br"}) {
|
||||
EXPECT_FALSE(mesh::http::acceptsGzip(header)) << header;
|
||||
}
|
||||
}
|
||||
|
||||
TEST(HttpContentEncoding, BrowserListsAndQualityValues) {
|
||||
for (const char* header : {"gzip", "gzip, deflate, br", "GZip", " gzip ; q=0.1 ",
|
||||
"gzip;q=1.000", "*", "*;q=.5, gzip;q=0.001"}) {
|
||||
EXPECT_TRUE(mesh::http::acceptsGzip(header)) << header;
|
||||
}
|
||||
for (const char* header : {"gzip;q=0", "gzip;q=0.000", "gzip;q=0, br",
|
||||
"gzip;q=2", "gzip;q=1.1", "gzip;q=0.1234", "gzip;q=oops",
|
||||
"gzip;q=", "gzip;q=.5", "notgzip", "gzipish", "*;q=0"}) {
|
||||
EXPECT_FALSE(mesh::http::acceptsGzip(header)) << header;
|
||||
}
|
||||
}
|
||||
|
||||
TEST(HttpContentEncoding, ExplicitCodingOverridesWildcardRegardlessOfOrder) {
|
||||
EXPECT_FALSE(mesh::http::acceptsGzip("*;q=1, gzip;q=0"));
|
||||
EXPECT_FALSE(mesh::http::acceptsGzip("gzip;q=0, *;q=1"));
|
||||
EXPECT_TRUE(mesh::http::acceptsGzip("*;q=0, gzip;q=1"));
|
||||
EXPECT_TRUE(mesh::http::acceptsGzip("gzip;q=1, *;q=0"));
|
||||
EXPECT_TRUE(mesh::http::acceptsGzip("gzip;other=value;q=0.5"));
|
||||
}
|
||||
|
||||
int main(int argc, char** argv) {
|
||||
::testing::InitGoogleTest(&argc, argv);
|
||||
return RUN_ALL_TESTS();
|
||||
}
|
||||
@@ -0,0 +1,90 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Qualify every generated HTML gzip stream and its compression settings."""
|
||||
import ast
|
||||
import gzip
|
||||
import hashlib
|
||||
import os
|
||||
from pathlib import Path
|
||||
import re
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
from unittest import mock
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
sys.path.insert(0, str(ROOT / "scripts"))
|
||||
import zopfli_compress
|
||||
from webconfig_minify import strip_source
|
||||
|
||||
|
||||
def arrays(header):
|
||||
return {name: bytes(int(value, 16) for value in re.findall(r"0x([0-9a-f]{2})", body))
|
||||
for name, body in re.findall(r"const uint8_t (\w+)\[\] PROGMEM = \{(.*?)\};",
|
||||
header, re.S)}
|
||||
|
||||
|
||||
class WebConfigGzipTest(unittest.TestCase):
|
||||
def test_embedded_pages_round_trip_and_match_cache_keys(self):
|
||||
header = (ROOT / "src/helpers/esp32/WebConfigHtml.h").read_text()
|
||||
self.assertIn("gzip, 1000 iterations", header)
|
||||
self.assertNotIn("WEBCONFIG_HTML_BR", header)
|
||||
blobs = arrays(header)
|
||||
self.assertEqual(set(blobs), {"WEBCONFIG_HTML_GZ", "WEBCONFIG_HTML_LOADER",
|
||||
"WEBCONFIG_HTML_SUCCESS"})
|
||||
page = gzip.decompress(blobs["WEBCONFIG_HTML_GZ"])
|
||||
self.assertEqual(page, strip_source((ROOT / "webui/index.html").read_text()).encode())
|
||||
version = hashlib.sha256(blobs["WEBCONFIG_HTML_GZ"]).hexdigest()[:16]
|
||||
self.assertIn('WEBCONFIG_HTML_VERSION[] = "%s"' % version, header)
|
||||
self.assertIn(("/ui?v=" + version).encode(), gzip.decompress(blobs["WEBCONFIG_HTML_LOADER"]))
|
||||
self.assertEqual(gzip.decompress(blobs["WEBCONFIG_HTML_SUCCESS"]),
|
||||
b"<HTML><HEAD><TITLE>Success</TITLE></HEAD><BODY>Success</BODY></HTML>")
|
||||
for name, blob in blobs.items():
|
||||
declared = re.search(name + r"_LEN = (\d+);", header)
|
||||
self.assertEqual(len(blob), int(declared[1]))
|
||||
self.assertEqual(blob[:3], b"\x1f\x8b\x08")
|
||||
self.assertEqual(blob[4:8], b"\0" * 4)
|
||||
self.assertEqual(blob[9], 255)
|
||||
|
||||
def test_compact_uploader_and_wifi_form_are_precompressed(self):
|
||||
header = (ROOT / "src/helpers/esp32/StaticHtml.h").read_text()
|
||||
self.assertIn("gzip, 1000 iterations", header)
|
||||
blobs = arrays(header)
|
||||
for page in (ROOT / "webui/pages").glob("*.html"):
|
||||
self.assertEqual(gzip.decompress(blobs["MESH_HTML_" + page.stem.upper()]), page.read_bytes())
|
||||
|
||||
def test_gzip_binding_uses_1000_iterations_without_changing_raw_ota_default(self):
|
||||
with mock.patch.object(zopfli_compress._gzip, "compress",
|
||||
return_value=gzip.compress(b"page", mtime=0)) as compress:
|
||||
encoded = zopfli_compress.gzip_compress(b"page")
|
||||
self.assertEqual(compress.call_args.kwargs["numiterations"], 1000)
|
||||
self.assertEqual(gzip.decompress(encoded), b"page")
|
||||
self.assertEqual(zopfli_compress.ITERATIONS, 15)
|
||||
|
||||
def test_cache_validates_contents_and_recovers_from_corruption(self):
|
||||
# Load the production cache helper without invoking its build entry point.
|
||||
source = ROOT / "scripts/generate_webconfig_html.py"
|
||||
tree = ast.parse(source.read_text(), filename=str(source))
|
||||
self.assertIsInstance(tree.body[-1], ast.Expr)
|
||||
self.assertEqual(tree.body[-1].value.func.id, "main")
|
||||
tree.body.pop()
|
||||
namespace = {}
|
||||
exec(compile(tree, str(source), "exec"), namespace)
|
||||
original = Path.cwd()
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
try:
|
||||
os.chdir(directory)
|
||||
compress = namespace["gzip_compress"]
|
||||
with mock.patch.dict(namespace, gzip_compress=mock.Mock(wraps=compress)):
|
||||
first = namespace["compress_page"](b"<html>cache test</html>")
|
||||
self.assertEqual(first, namespace["compress_page"](b"<html>cache test</html>"))
|
||||
self.assertEqual(namespace["gzip_compress"].call_count, 1)
|
||||
cache = next(Path(".pio/gzip-html").glob("*.gz"))
|
||||
cache.write_bytes(first[:-3])
|
||||
self.assertEqual(first, namespace["compress_page"](b"<html>cache test</html>"))
|
||||
self.assertEqual(namespace["gzip_compress"].call_count, 2)
|
||||
finally:
|
||||
os.chdir(original)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -1,6 +1,6 @@
|
||||
#!/usr/bin/env python3
|
||||
|
||||
import brotli
|
||||
import gzip
|
||||
from pathlib import Path
|
||||
import re
|
||||
import unittest
|
||||
@@ -19,14 +19,14 @@ def function_body(source: str, start: str, end: str) -> str:
|
||||
|
||||
def embedded_ui() -> str:
|
||||
header = EMBEDDED_UI.read_text(encoding="utf-8")
|
||||
length = int(re.search(r"WEBCONFIG_HTML_BR_LEN = (\d+);", header).group(1))
|
||||
length = int(re.search(r"WEBCONFIG_HTML_GZ_LEN = (\d+);", header).group(1))
|
||||
array = header.split(
|
||||
"const uint8_t WEBCONFIG_HTML_BR[] PROGMEM = {", 1
|
||||
"const uint8_t WEBCONFIG_HTML_GZ[] PROGMEM = {", 1
|
||||
)[1]
|
||||
blob = bytes(
|
||||
int(value, 16) for value in re.findall(r"0x([0-9a-f]{2})", array)
|
||||
)[:length]
|
||||
return brotli.decompress(blob).decode("utf-8")
|
||||
return gzip.decompress(blob).decode("utf-8")
|
||||
|
||||
|
||||
class WebConfigSecretMaskingTest(unittest.TestCase):
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
#!/usr/bin/env python3
|
||||
|
||||
import brotli
|
||||
import gzip
|
||||
from pathlib import Path
|
||||
import re
|
||||
import unittest
|
||||
@@ -226,15 +226,15 @@ class WebConfigUiTest(unittest.TestCase):
|
||||
|
||||
def test_generated_asset_contains_show_controls(self):
|
||||
header = HEADER.read_text(encoding="utf-8")
|
||||
self.assertIn("Compressed with Google Brotli", header)
|
||||
length = int(re.search(r"WEBCONFIG_HTML_BR_LEN = (\d+);", header).group(1))
|
||||
self.assertIn("Compressed with Google Zopfli", header)
|
||||
length = int(re.search(r"WEBCONFIG_HTML_GZ_LEN = (\d+);", header).group(1))
|
||||
array = header.split(
|
||||
"const uint8_t WEBCONFIG_HTML_BR[] PROGMEM = {", 1
|
||||
"const uint8_t WEBCONFIG_HTML_GZ[] PROGMEM = {", 1
|
||||
)[1]
|
||||
blob = bytes(
|
||||
int(value, 16) for value in re.findall(r"0x([0-9a-f]{2})", array)
|
||||
)[:length]
|
||||
page = brotli.decompress(blob).decode("utf-8")
|
||||
page = gzip.decompress(blob).decode("utf-8")
|
||||
self.assertEqual(page.count('onclick="toggleWiFiPassword(this)"'), 2)
|
||||
self.assertIn("function toggleWiFiPassword(btn)", page)
|
||||
self.assertIn("function wzSkipRadio()", page)
|
||||
@@ -249,13 +249,13 @@ class WebConfigUiTest(unittest.TestCase):
|
||||
self.assertIn('case"bluetooth.mac":return r.bluetooth_mac||"";', page)
|
||||
self.assertIn("random-every-boot", page)
|
||||
|
||||
def test_webconfig_serves_the_single_brotli_asset(self):
|
||||
def test_webconfig_serves_the_single_gzip_asset(self):
|
||||
server = (ROOT / "src/helpers/esp32/WebConfigServer.cpp").read_text(
|
||||
encoding="utf-8"
|
||||
)
|
||||
self.assertIn("WEBCONFIG_HTML_BR", server)
|
||||
self.assertIn('res->addHeader("Content-Encoding", "br")', server)
|
||||
self.assertNotIn("WEBCONFIG_HTML_GZ", server)
|
||||
self.assertIn("WEBCONFIG_HTML_GZ", server)
|
||||
self.assertIn('res->addHeader("Content-Encoding", "gzip")', server)
|
||||
self.assertNotIn("WEBCONFIG_HTML_BR", server)
|
||||
|
||||
def test_skipped_mqtt_does_not_pin_display_in_setup_mode(self):
|
||||
main = (ROOT / "examples" / "companion_radio" / "main.cpp").read_text(
|
||||
|
||||
@@ -2,7 +2,7 @@
|
||||
|
||||
"""Optional real-browser checks for WebConfig's asynchronous setup races."""
|
||||
|
||||
import brotli
|
||||
import gzip
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
@@ -68,12 +68,12 @@ def run_browser(args):
|
||||
|
||||
def embedded_page():
|
||||
header = HEADER.read_text(encoding="utf-8")
|
||||
length = int(re.search(r"WEBCONFIG_HTML_BR_LEN = (\d+);", header).group(1))
|
||||
array = header.split("const uint8_t WEBCONFIG_HTML_BR[] PROGMEM = {", 1)[1]
|
||||
length = int(re.search(r"WEBCONFIG_HTML_GZ_LEN = (\d+);", header).group(1))
|
||||
array = header.split("const uint8_t WEBCONFIG_HTML_GZ[] PROGMEM = {", 1)[1]
|
||||
blob = bytes(
|
||||
int(value, 16) for value in re.findall(r"0x([0-9a-f]{2})", array)
|
||||
)[:length]
|
||||
return brotli.decompress(blob).decode("utf-8")
|
||||
return gzip.decompress(blob).decode("utf-8")
|
||||
|
||||
|
||||
class WebConfigUiRuntimeTest(unittest.TestCase):
|
||||
|
||||
Reference in new issue
Block a user