August 2012 - tor-commits - lists.torproject.org

[flashproxy/master] Add readline abstraction.
by dcf＠torproject.org 31 Aug '12

31 Aug '12

commit d3e4263f443da5760fe7ed533fa1dbd48e4bc54a Author: David Fifield <david(a)bamsoftware.com> Date: Sat Jul 7 07:31:31 2012 -0700 Add readline abstraction. --- facilitator | 42 +++++++++++++++++++++++++++++++++++++++--- 1 files changed, 39 insertions(+), 3 deletions(-) diff --git a/facilitator b/facilitator index 26c964d..16782cc 100755 --- a/facilitator +++ b/facilitator @@ -16,6 +16,8 @@ DEFAULT_LOG_FILENAME = "facilitator.log" # Don't indulge clients for more than this many seconds. CLIENT_TIMEOUT = 1.0 +# Buffer no many than this many bytes when trying to read a line. +READLINE_MAX_LENGTH = 10240 LOG_DATE_FORMAT = "%Y-%m-%d %H:%M:%S" @@ -115,6 +117,8 @@ def format_addr(addr): class Handler(SocketServer.StreamRequestHandler): def __init__(self, *args, **kwargs): self.deadline = time.time() + CLIENT_TIMEOUT + # Buffer for readline. + self.buffer = "" SocketServer.StreamRequestHandler.__init__(self, *args, **kwargs) def recv(self): @@ -123,13 +127,45 @@ class Handler(SocketServer.StreamRequestHandler): return self.connection.recv(1024) def readline(self): + # A line already buffered? + i = self.buffer.find("\n") + if i >= 0: + line = self.buffer[:i+1] + self.buffer = self.buffer[i+1:] + return line + + auxbuf = [] + buflen = len(self.buffer) while True: - self.connection.settimeout(timeout) + data = self.recv() + if not data: + if self.buffer or auxbuf: + raise socket.error("readline: stream does not end with a newline") + else: + return "" + i = data.find("\n") + if i >= 0: + line = self.buffer + "".join(auxbuf) + data[:i+1] + self.buffer = data[i+1:] + return line + else: + auxbuf.append(data) + buflen += len(data) + if buflen >= READLINE_MAX_LENGTH: + raise socket.error("readline: refusing to buffer %d bytes (last read was %d bytes)" % (buflen, len(data))) def handle(self): + num_lines = 0 while True: - data = self.recv() - print repr(data) + try: + line = self.readline() + if not line: + break + num_lines += 1 + except socket.error, e: + log("socket error after reading %d lines: %s" % (num_lines, str(e))) + break + print "line", repr(line) class Server(SocketServer.ThreadingMixIn, SocketServer.TCPServer): allow_reuse_address = True

1 0

[flashproxy/master] Add skeleton facilitator test.
by dcf＠torproject.org 31 Aug '12

31 Aug '12

commit b9e9ed718794e485b7bf3079d9e4d5c30e4d6c14 Author: David Fifield <david(a)bamsoftware.com> Date: Sat Jul 7 02:14:07 2012 -0700 Add skeleton facilitator test. --- Makefile | 1 + facilitator-test | 45 +++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 46 insertions(+), 0 deletions(-) diff --git a/Makefile b/Makefile index a12a9f3..3a15e54 100644 --- a/Makefile +++ b/Makefile @@ -17,6 +17,7 @@ clean: rm -rf dist test: + ./facilitator-test ./flashproxy-client-test ./flashproxy-test.js diff --git a/facilitator-test b/facilitator-test new file mode 100755 index 0000000..bdc064d --- /dev/null +++ b/facilitator-test @@ -0,0 +1,45 @@ +#!/usr/bin/env python + +import socket +import subprocess +import time +import unittest + +FACILITATOR_HOST = "127.0.0.1" +FACILITATOR_PORT = 9002 + +def gimme_socket(host, port): + addrinfo = socket.getaddrinfo(host, port, 0, socket.SOCK_STREAM, socket.IPPROTO_TCP)[0] + s = socket.socket(addrinfo[0], addrinfo[1], addrinfo[2]) + s.settimeout(10.0) + s.connect(addrinfo[4]) + return s + +class FacilitatorTest(unittest.TestCase): + def gimme_socket(self): + return gimme_socket(FACILITATOR_HOST, FACILITATOR_PORT) + + def setUp(self): + self.process = subprocess.Popen(["./facilitator", "-d", "-p", str(FACILITATOR_PORT), "-r", "0.0.0.0:0", "-l", "/dev/null"]) + time.sleep(0.1) + + def tearDown(self): + self.process.terminate() + + def test_timeout(self): + """Test that the socket will not accept slow writes indefinitely. + Successive sends should not reset the timeout counter.""" + s = self.gimme_socket() + time.sleep(0.3) + s.send("w") + time.sleep(0.3) + s.send("w") + time.sleep(0.3) + s.send("w") + time.sleep(0.3) + s.send("w") + time.sleep(0.3) + self.assertRaises(socket.error, s.send, "w") + +if __name__ == "__main__": + unittest.main()

1 0

[flashproxy/master] Parse input commands.
by dcf＠torproject.org 31 Aug '12

31 Aug '12

