Blame for copilot/check.py

"""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+)$')

# Keys the form adds only when the admin sets them, with their types.
_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:
            # a channel body nests its config under channels.<type>; the default has it, so this is real
            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']:
            # the server links categories by name and silently skips unknown ones
            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