protocol.cc 12 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428
  1. /* vim:expandtab:shiftwidth=2:tabstop=2:smarttab:
  2. *
  3. * Gearmand client and server library.
  4. *
  5. * Copyright (C) 2012 Data Differential, http://datadifferential.com/
  6. * All rights reserved.
  7. *
  8. * Redistribution and use in source and binary forms, with or without
  9. * modification, are permitted provided that the following conditions are
  10. * met:
  11. *
  12. * * Redistributions of source code must retain the above copyright
  13. * notice, this list of conditions and the following disclaimer.
  14. *
  15. * * Redistributions in binary form must reproduce the above
  16. * copyright notice, this list of conditions and the following disclaimer
  17. * in the documentation and/or other materials provided with the
  18. * distribution.
  19. *
  20. * * The names of its contributors may not be used to endorse or
  21. * promote products derived from this software without specific prior
  22. * written permission.
  23. *
  24. * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
  25. * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
  26. * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
  27. * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
  28. * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
  29. * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
  30. * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
  31. * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
  32. * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
  33. * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
  34. * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  35. *
  36. */
  37. /**
  38. * @file
  39. * @brief Gear Protocol Definitions
  40. */
  41. #include "gear_config.h"
  42. #include "configmake.h"
  43. #include <libgearman-server/common.h>
  44. #include <libgearman/strcommand.h>
  45. #include <libgearman-server/packet.h>
  46. #include <cstdio>
  47. #include <cstdlib>
  48. #if defined(HAVE_CYASSL) && HAVE_CYASSL
  49. # include <cyassl/ssl.h>
  50. #endif
  51. #include <libgearman-server/plugins/protocol/gear/protocol.h>
  52. #include "libgearman/command.h"
  53. static gearmand_error_t gearmand_packet_unpack_header(gearmand_packet_st *packet)
  54. {
  55. uint32_t tmp;
  56. if (memcmp(packet->args, "\0REQ", 4) == 0)
  57. {
  58. packet->magic= GEARMAN_MAGIC_REQUEST;
  59. }
  60. else if (memcmp(packet->args, "\0RES", 4) == 0)
  61. {
  62. packet->magic= GEARMAN_MAGIC_RESPONSE;
  63. }
  64. else
  65. {
  66. gearmand_warning("invalid magic value");
  67. return GEARMAND_INVALID_MAGIC;
  68. }
  69. memcpy(&tmp, packet->args + 4, 4);
  70. packet->command= static_cast<gearman_command_t>(ntohl(tmp));
  71. if (packet->command == GEARMAN_COMMAND_TEXT ||
  72. packet->command >= GEARMAN_COMMAND_MAX)
  73. {
  74. gearmand_error("invalid command value");
  75. return GEARMAND_INVALID_COMMAND;
  76. }
  77. memcpy(&tmp, packet->args + 8, 4);
  78. packet->data_size= ntohl(tmp);
  79. return GEARMAND_SUCCESS;
  80. }
  81. class Geartext : public gearmand::protocol::Context {
  82. public:
  83. ~Geartext()
  84. { }
  85. bool is_owner()
  86. {
  87. return false;
  88. }
  89. void notify(gearman_server_con_st *)
  90. {
  91. gearmand_info("Gear connection disconnected");
  92. }
  93. size_t unpack(gearmand_packet_st *packet,
  94. gearman_server_con_st *,
  95. const void *data, const size_t data_size,
  96. gearmand_error_t& ret_ptr)
  97. {
  98. size_t used_size;
  99. gearmand_info("Gear unpack");
  100. if (packet->args_size == 0)
  101. {
  102. if (data_size > 0 && ((uint8_t *)data)[0] != 0)
  103. {
  104. /* Try to parse a text-based command. */
  105. uint8_t* ptr= (uint8_t *)memchr(data, '\n', data_size);
  106. if (ptr == NULL)
  107. {
  108. ret_ptr= GEARMAND_IO_WAIT;
  109. return 0;
  110. }
  111. packet->magic= GEARMAN_MAGIC_TEXT;
  112. packet->command= GEARMAN_COMMAND_TEXT;
  113. used_size= size_t(ptr - ((uint8_t *)data)) +1;
  114. *ptr= 0;
  115. if (used_size > 1 && *(ptr - 1) == '\r')
  116. {
  117. *(ptr - 1)= 0;
  118. }
  119. size_t arg_size;
  120. for (arg_size= used_size, ptr= (uint8_t *)data; ptr != NULL; data= ptr)
  121. {
  122. ptr= (uint8_t *)memchr(data, ' ', arg_size);
  123. if (ptr != NULL)
  124. {
  125. *ptr= 0;
  126. ptr++;
  127. while (*ptr == ' ')
  128. {
  129. ptr++;
  130. }
  131. arg_size-= size_t(ptr - ((uint8_t *)data));
  132. }
  133. ret_ptr= gearmand_packet_create(packet, data, ptr == NULL ? arg_size :
  134. size_t(ptr - ((uint8_t *)data)));
  135. if (ret_ptr != GEARMAND_SUCCESS)
  136. {
  137. return used_size;
  138. }
  139. }
  140. return used_size;
  141. }
  142. else if (data_size < GEARMAND_PACKET_HEADER_SIZE)
  143. {
  144. ret_ptr= GEARMAND_IO_WAIT;
  145. return 0;
  146. }
  147. packet->args= packet->args_buffer;
  148. packet->args_size= GEARMAND_PACKET_HEADER_SIZE;
  149. memcpy(packet->args, data, GEARMAND_PACKET_HEADER_SIZE);
  150. if (gearmand_failed(ret_ptr= gearmand_packet_unpack_header(packet)))
  151. {
  152. return 0;
  153. }
  154. used_size= GEARMAND_PACKET_HEADER_SIZE;
  155. }
  156. else
  157. {
  158. used_size= 0;
  159. }
  160. while (packet->argc != gearman_command_info(packet->command)->argc)
  161. {
  162. if (packet->argc != (gearman_command_info(packet->command)->argc - 1) or
  163. gearman_command_info(packet->command)->data)
  164. {
  165. uint8_t* ptr= (uint8_t *)memchr(((uint8_t *)data) +used_size, 0,
  166. data_size -used_size);
  167. if (ptr == NULL)
  168. {
  169. gearmand_log_debug(GEARMAN_DEFAULT_LOG_PARAM,
  170. "Possible protocol error for %s, received only %u args",
  171. gearman_command_info(packet->command)->name, packet->argc);
  172. ret_ptr= GEARMAND_IO_WAIT;
  173. return used_size;
  174. }
  175. size_t arg_size= size_t(ptr - (((uint8_t *)data) + used_size)) +1;
  176. if (gearmand_failed((ret_ptr= gearmand_packet_create(packet, ((uint8_t *)data) + used_size, arg_size))))
  177. {
  178. return used_size;
  179. }
  180. packet->data_size-= arg_size;
  181. used_size+= arg_size;
  182. }
  183. else
  184. {
  185. if ((data_size - used_size) < packet->data_size)
  186. {
  187. ret_ptr= GEARMAND_IO_WAIT;
  188. return used_size;
  189. }
  190. ret_ptr= gearmand_packet_create(packet, ((uint8_t *)data) + used_size, packet->data_size);
  191. if (gearmand_failed(ret_ptr))
  192. {
  193. return used_size;
  194. }
  195. used_size+= packet->data_size;
  196. packet->data_size= 0;
  197. }
  198. }
  199. if (packet->command == GEARMAN_COMMAND_ECHO_RES or packet->command == GEARMAN_COMMAND_ECHO_REQ)
  200. {
  201. gearmand_log_debug(GEARMAN_DEFAULT_LOG_PARAM,
  202. "GEAR length: %" PRIu64 " gearmand_command_t: %s echo: %.*s",
  203. uint64_t(packet->data_size),
  204. gearman_strcommand(packet->command),
  205. int(packet->data_size),
  206. packet->data);
  207. }
  208. else if (packet->command == GEARMAN_COMMAND_TEXT)
  209. {
  210. gearmand_log_debug(GEARMAN_DEFAULT_LOG_PARAM,
  211. "GEAR length: %" PRIu64 " gearmand_command_t: %s text: %.*s",
  212. uint64_t(packet->data_size),
  213. gearman_strcommand(packet->command),
  214. int(packet->data_size),
  215. packet->data);
  216. }
  217. else
  218. {
  219. gearmand_log_debug(GEARMAN_DEFAULT_LOG_PARAM,
  220. "GEAR length: %" PRIu64 " gearmand_command_t: %s",
  221. uint64_t(packet->data_size),
  222. gearman_strcommand(packet->command));
  223. }
  224. ret_ptr= GEARMAND_SUCCESS;
  225. return used_size;
  226. }
  227. size_t pack(const gearmand_packet_st *packet,
  228. gearman_server_con_st*,
  229. void *data, const size_t data_size,
  230. gearmand_error_t& ret_ptr)
  231. {
  232. if (packet->command == GEARMAN_COMMAND_ECHO_RES)
  233. {
  234. gearmand_log_debug(GEARMAN_DEFAULT_LOG_PARAM,
  235. "GEAR length: %" PRIu64 " gearmand_command_t: %s echo: %.*",
  236. uint64_t(packet->data_size),
  237. gearman_strcommand(packet->command),
  238. int(packet->data_size),
  239. packet->data);
  240. }
  241. else
  242. {
  243. gearmand_log_debug(GEARMAN_DEFAULT_LOG_PARAM,
  244. "GEAR length: %" PRIu64 " gearmand_command_t: %s",
  245. uint64_t(packet->data_size),
  246. gearman_strcommand(packet->command));
  247. }
  248. if (packet->args_size == 0)
  249. {
  250. ret_ptr= GEARMAND_SUCCESS;
  251. return 0;
  252. }
  253. if (packet->args_size > data_size)
  254. {
  255. ret_ptr= GEARMAND_FLUSH_DATA;
  256. return 0;
  257. }
  258. memcpy(data, packet->args, packet->args_size);
  259. ret_ptr= GEARMAND_SUCCESS;
  260. return packet->args_size;
  261. }
  262. private:
  263. };
  264. static Geartext gear_context;
  265. static gearmand_error_t _gear_con_add(gearman_server_con_st *connection)
  266. {
  267. #if defined(HAVE_CYASSL) && HAVE_CYASSL
  268. if (Gearmand()->ctx_ssl())
  269. {
  270. if ((connection->_ssl= CyaSSL_new(Gearmand()->ctx_ssl())) == NULL)
  271. {
  272. return gearmand_log_error(GEARMAN_DEFAULT_LOG_PARAM, "CyaSSL_new() failed");
  273. }
  274. CyaSSL_set_fd(connection->_ssl, connection->con.fd);
  275. bool connecting= true;
  276. while (connecting)
  277. {
  278. if (CyaSSL_accept(connection->_ssl) == SSL_SUCCESS)
  279. {
  280. connecting= false;
  281. break;
  282. }
  283. if (CyaSSL_get_error(connection->_ssl, 0) != SSL_ERROR_WANT_READ)
  284. {
  285. int cyassl_error= CyaSSL_get_error(connection->_ssl, 0);
  286. char cyassl_error_buffer[1024]= { 0 };
  287. CyaSSL_ERR_error_string(cyassl_error, cyassl_error_buffer);
  288. return gearmand_log_error(GEARMAN_DEFAULT_LOG_PARAM, "%s(%d)", cyassl_error_buffer, cyassl_error);
  289. }
  290. }
  291. gearmand_log_info(GEARMAN_DEFAULT_LOG_PARAM, "GearSSL connection made: %d", connection->con.fd);
  292. }
  293. #endif
  294. connection->set_protocol(&gear_context);
  295. return GEARMAND_SUCCESS;
  296. }
  297. namespace gearmand {
  298. namespace protocol {
  299. Gear::Gear() :
  300. Plugin("Gear"),
  301. _port(GEARMAN_DEFAULT_TCP_PORT_STRING),
  302. opt_ssl(false)
  303. {
  304. command_line_options().add_options()
  305. ("port,p", boost::program_options::value(&_port)->default_value(GEARMAN_DEFAULT_TCP_PORT_STRING),
  306. "Port the server should listen on.")
  307. ("ssl", boost::program_options::bool_switch(&opt_ssl)->default_value(false),
  308. "Enable ssl connections.")
  309. ;
  310. }
  311. Gear::~Gear()
  312. {
  313. }
  314. gearmand_error_t Gear::start(gearmand_st *gearmand)
  315. {
  316. gearmand_error_t rc;
  317. if (_port.compare(GEARMAN_DEFAULT_TCP_PORT_STRING) == 0)
  318. {
  319. char* service;
  320. if ((service= getenv("GEARMAND_PORT")) and service[0])
  321. {
  322. _port.clear();
  323. _port.append(service);
  324. }
  325. }
  326. if (_port.empty())
  327. {
  328. const char* service= GEARMAN_DEFAULT_TCP_PORT_STRING;
  329. struct servent *gearman_servent;
  330. if ((gearman_servent= getservbyname(GEARMAN_DEFAULT_TCP_SERVICE, NULL)))
  331. {
  332. if (gearman_servent and gearman_servent->s_name)
  333. {
  334. service= gearman_servent->s_name;
  335. }
  336. }
  337. _port.clear();
  338. _port.append(service);
  339. }
  340. gearmand_log_info(GEARMAN_DEFAULT_LOG_PARAM, "Initializing Gear on port %s with SSL: %s", _port.c_str(), opt_ssl ? "true" : "false");
  341. #if defined(HAVE_CYASSL) && HAVE_CYASSL
  342. if (opt_ssl)
  343. {
  344. gearmand->init_ssl();
  345. if (CyaSSL_CTX_load_verify_locations(gearmand->ctx_ssl(), GEARMAND_CA_CERTIFICATE, 0) != SSL_SUCCESS)
  346. {
  347. gearmand_log_fatal(GEARMAN_DEFAULT_LOG_PARAM, "CyaSSL_CTX_load_verify_locations() cannot local the ca certificate %s", GEARMAND_CA_CERTIFICATE);
  348. }
  349. if (CyaSSL_CTX_use_certificate_file(gearmand->ctx_ssl(), GEARMAND_SERVER_PEM, SSL_FILETYPE_PEM) != SSL_SUCCESS)
  350. {
  351. gearmand_log_fatal(GEARMAN_DEFAULT_LOG_PARAM, "CyaSSL_CTX_use_certificate_file() cannot obtain certificate %s", GEARMAND_SERVER_PEM);
  352. }
  353. if (CyaSSL_CTX_use_PrivateKey_file(gearmand->ctx_ssl(), GEARMAND_SERVER_KEY, SSL_FILETYPE_PEM) != SSL_SUCCESS)
  354. {
  355. gearmand_log_fatal(GEARMAN_DEFAULT_LOG_PARAM, "CyaSSL_CTX_use_PrivateKey_file() cannot obtain certificate %s", GEARMAND_SERVER_KEY);
  356. }
  357. assert(gearmand->ctx_ssl());
  358. }
  359. #endif
  360. rc= gearmand_port_add(gearmand, _port.c_str(), _gear_con_add);
  361. return rc;
  362. }
  363. } // namespace protocol
  364. } // namespace gearmand