feat(ec2): add ec2_instance_stopped_older_than_specific_days check (#12076)

Co-authored-by: Daniel Barranquero <danielbo2001@gmail.com>
This commit is contained in:
Nithin ReddyandDaniel Barranquero authored and GitHub committed 2026-07-27 09:13:18 +02:00
1 parent d933a8ecab
commit 339930ef13
15 files changed
+421

No files matched your search

@@ -133,6 +133,7 @@ Only fields with a numeric range, a fixed value set, or a length cap are listed.
| `max_unused_sagemaker_access_days` | `7..180` days | |
| `max_security_group_rules` | `1..1000` | AWS hard limit is 1000 rules per security group |
| `max_ec2_instance_age_in_days` | `1..1095` days | 3 years |
| `max_ec2_instance_stopped_days` | `1..1095` days | 3 years |
| `ec2_high_risk_ports` | each port `1..65535` | port 0 is reserved |
| `max_idle_disconnect_timeout_in_seconds` | `60..1800` s | NIST AC-12: cap at 30 min |
| `max_disconnect_timeout_in_seconds` | `60..3600` s | |
@@ -63,6 +63,7 @@ The following list includes all the AWS checks with configurable variables that
| `dynamodb_table_cross_account_access` | `trusted_account_ids` | List of Strings | `[]` |
| `ec2_elastic_ip_shodan` | `shodan_api_key` | String | `null` |
| `ec2_instance_older_than_specific_days` | `max_ec2_instance_age_in_days` | Integer | `180` |
| `ec2_instance_stopped_older_than_specific_days` | `max_ec2_instance_stopped_days` | Integer | `30` |
| `ec2_instance_secrets_user_data` | `secrets_ignore_patterns` | List of Strings | `[]` |
| `ec2_launch_template_no_secrets` | `secrets_ignore_patterns` | List of Strings | `[]` |
| `ec2_securitygroup_allow_ingress_from_internet_to_any_port` | `ec2_allowed_instance_owners` | List of Strings | `["amazon-elb"]` |
@@ -392,6 +393,8 @@ aws:
max_security_group_rules: 50
# aws.ec2_instance_older_than_specific_days --> by default is 6 months (180 days)
max_ec2_instance_age_in_days: 180
# aws.ec2_instance_stopped_older_than_specific_days --> by default is 30 days
max_ec2_instance_stopped_days: 30
# aws.ec2_securitygroup_allow_ingress_from_internet_to_any_port
# allowed network interface types for security groups open to the Internet
ec2_allowed_interface_types:
@@ -0,0 +1 @@
`ec2_instance_stopped_older_than_specific_days` check for AWS provider, detecting EC2 instances stopped longer than a configurable number of days (default 30)
+2
View File
@@ -63,6 +63,8 @@ aws:
max_security_group_rules: 50
# aws.ec2_instance_older_than_specific_days --> by default is 6 months (180 days)
max_ec2_instance_age_in_days: 180
# aws.ec2_instance_stopped_older_than_specific_days --> by default is 30 days
max_ec2_instance_stopped_days: 30
# aws.ec2_securitygroup_allow_ingress_from_internet_to_any_port
# allowed network interface types for security groups open to the Internet
ec2_allowed_interface_types:
+9
View File
@@ -213,6 +213,15 @@ class AWSProviderConfig(ProviderConfigBase):
"per NIST CM-3 — anything older is a security smell)."
),
)
max_ec2_instance_stopped_days: Optional[int] = Field(
default=None,
ge=1,
le=1095,
description=(
"Days an EC2 instance can remain stopped before being flagged. "
"Range: 1..1095 (3 years)."
),
)
ec2_allowed_interface_types: Optional[list[str]] = None
ec2_allowed_instance_owners: Optional[list[str]] = None
ec2_high_risk_ports: Annotated[
@@ -0,0 +1,41 @@
{
"Provider": "aws",
"CheckID": "ec2_instance_stopped_older_than_specific_days",
"CheckTitle": "EC2 instance has not been stopped longer than the configured maximum days",
"CheckType": [
"Software and Configuration Checks/AWS Security Best Practices",
"Software and Configuration Checks/Patch Management"
],
"ServiceName": "ec2",
"SubServiceName": "",
"ResourceIdTemplate": "arn:partition:ec2:region:account-id:instance/instance-id",
"Severity": "low",
"ResourceType": "AwsEc2Instance",
"ResourceGroup": "compute",
"Description": "**EC2 instances** in the `stopped` state are evaluated for how long they have remained stopped. Instances stopped beyond the configurable limit (`max_ec2_instance_stopped_days`, default `30`) are flagged. Running, pending, and other non-stopped instances pass.",
"Risk": "Long-stopped instances remain **unmonitored and unpatched** while still retaining EBS volumes, network configuration, IAM instance profiles, and SSH keys. This expands attack surface and incurs unnecessary storage cost. Stale AMIs and credentials attached to abandoned instances increase the chance of compromise if the instance is later started.",
"RelatedUrl": "",
"AdditionalURLs": [
"https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/ec2-instance-lifecycle.html",
"https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/Stop_Start.html",
"https://docs.aws.amazon.com/AWSEC2/latest/APIReference/API_DescribeInstances.html"
],
"Remediation": {
"Code": {
"CLI": "aws ec2 terminate-instances --instance-ids <example_resource_id>",
"NativeIaC": "",
"Other": "1. Sign in to the AWS Management Console and open EC2\n2. Go to Instances and select the long-stopped instance\n3. Review attached EBS volumes, tags, and ownership\n4. Choose Instance state > Terminate instance (or Start instance if still needed, then patch/rebuild)\n5. Confirm and verify the instance is terminated or returned to an actively managed lifecycle",
"Terraform": ""
},
"Recommendation": {
"Text": "Establish a lifecycle policy for stopped instances:\n- Tag instances with owner and expiry\n- Terminate instances no longer needed to reclaim EBS cost\n- If retained, start regularly for patching or rebuild from a hardened AMI\n- Prefer ephemeral, autoscaled workloads over long-lived stopped hosts\n\nAdjust `max_ec2_instance_stopped_days` to match policy.",
"Url": "https://hub.prowler.com/check/ec2_instance_stopped_older_than_specific_days"
}
},
"Categories": [],
"DependsOn": [],
"RelatedTo": [
"ec2_instance_older_than_specific_days"
],
"Notes": ""
}
@@ -0,0 +1,86 @@
import re
from datetime import datetime, timezone
from typing import Optional
from prowler.lib.check.models import Check, Check_Report_AWS
from prowler.providers.aws.services.ec2.ec2_client import ec2_client
# AWS StateTransitionReason for stopped instances, e.g.:
# "User initiated (2016-09-14 15:07:39 GMT)"
_STATE_TRANSITION_TIME_REGEX = re.compile(
r"\((\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}) GMT\)"
)
def _parse_state_transition_time(
state_transition_reason: Optional[str],
) -> Optional[datetime]:
"""Extract the stop timestamp from EC2 StateTransitionReason."""
if not state_transition_reason:
return None
match = _STATE_TRANSITION_TIME_REGEX.search(state_transition_reason)
if not match:
return None
try:
return datetime.strptime(match.group(1), "%Y-%m-%d %H:%M:%S").replace(
tzinfo=timezone.utc
)
except ValueError:
return None
class ec2_instance_stopped_older_than_specific_days(Check):
"""Ensure EC2 instances are not stopped longer than a configured number of days.
Evaluates each instance's stop duration using StateTransitionReason.
- PASS: Instance is not stopped, or has been stopped for at most the threshold.
- FAIL: Instance has been stopped longer than max_ec2_instance_stopped_days
(default 30).
"""
def execute(self) -> list[Check_Report_AWS]:
"""Execute the check logic.
Returns:
A list of reports containing the result of the check.
"""
findings = []
# max_ec2_instance_stopped_days, default: 30 days
max_ec2_instance_stopped_days = ec2_client.audit_config.get(
"max_ec2_instance_stopped_days", 30
)
for instance in ec2_client.instances:
report = Check_Report_AWS(metadata=self.metadata(), resource=instance)
report.resource_id = instance.id
report.resource_arn = instance.arn
report.resource_tags = instance.tags
report.status = "PASS"
report.status_extended = f"EC2 Instance {instance.id} is not stopped."
if instance.state == "stopped":
stop_time = _parse_state_transition_time(
instance.state_transition_reason
)
if not stop_time:
report.status_extended = (
f"EC2 Instance {instance.id} is stopped but stop time "
f"could not be determined."
)
else:
days_stopped = (datetime.now(timezone.utc) - stop_time).days
report.status_extended = (
f"EC2 Instance {instance.id} has not been stopped longer "
f"than {max_ec2_instance_stopped_days} days "
f"({days_stopped} days)."
)
if days_stopped > max_ec2_instance_stopped_days:
report.status = "FAIL"
report.status_extended = (
f"EC2 Instance {instance.id} has been stopped longer "
f"than {max_ec2_instance_stopped_days} days "
f"({days_stopped} days)."
)
findings.append(report)
return findings
@@ -100,6 +100,9 @@ class EC2(AWSService):
type=instance["InstanceType"],
image_id=instance["ImageId"],
launch_time=instance["LaunchTime"],
state_transition_reason=instance.get(
"StateTransitionReason"
),
private_dns=instance["PrivateDnsName"],
private_ip=instance.get("PrivateIpAddress"),
public_dns=instance.get("PublicDnsName"),
@@ -761,6 +764,7 @@ class Instance(BaseModel):
type: str
image_id: str
launch_time: datetime
state_transition_reason: Optional[str] = None
private_dns: str
private_ip: Optional[str]
public_dns: Optional[str]
+2
View File
@@ -35,6 +35,7 @@ old_config_aws = {
"shodan_api_key": None,
"max_security_group_rules": 50,
"max_ec2_instance_age_in_days": 180,
"max_ec2_instance_stopped_days": 30,
"ec2_allowed_interface_types": ["api_gateway_managed", "vpc_endpoint"],
"ec2_allowed_instance_owners": ["amazon-elb"],
"trusted_account_ids": [],
@@ -86,6 +87,7 @@ config_aws = {
"shodan_api_key": None,
"max_security_group_rules": 50,
"max_ec2_instance_age_in_days": 180,
"max_ec2_instance_stopped_days": 30,
"ec2_allowed_interface_types": ["api_gateway_managed", "vpc_endpoint"],
"ec2_allowed_instance_owners": ["amazon-elb"],
"ec2_high_risk_ports": [
+2
View File
@@ -31,6 +31,8 @@ aws:
max_security_group_rules: 50
# aws.ec2_instance_older_than_specific_days --> by default is 6 months (180 days)
max_ec2_instance_age_in_days: 180
# aws.ec2_instance_stopped_older_than_specific_days --> by default is 30 days
max_ec2_instance_stopped_days: 30
# aws.ec2_securitygroup_allow_ingress_from_internet_to_any_port
# allowed network interface types for security groups open to the Internet
ec2_allowed_interface_types:
+2
View File
@@ -5,6 +5,8 @@ shodan_api_key: null
max_security_group_rules: 50
# aws.ec2_instance_older_than_specific_days --> by default is 6 months (180 days)
max_ec2_instance_age_in_days: 180
# aws.ec2_instance_stopped_older_than_specific_days --> by default is 30 days
max_ec2_instance_stopped_days: 30
# aws.ec2_securitygroup_allow_ingress_from_internet_to_any_port
# allowed network interface types for security groups open to the Internet
ec2_allowed_interface_types:
+1
View File
@@ -27,6 +27,7 @@ INT_BOUND_CASES = [
("aws", "max_unused_sagemaker_access_days", 7, 180),
("aws", "max_security_group_rules", 1, 1000),
("aws", "max_ec2_instance_age_in_days", 1, 1095),
("aws", "max_ec2_instance_stopped_days", 1, 1095),
("aws", "recommended_cdk_bootstrap_version", 1, 100),
("aws", "max_idle_disconnect_timeout_in_seconds", 60, 1800),
("aws", "max_disconnect_timeout_in_seconds", 60, 3600),
@@ -0,0 +1,252 @@
from datetime import datetime, timedelta, timezone
from re import search
from unittest import mock
from boto3 import resource
from moto import mock_aws
from tests.providers.aws.utils import (
AWS_REGION_EU_WEST_1,
AWS_REGION_US_EAST_1,
set_mocked_aws_provider,
)
EXAMPLE_AMI_ID = "ami-12c6146b"
class Test_ec2_instance_stopped_older_than_specific_days:
@mock_aws
def test_ec2_no_instances(self):
from prowler.providers.aws.services.ec2.ec2_service import EC2
aws_provider = set_mocked_aws_provider(
[AWS_REGION_EU_WEST_1, AWS_REGION_US_EAST_1]
)
aws_provider._audit_config = {"max_ec2_instance_stopped_days": 30}
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
),
mock.patch(
"prowler.providers.aws.services.ec2.ec2_instance_stopped_older_than_specific_days.ec2_instance_stopped_older_than_specific_days.ec2_client",
new=EC2(aws_provider),
),
):
from prowler.providers.aws.services.ec2.ec2_instance_stopped_older_than_specific_days.ec2_instance_stopped_older_than_specific_days import (
ec2_instance_stopped_older_than_specific_days,
)
check = ec2_instance_stopped_older_than_specific_days()
result = check.execute()
assert len(result) == 0
@mock_aws
def test_running_ec2(self):
ec2 = resource("ec2", region_name=AWS_REGION_US_EAST_1)
instance = ec2.create_instances(
ImageId=EXAMPLE_AMI_ID,
MinCount=1,
MaxCount=1,
UserData="This is some user_data",
)[0]
from prowler.providers.aws.services.ec2.ec2_service import EC2
aws_provider = set_mocked_aws_provider(
[AWS_REGION_EU_WEST_1, AWS_REGION_US_EAST_1]
)
aws_provider._audit_config = {"max_ec2_instance_stopped_days": 30}
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
),
mock.patch(
"prowler.providers.aws.services.ec2.ec2_instance_stopped_older_than_specific_days.ec2_instance_stopped_older_than_specific_days.ec2_client",
new=EC2(aws_provider),
),
):
from prowler.providers.aws.services.ec2.ec2_instance_stopped_older_than_specific_days.ec2_instance_stopped_older_than_specific_days import (
ec2_instance_stopped_older_than_specific_days,
)
check = ec2_instance_stopped_older_than_specific_days()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert result[0].region == AWS_REGION_US_EAST_1
assert result[0].resource_tags is None
assert search(
f"EC2 Instance {instance.id} is not stopped",
result[0].status_extended,
)
assert result[0].resource_id == instance.id
assert (
result[0].resource_arn
== f"arn:{aws_provider.identity.partition}:ec2:{AWS_REGION_US_EAST_1}:{aws_provider.identity.account}:instance/{instance.id}"
)
@mock_aws
def test_stopped_ec2_within_threshold(self):
ec2 = resource("ec2", region_name=AWS_REGION_US_EAST_1)
instance = ec2.create_instances(
ImageId=EXAMPLE_AMI_ID,
MinCount=1,
MaxCount=1,
UserData="This is some user_data",
)[0]
from prowler.providers.aws.services.ec2.ec2_service import EC2
aws_provider = set_mocked_aws_provider(
[AWS_REGION_EU_WEST_1, AWS_REGION_US_EAST_1]
)
aws_provider._audit_config = {"max_ec2_instance_stopped_days": 30}
# Boundary: stopped exactly 30 days ago must remain PASS
# (threshold is exclusive: days_stopped > max_ec2_instance_stopped_days).
fixed_now = datetime(2024, 6, 15, 12, 0, 0, tzinfo=timezone.utc)
recent_stop = fixed_now - timedelta(days=30)
stop_reason = recent_stop.strftime("User initiated (%Y-%m-%d %H:%M:%S GMT)")
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
),
mock.patch(
"prowler.providers.aws.services.ec2.ec2_instance_stopped_older_than_specific_days.ec2_instance_stopped_older_than_specific_days.ec2_client",
new=EC2(aws_provider),
) as service_client,
mock.patch(
"prowler.providers.aws.services.ec2.ec2_instance_stopped_older_than_specific_days.ec2_instance_stopped_older_than_specific_days.datetime"
) as mock_datetime,
):
from prowler.providers.aws.services.ec2.ec2_instance_stopped_older_than_specific_days.ec2_instance_stopped_older_than_specific_days import (
ec2_instance_stopped_older_than_specific_days,
)
mock_datetime.now.return_value = fixed_now
mock_datetime.strptime = datetime.strptime
service_client.instances[0].state = "stopped"
service_client.instances[0].state_transition_reason = stop_reason
check = ec2_instance_stopped_older_than_specific_days()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert result[0].region == AWS_REGION_US_EAST_1
assert search(
f"EC2 Instance {instance.id} has not been stopped longer than",
result[0].status_extended,
)
assert result[0].resource_id == instance.id
assert (
result[0].resource_arn
== f"arn:{aws_provider.identity.partition}:ec2:{AWS_REGION_US_EAST_1}:{aws_provider.identity.account}:instance/{instance.id}"
)
@mock_aws
def test_stopped_ec2_older_than_threshold(self):
ec2 = resource("ec2", region_name=AWS_REGION_US_EAST_1)
instance = ec2.create_instances(
ImageId=EXAMPLE_AMI_ID,
MinCount=1,
MaxCount=1,
UserData="This is some user_data",
)[0]
from prowler.providers.aws.services.ec2.ec2_service import EC2
aws_provider = set_mocked_aws_provider(
[AWS_REGION_EU_WEST_1, AWS_REGION_US_EAST_1]
)
aws_provider._audit_config = {"max_ec2_instance_stopped_days": 30}
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
),
mock.patch(
"prowler.providers.aws.services.ec2.ec2_instance_stopped_older_than_specific_days.ec2_instance_stopped_older_than_specific_days.ec2_client",
new=EC2(aws_provider),
) as service_client,
):
from prowler.providers.aws.services.ec2.ec2_instance_stopped_older_than_specific_days.ec2_instance_stopped_older_than_specific_days import (
ec2_instance_stopped_older_than_specific_days,
)
service_client.instances[0].state = "stopped"
service_client.instances[0].state_transition_reason = (
"User initiated (2021-11-01 17:18:00 GMT)"
)
check = ec2_instance_stopped_older_than_specific_days()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert result[0].region == AWS_REGION_US_EAST_1
assert search(
f"EC2 Instance {instance.id} has been stopped longer than",
result[0].status_extended,
)
assert result[0].resource_id == instance.id
assert (
result[0].resource_arn
== f"arn:{aws_provider.identity.partition}:ec2:{AWS_REGION_US_EAST_1}:{aws_provider.identity.account}:instance/{instance.id}"
)
@mock_aws
def test_stopped_ec2_unknown_stop_time(self):
ec2 = resource("ec2", region_name=AWS_REGION_US_EAST_1)
instance = ec2.create_instances(
ImageId=EXAMPLE_AMI_ID,
MinCount=1,
MaxCount=1,
UserData="This is some user_data",
)[0]
from prowler.providers.aws.services.ec2.ec2_service import EC2
aws_provider = set_mocked_aws_provider(
[AWS_REGION_EU_WEST_1, AWS_REGION_US_EAST_1]
)
aws_provider._audit_config = {"max_ec2_instance_stopped_days": 30}
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
),
mock.patch(
"prowler.providers.aws.services.ec2.ec2_instance_stopped_older_than_specific_days.ec2_instance_stopped_older_than_specific_days.ec2_client",
new=EC2(aws_provider),
) as service_client,
):
from prowler.providers.aws.services.ec2.ec2_instance_stopped_older_than_specific_days.ec2_instance_stopped_older_than_specific_days import (
ec2_instance_stopped_older_than_specific_days,
)
service_client.instances[0].state = "stopped"
service_client.instances[0].state_transition_reason = ""
check = ec2_instance_stopped_older_than_specific_days()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert result[0].region == AWS_REGION_US_EAST_1
assert search(
f"EC2 Instance {instance.id} is stopped but stop time could not be determined",
result[0].status_extended,
)
assert result[0].resource_id == instance.id
@@ -26,6 +26,7 @@ from tests.providers.aws.utils import (
EXAMPLE_AMI_ID = "ami-12c6146b"
MOCK_DATETIME = datetime(2023, 1, 4, 7, 27, 30, tzinfo=tzutc())
MOCK_STATE_TRANSITION_REASON = "User initiated (2021-11-01 17:18:00 GMT)"
make_api_call = botocore.client.BaseClient._make_api_call
@@ -66,6 +67,15 @@ def mock_make_api_call(self, operation_name, kwarg):
return make_api_call(self, operation_name, kwarg)
def mock_make_api_call_with_state_transition_reason(self, operation_name, kwarg):
response = make_api_call(self, operation_name, kwarg)
if operation_name == "DescribeInstances":
for reservation in response.get("Reservations", []):
for instance in reservation.get("Instances", []):
instance["StateTransitionReason"] = MOCK_STATE_TRANSITION_REASON
return response
class Test_EC2_Service:
# Test EC2 Service
@mock_aws
@@ -319,6 +329,10 @@ class Test_EC2_Service:
assert ec2.images_by_id == {}
# Test EC2 Describe Instances
@mock.patch(
"botocore.client.BaseClient._make_api_call",
new=mock_make_api_call_with_state_transition_reason,
)
@mock_aws
@freeze_time(MOCK_DATETIME)
def test_describe_instances(self):
@@ -349,6 +363,7 @@ class Test_EC2_Service:
assert ec2.instances[0].state == "running"
assert re.match(r"ami-[0-9a-z]{8}", ec2.instances[0].image_id)
assert ec2.instances[0].launch_time == MOCK_DATETIME
assert ec2.instances[0].state_transition_reason == MOCK_STATE_TRANSITION_REASON
assert not ec2.instances[0].user_data
assert ec2.instances[0].http_tokens == "optional"
assert ec2.instances[0].http_endpoint == "enabled"