diff --git a/.gitattributes b/.gitattributes index 1372660..38e255e 100644 --- a/.gitattributes +++ b/.gitattributes @@ -2,6 +2,10 @@ # binary disables any text transformation git or editors might otherwise apply. *.raw binary +# Pre-generated test keystore — keep as opaque binary bytes. +*.p12 binary +*.jks binary + # Java sources, properties, markdown — normal LF normalisation in repo. *.java text eol=lf *.xml text eol=lf diff --git a/sip-transport-nio/src/main/java/com/sip/transport/nio/NioTlsTransport.java b/sip-transport-nio/src/main/java/com/sip/transport/nio/NioTlsTransport.java new file mode 100644 index 0000000..4267140 --- /dev/null +++ b/sip-transport-nio/src/main/java/com/sip/transport/nio/NioTlsTransport.java @@ -0,0 +1,347 @@ +package com.sip.transport.nio; + +import com.sip.codec.SipCodecException; +import com.sip.codec.SipEncoder; +import com.sip.codec.SipParser; +import com.sip.message.SipMessage; +import com.sip.transport.Endpoint; +import com.sip.transport.InboundMessage; +import com.sip.transport.MessageListener; +import com.sip.transport.Transport; +import com.sip.transport.TransportType; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.net.ssl.SSLContext; +import javax.net.ssl.SSLServerSocket; +import javax.net.ssl.SSLSocket; +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.net.InetSocketAddress; +import java.net.Socket; +import java.nio.charset.StandardCharsets; +import java.time.Instant; +import java.util.Objects; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.Executors; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.locks.ReentrantLock; + +/** + * TLS transport — SIPS / SIP over TLS (RFC 3261 §26.2, RFC 5630). + * + *

Architecture mirrors {@link NioTcpTransport}: per-connection virtual + * thread + Content-Length framing + connection reuse (RFC 5923). The + * difference is the socket factory comes from a caller-supplied + * {@link SSLContext}, so server-cert provisioning, mTLS, cipher suite + * selection, and protocol version pinning all live on the application + * side, exactly as they should.

+ * + *

Hostname verification (RFC 5922)

+ *

When dialing as a client, {@link SSLSocket}'s default does NOT verify + * the peer hostname against the certificate. We enable HTTPS-style + * endpoint identification, which RFC 5922 explicitly requires for SIPS.

