diff --git a/server/ota_endpoint.py b/server/ota_endpoint.py index 0f76c2a..c1ee4be 100644 --- a/server/ota_endpoint.py +++ b/server/ota_endpoint.py @@ -50,18 +50,27 @@ def ota_check_impl(current_version: str, firmware_dir: Path = FIRMWARE_DIR) -> d if not manifest_path.exists(): return {"update": False} - manifest = json.loads(manifest_path.read_text()) - if _parse_version(manifest["version"]) <= _parse_version(current_version): + try: + manifest = json.loads(manifest_path.read_text()) + 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} sig_path = firmware_dir / "current.sig" + if not sig_path.exists(): + return {"update": False} sig_b64 = base64.b64encode(sig_path.read_bytes()).decode() return { "update": True, - "version": manifest["version"], - "size": manifest["size"], - "sha256": manifest["sha256"], + "version": version, + "size": size, + "sha256": sha256, "sig_b64": sig_b64, } @@ -91,12 +100,8 @@ async def ota_firmware( # device_id: str = Depends(verify_device_hmac), # uncomment when wiring into app ): """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") - return FileResponse( - FIRMWARE_DIR / "current.bin", - media_type="application/octet-stream", - ) + return FileResponse(bin_path, media_type="application/octet-stream") diff --git a/server/test_ota_endpoint.py b/server/test_ota_endpoint.py index a85bc9b..8f7a924 100644 --- a/server/test_ota_endpoint.py +++ b/server/test_ota_endpoint.py @@ -68,3 +68,9 @@ def test_firmware_endpoint_missing_raises(tmp_path): import server.ota_endpoint as mod with pytest.raises(mod.FirmwareNotFoundError): 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