IBR-DTNSuite  0.12
EMailSmtpService.cpp
Go to the documentation of this file.
1 /*
2  * EMailSmtpService.cpp
3  *
4  * Copyright (C) 2013 IBR, TU Braunschweig
5  *
6  * Written-by: Björn Gernert <mail@bjoern-gernert.de>
7  *
8  * Licensed under the Apache License, Version 2.0 (the "License");
9  * you may not use this file except in compliance with the License.
10  * You may obtain a copy of the License at
11  *
12  * http://www.apache.org/licenses/LICENSE-2.0
13  *
14  * Unless required by applicable law or agreed to in writing, software
15  * distributed under the License is distributed on an "AS IS" BASIS,
16  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
17  * See the License for the specific language governing permissions and
18  * limitations under the License.
19  *
20  */
21 
22 #include "net/EMailSmtpService.h"
23 #include "net/EMailImapService.h"
24 
25 #include "Configuration.h"
26 #include "core/BundleCore.h"
27 #include "core/BundleEvent.h"
28 #include "net/TransferAbortedEvent.h"
29 #include "net/TransferCompletedEvent.h"
30 #include "storage/BundleStorage.h"
31 
32 #include <ibrcommon/Logger.h>
33 #include <vmime/platforms/posix/posixHandler.hpp>
34 
35 namespace dtn
36 {
37  namespace net
38  {
39  bool EMailSmtpService::_run(false);
40 
41  EMailSmtpService& EMailSmtpService::getInstance()
42  {
43  static EMailSmtpService instance;
44  return instance;
45  }
46 
47  EMailSmtpService::EMailSmtpService()
48  : _config(daemon::Configuration::getInstance().getEMail()),
49  _storage(dtn::core::BundleCore::getInstance().getStorage()),
50  _certificateVerifier(vmime::create<vmime::security::cert::defaultCertificateVerifier>())
51  {
52  // Load certificates
53  loadCerificates();
54 
55  try {
56  // Initialize VMime engine
57  vmime::platform::setHandler<vmime::platforms::posix::posixHandler>();
58 
59  // Set SMTP server for outgoing mails (use SSL or not)
60  std::string url;
61  if(_config.smtpUseSSL())
62  {
63  url = "smtps://" + _config.getSmtpServer();
64  }else{
65  url = "smtp://" + _config.getSmtpServer();
66  }
67  vmime::ref<vmime::net::session> session =
68  vmime::create<vmime::net::session>();
69 
70  // Create an instance of the transport service
71  _transport = session->getTransport(vmime::utility::url(url));
72 
73  // Set username, password and port
74  _transport->setProperty("options.need-authentication",
75  _config.smtpAuthenticationNeeded());
76  _transport->setProperty("auth.username", _config.getSmtpUsername());
77  _transport->setProperty("auth.password", _config.getSmtpPassword());
78  _transport->setProperty("server.port", _config.getSmtpPort());
79 
80  // Enable TLS
81  if(_config.smtpUseTLS())
82  {
83  _transport->setProperty("connection.tls", true);
84  _transport->setProperty("connection.tls.required", true);
85  }
86 
87  // Handle timeouts
88  _transport->setTimeoutHandlerFactory(vmime::create<TimeoutHandlerFactory>());
89 
90  // Add certificate verifier
91  if(_config.imapUseTLS() || _config.imapUseSSL())
92  {
93  _transport->setCertificateVerifier(_certificateVerifier);
94  }
95  } catch (vmime::exception &e) {
96  IBRCOMMON_LOGGER(error) << "EMail Convergence Layer: Unable to create SMTP service: " << e.what() << IBRCOMMON_LOGGER_ENDL;
97  }
98 
99  // Lock mutex
100  _threadMutex.enter();
101 
102  // Start the thread
103  _run = true;
104  this->start();
105  }
106 
107  EMailSmtpService::~EMailSmtpService()
108  {
109  this->stop();
110  this->join();
111 
112  // Delete remaining tasks
113  _queue.reset();
114  while(!_queue.empty()) {
115  Task *t = _queue.front();
116  _queue.pop();
117  delete t;
118  t = NULL;
119  }
120  _queue.abort();
121  }
122 
123  void EMailSmtpService::run() throw ()
124  {
125  while(true)
126  {
127  _threadMutex.enter();
128 
129  if(!_run)
130  break;
131 
132  try {
133  // Get Task
134  Task *t = _queue.getnpop(true, _config.getSmtpKeepAliveTimeout());
135 
136  //Check if bundle is still in the storage
137  if ( ! _storage.contains(t->getJob().getBundle()) )
138  {
139  // Destroy Task
140  delete t;
141  t = NULL;
142  continue;
143  }
144 
145  try {
146  // Submit Task
147  submit(t);
148  EMailImapService::getInstance().storeProcessedTask(t);
149  } catch(std::exception &e) {
150  if(!_run)
151  break;
152  if(dynamic_cast<vmime::exceptions::authentication_error*>(&e) != NULL)
153  {
154  dtn::net::TransferAbortedEvent::raise(t->getNode().getEID(), t->getJob().getBundle(), dtn::net::TransferAbortedEvent::REASON_CONNECTION_DOWN);
155  IBRCOMMON_LOGGER(error) << "EMail Convergence Layer: SMTP authentication error (username/password correct?)" << IBRCOMMON_LOGGER_ENDL;
156  }
157  else if(dynamic_cast<vmime::exceptions::connection_error*>(&e) != NULL)
158  {
159  dtn::net::TransferAbortedEvent::raise(t->getNode().getEID(), t->getJob().getBundle(), dtn::net::TransferAbortedEvent::REASON_CONNECTION_DOWN);
160  IBRCOMMON_LOGGER(error) << "EMail Convergence Layer: Unable to connect to the SMTP server" << IBRCOMMON_LOGGER_ENDL;
161  }
162  else
163  {
164  dtn::net::TransferAbortedEvent::raise(t->getNode().getEID(), t->getJob().getBundle(), dtn::net::TransferAbortedEvent::REASON_UNDEFINED);
165  IBRCOMMON_LOGGER(error) << "EMail Convergence Layer: SMTP error: " << e.what() << IBRCOMMON_LOGGER_ENDL;
166  }
167  // Delete task
168  delete t;
169  t = NULL;
170 
171  _threadMutex.leave();
172  continue;
173  }
174  } catch(ibrcommon::QueueUnblockedException&) {
175  // Disconnect from server
176  disconnect();
177  }
178  }
179  }
180 
181  void EMailSmtpService::__cancellation() throw ()
182  {
183  _run = false;
184  disconnect();
185  _queue.abort();
186  _threadMutex.leave();
187  }
188 
189  void EMailSmtpService::connect()
190  {
191  if(!isConnected())
192  _transport->connect();
193  }
194 
195  bool EMailSmtpService::isConnected()
196  {
197  return _transport->isConnected();
198  }
199 
200  void EMailSmtpService::disconnect()
201  {
202  if(isConnected())
203  _transport->disconnect();
204  }
205 
206  void EMailSmtpService::loadCerificates()
207  {
208  std::vector<std::string> certPath;
209  // Load CA certificates
210  std::vector<vmime::ref <vmime::security::cert::X509Certificate> > ca;
211  certPath = _config.getTlsCACerts();
212  if(!certPath.empty()) {
213  for(std::vector<std::string>::iterator it = certPath.begin() ; it != certPath.end(); ++it)
214  {
215  try {
216  ca.push_back(loadCertificateFromFile((*it)));
217  }catch(InvalidCertificate &e) {
218  IBRCOMMON_LOGGER(error) << "EMail Convergence Layer: Certificate load error: " << e.what() << IBRCOMMON_LOGGER_ENDL;
219  }
220  }
221  _certificateVerifier->setX509RootCAs(ca);
222  }
223 
224  // Load user certificates
225  std::vector<vmime::ref <vmime::security::cert::X509Certificate> > user;
226  certPath = _config.getTlsUserCerts();
227  if(!certPath.empty()) {
228  for(std::vector<std::string>::iterator it = certPath.begin() ; it != certPath.end(); ++it)
229  {
230  try {
231  user.push_back(loadCertificateFromFile((*it)));
232  }catch(InvalidCertificate &e) {
233  IBRCOMMON_LOGGER(error) << "EMail Convergence Layer: Certificate load error: " << e.what() << IBRCOMMON_LOGGER_ENDL;
234  }
235  }
236  _certificateVerifier->setX509TrustedCerts(user);
237  }
238  }
239 
240  vmime::ref<vmime::security::cert::X509Certificate> EMailSmtpService::loadCertificateFromFile(const std::string &path)
241  {
242  std::ifstream certFile;
243  certFile.open(path.c_str(), std::ios::in | std::ios::binary);
244  if(!certFile)
245  {
246  throw(InvalidCertificate("Unable to find certificate at \"" + path + "\""));
247  return NULL;
248  }
249 
250  vmime::utility::inputStreamAdapter is(certFile);
251  vmime::ref<vmime::security::cert::X509Certificate> cert;
252  cert = vmime::security::cert::X509Certificate::import(is);
253  if(cert != NULL)
254  {
255  return cert;
256  }else{
257  throw(InvalidCertificate("The certificate at \"" + path + "\" does not seem to be PEM or DER encoded"));
258  return NULL;
259  }
260  }
261 
262  void EMailSmtpService::queueTask(Task *t)
263  {
264  _queue.push(t);
265  }
266 
267  void EMailSmtpService::submitNow(Task *t)
268  {
269  _queue.push(t);
270  _threadMutex.leave();
271  }
272 
273  void EMailSmtpService::submitQueue()
274  {
275  _threadMutex.leave();
276  }
277 
278  void EMailSmtpService::submit(Task *t)
279  {
280  // Check connection
281  if(!isConnected())
282  connect();
283 
284  const dtn::data::Bundle bundle = _storage.get(t->getJob().getBundle());
285  const dtn::data::MetaBundle meta = dtn::data::MetaBundle::create(bundle);
286 
287  // Create new email
288  vmime::ref<vmime::message> msg = vmime::create <vmime::message>();
289  vmime::ref<vmime::header> header = msg->getHeader();
290  vmime::ref<vmime::body> body = msg->getBody();
291 
292  // Create header
293  vmime::headerFieldFactory* hfFactory =
294  vmime::headerFieldFactory::getInstance();
295 
296  // From
297  header->appendField(hfFactory->create(vmime::fields::FROM,
298  _config.getOwnAddress()));
299  // To
300  header->appendField(hfFactory->create(vmime::fields::TO,
301  t->getRecipient()));
302  // Subject
303  header->appendField(hfFactory->create(vmime::fields::SUBJECT,
304  "Bundle for " + t->getNode().getEID().getString()));
305 
306  // Insert header for primary block
307  header->appendField(hfFactory->create("Bundle-EMailCL-Version", "1"));
308  header->appendField(hfFactory->create("Bundle-Flags",
309  bundle.procflags.toString()));
310  header->appendField(hfFactory->create("Bundle-Destination",
311  bundle.destination.getString()));
312  header->appendField(hfFactory->create("Bundle-Source",
313  bundle.source.getString()));
314  header->appendField(hfFactory->create("Bundle-Report-To",
315  bundle.reportto.getString()));
316  header->appendField(hfFactory->create("Bundle-Custodian",
317  bundle.custodian.getString()));
318  header->appendField(hfFactory->create("Bundle-Creation-Time",
319  bundle.timestamp.toString()));
320  header->appendField(hfFactory->create("Bundle-Sequence-Number",
321  bundle.sequencenumber.toString()));
322  header->appendField(hfFactory->create("Bundle-Lifetime",
323  bundle.lifetime.toString()));
324  if(bundle.get(dtn::data::Bundle::FRAGMENT))
325  {
326  header->appendField(hfFactory->create("Bundle-Fragment-Offset",
327  bundle.fragmentoffset.toString()));
328  header->appendField(hfFactory->create("Bundle-Total-Application-Data-Unit-Length",
329  bundle.appdatalength.toString()));
330  }
331 
332  // Count additional blocks
333  unsigned int additionalBlocks = 0;
334 
335  // Set MIME-Header if we we need attachments
336  if(bundle.size() > 0)
337  header->appendField(hfFactory->create("MIME-Version","1.0"));
338 
339  for (dtn::data::Bundle::const_iterator iter = bundle.begin(); iter != bundle.end(); ++iter) {
340  // Payload Block
341  try {
342  const dtn::data::PayloadBlock &payload =
343  dynamic_cast<const dtn::data::PayloadBlock&>(**iter);
344 
345  header->appendField(hfFactory->create("Bundle-Payload-Flags",
346  toString(getProcFlags(payload))));
347  header->appendField(hfFactory->create("Bundle-Payload-Block-Length",
348  toString(payload.getLength())));
349  header->appendField(hfFactory->create("Bundle-Payload-Data-Name",
350  "payload.data"));
351 
352  // Create payload attachment
353  ibrcommon::BLOB::iostream data = payload.getBLOB().iostream();
354  std::stringstream ss;
355  ss << data->rdbuf();
356 
357  vmime::ref<vmime::utility::inputStream> dataStream =
358  vmime::create<vmime::utility::inputStreamStringAdapter>(ss.str());
359 
360  vmime::ref<vmime::attachment> payloadAttachement =
361  vmime::create<vmime::fileAttachment>(
362  dataStream,
363  vmime::word("payload.data"),
364  vmime::mediaType("application/octet-stream")
365  );
366  vmime::attachmentHelper::addAttachment(msg, payloadAttachement);
367  continue;
368  } catch(const std::bad_cast&) {};
369 
370  // Add extension blocks
371  try {
372  const dtn::data::Block &block = (**iter);
373 
374  // Create extension attachment
375  std::stringstream attachment;
376  size_t length = 0;
377  (**iter).serialize(attachment, length);
378 
379  vmime::ref<vmime::utility::inputStream> dataStream =
380  vmime::create<vmime::utility::inputStreamStringAdapter>(attachment.str());
381 
382  vmime::ref<vmime::attachment> extensionAttachement =
383  vmime::create<vmime::fileAttachment>(
384  dataStream,
385  vmime::word("block-" + toString(additionalBlocks) + ".data"),
386  vmime::mediaType("application/octet-stream")
387  );
388 
389  vmime::attachmentHelper::addAttachment(msg, extensionAttachement);
390 
391  // Get header of attachment
392  vmime::ref<vmime::header> extensionHeader = msg->getBody()->getPartAt(msg->getBody()->getPartCount()-1)->getHeader();
393  std::stringstream ss;
394  ss << (int)block.getType();
395 
396  extensionHeader->appendField(hfFactory->create("Block-Type", ss.str()));
397 
398  extensionHeader->appendField(hfFactory->create("Block-Processing-Flags",
399  toString(getProcFlags(block))));
400 
401  // Get EIDs
402  if(block.get(dtn::data::Block::BLOCK_CONTAINS_EIDS))
403  {
404  const Block::eid_list &eids = block.getEIDList();
405  for(Block::eid_list::const_iterator eidIt = eids.begin(); eidIt != eids.end(); eidIt++)
406  {
407  extensionHeader->appendField(hfFactory->create("Block-EID-Reference",
408  (*eidIt).getString()));
409  }
410  }
411 
412  // Add filename to header
413  header->appendField(hfFactory->create("Bundle-Additional-Block",
414  "block-" + toString(additionalBlocks++) + ".data"));
415  continue;
416  } catch(const std::bad_cast&) {};
417 
418  }
419  _transport->send(msg);
420  dtn::net::TransferCompletedEvent::raise(t->getNode().getEID(), meta);
421  dtn::core::BundleEvent::raise(meta, dtn::core::BUNDLE_FORWARDED);
422  IBRCOMMON_LOGGER(info) << "EMail Convergence Layer: Bundle " << t->getJob().getBundle().toString() << " for node " << t->getNode().getEID().getString() << " submitted via smtp" << IBRCOMMON_LOGGER_ENDL;
423  }
424 
425  unsigned int EMailSmtpService::getProcFlags(const dtn::data::Block &block)
426  {
427  unsigned int out = 0;
428 
429  if (block.get(dtn::data::Block::REPLICATE_IN_EVERY_FRAGMENT))
430  out |= dtn::data::Block::REPLICATE_IN_EVERY_FRAGMENT;
431  if (block.get(dtn::data::Block::TRANSMIT_STATUSREPORT_IF_NOT_PROCESSED))
432  out |= dtn::data::Block::TRANSMIT_STATUSREPORT_IF_NOT_PROCESSED;
433  if (block.get(dtn::data::Block::DELETE_BUNDLE_IF_NOT_PROCESSED))
434  out |= dtn::data::Block::DELETE_BUNDLE_IF_NOT_PROCESSED;
435  if (block.get(dtn::data::Block::LAST_BLOCK))
436  out |= dtn::data::Block::LAST_BLOCK;
437  if (block.get(dtn::data::Block::DISCARD_IF_NOT_PROCESSED))
438  out |= dtn::data::Block::DISCARD_IF_NOT_PROCESSED;
439  if (block.get(dtn::data::Block::FORWARDED_WITHOUT_PROCESSED))
440  out |= dtn::data::Block::FORWARDED_WITHOUT_PROCESSED;
441  if (block.get(dtn::data::Block::BLOCK_CONTAINS_EIDS))
442  out |= dtn::data::Block::BLOCK_CONTAINS_EIDS;
443 
444  return out;
445  }
446 
447  std::string EMailSmtpService::toString(int i)
448  {
449  std::stringstream ss;
450  ss << i;
451  return ss.str();
452  }
453 
454  EMailSmtpService::Task::Task(const dtn::core::Node &node, const dtn::net::BundleTransfer &job, std::string recipient)
455  : _node(node), _job(job), _recipient(recipient), _timesChecked(0) {}
456 
457  EMailSmtpService::Task::~Task() {}
458 
459  const dtn::core::Node EMailSmtpService::Task::getNode()
460  {
461  return _node;
462  }
463 
464  const dtn::net::BundleTransfer EMailSmtpService::Task::getJob()
465  {
466  return _job;
467  }
468 
469  std::string EMailSmtpService::Task::getRecipient()
470  {
471  return _recipient;
472  }
473 
474  bool EMailSmtpService::Task::checkForReturningMail()
475  {
476  return (!++_timesChecked > daemon::Configuration::getInstance().getEMail().getReturningMailChecks());
477  }
478 
479  bool EMailSmtpService::TimeoutHandler::isTimeOut()
480  {
481  if(!_run)
482  return true;
483 
484  return (getTime() >= last +
485  daemon::Configuration::getInstance().getEMail().getSmtpConnectionTimeout());
486  }
487 
488  void EMailSmtpService::TimeoutHandler::resetTimeOut()
489  {
490  last = getTime();
491  }
492 
493  bool EMailSmtpService::TimeoutHandler::handleTimeOut()
494  {
495  // true = continue, false = cancel
496  return false;
497  }
498 
499  }
500 }