+ */ +public final class NioTlsTransport implements Transport { + + private static final Logger log = LoggerFactory.getLogger(NioTlsTransport.class); + + private final SSLContext sslContext; + private final InetSocketAddress bindAddress; + private final AtomicBoolean running = new AtomicBoolean(false); + private final AtomicLong rxMessages = new AtomicLong(); + private final AtomicLong txMessages = new AtomicLong(); + private final AtomicLong parseErrors = new AtomicLong(); + private final ConcurrentHashMap connections = + new ConcurrentHashMap<>(); + + private volatile MessageListener listener; + private volatile SSLServerSocket serverSocket; + private volatile Endpoint local; + private volatile java.util.concurrent.ExecutorService executor; + + public NioTlsTransport(SSLContext sslContext, InetSocketAddress bindAddress) { + this.sslContext = Objects.requireNonNull(sslContext, "sslContext"); + this.bindAddress = Objects.requireNonNull(bindAddress, "bindAddress"); + } + + @Override + public TransportType type() { + return TransportType.TLS; + } + + @Override + public Endpoint local() { + return local; + } + + @Override + public CompletableFuture start() { + if (!running.compareAndSet(false, true)) { + return CompletableFuture.failedFuture( + new IllegalStateException("transport already started")); + } + try { + SSLServerSocket ss = (SSLServerSocket) sslContext.getServerSocketFactory() + .createServerSocket(); + ss.setReuseAddress(true); + ss.bind(bindAddress); + this.serverSocket = ss; + this.local = new Endpoint(TransportType.TLS, + (InetSocketAddress) ss.getLocalSocketAddress()); + this.executor = Executors.newThreadPerTaskExecutor( + Thread.ofVirtual().name("sip-tls-", 0).factory()); + executor.submit(this::acceptLoop); + log.info("TLS transport bound to {}", local.address()); + return CompletableFuture.completedFuture(null); + } catch (IOException e) { + running.set(false); + return CompletableFuture.failedFuture(e); + } + } + + @Override + public CompletableFuture send(SipMessage message, Endpoint peer) { + if (!running.get()) { + return CompletableFuture.failedFuture( + new IllegalStateException("transport not started")); + } + if (peer.transport() != TransportType.TLS) { + return CompletableFuture.failedFuture( + new IllegalArgumentException( + "TLS transport cannot send to " + peer.transport() + " endpoint")); + } + try { + byte[] bytes = SipEncoder.encode(message); + Connection conn = connectionFor(peer.address()); + conn.send(bytes); + txMessages.incrementAndGet(); + return CompletableFuture.completedFuture(null); + } catch (IOException | SipCodecException e) { + return CompletableFuture.failedFuture(e); + } + } + + @Override + public void listener(MessageListener listener) { + this.listener = listener; + } + + @Override + public void close() throws IOException { + if (!running.compareAndSet(true, false)) { + return; + } + SSLServerSocket ss = this.serverSocket; + if (ss != null) { + ss.close(); + } + for (Connection conn : connections.values()) { + conn.close(); + } + connections.clear(); + java.util.concurrent.ExecutorService exec = this.executor; + if (exec != null) { + exec.shutdownNow(); + } + log.info("TLS transport closed; rx={} tx={} parseErrors={}", + rxMessages.get(), txMessages.get(), parseErrors.get()); + } + + public long receivedMessages() { return rxMessages.get(); } + public long sentMessages() { return txMessages.get(); } + public long parseErrors() { return parseErrors.get(); } + + /* -------- internals -------- */ + + private void acceptLoop() { + while (running.get()) { + SSLSocket socket; + try { + socket = (SSLSocket) serverSocket.accept(); + } catch (java.net.SocketException closed) { + return; + } catch (IOException e) { + if (running.get()) { + log.warn("TLS accept error: {}", e.toString()); + } + continue; + } + InetSocketAddress peerAddr = (InetSocketAddress) socket.getRemoteSocketAddress(); + Connection conn = new Connection(socket, peerAddr); + connections.put(peerAddr, conn); + executor.submit(conn::readLoop); + } + } + + private Connection connectionFor(InetSocketAddress peer) throws IOException { + Connection existing = connections.get(peer); + if (existing != null && existing.alive()) { + return existing; + } + SSLSocket socket = (SSLSocket) sslContext.getSocketFactory().createSocket(); + // RFC 5922 — SIPS requires hostname verification. + var params = socket.getSSLParameters(); + params.setEndpointIdentificationAlgorithm("HTTPS"); + socket.setSSLParameters(params); + + socket.connect(peer, 5_000); + socket.startHandshake(); + + Connection conn = new Connection(socket, peer); + connections.put(peer, conn); + executor.submit(conn::readLoop); + return conn; + } + + final class Connection implements AutoCloseable { + private final Socket socket; + private final InetSocketAddress peerAddr; + private final InputStream in; + private final OutputStream out; + private final ReentrantLock writeLock = new ReentrantLock(); + private volatile boolean closed; + + Connection(Socket socket, InetSocketAddress peerAddr) { + this.socket = socket; + this.peerAddr = peerAddr; + try { + this.in = socket.getInputStream(); + this.out = socket.getOutputStream(); + } catch (IOException e) { + throw new RuntimeException("failed to open streams for " + peerAddr, e); + } + } + + boolean alive() { + return !closed && !socket.isClosed(); + } + + void send(byte[] bytes) throws IOException { + writeLock.lock(); + try { + out.write(bytes); + out.flush(); + } finally { + writeLock.unlock(); + } + } + + void readLoop() { + try { + while (running.get() && !closed) { + byte[] msgBytes = readOneMessage(); + if (msgBytes == null) { + return; + } + rxMessages.incrementAndGet(); + dispatch(msgBytes); + } + } catch (IOException e) { + if (running.get()) { + log.debug("TLS read error on {}: {}", peerAddr, e.toString()); + } + } finally { + close(); + } + } + + private byte[] readOneMessage() throws IOException { + java.io.ByteArrayOutputStream header = new java.io.ByteArrayOutputStream(512); + int crlfState = 0; + while (crlfState != 4) { + int b = in.read(); + if (b < 0) { + if (header.size() == 0) { + return null; + } + throw new IOException("EOF mid-header (read " + header.size() + " bytes)"); + } + header.write(b); + crlfState = switch (b) { + case '\r' -> (crlfState == 2) ? 3 : 1; + case '\n' -> (crlfState == 1) ? 2 : (crlfState == 3) ? 4 : 0; + default -> 0; + }; + if (header.size() > 64 * 1024) { + throw new IOException("header block exceeds 64 KiB"); + } + } + byte[] headerBytes = header.toByteArray(); + int contentLength = scanContentLength(headerBytes); + if (contentLength == 0) { + return headerBytes; + } + byte[] full = new byte[headerBytes.length + contentLength]; + System.arraycopy(headerBytes, 0, full, 0, headerBytes.length); + int read = 0; + while (read < contentLength) { + int n = in.read(full, headerBytes.length + read, contentLength - read); + if (n < 0) { + throw new IOException("EOF mid-body after " + + read + " of " + contentLength + " bytes"); + } + read += n; + } + return full; + } + + private int scanContentLength(byte[] hdr) throws IOException { + String text = new String(hdr, StandardCharsets.US_ASCII); + int idx = 0; + while (idx < text.length()) { + int eol = text.indexOf("\r\n", idx); + if (eol < 0) break; + String line = text.substring(idx, eol); + idx = eol + 2; + int colon = line.indexOf(':'); + if (colon <= 0) continue; + String name = line.substring(0, colon).trim(); + if (name.equalsIgnoreCase("Content-Length") || name.equalsIgnoreCase("l")) { + String value = line.substring(colon + 1).trim(); + try { + int v = Integer.parseInt(value); + if (v < 0) { + throw new IOException("Content-Length is negative on TLS: " + v); + } + return v; + } catch (NumberFormatException e) { + throw new IOException("Content-Length not numeric: '" + value + "'"); + } + } + } + return 0; + } + + private void dispatch(byte[] bytes) { + MessageListener l = listener; + Endpoint peer = new Endpoint(TransportType.TLS, peerAddr); + try { + SipMessage msg = SipParser.parse(bytes); + if (l != null) { + l.onMessage(new InboundMessage(msg, peer, local, Instant.now())); + } + } catch (SipCodecException e) { + parseErrors.incrementAndGet(); + if (l != null) { + try { l.onError(peer, e); } catch (Throwable ignore) { } + } + } catch (Throwable t) { + if (l != null) { + try { l.onError(peer, t); } catch (Throwable ignore) { } + } + } + } + + @Override + public void close() { + if (closed) return; + closed = true; + try { socket.close(); } catch (IOException ignore) { } + connections.remove(peerAddr, this); + } + } +} diff --git a/sip-transport-nio/src/main/java/com/sip/transport/nio/NioWebSocketTransport.java b/sip-transport-nio/src/main/java/com/sip/transport/nio/NioWebSocketTransport.java new file mode 100644 index 0000000..08c861a --- /dev/null +++ b/sip-transport-nio/src/main/java/com/sip/transport/nio/NioWebSocketTransport.java @@ -0,0 +1,348 @@ +package com.sip.transport.nio; + +import com.sip.codec.SipCodecException; +import com.sip.codec.SipEncoder; +import com.sip.codec.SipParser; +import com.sip.message.SipMessage; +import com.sip.transport.Endpoint; +import com.sip.transport.InboundMessage; +import com.sip.transport.MessageListener; +import com.sip.transport.Transport; +import com.sip.transport.TransportType; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.net.InetSocketAddress; +import java.net.ServerSocket; +import java.net.Socket; +import java.time.Instant; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.Executors; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.locks.ReentrantLock; + +/** + * WebSocket transport — SIP over WS / WSS (RFC 7118). + * + *

One TEXT WebSocket frame carries exactly one SIP message: this lets + * the receiver use the frame boundary as message boundary, so the + * {@code Content-Length}-driven framing logic from TCP is not needed.

+ * + *

Threading

+ *
    + *
  • Server: a virtual-thread accept loop; each connection then runs + * handshake + frame read loop on its own virtual thread.
    • + *
  • Client: outbound dial opens a TCP socket, runs the client-side + * handshake, then reuses the same virtual-thread read loop.
    • + *
  • Sends serialise per-connection with a {@link ReentrantLock} so + * interleaved frames are well-formed.
    • + *
+ * + *

Path / host

+ *

The server accepts upgrades on path {@code /}. The client transport + * uses path {@code /} and {@code Host: host:port} by default. SIP-aware + * intermediaries that key on the request URI can override later.

+ */ +public final class NioWebSocketTransport implements Transport { + + private static final Logger log = LoggerFactory.getLogger(NioWebSocketTransport.class); + + private final InetSocketAddress bindAddress; + private final boolean isClientRole; + private final AtomicBoolean running = new AtomicBoolean(false); + private final AtomicLong rxMessages = new AtomicLong(); + private final AtomicLong txMessages = new AtomicLong(); + private final AtomicLong parseErrors = new AtomicLong(); + private final ConcurrentHashMap connections = + new ConcurrentHashMap<>(); + + private volatile MessageListener listener; + private volatile ServerSocket serverSocket; + private volatile Endpoint local; + private volatile java.util.concurrent.ExecutorService executor; + + /** + * @param bindAddress local socket address. For server role pass the + * listening port; for pure client role pass 0 + * (an ephemeral port is allocated but never accepted on). + * @param clientRole {@code true} to disable the accept loop and only + * use the transport for outbound dials (which still + * receive responses on the dialed connection). + */ + public NioWebSocketTransport(InetSocketAddress bindAddress, boolean clientRole) { + this.bindAddress = bindAddress; + this.isClientRole = clientRole; + } + + public NioWebSocketTransport(InetSocketAddress bindAddress) { + this(bindAddress, false); + } + + @Override + public TransportType type() { + return TransportType.WS; + } + + @Override + public Endpoint local() { + return local; + } + + @Override + public CompletableFuture start() { + if (!running.compareAndSet(false, true)) { + return CompletableFuture.failedFuture( + new IllegalStateException("transport already started")); + } + try { + this.executor = Executors.newThreadPerTaskExecutor( + Thread.ofVirtual().name("sip-ws-", 0).factory()); + ServerSocket ss = new ServerSocket(); + ss.setReuseAddress(true); + ss.bind(bindAddress); + this.serverSocket = ss; + this.local = new Endpoint(TransportType.WS, + (InetSocketAddress) ss.getLocalSocketAddress()); + if (!isClientRole) { + executor.submit(this::acceptLoop); + } + log.info("WebSocket transport bound to {} (role={})", + local.address(), isClientRole ? "client" : "server"); + return CompletableFuture.completedFuture(null); + } catch (IOException e) { + running.set(false); + return CompletableFuture.failedFuture(e); + } + } + + @Override + public CompletableFuture send(SipMessage message, Endpoint peer) { + if (!running.get()) { + return CompletableFuture.failedFuture( + new IllegalStateException("transport not started")); + } + if (peer.transport() != TransportType.WS) { + return CompletableFuture.failedFuture( + new IllegalArgumentException( + "WS transport cannot send to " + peer.transport() + " endpoint")); + } + try { + byte[] bytes = SipEncoder.encode(message); + Connection conn = connectionFor(peer.address()); + conn.send(bytes); + txMessages.incrementAndGet(); + return CompletableFuture.completedFuture(null); + } catch (IOException | SipCodecException e) { + return CompletableFuture.failedFuture(e); + } + } + + @Override + public void listener(MessageListener listener) { + this.listener = listener; + } + + @Override + public void close() throws IOException { + if (!running.compareAndSet(true, false)) { + return; + } + ServerSocket ss = this.serverSocket; + if (ss != null) { + ss.close(); + } + for (Connection conn : connections.values()) { + conn.close(); + } + connections.clear(); + java.util.concurrent.ExecutorService exec = this.executor; + if (exec != null) { + exec.shutdownNow(); + } + log.info("WebSocket transport closed; rx={} tx={} parseErrors={}", + rxMessages.get(), txMessages.get(), parseErrors.get()); + } + + public long receivedMessages() { return rxMessages.get(); } + public long sentMessages() { return txMessages.get(); } + public long parseErrors() { return parseErrors.get(); } + + /* -------- internals -------- */ + + private void acceptLoop() { + while (running.get()) { + Socket socket; + try { + socket = serverSocket.accept(); + } catch (java.net.SocketException closed) { + return; + } catch (IOException e) { + if (running.get()) { + log.warn("WS accept error: {}", e.toString()); + } + continue; + } + InetSocketAddress peerAddr = (InetSocketAddress) socket.getRemoteSocketAddress(); + executor.submit(() -> handleServerSide(socket, peerAddr)); + } + } + + private void handleServerSide(Socket socket, InetSocketAddress peerAddr) { + try { + WebSocketHandshake.acceptServerHandshake( + socket.getInputStream(), socket.getOutputStream()); + } catch (IOException e) { + log.debug("WS handshake failed from {}: {}", peerAddr, e.toString()); + try { socket.close(); } catch (IOException ignore) { } + return; + } + Connection conn = new Connection(socket, peerAddr, /* maskOutbound */ false); + connections.put(peerAddr, conn); + conn.readLoop(); + } + + private Connection connectionFor(InetSocketAddress peer) throws IOException { + Connection existing = connections.get(peer); + if (existing != null && existing.alive()) { + return existing; + } + Socket socket = new Socket(); + socket.connect(peer, 5_000); + WebSocketHandshake.initiateClientHandshake( + socket.getInputStream(), socket.getOutputStream(), + peer.getHostString() + ":" + peer.getPort(), "/"); + Connection conn = new Connection(socket, peer, /* maskOutbound */ true); + connections.put(peer, conn); + executor.submit(conn::readLoop); + return conn; + } + + final class Connection implements AutoCloseable { + private final Socket socket; + private final InetSocketAddress peerAddr; + private final InputStream in; + private final OutputStream out; + private final boolean maskOutbound; + private final ReentrantLock writeLock = new ReentrantLock(); + private volatile boolean closed; + + Connection(Socket socket, InetSocketAddress peerAddr, boolean maskOutbound) { + this.socket = socket; + this.peerAddr = peerAddr; + this.maskOutbound = maskOutbound; + try { + this.in = socket.getInputStream(); + this.out = socket.getOutputStream(); + } catch (IOException e) { + throw new RuntimeException("failed to open streams for " + peerAddr, e); + } + } + + boolean alive() { + return !closed && !socket.isClosed(); + } + + void send(byte[] bytes) throws IOException { + writeLock.lock(); + try { + WebSocketFrame.write(out, WebSocketFrame.OPCODE_TEXT, bytes, maskOutbound); + } finally { + writeLock.unlock(); + } + } + + void readLoop() { + try { + while (running.get() && !closed) { + WebSocketFrame.Frame frame; + try { + frame = WebSocketFrame.read(in); + } catch (java.io.EOFException eof) { + return; + } + if (!handleFrame(frame)) { + return; + } + } + } catch (IOException e) { + if (running.get()) { + log.debug("WS read error on {}: {}", peerAddr, e.toString()); + } + } finally { + close(); + } + } + + /** Returns {@code false} if the connection should be torn down. */ + private boolean handleFrame(WebSocketFrame.Frame frame) throws IOException { + switch (frame.opcode()) { + case WebSocketFrame.OPCODE_TEXT, WebSocketFrame.OPCODE_BINARY -> { + dispatch(frame.payload()); + return true; + } + case WebSocketFrame.OPCODE_PING -> { + writeLock.lock(); + try { + WebSocketFrame.write(out, WebSocketFrame.OPCODE_PONG, + frame.payload(), maskOutbound); + } finally { + writeLock.unlock(); + } + return true; + } + case WebSocketFrame.OPCODE_PONG -> { + return true; + } + case WebSocketFrame.OPCODE_CLOSE -> { + writeLock.lock(); + try { + WebSocketFrame.write(out, WebSocketFrame.OPCODE_CLOSE, + new byte[0], maskOutbound); + } finally { + writeLock.unlock(); + } + return false; + } + default -> { + log.debug("ignoring unknown WS opcode 0x{} from {}", + Integer.toHexString(frame.opcode()), peerAddr); + return true; + } + } + } + + private void dispatch(byte[] bytes) { + MessageListener l = listener; + rxMessages.incrementAndGet(); + Endpoint peer = new Endpoint(TransportType.WS, peerAddr); + try { + SipMessage msg = SipParser.parse(bytes); + if (l != null) { + l.onMessage(new InboundMessage(msg, peer, local, Instant.now())); + } + } catch (SipCodecException e) { + parseErrors.incrementAndGet(); + if (l != null) { + try { l.onError(peer, e); } catch (Throwable ignore) { } + } + } catch (Throwable t) { + if (l != null) { + try { l.onError(peer, t); } catch (Throwable ignore) { } + } + } + } + + @Override + public void close() { + if (closed) return; + closed = true; + try { socket.close(); } catch (IOException ignore) { } + connections.remove(peerAddr, this); + } + } +} diff --git a/sip-transport-nio/src/main/java/com/sip/transport/nio/WebSocketFrame.java b/sip-transport-nio/src/main/java/com/sip/transport/nio/WebSocketFrame.java new file mode 100644 index 0000000..cea4593 --- /dev/null +++ b/sip-transport-nio/src/main/java/com/sip/transport/nio/WebSocketFrame.java @@ -0,0 +1,117 @@ +package com.sip.transport.nio; + +import java.io.DataInputStream; +import java.io.DataOutputStream; +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.security.SecureRandom; + +/** + * RFC 6455 §5.2 WebSocket frame codec — TEXT / BINARY / CLOSE / PING / PONG. + * + *

Tailored to RFC 7118 (SIP-over-WebSocket): SIP messages are transported + * as a single TEXT frame each, and the SIP framing is taken from the frame + * boundary rather than from {@code Content-Length}.

+ * + *

RFC 6455 mandates that client-originated frames be masked with a 32-bit + * key. The {@link #write(OutputStream, int, byte[], boolean) write} method + * generates a fresh random mask per frame when {@code maskOutbound = true}.

+ */ +public final class WebSocketFrame { + + public static final int OPCODE_CONTINUATION = 0x0; + public static final int OPCODE_TEXT = 0x1; + public static final int OPCODE_BINARY = 0x2; + public static final int OPCODE_CLOSE = 0x8; + public static final int OPCODE_PING = 0x9; + public static final int OPCODE_PONG = 0xA; + + private static final SecureRandom RNG = new SecureRandom(); + /** RFC 6455 §5.2 — frames bigger than this are likely abuse. */ + static final int MAX_PAYLOAD = 4 * 1024 * 1024; + + private WebSocketFrame() { } + + public record Frame(boolean fin, int opcode, byte[] payload) { + public Frame { + payload = payload.clone(); + } + @Override + public byte[] payload() { + return payload.clone(); + } + } + + public static Frame read(InputStream in) throws IOException { + DataInputStream dis = new DataInputStream(in); + int b1 = dis.read(); + if (b1 < 0) { + throw new java.io.EOFException("connection closed mid-frame"); + } + boolean fin = (b1 & 0x80) != 0; + int opcode = b1 & 0x0F; + int b2 = dis.readUnsignedByte(); + boolean masked = (b2 & 0x80) != 0; + long length = b2 & 0x7F; + if (length == 126) { + length = dis.readUnsignedShort(); + } else if (length == 127) { + length = dis.readLong(); + if (length < 0 || length > MAX_PAYLOAD) { + throw new IOException("payload too large: " + length); + } + } + if (length > MAX_PAYLOAD) { + throw new IOException("payload too large: " + length); + } + byte[] mask = null; + if (masked) { + mask = new byte[4]; + dis.readFully(mask); + } + byte[] payload = new byte[(int) length]; + dis.readFully(payload); + if (masked) { + for (int i = 0; i < payload.length; i++) { + payload[i] = (byte) (payload[i] ^ mask[i & 3]); + } + } + return new Frame(fin, opcode, payload); + } + + /** + * Write a single frame. {@code maskOutbound} should be {@code true} for + * client-originated frames, {@code false} for server-originated frames + * (RFC 6455 §5.1). + */ + public static void write(OutputStream out, int opcode, byte[] payload, + boolean maskOutbound) throws IOException { + DataOutputStream dos = new DataOutputStream(out); + dos.writeByte(0x80 | (opcode & 0x0F)); // FIN=1 + int len = payload.length; + int maskBit = maskOutbound ? 0x80 : 0; + if (len <= 125) { + dos.writeByte(maskBit | len); + } else if (len <= 0xFFFF) { + dos.writeByte(maskBit | 126); + dos.writeShort(len); + } else { + dos.writeByte(maskBit | 127); + dos.writeLong(len); + } + if (maskOutbound) { + byte[] mask = new byte[4]; + RNG.nextBytes(mask); + dos.write(mask); + byte[] masked = new byte[len]; + for (int i = 0; i < len; i++) { + masked[i] = (byte) (payload[i] ^ mask[i & 3]); + } + dos.write(masked); + } else { + dos.write(payload); + } + dos.flush(); + } +} diff --git a/sip-transport-nio/src/main/java/com/sip/transport/nio/WebSocketHandshake.java b/sip-transport-nio/src/main/java/com/sip/transport/nio/WebSocketHandshake.java new file mode 100644 index 0000000..9930786 --- /dev/null +++ b/sip-transport-nio/src/main/java/com/sip/transport/nio/WebSocketHandshake.java @@ -0,0 +1,191 @@ +package com.sip.transport.nio; + +import java.io.BufferedReader; +import java.io.IOException; +import java.io.InputStream; +import java.io.InputStreamReader; +import java.io.OutputStream; +import java.nio.charset.StandardCharsets; +import java.security.MessageDigest; +import java.security.NoSuchAlgorithmException; +import java.util.Base64; +import java.util.HashMap; +import java.util.Locale; +import java.util.Map; + +/** + * RFC 6455 §4 / RFC 7118 §5 WebSocket upgrade handshake. + * + *

The server reads the HTTP upgrade request and writes the matching + * 101 response with the {@code Sec-WebSocket-Accept} hash. RFC 7118 + * additionally requires negotiation of the {@code sip} sub-protocol.

+ * + *

Notes:

+ *
    + *
  • HTTP request line + headers are parsed; the body is ignored.
    • + *
  • Header names are case-insensitive (HTTP §4.2); we lowercase on + * insert.
    • + *
  • We don't bother with extensions ({@code Sec-WebSocket-Extensions}) + * — none are needed for SIP-over-WS.
    • + *
+ */ +public final class WebSocketHandshake { + + public static final String SIP_SUBPROTOCOL = "sip"; + + /** Magic GUID per RFC 6455 §4.2.2 step 5. */ + static final String GUID = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"; + + private WebSocketHandshake() { } + + /** Reads a client upgrade request off the wire and writes the 101 response. */ + public static void acceptServerHandshake(InputStream in, OutputStream out) + throws IOException { + Map headers = readHttpHeaders(in); + + String key = headers.get("sec-websocket-key"); + if (key == null) { + sendError(out, 400, "Bad Request", "missing Sec-WebSocket-Key"); + throw new IOException("missing Sec-WebSocket-Key"); + } + String upgrade = headers.getOrDefault("upgrade", ""); + if (!upgrade.equalsIgnoreCase("websocket")) { + sendError(out, 400, "Bad Request", "Upgrade header is not websocket"); + throw new IOException("not a WebSocket upgrade: '" + upgrade + "'"); + } + String version = headers.getOrDefault("sec-websocket-version", ""); + if (!"13".equals(version)) { + sendError(out, 426, "Upgrade Required", "Sec-WebSocket-Version 13 required"); + throw new IOException("unsupported version: '" + version + "'"); + } + String subprotocols = headers.getOrDefault("sec-websocket-protocol", ""); + boolean hasSip = false; + for (String p : subprotocols.split(",")) { + if (p.trim().equalsIgnoreCase(SIP_SUBPROTOCOL)) { + hasSip = true; + break; + } + } + + String accept = encodeAccept(key); + StringBuilder response = new StringBuilder(256); + response.append("HTTP/1.1 101 Switching Protocols\r\n"); + response.append("Upgrade: websocket\r\n"); + response.append("Connection: Upgrade\r\n"); + response.append("Sec-WebSocket-Accept: ").append(accept).append("\r\n"); + if (hasSip) { + response.append("Sec-WebSocket-Protocol: ").append(SIP_SUBPROTOCOL).append("\r\n"); + } + response.append("\r\n"); + out.write(response.toString().getBytes(StandardCharsets.US_ASCII)); + out.flush(); + } + + /** Writes a client upgrade request and validates the 101 response. */ + public static void initiateClientHandshake(InputStream in, OutputStream out, + String host, String resource) + throws IOException { + String key = generateClientKey(); + StringBuilder request = new StringBuilder(256); + request.append("GET ").append(resource).append(" HTTP/1.1\r\n"); + request.append("Host: ").append(host).append("\r\n"); + request.append("Upgrade: websocket\r\n"); + request.append("Connection: Upgrade\r\n"); + request.append("Sec-WebSocket-Key: ").append(key).append("\r\n"); + request.append("Sec-WebSocket-Version: 13\r\n"); + request.append("Sec-WebSocket-Protocol: ").append(SIP_SUBPROTOCOL).append("\r\n"); + request.append("\r\n"); + out.write(request.toString().getBytes(StandardCharsets.US_ASCII)); + out.flush(); + + // Read status line + headers. + BufferedReader reader = new BufferedReader( + new InputStreamReader(in, StandardCharsets.US_ASCII)); + String statusLine = reader.readLine(); + if (statusLine == null || !statusLine.contains("101")) { + throw new IOException("server did not switch protocols: '" + statusLine + "'"); + } + Map headers = new HashMap<>(); + String line; + while ((line = reader.readLine()) != null && !line.isEmpty()) { + int colon = line.indexOf(':'); + if (colon > 0) { + headers.put(line.substring(0, colon).trim().toLowerCase(Locale.ROOT), + line.substring(colon + 1).trim()); + } + } + String expected = encodeAccept(key); + String actual = headers.get("sec-websocket-accept"); + if (!expected.equals(actual)) { + throw new IOException( + "Sec-WebSocket-Accept mismatch (expected " + expected + ", got " + actual + ")"); + } + } + + static String encodeAccept(String clientKey) { + try { + MessageDigest sha1 = MessageDigest.getInstance("SHA-1"); + return Base64.getEncoder().encodeToString( + sha1.digest((clientKey + GUID).getBytes(StandardCharsets.US_ASCII))); + } catch (NoSuchAlgorithmException e) { + throw new IllegalStateException("SHA-1 unavailable", e); + } + } + + static String generateClientKey() { + byte[] random = new byte[16]; + new java.security.SecureRandom().nextBytes(random); + return Base64.getEncoder().encodeToString(random); + } + + private static Map readHttpHeaders(InputStream in) throws IOException { + // Lightweight HTTP request line + headers reader. + StringBuilder buf = new StringBuilder(512); + int crlfState = 0; + while (crlfState != 4) { + int b = in.read(); + if (b < 0) { + throw new IOException("EOF reading WebSocket upgrade request"); + } + buf.append((char) b); + crlfState = switch (b) { + case '\r' -> (crlfState == 2) ? 3 : 1; + case '\n' -> (crlfState == 1) ? 2 : (crlfState == 3) ? 4 : 0; + default -> 0; + }; + if (buf.length() > 64 * 1024) { + throw new IOException("HTTP request too large"); + } + } + String text = buf.toString(); + Map out = new HashMap<>(); + int idx = text.indexOf("\r\n"); + idx = text.indexOf("\r\n", idx + 2); // skip request line + int start = text.indexOf("\r\n") + 2; + while (start < text.length() - 2) { + int eol = text.indexOf("\r\n", start); + if (eol < 0) break; + String line = text.substring(start, eol); + int colon = line.indexOf(':'); + if (colon > 0) { + out.put(line.substring(0, colon).trim().toLowerCase(Locale.ROOT), + line.substring(colon + 1).trim()); + } + start = eol + 2; + } + return out; + } + + private static void sendError(OutputStream out, int code, String reason, String body) + throws IOException { + String payload = body + "\r\n"; + StringBuilder resp = new StringBuilder(); + resp.append("HTTP/1.1 ").append(code).append(' ').append(reason).append("\r\n"); + resp.append("Content-Type: text/plain\r\n"); + resp.append("Content-Length: ").append(payload.length()).append("\r\n"); + resp.append("Connection: close\r\n\r\n"); + resp.append(payload); + out.write(resp.toString().getBytes(StandardCharsets.US_ASCII)); + out.flush(); + } +} diff --git a/sip-transport-nio/src/test/java/com/sip/transport/nio/NioTlsTransportTest.java b/sip-transport-nio/src/test/java/com/sip/transport/nio/NioTlsTransportTest.java new file mode 100644 index 0000000..56a71f8 --- /dev/null +++ b/sip-transport-nio/src/test/java/com/sip/transport/nio/NioTlsTransportTest.java @@ -0,0 +1,143 @@ +package com.sip.transport.nio; + +import com.sip.message.SipMethod; +import com.sip.message.SipRequest; +import com.sip.message.SipResponse; +import com.sip.message.SipVersion; +import com.sip.message.header.Headers; +import com.sip.message.uri.SipUri; +import com.sip.transport.InboundMessage; +import com.sip.transport.TransportType; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; + +import javax.net.ssl.SSLContext; +import java.net.InetAddress; +import java.net.InetSocketAddress; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import static org.assertj.core.api.Assertions.assertThat; + +class NioTlsTransportTest { + + private final List open = new ArrayList<>(); + + @AfterEach + void closeAll() throws Exception { + for (int i = open.size() - 1; i >= 0; i--) { + open.get(i).close(); + } + } + + private NioTlsTransport bindServer() throws Exception { + SSLContext server = TestSslContext.serverContext(); + InetSocketAddress addr = new InetSocketAddress(InetAddress.getLoopbackAddress(), 0); + NioTlsTransport t = new NioTlsTransport(server, addr); + t.start().get(5, TimeUnit.SECONDS); + open.add(t); + return t; + } + + private NioTlsTransport bindClient() throws Exception { + SSLContext client = TestSslContext.clientContext(); + InetSocketAddress addr = new InetSocketAddress(InetAddress.getLoopbackAddress(), 0); + NioTlsTransport t = new NioTlsTransport(client, addr); + t.start().get(5, TimeUnit.SECONDS); + open.add(t); + return t; + } + + @Test + @Timeout(value = 15, unit = TimeUnit.SECONDS) + void clientCanSendInviteToServerOverTls() throws Exception { + NioTlsTransport server = bindServer(); + NioTlsTransport client = bindClient(); + + CountDownLatch latch = new CountDownLatch(1); + List received = new ArrayList<>(); + server.listener(in -> { + received.add(in); + latch.countDown(); + }); + + byte[] body = "v=0\r\no=- 0 0 IN IP4 127.0.0.1\r\n".getBytes(); + SipRequest req = new SipRequest( + SipMethod.INVITE, + SipUri.builder().secure(true).host("server").build(), + SipVersion.SIP_2_0, + Headers.builder() + .add("Via", "SIP/2.0/TLS host;branch=z9hG4bK1") + .add("From", ";tag=1") + .add("To", "") + .add("Call-ID", "tls@e") + .add("CSeq", "1 INVITE") + .add("Content-Type", "application/sdp") + .add("Max-Forwards", "70") + .build(), + body); + + client.send(req, server.local()).get(5, TimeUnit.SECONDS); + + assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(received).hasSize(1); + InboundMessage in = received.get(0); + assertThat(in.peer().transport()).isEqualTo(TransportType.TLS); + SipRequest got = (SipRequest) in.message(); + assertThat(got.method()).isEqualTo(SipMethod.INVITE); + assertThat(got.body()).isEqualTo(body); + } + + @Test + @Timeout(value = 15, unit = TimeUnit.SECONDS) + void serverCanReplyOverSameTlsConnection() throws Exception { + NioTlsTransport server = bindServer(); + NioTlsTransport client = bindClient(); + + CountDownLatch repliedLatch = new CountDownLatch(1); + List clientInbox = new ArrayList<>(); + client.listener(in -> { + clientInbox.add(in); + repliedLatch.countDown(); + }); + + server.listener(in -> { + SipResponse rsp = SipResponse.builder() + .status(200).reason("OK") + .headers(Headers.builder() + .add("Via", in.message().headers().first( + com.sip.message.header.HeaderName.VIA) + .orElseThrow().value()) + .add("From", ";tag=1") + .add("To", ";tag=2") + .add("Call-ID", "tls-reply@e") + .add("CSeq", "1 OPTIONS") + .build()) + .build(); + try { server.send(rsp, in.peer()).get(); } + catch (Exception ex) { throw new RuntimeException(ex); } + }); + + SipRequest req = new SipRequest( + SipMethod.OPTIONS, + SipUri.builder().secure(true).host("server").build(), + SipVersion.SIP_2_0, + Headers.builder() + .add("Via", "SIP/2.0/TLS host;branch=z9hG4bK1") + .add("From", ";tag=1") + .add("To", "") + .add("Call-ID", "tls-reply@e") + .add("CSeq", "1 OPTIONS") + .add("Max-Forwards", "70") + .build(), + new byte[0]); + client.send(req, server.local()).get(5, TimeUnit.SECONDS); + + assertThat(repliedLatch.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(clientInbox).hasSize(1); + assertThat(((SipResponse) clientInbox.get(0).message()).status()).isEqualTo(200); + } +} diff --git a/sip-transport-nio/src/test/java/com/sip/transport/nio/NioWebSocketTransportTest.java b/sip-transport-nio/src/test/java/com/sip/transport/nio/NioWebSocketTransportTest.java new file mode 100644 index 0000000..50094da --- /dev/null +++ b/sip-transport-nio/src/test/java/com/sip/transport/nio/NioWebSocketTransportTest.java @@ -0,0 +1,167 @@ +package com.sip.transport.nio; + +import com.sip.message.SipMethod; +import com.sip.message.SipRequest; +import com.sip.message.SipResponse; +import com.sip.message.SipVersion; +import com.sip.message.header.Headers; +import com.sip.message.uri.SipUri; +import com.sip.transport.InboundMessage; +import com.sip.transport.TransportType; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; + +import java.net.InetAddress; +import java.net.InetSocketAddress; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import static org.assertj.core.api.Assertions.assertThat; + +class NioWebSocketTransportTest { + + private final List open = new ArrayList<>(); + + @AfterEach + void closeAll() throws Exception { + for (int i = open.size() - 1; i >= 0; i--) { + open.get(i).close(); + } + } + + private NioWebSocketTransport bindServer() throws Exception { + InetSocketAddress addr = new InetSocketAddress(InetAddress.getLoopbackAddress(), 0); + NioWebSocketTransport t = new NioWebSocketTransport(addr, false); + t.start().get(5, TimeUnit.SECONDS); + open.add(t); + return t; + } + + private NioWebSocketTransport bindClient() throws Exception { + InetSocketAddress addr = new InetSocketAddress(InetAddress.getLoopbackAddress(), 0); + NioWebSocketTransport t = new NioWebSocketTransport(addr, true); + t.start().get(5, TimeUnit.SECONDS); + open.add(t); + return t; + } + + @Test + @Timeout(value = 10, unit = TimeUnit.SECONDS) + void clientCanSendSipOverWebSocket() throws Exception { + NioWebSocketTransport server = bindServer(); + NioWebSocketTransport client = bindClient(); + + CountDownLatch latch = new CountDownLatch(1); + List received = new ArrayList<>(); + server.listener(in -> { + received.add(in); + latch.countDown(); + }); + + SipRequest req = new SipRequest( + SipMethod.MESSAGE, + SipUri.builder().host("server").build(), + SipVersion.SIP_2_0, + Headers.builder() + .add("Via", "SIP/2.0/WS host;branch=z9hG4bK1") + .add("From", ";tag=1") + .add("To", "") + .add("Call-ID", "ws@e") + .add("CSeq", "1 MESSAGE") + .add("Content-Type", "text/plain") + .add("Max-Forwards", "70") + .build(), + "hello".getBytes()); + + client.send(req, server.local()).get(5, TimeUnit.SECONDS); + + assertThat(latch.await(5, TimeUnit.SECONDS)).isTrue(); + InboundMessage in = received.get(0); + assertThat(in.peer().transport()).isEqualTo(TransportType.WS); + SipRequest got = (SipRequest) in.message(); + assertThat(got.method()).isEqualTo(SipMethod.MESSAGE); + assertThat(new String(got.body())).isEqualTo("hello"); + } + + @Test + @Timeout(value = 10, unit = TimeUnit.SECONDS) + void multipleMessagesPerConnectionReuseHandshake() throws Exception { + NioWebSocketTransport server = bindServer(); + NioWebSocketTransport client = bindClient(); + + CountDownLatch latch = new CountDownLatch(3); + server.listener(in -> latch.countDown()); + + for (int i = 0; i < 3; i++) { + SipRequest req = new SipRequest( + SipMethod.OPTIONS, + SipUri.builder().host("server").build(), + SipVersion.SIP_2_0, + Headers.builder() + .add("Via", "SIP/2.0/WS host;branch=z9hG4bK" + i) + .add("From", ";tag=1") + .add("To", "") + .add("Call-ID", "ws-reuse@e") + .add("CSeq", (i + 1) + " OPTIONS") + .add("Max-Forwards", "70") + .build(), + new byte[0]); + client.send(req, server.local()).get(5, TimeUnit.SECONDS); + } + assertThat(latch.await(5, TimeUnit.SECONDS)).isTrue(); + assertThat(server.receivedMessages()).isEqualTo(3); + } + + @Test + @Timeout(value = 10, unit = TimeUnit.SECONDS) + void serverCanReplyOnSameWebSocket() throws Exception { + NioWebSocketTransport server = bindServer(); + NioWebSocketTransport client = bindClient(); + + CountDownLatch replied = new CountDownLatch(1); + List clientInbox = new ArrayList<>(); + client.listener(in -> { + clientInbox.add(in); + replied.countDown(); + }); + + server.listener(in -> { + SipResponse rsp = SipResponse.builder() + .status(200).reason("OK") + .headers(Headers.builder() + .add("Via", in.message().headers().first( + com.sip.message.header.HeaderName.VIA) + .orElseThrow().value()) + .add("From", ";tag=1") + .add("To", ";tag=2") + .add("Call-ID", "ws-reply@e") + .add("CSeq", "1 OPTIONS") + .build()) + .build(); + try { server.send(rsp, in.peer()).get(); } + catch (Exception ex) { throw new RuntimeException(ex); } + }); + + SipRequest req = new SipRequest( + SipMethod.OPTIONS, + SipUri.builder().host("server").build(), + SipVersion.SIP_2_0, + Headers.builder() + .add("Via", "SIP/2.0/WS host;branch=z9hG4bK1") + .add("From", ";tag=1") + .add("To", "") + .add("Call-ID", "ws-reply@e") + .add("CSeq", "1 OPTIONS") + .add("Max-Forwards", "70") + .build(), + new byte[0]); + client.send(req, server.local()).get(5, TimeUnit.SECONDS); + + assertThat(replied.await(5, TimeUnit.SECONDS)).isTrue(); + SipResponse rsp = (SipResponse) clientInbox.get(0).message(); + assertThat(rsp.status()).isEqualTo(200); + } +} diff --git a/sip-transport-nio/src/test/java/com/sip/transport/nio/TestSslContext.java b/sip-transport-nio/src/test/java/com/sip/transport/nio/TestSslContext.java new file mode 100644 index 0000000..f3d4235 --- /dev/null +++ b/sip-transport-nio/src/test/java/com/sip/transport/nio/TestSslContext.java @@ -0,0 +1,71 @@ +package com.sip.transport.nio; + +import javax.net.ssl.KeyManagerFactory; +import javax.net.ssl.SSLContext; +import javax.net.ssl.TrustManager; +import javax.net.ssl.X509ExtendedTrustManager; +import java.io.InputStream; +import java.net.Socket; +import java.security.KeyStore; +import java.security.SecureRandom; +import java.security.cert.X509Certificate; +import javax.net.ssl.SSLEngine; + +/** + * Helper that builds SSLContexts for integration tests from a pre-generated + * PKCS12 keystore shipped in {@code src/test/resources/test-keystore.p12}. + * + *

The keystore is created once via {@code keytool}; the test setup is + * therefore dependency-free at runtime. The certificate subject is + * {@code CN=localhost} with SANs for {@code localhost} and {@code 127.0.0.1} + * so SIPS hostname verification (RFC 5922) passes when peers dial the + * loopback interface.

+ */ +final class TestSslContext { + + private static final char[] PASSWORD = "changeit".toCharArray(); + private static final String KEYSTORE_RESOURCE = "/test-keystore.p12"; + + private TestSslContext() { } + + static SSLContext serverContext() throws Exception { + KeyStore keyStore = loadKeystore(); + KeyManagerFactory kmf = KeyManagerFactory.getInstance( + KeyManagerFactory.getDefaultAlgorithm()); + kmf.init(keyStore, PASSWORD); + SSLContext ctx = SSLContext.getInstance("TLSv1.3"); + ctx.init(kmf.getKeyManagers(), trustAll(), new SecureRandom()); + return ctx; + } + + static SSLContext clientContext() throws Exception { + SSLContext ctx = SSLContext.getInstance("TLSv1.3"); + ctx.init(null, trustAll(), new SecureRandom()); + return ctx; + } + + private static KeyStore loadKeystore() throws Exception { + try (InputStream in = TestSslContext.class.getResourceAsStream(KEYSTORE_RESOURCE)) { + if (in == null) { + throw new IllegalStateException( + "test keystore not found on classpath: " + KEYSTORE_RESOURCE); + } + KeyStore ks = KeyStore.getInstance("PKCS12"); + ks.load(in, PASSWORD); + return ks; + } + } + + /** Trust-everything manager. Strictly test-only — NEVER ship to production. */ + private static TrustManager[] trustAll() { + return new TrustManager[]{new X509ExtendedTrustManager() { + @Override public void checkClientTrusted(X509Certificate[] c, String t) { } + @Override public void checkClientTrusted(X509Certificate[] c, String t, Socket s) { } + @Override public void checkClientTrusted(X509Certificate[] c, String t, SSLEngine e) { } + @Override public void checkServerTrusted(X509Certificate[] c, String t) { } + @Override public void checkServerTrusted(X509Certificate[] c, String t, Socket s) { } + @Override public void checkServerTrusted(X509Certificate[] c, String t, SSLEngine e) { } + @Override public X509Certificate[] getAcceptedIssuers() { return new X509Certificate[0]; } + }}; + } +} diff --git a/sip-transport-nio/src/test/resources/test-keystore.p12 b/sip-transport-nio/src/test/resources/test-keystore.p12 new file mode 100644 index 0000000..fa96edd Binary files /dev/null and b/sip-transport-nio/src/test/resources/test-keystore.p12 differ