From 5247bd784ab3b33a13030b7846e1c01fc63aa1b0 Mon Sep 17 00:00:00 2001 From: LKuemmel Date: Thu, 22 Oct 2026 09:31:38 +0200 Subject: [PATCH 1/2] =?UTF-8?q?Bin=C3=A4rantworten=20nur=20mit=20Typ=20und?= =?UTF-8?q?=20Gr=C3=B6=C3=9Fe=20loggen?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- packages/modules/common/req.py | 18 ++++++++- packages/modules/common/req_test.py | 59 +++++++++++++++++++++++++++++ 2 files changed, 75 insertions(+), 2 deletions(-) create mode 100644 packages/modules/common/req_test.py diff --git a/packages/modules/common/req.py b/packages/modules/common/req.py index 3fe978deff..de65c0eed0 100644 --- a/packages/modules/common/req.py +++ b/packages/modules/common/req.py @@ -1,6 +1,6 @@ import copy import logging -from requests import Session +from requests import Response, Session import urllib3 from functools import wraps import warnings @@ -47,8 +47,22 @@ def __deepcopy__(self, memo): return new_copy +def _log_response(response: Response, *args, **kwargs) -> None: + if not log.isEnabledFor(logging.DEBUG): + return + content_type = response.headers.get('Content-Type', '').split(';', 1)[0].strip().lower() + binary = (content_type.startswith(('image/', 'audio/', 'video/', 'font/')) or + content_type in ('application/zip', 'application/x-zip-compressed', + 'application/octet-stream', 'application/pdf', 'application/gzip')) + # The portal can omit/mislabel the content type of ZIP downloads. + if binary or response.content.startswith((b'PK\x03\x04', b'PK\x05\x06', b'PK\x07\x08')): + log.debug("Get-Response: binary data (%s, %s bytes)", content_type or 'unknown type', len(response.content)) + else: + log.debug("Get-Response: %s", response.text) + + def get_http_session() -> CustomSession: session = CustomSession() session.hooks['response'].append(lambda r, *args, **kwargs: r.raise_for_status()) - session.hooks['response'].append(lambda r, *args, **kwargs: log.debug("Get-Response: " + r.text)) + session.hooks['response'].append(_log_response) return session diff --git a/packages/modules/common/req_test.py b/packages/modules/common/req_test.py new file mode 100644 index 0000000000..550ea42b91 --- /dev/null +++ b/packages/modules/common/req_test.py @@ -0,0 +1,59 @@ +import io +import logging +import zipfile +from unittest.mock import Mock, PropertyMock + +import pytest +from requests import HTTPError, Response + +from modules.common import req + + +@pytest.mark.parametrize('content_type', ['application/zip', 'application/octet-stream', '', 'text/plain']) +def test_zip_response_logs_metadata_without_decoding_or_changing_body(requests_mock, caplog, content_type): + archive = io.BytesIO() + with zipfile.ZipFile(archive, 'w') as handle: + handle.writestr('vehicle.json', '{"soc": 55}') + body = archive.getvalue() + requests_mock.get('https://example.org/data', content=body, headers={'Content-Type': content_type}) + caplog.set_level(logging.DEBUG, logger=req.__name__) + with req.get_http_session() as session: + response = session.get('https://example.org/data') + assert response.content == body + assert 'binary data' in caplog.text + assert f'{len(body)} bytes' in caplog.text + assert 'vehicle.json' not in caplog.text + assert '"soc"' not in caplog.text + + +@pytest.mark.parametrize('content_type', ['application/json', 'text/html; charset=utf-8', '']) +def test_text_response_is_still_logged(requests_mock, caplog, content_type): + body = '{"soc": 55}' + requests_mock.get('https://example.org/data', text=body, headers={'Content-Type': content_type}) + caplog.set_level(logging.DEBUG, logger=req.__name__) + with req.get_http_session() as session: + response = session.get('https://example.org/data') + assert response.json() == {'soc': 55} + assert 'Get-Response: ' + body in caplog.text + + +def test_binary_content_type_does_not_access_response_text(caplog): + response = Mock(spec=Response, headers={'Content-Type': 'application/octet-stream'}, content=b'\x00\xff') + type(response).text = PropertyMock(side_effect=AssertionError('Binary body must not be decoded')) + caplog.set_level(logging.DEBUG, logger=req.__name__) + req._log_response(response) + assert 'binary data (application/octet-stream, 2 bytes)' in caplog.text + + +def test_disabled_debug_does_not_read_response(caplog): + caplog.set_level(logging.INFO, logger=req.__name__) + req._log_response(object()) + + +def test_error_status_still_raises_with_full_response(requests_mock): + requests_mock.get('https://example.org/data', status_code=429, text='Too many requests') + with req.get_http_session() as session: + with pytest.raises(HTTPError) as error: + session.get('https://example.org/data') + assert error.value.response.status_code == 429 + assert error.value.response.text == 'Too many requests' From e722152369248b7a2d578c90561bad3fc0308ba1 Mon Sep 17 00:00:00 2001 From: LKuemmel Date: Wed, 4 Nov 2026 09:42:38 +0100 Subject: [PATCH 2/2] Review --- packages/modules/common/req.py | 8 ++++---- packages/modules/common/req_test.py | 27 +++++++++++++++++++++++---- 2 files changed, 27 insertions(+), 8 deletions(-) diff --git a/packages/modules/common/req.py b/packages/modules/common/req.py index de65c0eed0..5034367476 100644 --- a/packages/modules/common/req.py +++ b/packages/modules/common/req.py @@ -51,11 +51,11 @@ def _log_response(response: Response, *args, **kwargs) -> None: if not log.isEnabledFor(logging.DEBUG): return content_type = response.headers.get('Content-Type', '').split(';', 1)[0].strip().lower() - binary = (content_type.startswith(('image/', 'audio/', 'video/', 'font/')) or - content_type in ('application/zip', 'application/x-zip-compressed', - 'application/octet-stream', 'application/pdf', 'application/gzip')) + text = (content_type.startswith('text/') or + content_type in ('application/json', 'application/xml', 'application/x-www-form-urlencoded') or + (content_type.startswith('application/') and content_type.endswith(('+json', '+xml')))) # The portal can omit/mislabel the content type of ZIP downloads. - if binary or response.content.startswith((b'PK\x03\x04', b'PK\x05\x06', b'PK\x07\x08')): + if not text or response.content.startswith((b'PK\x03\x04', b'PK\x05\x06', b'PK\x07\x08')): log.debug("Get-Response: binary data (%s, %s bytes)", content_type or 'unknown type', len(response.content)) else: log.debug("Get-Response: %s", response.text) diff --git a/packages/modules/common/req_test.py b/packages/modules/common/req_test.py index 550ea42b91..dda5c36236 100644 --- a/packages/modules/common/req_test.py +++ b/packages/modules/common/req_test.py @@ -26,7 +26,11 @@ def test_zip_response_logs_metadata_without_decoding_or_changing_body(requests_m assert '"soc"' not in caplog.text -@pytest.mark.parametrize('content_type', ['application/json', 'text/html; charset=utf-8', '']) +@pytest.mark.parametrize('content_type', [ + 'application/json', 'text/plain', 'text/html; charset=utf-8', 'application/xml', + 'application/problem+json', 'application/vnd.example+xml', 'application/x-www-form-urlencoded', + ' Application/JSON ; charset=UTF-8', +]) def test_text_response_is_still_logged(requests_mock, caplog, content_type): body = '{"soc": 55}' requests_mock.get('https://example.org/data', text=body, headers={'Content-Type': content_type}) @@ -37,12 +41,27 @@ def test_text_response_is_still_logged(requests_mock, caplog, content_type): assert 'Get-Response: ' + body in caplog.text -def test_binary_content_type_does_not_access_response_text(caplog): - response = Mock(spec=Response, headers={'Content-Type': 'application/octet-stream'}, content=b'\x00\xff') +@pytest.mark.parametrize('content_type', [ + 'application/octet-stream', 'application/x-gzip', 'application/x-7z-compressed', + 'application/vnd.example.binary', 'image/png', 'audio/mpeg', 'video/mp4', 'font/woff2', + 'application/pdf', '', None, +]) +def test_binary_content_type_does_not_access_response_text(caplog, content_type): + headers = {'Content-Type': content_type} if content_type is not None else {} + response = Mock(spec=Response, headers=headers, content=b'\x00\xff') type(response).text = PropertyMock(side_effect=AssertionError('Binary body must not be decoded')) caplog.set_level(logging.DEBUG, logger=req.__name__) req._log_response(response) - assert 'binary data (application/octet-stream, 2 bytes)' in caplog.text + assert f'binary data ({content_type or "unknown type"}, 2 bytes)' in caplog.text + + +@pytest.mark.parametrize('signature', [b'PK\x03\x04', b'PK\x05\x06', b'PK\x07\x08']) +def test_mislabeled_zip_does_not_access_response_text(caplog, signature): + response = Mock(spec=Response, headers={'Content-Type': 'text/plain'}, content=signature) + type(response).text = PropertyMock(side_effect=AssertionError('ZIP body must not be decoded')) + caplog.set_level(logging.DEBUG, logger=req.__name__) + req._log_response(response) + assert 'binary data (text/plain, 4 bytes)' in caplog.text def test_disabled_debug_does_not_read_response(caplog):