diff --git a/PR_DESCRIPTION_9446.md b/PR_DESCRIPTION_9446.md new file mode 100644 index 0000000000..795e71dfc9 --- /dev/null +++ b/PR_DESCRIPTION_9446.md @@ -0,0 +1,109 @@ +# Issue #9446 - SSL Verification Downgrade Vulnerability Fix + +## 🔐 Security Fix + +This PR fixes a critical security vulnerability where `download_image_by_url()` and `download_file()` +would silently downgrade SSL certificate verification to `CERT_NONE` on SSL errors, allowing +man-in-the-middle (MITM) attacks. + +### Vulnerability Details + +**Before Fix:** +- When SSL certificate verification failed, the code would catch the exception and retry with `ssl.CERT_NONE` +- This fallback behavior was automatic and silent (only logged a warning) +- Attackers could exploit this by triggering certificate errors to force insecure connections + +**After Fix:** +- SSL certificate verification is always enforced +- SSL errors raise exceptions immediately without fallback +- No automatic downgrade to insecure connections + +## 🔧 Changes + +### Modified Functions + +#### `download_image_by_url()` (lines 116-144) +- **Removed:** SSL fallback try-except block that would retry with `CERT_NONE` +- **Behavior:** SSL certificate errors now raise immediately instead of falling back to insecure mode +- **Impact:** Functions calling this must handle SSL exceptions explicitly + +#### `download_file()` (lines 250-291) +- **Removed:** SSL fallback logic +- **Changed:** Default parameter value `allow_insecure_ssl_fallback=False` (was `True`) +- **Updated:** Parameter documentation to indicate it's deprecated and ignored +- **Behavior:** Always enforces SSL verification regardless of parameter value + +#### `download_dashboard()` (lines 452-549) +- **Updated:** Parameter documentation for `allow_insecure_ssl_fallback` +- **Documentation:** Now indicates the parameter is deprecated and ignored +- **Note:** Parameter retained for backward compatibility but has no effect + +### Breaking Change ⚠️ + +**Parameter default changed:** `download_file(allow_insecure_ssl_fallback=False)` (was `True`) + +**Impact:** Code that relied on automatic SSL verification downgrade will now raise exceptions +for certificate errors. This is intentional security hardening. + +**Migration Guide:** +- If you encounter SSL certificate errors, do NOT disable verification +- For self-signed certificates: Add them to your system's CA certificate store +- For development/testing: Use proper certificate configuration instead of bypassing verification +- The `allow_insecure_ssl_fallback` parameter is now deprecated and ignored + +## ✅ Testing + +### Test Coverage +- **13 new SSL verification tests** in `tests/unit/test_io_ssl_verification.py` +- **SSL error handling tests** in `tests/unit/test_io_download_file.py` +- All existing download tests updated and passing + +### Test Results +- ✅ SSL errors raise immediately without retry +- ✅ HTTPS downloads with valid certificates work correctly +- ✅ HTTP (non-encrypted) URLs continue to work +- ✅ Both `download_image_by_url()` and `download_file()` enforce SSL verification +- ✅ Deprecated parameter is properly ignored + +## 📝 Documentation Updates + +### Updated Documentation +1. **`download_file()`** - Parameter documentation updated to indicate deprecation +2. **`download_dashboard()`** - Parameter documentation updated to indicate deprecation +3. **Test files** - Include references to Issue #9446 with detailed explanations + +### Parameter Documentation +```python +allow_insecure_ssl_fallback: Deprecated parameter, ignored. SSL certificate + verification is always enforced for security. +``` + +## 🔍 Security Considerations + +### Why This Fix Matters +1. **Prevents MITM Attacks:** Ensures all HTTPS connections verify server identity +2. **No Silent Failures:** SSL errors are visible and must be handled explicitly +3. **Defense in Depth:** Removes automatic fallback to insecure mode +4. **Compliance:** Aligns with security best practices and standards + +### Affected Code Paths +- Dashboard downloads from official repositories +- Plugin downloads +- Image downloads from external URLs +- Any code using `download_file()` or `download_image_by_url()` + +## 📋 Checklist + +- [x] Security vulnerability fixed +- [x] Breaking change documented +- [x] All tests passing (13 new SSL tests added) +- [x] Documentation updated for affected functions +- [x] Migration guide provided +- [x] No new dependencies introduced +- [x] Code reviewed for security implications + +## 🔗 Related + +- Issue: #9446 +- Related commits: `218e88755` (earlier SSL error handling work) +- Security: High priority - MITM vulnerability diff --git a/astrbot/core/utils/io.py b/astrbot/core/utils/io.py index 41cc87639a..45d541989f 100644 --- a/astrbot/core/utils/io.py +++ b/astrbot/core/utils/io.py @@ -118,16 +118,31 @@ async def download_image_by_url( post: bool = False, post_data: dict | None = None, path: str | None = None, + timeout: float = 30.0, ) -> str: - """下载图片, 返回 path""" + """下载图片, 返回 path + + Args: + url: 图片 URL + post: 是否使用 POST 请求 + post_data: POST 请求数据 + path: 保存路径, 如果为 None 则保存到临时文件 + timeout: 下载超时时间(秒)，默认 30 秒 + + Returns: + 图片文件路径 + """ + ssl_context = ssl.create_default_context( + cafile=certifi.where(), + ) # 使用 certifi 提供的 CA 证书 + connector = aiohttp.TCPConnector(ssl=ssl_context) # 使用 certifi 的根证书 + client_timeout = aiohttp.ClientTimeout(total=timeout) + try: - ssl_context = ssl.create_default_context( - cafile=certifi.where(), - ) # 使用 certifi 提供的 CA 证书 - connector = aiohttp.TCPConnector(ssl=ssl_context) # 使用 certifi 的根证书 async with aiohttp.ClientSession( trust_env=True, connector=connector, + timeout=client_timeout, ) as session: if post: async with session.post(url, json=post_data) as resp: @@ -143,34 +158,18 @@ async def download_image_by_url( with open(path, "wb") as f: f.write(await resp.read()) return path - except (aiohttp.ClientConnectorSSLError, aiohttp.ClientConnectorCertificateError): - # 关闭SSL验证（仅在证书验证失败时作为fallback） - logger.warning( + except aiohttp.ClientConnectorSSLError as e: + logger.error( f"SSL certificate verification failed for {_safe_url_for_log(url)}. " - "Disabling SSL verification (CERT_NONE) as a fallback. " - "This is insecure and exposes the application to man-in-the-middle attacks. " - "Please investigate and resolve certificate issues." + "Please check the server's SSL certificate configuration." ) - ssl_context = ssl.create_default_context() - ssl_context.check_hostname = False - ssl_context.verify_mode = ssl.CERT_NONE - async with aiohttp.ClientSession() as session: - if post: - async with session.post(url, json=post_data, ssl=ssl_context) as resp: - if not path: - return save_temp_img(await resp.read()) - with open(path, "wb") as f: - f.write(await resp.read()) - return path - else: - async with session.get(url, ssl=ssl_context) as resp: - if not path: - return save_temp_img(await resp.read()) - with open(path, "wb") as f: - f.write(await resp.read()) - return path - except Exception as e: - raise e + raise + except aiohttp.ClientConnectorCertificateError as e: + logger.error( + f"SSL certificate error for {_safe_url_for_log(url)}. " + "Certificate may be expired or invalid." + ) + raise async def _emit_download_progress(progress_callback, payload: dict) -> None: @@ -281,7 +280,7 @@ async def download_file( path: str, show_progress: bool = False, progress_callback=None, - allow_insecure_ssl_fallback: bool = True, + allow_insecure_ssl_fallback: bool = False, ) -> None: """Download a remote file to a local path. @@ -290,18 +289,25 @@ async def download_file( path: Local destination path. show_progress: Whether to print progress to stdout. progress_callback: Optional callback for progress payloads. - allow_insecure_ssl_fallback: Whether certificate failures may retry with - TLS certificate verification disabled. + allow_insecure_ssl_fallback: Deprecated parameter, ignored. SSL verification + is always enforced. Returns: None. """ + if allow_insecure_ssl_fallback is not False: + logger.warning( + "Parameter 'allow_insecure_ssl_fallback' is deprecated and ignored. " + "SSL certificate verification is always enforced for security. " + "Please remove this parameter from your code." + ) + + ssl_context = ssl.create_default_context( + cafile=certifi.where(), + ) # 使用 certifi 提供的 CA 证书 + connector = aiohttp.TCPConnector(ssl=ssl_context) try: - ssl_context = ssl.create_default_context( - cafile=certifi.where(), - ) # 使用 certifi 提供的 CA 证书 - connector = aiohttp.TCPConnector(ssl=ssl_context) async with aiohttp.ClientSession( trust_env=True, connector=connector, @@ -316,35 +322,19 @@ async def download_file( show_progress, progress_callback, ) - except (aiohttp.ClientConnectorSSLError, aiohttp.ClientConnectorCertificateError): - if not allow_insecure_ssl_fallback: - raise - # 关闭SSL验证（仅在证书验证失败时作为fallback） - logger.warning( + except aiohttp.ClientConnectorSSLError as e: + logger.error( f"SSL certificate verification failed for {_safe_url_for_log(url)}. " - "Falling back to unverified connection (CERT_NONE). " + "Please check the server's SSL certificate configuration." ) - logger.warning( - f"SSL certificate verification failed for {_safe_url_for_log(url)}. " - "Falling back to unverified connection (CERT_NONE). " - "This is insecure and exposes the application to man-in-the-middle attacks. " - "Please investigate certificate issues with the remote server." + raise + except aiohttp.ClientConnectorCertificateError as e: + logger.error( + f"SSL certificate error for {_safe_url_for_log(url)}. " + "Certificate may be expired or invalid." ) - ssl_context = ssl.create_default_context() - ssl_context.check_hostname = False - ssl_context.verify_mode = ssl.CERT_NONE - async with aiohttp.ClientSession() as session: - async with session.get(url, ssl=ssl_context, timeout=120) as resp: - _raise_for_download_status(resp, url) - with open(path, "wb") as f: - await _download_response_to_file( - resp, - f, - url, - show_progress, - progress_callback, - show_downloading_label=False, - ) + raise + if show_progress: print() @@ -528,12 +518,19 @@ async def download_dashboard( proxy: Optional download proxy prefix. progress_callback: Optional callback for download progress payloads. extract: Whether to extract the archive after download. - allow_insecure_ssl_fallback: Whether certificate failures may retry with - TLS certificate verification disabled. + allow_insecure_ssl_fallback: Deprecated parameter, ignored. SSL certificate + verification is always enforced for security. Returns: None. """ + if allow_insecure_ssl_fallback is not True: + logger.warning( + "Parameter 'allow_insecure_ssl_fallback' is deprecated and ignored. " + "SSL certificate verification is always enforced for security. " + "Please remove this parameter from your code." + ) + if path is None: zip_path = Path(get_astrbot_data_path()).absolute() / "dashboard.zip" else: diff --git a/tests/unit/test_io_download_file.py b/tests/unit/test_io_download_file.py index 5cea2f0e50..971e44fe65 100644 --- a/tests/unit/test_io_download_file.py +++ b/tests/unit/test_io_download_file.py @@ -70,25 +70,27 @@ async def test_download_file_rejects_non_200_response(monkeypatch, tmp_path): @pytest.mark.asyncio -async def test_download_file_rejects_non_200_response_after_ssl_fallback( +async def test_download_file_raises_ssl_error_immediately( monkeypatch, tmp_path, ): + """Test that SSL errors are raised immediately without fallback retry. + + This test validates the fix for Issue #9446 - SSL verification should not + be downgraded when certificate validation fails. + """ class FakeSSLError(Exception): pass target_path = tmp_path / "missing.bin" - _patch_download_sessions( + _patch_download_session( monkeypatch, - [ - FakeSSLError(), - _FakeResponse(status=404, chunks=[b"not found"]), - ], + FakeSSLError(), ) monkeypatch.setattr(io.aiohttp, "ClientConnectorSSLError", FakeSSLError) monkeypatch.setattr(io.aiohttp, "ClientConnectorCertificateError", FakeSSLError) - with pytest.raises(io.DownloadFileHTTPError, match="HTTP status code: 404"): + with pytest.raises(FakeSSLError): await io.download_file("https://example.test/missing", str(target_path)) assert not target_path.exists() diff --git a/tests/unit/test_io_ssl_verification.py b/tests/unit/test_io_ssl_verification.py new file mode 100644 index 0000000000..9239db9aa9 --- /dev/null +++ b/tests/unit/test_io_ssl_verification.py @@ -0,0 +1,391 @@ +"""Tests for SSL verification in download functions. + +This test suite ensures that SSL certificate verification is properly enforced +and that the application does not silently downgrade to insecure connections +when SSL errors occur, which would expose users to MITM attacks. + +Related to Issue #9446 - SSL verification downgrade vulnerability. + +IMPORTANT: These tests are designed to FAIL on the current vulnerable code +and PASS after the vulnerability is fixed. The failing tests demonstrate that: +1. SSL errors trigger insecure fallback (CERT_NONE) - VULNERABLE BEHAVIOR +2. After the fix, SSL errors should raise exceptions immediately - SECURE BEHAVIOR + +Expected test results: +- BEFORE FIX: Some tests will fail, showing the vulnerability exists +- AFTER FIX: All tests should pass, confirming SSL errors raise exceptions +""" + +import ssl +import tempfile +from pathlib import Path +from unittest.mock import AsyncMock, MagicMock, patch + +import aiohttp +import pytest + +from astrbot.core.utils.io import download_file, download_image_by_url + + +class TestDownloadImageSSLVerification: + """Test SSL verification behavior in download_image_by_url function.""" + + @pytest.mark.asyncio + async def test_download_image_https_success(self): + """Test that HTTPS downloads work correctly with valid certificates.""" + mock_response = AsyncMock() + mock_response.read = AsyncMock(return_value=b"fake_image_data") + mock_response.__aenter__ = AsyncMock(return_value=mock_response) + mock_response.__aexit__ = AsyncMock(return_value=None) + + mock_session = MagicMock() + mock_session.get = MagicMock(return_value=mock_response) + mock_session.__aenter__ = AsyncMock(return_value=mock_session) + mock_session.__aexit__ = AsyncMock(return_value=None) + + with patch("aiohttp.ClientSession", return_value=mock_session): + result = await download_image_by_url("https://example.com/image.jpg") + + assert result is not None + assert Path(result).exists() + # Verify that SSL context was created properly + mock_session.get.assert_called_once_with("https://example.com/image.jpg") + # Clean up temp file + Path(result).unlink(missing_ok=True) + + @pytest.mark.asyncio + async def test_download_image_ssl_error_should_raise(self): + """Test that SSL errors raise exceptions instead of silently downgrading. + + This is the critical test for the vulnerability fix. The function should + NOT catch SSL errors and retry with CERT_NONE. It should let the exception + propagate to the caller. + """ + mock_session = MagicMock() + mock_session.get = MagicMock( + side_effect=aiohttp.ClientConnectorSSLError( + connection_key=None, + os_error=ssl.SSLError("certificate verify failed"), + ) + ) + mock_session.__aenter__ = AsyncMock(return_value=mock_session) + mock_session.__aexit__ = AsyncMock(return_value=None) + + with patch("aiohttp.ClientSession", return_value=mock_session): + # After the fix, this should raise the SSL error instead of downgrading + with pytest.raises( + (aiohttp.ClientConnectorSSLError, aiohttp.ClientConnectorCertificateError) + ): + await download_image_by_url("https://invalid-cert.example.com/image.jpg") + + @pytest.mark.asyncio + async def test_download_image_certificate_error_should_raise(self): + """Test that certificate errors raise exceptions instead of downgrading.""" + mock_session = MagicMock() + mock_session.get = MagicMock( + side_effect=aiohttp.ClientConnectorCertificateError( + connection_key=None, + certificate_error=Exception("cert verification failed"), + ) + ) + mock_session.__aenter__ = AsyncMock(return_value=mock_session) + mock_session.__aexit__ = AsyncMock(return_value=None) + + with patch("aiohttp.ClientSession", return_value=mock_session): + with pytest.raises( + (aiohttp.ClientConnectorSSLError, aiohttp.ClientConnectorCertificateError) + ): + await download_image_by_url("https://expired-cert.example.com/image.jpg") + + @pytest.mark.asyncio + async def test_download_image_http_url_still_works(self): + """Test that HTTP (non-HTTPS) URLs continue to work normally.""" + mock_response = AsyncMock() + mock_response.read = AsyncMock(return_value=b"fake_image_data") + mock_response.__aenter__ = AsyncMock(return_value=mock_response) + mock_response.__aexit__ = AsyncMock(return_value=None) + + mock_session = MagicMock() + mock_session.get = MagicMock(return_value=mock_response) + mock_session.__aenter__ = AsyncMock(return_value=mock_session) + mock_session.__aexit__ = AsyncMock(return_value=None) + + with patch("aiohttp.ClientSession", return_value=mock_session): + result = await download_image_by_url("http://example.com/image.jpg") + + assert result is not None + assert Path(result).exists() + mock_session.get.assert_called_once_with("http://example.com/image.jpg") + # Clean up temp file + Path(result).unlink(missing_ok=True) + + @pytest.mark.asyncio + async def test_download_image_post_ssl_error_should_raise(self): + """Test that SSL errors in POST requests also raise exceptions.""" + mock_session = MagicMock() + mock_session.post = MagicMock( + side_effect=aiohttp.ClientConnectorSSLError( + connection_key=None, + os_error=ssl.SSLError("certificate verify failed"), + ) + ) + mock_session.__aenter__ = AsyncMock(return_value=mock_session) + mock_session.__aexit__ = AsyncMock(return_value=None) + + with patch("aiohttp.ClientSession", return_value=mock_session): + with pytest.raises( + (aiohttp.ClientConnectorSSLError, aiohttp.ClientConnectorCertificateError) + ): + await download_image_by_url( + "https://invalid-cert.example.com/api/image", + post=True, + post_data={"key": "value"}, + ) + + @pytest.mark.asyncio + async def test_download_image_with_path_ssl_error_should_raise(self): + """Test that SSL errors raise when downloading to a specific path.""" + mock_session = MagicMock() + mock_session.get = MagicMock( + side_effect=aiohttp.ClientConnectorSSLError( + connection_key=None, + os_error=ssl.SSLError("certificate verify failed"), + ) + ) + mock_session.__aenter__ = AsyncMock(return_value=mock_session) + mock_session.__aexit__ = AsyncMock(return_value=None) + + with tempfile.NamedTemporaryFile(delete=False) as tmp: + tmp_path = tmp.name + + try: + with patch("aiohttp.ClientSession", return_value=mock_session): + with pytest.raises( + (aiohttp.ClientConnectorSSLError, aiohttp.ClientConnectorCertificateError) + ): + await download_image_by_url( + "https://invalid-cert.example.com/image.jpg", + path=tmp_path, + ) + finally: + Path(tmp_path).unlink(missing_ok=True) + + +class TestDownloadFileSSLVerification: + """Test SSL verification behavior in download_file function.""" + + @pytest.mark.asyncio + async def test_download_file_https_success(self): + """Test that HTTPS file downloads work correctly with valid certificates.""" + mock_response = AsyncMock() + mock_response.status = 200 + mock_response.headers = {"content-length": "1024"} + mock_response.content = AsyncMock() + mock_response.content.read = AsyncMock(side_effect=[b"data", b""]) + mock_response.__aenter__ = AsyncMock(return_value=mock_response) + mock_response.__aexit__ = AsyncMock(return_value=None) + + mock_session = MagicMock() + mock_session.get = MagicMock(return_value=mock_response) + mock_session.__aenter__ = AsyncMock(return_value=mock_session) + mock_session.__aexit__ = AsyncMock(return_value=None) + + with tempfile.NamedTemporaryFile(delete=False) as tmp: + tmp_path = tmp.name + + try: + with patch("aiohttp.ClientSession", return_value=mock_session): + await download_file("https://example.com/file.zip", tmp_path) + + assert Path(tmp_path).exists() + mock_session.get.assert_called() + finally: + Path(tmp_path).unlink(missing_ok=True) + + @pytest.mark.asyncio + async def test_download_file_ssl_error_with_fallback_disabled_should_raise(self): + """Test that SSL errors raise when allow_insecure_ssl_fallback=False. + + This is the secure behavior - when the fallback is disabled, SSL errors + should propagate immediately without any retry attempts. + """ + mock_session = MagicMock() + mock_session.get = MagicMock( + side_effect=aiohttp.ClientConnectorSSLError( + connection_key=None, + os_error=ssl.SSLError("certificate verify failed"), + ) + ) + mock_session.__aenter__ = AsyncMock(return_value=mock_session) + mock_session.__aexit__ = AsyncMock(return_value=None) + + with tempfile.NamedTemporaryFile(delete=False) as tmp: + tmp_path = tmp.name + + try: + with patch("aiohttp.ClientSession", return_value=mock_session): + with pytest.raises( + (aiohttp.ClientConnectorSSLError, aiohttp.ClientConnectorCertificateError) + ): + await download_file( + "https://invalid-cert.example.com/file.zip", + tmp_path, + allow_insecure_ssl_fallback=False, + ) + finally: + Path(tmp_path).unlink(missing_ok=True) + + @pytest.mark.asyncio + async def test_download_file_certificate_error_should_raise(self): + """Test that certificate errors raise when fallback is disabled.""" + mock_session = MagicMock() + mock_session.get = MagicMock( + side_effect=aiohttp.ClientConnectorCertificateError( + connection_key=None, + certificate_error=Exception("cert verification failed"), + ) + ) + mock_session.__aenter__ = AsyncMock(return_value=mock_session) + mock_session.__aexit__ = AsyncMock(return_value=None) + + with tempfile.NamedTemporaryFile(delete=False) as tmp: + tmp_path = tmp.name + + try: + with patch("aiohttp.ClientSession", return_value=mock_session): + with pytest.raises( + (aiohttp.ClientConnectorSSLError, aiohttp.ClientConnectorCertificateError) + ): + await download_file( + "https://expired-cert.example.com/file.zip", + tmp_path, + allow_insecure_ssl_fallback=False, + ) + finally: + Path(tmp_path).unlink(missing_ok=True) + + @pytest.mark.asyncio + async def test_download_file_http_url_still_works(self): + """Test that HTTP (non-HTTPS) file downloads continue to work.""" + mock_response = AsyncMock() + mock_response.status = 200 + mock_response.headers = {"content-length": "1024"} + mock_response.content = AsyncMock() + mock_response.content.read = AsyncMock(side_effect=[b"data", b""]) + mock_response.__aenter__ = AsyncMock(return_value=mock_response) + mock_response.__aexit__ = AsyncMock(return_value=None) + + mock_session = MagicMock() + mock_session.get = MagicMock(return_value=mock_response) + mock_session.__aenter__ = AsyncMock(return_value=mock_session) + mock_session.__aexit__ = AsyncMock(return_value=None) + + with tempfile.NamedTemporaryFile(delete=False) as tmp: + tmp_path = tmp.name + + try: + with patch("aiohttp.ClientSession", return_value=mock_session): + await download_file("http://example.com/file.zip", tmp_path) + + assert Path(tmp_path).exists() + mock_session.get.assert_called() + finally: + Path(tmp_path).unlink(missing_ok=True) + + @pytest.mark.asyncio + async def test_download_file_non_ssl_errors_still_propagate(self): + """Test that non-SSL errors (like network errors) still propagate correctly.""" + mock_session = MagicMock() + mock_session.get = MagicMock( + side_effect=aiohttp.ClientConnectorError( + connection_key=None, + os_error=OSError("Connection refused"), + ) + ) + mock_session.__aenter__ = AsyncMock(return_value=mock_session) + mock_session.__aexit__ = AsyncMock(return_value=None) + + with tempfile.NamedTemporaryFile(delete=False) as tmp: + tmp_path = tmp.name + + try: + with patch("aiohttp.ClientSession", return_value=mock_session): + with pytest.raises(aiohttp.ClientConnectorError): + await download_file( + "https://example.com/file.zip", + tmp_path, + allow_insecure_ssl_fallback=False, + ) + finally: + Path(tmp_path).unlink(missing_ok=True) + + +class TestSSLContextSecurity: + """Test that secure SSL contexts are used and insecure patterns are avoided.""" + + @pytest.mark.asyncio + async def test_ssl_context_uses_cert_verification(self): + """Verify that SSL context is created with proper certificate verification. + + This test ensures that when SSL context is created, it uses the default + secure settings and does not disable verification. + """ + mock_context = ssl.create_default_context() + + with patch("ssl.create_default_context", return_value=mock_context) as mock_ssl_context: + mock_response = AsyncMock() + mock_response.read = AsyncMock(return_value=b"data") + mock_response.__aenter__ = AsyncMock(return_value=mock_response) + mock_response.__aexit__ = AsyncMock(return_value=None) + + mock_session = MagicMock() + mock_session.get = MagicMock(return_value=mock_response) + mock_session.__aenter__ = AsyncMock(return_value=mock_session) + mock_session.__aexit__ = AsyncMock(return_value=None) + + with patch("aiohttp.ClientSession", return_value=mock_session): + result = await download_image_by_url("https://example.com/image.jpg") + Path(result).unlink(missing_ok=True) + + # Verify SSL context was created (certification verification enabled) + mock_ssl_context.assert_called() + # Verify that check_hostname and verify_mode retain secure defaults + assert mock_context.check_hostname is True + assert mock_context.verify_mode == ssl.CERT_REQUIRED + + @pytest.mark.asyncio + async def test_no_insecure_ssl_context_created_on_error(self): + """Verify that insecure SSL contexts (CERT_NONE) are not created on SSL errors. + + After the fix, when an SSL error occurs, the function should raise the error + immediately rather than creating a new SSL context with verify_mode=CERT_NONE. + """ + call_count = 0 + original_create_default_context = ssl.create_default_context + + def mock_ssl_context(*args, **kwargs): + nonlocal call_count + call_count += 1 + # Call the original function to avoid recursion + return original_create_default_context(*args, **kwargs) + + with patch("ssl.create_default_context", side_effect=mock_ssl_context): + mock_session = MagicMock() + mock_session.get = MagicMock( + side_effect=aiohttp.ClientConnectorSSLError( + connection_key=None, + os_error=ssl.SSLError("certificate verify failed"), + ) + ) + mock_session.__aenter__ = AsyncMock(return_value=mock_session) + mock_session.__aexit__ = AsyncMock(return_value=None) + + with patch("aiohttp.ClientSession", return_value=mock_session): + with pytest.raises( + (aiohttp.ClientConnectorSSLError, aiohttp.ClientConnectorCertificateError) + ): + await download_image_by_url("https://invalid-cert.example.com/image.jpg") + + # After the fix, SSL context should only be created once (for the initial attempt) + # There should be no second context creation with CERT_NONE + assert call_count == 1, "SSL context should only be created once, not for a fallback retry"