| """The last line of defence: is this exactly a body the platform accepts? |
| |
| Every draft body is checked here before it leaves the copilot. A failure is a copilot bug, |
| so the body is withheld (never returned half-right) and the failure is reported. |
| |
| Checks, in the order the server and the agent would hit them: |
| 1. shape: the same keys and value types as the form's own default body |
| 2. values: enums the server and agent accept (actions, modes, levels) |
| 3. references: groups, detection rules, block lists and connectors exist in the tenant |
| 4. the IDE section survives the server's sanitizeIdeControl() unchanged |
| 5. new detection rules and block lists pass their create routes' checks |
| """ |
| from __future__ import annotations |
|
|
| import json |
| import re |
| from typing import Any |
|
|
| from .platform import catalog |
| from .platform.context import PlatformContext |
|
|
| PLACEHOLDER = re.compile(r'^\$NEW_(RULE|LIST_\d+)$') |
|
|
| |
| _IDE_EXTRA = {'detectionRuleIds': list, 'storePromptContent': bool, 'interceptPrompts': bool, 'interceptHosts': list, |
| 'inspectResponses': bool, 'allowedProviders': list, 'blockedAiTools': list, 'deniedMcpServers': list, |
| 'claudeCodeHook': bool, 'watchedAiProcesses': list, 'sensitivePaths': list} |
| EXTRA_KEYS = { |
| 'email': {'channels.email.scheduleRestrict': bool, 'channels.email.allowedHoursStart': int, |
| 'channels.email.allowedHoursEnd': int}, |
| 'endpoint': {**{f'agentControl.ideControl.{k}': t for k, t in _IDE_EXTRA.items()}, |
| **{f'ideControl.{k}': t for k, t in _IDE_EXTRA.items()}}, |
| } |
|
|
| _HOST = re.compile(r'^(\*\.)?([a-z0-9-]+\.)+[a-z]{2,63}$', re.I) |
| _ID = re.compile(r'^[A-Za-z0-9_-]{1,64}$') |
| _GLOB = re.compile(r'^[^\x00-\x1f]+$') |
|
|
|
|
| def _enums() -> dict[str, dict[str, set]]: |
| ide = catalog.ide() |
| ide_enums = {'promptAction': set(ide['promptActions']), 'contextAction': set(ide['accessActions']), |
| 'commandAction': set(ide['accessActions']), 'fileAccessAction': set(ide['sensorActions']), |
| 'fileAccessThreshold': set(ide['riskLevels']), 'providerControl': {'all', 'allowlist'}, |
| 'mcpControl': {'monitor', 'denylist'}} |
| endpoint = {'action': set(catalog.action_rank()), |
| 'agentControl.clipboardMode': {'off', 'audit', 'targets', 'everywhere'}, |
| 'agentControl.sensitivityThreshold': {'low', 'medium', 'high'}} |
| for k, v in ide_enums.items(): |
| endpoint[f'agentControl.ideControl.{k}'] = v |
| endpoint[f'ideControl.{k}'] = v |
| scan = {'rules', 'discovery'} |
| return { |
| 'endpoint': endpoint, |
| 'email': {'channels.email.action': set(catalog.snapshot()['emailActions']), |
| 'channels.email.direction': {'outbound'}, |
| 'channels.email.provider': {'google', 'm365'}}, |
| 'slack': {'channels.slack.action': {'log', 'delete'}}, |
| 'drive': {'channels.drive.dspm.scanMode': scan}, |
| 'm365files': {'channels.m365files.dspm.scanMode': scan}, |
| 's3': {'channels.s3.action': {'log'}, 'channels.s3.scanMode': scan}, |
| 'gcs': {'channels.gcs.action': {'log'}, 'channels.gcs.scanMode': scan}, |
| 'azure': {'channels.azure.action': {'log'}, 'channels.azure.scanMode': scan}, |
| 'github': {'channels.github.scanMode': scan}, |
| 'teams': {}, 'chat': {}, |
| } |
|
|
|
|
| def _type_ok(value: Any, like: Any) -> bool: |
| if isinstance(like, bool): |
| return isinstance(value, bool) |
| if isinstance(like, int): |
| return isinstance(value, int) and not isinstance(value, bool) |
| if isinstance(like, str): |
| return isinstance(value, str) |
| if isinstance(like, list): |
| return isinstance(value, list) |
| if isinstance(like, dict): |
| return isinstance(value, dict) |
| return False |
|
|
|
|
| def _shape(body: dict, default: dict, extras: dict[str, type], path: str, errors: list[str]) -> None: |
| for k, dv in default.items(): |
| p = f'{path}{k}' |
| if k not in body: |
| errors.append(f'missing key {p}') |
| continue |
| v = body[k] |
| if not _type_ok(v, dv): |
| errors.append(f'{p}: expected {type(dv).__name__}, got {type(v).__name__}') |
| continue |
| if isinstance(dv, dict) and dv: |
| _shape(v, dv, extras, p + '.', errors) |
| if isinstance(dv, list) and isinstance(v, list): |
| for x in v: |
| if not isinstance(x, (str, dict)) or (isinstance(x, str) and not x.strip()): |
| errors.append(f'{p}: bad list element {x!r}') |
| for k, v in body.items(): |
| p = f'{path}{k}' |
| if k in default: |
| continue |
| t = extras.get(p) |
| if t is None: |
| |
| errors.append(f'unexpected key {p}') |
| elif t is int and (not isinstance(v, int) or isinstance(v, bool)): |
| errors.append(f'{p}: expected int') |
| elif t is not int and not isinstance(v, t): |
| errors.append(f'{p}: expected {t.__name__}') |
|
|
|
|
| def _get(d: dict, path: str) -> Any: |
| cur: Any = d |
| for p in path.split('.'): |
| if not isinstance(cur, dict) or p not in cur: |
| return None |
| cur = cur[p] |
| return cur |
|
|
|
|
| def _rule_ok(rid: str, ctx: PlatformContext) -> bool: |
| return bool(PLACEHOLDER.match(rid)) or ctx.rule(rid) is not None |
|
|
|
|
| def validate(ptype: str, body: dict, ctx: PlatformContext, new_rule: dict | None = None, |
| new_lists: list[dict] | None = None) -> list[str]: |
| errors: list[str] = [] |
| if ptype not in catalog.POLICY_TYPES: |
| return [f'unknown policy type {ptype}'] |
| default = catalog.default_body(ptype) |
| _shape(body, default, EXTRA_KEYS.get(ptype, {}), '', errors) |
| if errors: |
| return errors |
|
|
| name = body.get('name', '') |
| if not name.strip() or len(name) > 200: |
| errors.append('name must be 1-200 characters') |
| if body.get('active') is not False or body.get('enabled') is not False: |
| errors.append('copilot policies must be proposed inactive') |
|
|
| for path, allowed in _enums()[ptype].items(): |
| v = _get(body, path) |
| if v is not None and v not in allowed: |
| errors.append(f'{path}: {v!r} not in {sorted(allowed)}') |
|
|
| gid = body.get('groupId') |
| if ptype != 'endpoint' and gid != 'default': |
| errors.append('channel policies are saved with groupId "default"') |
| if gid != 'default' and not ctx.group(gid): |
| errors.append(f'unknown group {gid}') |
| if 'appliesToGroupIds' in body: |
| errors.append('appliesToGroupIds is set by the server, not the form') |
|
|
| if ptype == 'endpoint': |
| if body['channels'] != {}: |
| errors.append('endpoint policies have no channels') |
| for rid in body['extensionControl']['detectionRuleIds']: |
| if not _rule_ok(rid, ctx): |
| errors.append(f'unknown detection rule {rid}') |
| for x in body['extensionControl']['allowedExtensions']: |
| if not re.match(r'^[a-p]{32}$', x): |
| errors.append(f'allowedExtensions: {x} is not an extension id') |
| hosts = set(catalog.ai_hosts()) |
| for h in body['enabledSites']: |
| if h not in hosts: |
| errors.append(f'enabledSites: {h} is not a built-in AI site') |
| for h in body['customSites']: |
| if not _HOST.match(h): |
| errors.append(f'customSites: bad host {h}') |
| for name in body['activeWebCategories']: |
| |
| if not isinstance(name, str) or not ctx.web_category(name): |
| errors.append(f'activeWebCategories: {name!r} is not a Web Filtering category') |
| ac = body['agentControl'] |
| for key, kind in (('webBlockListIds', 'web'), ('appBlockListIds', 'app')): |
| for lid in ac[key]: |
| if PLACEHOLDER.match(lid): |
| nl = next((x for x in (new_lists or []) if x['placeholder'] == lid), None) |
| if not nl or nl['body']['kind'] != kind: |
| errors.append(f'{key}: placeholder {lid} has no {kind} list') |
| continue |
| bl = ctx.block_list(lid) |
| if not bl or bl.kind != kind: |
| errors.append(f'{key}: {lid} is not a {kind} block list') |
| if ac['removableStorageReadOnly'] and ac['blockRemovableStorage']: |
| errors.append('USB cannot be both blocked and read-only') |
| sr = ac['signInRestriction'] |
| if sr['enabled'] and not sr['allowedDomains']: |
| errors.append('sign-in restriction enabled without allowed domains') |
| for dom in sr['allowedDomains']: |
| if not _HOST.match(dom): |
| errors.append(f'sign-in domain {dom} is not a domain') |
| errors += _ide_errors(ac['ideControl'], ctx) |
| if body['ideControl'] != ac['ideControl']: |
| errors.append('ideControl must match agentControl.ideControl, as the form sends it') |
| else: |
| ch = body['channels'] |
| if list(ch) != [ptype] or ch[ptype].get('enabled') is not True: |
| errors.append(f'channel body must be channels.{ptype} with enabled: true') |
| if body['agentControl'] != {} or body['extensionControl'] != {}: |
| errors.append('channel policies send empty agentControl and extensionControl') |
| cfg = ch.get(ptype, {}) |
| for key in ('detectionRuleIds', 'sharingDetectionRuleIds'): |
| for rid in cfg.get(key, []) or []: |
| if not _rule_ok(rid, ctx): |
| errors.append(f'unknown detection rule {rid}') |
| for rid in (cfg.get('dspm') or {}).get('detectionRuleIds', []) or []: |
| if not _rule_ok(rid, ctx): |
| errors.append(f'unknown detection rule {rid}') |
| if ctx.connectorsKnown: |
| known = {c.id for c in ctx.connectors_for(ptype)} |
| for cid in cfg.get('connectorIds', []) or []: |
| if cid not in known: |
| errors.append(f'unknown {ptype} connector {cid}') |
| for key in ('allowedDomains', 'blockedDomains', 'trustedDomains', 'ownedDomains'): |
| for dom in cfg.get(key, []) or []: |
| if not _HOST.match(dom): |
| errors.append(f'{key}: {dom} is not a domain') |
| if ptype == 'email': |
| th = cfg['bulkForwardThreshold'] |
| if th < 1: |
| errors.append('bulkForwardThreshold must be at least 1') |
| for k in ('allowedHoursStart', 'allowedHoursEnd'): |
| if k in cfg and not 0 <= cfg[k] <= 23: |
| errors.append(f'{k} must be an hour 0-23') |
|
|
| if new_rule is not None: |
| errors += rule_body_errors(new_rule, ctx) |
| for nl in new_lists or []: |
| errors += list_body_errors(nl['body']) |
| try: |
| json.dumps(body) |
| except (TypeError, ValueError) as e: |
| errors.append(f'not JSON: {e}') |
| return errors |
|
|
|
|
| def _ide_errors(ide: dict, ctx: PlatformContext) -> list[str]: |
| """What server/src/modules/ide/policy.ts sanitizeIdeControl() would change or drop.""" |
| errs: list[str] = [] |
| cat = catalog.ide() |
| tools = {t['key'] for t in cat['tools']} |
| provs = {p['id'] for p in cat['providers']} |
| for rid in ide.get('detectionRuleIds', []): |
| if not (PLACEHOLDER.match(rid) or (_ID.match(rid) and ctx.rule(rid))): |
| errs.append(f'ideControl.detectionRuleIds: {rid}') |
| for t in ide.get('blockedAiTools', []): |
| if t not in tools: |
| errs.append(f'ideControl.blockedAiTools: unknown tool {t}') |
| for p in ide.get('allowedProviders', []): |
| if p not in provs: |
| errs.append(f'ideControl.allowedProviders: unknown provider {p}') |
| for h in ide.get('interceptHosts', []): |
| if not _HOST.match(h): |
| errs.append(f'ideControl.interceptHosts: {h}') |
| for g in ide.get('sensitivePaths', []): |
| if not _GLOB.match(g) or len(g) > 260: |
| errs.append(f'ideControl.sensitivePaths: {g!r}') |
| for r in ide.get('deniedMcpServers', []): |
| if not isinstance(r, dict) or r.get('type') not in ('url', 'command', 'name') or not str(r.get('value', '')).strip(): |
| errs.append(f'ideControl.deniedMcpServers: {r!r}') |
| elif r['type'] == 'url' and not re.match(r'^(\*|https?)://\S+$', r['value'], re.I): |
| errs.append(f'ideControl.deniedMcpServers: bad url {r["value"]}') |
| if ide.get('providerControl') == 'allowlist' and not ide.get('allowedProviders'): |
| errs.append('ideControl: an allowlist with no providers blocks every AI provider') |
| return errs |
|
|
|
|
| def rule_body_errors(r: dict, ctx: PlatformContext | None = None) -> list[str]: |
| """POST /api/admin/detection-rules (routes/detectionRules.ts). `ctx` adds the tenant's own detectors.""" |
| errs: list[str] = [] |
| if not str(r.get('name', '')).strip(): |
| errs.append('new rule: name required') |
| ents, kws, fts = r.get('entities') or [], r.get('keywords') or [], r.get('fileTypes') or [] |
| if not (ents or kws or fts): |
| errs.append('new rule: needs a detector, keyword or file type') |
| keys = ctx.entity_keys() if ctx else catalog.entity_keys() |
| for e in ents: |
| if e not in keys: |
| errs.append(f'new rule: unknown entity {e}') |
| for f in fts: |
| if f not in catalog.file_types(): |
| errs.append(f'new rule: unknown file type {f}') |
| if r.get('severity') not in ('LOW', 'MEDIUM', 'HIGH'): |
| errs.append('new rule: bad severity') |
| return errs |
|
|
|
|
| def list_body_errors(b: dict) -> list[str]: |
| """POST /api/admin/block-lists (routes/blockLists.ts cleanEntries).""" |
| errs: list[str] = [] |
| if not str(b.get('name', '')).strip(): |
| errs.append('new block list: name required') |
| if b.get('kind') not in ('web', 'app'): |
| errs.append('new block list: kind must be web or app') |
| entries = [str(x).strip().lower() for x in b.get('entries') or [] if str(x).strip()] |
| if not entries: |
| errs.append('new block list: no entries') |
| for e in entries: |
| if b.get('kind') == 'web' and not _HOST.match(e): |
| errs.append(f'new block list: {e} is not a website') |
| return errs |
|
|
