commit 760dee8a0f3efc1a71780bfde277ae7a5a7a6d9b Author: Serene Han <keroserene+git@gmail.com> Date: Wed Feb 17 17:39:09 2016 -0800 prepare snowflake client for buffered datachannel writes, separate out dialWebRTC (#12) --- client/snowflake.go | 196 ++++++++++++++++++++++++++++++---------------------- 1 file changed, 115 insertions(+), 81 deletions(-) diff --git a/client/snowflake.go b/client/snowflake.go index ba1561e..7c47fbb 100644 --- a/client/snowflake.go +++ b/client/snowflake.go @@ -28,7 +28,7 @@ var frontDomain string // ends, -1 is written. var handlerChan = make(chan int) -var signalChan = make(chan *webrtc.SessionDescription) +var answerChannel = make(chan *webrtc.SessionDescription) func copyLoop(a, b net.Conn) { var wg sync.WaitGroup @@ -46,10 +46,16 @@ func copyLoop(a, b net.Conn) { wg.Wait() } +// Implements net.Conn interface type webRTCConn struct { - pc *webrtc.PeerConnection - dc *webrtc.DataChannel - recvPipe *io.PipeReader + pc *webrtc.PeerConnection + dc *webrtc.DataChannel + broker *BrokerChannel + recvPipe *io.PipeReader + writePipe *io.PipeWriter + offerChannel chan *webrtc.SessionDescription + errorChannel chan error + openChannel chan struct{} } var webrtcRemote *webRTCConn @@ -61,6 +67,7 @@ func (c *webRTCConn) Read(b []byte) (int, error) { func (c *webRTCConn) Write(b []byte) (int, error) { // log.Printf("webrtc Write %d %+q", len(b), string(b)) log.Printf("Write %d bytes --> WebRTC", len(b)) + // Buffer in case datachannel isn't available. c.dc.Send(b) return len(b), nil } @@ -90,18 +97,87 @@ func (c *webRTCConn) SetWriteDeadline(t time.Time) error { return fmt.Errorf("SetWriteDeadline not implemented") } -func dialWebRTC(config *webrtc.Configuration, broker *BrokerChannel) ( - *webRTCConn, error) { +// Create a WebRTC DataChannel locally. +// This triggers "OnNegotiationNeeded" which should prepare an SDP offer. +func (c *webRTCConn) EstablishDataChannel() error { + dc, err := c.pc.CreateDataChannel("snowflake", webrtc.Init{}) + if err != nil { + log.Printf("CreateDataChannel: %s", err) + return err + } + dc.OnOpen = func() { + log.Println("OnOpen channel") + c.openChannel <- struct{}{} + } + dc.OnClose = func() { + log.Println("OnClose channel") + // writePipe.Close() + close(c.openChannel) + // TODO: (Issue #12) Should attempt to renegotiate at this point. + } + dc.OnMessage = func(msg []byte) { + log.Printf("OnMessage <--- %d bytes", len(msg)) + n, err := c.writePipe.Write(msg) + if err != nil { + // TODO: Maybe shouldn't actually close. + c.writePipe.CloseWithError(err) + } + if n != len(msg) { + panic("short write") + } + } + c.dc = dc + return nil +} - offerChan := make(chan *webrtc.SessionDescription) - errChan := make(chan error) - openChan := make(chan struct{}) +// Block until an offer is available, then send it to either +// the Broker or signal pipe. +func (c *webRTCConn) sendOffer() error { + select { + case offer := <-c.offerChannel: + if "" == brokerURL { + log.Printf("Please Copy & Paste the following to the peer:") + log.Printf("----------------") + fmt.Fprintln(logFile, "\n"+offer.Serialize()+"\n") + log.Printf("----------------") + return nil + } + // Use Broker... + go func() { + log.Println("Sending offer via BrokerChannel...\nTarget URL: ", brokerURL, + "\nFront URL: ", frontDomain) + answer, err := c.broker.Negotiate(c.pc.LocalDescription()) + if nil != err { + log.Printf("BrokerChannel signaling error: %s", err) + return + } + if nil == answer { + log.Printf("BrokerChannel: No answer received.") + return + // return errors.New("No answer received.") + } + answerChannel <- answer + }() + case err := <-c.errorChannel: + c.pc.Close() + return err + } + return nil +} +func dialWebRTC(config *webrtc.Configuration, broker *BrokerChannel) ( + *webRTCConn, error) { pc, err := webrtc.NewPeerConnection(config) if err != nil { log.Printf("NewPeerConnection: %s", err) return nil, err } + connection := new(webRTCConn) + connection.broker = broker + connection.pc = pc + connection.offerChannel = make(chan *webrtc.SessionDescription) + connection.errorChannel = make(chan error) + connection.openChannel = make(chan struct{}) // Triggered by CreateDataChannel. pc.OnNegotiationNeeded = func() { @@ -109,12 +185,12 @@ func dialWebRTC(config *webrtc.Configuration, broker *BrokerChannel) ( go func() { offer, err := pc.CreateOffer() if err != nil { - errChan <- err + connection.errorChannel <- err return } err = pc.SetLocalDescription(offer) if err != nil { - errChan <- err + connection.errorChannel <- err return } }() @@ -126,7 +202,7 @@ func dialWebRTC(config *webrtc.Configuration, broker *BrokerChannel) ( // TODO: This may soon be deprecated, consider OnIceGatheringStateChange. pc.OnIceComplete = func() { log.Printf("OnIceComplete") - offerChan <- pc.LocalDescription() + connection.offerChannel <- pc.LocalDescription() } // This callback is not expected, as the Client initiates the creation // of the data channel, not the remote peer. @@ -135,62 +211,14 @@ func dialWebRTC(config *webrtc.Configuration, broker *BrokerChannel) ( panic("OnDataChannel") } - pr, pw := io.Pipe() + // Pipes remain the same even when DataChannel gets switched. + connection.recvPipe, connection.writePipe = io.Pipe() - dc, err := pc.CreateDataChannel("test", webrtc.Init{}) - if err != nil { - log.Printf("CreateDataChannel: %s", err) - return nil, err - } - dc.OnOpen = func() { - log.Println("OnOpen channel") - openChan <- struct{}{} - } - dc.OnClose = func() { - log.Println("OnClose channel") - pw.Close() - close(openChan) - // TODO: (Issue #12) Should attempt to renegotiate at this point. - } - dc.OnMessage = func(msg []byte) { - log.Printf("OnMessage <--- %d bytes", len(msg)) - n, err := pw.Write(msg) - if err != nil { - pw.CloseWithError(err) - } - if n != len(msg) { - panic("short write") - } - } - - select { - case err := <-errChan: - pc.Close() - return nil, err - case offer := <-offerChan: - log.Printf("----------------") - fmt.Fprintln(logFile, "\n"+offer.Serialize()+"\n") - log.Printf("----------------") - go func() { - if "" != brokerURL { - log.Println("Sending offer via BrokerChannel...\nTarget URL: ", brokerURL, - "\nFront URL: ", frontDomain) - answer, err := broker.Negotiate(pc.LocalDescription()) - if nil != err { - log.Printf("BrokerChannel signaling error: %s", err) - } - if nil == answer { - log.Printf("BrokerChannel: No answer received.") - } else { - signalChan <- answer - } - } - }() - } - - log.Printf("waiting for answer") - answer, ok := <-signalChan + connection.EstablishDataChannel() + connection.sendOffer() + log.Printf("waiting for answer...") + answer, ok := <-answerChannel if !ok { pc.Close() return nil, fmt.Errorf("no answer received") @@ -205,13 +233,13 @@ func dialWebRTC(config *webrtc.Configuration, broker *BrokerChannel) ( // Wait until data channel is open; otherwise for example sends may get // lost. // TODO: Buffering *should* work though. - _, ok = <-openChan + _, ok = <-connection.openChannel if !ok { pc.Close() return nil, fmt.Errorf("failed to open data channel") } - return &webRTCConn{pc: pc, dc: dc, recvPipe: pr}, nil + return connection, nil } func endWebRTC() { @@ -229,6 +257,7 @@ func endWebRTC() { } } +// Establish a WebRTC channel for SOCKS connections. func handler(conn *pt.SocksConn) error { handlerChan <- 1 defer func() { @@ -259,7 +288,6 @@ func handler(conn *pt.SocksConn) error { } copyLoop(conn, remote) - return nil } @@ -293,10 +321,10 @@ func readSignalingMessages(f *os.File) { log.Printf("ignoring invalid signal message %+q", msg) continue } - signalChan <- sdp + answerChannel <- sdp } - log.Printf("close signalChan") - close(signalChan) + log.Printf("close answerChannel") + close(answerChannel) if err := s.Err(); err != nil { log.Printf("signal FIFO: %s", err) } @@ -308,19 +336,25 @@ func main() { flag.StringVar(&brokerURL, "url", "", "URL of signaling broker") flag.StringVar(&frontDomain, "front", "", "front domain") flag.Parse() - logFile, err = os.OpenFile("snowflake.log", os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0600) if err != nil { log.Fatal(err) } defer logFile.Close() log.SetOutput(logFile) - log.Println("starting") - - if "" == brokerURL { + log.Println("\nStarting Snowflake Client...") + + // Expect user to copy-paste if + // TODO: Maybe just get rid of copy-paste entirely. + if "" != brokerURL { + log.Println("Rendezvous using Broker at: ", brokerURL) + if "" != frontDomain { + log.Println("Domain fronting using:", frontDomain) + } + } else { log.Println("No HTTP signaling detected. Waiting for a \"signal\" pipe...") // This FIFO receives signaling messages. - err = syscall.Mkfifo("signal", 0600) + err := syscall.Mkfifo("signal", 0600) if err != nil { if err.(syscall.Errno) != syscall.EEXIST { log.Fatal(err) @@ -363,6 +397,7 @@ func main() { } } pt.CmethodsDone() + defer endWebRTC() var numHandlers int = 0 var sig os.Signal @@ -382,10 +417,9 @@ func main() { ln.Close() } - if syscall.SIGTERM == sig || syscall.SIGINT == sig { - endWebRTC() - return - } + // if syscall.SIGTERM == sig || syscall.SIGINT == sig { + // return + // } // wait for second signal or no more handlers sig = nil