13012068652

Committed 28 Jan 2025 01:59PM UTC coverage: 64.71% (+0.01%) from 64.699%

Build # 13012068652

Build Type

Pull #14724

github

Committed by

web-flow

Commit Message

Merge b15562560 into db18c3a17

Pull Request Pull Request #14724: dnsdist: Add meson support

Run Details

38328 of 90334 branches covered (42.43%)

Branch coverage included in aggregate %.

361 of 513 new or added lines in 35 files covered. (70.37%)

42 existing lines in 13 files now uncovered.

128150 of 166934 relevant lines covered (76.77%)

4540890.91 hits per line

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

65.38

/pdns/dnsdistdist/dnsdist-tcp-upstream.hh

#pragma once

#include "dolog.hh"
#include "dnsdist-tcp.hh"
#include "dnsdist-tcp-downstream.hh"

struct TCPCrossProtocolResponse;

class TCPClientThreadData
{
public:
  TCPClientThreadData():
    mplexer(std::unique_ptr<FDMultiplexer>(FDMultiplexer::getMultiplexerSilent()))
  {
  }

  std::unique_ptr<FDMultiplexer> mplexer{nullptr};
  pdns::channel::Receiver<ConnectionInfo> queryReceiver;
  pdns::channel::Receiver<CrossProtocolQuery> crossProtocolQueryReceiver;
  pdns::channel::Receiver<TCPCrossProtocolResponse> crossProtocolResponseReceiver;
  pdns::channel::Sender<TCPCrossProtocolResponse> crossProtocolResponseSender;
};

class IncomingTCPConnectionState : public TCPQuerySender, public std::enable_shared_from_this<IncomingTCPConnectionState>
{
public:
  enum class QueryProcessingResult : uint8_t { Forwarded, TooSmall, InvalidHeaders, Dropped, SelfAnswered, NoBackend, Asynchronous };
  enum class ProxyProtocolResult : uint8_t { Reading, Done, Error };

  IncomingTCPConnectionState(ConnectionInfo&& ci, TCPClientThreadData& threadData, const struct timeval& now): d_buffer(sizeof(uint16_t)), d_ci(std::move(ci)), d_handler(d_ci.fd, timeval{dnsdist::configuration::getCurrentRuntimeConfiguration().d_tcpRecvTimeout,0}, d_ci.cs->tlsFrontend ? d_ci.cs->tlsFrontend->getContext() : (d_ci.cs->dohFrontend ? d_ci.cs->dohFrontend->d_tlsContext.getContext() : nullptr), now.tv_sec), d_connectionStartTime(now), d_ioState(make_unique<IOStateHandler>(*threadData.mplexer, d_ci.fd)), d_threadData(threadData), d_creatorThreadID(std::this_thread::get_id())
  {
    d_origDest.reset();
    d_origDest.sin4.sin_family = d_ci.remote.sin4.sin_family;
    socklen_t socklen = d_origDest.getSocklen();
    if (getsockname(d_ci.fd, reinterpret_cast<sockaddr*>(&d_origDest), &socklen)) {
      d_origDest = d_ci.cs->local;
    }
    /* belongs to the handler now */
    d_ci.fd = -1;
    d_proxiedDestination = d_origDest;
    d_proxiedRemote = d_ci.remote;

    /* we manage the release of the downstream connection ourselves */
    d_releaseConnection = false;
  }

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

  virtual ~IncomingTCPConnectionState();

  void resetForNewQuery();

  boost::optional<struct timeval> getClientReadTTD(struct timeval now) const
  {
    const auto& runtimeConfiguration = dnsdist::configuration::getCurrentRuntimeConfiguration();
    if (runtimeConfiguration.d_maxTCPConnectionDuration == 0 && runtimeConfiguration.d_tcpRecvTimeout == 0) {
      return boost::none;
    }

    if (runtimeConfiguration.d_maxTCPConnectionDuration > 0) {
      auto elapsed = now.tv_sec - d_connectionStartTime.tv_sec;
      if (elapsed < 0 || (static_cast<size_t>(elapsed) >= runtimeConfiguration.d_maxTCPConnectionDuration)) {
        return now;
      }
      auto remaining = runtimeConfiguration.d_maxTCPConnectionDuration - elapsed;
      if (runtimeConfiguration.d_tcpRecvTimeout == 0 || remaining <= static_cast<size_t>(runtimeConfiguration.d_tcpRecvTimeout)) {
        now.tv_sec += remaining;
        return now;
      }
    }

    now.tv_sec += runtimeConfiguration.d_tcpRecvTimeout;
    return now;
  }

