- tor-commits - lists.torproject.org

[flashproxy/master] Rename DEFAULT_FAC_POLL_INTERVAL to FACILITATOR_POLL_INTERVAL.
by dcf＠torproject.org 12 Jun '11

12 Jun '11

commit efed6d80241205315ae3f79724d0cd027a2319c1 Author: David Fifield <david(a)bamsoftware.com> Date: Sat Jun 11 12:44:12 2011 -0700 Rename DEFAULT_FAC_POLL_INTERVAL to FACILITATOR_POLL_INTERVAL. --- swfcat.as | 12 ++++++------ 1 files changed, 6 insertions(+), 6 deletions(-) diff --git a/swfcat.as b/swfcat.as index dd2396d..d0aad9a 100644 --- a/swfcat.as +++ b/swfcat.as @@ -37,9 +37,9 @@ package host: "127.0.0.1", port: 9002 }; - - /* Poll facilitator every 10 sec */ - private const DEFAULT_FAC_POLL_INTERVAL:uint = 10000; + + // Milliseconds. + private const FACILITATOR_POLL_INTERVAL:int = 10000; // Socket to Cirrus server private var s_c:CirrusSocket; @@ -185,7 +185,7 @@ package s_f = new FacilitatorSocket(fac_addr.host, fac_addr.port); s_f.addEventListener(FacilitatorSocketEvent.CONNECT_FAILED, function (e:Event):void { puts("Facilitator: connect failed."); - setTimeout(establish_facilitator_connection, DEFAULT_FAC_POLL_INTERVAL); + setTimeout(establish_facilitator_connection, FACILITATOR_POLL_INTERVAL); }); if (proxy_mode) { @@ -210,14 +210,14 @@ package }); s_f.addEventListener(FacilitatorSocketEvent.REGISTRATIONS_EMPTY, function (e:Event):void { puts("Facilitator: no registrations available."); - setTimeout(establish_facilitator_connection, DEFAULT_FAC_POLL_INTERVAL); + setTimeout(establish_facilitator_connection, FACILITATOR_POLL_INTERVAL); }); puts("Facilitator: getting registration."); s_f.get_registration(); } else { s_f.addEventListener(FacilitatorSocketEvent.REGISTRATION_FAILED, function (e:Event):void { puts("Facilitator: registration failed."); - setTimeout(establish_facilitator_connection, DEFAULT_FAC_POLL_INTERVAL); + setTimeout(establish_facilitator_connection, FACILITATOR_POLL_INTERVAL); }); puts("Facilitator: posting registration."); s_f.post_registration(s_c.id);

1 0

[flashproxy/master] Switching swfcat.as to point to David's facilitator by default instead of Nate's
by dcf＠torproject.org 12 Jun '11

12 Jun '11

commit c16f7d998686c84d30b4f6e4177659e3845888b7 Author: Nate Hardison <hardison(a)stanford.edu> Date: Thu Jun 9 17:59:57 2011 -0700 Switching swfcat.as to point to David's facilitator by default instead of Nate's --- swfcat.as | 4 ++-- 1 files changed, 2 insertions(+), 2 deletions(-) diff --git a/swfcat.as b/swfcat.as index 627a3cd..1bb2ff5 100644 --- a/swfcat.as +++ b/swfcat.as @@ -27,9 +27,9 @@ package private const DEFAULT_CIRRUS_ADDR:String = "rtmfp://p2p.rtmfp.net"; private const DEFAULT_CIRRUS_KEY:String = RTMFP::CIRRUS_KEY; - /* Nate's facilitator -- serves a crossdomain policy */ + /* David's facilitator. */ private const DEFAULT_FACILITATOR_ADDR:Object = { - host: "128.12.179.80", + host: "173.255.221.44", port: 9002 };

1 0

[flashproxy/master] Use the same connector as in master branch.
by dcf＠torproject.org 12 Jun '11

12 Jun '11

commit e8f3ca392f37f326dac975a6827a9d4e71eb6c50 Author: David Fifield <david(a)bamsoftware.com> Date: Sat Jun 11 12:29:17 2011 -0700 Use the same connector as in master branch. Instead of coding different default ports, just document the command-line parameters you need to use to make the difference. --- connector.py | 6 +++--- 1 files changed, 3 insertions(+), 3 deletions(-) diff --git a/connector.py b/connector.py index 731fa19..27fff61 100755 --- a/connector.py +++ b/connector.py @@ -12,8 +12,8 @@ import time import urllib import xml.sax.saxutils -DEFAULT_REMOTE_ADDRESS = "127.0.0.1" -DEFAULT_REMOTE_PORT = 9002 +DEFAULT_REMOTE_ADDRESS = "0.0.0.0" +DEFAULT_REMOTE_PORT = 9000 DEFAULT_LOCAL_ADDRESS = "127.0.0.1" DEFAULT_LOCAL_PORT = 9001 DEFAULT_FACILITATOR_PORT = 9002 @@ -281,7 +281,7 @@ def register(): spec = format_addr((None, options.remote_addr[1])) log(u"Registering \"%s\" with %s." % (spec, format_addr(options.facilitator_addr))) http = httplib.HTTPConnection(*options.facilitator_addr) - http.request("POST", "/", urllib.urlencode({"client": spec}), {"Content-Type":"application/x-www-form-urlencoded"}) + http.request("POST", "/", urllib.urlencode({"client": spec})) http.close() return True

1 0

[flashproxy/master] Merge branch 'rtmfp' of ssh://upload.bamsoftware.com/~dcf/flashproxy into rtmfp
by dcf＠torproject.org 12 Jun '11

12 Jun '11

commit 07752417d9a42099bc0c869b0ebac1862f72baa4 Merge: 7c4ca7b a7cdef5 Author: Nate Hardison <nate(a)rescomp-09-154551.stanford.edu> Date: Thu Jun 2 03:46:27 2011 -0700 Merge branch 'rtmfp' of ssh://upload.bamsoftware.com/~dcf/flashproxy into rtmfp badge_con_counter.png | Bin 0 -> 270 bytes return_of_the_rtmfpcat.as | 234 +++++++++++++++++++++++++++++++++++++-------- 2 files changed, 192 insertions(+), 42 deletions(-)

1 0

[flashproxy/master] Adding some comments and removing clutter
by dcf＠torproject.org 12 Jun '11

12 Jun '11