commit 12fe309124e0f2099716db600fa0e5d68fac30b3 Author: David Fifield <david(a)bamsoftware.com> Date: Fri Jul 13 04:07:31 2012 -0700 Parse input commands. --- facilitator | 67 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++- 1 files changed, 66 insertions(+), 1 deletions(-) diff --git a/facilitator b/facilitator index 16782cc..90fb1d8 100755 --- a/facilitator +++ b/facilitator @@ -114,6 +114,65 @@ def format_addr(addr): else: return u"%s:%d" % (host, port) +def skip_space(pos, line): + """Skip a (possibly empty) sequence of space characters (the ASCII character + '\x20' exactly). Returns a pair (pos, num_skipped).""" + begin = pos + while pos < len(line) and line[pos] == "\x20": + pos += 1 + return pos, pos - begin + +TOKEN_CHARS = set("abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789_-") +def get_token(pos, line): + begin = pos + while pos < len(line) and line[pos] in TOKEN_CHARS: + pos += 1 + if begin == pos: + raise ValueError("No token found at position %d" % pos) + return pos, line[begin:pos] + +def get_quoted_string(pos, line): + chars = [] + if not (pos < len(line) and line[pos] == '"'): + raise ValueError("Expected '\"' at beginning of quoted string.") + pos += 1 + while pos < len(line) and line[pos] != '"': + if line[pos] == '\\': + pos += 1 + if not (pos < len(line)): + raise ValueError("End of line after backslash in quoted string") + chars.append(line[pos]) + pos += 1 + if not (pos < len(line) and line[pos] == '"'): + raise ValueError("Expected '\"' at end of quoted string.") + pos += 1 + return pos, "".join(chars) + +def parse_command(line): + """A line is a command followed by zero or more key-value pairs. Like so: + COMMAND KEY="VALUE" KEY="\"ESCAPED\" VALUE" + Values must be quoted. Any byte value may be escaped with a backslash. + Returns a pair: (COMMAND, ((KEY1, VALUE1), (KEY2, VALUE2), ...)). + """ + pos = 0 + pos, skipped = skip_space(pos, line) + pos, command = get_token(pos, line) + + pairs = [] + while True: + pos, skipped = skip_space(pos, line) + if not (pos < len(line)): + break + if skipped == 0: + raise ValueError("Expected space before key-value pair") + pos, key = get_token(pos, line) + if not (pos < len(line) and line[pos] == '='): + raise ValueError("No '=' found after key") + pos += 1 + pos, value = get_quoted_string(pos, line) + pairs.append((key, value)) + return command, tuple(pairs) + class Handler(SocketServer.StreamRequestHandler): def __init__(self, *args, **kwargs): self.deadline = time.time() + CLIENT_TIMEOUT @@ -165,7 +224,13 @@ class Handler(SocketServer.StreamRequestHandler): except socket.error, e: log("socket error after reading %d lines: %s" % (num_lines, str(e))) break - print "line", repr(line) + self.handle_line(line) + + def handle_line(self, line): + if not (len(line) > 0 and line[-1] == '\n'): + raise ValueError("No newline at end of string returned by readline") + command, pairs = parse_command(line[:-1]) + print command, pairs class Server(SocketServer.ThreadingMixIn, SocketServer.TCPServer): allow_reuse_address = True

1 0

[flashproxy/master] Add a test for readline limit.
by dcf＠torproject.org 31 Aug '12

31 Aug '12

commit ed2506d10deaf8979e56ac57be3b8d3dc25e5b7e Author: David Fifield <david(a)bamsoftware.com> Date: Sat Jul 7 07:48:39 2012 -0700 Add a test for readline limit. --- facilitator-test | 12 ++++++++++++ 1 files changed, 12 insertions(+), 0 deletions(-) diff --git a/facilitator-test b/facilitator-test index bdc064d..4e2ec56 100755 --- a/facilitator-test +++ b/facilitator-test @@ -41,5 +41,17 @@ class FacilitatorTest(unittest.TestCase): time.sleep(0.3) self.assertRaises(socket.error, s.send, "w") + def test_readline_limit(self): + """Test that reads won't buffer indefinitely.""" + s = self.gimme_socket() + buflen = 0 + try: + while buflen + 1024 < 200000: + s.send("X" * 1024) + buflen += 1024 + self.fail("should have raised a socket error") + except socket.error: + pass + if __name__ == "__main__": unittest.main()

1 0

[flashproxy/master] Handle PUT command.
by dcf＠torproject.org 31 Aug '12

31 Aug '12

