File size: 3,037 Bytes
a569fc1 df73137 a569fc1 c7c4a89 df73137 503d4ac df73137 c7c4a89 df73137 c7c4a89 b249057 df73137 c7c4a89 df73137 503d4ac df73137 6c9f45a df73137 c7c4a89 df73137 a569fc1 df73137 a569fc1 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 |
import logging
import socket
import json
import requests
from requests.adapters import HTTPAdapter
from requests.packages.urllib3.util.retry import Retry
logger = logging.getLogger(__name__)
# Cache for resolved IPs to avoid spamming DoH
_DNS_CACHE = {}
_ORIGINAL_GETADDRINFO = socket.getaddrinfo
def resolve_via_doh(hostname):
"""
Resolve hostname using Google DNS-over-HTTPS.
Bypasses local system resolver (/etc/resolv.conf).
"""
if hostname in _DNS_CACHE:
return _DNS_CACHE[hostname]
try:
# We must use an IP for the DoH server to avoid DNS loop!
# 8.8.8.8 is Google DNS.
doh_url = f"https://8.8.8.8/resolve?name={hostname}&type=A"
# Use a fresh simple session without adapters to avoid recursion
# But wait, requests might try to resolve 8.8.8.8? No, it's an IP.
resp = requests.get(doh_url, timeout=5, verify=False) # Skip verify for DoH to be fast/safe
data = resp.json()
if 'Answer' in data:
for answer in data['Answer']:
if answer['type'] == 1: # A Record
ip = answer['data']
logger.debug(f"🔍 DoH Resolved {hostname} -> {ip}")
_DNS_CACHE[hostname] = ip
return ip
except Exception as e:
logger.error(f"⚠️ DoH Resolution failed for {hostname}: {e}")
return None
def patched_getaddrinfo(host, port, family=0, type=0, proto=0, flags=0):
"""
Custom getaddrinfo that attempts DoH if system resolution fails or for specific hosts.
"""
# Only intervene for Facebook/Instagram APIs which are failing in this environment
if host in ('graph.facebook.com', 'www.facebook.com', 'api.instagram.com', 'graph.instagram.com'):
ip = resolve_via_doh(host)
if ip:
# Call original with IP instead of hostname
# Log usage for debugging
# logger.debug(f"Redirecting {host} to {ip}")
return _ORIGINAL_GETADDRINFO(ip, port, family, type, proto, flags)
# Fallback to system resolver
return _ORIGINAL_GETADDRINFO(host, port, family, type, proto, flags)
def force_ipv4_requests():
"""
Installs the DNS patch.
Named 'force_ipv4_requests' for backward compatibility with existing calls.
"""
if socket.getaddrinfo != patched_getaddrinfo:
socket.getaddrinfo = patched_getaddrinfo
logger.debug("🛡️ Installed DoH-based DNS patch for Facebook API")
else:
logger.debug("🛡️ DNS patch already active")
def get_resilient_session():
"""
Returns a session with retry logic.
"""
session = requests.Session()
retry = Retry(total=3, backoff_factor=1, status_forcelist=[500, 502, 503, 504])
adapter = HTTPAdapter(max_retries=retry)
session.mount('https://', adapter)
session.mount('http://', adapter)
# Ensure patch is applied whenever we get a session
force_ipv4_requests()
return session
|