File size: 9,169 Bytes
375880c | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 | """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,
}
|