playing around with connection management

Author: Dan Vittegleo

README.md

@@ -44,8 +44,8 @@ The easiest way is to download a pre-built binary from the [GitHub Releases](htt
 
 1. Copy `awslambdaproxy` binary to a publicly accessible linux host (e.g. EC2 instance). You will need to open the following ports on this host:
 
-    * Port 8080 - this port listens for user proxy connections and needs to only be opened to whatever your external IP address is where you plan to browse the web.
-    * Port 8081 - this port listens for tunnel connections from executing Lambda functions and needs to be opened to the world. This is a security concern and will be locked down in the future.
+    * Port 8080 - this port listens for user proxy connections and needs to only be opened to whatever your external IP address is where you plan to browse the web. Alternatively on OSX, I use [Secure Pipes](https://www.opoet.com/pyro/) to setup a SSH tunnel with port 8080 forwarded to localhost. This allows port 8080 to remain unexposed and instead relies on SSH being exposed.
+    * Port 8081 - this port listens for tunnel connections from executing Lambda functions and needs to be opened to the world. <b>This is a security concern and will be locked down in the future.</b>
 
 2. On publicly accessible host, run `awslambdaproxy`. You'll need to ensure AWS access key and secret key environment variables are defined. For now, this access key should have AdministratorAccess.
 
@@ -59,7 +59,7 @@ The easiest way is to download a pre-built binary from the [GitHub Releases](htt
 
 ## FAQ
 1. <b>Should I use awslambdaproxy?</b> That's up to you. Use at your own risk.
-2. <b>Will this make me completely anonymous?</b> No, absolutely not. The goal of this project is just to obfuscate your web traffic by rotating your IP address. All of your traffic is going through AWS which could be traced back to your account. You can also be tracked still with [browser fingerprinting](https://panopticlick.eff.org/), etc.
+2. <b>Will this make me completely anonymous?</b> No, absolutely not. The goal of this project is just to obfuscate your web traffic by rotating your IP address. All of your traffic is going through AWS which could be traced back to your account. You can also be tracked still with [browser fingerprinting](https://panopticlick.eff.org/), etc. Your [IP address may still leak](https://ipleak.net/) due to WebRTC, Flash, etc.
 3. <b>How often will my external IP address change?</b> For each region specified, the IP address will change roughly every 4 hours. This of course is subject to change at any moment as this is not something that is documented by AWS Lambda.
 4. <b>How big is the pool of IP addresses?</b> This I don't know, but I do know I did not have a duplicate IP while running the proxy for a week.
 5. <b>How much does this cost?</b> awslambdaproxy should be able to run mostly on the [AWS free tier](https://aws.amazon.com/free/) minus bandwidth costs. It can run on a t2.micro instance and the default 128MB Lambda function that is constantly running should also fall in the free tier usage. The bandwidth is what will cost you money; you will pay for bandwidth usage for both EC2 and Lambda.

cmd/awslambdaproxy/awslambdaproxy.go

@@ -15,7 +15,7 @@ const (
 
 func main() {
 	regionsPtr := flag.String("regions", "us-west-2", "Regions to run proxy (e.g. us-west-2) (can be comma separated list)")
-	frequencyPtr := flag.Int( "frequency", 10, "Frequency in seconds to execute Lambda function. If multiple regions are specified, this will cause traffic to rotate round robin at the interval specified here")
+	frequencyPtr := flag.Int( "frequency", 180, "Frequency in seconds to execute Lambda function. If multiple regions are specified, this will cause traffic to rotate round robin at the interval specified here")
 	proxyPortPtr := flag.String("proxy-port", "8080", "Port to listen for proxy connections")
 	tunnelPortPtr := flag.String("tunnel-port", "8081", "Port to listen for reverse connection from Lambda")
 	flag.Parse()

datacopy.go

@@ -18,21 +18,20 @@ func (d *DataCopyManager) run() {
 }
 
 func (d *DataCopyManager) handleClient(userConn net.Conn) {
-	d.tunnelConnectionManager.mutex.RLock()
-	tunnelStream, err := d.tunnelConnectionManager.lastTunnel.sess.Open()
-	d.tunnelConnectionManager.mutex.RUnlock()
+	tunnelStream, err := d.tunnelConnectionManager.getActiveSession()
 	if err != nil {
 		log.Println("Failed to open stream to remote")
 		return
 	}
 
-	log.Println("Opened stream to remote " + tunnelStream.RemoteAddr().String())
 	bidirectionalCopy(tunnelStream, userConn)
 }
 
 func newDataCopyManager(user *UserConnectionManager, tunnel *TunnelConnectionManager) *DataCopyManager {
-	return &DataCopyManager{
+	copyManager :=  &DataCopyManager{
 		userConnectionManager: user,
 		tunnelConnectionManager: tunnel,
 	}
+	go copyManager.run()
+	return copyManager
 }

init.go

@@ -4,10 +4,11 @@ import (
 	"time"
 	"os"
 	"log"
+	"runtime"
 )
 
 func ServerInit(proxyPort string, tunnelPort string, regions []string, lambdaExecutionFrequency time.Duration) {
-	lambdaExecutionTimeout := int64(lambdaExecutionFrequency.Seconds()) + int64(5)
+	lambdaExecutionTimeout := int64(lambdaExecutionFrequency.Seconds()) + int64(10)
 
 	log.Println("Setting up Lambda infrastructure")
 	err := setupLambdaInfrastructure(regions, lambdaExecutionTimeout)
@@ -17,20 +18,27 @@ func ServerInit(proxyPort string, tunnelPort string, regions []string, lambdaExe
 	}
 
 	log.Println("Starting TunnelConnectionManager")
-	tunnelConnectionManager, err := newTunnelConnectionManager(tunnelPort)
+	tunnelConnectionManager, err := newTunnelConnectionManager(tunnelPort, lambdaExecutionFrequency)
 	if err != nil {
 		log.Println("Failed to setup TunnelConnectionManager", err.Error())
 		os.Exit(1)
 	}
-	go tunnelConnectionManager.run()
 
 	log.Println("Starting LambdaExecutionManager")
 	lambdaExecutionManager, err := newLambdaExecutionManager(tunnelPort, regions, lambdaExecutionFrequency)
 	if err != nil {
 		log.Println("Failed to setup LambdaExecutionManager", err.Error())
 		os.Exit(1)
 	}
-	go lambdaExecutionManager.run()
+
+	go func(){
+		for {
+			<-tunnelConnectionManager.emergencyTunnel
+			log.Println("EMERGENCY TUNNEL STARTED")
+			lambdaExecutionManager.executeFunction(0)
+			time.Sleep(time.Second * 5)
+		}
+	}()
 
 	tunnelConnectionManager.waitUntilReady()
 
@@ -40,9 +48,9 @@ func ServerInit(proxyPort string, tunnelPort string, regions []string, lambdaExe
 		log.Println("Failed to setup UserConnectionManager", err.Error())
 		os.Exit(1)
 	}
-	go userConnectionManager.run()
 
 	log.Println("Starting DataCopyManager")
-	dataCopyManager := newDataCopyManager(userConnectionManager, tunnelConnectionManager)
-	dataCopyManager.run()
+	newDataCopyManager(userConnectionManager, tunnelConnectionManager)
+
+	runtime.Goexit()
 }

lambdaexecution.go

@@ -22,34 +22,41 @@ func (l *LambdaExecutionManager) run() {
 	log.Println("Using public IP", l.publicIp)
 	log.Println("Lambda execution frequency", l.frequency)
 	for {
-		sess := session.New(&aws.Config{})
 		for region := range l.regions {
-			svc := lambda.New(sess, &aws.Config{Region: aws.String(l.regions[region])})
-			payload, _ := json.Marshal(l.publicIp + ":" + l.port)
-			params := &lambda.InvokeInput{
-				FunctionName:   aws.String(lambdaFunctionName),
-				InvocationType: aws.String(lambda.InvocationTypeEvent),
-				Payload:        payload,
-			}
-			log.Println("Executing Lambda function in region", l.regions[region])
-			_, err := svc.Invoke(params)
-			if err != nil {
-				log.Println("Failed to execute Lambda function.", err.Error())
-			}
+			l.executeFunction(region)
 			time.Sleep(l.frequency)
 		}
 	}
 }
 
+func (l *LambdaExecutionManager) executeFunction(region int) error {
+	log.Println("Executing Lambda function in region", l.regions[region])
+	sess := session.New(&aws.Config{})
+	svc := lambda.New(sess, &aws.Config{Region: aws.String(l.regions[region])})
+	payload, _ := json.Marshal(l.publicIp + ":" + l.port)
+	params := &lambda.InvokeInput{
+		FunctionName:   aws.String(lambdaFunctionName),
+		InvocationType: aws.String(lambda.InvocationTypeEvent),
+		Payload:        payload,
+	}
+	_, err := svc.Invoke(params)
+	if err != nil {
+		return errors.Wrap(err, "Failed to execute Lambda function")
+	}
+	return nil
+}
+
 func newLambdaExecutionManager(port string, regions []string, frequency time.Duration) (*LambdaExecutionManager, error) {
 	publicIp, err := getPublicIp()
 	if err != nil {
-		return nil, errors.Wrap(err, "Error getting IP address")
+		return nil, errors.Wrap(err, "Error getting public IP address")
 	}
-	return &LambdaExecutionManager{
+	executionManager := &LambdaExecutionManager{
 		port: port,
 		regions: regions,
 		frequency: frequency,
 		publicIp: publicIp,
-	}, nil
+	}
+	go executionManager.run()
+	return executionManager, nil
 }

tunnelconnection.go

@@ -13,12 +13,16 @@ import (
 type TunnelConnection struct {
 	conn net.Conn
 	sess *yamux.Session
+	time time.Time
 }
 
 type TunnelConnectionManager struct {
-	listener net.Listener
-	lastTunnel TunnelConnection
-	mutex sync.RWMutex
+	listener      net.Listener
+	currentTunnel string
+	tunnels       map[string]TunnelConnection
+	mutex         sync.RWMutex
+	expectedTunnelSeconds float64
+	emergencyTunnel chan bool
 }
 
 func (t *TunnelConnectionManager) run() {
@@ -38,8 +42,44 @@ func (t *TunnelConnectionManager) run() {
 		log.Println("Established session to", tunnelSession.RemoteAddr())
 
 		t.mutex.Lock()
-		t.lastTunnel = TunnelConnection{c, tunnelSession}
+		t.currentTunnel = c.RemoteAddr().String()
+		t.tunnels[t.currentTunnel] = TunnelConnection{c, tunnelSession, time.Now()}
 		t.mutex.Unlock()
+		go t.monitorConnectionHealth(c.RemoteAddr().String())
+		log.Println("Active tunnel count: ", len(t.tunnels))
+	}
+}
+
+func (t *TunnelConnectionManager) monitorConnectionHealth(connectionId string) {
+	for {
+		_, err := t.tunnels[connectionId].sess.Ping()
+		if err != nil {
+			if time.Since(t.tunnels[connectionId].time).Seconds() < t.expectedTunnelSeconds {
+				log.Println("Signaling for emergency tunnel due to tunnel ending early: ", time.Since(t.tunnels[connectionId].time).Seconds())
+				t.emergencyTunnel <- true
+			}
+			t.tunnels[connectionId].sess.Close()
+			t.tunnels[connectionId].conn.Close()
+			t.mutex.Lock()
+			delete(t.tunnels, connectionId)
+			t.mutex.Unlock()
+			break
+		}
+		time.Sleep(time.Millisecond * 250)
+	}
+}
+
+func (t *TunnelConnectionManager) getActiveSession() (net.Conn, error) {
+	for {
+		t.mutex.RLock()
+		tunnel, ok := t.tunnels[t.currentTunnel]
+		t.mutex.RUnlock()
+		if ok {
+			sess, err := tunnel.sess.Open()
+			return sess, err
+		}
+		log.Println("TunnelConnectionManager.getActiveSession failed. Retrying..")
+		time.Sleep(time.Second * 1)
 	}
 }
 
@@ -55,19 +95,26 @@ func (t *TunnelConnectionManager) waitUntilReady() {
 }
 
 func (t *TunnelConnectionManager) isReady() bool {
-	if t.lastTunnel == (TunnelConnection{}) {
+	if t.currentTunnel == "" {
 		return false
 	} else {
 		return true
 	}
 }
 
-func newTunnelConnectionManager(port string) (*TunnelConnectionManager, error) {
+func newTunnelConnectionManager(port string, lambdaExecutionFrequency time.Duration) (*TunnelConnectionManager, error) {
 	listener, err := startTunnelListener(port)
 	if err != nil {
 		return nil, errors.Wrap(err, "Failed to start TunnelConnectionManager")
 	}
-	return &TunnelConnectionManager{listener: listener}, nil
+	connectionManager := &TunnelConnectionManager{
+		listener: listener,
+		tunnels: make(map[string]TunnelConnection),
+		emergencyTunnel: make(chan bool),
+		expectedTunnelSeconds: lambdaExecutionFrequency.Seconds(),
+	}
+	go connectionManager.run()
+	return connectionManager, nil
 }
 
 func startTunnelListener(tunnelPort string) (net.Listener, error) {

userconnection.go

@@ -15,7 +15,6 @@ type UserConnectionManager struct {
 func (u *UserConnectionManager) run() {
 	for {
 		c, err := u.listener.Accept()
-		log.Println("Accepted user connection..")
 		if err != nil {
 			log.Println("Failed to accept user connection")
 			return
@@ -33,10 +32,12 @@ func newUserConnectionManager(port string) (*UserConnectionManager, error) {
 	if err != nil {
 		return nil, errors.Wrap(err, "Failed to start UserConnectionManager")
 	}
-	return &UserConnectionManager{
+	connectionManager := &UserConnectionManager{
 		listener: listener,
 		connections: make(chan net.Conn),
-	}, nil
+	}
+	go connectionManager.run()
+	return connectionManager, nil
 }
 
 func startUserListener(proxyPort string) (net.Listener, error) {

util.go

@@ -12,7 +12,7 @@ import (
 )
 
 const (
-	getIPUrl = "http://myexternalip.com/raw"
+	getIPUrl = "http://checkip.amazonaws.com/"
 )
 
 func bidirectionalCopy(dst io.ReadWriteCloser, src io.ReadWriteCloser) {