T09 · Insecure Skill Coding Practices
Warning
- Location
- scripts/x_data_validator.py:203
- Finding
- Invalid X Records Remain Accepted After Integrity and Sanity-Check Failures<![CDATA[ ## Vulnerability Details **File Location**: `scripts/x_data_validator.py:203-294` **Vulnerability Type**: Fail-open validation and inconsistent validation status **Risk Level**: Medium ### Vulnerable Code ```python timeline_author = self.timeline_cache.get(tweet_id, {}).get("authorId") details_author = tweet.get("authorId") if timeline_author and details_author and timeline_author != details_author: checks.append({ "step": "cross_validation_author", "status": "ERROR", "tweet_id": tweet_id, "message": "Author ID mismatch between timeline and details", "severity": "ERROR" }) self._log_error(f"Author mismatch for tweet {tweet_id}", { "timeline": timeline_author, "details": details_author }) ``` ```python if likes is not None and impressions is not None and likes > impressions: checks.append({ "step": "sanity_check_likes", "status": "ERROR", "tweet_id": tweet_id, "message": f"IMPOSSIBLE: Likes ({likes}) > Impressions ({impressions})", "action": "FLAG_ANOMALY — data error detected but not modified", "severity": "ERROR" }) self._log_error("Data anomaly: likes > impressions", { "tweet_id": tweet_id, "likes": likes, "impressions": impressions }) if retweets is not None and impressions is not None and retweets > impressions: checks.append({ "step": "sanity_check_retweets", "status": "ERROR", "tweet_id": tweet_id, "message": f"IMPOSSIBLE: Retweets ({retweets}) > Impressions ({impressions})", "severity": "ERROR" }) self._log_error("Data anomaly: retweets > impressions", { "tweet_id": tweet_id, "retweets": retweets, "impressions": impressions }) for metric_name in required_metrics: value = metrics.get(metric_name) if value is not None and value < 0: checks.append({ "step": f"sanity_check_{metric_name}", "status": "ERROR", ...[truncated 3210 chars]
- Remediation
- <![CDATA[ ## Remediation Suggestions 1. Reject records with critical integrity failures: - Return `None` for author mismatches. - Return `None` for negative metrics. - Return `None` when likes or retweets exceed impressions, unless the API semantics explicitly permit that relationship. 2. Derive the final status from accumulated results instead of assigning `OK` unconditionally: ```python has_errors = any( check.get("severity") in {"ERROR", "CRITICAL"} for check in checks ) checks.append({ "step": "tweet_validation_complete", "status": "FAILED" if has_errors else "OK", "tweet_id": tweet_id, "message": ( "Tweet rejected because validation errors were detected" if has_errors else "Tweet data validated successfully" ), "has_warnings": any(c.get("severity") == "WARNING" for c in checks), "has_errors": has_errors, }) if has_errors: return None, checks return tweet, checks ``` 3. Distinguish explicitly between: - Valid records. - Valid records with non-fatal warnings. - Quarantined anomalous records. - Rejected records. 4. Require downstream analytics to consume only records carrying an explicit successful validation result. Do not rely solely on the record being non-null. 5. Avoid mutating the input response when adding missing metrics. Create a defensive copy before normalization so validation does not alter evidence supplied by the upstream source. 6. Add regression tests covering: - Author ID mismatch. - Negative metric values. - Likes greater than impressions. - Retweets greater than impressions. - Missing metrics. - Warning-only records. - Verification that rejected records cannot enter aggregate analytics. ]]>