diff --git a/README.md b/README.md index 51a35a3..c1f6d46 100644 --- a/README.md +++ b/README.md @@ -35,6 +35,7 @@ A modular, terminal-based toolkit for OSINT, reconnaissance, and scraping - buil | 15 | **Bluetooth Scanner** | Scans for nearby Bluetooth devices via `bluetoothctl` (Linux) and reports device names and MAC addresses. *(Windows support coming soon.)* | | 16 | **Local Users** | Enumerates all local user accounts on the system. On Linux: username, UID, GID, full name, home directory, shell, and group. On Windows: username, terminal, host, session start time, PID, SID, and domain. | | 17 | **Git Scrape** | Scrapes and analyzes a Git repository (remote/local) or a GitHub account for possible personal email leaks and other data. *(Other platforms will be supported soon.)* | +| 18 | **Proxy Manager** | Configure HTTP/SOCKS proxies to route H4X-Tools network traffic through, with optional round-robin rotation to avoid rate limiting. Proxies are persisted in `$HOME/.config/h4x-tools/config.json` and respected by all compatible tools automatically. | --- @@ -188,9 +189,94 @@ For larger tools, keep the UI wrapper in `tools/my_tool.py` and place reusable i ## Running with proxies -If you are encountering rate limits or wish to mask your traffic, you can route H4X-Tools through proxies using **ProxyChains**. This tool intercepts the network traffic generated by H4X-Tools and forces it through your specified proxy list. +H4X-Tools supports two proxy methods. The **built-in Proxy Manager** (method 1) is the recommended starting point. It requires no extra software and integrates directly with the toolkit. **ProxyChains** (method 2) is a system-level alternative that is useful when you need to cover tools that the built-in manager cannot reach. -### 1. Installation +--- + +## Method 1 — Built-in Proxy Manager + +The built-in Proxy Manager lets you add, test, and enable proxies from inside H4X-Tools. Once enabled, every compatible tool automatically routes its traffic through your configured proxies. No extra command or configuration file needed. + +### Supported proxy formats + +``` +http://host:port +http://username:password@host:port +https://host:port +socks4://host:port +socks5://host:port +socks5://username:password@host:port +``` + +### Opening the Proxy Manager + +Select **Proxy Manager** from the interactive menu, or run it directly: + +```sh +python h4xtools.py --proxy-manager +``` + +### Quick setup walkthrough + +**1. Add a proxy** + +Choose option `[2] Add proxy` and enter the proxy URL: + +``` +[?] Proxy URL : http://my-proxy.example.com:8080 +[+] Proxy added: http://my-proxy.example.com:8080 +``` + +**2. Test it** + +Choose option `[4] Test all proxies`. H4X-Tools will connect to [ipinfo.io](https://ipinfo.io) through each proxy and report the seen exit IP: + +``` +[*] Testing http://my-proxy.example.com:8080 ... +[+] OK http://my-proxy.example.com:8080 -> exit IP: 203.0.113.42 +[*] Results: 1/1 proxies working. +``` + +**3. Enable routing** + +Choose option `[5] Toggle proxy routing`. The status line at the top of the menu changes to `Enabled`. From this point, all compatible tools route their traffic through the proxy automatically. The main menu also shows a `[P]` indicator confirming that routing is active. + +### Rotation + +When multiple proxies are added, round-robin rotation is on by default — each outgoing request uses the next proxy in the list. Toggle it off with option `[6] Toggle rotation` to always use the first proxy instead. + +### Persisting settings + +All settings (proxy list, enabled state, rotation mode) are saved to `$HOME/.config/h4x-tools/config.json` and restored automatically on the next launch. + +### Coverage + +The built-in manager covers the following tools automatically when enabled: + +| Tool | Proxy support | +|---|---| +| IP Lookup | Full (HTTP + SOCKS) | +| IG Scrape | Full (HTTP + SOCKS) | +| Leak Search | Full (HTTP + SOCKS) | +| Git Scrape | Full (HTTP + SOCKS) | +| Username Search | Full (HTTP + SOCKS, via Maigret `--proxy`) | +| Web Reconnaissance | Full (HTTP + SOCKS) | +| Web Scrape | HTTP/HTTPS proxies only | +| Dir Buster | HTTP/HTTPS proxies only | +| Email Search | Not supported (holehe subprocess) | +| Phone Lookup | Not supported (ignorant subprocess) | +| WhoIs Lookup | Not supported (raw WHOIS socket protocol) | + +> [!TIP] +> Web Scrape and Dir Buster use `aiohttp`, which natively supports HTTP/HTTPS proxies only. SOCKS proxies are silently skipped for these two tools. Use ProxyChains (method 2) if you need SOCKS coverage for them. + +--- + +## Method 2 — ProxyChains + +If you are encountering rate limits or wish to mask your traffic, you can route H4X-Tools through proxies using **ProxyChains**. This tool intercepts the network traffic generated by H4X-Tools at the OS level and forces it through your specified proxy list, covering every tool including subprocess-based ones like holehe and ignorant. + +### Installation #### Debian / Ubuntu @@ -225,7 +311,7 @@ scoop install proxychains --- -### 2. Configuration +### Configuration Before running the tool, you need to tell ProxyChains which proxies to use. @@ -249,7 +335,7 @@ http 192.168.1.50 8080 # Public or private HTTP proxy --- -### 3. Usage +### Usage Once configured, simply prefix your standard startup command with `proxychains4` (or `proxychains` on Windows): diff --git a/h4xtools.py b/h4xtools.py index 286904b..5e2485e 100755 --- a/h4xtools.py +++ b/h4xtools.py @@ -24,7 +24,7 @@ from colorama import Fore, Style -from helper import config, printer +from helper import config, printer, proxymanager from tools import BaseTool, discover_tools QUIT_COMMANDS = {"quit", "exit", "q", "kill"} @@ -137,6 +137,22 @@ def _print_menu(tools: tuple[BaseTool, ...]) -> None: print(" " * 4, end="") print("\n") + + if proxymanager.is_enabled(): + proxy_list = proxymanager.list_proxies() + count = len(proxy_list) + mode = "rotating" if proxymanager.is_rotating() else "fixed" + if count: + print( + f"{Fore.LIGHTGREEN_EX}[P]{Style.RESET_ALL} " + f"Proxy routing active / {count} {'proxy' if count == 1 else 'proxies'} ({mode})" + ) + else: + print( + f"{Fore.LIGHTYELLOW_EX}[P]{Style.RESET_ALL} " + "Proxy routing enabled but no proxies configured." + ) + print(f"Type {Style.BRIGHT}?{Style.RESET_ALL} for help.") print(f"Type {Style.BRIGHT}exit{Style.RESET_ALL} to close the toolkit...") diff --git a/helper/proxymanager.py b/helper/proxymanager.py new file mode 100644 index 0000000..c24fe15 --- /dev/null +++ b/helper/proxymanager.py @@ -0,0 +1,310 @@ +""" +Copyright (c) 2023-2026. Vili and contributors. + +This program is free software: you can redistribute it and/or modify +it under the terms of the GNU General Public License as published by +the Free Software Foundation, either version 3 of the License, or +(at your option) any later version. + +This program is distributed in the hope that it will be useful, +but WITHOUT ANY WARRANTY; without even the implied warranty of +MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +GNU General Public License for more details. + +You should have received a copy of the GNU General Public License +along with this program. If not, see . +""" + +from urllib.parse import urlparse +from typing import Optional + +import requests + +from helper import config, printer + +_SECTION = "proxies" +_TEST_URL = "https://ipinfo.io" +_REQUEST_TIMEOUT = 10 + +# In-memory round-robin counter; modded against list length on each call. +_counter: list[int] = [0] + +_VALID_SCHEMES = ("http", "https", "socks4", "socks5") + + +# --------------------------------------------------------------------------- +# Internal helpers +# --------------------------------------------------------------------------- + + +def _load_proxy_list() -> list[str]: + """Load the saved proxy list from config.""" + raw = config.get_value(_SECTION, "list", default=[]) + if isinstance(raw, list): + return [str(p) for p in raw if p] + return [] + + +def _save_proxy_list(proxies: list[str]) -> bool: + """Persist the proxy list to config.""" + return config.set_value(_SECTION, "list", proxies) + + +# --------------------------------------------------------------------------- +# Enable / rotation flags +# --------------------------------------------------------------------------- + + +def is_enabled() -> bool: + """ + Return whether proxy routing is active. + + :return: ``True`` if proxies are enabled. + """ + return bool(config.get_value(_SECTION, "enabled", default=False)) + + +def set_enabled(enabled: bool) -> bool: + """ + Enable or disable proxy routing globally. + + :param enabled: ``True`` to route traffic through proxies. + :return: ``True`` when the setting was saved successfully. + """ + return config.set_value(_SECTION, "enabled", enabled) + + +def is_rotating() -> bool: + """ + Return whether proxies rotate in round-robin order. + + When rotation is off the first proxy in the list is always used. + + :return: ``True`` if rotation is enabled. + """ + return bool(config.get_value(_SECTION, "rotate", default=True)) + + +def set_rotating(rotate: bool) -> bool: + """ + Enable or disable round-robin proxy rotation. + + :param rotate: ``True`` to rotate proxies on each request. + :return: ``True`` when the setting was saved successfully. + """ + return config.set_value(_SECTION, "rotate", rotate) + + +# --------------------------------------------------------------------------- +# Proxy list management +# --------------------------------------------------------------------------- + + +def list_proxies() -> list[str]: + """ + Return the saved list of proxy URLs. + + :return: List of proxy URL strings. + """ + return _load_proxy_list() + + +def validate_proxy_url(url: str) -> bool: + """ + Check whether *url* looks like a valid proxy URL. + + Accepted schemes: ``http``, ``https``, ``socks4``, ``socks5``. + A hostname and explicit port are both required. + + :param url: Proxy URL to validate. + :return: ``True`` if the URL is well-formed. + """ + try: + parsed = urlparse(url) + return ( + parsed.scheme in _VALID_SCHEMES + and bool(parsed.hostname) + and parsed.port is not None + ) + except Exception: + return False + + +def add_proxy(url: str) -> bool: + """ + Validate and append a proxy URL to the saved list. + + :param url: Proxy URL to add (e.g. ``http://host:3128`` or ``socks5://user:pass@host:1080``). + :return: ``True`` if the proxy was added successfully. + """ + url = url.strip() + if not validate_proxy_url(url): + printer.error( + f"Invalid proxy URL: {url!r} " + f"Expected: scheme://[user:pass@]host:port " + f"(scheme is one of: {', '.join(_VALID_SCHEMES)})" + ) + return False + + proxies = _load_proxy_list() + if url in proxies: + printer.warning("Proxy is already in the list.") + return False + + proxies.append(url) + if _save_proxy_list(proxies): + printer.success(f"Proxy added: {url}") + return True + return False + + +def remove_proxy(index: int) -> bool: + """ + Remove a proxy by its 0-based position in the list. + + :param index: Zero-based index of the proxy to remove. + :return: ``True`` if removed successfully. + """ + proxies = _load_proxy_list() + if index < 0 or index >= len(proxies): + printer.error(f"No proxy at index {index}. List has {len(proxies)} entries.") + return False + + removed = proxies.pop(index) + if _save_proxy_list(proxies): + printer.success(f"Proxy removed: {removed}") + return True + return False + + +def clear_proxies() -> bool: + """ + Remove all saved proxies and reset the rotation counter. + + :return: ``True`` if cleared successfully. + """ + _counter[0] = 0 + if _save_proxy_list([]): + printer.success("All proxies cleared.") + return True + return False + + +# --------------------------------------------------------------------------- +# Proxy selection +# --------------------------------------------------------------------------- + + +def get_proxy() -> Optional[str]: + """ + Return the next proxy URL to use. + + Applies round-robin rotation when enabled, otherwise always returns the + first proxy. Returns ``None`` when proxies are disabled or none are + configured. + + :return: Proxy URL string or ``None``. + """ + if not is_enabled(): + return None + + proxies = _load_proxy_list() + if not proxies: + return None + + if is_rotating(): + index = _counter[0] % len(proxies) + _counter[0] += 1 + else: + index = 0 + + return proxies[index] + + +def get_aiohttp_proxy() -> Optional[str]: + """ + Return the next proxy URL for use with ``aiohttp``. + + aiohttp supports HTTP proxies natively via the ``proxy=`` keyword argument + on request calls. SOCKS proxies require the optional ``aiohttp-socks`` + package and are not returned by this helper; use ``get_requests_proxies()`` + with the ``requests`` library for full SOCKS proxy support. + + :return: HTTP(S) proxy URL string, or ``None``. + """ + proxy = get_proxy() + if proxy is None: + return None + if not proxy.startswith(("http://", "https://")): + return None + return proxy + + +def get_requests_proxies() -> Optional[dict[str, str]]: + """ + Return a proxies dict ready for use with the ``requests`` library. + + Pass the result directly to the ``proxies`` keyword argument of + ``requests.get``, ``requests.post``, or a ``requests.Session``. + Returns ``None`` when proxy use is disabled or no proxies are saved. + + :return: ``{"http": proxy_url, "https": proxy_url}`` or ``None``. + """ + proxy = get_proxy() + if proxy is None: + return None + return {"http": proxy, "https": proxy} + + +# --------------------------------------------------------------------------- +# Proxy testing +# --------------------------------------------------------------------------- + + +def test_proxy(url: str, timeout: int = _REQUEST_TIMEOUT) -> bool: + """ + Verify that *url* is a reachable, working proxy. + + Sends a request to ``ipinfo.io`` through the proxy and reports the + apparent exit IP if successful. + + :param url: Proxy URL to test. + :param timeout: Request timeout in seconds. + :return: ``True`` if the proxy responded successfully. + """ + proxies = {"http": url, "https": url} + try: + response = requests.get(_TEST_URL, proxies=proxies, timeout=timeout) + response.raise_for_status() + ip = response.json().get("ip", "?") + printer.success(f"OK {url} -> exit IP: {ip}") + return True + except requests.exceptions.ProxyError as exc: + printer.error(f"Proxy error for {url!r}: {exc}") + except requests.exceptions.Timeout: + printer.error(f"Timed out after {timeout}s: {url!r}") + except requests.exceptions.RequestException as exc: + printer.error(f"Request failed for {url!r}: {exc}") + return False + + +def test_all_proxies() -> dict[str, bool]: + """ + Test every saved proxy and return a pass/fail map. + + :return: Mapping of proxy URL to ``True`` (working) or ``False`` (failed). + """ + proxies = _load_proxy_list() + if not proxies: + printer.warning("No proxies configured.") + return {} + + results: dict[str, bool] = {} + for proxy in proxies: + printer.info(f"Testing {proxy} ...") + results[proxy] = test_proxy(proxy) + + passed = sum(results.values()) + printer.noprefix("") + printer.info(f"Results: {passed}/{len(results)} proxies working.") + return results diff --git a/requirements.txt b/requirements.txt index 73c06b4..be9c2ae 100644 --- a/requirements.txt +++ b/requirements.txt @@ -2,6 +2,7 @@ colorama ddgs phonenumbers requests +PySocks beautifulsoup4 urllib3 whoisdomain diff --git a/tools/proxy_manager.py b/tools/proxy_manager.py new file mode 100644 index 0000000..0b33125 --- /dev/null +++ b/tools/proxy_manager.py @@ -0,0 +1,146 @@ +""" +Copyright (c) 2023-2026. Vili and contributors. + +This program is free software: you can redistribute it and/or modify +it under the terms of the GNU General Public License as published by +the Free Software Foundation, either version 3 of the License, or +(at your option) any later version. + +This program is distributed in the hope that it will be useful, +but WITHOUT ANY WARRANTY; without even the implied warranty of +MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +GNU General Public License for more details. + +You should have received a copy of the GNU General Public License +along with this program. If not, see . +""" + +from helper import printer, proxymanager +from tools.base import BaseTool + +QUIT_COMMANDS = {"quit", "exit", "q", "kill", "0"} + +class ProxyManagerTool(BaseTool): + id = "proxy_manager" + name = "Proxy Manager" + order = 99 + aliases = ("--proxy-manager", "--proxies") + description = ( + "Configure HTTP/SOCKS proxies for H4X-Tools to route network traffic through, " + "with optional round-robin rotation to avoid rate limiting." + ) + + # No CLI arguments - the tool always opens its interactive sub-menu. + + def run(self) -> None: + self._menu() + + # ------------------------------------------------------------------ + # Interactive menu + # ------------------------------------------------------------------ + + def _menu(self) -> None: + while True: + proxies = proxymanager.list_proxies() + enabled = proxymanager.is_enabled() + rotating = proxymanager.is_rotating() + + printer.noprefix("") + printer.section("Proxy Manager") + printer.info(f"Status : {'Enabled' if enabled else 'Disabled'}") + printer.info(f"Rotation : {'On (round-robin)' if rotating else 'Off (first proxy only)'}") + printer.info(f"Proxies : {len(proxies)} configured") + printer.noprefix("") + printer.noprefix("[1] List proxies") + printer.noprefix("[2] Add proxy") + printer.noprefix("[3] Remove proxy") + printer.noprefix("[4] Test all proxies") + printer.noprefix("[5] Toggle proxy routing") + printer.noprefix("[6] Toggle rotation") + printer.noprefix("[7] Clear all proxies") + printer.noprefix("[0] Back") + printer.noprefix("") + + choice = printer.user_input("Choose an option : \t").strip() + + if choice == "1": + self._list_proxies() + elif choice == "2": + self._add_proxy() + elif choice == "3": + self._remove_proxy() + elif choice == "4": + self._test_proxies() + elif choice == "5": + self._toggle_enabled(enabled) + elif choice == "6": + self._toggle_rotation(rotating) + elif choice == "7": + self._clear_proxies() + elif choice in QUIT_COMMANDS: + break + else: + printer.error("Invalid option.") + + # ------------------------------------------------------------------ + # Actions + # ------------------------------------------------------------------ + + def _list_proxies(self) -> None: + proxies = proxymanager.list_proxies() + printer.noprefix("") + if not proxies: + printer.warning("No proxies configured.") + return + printer.section("Configured Proxies") + for i, proxy in enumerate(proxies): + printer.success(f"[{i}] {proxy}") + + def _add_proxy(self) -> None: + printer.noprefix("") + printer.info("Supported proxy formats:") + printer.noprefix(" http://host:port") + printer.noprefix(" http://user:pass@host:port") + printer.noprefix(" socks5://host:port") + printer.noprefix(" socks5://user:pass@host:port") + printer.noprefix("") + url = printer.user_input("Proxy URL : \t").strip() + if url: + proxymanager.add_proxy(url) + + def _remove_proxy(self) -> None: + proxies = proxymanager.list_proxies() + if not proxies: + printer.warning("No proxies to remove.") + return + self._list_proxies() + printer.noprefix("") + raw = printer.user_input("Index to remove : \t").strip() + try: + index = int(raw) + except ValueError: + printer.error("Enter a valid number.") + return + proxymanager.remove_proxy(index) + + def _test_proxies(self) -> None: + printer.noprefix("") + proxymanager.test_all_proxies() + + def _toggle_enabled(self, currently_enabled: bool) -> None: + new_state = not currently_enabled + proxymanager.set_enabled(new_state) + printer.success(f"Proxy routing {'enabled' if new_state else 'disabled'}.") + + def _toggle_rotation(self, currently_rotating: bool) -> None: + new_state = not currently_rotating + proxymanager.set_rotating(new_state) + printer.success(f"Proxy rotation turned {'on' if new_state else 'off'}.") + + def _clear_proxies(self) -> None: + printer.noprefix("") + confirm = printer.user_input("Remove ALL proxies? (y/N) : \t").strip().lower() + if confirm == "y": + proxymanager.clear_proxies() + else: + printer.info("Cancelled.") diff --git a/utils/dirbuster.py b/utils/dirbuster.py index 6535811..e57f335 100644 --- a/utils/dirbuster.py +++ b/utils/dirbuster.py @@ -20,7 +20,7 @@ import aiohttp from colorama import Style -from helper import printer, randomuser, timer, url_helper +from helper import printer, proxymanager, randomuser, timer, url_helper @timer.timer(require_input=True) @@ -68,7 +68,7 @@ async def _fetch_url( url = f"https://{domain}/{path}" headers = {"User-Agent": str(randomuser.GetUser())} try: - async with session.get(url, headers=headers) as response: + async with session.get(url, headers=headers, proxy=proxymanager.get_aiohttp_proxy()) as response: if response.status == 200: url_set.add(url) printer.success( diff --git a/utils/git_scrape.py b/utils/git_scrape.py index 2c8b780..f882e5f 100644 --- a/utils/git_scrape.py +++ b/utils/git_scrape.py @@ -28,7 +28,7 @@ import requests -from helper import printer +from helper import printer, proxymanager _SAVE_DIR = Path("scraped_data") @@ -170,7 +170,7 @@ def scrape_github_user(username: str, token: str, profile: GitHubProfile) -> Non profile_url = f"https://api.github.com/users/{username}" printer.verbose(f"Fetching profile data for: {username}") profile_data = _handle_api_response( - requests.get(profile_url, headers=headers, timeout=10) + requests.get(profile_url, headers=headers, proxies=proxymanager.get_requests_proxies(), timeout=10) ) if not profile_data: @@ -217,7 +217,7 @@ def scrape_github_user(username: str, token: str, profile: GitHubProfile) -> Non repos_url = f"https://api.github.com/users/{username}/repos?type=owner&sort=updated&per_page=5" repos_data = _handle_api_response( - requests.get(repos_url, headers=headers, timeout=10) + requests.get(repos_url, headers=headers, proxies=proxymanager.get_requests_proxies(), timeout=10) ) if not repos_data: @@ -232,7 +232,7 @@ def scrape_github_user(username: str, token: str, profile: GitHubProfile) -> Non f"https://api.github.com/repos/{username}/{repo_name}/commits?per_page=100" ) commits_data = _handle_api_response( - requests.get(commits_url, headers=headers, timeout=10) + requests.get(commits_url, headers=headers, proxies=proxymanager.get_requests_proxies(), timeout=10) ) if not commits_data or not isinstance(commits_data, list): @@ -284,7 +284,7 @@ def scrape_github_repo( f"https://api.github.com/repos/{owner}/{repo_name}/commits?per_page=100" ) commits_data = _handle_api_response( - requests.get(commits_url, headers=headers, timeout=10) + requests.get(commits_url, headers=headers, proxies=proxymanager.get_requests_proxies(), timeout=10) ) if not commits_data or not isinstance(commits_data, list): diff --git a/utils/ig_scrape.py b/utils/ig_scrape.py index 6ef96bd..265b549 100644 --- a/utils/ig_scrape.py +++ b/utils/ig_scrape.py @@ -31,7 +31,7 @@ from ensta import Guest from ensta.lib.Exceptions import APIError, NetworkError, RateLimitedError -from helper import config, printer, timer +from helper import config, printer, proxymanager, timer _KEY_WIDTH = 24 _SAVE_DIR = Path("scraped_data") @@ -498,6 +498,7 @@ def _get_instagram_user_id(username: str, session_id: str) -> dict: params={"username": username}, headers={"User-Agent": _WEB_UA, "x-ig-app-id": _WEB_IG_APP_ID}, cookies={"sessionid": session_id}, + proxies=proxymanager.get_requests_proxies(), timeout=20, ) except requests.RequestException as exc: @@ -554,6 +555,7 @@ def _get_instagram_info( f"https://i.instagram.com/api/v1/users/{user_id}/info/", headers=_mobile_headers(), cookies={"sessionid": session_id}, + proxies=proxymanager.get_requests_proxies(), timeout=20, ) except requests.RequestException: @@ -598,6 +600,7 @@ def _fetch_post_comments( params=params, headers=headers, cookies=_session_cookies(session_id), + proxies=proxymanager.get_requests_proxies(), timeout=20, ) except requests.RequestException as exc: @@ -650,6 +653,7 @@ def _fetch_authenticated_posts( params={"count": limit}, headers=_mobile_headers(), cookies=_session_cookies(session_id), + proxies=proxymanager.get_requests_proxies(), timeout=20, ) except requests.RequestException as exc: @@ -695,6 +699,7 @@ def _fetch_authenticated_stories(user_id: str, session_id: str) -> list[StoryIte params={"reel_ids": user_id}, headers=_mobile_headers(), cookies=_session_cookies(session_id), + proxies=proxymanager.get_requests_proxies(), timeout=20, ) except requests.RequestException as exc: @@ -736,6 +741,7 @@ def _fetch_authenticated_highlights(user_id: str, session_id: str) -> list[Story f"https://i.instagram.com/api/v1/highlights/{user_id}/highlights_tray/", headers=_mobile_headers(), cookies=_session_cookies(session_id), + proxies=proxymanager.get_requests_proxies(), timeout=20, ) except requests.RequestException as exc: @@ -768,6 +774,7 @@ def _fetch_authenticated_highlights(user_id: str, session_id: str) -> list[Story params={"reel_ids": highlight_id}, headers=_mobile_headers(), cookies=_session_cookies(session_id), + proxies=proxymanager.get_requests_proxies(), timeout=20, ) except requests.RequestException: @@ -814,6 +821,7 @@ def _fetch_authenticated_reels( params={"target_user_id": user_id, "page_size": limit}, headers=_mobile_headers(), cookies=_session_cookies(session_id), + proxies=proxymanager.get_requests_proxies(), timeout=20, ) except requests.RequestException: @@ -916,6 +924,9 @@ def _fetch_web_recovery_context() -> tuple[ """Fetch fresh guest browser cookies and tokens for Instagram web GraphQL.""" session = requests.Session() session.headers.update(_web_headers()) + _px = proxymanager.get_requests_proxies() + if _px: + session.proxies.update(_px) attempts: list[dict[str, object]] = [] for url in ( @@ -1207,6 +1218,7 @@ def _instagram_advanced_lookup(username: str, session_id: str = "") -> dict: headers=headers, cookies=cookies, data=data, + proxies=proxymanager.get_requests_proxies(), timeout=20, ) except requests.RequestException as exc: diff --git a/utils/ip_lookup.py b/utils/ip_lookup.py index 5ab8920..b13fb51 100644 --- a/utils/ip_lookup.py +++ b/utils/ip_lookup.py @@ -21,7 +21,7 @@ import requests from colorama import Style -from helper import printer, randomuser, timer +from helper import printer, proxymanager, randomuser, timer # Human-readable labels for the keys returned by ipinfo.io. # Any key not listed here falls back to key.replace("_", " ").title(). @@ -52,7 +52,7 @@ def lookup(ip_address: str) -> None: ip_address = socket.gethostbyname(ip_address) url = f"https://ipinfo.io/{ip_address}/json" headers = {"User-Agent": str(randomuser.GetUser())} - response = requests.get(url, headers=headers) + response = requests.get(url, headers=headers, proxies=proxymanager.get_requests_proxies()) response.raise_for_status() values = response.json() diff --git a/utils/leak_search.py b/utils/leak_search.py index 6621325..729b660 100644 --- a/utils/leak_search.py +++ b/utils/leak_search.py @@ -25,7 +25,7 @@ import requests from colorama import Style -from helper import printer, randomuser, timer +from helper import printer, proxymanager, randomuser, timer _REQUEST_TIMEOUT: int = 20 _MAX_RETRIES: int = 3 @@ -126,6 +126,7 @@ def _get(url: str, params: dict | None = None) -> dict | None: url, params=params, headers=headers, + proxies=proxymanager.get_requests_proxies(), timeout=_REQUEST_TIMEOUT, ) resp.raise_for_status() diff --git a/utils/search_username.py b/utils/search_username.py index ced7472..c2f8d92 100644 --- a/utils/search_username.py +++ b/utils/search_username.py @@ -28,7 +28,7 @@ from colorama import Style -from helper import printer, timer +from helper import printer, proxymanager, timer REPORT_DIR = Path("scraped_data/maigret") MAIGRET_DB_PATH = Path.home() / ".maigret" / "data.json" @@ -458,6 +458,10 @@ def _run_maigret(username: str, config: MaigretConfig) -> dict[str, Any] | None: if config.print_errors: command.append("--print-errors") + proxy = proxymanager.get_proxy() + if proxy: + command.extend(["--proxy", proxy]) + try: result = subprocess.run( command, diff --git a/utils/web_reconnaissance.py b/utils/web_reconnaissance.py index 43d11e8..50fda97 100644 --- a/utils/web_reconnaissance.py +++ b/utils/web_reconnaissance.py @@ -28,7 +28,7 @@ from ddgs import DDGS from ddgs.exceptions import DDGSException, RatelimitException, TimeoutException -from helper import printer, timer +from helper import printer, proxymanager, timer # Suppress verbose third-party warnings and raw info responses from filling the UI logging.getLogger("ddgs").setLevel(logging.ERROR) @@ -466,7 +466,7 @@ def _fetch_results( for backend in backends: for attempt in range(1, _MAX_RETRIES + 1): try: - with DDGS() as ddgs: + with DDGS(proxy=proxymanager.get_proxy()) as ddgs: raw: list[dict] = ( ddgs.text(query, max_results=max_results, backend=backend) or [] ) @@ -530,7 +530,7 @@ def _fetch_results( # Blanket fallback option: let the library attempt its automated resolution default try: - with DDGS() as ddgs: + with DDGS(proxy=proxymanager.get_proxy()) as ddgs: raw = ddgs.text(query, max_results=max_results, backend="auto") or [] return [ SearchResult( diff --git a/utils/web_scrape.py b/utils/web_scrape.py index e1a3771..a4ea82c 100644 --- a/utils/web_scrape.py +++ b/utils/web_scrape.py @@ -28,7 +28,7 @@ from bs4 import BeautifulSoup from colorama import Style -from helper import printer, randomuser, timer +from helper import printer, proxymanager, randomuser, timer REQUEST_TIMEOUT = aiohttp.ClientTimeout(total=15, connect=5, sock_read=10) MAX_CONCURRENT_REQUESTS = 12 @@ -352,7 +352,7 @@ async def _fetch(session: aiohttp.ClientSession, url: str) -> str: :return: Response body as text, or an empty string on error. """ try: - async with session.get(url, allow_redirects=True) as response: + async with session.get(url, proxy=proxymanager.get_aiohttp_proxy(), allow_redirects=True) as response: if response.status >= 400: return ""