al0ne/cloud-auditAutomated cloud security auditing tool that detects AK/SK credential misuse by periodically auditing cloud platform logs using anomaly detection, blacklist rules, and baseline analysis for Tencent Cloud and AWS.
cloud-audit (Cloud Security Audit Assistant) is a tool for detecting the exploitation of leaked AK/SK from public cloud vendors. It periodically calls cloud platform APIs to audit logs and identifies suspected intrusions based on abnormal behavior, black features, and baselines.
Currently only supports Tencent Cloud and AWS
Step 1: Run the command to install dependencies
cd /opt/ && git clone https://github.com/al0ne/cloud-audit
cd cloud-audit && pip3 install -r requirements.txt
Step 2: Create and modify the .env configuration
cp .env.example .env
Fill in the .env file according to your actual configuration
# Tencent Cloud AK/SK information, used to call API interfaces to obtain logs
TencentAccessKey="xxx"
TencentSecretKey="xxx"
# List of AK/SK to monitor on Tencent Cloud
AccesskeyList="xxx,xxx,xxx"
# AWS AK/SK information
AWS_ACCESS_KEY_ID="xxx"
AWS_SECRET_ACCESS_KEY="xxx"
# Discord notification information
discord_webhook_url=""
# WeCom robot notification
weixin_webhook_url=""
Step 3: Important program configuration
# Trusted network segments. Sensitive operations from untrusted segments will trigger alerts directly.
while_cidr = ["192.168.0.0/16", "172.16.0.0/12", "10.0.0.0/8"]
# AWS regions of interest. Events outside these regions will not be queried
aws_region = [
"us-east-1",
"us-east-2",
"ap-northeast-1",
"ap-southeast-1"
]
Run
python3 cloud-audit.py
The program executes and searches for results every 15 minutes
Since AWS logs are read and audited using the AWS CloudTrail service, please ensure your account's IAM permissions include AWSCloudTrail_ReadOnlyAccess to use this product, otherwise you may encounter the following error:
botocore.exceptions.ClientError: An error occurred (AccessDeniedException) when calling the LookupEvents operation: User: arn:aws:iam::xxxxx:user/test is not authorized to perform: cloudtrail:LookupEvents because no identity-based policy allows the cloudtrail:LookupEvents action
Go to the AWS IAM console, select Users -> Permissions to add permissions.
When using some cloud platform exploitation tools, typical actions include scanning all regions for RDS instances, ECS instances, or containers, creating/deleting sub-accounts, executing commands, etc. These actions themselves are relatively sensitive, so they are obtained through the log interfaces of each cloud platform.
Another approach is based on sensitive operations from untrusted network segments. Normally, calls come from the IDC internal network or IDC egress IPs. If AK/SK calls appear from non-enterprise trusted network segments and involve high-risk operations, an alert is triggered directly.
Common sensitive key operations include:
The biggest difference between Tencent Cloud and AWS API calls is that Tencent Cloud can query the detailed actions of a specific AK ID through an API, while AWS can only search on the platform and cannot query the recent execution information of a specific AK ID via AK/SK API calls.
Therefore, to detect Tencent Cloud AK/SK exploitation, you need to input a list of AK IDs used by online business to monitor, and periodically check the actions performed by a specific AK ID.
AWS can only obtain logs based on region and action, calling the CloudTrail API with AK/SK to check if there are logs for a specific action in a specific region.
Therefore, AWS detection relies more on baselines.
def DescribeInstances(CloudTrailEvent: dict):
"""
Detect AK/SK used to list EC2 instances
:param CloudTrailEvent:
:return: None
"""
accessKeyId = CloudTrailEvent.get('userIdentity').get('accessKeyId')
arn = CloudTrailEvent.get('userIdentity').get('arn')
eventTime = utc_to_china_tz(CloudTrailEvent.get('eventTime'))
eventName = CloudTrailEvent.get('eventName')
awsRegion = CloudTrailEvent.get('awsRegion')
sourceIPAddress = CloudTrailEvent.get('sourceIPAddress')
userAgent = CloudTrailEvent.get('userAgent')
sip_verify = False
for cidr in while_cidr:
if ipaddress.ip_address(sourceIPAddress) in ipaddress.ip_network(cidr):
sip_verify = True
if not sip_verify and 'aws-internal' not in userAgent:
message = f"{text_title}\nTime: {eventTime}\nAccount ID: {arn}\nRegion: {awsRegion}\nAccessKey ID:{accessKeyId}\n" \
f"Action: {eventName}\nIP Address: {sourceIPAddress}\nUser-Agent: {userAgent}\n\n" \
f"Detected an external IP request to list instance information. Please check if it is being exploited by an attacker!"
send_message(message)
Discord alert screenshot
Currently only detects based on behavior. In practice, it can also detect based on baselines.