Support/sp-hydra-veil-core
Support/sp-hydra-veil-core
2
0
Fork 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

Improved DNS evaluation and tools via Tor. We now support for being able to reach IPs that can't resolve through Tor, and evaluating if DNS resolutioin is blocked via Tor

Browse source
This commit is contained in:
SimplifiedPrivacy 2026-07-12 16:47:00 -04:00
parent c6a189c360
commit 882f374937
9 changed files with 330 additions and 114 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
change_log.md
View file

@ -5,5 +5,6 @@ July 11, 2026
Features: Revamp of GET/POST API requests for both Tor and Clearweb.
Why: Move towards more clear Tor API error feedback. And reduce redundancy.
Status: Stable, works for both clear/tor and GET/POST.
Still Needs: UI feedback, DNS Tor improvements
<br/>

5
core/services/networking/api_requests/ApiResponseModel.py
View file

@ -23,7 +23,9 @@ class ApiResponse:
message: Optional[str] = None
retry_now: bool = False
retry_later: bool = False
tor: bool = False
ip_address: str = None
def to_dict(self) -> dict:
"""Convert to dict for backwards compatibility."""
return {
@ -33,6 +35,7 @@ class ApiResponse:
"message": self.message,
"retry_now": self.retry_now,
"retry_later": self.retry_later,
"tor": self.tor,
}
def get(self, key: str, default=None):

10
core/services/networking/api_requests/step1_get_or_post.py
View file

