Publish immutable short firmware download links

Mirror pinned GitHub release bytes at shorter version-specific Pages URLs. Verify size, SHA-256 and firmware format before publishing, and preserve deployed alias identities across later site builds.

Add RAK4631 Full Companion 1.17.1.10 preview2 UF2 and native DFU ZIP aliases, a short-download directory, documentation and CI regressions. Existing flashing tools use the ZIP URL without losing board-family checks.

Validation: 16 short-link and Pages guard regressions, hosted-doc routing tests, firmware picker tests, full MkDocs build and exact UF2/ZIP byte checks.
This commit is contained in:
mikecarper committed 2026-10-12 21:47:08 -07:00
1 parent 015a3e8e05
commit 3e0ce84486
12 files changed
+755

No files matched your search

+8
View File
@@ -23,6 +23,12 @@ jobs:
with:
python-version: '3.13'
- name: Preserve published firmware short-link identities
run: |
python scripts/check_published_firmware_shortlinks.py \
--manifest docs/_data/firmware_shortlinks.json \
--previous "$RUNNER_TEMP/previous-firmware-shortlinks.json"
- name: Build
env:
GH_TOKEN: ${{ github.token }}
@@ -32,6 +38,8 @@ jobs:
MKDOCS_EDIT_URI: edit/${{ github.ref_name }}/docs/
run: |
pip install mkdocs-material
python -B test/test_firmware_shortlinks.py -v
python -B test/test_docs_firmware_shortlinks.py -v
node test/test_firmware_picker.js
node test/test_partition_catalog_viewer.js
python scripts/sync_bootloader_manifest.py
+5
View File
@@ -191,6 +191,11 @@ jobs:
- name: Verify build output cleanup safety
run: python3 -B test/test_build_output_safety.py -v
- name: Verify immutable short firmware downloads and Pages publishing
run: |
python3 -B test/test_firmware_shortlinks.py -v
python3 -B test/test_docs_firmware_shortlinks.py -v
- name: Verify serial mOTA response deadlines and RX fairness
run: |
python3 -B test/test_mota_serial_deadline.py -v
+1
View File
@@ -7,6 +7,7 @@
.vscode/ipch
out/
site/
.cache/firmware-shortlinks/
# Local firmware archives and build logs
.releases/
out.*/
+24
View File
@@ -0,0 +1,24 @@
{
"schemaVersion": 1,
"repository": "mikecarper/MeshCore",
"links": [
{
"filename": "rak4631-1.17.1.10-p2.uf2",
"version": "1.17.1.10",
"releaseTag": "rak4631-full-companion-v1.17.1.10-preview2",
"asset": "RAK_4631_companion_radio_full-v1.17.1.10-halo-keymind-cascade-preview2-015a3e8e.uf2",
"sha256": "b74243f9741133b0f0ae858fab02de9aaa999f84c474e0a659efce85fe35ad8c",
"size": 1408000,
"label": "RAK4631 Full Companion 1.17.1.10 preview2: UF2"
},
{
"filename": "rak4631-1.17.1.10-p2.zip",
"version": "1.17.1.10",
"releaseTag": "rak4631-full-companion-v1.17.1.10-preview2",
"asset": "RAK_4631_companion_radio_full-v1.17.1.10-halo-keymind-cascade-preview2-015a3e8e.zip",
"sha256": "11783d291f80c8084bb16f46ad2e35f0b1631da10606723488e2061320a5f6db",
"size": 704759,
"label": "RAK4631 Full Companion 1.17.1.10 preview2: Serial/Bluetooth DFU ZIP"
}
]
}
+35
View File
@@ -0,0 +1,35 @@
<!-- meshcore-hosted-doc-link:start -->
<p class="meshcore-hosted-doc-link"><a href="https://mikecarper.github.io/MeshCore/d/">View this page on MeshCore Docs</a>.</p>
<!-- meshcore-hosted-doc-link:end -->
# Short firmware links
These links download the same firmware bytes as the original GitHub release,
with shorter filenames. Each link is pinned to one version and verified against
its SHA-256 checksum when the site is built. Existing links keep their original
firmware; a new release gets a new link.
<!-- firmware-shortlinks:table -->
For **RAK4631 Full Companion**, use the **UF2** for a bootloader-drive
drag-and-drop update. Use the **DFU ZIP** when pasting a custom download URL
into the Windows or Linux MeshCore firmware tool. The ZIP is the native
application update package, not the release's download bundle. These files do
not install a bootloader or SoftDevice.
The 1.17.1.10 preview2 files are for MeshTalk testing with Keymind Cascade and
USA Cascade defaults. The original release includes instructions, qualification
reports and checksums. See the [firmware picker](firmware_picker.md) for other
board and release choices.
## Adding links for another release
Add an entry to `docs/_data/firmware_shortlinks.json` with a unique,
version-specific filename, exact release tag and asset name, byte size, label
and SHA-256 checksum from the validated release. Keep the original firmware
extension. Never change or remove a published filename's version, source or
checksum. Labels can be corrected without changing the download.
The documentation build downloads and verifies the pinned files before
publishing them under `d/`. A missing asset, incorrect checksum or changed
published link stops deployment instead of publishing a different image.
+4
View File
@@ -4,6 +4,10 @@
# Firmware picker
For shorter, version-pinned firmware download URLs, use
[Short firmware links](d.md). Those links serve the original verified firmware
files and include the RAK4631 Full Companion 1.17.1.10 preview2 for MeshTalk testing.
Each result now includes **Restore your settings after flashing**. Its commands
follow the selected logging mode and the exact image's verified role/hardware
controls. Click On, Off, or Check to view and copy the commands; Companion
+2
View File
@@ -8,6 +8,7 @@ edit_uri: !ENV [MKDOCS_EDIT_URI, 'edit/main/docs/']
hooks:
- scripts/docs_partition_catalog.py
- scripts/docs_firmware_shortlinks.py
# Keep historical URLs for bookmarks while showing only the organized copies.
not_in_nav: |
@@ -45,6 +46,7 @@ nav:
- Home: index.md
- Firmware and tools:
- firmware_picker.md
- Short firmware links: d.md
- notifications.md
- preset_test.md
- filter_tool.md
@@ -0,0 +1,54 @@
#!/usr/bin/env python3
"""Keep short download identities unchanged from the deployed Pages registry."""
import argparse
import os
from pathlib import Path
import subprocess
from firmware_shortlinks import load_manifest, validate_immutable
PUBLISHED_PATH = "_data/firmware_shortlinks.json"
def check_published(manifest, previous):
current = load_manifest(manifest)
previous = Path(previous)
previous.unlink(missing_ok=True)
found = subprocess.run(
["git", "ls-remote", "--exit-code", "--heads", "origin", "gh-pages"],
capture_output=True, text=True, check=False)
if found.returncode == 2:
return False # First Pages deployment: no published branch exists.
if found.returncode:
raise RuntimeError("Cannot check published Pages branch: " + found.stderr.strip())
subprocess.run(["git", "fetch", "--depth=1", "--no-tags", "origin", "gh-pages"], check=True)
names = subprocess.check_output(
["git", "ls-tree", "--name-only", "FETCH_HEAD", "--", PUBLISHED_PATH], text=True)
if not names.strip():
return False # Existing site predates short firmware links.
data = subprocess.check_output(["git", "show", "FETCH_HEAD:" + PUBLISHED_PATH])
previous.parent.mkdir(parents=True, exist_ok=True)
previous.write_bytes(data)
validate_immutable(load_manifest(previous), current)
return True
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--manifest", type=Path, required=True)
parser.add_argument("--previous", type=Path, required=True)
args = parser.parse_args()
try:
existing = check_published(args.manifest, args.previous)
except (OSError, ValueError, RuntimeError, subprocess.CalledProcessError) as error:
parser.exit(1, "Firmware short-link deployment rejected: " + str(error) + "\n")
if existing and os.environ.get("GITHUB_ENV"):
with open(os.environ["GITHUB_ENV"], "a", encoding="utf-8") as environment:
environment.write("MESHCORE_SHORTLINKS_PREVIOUS=" + str(args.previous.resolve()) + "\n")
print("Published firmware links preserved." if existing else "First firmware short-link deployment.")
if __name__ == "__main__":
main()
+50
View File
@@ -0,0 +1,50 @@
"""Publish version-pinned firmware downloads at short, real binary URLs."""
import html
import logging
import os
from pathlib import Path
import sys
sys.path.insert(0, str(Path(__file__).resolve().parent))
from firmware_shortlinks import download_url, load_manifest, mirror, validate_immutable
MARKER = "<!-- firmware-shortlinks:table -->"
def registry(config):
return load_manifest(Path(config["docs_dir"]) / "_data/firmware_shortlinks.json")
def on_page_markdown(markdown, page, config, **kwargs):
if MARKER not in markdown:
return markdown
data = registry(config)
base = config["site_url"].rstrip("/") + "/d/"
rows = ["<table><thead><tr><th>Firmware</th><th>Short download link</th>"
"<th>Original release</th></tr></thead><tbody>"]
for link in data["links"]:
label = html.escape(link["label"], quote=True)
alias = base + link["filename"]
original = download_url(data, link).split("/releases/download/", 1)[0]
original += "/releases/tag/" + link["releaseTag"]
rows.append(
"<tr><td>" + label + "</td><td><a href=\"" + html.escape(alias, quote=True)
+ "\">" + html.escape(link["filename"]) + "</a></td><td><a href=\""
+ html.escape(original, quote=True) + "\">Release</a></td></tr>")
rows.append("</tbody></table>")
return markdown.replace(MARKER, "\n".join(rows))
def on_post_build(config, **kwargs):
data = registry(config)
previous = os.environ.get("MESHCORE_SHORTLINKS_PREVIOUS")
if previous:
validate_immutable(load_manifest(Path(previous)), data)
cache = Path(os.environ.get(
"MESHCORE_SHORTLINKS_CACHE",
str(Path(config.config_file_path).resolve().parent / ".cache/firmware-shortlinks")))
outputs = mirror(data, Path(config["site_dir"]) / "d", cache)
logging.getLogger("mkdocs.plugins.firmware_shortlinks").info(
"Published %d checksum-verified short firmware downloads", len(outputs))
+223
View File
@@ -0,0 +1,223 @@
#!/usr/bin/env python3
"""Mirror pinned release firmware as immutable, version-specific short URLs."""
import argparse
import hashlib
import json
import os
from pathlib import Path
import re
import shutil
import struct
import tempfile
from urllib.parse import quote, urlsplit
from urllib.request import build_opener, HTTPRedirectHandler
REPOSITORY = "mikecarper/MeshCore"
EXTENSIONS = {".bin", ".uf2", ".zip", ".hex", ".mota"}
MAX_ASSET_SIZE = 32 * 1024 * 1024
IMMUTABLE_FIELDS = ("version", "releaseTag", "asset", "sha256", "size")
DELIVERY_HOSTS = {"github.com", "release-assets.githubusercontent.com", "objects.githubusercontent.com"}
def _version_in(value, version):
return re.search(r"(?<![0-9.])" + re.escape(version) + r"(?![0-9]|\.[0-9])", value) is not None
def _validate_manifest(manifest):
if (not isinstance(manifest, dict) or type(manifest.get("schemaVersion")) is not int
or manifest["schemaVersion"] != 1 or manifest.get("repository") != REPOSITORY):
raise ValueError("expected schemaVersion 1 and repository " + REPOSITORY)
links = manifest.get("links")
if not isinstance(links, list) or not links or len(links) > 128:
raise ValueError("links must contain 1..128 firmware aliases")
names = set()
for link in links:
if not isinstance(link, dict):
raise ValueError("firmware link must be an object")
filename, version = link.get("filename"), link.get("version")
if (not isinstance(filename, str)
or re.fullmatch(r"[a-z0-9][a-z0-9._-]{0,159}", filename) is None
or "latest" in filename):
raise ValueError("alias must be a safe flat filename without latest")
if (not isinstance(version, str)
or re.fullmatch(r"[0-9]+\.[0-9]+\.[0-9]+(?:\.[0-9]+)?", version) is None
or not _version_in(filename, version)):
raise ValueError(filename + ": alias must contain its exact numeric version")
asset, tag = link.get("asset"), link.get("releaseTag")
for field, value in (("asset", asset), ("releaseTag", tag)):
if (not isinstance(value, str)
or re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._-]{0,239}", value) is None
or not _version_in(value, version) or "latest" in value.lower()):
raise ValueError(filename + ": invalid version-specific " + field)
extension = Path(filename).suffix
if extension not in EXTENSIONS or Path(asset).suffix != extension:
raise ValueError(filename + ": alias and source need matching firmware extensions")
if (not isinstance(link.get("sha256"), str)
or re.fullmatch(r"[0-9a-f]{64}", link["sha256"]) is None):
raise ValueError(filename + ": invalid SHA-256")
if type(link.get("size")) is not int or not 1 <= link["size"] <= MAX_ASSET_SIZE:
raise ValueError(filename + ": size must be 1..32 MiB")
label = link.get("label")
if (not isinstance(label, str) or not label.strip() or len(label) > 160
or any(ord(char) < 32 or ord(char) > 126 for char in label)):
raise ValueError(filename + ": label must be printable ASCII")
if filename in names:
raise ValueError("duplicate firmware alias: " + filename)
names.add(filename)
return manifest
def load_manifest(path):
path = Path(path)
if path.stat().st_size > 1024 * 1024:
raise ValueError("firmware registry exceeds 1 MiB")
return _validate_manifest(json.loads(path.read_text(encoding="utf-8")))
def validate_immutable(previous, current):
"""Keep published aliases present and pinned; display labels may change."""
_validate_manifest(current)
if previous is None:
return
_validate_manifest(previous)
by_name = {link["filename"]: link for link in current["links"]}
for old in previous["links"]:
new = by_name.get(old["filename"])
if new is None or any(new[field] != old[field] for field in IMMUTABLE_FIELDS):
raise ValueError("published firmware alias removed or retargeted: " + old["filename"])
def download_url(manifest, link):
_validate_manifest(manifest)
if link not in manifest["links"]:
raise ValueError("link is not in the firmware registry")
return ("https://github.com/" + REPOSITORY + "/releases/download/"
+ quote(link["releaseTag"], safe="") + "/" + quote(link["asset"], safe=""))
def _delivery_url_allowed(url):
parsed = urlsplit(url)
return (parsed.scheme == "https" and parsed.hostname in DELIVERY_HOSTS
and parsed.port in (None, 443) and not parsed.username and not parsed.password)
class _ReleaseRedirect(HTTPRedirectHandler):
def redirect_request(self, request, response, code, message, headers, url):
if not _delivery_url_allowed(url):
raise ValueError("release download redirected outside GitHub HTTPS delivery")
return super().redirect_request(request, response, code, message, headers, url)
def _open_release(url):
return build_opener(_ReleaseRedirect()).open(url, timeout=30)
def _check_content(head, extension):
text = head.lstrip(b"\xef\xbb\xbf \t\r\n").lower()
if text.startswith((b"<!doctype html", b"<html", b"<script", b"window.location", b"location.href")):
raise ValueError("firmware download contains an HTML or JavaScript redirect")
if extension in {".bin", ".mota"} and head and all(
byte in b"\t\r\n" or 32 <= byte <= 126 for byte in head):
raise ValueError("binary firmware download contains only text")
if extension == ".uf2" and head[:8] != struct.pack("<II", 0x0A324655, 0x9E5D5157):
raise ValueError("firmware download lacks a UF2 header")
if extension == ".zip" and not head.startswith(b"PK\x03\x04"):
raise ValueError("firmware download lacks an application ZIP header")
if extension == ".hex" and not text.startswith(b":"):
raise ValueError("firmware download lacks an Intel HEX header")
def _verify_file(path, link):
if path.is_symlink() or not path.is_file() or path.stat().st_size != link["size"]:
raise ValueError("firmware size or file type mismatch: " + str(path))
digest = hashlib.sha256()
with path.open("rb") as source:
_check_content(source.read(1024), Path(link["filename"]).suffix)
source.seek(0)
for chunk in iter(lambda: source.read(65536), b""):
digest.update(chunk)
if digest.hexdigest() != link["sha256"]:
raise ValueError("firmware SHA-256 mismatch: " + str(path))
def _download(url, link, destination, opener):
with opener(url) as response, destination.open("wb") as output:
if hasattr(response, "geturl") and not _delivery_url_allowed(response.geturl()):
raise ValueError("release response is outside GitHub HTTPS delivery")
content_type = getattr(response, "headers", {}).get("Content-Type", "").lower()
if any(kind in content_type for kind in ("html", "javascript")):
raise ValueError("release response is HTML or JavaScript")
total = 0
while True:
chunk = response.read(min(65536, link["size"] - total + 1))
if not chunk:
break
total += len(chunk)
if total > link["size"]:
raise ValueError("release download exceeds its pinned size")
output.write(chunk)
output.flush()
os.fsync(output.fileno())
_verify_file(destination, link)
def mirror(manifest, output_dir, cache_dir, opener=None):
"""Validate cache and existing aliases before publishing real firmware files."""
_validate_manifest(manifest)
output_dir, cache_dir = Path(output_dir), Path(cache_dir)
output_dir.mkdir(parents=True, exist_ok=True)
cache_dir.mkdir(parents=True, exist_ok=True)
for link in manifest["links"]:
output = output_dir / link["filename"]
if output.exists() or output.is_symlink():
_verify_file(output, link) # An alias is immutable even across builds.
outputs = []
for link in manifest["links"]:
output = output_dir / link["filename"]
cached = cache_dir / (link["sha256"] + Path(link["filename"]).suffix)
try:
_verify_file(cached, link)
except ValueError:
with tempfile.TemporaryDirectory(prefix=".firmware-", dir=cache_dir) as temporary:
candidate = Path(temporary) / "download"
_download(download_url(manifest, link), link, candidate, opener or _open_release)
os.replace(candidate, cached)
if not output.exists():
with tempfile.TemporaryDirectory(prefix=".firmware-", dir=output_dir) as temporary:
candidate = Path(temporary) / "publish"
shutil.copyfile(cached, candidate)
_verify_file(candidate, link)
try:
os.link(candidate, output) # Atomic publication without overwriting a racing alias.
except FileExistsError:
_verify_file(output, link)
_verify_file(output, link)
outputs.append(output)
return outputs
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--manifest", required=True, type=Path)
parser.add_argument("--previous", type=Path)
parser.add_argument("--output", type=Path)
parser.add_argument("--cache", type=Path)
parser.add_argument("--check-only", action="store_true")
args = parser.parse_args()
try:
manifest = load_manifest(args.manifest)
previous = load_manifest(args.previous) if args.previous is not None else None
validate_immutable(previous, manifest)
if not args.check_only:
if args.output is None or args.cache is None:
parser.error("--output and --cache are required unless --check-only")
mirror(manifest, args.output, args.cache)
print("Firmware shortlinks validated: " + str(len(manifest["links"])))
except (OSError, ValueError, KeyError) as error:
parser.exit(1, "Firmware shortlink validation failed: " + str(error) + "\n")
if __name__ == "__main__":
main()
+121
View File
@@ -0,0 +1,121 @@
#!/usr/bin/env python3
"""Verify rendered download URLs and the deployed-link identity guard."""
import copy
import importlib.util
import json
from pathlib import Path
import subprocess
import sys
import tempfile
import unittest
from unittest import mock
ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "scripts"))
from firmware_shortlinks import load_manifest
def module(name, filename):
spec = importlib.util.spec_from_file_location(name, ROOT / "scripts" / filename)
loaded = importlib.util.module_from_spec(spec)
spec.loader.exec_module(loaded)
return loaded
HOOK = module("docs_shortlinks_hook", "docs_firmware_shortlinks.py")
GUARD = module("published_shortlinks_guard", "check_published_firmware_shortlinks.py")
MANIFEST = ROOT / "docs/_data/firmware_shortlinks.json"
class DocsFirmwareShortlinksTests(unittest.TestCase):
def test_table_uses_site_base_and_versioned_real_file_urls(self):
config = {"docs_dir": str(ROOT / "docs"), "site_url": "https://example.test/fork/"}
data = load_manifest(MANIFEST)
rendered = HOOK.on_page_markdown("Before\n\n" + HOOK.MARKER + "\n\nAfter", None, config)
self.assertNotIn(HOOK.MARKER, rendered)
for entry in data["links"]:
self.assertIn('href="https://example.test/fork/d/' + entry["filename"] + '"', rendered)
self.assertIn("/releases/tag/" + entry["releaseTag"], rendered)
self.assertTrue(rendered.startswith("Before"))
self.assertTrue(rendered.endswith("After"))
self.assertEqual(HOOK.on_page_markdown("Unrelated page", None, config), "Unrelated page")
def test_registry_labels_are_plain_text_not_html_or_markdown(self):
data = load_manifest(MANIFEST)
data["links"][0]["label"] = '<script>alert("bad")</script>|[bad](javascript:alert(1))'
with mock.patch.object(HOOK, "registry", return_value=data):
result = HOOK.on_page_markdown(HOOK.MARKER, None, {"site_url": "https://example.test/"})
self.assertNotIn("<script>", result)
self.assertNotIn('href="javascript:', result)
self.assertIn("&lt;script&gt;", result)
def test_pages_hook_mirrors_real_files_and_checks_previous_before_publication(self):
data = load_manifest(MANIFEST)
with tempfile.TemporaryDirectory() as directory:
previous = Path(directory) / "previous.json"
previous.write_text(json.dumps(data), encoding="ascii")
class Config(dict):
config_file_path = ROOT / "mkdocs.yml"
config = Config(docs_dir=ROOT / "docs", site_dir=Path(directory) / "site")
environment = {"MESHCORE_SHORTLINKS_PREVIOUS": str(previous),
"MESHCORE_SHORTLINKS_CACHE": str(Path(directory) / "cache")}
with mock.patch.dict(HOOK.os.environ, environment), mock.patch.object(HOOK, "mirror", return_value=[]) as mirror:
HOOK.on_post_build(config)
self.assertEqual(mirror.call_args.args,
(data, Path(directory) / "site/d", Path(directory) / "cache"))
changed = copy.deepcopy(data)
changed["links"][0]["sha256"] = "0" * 64
previous.write_text(json.dumps(changed), encoding="ascii")
with mock.patch.dict(HOOK.os.environ, environment), mock.patch.object(HOOK, "mirror") as mirror:
with self.assertRaises(ValueError):
HOOK.on_post_build(config)
mirror.assert_not_called()
def test_first_pages_deployment_does_not_reuse_an_old_previous_file(self):
with tempfile.TemporaryDirectory() as directory:
previous = Path(directory) / "previous.json"
previous.write_text("stale", encoding="ascii")
result = subprocess.CompletedProcess([], 2, "", "")
with mock.patch.object(GUARD.subprocess, "run", return_value=result):
self.assertFalse(GUARD.check_published(MANIFEST, previous))
self.assertFalse(previous.exists())
def test_existing_site_without_registry_is_a_first_shortlink_deployment(self):
with tempfile.TemporaryDirectory() as directory:
previous = Path(directory) / "previous.json"
with mock.patch.object(GUARD.subprocess, "run", return_value=subprocess.CompletedProcess([], 0, "sha", "")) as run, \
mock.patch.object(GUARD.subprocess, "check_output", return_value=""):
self.assertFalse(GUARD.check_published(MANIFEST, previous))
self.assertEqual(run.call_args_list[1].args[0],
["git", "fetch", "--depth=1", "--no-tags", "origin", "gh-pages"])
self.assertFalse(previous.exists())
def test_published_alias_retarget_and_removal_fail_the_deployment_guard(self):
original = load_manifest(MANIFEST)
for changed in (False, True):
with self.subTest(retarget=changed), tempfile.TemporaryDirectory() as directory:
current = copy.deepcopy(original)
if changed:
current["links"][0]["sha256"] = "0" * 64
else:
current["links"].pop()
manifest = Path(directory) / "new.json"
manifest.write_text(json.dumps(current), encoding="ascii")
previous = Path(directory) / "previous.json"
replies = [GUARD.PUBLISHED_PATH + "\n", json.dumps(original).encode("ascii")]
with mock.patch.object(GUARD.subprocess, "run", return_value=subprocess.CompletedProcess([], 0, "sha", "")), \
mock.patch.object(GUARD.subprocess, "check_output", side_effect=replies):
with self.assertRaises(ValueError):
GUARD.check_published(manifest, previous)
def test_network_failure_is_not_treated_as_an_empty_published_registry(self):
with tempfile.TemporaryDirectory() as directory:
result = subprocess.CompletedProcess([], 128, "", "Network unavailable")
with mock.patch.object(GUARD.subprocess, "run", return_value=result):
with self.assertRaisesRegex(RuntimeError, "Cannot check"):
GUARD.check_published(MANIFEST, Path(directory) / "previous.json")
if __name__ == "__main__":
unittest.main()
+228
View File
@@ -0,0 +1,228 @@
"""Keep short firmware URLs pinned, binary, bounded and safe to mirror."""
import copy
import hashlib
import importlib.util
import io
import json
from pathlib import Path
import struct
import subprocess
import sys
import tempfile
import unittest
from urllib.request import Request
ROOT = Path(__file__).resolve().parents[1]
SCRIPT = ROOT / "scripts/firmware_shortlinks.py"
SPEC = importlib.util.spec_from_file_location("firmware_shortlinks", SCRIPT)
SHORTLINKS = importlib.util.module_from_spec(SPEC)
SPEC.loader.exec_module(SHORTLINKS)
UF2 = struct.pack("<II", 0x0A324655, 0x9E5D5157) + bytes(504)
ZIP = b"PK\x03\x04" + b"application DFU fixture" * 7
TAG = "rak4631-full-companion-v1.17.1.10-preview2"
def registry(data=UF2, extension="uf2"):
return {"schemaVersion": 1, "repository": "mikecarper/MeshCore", "links": [{
"filename": "rak4631-1.17.1.10-p2." + extension,
"version": "1.17.1.10", "releaseTag": TAG,
"asset": "RAK_4631_companion_radio_full-v1.17.1.10-halo-keymind-cascade-preview2-015a3e8e." + extension,
"sha256": hashlib.sha256(data).hexdigest(), "size": len(data),
"label": "RAK4631 Full Companion preview2"}]}
class Response(io.BytesIO):
def __init__(self, data, content_type="application/octet-stream", url=None):
super().__init__(data)
self.headers = {"Content-Type": content_type}
self.url = url or "https://release-assets.githubusercontent.com/example"
def read(self, size=-1):
return super().read(min(size, 17) if size >= 0 else 17)
def geturl(self):
return self.url
class FirmwareShortlinkTests(unittest.TestCase):
def load(self, manifest):
with tempfile.TemporaryDirectory() as directory:
path = Path(directory) / "registry.json"
path.write_text(json.dumps(manifest), encoding="ascii")
return SHORTLINKS.load_manifest(path)
def test_real_preview_url_is_derived_from_fixed_repository(self):
manifest = self.load(registry())
self.assertEqual(SHORTLINKS.download_url(manifest, manifest["links"][0]),
"https://github.com/mikecarper/MeshCore/releases/download/" + TAG + "/"
"RAK_4631_companion_radio_full-v1.17.1.10-halo-keymind-cascade-preview2-015a3e8e.uf2")
foreign = dict(manifest["links"][0], asset="other-v1.17.1.10.uf2")
with self.assertRaisesRegex(ValueError, "not in"):
SHORTLINKS.download_url(manifest, foreign)
def test_registry_rejects_paths_mutable_names_wrong_sources_and_bad_types(self):
changes = [
("filename", "../rak4631-1.17.1.10.uf2"),
("filename", "directory/rak4631-1.17.1.10.uf2"),
("filename", "https://example.com/1.17.1.10.uf2"),
("filename", "rak4631-latest-1.17.1.10.uf2"),
("filename", "rak4631-1.17.1.100-p2.uf2"),
("filename", "rak4631-1.17.1.10-p2.html"),
("asset", "other-v1.17.1.10.zip"),
("asset", "https://example.com/v1.17.1.10.uf2"),
("asset", "firmware-v1.17.1.10.uf2?download=1"),
("releaseTag", "latest-v1.17.1.10"),
("releaseTag", "v1.17.1.9-preview2"),
("version", "latest"), ("version", "v1.17.1.10"),
("sha256", "a" * 63), ("sha256", "z" * 64),
("sha256", int("1" * 64)),
("size", 0), ("size", True), ("size", 32 * 1024 * 1024 + 1),
("label", "bad\nlabel"), ("label", "")]
for field, value in changes:
with self.subTest(field=field, value=value):
manifest = registry()
manifest["links"][0][field] = value
with self.assertRaises(ValueError):
self.load(manifest)
for field, value in (("repository", "attacker/MeshCore"), ("schemaVersion", True),
("schemaVersion", 2), ("links", [])):
manifest = registry()
manifest[field] = value
with self.subTest(field=field), self.assertRaises(ValueError):
self.load(manifest)
manifest = registry()
manifest["links"].append(dict(manifest["links"][0]))
with self.assertRaisesRegex(ValueError, "duplicate"):
self.load(manifest)
def test_published_aliases_cannot_disappear_or_change_but_labels_can(self):
previous = registry()
current = copy.deepcopy(previous)
current["links"][0]["label"] = "Improved description"
current["links"].extend(registry(ZIP, "zip")["links"])
SHORTLINKS.validate_immutable(previous, current)
SHORTLINKS.validate_immutable(None, current)
for field, value in (("releaseTag", TAG + "-changed"),
("asset", "different-v1.17.1.10.uf2"),
("sha256", "a" * 64), ("size", 1024)):
modified = copy.deepcopy(current)
modified["links"][0][field] = value
with self.subTest(field=field), self.assertRaisesRegex(ValueError, "retargeted"):
SHORTLINKS.validate_immutable(previous, modified)
removed = registry(ZIP, "zip")
with self.assertRaisesRegex(ValueError, "removed"):
SHORTLINKS.validate_immutable(previous, removed)
def test_mirror_streams_real_bytes_and_reuses_only_validated_cache(self):
manifest = registry()
manifest["links"].extend(registry(ZIP, "zip")["links"])
calls = []
def opener(url):
calls.append(url)
return Response(UF2 if url.endswith(".uf2") else ZIP)
with tempfile.TemporaryDirectory() as directory:
output, cache = Path(directory) / "out", Path(directory) / "cache"
published = SHORTLINKS.mirror(manifest, output, cache, opener)
self.assertEqual([path.read_bytes() for path in published], [UF2, ZIP])
self.assertEqual(len(calls), 2)
self.assertTrue(all(url.startswith("https://github.com/mikecarper/MeshCore/releases/download/")
for url in calls))
SHORTLINKS.mirror(manifest, output, cache, opener)
self.assertEqual(len(calls), 2)
# Corruption with the same size must not bypass the cache check.
cached = cache / (manifest["links"][0]["sha256"] + ".uf2")
cached.write_bytes(UF2[:20] + b"!" + UF2[21:])
published[0].unlink()
SHORTLINKS.mirror(manifest, output, cache, opener)
self.assertEqual(len(calls), 3)
self.assertEqual(published[0].read_bytes(), UF2)
self.assertEqual(cached.read_bytes(), UF2)
self.assertFalse(list(output.glob(".firmware-*")))
self.assertFalse(list(cache.glob(".firmware-*")))
def test_existing_alias_conflicts_fail_before_any_download(self):
manifest = registry()
with tempfile.TemporaryDirectory() as directory:
output, cache = Path(directory) / "out", Path(directory) / "cache"
output.mkdir()
alias = output / manifest["links"][0]["filename"]
conflict = UF2[:20] + b"!" + UF2[21:]
alias.write_bytes(conflict)
def opener(_url):
self.fail("existing alias conflicts must fail before network work")
with self.assertRaisesRegex(ValueError, "SHA-256"):
SHORTLINKS.mirror(manifest, output, cache, opener)
self.assertEqual(alias.read_bytes(), conflict)
def test_alias_created_during_download_is_not_overwritten(self):
manifest = registry()
with tempfile.TemporaryDirectory() as directory:
output, cache = Path(directory) / "out", Path(directory) / "cache"
alias = output / manifest["links"][0]["filename"]
def opener(_url):
alias.write_bytes(b"existing unrelated file")
return Response(UF2)
with self.assertRaises(ValueError):
SHORTLINKS.mirror(manifest, output, cache, opener)
self.assertEqual(alias.read_bytes(), b"existing unrelated file")
def test_bad_network_data_never_publishes_or_poison_cache(self):
cases = ((registry(), UF2[:-1], "application/octet-stream", None),
(registry(), UF2 + b"excess", "application/octet-stream", None),
(registry(), UF2[:20] + b"!" + UF2[21:], "application/octet-stream", None),
(registry(), UF2, "text/html", None),
(registry(), UF2, "application/javascript", None),
(registry(), UF2, "application/octet-stream", "http://release-assets.githubusercontent.com/example"),
(registry(), UF2, "application/octet-stream", "https://attacker.example/download"),
(registry(b"<!DOCTYPE html><html>redirect</html>", "bin"),
b"<!DOCTYPE html><html>redirect</html>", "application/octet-stream", None),
(registry(b"window.location = 'https://example.com';", "bin"),
b"window.location = 'https://example.com';", "application/octet-stream", None),
(registry(b"not firmware", "uf2"), b"not firmware", "application/octet-stream", None))
for manifest, data, content_type, url in cases:
with self.subTest(content_type=content_type, url=url, data=data[:20]):
with tempfile.TemporaryDirectory() as directory:
output, cache = Path(directory) / "out", Path(directory) / "cache"
with self.assertRaises(ValueError):
SHORTLINKS.mirror(manifest, output, cache,
lambda _url: Response(data, content_type, url))
self.assertFalse(list(output.iterdir()))
self.assertFalse(list(cache.iterdir()))
def test_redirects_allow_only_github_https_asset_delivery(self):
handler = SHORTLINKS._ReleaseRedirect()
request = Request("https://github.com/mikecarper/MeshCore/releases/download/" + TAG + "/app.uf2")
for url in ("http://github.com/app.uf2", "https://github.com.attacker.example/app.uf2",
"https://attacker.example/app.uf2", "https://user:secret@github.com/app.uf2"):
with self.subTest(url=url), self.assertRaises(ValueError):
handler.redirect_request(request, None, 302, "Found", {}, url)
allowed = handler.redirect_request(request, None, 302, "Found", {},
"https://release-assets.githubusercontent.com/example?token=signed")
self.assertEqual(allowed.full_url,
"https://release-assets.githubusercontent.com/example?token=signed")
def test_cli_check_only_runs_without_network_and_enforces_previous(self):
with tempfile.TemporaryDirectory() as directory:
current, previous = Path(directory) / "current.json", Path(directory) / "previous.json"
manifest = registry()
current.write_text(json.dumps(manifest), encoding="ascii")
previous.write_text(json.dumps(manifest), encoding="ascii")
command = [sys.executable, str(SCRIPT), "--manifest", str(current), "--check-only"]
result = subprocess.run(command, capture_output=True, text=True, timeout=10)
self.assertEqual(result.returncode, 0, result.stderr)
result = subprocess.run(command + ["--previous", str(previous)],
capture_output=True, text=True, timeout=10)
self.assertEqual(result.returncode, 0, result.stderr)
manifest["links"][0]["sha256"] = "a" * 64
current.write_text(json.dumps(manifest), encoding="ascii")
result = subprocess.run(command + ["--previous", str(previous)],
capture_output=True, text=True, timeout=10)
self.assertNotEqual(result.returncode, 0)
self.assertIn("retargeted", result.stderr)
if __name__ == "__main__":
unittest.main()