move utility functions to util.py, add new ones
This commit is contained in:
parent
0d30101d7c
commit
3850c164f7
39
src/wormhole/test/test_util.py
Normal file
39
src/wormhole/test/test_util.py
Normal file
|
@ -0,0 +1,39 @@
|
|||
import unicodedata
|
||||
from twisted.trial import unittest
|
||||
from .. import util
|
||||
|
||||
class Utils(unittest.TestCase):
|
||||
def test_to_bytes(self):
|
||||
b = util.to_bytes(u"abc")
|
||||
self.assertIsInstance(b, type(b""))
|
||||
self.assertEqual(b, b"abc")
|
||||
|
||||
A = unicodedata.lookup("LATIN SMALL LETTER A WITH DIAERESIS")
|
||||
b = util.to_bytes(A + u"bc")
|
||||
self.assertIsInstance(b, type(b""))
|
||||
self.assertEqual(b, b"\xc3\xa4\x62\x63")
|
||||
|
||||
def test_bytes_to_hexstr(self):
|
||||
b = b"\x00\x45\x91\xfe\xff"
|
||||
hexstr = util.bytes_to_hexstr(b)
|
||||
self.assertIsInstance(hexstr, type(u""))
|
||||
self.assertEqual(hexstr, u"004591feff")
|
||||
|
||||
def test_hexstr_to_bytes(self):
|
||||
hexstr = u"004591feff"
|
||||
b = util.hexstr_to_bytes(hexstr)
|
||||
hexstr = util.bytes_to_hexstr(b)
|
||||
self.assertIsInstance(b, type(b""))
|
||||
self.assertEqual(b, b"\x00\x45\x91\xfe\xff")
|
||||
|
||||
def test_dict_to_bytes(self):
|
||||
d = {"a": "b"}
|
||||
b = util.dict_to_bytes(d)
|
||||
self.assertIsInstance(b, type(b""))
|
||||
self.assertEqual(b, b'{"a": "b"}')
|
||||
|
||||
def test_bytes_to_dict(self):
|
||||
b = b'{"a": "b", "c": 2}'
|
||||
d = util.bytes_to_dict(b)
|
||||
self.assertIsInstance(d, dict)
|
||||
self.assertEqual(d, {u"a": u"b", u"c": 2})
|
25
src/wormhole/util.py
Normal file
25
src/wormhole/util.py
Normal file
|
@ -0,0 +1,25 @@
|
|||
import json, unicodedata
|
||||
from binascii import hexlify, unhexlify
|
||||
|
||||
def to_bytes(u):
|
||||
return unicodedata.normalize("NFC", u).encode("utf-8")
|
||||
def bytes_to_hexstr(b):
|
||||
assert isinstance(b, type(b""))
|
||||
hexstr = hexlify(b).decode("ascii")
|
||||
assert isinstance(hexstr, type(u""))
|
||||
return hexstr
|
||||
def hexstr_to_bytes(hexstr):
|
||||
assert isinstance(hexstr, type(u""))
|
||||
b = unhexlify(hexstr.encode("ascii"))
|
||||
assert isinstance(b, type(b""))
|
||||
return b
|
||||
def dict_to_bytes(d):
|
||||
assert isinstance(d, dict)
|
||||
b = json.dumps(d).encode("utf-8")
|
||||
assert isinstance(b, type(b""))
|
||||
return b
|
||||
def bytes_to_dict(b):
|
||||
assert isinstance(b, type(b""))
|
||||
d = json.loads(b.decode("utf-8"))
|
||||
assert isinstance(d, dict)
|
||||
return d
|
|
@ -1,5 +1,5 @@
|
|||
from __future__ import print_function, absolute_import
|
||||
import os, sys, json, re, unicodedata
|
||||
import os, sys, json, re
|
||||
from six.moves.urllib_parse import urlparse
|
||||
from binascii import hexlify, unhexlify
|
||||
from twisted.internet import defer, endpoints, error
|
||||
|
@ -18,6 +18,7 @@ from . import codes
|
|||
from .errors import (WrongPasswordError, UsageError, WelcomeError,
|
||||
WormholeClosedError)
|
||||
from .timing import DebugTiming
|
||||
from .util import to_bytes
|
||||
from hkdf import Hkdf
|
||||
|
||||
def HKDF(skm, outlen, salt=None, CTXinfo=b""):
|
||||
|
@ -28,8 +29,6 @@ CONFMSG_MAC_LENGTH = 256//8
|
|||
def make_confmsg(confkey, nonce):
|
||||
return nonce+HKDF(confkey, CONFMSG_MAC_LENGTH, nonce)
|
||||
|
||||
def to_bytes(u):
|
||||
return unicodedata.normalize("NFC", u).encode("utf-8")
|
||||
|
||||
# We send the following messages through the relay server to the far side (by
|
||||
# sending "add" commands to the server, and getting "message" responses):
|
||||
|
|
Loading…
Reference in New Issue
Block a user