19741624072

Committed 27 Nov 2025 03:45PM UTC coverage: 73.086% (+0.02%) from 73.065%

Build # 19741624072

Build Type

Pull #16570

github

Committed by

web-flow

Commit Message

Merge 08a2cdb1d into f94a3f63f

Pull Request Pull Request #16570: rec: rewrite all unwrap calls in web.rs

Run Details

38523 of 63408 branches covered (60.75%)

Branch coverage included in aggregate %.

128044 of 164496 relevant lines covered (77.84%)

6531485.83 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

79.35

/pdns/dnsdistdist/dnsdist-tcp.hh

/*
 * This file is part of PowerDNS or dnsdist.
 * Copyright -- PowerDNS.COM B.V. and its contributors
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of version 2 of the GNU General Public License as
 * published by the Free Software Foundation.
 *
 * In addition, for the avoidance of any doubt, permission is granted to
 * link this program with OpenSSL and to (re)distribute the binaries
 * produced as the result of such linking.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
 */
#pragma once

#include <optional>
#include <unistd.h>
#include "channel.hh"
#include "iputils.hh"
#include "dnsdist.hh"
#include "dnsdist-metrics.hh"

struct ConnectionInfo
{
  ConnectionInfo(ClientState* cs_) :
    cs(cs_), fd(-1)
  {
  }
  ConnectionInfo(ClientState* cs_, const ComboAddress remote_) :
    remote(remote_), cs(cs_), fd(-1)
  {
  }
  ConnectionInfo(ConnectionInfo&& rhs) :
    remote(rhs.remote), cs(rhs.cs), fd(rhs.fd)
  {
    rhs.cs = nullptr;
    rhs.fd = -1;
  }

  ConnectionInfo(const ConnectionInfo& rhs) = delete;
  ConnectionInfo& operator=(const ConnectionInfo& rhs) = delete;

  ConnectionInfo& operator=(ConnectionInfo&& rhs)
  {
    remote = rhs.remote;
    cs = rhs.cs;
    rhs.cs = nullptr;
    fd = rhs.fd;
    rhs.fd = -1;
    return *this;
  }

  ~ConnectionInfo()
  {
    if (fd != -1) {
      close(fd);
      fd = -1;
    }

    if (cs) {
      --cs->tcpCurrentConnections;
    }
  }

  ComboAddress remote;
  ClientState* cs{nullptr};
  int fd{-1};
  bool d_restricted{false};
};

class InternalQuery
{
public:
  InternalQuery()
  {
  }

  InternalQuery(PacketBuffer&& buffer, InternalQueryState&& state) :
    d_idstate(std::move(state)), d_buffer(std::move(buffer))
  {
  }

  InternalQuery(InternalQuery&& rhs) = default;
  InternalQuery& operator=(InternalQuery&& rhs) = default;

  InternalQuery(const InternalQuery& rhs) = delete;
  InternalQuery& operator=(const InternalQuery& rhs) = delete;

  bool isXFR() const
  {
    return d_idstate.qtype == QType::AXFR || d_idstate.qtype == QType::IXFR;
  }

  InternalQueryState d_idstate;
  std::string d_proxyProtocolPayload;
  PacketBuffer d_buffer;
  uint32_t d_ixfrQuerySerial{0};
  uint32_t d_xfrPrimarySerial{0};
  uint32_t d_xfrSerialCount{0};
  uint32_t d_downstreamFailures{0};
  uint8_t d_xfrPrimarySerialCount{0};
  bool d_xfrStarted{false};
  bool d_proxyProtocolPayloadAdded{false};
};

using TCPQuery = InternalQuery;

class ConnectionToBackend;

struct TCPResponse : public TCPQuery
{
  TCPResponse()
  {
    /* let's make Coverity happy */
    memset(&d_cleartextDH, 0, sizeof(d_cleartextDH));
  }

