| """Which detection rule(s) a policy should reference.
|
|
|
| Policies never hold data types themselves: they point at DetectionRule rows. To avoid
|
| filling the tenant's rule list with near-duplicates, an existing rule is preferred:
|
|
|
| reuse a rule detects exactly what was asked (custom rules before built-in ones)
|
| superset a rule detects what was asked plus at most 3 other things (the admin is asked)
|
| multi two or more rules together detect exactly what was asked
|
| create nothing fits: a new rule is proposed, created before the policy
|
| none the request names no data
|
| """
|
| from __future__ import annotations
|
|
|
| from dataclasses import dataclass, field
|
|
|
| from .interpret import Data
|
| from .lexicon import load
|
| from .platform import catalog
|
| from .platform.context import PlatformContext, Rule
|
|
|
| NEW_RULE_ID = '$NEW_RULE'
|
| MAX_EXTRAS = 3
|
|
|
|
|
| @dataclass
|
| class RuleDecision:
|
| mode: str = 'none'
|
| rule_ids: list[str] = field(default_factory=list)
|
| extras: list[str] = field(default_factory=list)
|
| new_rule: dict | None = None
|
| alternative: dict | None = None
|
| assumed: bool = False
|
|
|
| def to_json(self) -> dict:
|
| d = {'mode': self.mode, 'ruleIds': list(self.rule_ids)}
|
| if self.extras:
|
| d['extraEntities'] = list(self.extras)
|
| if self.new_rule:
|
| d['newRule'] = self.new_rule
|
| if self.alternative:
|
| d['alternative'] = self.alternative
|
| if self.assumed:
|
| d['assumed'] = True
|
| return d
|
|
|
|
|
| def _ordered(ctx: PlatformContext, minus: set[str] = frozenset()) -> list[Rule]:
|
|
|
|
|
| rules = [r for r in ctx.rules if not r.builtin] + [r for r in ctx.rules if r.builtin]
|
| return [r for r in rules if not (set(r.entities) & minus)]
|
|
|
|
|
| def resolve(data: Data, ctx: PlatformContext, answer: object = None) -> RuleDecision:
|
| if data.empty():
|
| return RuleDecision()
|
| lx = load()
|
| E = data.entity_set()
|
| K = sorted({k.lower() for k in data.keywords})
|
| F = sorted(set(data.file_types))
|
| refs = [rid for rid in data.rule_refs if ctx.rule(rid)]
|
| minus = set(data.minus)
|
| assumed = data.assumed
|
|
|
|
|
| if refs and not (E or K or F):
|
| return RuleDecision('reuse', refs)
|
|
|
| create = _new_rule(E, K, F, data, ctx)
|
| if answer == 'create' and create:
|
| return RuleDecision('create', refs + [NEW_RULE_ID], new_rule=create)
|
| if isinstance(answer, list) and answer:
|
| ids = [a for a in answer if isinstance(a, str) and ctx.rule(a)]
|
| if ids:
|
| return RuleDecision('reuse' if len(ids) == 1 else 'multi', refs + ids)
|
|
|
|
|
|
|
| if data.groups and data.entities and not data.minus and not K and not F and answer is None:
|
| group_rules: list[str] = []
|
| for g in data.groups:
|
| for rid in lx.data_groups[g]['rules']:
|
| if rid not in group_rules:
|
| group_rules.append(rid)
|
| covered = {e for rid in group_rules for e in (ctx.rule(rid).entities if ctx.rule(rid) else [])}
|
| rest = [e for e in data.entities if e not in covered]
|
| if not rest:
|
| data = Data(groups=list(data.groups), rule_refs=list(data.rule_refs), severity=data.severity)
|
| else:
|
| sub = resolve(Data(entities=rest, severity=data.severity), ctx)
|
| ids = refs + group_rules + [x for x in sub.rule_ids if x not in group_rules]
|
| mode = sub.mode if sub.mode in ('superset', 'create') else 'multi'
|
| return RuleDecision(mode, ids, extras=sub.extras, new_rule=sub.new_rule, assumed=assumed,
|
| alternative=sub.alternative)
|
|
|
|
|
| if data.groups and not data.entities and not data.minus and not K and not F:
|
| group_rules: list[str] = []
|
| for g in data.groups:
|
| for rid in lx.data_groups[g]['rules']:
|
| if rid not in group_rules:
|
| group_rules.append(rid)
|
| if len(group_rules) == 1:
|
| exact = _exact(set(E), K, F, ctx, minus)
|
| rid = exact.id if exact else group_rules[0]
|
| return RuleDecision('reuse', refs + [rid], assumed=assumed)
|
| exact = _exact(set(E), K, F, ctx, minus)
|
| if exact:
|
| return RuleDecision('reuse', refs + [exact.id], assumed=assumed)
|
| return RuleDecision('multi', refs + group_rules, assumed=assumed,
|
| alternative={'mode': 'create', 'newRule': create} if create else None)
|
|
|
| exact = _exact(set(E), K, F, ctx, minus)
|
| if exact:
|
| return RuleDecision('reuse', refs + [exact.id], assumed=assumed)
|
|
|
| if E:
|
| sup = _superset(set(E), K, F, ctx, minus)
|
| if sup:
|
| rule, extras = sup
|
| return RuleDecision('superset', refs + [rule.id], extras=extras, assumed=assumed,
|
| alternative={'mode': 'create', 'newRule': create} if create else None)
|
| if not K and not F:
|
| multi = _multi(set(E), ctx, minus)
|
| if multi:
|
| return RuleDecision('multi', refs + [r.id for r in multi], assumed=assumed,
|
| alternative={'mode': 'create', 'newRule': create} if create else None)
|
|
|
| if create is None:
|
| return RuleDecision('reuse', refs) if refs else RuleDecision()
|
| return RuleDecision('create', refs + [NEW_RULE_ID], new_rule=create, assumed=assumed,
|
| alternative=_broader(set(E), ctx))
|
|
|
|
|
| def _exact(E: set[str], K: list[str], F: list[str], ctx: PlatformContext, minus: set[str]) -> Rule | None:
|
| for r in _ordered(ctx, minus):
|
| if E:
|
| if set(r.entities) != E:
|
| continue
|
| if K and not set(K) <= set(r.keywords):
|
| continue
|
| if F and not set(F) <= set(r.fileTypes):
|
| continue
|
| return r
|
|
|
| if r.entities:
|
| continue
|
| if set(r.keywords) == set(K) and set(r.fileTypes) == set(F):
|
| return r
|
| return None
|
|
|
|
|
| def _superset(E: set[str], K: list[str], F: list[str], ctx: PlatformContext, minus: set[str]):
|
| best = None
|
| for r in _ordered(ctx, minus):
|
| ents = set(r.entities)
|
| if not E < ents:
|
| continue
|
| if K and not set(K) <= set(r.keywords):
|
| continue
|
| if F and not set(F) <= set(r.fileTypes):
|
| continue
|
| extras = [e for e in r.entities if e not in E]
|
| if len(extras) > MAX_EXTRAS:
|
| continue
|
| if best is None or len(extras) < len(best[1]):
|
| best = (r, extras)
|
| return best
|
|
|
|
|
| def _multi(E: set[str], ctx: PlatformContext, minus: set[str]) -> list[Rule] | None:
|
| """Two rules that together detect exactly E (each one only things in E)."""
|
| inside = [r for r in _ordered(ctx, minus) if r.entities and set(r.entities) <= E]
|
| for a in range(len(inside)):
|
| for b in range(a + 1, len(inside)):
|
| if set(inside[a].entities) | set(inside[b].entities) == E:
|
| return [inside[a], inside[b]]
|
| return None
|
|
|
|
|
| def _broader(E: set[str], ctx: PlatformContext) -> dict | None:
|
| """A built-in rule that covers the request with more than MAX_EXTRAS extras, offered as an option."""
|
| cands = [r for r in ctx.rules if r.builtin and E and E < set(r.entities)]
|
| if not cands:
|
| return None
|
| r = min(cands, key=lambda r: len(r.entities))
|
| return {'mode': 'reuse', 'ruleIds': [r.id], 'ruleName': r.name,
|
| 'extraEntities': [e for e in r.entities if e not in E]}
|
|
|
|
|
| def _new_rule(E: list[str], K: list[str], F: list[str], data: Data, ctx: PlatformContext) -> dict | None:
|
| if not (E or K or F):
|
| return None
|
| labels = [catalog.entity_label(e) for e in E] + [f'"{k}"' for k in K] + [f.upper() + ' files' for f in F]
|
| shown = ', '.join(labels[:3]) + (f' +{len(labels) - 3} more' if len(labels) > 3 else '')
|
| name = f'Copilot - {shown}'[:100]
|
| taken = {r.name.lower() for r in ctx.rules}
|
| base, n = name, 2
|
| while name.lower() in taken:
|
| name = f'{base[:94]} ({n})'
|
| n += 1
|
| hi = catalog.high_risk()
|
| severity = data.severity or ('HIGH' if any(e in hi for e in E) else 'MEDIUM')
|
| return {
|
| 'name': name,
|
| 'description': 'Created by Policy Copilot.',
|
| 'entities': list(E),
|
| 'keywords': [k.lower() for k in K],
|
| 'fileTypes': list(F),
|
| 'severity': severity,
|
| }
|
|
|
