feat(M1): TCP/TLS control plane -- auth, channels, ephemeral text

Implements the full M1 milestone. Two clients authenticate over TLS 1.3 (guest + Argon2id password) and exchange channel + private text messages through a real server. All five ctest --preset m1-dev tests pass in ~1 s. Key components added: - vcpkg baseline + m1-dev preset (protobuf/mbedTLS/libsodium/asio/sqlite3) - FrameCodec feed+emit, encode/decode_envelope, protobuf codegen - TcpServerConn with blocking TLS handshake thread + tls_read_loop - TlsContext (mbedTLS 1.3, ECDSA-P256 self-signed cert, TOFU on client) - WorkerPool (3 threads, used for Argon2id) - Database: SQLite + libsodium Argon2id, account lifecycle, bootstrap admin - ServerIdentityManager: Ed25519 key + cert generate/persist/fingerprint - ConnSession state machine: WaitingHello -> WaitingAuth -> Authenticated - SessionRegistry: channel tree, user map, text routing, broadcast - vc_client full M1 C ABI: connect/TLS/handshake/auth/text/disconnect - voicecat-admin CLI: account add/reset/del/list - test_m1_integration: M1 exit criterion, verified green Bug fixed: double-framing in ConnSession::send_envelope -- encode_envelope was adding the [4-byte len] prefix, then TcpServerConn::send_frame added a second one, causing the client to parse [len][proto] as protobuf (silent failure). Fixed by serializing raw protobuf bytes in send_envelope and letting send_frame apply the single length prefix. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-06-15 23:48:44 +02:00
parent b332b0972b
commit 63f457fc54
42 changed files with 4180 additions and 190 deletions
--- a/core/src/net/transport.cpp
+++ b/core/src/net/transport.cpp
@@ -1,8 +1,356 @@
 #include "net/transport.h"

