package test import ( "bytes" "encoding/base64" "encoding/binary" "fmt" "io" "math/rand" "net" "sync" "testing" "time" "github.com/cbeuw/Cloak/internal/client" "github.com/cbeuw/Cloak/internal/common" mux "github.com/cbeuw/Cloak/internal/multiplex" "github.com/cbeuw/Cloak/internal/server" "github.com/cbeuw/connutil" "github.com/stretchr/testify/assert" log "github.com/sirupsen/logrus" ) const numConns = 200 // -race option limits the number of goroutines to 8192 func serveTCPEcho(l net.Listener) { for { conn, err := l.Accept() if err != nil { log.Error(err) return } go func(conn net.Conn) { _, err := io.Copy(conn, conn) if err != nil { conn.Close() log.Error(err) return } }(conn) } } func serveUDPEcho(listener *connutil.PipeListener) { for { conn, err := listener.ListenPacket("udp", "") if err != nil { log.Error(err) return } const bufSize = 32 * 1024 go func(conn net.PacketConn) { defer conn.Close() buf := make([]byte, bufSize) for { r, _, err := conn.ReadFrom(buf) if err != nil { log.Error(err) return } w, err := conn.WriteTo(buf[:r], nil) if err != nil { log.Error(err) return } if r != w { log.Error("written not eqal to read") return } } }(conn) } } var bypassUID = [16]byte{0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15} var publicKey, _ = base64.StdEncoding.DecodeString("7f7TuKrs264VNSgMno8PkDlyhGhVuOSR8JHLE6H4Ljc=") var privateKey, _ = base64.StdEncoding.DecodeString("SMWeC6VuZF8S/id65VuFQFlfa7hTEJBpL6wWhqPP100=") var basicUDPConfig = client.RawConfig{ ServerName: "www.example.com", ProxyMethod: "openvpn", EncryptionMethod: "plain", UID: bypassUID[:], PublicKey: publicKey, NumConn: 4, UDP: true, Transport: "direct", RemoteHost: "fake.com", RemotePort: "9999", LocalHost: "127.0.0.1", LocalPort: "9999", } var basicTCPConfig = client.RawConfig{ ServerName: "www.example.com", ProxyMethod: "shadowsocks", EncryptionMethod: "plain", UID: bypassUID[:], PublicKey: publicKey, NumConn: 4, UDP: false, Transport: "direct", RemoteHost: "fake.com", RemotePort: "9999", LocalHost: "127.0.0.1", LocalPort: "9999", BrowserSig: "firefox", } var singleplexTCPConfig = client.RawConfig{ ServerName: "www.example.com", ProxyMethod: "shadowsocks", EncryptionMethod: "plain", UID: bypassUID[:], PublicKey: publicKey, NumConn: 0, UDP: false, Transport: "direct", RemoteHost: "fake.com", RemotePort: "9999", LocalHost: "127.0.0.1", LocalPort: "9999", BrowserSig: "safari", } func generateClientConfigs(rawConfig client.RawConfig, state common.WorldState) (client.LocalConnConfig, client.RemoteConnConfig, client.AuthInfo) { lcl, rmt, auth, err := rawConfig.ProcessRawConfig(state) if err != nil { log.Fatal(err) } return lcl, rmt, auth } func basicServerState(ws common.WorldState) *server.State { var serverConfig = server.RawConfig{ ProxyBook: map[string][]string{"shadowsocks": {"tcp", "fake.com:9999"}, "openvpn": {"udp", "fake.com:9999"}}, BindAddr: []string{"fake.com:9999"}, BypassUID: [][]byte{bypassUID[:]}, RedirAddr: "fake.com:9999", PrivateKey: privateKey, KeepAlive: 15, CncMode: false, } state, err := server.InitState(serverConfig, ws) if err != nil { log.Fatal(err) } return state } type mockUDPDialer struct { addrCh chan *net.UDPAddr raddr *net.UDPAddr } func (m *mockUDPDialer) Dial(network, address string) (net.Conn, error) { if m.raddr == nil { m.raddr = <-m.addrCh } return net.DialUDP("udp", nil, m.raddr) } func establishSession(lcc client.LocalConnConfig, rcc client.RemoteConnConfig, ai client.AuthInfo, serverState *server.State) (common.Dialer, *connutil.PipeListener, common.Dialer, net.Listener, error) { // redirecting web server // ^ // | // | // redirFromCkServerL // | // | // proxy client ----proxyToCkClientD----> ck-client ------> ck-server ----proxyFromCkServerL----> proxy server // ^ // | // | // netToCkServerD // | // | // whatever connection initiator (including a proper ck-client) netToCkServerD, ckServerListener := connutil.DialerListener(10 * 1024) clientSeshMaker := func() *mux.Session { ai := ai quad := make([]byte, 4) common.RandRead(ai.WorldState.Rand, quad) ai.SessionId = binary.BigEndian.Uint32(quad) return client.MakeSession(rcc, ai, netToCkServerD) } var proxyToCkClientD common.Dialer if ai.Unordered { // We can only "dial" a single UDP connection as we can't send packets from different context // to a single UDP listener addrCh := make(chan *net.UDPAddr, 1) mDialer := &mockUDPDialer{ addrCh: addrCh, } acceptor := func() (*net.UDPConn, error) { laddr, _ := net.ResolveUDPAddr("udp", "127.0.0.1:0") conn, err := net.ListenUDP("udp", laddr) addrCh <- conn.LocalAddr().(*net.UDPAddr) return conn, err } go client.RouteUDP(acceptor, lcc.Timeout, rcc.Singleplex, clientSeshMaker) proxyToCkClientD = mDialer } else { var proxyToCkClientL *connutil.PipeListener proxyToCkClientD, proxyToCkClientL = connutil.DialerListener(10 * 1024) go client.RouteTCP(proxyToCkClientL, lcc.Timeout, rcc.Singleplex, clientSeshMaker) } // set up server ckServerToProxyD, proxyFromCkServerL := connutil.DialerListener(10 * 1024) ckServerToWebD, redirFromCkServerL := connutil.DialerListener(10 * 1024) serverState.ProxyDialer = ckServerToProxyD serverState.RedirDialer = ckServerToWebD go server.Serve(ckServerListener, serverState) return proxyToCkClientD, proxyFromCkServerL, netToCkServerD, redirFromCkServerL, nil } func runEchoTest(t *testing.T, conns []net.Conn, msgLen int) { var wg sync.WaitGroup for _, conn := range conns { wg.Add(1) go func(conn net.Conn) { defer wg.Done() testData := make([]byte, msgLen) rand.Read(testData) // we cannot call t.Fatalf in concurrent contexts n, err := conn.Write(testData) if n != msgLen { t.Errorf("written only %v, err %v", n, err) return } recvBuf := make([]byte, msgLen) _, err = io.ReadFull(conn, recvBuf) if err != nil { t.Errorf("failed to read back: %v", err) return } if !bytes.Equal(testData, recvBuf) { t.Errorf("echoed data not correct") return } }(conn) } wg.Wait() } func TestUDP(t *testing.T) { log.SetLevel(log.ErrorLevel) worldState := common.WorldOfTime(time.Unix(10, 0)) lcc, rcc, ai := generateClientConfigs(basicUDPConfig, worldState) sta := basicServerState(worldState) proxyToCkClientD, proxyFromCkServerL, _, _, err := establishSession(lcc, rcc, ai, sta) if err != nil { t.Fatal(err) } t.Run("simple send", func(t *testing.T) { pxyClientConn, err := proxyToCkClientD.Dial("udp", "") if err != nil { t.Error(err) } const testDataLen = 1500 testData := make([]byte, testDataLen) rand.Read(testData) n, err := pxyClientConn.Write(testData) if n != testDataLen { t.Errorf("wrong length sent: %v", n) } if err != nil { t.Error(err) } pxyServerConn, err := proxyFromCkServerL.ListenPacket("", "") if err != nil { t.Error(err) } recvBuf := make([]byte, testDataLen+100) r, _, err := pxyServerConn.ReadFrom(recvBuf) if err != nil { t.Error(err) } if !bytes.Equal(testData, recvBuf[:r]) { t.Error("read wrong data") } }) const echoMsgLen = 1024 t.Run("user echo", func(t *testing.T) { go serveUDPEcho(proxyFromCkServerL) var conn [1]net.Conn conn[0], err = proxyToCkClientD.Dial("udp", "") if err != nil { t.Error(err) } runEchoTest(t, conn[:], echoMsgLen) }) } func TestTCPSingleplex(t *testing.T) { log.SetLevel(log.ErrorLevel) worldState := common.WorldOfTime(time.Unix(10, 0)) lcc, rcc, ai := generateClientConfigs(singleplexTCPConfig, worldState) sta := basicServerState(worldState) proxyToCkClientD, proxyFromCkServerL, _, _, err := establishSession(lcc, rcc, ai, sta) if err != nil { t.Fatal(err) } const echoMsgLen = 1 << 16 go serveTCPEcho(proxyFromCkServerL) proxyConn1, err := proxyToCkClientD.Dial("", "") if err != nil { t.Fatal(err) } runEchoTest(t, []net.Conn{proxyConn1}, echoMsgLen) user, err := sta.Panel.GetUser(ai.UID[:]) if err != nil { t.Fatalf("failed to fetch user: %v", err) } if user.NumSession() != 1 { t.Error("no session were made on first connection establishment") } proxyConn2, err := proxyToCkClientD.Dial("", "") if err != nil { t.Fatal(err) } runEchoTest(t, []net.Conn{proxyConn2}, echoMsgLen) if user.NumSession() != 2 { t.Error("no extra session were made on second connection establishment") } // Both conns should work runEchoTest(t, []net.Conn{proxyConn1, proxyConn2}, echoMsgLen) proxyConn1.Close() assert.Eventually(t, func() bool { return user.NumSession() == 1 }, time.Second, 10*time.Millisecond, "first session was not closed on connection close") // conn2 should still work runEchoTest(t, []net.Conn{proxyConn2}, echoMsgLen) var conns [numConns]net.Conn for i := 0; i < numConns; i++ { conns[i], err = proxyToCkClientD.Dial("", "") if err != nil { t.Fatal(err) } } runEchoTest(t, conns[:], echoMsgLen) } func TestTCPMultiplex(t *testing.T) { log.SetLevel(log.ErrorLevel) worldState := common.WorldOfTime(time.Unix(10, 0)) lcc, rcc, ai := generateClientConfigs(basicTCPConfig, worldState) sta := basicServerState(worldState) proxyToCkClientD, proxyFromCkServerL, netToCkServerD, redirFromCkServerL, err := establishSession(lcc, rcc, ai, sta) if err != nil { t.Fatal(err) } t.Run("user echo single", func(t *testing.T) { for i := 0; i < 18; i += 2 { dataLen := 1 << i writeData := make([]byte, dataLen) rand.Read(writeData) t.Run(fmt.Sprintf("data length %v", dataLen), func(t *testing.T) { go serveTCPEcho(proxyFromCkServerL) conn, err := proxyToCkClientD.Dial("", "") if err != nil { t.Error(err) } n, err := conn.Write(writeData) if err != nil { t.Error(err) } if n != dataLen { t.Errorf("write length doesn't match up: %v, expected %v", n, dataLen) } recvBuf := make([]byte, dataLen) _, err = io.ReadFull(conn, recvBuf) if err != nil { t.Error(err) } if !bytes.Equal(writeData, recvBuf) { t.Error("echoed data incorrect") } }) } }) const echoMsgLen = 16384 t.Run("user echo", func(t *testing.T) { go serveTCPEcho(proxyFromCkServerL) var conns [numConns]net.Conn for i := 0; i < numConns; i++ { conns[i], err = proxyToCkClientD.Dial("", "") if err != nil { t.Error(err) } } runEchoTest(t, conns[:], echoMsgLen) }) t.Run("redir echo", func(t *testing.T) { go serveTCPEcho(redirFromCkServerL) var conns [numConns]net.Conn for i := 0; i < numConns; i++ { conns[i], err = netToCkServerD.Dial("", "") if err != nil { t.Error(err) } } runEchoTest(t, conns[:], echoMsgLen) }) } func TestClosingStreamsFromProxy(t *testing.T) { log.SetLevel(log.ErrorLevel) worldState := common.WorldOfTime(time.Unix(10, 0)) for clientConfigName, clientConfig := range map[string]client.RawConfig{"basic": basicTCPConfig, "singleplex": singleplexTCPConfig} { clientConfig := clientConfig clientConfigName := clientConfigName t.Run(clientConfigName, func(t *testing.T) { lcc, rcc, ai := generateClientConfigs(clientConfig, worldState) sta := basicServerState(worldState) proxyToCkClientD, proxyFromCkServerL, _, _, err := establishSession(lcc, rcc, ai, sta) if err != nil { t.Fatal(err) } t.Run("closing from server", func(t *testing.T) { clientConn, _ := proxyToCkClientD.Dial("", "") clientConn.Write(make([]byte, 16)) serverConn, _ := proxyFromCkServerL.Accept() serverConn.Close() assert.Eventually(t, func() bool { _, err := clientConn.Read(make([]byte, 16)) return err != nil }, time.Second, 10*time.Millisecond, "closing stream on server side is not reflected to the client") }) t.Run("closing from client", func(t *testing.T) { // closing stream on client side clientConn, _ := proxyToCkClientD.Dial("", "") clientConn.Write(make([]byte, 16)) serverConn, _ := proxyFromCkServerL.Accept() clientConn.Close() assert.Eventually(t, func() bool { _, err := serverConn.Read(make([]byte, 16)) return err != nil }, time.Second, 10*time.Millisecond, "closing stream on client side is not reflected to the server") }) t.Run("send then close", func(t *testing.T) { testData := make([]byte, 24*1024) rand.Read(testData) clientConn, _ := proxyToCkClientD.Dial("", "") go func() { clientConn.Write(testData) // it takes time for this written data to be copied asynchronously // into ck-server's domain. If the pipe is closed before that, read // by ck-client in RouteTCP will fail as we have closed it. time.Sleep(700 * time.Millisecond) clientConn.Close() }() readBuf := make([]byte, len(testData)) serverConn, err := proxyFromCkServerL.Accept() if err != nil { t.Errorf("failed to accept a connection delievering data sent before closing: %v", err) } _, err = io.ReadFull(serverConn, readBuf) if err != nil { t.Errorf("failed to read data sent before closing: %v", err) } }) }) } } func BenchmarkIntegration(b *testing.B) { log.SetLevel(log.ErrorLevel) worldState := common.WorldOfTime(time.Unix(10, 0)) lcc, rcc, ai := generateClientConfigs(basicTCPConfig, worldState) sta := basicServerState(worldState) const bufSize = 16 * 1024 encryptionMethods := map[string]byte{ "plain": mux.EncryptionMethodPlain, "chacha20-poly1305": mux.EncryptionMethodChaha20Poly1305, "aes-256-gcm": mux.EncryptionMethodAES256GCM, "aes-128-gcm": mux.EncryptionMethodAES128GCM, } for name, method := range encryptionMethods { b.Run(name, func(b *testing.B) { ai.EncryptionMethod = method proxyToCkClientD, proxyFromCkServerL, _, _, err := establishSession(lcc, rcc, ai, sta) if err != nil { b.Fatal(err) } b.Run("single stream bandwidth", func(b *testing.B) { more := make(chan int, 10) go func() { // sender writeBuf := make([]byte, bufSize+100) serverConn, _ := proxyFromCkServerL.Accept() for { serverConn.Write(writeBuf) <-more } }() // receiver clientConn, _ := proxyToCkClientD.Dial("", "") readBuf := make([]byte, bufSize) clientConn.Write([]byte{1}) // to make server accept b.SetBytes(bufSize) b.ResetTimer() for i := 0; i < b.N; i++ { io.ReadFull(clientConn, readBuf) // ask for more more <- 0 } }) b.Run("single stream latency", func(b *testing.B) { clientConn, _ := proxyToCkClientD.Dial("", "") buf := []byte{1} clientConn.Write(buf) serverConn, _ := proxyFromCkServerL.Accept() serverConn.Read(buf) b.ResetTimer() for i := 0; i < b.N; i++ { clientConn.Write(buf) serverConn.Read(buf) } }) }) } }