fix(securityhub): preserve timestampt of existing findings

This commit is contained in:
MrCloudSec committed 2025-08-27 10:58:53 +02:00
1 parent 65e7e89d61
commit 4df5b878ca
5 files changed
+1126 -31

No files matched your search

+694
View File
@@ -584,3 +584,697 @@ class TestASFF:
assert ASFF.generate_status("FAIL") == "FAILED"
assert ASFF.generate_status("FAIL", True) == "WARNING"
assert ASFF.generate_status("SOMETHING ELSE") == "NOT_AVAILABLE"
def test_asff_preserves_existing_timestamps(self):
"""Test that ASFF preserves existing timestamps for findings that already exist in Security Hub."""
status = "PASS"
finding = generate_finding_output(
status=status,
status_extended="This is a test",
region=AWS_REGION_EU_WEST_1,
resource_details="Test resource details",
resource_name="test-resource",
resource_uid="test-arn",
resource_tags={"key1": "value1"},
)
# Mock existing finding timestamps from Security Hub
finding_id = f"prowler-{finding.metadata.CheckID}-{AWS_ACCOUNT_NUMBER}-{AWS_REGION_EU_WEST_1}-{hash_sha512(finding.resource_uid)}"
existing_timestamps = {
finding_id: {
"FirstObservedAt": "2023-01-01T00:00:00Z",
"CreatedAt": "2023-01-01T00:00:00Z",
"UpdatedAt": "2023-01-01T00:00:00Z",
}
}
current_timestamp = timestamp_utc.strftime("%Y-%m-%dT%H:%M:%SZ")
associated_standards, compliance_summary = ASFF.format_compliance(
finding.compliance
)
expected = AWSSecurityFindingFormat(
Id=finding_id,
ProductArn=f"arn:{AWS_COMMERCIAL_PARTITION}:securityhub:{AWS_REGION_EU_WEST_1}::product/prowler/prowler",
ProductFields=ProductFields(
ProviderVersion=prowler_version,
ProwlerResourceName=finding.resource_uid,
),
GeneratorId="prowler-" + finding.metadata.CheckID,
AwsAccountId=AWS_ACCOUNT_NUMBER,
Types=finding.metadata.CheckType,
FirstObservedAt="2023-01-01T00:00:00Z", # Should preserve existing timestamp
UpdatedAt=current_timestamp, # Should update with current timestamp
CreatedAt="2023-01-01T00:00:00Z", # Should preserve existing timestamp
Severity=Severity(Label=finding.metadata.Severity.value),
Title=finding.metadata.CheckTitle,
Resources=[
Resource(
Id=finding.resource_uid,
Type=finding.metadata.ResourceType,
Partition=AWS_COMMERCIAL_PARTITION,
Region=AWS_REGION_EU_WEST_1,
Tags={"key1": "value1"},
)
],
Compliance=Compliance(
Status=ASFF.generate_status(status),
RelatedRequirements=compliance_summary,
AssociatedStandards=associated_standards,
),
Remediation=Remediation(
Recommendation=Recommendation(
Text=finding.metadata.Remediation.Recommendation.Text,
Url=finding.metadata.Remediation.Recommendation.Url,
)
),
Description=finding.status_extended,
)
asff = ASFF(
findings=[finding], existing_findings_timestamps=existing_timestamps
)
assert len(asff.data) == 1
asff_finding = asff.data[0]
assert asff_finding == expected
# Verify that FirstObservedAt and CreatedAt are preserved
assert asff_finding.FirstObservedAt == "2023-01-01T00:00:00Z"
assert asff_finding.CreatedAt == "2023-01-01T00:00:00Z"
# Verify that UpdatedAt uses current timestamp
assert asff_finding.UpdatedAt == current_timestamp
def test_asff_uses_current_timestamps_for_new_findings(self):
"""Test that ASFF uses current timestamps for new findings when no existing timestamps are provided."""
status = "PASS"
finding = generate_finding_output(
status=status,
status_extended="This is a test",
region=AWS_REGION_EU_WEST_1,
resource_details="Test resource details",
resource_name="test-resource",
resource_uid="test-arn",
resource_tags={"key1": "value1"},
)
current_timestamp = timestamp_utc.strftime("%Y-%m-%dT%H:%M:%SZ")
associated_standards, compliance_summary = ASFF.format_compliance(
finding.compliance
)
expected = AWSSecurityFindingFormat(
Id=f"prowler-{finding.metadata.CheckID}-{AWS_ACCOUNT_NUMBER}-{AWS_REGION_EU_WEST_1}-{hash_sha512(finding.resource_uid)}",
ProductArn=f"arn:{AWS_COMMERCIAL_PARTITION}:securityhub:{AWS_REGION_EU_WEST_1}::product/prowler/prowler",
ProductFields=ProductFields(
ProviderVersion=prowler_version,
ProwlerResourceName=finding.resource_uid,
),
GeneratorId="prowler-" + finding.metadata.CheckID,
AwsAccountId=AWS_ACCOUNT_NUMBER,
Types=finding.metadata.CheckType,
FirstObservedAt=current_timestamp, # Should use current timestamp for new findings
UpdatedAt=current_timestamp,
CreatedAt=current_timestamp,
Severity=Severity(Label=finding.metadata.Severity.value),
Title=finding.metadata.CheckTitle,
Resources=[
Resource(
Id=finding.resource_uid,
Type=finding.metadata.ResourceType,
Partition=AWS_COMMERCIAL_PARTITION,
Region=AWS_REGION_EU_WEST_1,
Tags={"key1": "value1"},
)
],
Compliance=Compliance(
Status=ASFF.generate_status(status),
RelatedRequirements=compliance_summary,
AssociatedStandards=associated_standards,
),
Remediation=Remediation(
Recommendation=Recommendation(
Text=finding.metadata.Remediation.Recommendation.Text,
Url=finding.metadata.Remediation.Recommendation.Url,
)
),
Description=finding.status_extended,
)
asff = ASFF(findings=[finding])
assert len(asff.data) == 1
asff_finding = asff.data[0]
assert asff_finding == expected
# Verify that all timestamps use current timestamp for new findings
assert asff_finding.FirstObservedAt == current_timestamp
assert asff_finding.UpdatedAt == current_timestamp
assert asff_finding.CreatedAt == current_timestamp
def test_asff_constructor_with_existing_timestamps(self):
"""Test that ASFF constructor properly stores existing timestamps."""
status = "PASS"
finding = generate_finding_output(
status=status,
status_extended="This is a test",
region=AWS_REGION_EU_WEST_1,
resource_details="Test resource details",
resource_name="test-resource",
resource_uid="test-arn",
resource_tags={"key1": "value1"},
)
# Generate the actual finding ID that will be used
actual_finding_id = f"prowler-{finding.metadata.CheckID}-{finding.account_uid}-{finding.region}-{hash_sha512(finding.resource_uid)}"
# Existing timestamps with the correct ID
existing_timestamps = {
actual_finding_id: {
"FirstObservedAt": "2023-01-01T00:00:00Z",
"CreatedAt": "2023-01-01T00:00:00Z",
"UpdatedAt": "2023-01-01T00:00:00Z",
}
}
# Create ASFF output with timestamps in constructor
asff = ASFF(
findings=[finding], existing_findings_timestamps=existing_timestamps
)
# Verify that timestamps are preserved
assert len(asff.data) == 1
finding_asff = asff.data[0]
assert finding_asff.FirstObservedAt == "2023-01-01T00:00:00Z"
assert finding_asff.CreatedAt == "2023-01-01T00:00:00Z"
def test_asff_constructor_without_existing_timestamps(self):
"""Test that ASFF constructor works without existing timestamps."""
status = "PASS"
finding = generate_finding_output(
status=status,
status_extended="This is a test",
region=AWS_REGION_EU_WEST_1,
resource_details="Test resource details",
resource_name="test-resource",
resource_uid="test-arn",
resource_tags={"key1": "value1"},
)
# Create ASFF output without timestamps parameter
asff = ASFF(findings=[finding])
# Verify that current timestamps are used
assert len(asff.data) == 1
finding_asff = asff.data[0]
current_timestamp = timestamp_utc.strftime("%Y-%m-%dT%H:%M:%SZ")
assert finding_asff.FirstObservedAt == current_timestamp
assert finding_asff.CreatedAt == current_timestamp
assert finding_asff.UpdatedAt == current_timestamp
def test_asff_transform_method_parameter_override(self):
"""Test that transform method parameter overrides constructor parameter."""
status = "PASS"
finding = generate_finding_output(
status=status,
status_extended="This is a test",
region=AWS_REGION_EU_WEST_1,
resource_details="Test resource details",
resource_name="test-resource",
resource_uid="test-arn",
resource_tags={"key1": "value1"},
)
# Generate the actual finding ID
actual_finding_id = f"prowler-{finding.metadata.CheckID}-{finding.account_uid}-{finding.region}-{hash_sha512(finding.resource_uid)}"
# Constructor timestamps
constructor_timestamps = {
actual_finding_id: {
"FirstObservedAt": "2023-01-01T00:00:00Z",
"CreatedAt": "2023-01-01T00:00:00Z",
"UpdatedAt": "2023-01-01T00:00:00Z",
}
}
# Transform method timestamps (different)
transform_timestamps = {
actual_finding_id: {
"FirstObservedAt": "2023-02-01T00:00:00Z",
"CreatedAt": "2023-02-01T00:00:00Z",
"UpdatedAt": "2023-02-01T00:00:00Z",
}
}
# Create ASFF output with constructor timestamps
asff = ASFF(
findings=[finding], existing_findings_timestamps=constructor_timestamps
)
# Clear existing data and transform with different timestamps
asff._data = []
asff.transform([finding], transform_timestamps)
# Verify that transform method timestamps are used
assert len(asff.data) == 1
finding_asff = asff.data[0]
assert finding_asff.FirstObservedAt == "2023-02-01T00:00:00Z"
assert finding_asff.CreatedAt == "2023-02-01T00:00:00Z"
def test_asff_transform_method_without_parameter(self):
"""Test that transform method uses constructor timestamps when no parameter is provided."""
status = "PASS"
finding = generate_finding_output(
status=status,
status_extended="This is a test",
region=AWS_REGION_EU_WEST_1,
resource_details="Test resource details",
resource_name="test-resource",
resource_uid="test-arn",
resource_tags={"key1": "value1"},
)
# Generate the actual finding ID
actual_finding_id = f"prowler-{finding.metadata.CheckID}-{finding.account_uid}-{finding.region}-{hash_sha512(finding.resource_uid)}"
# Constructor timestamps
constructor_timestamps = {
actual_finding_id: {
"FirstObservedAt": "2023-01-01T00:00:00Z",
"CreatedAt": "2023-01-01T00:00:00Z",
"UpdatedAt": "2023-01-01T00:00:00Z",
}
}
# Create ASFF output with timestamps in constructor
asff = ASFF(
findings=[finding], existing_findings_timestamps=constructor_timestamps
)
# Clear existing data and transform without parameter (should use constructor timestamps)
asff._data = []
asff.transform([finding])
# Verify that constructor timestamps are used
assert len(asff.data) == 1
finding_asff = asff.data[0]
assert finding_asff.FirstObservedAt == "2023-01-01T00:00:00Z"
assert finding_asff.CreatedAt == "2023-01-01T00:00:00Z"
def test_asff_handles_partial_existing_timestamps(self):
"""Test that ASFF handles cases where only some timestamp fields exist in the existing data."""
status = "PASS"
finding = generate_finding_output(
status=status,
status_extended="This is a test",
region=AWS_REGION_EU_WEST_1,
resource_details="Test resource details",
resource_name="test-resource",
resource_uid="test-arn",
resource_tags={"key1": "value1"},
)
# Generate the actual finding ID
actual_finding_id = f"prowler-{finding.metadata.CheckID}-{finding.account_uid}-{finding.region}-{hash_sha512(finding.resource_uid)}"
# Mock existing finding with only FirstObservedAt timestamp
existing_timestamps = {
actual_finding_id: {
"FirstObservedAt": "2023-01-01T00:00:00Z",
# Missing CreatedAt and UpdatedAt
}
}
asff = ASFF(
findings=[finding], existing_findings_timestamps=existing_timestamps
)
# Should preserve FirstObservedAt, use current timestamp for missing fields
assert len(asff.data) == 1
finding_asff = asff.data[0]
assert finding_asff.FirstObservedAt == "2023-01-01T00:00:00Z"
current_timestamp = timestamp_utc.strftime("%Y-%m-%dT%H:%M:%SZ")
assert finding_asff.CreatedAt == current_timestamp
assert finding_asff.UpdatedAt == current_timestamp
def test_asff_handles_missing_timestamp_fields(self):
"""Test that ASFF handles cases where timestamp fields are missing from the dictionary."""
status = "PASS"
finding = generate_finding_output(
status=status,
status_extended="This is a test",
region=AWS_REGION_EU_WEST_1,
resource_details="Test resource details",
resource_name="test-resource",
resource_uid="test-arn",
resource_tags={"key1": "value1"},
)
# Generate the actual finding ID
actual_finding_id = f"prowler-{finding.metadata.CheckID}-{finding.account_uid}-{finding.region}-{hash_sha512(finding.resource_uid)}"
# Mock existing finding with missing keys (not None values)
existing_timestamps = {
actual_finding_id: {
# Missing FirstObservedAt, CreatedAt, and UpdatedAt keys entirely
}
}
asff = ASFF(
findings=[finding], existing_findings_timestamps=existing_timestamps
)
# Should use current timestamp for all fields when keys are missing
assert len(asff.data) == 1
finding_asff = asff.data[0]
current_timestamp = timestamp_utc.strftime("%Y-%m-%dT%H:%M:%SZ")
assert finding_asff.FirstObservedAt == current_timestamp
assert finding_asff.CreatedAt == current_timestamp
assert finding_asff.UpdatedAt == current_timestamp
def test_asff_handles_empty_existing_timestamps_dict(self):
"""Test that ASFF handles empty existing timestamps dictionary."""
status = "PASS"
finding = generate_finding_output(
status=status,
status_extended="This is a test",
region=AWS_REGION_EU_WEST_1,
resource_details="Test resource details",
resource_name="test-resource",
resource_uid="test-arn",
resource_tags={"key1": "value1"},
)
# Empty existing timestamps
existing_timestamps = {}
asff = ASFF(
findings=[finding], existing_findings_timestamps=existing_timestamps
)
# Should use current timestamp for all fields
assert len(asff.data) == 1
finding_asff = asff.data[0]
current_timestamp = timestamp_utc.strftime("%Y-%m-%dT%H:%M:%SZ")
assert finding_asff.FirstObservedAt == current_timestamp
assert finding_asff.CreatedAt == current_timestamp
assert finding_asff.UpdatedAt == current_timestamp
def test_asff_handles_none_existing_timestamps(self):
"""Test that ASFF handles None existing timestamps parameter."""
status = "PASS"
finding = generate_finding_output(
status=status,
status_extended="This is a test",
region=AWS_REGION_EU_WEST_1,
resource_details="Test resource details",
resource_name="test-resource",
resource_uid="test-arn",
resource_tags={"key1": "value1"},
)
# None existing timestamps
asff = ASFF(findings=[finding], existing_findings_timestamps=None)
# Should use current timestamp for all fields
assert len(asff.data) == 1
finding_asff = asff.data[0]
current_timestamp = timestamp_utc.strftime("%Y-%m-%dT%H:%M:%SZ")
assert finding_asff.FirstObservedAt == current_timestamp
assert finding_asff.CreatedAt == current_timestamp
assert finding_asff.UpdatedAt == current_timestamp
def test_asff_finding_id_generation_consistency(self):
"""Test that finding ID generation is consistent between timestamp lookup and ASFF creation."""
status = "PASS"
finding = generate_finding_output(
status=status,
status_extended="This is a test",
region=AWS_REGION_EU_WEST_1,
resource_details="Test resource details",
resource_name="test-resource",
resource_uid="test-arn",
resource_tags={"key1": "value1"},
)
# Generate the expected finding ID
expected_finding_id = f"prowler-{finding.metadata.CheckID}-{finding.account_uid}-{finding.region}-{hash_sha512(finding.resource_uid)}"
# Create existing timestamps with the expected ID
existing_timestamps = {
expected_finding_id: {
"FirstObservedAt": "2023-01-01T00:00:00Z",
"CreatedAt": "2023-01-01T00:00:00Z",
"UpdatedAt": "2023-01-01T00:00:00Z",
}
}
asff = ASFF(
findings=[finding], existing_findings_timestamps=existing_timestamps
)
# Verify that the finding ID matches and timestamps are preserved
assert len(asff.data) == 1
finding_asff = asff.data[0]
assert finding_asff.Id == expected_finding_id
assert finding_asff.FirstObservedAt == "2023-01-01T00:00:00Z"
assert finding_asff.CreatedAt == "2023-01-01T00:00:00Z"
def test_asff_multiple_findings_mixed_timestamps(self):
"""Test that ASFF handles multiple findings with mixed existing and new timestamps."""
# Create multiple findings
finding1 = generate_finding_output(
status="PASS",
status_extended="First finding",
region=AWS_REGION_EU_WEST_1,
resource_details="Test resource details",
resource_name="test-resource-1",
resource_uid="test-arn-1",
resource_tags={"key1": "value1"},
)
finding2 = generate_finding_output(
status="FAIL",
status_extended="Second finding",
region=AWS_REGION_EU_WEST_1,
resource_details="Test resource details",
resource_name="test-resource-2",
resource_uid="test-arn-2",
resource_tags={"key2": "value2"},
)
# Only first finding has existing timestamps
finding1_id = f"prowler-{finding1.metadata.CheckID}-{finding1.account_uid}-{finding1.region}-{hash_sha512(finding1.resource_uid)}"
existing_timestamps = {
finding1_id: {
"FirstObservedAt": "2023-01-01T00:00:00Z",
"CreatedAt": "2023-01-01T00:00:00Z",
"UpdatedAt": "2023-01-01T00:00:00Z",
}
}
asff = ASFF(
findings=[finding1, finding2],
existing_findings_timestamps=existing_timestamps,
)
# Verify that first finding preserves timestamps
assert len(asff.data) == 2
finding1_asff = asff.data[0]
assert finding1_asff.FirstObservedAt == "2023-01-01T00:00:00Z"
assert finding1_asff.CreatedAt == "2023-01-01T00:00:00Z"
# Verify that second finding uses current timestamps
finding2_asff = asff.data[1]
current_timestamp = timestamp_utc.strftime("%Y-%m-%dT%H:%M:%SZ")
assert finding2_asff.FirstObservedAt == current_timestamp
assert finding2_asff.CreatedAt == current_timestamp
assert finding2_asff.UpdatedAt == current_timestamp
def test_asff_updated_at_always_current(self):
"""Test that UpdatedAt is always set to current timestamp regardless of existing data."""
status = "PASS"
finding = generate_finding_output(
status=status,
status_extended="This is a test",
region=AWS_REGION_EU_WEST_1,
resource_details="Test resource details",
resource_name="test-resource",
resource_uid="test-arn",
resource_tags={"key1": "value1"},
)
# Generate the actual finding ID
actual_finding_id = f"prowler-{finding.metadata.CheckID}-{finding.account_uid}-{finding.region}-{hash_sha512(finding.resource_uid)}"
# Existing timestamps with old UpdatedAt
existing_timestamps = {
actual_finding_id: {
"FirstObservedAt": "2023-01-01T00:00:00Z",
"CreatedAt": "2023-01-01T00:00:00Z",
"UpdatedAt": "2023-01-01T00:00:00Z", # Old timestamp
}
}
asff = ASFF(
findings=[finding], existing_findings_timestamps=existing_timestamps
)
# Verify that UpdatedAt is current time, not the old timestamp
assert len(asff.data) == 1
finding_asff = asff.data[0]
assert finding_asff.FirstObservedAt == "2023-01-01T00:00:00Z"
assert finding_asff.CreatedAt == "2023-01-01T00:00:00Z"
current_timestamp = timestamp_utc.strftime("%Y-%m-%dT%H:%M:%SZ")
assert finding_asff.UpdatedAt == current_timestamp
assert finding_asff.UpdatedAt != "2023-01-01T00:00:00Z"
def test_asff_handles_duplicate_finding_ids(self):
"""Test that ASFF handles cases where multiple findings might have the same ID."""
# Create findings with same check ID but different resources
finding1 = generate_finding_output(
status="PASS",
status_extended="First finding",
region=AWS_REGION_EU_WEST_1,
resource_details="Test resource details",
resource_name="test-resource-1",
resource_uid="test-arn-1",
resource_tags={"key1": "value1"},
)
finding2 = generate_finding_output(
status="FAIL",
status_extended="Second finding",
region=AWS_REGION_EU_WEST_1,
resource_details="Test resource details",
resource_name="test-resource-2",
resource_uid="test-arn-2",
resource_tags={"key2": "value2"},
)
# Create existing timestamps for one finding
finding1_id = f"prowler-{finding1.metadata.CheckID}-{finding1.account_uid}-{finding1.region}-{hash_sha512(finding1.resource_uid)}"
existing_timestamps = {
finding1_id: {
"FirstObservedAt": "2023-01-01T00:00:00Z",
"CreatedAt": "2023-01-01T00:00:00Z",
"UpdatedAt": "2023-01-01T00:00:00Z",
}
}
asff = ASFF(
findings=[finding1, finding2],
existing_findings_timestamps=existing_timestamps,
)
# Verify that findings are processed correctly
assert len(asff.data) == 2
# First finding should preserve timestamps
finding1_asff = asff.data[0]
assert finding1_asff.FirstObservedAt == "2023-01-01T00:00:00Z"
assert finding1_asff.CreatedAt == "2023-01-01T00:00:00Z"
# Second finding should use current timestamps
finding2_asff = asff.data[1]
current_timestamp = timestamp_utc.strftime("%Y-%m-%dT%H:%M:%SZ")
assert finding2_asff.FirstObservedAt == current_timestamp
assert finding2_asff.CreatedAt == current_timestamp
def test_asff_handles_manual_status_findings(self):
"""Test that ASFF correctly skips MANUAL status findings even with existing timestamps."""
finding = generate_finding_output(
status="MANUAL",
status_extended="This is a manual finding",
region=AWS_REGION_EU_WEST_1,
resource_details="Test resource details",
resource_name="test-resource",
resource_uid="test-arn",
resource_tags={"key1": "value1"},
)
# Generate the actual finding ID
actual_finding_id = f"prowler-{finding.metadata.CheckID}-{finding.account_uid}-{finding.region}-{hash_sha512(finding.resource_uid)}"
# Existing timestamps
existing_timestamps = {
actual_finding_id: {
"FirstObservedAt": "2023-01-01T00:00:00Z",
"CreatedAt": "2023-01-01T00:00:00Z",
"UpdatedAt": "2023-01-01T00:00:00Z",
}
}
asff = ASFF(
findings=[finding], existing_findings_timestamps=existing_timestamps
)
# MANUAL findings should be skipped
assert len(asff.data) == 0
def test_asff_timestamp_format_consistency(self):
"""Test that all timestamps use consistent ISO 8601 format."""
finding = generate_finding_output(
status="PASS",
status_extended="This is a test",
region=AWS_REGION_EU_WEST_1,
resource_details="Test resource details",
resource_name="test-resource",
resource_uid="test-arn",
resource_tags={"key1": "value1"},
)
# Generate the actual finding ID
actual_finding_id = f"prowler-{finding.metadata.CheckID}-{finding.account_uid}-{finding.region}-{hash_sha512(finding.resource_uid)}"
# Existing timestamps in different formats
existing_timestamps = {
actual_finding_id: {
"FirstObservedAt": "2023-01-01T00:00:00Z",
"CreatedAt": "2023-01-01T00:00:00Z",
"UpdatedAt": "2023-01-01T00:00:00Z",
}
}
asff = ASFF(
findings=[finding], existing_findings_timestamps=existing_timestamps
)
# Verify that all timestamps use consistent format
assert len(asff.data) == 1
finding_asff = asff.data[0]
# Check format: YYYY-MM-DDTHH:MM:SSZ
import re
timestamp_pattern = r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z$"
assert re.match(timestamp_pattern, finding_asff.FirstObservedAt)
assert re.match(timestamp_pattern, finding_asff.CreatedAt)
assert re.match(timestamp_pattern, finding_asff.UpdatedAt)
def test_asff_backward_compatibility(self):
"""Test that ASFF maintains backward compatibility when no existing timestamps are provided."""
finding = generate_finding_output(
status="PASS",
status_extended="This is a test",
region=AWS_REGION_EU_WEST_1,
resource_details="Test resource details",
resource_name="test-resource",
resource_uid="test-arn",
resource_tags={"key1": "value1"},
)
# Test with no existing timestamps (backward compatibility)
asff = ASFF(findings=[finding])
# Verify that current timestamps are used
assert len(asff.data) == 1
finding_asff = asff.data[0]
current_timestamp = timestamp_utc.strftime("%Y-%m-%dT%H:%M:%SZ")
assert finding_asff.FirstObservedAt == current_timestamp
assert finding_asff.UpdatedAt == current_timestamp
assert finding_asff.CreatedAt == current_timestamp
@@ -5,7 +5,7 @@ import botocore
import pytest
from boto3 import session
from botocore.client import ClientError
from mock import patch
from mock import MagicMock, patch
from prowler.lib.outputs.asff.asff import ASFF
from prowler.providers.aws.lib.security_hub.exceptions.exceptions import (
@@ -1283,3 +1283,241 @@ class TestSecurityHub:
assert connection.error is None
assert len(connection.enabled_regions) == 1
assert len(connection.disabled_regions) == 1
@patch("prowler.providers.aws.lib.security_hub.security_hub.AwsSetUpSession")
def test_get_existing_findings_timestamps(self, mock_aws_setup):
"""Test that get_existing_findings_timestamps correctly retrieves existing findings timestamps."""
# Mock findings per region
mock_findings = [
MagicMock(
Id="prowler-test-check-123456789012-us-east-1-hash123",
Region="us-east-1",
Compliance=MagicMock(Status="FAILED"),
),
MagicMock(
Id="prowler-test-check-123456789012-us-west-2-hash456",
Region="us-west-2",
Compliance=MagicMock(Status="FAILED"),
),
]
# Mock enabled regions
mock_enabled_regions = {
"us-east-1": MagicMock(),
"us-west-2": MagicMock(),
}
# Mock paginator responses
mock_page1 = {
"Findings": [
{
"Id": "prowler-test-check-123456789012-us-east-1-hash123",
"FirstObservedAt": "2023-01-01T00:00:00Z",
"CreatedAt": "2023-01-01T00:00:00Z",
"UpdatedAt": "2023-01-01T00:00:00Z",
}
]
}
mock_page2 = {
"Findings": [
{
"Id": "prowler-test-check-123456789012-us-west-2-hash456",
"FirstObservedAt": "2023-01-15T00:00:00Z",
"CreatedAt": "2023-01-15T00:00:00Z",
"UpdatedAt": "2023-01-15T00:00:00Z",
}
]
}
# Mock paginator
mock_paginator = MagicMock()
mock_paginator.paginate.return_value = [mock_page1, mock_page2]
# Mock Security Hub client
mock_client = MagicMock()
mock_client.get_paginator.return_value = mock_paginator
# Create SecurityHub instance with mocked session
mock_session = MagicMock()
mock_aws_setup.return_value._session.current_session = mock_session
security_hub = SecurityHub(
aws_account_id="123456789012",
aws_partition="aws",
findings=mock_findings,
aws_security_hub_available_regions=["us-east-1", "us-west-2"],
)
# Mock the enabled regions
security_hub._enabled_regions = mock_enabled_regions
security_hub._enabled_regions["us-east-1"] = mock_client
security_hub._enabled_regions["us-west-2"] = mock_client
# Mock findings per region
security_hub._findings_per_region = {
"us-east-1": [mock_findings[0]],
"us-west-2": [mock_findings[1]],
}
# Call the method
result = security_hub.get_existing_findings_timestamps()
# Verify the result
expected_result = {
"prowler-test-check-123456789012-us-east-1-hash123": {
"FirstObservedAt": "2023-01-01T00:00:00Z",
"CreatedAt": "2023-01-01T00:00:00Z",
"UpdatedAt": "2023-01-01T00:00:00Z",
},
"prowler-test-check-123456789012-us-west-2-hash456": {
"FirstObservedAt": "2023-01-15T00:00:00Z",
"CreatedAt": "2023-01-15T00:00:00Z",
"UpdatedAt": "2023-01-15T00:00:00Z",
},
}
assert result == expected_result
# Verify that the paginator was called correctly
mock_client.get_paginator.assert_called_with("get_findings")
assert mock_paginator.paginate.call_count == 2
@patch("prowler.providers.aws.lib.security_hub.security_hub.AwsSetUpSession")
def test_get_existing_findings_timestamps_empty_regions(self, mock_aws_setup):
"""Test that get_existing_findings_timestamps handles empty regions correctly."""
# Mock session
mock_session = MagicMock()
mock_aws_setup.return_value._session.current_session = mock_session
security_hub = SecurityHub(
aws_account_id="123456789012",
aws_partition="aws",
findings=[],
aws_security_hub_available_regions=[],
)
# Mock empty findings per region
security_hub._findings_per_region = {}
result = security_hub.get_existing_findings_timestamps()
assert result == {}
@patch("prowler.providers.aws.lib.security_hub.security_hub.AwsSetUpSession")
def test_get_existing_findings_timestamps_with_error(self, mock_aws_setup):
"""Test that get_existing_findings_timestamps handles errors gracefully."""
# Mock findings per region
mock_findings = [
MagicMock(
Id="prowler-test-check-123456789012-us-east-1-hash123",
Region="us-east-1",
Compliance=MagicMock(Status="FAILED"),
),
]
# Mock enabled regions
mock_enabled_regions = {
"us-east-1": MagicMock(),
}
# Mock client that raises an exception
mock_client = MagicMock()
mock_client.get_paginator.side_effect = Exception("Test error")
# Mock session
mock_session = MagicMock()
mock_aws_setup.return_value._session.current_session = mock_session
# Create SecurityHub instance
security_hub = SecurityHub(
aws_account_id="123456789012",
aws_partition="aws",
findings=mock_findings,
aws_security_hub_available_regions=["us-east-1"],
)
# Mock the enabled regions
security_hub._enabled_regions = mock_enabled_regions
security_hub._enabled_regions["us-east-1"] = mock_client
# Mock findings per region
security_hub._findings_per_region = {
"us-east-1": [mock_findings[0]],
}
# Call the method - should not raise exception
result = security_hub.get_existing_findings_timestamps()
# Should return empty dict due to error
assert result == {}
@patch("prowler.providers.aws.lib.security_hub.security_hub.AwsSetUpSession")
def test_get_existing_findings_timestamps_partial_data(self, mock_aws_setup):
"""Test that get_existing_findings_timestamps handles partial timestamp data correctly."""
# Mock findings per region
mock_findings = [
MagicMock(
Id="prowler-test-check-123456789012-us-east-1-hash123",
Region="us-east-1",
Compliance=MagicMock(Status="FAILED"),
),
]
# Mock enabled regions
mock_enabled_regions = {
"us-east-1": MagicMock(),
}
# Mock paginator responses with partial data
mock_page = {
"Findings": [
{
"Id": "prowler-test-check-123456789012-us-east-1-hash123",
"FirstObservedAt": "2023-01-01T00:00:00Z",
# Missing CreatedAt and UpdatedAt
}
]
}
# Mock paginator
mock_paginator = MagicMock()
mock_paginator.paginate.return_value = [mock_page]
# Mock Security Hub client
mock_client = MagicMock()
mock_client.get_paginator.return_value = mock_paginator
# Mock session
mock_session = MagicMock()
mock_aws_setup.return_value._session.current_session = mock_session
# Create SecurityHub instance
security_hub = SecurityHub(
aws_account_id="123456789012",
aws_partition="aws",
findings=mock_findings,
aws_security_hub_available_regions=["us-east-1"],
)
# Mock the enabled regions
security_hub._enabled_regions = mock_enabled_regions
security_hub._enabled_regions["us-east-1"] = mock_client
# Mock findings per region
security_hub._findings_per_region = {
"us-east-1": [mock_findings[0]],
}
# Call the method
result = security_hub.get_existing_findings_timestamps()
# Verify the result handles missing fields gracefully
expected_result = {
"prowler-test-check-123456789012-us-east-1-hash123": {
"FirstObservedAt": "2023-01-01T00:00:00Z",
"CreatedAt": None,
"UpdatedAt": None,
},
}
assert result == expected_result