dify/api/core/helper/ssrf_proxy.py

48 lines
1.4 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):
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)