commit bd3c3708a404ccadb58ce8f70fdfc444799ec876 Author: David Fifield <david(a)bamsoftware.com> Date: Fri Jul 13 08:07:04 2012 -0700 Handle PUT command. --- facilitator | 132 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++- 1 files changed, 129 insertions(+), 3 deletions(-) diff --git a/facilitator b/facilitator index 90fb1d8..dc6b752 100755 --- a/facilitator +++ b/facilitator @@ -55,6 +55,13 @@ again. Listen on 127.0.0.1 and port PORT (by default %(port)d). "log": DEFAULT_LOG_FILENAME, } +def safe_str(s): + """Return s if options.safe_logging is true, and "[scrubbed]" otherwise.""" + if options.safe_logging: + return "[scrubbed]" + else: + return s + log_lock = threading.Lock() def log(msg): log_lock.acquire() @@ -173,6 +180,82 @@ def parse_command(line): pairs.append((key, value)) return command, tuple(pairs) +def param_first(key, params): + for k, v in params: + if key == k: + return v + return None + +class TCPReg(object): + def __init__(self, host, port): + self.host = host + self.port = port + + def __unicode__(self): + return format_addr((self.host, self.port)) + + def __str__(self): + return unicode(self).encode("UTF-8") + + def __cmp__(self, other): + if isinstance(other, TCPReg): + return cmp((self.host, self.port), (other.host, other.port)) + else: + return False + +class Reg(object): + @staticmethod + def parse(spec, defhost = None, defport = None): + try: + af, host, port = parse_addr_spec(spec, defhost, defport) + except ValueError: + pass + else: + try: + addrs = socket.getaddrinfo(host, port, af, socket.SOCK_STREAM, socket.IPPROTO_TCP, socket.AI_NUMERICHOST) + except socket.gaierror, e: + raise ValueError("Bad host or port: \"%s\" \"%s\": %s" % (host, port, str(e))) + if not addrs: + raise ValueError("Bad host or port: \"%s\" \"%s\"" % (host, port)) + + host, port = socket.getnameinfo(addrs[0][4], socket.NI_NUMERICHOST | socket.NI_NUMERICSERV) + return TCPReg(host, int(port)) + + raise ValueError("Bad spec format: %s" % repr(spec)) + +class RegSet(object): + def __init__(self): + self.set = [] + self.cv = threading.Condition() + + def add(self, reg): + self.cv.acquire() + try: + if reg not in list(self.set): + self.set.append(reg) + self.cv.notify() + return True + else: + return False + finally: + self.cv.release() + + def fetch(self): + self.cv.acquire() + try: + if not self.set: + return None + return self.set.pop(0) + finally: + self.cv.release() + + def __len__(self): + self.cv.acquire() + try: + return len(self.set) + finally: + self.cv.release() + class Handler(SocketServer.StreamRequestHandler): def __init__(self, *args, **kwargs): self.deadline = time.time() + CLIENT_TIMEOUT @@ -224,17 +307,60 @@ class Handler(SocketServer.StreamRequestHandler): except socket.error, e: log("socket error after reading %d lines: %s" % (num_lines, str(e))) break - self.handle_line(line) + if not self.handle_line(line): + break def handle_line(self, line): if not (len(line) > 0 and line[-1] == '\n'): raise ValueError("No newline at end of string returned by readline") - command, pairs = parse_command(line[:-1]) - print command, pairs + try: + command, params = parse_command(line[:-1]) + except ValueError, e: + log("parse_command: %s" % e) + self.send_error() + return False + + if command == "PUT": + return self.do_PUT(params) + else: + self.send_error() + return False + + def send_ok(self): + print >> self.wfile, "OK" + + def send_error(self): + print >> self.wfile, "ERROR" + + def do_PUT(self, params): + client_spec = param_first("CLIENT", params) + if client_spec is None: + log(u"PUT missing CLIENT param") + self.send_error() + return False + + # FROM + + try: + reg = Reg.parse(client_spec, self.client_address[0]) + except ValueError, e: + log(u"syntax error in %s: %s" % (safe_str(repr(client_spec)), repr(str(e)))) + self.send_error() + return False + + if REGS.add(reg): + log(u"client %s (now %d)" % (safe_str(unicode(reg)), len(REGS))) + else: + log(u"client %s (already present, now %d)" % (safe_str(unicode(reg)), len(REGS))) + + self.send_ok() + return True class Server(SocketServer.ThreadingMixIn, SocketServer.TCPServer): allow_reuse_address = True +REGS = RegSet() + def main(): opts, args = getopt.gnu_getopt(sys.argv[1:], "dhl:p:r:", ["debug", "help", "log=", "port=", "pidfile=", "relay=", "unsafe-logging"])

1 0

[flashproxy/master] Add GET command.
by dcf＠torproject.org 31 Aug '12

31 Aug '12

commit d5d45322d86dd80054ff66e37964de28f1eaf27c Author: David Fifield <david(a)bamsoftware.com> Date: Fri Jul 13 08:24:57 2012 -0700 Add GET command. --- facilitator | 29 +++++++++++++++++++++++++++++ 1 files changed, 29 insertions(+), 0 deletions(-) diff --git a/facilitator b/facilitator index dc6b752..2ab6491 100755 --- a/facilitator +++ b/facilitator @@ -186,6 +186,22 @@ def param_first(key, params): return v return None +def quote_string(s): + chars = [] + for c in s: + if c == "\\": + c = "\\\\" + elif c == "\"": + c = "\\\"" + chars.append(c) + return "\"" + "".join(chars) + "\"" + +def render_params(params): + parts = [] + for key, value in params: + parts.append("%s=%s" % (key, quote_string(value))) + return " ".join(parts) + class TCPReg(object): def __init__(self, host, port): self.host = host @@ -320,6 +336,8 @@ class Handler(SocketServer.StreamRequestHandler): self.send_error() return False + if command == "GET": + return self.do_GET(params) if command == "PUT": return self.do_PUT(params) else: @@ -332,6 +350,17 @@ class Handler(SocketServer.StreamRequestHandler): def send_error(self): print >> self.wfile, "ERROR" + def do_GET(self, params): + reg = REGS.fetch() + if reg: + log(u"proxy gets %s, relay %s (now %d)" % + (safe_str(unicode(reg)), options.relay_spec, len(REGS))) + print >> self.wfile, "OK", render_params((("CLIENT", str(reg)), ("RELAY", options.relay_spec))) + else: + log(u"proxy gets none") + print >> self.wfile, "NONE" + return True + def do_PUT(self, params): client_spec = param_first("CLIENT", params) if client_spec is None:

1 0

[flashproxy/master] Move some common code to fac.py.
by dcf＠torproject.org 31 Aug '12

31 Aug '12

