diff --git a/stubs/PySocks/@tests/stubtest_allowlist.txt b/stubs/PySocks/@tests/stubtest_allowlist.txt new file mode 100644 index 000000000..287b0a1f0 --- /dev/null +++ b/stubs/PySocks/@tests/stubtest_allowlist.txt @@ -0,0 +1,3 @@ +# Internal variables that were improperly leaked to the outside +socks.method +socks.name diff --git a/stubs/PySocks/METADATA.toml b/stubs/PySocks/METADATA.toml new file mode 100644 index 000000000..e99f8028c --- /dev/null +++ b/stubs/PySocks/METADATA.toml @@ -0,0 +1,2 @@ +version = "1.7.1" +upstream_repository = "https://github.com/Anorov/PySocks" diff --git a/stubs/PySocks/socks.pyi b/stubs/PySocks/socks.pyi new file mode 100644 index 000000000..931a8be77 --- /dev/null +++ b/stubs/PySocks/socks.pyi @@ -0,0 +1,143 @@ +import logging +import socket +import types +from _typeshed import ReadableBuffer +from collections.abc import Callable, Iterable, Mapping +from typing import Final, TypeVar, overload +from typing_extensions import ParamSpec, TypeAlias + +__version__: Final[str] + +log: logging.Logger # undocumented + +_ProxyType: TypeAlias = int + +PROXY_TYPE_SOCKS4: Final[_ProxyType] +SOCKS4: Final[_ProxyType] +PROXY_TYPE_SOCKS5: Final[_ProxyType] +SOCKS5: Final[_ProxyType] +PROXY_TYPE_HTTP: Final[_ProxyType] +HTTP: Final[_ProxyType] + +PROXY_TYPES: Final[dict[str, _ProxyType]] +PRINTABLE_PROXY_TYPES: Final[dict[_ProxyType, str]] + +_T = TypeVar("_T") +_P = ParamSpec("_P") + +def set_self_blocking(function: Callable[_P, _T]) -> Callable[_P, _T]: ... # undocumented + +class ProxyError(IOError): + msg: str + socket_err: socket.error + def __init__(self, msg: str, socket_err: socket.error | None = None) -> None: ... + +class GeneralProxyError(ProxyError): ... +class ProxyConnectionError(ProxyError): ... +class SOCKS5AuthError(ProxyError): ... +class SOCKS5Error(ProxyError): ... +class SOCKS4Error(ProxyError): ... +class HTTPError(ProxyError): ... + +SOCKS4_ERRORS: Final[Mapping[int, str]] +SOCKS5_ERRORS: Final[Mapping[int, str]] +DEFAULT_PORTS: Final[Mapping[_ProxyType, int]] + +_DefaultProxy: TypeAlias = tuple[_ProxyType | None, str | None, int | None, bool, bytes | None, bytes | None] + +def set_default_proxy( + proxy_type: _ProxyType | None = None, + addr: str | None = None, + port: int | None = None, + rdns: bool = True, + username: str | None = None, + password: str | None = None, +) -> None: ... +def setdefaultproxy( + proxy_type: _ProxyType | None = None, + addr: str | None = None, + port: int | None = None, + rdns: bool = True, + username: str | None = None, + password: str | None = None, + *, + proxytype: _ProxyType = ..., +) -> None: ... +def get_default_proxy() -> _DefaultProxy | None: ... + +getdefaultproxy = get_default_proxy + +def wrap_module(module: types.ModuleType) -> None: ... + +wrapmodule = wrap_module + +_Endpoint: TypeAlias = tuple[str, int] + +def create_connection( + dest_pair: _Endpoint, + timeout: int | None = None, + source_address: _Endpoint | None = None, + proxy_type: _ProxyType | None = None, + proxy_addr: str | None = None, + proxy_port: int | None = None, + proxy_rdns: bool = True, + proxy_username: str | None = None, + proxy_password: str | None = None, + socket_options: ( + Iterable[tuple[int, int, int | ReadableBuffer] | tuple[int, int, None, int]] | None + ) = None, # values passing to `socket.setsockopt` method +) -> socksocket: ... + +class _BaseSocket(socket.socket): # undocumented + ... + +class socksocket(_BaseSocket): + default_proxy: _DefaultProxy | None # undocumented + proxy: _DefaultProxy # undocumented + proxy_sockname: _Endpoint | None # undocumented + proxy_peername: _Endpoint | None # undocumented + def __init__( + self, family: socket.AddressFamily = ..., type: socket.SocketKind = ..., proto: int = 0, fileno: int | None = None + ) -> None: ... + def settimeout(self, timeout: float | None) -> None: ... + def gettimeout(self) -> float | None: ... + def setblocking(self, v: bool) -> None: ... + def set_proxy( + self, + proxy_type: _ProxyType | None = None, + addr: str | None = None, + port: int | None = None, + rdns: bool = True, + username: str | None = None, + password: str | None = None, + ) -> None: ... + def setproxy( + self, + proxy_type: _ProxyType | None = None, + addr: str | None = None, + port: int | None = None, + rdns: bool = True, + username: str | None = None, + password: str | None = None, + *, + proxytype: _ProxyType = ..., + ) -> None: ... + def bind(self, address: socket._Address, /) -> None: ... + @overload + def sendto(self, bytes: ReadableBuffer, address: socket._Address) -> int: ... + @overload + def sendto(self, bytes: ReadableBuffer, flags: int, address: socket._Address) -> int: ... + def send(self, bytes: ReadableBuffer, flags: int = 0) -> int: ... + def recvfrom(self, bufsize: int, flags: int = 0) -> tuple[bytes, _Endpoint]: ... + def recv(self, bufsize: int, flags: int = 0) -> bytes: ... + def close(self) -> None: ... + def get_proxy_sockname(self) -> _Endpoint | None: ... + getproxysockname = get_proxy_sockname + def get_proxy_peername(self) -> _Endpoint | None: ... + getproxypeername = get_proxy_peername + def get_peername(self) -> _Endpoint | None: ... + getpeername = get_peername + @set_self_blocking + def connect(self, dest_pair: _Endpoint, catch_errors: bool | None = None) -> None: ... # type: ignore[override] + @set_self_blocking + def connect_ex(self, dest_pair: _Endpoint) -> int: ... # type: ignore[override] diff --git a/stubs/PySocks/sockshandler.pyi b/stubs/PySocks/sockshandler.pyi new file mode 100644 index 000000000..7f13c16bf --- /dev/null +++ b/stubs/PySocks/sockshandler.pyi @@ -0,0 +1,102 @@ +import http.client +import ssl +import sys +import urllib.request +from _typeshed import Incomplete, SupportsKeysAndGetItem +from typing import Any, TypeVar +from typing_extensions import override + +import socks + +_K = TypeVar("_K") +_V = TypeVar("_V") + +def merge_dict(a: dict[_K, _V], b: SupportsKeysAndGetItem[_K, _V]) -> dict[_K, _V]: ... # undocumented +def is_ip(s: str) -> bool: ... # undocumented + +socks4_no_rdns: set[str] # undocumented + +class SocksiPyConnection(http.client.HTTPConnection): # undocumented + proxyargs: tuple[int, str, int | None, bool, str | None, str | None] + sock: socks.socksocket + def __init__( + self, + proxytype: int, + proxyaddr: str, + proxyport: int | None = None, + rdns: bool = True, + username: str | None = None, + password: str | None = None, + host: str | None = None, + port: int | None = None, + timeout: float | None = ..., + source_address: tuple[str, int] | None = None, + blocksize: int = 8192, + ) -> None: ... + @override + def connect(self) -> None: ... + +class SocksiPyConnectionS(http.client.HTTPSConnection): # undocumented + proxyargs: tuple[int, str, int | None, bool, str | None, str | None] + sock: socks.socksocket + if sys.version_info >= (3, 12): + def __init__( + self, + proxytype: int, + proxyaddr: str, + proxyport: int | None = None, + rdns: bool = True, + username: str | None = None, + password: str | None = None, + host: str | None = None, + port: int | None = None, + *, + timeout: float | None = ..., + source_address: tuple[str, int] | None = None, + context: ssl.SSLContext | None = None, + blocksize: int = 8192, + ) -> None: ... + else: + def __init__( + self, + proxytype: int, + proxyaddr: str, + proxyport: int | None = None, + rdns: bool = True, + username: str | None = None, + password: str | None = None, + host: str | None = None, + port: int | None = None, + key_file: str | None = None, + cert_file: str | None = None, + timeout: float | None = ..., + source_address: tuple[str, int] | None = None, + *, + context: ssl.SSLContext | None = None, + check_hostname: bool | None = None, + blocksize: int = 8192, + ) -> None: ... + + @override + def connect(self) -> None: ... + +class SocksiPyHandler(urllib.request.HTTPHandler, urllib.request.HTTPSHandler): + args: tuple[Incomplete, ...] # undocumented + kw: dict[str, Incomplete] # undocumented + def __init__( + self, + proxytype: int, + proxyaddr: str, + proxyport: int | None = None, + rdns: bool = True, + username: str | None = None, + password: str | None = None, + *, + source_address: tuple[str, int] | None = None, + blocksize: int = 8192, + **kwargs: Any, # any additional arguments to `SocksiPyConnection` or `SocksiPyConnectionS` + ) -> None: ... + @override + def http_open(self, req: urllib.request.Request) -> http.client.HTTPResponse: ... # undocumented + @override + def https_open(self, req: urllib.request.Request) -> http.client.HTTPResponse: ... # undocumented