Skip to content

Commit

Permalink
chore: removed unused code
Browse files Browse the repository at this point in the history
  • Loading branch information
leoslf committed Jun 9, 2024
1 parent 05fac3e commit dfb14f6
Show file tree
Hide file tree
Showing 4 changed files with 6 additions and 131 deletions.
27 changes: 6 additions & 21 deletions socks_router/models.py
Original file line number Diff line number Diff line change
Expand Up @@ -33,22 +33,6 @@ class SupportsUnbytes(Protocol):
def __unbytes__(cls, input: bytes) -> Self: ...


# class BytePackable(metaclass=ABCMeta):
# @classmethod
# def __pack_format__(cls) -> str:
# return "!B"

# class ByteEnum(IntEnum):
# def __new__(cls, value: int) -> Self:
# if not 0 <= value < 256:
# raise ValueError("value must be 0 <= value < 256")
# return super().__new__(cls, value)
#
# @classmethod
# def __pack_format__(cls) -> str:
# return "!B"


@dataclass(frozen=True)
class SocketAddress:
address: Any = field()
Expand Down Expand Up @@ -221,17 +205,18 @@ def address_type(cls, type: Socks5AddressType) -> Type[IPv4] | Type[Host] | Type

@dataclass(frozen=True)
class Socks5Request:
"""SEE: https://datatracker.ietf.org/doc/html/rfc1928#section-4"""
"""SEE: https://datatracker.ietf.org/doc/html/rfc1928#section-4
Request
-------
| version | cmd | rsv | atyp | dst.addr | dst.port |
| 1 byte | 1 byte | 0x00 | 1 byte | 4-255 bytes | 2 bytes |
"""

version: Annotated[int, "!B"]
command: Socks5Command
reserved: Annotated[int, "!B"]
destination: Socks5Address

# @property
# def destination(self):
# return Socks5Addresses[self.address_type](self.destination_address, self.destination_port)


class Socks5ReplyType(IntEnum):
"""SEE: https://datatracker.ietf.org/doc/html/rfc1928#section-6"""
Expand Down
44 changes: 0 additions & 44 deletions socks_router/router.py
Original file line number Diff line number Diff line change
Expand Up @@ -55,50 +55,6 @@ def free_port(address: str = "") -> tuple[str, int]:
return s.getsockname()


# def read_address(type: Socks5AddressType, connection: socket.socket) -> str:
# match type:
# case Socks5AddressType.IPv4:
# return socket.inet_ntop(socket.AF_INET, connection.recv(4))
# case Socks5AddressType.DOMAINNAME:
# (address_length,) = connection.recv(1)
# return connection.recv(address_length).decode("utf-8")
# case Socks5AddressType.IPv6:
# return socket.inet_ntop(socket.AF_INET6, connection.recv(6))
# case _ as unreachable:
# assert_never(unreachable)


# def read_method_selection_request(
# connection: socket.socket,
# ) -> Socks5MethodSelectionRequest:
# """Read Socks5 Method Selection Request.
# SEE: https://datatracker.ietf.org/doc/html/rfc1928#section-3
# Method Selection Request
# ------------------------
# | version | method_count | methods |
# | 1 byte | 1 byte | [method_count] bytes |
# """
# version, method_count = struct.unpack("!BB", connection.recv(2))
# # get available methods
# methods = connection.recv(method_count)
#
# return Socks5MethodSelectionRequest(version, list(map(int, methods)))


# def read_request(connection: socket.socket) -> Socks5Request:
# """Read Request.
# SEE: https://datatracker.ietf.org/doc/html/rfc1928#section-4
# Request
# -------
# | version | cmd | rsv | atyp | dst.addr | dst.port |
# | 1 byte | 1 byte | 0x00 | 1 byte | 4-255 bytes | 2 bytes |
# """
# version, command, reserved, address_type = struct.unpack("!BBBB", connection.recv(4))
# address = read_address(address_type, connection)
# (port,) = struct.unpack("!H", connection.recv(2))
# return Socks5Request(version, command, reserved, address_type, address, port)


def create_socket(type: Socks5AddressType) -> socks.socksocket:
match type:
case Socks5AddressType.IPv4 | Socks5AddressType.DOMAINNAME:
Expand Down
10 changes: 0 additions & 10 deletions tests/test_models.py
Original file line number Diff line number Diff line change
Expand Up @@ -47,13 +47,3 @@ def test_Host(address, port, host):
assert repr(host)
assert str(host)
assert host.pattern


# @pytest.mark.parametrize(
# "method_selection_request",
# [
# Socks5MethodSelectionRequest(SOCKS_VERSION, [Socks5Method.NO_AUTHENTICATION_REQUIRED]),
# ],
# )
# def test_Socks5MethodSelectionRequest(method_selection_request):
# assert bytes(method_selection_request)
56 changes: 0 additions & 56 deletions tests/test_router.py
Original file line number Diff line number Diff line change
Expand Up @@ -28,9 +28,6 @@
)
from socks_router.router import (
free_port,
# read_request,
# read_address,
# read_method_selection_request,
create_socket,
with_proxy,
connect_remote,
Expand Down Expand Up @@ -70,59 +67,6 @@ def it_should_return_a_free_port(mocker, address: str, port: int):
MocketSocket._address = None


# def describe_read_address():
# @pytest.mark.parametrize("ipv4", ["1.2.3.4"])
# def it_should_successfully_read_ipv4(mocker, client, ipv4):
# client.recv.side_effect = [socket.inet_pton(socket.AF_INET, ipv4)]
# assert read_address(Socks5AddressType.IPv4, client) == ipv4
#
# @pytest.mark.parametrize("domainname", ["foo.com"])
# def it_should_successfully_read_domainname(client, domainname):
# encoded = domainname.encode("utf-8")
# client.recv.side_effect = [struct.pack("!B", len(encoded)), encoded]
# assert read_address(Socks5AddressType.DOMAINNAME, client) == domainname
#
# @pytest.mark.parametrize("ipv6", ["::1"])
# def it_should_successfully_read_ipv6(client, ipv6):
# client.recv.return_value = socket.inet_pton(socket.AF_INET6, ipv6)
# assert read_address(Socks5AddressType.IPv6, client) == ipv6


# def describe_read_request():
# @pytest.mark.parametrize(
# "version,command,address_type,address,port",
# [
# (
# SOCKS_VERSION,
# Socks5Command.CONNECT,
# Socks5AddressType.IPv4,
# "127.0.0.1",
# 1234,
# ),
# ],
# )
# def it_should_successfully_read(mocker, client, version, command, address_type, address, port):
# client.recv.side_effect = [
# struct.pack("!BBBB", version, command, 0, address_type),
# struct.pack("!H", port),
# ]
# mocker.patch("socks_router.router.read_address").return_value = address
# assert read_request(client) == Socks5Request(version, command, 0x00, address_type, address, port)


# def describe_read_method_selection_request():
# @pytest.mark.parametrize(
# "version,methods",
# [
# (SOCKS_VERSION, []),
# (SOCKS_VERSION, [1, 2, 3]),
# ],
# )
# def it_should_correctly_read(client, version, methods):
# client.recv.side_effect = [struct.pack("!BB", version, len(methods)), methods]
# assert read_method_selection_request(client) == Socks5MethodSelectionRequest(version, methods)


def describe_create_socket():
@pytest.mark.parametrize(
"type",
Expand Down

0 comments on commit dfb14f6

Please sign in to comment.