generic-library/libzmq
1
0
Fork 0
mirror of https://github.com/zeromq/libzmq.git synced 2024-12-13 02:42:58 +01:00
Code Issues Projects Releases Wiki Activity

Merge pull request #2482 from evoskuil/master

Browse Source 
Problem: insufficient error handling relative to zap_connect.
This commit is contained in:
Luca Boccassi 2017-03-30 10:48:38 +01:00 committed by GitHub
commit 7ce68da212
7 changed files with 185 additions and 219 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

73
src/curve_server.cpp
View File

@ -491,23 +491,21 @@ int zmq::curve_server_t::process_initiate (msg_t *msg_)
// Use ZAP protocol (RFC 27) to authenticate the user.
rc = session->zap_connect ();
if (rc == 0) {
rc = send_zap_request (client_key);
if (rc != 0)
return -1;
rc = receive_and_process_zap_reply ();
if (rc == 0)
state = status_code == "200"
? send_ready
: send_error;
else
if (errno == EAGAIN)
state = expect_zap_reply;
else
return -1;
}
if (rc != 0)
return -1;
rc = send_zap_request (client_key);
if (rc != 0)
return -1;
rc = receive_and_process_zap_reply ();
if (rc == 0)
state = status_code == "200"
? send_ready
: send_error;
else
state = send_ready;
if (errno == EAGAIN)
state = expect_zap_reply;
else
return -1;
return parse_metadata (initiate_plaintext + crypto_box_ZEROBYTES + 128,
clen - crypto_box_ZEROBYTES - 128);
@ -582,7 +580,7 @@ int zmq::curve_server_t::send_zap_request (const uint8_t *key)
msg.set_flags (msg_t::more);
rc = session->write_zap_msg (&msg);
if (rc != 0)
return -1;
return send_failure (&msg);
// Version frame
rc = msg.init_size (3);
@ -591,7 +589,7 @@ int zmq::curve_server_t::send_zap_request (const uint8_t *key)
msg.set_flags (msg_t::more);
rc = session->write_zap_msg (&msg);
if (rc != 0)
return -1;
return send_failure (&msg);
// Request ID frame
rc = msg.init_size (1);
@ -600,7 +598,7 @@ int zmq::curve_server_t::send_zap_request (const uint8_t *key)
msg.set_flags (msg_t::more);
rc = session->write_zap_msg (&msg);
if (rc != 0)
return -1;
return send_failure (&msg);
// Domain frame
rc = msg.init_size (options.zap_domain.length ());
@ -609,7 +607,7 @@ int zmq::curve_server_t::send_zap_request (const uint8_t *key)
msg.set_flags (msg_t::more);
rc = session->write_zap_msg (&msg);
if (rc != 0)
return -1;
return send_failure (&msg);
// Address frame
rc = msg.init_size (peer_address.length ());
@ -618,7 +616,7 @@ int zmq::curve_server_t::send_zap_request (const uint8_t *key)
msg.set_flags (msg_t::more);
rc = session->write_zap_msg (&msg);
if (rc != 0)
return -1;
return send_failure (&msg);
// Identity frame
rc = msg.init_size (options.identity_size);
@ -627,7 +625,7 @@ int zmq::curve_server_t::send_zap_request (const uint8_t *key)
msg.set_flags (msg_t::more);
rc = session->write_zap_msg (&msg);
if (rc != 0)
return -1;
return send_failure (&msg);
// Mechanism frame
rc = msg.init_size (5);
@ -636,7 +634,7 @@ int zmq::curve_server_t::send_zap_request (const uint8_t *key)
msg.set_flags (msg_t::more);
rc = session->write_zap_msg (&msg);
if (rc != 0)
return -1;
return send_failure (&msg);
// Credentials frame
rc = msg.init_size (crypto_box_PUBLICKEYBYTES);
@ -644,7 +642,7 @@ int zmq::curve_server_t::send_zap_request (const uint8_t *key)
memcpy (msg.data (), key, crypto_box_PUBLICKEYBYTES);
rc = session->write_zap_msg (&msg);
if (rc != 0)
return -1;
return send_failure (&msg);
return 0;
}
@ -663,26 +661,21 @@ int zmq::curve_server_t::receive_and_process_zap_reply ()
for (int i = 0; i < 7; i++) {
rc = session->read_zap_msg (&msg [i]);
if (rc == -1)
break;
return send_failure (msg);
if ((msg [i].flags () & msg_t::more) == (i < 6? 0: msg_t::more)) {
// Temporary support for security debugging
puts ("CURVE I: ZAP handler sent incomplete reply message");
errno = EPROTO;
rc = -1;
break;
return send_failure (msg);
}
}
if (rc != 0)
goto error;
// Address delimiter frame
if (msg [0].size () > 0) {
// Temporary support for security debugging
puts ("CURVE I: ZAP handler sent malformed reply message");
errno = EPROTO;
rc = -1;
goto error;
return send_failure (msg);
}
// Version frame
@ -690,8 +683,7 @@ int zmq::curve_server_t::receive_and_process_zap_reply ()
// Temporary support for security debugging
puts ("CURVE I: ZAP handler sent bad version number");
errno = EPROTO;
rc = -1;
goto error;
return send_failure (msg);
}
// Request id frame
@ -699,8 +691,7 @@ int zmq::curve_server_t::receive_and_process_zap_reply ()
// Temporary support for security debugging
puts ("CURVE I: ZAP handler sent bad request ID");
errno = EPROTO;
rc = -1;
goto error;
return send_failure (msg);
}
// Status code frame
@ -708,8 +699,7 @@ int zmq::curve_server_t::receive_and_process_zap_reply ()
// Temporary support for security debugging
puts ("CURVE I: ZAP handler rejected client authentication");
errno = EACCES;
rc = -1;
goto error;
return send_failure (msg);
}
// Save status code
@ -722,13 +712,16 @@ int zmq::curve_server_t::receive_and_process_zap_reply ()
rc = parse_metadata (static_cast <const unsigned char*> (msg [6].data ()),
msg [6].size (), true);
error:
if (rc != 0)
return send_failure (msg);
// Close all reply frames
for (int i = 0; i < 7; i++) {
const int rc2 = msg [i].close ();
errno_assert (rc2 == 0);
}
return rc;
return 0;
}
#endif

