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

move parse_hint/encode_hint into _hints.py, add tests

Browse Source
This commit is contained in:
Brian Warner 2018-12-21 23:55:58 -05:00
parent d64c94a1dc
commit b4c90b40a2
4 changed files with 84 additions and 36 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

35
src/wormhole/_dilation/connector.py
View File

@ -21,42 +21,9 @@ from .roles import LEADER
from .._hints import (DirectTCPV1Hint, TorTCPV1Hint, RelayV1Hint,
parse_hint_argv, describe_hint_obj, endpoint_from_hint_obj,
parse_tcp_v1_hint)
encode_hint)
def parse_hint(hint_struct):
hint_type = hint_struct.get("type", "")
if hint_type == "relay-v1":
# the struct can include multiple ways to reach the same relay
rhints = filter(lambda h: h, # drop None (unrecognized)
[parse_tcp_v1_hint(rh) for rh in hint_struct["hints"]])
return RelayV1Hint(rhints)
return parse_tcp_v1_hint(hint_struct)
def encode_hint(h):
if isinstance(h, DirectTCPV1Hint):
return {"type": "direct-tcp-v1",
"priority": h.priority,
"hostname": h.hostname,
"port": h.port, # integer
}
elif isinstance(h, RelayV1Hint):
rhint = {"type": "relay-v1", "hints": []}
for rh in h.hints:
rhint["hints"].append({"type": "direct-tcp-v1",
"priority": rh.priority,
"hostname": rh.hostname,
"port": rh.port})
return rhint
elif isinstance(h, TorTCPV1Hint):
return {"type": "tor-tcp-v1",
"priority": h.priority,
"hostname": h.hostname,
"port": h.port, # integer
}
raise ValueError("unknown hint type", h)
def HKDF(skm, outlen, salt=None, CTXinfo=b""):
return Hkdf(salt, skm).expand(CTXinfo, outlen)

3
src/wormhole/_dilation/manager.py
View File

@ -15,7 +15,8 @@ from .encode import to_be4
from .subchannel import (SubChannel, _SubchannelAddress, _WormholeAddress,
ControlEndpoint, SubchannelConnectorEndpoint,
SubchannelListenerEndpoint)
from .connector import Connector, parse_hint
from .connector import Connector
from .._hints import parse_hint
from .roles import LEADER, FOLLOWER
from .connection import KCM, Ping, Pong, Open, Data, Close, Ack
from .inbound import Inbound

33
src/wormhole/_hints.py
View File

@ -103,3 +103,36 @@ def parse_tcp_v1_hint(hint): # hint_struct -> hint_obj
return DirectTCPV1Hint(hint["hostname"], hint["port"], priority)
else:
return TorTCPV1Hint(hint["hostname"], hint["port"], priority)
def parse_hint(hint_struct):
hint_type = hint_struct.get("type", "")
if hint_type == "relay-v1":
# the struct can include multiple ways to reach the same relay
rhints = filter(lambda h: h, # drop None (unrecognized)
[parse_tcp_v1_hint(rh) for rh in hint_struct["hints"]])
return RelayV1Hint(list(rhints))
return parse_tcp_v1_hint(hint_struct)
def encode_hint(h):
if isinstance(h, DirectTCPV1Hint):
return {"type": "direct-tcp-v1",
"priority": h.priority,
"hostname": h.hostname,
"port": h.port, # integer
}
elif isinstance(h, RelayV1Hint):
rhint = {"type": "relay-v1", "hints": []}
for rh in h.hints:
rhint["hints"].append({"type": "direct-tcp-v1",
"priority": rh.priority,
"hostname": rh.hostname,
"port": rh.port})
return rhint
elif isinstance(h, TorTCPV1Hint):
return {"type": "tor-tcp-v1",
"priority": h.priority,
"hostname": h.hostname,
"port": h.port, # integer
}
raise ValueError("unknown hint type", h)

49
src/wormhole/test/test_hints.py
View File

@ -5,7 +5,7 @@ import mock
from twisted.internet import endpoints, reactor
from twisted.trial import unittest
from .._hints import (endpoint_from_hint_obj, parse_hint_argv, parse_tcp_v1_hint,
describe_hint_obj,
describe_hint_obj, parse_hint, encode_hint,
DirectTCPV1Hint, TorTCPV1Hint, RelayV1Hint)
UnknownHint = namedtuple("UnknownHint", ["stuff"])
@ -89,6 +89,24 @@ class Hints(unittest.TestCase):
"port": "not a number"
}), None) # invalid port
def test_parse_hint(self):
p = parse_hint
self.assertEqual(p({"type": "direct-tcp-v1",
"hostname": "foo",
"port": 12}),
DirectTCPV1Hint("foo", 12, 0.0))
self.assertEqual(p({"type": "relay-v1",
"hints": [
{"type": "direct-tcp-v1",
"hostname": "foo",
"port": 12},
{"type": "unrecognized"},
{"type": "direct-tcp-v1",
"hostname": "bar",
"port": 13}]}),
RelayV1Hint([DirectTCPV1Hint("foo", 12, 0.0),
DirectTCPV1Hint("bar", 13, 0.0)]))
def test_parse_hint_argv(self):
def p(hint):
stderr = io.StringIO()
@ -147,3 +165,32 @@ class Hints(unittest.TestCase):
"->tor:host:1234")
self.assertEqual(d(UnknownHint("stuff"), False, False),
"->%s" % str(UnknownHint("stuff")))
def test_encode_hint(self):
e = encode_hint
self.assertEqual(e(DirectTCPV1Hint("host", 1234, 1.0)),
{"type": "direct-tcp-v1",
"priority": 1.0,
"hostname": "host",
"port": 1234})
self.assertEqual(e(RelayV1Hint([DirectTCPV1Hint("foo", 12, 0.0),
DirectTCPV1Hint("bar", 13, 0.0)])),
{"type": "relay-v1",
"hints": [
{"type": "direct-tcp-v1",
"hostname": "foo",
"port": 12,
"priority": 0.0},
{"type": "direct-tcp-v1",
"hostname": "bar",
"port": 13,
"priority": 0.0},
]})
self.assertEqual(e(TorTCPV1Hint("host", 1234, 1.0)),
{"type": "tor-tcp-v1",
"priority": 1.0,
"hostname": "host",
"port": 1234})
e = self.assertRaises(ValueError, e, "not a Hint")
self.assertIn("unknown hint type", str(e))
self.assertIn("not a Hint", str(e))