work on flood fill network

src/streamfill/__init__.py

@@ -0,0 +1 @@
+from .extracted25519 import extract_ed25519_from_onion_address

src/streamfill/extracted25519.py

@@ -0,0 +1,14 @@
+from base64 import b32decode
+from typing import TYPE_CHECKING
+
+from onionrutils.bytesconverter import str_to_bytes
+
+if TYPE_CHECKING:
+    from onionrtypes import Ed25519PublicKeyBytes, OnionAddressString
+
+
+def extract_ed25519_from_onion_address(
+        address: OnionAddressString) -> Ed25519PublicKeyBytes:
+    address = str_to_bytes(address).replace(b'.onion', b'').upper()
+    ed25519 = b32decode(address)[:-3]
+    return ed25519

src/streamfill/neighbors.py

@@ -0,0 +1,22 @@
+from onionrtypes import OnionAddressString
+from typing import Iterable
+from .extracted25519 import extract_ed25519_from_onion_address
+
+
+def identify_neighbors(
+        address: OnionAddressString,
+        peers: Iterable[OnionAddressString],
+        closest_n: int) -> OnionAddressString:
+
+    address = extract_ed25519_from_onion_address(address)
+    address_int = int.from_bytes(address, "big")
+
+    def _calc_closeness(y):
+        return abs(address_int - int.from_bytes(y, "big"))
+
+
+    peer_ed_keys = list(map(extract_ed25519_from_onion_address, peers))
+    differences = list(map(_calc_closeness, peer_ed_keys))
+
+    return sorted(differences)[:closest_n]
+

tests/test_streamfill_extract_ed25519.py

@@ -0,0 +1,53 @@
+#!/usr/bin/env python3
+import sys, os
+sys.path.append(".")
+sys.path.append("src/")
+import uuid
+import binascii
+from base64 import b32decode
+TEST_DIR = 'testdata/%s-%s' % (uuid.uuid4(), os.path.basename(__file__)) + '/'
+print("Test directory:", TEST_DIR)
+os.environ["ONIONR_HOME"] = TEST_DIR
+import unittest
+
+from streamfill import extract_ed25519_from_onion_address
+
+class TestStreamfillExtractEd25519(unittest.TestCase):
+    def test_extract_normal_onion(self):
+        hardcodedCorrect = b'y\xbc\xc6%\x18K\x05\x19Iu\xc2\x8bf\xb6k\x04i\xf7\xf6Uo\xb1\xac1\x89\xa7\x9b@\xdd\xa3/\x1f'
+        correct = b32decode(b'pg6mmjiyjmcrsslvykfwnntlaru7p5svn6y2ymmju6nubxndf4pscryd'.upper())[:-3]
+        self.assertEqual(correct, hardcodedCorrect)
+        self.assertEqual(correct, extract_ed25519_from_onion_address('pg6mmjiyjmcrsslvykfwnntlaru7p5svn6y2ymmju6nubxndf4pscryd.onion'))
+    def test_extract_normal_onion_bytes(self):
+        hardcodedCorrect = b'y\xbc\xc6%\x18K\x05\x19Iu\xc2\x8bf\xb6k\x04i\xf7\xf6Uo\xb1\xac1\x89\xa7\x9b@\xdd\xa3/\x1f'
+        correct = b32decode(b'pg6mmjiyjmcrsslvykfwnntlaru7p5svn6y2ymmju6nubxndf4pscryd'.upper())[:-3]
+        self.assertEqual(correct, hardcodedCorrect)
+        self.assertEqual(correct, extract_ed25519_from_onion_address(b'pg6mmjiyjmcrsslvykfwnntlaru7p5svn6y2ymmju6nubxndf4pscryd.onion'))
+    def test_extract_no_ext(self):
+        hardcodedCorrect = b'y\xbc\xc6%\x18K\x05\x19Iu\xc2\x8bf\xb6k\x04i\xf7\xf6Uo\xb1\xac1\x89\xa7\x9b@\xdd\xa3/\x1f'
+        correct = b32decode(b'pg6mmjiyjmcrsslvykfwnntlaru7p5svn6y2ymmju6nubxndf4pscryd'.upper())[:-3]
+        self.assertEqual(correct, hardcodedCorrect)
+        self.assertEqual(correct, extract_ed25519_from_onion_address('pg6mmjiyjmcrsslvykfwnntlaru7p5svn6y2ymmju6nubxndf4pscryd'))
+    def test_extract_no_ext_bytes(self):
+        hardcodedCorrect = b'y\xbc\xc6%\x18K\x05\x19Iu\xc2\x8bf\xb6k\x04i\xf7\xf6Uo\xb1\xac1\x89\xa7\x9b@\xdd\xa3/\x1f'
+        correct = b32decode(b'pg6mmjiyjmcrsslvykfwnntlaru7p5svn6y2ymmju6nubxndf4pscryd'.upper())[:-3]
+        self.assertEqual(correct, hardcodedCorrect)
+        self.assertEqual(correct, extract_ed25519_from_onion_address(b'pg6mmjiyjmcrsslvykfwnntlaru7p5svn6y2ymmju6nubxndf4pscryd'))
+    def test_fail_length(self):
+        hardcodedCorrect = b'y\xbc\xc6%\x18K\x05\x19Iu\xc2\x8bf\xb6k\x04i\xf7\xf6Uo\xb1\xac1\x89\xa7\x9b@\xdd\xa3/\x1f'
+        correct = b32decode(b'pg6mmjiyjmcrsslvykfwnntlaru7p5svn6y2ymmju6nubxndf4pscryd'.upper())[:-3]
+        self.assertEqual(correct, hardcodedCorrect)
+        try:
+            extract_ed25519_from_onion_address(b'pg6mmjiyjmcrsslvykfwnntlaru7p5svn6y2ymmju6nubxndf4pscry')
+        except binascii.Error:
+            pass
+    def test_fail_length_onion(self):
+        hardcodedCorrect = b'y\xbc\xc6%\x18K\x05\x19Iu\xc2\x8bf\xb6k\x04i\xf7\xf6Uo\xb1\xac1\x89\xa7\x9b@\xdd\xa3/\x1f'
+        correct = b32decode(b'pg6mmjiyjmcrsslvykfwnntlaru7p5svn6y2ymmju6nubxndf4pscryd'.upper())[:-3]
+        self.assertEqual(correct, hardcodedCorrect)
+        try:
+            extract_ed25519_from_onion_address(b'pg6mmjiyjmcrsslvykfwnntlaru7p5svn6y2ymmju6nubxndf4pscry.onion')
+        except binascii.Error:
+            pass
+
+unittest.main()