  boost::optional<struct timeval> getClientWriteTTD(const struct timeval& now) const
  {
    const auto& runtimeConfiguration = dnsdist::configuration::getCurrentRuntimeConfiguration();
    if (runtimeConfiguration.d_maxTCPConnectionDuration == 0 && runtimeConfiguration.d_tcpSendTimeout == 0) {
      return boost::none;
    }

    timeval res(now);

    if (runtimeConfiguration.d_maxTCPConnectionDuration > 0) {
      auto elapsed = res.tv_sec - d_connectionStartTime.tv_sec;
      if (elapsed < 0 || static_cast<size_t>(elapsed) >= runtimeConfiguration.d_maxTCPConnectionDuration) {
        return res;
      }
      auto remaining = runtimeConfiguration.d_maxTCPConnectionDuration - elapsed;
      if (runtimeConfiguration.d_tcpSendTimeout == 0 || remaining <= static_cast<size_t>(runtimeConfiguration.d_tcpSendTimeout)) {
        res.tv_sec += remaining;
        return res;
      }
    }

    res.tv_sec += runtimeConfiguration.d_tcpSendTimeout;
    return res;
  }

  bool maxConnectionDurationReached(unsigned int maxConnectionDuration, const struct timeval& now)
  {
    if (maxConnectionDuration) {
      time_t curtime = now.tv_sec;
      unsigned int elapsed = 0;
      if (curtime > d_connectionStartTime.tv_sec) { // To prevent issues when time goes backward
        elapsed = curtime - d_connectionStartTime.tv_sec;
      }
      if (elapsed >= maxConnectionDuration) {
        return true;
      }
    }

    return false;
  }

  std::shared_ptr<TCPConnectionToBackend> getOwnedDownstreamConnection(const std::shared_ptr<DownstreamState>& backend, const std::unique_ptr<std::vector<ProxyProtocolValue>>& tlvs);
  std::shared_ptr<TCPConnectionToBackend> getDownstreamConnection(std::shared_ptr<DownstreamState>& backend, const std::unique_ptr<std::vector<ProxyProtocolValue>>& tlvs, const struct timeval& now);
  void registerOwnedDownstreamConnection(std::shared_ptr<TCPConnectionToBackend>& conn);

  static size_t clearAllDownstreamConnections();

  static void handleIOCallback(int desc, FDMultiplexer::funcparam_t& param);
  static void handleAsyncReady(int desc, FDMultiplexer::funcparam_t& param);

  static void queueResponse(std::shared_ptr<IncomingTCPConnectionState>& state, const struct timeval& now, TCPResponse&& response, bool fromBackend);
  static void handleTimeout(std::shared_ptr<IncomingTCPConnectionState>& state, bool write);
  static void updateIOForAsync(std::shared_ptr<IncomingTCPConnectionState>& conn);

  virtual void handleIO();
  virtual void updateIO(IOState newState, const timeval& now);

  QueryProcessingResult handleQuery(PacketBuffer&& query, const struct timeval& now, std::optional<int32_t> streamID);
  virtual void handleResponse(const struct timeval& now, TCPResponse&& response) override;
  virtual void notifyIOError(const struct timeval& now, TCPResponse&& response) override;
  void handleXFRResponse(const struct timeval& now, TCPResponse&& response) override;

  virtual IOState sendResponse(const struct timeval& now, TCPResponse&& response);
  void handleResponseSent(TCPResponse& currentResponse, size_t sentBytes);
  virtual IOState handleHandshake(const struct timeval& now);
  void handleHandshakeDone(const struct timeval& now);
  ProxyProtocolResult handleProxyProtocolPayload();
  void handleCrossProtocolResponse(const struct timeval& now, TCPResponse&& response);

  void terminateClientConnection();

