chore: webhook with bin file should guess mimetype (#29704)

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
Co-authored-by: Maries <xh001x@hotmail.com>
This commit is contained in:
非法操作
2025-12-16 13:34:27 +08:00
committed by GitHub
parent bdccbb6e86
commit 7695f9151c
2 changed files with 83 additions and 1 deletions

View File

@@ -33,6 +33,11 @@ from services.errors.app import QuotaExceededError
from services.trigger.app_trigger_service import AppTriggerService
from services.workflow.entities import WebhookTriggerData
try:
import magic
except ImportError:
magic = None # type: ignore[assignment]
logger = logging.getLogger(__name__)
@@ -317,7 +322,8 @@ class WebhookService:
try:
file_content = request.get_data()
if file_content:
file_obj = cls._create_file_from_binary(file_content, "application/octet-stream", webhook_trigger)
mimetype = cls._detect_binary_mimetype(file_content)
file_obj = cls._create_file_from_binary(file_content, mimetype, webhook_trigger)
return {"raw": file_obj.to_dict()}, {}
else:
return {"raw": None}, {}
@@ -341,6 +347,18 @@ class WebhookService:
body = {"raw": ""}
return body, {}
@staticmethod
def _detect_binary_mimetype(file_content: bytes) -> str:
"""Guess MIME type for binary payloads using python-magic when available."""
if magic is not None:
try:
detected = magic.from_buffer(file_content[:1024], mime=True)
if detected:
return detected
except Exception:
logger.debug("python-magic detection failed for octet-stream payload")
return "application/octet-stream"
@classmethod
def _process_file_uploads(
cls, files: Mapping[str, FileStorage], webhook_trigger: WorkflowWebhookTrigger

View File

@@ -110,6 +110,70 @@ class TestWebhookServiceUnit:
assert webhook_data["method"] == "POST"
assert webhook_data["body"]["raw"] == "raw text content"
def test_extract_octet_stream_body_uses_detected_mime(self):
"""Octet-stream uploads should rely on detected MIME type."""
app = Flask(__name__)
binary_content = b"plain text data"
with app.test_request_context(
"/webhook", method="POST", headers={"Content-Type": "application/octet-stream"}, data=binary_content
):
webhook_trigger = MagicMock()
mock_file = MagicMock()
mock_file.to_dict.return_value = {"file": "data"}
with (
patch.object(WebhookService, "_detect_binary_mimetype", return_value="text/plain") as mock_detect,
patch.object(WebhookService, "_create_file_from_binary") as mock_create,
):
mock_create.return_value = mock_file
body, files = WebhookService._extract_octet_stream_body(webhook_trigger)
assert body["raw"] == {"file": "data"}
assert files == {}
mock_detect.assert_called_once_with(binary_content)
mock_create.assert_called_once()
args = mock_create.call_args[0]
assert args[0] == binary_content
assert args[1] == "text/plain"
assert args[2] is webhook_trigger
def test_detect_binary_mimetype_uses_magic(self, monkeypatch):
"""python-magic output should be used when available."""
fake_magic = MagicMock()
fake_magic.from_buffer.return_value = "image/png"
monkeypatch.setattr("services.trigger.webhook_service.magic", fake_magic)
result = WebhookService._detect_binary_mimetype(b"binary data")
assert result == "image/png"
fake_magic.from_buffer.assert_called_once()
def test_detect_binary_mimetype_fallback_without_magic(self, monkeypatch):
"""Fallback MIME type should be used when python-magic is unavailable."""
monkeypatch.setattr("services.trigger.webhook_service.magic", None)
result = WebhookService._detect_binary_mimetype(b"binary data")
assert result == "application/octet-stream"
def test_detect_binary_mimetype_handles_magic_exception(self, monkeypatch):
"""Fallback MIME type should be used when python-magic raises an exception."""
try:
import magic as real_magic
except ImportError:
pytest.skip("python-magic is not installed")
fake_magic = MagicMock()
fake_magic.from_buffer.side_effect = real_magic.MagicException("magic error")
monkeypatch.setattr("services.trigger.webhook_service.magic", fake_magic)
with patch("services.trigger.webhook_service.logger") as mock_logger:
result = WebhookService._detect_binary_mimetype(b"binary data")
assert result == "application/octet-stream"
mock_logger.debug.assert_called_once()
def test_extract_webhook_data_invalid_json(self):
"""Test webhook data extraction with invalid JSON."""
app = Flask(__name__)