Skip to content
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
80 changes: 41 additions & 39 deletions src/mock_vws/_query_validators/image_validators.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@

from beartype import beartype
from PIL import Image
from werkzeug.datastructures import FileStorage, MultiDict
from werkzeug.formparser import MultiPartParser

from mock_vws._query_validators.exceptions import (
Expand All @@ -19,19 +20,19 @@


@beartype
def validate_image_field_given(
def _parse_multipart_files(
*,
request_headers: Mapping[str, str],
request_body: bytes,
) -> None:
"""Validate that the image field is given.
) -> MultiDict[str, FileStorage]:
"""Parse the multipart body and return the files section.

Args:
request_headers: The headers sent with the request.
request_body: The body of the request.

Raises:
ImageNotGivenError: The image field is not given.
Returns:
The files parsed from the multipart body.
"""
email_message = EmailMessage()
email_message["Content-Type"] = request_headers["Content-Type"]
Expand All @@ -42,6 +43,28 @@ def validate_image_field_given(
boundary=boundary.encode(encoding="utf-8"),
content_length=len(request_body),
)
return files
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Duplicate multipart parsing logic across modules

Low Severity

The new _parse_multipart_files helper duplicates multipart parsing logic already present in get_query_match_response_text in _query_tools.py and validate_extra_fields in fields_validators.py. All three locations use identical code to extract boundary and parse multipart data. A shared utility function could eliminate this duplication and improve maintainability.

Fix in Cursor Fix in Web



@beartype
def validate_image_field_given(
*,
request_headers: Mapping[str, str],
request_body: bytes,
) -> None:
"""Validate that the image field is given.

Args:
request_headers: The headers sent with the request.
request_body: The body of the request.

Raises:
ImageNotGivenError: The image field is not given.
"""
files = _parse_multipart_files(
request_headers=request_headers,
request_body=request_body,
)
if files.get(key="image") is not None:
return

Expand All @@ -64,14 +87,9 @@ def validate_image_file_size(
Raises:
RequestEntityTooLargeError: The image file size is too large.
"""
email_message = EmailMessage()
email_message["Content-Type"] = request_headers["Content-Type"]
boundary = email_message.get_boundary(failobj="")
parser = MultiPartParser()
_, files = parser.parse(
stream=io.BytesIO(initial_bytes=request_body),
boundary=boundary.encode(encoding="utf-8"),
content_length=len(request_body),
files = _parse_multipart_files(
request_headers=request_headers,
request_body=request_body,
)
image_part = files["image"]
image_value = image_part.stream.read()
Expand Down Expand Up @@ -105,14 +123,9 @@ def validate_image_dimensions(
BadImageError: The image is given and is not within the maximum width
and height limits.
"""
email_message = EmailMessage()
email_message["Content-Type"] = request_headers["Content-Type"]
boundary = email_message.get_boundary(failobj="")
parser = MultiPartParser()
_, files = parser.parse(
stream=io.BytesIO(initial_bytes=request_body),
boundary=boundary.encode(encoding="utf-8"),
content_length=len(request_body),
files = _parse_multipart_files(
request_headers=request_headers,
request_body=request_body,
)
image_part = files["image"]
image_value = image_part.stream.read()
Expand Down Expand Up @@ -142,14 +155,9 @@ def validate_image_format(
Raises:
BadImageError: The image is given and is not either a PNG or a JPEG.
"""
email_message = EmailMessage()
email_message["Content-Type"] = request_headers["Content-Type"]
boundary = email_message.get_boundary(failobj="")
parser = MultiPartParser()
_, files = parser.parse(
stream=io.BytesIO(initial_bytes=request_body),
boundary=boundary.encode(encoding="utf-8"),
content_length=len(request_body),
files = _parse_multipart_files(
request_headers=request_headers,
request_body=request_body,
)
image_part = files["image"]
pil_image = Image.open(fp=image_part.stream)
Expand All @@ -175,17 +183,11 @@ def validate_image_is_image(
Raises:
BadImageError: Image data is given and it is not an image file.
"""
email_message = EmailMessage()
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Missing keyword-only argument enforcement

Medium Severity

The validate_image_is_image function signature is missing the *, parameter that enforces keyword-only arguments, while all four other validators (validate_image_field_given, validate_image_file_size, validate_image_dimensions, validate_image_format) correctly include it. This inconsistency breaks the API contract and allows positional arguments where they weren't previously permitted, potentially breaking external code that relied on this enforcement.

Fix in Cursor Fix in Web

email_message["Content-Type"] = request_headers["Content-Type"]
boundary = email_message.get_boundary(failobj="")
parser = MultiPartParser()
_, files = parser.parse(
stream=io.BytesIO(initial_bytes=request_body),
boundary=boundary.encode(encoding="utf-8"),
content_length=len(request_body),
files = _parse_multipart_files(
request_headers=request_headers,
request_body=request_body,
)
image_part = files["image"]
image_file = image_part.stream
image_file = files["image"].stream

try:
Image.open(fp=image_file)
Expand Down
Loading