Merge advisories based on their identifiers (advisory_id and aliases). Example: If two advisories share ``advisory_id`` or share an alias, they will be merged together.
(advisories)
| 903 | |
| 904 | |
| 905 | def get_merged_identifier_groups(advisories): |
| 906 | """ |
| 907 | Merge advisories based on their identifiers (advisory_id and aliases). |
| 908 | Example: If two advisories share ``advisory_id`` or share an alias, they will be merged together. |
| 909 | """ |
| 910 | from vulnerabilities.models import Group |
| 911 | |
| 912 | identifier_groups = defaultdict(set) |
| 913 | |
| 914 | advisories = list(advisories) |
| 915 | |
| 916 | for adv in advisories: |
| 917 | |
| 918 | identifier_groups[adv.advisory_id].add(adv) |
| 919 | |
| 920 | for alias in adv.aliases.values_list("alias", flat=True): |
| 921 | identifier_groups[alias].add(adv) |
| 922 | |
| 923 | groups = [set(advs) for advs in identifier_groups.values() if len(advs) > 1] |
| 924 | |
| 925 | merged = [] |
| 926 | |
| 927 | for group in groups: |
| 928 | group = set(group) |
| 929 | |
| 930 | i = 0 |
| 931 | while i < len(merged): |
| 932 | if group & merged[i]: |
| 933 | group |= merged[i] |
| 934 | merged.pop(i) |
| 935 | else: |
| 936 | i += 1 |
| 937 | |
| 938 | merged.append(group) |
| 939 | |
| 940 | all_grouped = set() |
| 941 | for g in merged: |
| 942 | all_grouped |= g |
| 943 | |
| 944 | for adv in advisories: |
| 945 | if adv not in all_grouped: |
| 946 | merged.append({adv}) |
| 947 | |
| 948 | final_groups: List[Group] = [] |
| 949 | |
| 950 | for group in merged: |
| 951 | identifiers = set() |
| 952 | for adv in group: |
| 953 | for alias in adv.aliases.all().order_by("alias"): |
| 954 | identifiers.add(alias) |
| 955 | |
| 956 | primary = max(group, key=lambda a: a.precedence if a.precedence is not None else -1) |
| 957 | |
| 958 | secondary = [a for a in group if a != primary] |
| 959 | |
| 960 | final_groups.append(Group(aliases=identifiers, primary=primary, secondaries=secondary)) |
| 961 | |
| 962 | return final_groups |