Leon4gr45 commited on
Commit
24d5c2d
·
verified ·
1 Parent(s): baf4253

Delete vrag/egress.py with huggingface_hub

Browse files
Files changed (1) hide show
  1. vrag/egress.py +0 -52
vrag/egress.py DELETED
@@ -1,52 +0,0 @@
1
- """Resolve public destinations at connect time, preserving TLS hostname checks.
2
-
3
- Direct HTTPS connects to a validated literal IP (Host and TLS SNI stay original).
4
- A configured environment proxy remains the sole outbound path; its administrator
5
- must also enforce public-only destination resolution inside that proxy.
6
- """
7
- import asyncio
8
- import ipaddress
9
- import socket
10
- from urllib.request import getproxies, proxy_bypass
11
-
12
- import httpx
13
-
14
- async def public_addresses(host: str, timeout: float = 5):
15
- records = await asyncio.wait_for(
16
- asyncio.to_thread(socket.getaddrinfo, host, 443, type=socket.SOCK_STREAM), timeout)
17
- addresses = list(dict.fromkeys(record[4][0] for record in records))
18
- if not addresses or any(not ipaddress.ip_address(address).is_global for address in addresses):
19
- raise ValueError('non-public destination')
20
- return addresses
21
-
22
- class PublicAddressTransport(httpx.AsyncBaseTransport):
23
- def __init__(self, inner=None):
24
- self.inner = inner or httpx.AsyncHTTPTransport(trust_env=True)
25
-
26
- async def handle_async_request(self, request):
27
- host = request.url.host
28
- try:
29
- addresses = await public_addresses(host)
30
- except TimeoutError:
31
- raise httpx.ConnectTimeout('destination lookup timed out') from None
32
- except (OSError, ValueError):
33
- raise httpx.ConnectError('destination denied') from None
34
- request.headers['Host'] = host
35
- request.extensions['sni_hostname'] = host
36
- request.url = request.url.copy_with(host=addresses[0])
37
- return await self.inner.handle_async_request(request)
38
-
39
- async def aclose(self):
40
- await self.inner.aclose()
41
-
42
- def proxy_for(base: str):
43
- host = httpx.URL(base).host
44
- proxies = getproxies()
45
- return None if proxy_bypass(host) else proxies.get('https', proxies.get('all'))
46
-
47
- def transport_for(base: str):
48
- proxy = proxy_for(base)
49
- if proxy:
50
- # Keep injected authentication and the trusted platform egress path.
51
- return httpx.AsyncHTTPTransport(proxy=proxy, trust_env=True)
52
- return PublicAddressTransport()