Phase 8：NioTlsTransport + NioWebSocketTransport（TLS + WS 完整实装）

.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

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).
+ *
+ * <p>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.</p>
+ *
+ * <h2>Hostname verification (RFC 5922)</h2>
+ * <p>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.</p>
+ */
+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<InetSocketAddress, Connection> 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<Void> 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<Void> 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);
+        }
+    }
+}