commit af7de115115516435388503fe21c22a03d3b7445 Author: David Fifield <david(a)bamsoftware.com> Date: Sat Aug 11 08:53:49 2012 -0700 Move some common code to fac.py. --- fac.py | 133 +++++++++++++++++++++++++++++++++++++++++++++++++++ facilitator | 154 +++++------------------------------------------------------ 2 files changed, 145 insertions(+), 142 deletions(-) diff --git a/fac.py b/fac.py new file mode 100644 index 0000000..ab8157c --- /dev/null +++ b/fac.py @@ -0,0 +1,133 @@ +import re +import socket + +def parse_addr_spec(spec, defhost = None, defport = None): + host = None + port = None + m = None + # IPv6 syntax. + if not m: + m = re.match(ur'^\[(.+)\]:(\d+)$', spec) + if m: + host, port = m.groups() + af = socket.AF_INET6 + if not m: + m = re.match(ur'^\[(.+)\]:?$', spec) + if m: + host, = m.groups() + af = socket.AF_INET6 + # IPv4 syntax. + if not m: + m = re.match(ur'^(.+):(\d+)$', spec) + if m: + host, port = m.groups() + af = socket.AF_INET + if not m: + m = re.match(ur'^:?(\d+)$', spec) + if m: + port, = m.groups() + af = 0 + if not m: + host = spec + af = 0 + host = host or defhost + port = port or defport + if not (host and port): + raise ValueError("Bad address specification \"%s\"" % spec) + return af, host, int(port) + +def format_addr(addr): + host, port = addr + if not host: + return u":%d" % port + # Numeric IPv6 address? + try: + addrs = socket.getaddrinfo(host, port, 0, socket.SOCK_STREAM, socket.IPPROTO_TCP, socket.AI_NUMERICHOST) + af = addrs[0][0] + except socket.gaierror, e: + af = 0 + if af == socket.AF_INET6: + return u"[%s]:%d" % (host, port) + else: + return u"%s:%d" % (host, port) + +def skip_space(pos, line): + """Skip a (possibly empty) sequence of space characters (the ASCII character + '\x20' exactly). Returns a pair (pos, num_skipped).""" + begin = pos + while pos < len(line) and line[pos] == "\x20": + pos += 1 + return pos, pos - begin + +TOKEN_CHARS = set("abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789_-") +def get_token(pos, line): + begin = pos + while pos < len(line) and line[pos] in TOKEN_CHARS: + pos += 1 + if begin == pos: + raise ValueError("No token found at position %d" % pos) + return pos, line[begin:pos] + +def get_quoted_string(pos, line): + chars = [] + if not (pos < len(line) and line[pos] == '"'): + raise ValueError("Expected '\"' at beginning of quoted string.") + pos += 1 + while pos < len(line) and line[pos] != '"': + if line[pos] == '\\': + pos += 1 + if not (pos < len(line)): + raise ValueError("End of line after backslash in quoted string") + chars.append(line[pos]) + pos += 1 + if not (pos < len(line) and line[pos] == '"'): + raise ValueError("Expected '\"' at end of quoted string.") + pos += 1 + return pos, "".join(chars) + +def parse_transaction(line): + """A transaction is a command followed by zero or more key-value pairs. Like so: + COMMAND KEY="VALUE" KEY="\"ESCAPED\" VALUE" + Values must be quoted. Any byte value may be escaped with a backslash. + Returns a pair: (COMMAND, ((KEY1, VALUE1), (KEY2, VALUE2), ...)). + """ + pos = 0 + pos, skipped = skip_space(pos, line) + pos, command = get_token(pos, line) + + pairs = [] + while True: + pos, skipped = skip_space(pos, line) + if not (pos < len(line)): + break + if skipped == 0: + raise ValueError("Expected space before key-value pair") + pos, key = get_token(pos, line) + if not (pos < len(line) and line[pos] == '='): + raise ValueError("No '=' found after key") + pos += 1 + pos, value = get_quoted_string(pos, line) + pairs.append((key, value)) + return command, tuple(pairs) + +def param_first(key, params): + for k, v in params: + if key == k: + return v + return None + +def quote_string(s): + chars = [] + for c in s: + if c == "\\": + c = "\\\\" + elif c == "\"": + c = "\\\"" + chars.append(c) + return "\"" + "".join(chars) + "\"" + +def render_transaction(command, *params): + parts = [command] + for key, value in params: + parts.append("%s=%s" % (key, quote_string(value))) + return " ".join(parts) diff --git a/facilitator b/facilitator index 2ab6491..d1e25a4 100755 --- a/facilitator +++ b/facilitator @@ -3,12 +3,13 @@ import SocketServer import getopt import os -import re import socket import sys import threading import time +import fac + LISTEN_ADDRESS = "127.0.0.1" DEFAULT_LISTEN_PORT = 9002 DEFAULT_RELAY_PORT = 9001 @@ -32,10 +33,10 @@ class options(object): @staticmethod def set_relay_spec(spec): - af, host, port = parse_addr_spec(spec, defport = DEFAULT_RELAY_PORT) + af, host, port = fac.parse_addr_spec(spec, defport = DEFAULT_RELAY_PORT) # Resolve to get an IP address. addrs = socket.getaddrinfo(host, port, af) - options.relay_spec = format_addr(addrs[0][4]) + options.relay_spec = fac.format_addr(addrs[0][4]) def usage(f = sys.stdout): print >> f, """\ @@ -71,144 +72,13 @@ def log(msg): finally: log_lock.release() -def parse_addr_spec(spec, defhost = None, defport = None): - host = None - port = None - m = None - # IPv6 syntax. - if not m: - m = re.match(ur'^\[(.+)\]:(\d+)$', spec) - if m: - host, port = m.groups() - af = socket.AF_INET6 - if not m: - m = re.match(ur'^\[(.+)\]:?$', spec) - if m: - host, = m.groups() - af = socket.AF_INET6 - # IPv4 syntax. - if not m: - m = re.match(ur'^(.+):(\d+)$', spec) - if m: - host, port = m.groups() - af = socket.AF_INET - if not m: - m = re.match(ur'^:?(\d+)$', spec) - if m: - port, = m.groups() - af = 0 - if not m: - host = spec - af = 0 - host = host or defhost - port = port or defport - if not (host and port): - raise ValueError("Bad address specification \"%s\"" % spec) - return af, host, int(port) - -def format_addr(addr): - host, port = addr - if not host: - return u":%d" % port - # Numeric IPv6 address? - try: - addrs = socket.getaddrinfo(host, port, 0, socket.SOCK_STREAM, socket.IPPROTO_TCP, socket.AI_NUMERICHOST) - af = addrs[0][0] - except socket.gaierror, e: - af = 0 - if af == socket.AF_INET6: - return u"[%s]:%d" % (host, port) - else: - return u"%s:%d" % (host, port) - -def skip_space(pos, line): - """Skip a (possibly empty) sequence of space characters (the ASCII character - '\x20' exactly). Returns a pair (pos, num_skipped).""" - begin = pos - while pos < len(line) and line[pos] == "\x20": - pos += 1 - return pos, pos - begin - -TOKEN_CHARS = set("abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789_-") -def get_token(pos, line): - begin = pos - while pos < len(line) and line[pos] in TOKEN_CHARS: - pos += 1 - if begin == pos: - raise ValueError("No token found at position %d" % pos) - return pos, line[begin:pos] - -def get_quoted_string(pos, line): - chars = [] - if not (pos < len(line) and line[pos] == '"'): - raise ValueError("Expected '\"' at beginning of quoted string.") - pos += 1 - while pos < len(line) and line[pos] != '"': - if line[pos] == '\\': - pos += 1 - if not (pos < len(line)): - raise ValueError("End of line after backslash in quoted string") - chars.append(line[pos]) - pos += 1 - if not (pos < len(line) and line[pos] == '"'): - raise ValueError("Expected '\"' at end of quoted string.") - pos += 1 - return pos, "".join(chars) - -def parse_command(line): - """A line is a command followed by zero or more key-value pairs. Like so: - COMMAND KEY="VALUE" KEY="\"ESCAPED\" VALUE" - Values must be quoted. Any byte value may be escaped with a backslash. - Returns a pair: (COMMAND, ((KEY1, VALUE1), (KEY2, VALUE2), ...)). - """ - pos = 0 - pos, skipped = skip_space(pos, line) - pos, command = get_token(pos, line) - - pairs = [] - while True: - pos, skipped = skip_space(pos, line) - if not (pos < len(line)): - break - if skipped == 0: - raise ValueError("Expected space before key-value pair") - pos, key = get_token(pos, line) - if not (pos < len(line) and line[pos] == '='): - raise ValueError("No '=' found after key") - pos += 1 - pos, value = get_quoted_string(pos, line) - pairs.append((key, value)) - return command, tuple(pairs) - -def param_first(key, params): - for k, v in params: - if key == k: - return v - return None - -def quote_string(s): - chars = [] - for c in s: - if c == "\\": - c = "\\\\" - elif c == "\"": - c = "\\\"" - chars.append(c) - return "\"" + "".join(chars) + "\"" - -def render_params(params): - parts = [] - for key, value in params: - parts.append("%s=%s" % (key, quote_string(value))) - return " ".join(parts) - class TCPReg(object): def __init__(self, host, port): self.host = host self.port = port def __unicode__(self): - return format_addr((self.host, self.port)) + return fac.format_addr((self.host, self.port)) def __str__(self): return unicode(self).encode("UTF-8") @@ -223,7 +93,7 @@ class Reg(object): @staticmethod def parse(spec, defhost = None, defport = None): try: - af, host, port = parse_addr_spec(spec, defhost, defport) + af, host, port = fac.parse_addr_spec(spec, defhost, defport) except ValueError: pass else: @@ -330,9 +200,9 @@ class Handler(SocketServer.StreamRequestHandler): if not (len(line) > 0 and line[-1] == '\n'): raise ValueError("No newline at end of string returned by readline") try: - command, params = parse_command(line[:-1]) + command, params = fac.parse_transaction(line[:-1]) except ValueError, e: - log("parse_command: %s" % e) + log("fac.parse_transaction: %s" % e) self.send_error() return False @@ -355,14 +225,14 @@ class Handler(SocketServer.StreamRequestHandler): if reg: log(u"proxy gets %s, relay %s (now %d)" % (safe_str(unicode(reg)), options.relay_spec, len(REGS))) - print >> self.wfile, "OK", render_params((("CLIENT", str(reg)), ("RELAY", options.relay_spec))) + print >> self.wfile, fac.render_transaction("OK", ("CLIENT", str(reg)), ("RELAY", options.relay_spec)) else: log(u"proxy gets none") - print >> self.wfile, "NONE" + print >> self.wfile, fac.render_transaction("NONE") return True def do_PUT(self, params): - client_spec = param_first("CLIENT", params) + client_spec = fac.param_first("CLIENT", params) if client_spec is None: log(u"PUT missing CLIENT param") self.send_error() @@ -433,7 +303,7 @@ The -r option is required. Give it the relay that will be sent to proxies. server = Server(addrinfo[4], Handler) - log(u"start on %s" % format_addr(addrinfo[4])) + log(u"start on %s" % fac.format_addr(addrinfo[4])) log(u"using relay address %s" % options.relay_spec) if options.daemonize:

