relay.py 16 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462
  1. #!/usr/bin/env python3
  2. import random # For simulation, not cryptography!
  3. import math
  4. import nacl.utils
  5. import nacl.signing
  6. import nacl.public
  7. import network
  8. import dirauth
  9. class RelayNetMsg(network.NetMsg):
  10. """The subclass of NetMsg for messages between relays and either
  11. relays or clients."""
  12. class RelayGetConsensusMsg(RelayNetMsg):
  13. """The subclass of RelayNetMsg for fetching the consensus."""
  14. class RelayConsensusMsg(RelayNetMsg):
  15. """The subclass of RelayNetMsg for returning the consensus."""
  16. def __init__(self, consensus):
  17. self.consensus = consensus
  18. class RelayRandomHopMsg(RelayNetMsg):
  19. """A message used for testing, that hops from relay to relay
  20. randomly until its TTL expires."""
  21. def __init__(self, ttl):
  22. self.ttl = ttl
  23. def __str__(self):
  24. return "RandomHop TTL=%d" % self.ttl
  25. class VanillaCreateCircuitMsg(RelayNetMsg):
  26. """The message for requesting circuit creation in Vanilla Onion
  27. Routing."""
  28. def __init__(self, circid, ntor_request):
  29. self.circid = circid
  30. self.ntor_request = ntor_request
  31. class VanillaCreatedCircuitMsg(RelayNetMsg):
  32. """The message for responding to circuit creation in Vanilla Onion
  33. Routing."""
  34. def __init__(self, circid, ntor_response):
  35. self.circid = circid
  36. self.ntor_response = ntor_response
  37. class CircuitCellMsg(RelayNetMsg):
  38. """Send a message tagged with a circuit id."""
  39. def __init__(self, circuitid, cell):
  40. self.circid = circuitid
  41. self.cell = cell
  42. def __str__(self):
  43. return "C%d:%s" % (self.circid, self.cell)
  44. class RelayFallbackTerminationError(Exception):
  45. """An exception raised when someone tries to terminate a fallback
  46. relay."""
  47. class CircuitHandler:
  48. """A class for managing sending and receiving encrypted cells on a
  49. particular circuit."""
  50. def __init__(self, channel, circid):
  51. self.channel = channel
  52. self.circid = circid
  53. self.send_cell = self.channel_send_cell
  54. self.received_cell = self.channel_received_cell
  55. def channel_send_cell(self, cell):
  56. """Send a cell on this circuit."""
  57. self.channel.send_msg(CircuitCellMsg(self.circid, cell))
  58. def channel_received_cell(self, cell, peeraddr, peer):
  59. """A cell has been received on this circuit. Forward it to the
  60. channel's received_cell callback."""
  61. self.channel.cellhandler.received_cell(self.circid, cell, peeraddr, peer)
  62. class Channel(network.Connection):
  63. """A class representing a channel between a relay and either a
  64. client or a relay, transporting cells from various circuits."""
  65. def __init__(self):
  66. super().__init__()
  67. # The CellRelay managing this Channel
  68. self.cellhandler = None
  69. # The Channel at the other end
  70. self.peer = None
  71. # The function to call when the connection closes
  72. self.closer = lambda: 0
  73. # The next circuit id to use on this channel. The party that
  74. # opened the channel uses even numbers; the receiving party uses
  75. # odd numbers.
  76. self.next_circid = None
  77. # A map for CircuitHandlers to use for each open circuit on the
  78. # channel
  79. self.circuithandlers = dict()
  80. def closed(self):
  81. self.closer()
  82. self.peer = None
  83. def close(self):
  84. if self.peer is not None and self.peer is not self:
  85. self.peer.closed()
  86. self.closed()
  87. def new_circuit(self):
  88. """Allocate a new circuit on this channel, returning the new
  89. circuit's id."""
  90. circid = self.next_circid
  91. self.next_circid += 2
  92. self.circuithandlers[circid] = CircuitHandler(self, circid)
  93. return circid
  94. def new_circuit_with_circid(self, circid):
  95. """Allocate a new circuit on this channel, with the circuit id
  96. received from our peer."""
  97. self.circuithandlers[circid] = CircuitHandler(self, circid)
  98. def send_cell(self, circid, cell):
  99. """Send the given message on the given circuit, encrypting or
  100. decrypting as needed."""
  101. self.circuithandlers[circid].send_cell(cell)
  102. def send_raw_cell(self, circid, cell):
  103. """Send the given message, tagged for the given circuit id. No
  104. encryption or decryption is done."""
  105. self.send_msg(CircuitCellMsg(self.circid, self.cell))
  106. def send_msg(self, msg):
  107. """Send the given NetMsg on the channel."""
  108. self.peer.received(self.cellhandler.myaddr, msg)
  109. def received(self, peeraddr, msg):
  110. """Callback when a message is received from the network."""
  111. if isinstance(msg, CircuitCellMsg):
  112. circid, cell = msg.circid, msg.cell
  113. self.circuithandlers[circid].received_cell(cell, peeraddr, self.peer)
  114. else:
  115. self.cellhandler.received_msg(msg, peeraddr, self.peer)
  116. class CellHandler:
  117. """The class that manages the channels to other relays and clients.
  118. Relays and clients both use subclasses of this class to both create
  119. on-demand channels to relays, to gracefully handle the closing of
  120. channels, and to handle commands received over the channels."""
  121. def __init__(self, myaddr, dirauthaddrs, perfstats):
  122. # A dictionary of Channels to other hosts, indexed by NetAddr
  123. self.channels = dict()
  124. self.myaddr = myaddr
  125. self.dirauthaddrs = dirauthaddrs
  126. self.consensus = None
  127. self.perfstats = perfstats
  128. def terminate(self):
  129. """Close all connections we're managing."""
  130. while self.channels:
  131. channelitems = iter(self.channels.items())
  132. addr, channel = next(channelitems)
  133. print('closing channel', addr, channel)
  134. channel.close()
  135. def add_channel(self, channel, peeraddr):
  136. """Add the given channel to the list of channels we are
  137. managing. If we are already managing a channel to the same
  138. peer, close it first."""
  139. if peeraddr in self.channels:
  140. self.channels[peeraddr].close()
  141. channel.cellhandler = self
  142. self.channels[peeraddr] = channel
  143. channel.closer = lambda: self.channels.pop(peeraddr)
  144. def get_channel_to(self, addr):
  145. """Get the Channel connected to the given NetAddr, creating one
  146. if none exists right now."""
  147. if addr in self.channels:
  148. return self.channels[addr]
  149. # Create the new channel
  150. newchannel = network.thenetwork.connect(self.myaddr, addr)
  151. self.channels[addr] = newchannel
  152. newchannel.closer = lambda: self.channels.pop(addr)
  153. newchannel.cellhandler = self
  154. return newchannel
  155. def received_msg(self, msg, peeraddr, peer):
  156. """Callback when a NetMsg not specific to a circuit is
  157. received."""
  158. print("CellHandler: Node %s received msg %s from %s" % (self.myaddr, msg, peeraddr))
  159. def received_cell(self, circid, cell, peeraddr, peer):
  160. """Callback with a circuit-specific cell is received."""
  161. print("CellHandler: Node %s received cell on circ %d: %s from %s" % (self.myaddr, circid, cell, peeraddr))
  162. def send_msg(self, msg, peeraddr):
  163. """Send a message to the peer with the given address."""
  164. channel = self.get_channel_to(peeraddr)
  165. channel.send_msg(msg)
  166. def send_cell(self, circid, cell, peeraddr):
  167. """Send a cell on the given circuit to the peer with the given
  168. address."""
  169. channel = self.get_channel_to(peeraddr)
  170. channel.send_cell(circid, cell)
  171. class CellRelay(CellHandler):
  172. """The subclass of CellHandler for relays."""
  173. def __init__(self, myaddr, dirauthaddrs, perfstats):
  174. super().__init__(myaddr, dirauthaddrs, perfstats)
  175. def get_consensus(self):
  176. """Download a fresh consensus from a random dirauth."""
  177. a = random.choice(self.dirauthaddrs)
  178. c = network.thenetwork.connect(self, a)
  179. self.consensus = c.getconsensus()
  180. dirauth.Consensus.verify(self.consensus, \
  181. network.thenetwork.dirauthkeys(), self.perfstats)
  182. c.close()
  183. def received_msg(self, msg, peeraddr, peer):
  184. """Callback when a NetMsg not specific to a circuit is
  185. received."""
  186. print("CellRelay: Node %s received msg %s from %s" % (self.myaddr, msg, peeraddr))
  187. if isinstance(msg, RelayRandomHopMsg):
  188. if msg.ttl > 0:
  189. # Pick a random next hop from the consensus
  190. nexthop = random.choice(self.consensus.consdict['relays'])
  191. nextaddr = nexthop.descdict['addr']
  192. self.send_msg(RelayRandomHopMsg(msg.ttl-1), nextaddr)
  193. elif isinstance(msg, RelayGetConsensusMsg):
  194. self.send_msg(RelayConsensusMsg(self.consensus), peeraddr)
  195. else:
  196. return super().received_msg(msg, peeraddr, peer)
  197. def received_cell(self, circid, cell, peeraddr, peer):
  198. """Callback with a circuit-specific cell is received."""
  199. print("CellRelay: Node %s received cell on circ %d: %s from %s" % (self.myaddr, circid, cell, peeraddr))
  200. return super().received_cell(circid, cell, peeraddr, peer)
  201. class Relay(network.Server):
  202. """The class representing an onion relay."""
  203. def __init__(self, dirauthaddrs, bw, flags):
  204. # Gather performance statistics
  205. self.perfstats = dirauth.PerfStats(dirauth.EntType.RELAY)
  206. self.perfstats.is_bootstrapping = True
  207. # Create the identity and onion keys
  208. self.idkey = nacl.signing.SigningKey.generate()
  209. self.onionkey = nacl.public.PrivateKey.generate()
  210. self.perfstats.keygens += 2
  211. self.name = self.idkey.verify_key.encode(encoder=nacl.encoding.HexEncoder).decode("ascii")
  212. # Bind to the network to get a network address
  213. self.netaddr = network.thenetwork.bind(self)
  214. self.perfstats.name = "Relay at %s" % self.netaddr
  215. # Our bandwidth and flags
  216. self.bw = bw
  217. self.flags = flags
  218. # Register for epoch change notification
  219. network.thenetwork.wantepochticks(self, True, end=True)
  220. network.thenetwork.wantepochticks(self, True)
  221. # Create the CellRelay connection manager
  222. self.cellhandler = CellRelay(self.netaddr, dirauthaddrs, self.perfstats)
  223. # Initially, we're not a fallback relay
  224. self.is_fallbackrelay = False
  225. self.uploaddesc()
  226. def terminate(self):
  227. """Stop this relay."""
  228. if self.is_fallbackrelay:
  229. # Fallback relays must not (for now) terminate
  230. raise RelayFallbackTerminationError(self)
  231. # Stop listening for epoch ticks
  232. network.thenetwork.wantepochticks(self, False, end=True)
  233. network.thenetwork.wantepochticks(self, False)
  234. # Tell the dirauths we're going away
  235. self.uploaddesc(False)
  236. # Close connections to other relays
  237. self.cellhandler.terminate()
  238. # Stop listening to our own bound port
  239. self.close()
  240. def set_is_fallbackrelay(self, isfallback = True):
  241. """Set this relay to be a fallback relay (or unset if passed
  242. False)."""
  243. self.is_fallbackrelay = isfallback
  244. def epoch_ending(self, epoch):
  245. # Download the new consensus, which will have been created
  246. # already since the dirauths' epoch_ending callbacks happened
  247. # before the relays'.
  248. self.cellhandler.get_consensus()
  249. def newepoch(self, epoch):
  250. self.uploaddesc()
  251. def uploaddesc(self, upload=True):
  252. # Upload the descriptor for the epoch to come, or delete a
  253. # previous upload if upload=False
  254. descdict = dict();
  255. descdict["epoch"] = network.thenetwork.getepoch() + 1
  256. descdict["idkey"] = self.idkey.verify_key
  257. descdict["onionkey"] = self.onionkey.public_key
  258. descdict["addr"] = self.netaddr
  259. descdict["bw"] = self.bw
  260. descdict["flags"] = self.flags
  261. desc = dirauth.RelayDescriptor(descdict)
  262. desc.sign(self.idkey, self.perfstats)
  263. dirauth.RelayDescriptor.verify(desc, self.perfstats)
  264. if upload:
  265. descmsg = dirauth.DirAuthUploadDescMsg(desc)
  266. else:
  267. # Note that this relies on signatures being deterministic;
  268. # otherwise we'd need to save the descriptor we uploaded
  269. # before so we could tell the airauths to delete the exact
  270. # one
  271. descmsg = dirauth.DirAuthDelDescMsg(desc)
  272. # Upload them
  273. for a in self.cellhandler.dirauthaddrs:
  274. c = network.thenetwork.connect(self, a)
  275. c.sendmsg(descmsg)
  276. c.close()
  277. def connected(self, peer):
  278. """Callback invoked when someone (client or relay) connects to
  279. us. Create a pair of linked Channels and return the peer half
  280. to the peer."""
  281. # Create the linked pair
  282. if peer is self.netaddr:
  283. # A self-loop? We'll allow it.
  284. peerchannel = Channel()
  285. peerchannel.peer = peerchannel
  286. peerchannel.next_circid = 2
  287. return peerchannel
  288. peerchannel = Channel()
  289. ourchannel = Channel()
  290. peerchannel.peer = ourchannel
  291. peerchannel.next_circid = 2
  292. ourchannel.peer = peerchannel
  293. ourchannel.next_circid = 1
  294. # Add our channel to the CellRelay
  295. self.cellhandler.add_channel(ourchannel, peer)
  296. return peerchannel
  297. if __name__ == '__main__':
  298. perfstats = dirauth.PerfStats(dirauth.EntType.NONE)
  299. # Start some dirauths
  300. numdirauths = 9
  301. dirauthaddrs = []
  302. for i in range(numdirauths):
  303. dira = dirauth.DirAuth(i, numdirauths)
  304. dirauthaddrs.append(network.thenetwork.bind(dira))
  305. # Start some relays
  306. numrelays = 10
  307. relays = []
  308. for i in range(numrelays):
  309. # Relay bandwidths (at least the ones fast enough to get used)
  310. # in the live Tor network (as of Dec 2019) are well approximated
  311. # by (200000-(200000-25000)/3*log10(x)) where x is a
  312. # uniform integer in [1,2500]
  313. x = random.randint(1,2500)
  314. bw = int(200000-(200000-25000)/3*math.log10(x))
  315. relays.append(Relay(dirauthaddrs, bw, 0))
  316. # The fallback relays are a hardcoded list of about 5% of the
  317. # relays, used by clients for bootstrapping
  318. numfallbackrelays = int(numrelays * 0.05) + 1
  319. fallbackrelays = random.sample(relays, numfallbackrelays)
  320. for r in fallbackrelays:
  321. r.set_is_fallbackrelay()
  322. network.thenetwork.setfallbackrelays(fallbackrelays)
  323. # Tick the epoch
  324. network.thenetwork.nextepoch()
  325. dirauth.Consensus.verify(dirauth.DirAuth.consensus, \
  326. network.thenetwork.dirauthkeys(), perfstats)
  327. print('ticked; epoch=', network.thenetwork.getepoch())
  328. relays[3].cellhandler.send_msg(RelayRandomHopMsg(30), relays[5].netaddr)
  329. # See what channels exist and do a consistency check
  330. for r in relays:
  331. print("%s: %s" % (r.netaddr, [ str(k) for k in r.cellhandler.channels.keys()]))
  332. raddr = r.netaddr
  333. for ad, ch in r.cellhandler.channels.items():
  334. if ch.peer.cellhandler.myaddr != ad:
  335. print('address mismatch:', raddr, ad, ch.peer.cellhandler.myaddr)
  336. if ch.peer.cellhandler.channels[raddr].peer is not ch:
  337. print('asymmetry:', raddr, ad, ch, ch.peer.cellhandler.channels[raddr].peer)
  338. # Stop some relays
  339. relays[3].terminate()
  340. del relays[3]
  341. relays[5].terminate()
  342. del relays[5]
  343. relays[7].terminate()
  344. del relays[7]
  345. # Tick the epoch
  346. network.thenetwork.nextepoch()
  347. print(dirauth.DirAuth.consensus)
  348. # See what channels exist and do a consistency check
  349. for r in relays:
  350. print("%s: %s" % (r.netaddr, [ str(k) for k in r.cellhandler.channels.keys()]))
  351. raddr = r.netaddr
  352. for ad, ch in r.cellhandler.channels.items():
  353. if ch.peer.cellhandler.myaddr != ad:
  354. print('address mismatch:', raddr, ad, ch.peer.cellhandler.myaddr)
  355. if ch.peer.cellhandler.channels[raddr].peer is not ch:
  356. print('asymmetry:', raddr, ad, ch, ch.peer.cellhandler.channels[raddr].peer)
  357. channel = relays[3].cellhandler.get_channel_to(relays[5].netaddr)
  358. circid = channel.new_circuit()
  359. peerchannel = relays[5].cellhandler.get_channel_to(relays[3].netaddr)
  360. peerchannel.new_circuit_with_circid(circid)
  361. relays[3].cellhandler.send_cell(circid, network.StringNetMsg("test"), relays[5].netaddr)