| 201 | |
| 202 | |
| 203 | def diff_archive(diff) -> AnalyzerReturnType: |
| 204 | if diff.operation not in "RM": |
| 205 | return |
| 206 | elif not (diff.a_scan.location.is_file() and diff.b_scan.location.is_file()): |
| 207 | return |
| 208 | elif diff.a_scan.metadata["md5"] == diff.b_scan.metadata["md5"]: |
| 209 | return |
| 210 | |
| 211 | a_hits = list(archive_analyzer(location=diff.a_scan)) |
| 212 | a_locations = [x for x in a_hits if type(x) == ScanLocation] |
| 213 | a_hits = [x for x in a_hits if type(x) != ScanLocation] |
| 214 | |
| 215 | b_hits = list(archive_analyzer(location=diff.b_scan)) |
| 216 | b_locations = [x for x in b_hits if type(x) == ScanLocation] |
| 217 | b_hits = [x for x in b_hits if type(x) != ScanLocation] |
| 218 | |
| 219 | # Yield all anomaly detections |
| 220 | yield from a_hits |
| 221 | yield from b_hits |
| 222 | |
| 223 | # Check if we should recurse diff into archives |
| 224 | if len(a_locations) == 0 and len(b_locations) == 0: |
| 225 | return |
| 226 | |
| 227 | # Create a new scan location |
| 228 | new_a_location = a_locations[0] if a_locations else diff.a_scan |
| 229 | new_b_location = b_locations[0] if b_locations else diff.b_scan |
| 230 | new_a_location.metadata["b_scan_location"] = new_b_location |
| 231 | yield new_a_location |