Signed-off-by: -LAN- <laipz8200@outlook.com>tags/0.14.2
| class FilenameNotExistsError(HTTPException): | class FilenameNotExistsError(HTTPException): | ||||
| code = 400 | code = 400 | ||||
| description = "The specified filename does not exist." | description = "The specified filename does not exist." | ||||
| class RemoteFileUploadError(HTTPException): | |||||
| code = 400 | |||||
| description = "Error uploading remote file." |
| import services | import services | ||||
| from controllers.common import helpers | from controllers.common import helpers | ||||
| from controllers.common.errors import RemoteFileUploadError | |||||
| from core.file import helpers as file_helpers | from core.file import helpers as file_helpers | ||||
| from core.helper import ssrf_proxy | from core.helper import ssrf_proxy | ||||
| from fields.file_fields import file_fields_with_signed_url, remote_file_info_fields | from fields.file_fields import file_fields_with_signed_url, remote_file_info_fields | ||||
| url = args["url"] | url = args["url"] | ||||
| resp = ssrf_proxy.head(url=url) | |||||
| if resp.status_code != httpx.codes.OK: | |||||
| resp = ssrf_proxy.get(url=url, timeout=3, follow_redirects=True) | |||||
| resp.raise_for_status() | |||||
| try: | |||||
| resp = ssrf_proxy.head(url=url) | |||||
| if resp.status_code != httpx.codes.OK: | |||||
| resp = ssrf_proxy.get(url=url, timeout=3, follow_redirects=True) | |||||
| if resp.status_code != httpx.codes.OK: | |||||
| raise RemoteFileUploadError(f"Failed to fetch file from {url}: {resp.text}") | |||||
| except httpx.RequestError as e: | |||||
| raise RemoteFileUploadError(f"Failed to fetch file from {url}: {str(e)}") | |||||
| file_info = helpers.guess_file_info_from_response(resp) | file_info = helpers.guess_file_info_from_response(resp) | ||||
| import services | import services | ||||
| from controllers.common import helpers | from controllers.common import helpers | ||||
| from controllers.common.errors import RemoteFileUploadError | |||||
| from controllers.web.wraps import WebApiResource | from controllers.web.wraps import WebApiResource | ||||
| from core.file import helpers as file_helpers | from core.file import helpers as file_helpers | ||||
| from core.helper import ssrf_proxy | from core.helper import ssrf_proxy | ||||
| url = args["url"] | url = args["url"] | ||||
| resp = ssrf_proxy.head(url=url) | |||||
| if resp.status_code != httpx.codes.OK: | |||||
| resp = ssrf_proxy.get(url=url, timeout=3) | |||||
| resp.raise_for_status() | |||||
| try: | |||||
| resp = ssrf_proxy.head(url=url) | |||||
| if resp.status_code != httpx.codes.OK: | |||||
| resp = ssrf_proxy.get(url=url, timeout=3, follow_redirects=True) | |||||
| if resp.status_code != httpx.codes.OK: | |||||
| raise RemoteFileUploadError(f"Failed to fetch file from {url}: {resp.text}") | |||||
| except httpx.RequestError as e: | |||||
| raise RemoteFileUploadError(f"Failed to fetch file from {url}: {str(e)}") | |||||
| file_info = helpers.guess_file_info_from_response(resp) | file_info = helpers.guess_file_info_from_response(resp) | ||||