+ finished basic client tests

* made connection var more meaningful in server
* added support for .onion-ending (or not) address format in client
This commit is contained in:
Kevin Froman 2020-04-24 18:43:24 -05:00
parent a10cc5e2e1
commit ef1b56a84b
4 changed files with 138 additions and 23 deletions

View File

@ -7,11 +7,14 @@ from .commands import garbage_character
from .commands import WELCOME_MESSAGE from .commands import WELCOME_MESSAGE
def client(delay: int, hs_id, socks_port, send_data: bytearray, recv_data: bytearray, connection: "Connection"): def client(delay: int, hs_id: str, socks_port, send_data: bytearray, recv_data: bytearray, connection: "Connection"):
s = socks.socksocket() # Same API as socket.socket in the standard lib s = socks.socksocket() # Same API as socket.socket in the standard lib
s.set_proxy(socks.SOCKS5, "127.0.0.1", socks_port, rdns=True) s.set_proxy(socks.SOCKS5, "127.0.0.1", socks_port, rdns=True)
if not hs_id.endswith('.onion'):
hs_id += '.onion'
def send_loop(): def send_loop():
for i in WELCOME_MESSAGE: for i in WELCOME_MESSAGE:
send_data.append(ord(i)) send_data.append(ord(i))

View File

@ -1,35 +1,39 @@
import socket import socket
import time import time
from threading import Thread
import typing
if typing.TYPE_CHECKING:
from stem.control import Controller
from.commands import garbage_character from.commands import garbage_character
from.commands import WELCOME_MESSAGE from.commands import WELCOME_MESSAGE
from threading import Thread
def server(delay: int, controller: 'Controller', socket_conn, send_data: bytearray, recv_data: bytearray, connection: "Connection"):
def server(delay: int, controller, conn, send_data: bytearray, recv_data: bytearray, connection: "Connection"):
def send_loop(): def send_loop():
while True: while True:
time.sleep(delay) time.sleep(delay)
try: try:
if not send_data: if not send_data:
conn.sendall(garbage_character) socket_conn.sendall(garbage_character)
else: else:
char = send_data.pop(0) char = send_data.pop(0)
try: try:
conn.sendall(ord(char)) socket_conn.sendall(ord(char))
except TypeError: except TypeError:
try: try:
conn.sendall(char) socket_conn.sendall(char)
except TypeError: except TypeError:
conn.sendall(chr(char).encode('utf-8')) socket_conn.sendall(chr(char).encode('utf-8'))
except OSError: except OSError:
connection.connected = False connection.connected = False
first_rec = True first_rec = True
with conn: with socket_conn:
Thread(target=send_loop, daemon=True).start() Thread(target=send_loop, daemon=True).start()
while True: while True:
try: try:
data = conn.recv(1) data = socket_conn.recv(1)
except ConnectionResetError: except ConnectionResetError:
connection.connected = False connection.connected = False
break break

View File

@ -1,6 +1,7 @@
import unittest import unittest
import socket import socket
from threading import Thread from threading import Thread
import sys
import time import time
import stem import stem
import stem.process import stem.process
@ -33,24 +34,49 @@ config = {
], ],
}, take_ownership=True) }, take_ownership=True)
def send_test_data(ip, port): class Address:
time.sleep(2) address = ""
s = socket.socket()
s.connect(('127.0.0.1', port)) def fake_server():
while True: with Controller.from_port(port=int(control_port)) as controller:
s.send(b"test") controller.authenticate()
for c in b"test2": with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
try: ip = '127.0.0.1'
s.send(chr(c).encode('utf8')) s.bind((ip, 0))
except TypeError: s.listen(1)
print(c) port = s.getsockname()[1]
s.close() serv = controller.create_ephemeral_hidden_service(
{1337: '127.0.0.1:' + str(port)},
key_content='ED25519-V3',
await_publication=True,
)
Address.address = serv.service_id
conn, addr = s.accept()
while True:
conn.send(chr(116).encode('utf-8'))
data = conn.recv(1)
class TestClient(unittest.TestCase): class TestClient(unittest.TestCase):
def test_client(self): def test_client(self):
Thread(target=fake_server, daemon=True).start()
send_data = bytearray() send_data = bytearray()
recv_data = bytearray() recv_data = bytearray()
#client(1, ) while Address.address == "":
time.sleep(1)
print(Address.address)
Thread(target=client.client, args=[0.01, Address.address, int(socks_port), send_data, recv_data, Connection], daemon=True).start()
start = time.time()
try:
while True:
try:
if chr(recv_data.pop(0)) in "t"*100:
break
except IndexError:
self.assertLess((time.time() - start), 20)
time.sleep(0.01)
except KeyboardInterrupt:
raise
unittest.main() unittest.main()

View File

@ -0,0 +1,82 @@
import unittest
import socket
from threading import Thread
import sys
import time
import stem
import stem.process
from stem.control import Controller
from youandme import client
class Connection:
connected = True
def get_open_port():
# taken from (but modified) https://stackoverflow.com/a/2838309 by https://stackoverflow.com/users/133374/albert ccy-by-sa-3 https://creativecommons.org/licenses/by-sa/3.0/
# changes from source: import moved to top of file, bind specifically to localhost
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.bind(("127.0.0.1",0))
s.listen(1)
port = s.getsockname()[1]
s.close()
return port
control_port = str(get_open_port())
socks_port = str(get_open_port())
assert control_port != socks_port
stem.process.launch_tor_with_config(
config = {
'ControlPort': control_port,
'SocksPort': socks_port,
'Log': [
'NOTICE stdout'
],
}, take_ownership=True)
class Address:
address = ""
def fake_server():
with Controller.from_port(port=int(control_port)) as controller:
controller.authenticate()
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
ip = '127.0.0.1'
s.bind((ip, 0))
s.listen(1)
port = s.getsockname()[1]
serv = controller.create_ephemeral_hidden_service(
{1337: '127.0.0.1:' + str(port)},
key_content='ED25519-V3',
await_publication=True,
)
Address.address = serv.service_id + '.onion'
conn, addr = s.accept()
while True:
conn.send(chr(116).encode('utf-8'))
data = conn.recv(1)
class TestClient(unittest.TestCase):
def test_client(self):
Thread(target=fake_server, daemon=True).start()
send_data = bytearray()
recv_data = bytearray()
while Address.address == "":
time.sleep(1)
print(Address.address)
Thread(target=client.client, args=[0.01, Address.address, int(socks_port), send_data, recv_data, Connection], daemon=True).start()
start = time.time()
try:
while True:
try:
if chr(recv_data.pop(0)) in "t"*100:
break
except IndexError:
self.assertLess((time.time() - start), 20)
time.sleep(0.01)
except KeyboardInterrupt:
raise
unittest.main()