Merge pull request #243 from davidmonterocrespo24/feat/picow-iot-gateway
Feat/picow iot gateway
This commit is contained in:
David Montero Crespo
GitHub
commit
319decf1bd
|
|
@ -18,6 +18,8 @@ from fastapi import APIRouter, Request, Response
|
|||
|
||||
from app.core.hooks import iot_gateway_gate
|
||||
from app.services.esp32_lib_manager import esp_lib_manager
|
||||
from app.services.picow_net.consts import STA_IP
|
||||
from app.services.picow_net_bridge import picow_net_manager
|
||||
|
||||
router = APIRouter()
|
||||
logger = logging.getLogger(__name__)
|
||||
|
|
@ -66,14 +68,26 @@ async def gateway_proxy(client_id: str, path: str, request: Request) -> Response
|
|||
media_type='application/json',
|
||||
)
|
||||
|
||||
# ── ESP32: the server runs in QEMU, reachable via slirp hostfwd. ──
|
||||
inst = esp_lib_manager.get_instance(client_id)
|
||||
if not inst or not inst.wifi_enabled or inst.wifi_hostfwd_port == 0:
|
||||
return Response(
|
||||
content='{"error":"No WiFi-enabled ESP32 instance found for this client"}',
|
||||
status_code=404,
|
||||
media_type='application/json',
|
||||
)
|
||||
if inst and inst.wifi_enabled and inst.wifi_hostfwd_port != 0:
|
||||
return await _proxy_esp32(inst, path, request)
|
||||
|
||||
# ── Pico W: the server runs in the browser-side lwIP, reachable only
|
||||
# by injecting TCP frames over the WebSocket bridge into the chip. ──
|
||||
picow = picow_net_manager.get_instance(client_id)
|
||||
if picow is not None and picow.wifi_enabled:
|
||||
return await _proxy_picow(picow, path, request)
|
||||
|
||||
return Response(
|
||||
content='{"error":"No WiFi-enabled board found for this client. Make sure your sketch connected to WiFi and started a server on port 80."}',
|
||||
status_code=404,
|
||||
media_type='application/json',
|
||||
)
|
||||
|
||||
|
||||
async def _proxy_esp32(inst, path: str, request: Request) -> Response:
|
||||
"""Reverse-proxy to an ESP32 web server via QEMU slirp hostfwd."""
|
||||
target_url = f'http://127.0.0.1:{inst.wifi_hostfwd_port}/{path}'
|
||||
body = await request.body()
|
||||
|
||||
|
|
@ -117,3 +131,88 @@ async def gateway_proxy(client_id: str, path: str, request: Request) -> Response
|
|||
headers=resp_headers,
|
||||
media_type=resp.headers.get('content-type'),
|
||||
)
|
||||
|
||||
|
||||
# Hop-by-hop / per-connection headers we never forward verbatim.
|
||||
_HOP_BY_HOP = {'host', 'transfer-encoding', 'connection', 'content-encoding',
|
||||
'keep-alive', 'proxy-connection', 'upgrade'}
|
||||
|
||||
|
||||
async def _proxy_picow(bridge, path: str, request: Request) -> Response:
|
||||
"""Reverse-proxy to a Pico W web server living in the browser-side lwIP.
|
||||
|
||||
There is no host-side socket to connect to — the server only exists
|
||||
inside the simulated chip — so we hand-build a raw HTTP/1.1 request and
|
||||
have the picow_net stack open a TCP connection INTO the chip over the
|
||||
WebSocket bridge, then parse the raw response back out."""
|
||||
body = await request.body()
|
||||
query = request.url.query
|
||||
target = '/' + path + (('?' + query) if query else '')
|
||||
|
||||
headers = {
|
||||
k: v for k, v in request.headers.items()
|
||||
if k.lower() not in _HOP_BY_HOP
|
||||
}
|
||||
headers['Host'] = STA_IP
|
||||
headers['Connection'] = 'close' # make the chip's server FIN when done
|
||||
if body and 'content-length' not in {k.lower() for k in headers}:
|
||||
headers['Content-Length'] = str(len(body))
|
||||
|
||||
req_line = f'{request.method} {target} HTTP/1.1\r\n'
|
||||
header_block = ''.join(f'{k}: {v}\r\n' for k, v in headers.items())
|
||||
raw_request = (req_line + header_block + '\r\n').encode('latin-1') + body
|
||||
|
||||
try:
|
||||
raw_response = await bridge.http_into_chip(raw_request, timeout=12.0)
|
||||
except Exception:
|
||||
logger.exception('[picow-gateway] request into chip failed')
|
||||
raw_response = None
|
||||
|
||||
if not raw_response:
|
||||
return Response(
|
||||
content='{"error":"Pico W HTTP server did not respond. Make sure your sketch connected to WiFi and is listening on port 80."}',
|
||||
status_code=502,
|
||||
media_type='application/json',
|
||||
)
|
||||
|
||||
status, resp_headers, resp_body = _parse_http_response(raw_response)
|
||||
for h in ('transfer-encoding', 'connection', 'content-encoding',
|
||||
'content-length', 'keep-alive'):
|
||||
resp_headers.pop(h, None)
|
||||
media_type = resp_headers.pop('content-type', None) or 'text/html'
|
||||
|
||||
return Response(
|
||||
content=resp_body,
|
||||
status_code=status,
|
||||
headers=resp_headers,
|
||||
media_type=media_type,
|
||||
)
|
||||
|
||||
|
||||
def _parse_http_response(raw: bytes) -> tuple[int, dict, bytes]:
|
||||
"""Split a raw HTTP/1.x response into (status, headers, body). Headers
|
||||
are returned with lower-cased keys (so callers can pop reliably)."""
|
||||
sep = raw.find(b'\r\n\r\n')
|
||||
sep_len = 4
|
||||
if sep < 0:
|
||||
sep = raw.find(b'\n\n')
|
||||
sep_len = 2
|
||||
if sep < 0:
|
||||
# No header terminator — treat the whole thing as a body.
|
||||
return 200, {}, raw
|
||||
|
||||
head = raw[:sep].decode('latin-1', 'replace')
|
||||
resp_body = raw[sep + sep_len:]
|
||||
lines = head.replace('\r\n', '\n').split('\n')
|
||||
|
||||
status = 200
|
||||
parts = lines[0].split(' ', 2)
|
||||
if len(parts) >= 2 and parts[1].isdigit():
|
||||
status = int(parts[1])
|
||||
|
||||
headers: dict = {}
|
||||
for line in lines[1:]:
|
||||
if ':' in line:
|
||||
k, v = line.split(':', 1)
|
||||
headers[k.strip().lower()] = v.strip()
|
||||
return status, headers, resp_body
|
||||
|
|
|
|||
|
|
@ -18,9 +18,12 @@ from typing import Awaitable, Callable
|
|||
|
||||
from .arp import ArpResponder
|
||||
from .consts import (
|
||||
ARP_REQUEST,
|
||||
BROADCAST_MAC,
|
||||
ETHERTYPE_ARP,
|
||||
ETHERTYPE_IPV4,
|
||||
GATEWAY_IP,
|
||||
GATEWAY_MAC,
|
||||
IPPROTO_ICMP,
|
||||
IPPROTO_TCP,
|
||||
IPPROTO_UDP,
|
||||
|
|
@ -35,7 +38,8 @@ from .dhcp import (
|
|||
)
|
||||
from .dns import DnsResolver, is_dns_traffic, make_dns_frame
|
||||
from .icmp import IcmpResponder
|
||||
from .protocols import Ethernet, IPv4, TCP, UDP
|
||||
from .protocols import Arp, Ethernet, IPv4, TCP, UDP, make_frame_arp
|
||||
from .tcp_inbound import TcpInbound
|
||||
from .tcp_nat import TcpNat
|
||||
from .udp_nat import UdpNat
|
||||
|
||||
|
|
@ -59,6 +63,7 @@ class PicowNetBridge:
|
|||
self._dns = DnsResolver()
|
||||
self._icmp = IcmpResponder()
|
||||
self._tcp = TcpNat(self._inject)
|
||||
self._tcp_in = TcpInbound(self._inject, lambda: self._chip_mac)
|
||||
self._udp = UdpNat(self._inject)
|
||||
|
||||
# ── lifecycle ──────────────────────────────────────────────────
|
||||
|
|
@ -74,6 +79,7 @@ class PicowNetBridge:
|
|||
self.running = False
|
||||
await asyncio.gather(
|
||||
self._tcp.shutdown(),
|
||||
self._tcp_in.shutdown(),
|
||||
self._udp.shutdown(),
|
||||
return_exceptions=True,
|
||||
)
|
||||
|
|
@ -126,6 +132,12 @@ class PicowNetBridge:
|
|||
tcp = TCP.parse(ip.payload)
|
||||
except ValueError:
|
||||
return
|
||||
# A reply to a connection WE opened into the chip's server (the
|
||||
# IoT gateway) takes priority over the chip-initiated NAT, which
|
||||
# would otherwise RST it as a stray segment.
|
||||
if self._tcp_in.matches(ip, tcp):
|
||||
await self._tcp_in.handle_chip_segment(ip, tcp)
|
||||
return
|
||||
await self._tcp.handle_chip_segment(self._chip_mac, ip, tcp)
|
||||
return
|
||||
|
||||
|
|
@ -152,6 +164,37 @@ class PicowNetBridge:
|
|||
# Anything else — generic UDP NAT.
|
||||
await self._udp.handle_chip_datagram(chip_mac, ip, udp)
|
||||
|
||||
# ── host → chip: inbound HTTP (IoT gateway) ────────────────────
|
||||
|
||||
async def ensure_chip_mac(self) -> bytes:
|
||||
"""Prime the chip's ARP cache for the gateway before we open a
|
||||
connection into it, so its SYN+ACK doesn't stall on a lookup.
|
||||
|
||||
The chip's on-wire MAC is deterministically DEFAULT_STA_MAC
|
||||
(frontend virtual-ap.ts), which equals our STA_MAC, so the default
|
||||
``_chip_mac`` is already the right destination — we don't need to
|
||||
learn it. If the chip ever sent an outbound frame, the learned MAC
|
||||
is used instead. We still emit one gratuitous ARP for the STA so the
|
||||
chip resolves the gateway promptly, then return without blocking."""
|
||||
req = Arp(
|
||||
opcode=ARP_REQUEST,
|
||||
sha=GATEWAY_MAC,
|
||||
spa=ip_to_bytes(GATEWAY_IP),
|
||||
tha=b'\x00' * 6,
|
||||
tpa=ip_to_bytes(STA_IP),
|
||||
)
|
||||
await self._inject(make_frame_arp(BROADCAST_MAC, GATEWAY_MAC, req))
|
||||
await asyncio.sleep(0.05)
|
||||
return self._chip_mac
|
||||
|
||||
async def http_into_chip(self, raw_http: bytes, timeout: float = 12.0) -> bytes | None:
|
||||
"""Open a TCP connection to the chip's :80 server, send a raw HTTP
|
||||
request, and return the raw HTTP response bytes (or None)."""
|
||||
if not self.running or not self.wifi_enabled:
|
||||
return None
|
||||
await self.ensure_chip_mac()
|
||||
return await self._tcp_in.request(raw_http, timeout=timeout)
|
||||
|
||||
# ── host → chip ────────────────────────────────────────────────
|
||||
|
||||
async def _inject(self, frame: bytes) -> None:
|
||||
|
|
|
|||
|
|
@ -0,0 +1,283 @@
|
|||
"""
|
||||
TCP inbound — host-initiated connections INTO the chip's listening server.
|
||||
|
||||
The mirror image of ``tcp_nat.py``. Where TcpNat plays the *server* for
|
||||
connections the chip opens outward, TcpInbound plays the *client* for
|
||||
connections we open inward — so a browser can reach an HTTP server the
|
||||
Pico W sketch is running on ``10.13.37.42:80``.
|
||||
|
||||
This is what makes Pico W web-server examples as useful as the ESP32
|
||||
ones: the ESP32 server lives in QEMU and is reachable via slirp hostfwd,
|
||||
but the Pico W server lives in the browser-side lwIP, reachable only by
|
||||
injecting frames over the WebSocket bridge. We synthesize a TCP client
|
||||
sourced from the gateway (``10.13.37.1``) and drive a one-shot HTTP
|
||||
request/response, exactly the per-request shape the IoT-gateway proxy
|
||||
already uses for the ESP32.
|
||||
|
||||
CLOSED
|
||||
│ we send SYN
|
||||
▼
|
||||
SYN_SENT ── await chip SYN+ACK
|
||||
│ chip SYN+ACK; we send ACK + request
|
||||
▼
|
||||
ESTABLISHED ── pump response bytes chip → us, ACK them
|
||||
│ chip FIN (or Content-Length satisfied)
|
||||
▼
|
||||
we ACK + FIN ──► CLOSED
|
||||
|
||||
Sequence arithmetic is modular-2³² and mirrors tcp_nat.py.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
import random
|
||||
import struct
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Awaitable, Callable, Dict, Optional, Tuple
|
||||
|
||||
from .consts import (
|
||||
GATEWAY_IP,
|
||||
GATEWAY_MAC,
|
||||
IPPROTO_TCP,
|
||||
STA_IP,
|
||||
TCP_ACK,
|
||||
TCP_FIN,
|
||||
TCP_MSS,
|
||||
TCP_PSH,
|
||||
TCP_RST,
|
||||
TCP_SYN,
|
||||
TCP_WINDOW,
|
||||
ip_to_bytes,
|
||||
)
|
||||
from .protocols import IPv4, TCP, make_frame_ipv4, parse_tcp_options
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
InjectFn = Callable[[bytes], Awaitable[None]]
|
||||
ChipMacFn = Callable[[], bytes]
|
||||
|
||||
_GW_IP = ip_to_bytes(GATEWAY_IP)
|
||||
_STA_IP = ip_to_bytes(STA_IP)
|
||||
|
||||
|
||||
def _seq_add(a: int, b: int) -> int:
|
||||
return (a + b) & 0xffffffff
|
||||
|
||||
|
||||
class _State:
|
||||
SYN_SENT = 'SYN_SENT'
|
||||
ESTABLISHED = 'ESTABLISHED'
|
||||
CLOSED = 'CLOSED'
|
||||
|
||||
|
||||
@dataclass
|
||||
class _InboundConn:
|
||||
chip_port: int # = 80 (server port on the chip)
|
||||
our_port: int # ephemeral gateway-side port
|
||||
state: str = _State.SYN_SENT
|
||||
our_isn: int = 0
|
||||
our_seq: int = 0 # next seq we put on the wire chipward
|
||||
chip_seq: int = 0 # next seq we expect from the chip
|
||||
rx: bytearray = field(default_factory=bytearray)
|
||||
established: asyncio.Event = field(default_factory=asyncio.Event)
|
||||
finished: asyncio.Event = field(default_factory=asyncio.Event)
|
||||
data_event: asyncio.Event = field(default_factory=asyncio.Event)
|
||||
reset: bool = False
|
||||
|
||||
|
||||
class TcpInbound:
|
||||
"""One-shot host→chip TCP client used by the IoT gateway."""
|
||||
|
||||
def __init__(self, inject: InjectFn, chip_mac: ChipMacFn) -> None:
|
||||
self._inject = inject
|
||||
self._chip_mac = chip_mac
|
||||
self._conns: Dict[int, _InboundConn] = {} # keyed by our ephemeral port
|
||||
|
||||
# ── routing predicate (called by the bridge before the outbound NAT) ──
|
||||
|
||||
def matches(self, ip: IPv4, tcp: TCP) -> bool:
|
||||
return (
|
||||
tcp.dst_port in self._conns
|
||||
and tcp.src_port == self._conns[tcp.dst_port].chip_port
|
||||
and bytes(ip.src) == _STA_IP
|
||||
and bytes(ip.dst) == _GW_IP
|
||||
)
|
||||
|
||||
# ── chip → us (segments from the chip's server) ────────────────────
|
||||
|
||||
async def handle_chip_segment(self, ip: IPv4, tcp: TCP) -> None:
|
||||
conn = self._conns.get(tcp.dst_port)
|
||||
if conn is None:
|
||||
return
|
||||
|
||||
if tcp.flags & TCP_RST:
|
||||
conn.reset = True
|
||||
conn.state = _State.CLOSED
|
||||
conn.established.set()
|
||||
conn.finished.set()
|
||||
return
|
||||
|
||||
if conn.state == _State.SYN_SENT:
|
||||
if (tcp.flags & TCP_SYN) and (tcp.flags & TCP_ACK):
|
||||
conn.chip_seq = _seq_add(tcp.seq, 1) # SYN consumes one seq
|
||||
conn.state = _State.ESTABLISHED
|
||||
await self._send(conn, TCP_ACK) # complete the handshake
|
||||
conn.established.set()
|
||||
return
|
||||
|
||||
if conn.state != _State.ESTABLISHED:
|
||||
return
|
||||
|
||||
# In-order data only; re-ACK and drop anything out of order so the
|
||||
# chip retransmits (these servers send tiny, in-order responses).
|
||||
if tcp.payload:
|
||||
if tcp.seq == conn.chip_seq:
|
||||
conn.rx.extend(tcp.payload)
|
||||
conn.chip_seq = _seq_add(conn.chip_seq, len(tcp.payload))
|
||||
await self._send(conn, TCP_ACK)
|
||||
conn.data_event.set()
|
||||
else:
|
||||
await self._send(conn, TCP_ACK) # force retransmit
|
||||
return
|
||||
|
||||
if tcp.flags & TCP_FIN:
|
||||
conn.chip_seq = _seq_add(conn.chip_seq, 1)
|
||||
# ACK the FIN, then send our own FIN to close cleanly.
|
||||
await self._send(conn, TCP_ACK)
|
||||
await self._send(conn, TCP_FIN | TCP_ACK)
|
||||
conn.our_seq = _seq_add(conn.our_seq, 1)
|
||||
conn.state = _State.CLOSED
|
||||
conn.finished.set()
|
||||
|
||||
# ── public one-shot request ────────────────────────────────────────
|
||||
|
||||
async def request(self, raw_http: bytes, timeout: float = 12.0) -> Optional[bytes]:
|
||||
"""Open a connection to the chip's :80 server, send ``raw_http``,
|
||||
return the raw HTTP response bytes (or None on failure)."""
|
||||
our_port = self._alloc_port()
|
||||
our_isn = random.randint(0, 0xffffffff)
|
||||
conn = _InboundConn(
|
||||
chip_port=80,
|
||||
our_port=our_port,
|
||||
our_isn=our_isn,
|
||||
our_seq=_seq_add(our_isn, 1), # our SYN consumes one seq
|
||||
)
|
||||
self._conns[our_port] = conn
|
||||
try:
|
||||
# SYN (advertise MSS, like the chip does).
|
||||
await self._send(conn, TCP_SYN, seq=our_isn,
|
||||
options=b'\x02\x04' + struct.pack('!H', TCP_MSS))
|
||||
try:
|
||||
await asyncio.wait_for(conn.established.wait(), timeout=4.0)
|
||||
except asyncio.TimeoutError:
|
||||
logger.info('[picow-tcp-in] SYN to chip:80 timed out')
|
||||
return None
|
||||
if conn.reset or conn.state != _State.ESTABLISHED:
|
||||
return None
|
||||
|
||||
# Send the HTTP request.
|
||||
await self._send(conn, TCP_PSH | TCP_ACK, payload=raw_http)
|
||||
conn.our_seq = _seq_add(conn.our_seq, len(raw_http))
|
||||
|
||||
# Collect the response until the chip FINs, the body is complete
|
||||
# per Content-Length, or we go idle.
|
||||
deadline = asyncio.get_event_loop().time() + timeout
|
||||
while not conn.finished.is_set():
|
||||
if _http_response_complete(conn.rx):
|
||||
break
|
||||
remaining = deadline - asyncio.get_event_loop().time()
|
||||
if remaining <= 0:
|
||||
break
|
||||
conn.data_event.clear()
|
||||
try:
|
||||
# Wake on new data; also poll so the idle/length checks run.
|
||||
await asyncio.wait_for(conn.data_event.wait(), timeout=min(remaining, 1.5))
|
||||
except asyncio.TimeoutError:
|
||||
if conn.rx and _http_headers_complete(conn.rx):
|
||||
break # got a full header block and went idle — good enough
|
||||
return bytes(conn.rx) if conn.rx else None
|
||||
finally:
|
||||
self._conns.pop(our_port, None)
|
||||
|
||||
# ── frame emission ─────────────────────────────────────────────────
|
||||
|
||||
async def _send(
|
||||
self,
|
||||
conn: _InboundConn,
|
||||
flags: int,
|
||||
seq: Optional[int] = None,
|
||||
options: bytes = b'',
|
||||
payload: bytes = b'',
|
||||
) -> None:
|
||||
tcp = TCP(
|
||||
src_port=conn.our_port,
|
||||
dst_port=conn.chip_port,
|
||||
seq=(conn.our_seq if seq is None else seq) & 0xffffffff,
|
||||
ack=conn.chip_seq,
|
||||
flags=flags,
|
||||
window=TCP_WINDOW,
|
||||
options=options,
|
||||
payload=payload,
|
||||
)
|
||||
l4 = tcp.to_bytes(_GW_IP, _STA_IP)
|
||||
frame = make_frame_ipv4(
|
||||
dst_mac=self._chip_mac(),
|
||||
src_mac=GATEWAY_MAC,
|
||||
src_ip=_GW_IP,
|
||||
dst_ip=_STA_IP,
|
||||
protocol=IPPROTO_TCP,
|
||||
l4_payload=l4,
|
||||
)
|
||||
await self._inject(frame)
|
||||
|
||||
def _alloc_port(self) -> int:
|
||||
for _ in range(64):
|
||||
port = random.randint(49152, 65535)
|
||||
if port not in self._conns:
|
||||
return port
|
||||
# Extremely unlikely; fall back to a linear scan.
|
||||
for port in range(49152, 65536):
|
||||
if port not in self._conns:
|
||||
return port
|
||||
raise RuntimeError('no free ephemeral port')
|
||||
|
||||
async def shutdown(self) -> None:
|
||||
for conn in list(self._conns.values()):
|
||||
conn.reset = True
|
||||
conn.finished.set()
|
||||
conn.established.set()
|
||||
self._conns.clear()
|
||||
|
||||
|
||||
# ─── HTTP framing helpers (just enough to know when a reply is done) ────
|
||||
|
||||
def _http_headers_complete(buf: bytearray) -> bool:
|
||||
return b'\r\n\r\n' in buf or b'\n\n' in buf
|
||||
|
||||
|
||||
def _http_response_complete(buf: bytearray) -> bool:
|
||||
"""True once we have a full header block plus a body matching
|
||||
Content-Length (if any). Without a length we wait for FIN/idle."""
|
||||
sep = buf.find(b'\r\n\r\n')
|
||||
sep_len = 4
|
||||
if sep < 0:
|
||||
sep = buf.find(b'\n\n')
|
||||
sep_len = 2
|
||||
if sep < 0:
|
||||
return False
|
||||
header_blob = bytes(buf[:sep]).lower()
|
||||
idx = header_blob.find(b'content-length:')
|
||||
if idx < 0:
|
||||
return False # no declared length — rely on FIN / idle
|
||||
try:
|
||||
line = header_blob[idx:].split(b'\n', 1)[0]
|
||||
length = int(line.split(b':', 1)[1].strip())
|
||||
except (ValueError, IndexError):
|
||||
return False
|
||||
body_len = len(buf) - (sep + sep_len)
|
||||
return body_len >= length
|
||||
|
||||
|
||||
__all__ = ['TcpInbound']
|
||||
|
|
@ -74,6 +74,9 @@ class PicowNetManager:
|
|||
def has_instance(self, client_id: str) -> bool:
|
||||
return client_id in self._instances
|
||||
|
||||
def get_instance(self, client_id: str) -> PicowNetBridge | None:
|
||||
return self._instances.get(client_id)
|
||||
|
||||
# ── Outbound traffic — chip → host ─────────────────────────────
|
||||
|
||||
async def deliver_packet_out(self, client_id: str, ether_b64: str) -> None:
|
||||
|
|
|
|||
|
|
@ -0,0 +1,146 @@
|
|||
/**
|
||||
* picow-gateway-e2e.investigate.test.ts (gated: CYW43_GATEWAY_E2E=1)
|
||||
*
|
||||
* End-to-end proof of the Pico W IoT gateway INBOUND path: a real
|
||||
* RP2040Simulator runs a MicroPython HTTP *server*, and a browser-style
|
||||
* HTTP request to /api/gateway/<clientId>/ is proxied by the backend INTO
|
||||
* the chip over the WebSocket bridge, returning the served page.
|
||||
*
|
||||
* Requires a backend with the inbound proxy + VELXIO_PICOW_NET=1 running;
|
||||
* point at it with VELXIO_E2E_API_BASE (default http://127.0.0.1:8011/api).
|
||||
*/
|
||||
import { describe, it, expect, vi } from 'vitest';
|
||||
import { readFileSync } from 'node:fs';
|
||||
import { WebSocket as NodeWS } from 'ws';
|
||||
|
||||
const FW_PATH = '/home/dave/velxio-prod/velxio/frontend/public/firmware/micropython-rp2040w.uf2';
|
||||
const WASM_PATH = '/home/dave/velxio-prod/velxio/frontend/node_modules/littlefs/dist/littlefs.wasm';
|
||||
|
||||
vi.mock('../simulation/MicroPythonLoader', async (orig) => {
|
||||
const actual = (await orig()) as Record<string, unknown>;
|
||||
return { ...actual, getFirmware: async () => new Uint8Array(readFileSync(FW_PATH)) };
|
||||
});
|
||||
vi.mock('littlefs', async (orig) => {
|
||||
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
||||
const actual = (await orig()) as any;
|
||||
const create = actual.default;
|
||||
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
||||
return { ...actual, default: (cfg: any = {}) => create({ ...cfg, wasmBinary: new Uint8Array(readFileSync(WASM_PATH)) }) };
|
||||
});
|
||||
|
||||
const SERVER_PY = [
|
||||
'import network, socket, time',
|
||||
'w = network.WLAN(network.STA_IF)',
|
||||
'w.active(True)',
|
||||
'w.connect("Velxio-GUEST", "")',
|
||||
'for i in range(80):',
|
||||
' if w.isconnected(): break',
|
||||
' time.sleep_ms(150)',
|
||||
'ip = w.ifconfig()[0]',
|
||||
'print("PYIP", ip)',
|
||||
's = socket.socket()',
|
||||
's.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)',
|
||||
's.bind(("0.0.0.0", 80))',
|
||||
's.listen(1)',
|
||||
'print("LISTEN", ip)',
|
||||
'while True:',
|
||||
' cl, addr = s.accept()',
|
||||
' try:',
|
||||
' cl.recv(512)',
|
||||
' cl.send(b"HTTP/1.1 200 OK\\r\\nContent-Type: text/html\\r\\n\\r\\n<html><body>VELXIO-PICO-OK</body></html>")',
|
||||
' except Exception as e:',
|
||||
' print("SRVERR", repr(e))',
|
||||
' cl.close()',
|
||||
].join('\n');
|
||||
|
||||
const API_BASE = process.env.VELXIO_E2E_API_BASE || 'http://127.0.0.1:8011/api';
|
||||
|
||||
describe.skipIf(!process.env.CYW43_GATEWAY_E2E)('Pico W IoT gateway inbound', () => {
|
||||
it('serves the chip web page through /api/gateway', async () => {
|
||||
const { RP2040Simulator } = await import('../simulation/RP2040Simulator');
|
||||
const { PinManager } = await import('../simulation/PinManager');
|
||||
const { Cyw43Bridge } = await import('../simulation/cyw43/Cyw43Bridge');
|
||||
|
||||
// Stable per-tab session id so bridge.clientId matches the WS registration.
|
||||
const ss = new Map<string, string>();
|
||||
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
||||
(globalThis as any).sessionStorage = {
|
||||
getItem: (k: string) => (ss.has(k) ? ss.get(k) : null),
|
||||
setItem: (k: string, v: string) => { ss.set(k, v); },
|
||||
removeItem: (k: string) => { ss.delete(k); },
|
||||
};
|
||||
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
||||
(globalThis as any).WebSocket = NodeWS;
|
||||
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
||||
(globalThis as any).window = { __VELXIO_API_BASE__: API_BASE };
|
||||
|
||||
const sim = new RP2040Simulator(new PinManager());
|
||||
let serial = '';
|
||||
sim.onSerialData = (ch: string) => { serial += ch; if (serial.length > 80000) serial = serial.slice(-40000); };
|
||||
|
||||
const bridge = new Cyw43Bridge('gw-board');
|
||||
sim.attachCyw43(bridge);
|
||||
bridge.wifiEnabled = true;
|
||||
const pktlog: string[] = [];
|
||||
const desc = (b: Uint8Array) => {
|
||||
if (b.length < 14) return 'short';
|
||||
const et = (b[12] << 8) | b[13];
|
||||
const src = Array.from(b.subarray(6, 12)).map((x) => x.toString(16).padStart(2, '0')).join('');
|
||||
if (et === 0x0806) return `ARP op${b[21]} src=${src}`;
|
||||
if (et !== 0x0800) return 'eth0x' + et.toString(16);
|
||||
const proto = b[23], l4 = 14 + (b[14] & 0xf) * 4;
|
||||
if (proto === 6) return `TCP ${(b[l4] << 8) | b[l4 + 1]}->${(b[l4 + 2] << 8) | b[l4 + 3]} fl${b[l4 + 13].toString(16)} src=${src}`;
|
||||
if (proto === 17) return `UDP ${(b[l4] << 8) | b[l4 + 1]}->${(b[l4 + 2] << 8) | b[l4 + 3]}`;
|
||||
return 'ip-proto' + proto;
|
||||
};
|
||||
const origSend = bridge.sendPacket.bind(bridge);
|
||||
bridge.sendPacket = (e: Uint8Array) => { if (pktlog.length < 120) pktlog.push('OUT ' + desc(e)); return origSend(e); };
|
||||
const innerIn = bridge.onPacketIn!;
|
||||
bridge.onPacketIn = (p) => { if (pktlog.length < 120) pktlog.push('IN ' + desc(p.ether)); return innerIn(p); };
|
||||
bridge.connect();
|
||||
const clientId = bridge.clientId;
|
||||
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
||||
delete (globalThis as any).window;
|
||||
|
||||
await sim.loadMicroPython([{ name: 'main.py', content: SERVER_PY }]);
|
||||
|
||||
// 1. Run until the chip's HTTP server is listening.
|
||||
const upDeadline = Date.now() + 90_000;
|
||||
while (Date.now() < upDeadline && !serial.includes('LISTEN')) {
|
||||
const n = serial.includes('PYIP') ? 1 : 16;
|
||||
for (let i = 0; i < n; i++) sim.runFrameForTime(n === 1 ? 10 : 50);
|
||||
await new Promise((r) => setTimeout(r, 0));
|
||||
}
|
||||
expect(serial).toMatch(/LISTEN 10\.13\.37\.42/);
|
||||
|
||||
// 2. Fire the gateway request and keep stepping the chip so it can
|
||||
// accept the inbound connection and serve the page.
|
||||
const gwUrl = `${API_BASE}/gateway/${encodeURIComponent(clientId)}/`;
|
||||
let result: { status: number; body: string } | null = null;
|
||||
let err: unknown = null;
|
||||
const fetchP = fetch(gwUrl)
|
||||
.then(async (r) => { result = { status: r.status, body: await r.text() }; })
|
||||
.catch((e) => { err = e; });
|
||||
|
||||
const reqDeadline = Date.now() + 30_000;
|
||||
while (Date.now() < reqDeadline && result === null && err === null) {
|
||||
sim.runFrameForTime(10);
|
||||
await new Promise((r) => setTimeout(r, 0));
|
||||
}
|
||||
await fetchP;
|
||||
|
||||
try { bridge.disconnect(); } catch { /* noop */ }
|
||||
try { sim.stop(); } catch { /* noop */ }
|
||||
|
||||
// eslint-disable-next-line no-console
|
||||
console.log('\n===== GATEWAY E2E =====\nclientId=' + clientId +
|
||||
'\nserial(PY)=' + serial.split('\n').filter((l) => l.startsWith('PY') || l.startsWith('LISTEN') || l.startsWith('SRV')).join(' | ') +
|
||||
'\nfetch=' + JSON.stringify(result) + (err ? ' err=' + String(err) : '') +
|
||||
'\npkts=\n' + pktlog.join('\n'));
|
||||
|
||||
expect(err).toBeNull();
|
||||
expect(result).not.toBeNull();
|
||||
expect(result!.status).toBe(200);
|
||||
expect(result!.body).toContain('VELXIO-PICO-OK');
|
||||
}, 140_000);
|
||||
});
|
||||
|
|
@ -261,7 +261,10 @@ export const SerialMonitor: React.FC = () => {
|
|||
// Strip them before any further processing — color isn't
|
||||
// worth a full parser here.
|
||||
const text = activeBoard.serialOutput.replace(/\x1b\[[0-9;]*m/g, '');
|
||||
const ipRegex = /http:\/\/192\.168\.4\.(\d+)(\/[^\s]*)?/g;
|
||||
// ESP32 (QEMU slirp) hands out 192.168.4.x; the Pico W virtual
|
||||
// net hands out 10.13.37.x. Both reach their emulated server
|
||||
// through the same /api/gateway proxy, so linkify either subnet.
|
||||
const ipRegex = /http:\/\/(?:192\.168\.4|10\.13\.37)\.(\d+)(\/[^\s]*)?/g;
|
||||
const matches = [...text.matchAll(ipRegex)];
|
||||
|
||||
if (matches.length > 0) {
|
||||
|
|
|
|||
|
|
@ -2166,12 +2166,23 @@ export const SimulatorCanvas = ({ headerSlot }: SimulatorCanvasProps = {}) => {
|
|||
<CameraToggle boardId={activeBoard.id} />
|
||||
)}
|
||||
|
||||
{/* WiFi status indicator (ESP32 boards only) */}
|
||||
{/* WiFi status indicator + IoT-gateway launcher (ESP32 + Pico W) */}
|
||||
{activeBoard &&
|
||||
isEsp32Kind(activeBoard.boardKind) &&
|
||||
(isEsp32Kind(activeBoard.boardKind) ||
|
||||
activeBoard.boardKind === 'pi-pico-w') &&
|
||||
activeBoard.wifiStatus &&
|
||||
(() => {
|
||||
const status = activeBoard.wifiStatus.status;
|
||||
// The Pico W virtual net assigns its IP deterministically when
|
||||
// the sketch connects; the bridge reports 'started' carrying the
|
||||
// IP. Treat that as got_ip so the badge matches the ESP32 (green,
|
||||
// clickable → the same /api/gateway proxy).
|
||||
const rawStatus = activeBoard.wifiStatus.status;
|
||||
const status =
|
||||
activeBoard.boardKind === 'pi-pico-w' &&
|
||||
rawStatus === 'started' &&
|
||||
activeBoard.wifiStatus.ip
|
||||
? 'got_ip'
|
||||
: rawStatus;
|
||||
const hasIp = status === 'got_ip';
|
||||
const sessionId = getTabSessionId();
|
||||
const clientId = `${sessionId}::${activeBoard.id}`;
|
||||
|
|
|
|||
|
|
@ -66,7 +66,9 @@ async def wifi_connect():
|
|||
print("Connecting WiFi...")
|
||||
while not wlan.isconnected():
|
||||
await asyncio.sleep(0.5)
|
||||
print("IP:", wlan.ifconfig()[0])
|
||||
ip = wlan.ifconfig()[0]
|
||||
print("IP:", ip)
|
||||
print("Open browser: http://%s/" % ip)
|
||||
|
||||
HTML = """<!DOCTYPE html>
|
||||
<html><body><h2>Pico W Async LED</h2>
|
||||
|
|
@ -190,7 +192,7 @@ s.listen(5)
|
|||
|
||||
current_pos = 90
|
||||
write_servo(current_pos)
|
||||
print("Web server started on port 80")
|
||||
print("Web server started. Open browser: http://%s/" % sta.ifconfig()[0])
|
||||
|
||||
while True:
|
||||
conn, addr = s.accept()
|
||||
|
|
|
|||
|
|
@ -0,0 +1,189 @@
|
|||
"""
|
||||
Unit tests for the Pico W inbound IoT-gateway path.
|
||||
|
||||
The gateway lets a browser reach an HTTP server running inside the
|
||||
browser-side lwIP of a simulated Pico W. The backend opens a TCP
|
||||
connection INTO the chip (``tcp_inbound.TcpInbound``) over the WebSocket
|
||||
bridge, sends a raw HTTP request, and parses the response back out.
|
||||
|
||||
These tests import the real modules (no re-implemented framing) so a
|
||||
refactor of the stack surfaces here immediately:
|
||||
|
||||
- the inbound TCP client drives a correct SYN → SYN+ACK → ACK →
|
||||
request → response → FIN exchange and returns the response bytes;
|
||||
- the gateway response parser splits status/headers/body;
|
||||
- the bridge learns the chip's real MAC via ARP before originating.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
|
||||
import pytest
|
||||
|
||||
from app.services.picow_net.bridge import PicowNetBridge
|
||||
from app.services.picow_net.consts import (
|
||||
GATEWAY_IP,
|
||||
STA_IP,
|
||||
STA_MAC,
|
||||
TCP_ACK,
|
||||
TCP_FIN,
|
||||
TCP_PSH,
|
||||
TCP_SYN,
|
||||
ip_to_bytes,
|
||||
)
|
||||
from app.services.picow_net.protocols import Ethernet, IPv4, TCP, Arp, make_frame_arp
|
||||
from app.services.picow_net.tcp_inbound import (
|
||||
TcpInbound,
|
||||
_http_response_complete,
|
||||
_seq_add,
|
||||
)
|
||||
from app.api.routes.iot_gateway import _parse_http_response
|
||||
|
||||
_STA = ip_to_bytes(STA_IP)
|
||||
_GW = ip_to_bytes(GATEWAY_IP)
|
||||
CHIP_MAC = bytes.fromhex('0242da0000aa')
|
||||
|
||||
|
||||
def _parse_injected_tcp(frame: bytes) -> TCP:
|
||||
eth = Ethernet.parse(frame)
|
||||
ip = IPv4.parse(eth.payload)
|
||||
return TCP.parse(ip.payload)
|
||||
|
||||
|
||||
def _chip_seg(src_port: int, dst_port: int, seq: int, ack: int,
|
||||
flags: int, payload: bytes = b'') -> tuple[IPv4, TCP]:
|
||||
"""A segment as if sent by the chip's server (STA:80 → gateway)."""
|
||||
ip = IPv4(protocol=6, src=_STA, dst=_GW)
|
||||
tcp = TCP(src_port=src_port, dst_port=dst_port, seq=seq, ack=ack,
|
||||
flags=flags, window=64240, payload=payload)
|
||||
return ip, tcp
|
||||
|
||||
|
||||
async def _wait_until(pred, timeout=1.0):
|
||||
loop = asyncio.get_event_loop()
|
||||
deadline = loop.time() + timeout
|
||||
while loop.time() < deadline:
|
||||
if pred():
|
||||
return True
|
||||
await asyncio.sleep(0.005)
|
||||
return False
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_inbound_tcp_http_roundtrip():
|
||||
injected: list[bytes] = []
|
||||
|
||||
async def inject(frame: bytes) -> None:
|
||||
injected.append(frame)
|
||||
|
||||
tin = TcpInbound(inject, lambda: CHIP_MAC)
|
||||
|
||||
request_bytes = b'GET /on HTTP/1.1\r\nHost: 10.13.37.42\r\nConnection: close\r\n\r\n'
|
||||
task = asyncio.create_task(tin.request(request_bytes, timeout=2.0))
|
||||
|
||||
# 1. The client should inject a SYN to the chip's :80.
|
||||
assert await _wait_until(lambda: len(injected) >= 1)
|
||||
syn = _parse_injected_tcp(injected[0])
|
||||
assert syn.flags & TCP_SYN and not (syn.flags & TCP_ACK)
|
||||
assert syn.dst_port == 80
|
||||
ephport = syn.src_port
|
||||
our_isn = syn.seq
|
||||
|
||||
# 2. Reply with SYN+ACK; expect the client to ACK and then send the request.
|
||||
chip_isn = 7000
|
||||
ip, tcp = _chip_seg(80, ephport, seq=chip_isn, ack=_seq_add(our_isn, 1),
|
||||
flags=TCP_SYN | TCP_ACK)
|
||||
await tin.handle_chip_segment(ip, tcp)
|
||||
|
||||
assert await _wait_until(
|
||||
lambda: any(_parse_injected_tcp(f).payload == request_bytes for f in injected))
|
||||
# The handshake ACK must have gone out before the request.
|
||||
assert any(_parse_injected_tcp(f).flags & TCP_ACK for f in injected)
|
||||
|
||||
# 3. Send the HTTP response, then FIN (a length-less, close-delimited body —
|
||||
# the common MicroPython "socket then conn.close()" shape).
|
||||
response = (b'HTTP/1.1 200 OK\r\nContent-Type: text/html\r\n\r\n'
|
||||
b'<html><body>LED ON</body></html>')
|
||||
ip, tcp = _chip_seg(80, ephport, seq=_seq_add(chip_isn, 1),
|
||||
ack=0, flags=TCP_PSH | TCP_ACK, payload=response)
|
||||
await tin.handle_chip_segment(ip, tcp)
|
||||
ip, tcp = _chip_seg(80, ephport, seq=_seq_add(chip_isn, 1 + len(response)),
|
||||
ack=0, flags=TCP_FIN | TCP_ACK)
|
||||
await tin.handle_chip_segment(ip, tcp)
|
||||
|
||||
result = await asyncio.wait_for(task, timeout=2.0)
|
||||
assert result == response
|
||||
# The client must close cleanly: a FIN should have been injected.
|
||||
assert any(_parse_injected_tcp(f).flags & TCP_FIN for f in injected)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_inbound_returns_none_when_chip_never_answers():
|
||||
async def inject(frame: bytes) -> None:
|
||||
pass
|
||||
|
||||
tin = TcpInbound(inject, lambda: CHIP_MAC)
|
||||
# No SYN+ACK ever arrives → request gives up (short SYN timeout path).
|
||||
result = await tin.request(b'GET / HTTP/1.1\r\n\r\n', timeout=0.5)
|
||||
assert result is None
|
||||
|
||||
|
||||
def test_http_response_complete_by_content_length():
|
||||
full = b'HTTP/1.1 200 OK\r\nContent-Length: 5\r\n\r\nhello'
|
||||
assert _http_response_complete(bytearray(full))
|
||||
# One byte short → not complete.
|
||||
assert not _http_response_complete(bytearray(full[:-1]))
|
||||
# No declared length → relies on FIN/idle, never "complete" here.
|
||||
assert not _http_response_complete(bytearray(b'HTTP/1.1 200 OK\r\n\r\nhi'))
|
||||
|
||||
|
||||
def test_parse_http_response_splits_status_headers_body():
|
||||
raw = (b'HTTP/1.1 404 Not Found\r\n'
|
||||
b'Content-Type: application/json\r\n'
|
||||
b'X-Foo: bar\r\n\r\n'
|
||||
b'{"missing":true}')
|
||||
status, headers, body = _parse_http_response(raw)
|
||||
assert status == 404
|
||||
assert headers['content-type'] == 'application/json'
|
||||
assert headers['x-foo'] == 'bar'
|
||||
assert body == b'{"missing":true}'
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_ensure_chip_mac_primes_gateway_arp():
|
||||
"""ensure_chip_mac emits an ARP for the STA (priming the chip's gateway
|
||||
lookup) and returns the chip MAC. The chip's on-wire MAC is
|
||||
deterministically STA_MAC, so the default is already the right target."""
|
||||
sent: list[tuple[str, dict]] = []
|
||||
|
||||
async def emit(event: str, data: dict) -> None:
|
||||
sent.append((event, data))
|
||||
|
||||
bridge = PicowNetBridge('sess::pico', emit, wifi_enabled=True)
|
||||
mac = await bridge.ensure_chip_mac()
|
||||
assert mac == STA_MAC
|
||||
|
||||
# An ARP request for the STA, sourced from the gateway, must have gone out.
|
||||
injected = [d['ether_b64'] for e, d in sent if e == 'picow_packet_in']
|
||||
assert injected, 'ensure_chip_mac should inject an ARP'
|
||||
import base64
|
||||
eth = Ethernet.parse(base64.b64decode(injected[0]))
|
||||
assert eth.ethertype == 0x0806
|
||||
arp = Arp.parse(eth.payload)
|
||||
assert arp.opcode == 1 and bytes(arp.tpa) == _STA
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_bridge_tracks_chip_mac_from_outbound_frame():
|
||||
"""If the chip ever sends an outbound frame, the bridge adopts its src
|
||||
MAC (used as the destination for injected replies)."""
|
||||
async def emit(event: str, data: dict) -> None:
|
||||
pass
|
||||
|
||||
bridge = PicowNetBridge('sess::pico', emit, wifi_enabled=True)
|
||||
assert bridge._chip_mac == STA_MAC
|
||||
# A gratuitous ARP from a chip that happens to use a different MAC.
|
||||
reply = Arp(opcode=2, sha=CHIP_MAC, spa=_STA, tha=STA_MAC, tpa=_GW)
|
||||
await bridge.deliver_packet_out(make_frame_arp(STA_MAC, CHIP_MAC, reply))
|
||||
assert bridge._chip_mac == CHIP_MAC
|
||||
Loading…
Reference in New Issue