1 0

[flashproxy/master] Add tests for parse_transaction.
by dcf＠torproject.org 31 Aug '12

31 Aug '12

commit 2ff059561529a7add125efc74231973fd312edc0 Author: David Fifield <david(a)bamsoftware.com> Date: Sat Aug 11 09:17:38 2012 -0700 Add tests for parse_transaction. --- facilitator-test | 36 ++++++++++++++++++++++++++++++++++++ 1 files changed, 36 insertions(+), 0 deletions(-) diff --git a/facilitator-test b/facilitator-test index 4e2ec56..35bb018 100755 --- a/facilitator-test +++ b/facilitator-test @@ -5,6 +5,8 @@ import subprocess import time import unittest +import fac + FACILITATOR_HOST = "127.0.0.1" FACILITATOR_PORT = 9002 @@ -53,5 +55,39 @@ class FacilitatorTest(unittest.TestCase): except socket.error: pass +class ParseTransactionTest(unittest.TestCase): + def test_empty_string(self): + self.assertRaises(ValueError, fac.parse_transaction, "") + + def test_correct(self): + self.assertEqual(fac.parse_transaction("COMMAND"), ("COMMAND", ())) + self.assertEqual(fac.parse_transaction("COMMAND X=\"\""), ("COMMAND", (("X", ""),))) + self.assertEqual(fac.parse_transaction("COMMAND X=\"ABC\""), ("COMMAND", (("X", "ABC"),))) + self.assertEqual(fac.parse_transaction("COMMAND X=\"\\A\\B\\C\""), ("COMMAND", (("X", "ABC"),))) + self.assertEqual(fac.parse_transaction("COMMAND X=\"\\\\\\\"\""), ("COMMAND", (("X", "\\\""),))) + self.assertEqual(fac.parse_transaction("COMMAND X=\"ABC\" Y=\"DEF\""), ("COMMAND", (("X", "ABC"), ("Y", "DEF")))) + self.assertEqual(fac.parse_transaction("COMMAND KEY-NAME=\"ABC\""), ("COMMAND", (("KEY-NAME", "ABC"),))) + self.assertEqual(fac.parse_transaction("COMMAND KEY_NAME=\"ABC\""), ("COMMAND", (("KEY_NAME", "ABC"),))) + + def test_missing_command(self): + self.assertRaises(ValueError, fac.parse_transaction, "X=\"ABC\"") + self.assertRaises(ValueError, fac.parse_transaction, " X=\"ABC\"") + + def test_missing_space(self): + self.assertRaises(ValueError, fac.parse_transaction, "COMMAND/X=\"ABC\"") + self.assertRaises(ValueError, fac.parse_transaction, "COMMAND X=\"ABC\"Y=\"DEF\"") + + def test_bad_quotes(self): + self.assertRaises(ValueError, fac.parse_transaction, "COMMAND X=\"") + self.assertRaises(ValueError, fac.parse_transaction, "COMMAND X=\"ABC") + self.assertRaises(ValueError, fac.parse_transaction, "COMMAND X=\"ABC\" Y=\"ABC") + self.assertRaises(ValueError, fac.parse_transaction, "COMMAND X=\"ABC\\") + + def test_truncated(self): + self.assertRaises(ValueError, fac.parse_transaction, "COMMAND X=") + + def test_newline(self): + self.assertRaises(ValueError, fac.parse_transaction, "COMMAND X=\"ABC\" \nY=\"DEF\"") + if __name__ == "__main__": unittest.main()