commit 7a302499ef39988d560790a220d4f05f5f1b2e9b Author: Nate Hardison <nate(a)rescomp-09-154551.stanford.edu> Date: Thu Jun 2 04:02:36 2011 -0700 Adding some comments and removing clutter --- rtmfp/CirrusSocket.as | 2 ++ rtmfp/RTMFPSocket.as | 17 +---------------- 2 files changed, 3 insertions(+), 16 deletions(-) diff --git a/rtmfp/CirrusSocket.as b/rtmfp/CirrusSocket.as index d9996f2..9e97c7d 100644 --- a/rtmfp/CirrusSocket.as +++ b/rtmfp/CirrusSocket.as @@ -49,6 +49,8 @@ package rtmfp connection.addEventListener(NetStatusEvent.NET_STATUS, on_net_status_event); connection.addEventListener(IOErrorEvent.IO_ERROR, on_io_error_event); connection.addEventListener(SecurityErrorEvent.SECURITY_ERROR, on_security_error_event); + + /* Set up a client object to handle the hello callback */ var client:Object = new Object(); client.onRelay = on_hello; connection.client = client; diff --git a/rtmfp/RTMFPSocket.as b/rtmfp/RTMFPSocket.as index 783a0e0..6e9a45e 100644 --- a/rtmfp/RTMFPSocket.as +++ b/rtmfp/RTMFPSocket.as @@ -45,8 +45,7 @@ package rtmfp } /* Tears down this RTMFPSocket, closing both its streams. - To be used when destroying this object. If you just want - to disconnect from a client, call disconnect() below */ + To be used when destroying this object. */ public function close():void { if (send_stream != null) { @@ -82,19 +81,6 @@ package rtmfp RTMFPSocketClient(recv_stream.client).connect_acknowledged); } - public function disconnect():void - { - if (recv_stream != null) { - if (recv_stream.client != null) { - recv_stream.client.removeEventListener(ProgressEvent.SOCKET_DATA, on_data_available); - recv_stream.client.removeEventListener(RTMFPSocketClient.CONNECT_ACKNOWLEDGED, on_connect_acknowledged); - } - recv_stream.removeEventListener(NetStatusEvent.NET_STATUS, on_recv_stream_event); - recv_stream.close(); - recv_stream = null; - } - } - /* In RTMFP, you open a listening socket by publishing a named stream that others can connect to instead of listening on a port. You register this stream with the Cirrus server via the Cirrus @@ -132,7 +118,6 @@ package rtmfp } } - public function writeBytes(bytes:ByteArray):void { if (send_stream != null && peer_connected) {

1 0

[flashproxy/master] Added badge
by dcf＠torproject.org 12 Jun '11

12 Jun '11

commit a7cdef5eaddee994257579f0b2db01c65164eda4 Author: ellitron <jdellit(a)stanford.edu> Date: Fri May 27 12:27:00 2011 -0700 Added badge --- badge_con_counter.png | Bin 0 -> 270 bytes return_of_the_rtmfpcat.as | 157 +++++++++++++++++++++++++++++++++++++++------ 2 files changed, 138 insertions(+), 19 deletions(-) diff --git a/badge_con_counter.png b/badge_con_counter.png new file mode 100644 index 0000000..d832e24 Binary files /dev/null and b/badge_con_counter.png differ diff --git a/return_of_the_rtmfpcat.as b/return_of_the_rtmfpcat.as index a1d14cc..e512408 100644 --- a/return_of_the_rtmfpcat.as +++ b/return_of_the_rtmfpcat.as @@ -54,13 +54,15 @@ package private static const DEFAULT_CIRCON_TIMEOUT:uint = 4000; + private static const DEFAULT_PEER_CON_TIMEOUT:uint = 4000; + /* Maximum connections. */ private const DEFAULT_MAXIMUM_RCP_PAIRS:uint = 1; /* Milliseconds. */ private const FACILITATOR_POLL_INTERVAL:int = 10000; - private var max_rcp_pairs; + private var max_rcp_pairs:uint; /* TextField for debug output. */ private var output_text:TextField; @@ -90,19 +92,42 @@ package /* Number of connected RTMFPConnectionPairs. */ private var rcp_pairs:uint; + private var rcp_pairs_total:uint; private var rtmfp_data_counter:uint; /* Keep track of facilitator polling timer. */ private var fac_poll_timeo_id:uint; + /* Badge with a client counter */ + [Embed(source="badge_con_counter.png")] + private var BadgeImage:Class; + private var tot_client_count_tf:TextField; + private var tot_client_count_fmt:TextFormat; + private var cur_client_count_tf:TextField; + private var cur_client_count_fmt:TextFormat; + /* Put a string to the screen. */ public function puts(s:String):void { output_text.appendText(s + "\n"); output_text.scrollV = output_text.maxScrollV; } - + + public function update_client_count():void + { + /* Update total client count. */ + if (String(rcp_pairs_total).length == 1) + tot_client_count_tf.text = "0" + String(rcp_pairs_total); + else + tot_client_count_tf.text = String(rcp_pairs_total); + + /* Update current client count. */ + cur_client_count_tf.text = ""; + for(var i:Number=0; i<rcp_pairs; i++) + cur_client_count_tf.appendText(".");; + } + public function return_of_the_rtmfpcat() { // Absolute positioning. @@ -115,6 +140,38 @@ package output_text.background = true; output_text.backgroundColor = 0x001f0f; output_text.textColor = 0x44cc44; + + /* Setup client counter for badge. */ + tot_client_count_fmt = new TextFormat(); + tot_client_count_fmt.color = 0xFFFFFF; + tot_client_count_fmt.align = "center"; + tot_client_count_fmt.font = "courier-new"; + tot_client_count_fmt.bold = true; + tot_client_count_fmt.size = 10; + tot_client_count_tf = new TextField(); + tot_client_count_tf.width = 20; + tot_client_count_tf.height = 17; + tot_client_count_tf.background = false; + tot_client_count_tf.defaultTextFormat = tot_client_count_fmt; + tot_client_count_tf.x=47; + tot_client_count_tf.y=0; + + cur_client_count_fmt = new TextFormat(); + cur_client_count_fmt.color = 0xFFFFFF; + cur_client_count_fmt.align = "center"; + cur_client_count_fmt.font = "courier-new"; + cur_client_count_fmt.bold = true; + cur_client_count_fmt.size = 10; + cur_client_count_tf = new TextField(); + cur_client_count_tf.width = 20; + cur_client_count_tf.height = 17; + cur_client_count_tf.background = false; + cur_client_count_tf.defaultTextFormat = cur_client_count_fmt; + cur_client_count_tf.x=47; + cur_client_count_tf.y=6; + + /* Update the client counter on badge. */ + update_client_count(); /* Initialize connection pair count. */ rcp_pairs = 0; @@ -136,7 +193,16 @@ package puts("Parameters loaded."); proxy_mode = (this.loaderInfo.parameters["proxy"] != null); - addChild(output_text); + + if(this.loaderInfo.parameters["debug"] != null) + addChild(output_text); + + addChild(new BadgeImage()); + /* Tried unsuccessfully to add counter to badge. */ + /* For now, need two addChilds :( */ + addChild(tot_client_count_tf); + addChild(cur_client_count_tf); + fac_spec = this.loaderInfo.parameters["facilitator"]; if (fac_spec) { @@ -180,6 +246,9 @@ package else max_rcp_pairs = DEFAULT_MAXIMUM_RCP_PAIRS; + if(this.loaderInfo.parameters["start"]) + rtmfp_data_counter = this.loaderInfo.parameters["start"]; + main(); } @@ -254,8 +323,14 @@ package var rcp:RTMFPConnectionPair = new RTMFPConnectionPair(circon, tor_addr, output_text); rcp.addEventListener(Event.CONNECT, rcp_connect_event); + rcp.addEventListener(Event.UNLOAD, function (e:Event):void { + /* Failed to connect to peer... continue loop. */ + puts("RTMFPConnectionPair: Timed out connecting to peer!"); + if(rcp_pairs < max_rcp_pairs) + poll_for_id(); + }); rcp.addEventListener(Event.CLOSE, rcp_close_event); - rcp.connect(client_id, client_data); + rcp.connect(client_id, client_data, DEFAULT_PEER_CON_TIMEOUT); } else { /* Need to clear any outstanding timers to ensure * that only one timer ever runs. */ @@ -277,6 +352,10 @@ package puts("RTMFPConnectionPair connected"); rcp_pairs++; + rcp_pairs_total++; + + /* Update the client count on the badge. */ + update_client_count(); if(proxy_mode) { if(rcp_pairs < max_rcp_pairs) { @@ -304,6 +383,9 @@ package rcp_pairs--; + /* Update the client count on the badge. */ + update_client_count(); + /* FIXME: Do I need to unregister the event listeners so * that the system can garbage collect the rcp object? */ if(proxy_mode) { @@ -402,6 +484,8 @@ class RTMFPSocket extends EventDispatcher /* Keeps the state of our connectedness. */ public var connected:Boolean; + private var connect_timeo_id:uint; + private var output_text:TextField; /* Put a string to the screen. */ @@ -437,15 +521,26 @@ class RTMFPSocket extends EventDispatcher send_stream.publish(data); } - public function connect(clientID:String, data:String):void + private function connect_timeout():void + { + puts("RTMFPSocket: Timed out connecting to peer"); + close(); + /* OK this is not so nice, because I wanted an Event.TIMEOUT event, but flash gives us none such event :( */ + dispatchEvent(new Event(Event.UNLOAD)); + } + + public function connect(clientID:String, data:String, timeout:uint):void { puts("RTMFPSocket: connecting to peer..."); + connect_timeo_id = setTimeout(connect_timeout, timeout); + this.data = data; send_stream = new NetStream(circon, NetStream.DIRECT_CONNECTIONS); var client:Object = new Object(); client.onPeerConnect = function (peer:NetStream):Boolean { + clearTimeout(connect_timeo_id); puts("RTMFPSocket: connected to peer"); connected = true; dispatchEvent(new Event(Event.CONNECT)); @@ -605,18 +700,48 @@ class RTMFPConnectionPair extends EventDispatcher this.output_text = output_text; } - public function connect(clientID:String, rtmfp_data:String):void + public function connect(clientID:String, rtmfp_data:String, timeout:uint):void { s_r = new RTMFPSocket(circon, output_text); s_r.addEventListener(Event.CONNECT, rtmfp_connect_event); + s_r.addEventListener(Event.UNLOAD, function (e:Event):void { + dispatchEvent(new Event(Event.UNLOAD)); + }); + s_r.addEventListener(ProgressEvent.SOCKET_DATA, function (e:ProgressEvent):void { + /* It's possible that we receive data before we're connected + * to the tor-side of the connection. In this case we want + * to buffer the data. */ + if(!s_t.connected) { + puts("MASSIVE ATTACK!"); + } else { + var bytes:ByteArray = new ByteArray(); + s_r.readBytes(bytes, 0, e.bytesLoaded); + puts("RTMFPConnectionPair: RTMFP: read " + bytes.length + " bytes."); + s_t.writeBytes(bytes); + } + }); s_r.addEventListener(Event.CLOSE, rtmfp_close_event); - s_r.connect(clientID, rtmfp_data); + s_r.connect(clientID, rtmfp_data, timeout); } public function listen(rtmfp_data:String):void { s_r = new RTMFPSocket(circon, output_text); s_r.addEventListener(Event.CONNECT, rtmfp_connect_event); + s_r.addEventListener(ProgressEvent.SOCKET_DATA, function (e:ProgressEvent):void { + /* It's possible that we receive data before we're connected + * to the tor-side of the connection. In this case we want + * to buffer the data. */ + if(!s_t.connected) { + puts("MASSIVE ATTACK!"); + } else { + var bytes:ByteArray = new ByteArray(); + s_r.readBytes(bytes, 0, e.bytesLoaded); + puts("RTMFPConnectionPair: RTMFP: read " + bytes.length + " bytes."); + s_t.writeBytes(bytes); + } + }); + s_r.addEventListener(Event.CLOSE, rtmfp_close_event); s_r.listen(rtmfp_data); } @@ -630,18 +755,6 @@ class RTMFPConnectionPair extends EventDispatcher s_t = new Socket(); s_t.addEventListener(Event.CONNECT, function (e:Event):void { puts("RTMFPConnectionPair: Tor: connected to " + tor_addr.host + ":" + tor_addr.port + "."); - s_r.addEventListener(ProgressEvent.SOCKET_DATA, function (e:ProgressEvent):void { - var bytes:ByteArray = new ByteArray(); - s_r.readBytes(bytes, 0, e.bytesLoaded); - puts("RTMFPConnectionPair: RTMFP: read " + bytes.length + " bytes."); - s_t.writeBytes(bytes); - }); - s_t.addEventListener(ProgressEvent.SOCKET_DATA, function (e:ProgressEvent):void { - var bytes:ByteArray = new ByteArray(); - s_t.readBytes(bytes, 0, e.bytesLoaded); - puts("RTMFPConnectionPair: Tor: read " + bytes.length + " bytes."); - s_r.writeBytes(bytes); - }); dispatchEvent(new Event(Event.CONNECT)); }); s_t.addEventListener(Event.CLOSE, function (e:Event):void { @@ -652,6 +765,12 @@ class RTMFPConnectionPair extends EventDispatcher s_r.close(); dispatchEvent(new Event(Event.CLOSE)); }); + s_t.addEventListener(ProgressEvent.SOCKET_DATA, function (e:ProgressEvent):void { + var bytes:ByteArray = new ByteArray(); + s_t.readBytes(bytes, 0, e.bytesLoaded); + puts("RTMFPConnectionPair: Tor: read " + bytes.length + " bytes."); + s_r.writeBytes(bytes); + }); s_t.addEventListener(IOErrorEvent.IO_ERROR, function (e:IOErrorEvent):void { puts("RTMFPConnectionPair: Tor: I/O error: " + e.text + "."); });

1 0

[flashproxy/master] Attempted to fix stream disconnection notification, also trying to fix whitespace
by dcf＠torproject.org 12 Jun '11

12 Jun '11

commit 60e2c5afa70435c9f336bf33097717ff24573515 Author: Nate Hardison <nate(a)rescomp-09-154551.stanford.edu> Date: Thu Jun 2 10:50:09 2011 -0700 Attempted to fix stream disconnection notification, also trying to fix whitespace --- rtmfp/CirrusSocket.as | 28 +++++++++++++------------- rtmfp/RTMFPSocket.as | 53 +++++++++++++++++++++++++----------------------- rtmfpcat.as | 3 ++ 3 files changed, 45 insertions(+), 39 deletions(-) diff --git a/rtmfp/CirrusSocket.as b/rtmfp/CirrusSocket.as index 9e97c7d..071ac0d 100644 --- a/rtmfp/CirrusSocket.as +++ b/rtmfp/CirrusSocket.as @@ -31,10 +31,10 @@ package rtmfp { private static const CONNECT_TIMEOUT:uint = 4000; // in milliseconds - /* We'll append a unique number to the DATA_STREAM_PREFIX for each + /* We'll append a unique number to the DATA_STREAM_PREFIX for each new stream we create so that we have unique streams per player. */ private static const DATA_STREAM_PREFIX:String = "DATA"; - private var data_stream_suffix:uint = 0; + private var data_stream_suffix:uint = 0; /* Connection to the Cirrus rendezvous service */ public var connection:NetConnection; @@ -94,18 +94,18 @@ package rtmfp return DATA_STREAM_PREFIX + data_stream_suffix; } - /* Sends a hello message to the Flash player with Cirrus ID "id" - We use this new call protocol outlined here: - http://forums.adobe.com/thread/780788?tstart=0 */ - public function send_hello(id:String):void - { - if (this.connected) { - connection.call("relay", null, id, local_stream_name); - } else { - throw new Error("Cannot send hello: Cirrus socket not connected."); - } - } - + /* Sends a hello message to the Flash player with Cirrus ID "id" + We use this new call protocol outlined here: + http://forums.adobe.com/thread/780788?tstart=0 */ + public function send_hello(id:String):void + { + if (this.connected) { + connection.call("relay", null, id, local_stream_name); + } else { + throw new Error("Cannot send hello: Cirrus socket not connected."); + } + } + /*************************** PRIVATE HELPER FUNCTIONS *************************/ private function fail():void diff --git a/rtmfp/RTMFPSocket.as b/rtmfp/RTMFPSocket.as index 6e9a45e..bcbee67 100644 --- a/rtmfp/RTMFPSocket.as +++ b/rtmfp/RTMFPSocket.as @@ -27,26 +27,28 @@ package rtmfp [Event(name=RTMFPSocketEvent.PUBLISH_FAILED, type="com.flashproxy.rtmfp.events.RTMFPSocketEvent")] public class RTMFPSocket extends EventDispatcher { - private const CONNECT_TIMEOUT:uint = 10000; + private const CONNECT_TIMEOUT:uint = 10000; private var s_c:CirrusSocket; - private var recv_stream:NetStream; + private var recv_stream:NetStream; private var send_stream:NetStream; - private var connect_timeout:int; + private var peer_stream:NetStream; + + private var connect_timeout:int; public function RTMFPSocket(s_c:CirrusSocket) { this.s_c = s_c; recv_stream = null; - send_stream = null; - connect_timeout = 0; + send_stream = null; + connect_timeout = 0; } - /* Tears down this RTMFPSocket, closing both its streams. - To be used when destroying this object. */ - public function close():void + /* Tears down this RTMFPSocket, closing both its streams. + To be used when destroying this object. */ + public function close():void { if (send_stream != null) { s_c.connection.removeEventListener(NetStatusEvent.NET_STATUS, on_stream_disconnection_event); @@ -81,23 +83,23 @@ package rtmfp RTMFPSocketClient(recv_stream.client).connect_acknowledged); } - /* In RTMFP, you open a listening socket by publishing a named + /* In RTMFP, you open a listening socket by publishing a named stream that others can connect to instead of listening on a port. You register this stream with the Cirrus server via the Cirrus socket so that it can redirect connection requests for an id/stream tuple to this socket. */ public function listen(stream:String):void { - // apparently streams don't get disconnection events, only the NetConnection - // object does...bleh. - s_c.connection.addEventListener(NetStatusEvent.NET_STATUS, on_stream_disconnection_event); + // apparently streams don't get disconnection events, only the NetConnection + // object does...bleh. + s_c.connection.addEventListener(NetStatusEvent.NET_STATUS, on_stream_disconnection_event); send_stream = new NetStream(s_c.connection, NetStream.DIRECT_CONNECTIONS); - send_stream.addEventListener(NetStatusEvent.NET_STATUS, on_send_stream_event); - var client:Object = new Object(); - client.onPeerConnect = on_peer_connect; - send_stream.client = client; - send_stream.publish(stream); + send_stream.addEventListener(NetStatusEvent.NET_STATUS, on_send_stream_event); + var client:Object = new Object(); + client.onPeerConnect = on_peer_connect; + send_stream.client = client; + send_stream.publish(stream); } public function get peer():String @@ -113,16 +115,16 @@ package rtmfp public function readBytes(bytes:ByteArray, offset:uint = 0, length:uint = 0):void { - if (recv_stream != null && recv_stream.client != null) { - recv_stream.client.bytes.readBytes(bytes, offset, length); - } + if (recv_stream != null && recv_stream.client != null) { + recv_stream.client.bytes.readBytes(bytes, offset, length); + } } public function writeBytes(bytes:ByteArray):void { if (send_stream != null && peer_connected) { - send_stream.send(RTMFPSocketClient.DATA_AVAILABLE, bytes); - } + send_stream.send(RTMFPSocketClient.DATA_AVAILABLE, bytes); + } } /* Listens for acknowledgement of a connection attempt to a @@ -130,7 +132,7 @@ package rtmfp private function on_connect_acknowledged(event:Event):void { clearTimeout(connect_timeout); - dispatchEvent(new RTMFPSocketEvent(RTMFPSocketEvent.CONNECT_SUCCESS, recv_stream)); + dispatchEvent(new RTMFPSocketEvent(RTMFPSocketEvent.CONNECT_SUCCESS, recv_stream)); } /* If we don't get a connection acknowledgement by the time this @@ -163,6 +165,7 @@ package rtmfp return false; } + peer_stream = peer; peer.send(RTMFPSocketClient.SET_CONNECT_ACKNOWLEDGED); // need to do this in a timeout so that this function can @@ -190,10 +193,10 @@ package rtmfp private function on_stream_disconnection_event(event:NetStatusEvent):void { - if (event.info.code == "NetStream.Connect.Closed") { + if (event.info.code == "NetStream.Connect.Closed" && event.info.stream === peer_stream) { dispatchEvent(new RTMFPSocketEvent(RTMFPSocketEvent.PEER_DISCONNECTED)); - //disconnect(); } } } } + diff --git a/rtmfpcat.as b/rtmfpcat.as index 866175e..25099dd 100644 --- a/rtmfpcat.as +++ b/rtmfpcat.as @@ -49,6 +49,8 @@ package private var s_f:FacilitatorSocket; // Handle local-remote traffic private var p_p:ProxyPair; + + private var proxy_pairs:Array; private var proxy_mode:Boolean; @@ -87,6 +89,7 @@ package puts("Parameters loaded."); proxy_mode = (this.loaderInfo.parameters["proxy"] != null); + proxy_pairs = new Array(); fac_spec = this.loaderInfo.parameters["facilitator"]; if (fac_spec) {

1 0

[flashproxy/master] Adding in David's rate limiting code to rtmfpcat
by dcf＠torproject.org 12 Jun '11

12 Jun '11

commit ab4e90626b6f6a7152c79b2d2aeeae601deb9d64 Author: Nate Hardison <nate(a)rescomp-09-154551.stanford.edu> Date: Thu Jun 2 13:23:08 2011 -0700 Adding in David's rate limiting code to rtmfpcat --- rtmfp/ProxyPair.as | 218 +++++++++++++++++++++++++++++++++++++------- rtmfp/RTMFPSocketClient.as | 5 +- rtmfpcat.as | 38 +++----- 3 files changed, 202 insertions(+), 59 deletions(-) diff --git a/rtmfp/ProxyPair.as b/rtmfp/ProxyPair.as index 42d4b81..fb184cc 100644 --- a/rtmfp/ProxyPair.as +++ b/rtmfp/ProxyPair.as @@ -7,6 +7,8 @@ package rtmfp import flash.events.SecurityErrorEvent; import flash.net.Socket; import flash.utils.ByteArray; + import flash.utils.clearTimeout; + import flash.utils.setTimeout; import rtmfp.CirrusSocket; import rtmfp.RTMFPSocket; @@ -14,20 +16,41 @@ package rtmfp public class ProxyPair extends EventDispatcher { - private var parent:rtmfpcat; + private var ui:rtmfpcat; private var s_r:RTMFPSocket; private var s_t:Socket; private var tor_host:String; private var tor_port:uint; + + private var p2t_schedule:Array; + private var t2p_schedule:Array; + + // Bytes per second. Set to undefined to disable limit. + public const RATE_LIMIT:Number = 10000; + // Seconds. + private const RATE_LIMIT_HISTORY:Number = 5.0; + + private var rate_limit:RateLimit; + + // Callback id. + private var flush_id:uint; - public function ProxyPair(parent:rtmfpcat, s_c:CirrusSocket, tor_host:String, tor_port:uint) + public function ProxyPair(ui:rtmfpcat, s_c:CirrusSocket, tor_host:String, tor_port:uint) { - this.parent = parent; + this.ui = ui; this.tor_host = tor_host; this.tor_port = tor_port; + this.p2t_schedule = new Array(); + this.t2p_schedule = new Array(); + + if (RATE_LIMIT) + rate_limit = new BucketRateLimit(RATE_LIMIT * RATE_LIMIT_HISTORY, RATE_LIMIT_HISTORY); + else + rate_limit = new RateUnlimit(); + setup_rtmfp_socket(s_c); setup_tor_socket(); } @@ -61,65 +84,196 @@ package rtmfp private function setup_rtmfp_socket(s_c:CirrusSocket):void { s_r = new RTMFPSocket(s_c); - s_r.addEventListener(RTMFPSocketEvent.PLAY_STARTED, function (e:RTMFPSocketEvent):void { - puts("Play started."); + s_r.addEventListener(RTMFPSocketEvent.CONNECT_FAILED, function (e:RTMFPSocketEvent):void { + ui.puts("Peering failed."); }); - s_r.addEventListener(RTMFPSocketEvent.PUBLISH_STARTED, function (e:RTMFPSocketEvent):void { - puts("Publishing started."); + s_r.addEventListener(RTMFPSocketEvent.CONNECT_SUCCESS, function (e:RTMFPSocketEvent):void { + ui.puts("Peering success."); + s_t.connect(tor_host, tor_port); }); s_r.addEventListener(RTMFPSocketEvent.PEER_CONNECTED, function (e:RTMFPSocketEvent):void { - puts("Peer connected."); + ui.puts("Peer connected."); }); s_r.addEventListener(RTMFPSocketEvent.PEER_DISCONNECTED, function (e:RTMFPSocketEvent):void { - puts("Peer disconnected."); + ui.puts("Peer disconnected."); close(); }); - s_r.addEventListener(RTMFPSocketEvent.CONNECT_SUCCESS, function (e:RTMFPSocketEvent):void { - puts("Peering success."); - s_t.connect(tor_host, tor_port); + s_r.addEventListener(RTMFPSocketEvent.PLAY_STARTED, function (e:RTMFPSocketEvent):void { + ui.puts("Play started."); }); - s_r.addEventListener(RTMFPSocketEvent.CONNECT_FAILED, function (e:RTMFPSocketEvent):void { - puts("Peering failed."); + s_r.addEventListener(RTMFPSocketEvent.PUBLISH_STARTED, function (e:RTMFPSocketEvent):void { + ui.puts("Publishing started."); }); + s_r.addEventListener(ProgressEvent.SOCKET_DATA, proxy_to_tor); } private function setup_tor_socket():void { s_t = new Socket(); s_t.addEventListener(Event.CONNECT, function (e:Event):void { - puts("Tor: connected to " + tor_host + ":" + tor_port + "."); - s_t.addEventListener(ProgressEvent.SOCKET_DATA, function (e:ProgressEvent):void { - var bytes:ByteArray = new ByteArray(); - s_t.readBytes(bytes, 0, e.bytesLoaded); - puts("RTMFPSocket: Tor: read " + bytes.length + " bytes."); - s_r.writeBytes(bytes); - }); - s_r.addEventListener(ProgressEvent.SOCKET_DATA, function (e:ProgressEvent):void { - var bytes:ByteArray = new ByteArray(); - s_r.readBytes(bytes, 0, e.bytesLoaded); - puts("RTMFPSocket: RTMFP: read " + bytes.length + " bytes."); - s_t.writeBytes(bytes); - }); + ui.puts("Tor: connected to " + tor_host + ":" + tor_port + "."); dispatchEvent(new Event(Event.CONNECT)); }); s_t.addEventListener(Event.CLOSE, function (e:Event):void { - puts("Tor: closed connection."); + ui.puts("Tor: closed connection."); close(); }); s_t.addEventListener(IOErrorEvent.IO_ERROR, function (e:IOErrorEvent):void { - puts("Tor: I/O error: " + e.text + "."); + ui.puts("Tor: I/O error: " + e.text + "."); close(); }); s_t.addEventListener(SecurityErrorEvent.SECURITY_ERROR, function (e:SecurityErrorEvent):void { - puts("Tor: security error: " + e.text + "."); + ui.puts("Tor: security error: " + e.text + "."); close(); }); + s_t.addEventListener(ProgressEvent.SOCKET_DATA, tor_to_proxy); } - private function puts(s:String):void + private function tor_to_proxy(e:ProgressEvent):void + { + t2p_schedule.push(e.bytesLoaded); + flush(); + } + + private function proxy_to_tor(e:ProgressEvent):void { - parent.puts(s); + p2t_schedule.push(e.bytesLoaded); + flush(); } + /* Send as much data as the rate limit currently allows. */ + private function flush():void + { + if (flush_id) + clearTimeout(flush_id); + flush_id = undefined; + + if (!(s_r.connected && s_t.connected)) + /* Can't do anything until both sockets are connected. */ + return; + + while (!rate_limit.is_limited() && (p2t_schedule.length > 0 || t2p_schedule.length > 0)) { + var numBytes:uint; + var bytes:ByteArray; + + if (p2t_schedule.length > 0) { + numBytes = p2t_schedule.shift(); + bytes = new ByteArray(); + s_r.readBytes(bytes, 0, numBytes); + ui.puts("ProxyPair: RTMFP: read " + bytes.length + " bytes."); + s_t.writeBytes(bytes); + rate_limit.update(numBytes); + } + if (t2p_schedule.length > 0) { + numBytes = t2p_schedule.shift(); + bytes = new ByteArray(); + s_t.readBytes(bytes, 0, numBytes); + ui.puts("ProxyPair: Tor: read " + bytes.length + " bytes."); + s_r.writeBytes(bytes); + rate_limit.update(numBytes); + } + } + + /* Call again when safe, if necessary. */ + if (p2t_schedule.length > 0 || t2p_schedule.length > 0) + flush_id = setTimeout(flush, rate_limit.when() * 1000); + } + } +} + +import flash.utils.getTimer; + +class RateLimit +{ + public function RateLimit() + { + } + + public function update(n:Number):Boolean + { + return true; + } + + public function when():Number + { + return 0.0; + } + + public function is_limited():Boolean + { + return false; + } +} + +class RateUnlimit extends RateLimit +{ + public function RateUnlimit() + { + } + + public override function update(n:Number):Boolean + { + return true; + } + + public override function when():Number + { + return 0.0; + } + + public override function is_limited():Boolean + { + return false; + } +} + +class BucketRateLimit extends RateLimit +{ + private var amount:Number; + private var capacity:Number; + private var time:Number; + private var last_update:uint; + + public function BucketRateLimit(capacity:Number, time:Number) + { + this.amount = 0.0; + /* capacity / time is the rate we are aiming for. */ + this.capacity = capacity; + this.time = time; + this.last_update = getTimer(); + } + + private function age():void + { + var now:uint; + var delta:Number; + + now = getTimer(); + delta = (now - last_update) / 1000.0; + last_update = now; + + amount -= delta * capacity / time; + if (amount < 0.0) + amount = 0.0; + } + + public override function update(n:Number):Boolean + { + age(); + amount += n; + + return amount <= capacity; + } + + public override function when():Number + { + age(); + return (amount - capacity) / (capacity / time); + } + + public override function is_limited():Boolean + { + age(); + return amount > capacity; } } \ No newline at end of file diff --git a/rtmfp/RTMFPSocketClient.as b/rtmfp/RTMFPSocketClient.as index 7469a53..b917bcb 100644 --- a/rtmfp/RTMFPSocketClient.as +++ b/rtmfp/RTMFPSocketClient.as @@ -28,9 +28,8 @@ package rtmfp public function data_available(bytes:ByteArray):void { - this._bytes.clear(); - bytes.readBytes(this._bytes); - dispatchEvent(new ProgressEvent(ProgressEvent.SOCKET_DATA, false, false, this._bytes.bytesAvailable, this._bytes.length)); + bytes.readBytes(_bytes, _bytes.length, 0); + dispatchEvent(new ProgressEvent(ProgressEvent.SOCKET_DATA, false, false, _bytes.bytesAvailable, _bytes.bytesAvailable)); } public function get connect_acknowledged():Boolean diff --git a/rtmfpcat.as b/rtmfpcat.as index 25099dd..2273bc9 100644 --- a/rtmfpcat.as +++ b/rtmfpcat.as @@ -6,8 +6,7 @@ package import flash.text.TextField; import flash.text.TextFormat; import flash.events.Event; - import flash.utils.clearInterval; - import flash.utils.setInterval; + import flash.utils.setTimeout; import rtmfp.CirrusSocket; import rtmfp.FacilitatorSocket; @@ -39,9 +38,8 @@ package port: 9001 }; - /* Poll facilitator every 3 sec if in proxy mode and haven't found - anyone to proxy */ - private const DEFAULT_FAC_POLL_INTERVAL:uint = 3000; + /* Poll facilitator every 10 sec */ + private const DEFAULT_FAC_POLL_INTERVAL:uint = 10000; // Socket to Cirrus server private var s_c:CirrusSocket; @@ -54,8 +52,6 @@ package private var proxy_mode:Boolean; - private var fac_poll_interval:uint; - /* TextField for debug output. */ private var output_text:TextField; @@ -80,6 +76,12 @@ package // Wait until the query string parameters are loaded. this.loaderInfo.addEventListener(Event.COMPLETE, loaderinfo_complete); } + + public function puts(s:String):void + { + output_text.appendText(s + "\n"); + output_text.scrollV = output_text.maxScrollV; + } private function loaderinfo_complete(e:Event):void { @@ -133,11 +135,7 @@ package s_c = new CirrusSocket(); s_c.addEventListener(CirrusSocketEvent.CONNECT_SUCCESS, function (e:CirrusSocketEvent):void { puts("Cirrus: connected with id " + s_c.id + "."); - if (proxy_mode) { - fac_poll_interval = setInterval(establish_facilitator_connection, DEFAULT_FAC_POLL_INTERVAL); - } else { - establish_facilitator_connection(); - } + establish_facilitator_connection(); }); s_c.addEventListener(CirrusSocketEvent.CONNECT_FAILED, function (e:CirrusSocketEvent):void { puts("Error: failed to connect to Cirrus."); @@ -180,6 +178,7 @@ package }); s_f.addEventListener(FacilitatorSocketEvent.CONNECT_FAILED, function (e:Event):void { puts("Facilitator: connect failed."); + setTimeout(establish_facilitator_connection, DEFAULT_FAC_POLL_INTERVAL); }); s_f.addEventListener(FacilitatorSocketEvent.CONNECT_CLOSED, function (e:Event):void { puts("Facilitator: connect closed."); @@ -188,16 +187,17 @@ package if (proxy_mode) { s_f.addEventListener(FacilitatorSocketEvent.REGISTRATION_RECEIVED, function (e:FacilitatorSocketEvent):void { puts("Facilitator: got registration " + e.client); - clearInterval(fac_poll_interval); start_proxy_pair(); s_c.send_hello(e.client); }); s_f.addEventListener(FacilitatorSocketEvent.REGISTRATIONS_EMPTY, function (e:Event):void { puts("Facilitator: no registrations available."); + setTimeout(establish_facilitator_connection, DEFAULT_FAC_POLL_INTERVAL); }); } else { s_f.addEventListener(FacilitatorSocketEvent.REGISTRATION_FAILED, function (e:Event):void { puts("Facilitator: registration failed."); + setTimeout(establish_facilitator_connection, DEFAULT_FAC_POLL_INTERVAL); }); } s_f.connect(fac_addr.host, fac_addr.port); @@ -213,11 +213,7 @@ package p_p.addEventListener(Event.CLOSE, function (e:Event):void { puts("ProxyPair: connection closed."); p_p = null; - if (proxy_mode) { - fac_poll_interval = setInterval(establish_facilitator_connection, DEFAULT_FAC_POLL_INTERVAL); - } else { - establish_facilitator_connection(); - } + establish_facilitator_connection(); }); p_p.listen(s_c.local_stream_name); } @@ -238,11 +234,5 @@ package return addr; } - - public function puts(s:String):void - { - output_text.appendText(s + "\n"); - output_text.scrollV = output_text.maxScrollV; - } } }

