fix(server): add error handling for malformed OTA manifest and missing sig file
Co-Authored-By: Claude Sonnet 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -50,18 +50,27 @@ def ota_check_impl(current_version: str, firmware_dir: Path = FIRMWARE_DIR) -> d
|
|||||||
if not manifest_path.exists():
|
if not manifest_path.exists():
|
||||||
return {"update": False}
|
return {"update": False}
|
||||||
|
|
||||||
|
try:
|
||||||
manifest = json.loads(manifest_path.read_text())
|
manifest = json.loads(manifest_path.read_text())
|
||||||
if _parse_version(manifest["version"]) <= _parse_version(current_version):
|
version = manifest["version"]
|
||||||
|
size = manifest["size"]
|
||||||
|
sha256 = manifest["sha256"]
|
||||||
|
except (json.JSONDecodeError, KeyError):
|
||||||
|
return {"update": False}
|
||||||
|
|
||||||
|
if _parse_version(version) <= _parse_version(current_version):
|
||||||
return {"update": False}
|
return {"update": False}
|
||||||
|
|
||||||
sig_path = firmware_dir / "current.sig"
|
sig_path = firmware_dir / "current.sig"
|
||||||
|
if not sig_path.exists():
|
||||||
|
return {"update": False}
|
||||||
sig_b64 = base64.b64encode(sig_path.read_bytes()).decode()
|
sig_b64 = base64.b64encode(sig_path.read_bytes()).decode()
|
||||||
|
|
||||||
return {
|
return {
|
||||||
"update": True,
|
"update": True,
|
||||||
"version": manifest["version"],
|
"version": version,
|
||||||
"size": manifest["size"],
|
"size": size,
|
||||||
"sha256": manifest["sha256"],
|
"sha256": sha256,
|
||||||
"sig_b64": sig_b64,
|
"sig_b64": sig_b64,
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -91,12 +100,8 @@ async def ota_firmware(
|
|||||||
# device_id: str = Depends(verify_device_hmac), # uncomment when wiring into app
|
# device_id: str = Depends(verify_device_hmac), # uncomment when wiring into app
|
||||||
):
|
):
|
||||||
"""Stream the staged firmware binary to the device."""
|
"""Stream the staged firmware binary to the device."""
|
||||||
try:
|
|
||||||
ota_firmware_impl() # validate existence before streaming
|
|
||||||
except FirmwareNotFoundError:
|
|
||||||
from fastapi import HTTPException
|
from fastapi import HTTPException
|
||||||
|
bin_path = FIRMWARE_DIR / "current.bin"
|
||||||
|
if not bin_path.exists():
|
||||||
raise HTTPException(status_code=404, detail="No firmware available")
|
raise HTTPException(status_code=404, detail="No firmware available")
|
||||||
return FileResponse(
|
return FileResponse(bin_path, media_type="application/octet-stream")
|
||||||
FIRMWARE_DIR / "current.bin",
|
|
||||||
media_type="application/octet-stream",
|
|
||||||
)
|
|
||||||
|
|||||||
@@ -68,3 +68,9 @@ def test_firmware_endpoint_missing_raises(tmp_path):
|
|||||||
import server.ota_endpoint as mod
|
import server.ota_endpoint as mod
|
||||||
with pytest.raises(mod.FirmwareNotFoundError):
|
with pytest.raises(mod.FirmwareNotFoundError):
|
||||||
ota_firmware_impl(firmware_dir=tmp_path)
|
ota_firmware_impl(firmware_dir=tmp_path)
|
||||||
|
|
||||||
|
|
||||||
|
def test_check_malformed_manifest(tmp_path):
|
||||||
|
(tmp_path / "manifest.json").write_text("not valid json{{{")
|
||||||
|
result = ota_check_impl(current_version="1.0.0", firmware_dir=tmp_path)
|
||||||
|
assert result["update"] is False
|
||||||
|
|||||||
Reference in New Issue
Block a user