Merge pull request #617 from hurtonm/zap_updates

Zap updates

src/curve_server.cpp

@@ -33,9 +33,11 @@
 #include "wire.hpp"
 
 zmq::curve_server_t::curve_server_t (session_base_t *session_,
+                                     const std::string &peer_address_,
                                      const options_t &options_) :
     mechanism_t (options_),
     session (session_),
+    peer_address (peer_address_),
     state (expect_hello),
     expecting_zap_reply (false),
     cn_nonce (1)
@@ -512,7 +514,7 @@ void zmq::curve_server_t::send_zap_request (const uint8_t *key)
     rc = session->write_zap_msg (&msg);
     errno_assert (rc == 0);
 
-    //  Sequence frame
+    //  Request ID frame
     rc = msg.init_size (1);
     errno_assert (rc == 0);
     memcpy (msg.data (), "1", 1);
@@ -527,6 +529,14 @@ void zmq::curve_server_t::send_zap_request (const uint8_t *key)
     rc = session->write_zap_msg (&msg);
     errno_assert (rc == 0);
 
+    //  Address frame
+    rc = msg.init_size (peer_address.length ());
+    errno_assert (rc == 0);
+    memcpy (msg.data (), peer_address.c_str (), peer_address.length ());
+    msg.set_flags (msg_t::more);
+    rc = session->write_zap_msg (&msg);
+    errno_assert (rc == 0);
+
     //  Mechanism frame
     rc = msg.init_size (5);
     errno_assert (rc == 0);
