[tor-commits] [snowflake/master] Start refactoring out a client and library
arlo at torproject.org
arlo at torproject.org
Thu Nov 22 16:28:38 UTC 2018
commit cce7ee64a77b89b8af6760a0e0a0682d2c9331b7
Author: Arlo Breault <arlolra at gmail.com>
Date: Tue Nov 20 22:17:24 2018 -0500
Start refactoring out a client and library
---
client/{ => lib}/interfaces.go | 2 +-
client/{client_test.go => lib/lib_test.go} | 4 +-
client/{ => lib}/peers.go | 2 +-
client/{ => lib}/rendezvous.go | 2 +-
client/lib/snowflake.go | 69 ++++++++++++++++++++++
client/{ => lib}/util.go | 50 ++++++++--------
client/{ => lib}/webrtc.go | 2 +-
client/snowflake.go | 94 +++++++-----------------------
8 files changed, 120 insertions(+), 105 deletions(-)
diff --git a/client/interfaces.go b/client/lib/interfaces.go
similarity index 98%
rename from client/interfaces.go
rename to client/lib/interfaces.go
index f18987a..f62d4f5 100644
--- a/client/interfaces.go
+++ b/client/lib/interfaces.go
@@ -1,4 +1,4 @@
-package main
+package lib
import (
"io"
diff --git a/client/client_test.go b/client/lib/lib_test.go
similarity index 99%
rename from client/client_test.go
rename to client/lib/lib_test.go
index cfc8cbf..5a9a2e5 100644
--- a/client/client_test.go
+++ b/client/lib/lib_test.go
@@ -1,4 +1,4 @@
-package main
+package lib
import (
"bytes"
@@ -179,7 +179,7 @@ func TestSnowflakeClient(t *testing.T) {
So(socks.rejected, ShouldEqual, false)
snowflakes.toRelease = nil
- handler(socks, snowflakes)
+ Handler(socks, snowflakes)
So(socks.rejected, ShouldEqual, true)
})
diff --git a/client/peers.go b/client/lib/peers.go
similarity index 99%
rename from client/peers.go
rename to client/lib/peers.go
index 3187f09..21411ed 100644
--- a/client/peers.go
+++ b/client/lib/peers.go
@@ -1,4 +1,4 @@
-package main
+package lib
import (
"container/list"
diff --git a/client/rendezvous.go b/client/lib/rendezvous.go
similarity index 99%
rename from client/rendezvous.go
rename to client/lib/rendezvous.go
index cab7f5a..7436f54 100644
--- a/client/rendezvous.go
+++ b/client/lib/rendezvous.go
@@ -9,7 +9,7 @@
//
// - Manual copy-paste signaling. User must create a signaling pipe.
// (The flags in torrc-manual allow this)
-package main
+package lib
import (
"bufio"
diff --git a/client/lib/snowflake.go b/client/lib/snowflake.go
new file mode 100644
index 0000000..900af88
--- /dev/null
+++ b/client/lib/snowflake.go
@@ -0,0 +1,69 @@
+package lib
+
+import (
+ "errors"
+ "io"
+ "log"
+ "net"
+ "sync"
+)
+
+const (
+ ReconnectTimeout = 10
+ SnowflakeTimeout = 30
+)
+
+// When a connection handler starts, +1 is written to this channel; when it
+// ends, -1 is written.
+var HandlerChan = make(chan int)
+
+// Given an accepted SOCKS connection, establish a WebRTC connection to the
+// remote peer and exchange traffic.
+func Handler(socks SocksConnector, snowflakes SnowflakeCollector) error {
+ HandlerChan <- 1
+ defer func() {
+ HandlerChan <- -1
+ }()
+ // Obtain an available WebRTC remote. May block.
+ snowflake := snowflakes.Pop()
+ if nil == snowflake {
+ socks.Reject()
+ return errors.New("handler: Received invalid Snowflake")
+ }
+ defer socks.Close()
+ defer snowflake.Close()
+ log.Println("---- Handler: snowflake assigned ----")
+ err := socks.Grant(&net.TCPAddr{IP: net.IPv4zero, Port: 0})
+ if err != nil {
+ return err
+ }
+
+ go func() {
+ // When WebRTC resets, close the SOCKS connection too.
+ snowflake.WaitForReset()
+ socks.Close()
+ }()
+
+ // Begin exchanging data. Either WebRTC or localhost SOCKS will close first.
+ // In eithercase, this closes the handler and induces a new handler.
+ copyLoop(socks, snowflake)
+ log.Println("---- Handler: closed ---")
+ return nil
+}
+
+// Exchanges bytes between two ReadWriters.
+// (In this case, between a SOCKS and WebRTC connection.)
+func copyLoop(a, b io.ReadWriter) {
+ var wg sync.WaitGroup
+ wg.Add(2)
+ go func() {
+ io.Copy(b, a)
+ wg.Done()
+ }()
+ go func() {
+ io.Copy(a, b)
+ wg.Done()
+ }()
+ wg.Wait()
+ log.Println("copy loop ended")
+}
diff --git a/client/util.go b/client/lib/util.go
similarity index 69%
rename from client/util.go
rename to client/lib/util.go
index 20817c3..028fb1c 100644
--- a/client/util.go
+++ b/client/lib/util.go
@@ -1,4 +1,4 @@
-package main
+package lib
import (
"fmt"
@@ -34,46 +34,46 @@ func (b BytesNullLogger) AddInbound(amount int) {}
// BytesSyncLogger uses channels to safely log from multiple sources with output
// occuring at reasonable intervals.
type BytesSyncLogger struct {
- outboundChan chan int
- inboundChan chan int
- outbound int
- inbound int
- outEvents int
- inEvents int
- isLogging bool
+ OutboundChan chan int
+ InboundChan chan int
+ Outbound int
+ Inbound int
+ OutEvents int
+ InEvents int
+ IsLogging bool
}
func (b *BytesSyncLogger) Log() {
- b.isLogging = true
+ b.IsLogging = true
var amount int
output := func() {
log.Printf("Traffic Bytes (in|out): %d | %d -- (%d OnMessages, %d Sends)",
- b.inbound, b.outbound, b.inEvents, b.outEvents)
- b.outbound = 0
- b.outEvents = 0
- b.inbound = 0
- b.inEvents = 0
+ b.Inbound, b.Outbound, b.InEvents, b.OutEvents)
+ b.Outbound = 0
+ b.OutEvents = 0
+ b.Inbound = 0
+ b.InEvents = 0
}
last := time.Now()
for {
select {
- case amount = <-b.outboundChan:
- b.outbound += amount
- b.outEvents++
+ case amount = <-b.OutboundChan:
+ b.Outbound += amount
+ b.OutEvents++
last := time.Now()
if time.Since(last) > time.Second*LogTimeInterval {
last = time.Now()
output()
}
- case amount = <-b.inboundChan:
- b.inbound += amount
- b.inEvents++
+ case amount = <-b.InboundChan:
+ b.Inbound += amount
+ b.InEvents++
if time.Since(last) > time.Second*LogTimeInterval {
last = time.Now()
output()
}
case <-time.After(time.Second * LogTimeInterval):
- if b.inEvents > 0 || b.outEvents > 0 {
+ if b.InEvents > 0 || b.OutEvents > 0 {
output()
}
}
@@ -81,15 +81,15 @@ func (b *BytesSyncLogger) Log() {
}
func (b *BytesSyncLogger) AddOutbound(amount int) {
- if !b.isLogging {
+ if !b.IsLogging {
return
}
- b.outboundChan <- amount
+ b.OutboundChan <- amount
}
func (b *BytesSyncLogger) AddInbound(amount int) {
- if !b.isLogging {
+ if !b.IsLogging {
return
}
- b.inboundChan <- amount
+ b.InboundChan <- amount
}
diff --git a/client/webrtc.go b/client/lib/webrtc.go
similarity index 99%
rename from client/webrtc.go
rename to client/lib/webrtc.go
index 8c7cb4c..e71a407 100644
--- a/client/webrtc.go
+++ b/client/lib/webrtc.go
@@ -1,4 +1,4 @@
-package main
+package lib
import (
"bytes"
diff --git a/client/snowflake.go b/client/snowflake.go
index a9841be..b2dea5c 100644
--- a/client/snowflake.go
+++ b/client/snowflake.go
@@ -2,7 +2,6 @@
package main
import (
- "errors"
"flag"
"io"
"io/ioutil"
@@ -12,36 +11,30 @@ import (
"os/signal"
"path/filepath"
"strings"
- "sync"
"syscall"
"time"
"git.torproject.org/pluggable-transports/goptlib.git"
+ sf "git.torproject.org/pluggable-transports/snowflake.git/client/lib"
"github.com/keroserene/go-webrtc"
)
const (
- ReconnectTimeout = 10
DefaultSnowflakeCapacity = 1
- SnowflakeTimeout = 30
)
-// When a connection handler starts, +1 is written to this channel; when it
-// ends, -1 is written.
-var handlerChan = make(chan int)
-
// Maintain |SnowflakeCapacity| number of available WebRTC connections, to
// transfer to the Tor SOCKS handler when needed.
-func ConnectLoop(snowflakes SnowflakeCollector) {
+func ConnectLoop(snowflakes sf.SnowflakeCollector) {
for {
// Check if ending is necessary.
_, err := snowflakes.Collect()
if nil != err {
log.Println("WebRTC:", err,
- " Retrying in", ReconnectTimeout, "seconds...")
+ " Retrying in", sf.ReconnectTimeout, "seconds...")
}
select {
- case <-time.After(time.Second * ReconnectTimeout):
+ case <-time.After(time.Second * sf.ReconnectTimeout):
continue
case <-snowflakes.Melted():
log.Println("ConnectLoop: stopped.")
@@ -51,7 +44,7 @@ func ConnectLoop(snowflakes SnowflakeCollector) {
}
// Accept local SOCKS connections and pass them to the handler.
-func socksAcceptLoop(ln *pt.SocksListener, snowflakes SnowflakeCollector) error {
+func socksAcceptLoop(ln *pt.SocksListener, snowflakes sf.SnowflakeCollector) error {
defer ln.Close()
log.Println("Started SOCKS listener.")
for {
@@ -64,64 +57,13 @@ func socksAcceptLoop(ln *pt.SocksListener, snowflakes SnowflakeCollector) error
return err
}
log.Println("SOCKS accepted: ", conn.Req)
- err = handler(conn, snowflakes)
+ err = sf.Handler(conn, snowflakes)
if err != nil {
log.Printf("handler error: %s", err)
}
}
}
-// Given an accepted SOCKS connection, establish a WebRTC connection to the
-// remote peer and exchange traffic.
-func handler(socks SocksConnector, snowflakes SnowflakeCollector) error {
- handlerChan <- 1
- defer func() {
- handlerChan <- -1
- }()
- // Obtain an available WebRTC remote. May block.
- snowflake := snowflakes.Pop()
- if nil == snowflake {
- socks.Reject()
- return errors.New("handler: Received invalid Snowflake")
- }
- defer socks.Close()
- defer snowflake.Close()
- log.Println("---- Handler: snowflake assigned ----")
- err := socks.Grant(&net.TCPAddr{IP: net.IPv4zero, Port: 0})
- if err != nil {
- return err
- }
-
- go func() {
- // When WebRTC resets, close the SOCKS connection too.
- snowflake.WaitForReset()
- socks.Close()
- }()
-
- // Begin exchanging data. Either WebRTC or localhost SOCKS will close first.
- // In eithercase, this closes the handler and induces a new handler.
- copyLoop(socks, snowflake)
- log.Println("---- Handler: closed ---")
- return nil
-}
-
-// Exchanges bytes between two ReadWriters.
-// (In this case, between a SOCKS and WebRTC connection.)
-func copyLoop(a, b io.ReadWriter) {
- var wg sync.WaitGroup
- wg.Add(2)
- go func() {
- io.Copy(b, a)
- wg.Done()
- }()
- go func() {
- io.Copy(a, b)
- wg.Done()
- }()
- wg.Wait()
- log.Println("copy loop ended")
-}
-
func main() {
iceServersCommas := flag.String("ice", "", "comma-separated list of ICE servers")
brokerURL := flag.String("url", "", "URL of signaling broker")
@@ -156,30 +98,34 @@ func main() {
log.Println("\n\n\n --- Starting Snowflake Client ---")
- var iceServers IceServerList
+ var iceServers sf.IceServerList
if len(strings.TrimSpace(*iceServersCommas)) > 0 {
option := webrtc.OptionIceServer(*iceServersCommas)
iceServers = append(iceServers, option)
}
// Prepare to collect remote WebRTC peers.
- snowflakes := NewPeers(*max)
+ snowflakes := sf.NewPeers(*max)
if "" != *brokerURL {
// Use potentially domain-fronting broker to rendezvous.
- broker := NewBrokerChannel(*brokerURL, *frontDomain, CreateBrokerTransport())
- snowflakes.Tongue = NewWebRTCDialer(broker, iceServers)
+ broker := sf.NewBrokerChannel(*brokerURL, *frontDomain, sf.CreateBrokerTransport())
+ snowflakes.Tongue = sf.NewWebRTCDialer(broker, iceServers)
} else {
// Otherwise, use manual copy and pasting of SDP messages.
- snowflakes.Tongue = NewCopyPasteDialer(iceServers)
+ snowflakes.Tongue = sf.NewCopyPasteDialer(iceServers)
}
if nil == snowflakes.Tongue {
log.Fatal("Unable to prepare rendezvous method.")
return
}
// Use a real logger to periodically output how much traffic is happening.
- snowflakes.BytesLogger = &BytesSyncLogger{
- inboundChan: make(chan int, 5), outboundChan: make(chan int, 5),
- inbound: 0, outbound: 0, inEvents: 0, outEvents: 0,
+ snowflakes.BytesLogger = &sf.BytesSyncLogger{
+ InboundChan: make(chan int, 5),
+ OutboundChan: make(chan int, 5),
+ Inbound: 0,
+ Outbound: 0,
+ InEvents: 0,
+ OutEvents: 0,
}
go snowflakes.BytesLogger.Log()
@@ -232,7 +178,7 @@ func main() {
sig = nil
for sig == nil {
select {
- case n := <-handlerChan:
+ case n := <-sf.HandlerChan:
numHandlers += n
case sig = <-sigChan:
}
@@ -244,7 +190,7 @@ func main() {
}
snowflakes.End()
for numHandlers > 0 {
- numHandlers += <-handlerChan
+ numHandlers += <-sf.HandlerChan
}
log.Println("snowflake is done.")
}
More information about the tor-commits
mailing list