"""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 # the data words were vague and the set was guessed 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]: # Custom rules first: an admin who made "Aadhaar & PAN" wants it used. A rule that detects # something the admin excluded ("except GSTIN") is never a candidate. 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 # Rules the admin named are used as they are; anything else asked for is added on top. 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) # Category words plus single items ("card numbers and bank details"): the category's preset(s), # plus whatever the single items need on their own. 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) # Category words ("financial data", "PII") name presets directly. 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 # keyword / file type only: the rule must detect exactly that and nothing else 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, }