@@ -546,18 +556,19 @@ void zmq::curve_server_t::send_zap_request (const uint8_t *key)
 int zmq::curve_server_t::receive_and_process_zap_reply ()
 {
     int rc = 0;
-    msg_t msg [6];
+    msg_t msg [7];  //  ZAP reply consists of 7 frames
 
-    for (int i = 0; i < 6; i++) {
+    //  Initialize all reply frames
+    for (int i = 0; i < 7; i++) {
         rc = msg [i].init ();
         errno_assert (rc == 0);
     }
 
-    for (int i = 0; i < 6; i++) {
+    for (int i = 0; i < 7; i++) {
         rc = session->read_zap_msg (&msg [i]);
         if (rc == -1)
             break;
-        if ((msg [i].flags () & msg_t::more) == (i < 5? 0: msg_t::more)) {
+        if ((msg [i].flags () & msg_t::more) == (i < 6? 0: msg_t::more)) {
             errno = EPROTO;
             rc = -1;
             break;
@@ -579,7 +590,7 @@ int zmq::curve_server_t::receive_and_process_zap_reply ()
         goto error;
     }
 
-    //  Sequence number frame
+    //  Request id frame
     if (msg [2].size () != 1 || memcmp (msg [2].data (), "1", 1)) {
         errno = EPROTO;
         goto error;
@@ -591,8 +602,12 @@ int zmq::curve_server_t::receive_and_process_zap_reply ()
         goto error;
     }
 
+    //  Process metadata frame
+    rc = parse_metadata (static_cast <const unsigned char*> (msg [6].data ()),
+                         msg [6].size ());
+
 error:
-    for (int i = 0; i < 6; i++) {
+    for (int i = 0; i < 7; i++) {
         const int rc2 = msg [i].close ();
         errno_assert (rc2 == 0);
     }

src/curve_server.hpp

@@ -50,6 +50,7 @@ namespace zmq
     public:
 
         curve_server_t (session_base_t *session_,
+                        const std::string &peer_address_,
                         const options_t &options_);
         virtual ~curve_server_t ();
 
@@ -74,6 +75,8 @@ namespace zmq
 
         session_base_t * const session;
 
+        const std::string peer_address;
+
         //  Current FSM state
         state_t state;
 

src/ip.cpp

@@ -27,6 +27,7 @@
 #include <fcntl.h>
 #include <sys/types.h>
 #include <sys/socket.h>
+#include <netdb.h>
 #include <netinet/in.h>
 #include <netinet/tcp.h>
 #endif
@@ -108,3 +109,42 @@ void zmq::enable_ipv4_mapping (fd_t s_)
 #endif
 }
 
+bool zmq::get_peer_ip_address (fd_t sockfd_, std::string &ip_addr_)
+{
+    int rc;
+    struct sockaddr_storage ss;
+
+#if defined ZMQ_HAVE_HPUX || defined ZMQ_HAVE_WINDOWS
+    int addrlen = static_cast <int> (sizeof ss);
+#else
+    socklen_t addrlen = sizeof ss;
+#endif
+    rc = getpeername (sockfd_, (struct sockaddr*) &ss, &addrlen);
+#ifdef ZMQ_HAVE_WINDOWS
+    if (rc == SOCKET_ERROR) {
+        wsa_assert (WSAGetLastError () != WSANOTINITIALISED &&
+                    WSAGetLastError () != WSAEFAULT &&
+                    WSAGetLastError () != WSAEINPROGRESS &&
+                    WSAGetLastError () != WSAENOTSOCK)
+        return false;
+    }
+#else
+    if (rc == -1) {
+        errno_assert (errno != EBADF &&
+                      errno != EFAULT &&
+                      errno != EINVAL &&
+                      errno != ENOTCONN &&
+                      errno != ENOTSOCK);
+        return false;
+    }
+#endif
+
+    char host [NI_MAXHOST];
+    rc = getnameinfo ((struct sockaddr*) &ss, addrlen, host, sizeof host,
+        NULL, 0, NI_NUMERICHOST);
+    if (rc != 0)
+        return false;
+
+    ip_addr_ = host;
+    return true;
+}

src/ip.hpp

@@ -20,6 +20,7 @@
 #ifndef __ZMQ_IP_HPP_INCLUDED__
 #define __ZMQ_IP_HPP_INCLUDED__
 
+#include <string>
 #include "fd.hpp"
 
 namespace zmq
@@ -34,6 +35,10 @@ namespace zmq
     //  Enable IPv4-mapping of addresses in case it is disabled by default.
     void enable_ipv4_mapping (fd_t s_);
 
+    //  Returns string representation of peer's address.
+    //  Socket sockfd_ must be connected. Returns true iff successful.
+    bool get_peer_ip_address (fd_t sockfd_, std::string &ip_addr_);
+
 }
 
 #endif

src/null_mechanism.cpp

@@ -28,13 +28,25 @@
 
 #include "err.hpp"
 #include "msg.hpp"
+#include "session_base.hpp"
 #include "wire.hpp"
 #include "null_mechanism.hpp"
 
-zmq::null_mechanism_t::null_mechanism_t (const options_t &options_) : mechanism_t (options_),
+zmq::null_mechanism_t::null_mechanism_t (session_base_t *session_,
+                                         const std::string &peer_address_,
+                                         const options_t &options_) :
+    mechanism_t (options_),
+    session (session_),
+    peer_address (peer_address_),
     ready_command_sent (false),
-  ready_command_received (false)
+    ready_command_received (false),
+    zap_connected (false),
+    zap_request_sent (false),
+    zap_reply_received (false)
 {
+    const int rc = session->zap_connect ();
+    if (rc == 0)
+        zap_connected = true;
 }
 
 zmq::null_mechanism_t::~null_mechanism_t ()
@@ -47,6 +59,18 @@ int zmq::null_mechanism_t::next_handshake_message (msg_t *msg_)
         errno = EAGAIN;
         return -1;
     }
+    if (zap_connected && !zap_reply_received) {
+        if (zap_request_sent) {
+            errno = EAGAIN;
+            return -1;
+        }
+        send_zap_request ();
+        zap_request_sent = true;
+        const int rc = receive_and_process_zap_reply ();
+        if (rc != 0)
+            return -1;
+        zap_reply_received = true;
+    }
 
     unsigned char * const command_buffer = (unsigned char *) malloc (512);
     alloc_assert (command_buffer);
@@ -112,7 +136,132 @@ int zmq::null_mechanism_t::process_handshake_message (msg_t *msg_)
     return rc;
 }
 
+int zmq::null_mechanism_t::zap_msg_available ()
+{
+    if (zap_reply_received) {
+        errno = EFSM;
+        return -1;
+    }
+    const int rc = receive_and_process_zap_reply ();
+    if (rc == 0)
+        zap_reply_received = true;
+    return rc;
+}
+
 bool zmq::null_mechanism_t::is_handshake_complete () const
 {
     return ready_command_received && ready_command_sent;
 }
+
+void zmq::null_mechanism_t::send_zap_request ()
+{
+    int rc;
+    msg_t msg;
+
+    //  Address delimiter frame
+    rc = msg.init ();
+    errno_assert (rc == 0);
+    msg.set_flags (msg_t::more);
+    rc = session->write_zap_msg (&msg);
+    errno_assert (rc == 0);
+
+    //  Version frame
+    rc = msg.init_size (3);
+    errno_assert (rc == 0);
+    memcpy (msg.data (), "1.0", 3);
+    msg.set_flags (msg_t::more);
+    rc = session->write_zap_msg (&msg);
+    errno_assert (rc == 0);
+
+    //  Request id frame
+    rc = msg.init_size (1);
+    errno_assert (rc == 0);
+    memcpy (msg.data (), "1", 1);
+    msg.set_flags (msg_t::more);
+    rc = session->write_zap_msg (&msg);
+    errno_assert (rc == 0);
+
+    //  Domain frame
+    rc = msg.init ();
+    errno_assert (rc == 0);
+    msg.set_flags (msg_t::more);
+    rc = session->write_zap_msg (&msg);
+    errno_assert (rc == 0);
+
+    //  Address frame
+    rc = msg.init_size (peer_address.length ());
+    errno_assert (rc == 0);
+    memcpy (msg.data (), peer_address.c_str (), peer_address.length ());
+    msg.set_flags (msg_t::more);
+    rc = session->write_zap_msg (&msg);
+    errno_assert (rc == 0);
+
+    //  Mechanism frame
+    rc = msg.init_size (5);
+    errno_assert (rc == 0);
+    memcpy (msg.data (), "NULL", 5);
+    rc = session->write_zap_msg (&msg);
+    errno_assert (rc == 0);
+}
+
+int zmq::null_mechanism_t::receive_and_process_zap_reply ()
+{
+    int rc = 0;
+    msg_t msg [7];  //  ZAP reply consists of 7 frames
+
+    //  Initialize all reply frames
+    for (int i = 0; i < 7; i++) {
+        rc = msg [i].init ();
+        errno_assert (rc == 0);
+    }
+
+    for (int i = 0; i < 7; i++) {
+        rc = session->read_zap_msg (&msg [i]);
+        if (rc == -1)
+            break;
+        if ((msg [i].flags () & msg_t::more) == (i < 6? 0: msg_t::more)) {
+            errno = EPROTO;
+            rc = -1;
+            break;
+        }
+    }
+
+    if (rc != 0)
+        goto error;
+
+    //  Address delimiter frame
+    if (msg [0].size () > 0) {
+        errno = EPROTO;
+        goto error;
+    }
+
+    //  Version frame
+    if (msg [1].size () != 3 || memcmp (msg [1].data (), "1.0", 3)) {
+        errno = EPROTO;
+        goto error;
+    }
+
+    //  Request id frame
+    if (msg [2].size () != 1 || memcmp (msg [2].data (), "1", 1)) {
+        errno = EPROTO;
+        goto error;
+    }
+
+    //  Status code frame
+    if (msg [3].size () != 3 || memcmp (msg [3].data (), "200", 3)) {
+        errno = EACCES;
+        goto error;
+    }
+
+    //  Process metadata frame
+    rc = parse_metadata (static_cast <const unsigned char*> (msg [6].data ()),
+                         msg [6].size ());
+
+error:
+    for (int i = 0; i < 7; i++) {
+        const int rc2 = msg [i].close ();
+        errno_assert (rc2 == 0);
+    }
+
+    return rc;
+}

src/null_mechanism.hpp

@@ -27,23 +27,37 @@ namespace zmq
 {
 
     class msg_t;
+    class session_base_t;
 
     class null_mechanism_t : public mechanism_t
     {
     public:
 
-        null_mechanism_t (const options_t &options_);
+        null_mechanism_t (session_base_t *session_,
+                          const std::string &peer_address,
+                          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 int zap_msg_available ();
         virtual bool is_handshake_complete () const;
 
     private:
 
+        session_base_t * const session;
+
+        const std::string peer_address;
+
         bool ready_command_sent;
         bool ready_command_received;
+        bool zap_connected;
+        bool zap_request_sent;
+        bool zap_reply_received;
+
+        void send_zap_request ();
+        int receive_and_process_zap_reply ();
     };
 
 }

src/plain_mechanism.cpp

@@ -32,9 +32,11 @@
 #include "wire.hpp"
 
 zmq::plain_mechanism_t::plain_mechanism_t (session_base_t *session_,
+                                           const std::string &peer_address_,
                                            const options_t &options_) :
     mechanism_t (options_),
     session (session_),
+    peer_address (peer_address_),
     expecting_zap_reply (false),
     state (options.as_server? waiting_for_hello: sending_hello)
 {
@@ -355,7 +357,7 @@ void zmq::plain_mechanism_t::send_zap_request (const std::string &username,
     rc = session->write_zap_msg (&msg);
     errno_assert (rc == 0);
 
-    //  Sequence frame
+    //  Request id frame
     rc = msg.init_size (1);
     errno_assert (rc == 0);
     memcpy (msg.data (), "1", 1);
@@ -370,6 +372,14 @@ void zmq::plain_mechanism_t::send_zap_request (const std::string &username,
     rc = session->write_zap_msg (&msg);
     errno_assert (rc == 0);
 
+    //  Address frame
+    rc = msg.init_size (peer_address.length ());
+    errno_assert (rc == 0);
+    memcpy (msg.data (), peer_address.c_str (), peer_address.length ());
+    msg.set_flags (msg_t::more);
+    rc = session->write_zap_msg (&msg);
+    errno_assert (rc == 0);
+
     //  Mechanism frame
     rc = msg.init_size (5);
     errno_assert (rc == 0);
@@ -397,18 +407,19 @@ void zmq::plain_mechanism_t::send_zap_request (const std::string &username,
 int zmq::plain_mechanism_t::receive_and_process_zap_reply ()
 {
     int rc = 0;
-    msg_t msg [6];
+    msg_t msg [7];  //  ZAP reply consists of 7 frames
 
-    for (int i = 0; i < 6; i++) {
+    //  Initialize all reply frames
+    for (int i = 0; i < 7; i++) {
         rc = msg [i].init ();
         errno_assert (rc == 0);
     }
 
-    for (int i = 0; i < 6; i++) {
+    for (int i = 0; i < 7; i++) {
         rc = session->read_zap_msg (&msg [i]);
         if (rc == -1)
             break;
-        if ((msg [i].flags () & msg_t::more) == (i < 5? 0: msg_t::more)) {
+        if ((msg [i].flags () & msg_t::more) == (i < 6? 0: msg_t::more)) {
             errno = EPROTO;
             rc = -1;
             break;
@@ -418,8 +429,6 @@ int zmq::plain_mechanism_t::receive_and_process_zap_reply ()
     if (rc != 0)
         goto error;
 
-    return 0;
-
     //  Address delimiter frame
     if (msg [0].size () > 0) {
         errno = EPROTO;
@@ -432,7 +441,7 @@ int zmq::plain_mechanism_t::receive_and_process_zap_reply ()
         goto error;
     }
 
-    //  Sequence number frame
+    //  Request id frame
     if (msg [2].size () != 1 || memcmp (msg [2].data (), "1", 1)) {
         errno = EPROTO;
         goto error;
@@ -444,8 +453,12 @@ int zmq::plain_mechanism_t::receive_and_process_zap_reply ()
         goto error;
     }
 
+    //  Process metadata frame
+    rc = parse_metadata (static_cast <const unsigned char*> (msg [6].data ()),
+                         msg [6].size ());
+
 error:
-    for (int i = 0; i < 6; i++) {
+    for (int i = 0; i < 7; i++) {
         const int rc2 = msg [i].close ();
         errno_assert (rc2 == 0);
     }

src/plain_mechanism.hpp

@@ -34,6 +34,7 @@ namespace zmq
     public:
 
         plain_mechanism_t (session_base_t *session_,
+                           const std::string &peer_address_,
                            const options_t &options_);
         virtual ~plain_mechanism_t ();
 
@@ -60,6 +61,8 @@ namespace zmq
 
         session_base_t * const session;
 
+        const std::string peer_address;
+
         //  True iff we are awaiting reply from ZAP reply.
         bool expecting_zap_reply;
 

src/stream_engine.cpp

@@ -84,6 +84,9 @@ zmq::stream_engine_t::stream_engine_t (fd_t fd_, const options_t &options_,
     //  Put the socket into non-blocking mode.
     unblock_socket (s);
 
+    if (!get_peer_ip_address (s, peer_address))
+        peer_address = "";
+
 #ifdef SO_NOSIGPIPE
     //  Make sure that SIGPIPE signal is not generated when writing to a
     //  connection that was already closed by the peer.
@@ -522,19 +525,22 @@ 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) {
-            mechanism = new (std::nothrow) null_mechanism_t (options);
+            mechanism = new (std::nothrow)
+                null_mechanism_t (session, peer_address, options);
             alloc_assert (mechanism);
         }
         else
         if (memcmp (greeting_recv + 12, "PLAIN\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0", 20) == 0) {
-            mechanism = new (std::nothrow) plain_mechanism_t (session, options);
+            mechanism = new (std::nothrow)
+                plain_mechanism_t (session, peer_address, options);
             alloc_assert (mechanism);
         }
 #ifdef HAVE_LIBSODIUM
         else
         if (memcmp (greeting_recv + 12, "CURVE\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0", 20) == 0) {
             if (options.as_server)
-                mechanism = new (std::nothrow) curve_server_t (session, options);
+                mechanism = new (std::nothrow)
+                    curve_server_t (session, peer_address, options);
             else
                 mechanism = new (std::nothrow) curve_client_t (options);
             alloc_assert (mechanism);

src/stream_engine.hpp

@@ -187,6 +187,8 @@ namespace zmq
         // Socket
         zmq::socket_base_t *socket;
 
+        std::string peer_address;
+
         stream_engine_t (const stream_engine_t&);
         const stream_engine_t &operator = (const stream_engine_t&);
     };

tests/test_security.cpp

@@ -28,6 +28,7 @@ zap_handler (void *zap)
     char *version = s_recv (zap);
     char *sequence = s_recv (zap);
     char *domain = s_recv (zap);
+    char *address = s_recv (zap);
     char *mechanism = s_recv (zap);
     char *username = s_recv (zap);
     char *password = s_recv (zap);
@@ -41,17 +42,20 @@ zap_handler (void *zap)
     &&  streq (password, "password")) {
         s_sendmore (zap, "200");
         s_sendmore (zap, "OK");
-        s_send (zap, "anonymous");
+        s_sendmore (zap, "anonymous");
+        s_send (zap, "");
     }
     else {
         s_sendmore (zap, "400");
         s_sendmore (zap, "Invalid username or password");
+        s_sendmore (zap, "");
         s_send (zap, "");
     }
     
     free (version);
     free (sequence);
     free (domain);
+    free (address);
     free (mechanism);
     free (username);
     free (password);

tests/test_security_curve.cpp

@@ -29,6 +29,7 @@ zap_handler (void *zap)
     char *version = s_recv (zap);
     char *sequence = s_recv (zap);
     char *domain = s_recv (zap);
+    char *address = s_recv (zap);
     char *mechanism = s_recv (zap);
     char *client_key = s_recv (zap);
     
@@ -39,11 +40,13 @@ zap_handler (void *zap)
     s_sendmore (zap, sequence);
     s_sendmore (zap, "200");
     s_sendmore (zap, "OK");
-    s_send     (zap, "anonymous");
+    s_sendmore (zap, "anonymous");
+    s_send (zap, "");
     
     free (version);
     free (sequence);
     free (domain);
+    free (address);
     free (mechanism);
     free (client_key);