diff --git a/src/Makefile.am b/src/Makefile.am index 41fa3fa2..530af01e 100644 --- a/src/Makefile.am +++ b/src/Makefile.am @@ -39,9 +39,11 @@ libzmq_la_SOURCES = \ lb.hpp \ likely.hpp \ mailbox.hpp \ + mechanism.hpp \ msg.hpp \ mtrie.hpp \ mutex.hpp \ + null_mechanism.hpp \ object.hpp \ options.hpp \ own.hpp \ @@ -100,8 +102,10 @@ libzmq_la_SOURCES = \ kqueue.cpp \ lb.cpp \ mailbox.cpp \ + mechanism.cpp \ msg.cpp \ mtrie.cpp \ + null_mechanism.cpp \ object.cpp \ options.cpp \ own.cpp \ diff --git a/src/mechanism.cpp b/src/mechanism.cpp new file mode 100644 index 00000000..c89c7817 --- /dev/null +++ b/src/mechanism.cpp @@ -0,0 +1,73 @@ +/* + Copyright (c) 2007-2013 Contributors as noted in the AUTHORS file + + This file is part of 0MQ. + + 0MQ is free software; you can redistribute it and/or modify it under + the terms of the GNU Lesser General Public License as published by + the Free Software Foundation; either version 3 of the License, or + (at your option) any later version. + + 0MQ is distributed in the hope that it will be useful, + but WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + GNU Lesser General Public License for more details. + + You should have received a copy of the GNU Lesser General Public License + along with this program. If not, see . +*/ + +#include + +#include "mechanism.hpp" +#include "options.hpp" +#include "msg.hpp" +#include "err.hpp" +#include "wire.hpp" + +zmq::mechanism_t::mechanism_t (const options_t &options_) : + options (options_) +{ +} + +zmq::mechanism_t::~mechanism_t () +{ +} + +void zmq::mechanism_t::set_peer_identity (const void *id_ptr, size_t id_size) +{ + identity = blob_t (static_cast (id_ptr), id_size); +} + +void zmq::mechanism_t::peer_identity (msg_t *msg_) +{ + const int rc = msg_->init_size (identity.size ()); + errno_assert (rc == 0); + memcpy (msg_->data (), identity.data (), identity.size ()); + msg_->set_flags (msg_t::identity); +} + +const char *zmq::mechanism_t::socket_type_string (int socket_type) const +{ + static const char *names [] = {"PAIR", "PUB", "SUB", "REQ", "REP", + "DEALER", "ROUTER", "PULL", "PUSH", + "XPUB", "XSUB"}; + zmq_assert (socket_type >= 0 && socket_type <= 10); + return names [socket_type]; +} + +size_t zmq::mechanism_t::add_property (unsigned char *ptr, const char *name, + const void *value, size_t value_len) const +{ + const size_t name_len = strlen (name); + zmq_assert (name_len <= 255); + *ptr++ = static_cast (name_len); + memcpy (ptr, name, name_len); + ptr += name_len; + zmq_assert (value_len <= (2^31) - 1); + put_uint32 (ptr, static_cast (value_len)); + ptr += 4; + memcpy (ptr, value, value_len); + + return 1 + name_len + 4 + value_len; +} diff --git a/src/mechanism.hpp b/src/mechanism.hpp new file mode 100644 index 00000000..28ed0e14 --- /dev/null +++ b/src/mechanism.hpp @@ -0,0 +1,73 @@ +/* + Copyright (c) 2007-2013 Contributors as noted in the AUTHORS file + + This file is part of 0MQ. + + 0MQ is free software; you can redistribute it and/or modify it under + the terms of the GNU Lesser General Public License as published by + the Free Software Foundation; either version 3 of the License, or + (at your option) any later version. + + 0MQ is distributed in the hope that it will be useful, + but WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + GNU Lesser General Public License for more details. + + You should have received a copy of the GNU Lesser General Public License + along with this program. If not, see . +*/ + +#ifndef __ZMQ_MECHANISM_HPP_INCLUDED__ +#define __ZMQ_MECHANISM_HPP_INCLUDED__ + +#include + +#include "options.hpp" +#include "blob.hpp" + +namespace zmq +{ + + // Abstract class representing security mechanism. + // Different mechanism extedns this class. + + class msg_t; + + class mechanism_t + { + public: + + mechanism_t (const options_t &options_); + + virtual ~mechanism_t (); + + // Prepare next handshake message that is to be sent to the peer. + virtual int next_handshake_message (msg_t *msg_) = 0; + + // Process the handshake message received from the peer. + virtual int process_handshake_message (msg_t *msg_) = 0; + + // True iff the handshake stage is complete? + virtual bool is_handshake_complete () const = 0; + + void set_peer_identity (const void *id_ptr, size_t id_size); + + void peer_identity (msg_t *msg_); + + protected: + + const char *socket_type_string (int socket_type) const; + + size_t add_property (unsigned char *ptr, const char *name, + const void *value, size_t value_len) const; + + options_t options; + + private: + + blob_t identity; + }; + +} + +#endif diff --git a/src/null_mechanism.cpp b/src/null_mechanism.cpp new file mode 100644 index 00000000..e0f7ef52 --- /dev/null +++ b/src/null_mechanism.cpp @@ -0,0 +1,152 @@ +/* + Copyright (c) 2007-2013 Contributors as noted in the AUTHORS file + + This file is part of 0MQ. + + 0MQ is free software; you can redistribute it and/or modify it under + the terms of the GNU Lesser General Public License as published by + the Free Software Foundation; either version 3 of the License, or + (at your option) any later version. + + 0MQ is distributed in the hope that it will be useful, + but WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + GNU Lesser General Public License for more details. + + You should have received a copy of the GNU Lesser General Public License + along with this program. If not, see . +*/ + +#include "platform.hpp" +#ifdef ZMQ_HAVE_WINDOWS +#include "windows.hpp" +#endif + +#include +#include +#include + +#include "err.hpp" +#include "msg.hpp" +#include "wire.hpp" +#include "null_mechanism.hpp" + +zmq::null_mechanism_t::null_mechanism_t (const options_t &options_) : mechanism_t (options_), + ready_command_sent (false), + ready_command_received (false) +{ +} + +zmq::null_mechanism_t::~null_mechanism_t () +{ +} + +int zmq::null_mechanism_t::next_handshake_message (msg_t *msg_) +{ + if (ready_command_sent) { + errno = EAGAIN; + return -1; + } + + unsigned char * const command_buffer = (unsigned char *) malloc (512); + alloc_assert (command_buffer); + + unsigned char *ptr = command_buffer; + + // Add mechanism string + memcpy (ptr, "READY ", 8); + ptr += 8; + + // Add socket type property + const char *socket_type = socket_type_string (options.type); + ptr += add_property (ptr, "Socket-Type", socket_type, strlen (socket_type)); + + // Add identity property + if (options.type == ZMQ_REQ + || options.type == ZMQ_DEALER + || options.type == ZMQ_ROUTER) { + ptr += add_property (ptr, "Identity", + options.identity, options.identity_size); + } + + const size_t command_size = ptr - command_buffer; + const int rc = msg_->init_size (command_size); + errno_assert (rc == 0); + memcpy (msg_->data (), command_buffer, command_size); + free (command_buffer); + + ready_command_sent = true; + + return 0; +} + +int zmq::null_mechanism_t::process_handshake_message (msg_t *msg_) +{ + if (ready_command_received) { + errno = EPROTO; + return -1; + } + + const unsigned char *ptr = + static_cast (msg_->data ()); + size_t bytes_left = msg_->size (); + + if (bytes_left < 8 || memcmp (ptr, "READY ", 8)) { + errno = EPROTO; + return -1; + } + + ptr += 8; + bytes_left -= 8; + + // Parse the property list + while (bytes_left > 1) { + const size_t name_length = static_cast (*ptr); + ptr += 1; + bytes_left -= 1; + + if (bytes_left < name_length) + break; + const std::string name = std::string((const char *) ptr, name_length); + ptr += name_length; + bytes_left -= name_length; + + if (bytes_left < 4) + break; + const size_t value_length = static_cast (get_uint32 (ptr)); + ptr += 4; + bytes_left -= 4; + + if (bytes_left < value_length) + break; + const unsigned char * const value = ptr; + ptr += value_length; + bytes_left -= value_length; + + if (name == "Socket-Type") { + // TODO: Implement socket type checking + } + else + if (name == "Identity" && options.recv_identity) + set_peer_identity (value, value_length); + } + + if (bytes_left > 0) { + errno = EPROTO; + return -1; + } + + int rc = msg_->close (); + errno_assert (rc == 0); + rc = msg_->init (); + errno_assert (rc == 0); + + ready_command_received = true; + + return 0; +} + +bool zmq::null_mechanism_t::is_handshake_complete () const +{ + return ready_command_received && ready_command_sent; +} diff --git a/src/null_mechanism.hpp b/src/null_mechanism.hpp new file mode 100644 index 00000000..2c5d9c27 --- /dev/null +++ b/src/null_mechanism.hpp @@ -0,0 +1,51 @@ +/* + Copyright (c) 2007-2013 Contributors as noted in the AUTHORS file + + This file is part of 0MQ. + + 0MQ is free software; you can redistribute it and/or modify it under + the terms of the GNU Lesser General Public License as published by + the Free Software Foundation; either version 3 of the License, or + (at your option) any later version. + + 0MQ is distributed in the hope that it will be useful, + but WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + GNU Lesser General Public License for more details. + + You should have received a copy of the GNU Lesser General Public License + along with this program. If not, see . +*/ + +#ifndef __ZMQ_NULL_MECHANISM_HPP_INCLUDED__ +#define __ZMQ_NULL_MECHANISM_HPP_INCLUDED__ + +#include "mechanism.hpp" +#include "options.hpp" + +namespace zmq +{ + + class msg_t; + + class null_mechanism_t : public mechanism_t + { + public: + + null_mechanism_t (const options_t &options_); + virtual ~null_mechanism_t (); + + // mechanism implementation + virtual int next_handshake_message (msg_t *msg_); + virtual int process_handshake_message (msg_t *msg_); + virtual bool is_handshake_complete () const; + + private: + + bool ready_command_sent; + bool ready_command_received; + }; + +} + +#endif diff --git a/src/stream_engine.cpp b/src/stream_engine.cpp index b894694a..6e4e69d1 100644 --- a/src/stream_engine.cpp +++ b/src/stream_engine.cpp @@ -40,6 +40,7 @@ #include "v1_decoder.hpp" #include "v2_encoder.hpp" #include "v2_decoder.hpp" +#include "null_mechanism.hpp" #include "raw_decoder.hpp" #include "raw_encoder.hpp" #include "config.hpp" @@ -69,8 +70,9 @@ zmq::stream_engine_t::stream_engine_t (fd_t fd_, const options_t &options_, cons io_error (false), congested (false), subscription_required (false), + mechanism (NULL), + input_paused (false), output_paused (false), - ready_command_received (false), socket (NULL) { int rc = tx_msg.init (); @@ -519,8 +521,10 @@ bool zmq::stream_engine_t::handshake () alloc_assert (decoder); if (memcmp (greeting_recv + 12, "NULL\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0", 20) == 0) { - read_msg = &stream_engine_t::send_ready_command; - write_msg = &stream_engine_t::receive_ready_command; + mechanism = new (std::nothrow) null_mechanism_t (options); + alloc_assert (mechanism); + read_msg = &stream_engine_t::next_handshake_message; + write_msg = &stream_engine_t::process_handshake_message; } else { error (); @@ -571,133 +575,59 @@ int zmq::stream_engine_t::write_identity (msg_t *msg_) return 0; } -int zmq::stream_engine_t::send_ready_command (msg_t *msg_) +int zmq::stream_engine_t::next_handshake_message (msg_t *msg_) { - unsigned char * const command_buffer = (unsigned char *) malloc (512); - alloc_assert (command_buffer); + zmq_assert (mechanism != NULL); - unsigned char *ptr = command_buffer; - - // Add mechanism string - memcpy (ptr, "READY ", 8); - ptr += 8; - - // Add socket type property - const char *socket_type = socket_type_string (options.type); - ptr += add_property (ptr, "Socket-Type", socket_type, strlen (socket_type)); - - // Add identity property - if (options.type == ZMQ_REQ - || options.type == ZMQ_DEALER - || options.type == ZMQ_ROUTER) { - ptr += add_property (ptr, "Identity", - options.identity, options.identity_size); + const int rc = mechanism->next_handshake_message (msg_); + if (rc == 0) { + if (mechanism->is_handshake_complete ()) + mechanism_ready (); + if (input_paused) { + activate_in (); + input_paused = false; + } } - - const size_t command_size = ptr - command_buffer; - const int rc = msg_->init_size (command_size); - errno_assert (rc == 0); - memcpy (msg_->data (), command_buffer, command_size); - free (command_buffer); - - if (ready_command_received) - read_msg = &stream_engine_t::pull_msg_from_session; else - read_msg = &stream_engine_t::wait; - - return 0; -} - -int zmq::stream_engine_t::receive_ready_command (msg_t *msg_) -{ - const unsigned char * const command_buffer = - static_cast (msg_->data ()); - const size_t command_size = msg_->size (); - - const unsigned char *ptr = command_buffer; - size_t bytes_left = command_size; - - if (bytes_left < 8 || memcmp(ptr, "READY ", 8)) { - errno = EPROTO; - return -1; - } - - ptr += 8; - bytes_left -= 8; - - // Parse the property list - while (bytes_left > 1) { - const size_t name_length = static_cast (*ptr); - ptr += 1; - bytes_left -= 1; - - if (bytes_left < name_length) - break; - const std::string name = std::string((const char *) ptr, name_length); - ptr += name_length; - bytes_left -= name_length; - - if (bytes_left < 4) - break; - const size_t value_length = static_cast (get_uint32 (ptr)); - ptr += 4; - bytes_left -= 4; - - if (bytes_left < value_length) - break; - const unsigned char * const value = ptr; - ptr += value_length; - bytes_left -= value_length; - - if (name == "Socket-Type") { - // Implement socket type checking - } - else - if (name == "Identity") { - if (options.recv_identity) { - msg_t identity; - int rc = identity.init_size (value_length); - errno_assert (rc == 0); - memcpy (identity.data (), value, value_length); - identity.set_flags (msg_t::identity); - rc = session->push_msg (&identity); - errno_assert (rc == 0); - } - } - } - - if (bytes_left > 0) { - errno = EPROTO; - return -1; - } - - int rc = msg_->close (); - errno_assert (rc == 0); - rc = msg_->init (); - errno_assert (rc == 0); - - write_msg = &stream_engine_t::push_msg_to_session; - - ready_command_received = true; - if (output_paused) { - activate_out (); - output_paused = false; - } - - return 0; -} - -int zmq::stream_engine_t::wait (msg_t *msg_) -{ - if (ready_command_received) { - read_msg = &stream_engine_t::pull_msg_from_session; - return pull_msg_from_session (msg_); - } - else { + if (rc == -1) { + zmq_assert (errno == EAGAIN); output_paused = true; - errno = EAGAIN; - return -1; } + + return rc; +} + +int zmq::stream_engine_t::process_handshake_message (msg_t *msg_) +{ + zmq_assert (mechanism != NULL); + + const int rc = mechanism->process_handshake_message (msg_); + if (rc == 0) { + if (mechanism->is_handshake_complete ()) + mechanism_ready (); + if (output_paused) { + activate_out (); + output_paused = false; + } + } + else + if (rc == -1 && errno == EAGAIN) + input_paused = true; + + return rc; +} + +void zmq::stream_engine_t::mechanism_ready () +{ + if (options.recv_identity) { + msg_t identity; + mechanism->peer_identity (&identity); + const int rc = session->push_msg (&identity); + errno_assert (rc == 0); + } + + read_msg = &stream_engine_t::pull_msg_from_session; + write_msg = &stream_engine_t::push_msg_to_session; } int zmq::stream_engine_t::pull_msg_from_session (msg_t *msg_) @@ -727,29 +657,6 @@ int zmq::stream_engine_t::write_subscription_msg (msg_t *msg_) return push_msg_to_session (msg_); } -size_t zmq::stream_engine_t::add_property (unsigned char *ptr, - const char *name, const void *value, size_t value_len) -{ - const size_t name_len = strlen (name); - zmq_assert (name_len <= 255); - *ptr++ = static_cast (name_len); - memcpy (ptr, name, name_len); - ptr += name_len; - zmq_assert (value_len <= (2^31) - 1); - put_uint32 (ptr, static_cast (value_len)); - ptr += 4; - memcpy (ptr, value, value_len); - - return 1 + name_len + 4 + value_len; -} - -const char *zmq::stream_engine_t::socket_type_string (int socket_type) { - const char *names [] = {"PAIR", "PUB", "SUB", "REQ", "REP", "DEALER", - "ROUTER", "PULL", "PUSH", "XPUB", "XSUB"}; - zmq_assert (socket_type >= 0 && socket_type <= 10); - return names [socket_type]; -} - void zmq::stream_engine_t::error () { zmq_assert (session); diff --git a/src/stream_engine.hpp b/src/stream_engine.hpp index cb8c75e1..91d641b8 100644 --- a/src/stream_engine.hpp +++ b/src/stream_engine.hpp @@ -43,6 +43,7 @@ namespace zmq class io_thread_t; class msg_t; class session_base_t; + class mechanism_t; // This engine handles any socket with SOCK_STREAM semantics, // e.g. TCP socket or an UNIX domain socket. @@ -93,13 +94,13 @@ namespace zmq int read_identity (msg_t *msg_); int write_identity (msg_t *msg_); - int send_ready_command (msg_t *msg); - int receive_ready_command (msg_t *msg); + int next_handshake_message (msg_t *msg); + int process_handshake_message (msg_t *msg); int pull_msg_from_session (msg_t *msg_); int push_msg_to_session (msg_t *msg); - int wait (msg_t *msg_); + void mechanism_ready (); int write_subscription_msg (msg_t *msg_); @@ -172,8 +173,10 @@ namespace zmq // Needed to support old peers. bool subscription_required; + mechanism_t *mechanism; + + bool input_paused; bool output_paused; - bool ready_command_received; // Socket zmq::socket_base_t *socket;