Utility Modules for OpalScript

# Check if user has any access to a resource
has_access = access.check_access(
    "550e8400-e29b-41d4-a716-446655440000",  # user ID
    "660e8400-e29b-41d4-a716-446655440001"   # resource ID
)

if has_access:
    print("User already has access")

# Check if user has a specific access level
is_admin = access.check_access(
    user_id,
    resource_id,
    "admin"
)

if is_admin:
    print("User is an admin")

# Check group membership
in_engineering = access.check_access(
    user_id,
    engineering_group_id
)

request = context.get_request()
notifications.notify_user(
    request.requester_id,
    "Access request flagged",
    "Your request has been flagged for manual review."
)

notifications.notify_admins(
    "Unusual access request",
    "A request for admin access was auto-denied due to an anomalous pattern."
)

notifications.notify_owner(
    "770e8400-e29b-41d4-a716-446655440002",
    "Request requires review",
    "A high-privilege access request has been submitted and needs manual approval."
)

provider = tickets.providers.JIRA

projects = tickets.list_projects(tickets.providers.JIRA)
eng_project = projects["ENG"]

request = context.get_request()
projects = tickets.list_projects(tickets.providers.JIRA)
ticket = tickets.create_ticket(
    projects["ENG"],
    "Access request review: " + request.id,
    "Requester: " + request.requester_id + ". Reason: " + request.reason
)
actions.comment("Ticket filed for review: " + ticket.url)

request = context.get_request()
ticket_id = request.custom_fields.get("ticket_id", None)
if ticket_id:
    ticket = tickets.get_ticket(tickets.providers.JIRA, ticket_id)
    if ticket.status == "CLOSED":
        actions.approve("Associated ticket is resolved")
    else:
        actions.comment("Waiting on open ticket: " + ticket.url)
else:
    actions.comment("No ticket ID found in request")

request = context.get_request()
projects = tickets.list_projects(tickets.providers.JIRA)
ticket = tickets.create_ticket(
    projects["ENG"],
    "Access request: " + request.id,
    request.reason
)
tickets.comment_ticket(
    tickets.providers.JIRA,
    ticket.identifier,
    "Automatically created by Opal for access request " + request.id
)
actions.comment("Ticket created and annotated: " + ticket.url)

request = context.get_request()
ticket_id = request.custom_fields.get("ticket_id", None)
if ticket_id:
    tickets.close_ticket(
        tickets.providers.JIRA,
        ticket_id,
        "Request denied by Opal automation."
    )
    actions.deny("Request denied and associated ticket closed.")
else:
    actions.deny("Request denied.")

# Create a ticket for high-risk requests, annotate it, then deny with a link
request = context.get_request()
resource = entity.get_resource(request.requested_resources[0].resource_id)
sensitivity = risk.resource_sensitivity(resource.id)

if sensitivity in ["CRITICAL", "HIGH"]:
    projects = tickets.list_projects(tickets.providers.JIRA)
    ticket = tickets.create_ticket(
        projects["SEC"],
        "High-risk access request: " + resource.name,
        "Request " + request.id + " for '" + resource.name + "' was auto-denied. Risk level: " + sensitivity
    )
    tickets.comment_ticket(
        tickets.providers.JIRA,
        ticket.identifier,
        "Requester: " + request.requester_id + ". Reason: " + request.reason
    )
    actions.deny("High-risk resource requires manual review. Ticket filed: " + ticket.url)
else:
    actions.approve("Resource risk level acceptable")

request = context.get_request()
resource_id = request.requested_resources[0]
sensitivity = risk.resource_sensitivity(resource_id)

if sensitivity == "CRITICAL":
    actions.deny("Cannot auto-approve access to critical resources")
elif sensitivity == "HIGH":
    actions.comment("High-risk resource - flagged for manual review")
else:
    actions.approve("Low-risk resource")

request = context.get_request()
group_sensitivity = risk.group_sensitivity(request.target_group_id)

if group_sensitivity in ["CRITICAL", "HIGH"]:
    actions.comment("This group contains high-risk assets - manual review required")
else:
    actions.approve("Group does not contain critical assets")

page = requests.by_resource(resource_id, "", "", "", 50)
page.next_cursor  # string — empty when no more pages

# All approved requests this user has been involved in
request = context.get_request()
prior = requests.by_user(request.requester_id, "APPROVED")

actions.comment("User has " + str(len(prior)) + " prior approved requests")

# Walk every request for this user one page at a time
cursor = ""
for _ in range(100):
    page = requests.by_user(user_id, "", cursor, 50)
    for request_id, prior_request in page.items():
        process(prior_request)
    if page.next_cursor == "":
        break
    cursor = page.next_cursor

# How many times has anyone been granted access to this resource?
request = context.get_request()
resource_id = request.requested_resources[0].resource_id
approved = requests.by_resource(resource_id, "", "APPROVED")
actions.comment("Approved " + str(len(approved)) + " times previously")

