Task: workarena.servicenow.impersonation
Goal: Impersonate the user Robin Grotz.
Task configuration: name: Robin Grotz

Validation code:
import json
import os
import time
from typing import Any, Dict, List

import requests

# Why a Scripted REST endpoint?
#
# The sn_vsc_impersonation_event table records all impersonation events (both
# UI and API-triggered), but it's protected by a Web Service Access Policy that
# blocks direct Table API access — even for admin. The only way to read it
# programmatically is via a server-side GlideRecord query, which we expose
# through a custom Scripted REST endpoint. The validator creates this endpoint
# automatically on first run since the instance environment may be refreshed.

ENDPOINT_PATH = "/api/global/impersonation_checker/check"

SCRIPTED_REST_SCRIPT = '''(function process(request, response) {
    var name = request.queryParams.name;
    if (!name) {
        response.setBody({result: []});
        return;
    }
    var userNames = [];
    var userGr = new GlideRecord('sys_user');
    userGr.addQuery('name', 'CONTAINS', name);
    userGr.setLimit(5);
    userGr.query();
    while (userGr.next()) {
        userNames.push(userGr.getValue('user_name'));
    }
    var gr = new GlideRecord('sn_vsc_impersonation_event');
    gr.orderByDesc('sys_created_on');
    gr.setLimit(10);
    if (userNames.length > 0) {
        gr.addQuery('impersonated', 'IN', userNames.join(','));
    } else {
        gr.addQuery('impersonated', 'CONTAINS', name);
    }
    gr.query();
    var results = [];
    while (gr.next()) {
        results.push({
            user: gr.getDisplayValue('user'),
            impersonated: gr.getValue('impersonated'),
            sys_created_on: gr.getValue('sys_created_on')
        });
    }
    response.setBody({result: results});
})(request, response);'''


def _ensure_endpoint_exists(instance_url: str, headers: dict) -> None:
    """Create the Scripted REST endpoint if it doesn't exist.
    Also ensures the impersonation event property is enabled."""

    # Ensure the impersonation event property is enabled
    prop_resp = requests.get(
        f"{instance_url}/api/now/table/sys_properties",
        headers=headers,
        params={
            "sysparm_query": "name=com.glide.identity.security.impersonation.event.enabled",
            "sysparm_fields": "sys_id,value",
            "sysparm_limit": 1,
        },
    )
    if prop_resp.status_code == 200:
        props = prop_resp.json().get("result", [])
        if props and props[0].get("value") != "true":
            requests.patch(
                f"{instance_url}/api/now/table/sys_properties/{props[0]['sys_id']}",
                headers={**headers, "Content-Type": "application/json"},
                json={"value": "true"},
            )

    # Check if the scripted REST service already exists
    resp = requests.get(
        f"{instance_url}/api/now/table/sys_ws_definition",
        headers=headers,
        params={
            "sysparm_query": "name=Impersonation Checker",
            "sysparm_fields": "sys_id",
            "sysparm_limit": 1,
        },
    )
    resp.raise_for_status()
    if resp.json().get("result"):
        return

    # Create the service
    resp = requests.post(
        f"{instance_url}/api/now/table/sys_ws_definition",
        headers={**headers, "Content-Type": "application/json"},
        json={
            "name": "Impersonation Checker",
            "api_id": "impersonation_checker",
            "namespace": "global",
            "short_description": "Check impersonation events",
            "active": "true",
        },
    )
    resp.raise_for_status()
    service_id = resp.json()["result"]["sys_id"]

    # Create the resource
    requests.post(
        f"{instance_url}/api/now/table/sys_ws_operation",
        headers={**headers, "Content-Type": "application/json"},
        json={
            "name": "Check Impersonation",
            "http_method": "GET",
            "relative_path": "/check",
            "operation_script": SCRIPTED_REST_SCRIPT,
            "web_service_definition": service_id,
            "active": "true",
        },
    ).raise_for_status()

    # Wait for ServiceNow to register the new endpoint
    time.sleep(5)


def validate(messages: List[Dict[str, str]], config: Dict[str, Any]) -> int:
    """
    Validate that a user impersonation was performed by checking the
    sn_vsc_impersonation_event table via a Scripted REST endpoint.

    The endpoint is created automatically if it doesn't exist (instance
    environments are refreshed regularly).

    Required in `config`:
      - name: str  # the display name of the user to impersonate

    Returns:
      1 if a matching impersonation event is found, else 0.
    """

    if "name" not in config:
        raise ValueError("Config must contain 'name'")

    if not messages:
        raise ValueError("Messages must not be empty")

    expected_name = config["name"]
    instance_url = os.getenv("SERVICENOW_INSTANCE_URL", "").rstrip("/")
    extra_headers = json.loads(os.getenv("SERVICENOW_EXTRA_HTTP_HEADERS", "{}"))
    headers = {"Accept": "application/json", **extra_headers}

    # Ensure the Scripted REST endpoint exists on this instance
    _ensure_endpoint_exists(instance_url, headers)

    # Query the endpoint with retries (endpoint may need a moment after creation)
    for attempt in range(3):
        response = requests.get(
            f"{instance_url}{ENDPOINT_PATH}",
            headers=headers,
            params={"name": expected_name},
            timeout=10,
        )

        if response.status_code == 200:
            results = response.json().get("result", {}).get("result", [])
            if results:
                return 1
            break
        elif response.status_code in (400, 404):
            # Endpoint not yet registered — wait and retry
            time.sleep(3)
        else:
            break

    print(f"[VALIDATOR] FAIL: no impersonation event found for {expected_name!r}")
    return 0