open.source/magic-wormhole
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

test_twisted: rewrite with inlineCallbacks

Browse Source 
Also, we now disable confirmation to exercise a verifier mismatch. An
upcoming implementation change can probably detect the mismatch too
early otherwise, and throw WrongPasswordError before the test can
compare the verifiers.
This commit is contained in:
Brian Warner 2016-04-20 00:41:53 -07:00
parent 28b3f685d6
commit 42106fd4f0
1 changed files with 140 additions and 191 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

331
tests/test_twisted.py
View File

@ -1,7 +1,7 @@
from __future__ import print_function from __future__ import print_function
import json import json
from twisted.trial import unittest from twisted.trial import unittest
from twisted.internet.defer import gatherResults, succeed from twisted.internet.defer import gatherResults, succeed, inlineCallbacks
from txwormhole.transcribe import (Wormhole, UsageError, ChannelManager, from txwormhole.transcribe import (Wormhole, UsageError, ChannelManager,
WrongPasswordError) WrongPasswordError)
from txwormhole.eventsource import EventSourceParser from txwormhole.eventsource import EventSourceParser
@ -136,138 +136,107 @@ class Basic(ServerBase, unittest.TestCase):
def doBoth(self, d1, d2): def doBoth(self, d1, d2):
return gatherResults([d1, d2], True) return gatherResults([d1, d2], True)
@inlineCallbacks
def test_basic(self): def test_basic(self):
w1 = Wormhole(APPID, self.relayurl) w1 = Wormhole(APPID, self.relayurl)
w2 = Wormhole(APPID, self.relayurl) w2 = Wormhole(APPID, self.relayurl)
d = w1.get_code() code = yield w1.get_code()
def _got_code(code): w2.set_code(code)
w2.set_code(code) yield self.doBoth(w1.send_data(b"data1"), w2.send_data(b"data2"))
return self.doBoth(w1.send_data(b"data1"), w2.send_data(b"data2")) dl = yield self.doBoth(w1.get_data(), w2.get_data())
d.addCallback(_got_code) (dataX, dataY) = dl
def _sent(res): self.assertEqual(dataX, b"data2")
return self.doBoth(w1.get_data(), w2.get_data()) self.assertEqual(dataY, b"data1")
d.addCallback(_sent) yield self.doBoth(w1.close(), w2.close())
def _done(dl):
(dataX, dataY) = dl
self.assertEqual(dataX, b"data2")
self.assertEqual(dataY, b"data1")
return self.doBoth(w1.close(), w2.close())
d.addCallback(_done)
return d
@inlineCallbacks
def test_same_message(self): def test_same_message(self):
# the two sides use random nonces for their messages, so it's ok for # the two sides use random nonces for their messages, so it's ok for
# both to try and send the same body: they'll result in distinct # both to try and send the same body: they'll result in distinct
# encrypted messages # encrypted messages
w1 = Wormhole(APPID, self.relayurl) w1 = Wormhole(APPID, self.relayurl)
w2 = Wormhole(APPID, self.relayurl) w2 = Wormhole(APPID, self.relayurl)
d = w1.get_code() code = yield w1.get_code()
def _got_code(code): w2.set_code(code)
w2.set_code(code) yield self.doBoth(w1.send_data(b"data"), w2.send_data(b"data"))
return self.doBoth(w1.send_data(b"data"), w2.send_data(b"data")) dl = yield self.doBoth(w1.get_data(), w2.get_data())
d.addCallback(_got_code) (dataX, dataY) = dl
def _sent(res): self.assertEqual(dataX, b"data")
return self.doBoth(w1.get_data(), w2.get_data()) self.assertEqual(dataY, b"data")
d.addCallback(_sent) yield self.doBoth(w1.close(), w2.close())
def _done(dl):
(dataX, dataY) = dl
self.assertEqual(dataX, b"data")
self.assertEqual(dataY, b"data")
return self.doBoth(w1.close(), w2.close())
d.addCallback(_done)
return d
@inlineCallbacks
def test_interleaved(self): def test_interleaved(self):
w1 = Wormhole(APPID, self.relayurl) w1 = Wormhole(APPID, self.relayurl)
w2 = Wormhole(APPID, self.relayurl) w2 = Wormhole(APPID, self.relayurl)
d = w1.get_code() code = yield w1.get_code()
def _got_code(code): w2.set_code(code)
w2.set_code(code) res = yield self.doBoth(w1.send_data(b"data1"), w2.get_data())
return self.doBoth(w1.send_data(b"data1"), w2.get_data()) (_, dataY) = res
d.addCallback(_got_code) self.assertEqual(dataY, b"data1")
def _sent(res): dl = yield self.doBoth(w1.get_data(), w2.send_data(b"data2"))
(_, dataY) = res (dataX, _) = dl
self.assertEqual(dataY, b"data1") self.assertEqual(dataX, b"data2")
return self.doBoth(w1.get_data(), w2.send_data(b"data2")) yield self.doBoth(w1.close(), w2.close())
d.addCallback(_sent)
def _done(dl):
(dataX, _) = dl
self.assertEqual(dataX, b"data2")
return self.doBoth(w1.close(), w2.close())
d.addCallback(_done)
return d
@inlineCallbacks
def test_fixed_code(self): def test_fixed_code(self):
w1 = Wormhole(APPID, self.relayurl) w1 = Wormhole(APPID, self.relayurl)
w2 = Wormhole(APPID, self.relayurl) w2 = Wormhole(APPID, self.relayurl)
w1.set_code(u"123-purple-elephant") w1.set_code(u"123-purple-elephant")
w2.set_code(u"123-purple-elephant") w2.set_code(u"123-purple-elephant")
d = self.doBoth(w1.send_data(b"data1"), w2.send_data(b"data2")) yield self.doBoth(w1.send_data(b"data1"), w2.send_data(b"data2"))
def _sent(res): dl = yield self.doBoth(w1.get_data(), w2.get_data())
return self.doBoth(w1.get_data(), w2.get_data()) (dataX, dataY) = dl
d.addCallback(_sent) self.assertEqual(dataX, b"data2")
def _done(dl): self.assertEqual(dataY, b"data1")
(dataX, dataY) = dl yield self.doBoth(w1.close(), w2.close())
self.assertEqual(dataX, b"data2")
self.assertEqual(dataY, b"data1")
return self.doBoth(w1.close(), w2.close())
d.addCallback(_done)
return d
@inlineCallbacks
def test_phases(self): def test_phases(self):
w1 = Wormhole(APPID, self.relayurl) w1 = Wormhole(APPID, self.relayurl)
w2 = Wormhole(APPID, self.relayurl) w2 = Wormhole(APPID, self.relayurl)
w1.set_code(u"123-purple-elephant") w1.set_code(u"123-purple-elephant")
w2.set_code(u"123-purple-elephant") w2.set_code(u"123-purple-elephant")
d = self.doBoth(w1.send_data(b"data1", u"p1"), yield self.doBoth(w1.send_data(b"data1", u"p1"),
w2.send_data(b"data2", u"p1")) w2.send_data(b"data2", u"p1"))
d.addCallback(lambda _: yield self.doBoth(w1.send_data(b"data3", u"p2"),
self.doBoth(w1.send_data(b"data3", u"p2"), w2.send_data(b"data4", u"p2"))
w2.send_data(b"data4", u"p2"))) dl = yield self.doBoth(w1.get_data(u"p2"),
d.addCallback(lambda _: w2.get_data(u"p1"))
self.doBoth(w1.get_data(u"p2"), (dataX, dataY) = dl
w2.get_data(u"p1"))) self.assertEqual(dataX, b"data4")
def _got_1(dl): self.assertEqual(dataY, b"data1")
(dataX, dataY) = dl dl = yield self.doBoth(w1.get_data(u"p1"),
self.assertEqual(dataX, b"data4")
self.assertEqual(dataY, b"data1")
return self.doBoth(w1.get_data(u"p1"),
w2.get_data(u"p2")) w2.get_data(u"p2"))
d.addCallback(_got_1) (dataX, dataY) = dl
def _got_2(dl): self.assertEqual(dataX, b"data2")
(dataX, dataY) = dl self.assertEqual(dataY, b"data3")
self.assertEqual(dataX, b"data2") yield self.doBoth(w1.close(), w2.close())
self.assertEqual(dataY, b"data3")
return self.doBoth(w1.close(), w2.close())
d.addCallback(_got_2)
return d
@inlineCallbacks
def test_wrong_password(self): def test_wrong_password(self):
w1 = Wormhole(APPID, self.relayurl) w1 = Wormhole(APPID, self.relayurl)
w2 = Wormhole(APPID, self.relayurl) w2 = Wormhole(APPID, self.relayurl)
d = w1.get_code() code = yield w1.get_code()
d.addCallback(lambda code: w2.set_code(code+"not")) w2.set_code(code+"not")
# w2 can't throw WrongPasswordError until it sees a CONFIRM message, # w2 can't throw WrongPasswordError until it sees a CONFIRM message,
# and w1 won't send CONFIRM until it sees a PAKE message, which w2 # and w1 won't send CONFIRM until it sees a PAKE message, which w2
# won't send until we call get_data. So we need both sides to be # won't send until we call get_data. So we need both sides to be
# running at the same time for this test. # running at the same time for this test.
def _w1_sends(): yield self.doBoth(w1.send_data(b"data1"),
return w1.send_data(b"data1") self.assertFailure(w2.get_data(), WrongPasswordError))
def _w2_gets():
return self.assertFailure(w2.get_data(), WrongPasswordError)
d.addCallback(lambda _: self.doBoth(_w1_sends(), _w2_gets()))
# and now w1 should have enough information to throw too # and now w1 should have enough information to throw too
d.addCallback(lambda _: self.assertFailure(w1.get_data(), yield self.assertFailure(w1.get_data(), WrongPasswordError)
WrongPasswordError))
def _done(_):
# both sides are closed automatically upon error, but it's still
# legal to call .close(), and should be idempotent
return self.doBoth(w1.close(), w2.close())
d.addCallback(_done)
return d
# both sides are closed automatically upon error, but it's still
# legal to call .close(), and should be idempotent
yield self.doBoth(w1.close(), w2.close())
@inlineCallbacks
def test_no_confirm(self): def test_no_confirm(self):
# newer versions (which check confirmations) should will work with # newer versions (which check confirmations) should will work with
# older versions (that don't send confirmations) # older versions (that don't send confirmations)
@ -275,131 +244,111 @@ class Basic(ServerBase, unittest.TestCase):
w1._send_confirm = False w1._send_confirm = False
w2 = Wormhole(APPID, self.relayurl) w2 = Wormhole(APPID, self.relayurl)
d = w1.get_code() code = yield w1.get_code()
d.addCallback(lambda code: w2.set_code(code)) w2.set_code(code)
d.addCallback(lambda _: self.doBoth(w1.send_data(b"data1"), dl = yield self.doBoth(w1.send_data(b"data1"), w2.get_data())
w2.get_data())) self.assertEqual(dl[1], b"data1")
d.addCallback(lambda dl: self.assertEqual(dl[1], b"data1")) dl = yield self.doBoth(w1.get_data(), w2.send_data(b"data2"))
d.addCallback(lambda _: self.doBoth(w1.get_data(), self.assertEqual(dl[0], b"data2")
w2.send_data(b"data2"))) yield self.doBoth(w1.close(), w2.close())
d.addCallback(lambda dl: self.assertEqual(dl[0], b"data2"))
d.addCallback(lambda _: self.doBoth(w1.close(), w2.close()))
return d
@inlineCallbacks
def test_verifier(self): def test_verifier(self):
w1 = Wormhole(APPID, self.relayurl) w1 = Wormhole(APPID, self.relayurl)
w2 = Wormhole(APPID, self.relayurl) w2 = Wormhole(APPID, self.relayurl)
d = w1.get_code() code = yield w1.get_code()
def _got_code(code): w2.set_code(code)
w2.set_code(code) res = yield self.doBoth(w1.get_verifier(), w2.get_verifier())
return self.doBoth(w1.get_verifier(), w2.get_verifier()) v1, v2 = res
d.addCallback(_got_code) self.failUnlessEqual(type(v1), type(b""))
def _check_verifier(res): self.failUnlessEqual(v1, v2)
v1, v2 = res yield self.doBoth(w1.send_data(b"data1"), w2.send_data(b"data2"))
self.failUnlessEqual(type(v1), type(b"")) dl = yield self.doBoth(w1.get_data(), w2.get_data())
self.failUnlessEqual(v1, v2) (dataX, dataY) = dl
return self.doBoth(w1.send_data(b"data1"), w2.send_data(b"data2")) self.assertEqual(dataX, b"data2")
d.addCallback(_check_verifier) self.assertEqual(dataY, b"data1")
def _sent(res): yield self.doBoth(w1.close(), w2.close())
return self.doBoth(w1.get_data(), w2.get_data())
d.addCallback(_sent)
def _done(dl):
(dataX, dataY) = dl
self.assertEqual(dataX, b"data2")
self.assertEqual(dataY, b"data1")
return self.doBoth(w1.close(), w2.close())
d.addCallback(_done)
return d
@inlineCallbacks
def test_verifier_mismatch(self): def test_verifier_mismatch(self):
w1 = Wormhole(APPID, self.relayurl) w1 = Wormhole(APPID, self.relayurl)
w2 = Wormhole(APPID, self.relayurl) w2 = Wormhole(APPID, self.relayurl)
d = w1.get_code() # we must disable confirmation messages, else the wormholes will
def _got_code(code): # figure out the mismatch by themselves and throw WrongPasswordError.
w2.set_code(code+"not") w1._send_confirm = w2._send_confirm = False
return self.doBoth(w1.get_verifier(), w2.get_verifier()) code = yield w1.get_code()
d.addCallback(_got_code) w2.set_code(code+"not")
def _check_verifier(res): res = yield self.doBoth(w1.get_verifier(), w2.get_verifier())
v1, v2 = res v1, v2 = res
self.failUnlessEqual(type(v1), type(b"")) self.failUnlessEqual(type(v1), type(b""))
self.failIfEqual(v1, v2) self.failIfEqual(v1, v2)
return self.doBoth(w1.close(), w2.close()) yield self.doBoth(w1.close(), w2.close())
d.addCallback(_check_verifier)
return d
@inlineCallbacks
def test_errors(self): def test_errors(self):
w1 = Wormhole(APPID, self.relayurl) w1 = Wormhole(APPID, self.relayurl)
d = self.assertFailure(w1.get_verifier(), UsageError) yield self.assertFailure(w1.get_verifier(), UsageError)
d.addCallback(lambda _: self.assertFailure(w1.send_data(b"data"), UsageError)) yield self.assertFailure(w1.send_data(b"data"), UsageError)
d.addCallback(lambda _: self.assertFailure(w1.get_data(), UsageError)) yield self.assertFailure(w1.get_data(), UsageError)
d.addCallback(lambda _: w1.set_code(u"123-purple-elephant")) w1.set_code(u"123-purple-elephant")
# this UsageError is synchronous, although most of the rest are async yield self.assertRaises(UsageError, w1.set_code, u"123-nope")
d.addCallback(lambda _: self.assertRaises(UsageError, w1.set_code, u"123-nope")) yield self.assertFailure(w1.get_code(), UsageError)
d.addCallback(lambda _: self.assertFailure(w1.get_code(), UsageError)) w2 = Wormhole(APPID, self.relayurl)
def _then(_): yield w2.get_code()
w2 = Wormhole(APPID, self.relayurl) yield self.assertFailure(w2.get_code(), UsageError)
d2 = w2.get_code() yield self.doBoth(w1.close(), w2.close())
d2.addCallback(lambda _: self.assertFailure(w2.get_code(), UsageError))
d2.addCallback(lambda _: self.doBoth(w1.close(), w2.close()))
return d2
d.addCallback(_then)
return d
@inlineCallbacks
def test_repeat_phases(self): def test_repeat_phases(self):
w1 = Wormhole(APPID, self.relayurl) w1 = Wormhole(APPID, self.relayurl)
w1.set_code(u"123-purple-elephant") w1.set_code(u"123-purple-elephant")
w2 = Wormhole(APPID, self.relayurl) w2 = Wormhole(APPID, self.relayurl)
w2.set_code(u"123-purple-elephant") w2.set_code(u"123-purple-elephant")
# we must let them establish a key before we can send data # we must let them establish a key before we can send data
d = self.doBoth(w1.get_verifier(), w2.get_verifier()) yield self.doBoth(w1.get_verifier(), w2.get_verifier())
d.addCallback(lambda _: w1.send_data(b"data1", phase=u"1")) yield w1.send_data(b"data1", phase=u"1")
# underscore-prefixed phases are reserved # underscore-prefixed phases are reserved
d.addCallback(lambda _: self.assertFailure(w1.send_data(b"data1", phase=u"_1"), yield self.assertFailure(w1.send_data(b"data1", phase=u"_1"),
UsageError)) UsageError)
d.addCallback(lambda _: self.assertFailure(w1.get_data(phase=u"_1"), UsageError)) yield self.assertFailure(w1.get_data(phase=u"_1"), UsageError)
# you can't send twice to the same phase # you can't send twice to the same phase
d.addCallback(lambda _: self.assertFailure(w1.send_data(b"data1", phase=u"1"), yield self.assertFailure(w1.send_data(b"data1", phase=u"1"),
UsageError)) UsageError)
# but you can send to a different one # but you can send to a different one
d.addCallback(lambda _: w1.send_data(b"data2", phase=u"2")) yield w1.send_data(b"data2", phase=u"2")
d.addCallback(lambda _: w2.get_data(phase=u"1")) res = yield w2.get_data(phase=u"1")
d.addCallback(lambda res: self.failUnlessEqual(res, b"data1")) self.failUnlessEqual(res, b"data1")
# and you can't read twice from the same phase # and you can't read twice from the same phase
d.addCallback(lambda _: self.assertFailure(w2.get_data(phase=u"1"), UsageError)) yield self.assertFailure(w2.get_data(phase=u"1"), UsageError)
# but you can read from a different one # but you can read from a different one
d.addCallback(lambda _: w2.get_data(phase=u"2")) res = yield w2.get_data(phase=u"2")
d.addCallback(lambda res: self.failUnlessEqual(res, b"data2")) self.failUnlessEqual(res, b"data2")
d.addCallback(lambda _: self.doBoth(w1.close(), w2.close())) yield self.doBoth(w1.close(), w2.close())
return d
@inlineCallbacks
def test_serialize(self): def test_serialize(self):
w1 = Wormhole(APPID, self.relayurl) w1 = Wormhole(APPID, self.relayurl)
self.assertRaises(UsageError, w1.serialize) # too early self.assertRaises(UsageError, w1.serialize) # too early
w2 = Wormhole(APPID, self.relayurl) w2 = Wormhole(APPID, self.relayurl)
d = w1.get_code() code = yield w1.get_code()
def _got_code(code): self.assertRaises(UsageError, w2.serialize) # too early
self.assertRaises(UsageError, w2.serialize) # too early w2.set_code(code)
w2.set_code(code) w2.serialize() # ok
w2.serialize() # ok s = w1.serialize()
s = w1.serialize() self.assertEqual(type(s), type(""))
self.assertEqual(type(s), type("")) unpacked = json.loads(s) # this is supposed to be JSON
unpacked = json.loads(s) # this is supposed to be JSON self.assertEqual(type(unpacked), dict)
self.assertEqual(type(unpacked), dict)
self.new_w1 = Wormhole.from_serialized(s) self.new_w1 = Wormhole.from_serialized(s)
return self.doBoth(self.new_w1.send_data(b"data1"), yield self.doBoth(self.new_w1.send_data(b"data1"),
w2.send_data(b"data2")) w2.send_data(b"data2"))
d.addCallback(_got_code) dl = yield self.doBoth(self.new_w1.get_data(), w2.get_data())
def _sent(res): (dataX, dataY) = dl
return self.doBoth(self.new_w1.get_data(), w2.get_data()) self.assertEqual((dataX, dataY), (b"data2", b"data1"))
d.addCallback(_sent) self.assertRaises(UsageError, w2.serialize) # too late
def _done(dl): yield gatherResults([w1.close(), w2.close(), self.new_w1.close()],
(dataX, dataY) = dl True)
self.assertEqual((dataX, dataY), (b"data2", b"data1"))
self.assertRaises(UsageError, w2.serialize) # too late
return gatherResults([w1.close(), w2.close(), self.new_w1.close()],
True)
d.addCallback(_done)
return d
data1 = b"""\ data1 = b"""\
event: welcome event: welcome