Shorten firmware picker selection links and keep original download names

Replace the download-file shortcut approach with compact self-contained picker URLs under /p/. Preserve long-link compatibility, selected install formats, and full published firmware filenames. Cover selection round trips and the redirect in the picker tests.
This commit is contained in:
mikecarper committed 2026-10-12 22:02:20 -07:00
1 parent 3e0ce84486
commit a31b19ed8a
16 files changed
+269 -773

No files matched your search

-8
View File
@@ -23,12 +23,6 @@ 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 }}
@@ -38,8 +32,6 @@ 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,11 +191,6 @@ 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,7 +7,6 @@
.vscode/ipch
out/
site/
.cache/firmware-shortlinks/
# Local firmware archives and build logs
.releases/
out.*/
-24
View File
@@ -1,24 +0,0 @@
{
"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"
}
]
}
+69 -7
View File
@@ -23,6 +23,11 @@
]);
const FACET_FIELDS = Object.freeze(FILTER_FIELDS.concat(["install"]));
// Versioned picker selections stay self-contained as release catalogs change.
const SHORT_SELECTION_FIELDS = Object.freeze({
c: "chipFamily", g: "hardwareFamily", h: "hardware", r: "role",
l: "logging", o: "ota", m: "mode", f: "feature", v: "variant", i: "install",
});
const PROFILE_FIELDS = Object.freeze(["ota", "feature", "variant"]);
const CHOICE_FIELDS = Object.freeze([
"chipFamily", "hardwareFamily", "hardware", "role", "logging",
@@ -1207,6 +1212,7 @@
function selectionUrl(url, filters, automaticChipFamily) {
const result = new URL(url);
result.searchParams.delete("s");
FACET_FIELDS.forEach(function (field) {
result.searchParams.delete(field);
if (filters[field]) result.searchParams.set(field, filters[field]);
@@ -1218,8 +1224,61 @@
return result.href;
}
function shortSelectionUrl(url, filters, automaticChipFamily) {
const result = new URL(url);
FACET_FIELDS.concat(["chipAuto", "s"]).forEach(function (field) {
result.searchParams.delete(field);
});
const parts = ["1"];
Object.keys(SHORT_SELECTION_FIELDS).forEach(function (key) {
const field = SHORT_SELECTION_FIELDS[key];
const value = filters[field];
if (!value || (field === "chipFamily" && automaticChipFamily) ||
(field === "hardwareFamily" && value === filters.hardware)) return;
// Dots delimit fields; escape them even though encodeURIComponent does not.
parts.push(key + encodeURIComponent(value).replace(/\./g, "%2E"));
});
if ((filters.hardwareFamily || filters.hardware) && !automaticChipFamily) {
parts.push("a0");
}
if (parts.length > 1) result.searchParams.set("s", parts.join("."));
return result.href;
}
function shortSelectionParams(params) {
const parts = params.get("s").split(".");
const result = new URLSearchParams();
if (params.getAll("s").length !== 1 || parts.shift() !== "1" || !parts.length) {
throw new Error("unsupported or incomplete short picker link");
}
const seen = new Set();
parts.forEach(function (part) {
const key = part.charAt(0);
if (seen.has(key) || (key !== "a" &&
!Object.prototype.hasOwnProperty.call(SHORT_SELECTION_FIELDS, key))) {
throw new Error("duplicate or unknown short picker field");
}
seen.add(key);
const value = decodeURIComponent(part.slice(1));
if (!value || (key === "a" && value !== "0" && value !== "1")) {
throw new Error("invalid short picker field");
}
result.set(key === "a" ? "chipAuto" : SHORT_SELECTION_FIELDS[key], value);
});
return result;
}
function selectionFromUrl(url, profiles) {
const params = new URL(url).searchParams;
let params = new URL(url).searchParams;
if (params.has("s")) {
try {
// The versioned selection is authoritative over any legacy fields.
params = shortSelectionParams(params);
} catch (error) {
return { filters: {}, automaticChipFamily: false,
unavailable: ["short link: " + error.message] };
}
}
const requested = {};
const filters = {};
const unavailable = [];
@@ -1952,7 +2011,7 @@
}
function updateSelectionUrl() {
const url = selectionUrl(global.location.href, filters, automaticChipFamily);
const url = shortSelectionUrl(global.location.href, filters, automaticChipFamily);
if (url !== global.location.href) {
try {
global.history.replaceState(global.history.state, "", url);
@@ -1962,9 +2021,11 @@
}
}
if (shareLink) {
const base = global.location.protocol === "file:"
? root.getAttribute("data-share-url") : url;
shareLink.href = selectionUrl(base || url, filters, automaticChipFamily);
const base = new URL(global.location.protocol === "file:"
? root.getAttribute("data-share-url") || url : url);
base.pathname = base.pathname.replace(/\/firmware_picker\/$/, "/p/");
if (global.location.protocol === "file:") base.hash = global.location.hash;
shareLink.href = shortSelectionUrl(base.href, filters, automaticChipFamily);
shareLink.hidden = false;
}
if (copyLinkButton) copyLinkButton.disabled = false;
@@ -1987,13 +2048,13 @@
if (copyLinkButton && shareLink) copyLinkButton.addEventListener("click", function () {
const copy = global.navigator.clipboard && global.navigator.clipboard.writeText;
if (!copy) {
linkStatus.textContent = "Copy the address bar, or copy the 'Link to these settings' link.";
linkStatus.textContent = "Copy the address bar, or copy the 'Short link to these settings' link.";
return;
}
global.navigator.clipboard.writeText(shareLink.href).then(function () {
linkStatus.textContent = "Link copied.";
}).catch(function () {
linkStatus.textContent = "Copy the address bar, or copy the 'Link to these settings' link.";
linkStatus.textContent = "Copy the address bar, or copy the 'Short link to these settings' link.";
});
});
global.addEventListener("popstate", restoreSelectionUrl);
@@ -2513,6 +2574,7 @@
choicesUseSameFirmware: choicesUseSameFirmware,
selectionRequirements: selectionRequirements,
selectionUrl: selectionUrl,
shortSelectionUrl: shortSelectionUrl,
selectionFromUrl: selectionFromUrl,
canonicalAsset: canonicalAsset,
resolveProfileAssets: resolveProfileAssets,
-35
View File
@@ -1,35 +0,0 @@
<!-- 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.
+17 -14
View File
@@ -4,10 +4,6 @@
# 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
@@ -45,7 +41,7 @@ The picker reads public release metadata from GitHub. It does not upload device
information. Hardware names, target names, and download links come directly
from the published firmware assets.
<div class="firmware-picker" data-firmware-picker data-release-repo="mikecarper/MeshCore" data-controls-url="../_data/firmware_controls.json?v=1.17.1.9-88c85108" data-bootloaders-url="../_data/bootloader_manifest.json?v=20261002-1" data-share-url="https://mikecarper.github.io/MeshCore/firmware_picker/">
<div class="firmware-picker" data-firmware-picker data-release-repo="mikecarper/MeshCore" data-controls-url="../_data/firmware_controls.json?v=1.17.1.9-88c85108" data-bootloaders-url="../_data/bootloader_manifest.json?v=20261002-1" data-share-url="https://mikecarper.github.io/MeshCore/p/">
<div class="firmware-picker-intro" role="note">
<strong>Current release set</strong>
<p data-role="release-set">Loading release information...</p>
@@ -74,7 +70,7 @@ from the published firmware assets.
<p class="firmware-picker-order-note">
Pick in any order. Use <strong>Any</strong> to clear one choice, or clear
everything with the button below. The address bar updates with your choices;
copy its URL to reopen or share the same selection.
use <strong>Copy link to settings</strong> for a shorter link to the same selection.
</p>
<form class="firmware-picker-form" data-role="form">
@@ -150,7 +146,7 @@ from the published firmware assets.
<div class="firmware-picker-form-actions firmware-picker-wide">
<button type="reset" data-action="clear" disabled>Clear all choices</button>
<button type="button" data-action="copy-link" disabled>Copy link to settings</button>
<a data-role="share-link" hidden>Link to these settings</a>
<a data-role="share-link" hidden>Short link to these settings</a>
<p data-role="link-status" aria-live="polite"></p>
</div>
</form>
@@ -253,31 +249,38 @@ changing roles or partition layouts.
## Share a selection
Select any combination of choices, then copy the address bar or use **Copy link
to settings**. Opening the link restores those choices after the release
Select any combination of choices, then use **Copy link to settings** for a
short URL under `https://mikecarper.github.io/MeshCore/p/`. Opening the link
restores those choices after the release
catalog loads. Partial selections work too, so you can share a board and role
while leaving other choices open. **Clear all choices** removes the picker
parameters from the URL. Changes update the current browser-history entry
without reloading the page or adding a Back-button entry for every click.
For example, [RAK3401 unified Full repeater](?hardware=RAK_3401&role=repeater&ota=lora-receiver&feature=full&variant=default)
For example, [RAK3401 unified Full repeater](https://mikecarper.github.io/MeshCore/p/?s=1.hRAK_3401.rrepeater.olora-receiver.ffull.vdefault)
preselects that board, role, OTA capability, and Full sensor policy. Supported
internal or external storage is detected by the unified image.
The query parameters are `chipFamily`, `hardwareFamily`, `hardware`, `role`,
Short links encode only the selected choices in a versioned `s` parameter.
Hardware identifiers stay readable, and the link omits redundant hardware-family
and automatically selected chip-family values. It keeps profile and install
choices explicitly rather than relying on the current catalog's defaults.
The address bar also uses the compact selection format; copying it works too.
Existing long links still work. Their query parameters are `chipFamily`, `hardwareFamily`, `hardware`, `role`,
`logging`, `ota`, `mode`, `feature`, `variant`, and `install`. Values use the
picker's internal identifiers rather than the displayed labels. One Firmware
profile button sets `ota`, `feature`, and `variant` together. Existing links
with only some of those choices still work and show their partial selection.
The generated
link also records `chipAuto` so automatic chip-family selection or an explicit
The selection also records when chip-family filtering was chosen manually so
automatic chip-family selection or an explicit
**Any** choice behaves the same after reopening. Existing section anchors and
unrelated query parameters are preserved.
Links use the current release catalog. If a linked choice is no longer available
or conflicts with another choice, the picker identifies it and asks you to
review the remaining selections. The downloadable HTML's **Copy link to
settings** button creates a public website link that other people can open.
settings** button creates a short public website link that other people can open.
## FULL versus standard
+22
View File
@@ -0,0 +1,22 @@
<!doctype html>
<html lang="en">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width,initial-scale=1">
<title>MeshCore firmware picker</title>
<link rel="canonical" href="../firmware_picker/">
</head>
<body>
<p>Opening the firmware picker. <a id="picker-link" href="../firmware_picker/">Continue to the picker</a>.</p>
<script>
(function () {
"use strict";
const target = new URL("../firmware_picker/", window.location.href);
target.search = window.location.search;
target.hash = window.location.hash;
document.getElementById("picker-link").href = target.href;
window.location.replace(target.href);
})();
</script>
</body>
</html>
+1 -3
View File
@@ -8,7 +8,6 @@ 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: |
@@ -46,7 +45,6 @@ nav:
- Home: index.md
- Firmware and tools:
- firmware_picker.md
- Short firmware links: d.md
- notifications.md
- preset_test.md
- filter_tool.md
@@ -133,7 +131,7 @@ extra_javascript:
- _javascript/hosted_doc_links.js?v=20261007-1
- _javascript/notification_builder.js?v=20261006-channel9
- https://unpkg.com/leaflet@1.9.4/dist/leaflet.js
- _javascript/firmware_picker.js?v=20261008-rak-uart2-1
- _javascript/firmware_picker.js?v=20261012-short-picker-1
- _javascript/telemetry_decoder.js?v=20261007-hosted-form
- _javascript/management_decoder.js
- _javascript/filter_tool.js?v=20261002-hashbytes-2plus
@@ -1,54 +0,0 @@
#!/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
@@ -1,50 +0,0 @@
"""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
@@ -1,223 +0,0 @@
#!/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
@@ -1,121 +0,0 @@
#!/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()
+158
View File
@@ -1,6 +1,9 @@
"use strict";
const assert = require("assert");
const fs = require("fs");
const path = require("path");
const vm = require("vm");
const picker = require("../docs/_javascript/firmware_picker.js");
const bootloaderManifest = require("../docs/_data/bootloader_manifest.json");
@@ -1629,6 +1632,161 @@ assert.deepStrictEqual(escapedShare.filters, {});
assert.deepStrictEqual(escapedShare.unavailable, ['role=<script>']);
console.log('shareable picker URL tests passed');
// Compact links navigate to the picker and retain the same selection semantics.
// Parameter values stay readable rather than depending on catalog row numbers.
const shortShareBase = 'https://example.com/p/?h=storage#installation-methods';
const shortShareUrl = picker.shortSelectionUrl(shortShareBase, shareFilters, true);
const shortShareParsedUrl = new URL(shortShareUrl);
assert.strictEqual(shortShareParsedUrl.pathname, '/p/');
assert.strictEqual(shortShareParsedUrl.searchParams.get('h'), 'storage');
assert.strictEqual(shortShareParsedUrl.hash, '#installation-methods');
assert(shortShareUrl.length < shareUrl.length);
assert.strictEqual(shortShareParsedUrl.searchParams.get('s'),
'1.hRAK_4631.rcompanion.lusb.olora-source.mfull.ffull.vdefault.iuf2');
assert(picker.FACET_FIELDS.every(field => !shortShareParsedUrl.searchParams.has(field)));
assert(!shortShareParsedUrl.searchParams.has('chipAuto'));
const restoredShortShare = picker.selectionFromUrl(shortShareUrl, controlled.profiles);
assert.deepStrictEqual(restoredShortShare, restoredShare);
assert.strictEqual(picker.shortSelectionUrl(shortShareUrl, {}, false), shortShareBase);
assert.strictEqual(picker.selectionUrl(shortShareUrl, {}, false), shortShareBase,
'Legacy serialization must also remove the compact selection when clearing');
const manualShortShare = picker.shortSelectionUrl(shortShareBase, shareFilters, false);
const manualShortSelection = picker.selectionFromUrl(manualShortShare, controlled.profiles);
assert.strictEqual(manualShortSelection.automaticChipFamily, false);
for (const field of picker.FACET_FIELDS) {
assert.strictEqual(manualShortSelection.filters[field], shareFilters[field], field);
}
assert(new URL(manualShortShare).searchParams.get('s').split('.').includes('a0'));
const anyChipShortSelection = picker.selectionFromUrl(picker.shortSelectionUrl(shortShareBase,
{...shareFilters, chipFamily: ''}, false), controlled.profiles);
assert.strictEqual(anyChipShortSelection.automaticChipFamily, false);
assert(!anyChipShortSelection.filters.chipFamily);
assert.strictEqual(anyChipShortSelection.filters.hardware, 'RAK_4631');
const partialShortShare = picker.shortSelectionUrl(shortShareBase,
{hardwareFamily: 'RAK_4631', hardware: 'RAK_4631', role: 'companion'}, true);
assert.strictEqual(new URL(partialShortShare).searchParams.get('s'), '1.hRAK_4631.rcompanion');
assert.deepStrictEqual(picker.selectionFromUrl(partialShortShare, controlled.profiles), {
filters: {hardwareFamily: 'RAK_4631', hardware: 'RAK_4631', role: 'companion'},
automaticChipFamily: true, unavailable: [],
});
const familyOnlyShortSelection = picker.selectionFromUrl(picker.shortSelectionUrl(shortShareBase,
{hardwareFamily: 'RAK_4631'}, true), controlled.profiles);
assert.deepStrictEqual(familyOnlyShortSelection.filters, {hardwareFamily: 'RAK_4631'});
assert.strictEqual(familyOnlyShortSelection.automaticChipFamily, true);
// Opening, editing and clearing a compact link must not resurrect stale choices.
const changedShortSelection = {...restoredShortShare.filters, install: ''};
const editedShortShare = picker.shortSelectionUrl(shortShareUrl, changedShortSelection, true);
const editedShortSelection = picker.selectionFromUrl(editedShortShare, controlled.profiles);
assert.deepStrictEqual(editedShortSelection.unavailable, []);
assert.strictEqual(editedShortSelection.filters.install, undefined);
assert.strictEqual(new URL(editedShortShare).searchParams.getAll('s').length, 1);
const editedLegacyShare = picker.selectionUrl(shortShareUrl, changedShortSelection, true);
assert(!new URL(editedLegacyShare).searchParams.has('s'));
assert.strictEqual(picker.selectionFromUrl(editedLegacyShare, controlled.profiles).filters.install, undefined);
const compactWithLegacy = shortShareUrl.replace('#installation-methods', '') +
'&hardware=RAK_3401&role=repeater&chipAuto=0';
assert.deepStrictEqual(picker.selectionFromUrl(compactWithLegacy, controlled.profiles), restoredShortShare,
'A compact selection must not inherit stale legacy picker fields');
const normalizedCompact = new URL(picker.shortSelectionUrl(compactWithLegacy, shareFilters, true));
assert(!normalizedCompact.searchParams.has('hardware'));
assert(!normalizedCompact.searchParams.has('role'));
assert(!normalizedCompact.searchParams.has('chipAuto'));
assert.strictEqual(normalizedCompact.searchParams.get('h'), 'storage');
const staleShortSelection = picker.selectionFromUrl(
'https://example.com/p/?s=1.hRAK_4631.rcompanion.vremoved.ibin', controlled.profiles);
assert.strictEqual(staleShortSelection.filters.hardware, 'RAK_4631');
assert.strictEqual(staleShortSelection.filters.role, 'companion');
assert.deepStrictEqual(staleShortSelection.unavailable, ['variant=removed', 'install=bin']);
const escapedShortSelection = picker.selectionFromUrl(
'https://example.com/p/?s=1.r%253Cscript%253E', controlled.profiles);
assert.deepStrictEqual(escapedShortSelection.filters, {});
assert.deepStrictEqual(escapedShortSelection.unavailable, ['role=<script>']);
for (const malformedQuery of [
's=', 's=2.hRAK_4631', 's=1.hRAK_4631.hRAK_3401', 's=1.xjunk',
's=1.h%252', 's=1.a2', 's=1.a0.a0', 's=1.hRAK_4631&s=1.hRAK_3401',
's=1.', 's=1.h',
]) {
const invalid = picker.selectionFromUrl('https://example.com/p/?' + malformedQuery, controlled.profiles);
assert.deepStrictEqual(invalid.filters, {}, malformedQuery);
assert(invalid.unavailable.length > 0, malformedQuery + ' must explain the invalid selection');
}
// Field delimiters and URL punctuation are data inside a catalog identifier.
const punctuationHardware = 'Board.rev/2 +100%?x#';
const punctuationProfile = {chipFamily: 'esp32', hardwareFamily: 'Board_Family',
hardware: punctuationHardware, role: 'repeater', logging: 'none', ota: 'none',
mode: 'standard', feature: 'standard', variant: 'rev.2+hot/cold 50% \u00e9',
installKinds: ['merged-bin']};
const punctuationFilters = Object.fromEntries(picker.FACET_FIELDS.map(field =>
[field, field === 'install' ? 'merged-bin' : punctuationProfile[field]]));
const punctuationShortUrl = picker.shortSelectionUrl(shortShareBase, punctuationFilters, false);
const punctuationTokens = new URL(punctuationShortUrl).searchParams.get('s').split('.');
assert.strictEqual(decodeURIComponent(punctuationTokens.find(token => token.startsWith('h')).slice(1)),
punctuationHardware);
assert(punctuationTokens.includes('gBoard_Family'), 'Different family and hardware IDs must both survive');
const punctuationSelection = picker.selectionFromUrl(punctuationShortUrl, [punctuationProfile]);
assert.deepStrictEqual(punctuationSelection.unavailable, []);
assert.deepStrictEqual(punctuationSelection.filters, punctuationFilters);
assert.strictEqual(punctuationSelection.automaticChipFamily, false);
for (const candidateCatalog of [catalog, controlled, currentCatalog, chipCatalog, withMigrations]) {
for (const profile of candidateCatalog.profiles) {
for (const install of profile.installKinds) {
for (const logging of picker.profileFieldValues(profile, 'logging')) {
for (const automatic of [false, true]) {
const chosen = Object.fromEntries(picker.FACET_FIELDS.map(field =>
[field, field === 'install' ? install : field === 'logging' ? logging :
picker.profileFieldValues(profile, field)[0] || '']));
const compactUrl = picker.shortSelectionUrl(shortShareBase, chosen, automatic);
const compactSelection = picker.selectionFromUrl(compactUrl, candidateCatalog.profiles);
const legacySelection = picker.selectionFromUrl(
picker.selectionUrl(shortShareBase, chosen, automatic), candidateCatalog.profiles);
const choiceDescription = profile.target + ' / ' + install + ' / logging=' + logging +
' / automatic=' + automatic;
assert.deepStrictEqual(compactSelection.unavailable, [], choiceDescription);
assert.deepStrictEqual(compactSelection, legacySelection, choiceDescription);
assert(candidateCatalog.profiles.filter(item => picker.profileMatchesFacets(item, compactSelection.filters))
.includes(profile), profile.target + ' must remain reachable');
}
}
}
}
}
console.log('compact picker URL compatibility, validation, escaping and catalog round trips passed');
// Exercise the actual short-route script, including its fallback navigation link.
const shortRouteHtml = fs.readFileSync(path.join(__dirname, '../docs/p/index.html'), 'utf8');
const shortRouteScript = shortRouteHtml.match(/<script>([\s\S]*?)<\/script>/);
assert(shortRouteScript, 'The short picker route must contain its navigation script');
for (const routeHref of [
'https://example.com/MeshCore/p/?s=1.hRAK_4631.rcompanion#installation-methods',
'https://example.com/MeshCore/p/?s=1.hRAK_4631&url=https%3A%2F%2Fother.test%2Fevil#installation-methods',
'https://example.com/MeshCore/p/',
]) {
const sourceUrl = new URL(routeHref);
const expectedDestination = 'https://example.com/MeshCore/firmware_picker/' +
sourceUrl.search + sourceUrl.hash;
const fallbackLink = {href: '../firmware_picker/'};
const redirectedUrls = [];
vm.runInNewContext(shortRouteScript[1], {
URL,
window: {location: {href: routeHref, search: sourceUrl.search, hash: sourceUrl.hash,
replace: href => redirectedUrls.push(href)}},
document: {getElementById: id => {
assert.strictEqual(id, 'picker-link');
return fallbackLink;
}},
});
assert.deepStrictEqual(redirectedUrls, [expectedDestination]);
assert.strictEqual(fallbackLink.href, expectedDestination);
assert.strictEqual(new URL(fallbackLink.href).origin, sourceUrl.origin);
}
console.log('short picker route preserves selections and redirects only to the same-origin picker');
// One choice carries OTA, feature set and sensor/storage tradeoffs together.
// Every existing profile remains reachable and resolves to its original files.
for (const candidateCatalog of [catalog, controlled, currentCatalog, chipCatalog, withMigrations]) {
+2
View File
@@ -40,6 +40,8 @@ class OfflinePickerPackageTests(unittest.TestCase):
self.assertTrue(all('uploader' not in asset for asset in embedded['releases'][0]['assets']))
self.assertIn('Full supported sensors', packaged)
self.assertIn('Reduced sensors', packaged)
self.assertIn('data-share-url="https://mikecarper.github.io/MeshCore/p/"', packaged)
self.assertIn('shortSelectionUrl', packaged)
def test_controls_must_belong_to_the_packaged_release_family(self):
with self.assertRaisesRegex(ValueError, 'match the supplied release family'):
-228
View File
@@ -1,228 +0,0 @@
"""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()