ci(ota): verify each published binary's native channel by tag

Every observer build now carries both channel URLs, so checking for a URL's
presence or absence can no longer tell prod from beta (and the old checks would
reject every build). Read the ota-base-* tags from every .bin in out/ instead
of a single ELF, with URLs shared by build.sh and both workflows.

.github/workflows/build-observer-firmwares-beta.yml

@@ -67,6 +67,10 @@ env:
   # The channel itself. Firmware fetches <OTA_MANIFEST_BASE>/<OTA_VARIANT>.json,
   # so this URL is what keeps beta nodes on beta.
   OTA_MANIFEST_BASE_URL: https://observer.gessaman.com/beta/v
+  # Both channel bases are baked into every build so `ota branch prod|beta` can
+  # move a node between channels. Identical in both observer workflows.
+  OTA_MANIFEST_BASE_STABLE_URL: https://observer.gessaman.com/v
+  OTA_MANIFEST_BASE_DEV_URL: https://observer.gessaman.com/beta/v
   # Marks the embedded version, e.g. v1.16.0.3-observer-beta-dev-abc1234, so `ver`
   # (and the MQTT firmware_version / SNMP) identify BOTH the channel and its
   # provenance: this channel is built from the upstream-dev-merged line, so "dev"
@@ -168,20 +172,9 @@ jobs:
         run: /usr/bin/env bash build.sh build-firmware ${{ matrix.shard.envs }}
 
       - name: Verify beta channel is baked in
-        shell: bash
-        run: |
-          # Fail fast rather than publish firmware that would OTA itself onto the
-          # production channel. Checks one built binary actually carries the beta
-          # manifest URL and does NOT carry the production one.
-          BIN=$(find .pio/build -name firmware.elf | head -1)
-          if [ -z "$BIN" ]; then echo "no ELF found to verify" >&2; exit 1; fi
-          if ! strings "$BIN" | grep -qF "$OTA_MANIFEST_BASE_URL"; then
-            echo "ERROR: beta manifest base missing from $BIN" >&2; exit 1
-          fi
-          if strings "$BIN" | grep -qE 'https://observer\.gessaman\.com/v"?$'; then
-            echo "ERROR: production manifest base present in a beta build" >&2; exit 1
-          fi
-          echo "OK: $BIN carries $OTA_MANIFEST_BASE_URL"
+        # Fail fast rather than publish firmware that would OTA itself onto the
+        # production channel: checks every .bin the shard publishes.
+        run: python3 scripts/verify_ota_channel.py --expect beta
 
       - name: Upload Shard Artifact
         uses: actions/upload-artifact@v4

.github/workflows/build-observer-firmwares.yml

@@ -48,6 +48,16 @@ env:
   # repo). Baked into the slim OTA manifests' file URLs; must stay consistent
   # with config.json's staticPath.
   STATIC_PATH: https://observer-fw.gessaman.com
+  # Where the firmware fetches <OTA_VARIANT>.json from — this URL *is* the release
+  # channel. Stated explicitly (it equals build.sh's default) rather than left
+  # unset, so "Verify production channel is baked in" below can assert against the
+  # same value the build was handed instead of a second hardcoded copy. Must match
+  # build.sh's OTA_MANIFEST_BASE_URL default and the Pages path serving flasher/v.
+  OTA_MANIFEST_BASE_URL: https://observer.gessaman.com/v
+  # Both channel bases are baked into every build so `ota branch prod|beta` can
+  # move a node between channels. Identical in both observer workflows.
+  OTA_MANIFEST_BASE_STABLE_URL: https://observer.gessaman.com/v
+  OTA_MANIFEST_BASE_DEV_URL: https://observer.gessaman.com/beta/v
 
 jobs:
 
@@ -137,6 +147,11 @@ jobs:
           FIRMWARE_BUILD_NUMBER: ${{ needs.enumerate.outputs.build_number }}
         run: /usr/bin/env bash build.sh build-firmware ${{ matrix.shard.envs }}
 
+      - name: Verify production channel is baked in
+        # Fail fast rather than publish firmware that would OTA itself onto the
+        # wrong channel: checks every .bin the shard publishes.
+        run: python3 scripts/verify_ota_channel.py --expect prod
+
       - name: Upload Shard Artifact
         uses: actions/upload-artifact@v4
         with:

scripts/verify_ota_channel.py

