Initial Commit

Author: MHProDev

.gitignore

@@ -0,0 +1,3 @@
+/venv/
+*/__pycache__
+/.idea/

PyRoxy/Exceptions/__init__.py

@@ -0,0 +1,14 @@
+class ProxyParseError(Exception):
+    pass
+
+
+class ProxyInvalidPort(ProxyParseError):
+    def __init__(self, port: int):
+        ProxyParseError.__init__(self, "'%d' is too %s" % (port, "small" if port < 1 else "long"))
+
+
+
+class ProxyInvalidHost(ProxyParseError):
+    def __init__(self, host: str):
+        ProxyParseError.__init__(self, "'%s' is an Invalid IP Address" % host)
+

PyRoxy/GeoIP/Sqlite/COPYRIGHT.txt

@@ -0,0 +1 @@
+Database and Contents Copyright (c) 2022 MaxMind, Inc.

PyRoxy/GeoIP/Sqlite/GeoLite2-Country.mmdb

@@ -0,0 +1,3 @@
+Use of this MaxMind product is governed by MaxMind's GeoLite2 End User License Agreement, which can be viewed at https://www.maxmind.com/en/geolite2/eula.
+
+This database incorporates GeoNames [https://www.geonames.org] geographical data, which is made available under the Creative Commons Attribution 4.0 License. To view a copy of this license, visit https://creativecommons.org/licenses/by/4.0/.

PyRoxy/GeoIP/Sqlite/LICENSE.txt

@@ -0,0 +1,13 @@
+from pathlib import Path as _path
+from maxminddb import open_database as _open
+
+__dir__ = _path(__file__).parent
+
+_reader = _open(__dir__ / 'Sqlite/GeoLite2-Country.mmdb')
+
+def get(ip: str):
+    return _reader.get(ip)
+
+
+def get_with_prefix_len(ip: str):
+    return _reader.get_with_prefix_len(ip)

PyRoxy/GeoIP/__init__.py

@@ -0,0 +1,41 @@
+from re import compile, IGNORECASE, MULTILINE
+from contextlib import suppress
+from os import urandom
+from socket import inet_ntop, inet_ntoa, AF_INET6
+from string import ascii_letters
+from struct import pack
+from struct import pack as data_pack
+from sys import maxsize
+from typing import Callable, Any, List
+
+
+class Random:
+    latters: List[str] = list(ascii_letters)
+    rand_str: Callable[[int], str] = lambda length=16: ''.join(Random.rand_choice(*Random.latters) for _ in range(length))
+    rand_ipv4: Callable[[], str] = lambda: inet_ntoa(data_pack('>I', Random.rand_int(1, 0xffffffff)))
+    rand_ipv6: Callable[[], str] = lambda: inet_ntop(AF_INET6, pack('>QQ', Random.rand_bits(64), Random.rand_bits(64)))
+    rand_int: Callable[[int, int], int] = lambda minimum=0, maximum=maxsize: int(Random.rand_float(minimum, maximum))
+    rand_choice: Callable[[List[Any]], Any] = lambda *data: data[(Random.rand_int(maximum=len(data) - 2) or 0)]
+    rand: Callable[[], int] = lambda: (int.from_bytes(urandom(7), 'big') >> 3) * (2 ** -53)
+
+    @staticmethod
+    def rand_bits(maximum: int = 255) -> int:
+        numbytes = (maximum + 7) // 8
+        return int.from_bytes(urandom(numbytes), 'big') >> (numbytes * 8 - maximum)
+
+    @staticmethod
+    def rand_float(minimum: float = 0.0, maximum: float = (maxsize * 1.0)) -> float:
+        with suppress(ZeroDivisionError):
+            return abs((Random.rand() * maximum) % (minimum - (maximum + 1))) + minimum
+        return 0.0
+
+class Patterns:
+    Port = compile("^((6553[0-5])|(655[0-2][0-9])|(65[0-4][0-9]"
+                      "{2})|(6[0-4][0-9]{3})|([1-5][0-9]{4})|([0-5]{0,5})|([0-9]{1,4}))$")
+    IP = compile("(25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)")
+    IPPort = compile("(25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?):(\d{1,5})")
+    Proxy = compile(r"^(?:\[|)(?:\s+|)(?:socks[45]|http(?:s|))(?:[]|]|)(?:\s+|)(?:](?:\s+|)|\|(?:\s+|)|://(?:\s+|)|)"
+                       r"((?:(?:\d+.){3}\d+|\S+[.]\w{2,3}))"
+                       r"(?:[:]|)((?:(\d+)|))"
+                       r"(?::(.+):(.+)|)$", IGNORECASE | MULTILINE)
+    URL = compile("\S+[.]\w{2,3}")

PyRoxy/Tools/__init__.py

@@ -0,0 +1,170 @@
+from contextlib import suppress
+from enum import IntEnum, auto
+from functools import partial
+from ipaddress import ip_address
+from pathlib import Path
+from socket import socket, SOCK_STREAM, AF_INET, gethostbyname
+from threading import Lock, Thread
+from typing import Match, AnyStr, Set, Collection
+
+from maxminddb.types import RecordDict
+from requests import Session
+from socks import socksocket, SOCKS4, SOCKS5, HTTP
+from yarl import URL
+
+from PyRoxy import GeoIP, Tools
+from PyRoxy.Exceptions import ProxyInvalidHost, ProxyInvalidPort, ProxyParseError
+from PyRoxy.Tools import Patterns
+
+
+class ProxyType(IntEnum):
+    HTTP = auto()
+    HTTPS = auto()
+    SOCKS4 = auto()
+    SOCKS5 = auto()
+
+    def asPySocksType(self):
+        return SOCKS5 if self == ProxyType.SOCKS5 else \
+            SOCKS4 if self == ProxyType.SOCKS4 else \
+                HTTP
+
+    @staticmethod
+    def stringToProxyType(n: str):
+        return ProxyType.HTTP if not (n.isdigit() and not (int(n) == 1)) else \
+            int(n) if int(n) in PRINTABLE_PROXY_TYPES else \
+                ProxyType.SOCKS5 if int(n) == 5 else \
+                    ProxyType.SOCKS4 if int(n) == 4 else \
+                        ProxyType.HTTP
+
+
+PROXY_TYPES = {"SOCKS4": ProxyType.SOCKS4, "SOCKS5": ProxyType.SOCKS5, "HTTP": ProxyType.HTTP, "HTTPS": ProxyType.HTTPS}
+PRINTABLE_PROXY_TYPES = dict(zip(PROXY_TYPES.values(), PROXY_TYPES.keys()))
+
+
+class Proxy(object):
+    user: AnyStr | None
+    password: AnyStr | None
+    country: AnyStr | RecordDict | None
+    port: int
+    type: ProxyType
+    host: AnyStr
+
+    def __init__(self, host: str, port: int = 0, proxy_type: ProxyType = ProxyType.HTTP, user=None, password=None):
+        if Patterns.URL.match(host): host = gethostbyname(host)
+        assert self.validate(host, port)
+        self.host = host
+        self.type = proxy_type
+        self.port = port
+        self.country = GeoIP.get(host)
+        if self.country: self.country = self.country["country"]["iso_code"]
+        self.user = user or None
+        self.password = password or None
+
+    def __str__(self):
+        return "%s://%s:%d%s" % (self.type.name.lower(), self.host, self.port, ("%s:%s" % (self.user, self.password)
+                                                                                if self.password and self.user else ""))
+
+    def __repr__(self):
+        return "<Proxy %s:%d>" % (self.host, self.port)
+
+    @staticmethod
+    def fromString(string: str):
+        with suppress(KeyboardInterrupt):
+            proxy: Match[str] | None = Patterns.Proxy.search(string)
+            return Proxy(proxy.group(1),
+                         int(proxy.group(2)) if proxy.group(3) and proxy.group(2).isdigit() else 80,
+                         ProxyType.stringToProxyType(proxy.group(1)),
+                         proxy.group(3),
+                         proxy.group(4))
+        raise ProxyParseError("'%s' is an Invalid Proxy URL" % string)
+
+    def ip_port(self):
+        return "%s:%d" % (self.host, self.port)
+
+    @staticmethod
+    def validate(host: str, port: int):
+        with suppress(ValueError):
+            if not ip_address(host):
+                raise ProxyInvalidHost(host)
+            if not Tools.Patterns.Port.match(str(port)):
+                raise ProxyInvalidPort(port)
+            return True
+        raise ProxyInvalidHost(host)
+
+    # noinspection PyShadowingBuiltins
+    def open_socket(self, family=AF_INET, type=SOCK_STREAM, proto=-1, fileno=None):
+        return ProxySocket(self, family, type, proto, fileno)
+
+    def wrap(self, sock: socket | Session):
+        if isinstance(sock, socket):
+            return self.open_socket(sock.family, sock.type, sock.proto, sock.fileno())
+        sock.proxies = self.asRequest()
+        return sock
+
+    def asRequest(self):
+        return {"http": self.__str__()}
+
+    # noinspection PyUnreachableCode
+    def check(self, url: str | URL = "https://httpbin.org/get", timeout=5):
+        if not isinstance(url, URL): url = URL(url)
+        with suppress(KeyboardInterrupt):
+            with self.open_socket() as sock:
+                sock.settimeout(timeout)
+                return sock.connect((url.host, url.port or 80))
+        return False
+
+
+# noinspection PyShadowingBuiltins
+class ProxySocket(socksocket):
+    def __init__(self, proxy: Proxy, family=-1, type=-1, proto=-1, fileno=None):
+        super().__init__(family, type, proto, fileno)
+        if proxy.port:
+            if proxy.user and proxy.password:
+                self.setproxy(proxy.type.asPySocksType(), proxy.host, proxy.port, username=proxy.user,
+                              password=proxy.password)
+                return
+            self.setproxy(proxy.type.asPySocksType(), proxy.host, proxy.port)
+            return
+        if proxy.user and proxy.password:
+            self.setproxy(proxy.type.asPySocksType(), proxy.host, username=proxy.user,
+                          password=proxy.password)
+            return
+        self.setproxy(proxy.type.asPySocksType(), proxy.host)
+
+
+class ProxyChecker:
+    result: Set[Proxy]
+    out_lock: Lock
+
+    def __init__(self):
+        self.out_lock = Lock()
+        self.result = set()
+
+    def check(self, proxy: Proxy, url: str | URL = "https://httpbin.org/get", timeout=5):
+        with suppress(Exception), self.out_lock:
+            if proxy.check(url, timeout):
+                self.result.add(proxy)
+
+    def checkAll(self, proxies: Collection[Proxy], url: str | URL = "https://httpbin.org/get", timeout=5):
+        threads = map(partial(Thread, target=self.check, args=(url, timeout,)), proxies)
+        for thr in threads:
+            thr.start()
+        for thr in threads:
+            thr.join()
+
+
+class ProxyUtiles:
+    @staticmethod
+    def parseAll(proxies: Collection[str]) -> Set[Proxy]:
+        return set(map(Proxy.fromString, proxies))
+
+    @staticmethod
+    def readFromFile(path: Path | str) -> Set[Proxy]:
+        if isinstance(path, Path):
+            with path.open("r+") as read:
+                lines = read.readlines()
+        else:
+            with open(path, "r+") as read:
+                lines = read.readlines()
+
+        return ProxyUtiles.parseAll([prox.strip() for prox in lines])

PyRoxy/__init__.py

@@ -0,0 +1,4 @@
+maxminddb>=2.2.0
+requests>=2.27.1
+yarl>=1.7.2
+pysocks>=1.7.1

requirements.txt

@@ -0,0 +1,12 @@
+from setuptools import setup
+
+setup(
+    name='PyRoxy',
+    version='',
+    packages=['PyRoxy', 'PyRoxy.GeoIP', 'PyRoxy.Tools', 'PyRoxy.Exceptions'],
+    url='https://github.com/MHProDev/PyRoxy',
+    license='',
+    author='MH_ProDev',
+    author_email='',
+    description=''
+)

setup.py

@@ -0,0 +1,4 @@
+from PyRoxy import Proxy, ProxyType, ProxyUtiles
+
+if __name__ == '__main__':
+    print(ProxyUtiles.readFromFile("test.txt"))

test.py

@@ -0,0 +1 @@
+[socks5] 209.127.191.180:9279:eclahjmw:sis9bjnxuds5