(row: dict[str, str], candidate: dict[str, str])
| 314 | |
| 315 | |
| 316 | def validate_result_row(row: dict[str, str], candidate: dict[str, str]) -> None: |
| 317 | scan_path = candidate["scan_path"] |
| 318 | if ( |
| 319 | row["scan_path"] != scan_path |
| 320 | or row["sha256"] != candidate["sha256"] |
| 321 | or row["size"] != candidate["size"] |
| 322 | or row["association_count"] != "1" |
| 323 | ): |
| 324 | raise ContractError(f"VT result is not bound to candidate object: {scan_path}") |
| 325 | if SHA256_RE.fullmatch(row["sha256"]) is None: |
| 326 | raise ContractError(f"VT result has invalid SHA-256: {scan_path}") |
| 327 | |
| 328 | completed = parse_nonnegative(row, "completed_engines", scan_path=scan_path) |
| 329 | total = parse_nonnegative(row, "total_engines", scan_path=scan_path) |
| 330 | malicious = parse_nonnegative(row, "malicious", scan_path=scan_path) |
| 331 | suspicious = parse_nonnegative(row, "suspicious", scan_path=scan_path) |
| 332 | # `total < completed` stays: that is an incoherent response. A low decisive |
| 333 | # count does NOT — how many engines answered is VirusTotal fleet |
| 334 | # availability on the day, not a property of this binary, and gating on it |
| 335 | # made an 8-target release fail on a candidate with zero detections. |
| 336 | if total < completed: |
| 337 | raise ContractError(f"VT result has incoherent engine coverage: {scan_path}") |
| 338 | if ANALYSIS_ID_RE.fullmatch(row["analysis_id"]) is None: |
| 339 | raise ContractError(f"VT result has invalid analysis id: {scan_path}") |
| 340 | for field in ("microsoft_engine_version", "microsoft_engine_update"): |
| 341 | if not row[field] or any(ord(character) < 32 for character in row[field]): |
| 342 | raise ContractError(f"VT result has incomplete Microsoft evidence: {scan_path}") |
| 343 | expected_url = f"https://www.virustotal.com/gui/file/{row['sha256']}/detection" |
| 344 | if row["virustotal_url"] != expected_url: |
| 345 | raise ContractError(f"VT result URL is not content-bound: {scan_path}") |
| 346 | |
| 347 | classification = row["policy_classification"] |
| 348 | if classification == "clean": |
| 349 | coherent = ( |
| 350 | malicious == 0 |
| 351 | and suspicious == 0 |
| 352 | and row["microsoft_category"] in {"undetected", "harmless"} |
| 353 | and row["microsoft_result"] == "" |
| 354 | ) |
| 355 | elif classification == "microsoft-ml": |
| 356 | coherent = ( |
| 357 | malicious == 1 |
| 358 | and suspicious == 0 |
| 359 | and row["microsoft_category"] == "malicious" |
| 360 | and bool(row["microsoft_result"]) |
| 361 | and row["microsoft_result"].endswith("!ml") |
| 362 | ) |
| 363 | else: |
| 364 | raise ContractError(f"VT result is a hard or unknown policy block: {scan_path}") |
| 365 | if not coherent: |
| 366 | raise ContractError(f"VT classification contradicts its result details: {scan_path}") |
| 367 | |
| 368 | |
| 369 | def load_results( |
no test coverage detected