| """Shared adapter helpers: actions, rule questions, groups, connectors, exceptions, thresholds.
|
|
|
| An adapter turns one Request into the exact body that type's dashboard form would POST.
|
| It starts from the form's defaults and only changes what the request decided, so a policy
|
| made by the copilot is indistinguishable from one made by hand.
|
| """
|
| from __future__ import annotations
|
|
|
| import difflib
|
| from dataclasses import dataclass, field
|
| from typing import Any
|
|
|
| from ..interpret import Content, Request
|
| from ..model import Draft, Question
|
| from ..platform import catalog
|
| from ..platform.context import PlatformContext
|
| from ..rules import RuleDecision
|
|
|
| VERB_WORD = {'BLOCK': 'block', 'WARN': 'warn', 'REDACT': 'redact', 'ALERT': 'alert', 'LOG': 'log/monitor',
|
| 'DELETE': 'delete', 'REMEDIATE': 'recall', 'REVOKE': 'revoke', 'READONLY': 'make read-only',
|
| 'ALLOW': 'allow', 'ENABLE': 'turn on', 'DISABLE': 'turn off'}
|
|
|
|
|
| @dataclass
|
| class Build:
|
| """What an adapter hands back."""
|
| draft: Draft
|
| consumed_exceptions: list[int] = field(default_factory=list)
|
| cancelled: bool = False
|
|
|
|
|
| def answer_for(answers: dict, slot: str, index: int) -> Any:
|
| """An answer for this draft ('slot@index') or for all drafts ('slot')."""
|
| if not isinstance(answers, dict):
|
| return None
|
| v = answers.get(f'{slot}@{index}')
|
| return v if v is not None else answers.get(slot)
|
|
|
|
|
| def choose_action(d: Draft, req: Request, index: int, answers: dict, *, mapping: dict[str, str],
|
| allowed: list[tuple[str, str]], default: str, type_label: str,
|
| impossible_default: str | None = None) -> str:
|
| """Pick the body action for the request's verb.
|
|
|
| mapping: verb -> action for verbs this type can honour.
|
| allowed: [(action, label)] the type accepts, for questions.
|
| A verb that is not in mapping is impossible on this type: the draft keeps a valid action
|
| and a blocking question asks which of the possible ones the admin wants.
|
| """
|
| valid = [a for a, _ in allowed]
|
| ans = answer_for(answers, 'action', index)
|
| if isinstance(ans, str) and ans in valid:
|
| return ans
|
| verb = req.verb
|
| options = [{'value': a, 'label': lbl} for a, lbl in allowed]
|
| if verb is None or verb in ('ENABLE', 'CREATE', 'DISABLE'):
|
| d.ask(Question('action', f'What should happen when this matches on {type_label}?', options, True, default))
|
| return default
|
| if req.verb_conf < 0.5:
|
| d.ask(Question('action', f'I could not tell whether you want to block or allow this. '
|
| f'What should happen on {type_label}?', options, True, default))
|
| return default
|
| if verb in mapping:
|
| return mapping[verb]
|
| fallback = impossible_default or default
|
| d.issue('ACTION_NOT_AVAILABLE', 'blocking',
|
| f'{type_label} policies cannot {VERB_WORD.get(verb, verb.lower())}. '
|
| f'They can: {", ".join(lbl for _, lbl in allowed)}.',
|
| requested=verb.lower(), type=d.policyType, available=valid)
|
| d.ask(Question('action', f'{type_label} cannot {VERB_WORD.get(verb, verb.lower())}. Which of these do you want?',
|
| options, True, fallback))
|
| return fallback
|
|
|
|
|
| def rule_questions(d: Draft, rd: RuleDecision, ctx: PlatformContext) -> None:
|
| """Questions about the detection-rule choice (never blocking unless the data was guessed)."""
|
| if rd.mode == 'superset' and rd.rule_ids:
|
| r = ctx.rule(rd.rule_ids[-1])
|
| extra = ', '.join(catalog.entity_label(e) for e in rd.extras)
|
| d.ask(Question('rule', f'Use the existing rule "{r.name if r else rd.rule_ids[-1]}" (it also detects {extra}), '
|
| f'or create a new rule with only what you asked for?',
|
| [{'value': rd.rule_ids[-1], 'label': f'Use "{r.name if r else rd.rule_ids[-1]}"'},
|
| {'value': 'create', 'label': 'Create a new rule'}], False, rd.rule_ids[-1]))
|
| if rd.mode == 'create' and rd.alternative and rd.alternative.get('mode') == 'reuse':
|
| alt = rd.alternative
|
| d.ask(Question('rule', f'I will create a new detection rule. The built-in rule "{alt["ruleName"]}" '
|
| f'already covers this and more. Use it instead?',
|
| [{'value': 'create', 'label': 'Create the new rule'},
|
| {'value': alt['ruleIds'], 'label': f'Use "{alt["ruleName"]}"'}], False, 'create'))
|
| if rd.assumed:
|
| names = [ctx.rule(r).name for r in rd.rule_ids if ctx.rule(r)]
|
| d.issue('DATA_ASSUMED', 'warning',
|
| f'"Sensitive data" can mean many things. I used these rules: {", ".join(names)}. '
|
| f'Change the list if you meant something narrower.', ruleIds=list(rd.rule_ids))
|
| d.ask(Question('data', 'Which data do you want this policy to look for?',
|
| [{'value': r.id, 'label': r.name} for r in ctx.rules], False, list(rd.rule_ids)))
|
|
|
|
|
| def need_data(d: Draft, rd: RuleDecision, ctx: PlatformContext, why: str) -> None:
|
| d.issue('NO_DETECTION_RULES', 'blocking', why)
|
| d.ask(Question('data', 'What data should this policy look for?',
|
| [{'value': r.id, 'label': r.name} for r in ctx.rules], True, []))
|
|
|
|
|
| def resolve_group(d: Draft, req: Request, ctx: PlatformContext, index: int, answers: dict) -> str:
|
| """The groupId for an endpoint policy. An unknown group is asked about, never dropped silently."""
|
| ans = answer_for(answers, 'group', index)
|
| if isinstance(ans, str) and (ans == 'default' or ctx.group(ans)):
|
| return ans
|
| if req.group_unknown and req.group == 'default':
|
| names = [g.name for g in ctx.groups]
|
| close = difflib.get_close_matches(req.group_unknown, [n.lower() for n in names], n=3, cutoff=0.5)
|
| opts = [{'value': g.id, 'label': g.name} for g in ctx.groups if g.name.lower() in close]
|
| opts += [{'value': g.id, 'label': g.name} for g in ctx.groups if g.name.lower() not in close][:8]
|
| opts.append({'value': 'default', 'label': 'Everyone (org-wide)'})
|
| d.issue('UNKNOWN_GROUP', 'blocking', f'There is no group called "{req.group_unknown}".',
|
| name=req.group_unknown)
|
| d.ask(Question('group', f'There is no group called "{req.group_unknown}". Which group did you mean?',
|
| opts, True, 'default'))
|
| return 'default'
|
| if req.group != 'default' and ctx.group(req.group):
|
| return req.group
|
| return 'default'
|
|
|
|
|
| def channel_group_check(d: Draft, req: Request, ctx: PlatformContext, index: int, answers: dict) -> None:
|
| """Channel policies are org-wide in the dashboard: a group can't narrow them."""
|
| asked = req.group != 'default' or req.group_unknown
|
| if not asked or answer_for(answers, 'scope', index) == 'org':
|
| return
|
| name = ctx.group(req.group).name if ctx.group(req.group) else req.group_unknown
|
| d.issue('GROUP_NOT_SUPPORTED', 'blocking',
|
| f'{catalog.TYPE_LABEL[d.policyType]} policies apply to the whole organisation; they cannot be limited '
|
| f'to "{name}". Only endpoint policies are scoped to a group.', group=name)
|
| d.ask(Question('scope', f'This policy will apply to everyone, not only {name}. Continue?',
|
| [{'value': 'org', 'label': 'Yes, apply to everyone'}, {'value': 'cancel', 'label': 'Cancel'}],
|
| True, 'org'))
|
|
|
|
|
| def pick_connectors(d: Draft, req: Request, ctx: PlatformContext, channel: str, index: int, answers: dict) -> list[str]:
|
| ans = answer_for(answers, 'connector', index)
|
| known = {c.id for c in ctx.connectors_for(channel)}
|
| if isinstance(ans, list):
|
| return [a for a in ans if a in known]
|
| if isinstance(ans, str) and ans in known:
|
| return [ans]
|
| ids = [cid for ch, cid in req.content.connectors if ch == channel and cid in known]
|
| if ctx.connectorsKnown and not ctx.connectors_for(channel):
|
| d.issue('NO_CONNECTOR', 'warning',
|
| f'No {catalog.TYPE_LABEL[channel]} connection is set up yet, so this policy will not see anything '
|
| f'until one is added in Settings.', channel=channel)
|
| return ids
|
|
|
|
|
| def thresholds_unsupported(d: Draft, req: Request, index: int, answers: dict, handled: set[str] = frozenset()) -> None:
|
| rest = [(n, noun) for n, noun in req.content.thresholds if noun not in handled]
|
| if not rest or answer_for(answers, 'threshold', index) == 'ignore':
|
| return
|
| n, noun = rest[0]
|
| d.issue('THRESHOLD_NOT_SUPPORTED', 'blocking',
|
| f'{catalog.TYPE_LABEL[d.policyType]} policies cannot count ("more than {n} {noun}"). '
|
| f'The policy acts on the first match.', count=n, noun=noun)
|
| d.ask(Question('threshold', f'The policy will act on the first match, not after {n} {noun}. Continue?',
|
| [{'value': 'ignore', 'label': 'Yes, act on every match'}, {'value': 'cancel', 'label': 'Cancel'}],
|
| True, 'ignore'))
|
|
|
|
|
| def schedule_unsupported(d: Draft, req: Request, index: int, answers: dict, after_hours_ok: bool = False) -> None:
|
| mods = req.content.mods
|
| if 'timed' in mods and answer_for(answers, 'schedule', index) != 'ignore':
|
| d.issue('START_END_NOT_SUPPORTED', 'blocking',
|
| 'Policies apply from the moment they are switched on until someone switches them off; they cannot '
|
| 'start or end on a date by themselves.')
|
| d.ask(Question('schedule', 'The policy will apply as soon as it is switched on, with no end date. Continue?',
|
| [{'value': 'ignore', 'label': 'Yes'}, {'value': 'cancel', 'label': 'Cancel'}], True, 'ignore'))
|
| return
|
| timed = 'during_hours' in mods or (not after_hours_ok and (req.content.hours or 'after_hours' in mods))
|
| if not timed:
|
| return
|
| if answer_for(answers, 'schedule', index) == 'ignore':
|
| return
|
| d.issue('SCHEDULE_NOT_SUPPORTED', 'blocking',
|
| f'{catalog.TYPE_LABEL[d.policyType]} policies apply at all times; they cannot be limited to certain hours.')
|
| d.ask(Question('schedule', 'The policy will apply at all hours. Continue?',
|
| [{'value': 'ignore', 'label': 'Yes, apply at all times'}, {'value': 'cancel', 'label': 'Cancel'}],
|
| True, 'ignore'))
|
|
|
|
|
| QUALIFIERS = {
|
| 'executables': 'kind of file (executables)',
|
| 'bulk': 'volume ("bulk")',
|
| }
|
|
|
|
|
| def qualifiers_unsupported(d: Draft, req: Request, index: int, answers: dict, handled: set[str] = frozenset()) -> None:
|
| """Words that narrow the request in a way this policy type cannot ("larger than 10 MB", "executables").
|
|
|
| Ignoring them would make the policy broader than asked, so the admin confirms first."""
|
| c = req.content
|
| found = []
|
| if c.sizes and 'sizes' not in handled:
|
| found.append(f'size ({", ".join(c.sizes)})')
|
| for mod in ('executables', 'bulk'):
|
| if mod in c.mods and mod not in handled:
|
| found.append(QUALIFIERS[mod])
|
| for s in c.situations:
|
| found.append(f'situation ("{s}")')
|
| if not found or answer_for(answers, 'qualifier', index) == 'ignore':
|
| return
|
| d.issue('QUALIFIER_NOT_SUPPORTED', 'blocking',
|
| f'This policy cannot be limited by {" or ".join(found)}; it would apply to everything it covers.',
|
| qualifiers=found)
|
| d.ask(Question('qualifier', f'The policy cannot be limited by {" or ".join(found)}. Apply it to everything?',
|
| [{'value': 'ignore', 'label': 'Yes, apply to everything'}, {'value': 'cancel', 'label': 'Cancel'}],
|
| True, 'ignore'))
|
|
|
|
|
| def size_bytes(s: str) -> int | None:
|
| """'5mb' -> 5242880."""
|
| import re
|
| m = re.match(r'^([\d.]+)(kb|mb|gb|tb|kilobytes|megabytes|gigabytes)$', s)
|
| if not m:
|
| return None
|
| mult = {'kb': 1024, 'kilobytes': 1024, 'mb': 1024 ** 2, 'megabytes': 1024 ** 2, 'gb': 1024 ** 3,
|
| 'gigabytes': 1024 ** 3, 'tb': 1024 ** 4}[m.group(2)]
|
| try:
|
| return max(1, int(float(m.group(1)) * mult))
|
| except ValueError:
|
| return None
|
|
|
|
|
| def domains_question(d: Draft, req: Request, index: int, answers: dict, doms: list[str],
|
| options: list[tuple[str, str]]) -> str | None:
|
| """What a domain named in a channel request is for. Returns the chosen role, or None (asked)."""
|
| if 'our_domain' in req.content.mods:
|
| return 'owned'
|
| ans = answer_for(answers, 'domains', index)
|
| if ans in [v for v, _ in options] or ans == 'ignore':
|
| return ans
|
| d.issue('DOMAINS_UNCLEAR', 'blocking', f'I am not sure what {", ".join(doms)} should do in this policy.',
|
| domains=doms)
|
| d.ask(Question('domains', f'What is {", ".join(doms)}?',
|
| [{'value': v, 'label': lbl} for v, lbl in options] + [{'value': 'ignore', 'label': 'Ignore it'}],
|
| True, 'ignore'))
|
| return None
|
|
|
|
|
| def exceptions_unsupported(d: Draft, req: Request, consumed: list[int], index: int, answers: dict,
|
| ctx: PlatformContext) -> None:
|
| left = [e for k, e in enumerate(req.exceptions) if k not in consumed and not e.says_nothing()]
|
| if not left or answer_for(answers, 'exception', index) == 'ignore':
|
| return
|
| desc = describe_content(left[0], ctx)
|
| groups = [ctx.group(g).name for e in left for g in e.groups if ctx.group(g)]
|
| hint = ''
|
| if groups and d.policyType == 'endpoint':
|
| hint = (f' For AI prompt rules, CLOAK\'s way to do this is a separate policy for {", ".join(groups)}: a '
|
| f'group\'s own policies replace the org-wide prompt rules for that group. Device and browser '
|
| f'controls (USB, printing, block lists) set org-wide cannot be lifted for one group.')
|
| d.issue('EXCEPTION_NOT_SUPPORTED', 'blocking',
|
| f'This kind of policy has no way to exempt {desc}.{hint}', exception=desc)
|
| d.ask(Question('exception', f'The policy cannot exempt {desc}; it would apply to them too. Continue?',
|
| [{'value': 'ignore', 'label': 'Yes, without the exception'}, {'value': 'cancel', 'label': 'Cancel'}],
|
| True, 'ignore'))
|
|
|
|
|
| def describe_content(c: Content, ctx: PlatformContext) -> str:
|
| parts: list[str] = []
|
| parts += [ctx.group(g).name for g in c.groups if ctx.group(g)]
|
| parts += c.unknown_groups
|
| parts += [catalog.entity_label(e) for e in c.data.entity_set()]
|
| parts += c.domains + c.emails + c.serials + c.sites + c.apps + c.ai_sites + c.mods + c.words
|
| return ', '.join(dict.fromkeys(parts)) or 'that'
|
|
|
|
|
| def domains_of(c: Content) -> list[str]:
|
| hosts = set(catalog.ai_hosts())
|
| return [x for x in c.domains if x not in hosts]
|
|
|
|
|
| def base_body(policy_type: str) -> dict:
|
| body = catalog.default_body(policy_type)
|
|
|
| body['active'] = False
|
| body['enabled'] = False
|
| return body
|
|
|
