Request Review Example Scripts

Learn to use Request Review scripts through examples.

The following examples illustrate the different use cases for our Request Review scripts, and how you might implement it in your environment.

​

Auto-approve based on duration

request = context.get_request()

if request.requested_duration_minutes and request.requested_duration_minutes <= 240:
    actions.approve("Auto-approved: 4 hours or less")
else:
    actions.comment("Duration exceeds auto-approval threshold")

​

Route based on resource type

request = context.get_request()

for resource in request.requested_resources:
    if resource.resource_type == "AWS_IAM_ROLE":
        if "prod" in resource.resource_name.lower():
            actions.comment("Production AWS access requires manual review")
            break
else:
    # No production resources found
    actions.approve("Auto-approved: non-production access")

​

Validate custom fields

request = context.get_request()
custom_fields = request.custom_fields

# Require ticket number for non-emergency requests
is_emergency = custom_fields.get("emergency_access", False)
ticket_number = custom_fields.get("ticket_number", "")

if is_emergency:
    actions.approve("Auto-approved: emergency access")
elif ticket_number:
    actions.approve("Auto-approved: ticket " + ticket_number)
else:
    actions.deny("A ticket number is required for non-emergency access")

​

Check prerequisite access

request = context.get_request()

PREREQUISITE_GROUP = "550e8400-e29b-41d4-a716-446655440000"

if request.target_user_id:
    has_prereq = access.check_access(
        request.target_user_id,
        PREREQUISITE_GROUP
    )

    if has_prereq:
        actions.approve("User has prerequisite access")
    else:
        actions.deny("User must first obtain access to the prerequisite group")
else:
    actions.comment("No target user specified")

​

Grant a required group on approval

Add the user to a group that the requested access depends on, such as the SSO group an app needs, and then approve. This example uses the groups module and is available as the Grant a required group on approval template in the OpalScript editor.

Use this script as the only reviewer in the final approval stage, after your human reviewers. It always approves. If the user can’t be added to the group, for example because they’re already a member, it still approves and includes the error in its comment.

Replace the group ID with your own.

REQUIRED_GROUP = "REPLACE-WITH-GROUP-ID"

request = context.get_request()
user_id = request.target_user_id or request.requester_id

err = groups.add_user(REQUIRED_GROUP, user_id, indefinite = True)
if not err:
    actions.approve("Approved and added the user to the required group.")
else:
    actions.approve("Approved, but the user could not be added to the required group. Check whether they are already a member, or add them manually: " + err)

​

Separation of duties: remove a conflicting group

Before approving, remove the user from a group that must not be combined with the access they’re requesting. This example uses the groups module and is available as the Separation of duties: remove conflicting group template in the OpalScript editor.

One groups.remove_user call removes all of the user’s direct memberships in the group, and does nothing if the user isn’t a member. If the removal fails, or if the user’s only access comes from a nested group, an on-call schedule, or break-glass access, the script doesn’t approve. It adds a comment and the request continues to its human reviewers.

Replace the group ID with your own conflicting group.

CONFLICTING_GROUP = "REPLACE-WITH-CONFLICTING-GROUP-ID"

request = context.get_request()
user_id = request.target_user_id or request.requester_id

err = groups.remove_user(CONFLICTING_GROUP, user_id, reason = "Separation of duties with the requested access.")
if err:
    actions.comment("Could not remove the user from the conflicting group; routing to a human: " + err)
else:
    actions.approve("Auto-approved: the user is out of the conflicting group.")

​

Complex multi-condition logic

def evaluate_request(request):
    reason_lower = request.reason.lower()

    # Check deny conditions
    if "permanent" in reason_lower:
        return ("deny", "Permanent access requires executive approval")

    # Calculate approval score
    score = 0

    # Short duration
    if request.requested_duration_minutes:
        if request.requested_duration_minutes <= 240:
            score = score + 2
        elif request.requested_duration_minutes <= 480:
            score = score + 1

    # Has detailed reason
    if len(request.reason) >= 50:
        score = score + 1

    # Urgent keyword
    if "urgent" in reason_lower or "emergency" in reason_lower:
        score = score + 2

    # Make decision
    if score >= 3:
        return ("approve", "Auto-approved: score " + str(score))
    else:
        return ("comment", "Score " + str(score) + ", requires manual review")

request = context.get_request()
decision, message = evaluate_request(request)

if decision == "approve":
    actions.approve(message)
elif decision == "deny":
    actions.deny(message)
else:
    actions.comment(message)

​

Block cross-region access

Deny a request when a Europe-based requester asks for a US-based resource. Location comes from tags that Opal already keeps in sync, so the script needs no external call and no secret.

The script fails closed to human review: when the requester’s country or a resource’s region can’t be determined, it takes no decision and the request continues to its normal reviewers. A deny applies to the whole request, so one US-based resource in a multi-resource request denies all of it.

