104 lines
5.2 KiB
Python
104 lines
5.2 KiB
Python
"""Offline cross-compiler tests: no hardware, network, or real credentials."""
|
|
import hashlib
|
|
import http.client
|
|
import io
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import ssl
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parent
|
|
COMPILER = (Path.home() / ".platformio" / "packages" / "toolchain-xtensa-esp32s3" /
|
|
"bin" / ("xtensa-esp32s3-elf-g++.exe" if os.name == "nt" else "xtensa-esp32s3-elf-g++"))
|
|
HEADER = ROOT / "include" / "cloud_config.h"
|
|
TOKEN = "a" * 64 # Synthetic test value, never registered in any hub.
|
|
|
|
|
|
class CloudIdentityTests(unittest.TestCase):
|
|
def compile(self, identity, token=TOKEN, expected=None):
|
|
if not COMPILER.is_file():
|
|
self.skipTest("ESP32-S3 toolchain not installed")
|
|
# Isolate header lookup: never read a developer's private credential file.
|
|
with tempfile.TemporaryDirectory(prefix="cloud-id-test-") as directory:
|
|
Path(directory, "cloud_config.h").write_bytes(HEADER.read_bytes())
|
|
arguments = [str(COMPILER), "-std=gnu++11", "-fsyntax-only", "-x", "c++", "-", "-I", directory]
|
|
if identity is not None:
|
|
arguments.append("-DPCB_ID=" + identity)
|
|
if token is not None:
|
|
arguments.append('-DCLOUD_DEVICE_TOKEN="' + token + '"')
|
|
source = '#include "cloud_config.h"\n'
|
|
if expected is not None:
|
|
source += (
|
|
"constexpr bool same(const char* a, const char* b) { "
|
|
"return *a == *b && (*a == '\\0' || same(a + 1, b + 1)); }\n"
|
|
'static_assert(same(CLOUD_DEVICE_ID, "' + expected + '"), "ID mismatch");\n'
|
|
)
|
|
return subprocess.run(arguments, input=source, text=True, capture_output=True, timeout=30, cwd=directory)
|
|
|
|
def test_01_exact_hub_ids(self):
|
|
for value, expected in (("0x01", "esp32-000000000001"), ("0x02", "esp32-000000000002"),
|
|
("0xabcdef123456ULL", "esp32-abcdef123456"),
|
|
("0xffffffffffffULL", "esp32-ffffffffffff")):
|
|
with self.subTest(value=value):
|
|
result = self.compile(value, expected=expected)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
|
|
def test_02_invalid_ids_cannot_produce_firmware(self):
|
|
for value in (None, "0", "-1", "0x1000000000000ULL", "1.5"):
|
|
with self.subTest(value=value):
|
|
result = self.compile(value)
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertIn("PCB_ID", result.stderr)
|
|
|
|
def test_03_missing_or_invalid_token_is_a_build_error(self):
|
|
for token in (None, "", "a" * 63, "a" * 65, "G" * 64, "A" * 64):
|
|
with self.subTest(length=None if token is None else len(token)):
|
|
result = self.compile("0x01", token)
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertIn("CLOUD_DEVICE_TOKEN", result.stderr)
|
|
|
|
def test_04_root_certificate_matches_isrg_root_x1(self):
|
|
text = HEADER.read_text(encoding="utf-8")
|
|
pem = re.search(r"-----BEGIN CERTIFICATE-----.*?-----END CERTIFICATE-----", text, re.S).group()
|
|
digest = hashlib.sha256(ssl.PEM_cert_to_DER_cert(pem)).hexdigest()
|
|
self.assertEqual(digest, "96bcec06264976f37460779acf28c5a7cfe8a3c0aae11a8ffcee05c0bddf08c6")
|
|
|
|
def test_05_transport_keeps_token_out_of_url_and_verifies_tls(self):
|
|
text = (ROOT / "src" / "cloud_client.cpp").read_text(encoding="utf-8")
|
|
self.assertIn('"?device_id="', text)
|
|
self.assertIn("devicePath += CLOUD_DEVICE_ID", text)
|
|
self.assertIn('setExtraHeaders("Authorization: Bearer " CLOUD_DEVICE_TOKEN)', text)
|
|
self.assertIn("beginSslWithCA(host, port, devicePath.c_str(), CLOUD_ROOT_CA_PEM)", text)
|
|
self.assertIn("client.setCACert(CLOUD_ROOT_CA_PEM)", text)
|
|
self.assertNotIn("setInsecure()", text)
|
|
|
|
def test_06_websocket_headers_have_no_bytes_after_http_terminator(self):
|
|
text = (ROOT / "src/cloud_client.cpp").read_text(encoding="utf-8")
|
|
arguments = re.search(r"setExtraHeaders\((.*?)\);", text, re.S).group(1)
|
|
before, after = arguments.split("CLOUD_DEVICE_TOKEN")
|
|
def literals(fragment):
|
|
return "".join(json.loads(value) for value in re.findall(r'"(?:[^"\\]|\\.)*"', fragment))
|
|
extra = literals(before) + TOKEN + literals(after)
|
|
# WebSockets 2.7.3 appends CRLF to extraHeaders, then User-Agent, then CRLF.
|
|
request = (
|
|
"GET /ws/esp32?device_id=esp32-000000000002 HTTP/1.1\r\n"
|
|
"Host: micro.ferrasdh.uk:443\r\nConnection: Upgrade\r\nUpgrade: websocket\r\n"
|
|
"Sec-WebSocket-Version: 13\r\nSec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\n"
|
|
+ extra + "\r\nUser-Agent: arduino-WebSocket-Client\r\n\r\n"
|
|
).encode("ascii")
|
|
stream = io.BytesIO(request)
|
|
stream.readline()
|
|
headers = http.client.parse_headers(stream)
|
|
self.assertEqual(headers.get("Authorization"), "Bearer " + TOKEN)
|
|
self.assertEqual(headers.get("User-Agent"), "arduino-WebSocket-Client")
|
|
self.assertEqual(stream.read(), b"", "HTTP header bytes would become invalid WebSocket frames")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main(verbosity=2)
|