src/corosio/src/tls/detail/engine_driver.hpp

100.0% Lines (38/0/38) 100.0% List of functions (13/0/13)

[ ] prev/next file

engine_driver.hpp

f(x) Functions (13)

Function Calls Lines Blocks

Line	Hits	Source Code
1		//
2		// Copyright (c) 2026 Steve Gerbino
3		//
4		// Distributed under the Boost Software License, Version 1.0. (See accompanying
5		// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
6		//
7		// Official repository: https://github.com/cppalliance/corosio
8		//
9
10		#ifndef SRC_TLS_DETAIL_ENGINE_DRIVER_HPP
11		#define SRC_TLS_DETAIL_ENGINE_DRIVER_HPP
12
13		#include <boost/corosio/tls_context.hpp>
14		#include <boost/corosio/tls_stream.hpp>
15
16		#include "src/tls/detail/engine_types.hpp"
17
18		#include <boost/capy/buffers.hpp>
19		#include <boost/capy/detail/buffer_array.hpp>
20		#include <boost/capy/ex/async_mutex.hpp>
21		#include <boost/capy/io/any_stream.hpp>
22		#include <boost/capy/io_task.hpp>
23		#include <boost/capy/task.hpp>
24		#include <boost/capy/write.hpp>
25
26		#include <concepts>
27		#include <cstddef>
28		#include <cstdint>
29		#include <cstring>
30		#include <string>
31		#include <string_view>
32		#include <system_error>
33		#include <tuple>
34		#include <utility>
35		#include <vector>
36
37		// This header is instantiated in both backends' TUs, whose vendor
38		// headers cannot coexist (WolfSSL's OpenSSL-compat layer clashes
39		// with genuine OpenSSL declarations on some builds); it must stay
40		// vendor-free.
41		#ifdef OPENSSL_VERSION_NUMBER
42		#error engine_driver.hpp must stay vendor-header-free
43		#endif
44		#ifdef WOLFSSL_VERSION
45		#error engine_driver.hpp must stay vendor-header-free
46		#endif
47
48		/*
49		TLS Driver Architecture
50
51		TLS layer wrapping an underlying stream (via any_stream). One
52		read_some and one write_some may be in flight concurrently.
53
54		Engine / Driver Split. The engine (each backend's detail/engine.hpp)
55		owns the TLS session, its byte interface, and every library-result-
56		to-error-code decision; it is synchronous and transport-free. This
57		driver owns the transport, the per-direction claims, and all
58		buffering, and loops on engine verdicts:
59
60		1. Call eng_.perform(op, data, len)
61		2. output_then_retry: flush pending ciphertext, then retry
62		3. input: feed transport bytes into the engine, then retry
63		4. done / output_then_done: flush if pending, report the
64		(already mapped) ec
65
66		Renegotiation causes cross-direction I/O: a read may need to
67		write handshake data, a write may need to read. Each operation
68		services whatever direction the engine requests.
69
70		Full-Duplex Concurrency. rd_cm_ and wr_cm_ are separate transport
71		claims: a read parked on the network cannot stall a concurrent
72		write's flush, and vice versa. A cross-direction request (e.g. a
73		read needing to flush queued handshake output) takes the other
74		direction's claim itself.
75
76		read_gen_ counts successful deposits into the engine. Each engine
77		call snapshots it before running; if fill_input later finds the
78		generation has moved (a concurrent operation's flush suspension let
79		a deposit land in the meantime) it retries the engine instead of
80		issuing a redundant transport read.
81
82		The transport reads ciphertext straight into the engine's input
83		staging via input_area(), so no driver-side buffer or deposit copy
84		sits on the read path. The write path stays copy-based on purpose:
85		ciphertext drained from the engine into out_buf_[0, out_len_) is
86		decoupled from the engine's shared output staging, so a concurrent
87		reader can still emit its own records (an alert, a key update) while
88		a write is in flight. out_len_ holds only the abandoned tail a failed
89		or canceled write kept for retry; the length being written is a local,
90		never published, so a concurrent flush probe cannot mistake in-flight
91		bytes for a tail and park on the write claim behind a healthy write.
92
93		A read or write that fully satisfies the caller's buffer still owes
94		a trailing flush (queued handshake or session-ticket output must
95		reach the peer before the call can report success). If that flush
96		fails, the transfer already succeeded and the stream contract
97		forbids reporting both a full count and an error in the same
98		result, so the error is stashed in pending_flush_ec_ and surfaced
99		by the next read, write, or shutdown call instead.
100		*/
101
102		namespace boost::corosio {
103
104		namespace detail {
105
106		/** The engine surface `engine_driver` drives.
107
108		A synchronous, transport-free TLS record engine: `perform`
109		advances one operation and returns a fully mapped verdict, while
110		`put_input` / `get_output` shuttle wire bytes. The handshake
111		hooks (`check_context`, `check_session`, `prepare`) let each
112		backend gate and configure a handshake at the protocol-mandated
113		points without the driver knowing the backend's session model.
114		*/
115		template<class Engine>
116		concept tls_engine = requires(
117		Engine& e,
118		Engine const& ce,
119		engine_op op,
120		void* buf,
121		unsigned char const* in,
122		unsigned char* out,
123		std::size_t n,
124		tls_context const& ctx,
125		tls_role role,
126		std::string const& hostname,
127		std::string& alpn) {
128		{ e.perform(op, buf, n) } -> std::same_as<engine_result>;
129		{ e.put_input(in, n) } -> std::same_as<std::size_t>;
130		{ e.input_area() } -> std::same_as<std::pair<unsigned char*, std::size_t>>;
131		{ e.input_committed(n) };
132		{ ce.pending_output() } -> std::same_as<std::size_t>;
133		{ e.get_output(out, n) } -> std::same_as<std::size_t>;
134		{ ce.received_shutdown() } -> std::same_as<bool>;
135		{ ce.capture_alpn(alpn) };
136		{ e.reset() };
137		{ ce.check_context() } -> std::same_as<std::error_code>;
138		{ ce.check_session() } -> std::same_as<std::error_code>;
139		{ e.prepare(ctx, role, hostname) } -> std::same_as<std::error_code>;
140		};
141
142		/** Coroutine driver shared by every TLS backend.
143
144		Hosts the transport, the per-direction claims, the buffering, and
145		the four operation loops; the engine type supplies all TLS
146		mechanics. Each backend's stream impl instantiates this template
147		with its own engine so the driver logic exists exactly once.
148
149		@par Thread Safety
150		Distinct objects: Safe.@n
151		Shared objects: Unsafe.
152		*/
153		template<tls_engine Engine>
154		class engine_driver
155		{
156		// Large enough to hold the largest possible TLS record.
157		static constexpr std::size_t buffer_size_ = std::size_t{17} * 1024;
158
159		capy::any_stream* s_;
160		tls_context ctx_;
161		Engine eng_;
162
163		// A handshake was attempted (successfully or not); the stream
164		// must be reset before the next handshake.
165		bool used_ = false;
166
167		// Per-stream SNI/verification hostname, set via set_hostname().
168		std::string hostname_;
169
170		// ALPN protocol negotiated during the handshake (empty if none).
171		std::string alpn_selected_;
172
173		std::vector<char> out_buf_;
174
175		// One transport claim per direction so a parked reader cannot block
176		// a writer's flush. read_gen_ counts deposits into the engine: each
177		// operation captures it at its engine call and passes it to
178		// fill_input, which retries the engine instead of issuing a stale
179		// transport read when a deposit landed anywhere after the engine's
180		// verdict (including during an intervening flush suspension).
181		capy::async_mutex rd_cm_;
182		capy::async_mutex wr_cm_;
183		std::uint64_t read_gen_ = 0;
184
185		// Engine ciphertext drained but not yet written: the abandoned
186		// tail a failed or canceled transport write kept for retry.
187		// Zero while a write is in flight (that length is a local).
188		std::size_t out_len_ = 0;
189
190		// A trailing flush that fails after the engine already accepted a
191		// full payload cannot be reported alongside n == size (the stream
192		// contract forbids error + full transfer, and capy::write's
193		// composed loop discards ec exactly in that case); the error is
194		// held here and surfaced by the next operation instead.
195		std::error_code pending_flush_ec_;
196
197	39048x	capy::task<std::error_code> flush_output()
198		{
199		if (eng_.pending_output() == 0 && out_len_ == 0)
200		co_return std::error_code{};
201
202		auto [lec] = co_await wr_cm_.lock();
203		if (lec)
204		co_return lec;
205		capy::async_mutex::lock_guard wr_guard(&wr_cm_);
206
207		// Drain under the claim so records reach the wire in engine
208		// order even when both directions produce output.
209		while (eng_.pending_output() > 0 \|\| out_len_ > 0)
210		{
211		// Take the abandoned tail (from an earlier failed write in
212		// this loop or a prior flush) into a local and append fresh
213		// engine ciphertext behind it, preserving record order.
214		// out_len_ drops to zero for the duration of the write: the
215		// in-flight length lives only in `n`, so a concurrent flush
216		// probe never sees these bytes as a tail and never parks on
217		// the write claim behind a healthy write.
218		std::size_t n = out_len_;
219		out_len_ = 0;
220		if (n < out_buf_.size())
221		n += eng_.get_output(
222		reinterpret_cast<unsigned char*>(out_buf_.data()) + n,
223		out_buf_.size() - n);
224		// The loop guard just confirmed pending bytes exist, so a
225		// drain failure here is unreachable in practice; fail loudly
226		// rather than silently drop already-accepted ciphertext.
227		if (n == 0) // LCOV_EXCL_LINE unreachable: pending bytes confirmed
228		co_return make_error_code(
229		std::errc::
230		no_buffer_space); // LCOV_EXCL_LINE unreachable: transport returned 0 with no error unreachable: pending bytes confirmed
231		auto [ec, wn] = co_await capy::write(
232		*s_, capy::const_buffer(out_buf_.data(), n));
233		if (ec)
234		{
235		// wn bytes already reached the peer; keep only the unsent
236		// remainder so a post-cancellation flush retry resends
237		// neither the delivered prefix nor loses the rest.
238		std::memmove(out_buf_.data(), out_buf_.data() + wn, n - wn);
239		out_len_ = n - wn;
240		co_return ec;
241		}
242		}
243		co_return std::error_code{};
244	78096x	}
245
246		// Flushes on a fatal/alert exit path: the engine may have queued a
247		// close_notify or alert that should still reach the peer, but a
248		// caller already reporting the triggering error cannot also report
249		// this flush's own failure. flush_output() itself no-ops when
250		// nothing is pending.
251	20x	capy::task<void> best_effort_flush()
252		{
253		std::ignore = co_await flush_output();
254	40x	}
255
256		// gen is the caller's read_gen_ snapshot from its engine call; a
257		// later snapshot would miss deposits landing while the caller's
258		// flush_output was suspended.
259	3317x	capy::task<std::error_code> fill_input(std::uint64_t gen)
260		{
261		// Input already arrived since the engine's verdict; retry the
262		// engine rather than queue behind a re-parked reader.
263		if (read_gen_ != gen)
264		co_return std::error_code{};
265
266		auto [lec] = co_await rd_cm_.lock();
267		if (lec)
268		co_return lec;
269		capy::async_mutex::lock_guard rd_guard(&rd_cm_);
270
271		// Input arrived while we queued for the claim; the caller's
272		// engine retry consumes it.
273		if (read_gen_ != gen)
274		co_return std::error_code{};
275
276		// Read the transport straight into the engine's input staging:
277		// input_area() hands back the contiguous writable run, so no
278		// staging buffer or deposit copy sits between the socket and the
279		// engine.
280		auto [dst, cap] = eng_.input_area();
281		// The staging is full while the caller still wants input: its own
282		// engine retry must decrypt the staged record to free space, so
283		// report success to re-run it rather than issue a zero-length read.
284		if (cap == 0)
285		co_return std::error_code{};
286
287		auto [ec, n] = co_await s_->read_some(capy::mutable_buffer(dst, cap));
288
289		// ReadStream permits n>0 alongside ec (IOCP forwards
290		// bytes_transferred on failed completions; a canceled read can
291		// likewise deliver bytes); commit whatever arrived before
292		// surfacing ec, or the record stream loses wire bytes the
293		// transport already handed over.
294		if (n > 0)
295		{
296		eng_.input_committed(n);
297		++read_gen_;
298		co_return ec;
299		}
300
301		if (ec)
302		co_return ec;
303
304		// The transport delivered nothing without an error, so it cannot
305		// make progress: fail loudly rather than spin the engine's input
306		// retry against a staging that will never fill.
307		co_return make_error_code(
308		std::errc::
309		no_buffer_space); // LCOV_EXCL_LINE unreachable: staging cannot stay empty
310	6634x	}
311
312		// A prior read/write already reported its full transfer as success;
313		// the trailing flush error it deferred surfaces on the next call
314		// instead. Each entry point takes it exactly once.
315	66966x	std::error_code take_pending_flush_ec() noexcept
316		{
317	66966x	std::error_code ec = pending_flush_ec_;
318	66966x	pending_flush_ec_ = {};
319	66966x	return ec;
320		}
321
322		public:
323		/** Construct a driver over a transport stream.
324
325		@param s The transport; the caller keeps it alive and repoints
326		it on move via `rebind_stream`.
327		@param ctx The TLS context handed to the engine's handshake
328		preparation.
329		*/
330	2307x	engine_driver(capy::any_stream& s, tls_context ctx)
331	2307x	: s_(&s)
332	2307x	, ctx_(std::move(ctx))
333		{
334	2307x	out_buf_.resize(buffer_size_);
335	2307x	}
336
337		/// Return the engine for backend-specific setup.
338		Engine& engine() noexcept
339		{
340		return eng_;
341		}
342
343		/// Return the TLS context this driver was constructed with.
344		tls_context const& context() const noexcept
345		{
346		return ctx_;
347		}
348
349		/// Point the driver at the transport's post-move location.
350	3x	void rebind_stream(capy::any_stream& s) noexcept
351		{
352	3x	s_ = &s;
353	3x	}
354
355		/// Set the hostname applied to the next client handshake.
356	13x	void set_hostname(std::string_view hostname)
357		{
358	13x	hostname_ = hostname;
359	13x	}
360
361		/// Return the ALPN protocol negotiated by the last handshake.
362	3x	std::string_view alpn_protocol() const noexcept
363		{
364	3x	return alpn_selected_;
365		}
366
367		/// Reset the engine and every per-connection driver state.
368	65x	void reset()
369		{
370	65x	eng_.reset();
371
372	65x	out_len_ = 0;
373
374	65x	alpn_selected_.clear();
375	65x	pending_flush_ec_ = {};
376	65x	used_ = false;
377	65x	}
378
379	33441x	capy::io_task<std::size_t> do_read_some(
380		capy::detail::mutable_buffer_array<capy::detail::max_iovec_> buffers)
381		{
382		if (auto ec = take_pending_flush_ec())
383		co_return {ec, 0};
384
385		std::error_code ec;
386		std::size_t total_read = 0;
387		std::size_t const bufs_size = capy::buffer_size(buffers);
388
389		for (auto& buf : buffers)
390		{
391		char* const dest = static_cast<char*>(buf.data());
392		int const remaining = static_cast<int>(buf.size());
393		if (remaining == 0)
394		continue;
395
396		// Exits by co_return: success after the first transferred
397		// chunk, or any terminal error.
398		while (true)
399		{
400		auto const gen = read_gen_;
401		auto r = eng_.perform(
402		engine_op::read, dest, static_cast<std::size_t>(remaining));
403
404		if (r.ec)
405		{
406		// Terminal, already mapped by the engine. eof (a
407		// received close_notify) arrives as a plain done:
408		// it queues no output, so no flush is needed.
409		if (r.want == engine_want::output_then_done)
410		co_await best_effort_flush();
411		co_return {r.ec, total_read};
412		}
413
414		if (r.want == engine_want::done \|\|
415		r.want == engine_want::output_then_done)
416		{
417		total_read += r.bytes;
418
419		// r.bytes > 0 already satisfies ReadStream's "at
420		// least one byte transferred" success condition;
421		// report now rather than loop for more (another
422		// engine call could park on input). out_len_ > 0
423		// covers a retained tail the engine cannot see.
424		if (r.want == engine_want::output_then_done \|\| out_len_ > 0)
425		ec = co_await flush_output();
426		if (ec && total_read == bufs_size)
427		{
428		// First failure wins: concurrent directions share
429		// one transport failure domain; the earliest
430		// error is the meaningful one.
431		if (!pending_flush_ec_)
432		pending_flush_ec_ = ec;
433		ec = {};
434		}
435		co_return {ec, total_read};
436		}
437
438		if (r.want == engine_want::output_then_retry)
439		{
440		ec = co_await flush_output();
441		if (ec)
442		co_return {ec, total_read};
443		continue;
444		}
445
446		// want == input. Flush first: a retained tail (an
447		// earlier failed write) may hold bytes the peer needs
448		// before it will send more; no-op when nothing pends.
449		ec = co_await flush_output();
450		if (ec)
451		co_return {ec, total_read};
452
453		ec = co_await fill_input(gen);
454		if (ec)
455		{
456		ec = map_fill_error(
457		engine_op::read, ec, eng_.received_shutdown());
458		co_return {ec, total_read};
459		}
460		}
461		}
462
463		co_return {std::error_code{}, total_read};
464	66882x	}
465
466	33390x	capy::io_task<std::size_t> do_write_some(
467		capy::detail::const_buffer_array<capy::detail::max_iovec_> buffers)
468		{
469		if (auto ec = take_pending_flush_ec())
470		co_return {ec, 0};
471
472		std::error_code ec;
473		std::size_t total_written = 0;
474		std::size_t const bufs_size = capy::buffer_size(buffers);
475
476		for (auto const& buf : buffers)
477		{
478		// The engine only reads through this pointer for a write
479		// op; the cast satisfies perform's single transfer
480		// signature.
481		void* const src = const_cast<void*>(buf.data());
482		int const remaining = static_cast<int>(buf.size());
483		if (remaining == 0)
484		continue;
485
486		// Exits by co_return: success after the first transferred
487		// chunk, or any terminal error.
488		while (true)
489		{
490		auto const gen = read_gen_;
491		auto r = eng_.perform(
492		engine_op::write, src, static_cast<std::size_t>(remaining));
493
494		if (r.ec)
495		{
496		// Terminal, already mapped. eof means the peer's
497		// close_notify WAS received (an announced close);
498		// it arrives as a plain done and needs no flush.
499		if (r.want == engine_want::output_then_done)
500		co_await best_effort_flush();
501		co_return {r.ec, total_written};
502		}
503
504		if (r.want == engine_want::done \|\|
505		r.want == engine_want::output_then_done)
506		{
507		total_written += r.bytes;
508
509		// r.bytes > 0 already satisfies WriteStream's "at
510		// least one byte transferred" success condition;
511		// report now rather than loop for more. out_len_ > 0
512		// covers a retained tail the engine cannot see.
513		if (r.want == engine_want::output_then_done \|\| out_len_ > 0)
514		ec = co_await flush_output();
515		if (ec && total_written == bufs_size)
516		{
517		// First failure wins: concurrent directions share
518		// one transport failure domain; the earliest
519		// error is the meaningful one.
520		if (!pending_flush_ec_)
521		pending_flush_ec_ = ec;
522		ec = {};
523		}
524		co_return {ec, total_written};
525		}
526
527		if (r.want == engine_want::output_then_retry)
528		{
529		ec = co_await flush_output();
530		if (ec)
531		co_return {ec, total_written};
532		continue;
533		}
534
535		// want == input: a write can need a read mid-rekey; the
536		// transport eof this surfaces means the same thing it
537		// means on the read path, so map it the same way. Flush
538		// first for the same retained-tail reason as the read
539		// path.
540		ec = co_await flush_output();
541		if (ec)
542		co_return {ec, total_written};
543
544		ec = co_await fill_input(gen);
545		if (ec)
546		{
547		ec = map_fill_error(
548		engine_op::write, ec, eng_.received_shutdown());
549		co_return {ec, total_written};
550		}
551		}
552		}
553
554		co_return {std::error_code{}, total_written};
555	66780x	}
556
557	1676x	capy::io_task<> do_handshake(tls_role role)
558		{
559		// Refuse the handshake while the engine's configuration is
560		// unusable, before consuming any per-connection state.
561		if (auto cec = eng_.check_context())
562		co_return {cec};
563
564		if (used_)
565		reset();
566
567		// reset() may not have restored a usable session; refuse to
568		// hand out a handshake on it.
569		if (auto sec = eng_.check_session())
570		co_return {sec};
571
572		// A failed attempt leaves the session in a dead state; any
573		// attempt, not just a completed handshake, consumes the stream
574		// so the next handshake() starts fresh.
575		used_ = true;
576
577		if (auto pec = eng_.prepare(ctx_, role, hostname_))
578		co_return {pec};
579
580		auto const op = role == tls_role::client ? engine_op::handshake_client
581		: engine_op::handshake_server;
582
583		std::error_code ec;
584
585		while (true)
586		{
587		auto const gen = read_gen_;
588		auto r = eng_.perform(op, nullptr, 0);
589
590		if (r.ec)
591		{
592		if (r.want == engine_want::output_then_done)
593		co_await best_effort_flush();
594		co_return {r.ec};
595		}
596
597		if (r.want == engine_want::done \|\|
598		r.want == engine_want::output_then_done)
599		{
600		eng_.capture_alpn(alpn_selected_);
601		ec = co_await flush_output();
602		co_return {ec};
603		}
604
605		if (r.want == engine_want::output_then_retry)
606		{
607		// Must flush (e.g. ClientHello) before reading the
608		// peer's reply.
609		ec = co_await flush_output();
610		if (ec)
611		co_return {ec};
612		continue;
613		}
614
615		// want == input. The flush is a no-op unless a retained
616		// tail pends; the fill error passes through map_fill_error
617		// unchanged (no close is clean before the session is
618		// established).
619		ec = co_await flush_output();
620		if (ec)
621		co_return {ec};
622
623		ec = co_await fill_input(gen);
624		if (ec)
625		{
626		ec = map_fill_error(op, ec, eng_.received_shutdown());
627		co_return {ec};
628		}
629		}
630	3352x	}
631
632	135x	capy::io_task<> do_shutdown()
633		{
634		if (auto ec = take_pending_flush_ec())
635		co_return {ec};
636
637		std::error_code ec;
638
639		while (true)
640		{
641		auto const gen = read_gen_;
642		auto r = eng_.perform(engine_op::shutdown, nullptr, 0);
643
644		if (r.ec)
645		{
646		if (r.want == engine_want::output_then_done)
647		co_await best_effort_flush();
648		co_return {r.ec};
649		}
650
651		if (r.want == engine_want::done \|\|
652		r.want == engine_want::output_then_done)
653		{
654		// Covers both the bidirectional-complete result and the
655		// engine's shutdown-after-close success mapping.
656		ec = co_await flush_output();
657		co_return {ec};
658		}
659
660		if (r.want == engine_want::output_then_retry)
661		{
662		// Sends our close_notify before parking for the peer's.
663		ec = co_await flush_output();
664		if (ec)
665		co_return {ec};
666		continue;
667		}
668
669		// want == input: awaiting the peer's close_notify. It may
670		// already have been deposited (and consumed by a concurrent
671		// reader) during a flush; fill_input then returns without
672		// reading and the loop retries the engine instead of parking.
673		ec = co_await flush_output();
674		if (ec)
675		co_return {ec};
676
677		ec = co_await fill_input(gen);
678		if (ec)
679		{
680		ec = map_fill_error(
681		engine_op::shutdown, ec, eng_.received_shutdown());
682		co_return {ec};
683		}
684		}
685	270x	}
686		};
687
688		} // namespace detail
689
690		} // namespace boost::corosio
691
692		#endif
693