Point the two tag keys at your own tagging scheme. Tag values are compared uppercased, so list whatever spellings your tags use in uppercase (for example both DE and GERMANY).

USER_COUNTRY_TAG = "country"
RESOURCE_REGION_TAG = "region"

EUROPE_COUNTRIES = [
    "AL", "AD", "AT", "BA", "BE", "BG", "BY", "CH", "CY", "CZ", "DE", "DK", "EE",
    "ES", "FI", "FR", "GB", "GR", "HR", "HU", "IE", "IS", "IT", "LI", "LT", "LU",
    "LV", "MC", "MD", "ME", "MK", "MT", "NL", "NO", "PL", "PT", "RO", "RS", "SE",
    "SI", "SK", "SM", "UA", "VA", "XK",
]

US_REGIONS = ["US", "USA", "US-EAST", "US-WEST", "AMER"]


# A tag key holds a single value, None, or a list when the key is multi-valued.
def tag_values(tags, key):
    if not tags:
        return []
    value = tags.get(key)
    if value == None:
        return []
    if type(value) == "list":
        return [v.upper() for v in value if v != None]
    return [value.upper()]


def evaluate():
    request = context.get_request()
    subject = entity.get_user(request.target_user_id or request.requester_id)

    countries = tag_values(subject.tags, USER_COUNTRY_TAG)
    if not countries:
        actions.comment(
            "Could not determine %s's country from the '%s' tag. Routing to a human."
            % (subject.full_name, USER_COUNTRY_TAG)
        )
        return

    if not any([country in EUROPE_COUNTRIES for country in countries]):
        return

    us_resources = []
    for requested in request.requested_resources:
        regions = tag_values(entity.get_resource(requested.resource_id).tags, RESOURCE_REGION_TAG)
        if any([region in US_REGIONS for region in regions]):
            us_resources.append(requested.resource_name)

    if us_resources:
        actions.deny(
            "%s is based in Europe and cannot be granted access to US-based resources: %s."
            % (subject.full_name, ", ".join(us_resources))
        )


evaluate()

​

Deny based on device posture with FleetDM

Deny an access request when the requester’s device is failing any FleetDM policy, and approve otherwise. It shows the full pattern for the http and secrets utility modules: secret composition for auth, a fail-closed helper, and terminal actions.deny / actions.approve.

Prerequisites (OpalScript → Advanced, on the Secrets and Allowed hosts tabs):

1. Egress allowlist: add your Fleet host, for example fleet.example.com with scheme https. It must be a public host. The SSRF-safe client blocks private and loopback addresses even when allowlisted.
2. Secret: add FLEET_API_TOKEN holding the raw token (no Bearer prefix). The script prepends Bearer with secret composition, so the combined value stays opaque.

Fleet’s host search matches hostname, serial, uuid, or ipv4, not email. The email-to-host link lives in Fleet’s device mapping, so this matches the requester’s email against each host’s mapping, making one call per host until it finds a match. The per-execution budget is 50 HTTP calls, so this suits small fleets and demos. For larger fleets, page through /hosts (page / per_page), narrow by team or status, or maintain an email-to-host index outside the script. Otherwise, a requester whose host is not on the first page is denied as unmapped.

FLEET_BASE_URL = "https://fleet.example.com"

def fleet_get(path, auth, params={}):
    """GET a Fleet endpoint and return parsed JSON. Fail closed on error."""
    resp = http.get(url=FLEET_BASE_URL + path, headers=auth, params=params)
    if not resp.ok:
        actions.deny(
            comment="Could not verify device posture: Fleet %s returned status %d"
            % (path, resp.status_code)
        )
    return resp.json()

def find_host_id_by_email(email, auth):
    """Find the Fleet host whose device mapping includes the email."""
    body = fleet_get("/api/v1/fleet/hosts", auth, {"per_page": "100"})
    hosts = body.get("hosts") or []

    for host in hosts:
        path = "/api/v1/fleet/hosts/" + str(host["id"]) + "/device_mapping"
        mapping = fleet_get(path, auth)
        for entry in mapping.get("device_mapping") or []:
            if entry["email"] == email:
                return host["id"]

    actions.deny(comment="No Fleet host mapped to " + email)

def failing_policies(host_id, auth):
    """Return the names of every failing policy on the host."""
    body = fleet_get("/api/v1/fleet/hosts/" + str(host_id), auth)
    host = body["host"]

    failing = []
    for policy in host.get("policies") or []:
        if policy["response"] == "fail":
            failing.append(policy["name"])
    return failing

def main():
    request = context.get_request()
    auth = {"Authorization": "Bearer " + secrets.get("FLEET_API_TOKEN")}

    requester = entity.get_user(request.requester_id)
    host_id = find_host_id_by_email(requester.email, auth)
    failing = failing_policies(host_id, auth)

    if failing:
        actions.deny(comment="Failing FleetDM policies: " + ", ".join(failing))
    else:
        actions.approve(comment="Device passed all FleetDM policies")

main()