1 0

[flashproxy/master] Make a recv timeout abstraction.
by dcf＠torproject.org 31 Aug '12

31 Aug '12

commit 9ed405581cb0e093e24cb2251f93911669e95e19 Author: David Fifield <david(a)bamsoftware.com> Date: Sat Jul 7 07:10:14 2012 -0700 Make a recv timeout abstraction. --- facilitator | 19 +++++++++++++++++-- 1 files changed, 17 insertions(+), 2 deletions(-) diff --git a/facilitator b/facilitator index 61129c7..26c964d 100755 --- a/facilitator +++ b/facilitator @@ -14,6 +14,9 @@ DEFAULT_LISTEN_PORT = 9002 DEFAULT_RELAY_PORT = 9001 DEFAULT_LOG_FILENAME = "facilitator.log" +# Don't indulge clients for more than this many seconds. +CLIENT_TIMEOUT = 1.0 + LOG_DATE_FORMAT = "%Y-%m-%d %H:%M:%S" class options(object): @@ -110,10 +113,22 @@ def format_addr(addr): return u"%s:%d" % (host, port) class Handler(SocketServer.StreamRequestHandler): + def __init__(self, *args, **kwargs): + self.deadline = time.time() + CLIENT_TIMEOUT + SocketServer.StreamRequestHandler.__init__(self, *args, **kwargs) + + def recv(self): + timeout = self.deadline - time.time() + self.connection.settimeout(timeout) + return self.connection.recv(1024) + + def readline(self): + while True: + self.connection.settimeout(timeout) + def handle(self): - self.connection.settimeout(1.0) while True: - data = self.connection.recv(1024) + data = self.recv() print repr(data) class Server(SocketServer.ThreadingMixIn, SocketServer.TCPServer):

1 0

[flashproxy/master] Add prototype of new facilitator.
by dcf＠torproject.org 31 Aug '12

31 Aug '12