70
src/gssapi_server.cpp
View File

@ -120,20 +120,21 @@ int zmq::gssapi_server_t::process_handshake_command (msg_t *msg_)
if (security_context_established) {
// Use ZAP protocol (RFC 27) to authenticate the user.
bool expecting_zap_reply = false;
int rc = session->zap_connect ();
if (rc == 0) {
rc = send_zap_request ();
if (rc != 0)
return -1;
rc = receive_and_process_zap_reply ();
if (rc != 0) {
if (errno != EAGAIN)
return -1;
expecting_zap_reply = true;
}
}
state = expecting_zap_reply? expect_zap_reply: send_ready;
if (rc != 0)
return -1;
rc = send_zap_request ();
if (rc != 0)
return -1;
rc = receive_and_process_zap_reply ();
if (rc == 0)
state = send_ready;
else
if (errno == EAGAIN)
state = expect_zap_reply;
else
return -1;
return 0;
}
@ -160,7 +161,7 @@ int zmq::gssapi_server_t::send_zap_request ()
msg.set_flags (msg_t::more);
rc = session->write_zap_msg (&msg);
if (rc != 0)
return -1;
return send_failure (&msg);
// Version frame
rc = msg.init_size (3);
@ -169,7 +170,7 @@ int zmq::gssapi_server_t::send_zap_request ()
msg.set_flags (msg_t::more);
rc = session->write_zap_msg (&msg);
if (rc != 0)
return -1;
return send_failure (&msg);
// Request ID frame
rc = msg.init_size (1);
@ -178,7 +179,7 @@ int zmq::gssapi_server_t::send_zap_request ()
msg.set_flags (msg_t::more);
rc = session->write_zap_msg (&msg);
if (rc != 0)
return -1;
return send_failure (&msg);
// Domain frame
rc = msg.init_size (options.zap_domain.length ());
@ -187,7 +188,7 @@ int zmq::gssapi_server_t::send_zap_request ()
msg.set_flags (msg_t::more);
rc = session->write_zap_msg (&msg);
if (rc != 0)
return -1;
return send_failure (&msg);
// Address frame
rc = msg.init_size (peer_address.length ());
@ -196,7 +197,7 @@ int zmq::gssapi_server_t::send_zap_request ()
msg.set_flags (msg_t::more);
rc = session->write_zap_msg (&msg);
if (rc != 0)
return -1;
return send_failure (&msg);
// Identity frame
rc = msg.init_size (options.identity_size);
@ -205,7 +206,7 @@ int zmq::gssapi_server_t::send_zap_request ()
msg.set_flags (msg_t::more);
rc = session->write_zap_msg (&msg);
if (rc != 0)
return -1;
return send_failure (&msg);
// Mechanism frame
rc = msg.init_size (6);
@ -214,7 +215,7 @@ int zmq::gssapi_server_t::send_zap_request ()
msg.set_flags (msg_t::more);
rc = session->write_zap_msg (&msg);
if (rc != 0)
return -1;
return send_failure (&msg);
// Principal frame
gss_buffer_desc principal;
@ -226,7 +227,7 @@ int zmq::gssapi_server_t::send_zap_request ()
rc = session->write_zap_msg (&msg);
gss_release_buffer(&min_stat, &principal);
if (rc != 0)
return -1;
return send_failure (&msg);
return 0;
}
@ -245,43 +246,35 @@ int zmq::gssapi_server_t::receive_and_process_zap_reply ()
for (int i = 0; i < 7; i++) {
rc = session->read_zap_msg (&msg [i]);
if (rc == -1)
break;
return send_failure (msg);
if ((msg [i].flags () & msg_t::more) == (i < 6? 0: msg_t::more)) {
errno = EPROTO;
rc = -1;
break;
return send_failure (msg);
}
}
if (rc != 0)
goto error;
// Address delimiter frame
if (msg [0].size () > 0) {
rc = -1;
errno = EPROTO;
goto error;
return send_failure (msg);
}
// Version frame
if (msg [1].size () != 3 || memcmp (msg [1].data (), "1.0", 3)) {
rc = -1;
errno = EPROTO;
goto error;
return send_failure (msg);
}
// Request id frame
if (msg [2].size () != 1 || memcmp (msg [2].data (), "1", 1)) {
rc = -1;
errno = EPROTO;
goto error;
return send_failure (msg);
}
// Status code frame
if (msg [3].size () != 3 || memcmp (msg [3].data (), "200", 3)) {
rc = -1;
errno = EACCES;
goto error;
return send_failure (msg);
}
// Save user id
@ -291,13 +284,16 @@ int zmq::gssapi_server_t::receive_and_process_zap_reply ()
rc = parse_metadata (static_cast <const unsigned char*> (msg [6].data ()),
msg [6].size (), true);
error:
if (rc != 0)
return send_failure (msg);
// Close all reply frames
for (int i = 0; i < 7; i++) {
const int rc2 = msg [i].close ();
errno_assert (rc2 == 0);
}
return rc;
return 0;
}

