mirror of
https://gitee.com/dify_ai/dify.git
synced 2024-12-04 20:28:12 +08:00
66 lines
2.1 KiB
Python
66 lines
2.1 KiB
Python
"""
|
|
Proxy requests to avoid SSRF
|
|
"""
|
|
|
|
import os
|
|
|
|
from httpx import get as _get
|
|
from httpx import head as _head
|
|
from httpx import options as _options
|
|
from httpx import patch as _patch
|
|
from httpx import post as _post
|
|
from httpx import put as _put
|
|
from requests import delete as _delete
|
|
|
|
SSRF_PROXY_HTTP_URL = os.getenv('SSRF_PROXY_HTTP_URL', '')
|
|
SSRF_PROXY_HTTPS_URL = os.getenv('SSRF_PROXY_HTTPS_URL', '')
|
|
|
|
requests_proxies = {
|
|
'http': SSRF_PROXY_HTTP_URL,
|
|
'https': SSRF_PROXY_HTTPS_URL
|
|
} if SSRF_PROXY_HTTP_URL and SSRF_PROXY_HTTPS_URL else None
|
|
|
|
httpx_proxies = {
|
|
'http://': SSRF_PROXY_HTTP_URL,
|
|
'https://': SSRF_PROXY_HTTPS_URL
|
|
} if SSRF_PROXY_HTTP_URL and SSRF_PROXY_HTTPS_URL else None
|
|
|
|
def get(url, *args, **kwargs):
|
|
return _get(url=url, *args, proxies=httpx_proxies, **kwargs)
|
|
|
|
def post(url, *args, **kwargs):
|
|
return _post(url=url, *args, proxies=httpx_proxies, **kwargs)
|
|
|
|
def put(url, *args, **kwargs):
|
|
return _put(url=url, *args, proxies=httpx_proxies, **kwargs)
|
|
|
|
def patch(url, *args, **kwargs):
|
|
return _patch(url=url, *args, proxies=httpx_proxies, **kwargs)
|
|
|
|
def delete(url, *args, **kwargs):
|
|
if 'follow_redirects' in kwargs:
|
|
if kwargs['follow_redirects']:
|
|
kwargs['allow_redirects'] = kwargs['follow_redirects']
|
|
kwargs.pop('follow_redirects')
|
|
if 'timeout' in kwargs:
|
|
timeout = kwargs['timeout']
|
|
if timeout is None:
|
|
kwargs.pop('timeout')
|
|
elif isinstance(timeout, tuple):
|
|
# check length of tuple
|
|
if len(timeout) == 2:
|
|
kwargs['timeout'] = timeout
|
|
elif len(timeout) == 1:
|
|
kwargs['timeout'] = timeout[0]
|
|
elif len(timeout) > 2:
|
|
kwargs['timeout'] = (timeout[0], timeout[1])
|
|
else:
|
|
kwargs['timeout'] = (timeout, timeout)
|
|
return _delete(url=url, *args, proxies=requests_proxies, **kwargs)
|
|
|
|
def head(url, *args, **kwargs):
|
|
return _head(url=url, *args, proxies=httpx_proxies, **kwargs)
|
|
|
|
def options(url, *args, **kwargs):
|
|
return _options(url=url, *args, proxies=httpx_proxies, **kwargs)
|