Skip to content

Suppressions for pyrefly #43

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
copybara-service merged 1 commit into from
Jul 24, 2026
Merged

Suppressions for pyrefly #43

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
2 changes: 1 addition & 1 deletion lib/utils.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -422,7 +422,7 @@ def is_pickle_file(
if is_seekable:
# Read a chunk to check text-based and non-pickle-magic prefixes safely
raw_bytes = pickle_bytes.read(1024)
pickle_bytes.seek(original_pos)
pickle_bytes.seek(original_pos) # pyrefly: ignore[bad-argument-type]
pickle_stream = pickle_bytes
else:
# Fallback for non-seekable stream: try to peek without advancing pointer
Expand Down
26 changes: 13 additions & 13 deletions saferpickle.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -443,7 +443,7 @@ def generate_ops(

try:
try:
for _, operand in _custom_genops(pickle_bytes):
for _, operand in _custom_genops(pickle_bytes): # pyrefly: ignore[bad-argument-type]
if operand is None:
continue
operand_str = str(operand)
Expand All @@ -456,7 +456,7 @@ def generate_ops(
return filtered_operands
finally:
if original_pos is not None:
pickle_bytes.seek(original_pos)
pickle_bytes.seek(original_pos) # pyrefly: ignore[missing-attribute]


def get_class_instantiations(
Expand Down Expand Up @@ -530,7 +530,7 @@ def get_class_instantiations(
logger.setLevel(original_level)
logger.propagate = original_propagate
if original_pos is not None:
pickle_bytes.seek(original_pos)
pickle_bytes.seek(original_pos) # pyrefly: ignore[missing-attribute]

was_unsafe_build_blocked = False
if unpickler:
Expand All @@ -551,7 +551,7 @@ def categorize_strings(
filtered_output
)
else:
safe, unsafe, suspicious, unknown = _categorize_genops(filtered_output)
safe, unsafe, suspicious, unknown = _categorize_genops(filtered_output) # pyrefly: ignore[bad-argument-type]
return _reclassify_with_resolution(safe, unsafe, suspicious, unknown)


Expand Down Expand Up @@ -683,7 +683,7 @@ def strict_security_scan(pickle_bytes: bytes | BinaryIO) -> bool:
unsafe_and_suspicious_strings = constants.UNSAFE_STRINGS.union(
constants.SUSPICIOUS_STRINGS
)
for _, operand in _custom_genops(pickle_bytes):
for _, operand in _custom_genops(pickle_bytes): # pyrefly: ignore[bad-argument-type]
if operand is None:
continue
stmt = str(operand)
Expand Down Expand Up @@ -810,7 +810,7 @@ def genops_scan(
A ScanResults object.
"""
resolved_pickle_length = (
pickle_length if pickle_length is not None else len(pickle_bytes)
pickle_length if pickle_length is not None else len(pickle_bytes) # pyrefly: ignore[bad-argument-type]
)
if shm_name:
genops_output = generate_ops_from_file(
Expand Down Expand Up @@ -984,7 +984,7 @@ def security_scan(

if is_archive:
# Temporarily archive streams fully to bytes
archive_bytes = pickle_bytes.read()
archive_bytes = pickle_bytes.read() # pyrefly: ignore[missing-attribute]
if archive_bytes.startswith(b"PK\x03\x04"):
archive_type = "zip"
elif archive_bytes.startswith(b"BZh"):
Expand Down Expand Up @@ -1060,7 +1060,7 @@ def security_scan(
)
finally:
if original_pos is not None:
pickle_bytes.seek(original_pos)
pickle_bytes.seek(original_pos) # pyrefly: ignore[missing-attribute]


def _merge_scores(total: Dict[str, int], new: Dict[str, int]):
Expand Down Expand Up @@ -1276,7 +1276,7 @@ def _security_scan_internal(

# Fast path for BytesIO, chunked fallback for other streams
if isinstance(stream, io.BytesIO):
shm.buf[:pickle_length] = stream.getbuffer()[
shm.buf[:pickle_length] = stream.getbuffer()[ # pyrefly: ignore[unsupported-operation]
start_offset : start_offset + pickle_length
]
else:
Expand All @@ -1288,7 +1288,7 @@ def _security_scan_internal(
chunk = stream.read(1024 * 1024)
if not chunk:
break
shm.buf[offset : offset + len(chunk)] = chunk
shm.buf[offset : offset + len(chunk)] = chunk # pyrefly: ignore[unsupported-operation]
offset += len(chunk)
finally:
stream.seek(current_pos)
Expand Down Expand Up @@ -1453,7 +1453,7 @@ def _scan_and_load(
load_args = (pickle_file,)
else:
load_func = loader_mod.loads # pyrefly: ignore[missing-attribute]
load_args = (data_bytes,)
load_args = (data_bytes,) # pyrefly: ignore[unbound-name]

if strict_check and allow_unsafe:
error_string_illegal_combination = (
Expand All @@ -1467,7 +1467,7 @@ def _scan_and_load(
if report_only:
logging.info("Loading pickle file with allow_unsafe set to True.")
elif strict_check:
if strict_security_scan(scan_source):
if strict_security_scan(scan_source): # pyrefly: ignore[bad-argument-type]
error_string_strict_check = "Pickle file failed strict security check."
if report_only:
logging.error(error_string_strict_check)
Expand All @@ -1476,7 +1476,7 @@ def _scan_and_load(
else:
# Default scanning routines
scan_scores = security_scan(
scan_source, force_scan=force_scan, check_magic_bytes=check_magic_bytes
scan_source, force_scan=force_scan, check_magic_bytes=check_magic_bytes # pyrefly: ignore[bad-argument-type]
)
number_of_unsafe_results = scan_scores["unsafe"]
number_of_suspicious_results = scan_scores["suspicious"]
Expand Down