15
src/msg.hpp
View File

@ -34,6 +34,7 @@
#include <stdio.h>
#include "config.hpp"
#include "err.hpp"
#include "fd.hpp"
#include "atomic_counter.hpp"
#include "metadata.hpp"
@ -246,6 +247,20 @@ namespace zmq
} u;
};
inline int send_failure (zmq::msg_t *msg)
{
const int rc = msg->close ();
errno_assert (rc == 0);
return -1;
}
inline int send_failure (zmq::msg_t msg[], int count)
{
for (int i = 0; i < count; i++)
send_failure (&msg [i]);
return -1;
}
}
#endif

41
src/null_mechanism.cpp
View File

@ -225,7 +225,7 @@ int zmq::null_mechanism_t::send_zap_request ()
msg.set_flags (msg_t::more);
rc = session->write_zap_msg (&msg);
if (rc != 0)
return -1;
return send_failure (&msg);
// Version frame
rc = msg.init_size (3);
@ -234,7 +234,7 @@ int zmq::null_mechanism_t::send_zap_request ()
msg.set_flags (msg_t::more);
rc = session->write_zap_msg (&msg);
if (rc != 0)
return -1;
return send_failure (&msg);
// Request id frame
rc = msg.init_size (1);
@ -243,7 +243,7 @@ int zmq::null_mechanism_t::send_zap_request ()
msg.set_flags (msg_t::more);
rc = session->write_zap_msg (&msg);
if (rc != 0)
return -1;
return send_failure (&msg);
// Domain frame
rc = msg.init_size (options.zap_domain.length ());
@ -252,7 +252,7 @@ int zmq::null_mechanism_t::send_zap_request ()
msg.set_flags (msg_t::more);
rc = session->write_zap_msg (&msg);
if (rc != 0)
return -1;
return send_failure (&msg);
// Address frame
rc = msg.init_size (peer_address.length ());
@ -261,7 +261,7 @@ int zmq::null_mechanism_t::send_zap_request ()
msg.set_flags (msg_t::more);
rc = session->write_zap_msg (&msg);
if (rc != 0)
return -1;
return send_failure (&msg);
// Identity frame
rc = msg.init_size (options.identity_size);
@ -270,7 +270,7 @@ int zmq::null_mechanism_t::send_zap_request ()
msg.set_flags (msg_t::more);
rc = session->write_zap_msg (&msg);
if (rc != 0)
return -1;
return send_failure (&msg);
// Mechanism frame
rc = msg.init_size (4);
@ -278,7 +278,7 @@ int zmq::null_mechanism_t::send_zap_request ()
memcpy (msg.data (), "NULL", 4);
rc = session->write_zap_msg (&msg);
if (rc != 0)
return -1;
return send_failure (&msg);
return 0;
}
@ -297,26 +297,21 @@ int zmq::null_mechanism_t::receive_and_process_zap_reply ()
for (int i = 0; i < 7; i++) {
rc = session->read_zap_msg (&msg [i]);
if (rc == -1)
break;
return send_failure (msg);
if ((msg [i].flags () & msg_t::more) == (i < 6? 0: msg_t::more)) {
// Temporary support for security debugging
puts ("NULL I: ZAP handler sent incomplete reply message");
errno = EPROTO;
rc = -1;
break;
return send_failure (msg);
}
}
if (rc != 0)
goto error;
// Address delimiter frame
if (msg [0].size () > 0) {
// Temporary support for security debugging
puts ("NULL I: ZAP handler sent malformed reply message");
errno = EPROTO;
rc = -1;
goto error;
return send_failure (msg);
}
// Version frame
@ -324,8 +319,7 @@ int zmq::null_mechanism_t::receive_and_process_zap_reply ()
// Temporary support for security debugging
puts ("NULL I: ZAP handler sent bad version number");
errno = EPROTO;
rc = -1;
goto error;
return send_failure (msg);
}
// Request id frame
@ -333,8 +327,7 @@ int zmq::null_mechanism_t::receive_and_process_zap_reply ()
// Temporary support for security debugging
puts ("NULL I: ZAP handler sent bad request ID");
errno = EPROTO;
rc = -1;
goto error;
return send_failure (msg);
}
// Status code frame
@ -342,8 +335,7 @@ int zmq::null_mechanism_t::receive_and_process_zap_reply ()
// Temporary support for security debugging
puts ("NULL I: ZAP handler rejected client authentication");
errno = EPROTO;
rc = -1;
goto error;
return send_failure (msg);
}
// Save status code
@ -356,11 +348,14 @@ int zmq::null_mechanism_t::receive_and_process_zap_reply ()
rc = parse_metadata (static_cast <const unsigned char*> (msg [6].data ()),
msg [6].size (), true);
error:
if (rc != 0)
return send_failure (msg);
// Close all reply frames
for (int i = 0; i < 7; i++) {
const int rc2 = msg [i].close ();
errno_assert (rc2 == 0);
}
return rc;
return 0;
}

