|
|
@@ -0,0 +1,49 @@ |
|
|
|
import aiosocks |
|
|
|
import asyncio |
|
|
|
from tests.test_functional import with_timeout |
|
|
|
|
|
|
|
machip = '192.168.0.29' |
|
|
|
socksport = 2345 |
|
|
|
targport = 20834 |
|
|
|
|
|
|
|
@with_timeout(2) |
|
|
|
async def xtest_live_datagram(): |
|
|
|
loop = asyncio.get_event_loop() |
|
|
|
addr = aiosocks.Socks5Addr('127.0.0.1', socksport) |
|
|
|
|
|
|
|
dst = (machip, targport) |
|
|
|
dgram = await aiosocks.open_datagram(addr, None, dst, loop=loop) |
|
|
|
|
|
|
|
dgram.send('this is a test\r\n') |
|
|
|
|
|
|
|
import socket |
|
|
|
|
|
|
|
def xtest_simpletcp(): |
|
|
|
s = socket.socket() |
|
|
|
s.connect(('127.0.0.1', socksport)) |
|
|
|
|
|
|
|
s.send(b'\x05\x01\x00') |
|
|
|
assert s.recv(2) == b'\x05\x00' |
|
|
|
|
|
|
|
s.send(b'\x05\x01\x00\x01' + socket.inet_pton(socket.AF_INET, machip) + (targport).to_bytes(2, 'big')) |
|
|
|
|
|
|
|
assert s.recv(4) == b'\x05\x00\x00\x01' |
|
|
|
print(socket.inet_ntop(socket.AF_INET, s.recv(4)), ':', int.from_bytes(s.recv(2), 'big')) |
|
|
|
|
|
|
|
s.send(b'foobar\r\n') |
|
|
|
|
|
|
|
def xtest_simpleudp(): |
|
|
|
s = socket.socket() |
|
|
|
s.connect(('127.0.0.1', socksport)) |
|
|
|
|
|
|
|
s.send(b'\x05\x01\x00') |
|
|
|
assert s.recv(2) == b'\x05\x00' |
|
|
|
|
|
|
|
s.send(b'\x05\x03\x00\x01' + socket.inet_pton(socket.AF_INET, machip) + (53).to_bytes(2, 'big')) |
|
|
|
print('a') |
|
|
|
|
|
|
|
assert s.recv(4) == b'\x05\x00\x00\x01' |
|
|
|
print('a') |
|
|
|
print(socket.inet_ntop(socket.AF_INET, s.recv(4)), ':', int.from_bytes(s.recv(2), 'big')) |
|
|
|
print('a') |
|
|
|
assert False |