Find anomalies in the Wheel packages that could be caused by manipulation or using a non-standard tools
(*, location: ScanLocation)
| 25 | |
| 26 | @Analyzer.ID("wheel") |
| 27 | def analyze_wheel(*, location: ScanLocation) -> AnalyzerReturnType: |
| 28 | """Find anomalies in the Wheel packages that could be caused by manipulation or using a non-standard tools""" |
| 29 | parts = location.location.parts |
| 30 | |
| 31 | if len(parts) < 3 or location.location.name != "WHEEL": |
| 32 | return |
| 33 | elif not parts[-2].endswith(".dist-info"): |
| 34 | return |
| 35 | |
| 36 | wheel_root = location.location.parents[1].absolute() |
| 37 | dist_info = location.location.parents[0] |
| 38 | |
| 39 | required_files = ("WHEEL", "METADATA", "RECORD") |
| 40 | for x in required_files: |
| 41 | if not (dist_info / x).is_file(): |
| 42 | continue |
| 43 | |
| 44 | record_entries = set() |
| 45 | record_path = dist_info / "RECORD" |
| 46 | |
| 47 | if not record_path.exists(): |
| 48 | yield Detection( |
| 49 | detection_type="Wheel", |
| 50 | location=location.location, |
| 51 | score = get_score_or_default("wheel-records-missing", 100), |
| 52 | message = f"Wheel anomaly, RECORD file is missing in dist-info", |
| 53 | tags = {"anomaly", "wheel", "wheel_missing_records"}, |
| 54 | signature = f"wheel#missing_records#{location.strip(record_path)}" |
| 55 | ) |
| 56 | return |
| 57 | |
| 58 | try: |
| 59 | with record_path.open(mode="r", newline=os.linesep) as rfd: |
| 60 | records_content = rfd.read() |
| 61 | except UnicodeDecodeError: |
| 62 | with record_path.open(mode="rb") as rfd: |
| 63 | records_raw: bytes = rfd.read() |
| 64 | try: |
| 65 | records_encoding = chardet.detect(records_raw)["encoding"] |
| 66 | records_content = records_raw.decode(records_encoding) |
| 67 | except (TypeError, UnicodeDecodeError): |
| 68 | yield Detection( |
| 69 | detection_type="Wheel", |
| 70 | location=location.location, |
| 71 | message="Unable to decode the wheel RECORDs file", |
| 72 | tags={"anomaly", "wheel", "unicode_decode_error"}, |
| 73 | signature=f"wheel#record_decode_err#{location.strip(record_path)}" |
| 74 | ) |
| 75 | return |
| 76 | |
| 77 | records_io = io.StringIO(records_content) |
| 78 | reader = csv.reader(records_io, delimiter=",", quotechar='"') |
| 79 | for record in reader: |
| 80 | full_pth = wheel_root.joinpath(record[0]) |
| 81 | |
| 82 | if not full_pth.exists(): |
| 83 | yield Detection( |
| 84 | location=location.location, |
nothing calls this directly
no test coverage detected