commit dfdc4fb57b2633359d41a993d78bc3ae8f7b92fc
Author: Zack Weinberg <zackw(a)panix.com>
Date: Wed Jul 27 17:06:41 2011 -0700
Add a whole bunch more logging and test framework tracking. Properly distinguish BEV_EVENT_EOF from BEV_EVENT_ERROR in error_cb (flush_error_cb and pending_socks_cb are still sloppy).
---
src/main.c | 1 +
src/network.c | 91 +++++++++++++++++++------
src/network.h | 2 +
src/test/tester.py.in | 183 ++++++++++++++++++++++++++++--------------------
src/util.c | 41 +++++++++++
src/util.h | 4 +
6 files changed, 226 insertions(+), 96 deletions(-)
diff --git a/src/main.c b/src/main.c
index f6280c2..1459af3 100644
--- a/src/main.c
+++ b/src/main.c
@@ -91,6 +91,7 @@ handle_signal_cb(evutil_socket_t fd, short what, void *arg)
void
finish_shutdown(void)
{
+ log_debug("Finishing shutdown.");
event_base_loopexit(the_event_base, NULL);
}
diff --git a/src/network.c b/src/network.c
index f05c5b8..7831a4e 100644
--- a/src/network.c
+++ b/src/network.c
@@ -92,6 +92,8 @@ static void pending_socks_cb(struct bufferevent *bev, short what, void *arg);
void
start_shutdown(int barbaric)
{
+ log_debug("Beginning %s shutdown.", barbaric ? "barbaric" : "normal");
+
if (!shutting_down)
shutting_down=1;
@@ -115,6 +117,7 @@ close_all_connections(void)
{
if (!connections)
return;
+ log_debug("Closing all connections.");
SMARTLIST_FOREACH(connections, conn_t *, conn,
{ conn_free(conn); });
smartlist_free(connections);
@@ -143,6 +146,8 @@ create_listener(struct event_base *base, protocol_params_t *params)
default: obfs_abort();
}
+ lsn->address = printable_address(params->listen_addr->ai_addr,
+ params->listen_addr->ai_addrlen);
lsn->proto_params = params;
lsn->listener =
evconnlistener_new_bind(base, callback, lsn, flags, -1,
@@ -156,6 +161,9 @@ create_listener(struct event_base *base, protocol_params_t *params)
return 0;
}
+ log_debug("Now listening on %s in mode %d, protocol %s.",
+ lsn->address, params->mode, params->vtable->name);
+
/* If we don't have a listener list, create one now. */
if (!listeners)
listeners = smartlist_create();
@@ -170,6 +178,8 @@ create_listener(struct event_base *base, protocol_params_t *params)
static void
listener_free(listener_t *lsn)
{
+ if (lsn->address)
+ free(lsn->address);
if (lsn->listener)
evconnlistener_free(lsn->listener);
if (lsn->proto_params)
@@ -206,7 +216,9 @@ simple_client_listener_cb(struct evconnlistener *evcl,
struct event_base *base;
conn_t *conn = xzalloc(sizeof(conn_t));
- log_debug("%s: connection attempt.", __func__);
+ conn->peername = printable_address(sourceaddr, socklen);
+ log_debug("%s: connection to %s from %s", __func__,
+ lsn->address, conn->peername);
conn->mode = lsn->proto_params->mode;
obfs_assert(conn->mode == LSN_SIMPLE_CLIENT);
@@ -280,7 +292,9 @@ socks_client_listener_cb(struct evconnlistener *evcl,
struct event_base *base;
conn_t *conn = xzalloc(sizeof(conn_t));
- log_debug("%s: connection attempt.", __func__);
+ conn->peername = printable_address(sourceaddr, socklen);
+ log_debug("%s: connection to %s from %s", __func__,
+ lsn->address, conn->peername);
conn->mode = lsn->proto_params->mode;
obfs_assert(conn->mode == LSN_SOCKS_CLIENT);
@@ -336,7 +350,9 @@ simple_server_listener_cb(struct evconnlistener *evcl,
struct event_base *base;
conn_t *conn = xzalloc(sizeof(conn_t));
- log_debug("%s: connection attempt.", __func__);
+ conn->peername = printable_address(sourceaddr, socklen);
+ log_debug("%s: connection to %s from %s", __func__,
+ lsn->address, conn->peername);
conn->mode = lsn->proto_params->mode;
obfs_assert(conn->mode == LSN_SIMPLE_SERVER);
@@ -403,6 +419,8 @@ simple_server_listener_cb(struct evconnlistener *evcl,
static void
conn_free(conn_t *conn)
{
+ if (conn->peername)
+ free(conn->peername);
if (conn->proto)
proto_destroy(conn->proto);
if (conn->socks_state)
@@ -423,10 +441,11 @@ static void
close_conn(conn_t *conn)
{
obfs_assert(connections);
+ log_debug("Closing connection from %s; %d remaining",
+ conn->peername, smartlist_len(connections) - 1);
+
smartlist_remove(connections, conn);
conn_free(conn);
- log_debug("Connection destroyed. "
- "We currently have %d connections!", smartlist_len(connections));
/* If this was the last connection AND we are shutting down,
finish shutdown. */
@@ -447,6 +466,7 @@ static void
close_conn_on_flush(struct bufferevent *bev, void *arg)
{
conn_t *conn = arg;
+ log_debug("%s for %s", __func__, conn->peername);
if (evbuffer_get_length(bufferevent_get_output(bev)) == 0)
close_conn(conn);
@@ -460,6 +480,7 @@ socks_read_cb(struct bufferevent *bev, void *arg)
{
conn_t *conn = arg;
enum socks_ret socks_ret;
+ log_debug("%s for %s", __func__, conn->peername);
/* socks only makes sense on the upstream side */
obfs_assert(bev == conn->upstream);
@@ -534,9 +555,9 @@ static void
upstream_read_cb(struct bufferevent *bev, void *arg)
{
conn_t *conn = arg;
+ log_debug("%s for %s", __func__, conn->peername);
obfs_assert(bev == conn->upstream);
- log_debug("Got data on upstream side");
if (proto_send(conn->proto,
bufferevent_get_input(conn->upstream),
bufferevent_get_output(conn->downstream)) < 0)
@@ -553,9 +574,9 @@ downstream_read_cb(struct bufferevent *bev, void *arg)
{
conn_t *conn = arg;
enum recv_ret r;
+ log_debug("%s for %s", __func__, conn->peername);
obfs_assert(bev == conn->downstream);
- log_debug("Got data on downstream side");
r = proto_recv(conn->proto,
bufferevent_get_input(conn->downstream),
bufferevent_get_output(conn->upstream));
@@ -576,12 +597,12 @@ static void
error_or_eof(conn_t *conn, struct bufferevent *bev_err)
{
struct bufferevent *bev_flush;
+ log_debug("%s for %s", __func__, conn->peername);
if (bev_err == conn->upstream) bev_flush = conn->downstream;
else if (bev_err == conn->downstream) bev_flush = conn->upstream;
else obfs_abort();
- log_debug("error_or_eof");
if (conn->flushing || !conn->is_open ||
evbuffer_get_length(bufferevent_get_output(bev_flush)) == 0) {
close_conn(conn);
@@ -607,17 +628,35 @@ error_or_eof(conn_t *conn, struct bufferevent *bev_err)
static void
error_cb(struct bufferevent *bev, short what, void *arg)
{
+ conn_t *conn = arg;
+ int errcode = EVUTIL_SOCKET_ERROR();
+ log_debug("%s for %s: what=%x err=%d", __func__, conn->peername,
+ what, errcode);
+
/* It should be impossible to get here with BEV_EVENT_CONNECTED. */
obfs_assert(what & (BEV_EVENT_EOF|BEV_EVENT_ERROR|BEV_EVENT_TIMEOUT));
obfs_assert(!(what & BEV_EVENT_CONNECTED));
- /* If we get EAGAIN or EINPROGRESS here, something has gone horribly
- wrong. */
- obfs_assert(EVUTIL_SOCKET_ERROR() != EAGAIN &&
- EVUTIL_SOCKET_ERROR() != EINPROGRESS);
-
- log_warn("Got error: %s",
- evutil_socket_error_to_string(EVUTIL_SOCKET_ERROR()));
+ if (what & BEV_EVENT_ERROR) {
+ /* If we get EAGAIN, EINTR, or EINPROGRESS here, something has
+ gone horribly wrong. */
+ obfs_assert(errcode != EAGAIN && errcode != EINTR &&
+ errcode != EINPROGRESS);
+
+ log_warn("Error on %s side of connection from %s: %s",
+ bev == conn->upstream ? "upstream" : "downstream",
+ conn->peername,
+ evutil_socket_error_to_string(errcode));
+ } else if (what & BEV_EVENT_EOF) {
+ log_info("EOF on %s side of connection from %s",
+ bev == conn->upstream ? "upstream" : "downstream",
+ conn->peername);
+ } else {
+ obfs_assert(what & BEV_EVENT_TIMEOUT);
+ log_info("Timeout on %s side of connection from %s",
+ bev == conn->upstream ? "upstream" : "downstream",
+ conn->peername);
+ }
error_or_eof(arg, bev);
}
@@ -629,6 +668,9 @@ static void
flush_error_cb(struct bufferevent *bev, short what, void *arg)
{
conn_t *conn = arg;
+ int errcode = EVUTIL_SOCKET_ERROR();
+ log_debug("%s for %s: what=%x err=%d", __func__, conn->peername,
+ what, errcode);
/* It should be impossible to get here with BEV_EVENT_CONNECTED. */
obfs_assert(what & (BEV_EVENT_EOF|BEV_EVENT_ERROR|BEV_EVENT_TIMEOUT));
@@ -636,8 +678,10 @@ flush_error_cb(struct bufferevent *bev, short what, void *arg)
obfs_assert(conn->flushing);
- log_warn("Error during flush: %s",
- evutil_socket_error_to_string(EVUTIL_SOCKET_ERROR()));
+ log_warn("Error during flush of %s side of connection from %s: %s",
+ bev == conn->upstream ? "upstream" : "downstream",
+ conn->peername,
+ evutil_socket_error_to_string(errcode));
close_conn(conn);
return;
}
@@ -653,6 +697,8 @@ pending_conn_cb(struct bufferevent *bev, short what, void *arg)
{
conn_t *conn = arg;
struct bufferevent *other;
+ log_debug("%s for %s", __func__, conn->peername);
+
if (bev == conn->upstream) other = conn->downstream;
else if (bev == conn->downstream) other = conn->upstream;
else obfs_abort();
@@ -663,13 +709,15 @@ pending_conn_cb(struct bufferevent *bev, short what, void *arg)
obfs_assert(!conn->flushing);
conn->is_open = 1;
- log_debug("Connection successful");
- bufferevent_enable(other, EV_READ|EV_WRITE);
+ log_debug("Successful %s connection for %s",
+ bev == conn->upstream ? "upstream" : "downstream",
+ conn->peername);
/* XXX Dirty access to bufferevent guts. There appears to be no
official API to retrieve the callback functions and/or change
just one callback while leaving the others intact. */
bufferevent_setcb(bev, bev->readcb, bev->writecb, error_cb, conn);
+ bufferevent_enable(other, EV_READ|EV_WRITE);
return;
}
@@ -686,13 +734,16 @@ static void
pending_socks_cb(struct bufferevent *bev, short what, void *arg)
{
conn_t *conn = arg;
+ log_debug("%s for %s", __func__, conn->peername);
obfs_assert(bev == conn->downstream);
obfs_assert(conn->socks_state);
/* If we got an error while in the ST_HAVE_ADDR state, chances are
that we failed connecting to the host requested by the CONNECT
call. This means that we should send a negative SOCKS reply back
- to the client and terminate the connection. */
+ to the client and terminate the connection.
+ XXX properly distinguish BEV_EVENT_EOF from BEV_EVENT_ERROR;
+ errno isn't meaningful in that case... */
if ((what & (BEV_EVENT_EOF|BEV_EVENT_ERROR|BEV_EVENT_TIMEOUT))) {
int err = EVUTIL_SOCKET_ERROR();
log_warn("Connection error: %s",
diff --git a/src/network.h b/src/network.h
index c5ef69e..4ae96fc 100644
--- a/src/network.h
+++ b/src/network.h
@@ -14,11 +14,13 @@ void start_shutdown(int barbaric);
#ifdef NETWORK_PRIVATE
typedef struct listener_t {
+ char *address;
protocol_params_t *proto_params;
struct evconnlistener *listener;
} listener_t;
typedef struct conn_t {
+ char *peername;
protocol_t *proto;
socks_state_t *socks_state;
struct bufferevent *upstream;
diff --git a/src/test/tester.py.in b/src/test/tester.py.in
index 104026c..d9040fe 100644
--- a/src/test/tester.py.in
+++ b/src/test/tester.py.in
@@ -7,16 +7,86 @@
# You need to be able to make connections to arbitrary high-numbered
# TCP ports on the loopback interface.
+import difflib
import errno
import multiprocessing
import Queue
+import re
import signal
import socket
import struct
import subprocess
import time
+import traceback
import unittest
+# Helper: generate unified-format diffs between two named strings.
+# Pythonic escaped-string syntax is used for unprintable characters.
+
+def diff(label, expected, received):
+ if expected == received:
+ return ""
+ else:
+ return (label + "\n"
+ + "\n".join(s.encode("string_escape")
+ for s in
+ difflib.unified_diff(expected.split("\n"),
+ received.split("\n"),
+ "expected", "received",
+ lineterm=""))
+ + "\n")
+
+# Helper: Run obfsproxy instances and confirm that they have
+# completed without any errors.
+
+class Obfsproxy(subprocess.Popen):
+ def __init__(self, *args, **kwargs):
+ argv = ["./obfsproxy", "--log-min-severity=debug"]
+ if len(args) == 1 and (isinstance(args[0], list) or
+ isinstance(args[0], tuple)):
+ argv.extend(args[0])
+ else:
+ argv.extend(args)
+
+ subprocess.Popen.__init__(self, argv,
+ stdin=open("/dev/null", "r"),
+ stdout=subprocess.PIPE,
+ stderr=subprocess.PIPE,
+ **kwargs)
+
+ severe_error_re = re.compile(r"\[(?:warn|err(?:or)?)\]")
+
+ def check_completion(self, label):
+ if self.poll() is None:
+ self.send_signal(signal.SIGINT)
+
+ (out, err) = self.communicate()
+
+ report = ""
+ def indent(s):
+ return "| " + "\n| ".join(s.strip().split("\n"))
+
+ # exit status should be zero
+ if self.returncode > 0:
+ report += label + " exit code: %d\n" % self.returncode
+ elif self.returncode < 0:
+ report += label + " killed: signal %d\n" % -self.returncode
+
+ # there should be nothing on stdout
+ if out != "":
+ report += label + " stdout:\n%s\n" % indent(out)
+
+ # there will be debugging messages on stderr, but there should be
+ # no [warn], [err], or [error] messages.
+ if self.severe_error_re.search(err):
+ report += label + " stderr:\n%s\n" % indent(err)
+
+ return report
+
+ def stop(self):
+ if self.poll() is None:
+ self.terminate()
+
# Helper: Repeatedly try to connect to the specified server socket
# until either it succeeds or one full second has elapsed. (Surely
# there is a better way to do this?)
@@ -89,32 +159,14 @@ EXIT_PORT = 5001
class DirectTest(object):
def setUp(self):
self.output_reader = ReadWorker(("127.0.0.1", EXIT_PORT))
- self.obfs = subprocess.Popen(
- self.obfs_args,
- stdin=open("/dev/null", "r"),
- stdout=subprocess.PIPE,
- stderr=subprocess.PIPE)
-
+ self.obfs = Obfsproxy(self.obfs_args)
self.input_chan = connect_with_retry(("127.0.0.1", ENTRY_PORT))
self.input_chan.settimeout(1.0)
def tearDown(self):
- if self.obfs.returncode is None:
- self.obfs.terminate()
+ self.obfs.stop()
self.output_reader.stop()
-
- def checkSubprocesses(self):
- if self.obfs.poll() is None:
- self.obfs.send_signal(signal.SIGINT)
-
- (out, err) = self.obfs.communicate()
-
- if (out != "" or err != "" or self.obfs.returncode != 0):
- self.fail("obfsproxy process failure:\n"
- "\treturn code: %d\n"
- "\tstdout: %s\n"
- "\tstderr: %s\n"
- % (self.obfs.returncode, out, err))
+ self.input_chan.close()
def test_direct_transfer(self):
# Open a server and a simple client (in the same process) and
@@ -127,8 +179,11 @@ class DirectTest(object):
except Queue.Empty:
output = ""
- self.checkSubprocesses()
- self.assertEqual(output, TEST_FILE)
+ report = self.obfs.check_completion("obfsproxy")
+ report += diff("errors in transfer:", TEST_FILE, output)
+
+ if report != "":
+ self.fail("\n" + report)
# Same as above, but we use a socks client instead of a simple client,
# and the server's a separate process.
@@ -136,47 +191,14 @@ class DirectTest(object):
class SocksTest(object):
def setUp(self):
self.output_reader = ReadWorker(("127.0.0.1", EXIT_PORT))
- self.obfs_server = subprocess.Popen(
- self.server_args,
- stdin=open("/dev/null", "r"),
- stdout=subprocess.PIPE,
- stderr=subprocess.PIPE)
- self.obfs_client = subprocess.Popen(
- self.client_args,
- stdin=open("/dev/null", "r"),
- stdout=subprocess.PIPE,
- stderr=subprocess.PIPE)
+ self.obfs_server = Obfsproxy(self.server_args)
+ self.obfs_client = Obfsproxy(self.client_args)
def tearDown(self):
- if self.obfs_server.returncode is None:
- self.obfs_server.terminate()
- if self.obfs_client.returncode is None:
- self.obfs_client.terminate()
+ self.obfs_server.stop()
+ self.obfs_client.stop()
self.output_reader.stop()
- def checkSubprocesses(self):
- if self.obfs_server.poll() is None:
- self.obfs_server.send_signal(signal.SIGINT)
- if self.obfs_client.poll() is None:
- self.obfs_client.send_signal(signal.SIGINT)
-
- (sout, serr) = self.obfs_server.communicate()
- (cout, cerr) = self.obfs_client.communicate()
-
- if (sout != "" or serr != "" or cout != "" or cerr != ""
- or self.obfs_server.returncode != 0
- or self.obfs_client.returncode != 0):
- self.fail("obfsproxy process failures:\n"
- "\tclient return code: %d\n"
- "\tserver return code: %d\n"
- "\tclient stdout: %s\n"
- "\tclient stderr: %s\n"
- "\tserver stdout: %s\n"
- "\tserver stderr: %s\n"
- % (self.obfs_client.returncode,
- self.obfs_server.returncode,
- cout, cerr, sout, serr))
-
# 'sequence' is a sequence of SOCKS[45] protocol messages
# which we will send or receive. Sends alternate with
# receives. Each entry may be a string, which is sent or
@@ -187,7 +209,7 @@ class SocksTest(object):
# the SOCKS sequence without the server having dropped the
# connection, we transmit the test file and expect to get it
# back from the far end.
- def socksTest(self, sequence):
+ def socksTestInner(self, sequence):
sending = True
good = True
input_chan = connect_with_retry(("127.0.0.1", ENTRY_PORT))
@@ -230,10 +252,25 @@ class SocksTest(object):
output = ""
input_chan.close()
- self.checkSubprocesses()
+ if good: return output
+ else: return None
- if good:
- self.assertEqual(output, TEST_FILE)
+ def socksTest(self, sequence):
+ try:
+ output = self.socksTestInner(sequence)
+ report = ""
+ except Exception:
+ output = None
+ report = traceback.format_exc()
+
+ report += self.obfs_server.check_completion("obfsproxy server")
+ report += self.obfs_client.check_completion("obfsproxy client")
+
+ if output is not None:
+ report += diff("errors in transfer:", TEST_FILE, output)
+
+ if report != "":
+ self.fail("\n" + report)
def test_illformed(self):
@@ -295,8 +332,7 @@ class SocksTest(object):
#
class DirectObfs2(DirectTest, unittest.TestCase):
- obfs_args = ("./obfsproxy", "--log-min-severity=warn",
- "obfs2",
+ obfs_args = ("obfs2",
"--dest=127.0.0.1:%d" % EXIT_PORT,
"server", "127.0.0.1:%d" % SERVER_PORT,
"+", "obfs2",
@@ -304,8 +340,7 @@ class DirectObfs2(DirectTest, unittest.TestCase):
"client", "127.0.0.1:%d" % ENTRY_PORT)
class DirectDummy(DirectTest, unittest.TestCase):
- obfs_args = ("./obfsproxy", "--log-min-severity=warn",
- "dummy", "server",
+ obfs_args = ("dummy", "server",
"127.0.0.1:%d" % SERVER_PORT,
"127.0.0.1:%d" % EXIT_PORT,
"+", "dummy", "client",
@@ -313,21 +348,17 @@ class DirectDummy(DirectTest, unittest.TestCase):
"127.0.0.1:%d" % SERVER_PORT)
class SocksObfs2(SocksTest, unittest.TestCase):
- server_args = ("./obfsproxy", "--log-min-severity=warn",
- "obfs2",
+ server_args = ("obfs2",
"--dest=127.0.0.1:%d" % EXIT_PORT,
"server", "127.0.0.1:%d" % SERVER_PORT)
- client_args = ("./obfsproxy", "--log-min-severity=warn",
- "obfs2",
+ client_args = ("obfs2",
"socks", "127.0.0.1:%d" % ENTRY_PORT)
class SocksDummy(SocksTest, unittest.TestCase):
- server_args = ("./obfsproxy", "--log-min-severity=warn",
- "dummy", "server",
+ server_args = ("dummy", "server",
"127.0.0.1:%d" % SERVER_PORT,
"127.0.0.1:%d" % EXIT_PORT)
- client_args = ("./obfsproxy", "--log-min-severity=warn",
- "dummy", "socks",
+ client_args = ("dummy", "socks",
"127.0.0.1:%d" % ENTRY_PORT)
TEST_FILE = """\
diff --git a/src/util.c b/src/util.c
index e44b131..46d4b42 100644
--- a/src/util.c
+++ b/src/util.c
@@ -10,6 +10,10 @@
#include <unistd.h>
#include <event2/dns.h>
+#include <arpa/inet.h>
+#ifdef AF_LOCAL
+#include <sys/un.h>
+#endif
/** Any size_t larger than this amount is likely to be an underflow. */
#define SIZE_T_CEILING (SIZE_MAX/2 - 16)
@@ -195,6 +199,43 @@ resolve_address_port(const char *address, int nodns, int passive,
return ai;
}
+char *
+printable_address(struct sockaddr *addr, socklen_t addrlen)
+{
+ char abuf[INET6_ADDRSTRLEN];
+ char apbuf[INET6_ADDRSTRLEN + 8]; /* []:65535 is 8 characters */
+
+ switch (addr->sa_family) {
+ case AF_INET: {
+ struct sockaddr_in *sin = (struct sockaddr_in*)addr;
+ if (!inet_ntop(AF_INET, &sin->sin_addr, abuf, INET6_ADDRSTRLEN))
+ break;
+ obfs_snprintf(apbuf, sizeof apbuf, "%s:%d", abuf, ntohs(sin->sin_port));
+ return xstrdup(apbuf);
+ }
+
+ case AF_INET6: {
+ struct sockaddr_in6 *sin6 = (struct sockaddr_in6*)addr;
+ if (!inet_ntop(AF_INET, &sin6->sin6_addr, abuf, INET6_ADDRSTRLEN))
+ break;
+ obfs_snprintf(apbuf, sizeof apbuf, "[%s]:%d", abuf,
+ ntohs(sin6->sin6_port));
+ return xstrdup(apbuf);
+ }
+
+#ifdef AF_LOCAL
+ case AF_LOCAL:
+ return xstrdup(((struct sockaddr_un*)addr)->sun_path);
+#endif
+ default:
+ break;
+ }
+
+ obfs_snprintf(apbuf, sizeof apbuf,
+ "<addr family %d>", addr->sa_family);
+ return xstrdup(apbuf);
+}
+
static struct evdns_base *the_evdns_base = NULL;
struct evdns_base *
diff --git a/src/util.h b/src/util.h
index 4fc1a33..5108b62 100644
--- a/src/util.h
+++ b/src/util.h
@@ -93,6 +93,10 @@ struct evutil_addrinfo *resolve_address_port(const char *address,
int nodns, int passive,
const char *default_port);
+/** Produce a printable name for this sockaddr. The result is in
+ malloced memory. */
+char *printable_address(struct sockaddr *addr, socklen_t addrlen);
+
struct evdns_base *get_evdns_base(void);
int init_evdns_base(struct event_base *base);
