| """One plain sentence per draft saying what the body does, plus the policy name. |
| |
| Many admins read only this sentence, so it describes the body, not the request. |
| """ |
| from __future__ import annotations |
|
|
| from .model import Draft |
| from .platform import catalog |
| from .platform.context import PlatformContext |
|
|
| ACTION_PHRASE = {'block': 'Block', 'warn': 'Warn about', 'redact': 'Redact', 'log': 'Log', |
| 'alert': 'Alert on', 'remediate': 'Recall', 'delete': 'Delete'} |
|
|
|
|
| def _rules(d: Draft, ctx: PlatformContext, ids: list[str]) -> str: |
| names = [] |
| for rid in ids: |
| if rid == '$NEW_RULE' and d.ruleDecision.new_rule: |
| names.append(f'new rule "{d.ruleDecision.new_rule["name"]}"') |
| elif ctx.rule(rid): |
| names.append(f'"{ctx.rule(rid).name}"') |
| return ', '.join(names) if names else 'no rule' |
|
|
|
|
| def _what(d: Draft, ctx: PlatformContext) -> str: |
| """A short name for the data a draft looks for: "Aadhaar, PAN" or "Payment Card Data".""" |
| rd = d.ruleDecision |
| if rd.mode == 'superset' and rd.rule_ids and ctx.rule(rd.rule_ids[-1]): |
| asked = [e for e in ctx.rule(rd.rule_ids[-1]).entities if e not in rd.extras] |
| return _list([_short(e) for e in asked], 3) |
| if rd.new_rule: |
| return rd.new_rule['name'].replace('Copilot - ', '') |
| names = [ctx.rule(r).name for r in rd.rule_ids if ctx.rule(r)] |
| if len(names) > 2: |
| return 'sensitive data' |
| return _list(names, 2) if names else 'data' |
|
|
|
|
| def _short(entity: str) -> str: |
| """'Aadhaar Number' -> 'Aadhaar', 'PAN Card' -> 'PAN', 'Credit Card Number' -> 'Credit Card'.""" |
| label = catalog.entity_label(entity) |
| return {'PAN Card': 'PAN', 'Card CVV': 'CVV', 'US SSN': 'SSN'}.get(label, label.replace(' Number', '')) |
|
|
|
|
| def _list(items: list[str], n: int = 4) -> str: |
| items = list(items) |
| if len(items) <= n: |
| return ', '.join(items) |
| return ', '.join(items[:n]) + f' and {len(items) - n} more' |
|
|
|
|
| def summarize(d: Draft, ctx: PlatformContext) -> tuple[str, str]: |
| """(summary sentence, policy name).""" |
| b = d.policyBody |
| t = d.policyType |
| parts: list[str] = [] |
| name_bits: list[str] = [] |
| if t == 'endpoint': |
| ac, ext = b['agentControl'], b['extensionControl'] |
| hosts = b['enabledSites'] |
| names = {s['host']: s['name'] for s in catalog.ai_sites()} |
| where = 'all AI sites' if set(hosts) == set(catalog.ai_hosts()) else _list([names.get(h, h) for h in hosts]) |
| if ext['detectionRuleIds']: |
| parts.append(f'{ACTION_PHRASE[b["action"]]} AI prompts on {where} that match ' |
| f'{_rules(d, ctx, ext["detectionRuleIds"])}') |
| short_where = 'AI' if where == 'all AI sites' else where |
| name_bits.append(f'{ACTION_PHRASE[b["action"]].split()[0]} {_what(d, ctx)} in {short_where}') |
| elif any(i.code == 'NO_DETECTION_RULES' for i in d.issues): |
| parts.append(f'{ACTION_PHRASE[b["action"]]} AI prompts on {where} (choose the data to look for)') |
| name_bits.append('AI prompts') |
| if not b['allowFileUpload']: |
| parts.append(f'block file uploads to {where}') |
| name_bits.append('No AI uploads') |
| if not b['allowCodePaste']: |
| parts.append(f'block pasting code into {where}') |
| name_bits.append('No code paste to AI') |
| switches = { |
| 'blockRemovableStorage': 'block USB storage', 'removableStorageReadOnly': 'make USB storage read-only', |
| 'blockPrint': 'block printing', 'blockBluetooth': 'block Bluetooth file transfer', |
| 'blockMobileSync': 'block phone sync', 'blockScreenCapture': 'block screenshots', |
| 'blockCloudUpload': 'block cloud storage uploads', 'blockPersonalEmail': 'block personal webmail', |
| 'blockFileSharing': 'block file-sharing sites', 'blockWebUpload': 'block web and FTP uploads', |
| 'blockDebugTools': 'block debug and proxy tools', 'watermarkPrints': 'watermark printed pages', |
| 'captureScreenshotOnViolation': 'take a screenshot when a policy is broken', |
| 'dlpScanBeforeTransfer': f'scan files before USB or cloud transfer (level {ac["sensitivityThreshold"]})', |
| } |
| for k, phrase in switches.items(): |
| if ac.get(k): |
| if k == 'blockCloudUpload': |
| svc = [s for s, f in (('Google Drive', 'blockGoogleDrive'), ('OneDrive', 'blockOneDrive'), |
| ('SharePoint', 'blockSharePoint'), ('Dropbox', 'blockDropbox'), |
| ('Box', 'blockBox'), ('iCloud', 'blockiCloud')) if ac.get(f)] |
| phrase = f'block uploads to {_list(svc)}' if svc else phrase |
| if k == 'blockPersonalEmail': |
| svc = [s for s, f in (('Gmail', 'blockGmail'), ('Outlook.com', 'blockOutlookPersonal'), |
| ('Yahoo Mail', 'blockYahooMail'), ('Proton Mail', 'blockProtonMail')) |
| if ac.get(f)] |
| phrase = f'block {_list(svc)}' if svc else phrase |
| parts.append(phrase) |
| name_bits.append(phrase[0].upper() + phrase[1:]) |
| if ac['allowedUsbDeviceIds']: |
| parts.append(f'except USB devices {_list(ac["allowedUsbDeviceIds"])}') |
| if ac['clipboardMode'] != 'off': |
| mode = {'audit': 'log copy-paste of personal data', |
| 'everywhere': 'clear the clipboard when it holds personal data', |
| 'targets': f'block pasting personal data into {_list(ac["pasteBlockedSites"] + ac["pasteBlockedApps"])}' |
| }[ac['clipboardMode']] |
| parts.append(mode) |
| name_bits.append('Clipboard protection') |
| for key, kind in (('webBlockListIds', 'websites'), ('appBlockListIds', 'apps')): |
| ids = ac[key] |
| if ids: |
| names_ = [] |
| for lid in ids: |
| nl = next((x for x in d.newBlockLists if x['placeholder'] == lid), None) |
| names_.append(f'"{nl["body"]["name"]}"' if nl else f'"{ctx.block_list(lid).name}"') |
| parts.append(f'block {kind} on {_list(names_)}') |
| name_bits.append(f'Block {_list([n.strip(chr(34)).replace("Copilot - Blocked sites: ", "").replace("Copilot - Blocked apps: ", "") for n in names_], 2)}') |
| if b['activeWebCategories']: |
| cats = b['activeWebCategories'] |
| parts.append(f'block websites in the {_list(cats)} categor{"y" if len(cats) == 1 else "ies"}') |
| name_bits.append(f'Block {_list(cats, 2)} sites') |
| ext_switches = {'enforceIncognito': 'block incognito windows', 'blockGuestMode': 'block guest mode', |
| 'blockOtherBrowsers': 'allow only Chrome, Edge and Brave', |
| 'blockDeveloperMode': 'block browser developer mode', 'blockDevTools': 'block browser developer tools', |
| 'blockDownloads': 'block browser downloads', |
| 'enforceExtensionAllowlist': 'allow only approved browser extensions'} |
| for k, phrase in ext_switches.items(): |
| if ext.get(k) and ext.get(k) != catalog.default_body('endpoint')['extensionControl'].get(k): |
| parts.append(phrase) |
| name_bits.append(phrase[0].upper() + phrase[1:]) |
| if not ext['monitorDownloads']: |
| parts.append('stop logging downloads') |
| sr = ac['signInRestriction'] |
| if sr['enabled']: |
| parts.append(f'allow sign-in only with {_list(sr["allowedDomains"])} accounts') |
| if sr['strictLogin']: |
| parts.append('refuse sign-in on websites that are not approved') |
| name_bits.append('Sign-in restriction') |
| ide = ac['ideControl'] |
| if ide.get('enabled'): |
| bits = [] |
| if ide.get('detectionRuleIds'): |
| bits.append(f'{ide["promptAction"]} prompts to AI coding tools that match ' |
| f'{_rules(d, ctx, ide["detectionRuleIds"])}') |
| if ide.get('blockedAiTools'): |
| tools = {t['key']: t['name'] for t in catalog.ide()['tools']} |
| bits.append(f'block {_list([tools.get(x, x) for x in ide["blockedAiTools"]])}') |
| if ide.get('providerControl') == 'allowlist': |
| provs = {p['id']: p['name'] for p in catalog.ide()['providers']} |
| bits.append(f'allow only {_list([provs.get(x, x) for x in ide.get("allowedProviders", [])], 5)} ' |
| f'in AI coding tools') |
| if ide.get('mcpControl') == 'denylist': |
| bits.append('block the named MCP servers') |
| if ide.get('commandAction') != 'warn': |
| bits.append(f'{ide["commandAction"]} shell commands run by AI agents') |
| if ide.get('contextAction') != 'warn': |
| bits.append(f'{ide["contextAction"]} AI agents reading sensitive files') |
| if ide.get('enforce') is False: |
| bits.append('monitor AI coding tools without enforcing') |
| if ide.get('storePromptContent'): |
| bits.append('store the text of prompts to AI coding tools') |
| if ide.get('interceptPrompts'): |
| bits.append('intercept prompts from AI command-line tools') |
| if ide.get('inspectResponses'): |
| bits.append('inspect AI responses') |
| if ide.get('claudeCodeHook'): |
| bits.append('use the Claude Code hook') |
| if ide.get('fileAccessAction', 'off') != 'off': |
| verb_ = {'close': 'close', 'warn': 'warn about', 'log': 'log'}[ide['fileAccessAction']] |
| bits.append(f'{verb_} AI tools that open sensitive files ({ide["fileAccessThreshold"].lower()} risk and up)') |
| if ide.get('inEditorAlerts') is False: |
| bits.append('turn off in-editor alerts') |
| if ide.get('discoverAiTools') is False: |
| bits.append('stop discovering AI tools') |
| parts += bits or ['turn on AI coding tool monitoring'] |
| name_bits.append('AI coding tools' + (': allow only ' + _list( |
| [p for p in ide.get('allowedProviders', [])], 3) if ide.get('providerControl') == 'allowlist' else '')) |
| if b['testMode']: |
| parts.append('in test mode (nothing is enforced)') |
| grp = ctx.group(b['groupId']).name if b['groupId'] != 'default' and ctx.group(b['groupId']) else None |
| scope = f'for {grp}' if grp else 'for everyone' |
| else: |
| cfg = b['channels'][t] |
| label = catalog.TYPE_LABEL[t] |
| rules = cfg.get('detectionRuleIds') or cfg.get('sharingDetectionRuleIds') or [] |
| if t == 'email': |
| act = ACTION_PHRASE[cfg['action']] |
| if rules: |
| parts.append(f'{act} emails that match {_rules(d, ctx, rules)}') |
| if cfg.get('allowedDomains'): |
| parts.append(f'except mail only to {_list(cfg["allowedDomains"])}') |
| if cfg.get('blockedDomains'): |
| parts.append(f'alert on every email to {_list(cfg["blockedDomains"])}') |
| if cfg.get('checkBulkForwarding'): |
| parts.append(f'flag anyone sending {cfg["bulkForwardThreshold"]}+ emails an hour') |
| if cfg.get('scheduleRestrict'): |
| parts.append(f'treat sensitive mail outside {cfg["allowedHoursStart"]}:00-{cfg["allowedHoursEnd"]}:00 UTC ' |
| f'as high risk') |
| name_bits.append(f'{act.split()[0]} {_what(d, ctx)} in email' if rules else 'Email monitoring') |
| elif t == 'slack': |
| act = 'Delete' if cfg['action'] == 'delete' else 'Log' |
| parts.append(f'{act} Slack messages that match {_rules(d, ctx, rules)}') |
| name_bits.append(f'{act} {_what(d, ctx)} in Slack') |
| elif t in ('teams', 'chat'): |
| parts.append(f'Record {label} messages that match {_rules(d, ctx, rules)}') |
| name_bits.append(f'Monitor {_what(d, ctx)} in {label}') |
| elif t in ('drive', 'm365files'): |
| if rules: |
| parts.append(f'Record {label} shares of files that match {_rules(d, ctx, rules)}') |
| else: |
| parts.append(f'Record risky {label} shares') |
| if cfg.get('autoRevokePublic'): |
| parts.append('remove public links') |
| if cfg.get('autoRevokeExternal'): |
| parts.append('remove external shares') |
| if cfg.get('trustedDomains'): |
| parts.append(f'except shares with {_list(cfg["trustedDomains"])}') |
| if cfg.get('excludedUsers'): |
| parts.append(f'except {_list(cfg["excludedUsers"])}') |
| dspm = cfg.get('dspm') or {} |
| if dspm.get('scanMode') == 'rules' and dspm.get('detectionRuleIds'): |
| parts.append(f'scan stored files for {_rules(d, ctx, dspm["detectionRuleIds"])}') |
| undo = ' + '.join(x for x, on in (('remove public links', cfg.get('autoRevokePublic')), |
| ('remove external shares', cfg.get('autoRevokeExternal'))) if on) |
| name_bits.append(f'{label}: {_what(d, ctx) if rules else "risky shares"}' + (f' ({undo})' if undo else '')) |
| else: |
| if cfg.get('scanMode') == 'rules': |
| parts.append(f'Scan {label} for {_rules(d, ctx, cfg["detectionRuleIds"])}') |
| else: |
| parts.append(f'Scan {label} for any sensitive data (discovery)') |
| if cfg.get('bulkRepoThreshold') and t == 'github': |
| parts.append(f'flag anyone cloning {cfg["bulkRepoThreshold"]}+ repositories an hour') |
| if (cfg.get('autoSync') or {}).get('enabled'): |
| parts.append(f're-scan changes every {cfg["autoSync"]["intervalMinutes"]} minutes') |
| name_bits.append(f'Scan {label} for {_what(d, ctx)}' if cfg.get('scanMode') == 'rules' |
| else f'Scan {label} (discovery)') |
| if t != 'email' and cfg.get('connectorIds'): |
| names_ = [c.name for c in ctx.connectors_for(t) if c.id in cfg['connectorIds']] |
| parts.append(f'in {_list(names_)}') |
| scope = 'for the whole organisation' |
| sentence = '; '.join(parts) if parts else 'No settings changed' |
| sentence = sentence[0].upper() + sentence[1:] + f', {scope}. Saved switched off.' |
| name = ' + '.join(dict.fromkeys(name_bits)) or catalog.TYPE_LABEL[t] |
| if t == 'endpoint' and b['groupId'] != 'default' and ctx.group(b['groupId']): |
| name += f' ({ctx.group(b["groupId"]).name})' |
| return sentence, _unique_name(name[:80], ctx) |
|
|
|
|
| def _unique_name(name: str, ctx: PlatformContext) -> str: |
| taken = {p.name.lower() for p in ctx.policies} | ctx.policyNames |
| out, n = name, 2 |
| while out.lower() in taken: |
| out = f'{name[:75]} ({n})' |
| n += 1 |
| return out |
|
|
