#!/usr/bin/env python
# $Id: blossom.py,v 1.164 2006-06-01 03:00:16 goodell Exp $
__license__ = """
Copyright (c) 2005-2006 Geoffrey Goodell.
Redistribution and use in source and binary forms, with or without
modification, are permitted provided that the following conditions are
met:
* Redistributions of source code must retain the above copyright
notice, this list of conditions and the following disclaimer.
* Redistributions in binary form must reproduce the above
copyright notice, this list of conditions and the following disclaimer
in the documentation and/or other materials provided with the
distribution.
* Neither the names of the copyright owners nor the names of its
contributors may be used to endorse or promote products derived from
this software without specific prior written permission.
"""
__version__ = "0.2.11"
__all__ = ["BlossomError",
"ClientRequestHandler",
"DirectoryRequestHandler",
"TorEventHandler",
"TorCloseHandler"]
import getopt
import httplib
import os
import random
import re
import select
import signal
import socket
import string
import struct
import sys
import threading
import time
from BaseHTTPServer import BaseHTTPRequestHandler, HTTPServer
from SocketServer import ThreadingMixIn
from TorCtl import *
# constants
PATH = 0
BUILT = 1
STREAMS = 2
T_CHECK = 0
T_METADATA = 1
T_GET = 2
T_PUBLISH = 3
T_PERIODIC = 4
LEFT = 0
RIGHT = 1
SUCCESS = 0
TARGET = 1
DESC = 1
META = 2
LOGLEVEL = ("err", "warn", "notice", "info")
# timing options
TIME_ACTION = 0.5 # time to wait following an action
TIME_BLOSSOM = 600 # interval for performing Blossom updates
TIME_CACHE = 600 # interval for fetching router lists
TIME_CHECK = 1200 # interval between periodic consistency checks
TIME_POLICY = 7200 # interval between exit policy checks
TIME_RETRY = 5 # interval for connection retries
TIME_REFRESH = 1 # interval for HTML refreshes
TIME_SAVE = 10 # how long to keep track of closed streams
DIR_DESC_EXPIRATION = 3600 # when to republish descriptors
DIR_DESC_DELETE = 86400 # when to delete descriptors
DIR_PEER_KEEPALIVE = 180 # when to conclude that a peer is dead
DIR_POLL_INTERVAL = 60 # interval for polling directory neighbors
TIMEOUT = 30 # timeout for select loops
# configuration options
BUFFER_SIZE = 16384
CONNECTION_CLOSED = 0
DEBUG = 0
DIRPORT = 0
DISCLOSE_TARGET = 1 # explicit query rather than generic download
ENABLE_DIR = 0
MAXINT = 4294967295 # a sufficiently large integer
MAXLEN = 16 # maximum length of blossom-path
MAXNICKLEN = 20 # maximum length of router nickname
MAXREATTACH = 16 # maximum number of times to reattach a stream
MAXRETRY = 60 # maximum retry interval (seconds)
BLOSSOM = []
NICK = ""
WEB_STATUS = "serifos.exit"
BLOSSOM_ARGS = "blossom=lefkada&"
POLICY_ARGS = "addr=1&textonly=1&ports"
STATUS_ARGS = "addr=1&textonly=fingerprint"
HTTP_PROXY = "localhost:8118"
DIR_SERVER = "0.0.0.0:9030"
TORCONTROL = "localhost:9051"
SERVER = "localhost:9052"
IMG_SIZE = "width=18 height=12"
URL_FLAGS = "/flags"
URL_ICONS = "/icons"
ICON_BLANK = "%s/v0.gif" % URL_ICONS
ICON_BUILT_0 = "%s/s0.gif" % URL_ICONS
ICON_BUILT_1 = "%s/s1.gif" % URL_ICONS
ICON_SMITE = "%s/ur.gif" % URL_ICONS
ICON_UNBUILT = "%s/hn.gif" % URL_ICONS
ICON_V0 = "%s/v0.gif" % URL_ICONS
ICON_V1 = "%s/v1.gif" % URL_ICONS
ICON_V2 = "%s/v2.gif" % URL_ICONS
ICON_V3 = "%s/v3.gif" % URL_ICONS
# global variables
AUTOREFRESH = 0
INIT = 0
PERSIST = 0
META_LOCAL = [__version__]
addr = {}
attempted = {}
bw = {}
cc = {}
cc_name = {}
circuits = {}
closed_streams = {}
counted_streams = {}
detached_streams = {}
failed_streams = {}
fingerprint = {}
local = {}
network = {}
path = {}
pending_streams = {}
policy = {}
policy_time = {}
port = {}
prop = {}
query_streams = {}
received_path = {}
semaphore = {}
streams = {}
tor_nodes = {}
interesting_ports = []
persist_nickname = {}
persist_id = {}
queue = ""
threads = {}
unestablished = {}
# data for individual routers
desc = {}
metadata = {}
router_adv = {}
update_time = {}
dir_fingerprint = {}
dir_metadata = {}
dir_path = {}
dir_summary = {}
# data for directory peers
full = {}
selection = {}
dir_port = {}
dir_prop = {}
dir_proxy = {}
metadata_pending = {}
neighbors_recv = {}
neighbors_send = {}
summary = {}
summary_pending = {}
summary_remote = {}
class BlossomError(Exception): pass
class MaxReattachError(Exception): pass
class ThreadingHTTPServer(ThreadingMixIn, HTTPServer): pass
class HTTPPostThread(threading.Thread):
def __init__(self, target, uri, payload, timeout=1):
self.target = target
self.uri = uri
self.payload = payload
self.timeout = timeout
self._stopevent = threading.Event()
threading.Thread.__init__(self)
def run(self):
name = self.getName()
threads[name] = "HTTPPost"
log_msg(3, "*** THREAD: %s %s" % (threads[name], name))
try:
dh, dp = self.target.split(":")
h = httplib.HTTP(dh, dp)
h.putrequest("POST", self.uri)
h.putheader("Content-length", "%d" % len(self.payload))
h.endheaders()
h.send(self.payload)
ec, em, headers = h.getreply()
log_msg(2, "--> HTTP-POST SUCCEEDED requesting %s from %s:%s [%s]" \
% (self.uri, dh, dp, len(self.payload)))
except socket.error:
log_msg(1, "HTTP-POST FAILED requesting %s from %s:%s" % (self.uri, dh, dp))
except:
log_msg(1, "HTTP-POST unexpected: %s" % sys.exc_info()[0])
del threads[name]
def join(self, timeout=None):
self._stopevent.set()
threading.Thread.join(self, timeout)
class OpenURLThread(threading.Thread):
def __init__(self, dh, dp, url, policy=-1, timeout=1):
self.dh = dh
self.dp = dp
self.url = url
self.policy = policy
self.timeout = timeout
self._stopevent = threading.Event()
threading.Thread.__init__(self)
def run(self):
name = self.getName()
threads[name] = "OpenURL"
log_msg(3, "*** THREAD: %s %s" % (threads[name], name))
try:
h = httplib.HTTP(self.dh, self.dp)
h.putrequest('GET', self.url)
h.endheaders()
errcode, errmsg, headers = h.getreply()
log_msg(2, "<-- HTTP-GET %s:%s %s: %s" % (self.dh, self.dp, self.url, str(errcode)))
h = h.file
if self.policy < 0:
process_descriptors(h, "%s:%s" % (self.dh, self.dp))
else:
while h:
try:
line = h.readline()
except AttributeError, e:
log_msg(1, repr(e))
break
except EOFError, e:
log_msg(1, repr(e))
break
if not line:
break
elif line:
if self.policy:
line = "POLICY %s" % line
log_msg(3, "--- %s" % line[:-1])
else:
if self.url == STATUS_URL_BLOSSOM:
line = "DATA+ %s" % line
else:
line = "DATA %s" % line
log_msg(3, "--- %s" % line[:-1])
process_line(line)
except AttributeError, e:
log_msg(1, "AttributeError")
except socket.error, (ec, em):
log_msg(1, "socket.error %s %s" % (repr(ec), repr(em)))
except IOError, (ec, em):
log_msg(1, "IOError %s %s" % (repr(ec), repr(em)))
del threads[name]
def join(self, timeout=None):
self._stopevent.set()
threading.Thread.join(self, timeout)
class SearchThread(threading.Thread):
def __init__(self, streamID, dest, target, timeout=1):
self.streamID = streamID
self.dest = dest
self.target = target
self.timeout = timeout
self._stopevent = threading.Event()
threading.Thread.__init__(self)
def run(self):
name = self.getName()
threads[name] = "Search"
log_msg(3, "*** THREAD: %s %s" % (threads[name], name))
search(self.streamID, self.dest, self.target)
log_msg(3, "search complete")
del threads[name]
def join(self, timeout=None):
self._stopevent.set()
threading.Thread.join(self, timeout)
class PeriodicClientThread(threading.Thread):
def __init__(self, last, interesting_ports, timeout=1):
self.last = last
self.interesting_ports = interesting_ports
self.timeout = timeout
self._stopevent = threading.Event()
threading.Thread.__init__(self)
def run(self):
global CONNECTION_CLOSED
global INIT
name = self.getName()
threads[name] = "PeriodicClient"
log_msg(3, "*** THREAD: %s %s" % (threads[name], name))
last = self.last
interesting_ports = self.interesting_ports
# periodically fetch parsed Tor metadata
if time.time() - last[T_METADATA] > TIME_CACHE:
log_msg(3, "*** fetching Tor metadata")
last[T_METADATA] = time.time()
urllist = [STATUS_URL]
if BLOSSOM:
urllist.append(STATUS_URL_BLOSSOM)
if not BLOSSOM_ARGS:
urllist = []
for url in urllist:
obtain_tor_metadata(url)
threads[name] = "PeriodicClient (phase 1)"
# periodically perform consistency check
if DEBUG and time.time() - last[T_CHECK] > TIME_CHECK:
last[T_CHECK] = time.time()
process_line("CONSISTENCY\n")
threads[name] = "PeriodicClient (phase 2)"
# periodically perform Blossom updates
if BLOSSOM and time.time() - last[T_GET] > TIME_BLOSSOM:
last[T_GET] = time.time()
for b in BLOSSOM:
try:
get_descriptors(b)
except socket.error, (ec, em):
log_msg(1, "socket.error %s %s" % (repr(ec), repr(em)))
if INIT and BLOSSOM and time.time() - last[T_PUBLISH] > TIME_BLOSSOM:
last[T_PUBLISH] = time.time()
for b in BLOSSOM:
try:
if NICK != "":
publish_descriptor(b)
except socket.error, (ec, em):
log_msg(1, "socket.error %s %s" % (repr(ec), repr(em)))
threads[name] = "PeriodicClient (phase 3)"
# establish initial persistent connections
if PERSIST:
to_establish = unestablished.keys()
log_msg(3, "--- BLOSSOM: %s" % BLOSSOM)
log_msg(3, "--- to_establish: %s" % to_establish)
for b in to_establish:
establish_persistent_connection(b)
if CONNECTION_CLOSED > 0:
conn = getConnection()
if conn:
CONNECTION_CLOSED = 0
for qp in interesting_ports:
get_tor_policy(qp)
threads[name] = "PeriodicClient (phase 4)"
if not ENABLE_DIR:
INIT = 1
del threads[name]
# report list of presently active threads
for thread in threads.keys():
try:
log_msg(3, "--- active: %s %s" % (thread, threads[thread]))
except:
# key deletion race condition: not critical
pass
def join(self, timeout=None):
self._stopevent.set()
threading.Thread.join(self, timeout)
class PeriodicDirectoryThread(threading.Thread):
def __init__(self, queue, timeout=1):
self.queue = queue
self.timeout = timeout
self._stopevent = threading.Event()
threading.Thread.__init__(self)
def run(self):
name = self.getName()
threads[name] = "PeriodicDirectory"
log_msg(3, "*** THREAD: %s %s" % (threads[name], name))
DIR_REG_LOCK = 1
send_updates(self.queue)
DIR_REG_LOCK = 0
del threads[name]
def join(self, timeout=None):
self._stopevent.set()
threading.Thread.join(self, timeout)
class DirectoryServiceThread(threading.Thread):
def __init__(self, timeout=1):
self.timeout = timeout
self._stopevent = threading.Event()
threading.Thread.__init__(self)
def run(self):
global INIT
name = self.getName()
threads[name] = "DirectoryService"
log_msg(3, "*** THREAD: %s %s" % (threads[name], name))
subthreads = []
for target in neighbors_recv.keys():
thread = GetBurstThread(target)
thread.setDaemon(1)
thread.start()
subthreads.append(thread)
# start the directory-side web server
DirectoryRequestHandler.protocol_version = "HTTP/1.0"
dir_httpd = ThreadingHTTPServer(('', DIR_PORT), DirectoryRequestHandler)
sa = dir_httpd.socket.getsockname()
INIT = 1
log_msg(2, "*** serving HTTP on %s:%s." % (sa[0], sa[1]))
handle_callbacks_individually(dir_httpd, processing_dir=1)
del threads[name]
def join(self, timeout=None):
self._stopevent.set()
threading.Thread.join(self, timeout)
class GetBurstThread(threading.Thread):
def __init__(self, target, timeout=1):
self.target = target
self.timeout = timeout
self._stopevent = threading.Event()
threading.Thread.__init__(self)
def run(self):
name = self.getName()
threads[name] = "GetBurst"
log_msg(3, "*** THREAD: %s %s" % (threads[name], name))
log_msg(2, "*** BURST PHASE 1: %s" % self.target)
if neighbors_send.has_key(self.target):
log_msg(2, "DIR BURST: %s" % self.target)
dh, dp = neighbors_send[self.target]
selector = "/blossom/burst"
try:
h = httplib.HTTP(dh, dp)
h.putrequest('GET', selector)
h.endheaders()
errcode, errmsg, headers = h.getreply()
log_msg(2, "<-- HTTP-GET %s:%s result: %s" % (dh, dp, str(errcode)))
except socket.error:
log_msg(1, "HTTP-GET FAILED connecting to %s:%s" % (dh, dp))
log_msg(1, "DIR BURST %s:%s FAILED" % (dh, dp))
return
log_msg(2, "DIR BURST %s:%s PROCEEDING" % (dh, dp))
queue = ""
while 1:
try:
line = h.file.readline()
except AttributeError, e:
log_msg(1, "%s" % repr(e))
break
except EOFError, e:
log_msg(1, "%s" % repr(e))
break
if not line:
break
# remove \r characters
m = re.search(r'^(.*)\r$', line)
if m:
line = m.group(1)
queue += line
entries = parse_queue(queue)
lines = ["directory-update %s %s" % (self.target, dp)]
log_msg(2, "<-- BURST %s:%s [%s]" % (dh, dp, len(queue)))
for node in entries.keys():
tokens = entries[node]
if tokens.has_key("summary"):
log_msg(3, "%s" % tokens["summary"])
lines.append(tokens["summary"])
if tokens.has_key("compiled-metadata"):
log_msg(3, "%s" % tokens["compiled-metadata"])
lines.append(tokens["compiled-metadata"])
if tokens.has_key("directory"):
log_msg(3, "%s" % tokens["directory"])
lines.append(tokens["directory"])
if tokens.has_key("router"):
log_msg(3, "router %s" % node)
lines.extend(tokens["router"].split("\n"))
if not tokens.has_key("blossom-path"):
tokens["blossom-path"] = "blossom-path %s" % node
log_msg(3, "%s" % tokens["blossom-path"])
lines.append(tokens["blossom-path"])
if tokens.has_key("metadata"):
log_msg(3, "%s" % tokens["metadata"])
lines.append(tokens["metadata"])
if not tokens.has_key("router-advertisement"):
tokens["router-advertisement"] = "router-advertisement %s" % node
log_msg(3, "%s" % tokens["router-advertisement"])
lines.append(tokens["router-advertisement"])
parse_update(lines)
log_msg(2, "DIR BURST %s:%s COMPLETED" % (dh, dp))
log_msg(2, "*** BURST PHASE 1: %s COMPLETED" % self.target)
log_msg(2, "*** BURST PHASE 2: %s" % self.target)
send_updates(generate_directory_report("", "/tor/"))
log_msg(2, "*** BURST PHASE 2: %s COMPLETED" % self.target)
del threads[name]
def join(self, timeout=None):
self._stopevent.set()
threading.Thread.join(self, timeout)
class TorEventHandler(EventHandler):
def circ_status(self, eventtype, circID, status, path):
"""Called when a circuit status changes if listening to CIRCSTATUS
events. 'status' is a member of CIRC_STATUS; circID is a numeric
circuit ID, and 'path' is the circuit's path so far as a list of
names.
"""
global semaphore
curr_time = int(time.time())
text = "CIRC %s %s %s %s\n" % (curr_time, circID, status, ",".join(path))
log_msg(2, "CIRC %s %s %s" % (circID, status, ",".join(path)))
process_line(text)
if len(path) > 0:
exit = path[-1]
else:
exit = ""
if status == "BUILT":
log_msg(3, "--- pending_streams: %s" % repr(pending_streams))
semaphore[circID] = 1
if pending_streams.has_key(circID):
for streamID in pending_streams[circID]:
attach_stream(streamID, circID)
del pending_streams[circID]
if status in ("FAILED", "CLOSED"):
if status == "FAILED":
semaphore[circID] = -1
if persist_id.has_key(circID):
establish_persistent_connection(persist_id[circID])
if pending_streams.has_key(circID):
for streamID in pending_streams[circID]:
if query_streams.has_key(streamID):
self.stream_status("STREAM", "DETACHED", streamID, query_streams[streamID])
def stream_status(self, eventtype, status, streamID, target, circID="0"):
"""Called when a stream status changes if listening to STREAMSTATUS
events. 'status' is a member of STREAM_STATUS; streamID is a
numeric stream ID, and 'target' is the destination of the stream.
"""
global conn
curr_time = int(time.time())
text = "STREAM %s %s %s %s %s\n" % (curr_time, status, streamID, target, circID)
log_msg(2, "STREAM %s %s %s %s" % (status, streamID, target, circID))
process_line(text)
if BLOSSOM and status in ("NEW", "NEWRESOLVE", "DETACHED"):
try:
dest = ""
fail = 0
query = 0
# avoid attempting to reattach the same stream infinitely many times
if status == "DETACHED":
if not detached_streams.has_key(streamID):
detached_streams[streamID] = 0
detached_streams[streamID] += 1
if detached_streams[streamID] > MAXREATTACH:
conn.close_stream(streamID, 1)
raise MaxReattachError
# preserve queries even if attachment failed previously
if query_streams.has_key(streamID):
target = query_streams[streamID]
log_msg(2, "--- target: %s" % target)
m = re.match(r'^([A-Za-z0-9-.]+\.)?q\.((([A-Za-z0-9-]+)\.)+)blossom:([0-9]+)$', target)
if m:
circID = 0
query = 1
query_streams[streamID] = target
rh = m.group(1)[:-1]
query = m.group(2)[:-1].split(".")
q_port = m.group(5)
log_msg(2, "--- QUERY: %s (%s)" % (",".join(query), q_port))
country = ""
isp = ""
for elt in query:
m = re.match(r'^([a-z])-(.*)$', elt.lower())
if m:
k = m.group(1)
v = m.group(2)
if k == "c":
country = v
if k == "i":
isp = v
allrtrs = {}
for rtr in cc.keys():
allrtrs[rtr] = 1
if country:
log_msg(2, "*** requested country: %s" % country)
for rtr in allrtrs.keys():
if cc[rtr] != country:
del allrtrs[rtr]
if isp:
log_msg(2, "*** requested ISP: %s" % isp)
for rtr in allrtrs.keys():
if network[rtr].lower() != isp:
del allrtrs[rtr]
log_msg(3, "--- allrtrs.keys(): %s" % allrtrs.keys())
if status != "DETACHED":
for c_circID in circuits.keys():
if len(circuits[c_circID][PATH]):
last_hop = circuits[c_circID][PATH][-1].lower()
log_msg(3, "--- last_hop: %s" % last_hop)
if allrtrs.has_key(last_hop):
circID = c_circID
dest = last_hop
break
if status == "DETACHED" or not dest:
dest = select_random(streamID, allrtrs.keys(), q_port)
interesting_ports.append(q_port)
if dest:
log_msg(2, "--- selected destination: %s" % dest)
target = rh
conn.redirect_stream(streamID, target)
else:
fail = 1
m = re.search(r'^(.*\.)?([A-Za-z0-9-]+)\.exit(:[0-9]+)?$', target)
if m:
dest = m.group(2)
if fingerprint.has_key(dest):
log_msg(2, "*** converting fingerprint %s -> %s" % (dest, fingerprint[dest]))
dest = fingerprint[dest]
# test for unconverted fingerprints
if len(dest) > MAXNICKLEN and re.match(r'^[0-9A-F]+$', dest):
dest = ""
log_msg(2, "--- normal destination: %s" % dest)
# exit to Blossom node
if dest and INIT and "%s.exit" % dest != WEB_STATUS and not tor_nodes.has_key(dest):
circID = 0
if not dest:
dest = m.group(1)
log_msg(2, "*** BLOSSOM: requested circuit to %s" % dest)
for c_circID in circuits.keys():
if len(circuits[c_circID][PATH]) and circuits[c_circID][PATH][-1] == dest:
circID = c_circID
break
log_msg(3, "--- INIT: %s, circID: %s, dest: %s" % (INIT, circID, dest))
if INIT and not circID and dest != NICK:
log_msg(3, "--- local: %s" % repr(local))
log_msg(3, "--- path: %s" % repr(path))
if not local.has_key(dest):
log_msg(2, "*** missing descriptor for router: %s" % dest)
log_msg(3, "--- summary.keys(): %s" % summary.keys())
text = ""
thread = SearchThread(streamID, dest, target)
thread.setDaemon(1)
thread.start()
circID = -1
if not circID and dest != NICK:
try:
seq = []
if path.has_key(dest):
log_msg(2, "*** path[%s]: %s" % (dest, path[dest]))
for rtr in path[dest]:
if rtr != NICK:
seq.append(rtr)
seq.append(dest)
circID = conn.extend_circuit(circID, seq)
except ErrorReply, e:
log_msg(1, "%s" % e)
if circuits.has_key(circID) and circuits[circID][BUILT]:
log_msg(2, "*** BLOSSOM: A attaching %s to %s" % (streamID, circID))
log_msg(2, "*** target: %s" % target)
m = re.search(r'^(.*\.)?([A-Za-z0-9-]+)\.exit:([0-9]+)?$', target)
if m:
cur_addr = m.group(1)
cur_exit = m.group(2)
try:
if m.group(1) and cur_addr:
conn.redirect_stream(streamID, cur_addr[:-1])
elif addr.has_key(cur_exit):
conn.redirect_stream(streamID, addr[cur_exit])
except ErrorReply:
log_msg(1, "cannot redirect stream %s" % streamID)
attach_stream(streamID, circID)
elif circID:
if not pending_streams.has_key(circID):
pending_streams[circID] = []
log_msg(2, "*** BLOSSOM: A queueing %s for attachment to %s" \
% (streamID, circID))
pending_streams[circID].append(streamID)
elif circID == 0:
log_msg(2, "*** BLOSSOM: cannot compose circuit for stream %s" % streamID)
elif circuits.has_key(circID) and circuits[circID][BUILT]:
log_msg(2, "*** BLOSSOM: B attaching %s to %s" % (streamID, circID))
attach_stream(streamID, circID)
elif query and dest:
# query Blossom request
log_msg(2, "*** BLOSSOM: received query %s %s %s" % (streamID, dest, target))
log_msg(2, "--- circID: %s" % circID)
if not int(circID):
circID = conn.extend_circuit(0, [dest])
log_msg(2, "--- circID: %s" % circID)
if not pending_streams.has_key(circID):
pending_streams[circID] = []
log_msg(2, "*** BLOSSOM: B queueing %s for attachment to %s" % (streamID, circID))
pending_streams[circID].append(streamID)
elif fail:
log_msg(2, "*** BLOSSOM: cannot create circuit for stream %s" % streamID)
conn.close_stream(streamID, 1)
else:
# ordinary Tor request
log_msg(2, "*** BLOSSOM: delegating management for stream %s" % streamID)
attach_stream(streamID, 0)
except ErrorReply, e:
log_msg(1, "%s" % e)
except MaxReattachError:
log_msg(1, "MAXIMUM NUMBER OF REATTACHMENTS EXCEEDED")
except TorCtlClosed:
log_msg(1, "CONTROLLER CONNECTION CLOSED: %s" % repr(ex))
CONNECTION_CLOSED = 1
except:
log_msg(1, "stream_status unexpected: %s" % sys.exc_info()[0])
log_msg(2, "*** internal processing complete for stream %s" % streamID)
if status in ("FAILED", "CLOSED"):
if query_streams.has_key(streamID):
del query_streams[streamID]
class ClientRequestHandler(BaseHTTPRequestHandler):
server_version = "Blossom/" + __version__
def do_GET(self):
global AUTOREFRESH
done = 0
output = ""
try:
if self.path[-4:] == ".css" or self.path[-4:] == ".gif":
try:
f = open("%s%s" % (F_ROOT, self.path))
data = f.read()
self.send_response(200)
if self.path[-4:] == ".css":
self.send_header("Content-type", "text/css")
else:
self.send_header("ETag", "0-0-0-0")
self.send_header("Content-type", "image/gif")
except:
log_msg(1, "secondary do_GET A unexpected: %s" % sys.exc_info()[0])
data = "404 File Not Found"
self.send_response(404)
self.send_header("Content-type", "text/plain")
elif self.path == "/":
AUTOREFRESH = 0
data = generate_output(AUTOREFRESH)
self.send_response(200)
self.send_header("Content-type", "text/html")
elif self.path == "/autorefresh":
AUTOREFRESH = -1
data = generate_output(AUTOREFRESH)
self.send_response(200)
self.send_header("Content-type", "text/html")
elif self.path == "/network-status":
data = generate_network_status()
self.send_response(200)
self.send_header("Content-type", "text/html")
elif len(self.path) > 9 and self.path[:10] == "/attach?q=":
data = generate_output(int(self.path[10:]))
self.send_response(200)
self.send_header("Content-type", "text/html")
elif len(self.path) > 8 and self.path[:9] == "/connect?":
vals = {}
args = self.path[9:].split("&")
for arg in args:
k, v = arg.split("=")
vals[k] = v
if vals["c"] and vals["s"]:
try:
sh, sp = parseHostAndPort(TORCONTROL)
attach_stream(int(vals["s"]), int(vals["c"]))
except:
log_msg(1, "secondary do_GET B unexpected: %s" % sys.exc_info()[0])
log_msg(2, "*** attach %s to %s unsuccessful" % (vals["s"], vals["c"]))
time.sleep(TIME_ACTION)
data = generate_output(AUTOREFRESH)
self.send_response(200)
self.send_header("Content-type", "text/html")
elif not done:
data = "404 File Not Found"
self.send_response(404)
self.send_header("Content-type", "text/plain")
if not done:
self.send_header("Content-Length", len(data))
self.end_headers()
self.wfile.write(data)
except KeyboardInterrupt:
log_msg(1, "exiting on ^C [%s]\n" % get_curr_time())
sys.exit(0)
except:
log_msg(1, "secondary do_GET C unexpected: %s" % sys.exc_info()[0])
def do_POST(self):
global conn
curr_time = time.time()
length = int(self.headers.getheader('content-length'), 10)
data = self.rfile.read(length)
lines = data.split("&")
circID = 0
for line in lines:
k, v = line.split("=")
# smite a circuit
if k == "c_smite":
circID = int(v)
try:
# sh, sp = parseHostAndPort(TORCONTROL)
conn.close_circuit(circID)
except:
log_msg(1, "secondary do_POST A unexpected: %s" % sys.exc_info()[0])
log_msg(2, "*** circuit smite %s unsuccessful" % circID)
# smite a stream
if k == "s_smite":
streamID = int(v)
try:
# sh, sp = parseHostAndPort(TORCONTROL)
conn.close_stream(streamID)
except:
log_msg(1, "secondary do_POST B unexpected: %s" % sys.exc_info()[0])
log_msg(2, "*** stream smite %s unsuccessful" % streamID)
time.sleep(TIME_ACTION)
data = generate_output(AUTOREFRESH)
self.send_response(200)
self.send_header("Content-type", "text/html")
self.send_header("Content-Length", len(data))
self.end_headers()
self.wfile.write(data)
def log_request(self, code='-', size='-'):
pass
class DirectoryRequestHandler(BaseHTTPRequestHandler):
server_version = "Blossom/" + __version__
def do_GET(self):
curr_time = time.time()
log_msg(2, "<-- HTTP-GET %s from %s:%s" \
% (self.path, self.client_address[0], self.client_address[1]))
# purge particularly old routers
for router in update_time.keys():
if curr_time - update_time[router] > DIR_DESC_DELETE:
del update_time[router]
if desc.has_key(router):
del desc[router]
if dir_fingerprint.has_key(router):
del dir_fingerprint[router]
if dir_path.has_key(router):
del dir_path[router]
if metadata.has_key(router):
del metadata[router]
if router_adv.has_key(router):
del router_adv[router]
reap_disconnected_neighbors(curr_time)
target = "%s:%s" % (self.client_address[0], self.client_address[1])
s_path = self.path
output = generate_directory_report(target, s_path)
try:
length = len(output)
self.send_response(200)
self.send_header("Content-type", "text/plain")
self.send_header("Content-Length", length)
self.end_headers()
self.wfile.write(output)
log_msg(2, "<-- HTTP-GET %s from %s:%s COMPLETED [%s]" \
% (self.path, self.client_address[0], self.client_address[1], len(output)))
except:
log_msg(1, "<-- HTTP-GET %s from %s:%s FAILED: %s" \
% (self.path, self.client_address[0], self.client_address[1],
sys.exc_info()[0]))
def do_POST(self):
log_msg(3, "<-- HTTP-POST %s from %s:%s" \
% (self.path, self.client_address[0], self.client_address[1]))
try:
curr_time = time.time()
length = int(self.headers.getheader('content-length'), 10)
data = self.rfile.read(length)
lines = data.split("\n")
client_addr, client_port = self.client_address
reap_disconnected_neighbors(curr_time)
if self.path == "/blossom/":
parse_blossom(lines, "")
elif self.path == "/blossom/directory-update": # BLOSSOM DIRECTORY
parse_update(lines)
log_msg(2, "<-- HTTP-POST %s from %s:%s COMPLETED [%s]" \
% (self.path, self.client_address[0], self.client_address[1], length))
except KeyboardInterrupt:
log_msg(1, "exiting on ^C [%s]\n" % get_curr_time())
sys.exit(0)
def log_request(self, code='-', size='-'):
pass
def TorCloseHandler(ex):
global CONNECTION_CLOSED
try:
raise ex
except TorCtlClosed:
log_msg(1, "*** CONTROLLER CONNECTION CLOSED: %s" % repr(ex))
CONNECTION_CLOSED = 1
except:
pass
def log_msg(debugval, msg):
if DEBUG >= debugval:
print "%s [%s] %s" % (get_curr_time()[11:], LOGLEVEL[debugval], msg)
def attach_stream(streamID, circID):
try:
conn.attach_stream(streamID, circID)
except ErrorReply, e:
e = "%s" % e
log_msg(1, "%s [%s][%s]" % (e, streamID, circID))
if re.search(r'^552 ', e):
pass
elif re.search(r'^555 ', e) and BLOSSOM:
conn.authenticate("")
conn.set_option("__leavestreamsunattached", "1")
else:
log_msg(1, "unknown error: ignoring")
def blossom():
if BLOSSOM:
return "Blossom"
else:
return "Tor"
def select_random(streamID, list, qp):
if policy_time.has_key(qp):
log_msg(2, "--- policy_time[%s]: %s" % (qp, policy_time[qp]))
log_msg(2, "--- candidates (before screening): %s" % len(list))
newlist = []
if policy_time.has_key(qp) and policy_time[qp] + TIME_POLICY > time.time():
for rtr in list:
if policy[qp].has_key(rtr):
newlist.append(rtr)
list = newlist
log_msg(2, "--- candidates (policy screening): %s" % len(list))
newlist = []
for rtr in list:
if not (attempted.has_key(streamID) and attempted[streamID].has_key(rtr)):
newlist.append(rtr)
list = newlist
log_msg(2, "--- candidates (full screening): %s" % len(list))
if list:
n = int(random.random()*len(list))
selection = list[n]
if not attempted.has_key(streamID):
attempted[streamID] = {}
attempted[streamID][selection] = 1
return selection
else:
return []
def format_summary(one_summary):
rtr_list = []
for rtr in one_summary.keys():
rtr_list.append("%s=%s" % (rtr, one_summary[rtr]))
rtr_list.sort()
return ",".join(rtr_list)
def stream_attach(streamID, attached):
if attached or closed_streams.has_key(streamID):
return streams[streamID]
else:
return "%s" % (streamID, streams[streamID])
def stream_status(streamID, message):
if closed_streams.has_key(streamID):
return " ![](\"%s\")
%s " \
% (IMG_SIZE, ICON_BLANK, message)
else:
return " %s " \
% (streamID, IMG_SIZE, ICON_SMITE, message)
def sort_numerically(array):
for i in range(0,len(array)):
array[i] = int(array[i])
array.sort()
for i in range(0,len(array)):
array[i] = "%d" % array[i]
return array
def get_curr_time():
curr_time = time.gmtime()
return "%04d-%02d-%02d %02d:%02d:%02d" % curr_time[0:6]
def icon_built(circID):
if circuits.has_key(circID) and circuits[circID][BUILT]:
if len(circuits[circID][STREAMS]) == 0:
icon = ICON_BUILT_0
else:
icon = ICON_BUILT_1
else:
icon = ICON_UNBUILT
return "" % (circID, IMG_SIZE, icon)
def icon_cc(path):
a = ""
for rtr in path:
ccs = "~~"
if cc.has_key(rtr.lower()):
ccs = cc[rtr.lower()]
icon = "![](\"%s/%s.gif\")
" % (IMG_SIZE, URL_FLAGS, ccs)
else:
icon = " " % (IMG_SIZE, ICON_BUILT_1)
if cc_name.has_key(ccs):
icon = "%s" % (cc_name[ccs], icon)
a += icon
return a
def icon_bw(path):
a = ""
min = 1<<16
for rtr in path:
if bw.has_key(rtr) and bw[rtr.lower()] < min:
min = bw[rtr.lower()]
if min >= 400:
r = "" % (IMG_SIZE, ICON_V3)
elif min >= 60:
r = "" % (IMG_SIZE, ICON_V2)
elif min >= 10:
r = "" % (IMG_SIZE, ICON_V1)
else:
r = "" % (IMG_SIZE, ICON_V0)
if min == 1<<16:
min = "unknown"
return "%s" % (min, r)
def auto_link(b):
if b == -1:
return "autorefresh stop"
else:
return "reload autorefresh start"
def auto_meta(b):
if b == -1:
return " 0:
return "%s" % (streamID, circID, a)
else:
return a
def generate_directory_report(target, s_path):
curr_time = time.time()
output = ""
s_pub = "published %s\n" % time.strftime("%Y-%m-%d %H:%M:%S", time.gmtime())
s_status = "router-status"
# report relatively recent routers
for router in update_time.keys():
router_status = ""
if curr_time - update_time[router] > DIR_DESC_EXPIRATION:
router_status = "!"
s_status += " %s%s=$%s" % (router_status, router, dir_fingerprint[router])
s_status += "\n"
if s_path in ["/tor/", "/blossom/burst"]:
# provide list of descriptors
output += "unsigned-directory %s %s\n" % (NICK, __version__)
output += s_pub
output += "running-routers"
for router in dir_fingerprint.keys():
output += " %s" % router
output += "\n"
output += s_status
dirs = selection.keys()
dirs.sort()
for dir in dirs:
if dir_port.has_key(dir) and dir_prop[selection[dir]].has_key(dir):
output += "directory %s %s %s\n" \
% (dir, dir_port[dir], ",".join(dir_prop[selection[dir]][dir]))
else:
log_msg(1, "dir_port[%s] or dir_prop[%s][%s] missing" \
% (dir, selection[dir], dir))
dirs = dir_summary.keys()
dirs.sort()
for dir in dirs:
output += "summary %s %s\n" % (dir, format_summary(dir_summary[dir]))
dirs = dir_metadata.keys()
dirs.sort()
for dir in dirs:
output += "compiled-metadata %s %s\n" % (dir, ",".join(dir_metadata[dir]))
output += "\n"
if s_path == "/tor/":
rtrs = dir_path.keys()
rtrs.sort()
for rtr in rtrs:
if dir_path[rtr]:
output += "blossom-path %s %s\n" % (rtr, ",".join(dir_path[rtr]))
rtrs = metadata.keys()
rtrs.sort()
for rtr in rtrs:
if metadata[rtr]:
output += "metadata %s %s\n" % (rtr, ",".join(metadata[rtr]))
rtrs = router_adv.keys()
rtrs.sort()
for rtr in rtrs:
output += "router-advertisement %s %s\n" % (rtr, ",".join(router_adv[rtr]))
output += "\n"
if desc.has_key(NICK):
log_msg(3, "<-- sending router per request: %s" % router)
output += desc[NICK]
if s_path == "/tor/":
for router in desc.keys():
if router != NICK:
log_msg(3, "<-- sending router per request: %s" % router)
output += desc[router]
elif s_path == "/tor/running-routers":
# provide forwarder availability information
output += "running-routers\n"
output += s_pub
output += s_status
else:
m = re.match(r'^/blossom/([0-9A-Za-z]*)$', s_path)
if m:
# perform Blossom query
output = ""
target = m.group(1)
if len(target) <= MAXNICKLEN:
if desc.has_key(target):
if dir_path.has_key(target):
output += "blossom-path %s %s\n" \
% (target, ",".join(dir_path[target]))
if metadata.has_key(target):
output += "metadata %s %s\n" \
% (target, ",".join(metadata[target]))
if router_adv.has_key(target):
output += "router-advertisement %s %s\n" \
% (target, ",".join(router_adv[target]))
output += "\n%s" % desc[target]
else:
relevant = {}
for dir in dir_summary.keys():
if dir_summary[dir].has_key(target):
relevant[dir] = 1
output += "summary %s %s\n" \
% (dir, format_summary(dir_summary[dir]))
for dir in relevant.keys():
if desc.has_key(dir):
if dir_path.has_key(dir):
output += "blossom-path %s %s\n" \
% (dir, ",".join(dir_path[dir]))
if metadata.has_key(dir):
output += "metadata %s %s\n" \
% (dir, ",".join(metadata[dir]))
if router_adv.has_key(dir):
output += "router-advertisement %s %s\n" \
% (dir, ",".join(router_adv[dir]))
if selection.has_key(dir):
output += "directory %s %s %s\n" \
% (dir, dir_port[dir], ",".join(dir_prop[selection[dir]][dir]))
output += "\n%s" % desc[dir]
output += "end\n\r"
return output
def establish_persistent_connection(b):
global conn
if persist_nickname.has_key(b):
try:
log_msg(2, "*** establishing persistent connection: %s" % b)
ret = conn.extend_circuit(0, [persist_nickname[b]])
persist_id[ret] = b
if unestablished.has_key(b):
del unestablished[b]
except ErrorReply, e:
log_msg(1, "%s" % e)
# publish my descriptor to the specified directory
def publish_descriptor(b):
f = 1
postable = ""
resource = "desc/name/" + NICK
log_msg(3, "*** publish descriptor for %s to %s" % (NICK, b))
try:
postable += get_info(resource)[0]
except struct.error, e:
log_msg(1, "publish_descriptor struct.error %s" % repr(e))
return
except ProtocolError, (ec, em):
log_msg(1, "publish_descriptor %s" % em)
return
except AttributeError:
log_msg(1, "publish_descriptor AttributeError")
return
postable += "\n"
if PERSIST:
if persist_nickname.has_key(b):
postable += "blossom-path %s %s\n" % (NICK, persist_nickname[b])
else:
log_msg(2, "warning: %s not in %s" % (b, persist_nickname.keys()))
f = 0
else:
postable += "blossom-path %s\n" % NICK
if META_LOCAL:
postable += "metadata %s %s\n" % (NICK, ",".join(META_LOCAL))
# publish descriptor and applicable Blossom metadata and path information
if f:
thread = HTTPPostThread(b, "/blossom/", postable)
thread.setDaemon(1)
thread.start()
log_msg(2, "*** publish descriptor for %s to %s SUCCEEDED" % (NICK, b))
else:
log_msg(1, "*** publish descriptor for %s to %s FAILED" % (NICK, b))
return
def report_streams(time, select_stream, s_streams, count):
content = ""
sorted_keys = sort_numerically(s_streams.keys())
attached = 0
for streamID in sorted_keys:
s_status = ""
if count:
attached = 1
s_status = "SUCCEEDED"
if streams.has_key(streamID) and not counted_streams.has_key(streamID):
td_class = ""
if int(streamID) == select_stream:
td_class = " class=\"heading\""
s_status = "SELECTED"
elif s_streams[streamID] == "SENTCONNECT":
td_class = " class=\"boldnormal\""
s_status = "SENTCONNECT"
attached = 0
if closed_streams.has_key(streamID):
if time - TIME_SAVE > closed_streams[streamID]:
log_msg(3, "*** time %s closed at %s" % (time, closed_streams[streamID]))
del streams[streamID]
del closed_streams[streamID]
if failed_streams.has_key(streamID):
del failed_streams[streamID]
continue
else:
if failed_streams.has_key(streamID):
td_class = " class=\"failedstream\""
s_status = "FAILED"
else:
td_class = " class=\"closedstream\""
s_status = "CLOSED"
content += """
%s
%s
%s
""" % ( td_class,
stream_status(streamID, s_status),
td_class,
streamID,
td_class,
stream_attach(streamID, attached)
)
if count:
counted_streams[streamID] = 1
return content
def generate_output(arg):
global circuits
global counted_streams
global pending_streams
global query_streams
global streams
counted_streams = {}
int_time = int(time.time())
curr_time = get_curr_time()
content = ""
select_stream = -1
refresh = 0
if arg == -1:
refresh = arg
elif arg > 0:
select_stream = arg
# report open circuits
s_circuits = circuits.keys()
s_circuits = sort_numerically(s_circuits)
for circID in s_circuits:
td_class = "dimleftentry"
if circuits.has_key(circID) and circuits[circID][BUILT]:
td_class = "leftentry"
content += """
| %s %s %s |
%s |
%s |
""" % (td_class,
icon_built(circID),
icon_bw(circuits[circID][PATH]),
icon_cc(circuits[circID][PATH]),
td_class,
circID,
td_class,
attach_link(select_stream, circID, ",".join(circuits[circID][PATH]))
)
# report streams associated with this circuit
content += report_streams(int_time, select_stream, circuits[circID][STREAMS], 1)
# report unattached streams
unattached_report = report_streams(int_time, select_stream, streams, 0)
if len(counted_streams.keys()) < len(streams.keys()):
content += """
 |
|
[unattached] |
%s""" % (ICON_BLANK, ICON_BLANK, unattached_report)
output = """
%s Client Status
%s
%s Client Status
current time: %s
%s version %s
""" % ( blossom(), \
auto_meta(refresh), \
blossom(), \
curr_time, \
content, \
auto_link(refresh), \
__version__ \
)
return output
def generate_network_status():
log_msg(2, "*** generating network status page")
try:
status_list = conn.get_info("dir/status/all")
server_list = conn.get_info("dir/server/all")
except ErrorReply, e:
log_msg(1, e)
except:
pass
data = status_list + server_list
return data
def search(streamID, dest, target):
global semaphore
global summary_remote # WARNING: this is very very bad
seq = []
desc = {}
summary_remote = {}
fail = 0
circID = 0
try:
sh, sp = parseHostAndPort(HTTP_PROXY)
except:
log_msg(1, "must use HTTP proxy for queries")
return
log_msg(2, "*** search to %s" % dest)
## -- THESIS EXPERIMENT SECTION --
explicit = ""
m = re.match(r'([0-9a-zA-Z]+)-e', dest)
if m:
log_msg(1, "EXPLICIT")
explicit = m.group(1)
num_hops = 0
query_server = 0
m = re.match(r'([0-9]+)-([qs])', dest)
if m:
if m.group(2) == 'q':
query_server = 1
num_hops = int(m.group(1))
# random nodeset 0
# target_array = ['80708b47', '80708b6a', '815d4438', '81aad6c0', '825c46fb', '82c240a3', '82cb7f28', '84e34a28', '84fc98c2']
# random nodeset 1
# target_array = ['8004240b', '8006c09e', '80708b6a', '80723f0e', '80df0671', '824b5753', '8441f067', '8441f068', '8a17cce8', '8ff88baa']
target_array = ['ithaca', 'ithaca', '80708b6a', '80723f0e', '80df0671', '824b5753', '8441f067', '8441f068', '8a17cce8', '8ff88baa']
## -- END THESIS EXPERIMENT SECTION --
while not num_hops and not explicit and not desc.has_key(dest) and not fail:
chosen_dir = ""
min = MAXINT
if not dest in target_array:
for dir in summary.keys():
if summary[dir].has_key(dest) and not seq.__contains__(dir):
log_msg(2, "--- prop.keys(): %s" % prop.keys())
if prop.has_key(dir):
if len(prop[dir]) < min:
min = len(prop[dir])
chosen_dir = dir
elif seq:
min = 0
chosen_dir = dir
if not chosen_dir:
for dir in summary_remote.keys():
if summary_remote[dir].has_key(dest) and not seq.__contains__(dir):
log_msg(2, "--- prop.keys(): %s" % prop.keys())
chosen_dir = dir
if chosen_dir == dest:
break
if not chosen_dir:
fail = 1
else:
log_msg(2, "*** selecting directory: %s" % chosen_dir)
seq.append(chosen_dir)
if not dest in target_array:
if prop.has_key(chosen_dir):
reversed = prop[chosen_dir]
reversed.reverse()
for elt in reversed:
seq.append(elt)
log_msg(2, "*** stream %s current route to %s: %s" % (streamID, dest, repr(seq)))
log_msg(3, "--- port: %s" % repr(port))
url = ""
if port.has_key(chosen_dir):
chosen_server = "%s:%s" % (chosen_dir, port[chosen_dir])
url = "http://%s.exit:%s/blossom/%s" % (chosen_dir, port[chosen_dir], dest)
else:
log_msg(1, "NO PORT AVAILABLE for %s" % chosen_dir)
try:
h = httplib.HTTP(sh, sp)
h.putrequest('GET', url)
h.endheaders()
errcode, errmsg, headers = h.getreply()
log_msg(2, "<-- HTTP-GET %s:%s %s: %s" % (sh, sp, url, str(errcode)))
h = h.file
desc = process_descriptors(h, "")
except AttributeError, e:
log_msg(1, "AttributeError")
except socket.error, (ec, em):
log_msg(1, "socket.error %s %s" % (repr(ec), repr(em)))
except IOError, (ec, em):
log_msg(1, "IOError %s %s" % (repr(ec), repr(em)))
if len(seq) > MAXLEN:
fail = 1
if fail:
log_msg(2, "*** BLOSSOM: unreachable destination %s" % dest)
try:
conn.close_stream(streamID, 1)
except ErrorReply, e:
log_msg(1, e)
## -- THESIS EXPERIMENT SECTION --
elif num_hops:
circID = 0
for i in range(1, num_hops):
chosen_dir = target_array[i % len(target_array)]
ff = 0
if circID:
ff = 1
semaphore[circID] = 0
circID = conn.extend_circuit(circID, [chosen_dir])
if ff:
start_time = time.time()
while(semaphore[circID] == 0):
time.sleep(0.1)
if semaphore.has_key(circID) and semaphore[circID] == -1:
log_msg(1, "--- circuit failed")
del semaphore[circID]
break
if query_server:
if port.has_key(chosen_dir):
chosen_server = "%s:%s" % (chosen_dir, port[chosen_dir])
url = "http://%s.exit:%s/blossom/%s" % (chosen_dir, port[chosen_dir], dest)
else:
log_msg(1, "NO PORT AVAILABLE for %s" % chosen_dir)
try:
h = httplib.HTTP(sh, sp)
h.putrequest('GET', url)
h.endheaders()
errcode, errmsg, headers = h.getreply()
log_msg(2, "<-- HTTP-GET %s:%s %s: %s" % (sh, sp, url, str(errcode)))
h = h.file
desc = process_descriptors(h, "")
except AttributeError, e:
log_msg(1, "AttributeError")
except socket.error, (ec, em):
log_msg(1, "socket.error %s %s" % (repr(ec), repr(em)))
except IOError, (ec, em):
log_msg(1, "IOError %s %s" % (repr(ec), repr(em)))
if semaphore.has_key(circID):
conn.close_circuit(circID)
conn.close_stream(streamID, 1)
## -- END THESIS EXPERIMENT SECTION --
else:
log_msg(2, "*** BLOSSOM: reachable destination %s" % dest)
try:
if explicit:
seq = [explicit]
else:
if path.has_key(dest):
log_msg(2, "*** path[%s]: %s" % (dest, path[dest]))
for rtr in path[dest]:
if rtr != chosen_dir and rtr != NICK:
seq.append(rtr)
seq.append(dest)
if not circID and len(seq) > 2:
log_msg(2, "*** EXTENDING for search")
c_from = seq[-2]
c_to = seq[-1]
for e_circID in circuits.keys():
log_msg(2, "--- e_circID: %s %s" % (e_circID, circuits[e_circID][PATH]))
if circuits[e_circID][PATH] and circuits[e_circID][PATH][-1] == c_from:
log_msg(2, "--- choosing e_circID: %s" % e_circID)
circID = e_circID
seq = [dest]
break
log_msg(2, "--- chosing seq: DONE")
# extend the circuit
log_msg(2, "*** final sequence for circ %s: %s" % (circID, repr(seq)))
circID = conn.extend_circuit(circID, seq)
# redirect stream if this is a first-time request to a new router
if re.match(r'^\.', target):
if desc.has_key(dest):
line = desc[dest].split("\n")[0]
log_msg(3, "--- line: %s" % line)
m = re.match(r'^\S+\s+\S+\s+([0-9]+\.[0-9]+\.[0-9]+\.[0-9]+)\s+', line)
if m:
conn.redirect_stream(streamID, m.group(1))
if circID:
if not pending_streams.has_key(circID):
pending_streams[circID] = []
log_msg(2, "*** BLOSSOM: queueing %s for attachment to %s" % (streamID, circID))
pending_streams[circID].append(streamID)
except:
log_msg(1, "search unexpected: %s" % sys.exc_info()[0])
log_msg(2, "*** external processing complete for stream %s" % streamID)
return
def process_line(line):
global circuits
global closed_streams
global counted_streams
global semaphore
global streams
# standard line format
line = re.sub(r" +", " ", line)
line = re.sub(r"\n", "", line)
item = line.split(" ")
code = item[0]
args = item[1:]
if code == "CIRC":
if len(args) == 3:
time, circID, status = args
elif len(args) == 4:
time, circID, status, c_path = args
if status == "LAUNCHED":
circuits[circID] = [[], 0, {}]
elif status == "EXTENDED":
circuits[circID] = [c_path.split(","), 0, {}]
elif status == "BUILT":
circuits[circID] = [c_path.split(","), 1, {}]
elif status == "FAILED" or status == "CLOSED":
if circuits.has_key(circID):
del circuits[circID]
elif code == "STREAM":
time, status, streamID, target, circID = args
if status == "NEW":
streams[streamID] = target
if status == "SENTCONNECT" or status == "SUCCEEDED":
if circuits.has_key(circID):
if not circuits[circID][STREAMS].has_key(streamID):
for circ in circuits:
if circuits[circ][STREAMS].has_key(streamID):
del circuits[circ][STREAMS][streamID]
circuits[circID][STREAMS][streamID] = status
if status == "DETACHED":
if circuits.has_key(circID):
if circuits[circID][STREAMS].has_key(streamID):
del circuits[circID][STREAMS][streamID]
if status == "CLOSED":
closed_streams[streamID] = int(time)
if status == "FAILED":
failed_streams[streamID] = int(time)
elif code in ["DATA", "DATA+"]:
if len(args) > 3:
name = re.sub(r"^\*", "", args[1])
cc[name] = args[0].lower()
if code == "DATA":
tor_nodes[name] = 1
if re.match(r'[0-9]+', args[2]):
bw[name] = int(int(args[2])/1000)
else:
bw[name] = 0
if len(args) > 6:
addr[name] = args[4]
network[name] = args[5]
fingerprint[args[6]] = name
elif code == "POLICY":
if len(args) > 6:
name = re.sub(r"^\*", "", args[1])
qp = args[6]
if qp != "-" and int(args[2]) > 0:
if not policy.has_key(qp):
policy[qp] = {}
log_msg(2, "--- SETTING POLICY: %s %s" % (qp, name))
policy[qp][name] = 1
elif code == "CC":
if len(args) > 1:
name = args[0]
cc_name[name] = " ".join(args[1:])
elif code == "CONSISTENCY":
check_consistency()
elif code == "FLUSH":
log_msg(2, "*** received FLUSH request")
circuits = {}
closed_streams = {}
counted_streams = {}
streams = {}
elif code == "EXIT":
raise BlossomError
return
def signal_handler(sig, id):
if sig == signal.SIGHUP:
log_msg(1, "*** received signal HUP; restarting")
main()
elif sig == signal.SIGTERM:
log_msg(1, "*** received signal TERM; exiting")
sys.exit(0)
elif sig == signal.SIGUSR1:
f = 1
c = 0
output = ""
while(f):
c = c + 1
try:
f = sys._getframe(c)
output += " %s" % f.f_lineno
except ValueError:
f = []
if output:
output = "line%s" % output
else:
output = "no traceback"
log_msg(1, "*** received signal USR1: %s" % output)
def getConnection(daemon=1):
"""
getConnection tries to open a socket to the tor server.
If a socket is established, and the daemon paramter is True (the default),
a thread is spawned to handle the communcation between us and the tor server.
"""
global NICK
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
try:
host, port = parseHostAndPort(TORCONTROL)
s.connect((host,port))
except socket.error, e:
log_msg(1, "CONNECTION FAILED: %s. Is Tor running? Is the ControlPort enabled?\n" % e)
try:
conn = get_connection(s)
except:
return
th = conn.launch_thread(daemon)
conn.authenticate("")
conn.set_event_handler(TorEventHandler())
conn.set_close_handler(TorCloseHandler)
k, v = conn.get_option("nickname")[0]
if v:
if NICK and NICK != v:
log_msg(1, "Tor nickname has changed. Please restart Blossom.")
raise BlossomError
NICK = v
else:
NICK = ""
if BLOSSOM:
conn.set_option("__leavestreamsunattached", "1")
else:
conn.set_option("__leavestreamsunattached", "0")
conn.set_events(["CIRCSTATUS", "STREAMSTATUS"])
return conn
def get_info(name):
global conn
try:
r = conn.get_info(name)
entries = r[name].split("\r\n")
n = 0
try:
while 1:
entries.remove("")
except:
pass
return entries
except AttributeError, e:
log_msg(1, "AttributeError %s" % repr(e))
conn = getConnection()
except TorCtlClosed:
log_msg(1, "*** retrying controller connection [%s]" % get_curr_time())
conn = getConnection()
time.sleep(TIME_RETRY)
return get_info(name)
def check_consistency():
global circuits
global closed_streams
err = 0
sh, sp = parseHostAndPort(TORCONTROL)
log_msg(2, "verifying consistency")
counted = {}
for ent in get_info("circuit-status"):
circID, status, path = ent.split(" ")
if circuits.has_key(circID):
log_msg(3, "circID %s OK" % circID)
counted[circID] = 1
else:
log_msg(1, "circID %s in Tor but not our records (INCONSISTENCY)" % circID)
err = 1
for circID in circuits.keys():
if not counted.has_key(circID):
log_msg(1, "circID %s in our records but not Tor (INCONSISTENCY)" % circID)
err = 1
counted = {}
for ent in get_info("stream-status"):
t = ent.split(" ")
if len(t) == 3:
streamID, status, target = t
elif len(t) == 4:
streamID, status, target, circID = t
else:
log_msg(1, "CONSISTENCY: ill-formed stream-status record")
err = 1
if streams.has_key(streamID):
log_msg(2, "streamID %s OK" % streamID)
counted[streamID] = 1
else:
log_msg(1, "streamID %s in Tor but not our records (INCONSISTENCY)" % streamID)
err = 1
for streamID in streams.keys():
if not counted.has_key(streamID) and not closed_streams.has_key(streamID):
log_msg(1, "streamID %s in our records but not Tor (INCONSISTENCY)" % streamID)
err = 1
if err:
log_msg(1, "*** CONSISTENCY CHECK FAILED")
else:
log_msg(2, "*** CONSISTENCY CHECK SUCCEEDED")
return err
# process descriptors from a file and export them to the Tor client
def process_descriptors(h, remote):
global INIT
global closed_streams
global conn
global summary_remote
desc = {}
record = ""
router = ""
log_msg(3, "*** parsing descriptors")
while 1:
try:
line = h.readline()
except AttributeError, e:
log_msg(1, repr(e))
break
except EOFError, e:
log_msg(1, repr(e))
break
if not line:
break
elif line == "\n":
if router:
desc[router] = record
if (not BLOSSOM) or (BLOSSOM and BLOSSOM.__contains__(remote)):
local[router] = 1
record = ""
router = ""
else:
# unsigned-directory
m = re.search(r'^unsigned-directory\s+(\S+)\s*(\s+(.*))?$', line)
if m:
persist_nickname[remote] = m.group(1)
INIT = 1
# blossom-path
m = re.search(r'^blossom-path\s+(\S+)\s*(\s+(\S+))?$', line)
if m:
btarget = m.group(1)
# delete path if one exists
if path.has_key(btarget):
del path[btarget]
# set a new path if one is available
if m.group(3):
bpath = m.group(3)
# enforce well-formedness of blossom-path
if re.match(r'^([0-9A-Za-z]+\,)*[0-9A-Za-z]+$', bpath):
log_msg(3, "setting path: %s %s" % (btarget, bpath))
path[btarget] = bpath.split(",")
else:
log_msg(2, "invalid path specification: %s %s" % (btarget, bpath))
continue
# metadata
m = re.search(r'^metadata\s+(\S+)\s+(\S+)$', line)
if m:
btarget = m.group(1)
bmeta = m.group(2)
if re.match(r'^[0-9a-z,-.]+$', bmeta):
log_msg(3, "setting metadata: %s %s" % (btarget, bmeta))
metadata[btarget] = bmeta.split(",")
continue
# router
m = re.search(r'^router\s+(\S+)\s+', line)
if m:
# determine router nickname
router = m.group(1).lower()
# summary
m = re.search(r'^summary\s+(\S+)\s*(\s+(\S+))?$', line)
if m:
# determine descriptors provided by this directory
dir = m.group(1)
rtr_list = []
if m.group(3):
rtr_list = m.group(3).split(",")
if remote:
summary[dir] = {}
else:
summary_remote[dir] = {}
for rtr_entry in rtr_list:
if re.search(r'=', rtr_entry):
rtr_entry = rtr_entry.split("=")
rtr = rtr_entry[0].lower()
rtr_dist = int(rtr_entry[1])
else:
rtr = rtr_entry
rtr_dist = 1
if remote:
summary[dir][rtr] = rtr_dist
else:
summary_remote[dir][rtr] = rtr_dist
# compiled-metadata
m = re.search(r'^compiled-metadata\s+(\S+)\s*(\s+(\S+))?$', line)
if m:
# determine metadata provided by this directory
dir = m.group(1)
metadata_list = []
if m.group(3):
metadata_list = m.group(3).split(",")
dir_metadata[dir] = {}
for metadata_entry in metadata_list:
dir_metadata[dir][metadata_entry] = 1
# directory
m = re.search(r'^directory\s+(\S+)\s+(\S+)\s*(\s+(\S+))?$', line)
if m:
# determine directory port and propagation information
dir = m.group(1)
port[dir] = m.group(2)
if remote:
if m.group(4):
prop[dir] = m.group(4).split(",")
else:
prop[dir] = []
record += line
# post descriptors
for router in desc.keys():
log_msg(3, "posting descriptor: " + router)
try:
conn.post_descriptor(desc[router])
except AttributeError, e:
log_msg(1, "post descriptor failed: %s" % e)
conn = getConnection()
except ErrorReply, e:
log_msg(1, e)
log_msg(3, "*** parsing descriptors COMPLETED")
return desc
# execute a routine directory fetch
def get_descriptors(b):
dh = "0.0.0.0"
dp = 80
url = "http://%s/tor/" % b
log_msg(2, "get_descriptors: %s" % url)
m = re.search(r'^([^/:]+)(:[0-9]+)?$', b)
if m:
dh = m.group(1)
if m.group(2):
dp = int(m.group(2)[1:])
url = "/tor/"
log_msg(2, "<-- HTTP-GET %s:%s %s" % (dh, dp, url))
thread = OpenURLThread(dh, dp, url)
thread.setDaemon(1)
thread.start()
return
def obtain_tor_metadata(url, policy=0):
log_msg(2, "*** attempting to retrieve Tor metadata (%s)" % policy)
dh = "0.0.0.0"
dp = 80
if HTTP_PROXY:
dh, dp = parseHostAndPort(HTTP_PROXY)
else:
m = re.search(r'^http:\/\/([^/:]+)(:[0-9]+)?(\/.*)$', url)
if m:
dh = m.group(1)
if m.group(2):
dp = int(m.group(2)[1:])
url = m.group(3)
log_msg(2, "<-- HTTP-GET %s:%s %s" % (dh, dp, url))
thread = OpenURLThread(dh, dp, url, policy=policy)
thread.setDaemon(1)
thread.start()
return
def get_tor_policy(qp):
log_msg(3, "*** get_tor_policy for port %s" % qp)
urllist = [POLICY_URL]
if BLOSSOM:
urllist.append(POLICY_URL_BLOSSOM)
if not BLOSSOM_ARGS:
urllist = []
if not policy_time.has_key(qp) or policy_time[qp] + TIME_POLICY < time.time():
for url in urllist:
p_url = "%s=%s" % (url, qp)
log_msg(2, "*** get_tor_policy: %s" % p_url)
obtain_tor_metadata(p_url, policy=1)
policy_time[qp] = time.time()
log_msg(3, "*** get_tor_policy for port %s SUCCEEDED" % qp)
def update_path(router, new_path, override):
global queue
# delete blossom-path and blossom-addr entries if applicable
if dir_path.has_key(router):
log_msg(3, "*** COMPARING (%s) path length %s <-> %s [%s <-> %s]" \
% (override, len(new_path), len(dir_path[router]), repr(new_path), repr(dir_path[router])))
if len(new_path) >= len(dir_path[router]) and not override:
return 0
del dir_path[router]
# create new blossom-path entry for this router, if applicable
log_msg(3, "*** SETTING PATH for %s: %s" % (router, repr(new_path)))
queue += "blossom-path %s %s\n" % (router, ",".join(new_path))
if new_path:
dir_path[router] = new_path
return 1
def parse_queue(queue):
lines = queue.split("\n")
entries = {}
processing_router = 0
for line in lines:
if processing_router:
if line == "":
entries[router]["router"] = new_desc
processing_router = 0
else:
new_desc += "%s\n" % line
else:
if len(line) == 0:
continue
m = re.match(r'^router\s+(\S+)\s+\S+', line)
if m:
processing_router = 1
router = m.group(1).lower()
if not entries.has_key(router):
entries[router] = {}
new_desc = "%s\n" % line
continue
m = re.match(r'^(\S+)\s+(\S+)\s*(\s+.*)?$', line)
if m:
entry_type = m.group(1)
router = m.group(2)
if not entries.has_key(router):
entries[router] = {}
entries[router][entry_type] = line
continue
if re.match(r'^\r$', line):
continue
if re.match(r'^end$', line):
continue
log_msg(1, "queue parse error: %s" % line)
return entries
def parse_blossom(lines, peer):
global metadata_pending
global queue
global summary_pending
pending = {}
new_path = []
new_meta = []
processing_router = 0
if len(lines) == 0:
return
for line in lines:
if processing_router:
new_desc += "%s\n" % line
m = re.match(r'opt\s+fingerprint\s+(\S+.*)$', line)
if m:
dir_fingerprint[router] = re.sub(" ", "", m.group(1))
if line == "-----END SIGNATURE-----":
f = 1
if line == "":
processing_router = 0
# publish if we have the entire descriptor
if f == 1:
log_msg(2, "--- VALID: %s" % router)
log_msg(3, "--- dir_summary: %s" % repr(dir_summary))
if peer:
if (neighbors_recv[peer] == "summarize" \
or neighbors_recv[peer] == "proxy") \
and not dir_summary.has_key(router) \
and not peer == router:
update_time[router] = time.time()
else:
if not pending.has_key(router):
pending[router] = ["", "", ""]
pending[router][DESC] = new_desc
else:
update_time[router] = time.time()
desc[router] = new_desc
queue += new_desc
queue += "router-advertisement %s\n" % router
else:
if len(line) == 0:
continue
log_msg(3, "--- line: %s" % line)
# router
m = re.match(r'^router\s+(\S+)\s+\S+', line)
if m:
processing_router = 1
router = m.group(1).lower()
new_desc = "%s\n" % line
continue
# summary
m = re.match(r'^summary\s+(\S+)\s*(\s+(\S+))?$', line)
if m and peer:
dir = m.group(1)
rtr_list = []
if m.group(3):
rtr_list = m.group(3).split(",")
summary_pending[dir] = rtr_list
log_msg(3, "--- received summary from %s: %s" % (peer, line))
continue
# compiled-metadata
m = re.match(r'^compiled-metadata\s+(\S+)\s*(\s+(\S+))?$', line)
if m and peer:
dir = m.group(1)
metadata_list = []
if m.group(3):
metadata_list = m.group(3).split(",")
metadata_pending[dir] = metadata_list
log_msg(3, "--- received compiled-metadata from %s: %s" % (peer, line))
continue
# router-advertisement
m = re.match(r'^router-advertisement\s+(\S+)(\s+(\S+))?$', line)
if m and peer:
router = m.group(1).lower()
if pending.has_key(router):
f = 1
if peer and neighbors_recv[peer] == "proxy":
if not dir_proxy.has_key(peer):
dir_proxy[peer] = {}
if not dir_proxy[peer].has_key(NICK):
dir_proxy[peer][NICK] = {}
dir_proxy[peer][NICK][router] = 1
if received_path.has_key(peer) and received_path[peer].has_key(router):
dir_proxy[peer][NICK][router] += len(received_path[peer][router])
log_msg(3, "--- received router-advertisement from %s: %s" % (peer, line))
new_adv = []
if m.group(3):
new_adv = m.group(3).split(",")
for index in range(len(new_adv)):
if new_adv[index] == NICK:
if index > 1:
new_adv = new_adv[:index-1]
else:
new_adv = []
log_msg(3, "*** BREAKING CYCLE at position %s: %s" % (index, new_adv))
f = 0
break
new_adv.append(peer)
# perform router summaries
if (peer != router) \
and (neighbors_recv[peer] == "summarize"\
or neighbors_recv[peer] == "proxy"):
# generate summary
if not dir_summary.has_key(peer):
dir_summary[peer] = {}
if not dir_summary[peer].has_key(router):
if router != NICK and router != peer:
dir_summary[peer][router] = 1
if received_path.has_key(peer) \
and received_path[peer].has_key(router):
dir_summary[peer][router] += len(received_path[peer][router])
rtr_list = []
for rtr in dir_summary[peer].keys():
rtr_list.append("%s=%s" % (rtr, dir_summary[peer][rtr]))
rtr_list.sort()
queue += "summary %s %s\n" \
% (peer, ",".join(rtr_list))
queue += "directory %s %s %s\n" \
% (peer, dir_port[peer], ",".join(dir_prop[peer][peer]))
if not dir_summary.has_key(router):
f = 0
# generate compiled-metadata
mf = 0
if not dir_metadata.has_key(peer):
dir_metadata[peer] = {}
mf = 1
if metadata.has_key(router):
for md in metadata[router]:
if not dir_metadata[peer].has_key(md):
dir_metadata[peer][md] = 1
mf = 1
if not mf:
all_metadata = {}
for md_source in metadata.keys():
for md in metadata[md_source]:
all_metadata[md] = 1
for md in dir_metadata[peer]:
if not all_metadata.has_key(md):
mf = 1
dir_metadata[peer] = all_metadata
if mf:
queue += "compiled-metadata %s %s\n" \
% (peer, ",".join(dir_metadata[peer]))
if f:
new_line = "router-advertisement %s %s" % (router, ",".join(new_adv))
override = 0
if router_adv.has_key(router):
log_msg(3, "--- router_adv [for override]: %s" \
% repr(router_adv[router]))
if len(router_adv[router]) and router_adv[router][-1] == peer:
override = 2
if update_path(router, pending[router][PATH], override):
router_adv[router] = new_adv
update_time[router] = time.time()
desc[router] = pending[router][DESC]
queue += desc[router]
metadata[router] = pending[router][META]
if metadata[router]:
log_msg(3, "--- metadata[%s]: %s" \
% (router, ",".join(metadata[router])))
new_metadata = "metadata %s %s\n" \
% (router, ",".join(metadata[router]))
queue += new_metadata
queue += "%s\n" % new_line
log_msg(3, "--- desc.keys(): %s" % desc.keys())
continue
# blossom-path
m = re.match(r'^blossom-path\s+(\S+)\s*(\s+(\S+))?$', line)
if m:
router = m.group(1).lower()
new_path = []
if m.group(3):
# enforce well-formedness of blossom-path
bpath = m.group(3)
if re.match(r'^([0-9A-Za-z]+\,)*[0-9A-Za-z]+$', bpath):
log_msg(3, "setting path: %s %s" % (router, bpath))
new_path = bpath.split(",")
else:
log_msg(2, "invalid path specification: %s %s" % (router, bpath))
if peer:
if not pending.has_key(router):
pending[router] = ["", "", ""]
if neighbors_recv[peer] == "prepend":
if new_path:
pending[router][PATH] = [NICK]
pending[router][PATH].extend(new_path)
else:
pending[router][PATH] = [NICK]
else:
pending[router][PATH] = new_path
if not received_path.has_key(peer):
received_path[peer] = {}
received_path[peer][router] = new_path
log_msg(3, "*** setting received_path[%s][%s] = %s" \
% (peer, router, received_path[peer][router]))
else:
update_path(router, new_path, 1)
continue
# metadata
m = re.match(r'^metadata\s+(\S+)\s+(\S+)$', line)
if m:
router = m.group(1).lower()
bmeta = m.group(2).lower()
if re.match(r'^[0-9a-z,-.]+$', bmeta):
log_msg(3, "setting metadata: %s %s" % (router, bmeta))
if peer:
pending[router][META] = bmeta.split(",")
else:
metadata[router] = bmeta.split(",")
new_metadata = "metadata %s %s\n" \
% (router, ",".join(metadata[router]))
queue += new_metadata
else:
log_msg(2, "invalid metadata specification: %s %s" % (router, bmeta))
continue
# directory
m = re.match(r'^directory\s+(\S+)\s+(\S+)\s*(\s+(\S+))?$', line)
if m and peer:
f = 1
dir = m.group(1)
port = m.group(2)
if m.group(4):
prop = m.group(4).split(",")
else:
prop = []
if dir == NICK:
prop = []
f = 0
if selection.has_key(dir) and dir_prop[selection[dir]].has_key(dir):
if len(dir_prop[selection[dir]][dir]) < len(prop) + 1:
f = 0
if not dir_prop.has_key(peer):
dir_prop[peer] = {}
prop.append(peer)
if prop.__contains__(NICK):
f = 0
if f:
dir_prop[peer][dir] = prop
log_msg(3, "--- dir_prop %s" % repr(dir_prop))
selection[dir] = peer
dir_port[dir] = port
if summary_pending.has_key(dir):
log_msg(3, "--- summary_pending[%s]: %s" \
% (dir, summary_pending[dir]))
if neighbors_recv[peer] == "proxy":
log_msg(3, "--- peer: %s" % peer)
log_msg(3, "--- dir_proxy: %s" % repr(dir_proxy))
log_msg(3, "--- summary_pending: %s" % repr(summary_pending))
dir_summary[peer] = {}
if not dir_proxy.has_key(peer):
dir_proxy[peer] = {}
dir_proxy[peer][dir] = {}
for rtr_entry in summary_pending[dir]:
if re.search(r'=', rtr_entry):
rtr_entry = rtr_entry.split("=")
rtr = rtr_entry[0].lower()
rtr_dist = int(rtr_entry[1])
else:
rtr = rtr_entry
rtr_dist = 1
log_msg(3, "--- dir_prop[%s][%s]: %s" \
% (peer, dir, dir_prop[peer][dir]))
dir_proxy[peer][dir][rtr] = rtr_dist + len(dir_prop[peer][dir])
if new_path and dir_proxy[peer][dir].has_key(rtr):
dir_proxy[peer][dir][rtr] += len(new_path)
log_msg(3, "--- dir_proxy[%s][%s][%s]: %s" \
% (peer, dir, rtr, dir_proxy[peer][dir][rtr]))
for s_dir in dir_proxy[peer]:
log_msg(3, "--- dir_proxy[%s][%s]: %s" \
% (peer, s_dir, repr(dir_proxy[peer][s_dir])))
for rtr in dir_proxy[peer][s_dir]:
if rtr != NICK \
and ((not dir_summary[peer].has_key(rtr)) \
or (dir_summary[peer].has_key(rtr) \
and dir_summary[peer][rtr] > dir_proxy[peer][s_dir][rtr])):
dir_summary[peer][rtr] = dir_proxy[peer][s_dir][rtr]
log_msg(3, "--- setting directory dir_summary[%s][%s]: %s" \
% (peer, rtr, repr(dir_summary[peer])))
queue += "summary %s %s\n" \
% (peer, ",".join(dir_summary[peer].keys()))
else:
dir_summary[dir] = {}
for rtr_entry in summary_pending[dir]:
if re.search(r'=', rtr_entry):
rtr_entry = rtr_entry.split("=")
rtr = rtr_entry[0].lower()
rtr_dist = int(rtr_entry[1])
else:
rtr = rtr_entry
rtr_dist = 1
dir_summary[dir][rtr] = rtr_dist
if new_path:
dir_summary[dir][rtr] += len(new_path)
queue += "summary %s %s\n" \
% (dir, ",".join(dir_summary[dir].keys()))
if metadata_pending.has_key(dir):
log_msg(3, "--- metadata_pending[%s]: %s" \
% (dir, metadata_pending[dir]))
if neighbors_recv[peer] == "proxy":
dir_metadata[peer] = {}
for metadata_entry in metadata_pending[dir]:
dir_metadata[peer][metadata_entry] = 1
queue += "compiled-metadata %s %s\n" \
% (peer, ",".join(dir_metadata[peer].keys()))
else:
dir_metadata[dir] = {}
for metadata_entry in metadata_pending[dir]:
dir_metadata[dir][metadata_entry] = 1
queue += "compiled-metadata %s %s\n" \
% (dir, ",".join(dir_metadata[dir].keys()))
queue += "directory %s %s %s\n" \
% (dir, dir_port[dir], ",".join(dir_prop[peer][dir]))
summary_pending = {}
metadata_pending = {}
continue
# withdraw
m = re.match(r'^withdraw\s+(\S+)\s*(\s+(\S+))?$', line)
if m and peer:
dir = m.group(1)
if selection.has_key(dir):
if dir_prop.has_key(peer) and dir_prop[peer].has_key(dir):
del dir_prop[peer][dir]
if m.group(3):
dir_prop[peer][dir] = m.group(3).split(",")
if selection[dir] == peer:
alt = ""
select_optimal(dir)
if selection.has_key(dir):
alt = ",".join(dir_prop[selection[dir]][dir])
queue += "withdraw %s %s\n" % (dir, alt)
continue
log_msg(1, "BLOSSOM: parse error: %s" % line)
log_msg(3, "*** parse_blossom done")
return
def parse_update(lines):
global queue
log_msg(3, "*** parse_update: %s" % lines[0])
m = re.match(r'^directory-update\s+(\S+)\s+(\S+)$', lines[0])
if m:
peer = m.group(1)
dir_port[peer] = m.group(2)
log_msg(3, "*** %s" % lines[0])
full[peer] = time.time()
selection[peer] = peer
if not dir_prop.has_key(peer):
dir_prop[peer] = {}
dir_prop[peer][peer] = []
queue += "directory %s %s\n" % (peer, dir_port[peer])
if not neighbors_recv.has_key(peer) or neighbors_recv[peer] == "none":
log_msg(3, "--- peer: %s" % peer)
log_msg(3, "--- neighbors_recv: %s" % repr(neighbors_recv))
log_msg(2, "<-- %s REJECTED" % lines[0])
return
peer = m.group(1)
log_msg(3, "<-- " + lines[0])
pending = {}
parse_blossom(lines[1:], peer)
log_msg(3, "<-- %s COMPLETED" % lines[0])
def parse_configuration(conflines):
global DIR_SERVER
for index in range(len(conflines)):
m = re.match(r'^directory\s+(\S+)\s*$', conflines[index])
if m:
DIR_SERVER = m.group(1)
m = re.match(r'^neighbor\s+(\S+)\s+(\S+)\s+(\S+)\s*$', conflines[index])
if m:
opt_name = m.group(1)
opt_recv = m.group(2)
opt_send = m.group(3)
neighbors_recv[opt_name] = opt_recv
if opt_send != "-":
neighbors_send[opt_name] = parseHostAndPort(opt_send)
log_msg(2, conflines[index])
def select_optimal(dir):
log_msg(3, "DIR select_optimal: %s" % repr(dir_prop))
min_len = MAXINT
new_selection = ""
for peer in dir_prop.keys():
if dir_prop[peer].has_key(dir):
opt_path = dir_prop[peer][dir]
if len(opt_path) < min_len:
min_len = len(opt_path)
new_selection = peer
if new_selection:
selection[dir] = new_selection
else:
if selection.has_key(dir):
del selection[dir]
log_msg(2, "DIR selection for %s: %s" % (dir, new_selection))
return new_selection
def reap_disconnected_neighbors(curr_time):
global queue
for dir in full.keys():
if curr_time - full[dir] > DIR_PEER_KEEPALIVE:
alt = ""
del full[dir]
if dir_prop.has_key(dir) and dir_prop[dir].has_key(dir):
del dir_prop[dir][dir]
select_optimal(dir)
if selection.has_key(dir):
alt = ",".join(dir_prop[selection[dir]][dir])
queue += "withdraw %s %s\n" % (dir, alt)
def handle_callbacks_individually(httpd, processing_dir=0):
global interesting_ports
global queue
queue = ""
DIR_REG_LOCK = 0
if processing_dir:
last = 0
else:
initialize()
last = [0, 0, 0, 0, 0]
log_msg(2, "*** entering main loop")
while 1:
try:
(r, w, o) = select.select([httpd], [], [], TIME_RETRY)
if r:
request, client_address = httpd.get_request()
if httpd.verify_request(request, client_address):
try:
httpd.process_request(request, client_address)
except SystemExit:
sys.exit(0)
except KeyboardInterrupt:
log_msg(1, "exiting on ^C [%s]\n" % get_curr_time())
except:
log_msg(1, "secondary unexpected: %s" % sys.exc_info()[0])
httpd.handle_error(request, client_address)
httpd.close_request(request)
if processing_dir and time.time() - last > DIR_POLL_INTERVAL and not DIR_REG_LOCK:
last = time.time()
thread = PeriodicDirectoryThread(queue)
thread.setDaemon(1)
thread.start()
queue = ""
if not processing_dir and time.time() - last[T_PERIODIC] > TIME_RETRY:
last[T_PERIODIC] = time.time()
thread = PeriodicClientThread(last, interesting_ports)
interesting_ports = []
thread.setDaemon(1)
thread.start()
except select.error, (ec, em):
log_msg(1, "select.error %s %s" % (ec, em))
except socket.error:
log_msg(1, "socket.error (aborting): %s" % sys.exc_info()[0])
return
def send_updates(queue):
log_msg(3, "*** send_updates: queue length %s" % len(queue.split("\n")))
entries = parse_queue(queue)
msg = "directory-update %s %s\n\n" % (NICK, DIR_PORT)
for node in entries.keys():
tokens = entries[node]
if tokens.has_key("withdraw"):
log_msg(3, "%s" % tokens["withdraw"])
msg += "%s\n" % tokens["withdraw"]
del tokens["withdraw"]
if tokens.has_key("summary"):
log_msg(3, "%s" % tokens["summary"])
msg += "%s\n" % tokens["summary"]
del tokens["summary"]
if tokens.has_key("compiled-metadata"):
log_msg(2, "%s" % tokens["compiled-metadata"])
msg += "%s\n" % tokens["compiled-metadata"]
del tokens["compiled-metadata"]
if tokens.has_key("directory"):
if len(tokens.keys()) > 1:
log_msg(2, "%s" % tokens["directory"])
msg += "%s\n" % tokens["directory"]
else:
log_msg(3, "*** superfluous: %s" % tokens["directory"])
del tokens["directory"]
if tokens.has_key("router") \
and tokens.has_key("router-advertisement") \
and tokens.has_key("blossom-path"):
log_msg(3, "router %s" % node)
msg += "\n%s\n" % tokens["router"]
del tokens["router"]
msg += "%s\n" % tokens["blossom-path"]
del tokens["blossom-path"]
if tokens.has_key("metadata"):
log_msg(3, "%s" % tokens["metadata"])
msg += "%s\n" % tokens["metadata"]
del tokens["metadata"]
for field in tokens.keys():
if field != "router" \
and field != "router-advertisement" \
and field != "metadata" \
and field != "blossom-path":
msg += "%s\n" % tokens[field]
log_msg(2, "-*- %s\n" % tokens[field])
del tokens[field]
msg += "%s\n" % tokens["router-advertisement"]
for peer in neighbors_send.keys():
(dh, dp) = neighbors_send[peer]
log_msg(3, "--> update %s:%s" % (dh, dp))
thread = HTTPPostThread("%s:%s" % (dh, dp), "/blossom/directory-update", msg)
thread.setDaemon(1)
thread.start()
queue = ""
def initialize():
global conn
global unestablished
curr_time = int(time.time())
interesting_ports = []
unestablished = {}
for elt in BLOSSOM:
unestablished[elt] = 1
try:
conn.close()
except:
pass
while 1:
try:
conn = getConnection()
log_msg(2, "*** OPENING CONTROLLER CONNECTION: initialize")
break
except socket.error, e:
err_code, err_msg = e
log_msg(1, "getConnection socket.error %s %s" % (repr(err_code), repr(err_msg)))
time.sleep(TIME_RETRY)
except IOError, (ec, em):
log_msg(1, "getConnection IOError: %s" % em)
raise BlossomError
for ent in get_info("circuit-status"):
text = "CIRC %s %s\n" % (curr_time, ent)
process_line(text)
for ent in get_info("stream-status"):
text = "STREAM %s %s\n" % (curr_time, ent)
process_line(text)
# parse the country codes file
log_msg(3, "*** parsing country codes file")
try:
f = open(F_COUNTRY_CODES)
except IOError, (errno, strerror):
log_msg(0, "cannot read country codes file: %s\n" % strerror)
raise BlossomError
while 1:
try:
line = f.readline()
except:
break
if not line:
break
m = re.match(r"^(\S\S)\s+(\S.*)$", line)
if m and m.group(1) and m.group(2):
cc_name[m.group(1).lower()] = m.group(2)
c = 0
for name in cc_name.keys():
c += 1
process_line("CC %s %s\n" % (name, cc_name[name]))
def main():
# declare configuration variables
global AUTOREFRESH
global BLOSSOM
global BLOSSOM_ARGS
global CONFFILE
global CONNECTION_CLOSED
global DEBUG
global DIR_POLL_INTERVAL
global DIR_PORT
global DIR_REG_LOCK
global DIR_SERVER
global DISCLOSE_TARGET
global ENABLE_DIR
global F_COUNTRY_CODES
global F_ROOT
global HTTP_PROXY
global INIT
global LOGLEVEL
global META_LOCAL
global NICK
global PERSIST
global POLICY_URL
global POLICY_URL_BLOSSOM
global SERVER
global STATUS_URL
global STATUS_URL_BLOSSOM
global TORCONTROL
global WEB_STATUS
# declare other global variables
global addr
global attempted
global circuits
global closed_streams
global conn
global counted_streams
global dir_metadata
global failed_streams
global fingerprint
global interesting_ports
global local
global metadata
global metadata_pending
global neighbors_send
global neighbors_recv
global network
global path
global pending_streams
global persist_id
global persist_nickname
global policy
global policy_time
global port
global prop
global query_streams
global queue
global semaphore
global streams
global summary
global summary_remote
global threads
global tor_nodes
global unestablished
global update_time
# declare local variables
dir_httpd = []
dh = ""
opts = []
pargs = []
usage = 0
try:
opts, pargs = getopt.getopt(sys.argv[1:], "b:c:d:f:gi:m:p:r:s:t:vw:xy")
except getopt.GetoptError, e:
usage = 1
if pargs:
usage = 1
# parse command-line options
for o, a in opts:
if o == "-b":
if ENABLE_DIR:
log_msg(0, "cannot specify both -b and -f options")
sys.exit(0)
else:
BLOSSOM = a.split(",")
if o == "-c":
TORCONTROL = a
if o == "-d":
DEBUG = int(a)
if o == "-f":
if BLOSSOM:
log_msg(0, "cannot specify both -b and -f options")
sys.exit(0)
else:
CONFFILE = a
ENABLE_DIR = 1
BLOSSOM = [DIR_SERVER]
if o == "-g":
DISCLOSE_TARGET = 0
if o == "-i":
DIR_POLL_INTERVAL = int(a)
if o == "-m":
META_LOCAL.extend(a.split(","))
if o == "-p":
if a == "-":
HTTP_PROXY = ""
else:
HTTP_PROXY = a
if o == "-r":
F_ROOT = a
if o == "-s":
SERVER = a
if o == "-v":
print "Blossom/%s" % __version__
sys.exit(0)
if o == "-w":
WEB_STATUS = a
if o == "-x":
PERSIST = 1
if o == "-y":
BLOSSOM_ARGS = ""
if ENABLE_DIR and not HTTP_PROXY:
log_msg(0, "directory servers require an HTTP Proxy (-p - disallowed)")
sys.exit(0)
if usage:
msg = "usage: %s [OPTIONS]\n" % sys.argv[0]
msg += " -b list comma-delimited list of Blossom servers \n"
msg += " -c host:port connect as Tor controller via \n"
msg += " -d num debug to stdout at level \n"
msg += " -f file run Blossom with local directory configured by \n"
msg += " -g force generic directory queries (slower)\n"
msg += " -i num directory poll interval\n"
msg += " -m list comma-delimited list of metadata strings\n"
msg += " -p host:port use as an HTTP proxy\n"
msg += " -r dir use as the root directory for data files\n"
msg += " -s host:port run client web server on host:port\n"
msg += " -t minutes time between periodic Blossom operations\n"
msg += " -v display version information\n"
msg += " -w host:port specify alternate web status page\n"
msg += " -x establish persistent connections\n"
msg += " -y suppress requests for external metadata\n"
print msg
sys.exit(0)
sh, sp = parseHostAndPort(TORCONTROL)
BASE_URL = "http://%s/cgi-bin/exit.pl?" % WEB_STATUS
POLICY_URL = BASE_URL + POLICY_ARGS
STATUS_URL = BASE_URL + STATUS_ARGS
POLICY_URL_BLOSSOM = BASE_URL + BLOSSOM_ARGS + POLICY_ARGS
STATUS_URL_BLOSSOM = BASE_URL + BLOSSOM_ARGS + STATUS_ARGS
# start the client-side web server
dh, dp = parseHostAndPort(SERVER)
ClientRequestHandler.protocol_version = "HTTP/1.0"
httpd = ThreadingHTTPServer((dh, dp), ClientRequestHandler)
sa = httpd.socket.getsockname()
log_msg(2, "serving HTTP on %s:%s." % (sa[0], sa[1]))
# parse configuration file
if ENABLE_DIR:
try:
conflines = open(CONFFILE, "r").read().split("\n")
log_msg(2, "*** reading configuration file: %s" % CONFFILE)
parse_configuration(conflines)
dh, DIR_PORT = parseHostAndPort(DIR_SERVER)
except IOError, e:
log_msg(1, e)
try:
# retrieve nickname directly
sh, sp = parseHostAndPort(TORCONTROL)
conn = getConnection()
log_msg(3, "*** OPENING CONTROLLER CONNECTION")
INIT = 0
if ENABLE_DIR:
thread = DirectoryServiceThread()
thread.setDaemon(1)
thread.start()
# CONTROL LOOP
while 1:
try:
handle_callbacks_individually(httpd)
except BlossomError:
log_msg(0, "FATAL ERROR")
sys.exit(0)
except TorCtlClosed:
log_msg(1, "*** retrying controller connection [%s]" % get_curr_time())
CONNECTION_CLOSED = 0
time.sleep(TIME_RETRY)
except KeyboardInterrupt:
log_msg(1, "exiting on ^C [%s]\n" % get_curr_time())
return
if __name__ == '__main__':
if os.name == "posix":
signal.signal(signal.SIGHUP, signal_handler)
signal.signal(signal.SIGTERM, signal_handler)
signal.signal(signal.SIGUSR1, signal_handler)
CONFFILE = os.environ['HOME'] + "/.blossomrc"
F_ROOT = os.environ['HOME'] + "/.blossom"
else:
CONFFILE = "./blossomrc"
F_ROOT = "."
F_COUNTRY_CODES = "%s/country-codes.txt" % F_ROOT
main()