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

100.0% Lines (38/38) 100.0% List of functions (13/13) 100.0% Branches (8/8)

[ ] prev/next file

engine_driver.hpp

f(x) Functions (13)

Function Calls Lines Branches Blocks

Line	Branch	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 =
117			requires(
118			Engine& e,
119			Engine const& ce,
120			engine_op op,
121			void* buf,
122			unsigned char const* in,
123			unsigned char* out,
124			std::size_t n,
125			tls_context const& ctx,
126			tls_role role,
127			std::string const& hostname,
128			std::string& alpn)
129			{
130			{ e.perform(op, buf, n) } -> std::same_as<engine_result>;
131			{ e.put_input(in, n) } -> std::same_as<std::size_t>;
132			{ e.input_area() }
133			-> std::same_as<std::pair<unsigned char*, std::size_t>>;
134			{ e.input_committed(n) };
135			{ ce.pending_output() } -> std::same_as<std::size_t>;
136			{ e.get_output(out, n) } -> std::same_as<std::size_t>;
137			{ ce.received_shutdown() } -> std::same_as<bool>;
138			{ ce.capture_alpn(alpn) };
139			{ e.reset() };
140			{ ce.check_context() } -> std::same_as<std::error_code>;
141			{ ce.check_session() } -> std::same_as<std::error_code>;
142			{ e.prepare(ctx, role, hostname) } -> std::same_as<std::error_code>;
143			};
144
145			/** Coroutine driver shared by every TLS backend.
146
147			Hosts the transport, the per-direction claims, the buffering, and
148			the four operation loops; the engine type supplies all TLS
149			mechanics. Each backend's stream impl instantiates this template
150			with its own engine so the driver logic exists exactly once.
151
152			@par Thread Safety
153			Distinct objects: Safe.@n
154			Shared objects: Unsafe.
155			*/
156			template<tls_engine Engine>
157			class engine_driver
158			{
159			// Large enough to hold the largest possible TLS record.
160			static constexpr std::size_t buffer_size_ = std::size_t{17} * 1024;
161
162			capy::any_stream* s_;
163			tls_context ctx_;
164			Engine eng_;
165
166			// A handshake was attempted (successfully or not); the stream
167			// must be reset before the next handshake.
168			bool used_ = false;
169
170			// Per-stream SNI/verification hostname, set via set_hostname().
171			std::string hostname_;
172
173			// ALPN protocol negotiated during the handshake (empty if none).
174			std::string alpn_selected_;
175
176			std::vector<char> out_buf_;
177
178			// One transport claim per direction so a parked reader cannot block
179			// a writer's flush. read_gen_ counts deposits into the engine: each
180			// operation captures it at its engine call and passes it to
181			// fill_input, which retries the engine instead of issuing a stale
182			// transport read when a deposit landed anywhere after the engine's
183			// verdict (including during an intervening flush suspension).
184			capy::async_mutex rd_cm_;
185			capy::async_mutex wr_cm_;
186			std::uint64_t read_gen_ = 0;
187
188			// Engine ciphertext drained but not yet written: the abandoned
189			// tail a failed or canceled transport write kept for retry.
190			// Zero while a write is in flight (that length is a local).
191			std::size_t out_len_ = 0;
192
193			// A trailing flush that fails after the engine already accepted a
194			// full payload cannot be reported alongside n == size (the stream
195			// contract forbids error + full transfer, and capy::write's
196			// composed loop discards ec exactly in that case); the error is
197			// held here and surfaced by the next operation instead.
198			std::error_code pending_flush_ec_;
199
200	1/1 ✓ Branch 5 → 6 taken 100916 times.	100916x	capy::task<std::error_code> flush_output()
201			{
202			if (eng_.pending_output() == 0 && out_len_ == 0)
203			co_return std::error_code{};
204
205			auto [lec] = co_await wr_cm_.lock();
206			if (lec)
207			co_return lec;
208			capy::async_mutex::lock_guard wr_guard(&wr_cm_);
209
210			// Drain under the claim so records reach the wire in engine
211			// order even when both directions produce output.
212			while (eng_.pending_output() > 0 \|\| out_len_ > 0)
213			{
214			// Take the abandoned tail (from an earlier failed write in
215			// this loop or a prior flush) into a local and append fresh
216			// engine ciphertext behind it, preserving record order.
217			// out_len_ drops to zero for the duration of the write: the
218			// in-flight length lives only in `n`, so a concurrent flush
219			// probe never sees these bytes as a tail and never parks on
220			// the write claim behind a healthy write.
221			std::size_t n = out_len_;
222			out_len_ = 0;
223			if (n < out_buf_.size())
224			n += eng_.get_output(
225			reinterpret_cast<unsigned char*>(out_buf_.data()) + n,
226			out_buf_.size() - n);
227			// The loop guard just confirmed pending bytes exist, so a
228			// drain failure here is unreachable in practice; fail loudly
229			// rather than silently drop already-accepted ciphertext.
230			if (n == 0)
231			co_return make_error_code(std::errc::no_buffer_space);
232			auto [ec, wn] = co_await capy::write(
233			*s_, capy::const_buffer(out_buf_.data(), n));
234			if (ec)
235			{
236			// wn bytes already reached the peer; keep only the unsent
237			// remainder so a post-cancellation flush retry resends
238			// neither the delivered prefix nor loses the rest.
239			std::memmove(
240			out_buf_.data(), out_buf_.data() + wn, n - wn);
241			out_len_ = n - wn;
242			co_return ec;
243			}
244			}
245			co_return std::error_code{};
246		201832x	}
247
248			// Flushes on a fatal/alert exit path: the engine may have queued a
249			// close_notify or alert that should still reach the peer, but a
250			// caller already reporting the triggering error cannot also report
251			// this flush's own failure. flush_output() itself no-ops when
252			// nothing is pending.
253	1/1 ✓ Branch 5 → 6 taken 19 times.	19x	capy::task<void> best_effort_flush()
254			{
255			std::ignore = co_await flush_output();
256		38x	}
257
258			// gen is the caller's read_gen_ snapshot from its engine call; a
259			// later snapshot would miss deposits landing while the caller's
260			// flush_output was suspended.
261	1/1 ✓ Branch 5 → 6 taken 50216 times.	50216x	capy::task<std::error_code> fill_input(std::uint64_t gen)
262			{
263			// Input already arrived since the engine's verdict; retry the
264			// engine rather than queue behind a re-parked reader.
265			if (read_gen_ != gen)
266			co_return std::error_code{};
267
268			auto [lec] = co_await rd_cm_.lock();
269			if (lec)
270			co_return lec;
271			capy::async_mutex::lock_guard rd_guard(&rd_cm_);
272
273			// Input arrived while we queued for the claim; the caller's
274			// engine retry consumes it.
275			if (read_gen_ != gen)
276			co_return std::error_code{};
277
278			// Read the transport straight into the engine's input staging:
279			// input_area() hands back the contiguous writable run, so no
280			// staging buffer or deposit copy sits between the socket and the
281			// engine.
282			auto [dst, cap] = eng_.input_area();
283			// The staging is full while the caller still wants input: its own
284			// engine retry must decrypt the staged record to free space, so
285			// report success to re-run it rather than issue a zero-length read.
286			if (cap == 0)
287			co_return std::error_code{};
288
289			auto [ec, n] =
290			co_await s_->read_some(capy::mutable_buffer(dst, cap));
291
292			// ReadStream permits n>0 alongside ec (IOCP forwards
293			// bytes_transferred on failed completions; a canceled read can
294			// likewise deliver bytes); commit whatever arrived before
295			// surfacing ec, or the record stream loses wire bytes the
296			// transport already handed over.
297			if (n > 0)
298			{
299			eng_.input_committed(n);
300			++read_gen_;
301			co_return ec;
302			}
303
304			if (ec)
305			co_return ec;
306
307			// The transport delivered nothing without an error, so it cannot
308			// make progress: fail loudly rather than spin the engine's input
309			// retry against a staging that will never fill.
310			co_return make_error_code(std::errc::no_buffer_space);
311		100432x	}
312
313			// A prior read/write already reported its full transfer as success;
314			// the trailing flush error it deferred surfaces on the next call
315			// instead. Each entry point takes it exactly once.
316		93109x	std::error_code take_pending_flush_ec() noexcept
317			{
318		93109x	std::error_code ec = pending_flush_ec_;
319		93109x	pending_flush_ec_ = {};
320		93109x	return ec;
321			}
322
323			public:
324			/** Construct a driver over a transport stream.
325
326			@param s The transport; the caller keeps it alive and repoints
327			it on move via `rebind_stream`.
328			@param ctx The TLS context handed to the engine's handshake
329			preparation.
330			*/
331		4191x	engine_driver(capy::any_stream& s, tls_context ctx)
332		4191x	: s_(&s)
333		4191x	, ctx_(std::move(ctx))
334			{
335	1/1 ✓ Branch 11 → 12 taken 4191 times.	4191x	out_buf_.resize(buffer_size_);
336		4191x	}
337
338			/// Return the engine for backend-specific setup.
339			Engine&
340			engine() noexcept
341			{
342			return eng_;
343			}
344
345			/// Return the TLS context this driver was constructed with.
346			tls_context const&
347			context() const noexcept
348			{
349			return ctx_;
350			}
351
352			/// Point the driver at the transport's post-move location.
353			void
354		3x	rebind_stream(capy::any_stream& s) noexcept
355			{
356		3x	s_ = &s;
357		3x	}
358
359			/// Set the hostname applied to the next client handshake.
360			void
361		13x	set_hostname(std::string_view hostname)
362			{
363		13x	hostname_ = hostname;
364		13x	}
365
366			/// Return the ALPN protocol negotiated by the last handshake.
367			std::string_view
368		3x	alpn_protocol() const noexcept
369			{
370		3x	return alpn_selected_;
371			}
372
373			/// Reset the engine and every per-connection driver state.
374		65x	void reset()
375			{
376		65x	eng_.reset();
377
378		65x	out_len_ = 0;
379
380		65x	alpn_selected_.clear();
381		65x	pending_flush_ec_ = {};
382		65x	used_ = false;
383		65x	}
384
385			capy::io_task<std::size_t>
386	1/1 ✓ Branch 6 → 7 taken 46539 times.	46539x	do_read_some(
387			capy::detail::mutable_buffer_array<capy::detail::max_iovec_> buffers)
388			{
389			if (auto ec = take_pending_flush_ec())
390			co_return {ec, 0};
391
392			std::error_code ec;
393			std::size_t total_read = 0;
394			std::size_t const bufs_size = capy::buffer_size(buffers);
395
396			for (auto& buf : buffers)
397			{
398			char* const dest = static_cast<char*>(buf.data());
399			int const remaining = static_cast<int>(buf.size());
400			if (remaining == 0)
401			continue;
402
403			// Exits by co_return: success after the first transferred
404			// chunk, or any terminal error.
405			while (true)
406			{
407			auto const gen = read_gen_;
408			auto r = eng_.perform(
409			engine_op::read, dest,
410			static_cast<std::size_t>(remaining));
411
412			if (r.ec)
413			{
414			// Terminal, already mapped by the engine. eof (a
415			// received close_notify) arrives as a plain done:
416			// it queues no output, so no flush is needed.
417			if (r.want == engine_want::output_then_done)
418			co_await best_effort_flush();
419			co_return {r.ec, total_read};
420			}
421
422			if (r.want == engine_want::done \|\|
423			r.want == engine_want::output_then_done)
424			{
425			total_read += r.bytes;
426
427			// r.bytes > 0 already satisfies ReadStream's "at
428			// least one byte transferred" success condition;
429			// report now rather than loop for more (another
430			// engine call could park on input). out_len_ > 0
431			// covers a retained tail the engine cannot see.
432			if (r.want == engine_want::output_then_done \|\|
433			out_len_ > 0)
434			ec = co_await flush_output();
435			if (ec && total_read == bufs_size)
436			{
437			// First failure wins: concurrent directions share
438			// one transport failure domain; the earliest
439			// error is the meaningful one.
440			if (!pending_flush_ec_)
441			pending_flush_ec_ = ec;
442			ec = {};
443			}
444			co_return {ec, total_read};
445			}
446
447			if (r.want == engine_want::output_then_retry)
448			{
449			ec = co_await flush_output();
450			if (ec)
451			co_return {ec, total_read};
452			continue;
453			}
454
455			// want == input. Flush first: a retained tail (an
456			// earlier failed write) may hold bytes the peer needs
457			// before it will send more; no-op when nothing pends.
458			ec = co_await flush_output();
459			if (ec)
460			co_return {ec, total_read};
461
462			ec = co_await fill_input(gen);
463			if (ec)
464			{
465			ec = map_fill_error(
466			engine_op::read, ec, eng_.received_shutdown());
467			co_return {ec, total_read};
468			}
469			}
470			}
471
472			co_return {std::error_code{}, total_read};
473		93078x	}
474
475			capy::io_task<std::size_t>
476	1/1 ✓ Branch 6 → 7 taken 46425 times.	46425x	do_write_some(
477			capy::detail::const_buffer_array<capy::detail::max_iovec_> buffers)
478			{
479			if (auto ec = take_pending_flush_ec())
480			co_return {ec, 0};
481
482			std::error_code ec;
483			std::size_t total_written = 0;
484			std::size_t const bufs_size = capy::buffer_size(buffers);
485
486			for (auto const& buf : buffers)
487			{
488			// The engine only reads through this pointer for a write
489			// op; the cast satisfies perform's single transfer
490			// signature.
491			void* const src = const_cast<void*>(buf.data());
492			int const remaining = static_cast<int>(buf.size());
493			if (remaining == 0)
494			continue;
495
496			// Exits by co_return: success after the first transferred
497			// chunk, or any terminal error.
498			while (true)
499			{
500			auto const gen = read_gen_;
501			auto r = eng_.perform(
502			engine_op::write, src,
503			static_cast<std::size_t>(remaining));
504
505			if (r.ec)
506			{
507			// Terminal, already mapped. eof means the peer's
508			// close_notify WAS received (an announced close);
509			// it arrives as a plain done and needs no flush.
510			if (r.want == engine_want::output_then_done)
511			co_await best_effort_flush();
512			co_return {r.ec, total_written};
513			}
514
515			if (r.want == engine_want::done \|\|
516			r.want == engine_want::output_then_done)
517			{
518			total_written += r.bytes;
519
520			// r.bytes > 0 already satisfies WriteStream's "at
521			// least one byte transferred" success condition;
522			// report now rather than loop for more. out_len_ > 0
523			// covers a retained tail the engine cannot see.
524			if (r.want == engine_want::output_then_done \|\|
525			out_len_ > 0)
526			ec = co_await flush_output();
527			if (ec && total_written == bufs_size)
528			{
529			// First failure wins: concurrent directions share
530			// one transport failure domain; the earliest
531			// error is the meaningful one.
532			if (!pending_flush_ec_)
533			pending_flush_ec_ = ec;
534			ec = {};
535			}
536			co_return {ec, total_written};
537			}
538
539			if (r.want == engine_want::output_then_retry)
540			{
541			ec = co_await flush_output();
542			if (ec)
543			co_return {ec, total_written};
544			continue;
545			}
546
547			// want == input: a write can need a read mid-rekey; the
548			// transport eof this surfaces means the same thing it
549			// means on the read path, so map it the same way. Flush
550			// first for the same retained-tail reason as the read
551			// path.
552			ec = co_await flush_output();
553			if (ec)
554			co_return {ec, total_written};
555
556			ec = co_await fill_input(gen);
557			if (ec)
558			{
559			ec = map_fill_error(
560			engine_op::write, ec, eng_.received_shutdown());
561			co_return {ec, total_written};
562			}
563			}
564			}
565
566			co_return {std::error_code{}, total_written};
567		92850x	}
568
569	1/1 ✓ Branch 5 → 6 taken 2934 times.	2934x	capy::io_task<> do_handshake(tls_role role)
570			{
571			// Refuse the handshake while the engine's configuration is
572			// unusable, before consuming any per-connection state.
573			if (auto cec = eng_.check_context())
574			co_return {cec};
575
576			if (used_)
577			reset();
578
579			// reset() may not have restored a usable session; refuse to
580			// hand out a handshake on it.
581			if (auto sec = eng_.check_session())
582			co_return {sec};
583
584			// A failed attempt leaves the session in a dead state; any
585			// attempt, not just a completed handshake, consumes the stream
586			// so the next handshake() starts fresh.
587			used_ = true;
588
589			if (auto pec = eng_.prepare(ctx_, role, hostname_))
590			co_return {pec};
591
592			auto const op = role == tls_role::client
593			? engine_op::handshake_client
594			: engine_op::handshake_server;
595
596			std::error_code ec;
597
598			while (true)
599			{
600			auto const gen = read_gen_;
601			auto r = eng_.perform(op, nullptr, 0);
602
603			if (r.ec)
604			{
605			if (r.want == engine_want::output_then_done)
606			co_await best_effort_flush();
607			co_return {r.ec};
608			}
609
610			if (r.want == engine_want::done \|\|
611			r.want == engine_want::output_then_done)
612			{
613			eng_.capture_alpn(alpn_selected_);
614			ec = co_await flush_output();
615			co_return {ec};
616			}
617
618			if (r.want == engine_want::output_then_retry)
619			{
620			// Must flush (e.g. ClientHello) before reading the
621			// peer's reply.
622			ec = co_await flush_output();
623			if (ec)
624			co_return {ec};
625			continue;
626			}
627
628			// want == input. The flush is a no-op unless a retained
629			// tail pends; the fill error passes through map_fill_error
630			// unchanged (no close is clean before the session is
631			// established).
632			ec = co_await flush_output();
633			if (ec)
634			co_return {ec};
635
636			ec = co_await fill_input(gen);
637			if (ec)
638			{
639			ec = map_fill_error(op, ec, eng_.received_shutdown());
640			co_return {ec};
641			}
642			}
643		5868x	}
644
645	1/1 ✓ Branch 5 → 6 taken 145 times.	145x	capy::io_task<> do_shutdown()
646			{
647			if (auto ec = take_pending_flush_ec())
648			co_return {ec};
649
650			std::error_code ec;
651
652			while (true)
653			{
654			auto const gen = read_gen_;
655			auto r = eng_.perform(engine_op::shutdown, nullptr, 0);
656
657			if (r.ec)
658			{
659			if (r.want == engine_want::output_then_done)
660			co_await best_effort_flush();
661			co_return {r.ec};
662			}
663
664			if (r.want == engine_want::done \|\|
665			r.want == engine_want::output_then_done)
666			{
667			// Covers both the bidirectional-complete result and the
668			// engine's shutdown-after-close success mapping.
669			ec = co_await flush_output();
670			co_return {ec};
671			}
672
673			if (r.want == engine_want::output_then_retry)
674			{
675			// Sends our close_notify before parking for the peer's.
676			ec = co_await flush_output();
677			if (ec)
678			co_return {ec};
679			continue;
680			}
681
682			// want == input: awaiting the peer's close_notify. It may
683			// already have been deposited (and consumed by a concurrent
684			// reader) during a flush; fill_input then returns without
685			// reading and the loop retries the engine instead of parking.
686			ec = co_await flush_output();
687			if (ec)
688			co_return {ec};
689
690			ec = co_await fill_input(gen);
691			if (ec)
692			{
693			ec = map_fill_error(
694			engine_op::shutdown, ec, eng_.received_shutdown());
695			co_return {ec};
696			}
697			}
698		290x	}
699			};
700
701			} // namespace detail
702
703			} // namespace boost::corosio
704
705			#endif
706