75
src/plain_server.cpp
View File

@ -190,23 +190,21 @@ int zmq::plain_server_t::process_hello (msg_t *msg_)
// Use ZAP protocol (RFC 27) to authenticate the user.
int rc = session->zap_connect ();
if (rc == 0) {
rc = send_zap_request (username, password);
if (rc != 0)
return -1;
rc = receive_and_process_zap_reply ();
if (rc == 0)
state = status_code == "200"
? sending_welcome
: sending_error;
else
if (errno == EAGAIN)
state = waiting_for_zap_reply;
else
return -1;
}
if (rc != 0)
return -1;
rc = send_zap_request (username, password);
if (rc != 0)
return -1;
rc = receive_and_process_zap_reply ();
if (rc == 0)
state = status_code == "200"
? sending_welcome
: sending_error;
else
state = sending_welcome;
if (errno == EAGAIN)
state = waiting_for_zap_reply;
else
return -1;
return 0;
}
@ -291,7 +289,7 @@ int zmq::plain_server_t::send_zap_request (const std::string &username,
msg.set_flags (msg_t::more);
rc = session->write_zap_msg (&msg);
if (rc != 0)
return -1;
return send_failure (&msg);
// Version frame
rc = msg.init_size (3);
@ -300,7 +298,7 @@ int zmq::plain_server_t::send_zap_request (const std::string &username,
msg.set_flags (msg_t::more);
rc = session->write_zap_msg (&msg);
if (rc != 0)
return -1;
return send_failure (&msg);
// Request id frame
rc = msg.init_size (1);
@ -309,7 +307,7 @@ int zmq::plain_server_t::send_zap_request (const std::string &username,
msg.set_flags (msg_t::more);
rc = session->write_zap_msg (&msg);
if (rc != 0)
return -1;
return send_failure (&msg);
// Domain frame
rc = msg.init_size (options.zap_domain.length ());
@ -318,7 +316,7 @@ int zmq::plain_server_t::send_zap_request (const std::string &username,
msg.set_flags (msg_t::more);
rc = session->write_zap_msg (&msg);
if (rc != 0)
return -1;
return send_failure (&msg);
// Address frame
rc = msg.init_size (peer_address.length ());
@ -327,7 +325,7 @@ int zmq::plain_server_t::send_zap_request (const std::string &username,
msg.set_flags (msg_t::more);
rc = session->write_zap_msg (&msg);
if (rc != 0)
return -1;
return send_failure (&msg);
// Identity frame
rc = msg.init_size (options.identity_size);
@ -336,7 +334,7 @@ int zmq::plain_server_t::send_zap_request (const std::string &username,
msg.set_flags (msg_t::more);
rc = session->write_zap_msg (&msg);
if (rc != 0)
return -1;
return send_failure (&msg);
// Mechanism frame
rc = msg.init_size (5);
@ -345,7 +343,7 @@ int zmq::plain_server_t::send_zap_request (const std::string &username,
msg.set_flags (msg_t::more);
rc = session->write_zap_msg (&msg);
if (rc != 0)
return -1;
return send_failure (&msg);
// Username frame
rc = msg.init_size (username.length ());
@ -354,7 +352,7 @@ int zmq::plain_server_t::send_zap_request (const std::string &username,
msg.set_flags (msg_t::more);
rc = session->write_zap_msg (&msg);
if (rc != 0)
return -1;
return send_failure (&msg);
// Password frame
rc = msg.init_size (password.length ());
@ -362,7 +360,7 @@ int zmq::plain_server_t::send_zap_request (const std::string &username,
memcpy (msg.data (), password.c_str (), password.length ());
rc = session->write_zap_msg (&msg);
if (rc != 0)
return -1;
return send_failure (&msg);
return 0;
}
@ -381,26 +379,21 @@ int zmq::plain_server_t::receive_and_process_zap_reply ()
for (int i = 0; i < 7; i++) {
rc = session->read_zap_msg (&msg [i]);
if (rc == -1)
break;
return send_failure (msg);
if ((msg [i].flags () & msg_t::more) == (i < 6? 0: msg_t::more)) {
// Temporary support for security debugging
puts ("PLAIN I: ZAP handler sent incomplete reply message");
errno = EPROTO;
rc = -1;
break;
return send_failure (msg);
}
}
if (rc != 0)
goto error;
// Address delimiter frame
if (msg [0].size () > 0) {
// Temporary support for security debugging
puts ("PLAIN I: ZAP handler sent malformed reply message");
errno = EPROTO;
rc = -1;
goto error;
return send_failure (msg);
}
// Version frame
@ -408,17 +401,15 @@ int zmq::plain_server_t::receive_and_process_zap_reply ()
// Temporary support for security debugging
puts ("PLAIN I: ZAP handler sent bad version number");
errno = EPROTO;
rc = -1;
goto error;
return send_failure (msg);
}
// Request id frame
if (msg [2].size () != 1 || memcmp (msg [2].data (), "1", 1)) {
// Temporary support for security debugging
puts ("PLAIN I: ZAP handler sent bad request ID");
rc = -1;
errno = EPROTO;
goto error;
return send_failure (msg);
}
// Status code frame
@ -426,8 +417,7 @@ int zmq::plain_server_t::receive_and_process_zap_reply ()
// Temporary support for security debugging
puts ("PLAIN I: ZAP handler rejected client authentication");
errno = EACCES;
rc = -1;
goto error;
return send_failure (msg);
}
// Save status code
@ -440,11 +430,14 @@ int zmq::plain_server_t::receive_and_process_zap_reply ()
rc = parse_metadata (static_cast <const unsigned char*> (msg [6].data ()),
msg [6].size (), true);
error:
if (rc != 0)
return send_failure (msg);
// Close all reply frames
for (int i = 0; i < 7; i++) {
const int rc2 = msg [i].close ();
errno_assert (rc2 == 0);
}
return rc;
return 0;
}

7
src/session_base.cpp
View File

@ -353,9 +353,10 @@ int zmq::session_base_t::zap_connect ()
rc = id.init ();
errno_assert (rc == 0);
id.set_flags (msg_t::identity);
bool ok = zap_pipe->write (&id);
zmq_assert (ok);
zap_pipe->flush ();
if (zap_pipe->write (&id))
zap_pipe->flush ();
else
return -1;
}
return 0;

123
tests/test_heartbeats.cpp
View File

@ -23,8 +23,10 @@
# include <ws2tcpip.h>
# include <stdexcept>
# define close closesocket
typedef SOCKET raw_socket;
#else
# include <arpa/inet.h>
typedef int raw_socket;
#endif
// Read one event off the monitor socket; return value and address
@ -34,10 +36,11 @@
static int
get_monitor_event (void *monitor)
{
for(int i = 0; i < 2; i++) {
for (int i = 0; i < 2; i++) {
// First frame in message contains event number and value
zmq_msg_t msg;
zmq_msg_init (&msg);
int rc = zmq_msg_init (&msg);
assert (rc == 0);
if (zmq_msg_recv (&msg, monitor, ZMQ_DONTWAIT) == -1) {
msleep (SETTLE_TIME);
continue; // Interruped, presumably
@ -48,7 +51,8 @@ get_monitor_event (void *monitor)
uint16_t event = *(uint16_t *) (data);
// Second frame in message contains event address
zmq_msg_init (&msg);
rc = zmq_msg_init (&msg);
assert (rc == 0);
if (zmq_msg_recv (&msg, monitor, 0) == -1) {
return -1; // Interruped, presumably
}
@ -60,7 +64,7 @@ get_monitor_event (void *monitor)
}
static void
recv_with_retry (int fd, char *buffer, int bytes) {
recv_with_retry (raw_socket fd, char *buffer, int bytes) {
int received = 0;
while (true) {
int rc = recv(fd, buffer + received, bytes - received, 0);
@ -72,18 +76,18 @@ recv_with_retry (int fd, char *buffer, int bytes) {
}
static void
mock_handshake (int fd) {
mock_handshake (raw_socket fd) {
const uint8_t zmtp_greeting[33] = { 0xff, 0, 0, 0, 0, 0, 0, 0, 0, 0x7f, 3, 0, 'N', 'U', 'L', 'L', 0 };
char buffer[128];
memset(buffer, 0, sizeof(buffer));
memcpy(buffer, zmtp_greeting, sizeof(zmtp_greeting));
char buffer [128];
memset (buffer, 0, sizeof(buffer));
memcpy (buffer, zmtp_greeting, sizeof(zmtp_greeting));
int rc = send(fd, buffer, 64, 0);
assert(rc == 64);
int rc = send (fd, buffer, 64, 0);
assert (rc == 64);
recv_with_retry(fd, buffer, 64);
recv_with_retry (fd, buffer, 64);
const uint8_t zmtp_ready[43] = {
const uint8_t zmtp_ready [43] = {
4, 41, 5, 'R', 'E', 'A', 'D', 'Y', 11, 'S', 'o', 'c', 'k', 'e', 't', '-', 'T', 'y', 'p', 'e',
0, 0, 0, 6, 'D', 'E', 'A', 'L', 'E', 'R', 8, 'I', 'd', 'e', 'n', 't', 'i', 't', 'y',
0, 0, 0, 0
@ -92,38 +96,13 @@ mock_handshake (int fd) {
memset(buffer, 0, sizeof(buffer));
memcpy(buffer, zmtp_ready, 43);
rc = send(fd, buffer, 43, 0);
assert(rc == 43);
assert (rc == 43);
recv_with_retry(fd, buffer, 43);
}
static void
setup_curve(void * socket, int is_server) {
const char *secret_key;
const char *public_key;
const char *server_key;
if(is_server) {
secret_key = "JTKVSB%%)wK0E.X)V>+}o?pNmC{O&4W4b!Ni{Lh6";
public_key = "rq:rM>}U?@Lns47E1%kR.o@n%FcmmsL/@{H8]yf7";
server_key = NULL;
}
else {
secret_key = "D:)Q[IlAW!ahhC2ac:9*A}h:p?([4%wOTJ%JR%cs";
public_key = "Yne@$w-vo<fVvi]a<NY6T1ed:M$fCG*[IaLV{hID";
server_key = "rq:rM>}U?@Lns47E1%kR.o@n%FcmmsL/@{H8]yf7";
}
zmq_setsockopt(socket, ZMQ_CURVE_SECRETKEY, secret_key, strlen(secret_key));
zmq_setsockopt(socket, ZMQ_CURVE_PUBLICKEY, public_key, strlen(public_key));
if(is_server)
zmq_setsockopt(socket, ZMQ_CURVE_SERVER, &is_server, sizeof(is_server));
else
zmq_setsockopt(socket, ZMQ_CURVE_SERVERKEY, server_key, strlen(server_key));
}
static void
prep_server_socket(void * ctx, int set_heartbeats, int is_curve, void ** server_out, void ** mon_out)
prep_server_socket(void * ctx, int set_heartbeats, void ** server_out, void ** mon_out)
{
int rc;
// We'll be using this socket in raw mode
@ -134,15 +113,12 @@ prep_server_socket(void * ctx, int set_heartbeats, int is_curve, void ** server_
rc = zmq_setsockopt (server, ZMQ_LINGER, &value, sizeof (value));
assert (rc == 0);
if(set_heartbeats) {
if (set_heartbeats) {
value = 50;
rc = zmq_setsockopt (server, ZMQ_HEARTBEAT_IVL, &value, sizeof(value));
assert (rc == 0);
}
if(is_curve)
setup_curve(server, 1);
rc = zmq_bind (server, "tcp://127.0.0.1:5556");
assert (rc == 0);
@ -175,10 +151,10 @@ test_heartbeat_timeout (void)
assert (ctx);
void * server, * server_mon;
prep_server_socket(ctx, 1, 0, &server, &server_mon);
prep_server_socket (ctx, 1, &server, &server_mon);
struct sockaddr_in ip4addr;
int s;
raw_socket s;
ip4addr.sin_family = AF_INET;
ip4addr.sin_port = htons (5556);
@ -193,15 +169,15 @@ test_heartbeat_timeout (void)
assert (rc > -1);
// Mock a ZMTP 3 client so we can forcibly time out a connection
mock_handshake(s);
mock_handshake (s);
// By now everything should report as connected
rc = get_monitor_event(server_mon);
assert(rc == ZMQ_EVENT_ACCEPTED);
assert (rc == ZMQ_EVENT_ACCEPTED);
// We should have been disconnected
rc = get_monitor_event(server_mon);
assert(rc == ZMQ_EVENT_DISCONNECTED);
assert (rc == ZMQ_EVENT_DISCONNECTED);
close(s);
@ -230,32 +206,34 @@ test_heartbeat_ttl (void)
assert (ctx);
void * server, * server_mon, *client;
prep_server_socket(ctx, 0, 0, &server, &server_mon);
prep_server_socket (ctx, 0, &server, &server_mon);
client = zmq_socket(ctx, ZMQ_DEALER);
assert(client != NULL);
client = zmq_socket (ctx, ZMQ_DEALER);
assert (client != NULL);
// Set the heartbeat TTL to 0.1 seconds
value = 100;
zmq_setsockopt(client, ZMQ_HEARTBEAT_TTL, &value, sizeof(value));
rc = zmq_setsockopt (client, ZMQ_HEARTBEAT_TTL, &value, sizeof (value));
assert (rc == 0);
// Set the heartbeat interval to much longer than the TTL so that
// the socket times out oon the remote side.
value = 250;
zmq_setsockopt(client, ZMQ_HEARTBEAT_IVL, &value, sizeof(value));
rc = zmq_setsockopt (client, ZMQ_HEARTBEAT_IVL, &value, sizeof (value));
assert (rc == 0);
rc = zmq_connect(client, "tcp://localhost:5556");
assert(rc == 0);
rc = zmq_connect (client, "tcp://localhost:5556");
assert (rc == 0);
// By now everything should report as connected
rc = get_monitor_event(server_mon);
assert(rc == ZMQ_EVENT_ACCEPTED);
rc = get_monitor_event (server_mon);
assert (rc == ZMQ_EVENT_ACCEPTED);
msleep (SETTLE_TIME);
// We should have been disconnected
rc = get_monitor_event(server_mon);
assert(rc == ZMQ_EVENT_DISCONNECTED);
rc = get_monitor_event (server_mon);
assert (rc == ZMQ_EVENT_DISCONNECTED);
rc = zmq_close (server);
assert (rc == 0);
@ -274,7 +252,7 @@ test_heartbeat_ttl (void)
// exchanged normally. There should be an accepted event on the server,
// and then no event afterwards.
static void
test_heartbeat_notimeout (int is_curve)
test_heartbeat_notimeout (void)
{
int rc;
@ -283,23 +261,21 @@ test_heartbeat_notimeout (int is_curve)
assert (ctx);
void * server, * server_mon;
prep_server_socket(ctx, 1, is_curve, &server, &server_mon);
prep_server_socket(ctx, 1, &server, &server_mon);
void * client = zmq_socket(ctx, ZMQ_DEALER);
if(is_curve)
setup_curve(client, 0);
rc = zmq_connect(client, "tcp://127.0.0.1:5556");
void * client = zmq_socket (ctx, ZMQ_DEALER);
rc = zmq_connect (client, "tcp://127.0.0.1:5556");
// Give it a sec to connect and handshake
msleep (SETTLE_TIME);
// By now everything should report as connected
rc = get_monitor_event(server_mon);
assert(rc == ZMQ_EVENT_ACCEPTED);
assert (rc == ZMQ_EVENT_ACCEPTED);
// We should still be connected because pings and pongs are happenin'
rc = get_monitor_event(server_mon);
assert(rc == -1);
rc = get_monitor_event (server_mon);
assert (rc == -1);
rc = zmq_close (client);
assert (rc == 0);
@ -316,11 +292,8 @@ test_heartbeat_notimeout (int is_curve)
int main (void)
{
setup_test_environment();
test_heartbeat_timeout();
test_heartbeat_ttl();
// Run this test without curve
test_heartbeat_notimeout(0);
// Then rerun it with curve
test_heartbeat_notimeout(1);
setup_test_environment ();
test_heartbeat_timeout ();
test_heartbeat_ttl ();
test_heartbeat_notimeout ();
}