[Author Prev][Author Next][Thread Prev][Thread Next][Author Index][Thread Index]
[tor-commits] [snowflake/master] USERADDR support for turbotunnel sessions.
commit 0790954020b550f5d5351d9e54b108c9d357fa50
Author: David Fifield <david@xxxxxxxxxxxxxxx>
Date: Tue Feb 4 22:27:58 2020 -0700
USERADDR support for turbotunnel sessions.
The difficulty here is that the whole point of turbotunnel sessions is
that they are not necessarily tied to a single WebSocket connection, nor
even a single client IP address. We use a heuristic: whenever a
WebSocket connection starts that has a new ClientID, we store a mapping
from that ClientID to the IP address attached to the WebSocket
connection in a lookup table. Later, when enough packets have arrived to
establish a turbotunnel session, we recover the ClientID associated with
the session (which kcp-go has stored in the RemoteAddr field), and look
it up in the table to get an IP address. We introduce a new data type,
clientIDMap, to store the clientID-to-IP mapping during the short time
between when a WebSocket connection starts and handleSession receives a
fully fledged KCP session.
---
server/server.go | 47 +++++++++++++++---
server/turbotunnel.go | 85 ++++++++++++++++++++++++++++++++
server/turbotunnel_test.go | 119 +++++++++++++++++++++++++++++++++++++++++++++
3 files changed, 244 insertions(+), 7 deletions(-)
diff --git a/server/server.go b/server/server.go
index 028a24d..1a53de7 100644
--- a/server/server.go
+++ b/server/server.go
@@ -43,6 +43,11 @@ const requestTimeout = 10 * time.Second
// indefinitely.
const clientMapTimeout = 1 * time.Minute
+// How big to make the map of ClientIDs to IP addresses. The map is used in
+// turbotunnelMode to store a reasonable IP address for a client session that
+// may outlive any single WebSocket connection.
+const clientIDAddrMapCapacity = 1024
+
// How long to wait for ListenAndServe or ListenAndServeTLS to return an error
// before deciding that it's not going to return.
const listenAndServeErrorTimeout = 100 * time.Millisecond
@@ -114,6 +119,15 @@ var upgrader = websocket.Upgrader{
CheckOrigin: func(r *http.Request) bool { return true },
}
+// clientIDAddrMap stores short-term mappings from ClientIDs to IP addresses.
+// When we call pt.DialOr, tor wants us to provide a USERADDR string that
+// represents the remote IP address of the client (for metrics purposes, etc.).
+// This data structure bridges the gap between ServeHTTP, which knows about IP
+// addresses, and handleStream, which is what calls pt.DialOr. The common piece
+// of information linking both ends of the chain is the ClientID, which is
+// attached to the WebSocket connection and every session.
+var clientIDAddrMap = newClientIDMap(clientIDAddrMapCapacity)
+
// overrideReadConn is a net.Conn with an overridden Read method. Compare to
// recordingConn at
// https://dave.cheney.net/2015/05/22/struct-composition-with-go.
@@ -203,8 +217,16 @@ func turbotunnelMode(conn net.Conn, addr string, pconn *turbotunnel.QueuePacketC
return fmt.Errorf("reading ClientID: %v", err)
}
- // TODO: ClientID-to-client_ip address mapping
- // Peek at the first read packet to get the KCP conv ID.
+ // Store a a short-term mapping from the ClientID to the client IP
+ // address attached to this WebSocket connection. tor will want us to
+ // provide a client IP address when we call pt.DialOr. But a KCP session
+ // does not necessarily correspond to any single IP address--it's
+ // composed of packets that are carried in possibly multiple WebSocket
+ // streams. We apply the heuristic that the IP address of the most
+ // recent WebSocket connection that has had to do with a session, at the
+ // time the session is established, is the IP address that should be
+ // credited for the entire KCP session.
+ clientIDAddrMap.Set(clientID, addr)
errCh := make(chan error)
@@ -249,10 +271,9 @@ func turbotunnelMode(conn net.Conn, addr string, pconn *turbotunnel.QueuePacketC
}
// handleStream bidirectionally connects a client stream with the ORPort.
-func handleStream(stream net.Conn) error {
- // TODO: This is where we need to provide the client IP address.
- statsChannel <- false
- or, err := pt.DialOr(&ptInfo, "", ptMethodName)
+func handleStream(stream net.Conn, addr string) error {
+ statsChannel <- addr != ""
+ or, err := pt.DialOr(&ptInfo, addr, ptMethodName)
if err != nil {
return fmt.Errorf("connecting to ORPort: %v", err)
}
@@ -266,6 +287,17 @@ func handleStream(stream net.Conn) error {
// acceptStreams layers an smux.Session on the KCP connection and awaits streams
// on it. Passes each stream to handleStream.
func acceptStreams(conn *kcp.UDPSession) error {
+ // Look up the IP address associated with this KCP session, via the
+ // ClientID that is returned by the session's RemoteAddr method.
+ addr, ok := clientIDAddrMap.Get(conn.RemoteAddr().(turbotunnel.ClientID))
+ if !ok {
+ // This means that the map is tending to run over capacity, not
+ // just that there was not client_ip on the incoming connection.
+ // We store "" in the map in the absence of client_ip. This log
+ // message means you should increase clientIDAddrMapCapacity.
+ log.Printf("no address in clientID-to-IP map (capacity %d)", clientIDAddrMapCapacity)
+ }
+
smuxConfig := smux.DefaultConfig()
smuxConfig.Version = 2
smuxConfig.KeepAliveTimeout = 10 * time.Minute
@@ -273,6 +305,7 @@ func acceptStreams(conn *kcp.UDPSession) error {
if err != nil {
return err
}
+
for {
stream, err := sess.AcceptStream()
if err != nil {
@@ -283,7 +316,7 @@ func acceptStreams(conn *kcp.UDPSession) error {
}
go func() {
defer stream.Close()
- err := handleStream(stream)
+ err := handleStream(stream, addr)
if err != nil {
log.Printf("handleStream: %v", err)
}
diff --git a/server/turbotunnel.go b/server/turbotunnel.go
new file mode 100644
index 0000000..1d00897
--- /dev/null
+++ b/server/turbotunnel.go
@@ -0,0 +1,85 @@
+package main
+
+import (
+ "sync"
+
+ "git.torproject.org/pluggable-transports/snowflake.git/common/turbotunnel"
+)
+
+// clientIDMap is a fixed-capacity mapping from ClientIDs to address strings.
+// Adding a new entry using the Set method causes the oldest existing entry to
+// be forgotten.
+//
+// This data type is meant to be used to remember the IP address associated with
+// a ClientID, during the short period of time between when a WebSocket
+// connection with that ClientID began, and when a KCP session is established.
+//
+// The design requirements of this type are that it needs to remember a mapping
+// for only a short time, and old entries should expire so as not to consume
+// unbounded memory. It is not a critical error if an entry is forgotten before
+// it is needed; better to forget entries than to use too much memory.
+type clientIDMap struct {
+ lock sync.Mutex
+ // entries is a circular buffer of (ClientID, addr) pairs.
+ entries []struct {
+ clientID turbotunnel.ClientID
+ addr string
+ }
+ // oldest is the index of the oldest member of the entries buffer, the
+ // one that will be overwritten at the next call to Set.
+ oldest int
+ // current points to the index of the most recent entry corresponding to
+ // each ClientID.
+ current map[turbotunnel.ClientID]int
+}
+
+// newClientIDMap makes a new clientIDMap with the given capacity.
+func newClientIDMap(capacity int) *clientIDMap {
+ return &clientIDMap{
+ entries: make([]struct {
+ clientID turbotunnel.ClientID
+ addr string
+ }, capacity),
+ oldest: 0,
+ current: make(map[turbotunnel.ClientID]int),
+ }
+}
+
+// Set adds a mapping from clientID to addr, replacing any previous mapping for
+// clientID. It may also cause the clientIDMap to forget at most one other
+// mapping, the oldest one.
+func (m *clientIDMap) Set(clientID turbotunnel.ClientID, addr string) {
+ m.lock.Lock()
+ defer m.lock.Unlock()
+ if len(m.entries) == 0 {
+ // The invariant m.oldest < len(m.entries) does not hold in this
+ // special case.
+ return
+ }
+ // m.oldest is the index of the entry we're about to overwrite. If it's
+ // the current entry for any ClientID, we need to delete that clientID
+ // from the current map (that ClientID is now forgotten).
+ if i, ok := m.current[m.entries[m.oldest].clientID]; ok && i == m.oldest {
+ delete(m.current, m.entries[m.oldest].clientID)
+ }
+ // Overwrite the oldest entry.
+ m.entries[m.oldest].clientID = clientID
+ m.entries[m.oldest].addr = addr
+ // Add the overwritten entry to the quick-lookup map.
+ m.current[clientID] = m.oldest
+ // What was the oldest entry is now the newest.
+ m.oldest = (m.oldest + 1) % len(m.entries)
+}
+
+// Get returns a previously stored mapping. The second return value indicates
+// whether clientID was actually present in the map. If it is false, then the
+// returned address string will be "".
+func (m *clientIDMap) Get(clientID turbotunnel.ClientID) (string, bool) {
+ m.lock.Lock()
+ defer m.lock.Unlock()
+ if i, ok := m.current[clientID]; ok {
+ return m.entries[i].addr, true
+ } else {
+ return "", false
+ }
+}
diff --git a/server/turbotunnel_test.go b/server/turbotunnel_test.go
new file mode 100644
index 0000000..c4bf02b
--- /dev/null
+++ b/server/turbotunnel_test.go
@@ -0,0 +1,119 @@
+package main
+
+import (
+ "encoding/binary"
+ "testing"
+
+ "git.torproject.org/pluggable-transports/snowflake.git/common/turbotunnel"
+)
+
+func TestClientIDMap(t *testing.T) {
+ // Convert a uint64 into a ClientID.
+ id := func(n uint64) turbotunnel.ClientID {
+ var clientID turbotunnel.ClientID
+ binary.PutUvarint(clientID[:], n)
+ return clientID
+ }
+
+ // Does m.Get(key) and checks that the output matches what is expected.
+ expectGet := func(m *clientIDMap, clientID turbotunnel.ClientID, expectedAddr string, expectedOK bool) {
+ t.Helper()
+ addr, ok := m.Get(clientID)
+ if addr != expectedAddr || ok != expectedOK {
+ t.Errorf("expected (%+q, %v), got (%+q, %v)", expectedAddr, expectedOK, addr, ok)
+ }
+ }
+
+ // Checks that the len of m.current is as expected.
+ expectSize := func(m *clientIDMap, expectedLen int) {
+ t.Helper()
+ if len(m.current) != expectedLen {
+ t.Errorf("expected map len %d, got %d %+v", expectedLen, len(m.current), m.current)
+ }
+ }
+
+ // Zero-capacity map can't remember anything.
+ {
+ m := newClientIDMap(0)
+ expectSize(m, 0)
+ expectGet(m, id(0), "", false)
+ expectGet(m, id(1234), "", false)
+
+ m.Set(id(0), "A")
+ expectSize(m, 0)
+ expectGet(m, id(0), "", false)
+ expectGet(m, id(1234), "", false)
+
+ m.Set(id(1234), "A")
+ expectSize(m, 0)
+ expectGet(m, id(0), "", false)
+ expectGet(m, id(1234), "", false)
+ }
+
+ {
+ m := newClientIDMap(1)
+ expectSize(m, 0)
+ expectGet(m, id(0), "", false)
+ expectGet(m, id(1), "", false)
+
+ m.Set(id(0), "A")
+ expectSize(m, 1)
+ expectGet(m, id(0), "A", true)
+ expectGet(m, id(1), "", false)
+
+ m.Set(id(1), "B") // forgets the (0, "A") entry
+ expectSize(m, 1)
+ expectGet(m, id(0), "", false)
+ expectGet(m, id(1), "B", true)
+
+ m.Set(id(1), "C") // forgets the (1, "B") entry
+ expectSize(m, 1)
+ expectGet(m, id(0), "", false)
+ expectGet(m, id(1), "C", true)
+ }
+
+ {
+ m := newClientIDMap(5)
+ m.Set(id(0), "A")
+ m.Set(id(1), "B")
+ m.Set(id(2), "C")
+ m.Set(id(0), "D") // shadows the (0, "D") entry
+ m.Set(id(3), "E")
+ expectSize(m, 4)
+ expectGet(m, id(0), "D", true)
+ expectGet(m, id(1), "B", true)
+ expectGet(m, id(2), "C", true)
+ expectGet(m, id(3), "E", true)
+ expectGet(m, id(4), "", false)
+
+ m.Set(id(4), "F") // forgets the (0, "A") entry but should preserve (0, "D")
+ expectSize(m, 5)
+ expectGet(m, id(0), "D", true)
+ expectGet(m, id(1), "B", true)
+ expectGet(m, id(2), "C", true)
+ expectGet(m, id(3), "E", true)
+ expectGet(m, id(4), "F", true)
+
+ m.Set(id(5), "G") // forgets the (1, "B") entry
+ m.Set(id(0), "H") // forgets the (2, "C") entry and shadows (0, "D")
+ expectSize(m, 4)
+ expectGet(m, id(0), "H", true)
+ expectGet(m, id(1), "", false)
+ expectGet(m, id(2), "", false)
+ expectGet(m, id(3), "E", true)
+ expectGet(m, id(4), "F", true)
+ expectGet(m, id(5), "G", true)
+
+ m.Set(id(0), "I") // forgets the (0, "D") entry and shadows (0, "H")
+ m.Set(id(0), "J") // forgets the (3, "E") entry and shadows (0, "I")
+ m.Set(id(0), "K") // forgets the (4, "F") entry and shadows (0, "J")
+ m.Set(id(0), "L") // forgets the (5, "G") entry and shadows (0, "K")
+ expectSize(m, 1)
+ expectGet(m, id(0), "L", true)
+ expectGet(m, id(1), "", false)
+ expectGet(m, id(2), "", false)
+ expectGet(m, id(3), "", false)
+ expectGet(m, id(4), "", false)
+ expectGet(m, id(5), "", false)
+ }
+}
_______________________________________________
tor-commits mailing list
tor-commits@xxxxxxxxxxxxxxxxxxxx
https://lists.torproject.org/cgi-bin/mailman/listinfo/tor-commits