Request Review Example Scripts

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")

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")

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")

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")

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)

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()