@ -18,6 +18,7 @@ def get_data_from_api(
full_request_url: str,
client_observer: Optional[ClientObserver] = None,
connection_observer: Optional[ConnectionObserver] = None,
clearweb_resolved_ip: str = None,
) -> ApiResponse:
"""
GET wrapper. Desktop passes observer for updates; Android passes None.
@ -39,7 +40,8 @@ def get_data_from_api(
None, # get request has no payload
connection_observer,
client_observer,
timeout=10
timeout=10,
clearweb_resolved_ip=clearweb_resolved_ip
)
else:
# Regular, no tor:
@ -64,6 +66,7 @@ def send_data_to_server(
original_url: str,
connection_observer: Optional[ConnectionObserver] = None,
client_observer: Optional[ClientObserver] = None,
clearweb_resolved_ip: str = None,
) -> ApiResponse:
"""
POST wrapper. Same pattern as GET.
@ -78,7 +81,8 @@ def send_data_to_server(
payload,
connection_observer,
client_observer,
timeout=10
timeout=10,
clearweb_resolved_ip=clearweb_resolved_ip
)
else:
result_object = _execute_regular_request(
@ -94,5 +98,5 @@ def send_data_to_server(
if not result_object.valid and client_observer:
error_msg = result_object.message
client_observer.notify('synchronizing', f"Connection Issue: {error_msg}")
return result_object

35
core/services/networking/api_requests/step2_execute.py
View file

@ -3,13 +3,13 @@ from typing import TYPE_CHECKING
if TYPE_CHECKING:
import requests
from core.Constants import Constants
# Networking Group
from core.services.networking.api_requests.step3_classify_http import classify_http_response
from core.services.networking.api_requests.step4_error_classifier import classify_request_error
from core.services.networking.api_requests.ApiResponseModel import ApiResponse, ErrorType
from core.services.networking.api_requests.subtools.custom_dns import prep_url_with_known_ip, load_custom_dns_resolver
# tor
from essentials.modules.TorModule import TorModule
@ -27,6 +27,8 @@ import json, os
from typing import Any
import time
def _execute_tor_request(
method: str,
url: str,
@ -34,6 +36,7 @@ def _execute_tor_request(
connection_observer: Optional[ConnectionObserver] = None,
client_observer: Optional[ClientObserver] = None,
timeout: int = 6,
clearweb_resolved_ip: str = None # if DNS via Tor is blocked
) -> ApiResponse:
"""
Tor request with optional intermediate UI feedback (desktop-only).
@ -57,12 +60,32 @@ def _execute_tor_request(
proxies = {"http": f"socks5h://127.0.0.1:{port_number}",
"https": f"socks5h://127.0.0.1:{port_number}"}
if method.lower() == "get":
response = requests.get(url, proxies=proxies, timeout=timeout)
# ================= OPTION: DNS OUTSIDE TOR. (IF DNS VIA TOR IS BLOCKED) ===============
if clearweb_resolved_ip:
CustomHostnameAdapter = load_custom_dns_resolver()
url_with_ip, original_hostname = prep_url_with_known_ip(clearweb_resolved_ip, url)
# Set up session with custom adapter for DNS resolution:
session = requests.Session()
session.mount('https://', CustomHostnameAdapter(original_hostname))
session.headers.update({"Host": original_hostname})
# Make the request to the raw IP, but tell the server it's the original domain
if method.lower() == "get":
response = session.get(url_with_ip, proxies=proxies, timeout=timeout, verify=True)
else:
response = requests.post(url_with_ip, json=payload, proxies=proxies, timeout=timeout, verify=True)
else:
response = requests.post(url, json=payload, proxies=proxies, timeout=timeout)
# ================= USING DNS VIA TOR (normal-use) ===============
if method.lower() == "get":
response = requests.get(url, proxies=proxies, timeout=timeout)
else:
response = requests.post(url, json=payload, proxies=proxies, timeout=timeout)
# ================= Working Response ===============
result = classify_http_response(response)
except Exception as e:

41
core/services/networking/api_requests/step4_error_classifier.py
View file

@ -1,9 +1,3 @@
# from __future__ import annotations
# from typing import TYPE_CHECKING
# if TYPE_CHECKING:
# from essentials.observers.ConnectionObserver import ConnectionObserver
from core.observers.ClientObserver import ClientObserver
from core.observers.ConnectionObserver import ConnectionObserver
@ -12,7 +6,7 @@ from core.services.networking.api_requests.ApiResponseModel import ApiResponse,
# diagnosis services
from core.services.networking.api_requests.subtools.internet_test import do_we_have_internet
from core.services.networking.api_requests.subtools.is_dns_problem import is_dns_problem, is_dns_problem_via_tor
from core.services.networking.api_requests.subtools.dns_evaluation import is_dns_problem, dns_works_via_tor
from core.services.networking.api_requests.subtools.extract_domain import extract_domain
from core.services.networking.api_requests.subtools.is_tor_working import is_tor_working
@ -31,8 +25,7 @@ def classify_request_error(
client_observer: Optional[ClientObserver] = None,
) -> ApiResponse:
"""
Unified diagnostic tree for all request failures.
Replaces evaluate_tor_networking_problem() and evaluate_regular_networking_problem().
Classify all error requests, Tor or Not.
"""
domain_only = extract_domain(url)
@ -43,23 +36,36 @@ def classify_request_error(
return ApiResponse(
valid=False,
error_type=ErrorType.TOR_NOT_WORKING,
tor=True,
message="Tor connection failed"
)
logger.debug("Skipping DNS resolution via Tor due to PySocks support.")
# if is_dns_problem_via_tor(domain_only, connection_observer):
# logger.debug("DNS resolution via Tor failed")
# return ApiResponse(
# valid=False,
# error_type=ErrorType.DNS_RESOLUTION,
# message=f"Cannot resolve {domain_only} via Tor"
# )
ip_via_tor = dns_works_via_tor(domain_only, connection_observer)
if not ip_via_tor:
logger.debug("DNS resolution via Tor failed")
return ApiResponse(
valid=False,
error_type=ErrorType.DNS_RESOLUTION,
tor=True,
message=f"Cannot resolve {domain_only} via Tor"
)
else:
return ApiResponse(
valid=False,
error_type=ErrorType.UNKNOWN,
tor=True,
ip_address=ip_via_tor,
message=f"For Unknown reasons, we can't connect, but we can resolve {domain_only} to {ip_via_tor} via Tor"
)
else:
# Regular (non-Tor) diagnostics
if not do_we_have_internet():
logger.error("No internet connection")
return ApiResponse(
valid=False,
tor=False,
error_type=ErrorType.NO_INTERNET,
message="No internet connectivity"
)
@ -68,6 +74,7 @@ def classify_request_error(
logger.error("Local DNS resolution failed")
return ApiResponse(
valid=False,
tor=False,
error_type=ErrorType.DNS_RESOLUTION,
message=f"Cannot resolve {domain_only}"
)

108
core/services/networking/api_requests/subtools/custom_dns.py Normal file
View file

@ -0,0 +1,108 @@
from __future__ import annotations
from typing import TYPE_CHECKING
if TYPE_CHECKING:
import requests
# errors
from core.errors.exceptions import *
from core.errors.logger import logger
from urllib.parse import urlparse, urlunparse
def prep_url_with_known_ip(known_ip: str, original_url: str):
"""
Purpose:
Prepares URLs for custom DNS checking
Rank:
Helper
Called by:
step2_execute's _execute_tor_request
Raises Errors?
False
"""
# Parse the original URL to extract the hostname
parsed = urlparse(original_url)
original_hostname = parsed.hostname
# Reconstruct the URL with the IP instead of the domain
url_with_ip = original_url.replace(parsed.hostname, known_ip)
return url_with_ip, original_hostname
def load_custom_dns_resolver():
from requests.adapters import HTTPAdapter
from urllib3.util.ssl_ import create_urllib3_context
import ssl
import certifi
class CustomHostnameAdapter(HTTPAdapter):
"""
Purpose:
If Tor (or ANY) DNS server is blocked, we can go directly to the IP,
and use this module to verify the SSL against local certs.
Method:
Adapter that connects to an IP but verifies the SSL cert against the original hostname (via SNI).
then uses that with the proxy, essentially intercepting/modifying the proxy behavior
Rank:
Coordinator for the SSL verification Task,
But the caller is orchestrating the actual API calls.
Called by:
step2_execute's _execute_tor_request
Raises Errors?
False
"""
def __init__(self, original_hostname, *args, **kwargs):
logger.debug(f"\n[ADAPTER INIT] Creating CustomHostnameAdapter")
logger.debug(f" original_hostname: {original_hostname}")
self.original_hostname = original_hostname
super().__init__(*args, **kwargs)
logger.debug(f"[ADAPTER INIT] ✓ Adapter initialized")
def init_poolmanager(self, *args, **kwargs):
logger.debug(f"\n[POOLMANAGER INIT] Creating standard pool manager (for direct connections)")
ctx = create_urllib3_context()
ctx.check_hostname = True
ctx.verify_mode = ssl.CERT_REQUIRED
ctx.load_verify_locations(certifi.where())
kwargs['ssl_context'] = ctx
logger.debug(f" ✓ Standard pool manager configured")
return super().init_poolmanager(*args, **kwargs)
def proxy_manager_for(self, proxy, **proxy_kwargs):
"""
This is called when making requests through a proxy.
This is where we inject the custom SSL context.
"""
logger.debug(f"\n[PROXY MANAGER] Creating proxy manager")
logger.debug(f" proxy: {proxy}")
logger.debug(f" original_hostname (for verification): {self.original_hostname}")
ctx = create_urllib3_context()
ctx.check_hostname = True
ctx.verify_mode = ssl.CERT_REQUIRED
ctx.load_verify_locations(certifi.where())
# KEY: Set server_hostname on the context for SNI
logger.debug(f" Setting server_hostname for SNI: {self.original_hostname}")
proxy_kwargs['ssl_context'] = ctx
proxy_kwargs['server_hostname'] = self.original_hostname
logger.debug(f" ✓ Proxy manager will verify cert against: {self.original_hostname}")
result = super().proxy_manager_for(proxy, **proxy_kwargs)
logger.debug(f" ✓ Proxy manager created successfully")
return result
return CustomHostnameAdapter

115
core/services/networking/api_requests/subtools/dns_evaluation.py Normal file
View file

@ -0,0 +1,115 @@
# for tor:
from essentials.modules.TorModule import TorModule
from essentials.observers.ConnectionObserver import ConnectionObserver
from essentials.services.ConnectionService import ConnectionService
from core.observers.ClientObserver import ClientObserver
from core.Constants import Constants
from core.services.networking.api_requests.subtools.extract_domain import extract_domain
from core.errors.logger import logger
from core.errors.exceptions import *
import dns.resolver
import dns.exception
import dns.rdatatype
from typing import Optional
import os
import socket
import socks
import subprocess
"""
Check if there's a DNS problem with a domain.
Returns True if DNS fails (problem exists), False if DNS resolves successfully.
"""
def is_dns_problem(url: str) -> bool:
"""Regular DNS check for non-Tor domains"""
domain = extract_domain(url)
try:
socket.gethostbyname(domain)
return False # DNS resolved successfully
except socket.gaierror:
return True # DNS resolution failed
except Exception:
return True
def dns_works_via_tor(domain, connection_observer, client_observer=None):
"""
Check if domain's resolver is blocking Tor by querying via proxychains.
Dynamically configures proxychains with the given port.
"""
port_number = ConnectionService.get_random_available_port_number()
logger.debug(f"[DNS PROBLEM OF TOR SERVICE] Using Tor on port number {port_number}")
if client_observer is not None:
client_observer.notify('synchronizing', f"Using Tor on Port {port_number}..")
try:
tor_module = TorModule(Constants.HV_TOR_STATE_HOME)
tor_module.create_session(port_number, connection_observer)
logger.debug(f"[DNS PROBLEM OF TOR SERVICE] Tor Started a Session")
if client_observer is not None:
client_observer.notify('synchronizing', f"Initialized Tor..")
except TorServiceInitializationError as e:
error_msg = f"[DNS PROBLEM OF TOR SERVICE] Tor failed to start: {e}"
tor_module.destroy_session(port_number)
logger.error(error_msg, exc_info=True)
if client_observer is not None:
client_observer.notify('synchronizing', f"Tor Failed to Initialize")
proxies = {
"http": f"socks5h://127.0.0.1:{port_number}",
"https": f"socks5h://127.0.0.1:{port_number}",
}
config_path = f"{Constants.HV_CONFIG_HOME}/proxychains_dynamic.conf"
try:
# Write proxychains config on the fly
os.makedirs(os.path.dirname(config_path), exist_ok=True)
with open(config_path, 'w') as f:
f.write(f"""
strict_chain
proxy_dns
tcp_read_time_out 10000
tcp_connect_time_out 10000
[ProxyList]
socks5 127.0.0.1 {port_number}
""")
f.flush()
os.fsync(f.fileno()) # Force write to disk
print("running command...")
# Run dig through proxychains with the dynamic config
result = subprocess.run(
['proxychains4', '-f', config_path, 'dig', domain, '+short'],
capture_output=True,
text=True,
timeout=10
)
print(f"result: {result}")
output = result.stdout.strip()
print(f"output: {output}")
if not output or result.returncode != 0:
logger.debug(f"[DNS PROBLEM] {domain}: No response")
return False
logger.debug(f"[DNS OK] {domain}: {output}")
return output
except subprocess.TimeoutExpired:
logger.debug(f"[DNS PROBLEM] {domain}: Timeout")
return False
except Exception as e:
logger.debug(f"[DNS PROBLEM] {domain}: {type(e).__name__}: {e}")
return False
finally:
tor_module.destroy_session(port_number)

87
core/services/networking/api_requests/subtools/is_dns_problem.py
View file

@ -1,87 +0,0 @@
# for tor:
from essentials.modules.TorModule import TorModule
from essentials.observers.ConnectionObserver import ConnectionObserver
from essentials.services.ConnectionService import ConnectionService
from core.observers.ClientObserver import ClientObserver
from core.Constants import Constants
import json, os
from typing import Optional
from core.services.networking.api_requests.subtools.extract_domain import extract_domain
from core.errors.logger import logger
import socket
import socks
"""
Check if there's a DNS problem with a domain.
Returns True if DNS fails (problem exists), False if DNS resolves successfully.
"""
def is_dns_problem(url: str) -> bool:
domain = extract_domain(url)
try:
socket.gethostbyname(domain)
return False # DNS resolved successfully, no problem
except socket.gaierror:
return True # DNS resolution failed, problem exists
except Exception:
return True # Any other error is treated as a DNS problem
def is_dns_problem_via_tor(
domain: str,
connection_observer: ConnectionObserver | None = None,
client_observer: Optional[ClientObserver] = None,
) -> bool:
logger.debug("[DNS PROBLEM OF TOR SERVICE] We've triggered EVALUATING IF it's a DNS problem via Tor")
port_number = ConnectionService.get_random_available_port_number()
logger.debug(f"[DNS PROBLEM OF TOR SERVICE] Using Tor on port number {port_number}")
if client_observer is not None:
client_observer.notify('synchronizing', f"Using Tor on Port {port_number}..")
try:
tor_module = TorModule(Constants.HV_TOR_STATE_HOME)
tor_module.create_session(port_number, connection_observer)
logger.debug(f"[DNS PROBLEM OF TOR SERVICE] Tor Started a Session")
if client_observer is not None:
client_observer.notify('synchronizing', f"Initialized Tor..")
except TorServiceInitializationError as e:
error_msg = f"[DNS PROBLEM OF TOR SERVICE] Tor failed to start: {e}"
tor_module.destroy_session(port_number)
logger.error(error_msg, exc_info=True)
if client_observer is not None:
client_observer.notify('synchronizing', f"Tor Failed to Initialize")
# proxies = {
# "http": f"socks5h://127.0.0.1:{port_number}",
# "https": f"socks5h://127.0.0.1:{port_number}",
# }
logger.debug(f"[DNS PROBLEM OF TOR SERVICE] About to try to apply the proxy but may get a PySocks error...")
try:
# Set up SOCKS5 proxy using the same port as the existing Tor proxy,
socks.set_default_proxy(socks.SOCKS5, "127.0.0.1", port_number)
socket.socket = socks.socksocket
except:
logger.error(f"[DNS PROBLEM OF TOR SERVICE] Error with setting up the socks5 proxy using socks.set_deault0")
tor_module.destroy_session(port_number)
return True
# DNS query check
try:
logger.debug("checking dns via socket..")
socket.gethostbyname(domain)
tor_module.destroy_session(port_number)
return False # DNS resolved successfully, no problem
except socket.gaierror:
return True # DNS resolution failed, problem exists
except Exception:
return True # Any other error is treated as a DNS problem

42
core/test_dns.py Normal file
View file

@ -0,0 +1,42 @@
print("hi")
import httpx
import ssl
import certifi
def load_custom_dns_resolver():
"""Create a client with custom transports for Tor DNS bypass."""
def create_client_with_custom_dns(hostname_to_ip_map: dict):
"""
hostname_to_ip_map: {"example.com": "203.0.113.45", "other.com": "203.0.113.46"}
"""
mounts = {}
for hostname, ip in hostname_to_ip_map.items():
ctx = ssl.create_default_context(cafile=certifi.where())
ctx.check_hostname = True
ctx.verify_mode = ssl.CERT_REQUIRED
# transport = httpx.HTTPTransport(ssl_context=ctx, http2=True)
transport = httpx.HTTPTransport(ssl_context=context)
# Mount for both hostname and IP patterns
mounts[f"https://{hostname}"] = transport
mounts[f"https://{ip}"] = transport
# client = httpx.Client(mounts=mounts, http2=True)
client = httpx.Client(mounts=mounts)
return client
return create_client_with_custom_dns
# Usage
create_client = load_custom_dns_resolver()
client = create_client({"api.hydraveil.net": "185.165.169.91"})
# Connect to IP, but Host header and cert verification use original hostname
response = client.get("https://185.165.169.91/api/v1/cachedsync", headers={"Host": "api.hydraveil.net"})
print(response)
print(response.json)