+#ifdef VOICECAT_HAS_NET
+
+#include <cstring>
+
+#include "crypto/crypto.h"
+
 namespace voicecat::net {

-// M0 stub. Subsystem brought up in M1 (TCP/TLS) and M2 (UDP). See docs/protocol.md,
-// docs/voice.md, and AGENTS.md "Suggested first steps".
+// ── TcpControlChannel ────────────────────────────────────────────────────────
+
+TcpControlChannel::TcpControlChannel(TcpChannelCallbacks cbs)
+    : work_guard_(asio::make_work_guard(io_)),
+      socket_(io_),
+      strand_(io_.get_executor()),
+      cbs_(std::move(cbs)) {
+    net_thread_ = std::thread([this] { run_loop(); });
+}
+
+TcpControlChannel::~TcpControlChannel() { close(); }
+
+void TcpControlChannel::run_loop() { io_.run(); }
+
+void TcpControlChannel::async_connect(const std::string& host, uint16_t port) {
+    auto resolver = std::make_shared<asio::ip::tcp::resolver>(io_);
+    resolver->async_resolve(
+        host, std::to_string(port),
+        [this, resolver](std::error_code ec, asio::ip::tcp::resolver::results_type eps) {
+            if (ec) {
+                if (cbs_.on_connect_error) cbs_.on_connect_error(ec);
+                return;
+            }
+            asio::async_connect(socket_, eps,
+                [this](std::error_code ec2, const asio::ip::tcp::endpoint&) {
+                    if (ec2) {
+                        if (cbs_.on_connect_error) cbs_.on_connect_error(ec2);
+                        return;
+                    }
+                    connected_.store(true, std::memory_order_release);
+                    if (cbs_.on_connected) cbs_.on_connected();
+                    start_read();
+                });
+        });
+}
+
+void TcpControlChannel::send_frame(std::vector<uint8_t> payload) {
+    std::vector<uint8_t> wire;
+    protocol::FrameCodec::emit(payload, wire);
+    asio::post(strand_, [this, w = std::move(wire)]() mutable {
+        send_queue_.push_back(std::move(w));
+        if (!sending_) do_send();
+    });
+}
+
+void TcpControlChannel::do_send() {
+    if (send_queue_.empty()) { sending_ = false; return; }
+    sending_ = true;
+    auto& front = send_queue_.front();
+    asio::async_write(socket_,
+        asio::buffer(front),
+        asio::bind_executor(strand_,
+            [this](std::error_code ec, std::size_t) {
+                if (ec) {
+                    connected_.store(false, std::memory_order_release);
+                    if (cbs_.on_error) cbs_.on_error(ec);
+                    return;
+                }
+                send_queue_.pop_front();
+                do_send();
+            }));
+}
+
+void TcpControlChannel::start_read() {
+    asio::async_read(socket_, asio::buffer(len_buf_, 4),
+        [this](std::error_code ec, std::size_t n) { handle_length(ec, n); });
+}
+
+void TcpControlChannel::handle_length(std::error_code ec, std::size_t) {
+    if (ec) {
+        connected_.store(false, std::memory_order_release);
+        if (ec == asio::error::eof || ec == asio::error::connection_reset) {
+            if (cbs_.on_disconnected) cbs_.on_disconnected();
+        } else {
+            if (cbs_.on_error) cbs_.on_error(ec);
+        }
+        return;
+    }
+    uint32_t length =
+        (static_cast<uint32_t>(len_buf_[0]) << 24) |
+        (static_cast<uint32_t>(len_buf_[1]) << 16) |
+        (static_cast<uint32_t>(len_buf_[2]) <<  8) |
+         static_cast<uint32_t>(len_buf_[3]);
+
+    if (length > protocol::kMaxFrameBytes) {
+        if (cbs_.on_error) cbs_.on_error(asio::error::message_size);
+        return;
+    }
+    if (length == 0) {
+        if (cbs_.on_frame) cbs_.on_frame({});
+        start_read();
+        return;
+    }
+    body_buf_.resize(length);
+    asio::async_read(socket_, asio::buffer(body_buf_),
+        [this, length](std::error_code ec, std::size_t n) { handle_body(length, ec, n); });
+}
+
+void TcpControlChannel::handle_body(uint32_t, std::error_code ec, std::size_t) {
+    if (ec) {
+        connected_.store(false, std::memory_order_release);
+        if (ec == asio::error::eof || ec == asio::error::connection_reset) {
+            if (cbs_.on_disconnected) cbs_.on_disconnected();
+        } else {
+            if (cbs_.on_error) cbs_.on_error(ec);
+        }
+        return;
+    }
+    if (cbs_.on_frame) cbs_.on_frame(body_buf_);
+    start_read();
+}
+
+void TcpControlChannel::close() {
+    if (closing_.exchange(true)) return;
+    asio::post(io_, [this] {
+        std::error_code ignored;
+        socket_.shutdown(asio::ip::tcp::socket::shutdown_both, ignored);
+        socket_.close(ignored);
+    });
+    work_guard_.reset();
+    if (net_thread_.joinable()) net_thread_.join();
+}
+
+// ── TcpServerConn ────────────────────────────────────────────────────────────
+
+TcpServerConn::TcpServerConn(asio::ip::tcp::socket socket, TcpChannelCallbacks cbs)
+    : socket_(std::move(socket)),
+      strand_(asio::make_strand(socket_.get_executor())),
+      cbs_(std::move(cbs)) {}
+
+TcpServerConn::TcpServerConn(asio::ip::tcp::socket socket, TcpChannelCallbacks cbs,
+                              std::unique_ptr<crypto::TlsContext> tls)
+    : socket_(std::move(socket)),
+      strand_(asio::make_strand(socket_.get_executor())),
+      cbs_(std::move(cbs)),
+      tls_(std::move(tls)) {}
+
+TcpServerConn::~TcpServerConn() {
+    close();
+    if (tls_thread_.joinable()) {
+        if (std::this_thread::get_id() == tls_thread_.get_id()) {
+            tls_thread_.detach();  // being destroyed from our own TLS thread — detach safely
+        } else {
+            tls_thread_.join();
+        }
+    }
+}
+
+void TcpServerConn::start() {
+    if (tls_) {
+        // Run TLS handshake on a temporary thread so we don't block the io_context.
+        auto self = shared_from_this();
+        std::thread([self] {
+            std::string err;
+            int fd = static_cast<int>(self->socket_.native_handle());
+            if (!self->tls_->handshake(fd, err)) {
+                if (!self->closing_.exchange(true)) {
+                    if (self->cbs_.on_error) {
+                        asio::post(self->strand_, [self] {
+                            self->cbs_.on_error(
+                                std::make_error_code(std::errc::connection_reset));
+                        });
+                    }
+                }
+                return;
+            }
+            self->connected_.store(true, std::memory_order_release);
+            // 50 ms timeout so tls_read_loop can drain the send queue between reads.
+            self->tls_->set_read_timeout(50);
+            self->tls_thread_ = std::thread([self] { self->tls_read_loop(); });
+        }).detach();
+    } else {
+        connected_.store(true, std::memory_order_release);
+        start_read();
+    }
+}
+
+void TcpServerConn::tls_read_loop() {
+    std::vector<uint8_t> buf(16384);
+    while (!closing_.load(std::memory_order_acquire)) {
+        tls_drain_sends();
+
+        int n = tls_->read(buf.data(), buf.size());
+        if (crypto::TlsContext::is_timeout_error(n)) continue;
+        if (n <= 0) break;
+
+        std::vector<std::vector<uint8_t>> frames;
+        if (!codec_.feed(buf.data(), static_cast<size_t>(n), frames)) break;
+        for (auto& frame : frames) {
+            if (cbs_.on_frame) cbs_.on_frame(std::move(frame));
+        }
+    }
+    connected_.store(false, std::memory_order_release);
+    if (cbs_.on_disconnected) cbs_.on_disconnected();
+}
+
+void TcpServerConn::tls_drain_sends() {
+    while (true) {
+        std::vector<uint8_t> frame;
+        {
+            std::lock_guard lk(tls_send_mutex_);
+            if (tls_send_queue_.empty()) return;
+            frame = std::move(tls_send_queue_.front());
+            tls_send_queue_.pop_front();
+        }
+        size_t off = 0;
+        while (off < frame.size()) {
+            int n = tls_->write(frame.data() + off, frame.size() - off);
+            if (n <= 0) { closing_.store(true, std::memory_order_release); return; }
+            off += static_cast<size_t>(n);
+        }
+    }
+}
+
+void TcpServerConn::start_read() {
+    auto self = shared_from_this();
+    asio::async_read(socket_, asio::buffer(len_buf_, 4),
+        asio::bind_executor(strand_,
+            [this, self](std::error_code ec, std::size_t n) { handle_length(ec, n); }));
+}
+
+void TcpServerConn::handle_length(std::error_code ec, std::size_t) {
+    if (ec) {
+        connected_.store(false, std::memory_order_release);
+        if (ec == asio::error::eof || ec == asio::error::connection_reset) {
+            if (cbs_.on_disconnected) cbs_.on_disconnected();
+        } else {
+            if (cbs_.on_error) cbs_.on_error(ec);
+        }
+        return;
+    }
+    uint32_t length =
+        (static_cast<uint32_t>(len_buf_[0]) << 24) |
+        (static_cast<uint32_t>(len_buf_[1]) << 16) |
+        (static_cast<uint32_t>(len_buf_[2]) <<  8) |
+         static_cast<uint32_t>(len_buf_[3]);
+
+    if (length > protocol::kMaxFrameBytes) {
+        if (cbs_.on_error) cbs_.on_error(asio::error::message_size);
+        return;
+    }
+    if (length == 0) {
+        if (cbs_.on_frame) cbs_.on_frame({});
+        start_read();
+        return;
+    }
+    body_buf_.resize(length);
+    auto self = shared_from_this();
+    asio::async_read(socket_, asio::buffer(body_buf_),
+        asio::bind_executor(strand_,
+            [this, self, length](std::error_code ec, std::size_t n) {
+                handle_body(length, ec, n);
+            }));
+}
+
+void TcpServerConn::handle_body(uint32_t, std::error_code ec, std::size_t) {
+    if (ec) {
+        connected_.store(false, std::memory_order_release);
+        if (ec == asio::error::eof || ec == asio::error::connection_reset) {
+            if (cbs_.on_disconnected) cbs_.on_disconnected();
+        } else {
+            if (cbs_.on_error) cbs_.on_error(ec);
+        }
+        return;
+    }
+    if (cbs_.on_frame) cbs_.on_frame(body_buf_);
+    start_read();
+}
+
+void TcpServerConn::send_frame(std::vector<uint8_t> payload) {
+    std::vector<uint8_t> wire;
+    protocol::FrameCodec::emit(payload, wire);
+    if (tls_) {
+        std::lock_guard lk(tls_send_mutex_);
+        tls_send_queue_.push_back(std::move(wire));
+    } else {
+        auto self = shared_from_this();
+        asio::post(strand_, [this, self, w = std::move(wire)]() mutable {
+            send_queue_.push_back(std::move(w));
+            if (!sending_) do_send();
+        });
+    }
+}
+
+void TcpServerConn::do_send() {
+    if (send_queue_.empty()) { sending_ = false; return; }
+    sending_ = true;
+    auto self = shared_from_this();
+    auto& front = send_queue_.front();
+    asio::async_write(socket_,
+        asio::buffer(front),
+        asio::bind_executor(strand_,
+            [this, self](std::error_code ec, std::size_t) {
+                if (ec) {
+                    connected_.store(false, std::memory_order_release);
+                    if (cbs_.on_error) cbs_.on_error(ec);
+                    return;
+                }
+                send_queue_.pop_front();
+                do_send();
+            }));
+}
+
+void TcpServerConn::close() {
+    if (closing_.exchange(true)) return;
+    std::error_code ignored;
+    socket_.shutdown(asio::ip::tcp::socket::shutdown_both, ignored);
+    socket_.close(ignored);
+    connected_.store(false, std::memory_order_release);
+    // In non-TLS mode, the Asio async chain will naturally stop when the socket closes.
+}
+
+// ── TcpAcceptor ─────────────────────────────────────────────────────────────
+
+TcpAcceptor::TcpAcceptor(asio::io_context& io, uint16_t port, ConnFactory factory)
+    : acceptor_(io, asio::ip::tcp::endpoint(asio::ip::tcp::v4(), port)),
+      factory_(std::move(factory)) {
+    acceptor_.set_option(asio::ip::tcp::acceptor::reuse_address(true));
+}
+
+void TcpAcceptor::start() { do_accept(); }
+
+void TcpAcceptor::stop() {
+    stopped_ = true;
+    std::error_code ignored;
+    acceptor_.close(ignored);
+}
+
+void TcpAcceptor::do_accept() {
+    if (stopped_) return;
+    acceptor_.async_accept(
+        [this](std::error_code ec, asio::ip::tcp::socket socket) {
+            if (ec) {
+                if (!stopped_) do_accept();
+                return;
+            }
+            socket.set_option(asio::ip::tcp::no_delay(true));
+            auto conn = factory_(std::move(socket));
+            if (conn) conn->start();
+            do_accept();
+        });
+}

 }  // namespace voicecat::net
+
+#endif  // VOICECAT_HAS_NET