commit 434638a828201c286e5fef158761dc2c4c67e47d Author: Kevin Froman beardog@mailbox.org Date: Fri Feb 5 23:31:32 2021 +0000
Now validate Tor v3 address checksum in hidden service address validator --- stem/util/tor_tools.py | 19 ++++++++++++++++ test/unit/util/tor_tools.py | 55 +++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 74 insertions(+)
diff --git a/stem/util/tor_tools.py b/stem/util/tor_tools.py index e2df4312..c9a04e33 100644 --- a/stem/util/tor_tools.py +++ b/stem/util/tor_tools.py @@ -20,6 +20,8 @@ Miscellaneous utility functions for working with tor. """
import re +from base64 import b32decode +from hashlib import sha3_256
import stem.util.str_tools
@@ -45,6 +47,7 @@ CIRC_ID_PATTERN = re.compile('^[a-zA-Z0-9]{1,16}$')
HS_V2_ADDRESS_PATTERN = re.compile('^[a-z2-7]{16}$') HS_V3_ADDRESS_PATTERN = re.compile('^[a-z2-7]{56}$') +HS_V3_CHECKSUM_CONSTANT = ".onion checksum"
def is_valid_fingerprint(entry: str, check_prefix: bool = False) -> bool: @@ -153,6 +156,14 @@ def is_valid_hidden_service_address(entry: str, version: Optional[Union[int, Seq otherwise """
+ def _extract_v3_parts(address): + decoded = b32decode(address.upper()) + return (decoded[:32], decoded[32:34]) + + v3_pubkey = None + v3_checksum = None + v3_version = int(3).to_bytes(1, 'little') + if isinstance(entry, bytes): entry = stem.util.str_tools._to_unicode(entry)
@@ -168,6 +179,14 @@ def is_valid_hidden_service_address(entry: str, version: Optional[Union[int, Seq return True
if 3 in version and bool(HS_V3_ADDRESS_PATTERN.match(entry)): + # v3+ onions have a consistent version at end of address + if not entry.endswith('d'): + return False + # Test that the checksum (part of every v3 address) is valid + v3_pubkey, v3_checksum = _extract_v3_parts(entry) + expected_checksum = sha3_256(HS_V3_CHECKSUM_CONSTANT.encode('utf-8') + v3_pubkey + v3_version).digest()[:2] + if expected_checksum != v3_checksum: + return False return True
return False diff --git a/test/unit/util/tor_tools.py b/test/unit/util/tor_tools.py index 5035d7b4..285b4dde 100644 --- a/test/unit/util/tor_tools.py +++ b/test/unit/util/tor_tools.py @@ -3,12 +3,67 @@ Unit tests for the stem.util.tor_tools functions. """
import unittest +import os
import stem.util.str_tools import stem.util.tor_tools
class TestTorTools(unittest.TestCase): + + def test_is_valid_hidden_service_address(self): + """ + Check hidden service addresses are valid (no .onion) + """ + valid_v2_addresses = [ + 'facebookcorewwwi', + 'aaaaaaaaaaaaaaaa', + ] + invalid_v2_addresses = [ + 'facebookcorewww', + 'facebookcorewwyi' + 'facebookc0rewwwi', + 'facebookcorew wi', + None, + 0, + 1, + -1, + os.urandom(8) + ] + + for address in valid_v2_addresses: + self.assertTrue(stem.util.tor_tools.is_valid_hidden_service_address(address)) + self.assertTrue(stem.util.tor_tools.is_valid_hidden_service_address(address, version=2)) + + for address in invalid_v2_addresses: + self.assertFalse(stem.util.tor_tools.is_valid_hidden_service_address(address)) + self.assertFalse(stem.util.tor_tools.is_valid_hidden_service_address(address, version=2)) + + # Test version 3 addresses + valid_v3_addresses = [ + 'pg6mmjiyjmcrsslvykfwnntlaru7p5svn6y2ymmju6nubxndf4pscryd', + 'sp3k262uwy4r2k3ycr5awluarykdpag6a7y33jxop4cs2lu5uz5sseqd', + 'xa4r2iadxm55fbnqgwwi5mymqdcofiu3w6rpbtqn7b2dyn7mgwj64jyd' + ] + invalid_v3_addresses = [ + 'pg6mmjiyjmcrsslvykfwnntlaru7p5svn6y2ymmju6nubxndf4pscryc', # bad version + 'xa4r2iadxm55fbnqgwwi5mymqdcofiu3w6rpbtqn7b2dyn7mgwj64jy', # too short + 'sp3k262uwy4r2k4ycr5awluarykdpag6a7y33jxop4cs2lu5uz5sseqd', # checksum mismatch + 'pg6mmjiyjmcrsslvykfwnntlaru7p5svn6y2ymmju6nubxndf4pscrybd', # too long + None, + 0, + 1, + -1, + os.urandom(56) + ] + + for address in valid_v3_addresses: + self.assertTrue(stem.util.tor_tools.is_valid_hidden_service_address(address)) + self.assertTrue(stem.util.tor_tools.is_valid_hidden_service_address(address, version=3)) + for address in invalid_v3_addresses: + self.assertFalse(stem.util.tor_tools.is_valid_hidden_service_address(address)) + self.assertFalse(stem.util.tor_tools.is_valid_hidden_service_address(address, version=3)) + def test_is_valid_fingerprint(self): """ Checks the is_valid_fingerprint function.