def endpoint_by_attack()

in detection_rules/devtools.py [0:0]

• 43 lines of code
• 16 McCabe index (conditional complexity)

def endpoint_by_attack(ctx: click.Context, pre: str, post: str, force: bool, remote: Optional[str] = 'origin'):
    """Rule diffs across tagged branches, broken down by ATT&CK tactics."""
    if not force:
        if not click.confirm(f'This will refresh tags and may overwrite local tags for: {pre} and {post}. Continue?'):
            ctx.exit(1)

    changed, new, deprecated = get_release_diff(pre, post, remote)
    oses = ('windows', 'linux', 'macos')

    def delta_stats(rule_map) -> List[dict]:
        stats = defaultdict(lambda: defaultdict(int))
        os_totals = defaultdict(int)
        tactic_totals = defaultdict(int)

        for rule_id, rule in rule_map.items():
            threat = rule.contents.data.get('threat')
            os_types = [i.lower() for i in rule.contents.data.get('tags') or [] if i.lower() in oses]
            if not threat or not os_types:
                continue

            if isinstance(threat[0], dict):
                tactics = sorted(set(e['tactic']['name'] for e in threat))
            else:
                tactics = ThreatMapping.flatten(threat).tactic_names
            for tactic in tactics:
                tactic_totals[tactic] += 1
                for os_type in os_types:
                    os_totals[os_type] += 1
                    stats[tactic][os_type] += 1

        # structure stats for table
        rows = []
        for tac, stat in stats.items():
            row = {'tactic': tac, 'total': tactic_totals[tac]}
            for os_type, count in stat.items():
                row[os_type] = count
            rows.append(row)

        rows.append(dict(tactic='total_by_os', **os_totals))

        return rows

    fields = ['tactic', 'linux', 'macos', 'windows', 'total']

    changed_stats = delta_stats(changed)
    table = Table.from_list(fields, changed_stats)
    click.echo(f'Changed rules {len(changed)}\n{table}\n')

    new_stats = delta_stats(new)
    table = Table.from_list(fields, new_stats)
    click.echo(f'New rules {len(new)}\n{table}\n')

    dep_stats = delta_stats(deprecated)
    table = Table.from_list(fields, dep_stats)
    click.echo(f'Deprecated rules {len(deprecated)}\n{table}\n')

    return changed_stats, new_stats, dep_stats