  TCPResponse(PacketBuffer&& buffer, InternalQueryState&& state, std::shared_ptr<ConnectionToBackend> conn, std::shared_ptr<DownstreamState> ds) :
    TCPQuery(std::move(buffer), std::move(state)), d_connection(std::move(conn)), d_ds(std::move(ds))
  {
    if (d_buffer.size() >= sizeof(dnsheader)) {
      dnsheader_aligned header(d_buffer.data());
      memcpy(&d_cleartextDH, header.get(), sizeof(d_cleartextDH));
    }
    else {
      memset(&d_cleartextDH, 0, sizeof(d_cleartextDH));
    }
  }

  TCPResponse(TCPQuery&& query) :
    TCPQuery(std::move(query))
  {
    if (d_buffer.size() >= sizeof(dnsheader)) {
      dnsheader_aligned header(d_buffer.data());
      memcpy(&d_cleartextDH, header.get(), sizeof(d_cleartextDH));
    }
    else {
      memset(&d_cleartextDH, 0, sizeof(d_cleartextDH));
    }
  }

  bool isAsync() const
  {
    return d_async;
  }

  std::shared_ptr<ConnectionToBackend> d_connection{nullptr};
  std::shared_ptr<DownstreamState> d_ds{nullptr};
  dnsheader d_cleartextDH;
  bool d_async{false};
};

class TCPQuerySender
{
public:
  virtual ~TCPQuerySender()
  {
  }

  virtual bool active() const = 0;
  virtual void handleResponse(const struct timeval& now, TCPResponse&& response) = 0;
  virtual void handleXFRResponse(const struct timeval& now, TCPResponse&& response) = 0;
  virtual void notifyIOError(const struct timeval& now, TCPResponse&& response) = 0;

  /* whether the connection should be automatically released to the pool after handleResponse()
     has been called */
  bool releaseConnection() const
  {
    return d_releaseConnection;
  }

protected:
  bool d_releaseConnection{true};
};

struct CrossProtocolQuery
{
  CrossProtocolQuery()
  {
  }
  CrossProtocolQuery(InternalQuery&& query_, std::shared_ptr<DownstreamState>& downstream_) :
    query(std::move(query_)), downstream(downstream_)
  {
  }

  CrossProtocolQuery(CrossProtocolQuery&& rhs) = delete;
  virtual ~CrossProtocolQuery()
  {
  }

  virtual std::shared_ptr<TCPQuerySender> getTCPQuerySender() = 0;
  virtual DNSQuestion getDQ()
  {
    auto& ids = query.d_idstate;
    DNSQuestion dq(ids, query.d_buffer);
    return dq;
  }

  virtual DNSResponse getDR()
  {
    auto& ids = query.d_idstate;
    DNSResponse dr(ids, query.d_buffer, downstream);
    return dr;
  }

  InternalQuery query;
  std::shared_ptr<DownstreamState> downstream{nullptr};
  bool d_isResponse{false};
};

class TCPClientCollection
{
public:
  TCPClientCollection(size_t maxThreads, std::vector<ClientState*> tcpAcceptStates);

  bool passConnectionToThread(std::unique_ptr<ConnectionInfo>&& conn)
  {
    if (d_numthreads == 0) {
      throw std::runtime_error("No TCP worker thread yet");
    }

    uint64_t pos = d_pos++;
    /* we need to increment this counter _before_ writing to the pipe,
       otherwise there is a very real possibility that the other end
       decrement the counter before we can increment it, leading to an underflow */
    ++d_queued;
    if (!d_tcpclientthreads.at(pos % d_numthreads).d_querySender.send(std::move(conn))) {
      --d_queued;
      ++dnsdist::metrics::g_stats.tcpQueryPipeFull;
      return false;
    }

    return true;
  }

