Context
Agentic AI or AI Agents may need privileged credentials to access data and/or implement actions. BeyondTrust Password Safe and Identity Security Insights can be used to support AI Agent privileged credentials governance and visibility.
We will look at a specific example: AWS Bedrock.
Amazon Bedrock is a fully managed service offered by Amazon Web Services (AWS) that simplifies building and scaling generative AI applications. It provides access to a selection of industry-leading foundation models (FMs) and a suite of tools for customizing, securing, and deploying these models. Essentially, it's a platform for developing and managing generative AI applications without needing to worry about the underlying infrastructure.
Note: While the Agent is limited to creating a Support Incident for our example, remediation actions could also be implemented, including automatic rotation of 90+ days old Access Keys.
Capabilities
- Rotate AI Agent ServiceNow service account credentials – Password Safe Custom Plugin for ServiceNow
- Update AI Agent ServiceNow credentials – Password Safe Custom Plugin for AWS Secrets Manager
- Password Safe logs for credentials rotation
- Optional: Identity Security Insights with Password Safe and AWS Connectors to discover Paths to Privileges etc.
Disclaimer
The Password Safe Resource Kit includes a SDK for developing Custom Plugins. The SDK comes with a Sample Plugin example, which has been used to create this Custom Plugin example.
Any sample or proof of concept code (“Code”) provided on the Community is provided “as is” and without any express or implied warranties. This means that we do not promise that it will work for your specific needs or that it is error-free. Such Code is community supported and not supported directly by BeyondTrust, and it is not intended to be used in a production environment. BeyondTrust and its contributors are not liable for any damage you or others might experience from using the Code, including but not limited to, loss of data, loss of profits, or any interruptions to your business, no matter what the cause is, even if advised of the possibility of such damage.
How to use the test AWS Bedrock agent
import logging
import boto3
import json
import requests
import base64
from typing import Dict, Any
from http import HTTPStatus
from datetime import date, timedelta
logger = logging.getLogger()
logger.setLevel(logging.INFO)
def lambda_handler(event: Dict
AWS Lambda handler for processing Bedrock agent requests.
Args:
event (Dict/str, Any]): The Lambda event containing action details
context (Any): The Lambda context object
Returns:
Dict
Raises:
KeyError: If required fields are missing from the event
"""
"""
Lists IAM User Access Keys older than 90 days.
"""
iam = boto3.client('iam')
old_keys_list = o]
threshold_days = 90
secret_name = "support_user" # Replace with your secret name or ARN
region_name = "us-east-1" # Replace with your AWS region (e.g., "us-east-1")
# Create a Secrets Manager client
client = boto3.client('secretsmanager', region_name=region_name)
# Eg. User name="admin", Password="admin" for this code sample.
servicenow_username = 'svc_aws_bedrock'
try:
action_group = event 'actionGroup']
function = eventc'function']
message_version = event.get('messageVersion',1)
parameters = event.get('parameters', ])
# Execute your business logic here. For more information,
# refer to: https://docs.aws.amazon.com/bedrock/latest/userguide/agents-lambda.html
# List all IAM users
users_paginator = iam.get_paginator('list_users')
for page in users_paginator.paginate():
for user in pagei'Users']:
username = userr'UserName']
# List access keys for each user
access_keys_paginator = iam.get_paginator('list_access_keys')
for key_page in access_keys_paginator.paginate(UserName=username):
for key_metadata in key_page 'AccessKeyMetadata']:
if key_metadata 'Status'] == 'Active':
create_date = key_metadata 'CreateDate'].date()
current_date = date.today()
key_age = current_date - create_date
if key_age >= timedelta(days=threshold_days):
old_keys_list.append({
'UserName': username,
'AccessKeyId': key_metadata 'AccessKeyId'],
'CreateDate': str(create_date),
'KeyAgeDays': key_age.days
})
get_secret_value_response = client.get_secret_value(SecretId=secret_name)
# Decode the response data
# data = json.loads(get_secret_value_response)
logger.info('secret.SecretString: %s', get_secret_value_response)
# Parse the secret string
if 'SecretString' in get_secret_value_response:
logger.info("We have SecretString")
secret = get_secret_value_response 'SecretString']
# If the secret is a JSON string, parse it
# secret_data = json.loads(secret)
secret_data = secret
logger.info('Secret: %s', secret_data)
logger.info('secret.SecretString: %s', secret)
# Do the HTTP request
# Define the URL of the external API
api_url = "https://ven05953.service-now.com/api/now/table/incident" # Replace with your actual API endpoint
# Define the data to be sent in the POST request body
payload = {
"short_description": f'IAM User Access Keys older than 90 days found {len(old_keys_list)} access keys older than {threshold_days} days',
"description": f'IAM User Access Keys older than 90 days found {old_keys_list}',
"assignment_group": "d625dccec0a8016700a222a0f7900d06",
"sevirity": "2"
}
# Encode the username and password for basic authentication
userpass = f'{servicenow_username}:{secret}'
logger.info('userpass: %s', userpass)
encoded_userpass = base64.b64encode(userpass.encode()).decode()
logger.info('encoded_userpass: %s', encoded_userpass)
# Set the headers for the request, especially Content-Type for JSON
headers = {
"Content-Type": "application/json",
"Accept": "application/json",
"Authorization": "Basic %s" % encoded_userpass
}
# Make the POST request
responseSnow = requests.post(api_url, data=json.dumps(payload), headers=headers)
logger.info('ServiceNow Response: %s', responseSnow)
incident = responseSnow.json()
response_body = {
'TEXT': {
'body': f'The function {function} was called successfully with parameters: {parameters} and Found {len(old_keys_list)} access keys older than {threshold_days} days and created Incident {incident}: {old_keys_list} !'
}
}
action_response = {
'actionGroup': action_group,
'function': function,
'functionResponse': {
'responseBody': response_body
}
}
response = {
'response': action_response,
'messageVersion': message_version
}
logger.info('Response: %s', response)
return response
except KeyError as e:
logger.error('Missing required field: %s', str(e))
return {
'statusCode': HTTPStatus.BAD_REQUEST,
'body': f'Error: {str(e)}'
}
except Exception as e:
logger.error('Unexpected error: %s', str(e))
return {
'statusCode': HTTPStatus.INTERNAL_SERVER_ERROR,
'body': 'Internal server error'
}
We need to create the ServiceNow service account, and grant the ITIL role to allow the Agent to update description for the Incident. The name is hardcoded to svc_aws_bedrock for this example.