# Has the requester ever been denied this resource before?
request = context.get_request()
resource_id = request.requested_resources[0].resource_id
prior_denials = requests.by_resource(
    resource_id,
    request.requester_id,
    "DENIED",
)
if len(prior_denials) > 0:
    actions.comment("Requester previously denied this resource")

# Walk all approved requests one page at a time
request = context.get_request()
resource_id = request.requested_resources[0].resource_id
cursor = ""
for _ in range(100):
    page = requests.by_resource(resource_id, "", "APPROVED", cursor, 50)
    for request_id, prior_request in page.items():
        process(prior_request)
    if page.next_cursor == "":
        break
    cursor = page.next_cursor

# Pending requests for this group from the same requester
request = context.get_request()
overlapping = 0
for requested_group in request.requested_groups:
    prior = requests.by_group(
        requested_group.group_id,
        request.requester_id,
        "PENDING",
    )
    if len(prior) > 1:
        overlapping += 1
if overlapping > 0:
    actions.comment("Requester has overlapping pending requests for " + str(overlapping) + " groups")

current_time = time.now()
actions.comment("Request processed at " + time.from_unix(current_time))

request = context.get_request()
request_time_str = time.from_unix(request.created_at)
actions.comment("Request was created at " + request_time_str)

request = context.get_request()
now = time.now()

# Deny requests older than 30 days
if time.is_before(request.created_at, now - time.days(30)):
    actions.deny("Request has expired (older than 30 days)")

# Auto-approve short-duration recent requests
if time.is_after(request.created_at, now - time.days(7)):
    if request.requested_duration_minutes <= 60:
        actions.approve("Recent short-duration request")

request = context.get_request()
age_seconds = time.seconds_since(time.now(), request.created_at)
actions.comment("Request is " + str(age_seconds) + " seconds old")

# Convert time intervals to seconds
five_minutes = time.minutes(5)      # Returns 300
two_hours = time.hours(2)           # Returns 7200
thirty_days = time.days(30)         # Returns 2592000

# Use in time comparisons
request = context.get_request()
deadline = request.created_at + time.days(7)

if time.is_after(time.now(), deadline):
    actions.deny("Request expired after 7 days")
else:
    actions.approve("Request is within the 7-day window")

# Check request duration
if request.requested_duration_minutes < 5:
    actions.approve("Very short-duration request")
elif request.requested_duration_minutes <= time.hours(24) // 60:
    actions.comment("Duration is 24 hours or less - acceptable")
else:
    actions.comment("Duration exceeds 24 hours - requires review")

request = context.get_request()
user = entity.get_user(request.requester_id)
department = user.tags.get("department", None)

# Look up a user's manager
request = context.get_request()
user = entity.get_user(request.requester_id)
if user.manager_id:
    manager = entity.get_user(user.manager_id)
    notifications.notify_user(manager.id, "Access request pending", "A member of your team has submitted an access request.")

request = context.get_request()
for requested_group in request.requested_groups:
    group = entity.get_group(requested_group.group_id)
    cost_center = group.tags.get("cost_center", None)

request = context.get_request()
for requested_resource in request.requested_resources:
    resource = entity.get_resource(requested_resource.resource_id)
    env = resource.tags.get("env", None)
    if env == "production":
        actions.deny("Production access requires manual review.")

request = context.get_request()
user = entity.get_user(request.requester_id)

# Safe access with default
department = user.tags.get("department", "Unknown")

# Check if a tag exists before accessing
if "department" in user.tags:
    actions.comment("Department: " + user.tags["department"])

request = context.get_request()
requester = entity.get_user(request.requester_id)
auth = {"Authorization": "Bearer " + secrets.get("example_api_token")}

# GET with query params and an auth header, failing closed on error
resp = http.get(
    url="https://api.example.com/v1/devices",
    headers=auth,
    params={"user_email": requester.email},
)
if not resp.ok:
    actions.deny("Could not reach the device API (status %d)" % resp.status_code)

# POST JSON
http.post(
    url="https://api.example.com/v1/tickets",
    headers=auth,
    json={"title": "Access granted for " + requester.email, "priority": "low"},
)

resp = http.get(url="https://api.example.com/v1/devices/123")
if resp.ok:
    device = resp.json()
    if not device["is_compliant"]:
        actions.deny("Device is not compliant with security policy")

# Composed into an auth header (the result stays opaque)
auth = {"Authorization": "Bearer " + secrets.get("example_api_token")}
http.get(url="https://api.example.com/v1/things", headers=auth)

# As the URL itself (for example a chat webhook stored as a secret)
http.post(
    url=secrets.get("slack_webhook_url"),
    json={"text": "Access request approved"},
)

# The value is never visible, even by accident:
print(secrets.get("example_api_token"))   # logs ***