| """Pure Python executor for Sentinel Session Observer programs. |
| |
| The Sentinel SDK receives two encrypted VM programs. The collector installs |
| event handlers and initializes state; the snapshot program must then run in |
| the same process map. ``build_session_observer_token`` mirrors that sequence |
| and returns the JSON value used by ``OpenAI-Sentinel-SO-Token``. |
| |
| Malformed programs raise ``ValueError``. Runtime JavaScript-like errors are |
| handled by the VM in the same places as the browser SDK and normally produce |
| an encoded diagnostic value instead of escaping to the caller. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import base64 |
| import json |
| import math |
| import random |
| import time |
| from collections import defaultdict |
| from collections.abc import Callable |
| from dataclasses import dataclass |
| from typing import Any |
|
|
|
|
| class _Undefined: |
| def __repr__(self) -> str: |
| return "undefined" |
|
|
|
|
| UNDEFINED = _Undefined() |
|
|
|
|
| def _js_string(value: Any) -> str: |
| if value is UNDEFINED: |
| return "undefined" |
| if value is None: |
| return "null" |
| if value is True: |
| return "true" |
| if value is False: |
| return "false" |
| if isinstance(value, str): |
| return value |
| if isinstance(value, float): |
| if math.isnan(value): |
| return "NaN" |
| if math.isinf(value): |
| return "Infinity" if value > 0 else "-Infinity" |
| if value.is_integer(): |
| return str(int(value)) |
| if isinstance(value, (list, tuple)): |
| return ",".join("" if item is None or item is UNDEFINED else _js_string(item) for item in value) |
| if isinstance(value, _JSObject): |
| return "[object Object]" |
| if callable(value): |
| return "function () { [native code] }" |
| return str(value) |
|
|
|
|
| def _js_number(value: Any) -> float: |
| if value is None: |
| return 0.0 |
| if value is True: |
| return 1.0 |
| if value is False: |
| return 0.0 |
| if isinstance(value, (int, float)): |
| return float(value) |
| try: |
| text = _js_string(value).strip() |
| return 0.0 if not text else float(text) |
| except (TypeError, ValueError): |
| return math.nan |
|
|
|
|
| def _latin1_btoa(value: Any) -> str: |
| return base64.b64encode(_js_string(value).encode("latin1", errors="replace")).decode("ascii") |
|
|
|
|
| def _latin1_atob(value: Any) -> str: |
| return base64.b64decode(_js_string(value), validate=True).decode("latin1") |
|
|
|
|
| def _xor_text(value: str, key: str) -> str: |
| if not key: |
| raise ValueError("requirements_token must not be empty") |
| return "".join(chr(ord(char) ^ ord(key[index % len(key)])) for index, char in enumerate(value)) |
|
|
|
|
| class _JSObject(dict[str, Any]): |
| pass |
|
|
|
|
| class _JSMap: |
| def __init__(self) -> None: |
| self.values: dict[Any, Any] = {} |
|
|
|
|
| class _EventTarget(_JSObject): |
| def __init__(self, values: dict[str, Any] | None = None) -> None: |
| super().__init__(values or {}) |
| self.listeners: dict[str, list[Callable[..., Any]]] = defaultdict(list) |
|
|
| def add_event_listener(self, event_type: Any, callback: Any, *_: Any) -> None: |
| if callable(callback): |
| self.listeners[_js_string(event_type)].append(callback) |
|
|
| def remove_event_listener(self, event_type: Any, callback: Any, *_: Any) -> None: |
| listeners = self.listeners.get(_js_string(event_type), []) |
| self.listeners[_js_string(event_type)] = [item for item in listeners if item is not callback] |
|
|
| def dispatch(self, event_type: str, **values: Any) -> None: |
| event = _JSObject({"type": event_type, **values}) |
| for callback in tuple(self.listeners.get(event_type, ())): |
| try: |
| callback(event) |
| except Exception: |
| continue |
|
|
|
|
| @dataclass |
| class _Clock: |
| start_epoch_ms: float |
| elapsed_ms: float = 12.0 |
|
|
| def now(self) -> float: |
| self.elapsed_ms += 0.37 |
| return self.elapsed_ms |
|
|
| def date_now(self) -> int: |
| return int(self.start_epoch_ms + self.now()) |
|
|
|
|
| def _json_value(value: Any) -> Any: |
| if value is UNDEFINED: |
| return None |
| if isinstance(value, _JSObject): |
| return {key: _json_value(item) for key, item in value.items() if item is not UNDEFINED} |
| if isinstance(value, _JSMap): |
| return {} |
| if isinstance(value, (list, tuple)): |
| return [_json_value(item) for item in value] |
| if isinstance(value, float) and (math.isnan(value) or math.isinf(value)): |
| return None |
| return value |
|
|
|
|
| def _json_stringify(value: Any) -> Any: |
| if value is UNDEFINED or callable(value): |
| return UNDEFINED |
| return json.dumps(_json_value(value), ensure_ascii=False, separators=(",", ":")) |
|
|
|
|
| class _SessionObserverVM: |
| def __init__(self, *, device_id: str, flow: str, seed: int | None = None) -> None: |
| self.device_id = device_id |
| self.flow = flow |
| self.random = random.Random(seed) |
| self.clock = _Clock(time.time() * 1000) |
| self.window = self._make_window() |
| self.slots: dict[Any, Any] = {} |
| self.operation_count = 0 |
| self.result: str | None = None |
| self._settled = False |
|
|
| def _make_window(self) -> _EventTarget: |
| window = _EventTarget() |
| performance = _JSObject({"now": self.clock.now, "timeOrigin": self.clock.start_epoch_ms}) |
| page_path = { |
| "authorize_continue": "/email-verification", |
| "oauth_create_account": "/about-you", |
| "password_verify": "/log-in/password", |
| "username_password_create": "/create-account/password", |
| }.get(self.flow, "/create-account/password") |
| location = _JSObject( |
| { |
| "href": f"https://auth.openai.com{page_path}", |
| "origin": "https://auth.openai.com", |
| "pathname": page_path, |
| "search": "", |
| } |
| ) |
| navigator = _JSObject( |
| { |
| "userAgent": ( |
| "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " |
| "(KHTML, like Gecko) Chrome/145.0.0.0 Safari/537.36" |
| ), |
| "language": "en-US", |
| "languages": ["en-US", "en"], |
| "platform": "Win32", |
| "hardwareConcurrency": 8, |
| "deviceMemory": 8, |
| "maxTouchPoints": 0, |
| "webdriver": False, |
| "cookieEnabled": True, |
| "vendor": "Google Inc.", |
| } |
| ) |
| document = _EventTarget( |
| { |
| "location": location, |
| "visibilityState": "visible", |
| "hidden": False, |
| "readyState": "complete", |
| "referrer": "https://auth.openai.com/", |
| "cookie": f"oai-did={self.device_id}", |
| "scripts": [], |
| "documentElement": _JSObject(), |
| "body": _JSObject(), |
| } |
| ) |
| local_storage = _JSObject({"oai-did": self.device_id}) |
| math_object = _JSObject( |
| { |
| "abs": abs, |
| "hypot": math.hypot, |
| "sqrt": math.sqrt, |
| "round": round, |
| "floor": math.floor, |
| "random": self.random.random, |
| } |
| ) |
| object_object = _JSObject( |
| { |
| "create": lambda *_: _JSObject(), |
| "keys": lambda value: list(value.keys()) if isinstance(value, (dict, _JSObject)) else [], |
| } |
| ) |
| reflect = _JSObject({"set": self._reflect_set}) |
| date_object = _JSObject({"now": self.clock.date_now}) |
| json_object = _JSObject({"parse": lambda value: json.loads(_js_string(value)), "stringify": _json_stringify}) |
| window.update( |
| { |
| "window": window, |
| "self": window, |
| "top": window, |
| "parent": window, |
| "globalThis": window, |
| "document": document, |
| "navigator": navigator, |
| "location": location, |
| "performance": performance, |
| "screen": _JSObject({"width": 1920, "height": 1080, "availWidth": 1920, "availHeight": 1040}), |
| "localStorage": local_storage, |
| "sessionStorage": _JSObject(), |
| "Math": math_object, |
| "Object": object_object, |
| "Reflect": reflect, |
| "Date": date_object, |
| "JSON": json_object, |
| "btoa": _latin1_btoa, |
| "atob": _latin1_atob, |
| "setTimeout": lambda *_: 1, |
| "clearTimeout": lambda *_: None, |
| "scrollX": 0, |
| "scrollY": 0, |
| "innerWidth": 1920, |
| "innerHeight": 937, |
| } |
| ) |
| window["addEventListener"] = window.add_event_listener |
| window["removeEventListener"] = window.remove_event_listener |
| document["addEventListener"] = document.add_event_listener |
| document["removeEventListener"] = document.remove_event_listener |
| return window |
|
|
| @staticmethod |
| def _reflect_set(target: Any, key: Any, value: Any) -> bool: |
| _js_set(target, key, value) |
| return True |
|
|
| def _reset(self, requirements_token: str) -> None: |
| self.slots.clear() |
| self.operation_count = 0 |
| self.result = None |
| self._settled = False |
| self.slots.update( |
| { |
| 0: self._run_nested_dx, |
| 1: self._op_xor, |
| 2: self._op_literal, |
| 3: self._op_resolve, |
| 4: self._op_reject, |
| 5: self._op_add, |
| 6: self._op_get_property, |
| 7: self._op_call, |
| 8: self._op_copy, |
| 9: [], |
| 10: self.window, |
| 11: self._op_find_script, |
| 12: self._op_process_map, |
| 13: self._op_call_raw, |
| 14: self._op_json_parse, |
| 15: self._op_json_stringify, |
| 16: requirements_token, |
| 17: self._op_call_assign, |
| 18: self._op_atob, |
| 19: self._op_btoa, |
| 20: self._op_if_equal, |
| 21: self._op_if_distance, |
| 22: self._op_nested_queue, |
| 23: self._op_if_defined, |
| 24: self._op_bind, |
| 25: self._op_noop, |
| 26: self._op_noop, |
| 27: self._op_subtract, |
| 28: self._op_noop, |
| 29: self._op_less_than, |
| 30: self._op_make_function, |
| 33: self._op_multiply, |
| 34: self._op_resolve_value, |
| 35: self._op_divide, |
| } |
| ) |
|
|
| def run(self, dx: str, *, requirements_token: str | None = None) -> str: |
| if requirements_token is not None: |
| self._reset(requirements_token) |
| if 16 not in self.slots: |
| raise ValueError("collector must run before snapshot") |
| try: |
| decoded = base64.b64decode(dx, validate=True).decode("latin1") |
| instructions = json.loads(_xor_text(decoded, _js_string(self.slots[16]))) |
| except (ValueError, TypeError, json.JSONDecodeError) as exc: |
| raise ValueError("invalid Session Observer dx program") from exc |
| if not isinstance(instructions, list): |
| raise ValueError("Session Observer program must be an instruction array") |
| self.result = None |
| self._settled = False |
| self.slots[9] = list(instructions) |
| try: |
| self._execute_queue() |
| except Exception as exc: |
| if not self._settled: |
| self.result = _latin1_btoa(f"{self.operation_count}: {exc}") |
| self._settled = True |
| return self.result or "" |
|
|
| def emit_synthetic_activity(self) -> None: |
| x = 420 + self.random.randint(-20, 20) |
| y = 280 + self.random.randint(-20, 20) |
| self.window.dispatch("pointermove", clientX=x, clientY=y) |
| self.window.dispatch("pointermove", clientX=x + 37, clientY=y + 19) |
| self.window.dispatch("click", clientX=x + 37, clientY=y + 19) |
| self.window.dispatch("keydown", key="a", altKey=False, ctrlKey=False, metaKey=False) |
| self.window["scrollY"] = 184 |
| self.window.dispatch("scroll") |
| self.window.dispatch("wheel") |
|
|
| def _execute_queue(self) -> None: |
| while isinstance(self.slots.get(9), list) and self.slots[9]: |
| queue = self.slots[9] |
| row = queue.pop(0) |
| if not isinstance(row, list) or not row: |
| self.operation_count += 1 |
| continue |
| operation = self.slots.get(row[0], UNDEFINED) |
| if callable(operation): |
| operation(*row[1:]) |
| self.operation_count += 1 |
|
|
| def _run_nested_dx(self, dx: Any) -> str: |
| return self.run(_js_string(dx)) |
|
|
| def _op_xor(self, target: Any, other: Any) -> None: |
| self.slots[target] = _xor_text(_js_string(self.slots.get(target, UNDEFINED)), _js_string(self.slots.get(other, UNDEFINED))) |
|
|
| def _op_literal(self, target: Any, value: Any) -> None: |
| self.slots[target] = value |
|
|
| def _op_resolve(self, value: Any) -> None: |
| if not self._settled: |
| self.result = _latin1_btoa(value) |
| self._settled = True |
|
|
| def _op_reject(self, value: Any) -> None: |
| if not self._settled: |
| self.result = _latin1_btoa(value) |
| self._settled = True |
|
|
| def _op_add(self, target: Any, other: Any) -> None: |
| current = self.slots.get(target, UNDEFINED) |
| incoming = self.slots.get(other, UNDEFINED) |
| if isinstance(current, list): |
| current.append(incoming) |
| elif isinstance(current, str) or isinstance(incoming, str): |
| self.slots[target] = _js_string(current) + _js_string(incoming) |
| else: |
| self.slots[target] = _js_number(current) + _js_number(incoming) |
|
|
| def _op_get_property(self, target: Any, owner: Any, key: Any) -> None: |
| self.slots[target] = _js_get(self.slots.get(owner, UNDEFINED), self.slots.get(key, UNDEFINED)) |
|
|
| def _op_call(self, function_slot: Any, *argument_slots: Any) -> Any: |
| function = self.slots.get(function_slot, UNDEFINED) |
| if not callable(function): |
| raise TypeError(f"{_js_string(function)} is not a function") |
| return function(*(self.slots.get(slot, UNDEFINED) for slot in argument_slots)) |
|
|
| def _op_copy(self, target: Any, source: Any) -> None: |
| self.slots[target] = self.slots.get(source, UNDEFINED) |
|
|
| def _op_find_script(self, target: Any, pattern_slot: Any) -> None: |
| self.slots[target] = None |
|
|
| def _op_process_map(self, target: Any) -> None: |
| self.slots[target] = self.slots |
|
|
| def _op_call_raw(self, error_slot: Any, function_slot: Any, *raw_arguments: Any) -> None: |
| try: |
| function = self.slots.get(function_slot, UNDEFINED) |
| if not callable(function): |
| raise TypeError(f"{_js_string(function)} is not a function") |
| function(*raw_arguments) |
| except Exception as exc: |
| self.slots[error_slot] = f"{type(exc).__name__}: {exc}" |
|
|
| def _op_json_parse(self, target: Any, source: Any) -> None: |
| self.slots[target] = json.loads(_js_string(self.slots.get(source, UNDEFINED))) |
|
|
| def _op_json_stringify(self, target: Any, source: Any) -> None: |
| self.slots[target] = _json_stringify(self.slots.get(source, UNDEFINED)) |
|
|
| def _op_call_assign(self, target: Any, function_slot: Any, *argument_slots: Any) -> None: |
| try: |
| function = self.slots.get(function_slot, UNDEFINED) |
| if not callable(function): |
| raise TypeError(f"{_js_string(function)} is not a function") |
| self.slots[target] = function(*(self.slots.get(slot, UNDEFINED) for slot in argument_slots)) |
| except Exception as exc: |
| self.slots[target] = f"{type(exc).__name__}: {exc}" |
|
|
| def _op_atob(self, target: Any) -> None: |
| self.slots[target] = _latin1_atob(self.slots.get(target, UNDEFINED)) |
|
|
| def _op_btoa(self, target: Any) -> None: |
| self.slots[target] = _latin1_btoa(self.slots.get(target, UNDEFINED)) |
|
|
| def _op_if_equal(self, left: Any, right: Any, function_slot: Any, *raw_arguments: Any) -> None: |
| if _js_strict_equal(self.slots.get(left, UNDEFINED), self.slots.get(right, UNDEFINED)): |
| function = self.slots.get(function_slot, UNDEFINED) |
| if callable(function): |
| function(*raw_arguments) |
|
|
| def _op_if_distance(self, left: Any, right: Any, distance: Any, function_slot: Any, *raw_arguments: Any) -> None: |
| if abs(_js_number(self.slots.get(left, UNDEFINED)) - _js_number(self.slots.get(right, UNDEFINED))) > _js_number( |
| self.slots.get(distance, UNDEFINED) |
| ): |
| function = self.slots.get(function_slot, UNDEFINED) |
| if callable(function): |
| function(*raw_arguments) |
|
|
| def _op_nested_queue(self, error_slot: Any, instructions: Any) -> None: |
| old_queue = self.slots.get(9, []) |
| try: |
| self.slots[9] = list(instructions) |
| self._execute_queue() |
| except Exception as exc: |
| self.slots[error_slot] = str(exc) |
| finally: |
| self.slots[9] = old_queue |
|
|
| def _op_if_defined(self, value_slot: Any, function_slot: Any, *raw_arguments: Any) -> None: |
| if self.slots.get(value_slot, UNDEFINED) is not UNDEFINED: |
| function = self.slots.get(function_slot, UNDEFINED) |
| if callable(function): |
| function(*raw_arguments) |
|
|
| def _op_bind(self, target: Any, owner: Any, key: Any) -> None: |
| self.slots[target] = _js_get(self.slots.get(owner, UNDEFINED), self.slots.get(key, UNDEFINED)) |
|
|
| @staticmethod |
| def _op_noop(*_: Any) -> None: |
| return None |
|
|
| def _op_subtract(self, target: Any, other: Any) -> None: |
| current = self.slots.get(target, UNDEFINED) |
| incoming = self.slots.get(other, UNDEFINED) |
| if isinstance(current, list): |
| try: |
| current.remove(incoming) |
| except ValueError: |
| pass |
| else: |
| self.slots[target] = _js_number(current) - _js_number(incoming) |
|
|
| def _op_less_than(self, target: Any, left: Any, right: Any) -> None: |
| self.slots[target] = _js_number(self.slots.get(left, UNDEFINED)) < _js_number(self.slots.get(right, UNDEFINED)) |
|
|
| def _op_make_function(self, target: Any, result_slot: Any, parameter_slots: Any, body: Any = UNDEFINED) -> None: |
| has_parameters = isinstance(body, list) |
| parameters = list(parameter_slots) if has_parameters and isinstance(parameter_slots, list) else [] |
| instructions = list(body if has_parameters else (parameter_slots or [])) |
|
|
| def generated(*arguments: Any) -> Any: |
| old_queue = self.slots.get(9, []) |
| for index, slot in enumerate(parameters): |
| self.slots[slot] = arguments[index] if index < len(arguments) else UNDEFINED |
| try: |
| self.slots[9] = list(instructions) |
| self._execute_queue() |
| return self.slots.get(result_slot, UNDEFINED) |
| except Exception as exc: |
| return str(exc) |
| finally: |
| self.slots[9] = old_queue |
|
|
| self.slots[target] = generated |
|
|
| def _op_multiply(self, target: Any, left: Any, right: Any) -> None: |
| self.slots[target] = _js_number(self.slots.get(left, UNDEFINED)) * _js_number(self.slots.get(right, UNDEFINED)) |
|
|
| def _op_resolve_value(self, target: Any, source: Any) -> None: |
| self.slots[target] = self.slots.get(source, UNDEFINED) |
|
|
| def _op_divide(self, target: Any, left: Any, right: Any) -> None: |
| numerator = _js_number(self.slots.get(left, UNDEFINED)) |
| denominator = _js_number(self.slots.get(right, UNDEFINED)) |
| self.slots[target] = 0 if denominator == 0 else numerator / denominator |
|
|
|
|
| def _js_strict_equal(left: Any, right: Any) -> bool: |
| if left is UNDEFINED or right is UNDEFINED: |
| return left is right |
| if type(left) is not type(right) and not ( |
| isinstance(left, (int, float)) and not isinstance(left, bool) and isinstance(right, (int, float)) and not isinstance(right, bool) |
| ): |
| return False |
| return left == right |
|
|
|
|
| def _js_set(target: Any, key: Any, value: Any) -> None: |
| key_text = _js_string(key) |
| if isinstance(target, _JSMap): |
| target.values[key] = value |
| elif isinstance(target, dict): |
| target[key_text] = value |
| elif isinstance(target, list) and key_text.isdigit(): |
| index = int(key_text) |
| while len(target) <= index: |
| target.append(UNDEFINED) |
| target[index] = value |
| else: |
| setattr(target, key_text, value) |
|
|
|
|
| def _js_get(owner: Any, key: Any) -> Any: |
| key_text = _js_string(key) |
| if owner is UNDEFINED or owner is None: |
| raise TypeError(f"Cannot read properties of {_js_string(owner)}") |
| if isinstance(owner, _JSMap): |
| methods = { |
| "get": lambda item: owner.values.get(item, UNDEFINED), |
| "set": lambda item, value: owner.values.__setitem__(item, value) or owner, |
| "has": lambda item: item in owner.values, |
| "delete": lambda item: owner.values.pop(item, UNDEFINED) is not UNDEFINED, |
| "clear": owner.values.clear, |
| "keys": lambda: list(owner.values.keys()), |
| "values": lambda: list(owner.values.values()), |
| "size": len(owner.values), |
| } |
| return methods.get(key_text, UNDEFINED) |
| if isinstance(owner, dict): |
| return owner.get(key_text, UNDEFINED) |
| if isinstance(owner, (list, tuple)): |
| if key_text == "length": |
| return len(owner) |
| if key_text.lstrip("-").isdigit(): |
| index = int(key_text) |
| return owner[index] if -len(owner) <= index < len(owner) else UNDEFINED |
| methods = { |
| "push": lambda *items: owner.extend(items) or len(owner), |
| "indexOf": lambda item: owner.index(item) if item in owner else -1, |
| "join": lambda separator=",": _js_string(separator).join(_js_string(item) for item in owner), |
| "slice": lambda start=0, end=None: list(owner)[int(_js_number(start)) : None if end is None or end is UNDEFINED else int(_js_number(end))], |
| "shift": lambda: owner.pop(0) if owner else UNDEFINED, |
| "map": lambda function: [function(item, index, owner) for index, item in enumerate(owner)], |
| "filter": lambda function: [item for index, item in enumerate(owner) if function(item, index, owner)], |
| } |
| return methods.get(key_text, UNDEFINED) |
| if isinstance(owner, str): |
| if key_text == "length": |
| return len(owner) |
| if key_text.lstrip("-").isdigit(): |
| index = int(key_text) |
| return owner[index] if -len(owner) <= index < len(owner) else UNDEFINED |
| methods = { |
| "charCodeAt": lambda index=0: ord(owner[int(_js_number(index))]) if 0 <= int(_js_number(index)) < len(owner) else math.nan, |
| "indexOf": lambda value: owner.find(_js_string(value)), |
| "includes": lambda value: _js_string(value) in owner, |
| "slice": lambda start=0, end=None: owner[int(_js_number(start)) : None if end is None or end is UNDEFINED else int(_js_number(end))], |
| "substring": lambda start=0, end=None: owner[max(0, int(_js_number(start))) : None if end is None or end is UNDEFINED else max(0, int(_js_number(end)))], |
| "toLowerCase": owner.lower, |
| "toUpperCase": owner.upper, |
| "split": lambda separator=None: list(owner) if separator == "" else owner.split(None if separator is None or separator is UNDEFINED else _js_string(separator)), |
| "toString": lambda: owner, |
| } |
| return methods.get(key_text, UNDEFINED) |
| if isinstance(owner, (int, float, bool)): |
| if key_text in {"toString", "valueOf"}: |
| return lambda: _js_string(owner) if key_text == "toString" else owner |
| return UNDEFINED |
| if callable(owner): |
| if key_text == "bind": |
| return lambda *_: owner |
| if key_text == "call": |
| return lambda _this=UNDEFINED, *args: owner(*args) |
| if key_text == "apply": |
| return lambda _this=UNDEFINED, args=(): owner(*args) |
| if key_text == "toString": |
| return lambda: _js_string(owner) |
| return getattr(owner, key_text, UNDEFINED) |
|
|
|
|
| def build_session_observer_token( |
| *, |
| collector_dx: str, |
| snapshot_dx: str, |
| requirements_token: str, |
| response_token: str, |
| device_id: str, |
| flow: str, |
| ) -> str: |
| """Build the JSON header value for a required Session Observer challenge. |
| |
| ``requirements_token`` is the ``p`` value sent to Sentinel ``/req``. It is |
| the XOR key for both programs; it is not the subsequent proof-of-work |
| answer. Invalid encrypted programs raise ``ValueError``. |
| """ |
|
|
| if not all(isinstance(value, str) and value for value in (collector_dx, snapshot_dx, requirements_token)): |
| raise ValueError("collector_dx, snapshot_dx and requirements_token are required") |
| vm = _SessionObserverVM(device_id=device_id, flow=flow) |
| vm.run(collector_dx, requirements_token=requirements_token) |
| vm.emit_synthetic_activity() |
| snapshot = vm.run(snapshot_dx) |
| if not snapshot: |
| return "" |
| return json.dumps( |
| {"so": snapshot, "c": response_token, "id": device_id, "flow": flow}, |
| ensure_ascii=False, |
| separators=(",", ":"), |
| ) |
|
|
|
|
| def execute_sentinel_vm_token( |
| *, |
| dx: str, |
| requirements_token: str, |
| device_id: str = "", |
| flow: str = "", |
| ) -> str: |
| """Execute one current Sentinel VM program and return its encoded result.""" |
|
|
| if not isinstance(dx, str) or not dx or not isinstance(requirements_token, str) or not requirements_token: |
| raise ValueError("dx and requirements_token are required") |
| vm = _SessionObserverVM(device_id=device_id, flow=flow) |
| return vm.run(dx, requirements_token=requirements_token) |
|
|
|
|
| __all__ = ["build_session_observer_token", "execute_sentinel_vm_token"] |
|
|