1 0

[flashproxy/master] Using URLLoader in the FacilitatorSocket, Facilitator now speaks true HTTP
by dcf＠torproject.org 12 Jun '11

12 Jun '11

commit fbffb6fce47047c877d8dc4d2a83bcb4ee2926ea Author: Nate Hardison <nate(a)rescomp-09-154551.stanford.edu> Date: Thu Jun 2 19:49:49 2011 -0700 Using URLLoader in the FacilitatorSocket, Facilitator now speaks true HTTP --- facilitator.py | 55 +++++++++++++++++--- rtmfp/FacilitatorSocket.as | 121 ++++++++++++++++++++------------------------ rtmfpcat.as | 19 ++----- 3 files changed, 106 insertions(+), 89 deletions(-) diff --git a/facilitator.py b/facilitator.py index c9859db..0ea86a1 100755 --- a/facilitator.py +++ b/facilitator.py @@ -10,6 +10,7 @@ import socket import sys import threading import time +import urllib DEFAULT_ADDRESS = "0.0.0.0" DEFAULT_PORT = 9002 @@ -190,31 +191,47 @@ class Handler(BaseHTTPServer.BaseHTTPRequestHandler): def do_GET(self): log(u"proxy %s connects" % format_addr(self.client_address)) + if self.path == "/crossdomain.xml": + self.send_crossdomain() + return + + client = "" reg = REGS.fetch() if reg: log(u"proxy %s gets %s (now %d)" % (format_addr(self.client_address), unicode(reg), len(REGS))) - self.request.send(str(reg)) + client = str(reg) else: log(u"proxy %s gets none" % format_addr(self.client_address)) + client = "Registration list empty" + + response = "client=%s" % urllib.quote(client) + self.send_response(200) + self.send_header('Content-Type', 'text/html') + self.send_header('Content-Length', str(len(response))) + self.end_headers() + self.wfile.write(response) def do_POST(self): - data = self.rfile.readline(1024).strip() - try: - vals = cgi.parse_qs(data, False, True) - except ValueError, e: - log(u"client %s POST syntax error: %s" % (format_addr(self.client_address), repr(str(e)))) + data = cgi.FieldStorage(fp = self.rfile, headers = self.headers, + environ = {'REQUEST_METHOD' : 'POST', + 'CONTENT_TYPE' : self.headers['Content-Type']}) + + if self.path == "/crossdomain.xml": + self.send_crossdomain() return - client_specs = vals.get("client") - if client_specs is None or len(client_specs) != 1: + client_specs = data["client"] + if client_specs is None or client_specs.value is None: log(u"client %s missing \"client\" param" % format_addr(self.client_address)) + self.send_error(404) return - val = client_specs[0] + val = client_specs.value try: reg = Reg.parse(val, self.client_address[0]) except ValueError, e: log(u"client %s syntax error in %s: %s" % (format_addr(self.client_address), repr(val), repr(str(e)))) + self.send_error(404) return log(u"client %s regs %s -> %s" % (format_addr(self.client_address), val, unicode(reg))) @@ -222,10 +239,30 @@ class Handler(BaseHTTPServer.BaseHTTPRequestHandler): log(u"client %s %s (now %d)" % (format_addr(self.client_address), unicode(reg), len(REGS))) else: log(u"client %s %s (already present, now %d)" % (format_addr(self.client_address), unicode(reg), len(REGS))) + + response = "" + + self.send_response(200) + self.send_header('Content-Type', 'text/html') + self.send_header('Content-Length', str(len(response))) + self.send_header('Connection', 'close') + self.end_headers() + self.wfile.write(response) 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_crossdomain(self): + crossdomain = """\ +<cross-domain-policy> + <allow-access-from domain="*" to-ports="%s"/> +</cross-domain-policy>\r\n""" % (address[1]) + self.send_response(200) + self.send_header('Content-Type', 'application/xml') + self.send_header('Content-Length', str(len(crossdomain))) + self.end_headers() + self.wfile.write(crossdomain) REGS = RegSet() diff --git a/rtmfp/FacilitatorSocket.as b/rtmfp/FacilitatorSocket.as index e175290..32c8a7d 100644 --- a/rtmfp/FacilitatorSocket.as +++ b/rtmfp/FacilitatorSocket.as @@ -1,108 +1,94 @@ package rtmfp { - import flash.net.Socket; import flash.events.Event; import flash.events.EventDispatcher; + import flash.events.HTTPStatusEvent; import flash.events.IOErrorEvent; - import flash.events.NetStatusEvent; - import flash.events.ProgressEvent; import flash.events.SecurityErrorEvent; - import flash.utils.clearInterval; - import flash.utils.setInterval; + import flash.net.URLLoader; + import flash.net.URLLoaderDataFormat; + import flash.net.URLRequest; + import flash.net.URLRequestMethod; + import flash.net.URLVariables; + import flash.system.Security; import rtmfp.events.FacilitatorSocketEvent; - [Event(name=FacilitatorSocketEvent.CONNECT_CLOSED, type="com.flashproxy.rtmfp.events.FacilitatorSocketEvent")] [Event(name=FacilitatorSocketEvent.CONNECT_FAILED, type="com.flashproxy.rtmfp.events.FacilitatorSocketEvent")] - [Event(name=FacilitatorSocketEvent.CONNECT_SUCCESS, type="com.flashproxy.rtmfp.events.FacilitatorSocketEvent")] [Event(name=FacilitatorSocketEvent.REGISTRATION_FAILED, type="com.flashproxy.rtmfp.events.FacilitatorSocketEvent")] [Event(name=FacilitatorSocketEvent.REGISTRATION_RECEIVED, type="com.flashproxy.rtmfp.events.FacilitatorSocketEvent")] [Event(name=FacilitatorSocketEvent.REGISTRATIONS_EMPTY, type="com.flashproxy.rtmfp.events.FacilitatorSocketEvent")] public class FacilitatorSocket extends EventDispatcher { - private var socket:Socket; - private var connected:Boolean; - private var connection_timeout:uint; + private var host:String; + private var port:uint; - public function FacilitatorSocket() - { - socket = null; - connected = false; - } - - public function close():void - { - connected = false; - if (socket != null) { - socket.removeEventListener(Event.CONNECT, on_connect_event); - socket.removeEventListener(Event.CLOSE, on_close_event); - socket.removeEventListener(IOErrorEvent.IO_ERROR, on_io_error_event); - socket.removeEventListener(ProgressEvent.SOCKET_DATA, on_progress_event); - socket.removeEventListener(SecurityErrorEvent.SECURITY_ERROR, on_security_error_event); - if (connected) { - socket.close(); - } - } - } - - public function connect(host:String, port:uint):void - { - if (socket != null || connected) { - return; - } - - socket = new Socket(); - socket.addEventListener(Event.CONNECT, on_connect_event); - socket.addEventListener(Event.CLOSE, on_close_event); - socket.addEventListener(IOErrorEvent.IO_ERROR, on_io_error_event); - socket.addEventListener(ProgressEvent.SOCKET_DATA, on_progress_event); - socket.addEventListener(SecurityErrorEvent.SECURITY_ERROR, on_security_error_event); - socket.connect(host, port); + public function FacilitatorSocket(host:String, port:uint) + { + this.host = host; + this.port = port; } public function get_registration():void { - if (!connected) return; - socket.writeUTFBytes("GET / HTTP/1.0\r\n\r\n"); + make_request(URLRequestMethod.GET); } public function post_registration(registration_data:String):void - { - if (!connected) return; - socket.writeUTFBytes("POST / HTTP/1.0\r\n\r\nclient=" + registration_data + "\r\n"); + { + var data:URLVariables = new URLVariables(); + data.client = registration_data; + make_request(URLRequestMethod.POST, data); } private function fail():void { - clearInterval(connection_timeout); dispatchEvent(new FacilitatorSocketEvent(FacilitatorSocketEvent.CONNECT_FAILED)); } - private function on_close_event(event:Event):void + private function make_request(method:String, data:URLVariables = null):void { - close(); - dispatchEvent(new FacilitatorSocketEvent(FacilitatorSocketEvent.CONNECT_CLOSED)); + var request:URLRequest = new URLRequest(url) + request.data = data; + request.method = method; + + var url_loader:URLLoader = new URLLoader(); + url_loader = new URLLoader(); + url_loader.dataFormat = URLLoaderDataFormat.VARIABLES; + + url_loader.addEventListener(Event.COMPLETE, on_complete_event); + url_loader.addEventListener(HTTPStatusEvent.HTTP_STATUS, on_http_status_event); + url_loader.addEventListener(SecurityErrorEvent.SECURITY_ERROR, on_security_error_event); + url_loader.addEventListener(IOErrorEvent.IO_ERROR, on_io_error_event); + + url_loader.load(request); } - private function on_connect_event(event:Event):void + private function on_complete_event(event:Event):void { - connected = true; - dispatchEvent(new FacilitatorSocketEvent(FacilitatorSocketEvent.CONNECT_SUCCESS)); + try { + var client_id:String = event.target.data.client; + if (client_id == "Registration list empty") { + dispatchEvent(new FacilitatorSocketEvent(FacilitatorSocketEvent.REGISTRATIONS_EMPTY)); + } else { + dispatchEvent(new FacilitatorSocketEvent(FacilitatorSocketEvent.REGISTRATION_RECEIVED, client_id)); + } + } catch (e:Error) { + /* error is thrown for POST when we don't care about + the response anyways */ + } + + event.target.close() } - private function on_io_error_event(event:IOErrorEvent):void + private function on_http_status_event(event:HTTPStatusEvent):void { - fail(); + /* empty for now */ } - private function on_progress_event(event:ProgressEvent):void + private function on_io_error_event(event:IOErrorEvent):void { - var client_id:String = socket.readUTFBytes(event.bytesLoaded); - if (client_id == "Registration list empty") { - dispatchEvent(new FacilitatorSocketEvent(FacilitatorSocketEvent.REGISTRATIONS_EMPTY)); - } else { - dispatchEvent(new FacilitatorSocketEvent(FacilitatorSocketEvent.REGISTRATION_RECEIVED, client_id)); - } + fail(); } private function on_security_error_event(event:SecurityErrorEvent):void @@ -110,6 +96,9 @@ package rtmfp fail(); } - + private function get url():String + { + return "http://" + host + ":" + port; + } } } \ No newline at end of file diff --git a/rtmfpcat.as b/rtmfpcat.as index 713fc45..224fe56 100644 --- a/rtmfpcat.as +++ b/rtmfpcat.as @@ -166,23 +166,11 @@ package private function establish_facilitator_connection():void { - s_f = new FacilitatorSocket(); - s_f.addEventListener(FacilitatorSocketEvent.CONNECT_SUCCESS, function (e:Event):void { - if (proxy_mode) { - puts("Facilitator: getting registration."); - s_f.get_registration(); - } else { - puts("Facilitator: posting registration."); - s_f.post_registration(s_c.id); - } - }); + s_f = new FacilitatorSocket(fac_addr.host, fac_addr.port); s_f.addEventListener(FacilitatorSocketEvent.CONNECT_FAILED, function (e:Event):void { puts("Facilitator: connect failed."); setTimeout(establish_facilitator_connection, DEFAULT_FAC_POLL_INTERVAL); }); - s_f.addEventListener(FacilitatorSocketEvent.CONNECT_CLOSED, function (e:Event):void { - puts("Facilitator: connect closed."); - }); if (proxy_mode) { s_f.addEventListener(FacilitatorSocketEvent.REGISTRATION_RECEIVED, function (e:FacilitatorSocketEvent):void { @@ -194,13 +182,16 @@ package puts("Facilitator: no registrations available."); setTimeout(establish_facilitator_connection, DEFAULT_FAC_POLL_INTERVAL); }); + puts("Facilitator: getting registration."); + s_f.get_registration(); } else { s_f.addEventListener(FacilitatorSocketEvent.REGISTRATION_FAILED, function (e:Event):void { puts("Facilitator: registration failed."); setTimeout(establish_facilitator_connection, DEFAULT_FAC_POLL_INTERVAL); }); + puts("Facilitator: posting registration."); + s_f.post_registration(s_c.id); } - s_f.connect(fac_addr.host, fac_addr.port); } private function start_proxy_pair():void

1 0

[flashproxy/master] Changing terminology to use relay/proxy instead of Tor/RTMFP
by dcf＠torproject.org 12 Jun '11

12 Jun '11

commit 3a8df84e3bb8dcc16d411a4c2fb9f4e005f313c8 Author: Nate Hardison <nate(a)rescomp-09-154551.stanford.edu> Date: Thu Jun 2 13:47:34 2011 -0700 Changing terminology to use relay/proxy instead of Tor/RTMFP --- rtmfp/ProxyPair.as | 112 ++++++++++++++++++++++++++-------------------------- rtmfpcat.as | 26 ++++++------ 2 files changed, 69 insertions(+), 69 deletions(-) diff --git a/rtmfp/ProxyPair.as b/rtmfp/ProxyPair.as index fb184cc..159abde 100644 --- a/rtmfp/ProxyPair.as +++ b/rtmfp/ProxyPair.as @@ -18,126 +18,126 @@ package rtmfp { private var ui:rtmfpcat; - private var s_r:RTMFPSocket; - private var s_t:Socket; + private var s_p:RTMFPSocket; + private var s_r:Socket; - private var tor_host:String; - private var tor_port:uint; + private var relay_host:String; + private var relay_port:uint; - private var p2t_schedule:Array; - private var t2p_schedule:Array; + private var p2r_schedule:Array; + private var r2p_schedule:Array; // Bytes per second. Set to undefined to disable limit. public const RATE_LIMIT:Number = 10000; // Seconds. - private const RATE_LIMIT_HISTORY:Number = 5.0; + private const RATE_LIMIT_HISrelayY:Number = 5.0; private var rate_limit:RateLimit; // Callback id. private var flush_id:uint; - public function ProxyPair(ui:rtmfpcat, s_c:CirrusSocket, tor_host:String, tor_port:uint) + public function ProxyPair(ui:rtmfpcat, s_c:CirrusSocket, relay_host:String, relay_port:uint) { this.ui = ui; - this.tor_host = tor_host; - this.tor_port = tor_port; + this.relay_host = relay_host; + this.relay_port = relay_port; - this.p2t_schedule = new Array(); - this.t2p_schedule = new Array(); + this.p2r_schedule = new Array(); + this.r2p_schedule = new Array(); if (RATE_LIMIT) - rate_limit = new BucketRateLimit(RATE_LIMIT * RATE_LIMIT_HISTORY, RATE_LIMIT_HISTORY); + rate_limit = new BucketRateLimit(RATE_LIMIT * RATE_LIMIT_HISrelayY, RATE_LIMIT_HISrelayY); else rate_limit = new RateUnlimit(); setup_rtmfp_socket(s_c); - setup_tor_socket(); + setup_relay_socket(); } public function close():void { + if (s_p.connected) { + s_p.close(); + } if (s_r.connected) { s_r.close(); } - if (s_t.connected) { - s_t.close(); - } dispatchEvent(new Event(Event.CLOSE)); } public function connect(peer:String, stream:String):void { - s_r.connect(peer, stream); + s_p.connect(peer, stream); } public function get connected():Boolean { - return (s_r.connected && s_t.connected); + return (s_p.connected && s_r.connected); } public function listen(stream:String):void { - s_r.listen(stream); + s_p.listen(stream); } private function setup_rtmfp_socket(s_c:CirrusSocket):void { - s_r = new RTMFPSocket(s_c); - s_r.addEventListener(RTMFPSocketEvent.CONNECT_FAILED, function (e:RTMFPSocketEvent):void { + s_p = new RTMFPSocket(s_c); + s_p.addEventListener(RTMFPSocketEvent.CONNECT_FAILED, function (e:RTMFPSocketEvent):void { ui.puts("Peering failed."); }); - s_r.addEventListener(RTMFPSocketEvent.CONNECT_SUCCESS, function (e:RTMFPSocketEvent):void { + s_p.addEventListener(RTMFPSocketEvent.CONNECT_SUCCESS, function (e:RTMFPSocketEvent):void { ui.puts("Peering success."); - s_t.connect(tor_host, tor_port); + s_r.connect(relay_host, relay_port); }); - s_r.addEventListener(RTMFPSocketEvent.PEER_CONNECTED, function (e:RTMFPSocketEvent):void { + s_p.addEventListener(RTMFPSocketEvent.PEER_CONNECTED, function (e:RTMFPSocketEvent):void { ui.puts("Peer connected."); }); - s_r.addEventListener(RTMFPSocketEvent.PEER_DISCONNECTED, function (e:RTMFPSocketEvent):void { + s_p.addEventListener(RTMFPSocketEvent.PEER_DISCONNECTED, function (e:RTMFPSocketEvent):void { ui.puts("Peer disconnected."); close(); }); - s_r.addEventListener(RTMFPSocketEvent.PLAY_STARTED, function (e:RTMFPSocketEvent):void { + s_p.addEventListener(RTMFPSocketEvent.PLAY_STARTED, function (e:RTMFPSocketEvent):void { ui.puts("Play started."); }); - s_r.addEventListener(RTMFPSocketEvent.PUBLISH_STARTED, function (e:RTMFPSocketEvent):void { + s_p.addEventListener(RTMFPSocketEvent.PUBLISH_STARTED, function (e:RTMFPSocketEvent):void { ui.puts("Publishing started."); }); - s_r.addEventListener(ProgressEvent.SOCKET_DATA, proxy_to_tor); + s_p.addEventListener(ProgressEvent.SOCKET_DATA, proxy_to_relay); } - private function setup_tor_socket():void + private function setup_relay_socket():void { - s_t = new Socket(); - s_t.addEventListener(Event.CONNECT, function (e:Event):void { - ui.puts("Tor: connected to " + tor_host + ":" + tor_port + "."); + s_r = new Socket(); + s_r.addEventListener(Event.CONNECT, function (e:Event):void { + ui.puts("Relay: connected to " + relay_host + ":" + relay_port + "."); dispatchEvent(new Event(Event.CONNECT)); }); - s_t.addEventListener(Event.CLOSE, function (e:Event):void { - ui.puts("Tor: closed connection."); + s_r.addEventListener(Event.CLOSE, function (e:Event):void { + ui.puts("Relay: closed connection."); close(); }); - s_t.addEventListener(IOErrorEvent.IO_ERROR, function (e:IOErrorEvent):void { - ui.puts("Tor: I/O error: " + e.text + "."); + s_r.addEventListener(IOErrorEvent.IO_ERROR, function (e:IOErrorEvent):void { + ui.puts("Relay: I/O error: " + e.text + "."); close(); }); - s_t.addEventListener(SecurityErrorEvent.SECURITY_ERROR, function (e:SecurityErrorEvent):void { - ui.puts("Tor: security error: " + e.text + "."); + s_r.addEventListener(SecurityErrorEvent.SECURITY_ERROR, function (e:SecurityErrorEvent):void { + ui.puts("Relay: security error: " + e.text + "."); close(); }); - s_t.addEventListener(ProgressEvent.SOCKET_DATA, tor_to_proxy); + s_r.addEventListener(ProgressEvent.SOCKET_DATA, relay_to_proxy); } - private function tor_to_proxy(e:ProgressEvent):void + private function relay_to_proxy(e:ProgressEvent):void { - t2p_schedule.push(e.bytesLoaded); + r2p_schedule.push(e.bytesLoaded); flush(); } - private function proxy_to_tor(e:ProgressEvent):void + private function proxy_to_relay(e:ProgressEvent):void { - p2t_schedule.push(e.bytesLoaded); + p2r_schedule.push(e.bytesLoaded); flush(); } @@ -148,34 +148,34 @@ package rtmfp clearTimeout(flush_id); flush_id = undefined; - if (!(s_r.connected && s_t.connected)) + if (!(s_p.connected && s_r.connected)) /* Can't do anything until both sockets are connected. */ return; - while (!rate_limit.is_limited() && (p2t_schedule.length > 0 || t2p_schedule.length > 0)) { + while (!rate_limit.is_limited() && (p2r_schedule.length > 0 || r2p_schedule.length > 0)) { var numBytes:uint; var bytes:ByteArray; - if (p2t_schedule.length > 0) { - numBytes = p2t_schedule.shift(); + if (p2r_schedule.length > 0) { + numBytes = p2r_schedule.shift(); bytes = new ByteArray(); - s_r.readBytes(bytes, 0, numBytes); + s_p.readBytes(bytes, 0, numBytes); ui.puts("ProxyPair: RTMFP: read " + bytes.length + " bytes."); - s_t.writeBytes(bytes); + s_r.writeBytes(bytes); rate_limit.update(numBytes); } - if (t2p_schedule.length > 0) { - numBytes = t2p_schedule.shift(); + if (r2p_schedule.length > 0) { + numBytes = r2p_schedule.shift(); bytes = new ByteArray(); - s_t.readBytes(bytes, 0, numBytes); - ui.puts("ProxyPair: Tor: read " + bytes.length + " bytes."); - s_r.writeBytes(bytes); + s_r.readBytes(bytes, 0, numBytes); + ui.puts("ProxyPair: Relay: read " + bytes.length + " bytes."); + s_p.writeBytes(bytes); rate_limit.update(numBytes); } } /* Call again when safe, if necessary. */ - if (p2t_schedule.length > 0 || t2p_schedule.length > 0) + if (p2r_schedule.length > 0 || r2p_schedule.length > 0) flush_id = setTimeout(flush, rate_limit.when() * 1000); } } diff --git a/rtmfpcat.as b/rtmfpcat.as index 2273bc9..713fc45 100644 --- a/rtmfpcat.as +++ b/rtmfpcat.as @@ -33,7 +33,7 @@ package /* David's relay (nickname 3VXRyxz67OeRoqHn) that also serves a crossdomain policy. */ - private const DEFAULT_TOR_PROXY_ADDR:Object = { + private const DEFAULT_TOR_RELAY_ADDR:Object = { host: "173.255.221.44", port: 9001 }; @@ -56,7 +56,7 @@ package private var output_text:TextField; private var fac_addr:Object; - private var tor_addr:Object; + private var relay_addr:Object; public function rtmfpcat() { @@ -86,7 +86,7 @@ package private function loaderinfo_complete(e:Event):void { var fac_spec:String; - var tor_spec:String; + var relay_spec:String; puts("Parameters loaded."); @@ -104,20 +104,20 @@ package } else { fac_addr = DEFAULT_FACILITATOR_ADDR; } - - tor_spec = this.loaderInfo.parameters["tor"]; - if (tor_spec) { - puts("Tor spec: \"" + tor_spec + "\""); - tor_addr = parse_addr_spec(tor_spec); - if (!tor_addr) { - puts("Error: Tor spec must be in the form \"host:port\"."); + + relay_spec = this.loaderInfo.parameters["relay"]; + if (relay_spec) { + puts("Relay spec: \"" + relay_spec + "\""); + relay_addr = parse_addr_spec(relay_spec); + if (!relay_addr) { + puts("Error: Relay spec must be in the form \"host:port\"."); return; } } else { if (proxy_mode) { - tor_addr = DEFAULT_TOR_PROXY_ADDR; + relay_addr = DEFAULT_TOR_RELAY_ADDR; } else { - tor_addr = DEFAULT_TOR_CLIENT_ADDR; + relay_addr = DEFAULT_TOR_CLIENT_ADDR; } } @@ -206,7 +206,7 @@ package private function start_proxy_pair():void { puts("Starting proxy pair on stream " + s_c.local_stream_name); - p_p = new ProxyPair(this, s_c, tor_addr.host, tor_addr.port); + p_p = new ProxyPair(this, s_c, relay_addr.host, relay_addr.port); p_p.addEventListener(Event.CONNECT, function (e:Event):void { puts("ProxyPair: connected!"); });

1 0