"""From matched items to Requests: what the admin asked for, still independent of how the platform stores it. Steps: intent -> clauses (main and exception) -> verb with negation -> content -> policy type(s). Nothing here knows field names of a policy body; adapters do that. """ from __future__ import annotations import copy import re from dataclasses import dataclass, field from .lexicon import Lexicon, load from .matcher import Item from .platform import catalog from .platform.context import PlatformContext from .text import is_domain STRONG_VERBS = ('BLOCK', 'WARN', 'REDACT', 'ALERT', 'LOG', 'DELETE', 'REMEDIATE', 'REVOKE', 'ALLOW', 'READONLY', 'UNSUPPORTED') # When one clause has several strong verbs ("monitor and alert"), the strongest is meant. VERB_STRENGTH = {'ALLOW': 0, 'LOG': 1, 'ALERT': 2, 'WARN': 3, 'READONLY': 4, 'REDACT': 5, 'REVOKE': 6, 'REMEDIATE': 6, 'DELETE': 7, 'BLOCK': 8, 'UNSUPPORTED': 9} SENTENCE_BREAKS = {'.', ';', '!', '?', '\n'} SPLIT_WORDS = {'but', 'and', 'also', 'then', 'while', 'plus', 'additionally', 'or', 'whereas', 'however'} SOFT_WORDS = {'only', 'just', 'simply', 'also', 'please', 'should', 'to', 'we', 'i', 'you', 'they', 'it'} GENERIC_PEOPLE = {'users', 'user', 'people', 'employees', 'employee', 'staff', 'members', 'folks', 'guys', 'everyone', 'anyone', 'someone', 'somebody', 'anybody', 'person', 'persons', 'workers'} DATA_NOUNS = {'documents', 'document', 'docs', 'doc', 'files', 'file', 'data', 'records', 'record', 'information', 'info', 'reports', 'report', 'sheets', 'spreadsheets', 'details', 'folders', 'folder', 'contracts'} KEYWORD_LEADS = {'word', 'words', 'keyword', 'keywords', 'term', 'terms', 'phrase', 'phrases', 'labelled', 'labeled', 'tagged', 'titled', 'named', 'mentioning', 'string', 'text'} # Unknown words before a data noun that only describe it ("important files"), not what it is about. PLAIN_ADJECTIVES = {'important', 'internal', 'external', 'company', 'corporate', 'work', 'office', 'official', 'business', 'sensitive', 'private', 'secret', 'restricted', 'classified', 'critical', 'large', 'big', 'small', 'huge', 'old', 'new', 'latest', 'shared', 'local', 'personal', 'user', 'customer', 'client', 'employee', 'staff', 'team', 'other', 'random', 'certain', 'specific', 'those', 'such', 'their', 'your', 'our', 'his', 'her', 'any', 'all', 'some', 'many', 'multiple', 'several', 'various', 'different', 'same', 'these', 'protected', 'regulated', 'key', 'main'} CONTEXT_LEADS = {'during'} # "during zoom calls": a situation a policy cannot detect COPY_WORDS = {'copying', 'copy', 'copied', 'copies'} COPY_TARGETS = {'usb', 'mobile_sync', 'bluetooth', 'print', 'cloud_upload', 'dropbox', 'box', 'icloud'} THRESHOLD_LEADS = {'more', 'over', 'above', 'exceeds', 'exceeding', 'exceed', 'greater', 'beyond', 'least', 'than', '>', 'upto', 'max', 'maximum', 'limit', 'threshold', 'crosses', 'cross', 'past', 'after'} SIZE_UNITS = {'mb', 'gb', 'kb', 'tb', 'bytes', 'megabytes', 'gigabytes', 'kilobytes', 'percent', '%'} TIME_UNITS = {'minute': 1, 'minutes': 1, 'min': 1, 'mins': 1, 'hour': 60, 'hours': 60, 'hr': 60, 'hrs': 60, 'day': 1440, 'days': 1440, 'second': 0, 'seconds': 0, 'sec': 0, 'secs': 0, 'week': 10080, 'weeks': 10080} COUNTED_NOUNS = { 'emails': 'emails', 'email': 'emails', 'mails': 'emails', 'messages': 'messages', 'message': 'messages', 'recipients': 'recipients', 'recipient': 'recipients', 'people': 'recipients', 'persons': 'recipients', 'addresses': 'recipients', 'repos': 'repos', 'repo': 'repos', 'repositories': 'repos', 'repository': 'repos', 'records': 'records', 'rows': 'records', 'numbers': 'records', 'entries': 'records', 'items': 'records', 'files': 'files', 'file': 'files', 'documents': 'files', 'times': 'times', 'forwards': 'emails', 'objects': 'objects', 'pages': 'records', 'card': 'records', 'cards': 'records', 'ids': 'records', 'violations': 'times', 'violation': 'times', 'attempts': 'times', 'attempt': 'times', 'tries': 'times', 'incidents': 'times', 'offences': 'times', 'offenses': 'times', 'strikes': 'times', } # Words never worth reporting as "not understood". STOPWORDS = set('''a an the to of in on at for from by with into onto via over under than as is are was were be been being am do does did doing have has had having it its this that these those there here what which who whom whose when where why how all any each every some such no nor not only own same so too very can will just should would could may might must shall i me my we our ours you your he she they them their his her him if then else while until because about against between through during before after above below up down out off again further once also please kindly someone anyone somebody anybody everyone people person user users employee employees staff team member members able allowed allow let make sure ensure going go goes get gets got being want wants need needs like using use used data info information thing things stuff something anything everything whenever wherever whoever whatever okay ok hey hi hello thanks thank via within without across along around inside outside anything policy policies rule rules new create add set setup up our org organisation organization company companies employee's user's everyone's mein ko ke ki ka hai hain se pe par me bhi aur ya karo karna kare kar do dena chahiye hona sab koi liye lie usage use usages activity activities link links more less hour hours minute minutes day days details detail compliant compliance needful regulation regulations requirement requirements serial serials existing current also too as well and or but nor so yet folder folders directory directories dir dirs too either neither'''.split()) @dataclass class Data: entities: list[str] = field(default_factory=list) groups: list[str] = field(default_factory=list) # data group names from the vocabulary keywords: list[str] = field(default_factory=list) file_types: list[str] = field(default_factory=list) rule_refs: list[str] = field(default_factory=list) # custom rules named by the admin minus: list[str] = field(default_factory=list) # entities to leave out ("all PII except names") severity: str | None = None def empty(self) -> bool: return not (self.entities or self.groups or self.keywords or self.file_types or self.rule_refs) @property def assumed(self) -> bool: lx = load() return any(lx.data_groups.get(g, {}).get('assumed') for g in self.groups) def entity_set(self) -> list[str]: """Every entity the request names, groups expanded, exclusions removed, in mention order.""" lx = load() rules = {r['id']: r for r in catalog.builtin_rules()} out: list[str] = [] for e in self.entities: if e not in out: out.append(e) for g in self.groups: for rid in lx.data_groups[g]['rules']: for e in rules[rid]['entities']: if e not in out: out.append(e) return [e for e in out if e not in self.minus] @dataclass class Content: """Everything one clause (or one exception clause) mentions.""" data: Data = field(default_factory=Data) ai_sites: list[str] = field(default_factory=list) ai_any: bool = False prompt: bool = False channels: list[str] = field(default_factory=list) duals: list[str] = field(default_factory=list) controls: list[str] = field(default_factory=list) blists: list[str] = field(default_factory=list) web_categories: list[str] = field(default_factory=list) # vocabulary kinds or tenant category names sites: list[str] = field(default_factory=list) # named websites (block-list entries) apps: list[str] = field(default_factory=list) # named applications (.exe) ide_tools: list[str] = field(default_factory=list) ide_providers: list[str] = field(default_factory=list) ide_any: bool = False ide_features: list[str] = field(default_factory=list) groups: list[str] = field(default_factory=list) unknown_groups: list[str] = field(default_factory=list) excluded_groups: list[str] = field(default_factory=list) # "non-engineers": everyone but them all_users: bool = False connectors: list[tuple[str, str]] = field(default_factory=list) mods: list[str] = field(default_factory=list) domains: list[str] = field(default_factory=list) emails: list[str] = field(default_factory=list) paths: list[str] = field(default_factory=list) serials: list[str] = field(default_factory=list) ext_ids: list[str] = field(default_factory=list) guessed_domains: list[str] = field(default_factory=list) # "google domain" -> google.com (a guess) thresholds: list[tuple[int, str]] = field(default_factory=list) # (n, counted noun) interval_minutes: int | None = None hours: tuple[int | None, int | None] | None = None sizes: list[str] = field(default_factory=list) transfer: bool = False transfer_words: list[str] = field(default_factory=list) domain_words: list[str] = field(default_factory=list) # dual/control words written as a domain situations: list[str] = field(default_factory=list) # "during zoom calls": cannot be detected words: list[str] = field(default_factory=list) # unmatched content words def has_endpoint_surface(self) -> bool: return bool(self.ai_sites or self.ai_any or self.prompt or self.controls or self.blists or self.sites or self.web_categories or self.apps or self.ide_tools or self.ide_providers or self.ide_any or self.ide_features) def says_nothing(self) -> bool: """Empty even of unknown words: safe to ignore.""" return self.is_empty() and not self.words def is_empty(self) -> bool: return self.data.empty() and not (self.has_endpoint_surface() or self.channels or self.duals or self.groups or self.unknown_groups or self.domains or self.emails or self.serials or self.mods or self.connectors or self.thresholds or self.all_users) @dataclass class Request: type: str | None verb: str | None content: Content exceptions: list[Content] = field(default_factory=list) verb_conf: float = 1.0 verb_text: str = '' only: bool = False # "only allow X": an allowlist only_group_can: bool = False # "only HR can print": everyone else is restricted candidates: list[str] = field(default_factory=list) # when type is None: the plausible types group: str = 'default' group_unknown: str | None = None spans: list[tuple[int, int, str]] = field(default_factory=list) clause: str = '' dual_endpoint: list[str] = field(default_factory=list) # dual words used in their endpoint sense notes: list[str] = field(default_factory=list) consumed: list[int] = field(default_factory=list) # exceptions already applied (data exclusions) control_verb: str | None = None # verb for device/web switches when merged with a prompt request sentence: int = 0 negated: bool = False # the clause contained a negation ("... and not print") @dataclass class Understanding: intent: str requests: list[Request] fuzzy: list[tuple[str, str]] = field(default_factory=list) ignored: list[str] = field(default_factory=list) text: str = '' # ----------------------------------------------------------------------------- entry point def interpret(items: list[Item], text: str, ctx: PlatformContext) -> Understanding: lx = load() items = _reinterpret_teams(items) fuzzy = [(it.typed, it.text) for it in items if it.method == 'fuzzy'] intent = _intent(items) und = Understanding(intent=intent, requests=[], fuzzy=fuzzy, text=text) if intent != 'CREATE': return und clauses = _clauses(items) built: list[tuple[str | None, Content, list[Content], dict]] = [] nobody_else = bool(re.search(r'\b(nobody|no one|noone|no-one) else\b', text)) for main, excs, sno in clauses: verb, vconf, vtext, flags = _verb(main) flags['sentence'] = sno content = _content(main, ctx, lx) exc_contents = [_content(e, ctx, lx, exception=True) for e in excs] if content.excluded_groups: # "for non-engineers": everyone, that group excepted exc_contents.append(Content(groups=list(content.excluded_groups))) content.excluded_groups, content.all_users = [], True if (verb in (None, 'ALLOW', 'ENABLE') and content.all_users and not _has_surface(content) and len(exc_contents) == 1 and exc_contents[0].groups and exc_contents[0].has_endpoint_surface()): # "everyone except contractors can print": contractors may not content, exc_contents = exc_contents[0], [] verb, vconf = 'BLOCK', 0.8 if (flags.get('only_group_can') or (nobody_else and verb in ('ALLOW', 'ENABLE'))) and content.groups: # "only HR can print" = restrict everyone, HR excepted. exc_contents.append(Content(groups=list(content.groups))) content.groups = [] verb, vconf = 'BLOCK', 0.8 built.append((verb, content, exc_contents, dict(flags, vconf=vconf, vtext=vtext, items=main))) built = _merge_and_inherit(built) for verb, content, excs, flags in built: und.ignored += content.words if verb == 'ALLOW' and _looks_like_exception(content) and und.requests: # "Block USB for everyone. Allow HR." -> HR is an exception to the previous request. und.requests[-1].exceptions.append(content) continue for req in _route(verb, content, excs, flags, ctx, lx): req.sentence = flags.get('sentence', 0) und.requests.append(req) und.requests = _merge_endpoint(und.requests) und.ignored = list(dict.fromkeys(w for w in und.ignored))[:8] return und def _switches_only(r: Request) -> bool: """A data-less endpoint request that only flips switches (no AI sites to block or watch).""" c = r.content return not (c.ai_sites or c.ai_any or c.prompt) def _merge_endpoint(reqs: list[Request]) -> list[Request]: """One sentence, one form: "block cards in ChatGPT and disable USB for finance" is one policy, and "scan GitHub for keys and alert on mass cloning" is one GitHub policy (a channel applies only one). Requests merge when they are the same type, sentence and group and at most one of them names data (two prompt actions cannot share one policy).""" out: list[Request] = [] for r in reqs: target = next((o for o in out if o.type == r.type and o.type is not None and o.sentence == r.sentence and o.group == r.group and o.group_unknown == r.group_unknown and (o.content.data.empty() or r.content.data.empty()) and (o.verb != 'ALLOW' or o.only) and (r.verb != 'ALLOW' or r.only) and (o.type != 'endpoint' or _switches_only(o if o.content.data.empty() else r))), None) if target is None: out.append(r) continue base, other = (target, r) if not target.content.data.empty() or r.content.data.empty() else (r, target) if base is r: out[out.index(target)] = r oc, bc = other.content, base.content for f in ('controls', 'blists', 'web_categories', 'sites', 'apps', 'ide_tools', 'ide_providers', 'ide_features', 'mods', 'serials', 'paths', 'domains', 'ai_sites', 'emails', 'thresholds', 'connectors', 'sizes', 'transfer_words', 'ext_ids'): for v in getattr(oc, f): _add(getattr(bc, f), v) bc.ai_any = bc.ai_any or oc.ai_any bc.ide_any = bc.ide_any or oc.ide_any bc.interval_minutes = bc.interval_minutes or oc.interval_minutes bc.hours = bc.hours or oc.hours if oc.controls or oc.blists or oc.web_categories or oc.sites or oc.apps: base.control_verb = other.verb base.exceptions += other.exceptions base.dual_endpoint += [d for d in other.dual_endpoint if d not in base.dual_endpoint] base.spans += other.spans base.clause = f'{base.clause}; {other.clause}' return out # ----------------------------------------------------------------------------- intent def _sig(items: list[Item]) -> list[Item]: return [it for it in items if not it.is_punct and it.text not in ('please', 'kindly', 'hey', 'hi', 'can', 'you', 'could', 'would', 'pls', 'plz')] def _intent(items: list[Item]) -> str: sig = _sig(items) if not sig: return 'NONE' actionable = any(it.has('verb', 'neg', 'control', 'entity', 'dgroup', 'channel', 'dual', 'ai_site', 'blist', 'site', 'app', 'ide_tool', 'ide_any', 'ide_provider', 'ide_feature', 'rule_ref') for it in sig) first = sig[0] head = sig[:3] has_policy_noun = any(it.has('policy_noun') for it in sig) if first.text in ('why', 'explain') or (first.get('mgmt') and first.get('mgmt').value == 'EXPLAIN' and first.text not in ('how do', 'how can i')): return 'EXPLAIN' for it in head: m = it.get('mgmt') if m and m.value == 'LIST' and not it.has('verb'): if has_policy_noun or not actionable: return 'LIST' # "make the existing USB policy apply to HR": about a policy that already exists for k, it in enumerate(sig): if it.has('policy_noun') and any(w.text in ('existing', 'current', 'already', 'old', 'previous') for w in sig[max(0, k - 3):k]): return 'MODIFY' # "turn off USB blocking", "remove the printing restriction": undoing something already in place undo = {'turn off', 'switch off', 'disable', 'remove', 'lift', 'undo', 'stop', 'cancel', 'drop'} for k, it in enumerate(sig): if it.text in undo and any(w.text in ('blocking', 'block', 'restriction', 'restrictions', 'ban', 'blocks') for w in sig[k + 2:k + 5]) and any(w.has('control', 'blist', 'site', 'app', 'ai_site', 'ide_tool') for w in sig[k + 1:k + 4]): return 'REMOVE' # "delete / disable / edit the USB policy" manages an existing policy. for k, it in enumerate(sig): m = it.get('mgmt') if m and m.value in ('MODIFY', 'REMOVE'): window = sig[k + 1:k + 5] if any(w.has('policy_noun') for w in window): return m.value if not actionable: if any(it.has('mod', 'group', 'number', 'domain', 'quoted') for it in sig): return 'CREATE' # something to ask about return 'NONE' return 'CREATE' def _reinterpret_teams(items: list[Item]) -> list[Item]: """'teams' after a group name or a quantifier means teams of people, not Microsoft Teams.""" out = [] for k, it in enumerate(items): if it.text == 'teams' and it.has('channel'): prev = items[k - 1] if k else None if prev is not None and (prev.has('group', 'all_users') or prev.text in ( 'all', 'both', 'these', 'those', 'other', 'our', 'different', 'various', 'multiple', 'some', 'many', 'every', 'two', 'three', 'the')): it = copy.copy(it) it.senses = [s for s in it.senses if s.slot != 'channel'] out.append(it) return out # ----------------------------------------------------------------------------- clauses def _is_strong_verb(it: Item) -> bool: v = it.get('verb') return bool(v and v.value in STRONG_VERBS) SURFACE_SLOTS = ('ai_site', 'ai_any', 'prompt', 'channel', 'dual', 'control', 'blist', 'ide_tool', 'ide_any', 'ide_provider', 'group', 'all_users') PREPOSITIONS = {'in', 'on', 'into', 'to', 'for', 'from', 'via', 'through', 'inside', 'onto', 'over', 'across', 'within', 'mein', 'pe', 'par'} def _clauses(items: list[Item]) -> list[tuple[list[Item], list[list[Item]], int]]: """Split into (main clause, [exception clauses], sentence number) triples.""" sentences: list[list[Item]] = [[]] for it in items: if it.is_punct and it.text in SENTENCE_BREAKS: if sentences[-1]: sentences.append([]) continue sentences[-1].append(it) sentences = [s for s in sentences if s] out: list[tuple[list[Item], list[list[Item]], int]] = [] for sno, sent in enumerate(sentences): # 1) split into action clauses at "but / and / , " when a verb already appeared parts: list[list[Item]] = [[]] seen_verb = False for k, it in enumerate(sent): if (seen_verb and _is_split_verb(it) and not it.has('exc') and _preceded_by_split(sent, k) and not _in_exception(parts[-1])): parts.append([]) seen_verb = False parts[-1].append(it) if _is_split_verb(it) or (it.has('neg') and not it.has('exc')): seen_verb = True # 2) inside each part, carve out exception segments for part in parts: main: list[Item] = [] excs: list[list[Item]] = [] in_exc = False for k, it in enumerate(part): if it.has('exc'): in_exc = True excs.append([]) continue if in_exc: # "except HR, and warn ..." - a split word followed by a strong verb ends it if it.text in SPLIT_WORDS and _next_strong_verb(part, k): in_exc = False main.append(it) continue # "except GSTIN in chatgpt": once the exception names something, a preposition # followed by a surface or a group starts the main clause again if (it.text in PREPOSITIONS and _substantive(excs[-1]) and k + 1 < len(part) and part[k + 1].has(*SURFACE_SLOTS) and not part[k + 1].has('exc')): in_exc = False main.append(it) continue # "except HR, block ..." - a comma followed by a verb ends it; "except HR, Finance # and Sales" and "except chatgpt, claude and gemini" are lists and do not if (it.is_punct and it.text == ',' and _substantive(excs[-1]) and k + 1 < len(part) and (_is_strong_verb(part[k + 1]) or part[k + 1].has('neg'))): in_exc = False continue excs[-1].append(it) else: main.append(it) out.append((main, [e for e in excs if e], sno)) return out def _is_split_verb(it: Item) -> bool: """Verbs that start a new action clause after "and / but / ,".""" v = it.get('verb') return bool(v and (v.value in STRONG_VERBS or v.value == 'DISABLE')) _NOT_SUBSTANTIVE = {'group_noun', 'scope', 'all_users', 'transfer', 'neg', 'verb', 'mgmt', 'policy_noun'} def _substantive(items: list[Item]) -> bool: """Whether an exception clause already names something (not just "the user is ...").""" return any(x.senses and not all(s.slot in _NOT_SUBSTANTIVE for s in x.senses) for x in items) def _in_exception(part: list[Item]) -> bool: return any(it.has('exc') for it in part[-4:]) def _preceded_by_split(sent: list[Item], k: int) -> bool: j = k - 1 while j >= 0 and sent[j].text in ('only', 'just', 'also', 'simply', 'should', 'we', 'please'): j -= 1 if j < 0: return False return sent[j].text in SPLIT_WORDS or (sent[j].is_punct and sent[j].text in (',', '-', '&')) def _next_strong_verb(part: list[Item], k: int) -> bool: for it in part[k + 1:k + 4]: if _is_strong_verb(it): return True return False # ----------------------------------------------------------------------------- verbs def _verb(items: list[Item]) -> tuple[str | None, float, str, dict]: flags: dict = {} verbs: list[tuple[int, str, str]] = [] weak: list[tuple[int, str]] = [] negs = [k for k, it in enumerate(items) if it.has('neg')] for k, it in enumerate(items): v = it.get('verb') if not v: continue if v.value in STRONG_VERBS: verbs.append((k, v.value, it.text)) elif v.value in ('ENABLE', 'DISABLE'): weak.append((k, v.value)) conf = 1.0 # Negation: a negation cue governs the next verb, transfer word or thing within a few words. negated_verb: tuple[int, str] | None = None for n in negs: # "remind people not to share X": the negation is what the warning says, not the action. # "warn, don't block" is different: there the negation governs another verb. governs_verb = any(_is_strong_verb(it) for it in items[n + 1:n + 3]) if not governs_verb and any(it.get('verb') and it.get('verb').value in ('WARN', 'ALERT') for it in items[max(0, n - 4):n]): continue # Hindi puts the negation after the verb: "share nahi", "paste mat karna" back = [it for it in items[max(0, n - 2):n] if it.has('transfer')] if back and items[n].text in ('nahi', 'nahin', 'mat', 'na'): negated_verb = (n - 1, 'BLOCK') break window = items[n + 1:n + 6] neg_count = sum(1 for k in negs if n <= k < n + 6) for off, it in enumerate(window): k = n + 1 + off v = it.get('verb') if v and v.value in ('ALLOW', 'ENABLE'): negated_verb = (k, 'BLOCK') break if v and v.value in ('BLOCK', 'REDACT', 'WARN', 'DELETE', 'REVOKE', 'REMEDIATE'): negated_verb = (k, 'ALLOW') break if v and v.value in ('ALERT', 'LOG'): negated_verb = (k, 'ALLOW') break if it.has('transfer') or it.has('control', 'dual', 'entity', 'dgroup', 'ai_site', 'channel', 'blist', 'site', 'app', 'ide_tool'): negated_verb = (k, 'BLOCK') break if neg_count > 1: conf = 0.3 # "do not not allow": never guess if negated_verb: break if negated_verb: k, val = negated_verb verbs = [(i, v, t) for (i, v, t) in verbs if i != k] verbs.append((k, val, items[k].text)) flags['negated'] = True # "only allow X" / "allow only X": an allowlist for k, it in enumerate(items): v = it.get('verb') if v and v.value == 'ALLOW': near = [x.text for x in items[max(0, k - 2):k + 3]] if 'only' in near: flags['only'] = True # "only HR can print" for k, it in enumerate(items): if it.text == 'only' and k + 2 < len(items) and items[k + 1].has('group'): if items[k + 2].text in ('can', 'may', 'should', 'are', 'is', 'allowed', 'gets', 'get'): flags['only_group_can'] = True # "allow only HR to print", "only allow HR to use ChatGPT" v = it.get('verb') if v and v.value == 'ALLOW' and flags.get('only'): nxt = [x for x in items[k + 1:k + 4] if x.text not in ('only', 'the')] if nxt and nxt[0].has('group'): flags['only_group_can'] = True # "restrict / limit ChatGPT to the engineering team (only)": everyone else is blocked if it.text in ('restrict', 'restricted', 'limit', 'limited', 'restricting', 'limiting'): for j in range(k + 1, len(items) - 1): if items[j].text in ('to', 'to the', 'to all'): after = [x for x in items[j + 1:j + 3] if x.text != 'the'] if after and after[0].has('group'): flags['only_group_can'] = True break if not verbs: if weak: k, v = weak[0] return v, conf, items[k].text, flags return None, conf, '', flags distinct = {v for _, v, _ in verbs} if any(v == 'DISABLE' for _, v in weak) and distinct <= {'LOG', 'ALERT'} and all( t.endswith(('ing', 'ion', 'ions', 'alerts', 'logs')) for _, _, t in verbs): # "turn off AI coding tool monitoring": the noun names what is switched off k = next(i for i, v in weak if v == 'DISABLE') return 'DISABLE', conf, items[k].text, flags if 'READONLY' in distinct: # "read from USB, not write to it": the specific verb beats the negation it contains distinct = {'READONLY'} if 'ALLOW' in distinct and len(distinct) > 1 and not flags.get('only'): # "allow X but block Y" was split already; here both sit together: trust the restrictive one distinct.discard('ALLOW') best = max(distinct, key=lambda v: VERB_STRENGTH[v]) text = next(t for _, v, t in verbs if v == best) return best, conf, text, flags # ----------------------------------------------------------------------------- content def _content(items: list[Item], ctx: PlatformContext, lx: Lexicon, exception: bool = False) -> Content: c = Content() n = len(items) copy_verb = email_as_data = False for k, it in enumerate(items): if it.is_punct: continue s = {x.slot for x in it.senses} prev = items[k - 1] if k else None if is_domain(it.text) and s & {'dual', 'control', 'site', 'ai_site'}: # "gmail.com", "hotmail.com", "box.com": a vocabulary word that is also a domain _add(c.domains, it.text) if s & {'dual', 'control'}: _add(c.domain_words, it.text) nxt = items[k + 1] if k + 1 < n else None # modifiers and transfer words are flags: record them whatever else the word means if 'mod' in s: for v in it.values('mod'): _add(c.mods, v) if 'transfer' in s: c.transfer = True _add(c.transfer_words, it.text) if 'group' in s: if nxt is not None and nxt.text in DATA_NOUNS: # "HR documents", "finance files": describes data, not who the policy is for continue for v in it.values('group'): _add(c.excluded_groups if prev is not None and prev.text in ('non', 'non-') else c.groups, v) continue if 'rule_ref' in s: for v in it.values('rule_ref'): _add(c.data.rule_refs, v) continue if 'connector' in s: for v in it.values('connector'): _add(c.connectors, v) continue if 'entity' in s: for v in it.values('entity'): _add(c.data.entities, v) continue if 'dgroup' in s: for v in it.values('dgroup'): _add(c.data.groups, v) continue if 'ftype' in s: for v in it.values('ftype'): _add(c.data.file_types, v) continue if 'ide_tool' in s: for v in it.values('ide_tool'): _add(c.ide_tools, v) continue if 'ide_provider' in s: for v in it.values('ide_provider'): _add(c.ide_providers, v) continue if 'ai_site' in s: for v in it.values('ai_site'): _add(c.ai_sites, v) continue if 'ide_any' in s: c.ide_any = True continue if 'ide_feature' in s: for v in it.values('ide_feature'): _add(c.ide_features, v) continue if 'ai_any' in s: c.ai_any = True continue if 'prompt' in s: c.prompt = True continue if 'dual' in s: for v in it.values('dual'): _add(c.duals, v) continue if 'channel' in s: for v in it.values('channel'): if v == 'email' and _listed_with_data(items, k): email_as_data = True # "phone numbers and emails": e-mail addresses, maybe _add(c.channels, v) if 'transfer' in s: c.transfer = True continue if 'control' in s: for v in it.values('control'): if v == 'clipboard' and it.text in COPY_WORDS: copy_verb = True _add(c.controls, v) if v in ('file_uploads_ai', 'web_upload', 'cloud_upload'): _add(c.transfer_words, 'upload') if 'transfer' in s: c.transfer = True continue if 'blist' in s: for v in it.values('blist'): _add(c.blists, v) continue if 'wcat' in s: for v in it.values('wcat'): _add(c.web_categories, v) continue if 'site' in s: for v in it.values('site'): _add(c.sites, v) continue if 'app' in s: for v in it.values('app'): _add(c.apps, v) continue if 'all_users' in s: # bare "all" only counts as everyone right after a scope word ("for all") if it.text in ('all', 'sab') and not (prev and prev.has('scope')): continue c.all_users = True continue if 'mod' in s: for v in it.values('mod'): _add(c.mods, v) continue if 'transfer' in s: c.transfer = True continue if 'quoted' in s: q = it.get('quoted').value.strip().lower() g = _group_by_name(q, ctx) if g: _add(c.groups, g) elif q: _add(c.data.keywords, q) continue if 'domain' in s: d = it.get('domain').value _add(c.domains, d) continue if 'email_addr' in s: _add(c.emails, it.get('email_addr').value) continue if 'path' in s: _add(c.paths, it.get('path').value) continue if 'serial' in s: _add(c.serials, it.get('serial').value) continue if 'ext_id' in s: _add(c.ext_ids, it.get('ext_id').value) continue if 'time' in s: _hours(c, items, k) continue if 'number' in s: _number(c, items, k) continue if 'group_noun' in s and it.text not in GENERIC_PEOPLE: # "for the growth team": the word before the group noun names a group we do not know if prev is not None and not prev.senses and not prev.is_punct and prev.text not in STOPWORDS: _add(c.unknown_groups, prev.text) if prev.text in c.words: c.words.remove(prev.text) elif prev is not None and not prev.senses and prev.text == 'it': _add(c.unknown_groups, 'IT') # "the IT team": "it" is a name here, not a pronoun elif it.text in ('interns', 'contractors') and not ctx.groups: _add(c.unknown_groups, it.text) elif it.text in ('interns', 'contractors'): _add(c.unknown_groups, it.text) continue if not it.senses: w = it.text if exception and k == 0 and w not in STOPWORDS and len(w) >= 3 and not w.isdigit(): _add(c.unknown_groups, w) _add(c.words, w) continue # keyword leads: "containing the word salary", "files labelled 'restricted'" if prev is not None and prev.text in KEYWORD_LEADS and w not in STOPWORDS: _add(c.data.keywords, w) continue if prev is not None and prev.text in CONTEXT_LEADS and w not in STOPWORDS: # "during zoom calls", "while travelling": a situation no policy can detect rest = [x.text for x in items[k:] if not x.senses and not x.is_punct][:3] _add(c.situations, f'{prev.text} {" ".join(rest)}') continue if prev is not None and prev.text == 'every' and w in TIME_UNITS and TIME_UNITS[w]: c.interval_minutes = TIME_UNITS[w] # "every hour", "every day" continue if w in STOPWORDS or len(w) < 3 or w.isdigit(): continue if (nxt is not None and nxt.text in DATA_NOUNS and w not in PLAIN_ADJECTIVES and not exception and w.isalpha() and not c.situations): # "pricing documents", "payroll files": the word is what the data is about _add(c.data.keywords, w) continue if (prev is not None and 'ALLOW' in prev.values('verb') and nxt is not None and (nxt.text == 'to' or nxt.has('transfer') or nxt.has('verb')) and w.endswith('s')): # "let developers use ...", "allow designers to ...": people we have no group for _add(c.unknown_groups, w) continue # "for " right after a scope word, then the end or a group noun: a group name we lack if prev is not None and prev.has('scope') and (nxt is None or nxt.is_punct or nxt.has('group_noun') or nxt.text in ('and', 'or', 'only')): _add(c.unknown_groups, w) continue _add(c.words, w) # severity words if 'severity_high' in c.mods: c.data.severity = 'HIGH' elif 'severity_medium' in c.mods: c.data.severity = 'MEDIUM' elif 'severity_low' in c.mods: c.data.severity = 'LOW' # "except wardline, google and microsoft domain": bare names next to the word domain are domains if 'domain_word' in c.mods: for w in list(c.words): if w.replace('-', '').isalnum() and not w.isdigit() and w not in STOPWORDS: guess = f'{w}.com' _add(c.domains, guess) _add(c.guessed_domains, guess) c.words.remove(w) if w in c.unknown_groups: c.unknown_groups.remove(w) # remove unknown group candidates that turned out to be known things c.unknown_groups = [g for g in c.unknown_groups if g not in GENERIC_PEOPLE] if copy_verb and set(c.controls) & COPY_TARGETS: # "copying files to USB": copying is how the data leaves, not the clipboard switch c.controls.remove('clipboard') if email_as_data and (c.has_endpoint_surface() or len(c.channels) > 1 or c.duals): # "redact phone numbers and emails in AI prompts": the place is elsewhere; emails are data c.channels.remove('email') _add(c.data.entities, 'EMAIL') return c def _listed_with_data(items: list[Item], k: int) -> bool: """'X and emails' / 'emails, X': the word sits in a list of data types.""" def is_data(j: int) -> bool: return 0 <= j < len(items) and (items[j].has('entity') or items[j].has('dgroup')) joiners = {'and', 'or', ',', '&'} return ((k >= 2 and items[k - 1].text in joiners and is_data(k - 2)) or (k + 2 < len(items) and items[k + 1].text in joiners and is_data(k + 2))) def _group_by_name(name: str, ctx: PlatformContext) -> str | None: n = name.strip().lower() for g in ctx.groups: if g.name.lower() == n: return g.id return None def _add(lst: list, v) -> None: if v not in lst: lst.append(v) def _number(c: Content, items: list[Item], k: int) -> None: v = items[k].get('number').value if not isinstance(v, (int, float)): return nxt = items[k + 1].text if k + 1 < len(items) else '' nxt2 = items[k + 2].text if k + 2 < len(items) else '' prev = [it.text for it in items[max(0, k - 3):k]] if nxt in SIZE_UNITS: c.sizes.append(f'{v}{nxt}') return if nxt in TIME_UNITS: if 'every' in prev or 'each' in prev or any(p in ('interval', 'frequency') for p in prev): c.interval_minutes = int(v * TIME_UNITS[nxt]) if TIME_UNITS[nxt] else None elif (any(p in ('for', 'next', 'after') for p in prev[-2:]) and TIME_UNITS[nxt] >= 1) or \ (any(p == 'within' for p in prev[-2:]) and TIME_UNITS[nxt] >= 1440): _add(c.mods, 'timed') # "for 2 weeks", "after 3 days": a policy with an end or start date return # "in an hour" / "per hour" after the count is part of the threshold phrase, not a time noun = COUNTED_NOUNS.get(nxt) or COUNTED_NOUNS.get(nxt2) if not noun: for it in items[k + 1:k + 4]: if it.has('entity', 'dgroup'): noun = 'records' break if it.has('channel') and it.text in ('repos', 'repositories', 'repository', 'repo'): noun = 'repos' break if any(p in THRESHOLD_LEADS for p in prev) or noun: c.thresholds.append((int(v), noun or 'items')) def _hours(c: Content, items: list[Item], k: int) -> None: h = items[k].get('time').value prev = [it.text for it in items[max(0, k - 3):k]] start, end = c.hours or (None, None) if 'after' in prev or 'past' in prev or 'beyond' in prev: end = h elif 'before' in prev: start = h elif start is None: start = h else: end = h c.hours = (start, end) def _has_surface(c: Content) -> bool: return bool(c.has_endpoint_surface() or c.channels or c.duals) def _looks_like_exception(c: Content) -> bool: """A content that only names who/what is exempt, with no surface of its own.""" return bool(c.groups or c.domains or c.emails or c.serials) and not ( c.has_endpoint_surface() or c.channels or c.duals or not c.data.empty()) # ----------------------------------------------------------------------------- merging def _merge_and_inherit(built): """Clauses without a verb take the neighbour's; clauses with only a verb lend it forward.""" out = [] pending_verb = None for verb, content, excs, flags in built: if content.is_empty() and not excs: if verb == 'READONLY' and out: # "allow reading from usb but block writing": the trailing verb decides the clause before out[-1][0] = 'READONLY' continue if verb: pending_verb = (verb, flags) continue if verb is None and pending_verb: verb, f = pending_verb flags = dict(flags, vconf=f.get('vconf', 1.0), vtext=f.get('vtext', '')) pending_verb = None out.append([verb, content, excs, flags]) # A clause that names data but no surface takes the surface (and group) of the clause before it # in the same sentence: "block aadhaar in chatgpt but only warn for phone numbers". drop: set[int] = set() for k in range(1, len(out)): cur, prev = out[k][1], out[k - 1][1] if out[k][3].get('sentence') != out[k - 1][3].get('sentence'): continue words = {it.text for it in out[k][3].get('items', [])} if (out[k - 1][0] == 'ALLOW' and out[k][0] in ('BLOCK', 'DISABLE') and not _has_surface(cur) and cur.data.empty() and not cur.groups and words & {'else', 'rest', 'others', 'other'} and not (prev.groups or prev.unknown_groups)): # "Only allow Claude for AI coding, block everything else": an allowlist, said twice out[k - 1][3] = dict(out[k - 1][3], only=True) drop.add(k) continue if (out[k - 1][0] == 'ALLOW' and out[k][0] in ('BLOCK', 'DISABLE') and prev.ai_sites and cur.ai_any and not cur.ai_sites and cur.data.empty() and not (cur.groups or cur.unknown_groups) and words & {'else', 'rest', 'others', 'other'}): # "Only allow Claude for engineering, block other AI": the same allow-list, for the same people out[k - 1][3] = dict(out[k - 1][3], only=True) drop.add(k) continue if (_has_surface(cur) and (cur.channels or cur.duals) and cur.data.empty() and not prev.data.empty() and not (cur.controls or cur.blists or cur.web_categories or cur.sites or cur.apps) and not set(cur.channels + cur.duals) <= set(prev.channels + prev.duals) and out[k][0] is not None): # "block aadhaar in chatgpt and alert on slack": the same data, another place cur.data = copy.deepcopy(prev.data) if (cur.prompt and not (cur.ai_sites or cur.ai_any or cur.ide_tools or cur.ide_providers or cur.ide_any) and (prev.ide_tools or prev.ide_providers or prev.ide_any) and not (prev.ai_sites or prev.ai_any)): # "let developers use Claude Code but log every prompt": the prompts are the coding tool's cur.ide_tools, cur.ide_providers, cur.ide_any = list(prev.ide_tools), list(prev.ide_providers), prev.ide_any if not cur.groups and not cur.unknown_groups: cur.groups, cur.unknown_groups = list(prev.groups), list(prev.unknown_groups) continue if _has_surface(cur) or not _has_surface(prev): continue words = {it.text for it in out[k][3].get('items', [])} pronoun = bool(words & {'it', 'them', 'this', 'that', 'those', 'these', 'same'}) fields = ['ai_sites', 'channels', 'duals', 'ide_tools', 'ide_providers', 'connectors'] if pronoun or cur.data.empty(): # "... but block it for everyone else": "it" is the whole surface, switches included fields += ['controls', 'blists', 'web_categories', 'sites', 'apps', 'ide_features'] for f in fields: setattr(cur, f, list(getattr(prev, f))) cur.ai_any, cur.prompt, cur.ide_any = prev.ai_any, prev.prompt, prev.ide_any if (cur.data.empty() and words & {'else', 'rest'} and not (cur.groups or cur.unknown_groups or cur.all_users) and not (prev.data.empty() or prev.controls or prev.blists)): # "block aadhaar in chatgpt, warn for phone numbers, log everything else": all other personal data said = [e for _, c2, _, f2 in out[:k] if f2.get('sentence') == out[k][3].get('sentence') for e in c2.data.entity_set()] cur.data = Data(groups=['pii'], minus=list(dict.fromkeys(said))) elif cur.data.empty() and (pronoun or not (prev.controls or prev.blists)): # "warn about secrets in X, block for interns": a clause with no surface of its own repeats the # data too (but "block aadhaar in chatgpt and disable usb" does not move Aadhaar onto USB) cur.data = copy.deepcopy(prev.data) if out[k - 1][0] == 'ALLOW' and out[k][0] in ('BLOCK', 'DISABLE') and (prev.groups or prev.unknown_groups): # "Allow HR to use USB but block it for everyone else": one restriction, HR excepted out[k][2].append(Content(groups=list(prev.groups), unknown_groups=list(prev.unknown_groups))) drop.add(k - 1) elif not cur.groups and not cur.unknown_groups: cur.groups, cur.unknown_groups = list(prev.groups), list(prev.unknown_groups) out = [row for k, row in enumerate(out) if k not in drop] # "For the finance team, block X and also disable USB": a group named once covers the sentence for k in range(1, len(out)): cur, prev = out[k][1], out[k - 1][1] if (out[k][3].get('sentence') == out[k - 1][3].get('sentence') and not cur.groups and not cur.unknown_groups and not cur.all_users and (prev.groups or prev.unknown_groups) and out[k - 1][0] != 'ALLOW'): cur.groups, cur.unknown_groups = list(prev.groups), list(prev.unknown_groups) # a clause still without a verb inherits from the previous one ("block aadhaar in chatgpt and pan in claude") for k in range(1, len(out)): if out[k][0] is None and out[k - 1][0] is not None: out[k][0] = out[k - 1][0] out[k][3] = dict(out[k][3], inherited=True) # ... or from the next one ("aadhaar in chatgpt, block it") for k in range(len(out) - 2, -1, -1): if out[k][0] is None and out[k + 1][0] is not None: out[k][0] = out[k + 1][0] # A clause that is only "for " / only an exception continues the previous clause. merged = [] for row in out: verb, content, excs, flags = row if merged and not (content.has_endpoint_surface() or content.channels or content.duals or not content.data.empty()) and (content.groups or content.unknown_groups or content.all_users) and verb in (None, merged[-1][0]): prev = merged[-1][1] for g in content.groups: _add(prev.groups, g) for g in content.unknown_groups: _add(prev.unknown_groups, g) prev.all_users = prev.all_users or content.all_users merged[-1][2].extend(excs) continue # A clause that only names a surface ("... in chatgpt and gemini") but whose data and verb # are in the previous clause belongs to it. merged.append(row) return [tuple(r) for r in merged] # ----------------------------------------------------------------------------- routing # Switch words that are only switches when nothing else in the clause names a channel # ("downloading repositories" is GitHub, not the browser-download switch). WEAK_CONTROLS = {'downloads', 'file_uploads_ai', 'code_paste'} def _dual_sense(name: str, verb: str | None, c: Content) -> str: """'endpoint' or 'channel' for a dual-meaning word in this clause.""" sharing = any(m in c.mods for m in ('public', 'external', 'domain_wide', 'internal', 'at_rest', 'user_drives', 'shared_drives', 'suspended')) sharing = sharing and not any(w.startswith(('upload', 'sync', 'sav', 'copy', 'cop')) for w in c.transfer_words) share_words = any(w.startswith('shar') for w in c.transfer_words) has_data = not c.data.empty() if name in ('gmail', 'outlook'): if has_data or verb in ('ALERT', 'LOG', 'REMEDIATE') or 'channel' in c.channels: return 'channel' return 'endpoint' # google drive / onedrive / sharepoint if sharing or share_words or verb in ('REVOKE', 'DELETE'): return 'channel' if verb in ('LOG', 'ALERT') and not c.transfer: return 'channel' if verb in ('BLOCK', 'DISABLE', 'WARN') or c.transfer: if has_data and not c.transfer and verb not in ('BLOCK', 'DISABLE'): return 'channel' return 'endpoint' return 'channel' if has_data else 'endpoint' def _route(verb, content: Content, excs: list[Content], flags: dict, ctx: PlatformContext, lx: Lexicon): types: list[str] = [] dual_endpoint: list[str] = [] for ch in content.channels: _add(types, ch) for ch, _cid in content.connectors: # "the Prod AWS buckets": a connector names its channel _add(types, ch) if verb in ('WARN', 'LOG', 'ALERT', 'REDACT') and 'code_paste' in content.controls: # "warn when they paste code into ChatGPT": the paste switch can only block; warning means # checking prompts for source code content.controls.remove('code_paste') _add(content.data.groups, 'source_code') content.prompt = True if (content.channels or content.duals) and 'personal_recipients' in content.mods: # "emails sent to personal accounts": recipients, not the sign-in restriction content.controls = [x for x in content.controls if x not in ('sign_in', 'social_login')] if content.channels or content.duals: # "downloading repositories", "uploads to onedrive": these words are the action, not the # browser-download / AI-upload switches content.controls = [x for x in content.controls if x not in WEAK_CONTROLS] if content.channels and content.domain_words: # "emails to hotmail.com or gmail.com": recipients' domains, not webmail switches lxc = lx.controls dom_controls = {name for name in content.controls if any(w in lxc[name]['say'] for w in content.domain_words)} content.controls = [x for x in content.controls if x not in dom_controls] content.duals = [x for x in content.duals if not any(w in lx.dual_words[x]['say'] for w in content.domain_words)] for d in content.duals: sense = _dual_sense(d, verb, content) if sense == 'channel': _add(types, lx.dual_words[d]['channel']) else: dual_endpoint.append(d) if types and verb in ('LOG', 'ALERT', None): # "monitor Drive for files shared with personal gmail accounts": the gmail words say what to watch content.controls = [x for x in content.controls if lx.controls[x]['polarity'] != 'restrict'] if content.has_endpoint_surface() or dual_endpoint: # Endpoint words add an endpoint request next to any channel ones ("block aadhaar on slack and in # chatgpt"); the bare word "prompt" alone does not, when a channel is named. if not types or content.controls or content.blists or content.web_categories or content.ai_sites \ or content.ai_any or content.ide_tools or content.ide_any or content.sites or content.apps or dual_endpoint: types.insert(0, 'endpoint') if not types: if content.domains and verb in ('BLOCK', 'DISABLE', 'WARN', 'REDACT'): # "block example.com": a website; "block aadhaar on internal-gpt.acme.com": a site to watch types = ['endpoint'] elif content.thresholds and any(t[1] == 'repos' for t in content.thresholds): types = ['github'] elif 'clone' in content.mods: types = ['github'] elif 'after_hours' in content.mods or 'personal_recipients' in content.mods: types = ['email'] base = dict(verb=verb, verb_conf=flags.get('vconf', 1.0), verb_text=flags.get('vtext', ''), only=bool(flags.get('only')), only_group_can=bool(flags.get('only_group_can')), negated=bool(flags.get('negated'))) spans = _spans(flags.get('items', [])) clause_text = ' '.join(it.text for it in flags.get('items', []) if not it.is_punct) if not types: cands = _candidates(ctx) if content.is_empty() and not verb: return [] return [Request(type=None, content=content, exceptions=excs, candidates=cands, spans=spans, clause=clause_text, **base)] reqs: list[Request] = [] for t in types: c = copy.deepcopy(content) if t != 'endpoint': # endpoint-only content does not travel to channel requests c.controls, c.blists, c.sites, c.apps, c.web_categories = [], [], [], [], [] c.ai_sites, c.ai_any, c.prompt = [], False, False c.ide_tools, c.ide_providers, c.ide_any, c.ide_features = [], [], False, [] c.connectors = [x for x in c.connectors if x[0] == t] else: c.connectors = [] if t == 'endpoint' and len(c.groups) > 1: # one endpoint policy per group: the form saves a single groupId for g in c.groups: cg = copy.deepcopy(c) cg.groups = [g] reqs.append(Request(type=t, content=cg, exceptions=copy.deepcopy(excs), spans=spans, clause=clause_text, group=g, dual_endpoint=list(dual_endpoint), **base)) continue r = Request(type=t, content=c, exceptions=copy.deepcopy(excs), spans=spans, clause=clause_text, dual_endpoint=list(dual_endpoint) if t == 'endpoint' else [], **base) if c.groups: r.group = c.groups[0] if c.unknown_groups: r.group_unknown = c.unknown_groups[0] reqs.append(r) return reqs def _candidates(ctx: PlatformContext) -> list[str]: """Policy types that could carry a data-only request, best first.""" cands = ['endpoint', 'email', 'slack', 'teams', 'chat', 'drive', 'm365files', 's3', 'gcs', 'azure', 'github'] if ctx.connectorsKnown: have = [t for t in cands[1:] if ctx.connectors_for(t)] return ['endpoint'] + have return cands[:4] def _spans(items: list[Item]) -> list[tuple[int, int, str]]: out = [] for it in items: if it.senses and not it.is_punct: out.append((it.start, it.end, it.senses[0].slot)) return out