@@ -0,0 +1,125 @@
+#!/usr/bin/env python3
+"""Verify every built observer binary is baked for the expected OTA channel.
+
+Observer builds carry three manifest bases (src/helpers/OtaChannel.h), each stored
+behind a tag so it can be read back from the binary:
+
+  ota-base-native:<url>   the channel the build OTAs from by default
+  ota-base-stable:<url>   production, the target of `ota branch prod`
+  ota-base-dev:<url>      beta, the target of `ota branch beta`
+
+Every build contains both channel URLs, so checking for the presence or absence
+of a URL can no longer tell production from beta. This reads the tags instead and
+fails unless, in every .bin in out/ (the files that get published), each tag appears with exactly one value, the stable
+and dev tags match the workflow's URLs, and the native tag is the URL of the
+channel named by --expect.
+
+  python3 scripts/verify_ota_channel.py --expect prod
+  python3 scripts/verify_ota_channel.py --self-test
+
+URLs default to OTA_MANIFEST_BASE_URL, OTA_MANIFEST_BASE_STABLE_URL and
+OTA_MANIFEST_BASE_DEV_URL from the environment. Stdlib only.
+"""
+import argparse
+import os
+import re
+import sys
+
+TAG_RE = re.compile(rb"ota-base-(native|stable|dev):([\x21-\x7e]*)\x00")
+
+
+def read_tags(data):
+    tags = {"native": set(), "stable": set(), "dev": set()}
+    for m in TAG_RE.finditer(data):
+        tags[m.group(1).decode()].add(m.group(2).decode())
+    return tags
+
+
+def check(data, expect, native_url, stable_url, dev_url):
+    """Return a list of problems with one binary's tags (empty when it passes)."""
+    problems = []
+    if stable_url == dev_url:
+        problems.append("stable and dev URLs are identical (%s)" % stable_url)
+    expected_native = stable_url if expect == "prod" else dev_url
+    if native_url != expected_native:
+        problems.append("OTA_MANIFEST_BASE_URL %s is not the %s URL %s"
+                        % (native_url, expect, expected_native))
+    tags = read_tags(data)
+    for name, want in (("native", expected_native), ("stable", stable_url), ("dev", dev_url)):
+        found = sorted(tags[name])
+        if found != [want]:
+            problems.append("ota-base-%s: expected [%s], found %s" % (name, want, found or "nothing"))
+    return problems
+
+
+def find_bins(root):
+    for name in os.listdir(root):
+        if name.endswith(".bin"):
+            yield os.path.join(root, name)
+
+
+def self_test():
+    P, B = "https://h/v", "https://h/beta/v"
+
+    def blob(native, stable=P, dev=B):
+        return b"\x00junk\x00ota-base-native:%s\x00ota-base-stable:%s\x00ota-base-dev:%s\x00" % (
+            native.encode(), stable.encode(), dev.encode())
+
+    cases = [
+        ("prod build passes", blob(P), "prod", P, True),
+        ("beta build passes", blob(B), "beta", B, True),
+        ("beta build fails prod check", blob(B), "prod", P, False),
+        ("prod build fails beta check", blob(P), "beta", B, False),
+        ("missing tags fail", b"\x00https://h/v\x00https://h/beta/v\x00", "prod", P, False),
+        ("wrong dev URL fails", blob(P, dev="https://other/v"), "prod", P, False),
+        ("conflicting native tags fail", blob(P) + blob(B), "prod", P, False),
+        ("workflow URL off-channel fails", blob(P), "prod", B, False),
+    ]
+    failed = 0
+    for name, data, expect, native, ok in cases:
+        got = not check(data, expect, native, P, B)
+        if got != ok:
+            failed += 1
+            print("FAIL: %s" % name)
+    print("self-test: %d/%d passed" % (len(cases) - failed, len(cases)))
+    return 1 if failed else 0
+
+
+def main():
+    ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
+    ap.add_argument("--expect", choices=("prod", "beta"))
+    ap.add_argument("--bin-dir", default="out")
+    ap.add_argument("--native-url", default=os.environ.get("OTA_MANIFEST_BASE_URL"))
+    ap.add_argument("--stable-url", default=os.environ.get("OTA_MANIFEST_BASE_STABLE_URL"))
+    ap.add_argument("--dev-url", default=os.environ.get("OTA_MANIFEST_BASE_DEV_URL"))
+    ap.add_argument("--self-test", action="store_true")
+    args = ap.parse_args()
+
+    if args.self_test:
+        return self_test()
+    if not args.expect:
+        ap.error("--expect is required")
+    for opt in ("native_url", "stable_url", "dev_url"):
+        if not getattr(args, opt):
+            ap.error("--%s (or its environment variable) is required" % opt.replace("_", "-"))
+
+    bins = sorted(find_bins(args.bin_dir)) if os.path.isdir(args.bin_dir) else []
+    if not bins:
+        print("ERROR: no .bin files in %s" % args.bin_dir, file=sys.stderr)
+        return 1
+    bad = 0
+    for path in bins:
+        with open(path, "rb") as f:
+            problems = check(f.read(), args.expect, args.native_url, args.stable_url, args.dev_url)
+        for p in problems:
+            print("ERROR: %s: %s" % (path, p), file=sys.stderr)
+        bad += bool(problems)
+    if bad:
+        print("ERROR: %d of %d builds are not baked for %s" % (bad, len(bins), args.expect), file=sys.stderr)
+        return 1
+    print("OK: %d builds default to %s (%s) and carry both channels" % (len(bins), args.expect, args.native_url))
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())