1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
|
#pragma once
#include "dolog.hh"
#include "dnsdist-tcp.hh"
#include "dnsdist-tcp-downstream.hh"
struct TCPCrossProtocolResponse;
class TCPClientThreadData
{
public:
TCPClientThreadData():
localRespRuleActions(g_respruleactions.getLocal()), localCacheInsertedRespRuleActions(g_cacheInsertedRespRuleActions.getLocal()), mplexer(std::unique_ptr<FDMultiplexer>(FDMultiplexer::getMultiplexerSilent()))
{
}
LocalHolders holders;
LocalStateHolder<vector<DNSDistResponseRuleAction>> localRespRuleActions;
LocalStateHolder<vector<DNSDistResponseRuleAction>> localCacheInsertedRespRuleActions;
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{g_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
{
if (g_maxTCPConnectionDuration == 0 && g_tcpRecvTimeout == 0) {
return boost::none;
}
if (g_maxTCPConnectionDuration > 0) {
auto elapsed = now.tv_sec - d_connectionStartTime.tv_sec;
if (elapsed < 0 || (static_cast<size_t>(elapsed) >= g_maxTCPConnectionDuration)) {
return now;
}
auto remaining = g_maxTCPConnectionDuration - elapsed;
if (g_tcpRecvTimeout == 0 || remaining <= static_cast<size_t>(g_tcpRecvTimeout)) {
now.tv_sec += remaining;
return now;
}
}
now.tv_sec += g_tcpRecvTimeout;
return now;
}
boost::optional<struct timeval> getClientWriteTTD(const struct timeval& now) const
{
if (g_maxTCPConnectionDuration == 0 && g_tcpSendTimeout == 0) {
return boost::none;
}
struct timeval res = now;
if (g_maxTCPConnectionDuration > 0) {
auto elapsed = res.tv_sec - d_connectionStartTime.tv_sec;
if (elapsed < 0 || static_cast<size_t>(elapsed) >= g_maxTCPConnectionDuration) {
return res;
}
auto remaining = g_maxTCPConnectionDuration - elapsed;
if (g_tcpSendTimeout == 0 || remaining <= static_cast<size_t>(g_tcpSendTimeout)) {
res.tv_sec += remaining;
return res;
}
}
res.tv_sec += g_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 updateIO(std::shared_ptr<IncomingTCPConnectionState>& state, IOState newState, const struct timeval& now);
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);
virtual void handleIO();
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)
{
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};
};
|