  bool passCrossProtocolQueryToThread(std::unique_ptr<CrossProtocolQuery>&& cpq)
  {
    if (d_numthreads == 0) {
      throw std::runtime_error("No TCP worker thread yet");
    }

    uint64_t pos = d_pos++;
    if (!d_tcpclientthreads.at(pos % d_numthreads).d_crossProtocolQuerySender.send(std::move(cpq))) {
      ++dnsdist::metrics::g_stats.tcpCrossProtocolQueryPipeFull;
      return false;
    }

    return true;
  }

  bool hasReachedMaxThreads() const
  {
    return d_numthreads >= d_maxthreads;
  }

  uint64_t getThreadsCount() const
  {
    return d_numthreads;
  }

  uint64_t getQueuedCount() const
  {
    return d_queued;
  }

  void decrementQueuedCount()
  {
    --d_queued;
  }

private:
  void addTCPClientThread(std::vector<ClientState*>& tcpAcceptStates);

  struct TCPWorkerThread
  {
    TCPWorkerThread()
    {
    }

    TCPWorkerThread(pdns::channel::Sender<ConnectionInfo>&& querySender, pdns::channel::Sender<CrossProtocolQuery>&& crossProtocolQuerySender) :
      d_querySender(std::move(querySender)), d_crossProtocolQuerySender(std::move(crossProtocolQuerySender))
    {
    }

    TCPWorkerThread(TCPWorkerThread&& rhs) = default;
    TCPWorkerThread& operator=(TCPWorkerThread&& rhs) = default;
    TCPWorkerThread(const TCPWorkerThread& rhs) = delete;
    TCPWorkerThread& operator=(const TCPWorkerThread&) = delete;

    pdns::channel::Sender<ConnectionInfo> d_querySender;
    pdns::channel::Sender<CrossProtocolQuery> d_crossProtocolQuerySender;
  };

  std::vector<TCPWorkerThread> d_tcpclientthreads;
  stat_t d_numthreads{0};
  stat_t d_pos{0};
  stat_t d_queued{0};
  const uint64_t d_maxthreads{0};
};

namespace dnsdist::tcp
{
class HandlingIOGuard
{
public:
  HandlingIOGuard(bool& handlingIO) :
    d_handlingIO(handlingIO)
  {
    d_handlingIO = true;
  }
  HandlingIOGuard(const HandlingIOGuard&) = delete;
  HandlingIOGuard(HandlingIOGuard&&) = delete;
  HandlingIOGuard& operator=(const HandlingIOGuard& rhs) = delete;
  HandlingIOGuard& operator=(HandlingIOGuard&&) = delete;
  ~HandlingIOGuard()
  {
    d_handlingIO = false;
  }

private:
  bool& d_handlingIO;
};
}

extern std::unique_ptr<TCPClientCollection> g_tcpclientthreads;

std::unique_ptr<CrossProtocolQuery> getTCPCrossProtocolQueryFromDQ(DNSQuestion& dnsQuestion);