commit a482cca6aff72b40529d02dc1c6f0479dc2a76ee Author: David Fifield <david(a)bamsoftware.com> Date: Thu Aug 30 10:31:20 2012 -0700 Add prototype of new facilitator. --- facilitator | 316 +++++++++++++---------------------------------------------- 1 files changed, 68 insertions(+), 248 deletions(-) diff --git a/facilitator b/facilitator index 2a04e71..61129c7 100755 --- a/facilitator +++ b/facilitator @@ -1,9 +1,6 @@ #!/usr/bin/env python -import BaseHTTPServer import SocketServer -import cgi -import errno import getopt import os import re @@ -11,17 +8,16 @@ import socket import sys import threading import time -import urllib -import urlparse -DEFAULT_ADDRESS = "0.0.0.0" -DEFAULT_PORT = 9002 +LISTEN_ADDRESS = "127.0.0.1" +DEFAULT_LISTEN_PORT = 9002 DEFAULT_RELAY_PORT = 9001 DEFAULT_LOG_FILENAME = "facilitator.log" LOG_DATE_FORMAT = "%Y-%m-%d %H:%M:%S" class options(object): + listen_port = DEFAULT_LISTEN_PORT log_filename = DEFAULT_LOG_FILENAME log_file = sys.stdout relay_spec = None @@ -38,30 +34,22 @@ class options(object): def usage(f = sys.stdout): print >> f, """\ -Usage: %(progname)s -r RELAY <OPTIONS> [HOST] [PORT] -Flash proxy facilitator: Register client addresses with HTTP POST requests -and serve them out again with HTTP GET. Listen on HOST and PORT, by default -%(addr)s %(port)d. +Usage: %(progname)s -r RELAY <OPTIONS> +Flash proxy facilitator: Register client addresses and serve them out +again. Listen on 127.0.0.1 and port PORT (by default %(port)d). -d, --debug don't daemonize, log to stdout. -h, --help show this help. -l, --log FILENAME write log to FILENAME (default \"%(log)s\"). + -p, --port PORT listen on PORT (by default %(port)d). --pidfile FILENAME write PID to FILENAME after daemonizing. -r, --relay RELAY send RELAY (host:port) to proxies as the relay to use. - --unsafe-logging don't scrub IP addresses from logs.\ + --unsafe-logging don't scrub IP addresses from logs.\ """ % { "progname": sys.argv[0], - "addr": DEFAULT_ADDRESS, - "port": DEFAULT_PORT, + "port": DEFAULT_LISTEN_PORT, "log": DEFAULT_LOG_FILENAME, } -def safe_str(s): - """Return s if options.safe_logging is true, and "[scrubbed]" otherwise.""" - if options.safe_logging: - return "[scrubbed]" - else: - return s - log_lock = threading.Lock() def log(msg): log_lock.acquire() @@ -121,244 +109,76 @@ def format_addr(addr): else: return u"%s:%d" % (host, port) -class TCPReg(object): - def __init__(self, host, port): - self.host = host - self.port = port - - def __unicode__(self): - return format_addr((self.host, self.port)) - - def __str__(self): - return unicode(self).encode("UTF-8") - - def __cmp__(self, other): - if isinstance(other, TCPReg): - return cmp((self.host, self.port), (other.host, other.port)) - else: - return False - -class Reg(object): - @staticmethod - def parse(spec, defhost = None, defport = None): - try: - af, host, port = parse_addr_spec(spec, defhost, defport) - except ValueError: - pass - else: +class Handler(SocketServer.StreamRequestHandler): + def handle(self): + self.connection.settimeout(1.0) + while True: + data = self.connection.recv(1024) + print repr(data) + +class Server(SocketServer.ThreadingMixIn, SocketServer.TCPServer): + allow_reuse_address = True + +def main(): + opts, args = getopt.gnu_getopt(sys.argv[1:], "dhl:p:r:", + ["debug", "help", "log=", "port=", "pidfile=", "relay=", "unsafe-logging"]) + for o, a in opts: + if o == "-d" or o == "--debug": + options.daemonize = False + options.log_filename = None + elif o == "-h" or o == "--help": + usage() + sys.exit() + elif o == "-l" or o == "--log": + options.log_filename = a + elif o == "-p" or o == "--port": + options.listen_port = int(a) + elif o == "--pidfile": + options.pid_filename = a + elif o == "-r" or o == "--relay": try: - addrs = socket.getaddrinfo(host, port, af, socket.SOCK_STREAM, socket.IPPROTO_TCP, socket.AI_NUMERICHOST) + options.set_relay_spec(a) except socket.gaierror, e: - raise ValueError("Bad host or port: \"%s\" \"%s\": %s" % (host, port, str(e))) - if not addrs: - raise ValueError("Bad host or port: \"%s\" \"%s\"" % (host, port)) - - host, port = socket.getnameinfo(addrs[0][4], socket.NI_NUMERICHOST | socket.NI_NUMERICSERV) - return TCPReg(host, int(port)) - - raise ValueError("Bad spec format: %s" % repr(spec)) - -class RegSet(object): - def __init__(self): - self.set = [] - self.cv = threading.Condition() - - def add(self, reg): - self.cv.acquire() - try: - if reg not in list(self.set): - self.set.append(reg) - self.cv.notify() - return True - else: - return False - finally: - self.cv.release() - - def fetch(self): - self.cv.acquire() - try: - if not self.set: - return None - return self.set.pop(0) - finally: - self.cv.release() - - def __len__(self): - self.cv.acquire() - try: - return len(self.set) - finally: - self.cv.release() - -class Handler(BaseHTTPServer.BaseHTTPRequestHandler): - def do_GET(self): - proxy_addr_s = format_addr(self.client_address) - - path = urlparse.urlsplit(self.path)[2] - - reg = REGS.fetch() - if reg: - log(u"proxy %s gets %s, relay %s (now %d)" % - (safe_str(proxy_addr_s), safe_str(unicode(reg)), options.relay_spec, len(REGS))) - else: - log(u"proxy %s gets none" % safe_str(proxy_addr_s)) - self.send_client(reg) - - def do_POST(self): - client_addr_s = format_addr(self.client_address) + print >> sys.stderr, u"Can't resolve relay %s: %s" % (repr(a), str(e)) + sys.exit(1) + elif o == "--unsafe-logging": + options.safe_logging = False - data = cgi.FieldStorage(fp = self.rfile, headers = self.headers, - environ = {"REQUEST_METHOD": "POST"}) - - client_spec = data.getfirst("client") - if client_spec is None: - log(u"client %s missing \"client\" param" % safe_str(client_addr_s)) - self.send_error(400) - return - - try: - reg = Reg.parse(client_spec, self.client_address[0]) - except ValueError, e: - log(u"client %s syntax error in %s: %s" - % (safe_str(client_addr_s), safe_str(repr(client_spec)), repr(str(e)))) - self.send_error(400) - return - - if REGS.add(reg): - log(u"client %s %s (now %d)" - % (safe_str(client_addr_s), safe_str(unicode(reg)), len(REGS))) - else: - log(u"client %s %s (already present, now %d)" - % (safe_str(client_addr_s), safe_str(unicode(reg)), len(REGS))) - - self.send_response(200) - self.end_headers() - - def send_error(self, code, message = None): - self.send_response(code) - self.end_headers() - if message: - self.wfile.write(message) - - def log_request(self, code): - pass - - def log_message(self, format, *args): - msg = format % args - log(u"message from HTTP handler for %s: %s" - % (format_addr(self.client_address), repr(msg))) - - def send_client(self, reg): - if reg: - client_str = str(reg) - else: - client_str = "" - self.send_response(200) - self.send_header("Content-Type", "application/x-www-form-urlencoded") - self.send_header("Cache-Control", "no-cache") - # Allow XMLHttpRequest from any domain. http://www.w3.org/TR/cors/. - self.send_header("Access-Control-Allow-Origin", "*") - self.end_headers() - - data = {} - data["client"] = client_str - data["relay"] = options.relay_spec - self.request.send(urllib.urlencode(data)) - - # Catch "broken pipe" errors that otherwise cause a stack trace in the log. - def catch_epipe(fn): - def ret(self, *args): - try: - fn(self, *args) - except socket.error, e: - try: - err_num = e.errno - except AttributeError: - # Before Python 2.6, exception can be a pair. - err_num, errstr = e - except: - raise - if err_num != errno.EPIPE: - raise - return ret - handle = catch_epipe(BaseHTTPServer.BaseHTTPRequestHandler.handle) - finish = catch_epipe(BaseHTTPServer.BaseHTTPRequestHandler.finish) - -REGS = RegSet() - -opts, args = getopt.gnu_getopt(sys.argv[1:], "dhl:r:", - ["debug", "help", "log=", "pidfile=", "relay=", "unsafe-logging"]) -for o, a in opts: - if o == "-d" or o == "--debug": - options.daemonize = False - options.log_filename = None - elif o == "-h" or o == "--help": - usage() - sys.exit() - elif o == "-l" or o == "--log": - options.log_filename = a - elif o == "--pidfile": - options.pid_filename = a - elif o == "-r" or o == "--relay": - try: - options.set_relay_spec(a) - except socket.gaierror, e: - print >> sys.stderr, u"Can't resolve relay %s: %s" % (repr(a), str(e)) - sys.exit(1) - elif o == "--unsafe-logging": - options.safe_logging = False - -if not options.relay_spec: - print >> sys.stderr, """\ + if not options.relay_spec: + print >> sys.stderr, """\ The -r option is required. Give it the relay that will be sent to proxies. -r HOST[:PORT]\ -""" - sys.exit(1) + """ + sys.exit(1) -if len(args) == 0: - address = (DEFAULT_ADDRESS, DEFAULT_PORT) -elif len(args) == 1: - # Either HOST or PORT may be omitted; figure out which one. - if args[0].isdigit(): - address = (DEFAULT_ADDRESS, args[0]) + if options.log_filename: + options.log_file = open(options.log_filename, "a") + # Send error tracebacks to the log. + sys.stderr = options.log_file else: - address = (args[0], DEFAULT_PORT) -elif len(args) == 2: - address = (args[0], args[1]) -else: - usage(sys.stderr) - sys.exit(1) + options.log_file = sys.stdout -if options.log_filename: - options.log_file = open(options.log_filename, "a") - # Send error tracebacks to the log. - sys.stderr = options.log_file -else: - options.log_file = sys.stdout + addrinfo = socket.getaddrinfo(LISTEN_ADDRESS, options.listen_port, 0, socket.SOCK_STREAM, socket.IPPROTO_TCP)[0] -addrinfo = socket.getaddrinfo(address[0], address[1], 0, socket.SOCK_STREAM, socket.IPPROTO_TCP)[0] + server = Server(addrinfo[4], Handler) -class Server(SocketServer.ThreadingMixIn, BaseHTTPServer.HTTPServer): - pass + log(u"start on %s" % format_addr(addrinfo[4])) + log(u"using relay address %s" % options.relay_spec) -# Setup the server -server = Server(addrinfo[4], Handler) + if options.daemonize: + log(u"daemonizing") + pid = os.fork() + if pid != 0: + if options.pid_filename: + f = open(options.pid_filename, "w") + print >> f, pid + f.close() + sys.exit(0) -log(u"start on %s" % format_addr(addrinfo[4])) -log(u"using relay address %s" % options.relay_spec) - -if options.daemonize: - log(u"daemonizing") - pid = os.fork() - if pid != 0: - if options.pid_filename: - f = open(options.pid_filename, "w") - print >> f, pid - f.close() + try: + server.serve_forever() + except KeyboardInterrupt: sys.exit(0) -try: - server.serve_forever() -except KeyboardInterrupt: - sys.exit(0) +if __name__ == "__main__": + main()

1 0