Skip to content


Subversion checkout URL

You can clone with
Download ZIP
Fetching contributors…

Cannot retrieve contributors at this time

107 lines (91 sloc) 3.85 KB
from twisted.internet import pollreactor
except: pass
from twisted.internet import protocol, reactor, defer, task
from twisted.web import http, proxy, resource, server
from twisted.python import log
import sys, time
import urlparse
import socket
import simplejson
import re
SSH_USER = 'localtunnel'
AUTHORIZED_KEYS = '/home/localtunnel/.ssh/authorized_keys'
PORT_RANGE = [32000, 64000]
BANNER = "This localtunnel service is brought to you by Twilio."
SSH_OPTIONS = 'command="/bin/echo Shell access denied",no-agent-forwarding,no-pty,no-user-rc,no-X11-forwarding '
KEY_REGEX = re.compile(r'^ssh-(\w{3}) [^\n]+$')
def port_available(port):
socket.create_connection(['', port]).close()
return False
except socket.error:
return True
def baseN(num,b=32,numerals="23456789abcdefghijkmnpqrstuvwxyz"):
return ((num == 0) and "0" ) or (baseN(num // b, b).lstrip("0") + numerals[num % b])
class LocalTunnelReverseProxy(proxy.ReverseProxyResource):
isLeaf = True
def __init__(self, user, host=''):
self.user = user
self.tunnels = {}
proxy.ReverseProxyResource.__init__(self, host, None, None)
def find_tunnel_name(self):
name = baseN(abs(hash(time.time())))[0:4]
if (name in self.tunnels and not port_available(self.tunnels[name])) or name == 'open':
return self.find_tunnel_name()
return name
def find_tunnel_port(self):
port = PORT_RANGE[0]
start_time = time.time()
while not port_available(port):
if time.time()-start_time > 3:
raise Exception("No port available")
port += 1
if port >= PORT_RANGE[1]: port = PORT_RANGE[0]
return port
def garbage_collect(self):
for name in self.tunnels:
if port_available(self.tunnels[name]):
del self.tunnels[name]
def install_key(self, key):
if not KEY_REGEX.match(key.strip()):
return False
key = ''.join([SSH_OPTIONS, key.strip(), "\n"])
fr = open(AUTHORIZED_KEYS, 'r')
if not key in fr.readlines():
fa = open(AUTHORIZED_KEYS, 'a')
return True
def register_tunnel(self, superhost, key=None):
if key and not self.install_key(key): return simplejson.dumps(dict(error="Invalid key."))
name = self.find_tunnel_name()
port = self.find_tunnel_port()
self.tunnels[name] = port
return simplejson.dumps(
dict(through_port=port, user=self.user, host='%s.%s' % (name, superhost), banner=BANNER))
def render(self, request):
host = request.getHeader('host')
name, superhost = host.split('.', 1)
if host.startswith('open.'):
request.setHeader('Content-Type', 'application/json')
return self.register_tunnel(superhost, request.args.get('key', [None])[0])
if not name in self.tunnels: return "Not found", 0)
clientFactory = self.proxyClientFactoryClass(
request.method, request.uri, request.clientproto,
request.getAllHeaders(),, request)
self.reactor.connectTCP(, self.tunnels[name], clientFactory)
return server.NOT_DONE_YET
#if 'location' in request.responseHeaders and host in request.responseHeaders['location']:
# # Strip out the port they think they need
# p = re.compile(r'%s\:\d+' % host)
# location = p.sub(host, request.responseHeaders['location'])
# request.responseHeaders['location'] = location
reactor.listenTCP(80, server.Site(LocalTunnelReverseProxy(SSH_USER)))
Jump to Line
Something went wrong with that request. Please try again.