diff --git a/CLIENT_MIGRATION.md b/CLIENT_MIGRATION.md new file mode 100644 index 0000000..1f57630 --- /dev/null +++ b/CLIENT_MIGRATION.md @@ -0,0 +1,248 @@ +# Client migration: `/api/refresh-musallahboard` now requires authentication + +**Applies to:** the MusallahBoard kiosk agent (Raspberry Pi) and anything else that opens the +board refresh stream. +**Breaking:** yes. An existing client connects and then sits silent; after ~15 s the server +closes it with `4008 auth_timeout`. It will never receive a refresh again. + +--- + +## 1. What changed, in one paragraph + +The refresh stream used to accept `ws://host/api/refresh-musallahboard?deviceId=` and +treat the presence of a known, unrevoked device id as sufficient. A device id is an +identifier, not a secret — it is in the board's own URL, on the physical screen, and in +`GET /api/admin/board/devices` — so anyone who had it could take a kiosk's place on the +channel and read its pushes. The endpoint now runs the same Ed25519 challenge–response the +agent command channel (`/api/agent/ws`) already used. **The `?deviceId=` query parameter is +ignored entirely**; the device a connection speaks for is decided by a signature the server +verifies against the public key stored on that device's enrollment row. + +No new credential is issued. The agent already holds the Ed25519 private key whose public +half it registered at `POST /api/agent/enroll`. That is the credential. If your device +enrolled successfully, you have everything you need. + +--- + +## 2. The handshake + +``` +client server + │ WS upgrade GET /api/refresh-musallahboard │ + ├───────────────────────────────────────────────►│ + │ │ + │◄────────────── hello (seq 1) ────────────────┤ 32-byte challenge + │ │ + ├──────────────► auth (seq 1) ────────────────┤ Ed25519 signature + │ │ + │◄────────────── auth_ok (seq 2) ───────────────┤ stream is now live + │ │ + │◄────────────── refresh (seq 3..) ─────────────┤ 0..n, indefinitely +``` + +Every frame in both directions is a single JSON text frame. Binary frames are not used. + +### 2.1 Connect + +``` +ws:///api/refresh-musallahboard (wss:// in production) +``` + +- Query parameters are ignored. Keep sending `?deviceId=` if it is convenient for your logs; + it grants nothing. +- **Origin header:** the endpoint restricts browser origins to the configured frontend and + board base URLs. Send **no** `Origin` header at all (the default for Go's + `gorilla/websocket` and `nhooyr.io/websocket`, and for Python `websockets`) — Spring allows + originless upgrades. If your HTTP stack forces one, it must exactly match + `musallahboard.baseurl` or `frontend.baseurl`, or the upgrade fails with HTTP 403 before + any WebSocket frame is exchanged. +- No `Authorization` header is required or read. Do not send the enrollment token. + +### 2.2 `hello` (server → client, always the first frame) + +```json +{ + "type": "hello", + "seq": 1, + "sessionId": "0d0a2e6a-...-a1b2", + "challenge": "5Xy2...c9Q", + "serverTime": 1774483200123 +} +``` + +| field | type | meaning | +|--------------|-------------|-------------------------------------------------------------| +| `sessionId` | UUID string | Server-issued. Echo it on every frame you send. | +| `challenge` | string | Base64 (**standard alphabet, no padding**) of 32 random bytes. Opaque — do **not** decode it; sign the string exactly as received. | +| `serverTime` | int64 | Server clock, Unix milliseconds. Use it to detect your own clock skew. | + +`sessionId` and `challenge` are fresh per connection. Nothing from a previous connection is +reusable. + +### 2.3 `auth` (client → server) + +Send this as soon as `hello` arrives. You have `musallahboard.stream.authTimeoutMs` +(default **15 000 ms**) from connection open. The deadline is enforced by a sweep every +`musallahboard.stream.authReaperIntervalMs` (default 5 000 ms), so in the worst case a silent +connection survives about 20 s — treat 15 s as the budget, not 20. + +```json +{ + "type": "auth", + "seq": 1, + "sessionId": "0d0a2e6a-...-a1b2", + "deviceId": "9f14c0b1-...-77de", + "timestamp": 1774483200456, + "signature": "MEUCIQ...==" +} +``` + +| field | type | rule | +|-------------|-------------|-------------------------------------------------------------------------| +| `type` | string | Must be `"auth"`. It is the JSON type discriminator; the subtype set is closed and `"auth"` is currently the only member. Anything else is `4004`. | +| `seq` | int64 | Must be `1` on your first frame, strictly increasing thereafter. | +| `sessionId` | UUID string | Exactly the `sessionId` from `hello`. A mismatch is `4001`. | +| `deviceId` | UUID string | The device id returned by `POST /api/agent/enroll`. | +| `timestamp` | int64 | Unix **milliseconds**, not seconds. Must be within ±5 min of server time. | +| `signature` | string | Base64 (standard alphabet, padding optional) of the 64-byte raw Ed25519 signature — detached, **not** ASN.1-wrapped. | + +#### The signed bytes + +UTF-8 encoding of exactly this string, `\n` (LF, `0x0A`) separators, **no trailing newline**: + +``` +musallahboard-stream-auth-v1 + + + + +``` + +- UUIDs are lowercase, hyphenated, canonical 36-character form. +- `` is the base64 **string** from `hello`, copied verbatim. +- `` is the decimal integer, no separators, and must equal the `timestamp` field + in the frame byte for byte. + +The version prefix is `musallahboard-stream-auth-v1`. It differs from the agent channel's +`musallahboard-auth-v1` on purpose: a signature made for one channel must not verify on the +other. **Do not reuse your agent-channel signing helper without changing the prefix** — the +server will reject it with `4001` and it will look, unhelpfully, like a key problem. + +Worked example (Go, `crypto/ed25519`): + +```go +payload := fmt.Sprintf("musallahboard-stream-auth-v1\n%s\n%s\n%s\n%d", + hello.SessionID, hello.Challenge, deviceID, ts) +sig := ed25519.Sign(privKey, []byte(payload)) +frame := authFrame{ + Type: "auth", Seq: 1, + SessionID: hello.SessionID, DeviceID: deviceID, + Timestamp: ts, + Signature: base64.StdEncoding.EncodeToString(sig), +} +``` + +Python (`cryptography`): + +```python +payload = f"musallahboard-stream-auth-v1\n{session_id}\n{challenge}\n{device_id}\n{ts}".encode() +signature = base64.b64encode(private_key.sign(payload)).decode() +``` + +### 2.4 `auth_ok` (server → client) + +```json +{ "type": "auth_ok", "seq": 2, "sessionId": "0d0a...", "deviceId": "9f14...", "serverTime": 1774483200460 } +``` + +The stream is live. Reset your reconnect backoff here, not on socket open — an open socket +that never authenticates is worthless and you would otherwise hammer the server on a +credential bug. + +### 2.5 `refresh` (server → client, 0..n) + +**Unchanged in every field an existing client reads.** `seq` and `sessionId` are additive. + +```json +{ + "type": "refresh", + "seq": 7, + "sessionId": "0d0a...", + "reason": "posters", + "deviceId": "9f14...", + "at": "2026-08-25T18:40:00.123Z" +} +``` + +| field | notes | +|------------|---------------------------------------------------------------------------| +| `reason` | `posters` \| `events` \| `socials` \| `weekly-content` \| `config` \| `manual-refresh`. Treat unknown values as "refetch everything". | +| `deviceId` | Present only when the push is addressed to one board (`reason: "config"`). **Omitted, not null**, on a fleet-wide broadcast. | +| `at` | ISO-8601 instant, UTC. | + +Reaction is unchanged: refetch your payload in place. Do not reload the page. + +### 2.6 Frame size + +Anything you send longer than **4096 characters** is closed with `4004` before it is parsed. +A correct `auth` frame is roughly 300 characters, so this only bites a malformed client. + +### 2.7 Client → server after `auth_ok` + +Nothing. There are no heartbeats on this channel — heartbeats live on `/api/agent/ws`. Any +frame sent after `auth_ok`, including a second `auth`, closes the connection with `4003`. +Rely on WebSocket ping/pong for liveness. + +--- + +## 3. Close codes + +| code | reason | when | what to do | +|--------|--------------------|--------------------------------------------------------------------------------------|------------| +| `4001` | `auth_failed` | Bad signature, unknown `deviceId`, revoked device, device has no key on file, timestamp skew > 5 min, `sessionId` mismatch, or malformed base64 signature. **Deliberately one code for all of these** — the server will not tell an unauthenticated caller which device ids exist. | Back off hard (minutes). Check clock sync and the signing prefix before assuming key loss. Do not fast-retry; it will not start working. | +| `4002` | `seq_violation` | `seq` not strictly increasing. | Bug in your client. Reconnect and restart `seq` at 1. | +| `4003` | `bad_state` | Any frame after `auth_ok`. | Bug in your client. | +| `4004` | `bad_frame` | Unparseable JSON, a `type` this channel does not know, or a frame over 4096 characters. | Bug in your client. | +| `4008` | `auth_timeout` | Connected but did not authenticate within the deadline (default 15 s, swept every 5 s). | Reconnect with normal backoff. | +| `1000` | `superseded` | The same device authenticated on a newer connection. | **Expected during your own reload.** If you did not open a second socket, someone else holds your private key — escalate. | +| `1008` | `device_revoked` | An operator revoked this device while the session was live. Also closes your `/api/agent/ws` session. | Stop reconnecting. The device needs re-enrollment; retrying will only produce `4001`. | +| `1000` | (no reason) / `1006` | Ordinary close or transport drop. | Reconnect with jittered backoff. | + +`4001` and `4008` are the two you will hit during migration. `4008` means the new frames are +not being sent at all; `4001` means they are being sent wrong. + +Codes changed meaning. `4001` used to be `device_id_required`, `4002` `unknown_device`, +`4003` `device_revoked`. Anything matching on the old numbers must be updated. + +### Reconnect behaviour + +Keep full-jitter exponential backoff, capped at 60 s. Add two rules: + +1. Restart the whole handshake on every reconnect. `sessionId`, `challenge` and the signature + are single-connection values. +2. Treat `4001` as a distinct, slower class of failure. A misconfigured fleet retrying a + rejected credential every second is a self-inflicted outage. + +--- + +## 4. Migration order + +The server rejects unauthenticated connections the moment it is deployed; there is no +grace period and no dual-accept mode. + +1. Ship the agent change first, or accept a refresh-stream outage between the two deploys. + Boards keep working during an outage — they fall back to the ten-minute poll. Screens go + stale, they do not go blank. +2. Confirm each board's clock is disciplined (`chronyd`/`systemd-timesyncd`). The ±5 min skew + cap is the most likely field failure, and it presents as `4001`. +3. Verify with `journalctl -u musallahboard-agent | grep auth_ok` (or your equivalent) rather + than by watching for a socket to stay open. + +## 5. Browser-hosted boards + +The kiosk page at `MusallahBoard/frontend/src/api/refreshSocket.js` cannot complete this +handshake — a browser has no access to the device private key, and putting one there would +recreate the flaw in a new place. A page-hosted board needs the local agent to own the socket +and relay refresh events to the page over the loopback interface. That work belongs in the +board repo and is out of scope for this change; until it lands, a browser-only board falls +back to its poll. diff --git a/src/main/java/com/ibrasoft/lensbridge/config/WebSocketConfig.java b/src/main/java/com/ibrasoft/lensbridge/config/WebSocketConfig.java index 09520a7..3f12fc6 100644 --- a/src/main/java/com/ibrasoft/lensbridge/config/WebSocketConfig.java +++ b/src/main/java/com/ibrasoft/lensbridge/config/WebSocketConfig.java @@ -34,7 +34,11 @@ public WebSocketConfig(BoardStreamHandler boardStreamHandler, @Override public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) { - // Path is the deployed kiosk's; boards scope themselves with ?deviceId=. + // Boards authenticate per-frame inside the channel (Ed25519 challenge-response), the + // same way agents do. The origin list is kept as browser-only defence in depth: it + // constrains a page running in someone's browser and nothing else, so it must never + // be the reason a connection is trusted. Non-browser clients send no Origin header, + // which Spring allows. registry.addHandler(boardStreamHandler, "/api/refresh-musallahboard") .setAllowedOrigins(frontendBaseUrl, musallahBoardBaseUrl); diff --git a/src/main/java/com/ibrasoft/lensbridge/controller/DeviceAdminController.java b/src/main/java/com/ibrasoft/lensbridge/controller/DeviceAdminController.java index 8611f28..2ffbb8c 100644 --- a/src/main/java/com/ibrasoft/lensbridge/controller/DeviceAdminController.java +++ b/src/main/java/com/ibrasoft/lensbridge/controller/DeviceAdminController.java @@ -20,6 +20,7 @@ import com.ibrasoft.lensbridge.service.agent.CommandDispatcher; import com.ibrasoft.lensbridge.service.agent.EnrollmentTokenService; import com.ibrasoft.lensbridge.exception.ApiResponseException; +import com.ibrasoft.lensbridge.handler.BoardStreamHandler; import com.ibrasoft.lensbridge.service.agent.EnrollmentTokenService.Issued; import io.swagger.v3.oas.annotations.Operation; import io.swagger.v3.oas.annotations.media.Content; @@ -60,6 +61,7 @@ public class DeviceAdminController { private final CommandDispatcher commandDispatcher; private final ObjectMapper objectMapper; private final AgentSessionRegistry sessionRegistry; + private final BoardStreamHandler boardStream; private final AdminAuditService auditService; @Operation(operationId = "listDevices", summary = "List every enrolled board device") @@ -118,7 +120,7 @@ public ResponseEntity issueEnrollmentToken( @Operation(operationId = "revokeDevice", summary = "Revoke a device's enrollment", - description = "Closes any live agent session. Idempotent: re-revoking an already revoked device is a no-op.") + description = "Closes any live agent session and refresh-stream session. Idempotent: re-revoking an already revoked device is a no-op.") @ApiResponses({ @ApiResponse(responseCode = "200", description = "Device revoked"), @ApiResponse(responseCode = "404", description = "No device with that id", @@ -137,7 +139,10 @@ public ResponseEntity revoke(@PathVariable UUID deviceId, if (d.getRevokedAt() == null) { d.setRevokedAt(Instant.now()); deviceRepository.save(d); - sessionRegistry.closeIfPresent(deviceId, CloseStatus.POLICY_VIOLATION.withReason("device_revoked")); + CloseStatus revoked = CloseStatus.POLICY_VIOLATION.withReason("device_revoked"); + sessionRegistry.closeIfPresent(deviceId, revoked); + // A revoked board must lose the refresh stream too, not just the command channel. + boardStream.closeIfPresent(deviceId, revoked); log.warn("Device {} revoked by {}", deviceId, getCurrentUserEmail()); auditService.logAuditEvent(getCurrentUserEmail(), AuditAction.REVOKE_DEVICE, diff --git a/src/main/java/com/ibrasoft/lensbridge/dto/board/stream/BoardStreamAuthFrame.java b/src/main/java/com/ibrasoft/lensbridge/dto/board/stream/BoardStreamAuthFrame.java new file mode 100644 index 0000000..f852a29 --- /dev/null +++ b/src/main/java/com/ibrasoft/lensbridge/dto/board/stream/BoardStreamAuthFrame.java @@ -0,0 +1,29 @@ +package com.ibrasoft.lensbridge.dto.board.stream; + +import lombok.EqualsAndHashCode; +import lombok.Getter; +import lombok.Setter; + +import java.util.UUID; + +/** + * Board → backend authentication frame, sent in response to a {@code hello}. + *