  bool canAcceptNewQueries(const struct timeval& now);

  bool active() const override
  {
    return d_ioState != nullptr;
  }
  bool isProxyPayloadOutsideTLS() const
  {
    if (!d_ci.cs->hasTLS()) {
      return false;
    }
    return d_ci.cs->getTLSFrontend().d_proxyProtocolOutsideTLS;
  }

  virtual bool forwardViaUDPFirst() const
  {
    return false;
  }
  virtual std::unique_ptr<DOHUnitInterface> getDOHUnit(uint32_t streamID)
  {
    (void)streamID;
    throw std::runtime_error("Getting a DOHUnit state from a generic TCP/DoT connection is not supported");
  }
  virtual void restoreDOHUnit(std::unique_ptr<DOHUnitInterface>&&)
  {
    throw std::runtime_error("Restoring a DOHUnit state to a generic TCP/DoT connection is not supported");
  }

  std::unique_ptr<CrossProtocolQuery> getCrossProtocolQuery(PacketBuffer&& query, InternalQueryState&& state, const std::shared_ptr<DownstreamState>& backend);

  std::string toString() const
  {
    ostringstream o;
    o << "Incoming TCP connection from "<<d_ci.remote.toStringWithPort()<<" over FD "<<d_handler.getDescriptor()<<", state is "<<(int)d_state<<", io state is "<<(d_ioState ? d_ioState->getState() : "empty")<<", queries count is "<<d_queriesCount<<", current queries count is "<<d_currentQueriesCount<<", "<<d_queuedResponses.size()<<" queued responses, "<<d_ownedConnectionsToBackend.size()<<" owned connections to a backend";
    return o.str();
  }

  dnsdist::Protocol getProtocol() const;
  IOState handleIncomingQueryReceived(const struct timeval& now);
  void handleExceptionDuringIO(const std::exception& exp);
  bool readIncomingQuery(const timeval& now, IOState& iostate);

  enum class State : uint8_t { starting, doingHandshake, readingProxyProtocolHeader, waitingForQuery, readingQuerySize, readingQuery, sendingResponse, idle /* in case of XFR, we stop processing queries */ };

  TCPResponse d_currentResponse;
  std::map<std::shared_ptr<DownstreamState>, std::deque<std::shared_ptr<TCPConnectionToBackend>>> d_ownedConnectionsToBackend;
  std::deque<TCPResponse> d_queuedResponses;
  PacketBuffer d_buffer;
  ConnectionInfo d_ci;
  ComboAddress d_origDest;
  ComboAddress d_proxiedRemote;
  ComboAddress d_proxiedDestination;
  TCPIOHandler d_handler;
  struct timeval d_connectionStartTime;
  struct timeval d_handshakeDoneTime;
  struct timeval d_firstQuerySizeReadTime;
  struct timeval d_querySizeReadTime;
  struct timeval d_queryReadTime;
  std::unique_ptr<IOStateHandler> d_ioState{nullptr};
  std::unique_ptr<std::vector<ProxyProtocolValue>> d_proxyProtocolValues{nullptr};
  TCPClientThreadData& d_threadData;
  size_t d_currentPos{0};
  size_t d_proxyProtocolNeed{0};
  size_t d_queriesCount{0};
  size_t d_currentQueriesCount{0};
  std::thread::id d_creatorThreadID;
  uint16_t d_querySize{0};
  State d_state{State::starting};
  bool d_isXFR{false};
  bool d_proxyProtocolPayloadHasTLV{false};
  bool d_lastIOBlocked{false};
  bool d_hadErrors{false};
};

