Fix py3.9 ci failure for directed-inputs-class#252
Fix py3.9 ci failure for directed-inputs-class#252jbdevprimary wants to merge 5 commits intofeat/directed-inputs-decorator-apifrom
Conversation
Add @directed_inputs class decorator and @input_config method decorator as modern alternatives to DirectedInputsClass inheritance. ## New Features - `@directed_inputs` class decorator for automatic input loading - `@input_config` method decorator for per-parameter configuration - Automatic type coercion (bool, int, float, Path, datetime, dict, list) - Case-insensitive key lookup - Full backward compatibility with legacy DirectedInputsClass API ## Components - `decorators.py`: New decorator implementations - `InputContext`: Runtime input storage and lookup - `InputConfig`: Per-parameter configuration dataclass ## Tests - 23 new tests for decorator API - 39 total tests passing (16 legacy + 23 new) Part of terraform-modules migration architectural refactor. Depends on: PR #246 (docs/wiki-orchestration-update)
Co-authored-by: jon <jon@jonbogaty.com>
Co-authored-by: jon <jon@jonbogaty.com>
|
Cursor Agent can help with this pull request. Just |
|
⏳ Code review in progress. Analyzing for code quality issues and best practices. You can monitor the review status in the checks section at the bottom of this pull request. Detailed findings will be posted upon completion. Using Amazon Q Developer for GitHubAmazon Q Developer1 is an AI-powered assistant that integrates directly into your GitHub workflow, enhancing your development process with intelligent features for code development, review, and transformation. Slash Commands
FeaturesAgentic Chat Code Review CustomizationYou can create project-specific rules for Amazon Q Developer to follow:
Example rule: FeedbackTo provide feedback on Amazon Q Developer, create an issue in the Amazon Q Developer public repository. For more detailed information, visit the Amazon Q for GitHub documentation. Footnotes
|
![amazon-q-developer[bot]](https://avatars.githubusercontent.com/in/1220912?s=60&v=4){kind=small}
There was a problem hiding this comment.
Review Summary
This PR addresses Python 3.9 compatibility issues and adds important security improvements for stdin handling. However, there are several critical issues that need to be resolved before merge:
Critical Issues Found:
- Security Vulnerability: The stdin size limit implementation has a race condition that allows memory consumption beyond the intended limit
- Logic Errors: Duplicate and inconsistent union type handling logic that could cause unpredictable type resolution behavior
- Test Coverage: Missing boundary condition testing for the security feature
Positive Changes:
- ✅ Proper Python 3.9 compatibility with
types.UnionTypehandling - ✅ Addition of stdin size limits for security (implementation needs fixing)
- ✅ Correct import fix in docstring example
- ✅ Parameter renaming for clarity (
decode_from_json/decode_from_yaml)
Required Actions:
- Fix the stdin security implementation to prevent the race condition
- Consolidate the duplicate union type handling logic
- Enhance test coverage for boundary conditions
The core approach is sound, but the implementation details need refinement to ensure security and consistency.
You can now have the agent implement changes and create commits directly on your pull request's source branch. Simply comment with /q followed by your request in natural language to ask the agent to make changes.
| stdin_data = sys.stdin.read(STDIN_MAX_BYTES + 1) | ||
| except OSError: | ||
| return {} | ||
|
|
||
| if len(stdin_data) > STDIN_MAX_BYTES: | ||
| msg = f"Stdin input exceeds maximum size limit ({STDIN_MAX_BYTES} bytes)" | ||
| raise ValueError(msg) |
There was a problem hiding this comment.
🛑 Security Vulnerability: The stdin size limit check has a race condition that allows bypassing the security control. Reading STDIN_MAX_BYTES + 1 bytes and then checking the length allows an attacker to consume more memory than intended before the check occurs. This could lead to memory exhaustion attacks1.
| stdin_data = sys.stdin.read(STDIN_MAX_BYTES + 1) | |
| except OSError: | |
| return {} | |
| if len(stdin_data) > STDIN_MAX_BYTES: | |
| msg = f"Stdin input exceeds maximum size limit ({STDIN_MAX_BYTES} bytes)" | |
| raise ValueError(msg) | |
| stdin_data = sys.stdin.read(STDIN_MAX_BYTES) | |
| except OSError: | |
| return {} | |
| if len(stdin_data) == STDIN_MAX_BYTES: | |
| # Check if there's more data available | |
| try: | |
| extra = sys.stdin.read(1) | |
| if extra: | |
| msg = f"Stdin input exceeds maximum size limit ({STDIN_MAX_BYTES} bytes)" | |
| raise ValueError(msg) | |
| except OSError: | |
| pass |
Footnotes
-
CWE-400: Uncontrolled Resource Consumption - https://cwe.mitre.org/data/definitions/400.html ↩
| if "|" in cleaned: | ||
| parts = [part.strip() for part in cleaned.split("|")] | ||
| for part in parts: | ||
| resolved = _normalize_string_type(part) | ||
| if resolved is not None: | ||
| return resolved | ||
| return None |
There was a problem hiding this comment.
🛑 Logic Error: The union type parsing logic has a critical flaw that could cause incorrect type resolution. When parsing str | None, if the first part fails to resolve, it continues to the next part and may return None as the resolved type, leading to no type coercion being applied when string coercion was intended.
| if "|" in cleaned: | |
| parts = [part.strip() for part in cleaned.split("|")] | |
| for part in parts: | |
| resolved = _normalize_string_type(part) | |
| if resolved is not None: | |
| return resolved | |
| return None | |
| if "|" in cleaned: | |
| parts = [part.strip() for part in cleaned.split("|")] | |
| # Filter out None types first, then try to resolve the first non-None type | |
| non_none_parts = [part for part in parts if part.lower() not in {"none", "nonetype", "type(none)"}] | |
| for part in non_none_parts: | |
| resolved = _normalize_string_type(part) | |
| if resolved is not None: | |
| return resolved | |
| return None |
| if origin is Union or (_UNION_TYPE is not None and origin is _UNION_TYPE): | ||
| for arg in get_args(target_type): | ||
| resolved = _resolve_type_hint(arg) | ||
| if resolved is not None: | ||
| return resolved | ||
| return None | ||
|
|
||
| if _UNION_TYPE is not None and isinstance(target_type, _UNION_TYPE): | ||
| for arg in getattr(target_type, "__args__", ()): | ||
| resolved = _resolve_type_hint(arg) | ||
| if resolved is not None: | ||
| return resolved | ||
| return None |
There was a problem hiding this comment.
🛑 Logic Error: Duplicate union type handling logic creates inconsistent behavior. The same union type resolution is implemented twice with slightly different approaches, which could lead to different results for the same input type depending on which path is taken.
| if origin is Union or (_UNION_TYPE is not None and origin is _UNION_TYPE): | |
| for arg in get_args(target_type): | |
| resolved = _resolve_type_hint(arg) | |
| if resolved is not None: | |
| return resolved | |
| return None | |
| if _UNION_TYPE is not None and isinstance(target_type, _UNION_TYPE): | |
| for arg in getattr(target_type, "__args__", ()): | |
| resolved = _resolve_type_hint(arg) | |
| if resolved is not None: | |
| return resolved | |
| return None | |
| origin = get_origin(target_type) | |
| if origin is Union or (_UNION_TYPE is not None and (origin is _UNION_TYPE or isinstance(target_type, _UNION_TYPE))): | |
| # Handle both typing.Union and types.UnionType uniformly | |
| if hasattr(target_type, "__args__"): | |
| args = target_type.__args__ | |
| else: | |
| args = get_args(target_type) | |
| for arg in args: | |
| resolved = _resolve_type_hint(arg) | |
| if resolved is not None: | |
| return resolved | |
| return None |
| if origin is Union or (_UNION_TYPE is not None and origin is _UNION_TYPE): | ||
| args = get_args(target_type) | ||
| non_none_types = [t for t in args if t is not type(None)] | ||
| if non_none_types: | ||
| target_type = non_none_types[0] | ||
| else: | ||
| return value | ||
| elif _UNION_TYPE is not None and isinstance(target_type, _UNION_TYPE): | ||
| args = getattr(target_type, "__args__", ()) | ||
| non_none_types = [t for t in args if t is not type(None)] | ||
| if non_none_types: | ||
| target_type = non_none_types[0] | ||
| else: | ||
| return value |
There was a problem hiding this comment.
🛑 Logic Error: Duplicate union type handling in _coerce_value creates the same inconsistency issue as in _resolve_type_hint. The logic should be unified to prevent different behavior for the same union types.
| if origin is Union or (_UNION_TYPE is not None and origin is _UNION_TYPE): | |
| args = get_args(target_type) | |
| non_none_types = [t for t in args if t is not type(None)] | |
| if non_none_types: | |
| target_type = non_none_types[0] | |
| else: | |
| return value | |
| elif _UNION_TYPE is not None and isinstance(target_type, _UNION_TYPE): | |
| args = getattr(target_type, "__args__", ()) | |
| non_none_types = [t for t in args if t is not type(None)] | |
| if non_none_types: | |
| target_type = non_none_types[0] | |
| else: | |
| return value | |
| # Handle Union types (including X | None via types.UnionType) | |
| origin = get_origin(target_type) | |
| if origin is Union or (_UNION_TYPE is not None and (origin is _UNION_TYPE or isinstance(target_type, _UNION_TYPE))): | |
| # Handle both typing.Union and types.UnionType uniformly | |
| if hasattr(target_type, "__args__"): | |
| args = target_type.__args__ | |
| else: | |
| args = get_args(target_type) | |
| non_none_types = [t for t in args if t is not type(None)] | |
| if non_none_types: | |
| target_type = non_none_types[0] | |
| else: | |
| return value |
| oversized = "x" * (decorators_mod.STDIN_MAX_BYTES + 1) | ||
| with patch( | ||
| "directed_inputs_class.decorators.sys.stdin.read", | ||
| return_value=oversized, | ||
| ), pytest.raises(ValueError, match=r"exceeds maximum size"): |
There was a problem hiding this comment.
Test coverage gap: The test only verifies the error message but doesn't test the actual memory protection. Consider adding a test that verifies the exact boundary condition (data exactly at the limit should pass).
| oversized = "x" * (decorators_mod.STDIN_MAX_BYTES + 1) | |
| with patch( | |
| "directed_inputs_class.decorators.sys.stdin.read", | |
| return_value=oversized, | |
| ), pytest.raises(ValueError, match=r"exceeds maximum size"): | |
| oversized = "x" * (decorators_mod.STDIN_MAX_BYTES + 1) | |
| exactly_at_limit = "x" * decorators_mod.STDIN_MAX_BYTES | |
| # Test that exactly at limit works | |
| with patch( | |
| "directed_inputs_class.decorators.sys.stdin.read", | |
| return_value=exactly_at_limit, | |
| ): | |
| MyService() # Should not raise | |
| # Test that over limit fails | |
| with patch( | |
| "directed_inputs_class.decorators.sys.stdin.read", | |
| return_value=oversized, | |
| ), pytest.raises(ValueError, match=r"exceeds maximum size"): |
|
Closing - stale fix attempt. Original issue addressed differently. |
cursor
bot
force-pushed
the
feat/directed-inputs-decorator-api
branch
from
43ca607 to
8b9889b
Compare
…dling - Fix stdin size limit race condition: read limit first, then check for extra - Remove duplicate union type handling in _coerce_value (handled by _resolve_type_hint) - Add error handling for type coercion failures Addresses Amazon Q Developer feedback.
jbdevprimary
deleted the
cursor/fix-py3-9-ci-failure-for-directed-inputs-class-9453
branch
2 participants
Fix Python 3.9 CI and address review feedback for
directed-inputs-classDescription
This PR resolves the Python 3.9 CI failure for
directed-inputs-classand addresses critical review feedback.Motivation:
The
directed-inputs-classCI was failing on Python 3.9 due to incompatibilities with newer typing features (e.g.,str | Nonesyntax). Additionally, review feedback highlighted the need for stdin input safety and correct YAML decoding.Changes:
_normalize_string_type,_resolve_type_hint) to correctly parse string-based andUniontype annotations across Python versions, specifically addressingtypes.UnionTypein Python 3.9.ValueErrorfor oversized payloads._load_stdinnow also includes a graceful JSON decoding fallback.decode_yamlwithin_decode_valueto ensure proper YAML processing and prevent shadowing. Parameters were renamed todecode_from_jsonanddecode_from_yamlfor clarity.directed_inputs_classimport in the docstring.Fixes #247
Type of change
How has this been tested?
Local tests were run to verify the changes:
uv run --with pytest pytest packages/directed-inputs-class/tests -q(for Python 3.13).venv-py39/bin/pytest packages/directed-inputs-class/tests -q(for Python 3.9)test_stdin_size_limitwas added to confirm the stdin safety feature.uv run --with ruff ruff check packages/directed-inputs-class/src/directed_inputs_class/decorators.py packages/directed-inputs-class/tests/test_decorators.pyTest Configuration:
Checklist
Note
Adds Py3.9-safe type-hint resolution, stdin size limits, and decoding fixes to
directed_inputs_classwith a new test.packages/directed-inputs-class/src/directed_inputs_class/decorators.py):_normalize_string_type,_resolve_type_hint, handleUnion/types.UnionType, string annotations, andForwardRef; integrate into_coerce_value.STDIN_MAX_BYTES, limitsys.stdin.read, handleOSError, and raiseValueErrorfor oversized payloads; retain JSON fallback.decode_yamlcorrectly; rename internal flags todecode_from_json/decode_from_yaml; adjust base64 unwrap behavior.directed_inputs_class.packages/directed-inputs-class/tests/test_decorators.py):test_stdin_size_limitand importdecorators_mod; minor test updates to reflect new behavior.Written by Cursor Bugbot for commit b787ff7. This will update automatically on new commits. Configure here.