""" Unit tests for stem.descriptor.remote. """ import io import socket import time import unittest import stem import stem.descriptor.remote import stem.prereq import stem.util.str_tools from stem.descriptor.remote import Compression from test.unit.descriptor import read_resource try: from http.client import HTTPMessage # python3 except ImportError: from httplib import HTTPMessage # python2 try: # added in python 3.3 from unittest.mock import patch, Mock, MagicMock except ImportError: from mock import patch, Mock, MagicMock # The urlopen() method is in a different location depending on if we're using # python 2.x or 3.x. The 2to3 converter accounts for this in imports, but not # mock annotations. URL_OPEN = 'urllib.request.urlopen' if stem.prereq.is_python_3() else 'urllib2.urlopen' TEST_RESOURCE = '/tor/server/fp/9695DFC35FFEB861329B9F1AB04C46397020CE31' # Output from requesting moria1's descriptor from itself... # % curl http://128.31.0.39:9131/tor/server/fp/9695DFC35FFEB861329B9F1AB04C46397020CE31 TEST_DESCRIPTOR = b"""\ router moria1 128.31.0.34 9101 0 9131 platform Tor 0.2.5.0-alpha-dev on Linux protocols Link 1 2 Circuit 1 published 2013-07-05 23:48:52 fingerprint 9695 DFC3 5FFE B861 329B 9F1A B04C 4639 7020 CE31 uptime 1818933 bandwidth 512000 62914560 1307929 extra-info-digest 17D0142F6EBCDF60160EB1794FA6C9717D581F8C caches-extra-info onion-key -----BEGIN RSA PUBLIC KEY----- MIGJAoGBALzd4bhz1usB7wpoaAvP+BBOnNIk7mByAKV6zvyQ0p1M09oEmxPMc3qD AAm276oJNf0eq6KWC6YprzPWFsXEIdXSqA6RWXCII1JG/jOoy6nt478BkB8TS9I9 1MJW27ppRaqnLiTmBmM+qzrsgJGwf+onAgUKKH2GxlVgahqz8x6xAgMBAAE= -----END RSA PUBLIC KEY----- signing-key -----BEGIN RSA PUBLIC KEY----- MIGJAoGBALtJ9uD7cD7iHjqNA3AgsX9prES5QN+yFQyr2uOkxzhvunnaf6SNhzWW bkfylnMrRm/qCz/czcjZO6N6EKHcXmypehvP566B7gAQ9vDsb+l7VZVWgXvzNc2s tl3P7qpC08rgyJh1GqmtQTCesIDqkEyWxwToympCt09ZQRq+fIttAgMBAAE= -----END RSA PUBLIC KEY----- hidden-service-dir contact 1024D/28988BF5 arma mit edu ntor-onion-key 9ZVjNkf/iLEnD685SpC5kcDytQ7u5ViiI9JOftdbE0k= reject *:* router-signature -----BEGIN SIGNATURE----- Y8Tj2e7mPbFJbguulkPEBVYzyO57p4btpWEXvRMD6vxIh/eyn25pehg5dUVBtZlL iO3EUE0AEYah2W9gdz8t+i3Dtr0zgqLS841GC/TyDKCm+MKmN8d098qnwK0NGF9q 01NZPuSqXM1b6hnl2espFzL7XL8XEGRU+aeg+f/ukw4= -----END SIGNATURE----- """ HEADER = '\r\n'.join([ 'Date: Fri, 13 Apr 2018 16:35:50 GMT', 'Content-Type: application/octet-stream', 'X-Your-Address-Is: 97.103.17.56', 'Pragma: no-cache', 'Content-Encoding: %s', ]) def _orport_mock(data, encoding = 'identity', response_code_header = None): if response_code_header is None: response_code_header = b'HTTP/1.0 200 OK\r\n' data = response_code_header + stem.util.str_tools._to_bytes(HEADER % encoding) + b'\r\n\r\n' + data cells = [] for hunk in [data[i:i + 50] for i in range(0, len(data), 50)]: cell = Mock() cell.data = hunk cells.append(cell) connect_mock = MagicMock() relay_mock = connect_mock().__enter__() circ_mock = relay_mock.create_circuit().__enter__() circ_mock.send.return_value = cells return connect_mock def _dirport_mock(data, encoding = 'identity'): dirport_mock = Mock() dirport_mock().read.return_value = data if stem.prereq.is_python_3(): headers = HTTPMessage() for line in HEADER.splitlines(): key, value = line.split(': ', 1) headers.add_header(key, encoding if key == 'Content-Encoding' else value) dirport_mock().headers = headers else: dirport_mock().headers = HTTPMessage(io.BytesIO(HEADER % encoding)) return dirport_mock class TestDescriptorDownloader(unittest.TestCase): def tearDown(self): # prevent our mocks from impacting other tests stem.descriptor.remote.SINGLETON_DOWNLOADER = None @patch('stem.client.Relay.connect', _orport_mock(TEST_DESCRIPTOR)) def test_using_orport(self): """ Download a descriptor through the ORPort. """ reply = stem.descriptor.remote.their_server_descriptor( endpoints = [stem.ORPort('12.34.56.78', 1100)], validate = True, ) self.assertEqual(1, len(list(reply))) self.assertEqual('moria1', list(reply)[0].nickname) self.assertEqual(5, len(reply.reply_headers)) def test_orport_response_code_headers(self): """ When successful Tor provides a '200 OK' status, but we should accept other 2xx response codes, reason text, and recognize HTTP errors. """ response_code_headers = ( b'HTTP/1.0 200 OK\r\n', b'HTTP/1.0 205 OK\r\n', b'HTTP/1.0 200 This is also alright\r\n', ) for header in response_code_headers: with patch('stem.client.Relay.connect', _orport_mock(TEST_DESCRIPTOR, response_code_header = header)): stem.descriptor.remote.their_server_descriptor( endpoints = [stem.ORPort('12.34.56.78', 1100)], validate = True, ).run() with patch('stem.client.Relay.connect', _orport_mock(TEST_DESCRIPTOR, response_code_header = b'HTTP/1.0 500 Kaboom\r\n')): request = stem.descriptor.remote.their_server_descriptor( endpoints = [stem.ORPort('12.34.56.78', 1100)], validate = True, ) self.assertRaisesWith(stem.ProtocolError, "Response should begin with HTTP success, but was 'HTTP/1.0 500 Kaboom'", request.run) @patch(URL_OPEN, _dirport_mock(TEST_DESCRIPTOR)) def test_using_dirport(self): """ Download a descriptor through the DirPort. """ reply = stem.descriptor.remote.their_server_descriptor( endpoints = [stem.DirPort('12.34.56.78', 1100)], validate = True, ) self.assertEqual(1, len(list(reply))) self.assertEqual('moria1', list(reply)[0].nickname) self.assertEqual(5, len(reply.reply_headers)) def test_gzip_url_override(self): query = stem.descriptor.remote.Query(TEST_RESOURCE, start = False) self.assertEqual([Compression.PLAINTEXT], query.compression) self.assertEqual(TEST_RESOURCE, query.resource) query = stem.descriptor.remote.Query(TEST_RESOURCE + '.z', compression = Compression.PLAINTEXT, start = False) self.assertEqual([Compression.GZIP], query.compression) self.assertEqual(TEST_RESOURCE, query.resource) def test_zstd_support_check(self): with patch('stem.prereq.is_zstd_available', Mock(return_value = True)): query = stem.descriptor.remote.Query(TEST_RESOURCE, compression = Compression.ZSTD, start = False) self.assertEqual([Compression.ZSTD], query.compression) with patch('stem.prereq.is_zstd_available', Mock(return_value = False)): query = stem.descriptor.remote.Query(TEST_RESOURCE, compression = Compression.ZSTD, start = False) self.assertEqual([Compression.PLAINTEXT], query.compression) def test_lzma_support_check(self): with patch('stem.prereq.is_lzma_available', Mock(return_value = True)): query = stem.descriptor.remote.Query(TEST_RESOURCE, compression = Compression.LZMA, start = False) self.assertEqual([Compression.LZMA], query.compression) with patch('stem.prereq.is_lzma_available', Mock(return_value = False)): query = stem.descriptor.remote.Query(TEST_RESOURCE, compression = Compression.LZMA, start = False) self.assertEqual([Compression.PLAINTEXT], query.compression) @patch(URL_OPEN, _dirport_mock(read_resource('compressed_identity'), encoding = 'identity')) def test_compression_plaintext(self): """ Download a plaintext descriptor. """ descriptors = list(stem.descriptor.remote.get_server_descriptors( '9695DFC35FFEB861329B9F1AB04C46397020CE31', compression = Compression.PLAINTEXT, validate = True, )) self.assertEqual(1, len(descriptors)) self.assertEqual('moria1', descriptors[0].nickname) @patch(URL_OPEN, _dirport_mock(read_resource('compressed_gzip'), encoding = 'gzip')) def test_compression_gzip(self): """ Download a gip compressed descriptor. """ descriptors = list(stem.descriptor.remote.get_server_descriptors( '9695DFC35FFEB861329B9F1AB04C46397020CE31', compression = Compression.GZIP, validate = True, )) self.assertEqual(1, len(descriptors)) self.assertEqual('moria1', descriptors[0].nickname) @patch(URL_OPEN, _dirport_mock(read_resource('compressed_zstd'), encoding = 'x-zstd')) def test_compression_zstd(self): """ Download a zstd compressed descriptor. """ if not stem.prereq.is_zstd_available(): self.skipTest('(requires zstd module)') return descriptors = list(stem.descriptor.remote.get_server_descriptors( '9695DFC35FFEB861329B9F1AB04C46397020CE31', compression = Compression.ZSTD, validate = True, )) self.assertEqual(1, len(descriptors)) self.assertEqual('moria1', descriptors[0].nickname) @patch(URL_OPEN, _dirport_mock(read_resource('compressed_lzma'), encoding = 'x-tor-lzma')) def test_compression_lzma(self): """ Download a lzma compressed descriptor. """ if not stem.prereq.is_lzma_available(): self.skipTest('(requires lzma module)') return descriptors = list(stem.descriptor.remote.get_server_descriptors( '9695DFC35FFEB861329B9F1AB04C46397020CE31', compression = Compression.LZMA, validate = True, )) self.assertEqual(1, len(descriptors)) self.assertEqual('moria1', descriptors[0].nickname) @patch(URL_OPEN, _dirport_mock(TEST_DESCRIPTOR)) def test_reply_headers(self): query = stem.descriptor.remote.get_server_descriptors('9695DFC35FFEB861329B9F1AB04C46397020CE31', start = False) self.assertEqual(None, query.reply_headers) # initially we don't have a reply query.run() self.assertEqual('Fri, 13 Apr 2018 16:35:50 GMT', query.reply_headers.get('date')) self.assertEqual('application/octet-stream', query.reply_headers.get('content-type')) self.assertEqual('97.103.17.56', query.reply_headers.get('x-your-address-is')) self.assertEqual('no-cache', query.reply_headers.get('pragma')) self.assertEqual('identity', query.reply_headers.get('content-encoding')) # getting headers should be case insensitive self.assertEqual('identity', query.reply_headers.get('CoNtEnT-ENCODING')) # request a header that isn't present self.assertEqual(None, query.reply_headers.get('no-such-header')) self.assertEqual('default', query.reply_headers.get('no-such-header', 'default')) descriptors = list(query) self.assertEqual(1, len(descriptors)) self.assertEqual('moria1', descriptors[0].nickname) @patch(URL_OPEN, _dirport_mock(TEST_DESCRIPTOR)) def test_query_download(self): """ Check Query functionality when we successfully download a descriptor. """ query = stem.descriptor.remote.Query( TEST_RESOURCE, 'server-descriptor 1.0', endpoints = [('128.31.0.39', 9131)], compression = Compression.PLAINTEXT, validate = True, ) self.assertEqual(stem.DirPort('128.31.0.39', 9131), query._pick_endpoint()) descriptors = list(query) self.assertEqual(1, len(descriptors)) desc = descriptors[0] self.assertEqual('moria1', desc.nickname) self.assertEqual('128.31.0.34', desc.address) self.assertEqual('9695DFC35FFEB861329B9F1AB04C46397020CE31', desc.fingerprint) self.assertEqual(TEST_DESCRIPTOR.strip(), desc.get_bytes()) @patch(URL_OPEN, _dirport_mock(b'some malformed stuff')) def test_query_with_malformed_content(self): """ Query with malformed descriptor content. """ query = stem.descriptor.remote.Query( TEST_RESOURCE, 'server-descriptor 1.0', endpoints = [('128.31.0.39', 9131)], compression = Compression.PLAINTEXT, validate = True, ) # checking via the iterator expected_error_msg = 'Content conform to being a server descriptor:\nsome malformed stuff' descriptors = list(query) self.assertEqual(0, len(descriptors)) self.assertEqual(ValueError, type(query.error)) self.assertEqual(expected_error_msg, str(query.error)) # check via the run() method self.assertRaises(ValueError, query.run) @patch(URL_OPEN) def test_query_with_timeout(self, dirport_mock): def urlopen_call(*args, **kwargs): time.sleep(0.06) raise socket.timeout('connection timed out') dirport_mock.side_effect = urlopen_call query = stem.descriptor.remote.Query( TEST_RESOURCE, 'server-descriptor 1.0', endpoints = [('128.31.0.39', 9131)], fall_back_to_authority = False, timeout = 0.1, validate = True, ) # After two requests we'll have reached our total permissable timeout. # Check that we don't make a third. self.assertRaises(socket.timeout, query.run) self.assertEqual(2, dirport_mock.call_count) def test_query_with_invalid_endpoints(self): invalid_endpoints = { 'hello': "'h' is a str.", ('hello',): "'hello' is a str.", (15,): "'15' is a int.", (('12.34.56.78', 15, 'third arg'),): "'('12.34.56.78', 15, 'third arg')' is a tuple.", } for endpoints, error_suffix in invalid_endpoints.items(): expected_error = 'Endpoints must be an stem.ORPort, stem.DirPort, or two value tuple. ' + error_suffix self.assertRaisesWith(ValueError, expected_error, stem.descriptor.remote.Query, TEST_RESOURCE, 'server-descriptor 1.0', endpoints = endpoints) @patch(URL_OPEN, _dirport_mock(TEST_DESCRIPTOR)) def test_can_iterate_multiple_times(self): query = stem.descriptor.remote.Query( TEST_RESOURCE, 'server-descriptor 1.0', endpoints = [('128.31.0.39', 9131)], compression = Compression.PLAINTEXT, validate = True, ) # check that iterating over the query provides the descriptors each time self.assertEqual(1, len(list(query))) self.assertEqual(1, len(list(query))) self.assertEqual(1, len(list(query)))