1	#pragma once
2
3	#include "dolog.hh"
4	#include "dnsdist-tcp.hh"
5	#include "dnsdist-tcp-downstream.hh"
6
7	struct TCPCrossProtocolResponse;
8
9	class TCPClientThreadData
10	{
11	public:
12	TCPClientThreadData():
13	mplexer(std::unique_ptr<FDMultiplexer>(FDMultiplexer::getMultiplexerSilent()))
14	{	3,165✔
15	}	3,165✔
16
17	std::unique_ptr<FDMultiplexer> mplexer{nullptr};
18	pdns::channel::Receiver<ConnectionInfo> queryReceiver;
19	pdns::channel::Receiver<CrossProtocolQuery> crossProtocolQueryReceiver;
20	pdns::channel::Receiver<TCPCrossProtocolResponse> crossProtocolResponseReceiver;
21	pdns::channel::Sender<TCPCrossProtocolResponse> crossProtocolResponseSender;
22	};
23
24	class IncomingTCPConnectionState : public TCPQuerySender, public std::enable_shared_from_this<IncomingTCPConnectionState>
25	{
26	public:
27	enum class QueryProcessingResult : uint8_t { Forwarded, TooSmall, InvalidHeaders, Dropped, SelfAnswered, NoBackend, Asynchronous };
28	enum class ProxyProtocolResult : uint8_t { Reading, Done, Error };
29
30	IncomingTCPConnectionState(ConnectionInfo&& ci, TCPClientThreadData& threadData, const struct timeval& now): d_buffer(sizeof(uint16_t)), d_ci(std::move(ci)), d_handler(d_ci.fd, timeval{dnsdist::configuration::getCurrentRuntimeConfiguration().d_tcpRecvTimeout,0}, d_ci.cs->tlsFrontend ? d_ci.cs->tlsFrontend->getContext() : (d_ci.cs->dohFrontend ? d_ci.cs->dohFrontend->d_tlsContext.getContext() : nullptr), now.tv_sec), d_connectionStartTime(now), d_ioState(make_unique<IOStateHandler>(*threadData.mplexer, d_ci.fd)), d_threadData(threadData), d_creatorThreadID(std::this_thread::get_id())
31	{	4,805✔
32	d_origDest.reset();	4,805✔
33	d_origDest.sin4.sin_family = d_ci.remote.sin4.sin_family;	4,805✔
34	socklen_t socklen = d_origDest.getSocklen();	4,805✔
35	if (getsockname(d_ci.fd, reinterpret_cast<sockaddr*>(&d_origDest), &socklen)) {	4,805✔
36	d_origDest = d_ci.cs->local;	251✔
37	}	251✔
38	/* belongs to the handler now */
39	d_ci.fd = -1;	4,805✔
40	d_proxiedDestination = d_origDest;	4,805✔
41	d_proxiedRemote = d_ci.remote;	4,805✔
42
43	/* we manage the release of the downstream connection ourselves */
44	d_releaseConnection = false;	4,805✔
45	}	4,805✔
46
47	IncomingTCPConnectionState(const IncomingTCPConnectionState& rhs) = delete;
48	IncomingTCPConnectionState& operator=(const IncomingTCPConnectionState& rhs) = delete;
49
50	virtual ~IncomingTCPConnectionState();
51
52	void resetForNewQuery();
53
54	boost::optional<struct timeval> getClientReadTTD(struct timeval now) const
55	{	72,300✔
56	const auto& runtimeConfiguration = dnsdist::configuration::getCurrentRuntimeConfiguration();	72,300✔
57	if (runtimeConfiguration.d_maxTCPConnectionDuration == 0 && runtimeConfiguration.d_tcpRecvTimeout == 0) {	72,300!
58	return boost::none;	×
59	}	×
60
61	if (runtimeConfiguration.d_maxTCPConnectionDuration > 0) {	72,300✔
62	auto elapsed = now.tv_sec - d_connectionStartTime.tv_sec;	1,022✔
63	if (elapsed < 0 \|\| (static_cast<size_t>(elapsed) >= runtimeConfiguration.d_maxTCPConnectionDuration)) {	1,022!
64	return now;	×
65	}	×
66	auto remaining = runtimeConfiguration.d_maxTCPConnectionDuration - elapsed;	1,022✔
67	if (runtimeConfiguration.d_tcpRecvTimeout == 0 \|\| remaining <= static_cast<size_t>(runtimeConfiguration.d_tcpRecvTimeout)) {	1,022!
68	now.tv_sec += remaining;	40✔
69	return now;	40✔
70	}	40✔
71	}	1,022✔
72
73	now.tv_sec += runtimeConfiguration.d_tcpRecvTimeout;	72,260✔
74	return now;	72,260✔
75	}	72,300✔
76
77	boost::optional<struct timeval> getClientWriteTTD(const struct timeval& now) const
78	{	42✔
79	const auto& runtimeConfiguration = dnsdist::configuration::getCurrentRuntimeConfiguration();	42✔
80	if (runtimeConfiguration.d_maxTCPConnectionDuration == 0 && runtimeConfiguration.d_tcpSendTimeout == 0) {	42!
81	return boost::none;	×
82	}	×
83
84	timeval res(now);	42✔
85
86	if (runtimeConfiguration.d_maxTCPConnectionDuration > 0) {	42!
87	auto elapsed = res.tv_sec - d_connectionStartTime.tv_sec;	×
88	if (elapsed < 0 \|\| static_cast<size_t>(elapsed) >= runtimeConfiguration.d_maxTCPConnectionDuration) {	×
89	return res;	×
90	}	×
91	auto remaining = runtimeConfiguration.d_maxTCPConnectionDuration - elapsed;	×
92	if (runtimeConfiguration.d_tcpSendTimeout == 0 \|\| remaining <= static_cast<size_t>(runtimeConfiguration.d_tcpSendTimeout)) {	×
93	res.tv_sec += remaining;	×
94	return res;	×
95	}	×
96	}	×
97
98	res.tv_sec += runtimeConfiguration.d_tcpSendTimeout;	42✔
99	return res;	42✔
100	}	42✔
101
102	bool maxConnectionDurationReached(unsigned int maxConnectionDuration, const struct timeval& now)
103	{	81,874✔
104	if (maxConnectionDuration) {	81,874✔
105	time_t curtime = now.tv_sec;	1,572✔
106	unsigned int elapsed = 0;	1,572✔
107	if (curtime > d_connectionStartTime.tv_sec) { // To prevent issues when time goes backward	1,572✔
108	elapsed = curtime - d_connectionStartTime.tv_sec;	82✔
109	}	82✔
110	if (elapsed >= maxConnectionDuration) {	1,572✔
111	return true;	2✔
112	}	2✔
113	}	1,572✔
114
115	return false;	81,872✔
116	}	81,874✔
117
118	std::shared_ptr<TCPConnectionToBackend> getOwnedDownstreamConnection(const std::shared_ptr<DownstreamState>& backend, const std::unique_ptr<std::vector<ProxyProtocolValue>>& tlvs);
119	std::shared_ptr<TCPConnectionToBackend> getDownstreamConnection(std::shared_ptr<DownstreamState>& backend, const std::unique_ptr<std::vector<ProxyProtocolValue>>& tlvs, const struct timeval& now);
120	void registerOwnedDownstreamConnection(std::shared_ptr<TCPConnectionToBackend>& conn);
121
122	static size_t clearAllDownstreamConnections();
123
124	static void handleIOCallback(int desc, FDMultiplexer::funcparam_t& param);
125	static void handleAsyncReady(int desc, FDMultiplexer::funcparam_t& param);
126
127	static void queueResponse(std::shared_ptr<IncomingTCPConnectionState>& state, const struct timeval& now, TCPResponse&& response, bool fromBackend);
128	static void handleTimeout(std::shared_ptr<IncomingTCPConnectionState>& state, bool write);
129	static void updateIOForAsync(std::shared_ptr<IncomingTCPConnectionState>& conn);
130
131	virtual void handleIO();
132	virtual void updateIO(IOState newState, const timeval& now);
133
134	QueryProcessingResult handleQuery(PacketBuffer&& query, const struct timeval& now, std::optional<int32_t> streamID);
135	virtual void handleResponse(const struct timeval& now, TCPResponse&& response) override;
136	virtual void notifyIOError(const struct timeval& now, TCPResponse&& response) override;
137	void handleXFRResponse(const struct timeval& now, TCPResponse&& response) override;
138
139	virtual IOState sendResponse(const struct timeval& now, TCPResponse&& response);
140	void handleResponseSent(TCPResponse& currentResponse, size_t sentBytes);
141	virtual IOState handleHandshake(const struct timeval& now);
142	void handleHandshakeDone(const struct timeval& now);
143	ProxyProtocolResult handleProxyProtocolPayload();
144	void handleCrossProtocolResponse(const struct timeval& now, TCPResponse&& response);
145
146	void terminateClientConnection();
147
148	bool canAcceptNewQueries(const struct timeval& now);
149
150	bool active() const override
151	{	178,280✔
152	return d_ioState != nullptr;	178,280✔
153	}	178,280✔
154	bool isProxyPayloadOutsideTLS() const
155	{	9,214✔
156	if (!d_ci.cs->hasTLS()) {	9,214✔
157	return false;	7,836✔
158	}	7,836✔
159	return d_ci.cs->getTLSFrontend().d_proxyProtocolOutsideTLS;	1,378✔
160	}	9,214✔
161
162	virtual bool forwardViaUDPFirst() const
163	{	24,892✔
164	return false;	24,892✔
165	}	24,892✔
166	virtual std::unique_ptr<DOHUnitInterface> getDOHUnit(uint32_t streamID)
167	{	×
NEW 168	(void)streamID;	×
169	throw std::runtime_error("Getting a DOHUnit state from a generic TCP/DoT connection is not supported");	×
170	}	×
171	virtual void restoreDOHUnit(std::unique_ptr<DOHUnitInterface>&&)
172	{	×
173	throw std::runtime_error("Restoring a DOHUnit state to a generic TCP/DoT connection is not supported");	×
174	}	×
175
176	std::unique_ptr<CrossProtocolQuery> getCrossProtocolQuery(PacketBuffer&& query, InternalQueryState&& state, const std::shared_ptr<DownstreamState>& backend);
177
178	std::string toString() const
179	{	×
180	ostringstream o;	×
181	o << "Incoming TCP connection from "<<d_ci.remote.toStringWithPort()<<" over FD "<<d_handler.getDescriptor()<<", state is "<<(int)d_state<<", io state is "<<(d_ioState ? d_ioState->getState() : "empty")<<", queries count is "<<d_queriesCount<<", current queries count is "<<d_currentQueriesCount<<", "<<d_queuedResponses.size()<<" queued responses, "<<d_ownedConnectionsToBackend.size()<<" owned connections to a backend";	×
182	return o.str();	×
183	}	×
184
185	dnsdist::Protocol getProtocol() const;
186	IOState handleIncomingQueryReceived(const struct timeval& now);
187	void handleExceptionDuringIO(const std::exception& exp);
188	bool readIncomingQuery(const timeval& now, IOState& iostate);
189
190	enum class State : uint8_t { starting, doingHandshake, readingProxyProtocolHeader, waitingForQuery, readingQuerySize, readingQuery, sendingResponse, idle /* in case of XFR, we stop processing queries */ };
191
192	TCPResponse d_currentResponse;
193	std::map<std::shared_ptr<DownstreamState>, std::deque<std::shared_ptr<TCPConnectionToBackend>>> d_ownedConnectionsToBackend;
194	std::deque<TCPResponse> d_queuedResponses;
195	PacketBuffer d_buffer;
196	ConnectionInfo d_ci;
197	ComboAddress d_origDest;
198	ComboAddress d_proxiedRemote;
199	ComboAddress d_proxiedDestination;
200	TCPIOHandler d_handler;
201	struct timeval d_connectionStartTime;
202	struct timeval d_handshakeDoneTime;
203	struct timeval d_firstQuerySizeReadTime;
204	struct timeval d_querySizeReadTime;
205	struct timeval d_queryReadTime;
206	std::unique_ptr<IOStateHandler> d_ioState{nullptr};
207	std::unique_ptr<std::vector<ProxyProtocolValue>> d_proxyProtocolValues{nullptr};
208	TCPClientThreadData& d_threadData;
209	size_t d_currentPos{0};
210	size_t d_proxyProtocolNeed{0};
211	size_t d_queriesCount{0};
212	size_t d_currentQueriesCount{0};
213	std::thread::id d_creatorThreadID;
214	uint16_t d_querySize{0};
215	State d_state{State::starting};
216	bool d_isXFR{false};
217	bool d_proxyProtocolPayloadHasTLV{false};
218	bool d_lastIOBlocked{false};
219	bool d_hadErrors{false};
220	};

PowerDNS / pdns / 13012068652

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