Commit e2f2193b authored by Luca Boccassi's avatar Luca Boccassi Committed by GitHub

Merge pull request #2752 from sigiesec/test-no-zap-handler

Problem: missing tests and inconsistent behaviour for ZAP corner cases
parents 9071265c 77f76a49
...@@ -390,18 +390,29 @@ int zmq::curve_server_t::process_initiate (msg_t *msg_) ...@@ -390,18 +390,29 @@ int zmq::curve_server_t::process_initiate (msg_t *msg_)
rc = crypto_box_beforenm (cn_precom, cn_client, cn_secret); rc = crypto_box_beforenm (cn_precom, cn_client, cn_secret);
zmq_assert (rc == 0); zmq_assert (rc == 0);
if (zap_required ()) {
// Use ZAP protocol (RFC 27) to authenticate the user. // Use ZAP protocol (RFC 27) to authenticate the user.
// Note that rc will be -1 only if ZAP is not set up (Stonehouse pattern -
// encryption without authentication), but if it was requested and it does
// not work properly the program will abort.
rc = session->zap_connect (); rc = session->zap_connect ();
if (rc == 0) { if (rc == 0) {
send_zap_request (client_key); send_zap_request (client_key);
state = waiting_for_zap_reply;
// TODO actually, it is quite unlikely that we can read the ZAP
// reply already, but removing this has some strange side-effect
// (probably because the pipe's in_active flag is true until a read
// is attempted)
rc = receive_and_process_zap_reply (); rc = receive_and_process_zap_reply ();
if (rc == -1) if (rc == -1)
return -1; return -1;
} else } else {
session->get_socket ()->event_handshake_failed_no_detail (
session->get_endpoint (), EFAULT);
return -1;
}
} else {
// This supports the Stonehouse pattern (encryption without authentication).
state = sending_ready; state = sending_ready;
}
return parse_metadata (initiate_plaintext + crypto_box_ZEROBYTES + 128, return parse_metadata (initiate_plaintext + crypto_box_ZEROBYTES + 128,
clen - crypto_box_ZEROBYTES - 128); clen - crypto_box_ZEROBYTES - 128);
......
...@@ -63,3 +63,8 @@ void zmq::mechanism_base_t::handle_error_reason (const char *error_reason, ...@@ -63,3 +63,8 @@ void zmq::mechanism_base_t::handle_error_reason (const char *error_reason,
session->get_endpoint (), (error_reason[0] - '0') * 100); session->get_endpoint (), (error_reason[0] - '0') * 100);
} }
} }
bool zmq::mechanism_base_t::zap_required() const
{
return !options.zap_domain.empty ();
}
...@@ -45,6 +45,8 @@ class mechanism_base_t : public mechanism_t ...@@ -45,6 +45,8 @@ class mechanism_base_t : public mechanism_t
int check_basic_command_structure (msg_t *msg_); int check_basic_command_structure (msg_t *msg_);
void handle_error_reason (const char *error_reason, size_t error_reason_len); void handle_error_reason (const char *error_reason, size_t error_reason_len);
bool zap_required() const;
}; };
} }
......
...@@ -48,15 +48,9 @@ zmq::null_mechanism_t::null_mechanism_t (session_base_t *session_, ...@@ -48,15 +48,9 @@ zmq::null_mechanism_t::null_mechanism_t (session_base_t *session_,
error_command_sent (false), error_command_sent (false),
ready_command_received (false), ready_command_received (false),
error_command_received (false), error_command_received (false),
zap_connected (false),
zap_request_sent (false), zap_request_sent (false),
zap_reply_received (false) zap_reply_received (false)
{ {
// NULL mechanism only uses ZAP if there's a domain defined
// This prevents ZAP requests on naive sockets
if (options.zap_domain.size () > 0
&& session->zap_connect () == 0)
zap_connected = true;
} }
zmq::null_mechanism_t::~null_mechanism_t () zmq::null_mechanism_t::~null_mechanism_t ()
...@@ -69,16 +63,31 @@ int zmq::null_mechanism_t::next_handshake_command (msg_t *msg_) ...@@ -69,16 +63,31 @@ int zmq::null_mechanism_t::next_handshake_command (msg_t *msg_)
errno = EAGAIN; errno = EAGAIN;
return -1; return -1;
} }
if (zap_connected && !zap_reply_received) {
if (zap_required() && !zap_reply_received) {
if (zap_request_sent) { if (zap_request_sent) {
errno = EAGAIN; errno = EAGAIN;
return -1; return -1;
} }
int rc = session->zap_connect();
if (rc == -1)
{
session->get_socket()->event_handshake_failed_no_detail (
session->get_endpoint(),
EFAULT);
return -1;
}
send_zap_request (); send_zap_request ();
zap_request_sent = true; zap_request_sent = true;
int rc = receive_and_process_zap_reply ();
if (rc == -1 || rc == 1) // TODO actually, it is quite unlikely that we can read the ZAP
// reply already, but removing this has some strange side-effect
// (probably because the pipe's in_active flag is true until a read
// is attempted)
rc = receive_and_process_zap_reply ();
if (rc != 0)
return -1; return -1;
zap_reply_received = true; zap_reply_received = true;
} }
......
...@@ -61,7 +61,6 @@ namespace zmq ...@@ -61,7 +61,6 @@ namespace zmq
bool error_command_sent; bool error_command_sent;
bool ready_command_received; bool ready_command_received;
bool error_command_received; bool error_command_received;
bool zap_connected;
bool zap_request_sent; bool zap_request_sent;
bool zap_reply_received; bool zap_reply_received;
......
...@@ -44,6 +44,10 @@ zmq::plain_server_t::plain_server_t (session_base_t *session_, ...@@ -44,6 +44,10 @@ zmq::plain_server_t::plain_server_t (session_base_t *session_,
zap_client_common_handshake_t ( zap_client_common_handshake_t (
session_, peer_address_, options_, sending_welcome) session_, peer_address_, options_, sending_welcome)
{ {
// Note that there is no point to PLAIN if ZAP is not set up to handle the
// username and password, so if ZAP is not configured it is considered a
// failure.
zmq_assert (zap_required());
} }
zmq::plain_server_t::~plain_server_t () zmq::plain_server_t::~plain_server_t ()
...@@ -173,13 +177,20 @@ int zmq::plain_server_t::process_hello (msg_t *msg_) ...@@ -173,13 +177,20 @@ int zmq::plain_server_t::process_hello (msg_t *msg_)
} }
// Use ZAP protocol (RFC 27) to authenticate the user. // Use ZAP protocol (RFC 27) to authenticate the user.
// Note that there is no point to PLAIN if ZAP is not set up to handle the
// username and password, so if ZAP is not configured it is considered a
// failure.
rc = session->zap_connect (); rc = session->zap_connect ();
if (rc != 0) if (rc != 0) {
session->get_socket ()->event_handshake_failed_no_detail (
session->get_endpoint (), EFAULT);
return -1; return -1;
}
send_zap_request (username, password); send_zap_request (username, password);
state = waiting_for_zap_reply;
// TODO actually, it is quite unlikely that we can read the ZAP
// reply already, but removing this has some strange side-effect
// (probably because the pipe's in_active flag is true until a read
// is attempted)
return receive_and_process_zap_reply () == -1 ? -1 : 0; return receive_and_process_zap_reply () == -1 ? -1 : 0;
} }
......
...@@ -286,8 +286,10 @@ void zmq::session_base_t::read_activated (pipe_t *pipe_) ...@@ -286,8 +286,10 @@ void zmq::session_base_t::read_activated (pipe_t *pipe_)
if (likely (pipe_ == pipe)) if (likely (pipe_ == pipe))
engine->restart_output (); engine->restart_output ();
else else {
// i.e. pipe_ == zap_pipe
engine->zap_msg_available (); engine->zap_msg_available ();
}
} }
void zmq::session_base_t::write_activated (pipe_t *pipe_) void zmq::session_base_t::write_activated (pipe_t *pipe_)
......
...@@ -299,10 +299,7 @@ void zap_client_common_handshake_t::handle_zap_status_code () ...@@ -299,10 +299,7 @@ void zap_client_common_handshake_t::handle_zap_status_code ()
int zap_client_common_handshake_t::receive_and_process_zap_reply () int zap_client_common_handshake_t::receive_and_process_zap_reply ()
{ {
int rc = zap_client_t::receive_and_process_zap_reply (); zmq_assert (state == waiting_for_zap_reply);
if (rc == 1) return zap_client_t::receive_and_process_zap_reply ();
// TODO shouldn't the state already be this?
state = waiting_for_zap_reply;
return rc;
} }
} }
...@@ -109,6 +109,9 @@ int main (void) ...@@ -109,6 +109,9 @@ int main (void)
void *server = zmq_socket (ctx, ZMQ_DEALER); void *server = zmq_socket (ctx, ZMQ_DEALER);
assert (server); assert (server);
int rc = zmq_setsockopt (server, ZMQ_IDENTITY, "IDENT", 6); int rc = zmq_setsockopt (server, ZMQ_IDENTITY, "IDENT", 6);
const char domain[] = "test";
assert (rc == 0);
rc = zmq_setsockopt (server, ZMQ_ZAP_DOMAIN, domain, strlen (domain));
assert (rc == 0); assert (rc == 0);
int as_server = 1; int as_server = 1;
rc = zmq_setsockopt (server, ZMQ_PLAIN_SERVER, &as_server, sizeof (int)); rc = zmq_setsockopt (server, ZMQ_PLAIN_SERVER, &as_server, sizeof (int));
...@@ -141,6 +144,8 @@ int main (void) ...@@ -141,6 +144,8 @@ int main (void)
client = zmq_socket (ctx, ZMQ_DEALER); client = zmq_socket (ctx, ZMQ_DEALER);
assert (client); assert (client);
as_server = 1; as_server = 1;
rc = zmq_setsockopt(client, ZMQ_ZAP_DOMAIN, domain, strlen (domain));
assert (rc == 0);
rc = zmq_setsockopt (client, ZMQ_PLAIN_SERVER, &as_server, sizeof (int)); rc = zmq_setsockopt (client, ZMQ_PLAIN_SERVER, &as_server, sizeof (int));
assert (rc == 0); assert (rc == 0);
rc = zmq_connect (client, my_endpoint); rc = zmq_connect (client, my_endpoint);
......
...@@ -59,15 +59,31 @@ static void zap_handler_too_many_parts (void *ctx) ...@@ -59,15 +59,31 @@ static void zap_handler_too_many_parts (void *ctx)
zap_handler_generic (ctx, zap_too_many_parts); zap_handler_generic (ctx, zap_too_many_parts);
} }
void test_zap_unsuccessful (void *ctx, static void zap_handler_disconnect (void *ctx)
{
zap_handler_generic (ctx, zap_disconnect);
}
static void zap_handler_do_not_recv (void *ctx)
{
zap_handler_generic (ctx, zap_do_not_recv);
}
static void zap_handler_do_not_send (void *ctx)
{
zap_handler_generic (ctx, zap_do_not_send);
}
int expect_new_client_bounce_fail_and_count_monitor_events (
void *ctx,
char *my_endpoint, char *my_endpoint,
void *server, void *server,
void *server_mon,
int expected_event,
int expected_err,
socket_config_fn socket_config_, socket_config_fn socket_config_,
void *socket_config_data_, void *socket_config_data_,
void **client_mon = NULL) void **client_mon,
void *server_mon,
int expected_event,
int expected_err)
{ {
expect_new_client_bounce_fail (ctx, my_endpoint, server, socket_config_, expect_new_client_bounce_fail (ctx, my_endpoint, server, socket_config_,
socket_config_data_, client_mon); socket_config_data_, client_mon);
...@@ -78,12 +94,54 @@ void test_zap_unsuccessful (void *ctx, ...@@ -78,12 +94,54 @@ void test_zap_unsuccessful (void *ctx,
expect_monitor_event_multiple (server_mon, expected_event, expected_err); expect_monitor_event_multiple (server_mon, expected_event, expected_err);
#endif #endif
return events_received;
}
void test_zap_unsuccessful (void *ctx,
char *my_endpoint,
void *server,
void *server_mon,
int expected_event,
int expected_err,
socket_config_fn socket_config_,
void *socket_config_data_,
void **client_mon = NULL)
{
int events_received =
expect_new_client_bounce_fail_and_count_monitor_events (
ctx, my_endpoint, server, socket_config_, socket_config_data_,
client_mon, server_mon, expected_event, expected_err);
// there may be more than one ZAP request due to repeated attempts by the // there may be more than one ZAP request due to repeated attempts by the
// client (actually only in case if ZAP status code 300) // client (actually only in case if ZAP status code 300)
assert (events_received == 0 assert (events_received == 0
|| 1 <= zmq_atomic_counter_value (zap_requests_handled)); || 1 <= zmq_atomic_counter_value (zap_requests_handled));
} }
void test_zap_unsuccessful_no_handler (void *ctx,
char *my_endpoint,
void *server,
void *server_mon,
int expected_event,
int expected_err,
socket_config_fn socket_config_,
void *socket_config_data_,
void **client_mon = NULL)
{
int events_received =
expect_new_client_bounce_fail_and_count_monitor_events (
ctx, my_endpoint, server, socket_config_, socket_config_data_,
client_mon, server_mon, expected_event, expected_err);
#ifdef ZMQ_BUILD_DRAFT_API
// there may be more than one ZAP request due to repeated attempts by the
// client
assert (events_received > 0);
#else
LIBZMQ_UNUSED (events_received);
#endif
}
void test_zap_protocol_error (void *ctx, void test_zap_protocol_error (void *ctx,
char *my_endpoint, char *my_endpoint,
void *server, void *server,
...@@ -266,6 +324,72 @@ void test_zap_errors (socket_config_fn server_socket_config_, ...@@ -266,6 +324,72 @@ void test_zap_errors (socket_config_fn server_socket_config_,
client_socket_config_data_); client_socket_config_data_);
shutdown_context_and_server_side (ctx, zap_thread, server, server_mon, shutdown_context_and_server_side (ctx, zap_thread, server, server_mon,
handler); handler);
// no ZAP handler
fprintf (stderr, "test_zap_unsuccessful no ZAP handler started\n");
setup_context_and_server_side (&ctx, &handler, &zap_thread, &server,
&server_mon, my_endpoint, NULL,
server_socket_config_);
test_zap_unsuccessful_no_handler (
ctx, my_endpoint, server, server_mon,
#ifdef ZMQ_BUILD_DRAFT_API
ZMQ_EVENT_HANDSHAKE_FAILED_NO_DETAIL, EFAULT,
#else
0, 0,
#endif
client_socket_config_, client_socket_config_data_);
shutdown_context_and_server_side (ctx, zap_thread, server, server_mon,
handler);
// ZAP handler disconnecting on first message
fprintf(stderr, "test_zap_unsuccessful ZAP handler disconnects\n");
setup_context_and_server_side(&ctx, &handler, &zap_thread, &server,
&server_mon, my_endpoint, &zap_handler_disconnect,
server_socket_config_);
test_zap_unsuccessful_no_handler (
ctx, my_endpoint, server, server_mon,
#ifdef ZMQ_BUILD_DRAFT_API
ZMQ_EVENT_HANDSHAKE_FAILED_NO_DETAIL, EPIPE,
#else
0, 0,
#endif
client_socket_config_, client_socket_config_data_);
shutdown_context_and_server_side (ctx, zap_thread, server, server_mon,
handler, true);
// ZAP handler does not read request
fprintf (stderr,
"test_zap_unsuccessful ZAP handler does not read request\n");
setup_context_and_server_side (&ctx, &handler, &zap_thread, &server,
&server_mon, my_endpoint, &zap_handler_do_not_recv,
server_socket_config_);
test_zap_unsuccessful_no_handler (
ctx, my_endpoint, server, server_mon,
#ifdef ZMQ_BUILD_DRAFT_API
ZMQ_EVENT_HANDSHAKE_FAILED_NO_DETAIL, EPIPE,
#else
0, 0,
#endif
client_socket_config_, client_socket_config_data_);
shutdown_context_and_server_side (ctx, zap_thread, server, server_mon,
handler);
// ZAP handler does not send reply
fprintf (stderr,
"test_zap_unsuccessful ZAP handler does not write reply\n");
setup_context_and_server_side (
&ctx, &handler, &zap_thread, &server, &server_mon, my_endpoint,
&zap_handler_do_not_send, server_socket_config_);
test_zap_unsuccessful_no_handler (
ctx, my_endpoint, server, server_mon,
#ifdef ZMQ_BUILD_DRAFT_API
ZMQ_EVENT_HANDSHAKE_FAILED_NO_DETAIL, EPIPE,
#else
0, 0,
#endif
client_socket_config_, client_socket_config_data_);
shutdown_context_and_server_side (ctx, zap_thread, server, server_mon,
handler);
} }
int main (void) int main (void)
......
...@@ -49,7 +49,8 @@ void socket_config_null_server (void *server, void *server_secret) ...@@ -49,7 +49,8 @@ void socket_config_null_server (void *server, void *server_secret)
{ {
LIBZMQ_UNUSED (server_secret); LIBZMQ_UNUSED (server_secret);
int rc = zmq_setsockopt (server, ZMQ_ZAP_DOMAIN, test_zap_domain, 7); int rc = zmq_setsockopt (server, ZMQ_ZAP_DOMAIN, test_zap_domain,
strlen (test_zap_domain));
assert (rc == 0); assert (rc == 0);
} }
...@@ -61,7 +62,8 @@ void socket_config_plain_client (void *server, void *server_secret) ...@@ -61,7 +62,8 @@ void socket_config_plain_client (void *server, void *server_secret)
{ {
LIBZMQ_UNUSED (server_secret); LIBZMQ_UNUSED (server_secret);
int rc = zmq_setsockopt (server, ZMQ_PLAIN_PASSWORD, test_plain_password, 8); int rc =
zmq_setsockopt (server, ZMQ_PLAIN_PASSWORD, test_plain_password, 8);
assert (rc == 0); assert (rc == 0);
rc = zmq_setsockopt (server, ZMQ_PLAIN_USERNAME, test_plain_username, 8); rc = zmq_setsockopt (server, ZMQ_PLAIN_USERNAME, test_plain_username, 8);
...@@ -73,8 +75,13 @@ void socket_config_plain_server (void *server, void *server_secret) ...@@ -73,8 +75,13 @@ void socket_config_plain_server (void *server, void *server_secret)
LIBZMQ_UNUSED (server_secret); LIBZMQ_UNUSED (server_secret);
int as_server = 1; int as_server = 1;
int rc = zmq_setsockopt (server, ZMQ_PLAIN_SERVER, &as_server, sizeof (int)); int rc =
zmq_setsockopt (server, ZMQ_PLAIN_SERVER, &as_server, sizeof (int));
assert (rc == 0); assert (rc == 0);
rc = zmq_setsockopt (server, ZMQ_ZAP_DOMAIN, test_zap_domain,
strlen (test_zap_domain));
assert(rc == 0);
} }
// CURVE specific functions // CURVE specific functions
...@@ -97,11 +104,16 @@ void setup_testutil_security_curve () ...@@ -97,11 +104,16 @@ void setup_testutil_security_curve ()
void socket_config_curve_server (void *server, void *server_secret) void socket_config_curve_server (void *server, void *server_secret)
{ {
int as_server = 1; int as_server = 1;
int rc = zmq_setsockopt (server, ZMQ_CURVE_SERVER, &as_server, sizeof (int)); int rc =
zmq_setsockopt (server, ZMQ_CURVE_SERVER, &as_server, sizeof (int));
assert (rc == 0); assert (rc == 0);
rc = zmq_setsockopt (server, ZMQ_CURVE_SECRETKEY, server_secret, 41); rc = zmq_setsockopt (server, ZMQ_CURVE_SECRETKEY, server_secret, 41);
assert (rc == 0); assert (rc == 0);
rc = zmq_setsockopt (server, ZMQ_ZAP_DOMAIN, test_zap_domain,
strlen (test_zap_domain));
assert(rc == 0);
} }
struct curve_client_data_t struct curve_client_data_t
...@@ -141,7 +153,10 @@ enum zap_protocol_t ...@@ -141,7 +153,10 @@ enum zap_protocol_t
zap_wrong_version, zap_wrong_version,
zap_wrong_request_id, zap_wrong_request_id,
zap_status_invalid, zap_status_invalid,
zap_too_many_parts zap_too_many_parts,
zap_disconnect,
zap_do_not_recv,
zap_do_not_send
}; };
void *zap_requests_handled; void *zap_requests_handled;
...@@ -165,11 +180,15 @@ void zap_handler_generic (void *ctx, ...@@ -165,11 +180,15 @@ void zap_handler_generic (void *ctx,
assert (rc == 2); assert (rc == 2);
zmq_pollitem_t items[] = { zmq_pollitem_t items[] = {
{control, 0, ZMQ_POLLIN, 0}, {handler, 0, ZMQ_POLLIN, 0}, {control, 0, ZMQ_POLLIN, 0},
{handler, 0, ZMQ_POLLIN, 0},
}; };
// if ordered not to receive the request, ignore the second poll item
const int numitems = (zap_protocol == zap_do_not_recv) ? 1 : 2;
// Process ZAP requests forever // Process ZAP requests forever
while (zmq_poll (items, 2, -1) >= 0) { while (zmq_poll (items, numitems, -1) >= 0) {
if (items[0].revents & ZMQ_POLLIN) { if (items[0].revents & ZMQ_POLLIN) {
char *buf = s_recv (control); char *buf = s_recv (control);
assert (buf); assert (buf);
...@@ -183,6 +202,11 @@ void zap_handler_generic (void *ctx, ...@@ -183,6 +202,11 @@ void zap_handler_generic (void *ctx,
char *version = s_recv (handler); char *version = s_recv (handler);
if (!version) if (!version)
break; // Terminating - peer's socket closed break; // Terminating - peer's socket closed
if (zap_protocol == zap_disconnect)
{
free (version);
break;
}
char *sequence = s_recv (handler); char *sequence = s_recv (handler);
char *domain = s_recv (handler); char *domain = s_recv (handler);
...@@ -200,15 +224,13 @@ void zap_handler_generic (void *ctx, ...@@ -200,15 +224,13 @@ void zap_handler_generic (void *ctx,
authentication_succeeded = authentication_succeeded =
streq (client_key_text, valid_client_public); streq (client_key_text, valid_client_public);
} } else if (streq (mechanism, "PLAIN")) {
else if (streq(mechanism, "PLAIN")) char client_username [32];
{
char client_username[32];
int size = zmq_recv (handler, client_username, 32, 0); int size = zmq_recv (handler, client_username, 32, 0);
assert (size > 0); assert (size > 0);
client_username [size] = 0; client_username [size] = 0;
char client_password[32]; char client_password [32];
size = zmq_recv (handler, client_password, 32, 0); size = zmq_recv (handler, client_password, 32, 0);
assert (size > 0); assert (size > 0);
client_password [size] = 0; client_password [size] = 0;
...@@ -216,13 +238,9 @@ void zap_handler_generic (void *ctx, ...@@ -216,13 +238,9 @@ void zap_handler_generic (void *ctx,
authentication_succeeded = authentication_succeeded =
streq (test_plain_username, client_username) streq (test_plain_username, client_username)
&& streq (test_plain_password, client_password); && streq (test_plain_password, client_password);
} } else if (streq (mechanism, "NULL")) {
else if (streq(mechanism, "NULL"))
{
authentication_succeeded = true; authentication_succeeded = true;
} } else {
else
{
fprintf (stderr, "Unsupported mechanism: %s\n", mechanism); fprintf (stderr, "Unsupported mechanism: %s\n", mechanism);
assert (false); assert (false);
} }
...@@ -258,12 +276,14 @@ void zap_handler_generic (void *ctx, ...@@ -258,12 +276,14 @@ void zap_handler_generic (void *ctx,
if (zap_protocol == zap_too_many_parts) { if (zap_protocol == zap_too_many_parts) {
s_sendmore (handler, ""); s_sendmore (handler, "");
} }
if (zap_protocol != zap_do_not_send)
s_send (handler, ""); s_send (handler, "");
} else { } else {
s_sendmore (handler, "400"); s_sendmore (handler, "400");
s_sendmore (handler, "Invalid client public key"); s_sendmore (handler, "Invalid client public key");
s_sendmore (handler, ""); s_sendmore (handler, "");
s_send (handler, ""); if (zap_protocol != zap_do_not_send)
s_send(handler, "");
} }
free (version); free (version);
free (sequence); free (sequence);
...@@ -278,8 +298,11 @@ void zap_handler_generic (void *ctx, ...@@ -278,8 +298,11 @@ void zap_handler_generic (void *ctx,
assert (rc == 0); assert (rc == 0);
close_zero_linger (handler); close_zero_linger (handler);
rc = s_send (control, "STOPPED"); if (zap_protocol != zap_disconnect)
assert (rc == 7); {
rc = s_send(control, "STOPPED");
assert(rc == 7);
}
close_zero_linger (control); close_zero_linger (control);
} }
...@@ -314,7 +337,7 @@ void setup_handshake_socket_monitor (void *ctx, ...@@ -314,7 +337,7 @@ void setup_handshake_socket_monitor (void *ctx,
void setup_context_and_server_side ( void setup_context_and_server_side (
void **ctx, void **ctx,
void **handler, void **zap_control,
void **zap_thread, void **zap_thread,
void **server, void **server,
void **server_mon, void **server_mon,
...@@ -331,17 +354,20 @@ void setup_context_and_server_side ( ...@@ -331,17 +354,20 @@ void setup_context_and_server_side (
zap_requests_handled = zmq_atomic_counter_new (); zap_requests_handled = zmq_atomic_counter_new ();
assert (zap_requests_handled != NULL); assert (zap_requests_handled != NULL);
*handler = zmq_socket (*ctx, ZMQ_REP); *zap_control = zmq_socket (*ctx, ZMQ_REP);
assert (*handler); assert (*zap_control);
int rc = zmq_bind (*handler, "inproc://handler-control"); int rc = zmq_bind (*zap_control, "inproc://handler-control");
assert (rc == 0); assert (rc == 0);
if (zap_handler_) {
*zap_thread = zmq_threadstart (zap_handler_, *ctx); *zap_thread = zmq_threadstart (zap_handler_, *ctx);
char *buf = s_recv (*handler); char *buf = s_recv (*zap_control);
assert (buf); assert (buf);
assert (streq (buf, "GO")); assert (streq (buf, "GO"));
free (buf); free (buf);
} else
*zap_thread = NULL;
// Server socket will accept connections // Server socket will accept connections
*server = zmq_socket (*ctx, ZMQ_DEALER); *server = zmq_socket (*ctx, ZMQ_DEALER);
...@@ -349,7 +375,7 @@ void setup_context_and_server_side ( ...@@ -349,7 +375,7 @@ void setup_context_and_server_side (
socket_config_ (*server, socket_config_data_); socket_config_ (*server, socket_config_data_);
rc = zmq_setsockopt (*server, ZMQ_IDENTITY, identity, strlen(identity)); rc = zmq_setsockopt (*server, ZMQ_IDENTITY, identity, strlen (identity));
assert (rc == 0); assert (rc == 0);
rc = zmq_bind (*server, "tcp://127.0.0.1:*"); rc = zmq_bind (*server, "tcp://127.0.0.1:*");
...@@ -368,17 +394,20 @@ void shutdown_context_and_server_side (void *ctx, ...@@ -368,17 +394,20 @@ void shutdown_context_and_server_side (void *ctx,
void *zap_thread, void *zap_thread,
void *server, void *server,
void *server_mon, void *server_mon,
void *handler) void *zap_control,
bool zap_handler_stopped = false)
{ {
int rc = s_send (handler, "STOP"); if (zap_thread && !zap_handler_stopped) {
int rc = s_send (zap_control, "STOP");
assert (rc == 4); assert (rc == 4);
char *buf = s_recv (handler); char *buf = s_recv (zap_control);
assert (buf); assert (buf);
assert (streq (buf, "STOPPED")); assert (streq (buf, "STOPPED"));
free (buf); free (buf);
rc = zmq_unbind (handler, "inproc://handler-control"); rc = zmq_unbind (zap_control, "inproc://handler-control");
assert (rc == 0); assert (rc == 0);
close_zero_linger (handler); }
close_zero_linger(zap_control);
#ifdef ZMQ_BUILD_DRAFT_API #ifdef ZMQ_BUILD_DRAFT_API
close_zero_linger (server_mon); close_zero_linger (server_mon);
...@@ -386,9 +415,10 @@ void shutdown_context_and_server_side (void *ctx, ...@@ -386,9 +415,10 @@ void shutdown_context_and_server_side (void *ctx,
close_zero_linger (server); close_zero_linger (server);
// Wait until ZAP handler terminates // Wait until ZAP handler terminates
if (zap_thread)
zmq_threadclose (zap_thread); zmq_threadclose (zap_thread);
rc = zmq_ctx_term (ctx); int rc = zmq_ctx_term (ctx);
assert (rc == 0); assert (rc == 0);
zmq_atomic_counter_destroy (&zap_requests_handled); zmq_atomic_counter_destroy (&zap_requests_handled);
...@@ -408,8 +438,7 @@ void *create_and_connect_client (void *ctx, ...@@ -408,8 +438,7 @@ void *create_and_connect_client (void *ctx,
int rc = zmq_connect (client, my_endpoint); int rc = zmq_connect (client, my_endpoint);
assert (rc == 0); assert (rc == 0);
if (client_mon) if (client_mon) {
{
setup_handshake_socket_monitor (ctx, client, client_mon, setup_handshake_socket_monitor (ctx, client, client_mon,
"inproc://client-monitor"); "inproc://client-monitor");
} }
...@@ -436,8 +465,10 @@ void expect_new_client_bounce_fail (void *ctx, ...@@ -436,8 +465,10 @@ void expect_new_client_bounce_fail (void *ctx,
// by reference, if not null, and event number by value. Returns -1 // by reference, if not null, and event number by value. Returns -1
// in case of error. // in case of error.
static int static int get_monitor_event_internal (void *monitor,
get_monitor_event_internal (void *monitor, int *value, char **address, int recv_flag) int *value,
char **address,
int recv_flag)
{ {
// First frame in message contains event number and value // First frame in message contains event number and value
zmq_msg_t msg; zmq_msg_t msg;
...@@ -507,8 +538,7 @@ int get_monitor_event (void *monitor, int *value, char **address) ...@@ -507,8 +538,7 @@ int get_monitor_event (void *monitor, int *value, char **address)
void expect_monitor_event (void *monitor, int expected_event) void expect_monitor_event (void *monitor, int expected_event)
{ {
int event = get_monitor_event (monitor, NULL, NULL); int event = get_monitor_event (monitor, NULL, NULL);
if (event != expected_event) if (event != expected_event) {
{
fprintf (stderr, "Expected monitor event %x but received %x\n", fprintf (stderr, "Expected monitor event %x but received %x\n",
expected_event, event); expected_event, event);
assert (event == expected_event); assert (event == expected_event);
...@@ -522,8 +552,7 @@ void print_unexpected_event (int event, ...@@ -522,8 +552,7 @@ void print_unexpected_event (int event,
int expected_event, int expected_event,
int expected_err) int expected_err)
{ {
fprintf( fprintf (stderr,
stderr,
"Unexpected event: 0x%x, value = %i/0x%x (expected: 0x%x, value " "Unexpected event: 0x%x, value = %i/0x%x (expected: 0x%x, value "
"= %i/0x%x)\n", "= %i/0x%x)\n",
event, err, err, expected_event, expected_err, expected_err); event, err, err, expected_event, expected_err, expected_err);
...@@ -550,14 +579,15 @@ int expect_monitor_event_multiple (void *server_mon, ...@@ -550,14 +579,15 @@ int expect_monitor_event_multiple (void *server_mon,
int err; int err;
while ( while (
(event = get_monitor_event_with_timeout (server_mon, &err, NULL, timeout)) (event = get_monitor_event_with_timeout (server_mon, &err, NULL, timeout))
!= -1 || !count_of_expected_events) { != -1
|| !count_of_expected_events) {
if (event == -1) { if (event == -1) {
if (optional) if (optional)
break; break;
wait_time += timeout; wait_time += timeout;
fprintf (stderr, fprintf (stderr,
"Still waiting for first event after %ims (expected event " "Still waiting for first event after %ims (expected event "
"%x (value %i/%x))\n", "%x (value %i/0x%x))\n",
wait_time, expected_event, expected_err, expected_err); wait_time, expected_event, expected_err, expected_err);
continue; continue;
} }
...@@ -565,12 +595,13 @@ int expect_monitor_event_multiple (void *server_mon, ...@@ -565,12 +595,13 @@ int expect_monitor_event_multiple (void *server_mon,
// ECONNRESET can happen on very slow machines, when the engine writes // ECONNRESET can happen on very slow machines, when the engine writes
// to the peer and then tries to read the socket before the peer reads // to the peer and then tries to read the socket before the peer reads
// ECONNABORTED happens when a client aborts a connection via RST/timeout // ECONNABORTED happens when a client aborts a connection via RST/timeout
if (event == ZMQ_EVENT_HANDSHAKE_FAILED_NO_DETAIL && if (event == ZMQ_EVENT_HANDSHAKE_FAILED_NO_DETAIL
(err == EPIPE || err == ECONNRESET || err == ECONNABORTED)) { && ((err == EPIPE && expected_err != EPIPE) || err == ECONNRESET
fprintf ( || err == ECONNABORTED)) {
stderr, fprintf (stderr,
"Ignored event (skipping any further events): %x (err = %i)\n", "Ignored event (skipping any further events): %x (err = "
event, err); "%i == %s)\n",
event, err, zmq_strerror (err));
client_closed_connection = 1; client_closed_connection = 1;
break; break;
} }
...@@ -581,7 +612,8 @@ int expect_monitor_event_multiple (void *server_mon, ...@@ -581,7 +612,8 @@ int expect_monitor_event_multiple (void *server_mon,
} }
++count_of_expected_events; ++count_of_expected_events;
} }
assert (optional || count_of_expected_events > 0 || client_closed_connection); assert (optional || count_of_expected_events > 0
|| client_closed_connection);
return count_of_expected_events; return count_of_expected_events;
} }
...@@ -600,8 +632,8 @@ int expect_monitor_event_multiple (void *server_mon, ...@@ -600,8 +632,8 @@ int expect_monitor_event_multiple (void *server_mon,
|| err == ECONNABORTED)) { \ || err == ECONNABORTED)) { \
fprintf (stderr, \ fprintf (stderr, \
"Ignored event (skipping any further events): %x " \ "Ignored event (skipping any further events): %x " \
"(err = %i)\n", \ "(err = %i == %s)\n", \
event, err); \ event, err, zmq_strerror (err)); \
continue; \ continue; \
} \ } \
++event_count; \ ++event_count; \
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment