Run Yara rules on all input files recursively
(*, location: ScanLocation)
| 29 | |
| 30 | @Analyzer.ID("yara") |
| 31 | def analyze(*, location: ScanLocation): |
| 32 | """Run Yara rules on all input files recursively""" |
| 33 | start = time.time() |
| 34 | |
| 35 | try: |
| 36 | for m in rules.match(os.fspath(location.location), timeout=10): |
| 37 | strings = tuple(set(x[-1] for x in m.strings)) |
| 38 | score_each = m.meta.get("score_each", False) |
| 39 | rule_score = m.meta.get("score", 0) |
| 40 | if score_each: |
| 41 | total_score = len(strings)*rule_score |
| 42 | else: |
| 43 | total_score = rule_score |
| 44 | |
| 45 | yield Detection( |
| 46 | detection_type = "YaraMatch", |
| 47 | message = f"Yara match '{m.rule}' signature", |
| 48 | signature = f"yara#{str(location)}#{m.rule}#{hash(strings)}", |
| 49 | location = location.location, |
| 50 | score = total_score, |
| 51 | extra = { |
| 52 | "rule": m.rule, |
| 53 | "strings": strings, |
| 54 | "meta": m.meta, |
| 55 | }, |
| 56 | tags = set(m.tags) |
| 57 | ) |
| 58 | except yara.Error as exc: |
| 59 | yield Detection( |
| 60 | detection_type="YaraError", |
| 61 | message=exc.args[0], |
| 62 | signature=f"yara_error#{str(location)}", |
| 63 | location=location.location, |
| 64 | tags = {"yara_error"} |
| 65 | ) |
| 66 | |
| 67 | end = time.time() - start |
| 68 | if end >= 1: |
| 69 | logger.info(f"Yara scan of {str(location)} took {end} s") |