+ * The signature covers a fixed canonical string; see + * {@link com.ibrasoft.lensbridge.service.board.stream.BoardStreamAuthPayload}. The public key + * used to verify it is always the one on the enrolled {@code Device} row — never anything + * carried in this frame. + */ +@Getter +@Setter +@EqualsAndHashCode(callSuper = true) +public class BoardStreamAuthFrame extends IncomingBoardStreamFrame { + + private UUID deviceId; + + /** Unix milliseconds; server enforces |timestamp - now| within a skew window. */ + private long timestamp; + + /** Base64-encoded 64-byte Ed25519 signature. */ + private String signature; +} diff --git a/src/main/java/com/ibrasoft/lensbridge/dto/board/stream/IncomingBoardStreamFrame.java b/src/main/java/com/ibrasoft/lensbridge/dto/board/stream/IncomingBoardStreamFrame.java new file mode 100644 index 0000000..025bcee --- /dev/null +++ b/src/main/java/com/ibrasoft/lensbridge/dto/board/stream/IncomingBoardStreamFrame.java @@ -0,0 +1,43 @@ +package com.ibrasoft.lensbridge.dto.board.stream; + +import com.fasterxml.jackson.annotation.JsonSubTypes; +import com.fasterxml.jackson.annotation.JsonTypeInfo; + +import java.util.UUID; + +/** + * Polymorphic envelope for frames sent from a board to the backend on the refresh stream. + * Discriminated by the {@code type} field. + *

+ * The subtype list is closed on purpose — {@code Id.NAME} with an explicit + * {@link JsonSubTypes} set, never {@code Id.CLASS} or default typing. + */ +@JsonTypeInfo( + use = JsonTypeInfo.Id.NAME, + include = JsonTypeInfo.As.PROPERTY, + property = "type", + visible = true +) +@JsonSubTypes({ + @JsonSubTypes.Type(value = BoardStreamAuthFrame.class, name = "auth") +}) +public abstract class IncomingBoardStreamFrame { + + /** Discriminator (kept for logging even though Jackson uses it as the type id). */ + private String type; + + /** Monotonic sequence number, starts at 1 from the board. */ + private long seq; + + /** Session identifier issued by the backend in the {@code hello} frame. */ + private UUID sessionId; + + public String getType() { return type; } + public void setType(String type) { this.type = type; } + + public long getSeq() { return seq; } + public void setSeq(long seq) { this.seq = seq; } + + public UUID getSessionId() { return sessionId; } + public void setSessionId(UUID sessionId) { this.sessionId = sessionId; } +} diff --git a/src/main/java/com/ibrasoft/lensbridge/dto/board/stream/OutgoingBoardStreamFrame.java b/src/main/java/com/ibrasoft/lensbridge/dto/board/stream/OutgoingBoardStreamFrame.java new file mode 100644 index 0000000..7753fdc --- /dev/null +++ b/src/main/java/com/ibrasoft/lensbridge/dto/board/stream/OutgoingBoardStreamFrame.java @@ -0,0 +1,45 @@ +package com.ibrasoft.lensbridge.dto.board.stream; + +import com.fasterxml.jackson.annotation.JsonInclude; +import lombok.Builder; +import lombok.Data; + +import java.util.UUID; + +/** + * Backend → board frame envelope for the refresh stream. + *

+ * Three shapes ride this envelope: {@code hello} (challenge), {@code auth_ok}, and + * {@code refresh}. The {@code refresh} shape keeps the exact field names the deployed kiosk + * already reads ({@code type}, {@code reason}, {@code deviceId}, {@code at}); {@code seq} and + * {@code sessionId} are additive and safe for an older reader to ignore. + */ +@Data +@Builder +@JsonInclude(JsonInclude.Include.NON_NULL) +public class OutgoingBoardStreamFrame { + + private String type; + private long seq; + private UUID sessionId; + + // ── hello ──────────────────────────────────────────────────────────── + /** Base64-encoded 32 random bytes. Single-use, scoped to this session. */ + private String challenge; + private Long serverTime; + + // ── auth_ok / refresh ──────────────────────────────────────────────── + /** On auth_ok: the device this connection is now bound to. On refresh: the addressed + * board, or absent for a broadcast. */ + private UUID deviceId; + + // ── refresh ────────────────────────────────────────────────────────── + /** Why the board should refetch. */ + private String reason; + /** + * ISO-8601 instant, pre-formatted as a String rather than left as an {@code Instant}: a + * mapper without JavaTimeModule throws on one, and the only symptom would be pushes + * silently going nowhere. + */ + private String at; +} diff --git a/src/main/java/com/ibrasoft/lensbridge/handler/BoardStreamHandler.java b/src/main/java/com/ibrasoft/lensbridge/handler/BoardStreamHandler.java index c0f9595..5787bf3 100644 --- a/src/main/java/com/ibrasoft/lensbridge/handler/BoardStreamHandler.java +++ b/src/main/java/com/ibrasoft/lensbridge/handler/BoardStreamHandler.java @@ -1,99 +1,294 @@ package com.ibrasoft.lensbridge.handler; import com.fasterxml.jackson.databind.ObjectMapper; +import com.ibrasoft.lensbridge.dto.board.stream.BoardStreamAuthFrame; +import com.ibrasoft.lensbridge.dto.board.stream.IncomingBoardStreamFrame; +import com.ibrasoft.lensbridge.dto.board.stream.OutgoingBoardStreamFrame; import com.ibrasoft.lensbridge.model.minbar.board.Device; import com.ibrasoft.lensbridge.repository.sql.DeviceRepository; +import com.ibrasoft.lensbridge.service.agent.handshake.Ed25519Verifier; +import com.ibrasoft.lensbridge.service.board.stream.BoardStreamAuthPayload; +import com.ibrasoft.lensbridge.service.board.stream.BoardStreamSession; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Value; import org.springframework.lang.NonNull; +import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Component; import org.springframework.web.socket.CloseStatus; import org.springframework.web.socket.TextMessage; import org.springframework.web.socket.WebSocketSession; import org.springframework.web.socket.handler.TextWebSocketHandler; +import java.security.SecureRandom; import java.time.Instant; -import java.util.LinkedHashMap; +import java.util.Base64; import java.util.Map; import java.util.Optional; import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; /** - * Push channel for kiosk pages: tells a board its content or config changed so it can + * Push channel for kiosk boards: tells a board its content or config changed so it can * refetch without waiting out its ten-minute poll. *

- * Every board on this channel is an enrolled device. A connection must name a known, - * unrevoked {@code ?deviceId=} or it is closed on the spot — there is no anonymous - * or pre-enrollment state here. A board that has not been enrolled has no business on the - * stream: it has no payload to refetch and nothing can be addressed to it. + * Every board on this channel is an enrolled device, and it must prove it — the + * device id alone is an identifier, not a credential. It is printed in the board's own URL, + * visible on the physical display, and listed by {@code GET /api/admin/board/devices}, so + * accepting it as proof let anyone who had ever glanced at a screen evict that kiosk from + * the stream and read its pushes. *

+ * The handshake mirrors {@link AgentWebSocketHandler} exactly: + * 

+ *   open      → server generates a 32-byte challenge, sends hello (seq:1)
+ *   recv auth → verify Ed25519 sig over sessionId+challenge+deviceId+timestamp, check skew
+ *               and revocation; on success bind deviceId and send auth_ok (seq:2)
+ *   (no auth) → closed 4008 once the auth deadline passes
+ *   recv ?    → close 4001 (auth) / 4002 (seq replay) / 4003 (bad-state) / 4004 (bad frame)
+ *   close     → deregister
+ *
+ * The verifying public key always comes from the enrolled {@link Device} row; nothing in the + * incoming frame can influence which key is used. */ @Component @RequiredArgsConstructor @Slf4j public class BoardStreamHandler extends TextWebSocketHandler { - // Should probably be enums but this is sufficient for now - static final CloseStatus CLOSE_NO_DEVICE = new CloseStatus(4001, "device_id_required"); - static final CloseStatus CLOSE_UNKNOWN_DEVICE = new CloseStatus(4002, "unknown_device"); - static final CloseStatus CLOSE_REVOKED = new CloseStatus(4003, "device_revoked"); + static final int CHALLENGE_BYTES = 32; + static final long TIMESTAMP_SKEW_MS = 5 * 60 * 1000L; + /** An auth frame is ~300 characters; anything near this is not a board. */ + static final int MAX_INCOMING_FRAME_CHARS = 4096; + + static final CloseStatus CLOSE_AUTH_FAILED = new CloseStatus(4001, "auth_failed"); + static final CloseStatus CLOSE_SEQ_VIOLATION = new CloseStatus(4002, "seq_violation"); + static final CloseStatus CLOSE_BAD_STATE = new CloseStatus(4003, "bad_state"); + static final CloseStatus CLOSE_BAD_FRAME = new CloseStatus(4004, "bad_frame"); + static final CloseStatus CLOSE_AUTH_TIMEOUT = new CloseStatus(4008, "auth_timeout"); static final CloseStatus CLOSE_SUPERSEDED = CloseStatus.NORMAL.withReason("superseded"); - /** Session attribute holding the device a connection was accepted as. */ - private static final String DEVICE_ATTR = "board.deviceId"; + /** Session attribute holding this connection's server-side state. */ + private static final String SESSION_ATTR = "board.stream.session"; private final ObjectMapper objectMapper; + private final Ed25519Verifier ed25519; private final DeviceRepository deviceRepository; + private final SecureRandom random = new SecureRandom(); /** - * Live boards keyed by device. One session per device: a reloading kiosk can briefly - * open its replacement before the old socket finishes closing, so the newcomer wins - * and the stale one is closed rather than left to accumulate. + * How long a connection may sit unauthenticated before it is closed. Without this an + * anonymous client could hold sockets open indefinitely and never prove anything. */ - private final Map sessions = new ConcurrentHashMap<>(); + @Value("${musallahboard.stream.authTimeoutMs:15000}") + private long authTimeoutMs = 15_000L; + + /** + * Authenticated boards keyed by device. One session per device: a reloading kiosk can + * briefly open its replacement before the old socket finishes closing, so the newcomer + * wins and the stale one is closed rather than left to accumulate. + */ + private final Map sessions = new ConcurrentHashMap<>(); + + /** Connections that have not authenticated yet, keyed by their server-issued session id. */ + private final Map pending = new ConcurrentHashMap<>(); + + @Override + public void afterConnectionEstablished(@NonNull WebSocketSession transport) { + String challenge = newChallenge(); + BoardStreamSession session = new BoardStreamSession( + transport, challenge, objectMapper, System.currentTimeMillis() + authTimeoutMs); + transport.getAttributes().put(SESSION_ATTR, session); + pending.put(session.getSessionId(), session); + + session.send(OutgoingBoardStreamFrame.builder() + .type("hello") + .challenge(challenge) + .serverTime(System.currentTimeMillis())); + } @Override - public void afterConnectionEstablished(@NonNull WebSocketSession session) { - UUID deviceId = parseDeviceId(session); - if (deviceId == null) { - close(session, CLOSE_NO_DEVICE); + protected void handleTextMessage(@NonNull WebSocketSession transport, @NonNull TextMessage message) { + BoardStreamSession session = (BoardStreamSession) transport.getAttributes().get(SESSION_ATTR); + if (session == null) { + safeClose(transport, CLOSE_BAD_STATE); + return; + } + + // Nothing this channel accepts is large — an auth frame is a few hundred characters. + // The container's text buffer is sized for the media paths (10 MB when the customizer + // is on), which is not a budget an unauthenticated caller should get to spend. + // Measured in characters rather than bytes on purpose: TextMessage.getPayloadLength() + // UTF-8-encodes the whole payload to answer, allocating a second copy of exactly the + // oversized frame we are refusing to spend memory on. + if (message.getPayload().length() > MAX_INCOMING_FRAME_CHARS) { + log.warn("board stream session={} sent an oversized frame ({} chars)", + session.getSessionId(), message.getPayload().length()); + closeAndDeregister(session, CLOSE_BAD_FRAME); + return; + } + + IncomingBoardStreamFrame frame; + try { + frame = objectMapper.readValue(message.getPayload(), IncomingBoardStreamFrame.class); + } catch (Exception e) { + log.warn("board stream session={} unparseable frame: {}", session.getSessionId(), e.getMessage()); + closeAndDeregister(session, CLOSE_BAD_FRAME); + return; + } + + if (frame.getSessionId() == null || !frame.getSessionId().equals(session.getSessionId())) { + log.warn("board stream session={} frame sessionId mismatch (got {})", + session.getSessionId(), frame.getSessionId()); + closeAndDeregister(session, CLOSE_AUTH_FAILED); + return; + } + if (!session.acceptIncomingSeq(frame.getSeq())) { + closeAndDeregister(session, CLOSE_SEQ_VIOLATION); + return; + } + + switch (session.getPhase()) { + case UNAUTH -> { + if (frame instanceof BoardStreamAuthFrame auth) { + handleAuth(session, auth); + } else { + log.warn("board stream session={} expected auth frame, got {}", + session.getSessionId(), frame.getType()); + closeAndDeregister(session, CLOSE_BAD_STATE); + } + } + case AUTHED -> { + // Nothing an authenticated board can usefully say on this channel yet; a + // second auth is a protocol error rather than a re-bind opportunity. + log.warn("board stream session={} sent {} after auth_ok", + session.getSessionId(), frame.getType()); + closeAndDeregister(session, CLOSE_BAD_STATE); + } + case CLOSED -> { /* drop */ } + } + } + + private void handleAuth(BoardStreamSession session, BoardStreamAuthFrame frame) { + long now = System.currentTimeMillis(); + if (Math.abs(now - frame.getTimestamp()) > TIMESTAMP_SKEW_MS) { + log.warn("board stream session={} timestamp skew {} ms (limit {})", + session.getSessionId(), now - frame.getTimestamp(), TIMESTAMP_SKEW_MS); + closeAndDeregister(session, CLOSE_AUTH_FAILED); + return; + } + if (frame.getDeviceId() == null || frame.getSignature() == null) { + closeAndDeregister(session, CLOSE_AUTH_FAILED); + return; + } + + // Unknown, revoked, keyless and badly-signed all close the same way on purpose: the + // close code must not tell an unauthenticated caller which device ids are real. + Optional deviceOpt = deviceRepository.findById(frame.getDeviceId()); + if (deviceOpt.isEmpty()) { + log.warn("board stream session={} named unknown device {}", session.getSessionId(), frame.getDeviceId()); + closeAndDeregister(session, CLOSE_AUTH_FAILED); + return; + } + Device device = deviceOpt.get(); + if (device.getRevokedAt() != null) { + log.warn("board stream session={} revoked device {} attempting to auth", + session.getSessionId(), device.getId()); + closeAndDeregister(session, CLOSE_AUTH_FAILED); + return; + } + if (device.getPublicKey() == null) { + log.error("board stream session={} device {} has no public key on file", + session.getSessionId(), device.getId()); + closeAndDeregister(session, CLOSE_AUTH_FAILED); return; } - Optional device = deviceRepository.findById(deviceId); - if (device.isEmpty()) { - log.warn("Board stream session={} named unknown device {}", session.getId(), deviceId); - close(session, CLOSE_UNKNOWN_DEVICE); + byte[] signed = BoardStreamAuthPayload.build( + session.getSessionId(), + session.getChallenge(), + frame.getDeviceId(), + frame.getTimestamp()); + byte[] signature; + try { + signature = Base64.getDecoder().decode(frame.getSignature()); + } catch (IllegalArgumentException e) { + closeAndDeregister(session, CLOSE_AUTH_FAILED); return; } - if (device.get().getRevokedAt() != null) { - log.warn("Board stream session={} is a revoked device {}", session.getId(), deviceId); - close(session, CLOSE_REVOKED); + + if (!ed25519.verify(device.getPublicKey(), signed, signature)) { + log.warn("board stream session={} bad Ed25519 signature for device {}", + session.getSessionId(), device.getId()); + closeAndDeregister(session, CLOSE_AUTH_FAILED); + return; + } + + if (!session.markAuthenticated(device.getId())) { + // The transport dropped while we were verifying; the close callback already ran. + log.debug("board stream session={} closed before auth could be bound", session.getSessionId()); + deregister(session); return; } + pending.remove(session.getSessionId(), session); - session.getAttributes().put(DEVICE_ATTR, deviceId); - WebSocketSession previous = sessions.put(deviceId, session); + BoardStreamSession previous = sessions.put(device.getId(), session); if (previous != null && previous != session) { - log.info("Board stream device {} superseded session {} with {}", - deviceId, previous.getId(), session.getId()); - close(previous, CLOSE_SUPERSEDED); + log.info("board stream device {} superseded session {} with {}", + device.getId(), previous.getSessionId(), session.getSessionId()); + previous.close(CLOSE_SUPERSEDED); } - log.info("Board stream connected: session={} device={}", session.getId(), deviceId); + log.info("board stream session={} authenticated as device {}", session.getSessionId(), device.getId()); + + session.send(OutgoingBoardStreamFrame.builder() + .type("auth_ok") + .deviceId(device.getId()) + .serverTime(now)); } @Override - public void afterConnectionClosed(@NonNull WebSocketSession session, @NonNull CloseStatus status) { - UUID deviceId = (UUID) session.getAttributes().get(DEVICE_ATTR); - if (deviceId == null) return; // rejected before it was ever registered - - // Two-arg remove: a superseded session closing later must not evict the - // replacement that has already taken its place. - sessions.remove(deviceId, session); - log.info("Board stream disconnected: session={} device={} status={}", - session.getId(), deviceId, status); + public void afterConnectionClosed(@NonNull WebSocketSession transport, @NonNull CloseStatus status) { + BoardStreamSession session = (BoardStreamSession) transport.getAttributes().get(SESSION_ATTR); + if (session == null) return; + + session.markClosed(); + deregister(session); + log.info("board stream disconnected: session={} device={} status={}", + session.getSessionId(), session.getDeviceId(), status); + } + + /** + * Closes connections that connected but never proved who they were. Runs on a timer + * rather than a per-session task so one sweep covers the whole fleet. + */ + @Scheduled(fixedDelayString = "${musallahboard.stream.authReaperIntervalMs:5000}") + public void reapUnauthenticatedSessions() { + long now = System.currentTimeMillis(); + for (BoardStreamSession session : pending.values()) { + if (session.getPhase() == BoardStreamSession.Phase.UNAUTH && now > session.getAuthDeadlineMs()) { + log.warn("board stream session={} never authenticated within {} ms; closing", + session.getSessionId(), authTimeoutMs); + closeAndDeregister(session, CLOSE_AUTH_TIMEOUT); + } else if (session.getPhase() == BoardStreamSession.Phase.CLOSED) { + pending.remove(session.getSessionId(), session); + } + } + } + + /** + * Closes a device's live stream session, if it has one. + *

+ * Revocation is checked at handshake time, so without this a board revoked mid-session + * would keep receiving pushes until it happened to reconnect. The agent channel already + * closes on revoke via {@code AgentSessionRegistry.closeIfPresent}; now that this channel + * also binds a device identity, it owes the same guarantee. + */ + public void closeIfPresent(UUID deviceId, CloseStatus status) { + BoardStreamSession session = sessions.remove(deviceId); + if (session != null) { + log.info("board stream closing session {} for device {}", session.getSessionId(), deviceId); + session.close(status); + } } /** Content every board shares — posters, events, weekly content. */ @@ -112,89 +307,74 @@ public void refreshAll() { } /** - * Push a message to every connected board, or to a single one. + * Push a message to every authenticated board, or to a single one. * * @param reason why the board should refetch. Carried on the wire for logging and for * a frontend that wants to react selectively. * @param target device to notify, or null to reach every board. */ private void push(String reason, UUID target) { - String json = serialize(reason, target); - if (json == null) return; + String at = Instant.now().toString(); if (target != null) { - WebSocketSession session = sessions.get(target); + BoardStreamSession session = sessions.get(target); if (session == null) { - log.debug("Board stream reason={} device={} not connected — it will pick the change up on its next poll", + log.debug("board stream reason={} device={} not connected — it will pick the change up on its next poll", reason, target); return; } - if (send(target, session, json)) { - log.debug("Board stream push reason={} device={}", reason, target); + if (send(target, session, reason, target, at)) { + log.debug("board stream push reason={} device={}", reason, target); } return; } int sent = 0; - for (Map.Entry entry : sessions.entrySet()) { - if (send(entry.getKey(), entry.getValue(), json)) sent++; + for (Map.Entry entry : sessions.entrySet()) { + if (send(entry.getKey(), entry.getValue(), reason, null, at)) sent++; } - log.debug("Board stream broadcast reason={} boards={}", reason, sent); + log.debug("board stream broadcast reason={} boards={}", reason, sent); } /** @return true when the frame was written; drops the session if it has gone away. */ - private boolean send(UUID deviceId, WebSocketSession session, String json) { - if (!session.isOpen()) { + private boolean send(UUID deviceId, BoardStreamSession session, String reason, UUID target, String at) { + if (!session.getTransport().isOpen()) { sessions.remove(deviceId, session); return false; } - try { - session.sendMessage(new TextMessage(json)); - return true; - } catch (Exception e) { - log.warn("Board stream send failed: session={} device={} err={}", - session.getId(), deviceId, e.getMessage()); - return false; - } + return session.send(OutgoingBoardStreamFrame.builder() + .type("refresh") + .reason(reason) + .deviceId(target) + .at(at)); } - private String serialize(String reason, UUID target) { - Map body = new LinkedHashMap<>(); - body.put("type", "refresh"); - body.put("reason", reason); - body.put("deviceId", target); - // Pre-formatted rather than a raw Instant: a mapper without JavaTimeModule throws on - // one, and the only symptom would be pushes silently going nowhere. - body.put("at", Instant.now().toString()); - try { - return objectMapper.writeValueAsString(body); - } catch (Exception e) { - log.error("Board stream message could not be serialized", e); - return null; - } + private void closeAndDeregister(BoardStreamSession session, CloseStatus status) { + session.close(status); + deregister(session); } - private static UUID parseDeviceId(WebSocketSession session) { - String query = session.getUri() == null ? null : session.getUri().getQuery(); - if (query == null) return null; - for (String param : query.split("&")) { - if (!param.startsWith("deviceId=")) continue; - String raw = param.substring("deviceId=".length()); - try { - return UUID.fromString(raw); - } catch (IllegalArgumentException e) { - log.warn("Board stream session={} sent a malformed deviceId: {}", session.getId(), raw); - return null; - } - } - return null; + /** + * Two-arg removes throughout: a superseded session closing later must not evict the + * replacement that has already taken its place. + */ + private void deregister(BoardStreamSession session) { + pending.remove(session.getSessionId(), session); + UUID deviceId = session.getDeviceId(); + if (deviceId != null) sessions.remove(deviceId, session); + } + + private String newChallenge() { + byte[] raw = new byte[CHALLENGE_BYTES]; + random.nextBytes(raw); + return Base64.getEncoder().withoutPadding().encodeToString(raw); } - private static void close(WebSocketSession session, CloseStatus status) { + private static void safeClose(WebSocketSession transport, CloseStatus status) { try { - if (session.isOpen()) session.close(status); + if (transport.isOpen()) transport.close(status); } catch (Exception e) { - log.debug("Board stream close failed: session={} err={}", session.getId(), e.getMessage()); + log.debug("board stream close failed: session={} err={}", transport.getId(), e.getMessage()); } } } diff --git a/src/main/java/com/ibrasoft/lensbridge/service/board/stream/BoardStreamAuthPayload.java b/src/main/java/com/ibrasoft/lensbridge/service/board/stream/BoardStreamAuthPayload.java new file mode 100644 index 0000000..f9cfb01 --- /dev/null +++ b/src/main/java/com/ibrasoft/lensbridge/service/board/stream/BoardStreamAuthPayload.java @@ -0,0 +1,40 @@ +package com.ibrasoft.lensbridge.service.board.stream; + +import java.nio.charset.StandardCharsets; +import java.util.UUID; + +/** + * Canonical byte sequence a board signs to join the refresh stream. + *

+ * Deliberately a different domain string from the agent channel's + * {@link com.ibrasoft.lensbridge.service.agent.handshake.AuthSignaturePayload}: a signature + * produced for one channel must never verify on the other. The server-issued + * {@code sessionId} and {@code challenge} already make cross-channel replay fail, but domain + * separation is the backstop if either channel's handshake is ever relaxed. + * + * 

+ * musallahboard-stream-auth-v1\n
+ * <sessionId>\n
+ * <challenge-base64>\n
+ * <deviceId>\n
+ * <timestamp-millis>
+ *
+ * + * Both sides MUST construct it identically. The version prefix lets the format change later + * without silently accepting older agents — they will simply fail verification. + */ +public final class BoardStreamAuthPayload { + + public static final String VERSION_PREFIX = "musallahboard-stream-auth-v1"; + + private BoardStreamAuthPayload() {} + + public static byte[] build(UUID sessionId, String challengeBase64, UUID deviceId, long timestampMillis) { + String s = VERSION_PREFIX + "\n" + + sessionId + "\n" + + challengeBase64 + "\n" + + deviceId + "\n" + + timestampMillis; + return s.getBytes(StandardCharsets.UTF_8); + } +} diff --git a/src/main/java/com/ibrasoft/lensbridge/service/board/stream/BoardStreamSession.java b/src/main/java/com/ibrasoft/lensbridge/service/board/stream/BoardStreamSession.java new file mode 100644 index 0000000..ffd53d4 --- /dev/null +++ b/src/main/java/com/ibrasoft/lensbridge/service/board/stream/BoardStreamSession.java @@ -0,0 +1,150 @@ +package com.ibrasoft.lensbridge.service.board.stream; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.ibrasoft.lensbridge.dto.board.stream.OutgoingBoardStreamFrame; +import lombok.Getter; +import lombok.extern.slf4j.Slf4j; +import org.springframework.web.socket.CloseStatus; +import org.springframework.web.socket.TextMessage; +import org.springframework.web.socket.WebSocketSession; + +import java.io.IOException; +import java.util.UUID; +import java.util.concurrent.atomic.AtomicLong; + +/** + * Server-side state for one open board refresh-stream connection. + *

+ * Mirrors {@link com.ibrasoft.lensbridge.service.agent.AgentSession}: a connection carries no + * authority until an Ed25519 challenge–response succeeds, and the device it speaks for is + * decided here, not by anything the client asserts in a URL. + *

    + *
  1. UNAUTH — connected, server has issued a {@code hello} with a challenge. Only an + * {@code auth} frame is allowed, and only until {@code authDeadline} passes.
    2. + *
  2. AUTHED — signature verified; {@link #deviceId} is bound and the session is + * eligible to receive pushes.
    3. + *
  3. CLOSED — underlying transport closed.
    4. + *
+ */ +@Slf4j +public class BoardStreamSession { + + public enum Phase { UNAUTH, AUTHED, CLOSED } + + @Getter private final UUID sessionId; + @Getter private final WebSocketSession transport; + @Getter private final String challenge; + /** Epoch millis after which an unauthenticated session is reaped. */ + @Getter private final long authDeadlineMs; + + private final ObjectMapper objectMapper; + + private final AtomicLong outgoingSeq = new AtomicLong(0); + + /** Last seq received from the client (must be strictly increasing). 0 means none yet. */ + private long lastIncomingSeq = 0; + + @Getter private volatile Phase phase = Phase.UNAUTH; + @Getter private volatile UUID deviceId; + + public BoardStreamSession(WebSocketSession transport, + String challenge, + ObjectMapper objectMapper, + long authDeadlineMs) { + this.transport = transport; + this.objectMapper = objectMapper; + this.sessionId = UUID.randomUUID(); + this.challenge = challenge; + this.authDeadlineMs = authDeadlineMs; + } + + /** Rejects replayed or regressed sequence numbers. */ + public synchronized boolean acceptIncomingSeq(long seq) { + if (seq <= lastIncomingSeq) { + log.warn("board stream session={} rejecting non-monotonic seq {} (last seen {})", + sessionId, seq, lastIncomingSeq); + return false; + } + lastIncomingSeq = seq; + return true; + } + + /** + * Binds the device, but only from UNAUTH. + *

+ * Returns false rather than throwing: the transport can drop while the handshake is + * mid-verification, in which case the close callback has already moved this session to + * CLOSED and the caller simply has nothing left to do. Throwing there would escape + * {@code handleTextMessage} as an unchecked error for an entirely ordinary disconnect. + * + * @return true if this call performed the transition + */ + public synchronized boolean markAuthenticated(UUID deviceId) { + if (phase != Phase.UNAUTH) return false; + this.deviceId = deviceId; + this.phase = Phase.AUTHED; + return true; + } + + public synchronized void markClosed() { + this.phase = Phase.CLOSED; + } + + /** + * Allocates the next outbound seq and writes the frame. + *

+ * Seq allocation happens inside the transport monitor, not before it. Two request + * threads can push to the same board at once (a poster edit and a config change), and + * allocating outside the lock lets them interleave so that seq 5 reaches the wire ahead + * of seq 4 — which is exactly the regression a client validating the sequence would + * reconnect over. + * + * @return true when the frame reached the wire. + */ + public boolean send(OutgoingBoardStreamFrame.OutgoingBoardStreamFrameBuilder builder) { + String type = null; + synchronized (transport) { + if (!transport.isOpen()) { + log.debug("board stream session={} transport closed; frame was not sent", sessionId); + markClosed(); + return false; + } + OutgoingBoardStreamFrame frame = builder + .seq(outgoingSeq.incrementAndGet()) + .sessionId(sessionId) + .build(); + type = frame.getType(); + try { + transport.sendMessage(new TextMessage(objectMapper.writeValueAsString(frame))); + return true; + } catch (JsonProcessingException e) { + log.error("board stream session={} failed to serialize frame {}", sessionId, type, e); + close(CloseStatus.SERVER_ERROR); + return false; + } catch (IOException e) { + log.warn("board stream session={} transport send failed: {}", sessionId, e.getMessage()); + close(CloseStatus.SESSION_NOT_RELIABLE); + return false; + } catch (Exception e) { + // Tomcat's endpoint throws unchecked IllegalStateException when the socket + // moves to a bad state between the isOpen() check and the write. One dying + // board must not abort a fleet-wide broadcast, nor fail the admin's HTTP + // request that triggered it. + log.warn("board stream session={} send failed on frame {}: {}", sessionId, type, e.toString()); + close(CloseStatus.SESSION_NOT_RELIABLE); + return false; + } + } + } + + public void close(CloseStatus status) { + try { + if (transport.isOpen()) transport.close(status); + } catch (IOException ignored) { + // best effort + } finally { + markClosed(); + } + } +} diff --git a/src/main/resources/application.properties.example b/src/main/resources/application.properties.example index 165d227..f080cd6 100644 --- a/src/main/resources/application.properties.example +++ b/src/main/resources/application.properties.example @@ -45,6 +45,11 @@ musallahboard.enrollment.tokenPepper= musallahboard.defaultTimezone=America/Toronto # How often undelivered device commands are expired and in-flight ones timed out. musallahboard.command.reaperIntervalMs=60000 +# Board refresh stream (/api/refresh-musallahboard). A connection has no authority until it +# completes the Ed25519 challenge-response; these bound how long it may sit there trying. +# The deadline is enforced by a sweep, so the effective ceiling is the sum of the two. +musallahboard.stream.authTimeoutMs=15000 +musallahboard.stream.authReaperIntervalMs=5000 # Cloudflare R2 cloudflare.r2.endpoint=https://.r2.cloudflarestorage.com diff --git a/src/test/java/com/ibrasoft/lensbridge/controller/DeviceAdminControllerTest.java b/src/test/java/com/ibrasoft/lensbridge/controller/DeviceAdminControllerTest.java index 64a7cec..6afae32 100644 --- a/src/test/java/com/ibrasoft/lensbridge/controller/DeviceAdminControllerTest.java +++ b/src/test/java/com/ibrasoft/lensbridge/controller/DeviceAdminControllerTest.java @@ -4,6 +4,7 @@ import com.ibrasoft.lensbridge.model.audit.AuditAction; import com.ibrasoft.lensbridge.model.minbar.Audience; import com.ibrasoft.lensbridge.model.minbar.board.Device; +import com.ibrasoft.lensbridge.handler.BoardStreamHandler; import com.ibrasoft.lensbridge.service.AdminAuditService; import com.ibrasoft.lensbridge.repository.sql.DeviceCommandRepository; import com.ibrasoft.lensbridge.repository.sql.DeviceRepository; @@ -43,6 +44,7 @@ void revoke_closesRegisteredAgentSession() throws Exception { DeviceCommandRepository commandRepository = mock(DeviceCommandRepository.class); CommandDispatcher commandDispatcher = mock(CommandDispatcher.class); AgentSessionRegistry registry = new AgentSessionRegistry(); + BoardStreamHandler boardStream = mock(BoardStreamHandler.class); AdminAuditService auditService = mock(AdminAuditService.class); DeviceAdminController controller = new DeviceAdminController( @@ -52,6 +54,7 @@ void revoke_closesRegisteredAgentSession() throws Exception { commandDispatcher, new ObjectMapper(), registry, + boardStream, auditService); UUID deviceId = UUID.randomUUID(); @@ -81,6 +84,11 @@ void revoke_closesRegisteredAgentSession() throws Exception { verify(transport).close(argThat(status -> status.getCode() == CloseStatus.POLICY_VIOLATION.getCode() && "device_revoked".equals(status.getReason()))); + // Revocation must drop the refresh stream too, not just the command channel — + // otherwise a revoked board keeps receiving pushes until it happens to reconnect. + verify(boardStream).closeIfPresent(eq(deviceId), argThat(status -> + status.getCode() == CloseStatus.POLICY_VIOLATION.getCode() + && "device_revoked".equals(status.getReason()))); verify(auditService).logAuditEvent( eq("root@example.com"), eq(AuditAction.REVOKE_DEVICE), eq("Device"), eq(deviceId), eq("10.0.0.1")); @@ -98,6 +106,7 @@ void revoke_isANoOpAndDoesNotReAuditWhenAlreadyRevoked() throws Exception { mock(CommandDispatcher.class), new ObjectMapper(), new AgentSessionRegistry(), + mock(BoardStreamHandler.class), auditService); UUID deviceId = UUID.randomUUID(); diff --git a/src/test/java/com/ibrasoft/lensbridge/handler/BoardStreamHandlerTest.java b/src/test/java/com/ibrasoft/lensbridge/handler/BoardStreamHandlerTest.java index dc901d0..2b5b059 100644 --- a/src/test/java/com/ibrasoft/lensbridge/handler/BoardStreamHandlerTest.java +++ b/src/test/java/com/ibrasoft/lensbridge/handler/BoardStreamHandlerTest.java @@ -4,17 +4,27 @@ import com.fasterxml.jackson.databind.ObjectMapper; import com.ibrasoft.lensbridge.model.minbar.board.Device; import com.ibrasoft.lensbridge.repository.sql.DeviceRepository; +import com.ibrasoft.lensbridge.service.agent.Ed25519TestUtil; +import com.ibrasoft.lensbridge.service.agent.handshake.AuthSignaturePayload; +import com.ibrasoft.lensbridge.service.agent.handshake.Ed25519Verifier; +import com.ibrasoft.lensbridge.service.board.stream.BoardStreamAuthPayload; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import org.springframework.test.util.ReflectionTestUtils; import org.springframework.web.socket.CloseStatus; import org.springframework.web.socket.TextMessage; import org.springframework.web.socket.WebSocketSession; import java.net.URI; +import java.security.KeyPair; +import java.security.PrivateKey; import java.time.Instant; import java.util.ArrayList; +import java.util.Base64; import java.util.HashMap; +import java.util.LinkedHashMap; import java.util.List; +import java.util.Map; import java.util.Optional; import java.util.UUID; @@ -24,6 +34,13 @@ import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; +/** + * The refresh stream used to accept anyone who could name a device id. A device id is + * printed in the board's own URL and listed by the admin API, so that let a stranger evict + * a physical kiosk from the channel and read its pushes. These tests pin the replacement: + * nothing is delivered until an Ed25519 challenge–response over the enrolled device key + * succeeds, and everything the old handler got right about session lifecycle still holds. + */ class BoardStreamHandlerTest { private final ObjectMapper mapper = new ObjectMapper(); @@ -34,112 +51,360 @@ class BoardStreamHandlerTest { void setUp() { deviceRepository = mock(DeviceRepository.class); when(deviceRepository.findById(any())).thenReturn(Optional.empty()); - handler = new BoardStreamHandler(mapper, deviceRepository); + handler = new BoardStreamHandler(mapper, new Ed25519Verifier(), deviceRepository); + } + + // ── handshake ───────────────────────────────────────────────────────── + + @Test + void connectingIssuesAHelloCarryingAChallenge() throws Exception { + Session s = connect(); + + assertThat(s.sent).hasSize(1); + JsonNode hello = mapper.readTree(s.sent.get(0)); + assertThat(hello.get("type").asText()).isEqualTo("hello"); + assertThat(hello.get("seq").asLong()).isEqualTo(1); + assertThat(UUID.fromString(hello.get("sessionId").asText())).isNotNull(); + // 32 random bytes, base64 without padding. + assertThat(Base64.getDecoder().decode(hello.get("challenge").asText())).hasSize(32); + assertThat(hello.get("serverTime").asLong()).isGreaterThan(0); } @Test - void contentPushReachesEveryConnectedBoard() throws Exception { - Session a = connect(enrolled()); - Session b = connect(enrolled()); + void authenticatedBoardReceivesAuthOkAndThenPushes() throws Exception { + Board board = enrolled(); + Session s = connect(); + + authenticate(s, board); + + assertThat(s.open).isTrue(); + assertThat(s.sent).hasSize(2); + JsonNode ok = mapper.readTree(s.sent.get(1)); + assertThat(ok.get("type").asText()).isEqualTo("auth_ok"); + assertThat(ok.get("seq").asLong()).isEqualTo(2); + assertThat(ok.get("deviceId").asText()).isEqualTo(board.id.toString()); handler.contentChanged("posters"); - assertThat(a.sent).hasSize(1); - assertThat(b.sent).hasSize(1); + assertThat(s.sent).hasSize(3); + assertThat(mapper.readTree(s.sent.get(2)).get("type").asText()).isEqualTo("refresh"); } + // ── the hole this change closes ─────────────────────────────────────── + + /** + * The exact pre-fix exploit: name a real device id, receive that board's stream. The + * device id is no longer accepted anywhere, so there is nothing to name. + */ @Test - void configPushReachesOnlyTheTargetBoard() throws Exception { - UUID target = enrolled(); - Session mine = connect(target); - Session theirs = connect(enrolled()); + void namingARealDeviceIdWithoutSigningDeliversNothing() throws Exception { + Board board = enrolled(); + Session attacker = connectWithQuery("deviceId=" + board.id); - handler.configChanged(target); + handler.contentChanged("posters"); + handler.configChanged(board.id); + handler.refreshAll(); - assertThat(mine.sent).hasSize(1); - assertThat(theirs.sent).isEmpty(); + // Only the hello it was issued on connect; no refresh ever reaches it. + assertThat(attacker.sent).hasSize(1); + assertThat(mapper.readTree(attacker.sent.get(0)).get("type").asText()).isEqualTo("hello"); } + /** + * The denial-of-service half: an unauthenticated connection naming a live board's id + * must not supersede it. + */ @Test - void messageIsJsonCarryingTypeReasonAndTarget() throws Exception { - UUID target = enrolled(); - Session s = connect(target); + void unauthenticatedConnectionCannotSupersedeALiveBoard() throws Exception { + Board board = enrolled(); + Session kiosk = connect(); + authenticate(kiosk, board); - handler.configChanged(target); + Session attacker = connectWithQuery("deviceId=" + board.id); - JsonNode body = mapper.readTree(s.sent.get(0)); - assertThat(body.get("type").asText()).isEqualTo("refresh"); - assertThat(body.get("reason").asText()).isEqualTo("config"); - assertThat(body.get("deviceId").asText()).isEqualTo(target.toString()); - assertThat(body.hasNonNull("at")).isTrue(); + assertThat(kiosk.closedWith).isNull(); + assertThat(kiosk.open).isTrue(); + + handler.configChanged(board.id); + + assertThat(mapper.readTree(kiosk.sent.get(kiosk.sent.size() - 1)).get("reason").asText()) + .isEqualTo("config"); + assertThat(attacker.sent).hasSize(1); // hello only } - /** The deployed kiosk substring-matches rather than parsing, so JSON must still hit it. */ + // ── rejected credentials ────────────────────────────────────────────── + @Test - void messageStillMatchesTheLegacyRefreshSubstring() throws Exception { - Session s = connect(enrolled()); + void signatureFromAKeyTheDeviceDoesNotOwnIsRejected() throws Exception { + Board board = enrolled(); + KeyPair impostor = Ed25519TestUtil.generate(); + Session s = connect(); - handler.refreshAll(); + authenticate(s, board.id, impostor.getPrivate(), System.currentTimeMillis()); - assertThat(s.sent.get(0).toUpperCase()).contains("REFRESH"); + assertThat(s.closedWith.getCode()).isEqualTo(4001); + assertRejected(s, board.id); } - // ── enrollment is a precondition, not a state ───────────────────────── + @Test + void unknownDeviceIsRejectedWithoutRevealingThatItIsUnknown() throws Exception { + KeyPair kp = Ed25519TestUtil.generate(); + Session s = connect(); + + authenticate(s, UUID.randomUUID(), kp.getPrivate(), System.currentTimeMillis()); + + // Same 4001 as a bad signature: the close code must not be a device-id oracle. + assertThat(s.closedWith.getCode()).isEqualTo(4001); + } @Test - void connectionWithoutADeviceIdIsRejected() throws Exception { - Session s = connectWithQuery(null); + void revokedDeviceIsRejected() throws Exception { + Board board = enrolled(); + board.device.setRevokedAt(Instant.now().minusSeconds(60)); + Session s = connect(); + + authenticate(s, board); - assertThat(s.closedWith).isNotNull(); assertThat(s.closedWith.getCode()).isEqualTo(4001); + assertRejected(s, board.id); + } - handler.contentChanged("events"); - assertThat(s.sent).isEmpty(); + @Test + void deviceWithNoPublicKeyOnFileIsRejected() throws Exception { + Board board = enrolled(); + board.device.setPublicKey(null); + Session s = connect(); + + authenticate(s, board); + + assertThat(s.closedWith.getCode()).isEqualTo(4001); + } + + @Test + void staleTimestampIsRejected() throws Exception { + Board board = enrolled(); + Session s = connect(); + + long tenMinutesAgo = System.currentTimeMillis() - (10 * 60 * 1000L); + authenticate(s, board.id, board.keyPair.getPrivate(), tenMinutesAgo); + + assertThat(s.closedWith.getCode()).isEqualTo(4001); + } + + /** A signature is bound to the session that issued the challenge; it does not travel. */ + @Test + void aSignatureCapturedOnOneSessionDoesNotWorkOnAnother() throws Exception { + Board board = enrolled(); + Session first = connect(); + Session second = connect(); + + long now = System.currentTimeMillis(); + byte[] signed = BoardStreamAuthPayload.build( + sessionIdOf(first), challengeOf(first), board.id, now); + String signature = Base64.getEncoder() + .encodeToString(Ed25519TestUtil.sign(board.keyPair.getPrivate(), signed)); + + // Replayed against the second session, whose sessionId+challenge differ. + deliver(second, authFrame(sessionIdOf(second), 1, board.id, now, signature)); + + assertThat(second.closedWith.getCode()).isEqualTo(4001); } + /** Domain separation: an agent-channel signature must not open the board stream. */ @Test - void malformedDeviceIdIsRejected() throws Exception { - Session s = connectWithQuery("deviceId=not-a-uuid"); + void anAgentChannelSignatureDoesNotAuthenticateOnTheBoardStream() throws Exception { + Board board = enrolled(); + Session s = connect(); + + long now = System.currentTimeMillis(); + byte[] agentPayload = AuthSignaturePayload.build( + sessionIdOf(s), challengeOf(s), board.id, now); + String signature = Base64.getEncoder() + .encodeToString(Ed25519TestUtil.sign(board.keyPair.getPrivate(), agentPayload)); + + deliver(s, authFrame(sessionIdOf(s), 1, board.id, now, signature)); assertThat(s.closedWith.getCode()).isEqualTo(4001); } @Test - void unknownDeviceIsRejected() throws Exception { - // deviceRepository returns empty for anything not explicitly enrolled. - Session s = connectWithQuery("deviceId=" + UUID.randomUUID()); + void frameNamingSomeoneElsesSessionIdIsRejected() throws Exception { + Board board = enrolled(); + Session s = connect(); + + long now = System.currentTimeMillis(); + UUID wrongSession = UUID.randomUUID(); + byte[] signed = BoardStreamAuthPayload.build(wrongSession, challengeOf(s), board.id, now); + String signature = Base64.getEncoder() + .encodeToString(Ed25519TestUtil.sign(board.keyPair.getPrivate(), signed)); + + deliver(s, authFrame(wrongSession, 1, board.id, now, signature)); + + assertThat(s.closedWith.getCode()).isEqualTo(4001); + } + + @Test + void nonMonotonicSequenceNumberIsRejected() throws Exception { + Board board = enrolled(); + Session s = connect(); + + // The client's first frame must be seq 1; 0 is a regression against the initial 0. + deliver(s, authFrame(sessionIdOf(s), 0, board.id, System.currentTimeMillis(), "x")); assertThat(s.closedWith.getCode()).isEqualTo(4002); + } - handler.contentChanged("events"); - assertThat(s.sent).isEmpty(); + @Test + void replayedSequenceNumberIsRejectedAfterAuth() throws Exception { + Board board = enrolled(); + Session s = connect(); + authenticate(s, board); // consumes seq 1 + + // Seq is checked before the frame is dispatched, so a replay closes on 4002 rather + // than falling through to the bad-state path. + deliver(s, authFrame(sessionIdOf(s), 1, board.id, System.currentTimeMillis(), "x")); + + assertThat(s.closedWith.getCode()).isEqualTo(4002); } @Test - void revokedDeviceIsRejected() throws Exception { - UUID id = UUID.randomUUID(); - Device revoked = Device.builder().id(id).displayName("d").build(); - revoked.setRevokedAt(Instant.now().minusSeconds(60)); - when(deviceRepository.findById(id)).thenReturn(Optional.of(revoked)); + void anOversizedFrameIsRejectedWithoutBeingParsed() throws Exception { + Session s = connect(); - Session s = connect(id); + deliver(s, "{\"type\":\"auth\",\"pad\":\"" + "A".repeat(8192) + "\"}"); + + assertThat(s.closedWith.getCode()).isEqualTo(4004); + } + + @Test + void unparseableFrameIsRejected() throws Exception { + Session s = connect(); + + deliver(s, "{not json"); + + assertThat(s.closedWith.getCode()).isEqualTo(4004); + } + + /** + * The subtype set is closed. A frame borrowed from the agent channel does not even + * deserialize here, so it never reaches the phase machine. + */ + @Test + void aFrameTypeThisChannelDoesNotKnowIsRejected() throws Exception { + Session s = connect(); + + deliver(s, heartbeatShaped(sessionIdOf(s), 1)); + + assertThat(s.closedWith.getCode()).isEqualTo(4004); + } + + @Test + void reAuthenticatingAfterAuthOkIsRejected() throws Exception { + Board board = enrolled(); + Session s = connect(); + authenticate(s, board); + + long now = System.currentTimeMillis(); + byte[] signed = BoardStreamAuthPayload.build(sessionIdOf(s), challengeOf(s), board.id, now); + String signature = Base64.getEncoder() + .encodeToString(Ed25519TestUtil.sign(board.keyPair.getPrivate(), signed)); + deliver(s, authFrame(sessionIdOf(s), 2, board.id, now, signature)); assertThat(s.closedWith.getCode()).isEqualTo(4003); } - // ── session lifecycle ───────────────────────────────────────────────── + @Test + void aConnectionThatNeverAuthenticatesIsReaped() throws Exception { + ReflectionTestUtils.setField(handler, "authTimeoutMs", -1L); // deadline already past + Session s = connect(); + + handler.reapUnauthenticatedSessions(); + + assertThat(s.closedWith.getCode()).isEqualTo(4008); + } + + @Test + void theReaperLeavesAuthenticatedBoardsAlone() throws Exception { + ReflectionTestUtils.setField(handler, "authTimeoutMs", -1L); + Board board = enrolled(); + Session s = connect(); + authenticate(s, board); + + handler.reapUnauthenticatedSessions(); + + assertThat(s.closedWith).isNull(); + handler.contentChanged("events"); + assertThat(s.sent).hasSize(3); + } + + // ── delivery ────────────────────────────────────────────────────────── + + @Test + void contentPushReachesEveryAuthenticatedBoard() throws Exception { + Session a = connectAuthenticated(); + Session b = connectAuthenticated(); + + handler.contentChanged("posters"); + + assertThat(refreshCount(a)).isEqualTo(1); + assertThat(refreshCount(b)).isEqualTo(1); + } + + @Test + void configPushReachesOnlyTheTargetBoard() throws Exception { + Board target = enrolled(); + Session mine = connect(); + authenticate(mine, target); + Session theirs = connectAuthenticated(); + + handler.configChanged(target.id); + + assertThat(refreshCount(mine)).isEqualTo(1); + assertThat(refreshCount(theirs)).isZero(); + } + + @Test + void messageIsJsonCarryingTypeReasonAndTarget() throws Exception { + Board target = enrolled(); + Session s = connect(); + authenticate(s, target); + + handler.configChanged(target.id); + + JsonNode body = mapper.readTree(s.sent.get(2)); + assertThat(body.get("type").asText()).isEqualTo("refresh"); + assertThat(body.get("reason").asText()).isEqualTo("config"); + assertThat(body.get("deviceId").asText()).isEqualTo(target.id.toString()); + assertThat(body.hasNonNull("at")).isTrue(); + } + + /** The deployed kiosk substring-matches rather than parsing, so JSON must still hit it. */ + @Test + void messageStillMatchesTheLegacyRefreshSubstring() throws Exception { + Session s = connectAuthenticated(); + + handler.refreshAll(); + + assertThat(s.sent.get(s.sent.size() - 1).toUpperCase()).contains("REFRESH"); + } + + // ── session lifecycle (unchanged semantics) ─────────────────────────── @Test void reconnectSupersedesThePriorSessionForTheSameDevice() throws Exception { - UUID device = enrolled(); - Session first = connect(device); - Session second = connect(device); + Board board = enrolled(); + Session first = connect(); + authenticate(first, board); + Session second = connect(); + authenticate(second, board); assertThat(first.closedWith).isNotNull(); + assertThat(first.closedWith.getCode()).isEqualTo(1000); + assertThat(first.closedWith.getReason()).isEqualTo("superseded"); - handler.configChanged(device); - assertThat(second.sent).hasSize(1); - assertThat(first.sent).isEmpty(); + handler.configChanged(board.id); + assertThat(refreshCount(second)).isEqualTo(1); + assertThat(refreshCount(first)).isZero(); } /** @@ -148,48 +413,118 @@ void reconnectSupersedesThePriorSessionForTheSameDevice() throws Exception { */ @Test void lateCloseOfASupersededSessionLeavesTheReplacementRegistered() throws Exception { - UUID device = enrolled(); - Session first = connect(device); - Session second = connect(device); + Board board = enrolled(); + Session first = connect(); + authenticate(first, board); + Session second = connect(); + authenticate(second, board); handler.afterConnectionClosed(first.ws, CloseStatus.NORMAL); - handler.configChanged(device); - assertThat(second.sent).hasSize(1); + handler.configChanged(board.id); + assertThat(refreshCount(second)).isEqualTo(1); } @Test void disconnectStopsDelivery() throws Exception { - Session s = connect(enrolled()); + Session s = connectAuthenticated(); handler.afterConnectionClosed(s.ws, CloseStatus.NORMAL); handler.contentChanged("events"); - assertThat(s.sent).isEmpty(); + assertThat(refreshCount(s)).isZero(); } @Test void closedSessionsAreDroppedAndNeverWrittenTo() throws Exception { - Session s = connect(enrolled()); + Session s = connectAuthenticated(); s.open = false; handler.contentChanged("events"); - assertThat(s.sent).isEmpty(); + assertThat(refreshCount(s)).isZero(); + } + + /** + * Tomcat's endpoint throws an unchecked {@link IllegalStateException} when a socket goes + * bad between the open check and the write. One dying board must not abort the broadcast + * for every board after it, nor fail the admin request that triggered the push. + */ + @Test + void aBoardThatThrowsOnWriteDoesNotStopTheBroadcast() throws Exception { + Session broken = connectAuthenticated(); + Session healthy = connectAuthenticated(); + broken.throwOnSend = true; + + handler.contentChanged("posters"); + + assertThat(refreshCount(healthy)).isEqualTo(1); + assertThat(refreshCount(broken)).isZero(); + } + + /** Server-side seq is what a client validates; it must never go backwards. */ + @Test + void outboundSeqIncreasesMonotonicallyAcrossPushes() throws Exception { + Session s = connectAuthenticated(); + + handler.contentChanged("posters"); + handler.contentChanged("events"); + handler.refreshAll(); + + List seqs = new ArrayList<>(); + for (String json : s.sent) seqs.add(mapper.readTree(json).get("seq").asLong()); + assertThat(seqs).containsExactly(1L, 2L, 3L, 4L, 5L); + } + + @Test + void closeIfPresentDropsTheBoardAndStopsDelivery() throws Exception { + Board board = enrolled(); + Session s = connect(); + authenticate(s, board); + + handler.closeIfPresent(board.id, CloseStatus.POLICY_VIOLATION.withReason("device_revoked")); + + assertThat(s.closedWith.getCode()).isEqualTo(CloseStatus.POLICY_VIOLATION.getCode()); + assertThat(s.closedWith.getReason()).isEqualTo("device_revoked"); + + handler.contentChanged("events"); + assertThat(refreshCount(s)).isZero(); + } + + @Test + void closeIfPresentIsANoOpForADeviceWithNoLiveSession() { + handler.closeIfPresent(UUID.randomUUID(), CloseStatus.POLICY_VIOLATION); + } + + @Test + void closeOfAConnectionThatNeverAuthenticatedIsHandled() throws Exception { + Session s = connect(); + + handler.afterConnectionClosed(s.ws, CloseStatus.NORMAL); + + // No device was ever bound, so there is nothing to deregister and nothing to throw. + assertThat(s.sent).hasSize(1); } // ── helpers ─────────────────────────────────────────────────────────── - /** Registers a device the repository will vouch for, and returns its id. */ - private UUID enrolled() { + /** A device the repository will vouch for, with a keypair the test can sign as. */ + private record Board(UUID id, KeyPair keyPair, Device device) {} + + private Board enrolled() { UUID id = UUID.randomUUID(); - when(deviceRepository.findById(id)) - .thenReturn(Optional.of(Device.builder().id(id).displayName("board").build())); - return id; + KeyPair kp = Ed25519TestUtil.generate(); + Device device = Device.builder() + .id(id) + .displayName("board") + .publicKey(Ed25519TestUtil.rawPublicKey(kp.getPublic())) + .build(); + when(deviceRepository.findById(id)).thenReturn(Optional.of(device)); + return new Board(id, kp, device); } - private Session connect(UUID deviceId) throws Exception { - return connectWithQuery("deviceId=" + deviceId); + private Session connect() throws Exception { + return connectWithQuery(null); } private Session connectWithQuery(String query) throws Exception { @@ -198,11 +533,73 @@ private Session connectWithQuery(String query) throws Exception { return session; } + private Session connectAuthenticated() throws Exception { + Session s = connect(); + authenticate(s, enrolled()); + return s; + } + + private void authenticate(Session s, Board board) throws Exception { + authenticate(s, board.id, board.keyPair.getPrivate(), System.currentTimeMillis()); + } + + private void authenticate(Session s, UUID deviceId, PrivateKey key, long timestamp) throws Exception { + byte[] signed = BoardStreamAuthPayload.build(sessionIdOf(s), challengeOf(s), deviceId, timestamp); + String signature = Base64.getEncoder().encodeToString(Ed25519TestUtil.sign(key, signed)); + deliver(s, authFrame(sessionIdOf(s), 1, deviceId, timestamp, signature)); + } + + /** Asserts nothing is delivered to a rejected session, including targeted pushes. */ + private void assertRejected(Session s, UUID deviceId) { + handler.contentChanged("events"); + handler.configChanged(deviceId); + assertThat(refreshCount(s)).isZero(); + } + + private long refreshCount(Session s) { + return s.sent.stream().filter(json -> json.contains("\"type\":\"refresh\"")).count(); + } + + private UUID sessionIdOf(Session s) throws Exception { + return UUID.fromString(mapper.readTree(s.sent.get(0)).get("sessionId").asText()); + } + + private String challengeOf(Session s) throws Exception { + return mapper.readTree(s.sent.get(0)).get("challenge").asText(); + } + + private String authFrame(UUID sessionId, long seq, UUID deviceId, long timestamp, String signature) + throws Exception { + Map body = new LinkedHashMap<>(); + body.put("type", "auth"); + body.put("seq", seq); + body.put("sessionId", sessionId); + body.put("deviceId", deviceId); + body.put("timestamp", timestamp); + body.put("signature", signature); + return mapper.writeValueAsString(body); + } + + /** A frame shaped like the agent channel's heartbeat — not a subtype this channel knows. */ + private String heartbeatShaped(UUID sessionId, long seq) throws Exception { + Map body = new LinkedHashMap<>(); + body.put("type", "heartbeat"); + body.put("seq", seq); + body.put("sessionId", sessionId); + return mapper.writeValueAsString(body); + } + + private void deliver(Session s, String json) throws Exception { + handler.handleMessage(s.ws, new TextMessage(json)); + } + /** Minimal stand-in that records what the handler writes and how it closes. */ private static class Session { final WebSocketSession ws = mock(WebSocketSession.class); final List sent = new ArrayList<>(); boolean open = true; + /** Simulates the container throwing an unchecked error mid-write. */ + boolean throwOnSend = false; CloseStatus closedWith; Session(String query) throws Exception { @@ -212,6 +609,7 @@ private static class Session { when(ws.isOpen()).thenAnswer(inv -> open); when(ws.getAttributes()).thenReturn(new HashMap<>()); doAnswer(inv -> { + if (throwOnSend) throw new IllegalStateException("endpoint in bad state"); sent.add(((TextMessage) inv.getArgument(0)).getPayload()); return null; }).when(ws).sendMessage(any());