1	/*
2	* This file is part of PowerDNS or dnsdist.
3	* Copyright -- PowerDNS.COM B.V. and its contributors
4	*
5	* This program is free software; you can redistribute it and/or modify
6	* it under the terms of version 2 of the GNU General Public License as
7	* published by the Free Software Foundation.
8	*
9	* In addition, for the avoidance of any doubt, permission is granted to
10	* link this program with OpenSSL and to (re)distribute the binaries
11	* produced as the result of such linking.
12	*
13	* This program is distributed in the hope that it will be useful,
14	* but WITHOUT ANY WARRANTY; without even the implied warranty of
15	* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
16	* GNU General Public License for more details.
17	*
18	* You should have received a copy of the GNU General Public License
19	* along with this program; if not, write to the Free Software
20	* Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
21	*/
22	#pragma once
23
24	#include <optional>
25	#include <unistd.h>
26	#include "channel.hh"
27	#include "iputils.hh"
28	#include "dnsdist.hh"
29	#include "dnsdist-metrics.hh"
30
31	struct ConnectionInfo
32	{
33	ConnectionInfo(ClientState* cs_) :
34	cs(cs_), fd(-1)	3,179✔
35	{	3,179✔
36	}	3,179✔
37	ConnectionInfo(ClientState* cs_, const ComboAddress remote_) :
38	remote(remote_), cs(cs_), fd(-1)	155✔
39	{	155✔
40	}	155✔
41	ConnectionInfo(ConnectionInfo&& rhs) :
42	remote(rhs.remote), cs(rhs.cs), fd(rhs.fd)	5,507✔
43	{	5,507✔
44	rhs.cs = nullptr;	5,507✔
45	rhs.fd = -1;	5,507✔
46	}	5,507✔
47
48	ConnectionInfo(const ConnectionInfo& rhs) = delete;
49	ConnectionInfo& operator=(const ConnectionInfo& rhs) = delete;
50
51	ConnectionInfo& operator=(ConnectionInfo&& rhs)
52	{	×
53	remote = rhs.remote;	×
54	cs = rhs.cs;	×
55	rhs.cs = nullptr;	×
56	fd = rhs.fd;	×
57	rhs.fd = -1;	×
58	return *this;	×
59	}	×
60
61	~ConnectionInfo()
62	{	8,359✔
63	if (fd != -1) {	8,359✔
64	close(fd);	18✔
65	fd = -1;	18✔
66	}	18✔
67
68	if (cs) {	8,359✔
69	--cs->tcpCurrentConnections;	2,852✔
70	}	2,852✔
71	}	8,359✔
72
73	ComboAddress remote;
74	ClientState* cs{nullptr};
75	int fd{-1};
76	bool d_restricted{false};
77	};
78
79	class InternalQuery
80	{
81	public:
82	InternalQuery()
83	{	35,161✔
84	}	35,161✔
85
86	InternalQuery(PacketBuffer&& buffer, InternalQueryState&& state) :
87	d_idstate(std::move(state)), d_buffer(std::move(buffer))	67,014✔
88	{	67,014✔
89	}	67,014✔
90
91	InternalQuery(InternalQuery&& rhs) = default;	233,404✔
92	InternalQuery& operator=(InternalQuery&& rhs) = default;	98,570✔
93
94	InternalQuery(const InternalQuery& rhs) = delete;
95	InternalQuery& operator=(const InternalQuery& rhs) = delete;
96
97	bool isXFR() const
98	{	34,870✔
99	return d_idstate.qtype == QType::AXFR \|\| d_idstate.qtype == QType::IXFR;	34,870✔
100	}	34,870✔
101
102	InternalQueryState d_idstate;
103	std::string d_proxyProtocolPayload;
104	PacketBuffer d_buffer;
105	uint32_t d_ixfrQuerySerial{0};
106	uint32_t d_xfrPrimarySerial{0};
107	uint32_t d_xfrSerialCount{0};
108	uint32_t d_downstreamFailures{0};
109	uint8_t d_xfrPrimarySerialCount{0};
110	bool d_xfrStarted{false};
111	bool d_proxyProtocolPayloadAdded{false};
112	};
113
114	using TCPQuery = InternalQuery;
115
116	class ConnectionToBackend;
117
118	struct TCPResponse : public TCPQuery
119	{
120	TCPResponse()
121	{	33,787✔
122	/* let's make Coverity happy */
123	memset(&d_cleartextDH, 0, sizeof(d_cleartextDH));	33,787✔
124	}	33,787✔
125
126	TCPResponse(PacketBuffer&& buffer, InternalQueryState&& state, std::shared_ptr<ConnectionToBackend> conn, std::shared_ptr<DownstreamState> ds) :
127	TCPQuery(std::move(buffer), std::move(state)), d_connection(std::move(conn)), d_ds(std::move(ds))	33,227✔
128	{	33,227✔
129	if (d_buffer.size() >= sizeof(dnsheader)) {	33,227✔
130	dnsheader_aligned header(d_buffer.data());	33,191✔
131	memcpy(&d_cleartextDH, header.get(), sizeof(d_cleartextDH));	33,191✔
132	}	33,191✔
133	else {	36✔
134	memset(&d_cleartextDH, 0, sizeof(d_cleartextDH));	36✔
135	}	36✔
136	}	33,227✔
137
138	TCPResponse(TCPQuery&& query) :
139	TCPQuery(std::move(query))	216✔
140	{	216✔
141	if (d_buffer.size() >= sizeof(dnsheader)) {	216✔
142	dnsheader_aligned header(d_buffer.data());	210✔
143	memcpy(&d_cleartextDH, header.get(), sizeof(d_cleartextDH));	210✔
144	}	210✔
145	else {	6✔
146	memset(&d_cleartextDH, 0, sizeof(d_cleartextDH));	6✔
147	}	6✔
148	}	216✔
149
150	bool isAsync() const
151	{	66,259✔
152	return d_async;	66,259✔
153	}	66,259✔
154
155	std::shared_ptr<ConnectionToBackend> d_connection{nullptr};
156	std::shared_ptr<DownstreamState> d_ds{nullptr};
157	dnsheader d_cleartextDH;
158	bool d_async{false};
159	};
160
161	class TCPQuerySender
162	{
163	public:
164	virtual ~TCPQuerySender()
165	{	2,918✔
166	}	2,918✔
167
168	virtual bool active() const = 0;
169	virtual void handleResponse(const struct timeval& now, TCPResponse&& response) = 0;
170	virtual void handleXFRResponse(const struct timeval& now, TCPResponse&& response) = 0;
171	virtual void notifyIOError(const struct timeval& now, TCPResponse&& response) = 0;
172
173	/* whether the connection should be automatically released to the pool after handleResponse()
174	has been called */
175	bool releaseConnection() const
176	{	×
177	return d_releaseConnection;	×
178	}	×
179
180	protected:
181	bool d_releaseConnection{true};
182	};
183
184	struct CrossProtocolQuery
185	{
186	CrossProtocolQuery()
187	{	184✔
188	}	184✔
189	CrossProtocolQuery(InternalQuery&& query_, std::shared_ptr<DownstreamState>& downstream_) :
190	query(std::move(query_)), downstream(downstream_)	734✔
191	{	734✔
192	}	734✔
193
194	CrossProtocolQuery(CrossProtocolQuery&& rhs) = delete;
195	virtual ~CrossProtocolQuery()
196	{	918✔
197	}	918✔
198
199	virtual std::shared_ptr<TCPQuerySender> getTCPQuerySender() = 0;
200	virtual DNSQuestion getDQ()
201	{	361✔
202	auto& ids = query.d_idstate;	361✔
203	DNSQuestion dq(ids, query.d_buffer);	361✔
204	return dq;	361✔
205	}	361✔
206
207	virtual DNSResponse getDR()
208	{	24✔
209	auto& ids = query.d_idstate;	24✔
210	DNSResponse dr(ids, query.d_buffer, downstream);	24✔
211	return dr;	24✔
212	}	24✔
213
214	InternalQuery query;
215	std::shared_ptr<DownstreamState> downstream{nullptr};
216	bool d_isResponse{false};
217	};
218
219	class TCPClientCollection
220	{
221	public:
222	TCPClientCollection(size_t maxThreads, std::vector<ClientState*> tcpAcceptStates);
223
224	bool passConnectionToThread(std::unique_ptr<ConnectionInfo>&& conn)
225	{	2,676✔
226	if (d_numthreads == 0) {	2,676!
227	throw std::runtime_error("No TCP worker thread yet");	×
228	}	×
229
230	uint64_t pos = d_pos++;	2,676✔
231	/* we need to increment this counter _before_ writing to the pipe,
232	otherwise there is a very real possibility that the other end
233	decrement the counter before we can increment it, leading to an underflow */
234	++d_queued;	2,676✔
235	if (!d_tcpclientthreads.at(pos % d_numthreads).d_querySender.send(std::move(conn))) {	2,676!
236	--d_queued;	×
237	++dnsdist::metrics::g_stats.tcpQueryPipeFull;	×
238	return false;	×
239	}	×
240
241	return true;	2,676✔
242	}	2,676✔
243
244	bool passCrossProtocolQueryToThread(std::unique_ptr<CrossProtocolQuery>&& cpq)
245	{	228✔
246	if (d_numthreads == 0) {	228!
247	throw std::runtime_error("No TCP worker thread yet");	×
248	}	×
249
250	uint64_t pos = d_pos++;	228✔
251	if (!d_tcpclientthreads.at(pos % d_numthreads).d_crossProtocolQuerySender.send(std::move(cpq))) {	228!
252	++dnsdist::metrics::g_stats.tcpCrossProtocolQueryPipeFull;	×
253	return false;	×
254	}	×
255
256	return true;	228✔
257	}	228✔
258
259	bool hasReachedMaxThreads() const
260	{	×
261	return d_numthreads >= d_maxthreads;	×
262	}	×
263
264	uint64_t getThreadsCount() const
265	{	×
266	return d_numthreads;	×
267	}	×
268
269	uint64_t getQueuedCount() const
270	{	2,682✔
271	return d_queued;	2,682✔
272	}	2,682✔
273
274	void decrementQueuedCount()
275	{	2,676✔
276	--d_queued;	2,676✔
277	}	2,676✔
278
279	private:
280	void addTCPClientThread(std::vector<ClientState*>& tcpAcceptStates);
281
282	struct TCPWorkerThread
283	{
284	TCPWorkerThread()
285	{	3,811✔
286	}	3,811✔
287
288	TCPWorkerThread(pdns::channel::Sender<ConnectionInfo>&& querySender, pdns::channel::Sender<CrossProtocolQuery>&& crossProtocolQuerySender) :
289	d_querySender(std::move(querySender)), d_crossProtocolQuerySender(std::move(crossProtocolQuerySender))	3,811✔
290	{	3,811✔
291	}	3,811✔
292
293	TCPWorkerThread(TCPWorkerThread&& rhs) = default;
294	TCPWorkerThread& operator=(TCPWorkerThread&& rhs) = default;	3,811✔
295	TCPWorkerThread(const TCPWorkerThread& rhs) = delete;
296	TCPWorkerThread& operator=(const TCPWorkerThread&) = delete;
297
298	pdns::channel::Sender<ConnectionInfo> d_querySender;
299	pdns::channel::Sender<CrossProtocolQuery> d_crossProtocolQuerySender;
300	};
301
302	std::vector<TCPWorkerThread> d_tcpclientthreads;
303	stat_t d_numthreads{0};
304	stat_t d_pos{0};
305	stat_t d_queued{0};
306	const uint64_t d_maxthreads{0};
307	};
308
309	namespace dnsdist::tcp
310	{
311	class HandlingIOGuard
312	{
313	public:
314	HandlingIOGuard(bool& handlingIO) :
315	d_handlingIO(handlingIO)	44,522✔
316	{	44,522✔
317	d_handlingIO = true;	44,522✔
318	}	44,522✔
319	HandlingIOGuard(const HandlingIOGuard&) = delete;
320	HandlingIOGuard(HandlingIOGuard&&) = delete;
321	HandlingIOGuard& operator=(const HandlingIOGuard& rhs) = delete;
322	HandlingIOGuard& operator=(HandlingIOGuard&&) = delete;
323	~HandlingIOGuard()
324	{	44,521✔
325	d_handlingIO = false;	44,521✔
326	}	44,521✔
327
328	private:
329	bool& d_handlingIO;
330	};
331	}
332
333	extern std::unique_ptr<TCPClientCollection> g_tcpclientthreads;
334
335	std::unique_ptr<CrossProtocolQuery> getTCPCrossProtocolQueryFromDQ(DNSQuestion& dnsQuestion);

PowerDNS / pdns / 19741624072

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous