source: trunk/FACT++/src/ConnectionSSL.cc@ 19164

Last change on this file since 19164 was 19050, checked in by tbretz, 6 years ago
Unified policy for info output
File size: 15.3 KB
Line 
1// **************************************************************************
2/** @class Connection
3
4@brief Maintains an ansynchronous TCP/IP client connection
5
6@todo
7 Unify with ConnectionUSB
8
9*/
10// **************************************************************************
11#include "ConnectionSSL.h"
12
13using namespace std;
14
15namespace ba = boost::asio;
16namespace ssl = boost::asio::ssl;
17namespace bs = boost::system;
18namespace dummy = ba::placeholders;
19
20typedef ssl::stream<boost::asio::ip::tcp::socket> stream;
21
22using ba::ip::tcp;
23
24 // -------- Abbreviations for starting async tasks ---------
25
26int ConnectionSSL::Write(const Time &t, const string &txt, int qos)
27{
28 if (fLog)
29 return fLog->Write(t, txt, qos);
30
31 return MessageImp::Write(t, txt, qos);
32}
33
34void ConnectionSSL::AsyncRead(const ba::mutable_buffers_1 buffers, int type)
35{
36 ba::async_read(*this, buffers,
37 boost::bind(&ConnectionSSL::HandleReceivedData, this,
38 dummy::error, dummy::bytes_transferred, type));
39}
40
41void ConnectionSSL::AsyncWrite(const ba::const_buffers_1 &buffers)
42{
43 ba::async_write(*this, buffers,
44 boost::bind(&ConnectionSSL::HandleSentData, this,
45 dummy::error, dummy::bytes_transferred));
46}
47
48/*
49void ConnectionSSL::AsyncWait(ba::deadline_timer &timer, int millisec,
50 void (ConnectionSSL::*handler)(const bs::error_code&))
51{
52 // - The boost::asio::basic_deadline_timer::expires_from_now()
53 // function cancels any pending asynchronous waits, and returns
54 // the number of asynchronous waits that were cancelled. If it
55 // returns 0 then you were too late and the wait handler has
56 // already been executed, or will soon be executed. If it
57 // returns 1 then the wait handler was successfully cancelled.
58 // - If a wait handler is cancelled, the bs::error_code passed to
59 // it contains the value bs::error::operation_aborted.
60 timer.expires_from_now(boost::posix_time::milliseconds(millisec));
61
62 timer.async_wait(boost::bind(handler, this, dummy::error));
63}
64*/
65
66void ConnectionSSL::AsyncConnect(tcp::resolver::iterator iterator)
67{
68 tcp::endpoint endpoint = *iterator;
69
70 // AsyncConnect + Deadline
71 lowest_layer().async_connect(endpoint,
72 boost::bind(&ConnectionSSL::ConnectIter,
73 this, iterator, ba::placeholders::error));
74
75 // We will get a "Connection timeout anyway"
76 //AsyncWait(fConnectTimeout, 5, &ConnectionSSL::HandleConnectTimeout);
77}
78
79void ConnectionSSL::AsyncConnect()
80{
81 // AsyncConnect + Deadline
82 lowest_layer().async_connect(fEndpoint,
83 boost::bind(&ConnectionSSL::ConnectAddr,
84 this, fEndpoint, ba::placeholders::error));
85
86 // We will get a "Connection timeout anyway"
87 //AsyncWait(fConnectTimeout, 5, &ConnectionSSL::HandleConnectTimeout);
88}
89
90// ------------------------ close --------------------------
91// close from another thread
92void ConnectionSSL::CloseImp(bool restart)
93{
94 if (IsConnected() && fVerbose)
95 {
96 ostringstream str;
97 str << "Connection closed to " << URL() << ".";
98 Info(str);
99 }
100
101 // Stop any pending connection attempt
102 fConnectionTimer.cancel();
103
104 // Close possible open connections
105 lowest_layer().close();
106
107 // Reset the connection status
108 fQueueSize = 0;
109 fConnectionStatus = kDisconnected;
110
111 // Stop deadline counters
112 fInTimeout.cancel();
113 fOutTimeout.cancel();
114
115 if (!restart || IsConnecting())
116 return;
117
118 // We need some timeout before reconnecting!
119 // And we have to check if we are alreayd trying to connect
120 // We shoudl wait until all operations in progress were canceled
121
122 // Start trying to reconnect
123 fMsgConnect = "";
124 fErrConnect = "";
125 StartConnect();
126}
127
128void ConnectionSSL::PostClose(bool restart)
129{
130 get_io_service().post(boost::bind(&ConnectionSSL::CloseImp, this, restart));
131}
132
133// ------------------------ write --------------------------
134void ConnectionSSL::HandleWriteTimeout(const bs::error_code &error)
135{
136 if (error==ba::error::basic_errors::operation_aborted)
137 return;
138
139 // 125: Operation canceled (bs::error_code(125, bs::system_category))
140 if (error)
141 {
142 ostringstream str;
143 str << "Write timeout of " << URL() << ": " << error.message() << " (" << error << ")";// << endl;
144 Error(str);
145
146 CloseImp();
147 return;
148 }
149
150 if (IsClosed())
151 {
152 // For example: Here we could schedule a new accept if we
153 // would not want to allow two connections at the same time.
154 return;
155 }
156
157 // Check whether the deadline has passed. We compare the deadline
158 // against the current time since a new asynchronous operation
159 // may have moved the deadline before this actor had a chance
160 // to run.
161 if (fOutTimeout.expires_at() > ba::deadline_timer::traits_type::now())
162 return;
163
164 Error("fOutTimeout has expired, writing data to "+URL());
165
166 CloseImp();
167}
168
169void ConnectionSSL::HandleSentData(const bs::error_code& error, size_t n)
170{
171 if (error==ba::error::basic_errors::operation_aborted)
172 return;
173
174 if (error && error != ba::error::not_connected)
175 {
176 ostringstream str;
177 str << "Writing to " << URL() << ": " << error.message() << " (" << error << ")";// << endl;
178 Error(str);
179
180 CloseImp();
181 return;
182 }
183
184 if (error == ba::error::not_connected)
185 {
186 ostringstream msg;
187 msg << n << " bytes could not be sent to " << URL() << " due to missing connection.";
188 Warn(msg);
189
190 return;
191 }
192
193 if (--fQueueSize==0)
194 fOutTimeout.cancel();
195
196 if (fDebugTx)
197 {
198 ostringstream msg;
199 msg << n << " bytes successfully sent to " << URL();
200 Debug(msg);
201 }
202}
203
204void ConnectionSSL::PostMessage(const void *ptr, size_t sz)
205{
206 // This function can be called from a different thread...
207 if (IsClosed())
208 return;
209
210 // ... this is why we have to increase fQueueSize first
211 fQueueSize++;
212
213 // ... and shift the deadline timer
214 // This is not ideal, because if we are continously
215 // filling the buffer, it will never timeout
216 AsyncWait(fOutTimeout, 5000, &ConnectionSSL::HandleWriteTimeout);
217
218 // Now we can schedule the buffer to be sent
219 AsyncWrite(ba::const_buffers_1(ptr, sz));
220
221 // If a socket is closed, all pending asynchronous
222 // operation will be aborted.
223}
224
225void ConnectionSSL::PostMessage(const string &cmd, size_t max)
226{
227 if (max==size_t(-1))
228 max = cmd.length()+1;
229
230 PostMessage(cmd.c_str(), min(cmd.length()+1, max));
231}
232
233void ConnectionSSL::HandleConnectionTimer(const bs::error_code &error)
234{
235 if (error==ba::error::basic_errors::operation_aborted)
236 return;
237
238 if (error)
239 {
240 ostringstream str;
241 str << "Connetion timer of " << URL() << ": " << error.message() << " (" << error << ")";// << endl;
242 Error(str);
243 }
244
245 if (!IsClosed())
246 {
247 // For example: Here we could schedule a new accept if we
248 // would not want to allow two connections at the same time.
249 return;
250 }
251
252 // Check whether the deadline has passed. We compare the deadline
253 // against the current time since a new asynchronous operation
254 // may have moved the deadline before this actor had a chance
255 // to run.
256 if (fConnectionTimer.expires_at() < ba::deadline_timer::traits_type::now())
257 StartConnect();
258}
259
260void ConnectionSSL::HandleHandshake(const tcp::endpoint &endpoint, const bs::error_code& error)
261{
262 const string host = endpoint.port()==0 ? "" :
263 endpoint.address().to_string()+':'+to_string((long long unsigned int)endpoint.port());
264
265 if (!error)
266 {
267 lowest_layer().set_option(boost::asio::socket_base::keep_alive(true));
268
269 const int optval = 30;
270 // First keep alive after 30s
271 setsockopt(lowest_layer().native(), SOL_TCP, TCP_KEEPIDLE, &optval, sizeof(optval));
272 // New keep alive after 30s
273 setsockopt(lowest_layer().native(), SOL_TCP, TCP_KEEPINTVL, &optval, sizeof(optval));
274
275 if (fVerbose)
276 Info("Connection established to "+host+"...");
277
278 fQueueSize = 0;
279 fConnectionStatus = kConnected;
280
281 ConnectionEstablished();
282 return;
283 }
284
285 // If returning from run will lead to deletion of this
286 // instance, close() is not needed (maybe implicitly called).
287 // If run is called again, close() is needed here. Otherwise:
288 // Software caused connection abort when we try to resolve
289 // the endpoint again.
290 CloseImp(false);
291
292 ostringstream msg;
293 if (!host.empty())
294 msg << "Handshake with " << host << ": " << error.message() << " (" << error << ")";
295
296 if (fErrConnect!=msg.str())
297 {
298 if (error!=ba::error::basic_errors::connection_refused)
299 fMsgConnect = "";
300 fErrConnect = msg.str();
301 Warn(fErrConnect);
302 }
303
304 if (error==ba::error::basic_errors::operation_aborted)
305 return;
306
307 fConnectionStatus = kConnecting;
308}
309
310bool ConnectionSSL::ConnectImp(const tcp::endpoint &endpoint, const bs::error_code& error)
311{
312 const string addr = endpoint.address().to_string();
313 const string host = endpoint.port()==0 ? "" : addr+':'+to_string((long long unsigned int)endpoint.port());
314
315 // Connection established
316 if (!error)
317 {
318 if (fVerbose)
319 Info("Starting handshake for "+fAddress);
320
321 //stream::set_verify_mode(ssl::verify_peer);
322 //stream::set_verify_callback(ssl::rfc2818_verification(fAddress));
323 //set_default_verify_paths();
324
325 async_handshake(stream::client,
326 boost::bind(&ConnectionSSL::HandleHandshake, this,
327 endpoint, dummy::error));
328 return true;
329 }
330
331 /*
332 handshake(stream::client);
333
334 lowest_layer().set_option(boost::asio::socket_base::keep_alive(true));
335
336 const int optval = 30;
337 // First keep alive after 30s
338 setsockopt(lowest_layer().native(), SOL_TCP, TCP_KEEPIDLE, &optval, sizeof(optval));
339 // New keep alive after 30s
340 setsockopt(lowest_layer().native(), SOL_TCP, TCP_KEEPINTVL, &optval, sizeof(optval));
341
342 if (fVerbose)
343 Info("Connection established to "+host+"...");
344
345 fQueueSize = 0;
346 fConnectionStatus = kConnected;
347
348 ConnectionEstablished();
349 return true;
350 }*/
351
352 // If returning from run will lead to deletion of this
353 // instance, close() is not needed (maybe implicitly called).
354 // If run is called again, close() is needed here. Otherwise:
355 // Software caused connection abort when we try to resolve
356 // the endpoint again.
357 CloseImp(false);
358
359 ostringstream msg;
360 if (!host.empty())
361 msg << "Connecting to " << host << ": " << error.message() << " (" << error << ")";
362
363 if (fErrConnect!=msg.str())
364 {
365 if (error!=ba::error::basic_errors::connection_refused)
366 fMsgConnect = "";
367 fErrConnect = msg.str();
368 Warn(fErrConnect);
369 }
370
371 if (error==ba::error::basic_errors::operation_aborted)
372 return true;
373
374 fConnectionStatus = kConnecting;
375
376 return false;
377/*
378 // Go on with the next
379 if (++iterator != tcp::resolver::iterator())
380 {
381 AsyncConnect(iterator);
382 return;
383 }
384*/
385 // No more entries to try, if we would not put anything else
386 // into the queue anymore it would now return (run() would return)
387
388 // Since we don't want to block the main loop, we wait using an
389 // asnychronous timer
390
391 // FIXME: Should we move this before AsyncConnect() ?
392// AsyncWait(fConnectionTimer, 250, &ConnectionSSL::HandleConnectionTimer);
393}
394
395void ConnectionSSL::ConnectIter(tcp::resolver::iterator iterator, const bs::error_code& error)
396{
397 if (ConnectImp(*iterator, error))
398 return;
399
400 // Go on with the next
401 if (++iterator != tcp::resolver::iterator())
402 {
403 AsyncConnect(iterator);
404 return;
405 }
406
407 // No more entries to try, if we would not put anything else
408 // into the queue anymore it would now return (run() would return)
409 AsyncWait(fConnectionTimer, 250, &ConnectionSSL::HandleConnectionTimer);
410}
411
412void ConnectionSSL::ConnectAddr(const tcp::endpoint &endpoint, const bs::error_code& error)
413{
414 if (ConnectImp(endpoint, error))
415 return;
416
417 AsyncWait(fConnectionTimer, 250, &ConnectionSSL::HandleConnectionTimer);
418}
419
420// FIXME: Async connect should get address and port as an argument
421void ConnectionSSL::StartConnect()
422{
423 fConnectionStatus = kConnecting;
424
425 if (fEndpoint!=tcp::endpoint())
426 {
427 ostringstream msg;
428 msg << "Trying to connect to " << fEndpoint << "...";
429 if (fMsgConnect!=msg.str())
430 {
431 fMsgConnect = msg.str();
432 Info(msg);
433 }
434
435 AsyncConnect();
436 return;
437 }
438
439 const bool valid = !fAddress.empty() || !fPort.empty();
440
441 boost::system::error_code ec;
442
443 ostringstream msg;
444 if (!valid)
445 msg << "No target address... connection attempt postponed.";
446 else
447 {
448 tcp::resolver resolver(get_io_service());
449
450 tcp::resolver::query query(fAddress, fPort);
451 tcp::resolver::iterator iterator = resolver.resolve(query, ec);
452
453 msg << "Trying to connect to " << URL() << "...";
454
455 // Start connection attempts (will also reset deadline counter)
456 if (!ec)
457 AsyncConnect(iterator);
458 else
459 msg << " " << ec.message() << " (" << ec << ")";
460 }
461
462 // Only output message if it has changed
463 if (fMsgConnect!=msg.str())
464 {
465 fMsgConnect = msg.str();
466 if (ec)
467 Error(msg);
468 if (!ec && fVerbose)
469 Info(msg);
470 }
471
472 if (!valid || ec)
473 AsyncWait(fConnectionTimer, 250, &ConnectionSSL::HandleConnectionTimer);
474}
475
476void ConnectionSSL::SetEndpoint(const string &addr, int port)
477{
478 if (fConnectionStatus>=1)
479 Warn("Connection or connection attempt in progress. New endpoint only valid for next connection.");
480
481 fAddress = addr;
482 fPort = to_string((long long)port);
483}
484
485void ConnectionSSL::SetEndpoint(const string &addr, const string &port)
486{
487 if (fConnectionStatus>=1 && URL()!=":")
488 Warn("Connection or connection attempt in progress. New endpoint only valid for next connection.");
489
490 fAddress = addr;
491 fPort = port;
492}
493
494void ConnectionSSL::SetEndpoint(const string &addr)
495{
496 const size_t p0 = addr.find_first_of(':');
497 const size_t p1 = addr.find_last_of(':');
498
499 if (p0==string::npos || p0!=p1)
500 {
501 Error("ConnectionSSL::SetEndpoint - Wrong format of argument '"+addr+"' ('host:port' expected)");
502 return;
503 }
504
505 SetEndpoint(addr.substr(0, p0), addr.substr(p0+1));
506}
507
508void ConnectionSSL::SetEndpoint(const tcp::endpoint &ep)
509{
510 const ba::ip::address addr = ep.address();
511
512 const ba::ip::address use =
513 addr.is_v6() && addr.to_v6().is_loopback() ?
514 ba::ip::address(ba::ip::address_v4::loopback()) :
515 addr;
516
517 SetEndpoint(use.to_string(), ep.port());
518
519 fEndpoint = tcp::endpoint(use, ep.port());
520}
521
522ConnectionSSL::ConnectionSSL(ba::io_service& ioservice, ostream &out) : MessageImp(out),
523ssl::context(ioservice, boost::asio::ssl::context::method::sslv23_client), stream(ioservice, *this),
524fLog(0), fVerbose(true), fDebugTx(false),
525fInTimeout(ioservice), fOutTimeout(ioservice), fConnectionTimer(ioservice),
526fQueueSize(0), fConnectionStatus(kDisconnected)
527{
528 //stream::set_verify_mode(ssl::verify_none);
529}
Note: See TracBrowser for help on using the repository browser.