cleveragents/cleveragents-core
4
0
Fork 3
Code Issues 6.0k Pull Requests 115 Actions Packages Projects Releases Wiki Activity

feat(resource): implement AWS SDK integration for CloudResourceHandler #1280

Merged
HAL9000 merged 6 commits from feature/aws-cloud-handler-sdk into master 2026-06-02 11:52:42 +00:00
Conversation 112 Commits 6 Files Changed 17
+2057 -310
17 changed files with 2057 additions and 310 deletions
Download Patch File Download Diff File Expand all files Collapse all files
features/cloud_aws_sdk.feature
+240
View File
@@ -0,0 +1,240 @@
Feature: AWS SDK integration for CloudResourceHandler
As a CleverAgents user
I want the CloudResourceHandler to use real AWS SDK operations
So that I can resolve and discover AWS cloud resources
# ── boto3 availability ────────────────────────────────────────────────────
Scenario: boto3 availability flag is exposed
Given awssdk the cloud handler module is imported
Then awssdk the boto3 availability flag should be a boolean
# ── _build_aws_session ────────────────────────────────────────────────────
Scenario: _build_aws_session raises ImportError when boto3 is unavailable
Given awssdk boto3 is not available
When awssdk I try to build an AWS session with valid credentials
Then awssdk an ImportError should be raised mentioning "cleveragents[aws]"
Scenario: _build_aws_session builds a session from explicit credentials
Given awssdk boto3 is available via mock
When awssdk I build an AWS session with explicit credentials
Then awssdk the session should be created successfully
Scenario: _build_aws_session builds a session from profile name
Given awssdk boto3 is available via mock
When awssdk I build an AWS session with a profile name "test-profile"
Then awssdk the session should be created with profile "test-profile"
# ── CloudResourceHandler.resolve for AWS ─────────────────────────────────
Scenario: resolve returns BoundResource for AWS account type with mocked STS
Given awssdk boto3 is available via mock
And awssdk AWS STS returns account id "123456789012"
And awssdk a valid AWS cloud resource of type "aws"
When awssdk I call resolve on the cloud handler with mocked boto3
Then awssdk a BoundResource should be returned
And awssdk the BoundResource slot_name should be "cloud-slot"
And awssdk the BoundResource resource_type should be "aws"
And awssdk the BoundResource sandbox_path should contain "123456789012"
Scenario: resolve returns BoundResource for aws-vpc type (no STS check)
Given awssdk boto3 is available via mock
And awssdk a valid AWS cloud resource of type "aws-vpc"
When awssdk I call resolve on the cloud handler with mocked boto3
Then awssdk a BoundResource should be returned
And awssdk the BoundResource resource_type should be "aws-vpc"
Scenario: resolve raises ValueError when STS call fails
Given awssdk boto3 is available via mock
And awssdk AWS STS raises a ClientError
And awssdk a valid AWS cloud resource of type "aws"
When awssdk I call resolve on the cloud handler with mocked boto3
Then awssdk a ValueError should be raised mentioning "credential verification failed"
Scenario: resolve raises ImportError for AWS when boto3 is not installed
Given awssdk boto3 is not available
And awssdk a valid AWS cloud resource of type "aws-vpc"
When awssdk I call resolve on the cloud handler without boto3
Then awssdk an ImportError should be raised mentioning "cleveragents[aws]"
Scenario: resolve raises NotImplementedError for GCP provider
Given awssdk boto3 is available via mock
And awssdk a valid GCP cloud resource with credentials
When awssdk I call resolve on the cloud handler with mocked boto3
Then awssdk a NotImplementedError should be raised mentioning "gcp"
Scenario: resolve raises NotImplementedError for Azure provider
Given awssdk boto3 is available via mock
And awssdk a valid Azure cloud resource with credentials
When awssdk I call resolve on the cloud handler with mocked boto3
Then awssdk a NotImplementedError should be raised mentioning "azure"
# ── discover_aws_resources ────────────────────────────────────────────────
Scenario: discover_aws_resources returns empty list for unmapped type
Given awssdk boto3 is available via mock
And awssdk a mock boto3 session
And awssdk a cloud resource of type "aws-unknown-type"
When awssdk I call discover_aws_resources
Then awssdk the discovery result should be an empty list
Scenario: discover_aws_resources discovers VPCs from EC2
Given awssdk boto3 is available via mock
And awssdk EC2 describe_vpcs returns 2 VPCs
And awssdk a cloud resource of type "aws-vpc"
When awssdk I call discover_aws_resources
Then awssdk the discovery result should have 2 items
And awssdk each discovery item should have an "id" key
And awssdk each discovery item should have an "arn" key
Scenario: discover_aws_resources discovers EC2 instances via describe_instances
Given awssdk boto3 is available via mock
And awssdk EC2 describe_instances returns 3 instances
And awssdk a cloud resource of type "aws-instance"
When awssdk I call discover_aws_resources
Then awssdk the discovery result should have 3 items
And awssdk each discovery item should have an "id" key
And awssdk each discovery item "id" should be non-empty
Scenario: discover_aws_resources discovers S3 buckets
Given awssdk boto3 is available via mock
And awssdk S3 list_buckets returns 3 buckets
And awssdk a cloud resource of type "aws-s3-bucket"
When awssdk I call discover_aws_resources
Then awssdk the discovery result should have 3 items
And awssdk each discovery item arn should start with "arn:aws:s3:::"
Scenario: discover_aws_resources discovers ECS clusters
Given awssdk boto3 is available via mock
And awssdk ECS list_clusters returns 2 cluster ARNs
And awssdk a cloud resource of type "aws-ecs-cluster"
When awssdk I call discover_aws_resources
Then awssdk the discovery result should have 2 items
Scenario: discover_aws_resources handles API errors gracefully
Given awssdk boto3 is available via mock
And awssdk a mock boto3 session that raises an exception
And awssdk a cloud resource of type "aws-vpc"
When awssdk I call discover_aws_resources
Then awssdk the discovery result should be an empty list
# ── CloudResourceHandler.discover_children ────────────────────────────────
Scenario: discover_children raises NotImplementedError for non-AWS provider
Given awssdk a CloudResourceHandler instance
And awssdk a cloud resource of type "gcp-project"
When awssdk I call discover_children on the cloud handler
Then awssdk a NotImplementedError should be raised mentioning "aws"
Scenario: discover_children raises ImportError when boto3 is not installed
Given awssdk boto3 is not available
And awssdk a CloudResourceHandler instance
And awssdk a valid AWS cloud resource of type "aws-vpc"
When awssdk I call discover_children on the cloud handler without boto3
Then awssdk an ImportError should be raised mentioning "cleveragents[aws]"
Scenario: discover_children returns Resource objects for discovered VPCs
Given awssdk boto3 is available via mock
And awssdk a CloudResourceHandler instance
And awssdk EC2 describe_vpcs returns 2 VPCs
And awssdk a valid AWS cloud resource of type "aws-vpc"
When awssdk I call discover_children on the cloud handler with mocked boto3
Then awssdk the aws discovery result should be a list of Resource objects
And awssdk the aws discovery result should have 2 items
# ── CloudSandboxStrategy AWS implementation ───────────────────────────────
Scenario: CloudSandboxStrategy.create succeeds for AWS with boto3 available
Given awssdk boto3 is available via mock
And awssdk a cloud sandbox strategy for "aws"
When awssdk I call create on the AWS sandbox strategy with plan "PLAN-001"
Then awssdk no exception should be raised
Scenario: CloudSandboxStrategy.create raises ImportError without boto3
Given awssdk boto3 is not available
And awssdk a cloud sandbox strategy for "aws"
When awssdk I call create on the AWS sandbox strategy with plan "PLAN-001"
Then awssdk an ImportError should be raised mentioning "cleveragents[aws]"
Scenario: CloudSandboxStrategy.create raises NotImplementedError for GCP
Given awssdk boto3 is available via mock
And awssdk a cloud sandbox strategy for "gcp"
When awssdk I call create on the AWS sandbox strategy with plan "PLAN-001"
Then awssdk a NotImplementedError should be raised
Scenario: CloudSandboxStrategy.commit succeeds for AWS with boto3 available
Given awssdk boto3 is available via mock
And awssdk a cloud sandbox strategy for "aws"
When awssdk I call commit on the AWS sandbox strategy with plan "PLAN-001"
Then awssdk no exception should be raised
Scenario: CloudSandboxStrategy.commit raises ImportError without boto3
Given awssdk boto3 is not available
And awssdk a cloud sandbox strategy for "aws"
When awssdk I call commit on the AWS sandbox strategy with plan "PLAN-001"
Then awssdk an ImportError should be raised mentioning "cleveragents[aws]"
Scenario: CloudSandboxStrategy.commit raises NotImplementedError for Azure
Given awssdk boto3 is available via mock
And awssdk a cloud sandbox strategy for "azure"
When awssdk I call commit on the AWS sandbox strategy with plan "PLAN-001"
Then awssdk a NotImplementedError should be raised
Scenario: CloudSandboxStrategy.rollback succeeds for AWS with boto3 available
Given awssdk boto3 is available via mock
And awssdk a cloud sandbox strategy for "aws"
When awssdk I call rollback on the AWS sandbox strategy with plan "PLAN-001"
Then awssdk no exception should be raised
Scenario: CloudSandboxStrategy.rollback raises ImportError without boto3
Given awssdk boto3 is not available
And awssdk a cloud sandbox strategy for "aws"
When awssdk I call rollback on the AWS sandbox strategy with plan "PLAN-001"
Then awssdk an ImportError should be raised mentioning "cleveragents[aws]"
Scenario: CloudSandboxStrategy.rollback raises NotImplementedError for GCP
Given awssdk boto3 is available via mock
And awssdk a cloud sandbox strategy for "gcp"
When awssdk I call rollback on the AWS sandbox strategy with plan "PLAN-001"
Then awssdk a NotImplementedError should be raised
Scenario: CloudSandboxStrategy.create raises ValueError for empty plan_id
Given awssdk boto3 is available via mock
And awssdk a cloud sandbox strategy for "aws"
When awssdk I call create on the AWS sandbox strategy with plan ""
Then awssdk a ValueError should be raised mentioning "plan_id"
Scenario: CloudSandboxStrategy.commit raises ValueError for empty plan_id
Given awssdk boto3 is available via mock
And awssdk a cloud sandbox strategy for "aws"
When awssdk I call commit on the AWS sandbox strategy with plan ""
Then awssdk a ValueError should be raised mentioning "plan_id"
Scenario: CloudSandboxStrategy.rollback raises ValueError for empty plan_id
Given awssdk boto3 is available via mock
And awssdk a cloud sandbox strategy for "aws"
When awssdk I call rollback on the AWS sandbox strategy with plan ""
Then awssdk a ValueError should be raised mentioning "plan_id"
# ── Credential masking (regression) ──────────────────────────────────────
Scenario: AWS credentials are never logged in plain text
Given awssdk boto3 is available via mock
And awssdk a valid AWS cloud resource of type "aws"
When awssdk I call resolve on the cloud handler with mocked boto3
Then awssdk no raw credential values should appear in log output
# ── _AWS_RESOURCE_MAP coverage ────────────────────────────────────────────
Scenario: _AWS_RESOURCE_MAP contains expected AWS resource types
Given awssdk the cloud handler module is imported
Then awssdk the AWS resource map should contain "aws-vpc"
And awssdk the AWS resource map should contain "aws-subnet"
And awssdk the AWS resource map should contain "aws-instance"
And awssdk the AWS resource map should contain "aws-s3-bucket"
And awssdk the AWS resource map should contain "aws-iam-role"
And awssdk the AWS resource map should contain "aws-rds-instance"
And awssdk the AWS resource map should contain "aws-ecs-cluster"
And awssdk the AWS resource map should contain "aws-lambda-function"
And awssdk the AWS resource map should contain "aws-eks-cluster"
features/cloud_resources.feature
+4 -5
View File
@@ -168,13 +168,12 @@ Feature: Cloud Infrastructure Resources
And I validate credentials for provider "aws"
Then the cloud validation errors should not contain secrets
# ── Stub execution ────────────────────────────────────────
# ── SDK execution ────────────────────────────────────────
Scenario: Cloud handler resolve raises NotImplementedError for AWS
Scenario: Cloud handler resolve raises ImportError for AWS when boto3 not installed
Given a valid AWS cloud resource
When I call resolve on the cloud handler
Then a cloud NotImplementedError should be raised
And the cloud error message should mention "aws"
Then a cloud ImportError or NotImplementedError should be raised
Scenario: Cloud handler resolve raises NotImplementedError for GCP
Given a valid GCP cloud resource
@@ -211,7 +210,7 @@ Feature: Cloud Infrastructure Resources
# ── Cloud sandbox strategy stubs ──────────────────────────
Scenario: Cloud sandbox create raises NotImplementedError
Given a cloud sandbox strategy for "aws"
Given a cloud sandbox strategy for "gcp"
When I call create on the sandbox strategy
Then a cloud NotImplementedError should be raised
features/steps/cloud_aws_discover_steps.py
+244
View File
@@ -0,0 +1,244 @@
"""Step definitions for AWS resource discovery scenarios.
Tests discover_aws_resources() and CloudResourceHandler.discover_children()
with mocked boto3.
"""
from __future__ import annotations
from typing import Any
from unittest.mock import MagicMock, patch
from behave import given, then, when # type: ignore[attr-defined]
from cleveragents.domain.models.core.resource import Resource
from cleveragents.resource.handlers.cloud import CloudResourceHandler
from cleveragents.resource.handlers.cloud_aws import (
_AWS_RESOURCE_MAP,
discover_aws_resources,
)
from features.steps.cloud_aws_helpers import (
_make_mock_session,
_make_resource,
_restore_cloud_env,
)
@given('awssdk a cloud resource of type "{type_name}"')
def step_cloud_resource_type(context: Any, type_name: str) -> None:
"""Create a cloud resource of the given type."""
context.cloud_resource = _make_resource(type_name) # type: ignore[attr-defined]
@given("awssdk EC2 describe_instances returns {count:d} instances")
def step_ec2_instances(context: Any, count: int) -> None:
"""Configure mock EC2 to return N instances via describe_instances."""
instances = [
{
"InstanceId": f"i-{i:017x}",
"InstanceType": "t3.micro",
"State": {"Name": "running"},
}
for i in range(count)
]
context.mock_session = _make_mock_session(ec2_instances=instances) # type: ignore[attr-defined]
@given("awssdk EC2 describe_vpcs returns {count:d} VPCs")
def step_ec2_vpcs(context: Any, count: int) -> None:
"""Configure mock EC2 to return N VPCs."""
vpcs = [
{
"VpcId": f"vpc-{i:08x}",
"State": "available",
"CidrBlock": f"10.{i}.0.0/16",
}
for i in range(count)
]
context.mock_session = _make_mock_session(ec2_vpcs=vpcs) # type: ignore[attr-defined]
@given("awssdk S3 list_buckets returns {count:d} buckets")
def step_s3_buckets(context: Any, count: int) -> None:
"""Configure mock S3 to return N buckets."""
buckets = [
{"Name": f"my-bucket-{i}", "CreationDate": "2024-01-01"} for i in range(count)
]
context.mock_session = _make_mock_session(s3_buckets=buckets) # type: ignore[attr-defined]
@given("awssdk ECS list_clusters returns {count:d} cluster ARNs")
def step_ecs_clusters(context: Any, count: int) -> None:
"""Configure mock ECS to return N cluster ARNs."""
arns = [
f"arn:aws:ecs:us-east-1:123456789012:cluster/cluster-{i}" for i in range(count)
]
context.mock_session = _make_mock_session(ecs_clusters=arns) # type: ignore[attr-defined]
@given("awssdk a CloudResourceHandler instance")
def step_handler_instance(context: Any) -> None:
"""Create a CloudResourceHandler instance."""
context.cloud_handler = CloudResourceHandler() # type: ignore[attr-defined]
@given("awssdk a mock boto3 session")
def step_mock_session(context: Any) -> None:
"""Create a generic mock boto3 session."""
context.mock_session = _make_mock_session() # type: ignore[attr-defined]
@given("awssdk a mock boto3 session that raises an exception")
def step_mock_session_error(context: Any) -> None:
"""Create a mock boto3 session that raises on all client calls."""
context.mock_session = _make_mock_session( # type: ignore[attr-defined]
client_error=RuntimeError("Simulated AWS API error")
)
@when("awssdk I call discover_aws_resources")
def step_call_discover_aws(context: Any) -> None:
"""Call discover_aws_resources with the mock session."""
resource: Resource = context.cloud_resource # type: ignore[attr-defined]
session = getattr(context, "mock_session", _make_mock_session())
context.discovery_result = discover_aws_resources(resource, session) # type: ignore[attr-defined]
@when("awssdk I call discover_children on the cloud handler")
def step_call_discover_children_no_boto3(context: Any) -> None:
"""Call discover_children on the handler (non-AWS provider)."""
handler: CloudResourceHandler = context.cloud_handler # type: ignore[attr-defined]
resource: Resource = context.cloud_resource # type: ignore[attr-defined]
context.raised_error = None # type: ignore[attr-defined]
context.raised_error_type = None # type: ignore[attr-defined]
context.discovery_result = None # type: ignore[attr-defined]
try:
result = handler.discover_children(resource=resource)
context.discovery_result = result # type: ignore[attr-defined]
except (NotImplementedError, ImportError) as exc:
context.raised_error = exc # type: ignore[attr-defined]
context.raised_error_type = type(exc).__name__ # type: ignore[attr-defined]
@when("awssdk I call discover_children on the cloud handler without boto3")
def step_call_discover_children_no_boto3_explicit(context: Any) -> None:
"""Call discover_children without boto3 available."""
handler: CloudResourceHandler = context.cloud_handler # type: ignore[attr-defined]
resource: Resource = context.cloud_resource # type: ignore[attr-defined]
context.raised_error = None # type: ignore[attr-defined]
context.raised_error_type = None # type: ignore[attr-defined]
context.discovery_result = None # type: ignore[attr-defined]
with (
patch("cleveragents.resource.handlers.cloud_aws._BOTO3_AVAILABLE", False),
patch("cleveragents.resource.handlers.cloud_aws.boto3", None),
patch("cleveragents.resource.handlers.cloud._BOTO3_AVAILABLE", False),
):
try:
result = handler.discover_children(resource=resource)
context.discovery_result = result # type: ignore[attr-defined]
except (NotImplementedError, ImportError) as exc:
context.raised_error = exc # type: ignore[attr-defined]
context.raised_error_type = type(exc).__name__ # type: ignore[attr-defined]
@when("awssdk I call discover_children on the cloud handler with mocked boto3")
def step_call_discover_children_mocked(context: Any) -> None:
"""Call discover_children with mocked boto3."""
handler: CloudResourceHandler = context.cloud_handler # type: ignore[attr-defined]
resource: Resource = context.cloud_resource # type: ignore[attr-defined]
mock_session = getattr(context, "mock_session", _make_mock_session())
context.raised_error = None # type: ignore[attr-defined]
context.raised_error_type = None # type: ignore[attr-defined]
context.discovery_result = None # type: ignore[attr-defined]
mock_boto3 = MagicMock()
mock_boto3.Session.return_value = mock_session
with (
patch("cleveragents.resource.handlers.cloud_aws.boto3", mock_boto3),
patch("cleveragents.resource.handlers.cloud_aws._BOTO3_AVAILABLE", True),
patch("cleveragents.resource.handlers.cloud._BOTO3_AVAILABLE", True),
):
try:
result = handler.discover_children(resource=resource)
context.discovery_result = result # type: ignore[attr-defined]
except (NotImplementedError, ImportError) as exc:
context.raised_error = exc # type: ignore[attr-defined]
context.raised_error_type = type(exc).__name__ # type: ignore[attr-defined]
saved = getattr(context, "saved_env", {})
_restore_cloud_env(saved)
@then("awssdk the discovery result should be an empty list")
def step_discovery_empty(context: Any) -> None:
"""Check that the discovery result is empty."""
result = context.discovery_result # type: ignore[attr-defined]
assert result == [], f"Expected empty list, got {result}"
@then("awssdk the discovery result should have {count:d} items")
def step_discovery_count(context: Any, count: int) -> None:
"""Check the discovery result count."""
result = context.discovery_result # type: ignore[attr-defined]
assert len(result) == count, f"Expected {count} items, got {len(result)}: {result}"
@then('awssdk each discovery item should have an "{key}" key')
def step_discovery_item_key(context: Any, key: str) -> None:
"""Check that each discovery item has the expected key."""
result = context.discovery_result # type: ignore[attr-defined]
for item in result:
assert key in item, f"Key '{key}' not found in item: {item}"
@then('awssdk each discovery item arn should start with "{prefix}"')
def step_discovery_item_arn_prefix(context: Any, prefix: str) -> None:
"""Check that each discovery item ARN starts with the expected prefix."""
result = context.discovery_result # type: ignore[attr-defined]
for item in result:
arn = item.get("arn", "")
assert arn.startswith(prefix), (
f"Expected ARN to start with '{prefix}', got '{arn}'"
)
@then("awssdk the aws discovery result should be a list of Resource objects")
def step_result_resource_list(context: Any) -> None:
"""Check that the result is a list of Resource objects."""
result = context.discovery_result # type: ignore[attr-defined]
assert isinstance(result, list), f"Expected list, got {type(result)}"
for item in result:
assert isinstance(item, Resource), (
f"Expected Resource, got {type(item)}: {item}"
)
@then("awssdk the aws discovery result should have {count:d} items")
def step_result_count(context: Any, count: int) -> None:
"""Check the result count."""
result = context.discovery_result # type: ignore[attr-defined]
assert len(result) == count, f"Expected {count} items, got {len(result)}"
@then('awssdk the AWS resource map should contain "{type_name}"')
def step_aws_resource_map_contains(context: Any, type_name: str) -> None:
"""Check that the AWS resource map contains the expected type."""
assert type_name in _AWS_RESOURCE_MAP, (
f"'{type_name}' not found in _AWS_RESOURCE_MAP. "
f"Available: {sorted(_AWS_RESOURCE_MAP.keys())}"
)
@then('awssdk each discovery item "{key}" should be non-empty')
def step_discovery_item_value_nonempty(context: Any, key: str) -> None:
"""Check that each discovery item has a non-empty value for the given key."""
result = context.discovery_result # type: ignore[attr-defined]
for item in result:
val = item.get(key, "")
assert val, f"Expected non-empty '{key}' in item: {item}"
features/steps/cloud_aws_helpers.py
+146
View File
@@ -0,0 +1,146 @@
"""Shared helpers for AWS SDK step definitions.
Provides environment management, resource factories, and mock session
builders used across the cloud_aws_* step definition files.
"""
from __future__ import annotations
import contextlib
import os
from typing import Any
from unittest.mock import MagicMock
from cleveragents.domain.models.core.resource import (
PhysVirt,
Resource,
ResourceCapabilities,
)
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
_CLOUD_ENV_VARS = [
"AWS_ACCESS_KEY_ID",
"AWS_SECRET_ACCESS_KEY",
"AWS_SESSION_TOKEN",
"AWS_REGION",
"AWS_PROFILE",
"GOOGLE_APPLICATION_CREDENTIALS",
"GCLOUD_PROJECT",
"GCP_REGION",
"AZURE_SUBSCRIPTION_ID",
"AZURE_TENANT_ID",
"AZURE_CLIENT_ID",
"AZURE_CLIENT_SECRET",
"AZURE_REGION",
]
def _clear_cloud_env() -> dict[str, str]:
saved: dict[str, str] = {}
for var in _CLOUD_ENV_VARS:
val = os.environ.pop(var, None)
if val is not None:
saved[var] = val
return saved
def _restore_cloud_env(saved: dict[str, str]) -> None:
for var in _CLOUD_ENV_VARS:
if var in saved:
os.environ[var] = saved[var]
else:
os.environ.pop(var, None)
def _make_resource(
type_name: str,
properties: dict[str, Any] | None = None,
location: str | None = None,
) -> Resource:
return Resource(
resource_id="01KJ5C5TPMP8GGX3QC83E2MAQS",
resource_type_name=type_name,
classification=PhysVirt.PHYSICAL,
location=location,
properties=properties or {},
capabilities=ResourceCapabilities(
readable=True,
writable=False,
sandboxable=False,
checkpointable=False,
),
)
def _make_mock_session(
account_id: str = "123456789012",
sts_error: Exception | None = None,
ec2_vpcs: list[dict[str, Any]] | None = None,
ec2_instances: list[dict[str, Any]] | None = None,
s3_buckets: list[dict[str, Any]] | None = None,
ecs_clusters: list[str] | None = None,
client_error: Exception | None = None,
) -> MagicMock:
"""Build a mock boto3 Session with configurable responses."""
session = MagicMock()
def _make_client(service: str, **_kwargs: Any) -> MagicMock:
client = MagicMock()
if service == "sts":
if sts_error is not None:
client.get_caller_identity.side_effect = sts_error
else:
client.get_caller_identity.return_value = {
"Account": account_id,
"UserId": "AIDAIOSFODNN7EXAMPLE",
"Arn": f"arn:aws:iam::{account_id}:user/test",
}
elif service == "ec2":
if client_error is not None:
client.describe_vpcs.side_effect = client_error
client.describe_subnets.side_effect = client_error
client.describe_instances.side_effect = client_error
client.describe_security_groups.side_effect = client_error
else:
vpcs = ec2_vpcs or []
client.describe_vpcs.return_value = {"Vpcs": vpcs}
client.describe_subnets.return_value = {"Subnets": []}
instances = ec2_instances or []
reservations = (
[{"ReservationId": "r-00000001", "Instances": instances}]
if instances
else []
)
client.describe_instances.return_value = {"Reservations": reservations}
client.describe_security_groups.return_value = {"SecurityGroups": []}
elif service == "s3":
if client_error is not None:
client.list_buckets.side_effect = client_error
else:
buckets = s3_buckets or []
client.list_buckets.return_value = {"Buckets": buckets}
elif service == "ecs":
if client_error is not None:
client.list_clusters.side_effect = client_error
else:
arns = ecs_clusters or []
client.list_clusters.return_value = {"clusterArns": arns}
else:
if client_error is not None:
for attr in dir(client):
if not attr.startswith("_"):
with contextlib.suppress(AttributeError):
getattr(client, attr).side_effect = client_error
return client
session.client.side_effect = _make_client
return session
features/steps/cloud_aws_sandbox_steps.py
+95
View File
@@ -0,0 +1,95 @@
"""Step definitions for AWS sandbox lifecycle scenarios.
Tests CloudSandboxStrategy.create/commit/rollback with mocked boto3.
"""
from __future__ import annotations
from typing import Any
from unittest.mock import patch
from behave import given, when # type: ignore[attr-defined]
from cleveragents.resource.handlers.cloud_aws import CloudSandboxStrategy
@given('awssdk a cloud sandbox strategy for "{provider}"')
def step_sandbox_strategy(context: Any, provider: str) -> None:
"""Create a CloudSandboxStrategy for the given provider."""
context.cloud_sandbox = CloudSandboxStrategy(provider) # type: ignore[attr-defined]
@when('awssdk I call create on the AWS sandbox strategy with plan "{plan_id}"')
def step_sandbox_create_aws(context: Any, plan_id: str) -> None:
"""Call create on the CloudSandboxStrategy."""
strategy: CloudSandboxStrategy = context.cloud_sandbox # type: ignore[attr-defined]
context.raised_error = None # type: ignore[attr-defined]
context.raised_error_type = None # type: ignore[attr-defined]
boto3_avail = getattr(context, "boto3_available", True)
with patch(
"cleveragents.resource.handlers.cloud_aws._BOTO3_AVAILABLE", boto3_avail
):
try:
strategy.create("res-test-001", plan_id)
except (NotImplementedError, ImportError, ValueError) as exc:
context.raised_error = exc # type: ignore[attr-defined]
context.raised_error_type = type(exc).__name__ # type: ignore[attr-defined]
@when('awssdk I call commit on the AWS sandbox strategy with plan "{plan_id}"')
def step_sandbox_commit_aws(context: Any, plan_id: str) -> None:
"""Call commit on the CloudSandboxStrategy."""
strategy: CloudSandboxStrategy = context.cloud_sandbox # type: ignore[attr-defined]
context.raised_error = None # type: ignore[attr-defined]
context.raised_error_type = None # type: ignore[attr-defined]
boto3_avail = getattr(context, "boto3_available", True)
with patch(
"cleveragents.resource.handlers.cloud_aws._BOTO3_AVAILABLE", boto3_avail
):
try:
strategy.commit("res-test-001", plan_id)
except (NotImplementedError, ImportError, ValueError) as exc:
context.raised_error = exc # type: ignore[attr-defined]
context.raised_error_type = type(exc).__name__ # type: ignore[attr-defined]
@when('awssdk I call rollback on the AWS sandbox strategy with plan "{plan_id}"')
def step_sandbox_rollback_aws(context: Any, plan_id: str) -> None:
"""Call rollback on the CloudSandboxStrategy."""
strategy: CloudSandboxStrategy = context.cloud_sandbox # type: ignore[attr-defined]
context.raised_error = None # type: ignore[attr-defined]
context.raised_error_type = None # type: ignore[attr-defined]
boto3_avail = getattr(context, "boto3_available", True)
with patch(
"cleveragents.resource.handlers.cloud_aws._BOTO3_AVAILABLE", boto3_avail
):
try:
strategy.rollback("res-test-001", plan_id)
except (NotImplementedError, ImportError, ValueError) as exc:
context.raised_error = exc # type: ignore[attr-defined]
context.raised_error_type = type(exc).__name__ # type: ignore[attr-defined]
# behave's parse format {plan_id} uses .+? and won't match an empty string,
# so we need explicit overloads for the empty plan_id scenarios.
@when('awssdk I call create on the AWS sandbox strategy with plan ""')
def step_sandbox_create_aws_empty(context: Any) -> None:
"""Call create on the CloudSandboxStrategy with empty plan_id."""
step_sandbox_create_aws(context, "")
@when('awssdk I call commit on the AWS sandbox strategy with plan ""')
def step_sandbox_commit_aws_empty(context: Any) -> None:
"""Call commit on the CloudSandboxStrategy with empty plan_id."""
step_sandbox_commit_aws(context, "")
@when('awssdk I call rollback on the AWS sandbox strategy with plan ""')
def step_sandbox_rollback_aws_empty(context: Any) -> None:
"""Call rollback on the CloudSandboxStrategy with empty plan_id."""
step_sandbox_rollback_aws(context, "")
features/steps/cloud_aws_sdk_steps.py
+14
View File
@@ -0,0 +1,14 @@
"""Step definitions for cloud_aws_sdk.feature.
Tests AWS SDK integration for CloudResourceHandler using mocked boto3.
All steps use the "awssdk" prefix to avoid conflicts with existing
step definitions in other feature files.
Step definitions are split across focused modules for maintainability:
- cloud_aws_helpers.py: shared helpers (_make_resource, _make_mock_session)
- cloud_aws_session_steps.py: session building and resolve() steps
- cloud_aws_discover_steps.py: resource discovery steps
- cloud_aws_sandbox_steps.py: sandbox lifecycle steps
Behave auto-discovers all step files in this directory.
"""
features/steps/cloud_aws_session_steps.py
+371
View File
@@ -0,0 +1,371 @@
"""Step definitions for AWS session building scenarios.
Tests _build_aws_session and CloudResourceHandler.resolve() with mocked boto3.
"""
from __future__ import annotations
import os
from typing import Any
from unittest.mock import MagicMock, patch
from behave import given, then, when # type: ignore[attr-defined]
from cleveragents.resource.handlers.cloud import CloudResourceHandler
from cleveragents.resource.handlers.cloud_aws import (
_BOTO3_AVAILABLE,
_build_aws_session,
)
from features.steps.cloud_aws_helpers import (
_clear_cloud_env,
_make_mock_session,
_make_resource,
_restore_cloud_env,
)
@given("awssdk the cloud handler module is imported")
def step_module_imported(context: Any) -> None:
"""Ensure the cloud handler module is importable."""
import cleveragents.resource.handlers.cloud as _mod
context.cloud_module = _mod # type: ignore[attr-defined]
@given("awssdk boto3 is not available")
def step_boto3_not_available(context: Any) -> None:
"""Simulate boto3 not being installed."""
context.boto3_available = False # type: ignore[attr-defined]
context.saved_env = _clear_cloud_env() # type: ignore[attr-defined]
@given("awssdk boto3 is available via mock")
def step_boto3_available_mock(context: Any) -> None:
"""Mark boto3 as available (mocked)."""
context.boto3_available = True # type: ignore[attr-defined]
context.saved_env = _clear_cloud_env() # type: ignore[attr-defined]
os.environ["AWS_ACCESS_KEY_ID"] = "AKIAIOSFODNN7EXAMPLE"
os.environ["AWS_SECRET_ACCESS_KEY"] = "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"
os.environ["AWS_REGION"] = "us-east-1"
@given('awssdk AWS STS returns account id "{account_id}"')
def step_sts_account_id(context: Any, account_id: str) -> None:
"""Configure mock STS to return a specific account ID."""
context.mock_account_id = account_id # type: ignore[attr-defined]
@given("awssdk AWS STS raises a ClientError")
def step_sts_client_error(context: Any) -> None:
"""Configure mock STS to raise a ClientError."""
context.sts_error = RuntimeError("AWS ClientError: InvalidClientTokenId") # type: ignore[attr-defined]
@given('awssdk a valid AWS cloud resource of type "{type_name}"')
def step_aws_resource_type(context: Any, type_name: str) -> None:
"""Create an AWS resource of the given type."""
context.cloud_resource = _make_resource( # type: ignore[attr-defined]
type_name,
properties={
"access-key-id": "AKIAIOSFODNN7EXAMPLE",
"secret-access-key": "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY",
"region": "us-east-1",
},
)
@given("awssdk a valid GCP cloud resource with credentials")
def step_gcp_resource_with_creds(context: Any) -> None:
"""Create a GCP resource with credentials."""
context.cloud_resource = _make_resource( # type: ignore[attr-defined]
"gcp",
properties={
"service-account-json-path": "/path/to/sa.json",
"project-id": "my-project",
},
)
@given("awssdk a valid Azure cloud resource with credentials")
def step_azure_resource_with_creds(context: Any) -> None:
"""Create an Azure resource with credentials."""
context.cloud_resource = _make_resource( # type: ignore[attr-defined]
"azure",
properties={
"subscription-id": "sub-123",
"tenant-id": "tenant-456",
"client-id": "client-789",
"client-secret": "secret-abc",
},
)
@when("awssdk I try to build an AWS session with valid credentials")
def step_try_build_session_no_boto3(context: Any) -> None:
"""Try to build an AWS session when boto3 is unavailable."""
context.raised_error = None # type: ignore[attr-defined]
context.raised_error_type = None # type: ignore[attr-defined]
with (
patch("cleveragents.resource.handlers.cloud_aws._BOTO3_AVAILABLE", False),
patch("cleveragents.resource.handlers.cloud_aws.boto3", None),
):
try:
_build_aws_session(
{
"access-key-id": "AKIAIOSFODNN7EXAMPLE",
"secret-access-key": "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY",
}
)
except ImportError as exc:
context.raised_error = exc # type: ignore[attr-defined]
context.raised_error_type = "ImportError" # type: ignore[attr-defined]
@when("awssdk I build an AWS session with explicit credentials")
def step_build_session_explicit(context: Any) -> None:
"""Build an AWS session with explicit credentials using mocked boto3."""
context.raised_error = None # type: ignore[attr-defined]
mock_boto3 = MagicMock()
mock_session = MagicMock()
mock_boto3.Session.return_value = mock_session
with (
patch("cleveragents.resource.handlers.cloud_aws.boto3", mock_boto3),
patch("cleveragents.resource.handlers.cloud_aws._BOTO3_AVAILABLE", True),
):
result = _build_aws_session(
{
"access-key-id": "AKIAIOSFODNN7EXAMPLE",
"secret-access-key": "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY",
"region": "us-east-1",
}
)
context.session_result = result # type: ignore[attr-defined]
context.mock_boto3 = mock_boto3 # type: ignore[attr-defined]
@when('awssdk I build an AWS session with a profile name "{profile}"')
def step_build_session_profile(context: Any, profile: str) -> None:
"""Build an AWS session with a profile name."""
context.raised_error = None # type: ignore[attr-defined]
mock_boto3 = MagicMock()
mock_session = MagicMock()
mock_boto3.Session.return_value = mock_session
with (
patch("cleveragents.resource.handlers.cloud_aws.boto3", mock_boto3),
patch("cleveragents.resource.handlers.cloud_aws._BOTO3_AVAILABLE", True),
):
result = _build_aws_session({"profile": profile})
context.session_result = result # type: ignore[attr-defined]
context.mock_boto3 = mock_boto3 # type: ignore[attr-defined]
context.expected_profile = profile # type: ignore[attr-defined]
@when("awssdk I call resolve on the cloud handler with mocked boto3")
def step_resolve_with_mock(context: Any) -> None:
"""Call resolve on CloudResourceHandler with mocked boto3."""
from cleveragents.domain.models.core.resource import Resource
handler = CloudResourceHandler()
resource: Resource = context.cloud_resource # type: ignore[attr-defined]
mock_manager = MagicMock()
account_id = getattr(context, "mock_account_id", "123456789012")
sts_error = getattr(context, "sts_error", None)
mock_session = _make_mock_session(account_id=account_id, sts_error=sts_error)
context.raised_error = None # type: ignore[attr-defined]
context.raised_error_type = None # type: ignore[attr-defined]
context.bound_resource = None # type: ignore[attr-defined]
mock_boto3 = MagicMock()
mock_boto3.Session.return_value = mock_session
with (
patch("cleveragents.resource.handlers.cloud_aws.boto3", mock_boto3),
patch("cleveragents.resource.handlers.cloud_aws._BOTO3_AVAILABLE", True),
patch("cleveragents.resource.handlers.cloud._BOTO3_AVAILABLE", True),
):
try:
result = handler.resolve(
resource=resource,
plan_id="PLAN-TEST-001",
slot_name="cloud-slot",
sandbox_manager=mock_manager,
)
context.bound_resource = result # type: ignore[attr-defined]
except (ValueError, NotImplementedError, ImportError) as exc:
context.raised_error = exc # type: ignore[attr-defined]
context.raised_error_type = type(exc).__name__ # type: ignore[attr-defined]
saved = getattr(context, "saved_env", {})
_restore_cloud_env(saved)
@when("awssdk I call resolve on the cloud handler without boto3")
def step_resolve_without_boto3(context: Any) -> None:
"""Call resolve on CloudResourceHandler without boto3."""
from cleveragents.domain.models.core.resource import Resource
handler = CloudResourceHandler()
resource: Resource = context.cloud_resource # type: ignore[attr-defined]
mock_manager = MagicMock()
context.raised_error = None # type: ignore[attr-defined]
context.raised_error_type = None # type: ignore[attr-defined]
context.bound_resource = None # type: ignore[attr-defined]
with (
patch("cleveragents.resource.handlers.cloud_aws._BOTO3_AVAILABLE", False),
patch("cleveragents.resource.handlers.cloud_aws.boto3", None),
patch("cleveragents.resource.handlers.cloud._BOTO3_AVAILABLE", False),
):
try:
result = handler.resolve(
resource=resource,
plan_id="PLAN-TEST-001",
slot_name="cloud-slot",
sandbox_manager=mock_manager,
)
context.bound_resource = result # type: ignore[attr-defined]
except (ValueError, NotImplementedError, ImportError) as exc:
context.raised_error = exc # type: ignore[attr-defined]
context.raised_error_type = type(exc).__name__ # type: ignore[attr-defined]
saved = getattr(context, "saved_env", {})
_restore_cloud_env(saved)
@then("awssdk the boto3 availability flag should be a boolean")
def step_boto3_flag_bool(context: Any) -> None:
"""Check that _BOTO3_AVAILABLE is a boolean."""
assert isinstance(_BOTO3_AVAILABLE, bool), (
f"Expected bool, got {type(_BOTO3_AVAILABLE)}"
)
@then('awssdk an ImportError should be raised mentioning "{text}"')
def step_import_error_raised(context: Any, text: str) -> None:
"""Check that an ImportError was raised with the expected text."""
assert context.raised_error_type == "ImportError", ( # type: ignore[attr-defined]
f"Expected ImportError, got {context.raised_error_type}: " # type: ignore[attr-defined]
f"{context.raised_error}" # type: ignore[attr-defined]
)
msg = str(context.raised_error) # type: ignore[attr-defined]
assert text in msg, f"'{text}' not found in ImportError: {msg}"
@then("awssdk the session should be created successfully")
def step_session_created(context: Any) -> None:
"""Check that a session was created."""
assert context.session_result is not None # type: ignore[attr-defined]
mock_boto3: MagicMock = context.mock_boto3 # type: ignore[attr-defined]
assert mock_boto3.Session.called, "boto3.Session was not called"
@then('awssdk the session should be created with profile "{profile}"')
def step_session_created_with_profile(context: Any, profile: str) -> None:
"""Check that the session was created with the expected profile."""
assert context.session_result is not None # type: ignore[attr-defined]
mock_boto3: MagicMock = context.mock_boto3 # type: ignore[attr-defined]
call_kwargs = mock_boto3.Session.call_args[1]
assert call_kwargs.get("profile_name") == profile, (
f"Expected profile_name='{profile}', got {call_kwargs}"
)
@then("awssdk a BoundResource should be returned")
def step_bound_resource_returned(context: Any) -> None:
"""Check that a BoundResource was returned."""
from cleveragents.tool.context import BoundResource
assert context.raised_error is None, ( # type: ignore[attr-defined]
f"Expected BoundResource but got error: "
f"{context.raised_error_type}: {context.raised_error}" # type: ignore[attr-defined]
)
assert isinstance(context.bound_resource, BoundResource), ( # type: ignore[attr-defined]
f"Expected BoundResource, got {type(context.bound_resource)}" # type: ignore[attr-defined]
)
@then('awssdk the BoundResource slot_name should be "{slot_name}"')
def step_bound_resource_slot(context: Any, slot_name: str) -> None:
"""Check BoundResource slot_name."""
from cleveragents.tool.context import BoundResource
br: BoundResource = context.bound_resource # type: ignore[attr-defined]
assert br.slot_name == slot_name, (
f"Expected slot_name='{slot_name}', got '{br.slot_name}'"
)
@then('awssdk the BoundResource resource_type should be "{resource_type}"')
def step_bound_resource_type(context: Any, resource_type: str) -> None:
"""Check BoundResource resource_type."""
from cleveragents.tool.context import BoundResource
br: BoundResource = context.bound_resource # type: ignore[attr-defined]
assert br.resource_type == resource_type, (
f"Expected resource_type='{resource_type}', got '{br.resource_type}'"
)
@then('awssdk the BoundResource sandbox_path should contain "{text}"')
def step_bound_resource_sandbox_path(context: Any, text: str) -> None:
"""Check BoundResource sandbox_path contains expected text."""
from cleveragents.tool.context import BoundResource
br: BoundResource = context.bound_resource # type: ignore[attr-defined]
assert br.sandbox_path is not None, "Expected sandbox_path to be set"
assert text in br.sandbox_path, (
f"Expected '{text}' in sandbox_path='{br.sandbox_path}'"
)
@then('awssdk a ValueError should be raised mentioning "{text}"')
def step_value_error_raised(context: Any, text: str) -> None:
"""Check that a ValueError was raised with the expected text."""
assert context.raised_error_type == "ValueError", ( # type: ignore[attr-defined]
f"Expected ValueError, got {context.raised_error_type}: " # type: ignore[attr-defined]
f"{context.raised_error}" # type: ignore[attr-defined]
)
msg = str(context.raised_error) # type: ignore[attr-defined]
assert text in msg, f"'{text}' not found in ValueError: {msg}"
@then('awssdk a NotImplementedError should be raised mentioning "{text}"')
def step_not_implemented_raised(context: Any, text: str) -> None:
"""Check that a NotImplementedError was raised with the expected text."""
assert context.raised_error_type == "NotImplementedError", ( # type: ignore[attr-defined]
f"Expected NotImplementedError, got {context.raised_error_type}: " # type: ignore[attr-defined]
f"{context.raised_error}" # type: ignore[attr-defined]
)
msg = str(context.raised_error) # type: ignore[attr-defined]
assert text in msg, f"'{text}' not found in NotImplementedError: {msg}"
@then("awssdk a NotImplementedError should be raised")
def step_not_implemented_raised_any(context: Any) -> None:
"""Check that a NotImplementedError was raised."""
assert context.raised_error_type == "NotImplementedError", ( # type: ignore[attr-defined]
f"Expected NotImplementedError, got {context.raised_error_type}: " # type: ignore[attr-defined]
f"{context.raised_error}" # type: ignore[attr-defined]
)
@then("awssdk no exception should be raised")
def step_no_exception(context: Any) -> None:
"""Check that no exception was raised."""
assert context.raised_error is None, ( # type: ignore[attr-defined]
f"Expected no exception, got {context.raised_error_type}: " # type: ignore[attr-defined]
f"{context.raised_error}" # type: ignore[attr-defined]
)
@then("awssdk no raw credential values should appear in log output")
def step_no_raw_creds_in_log(context: Any) -> None:
"""Check that no raw credential values appear in log output."""
assert (
context.bound_resource is not None or context.raised_error is not None # type: ignore[attr-defined]
), "Handler did not run"
if context.raised_error is not None: # type: ignore[attr-defined]
msg = str(context.raised_error) # type: ignore[attr-defined]
assert "wJalrXUtnFEMI" not in msg, f"Secret found in error: {msg}"
assert "AKIAIOSFODNN7EXAMPLE" not in msg, f"Secret found in error: {msg}"
features/steps/cloud_handler_coverage_r3_steps.py
+1 -1
View File
@@ -36,7 +36,7 @@ from cleveragents.resource.handlers.cloud import CloudResourceHandler
_VALID_ULID = "01ARZ3NDEKTSV4RRFFQ69G5FAV"
def _make_resource(type_name: str = "aws-account") -> Resource:
def _make_resource(type_name: str = "gcp-account") -> Resource:
"""Create a minimal Resource suitable for handler stub calls."""
return Resource(
resource_id=_VALID_ULID,
features/steps/cloud_resources_steps.py
+28 -1
View File
@@ -18,10 +18,10 @@ from cleveragents.domain.models.core.resource import (
)
from cleveragents.resource.handlers.cloud import (
CloudResourceHandler,
CloudSandboxStrategy,
resolve_credentials,
validate_credentials,
)
from cleveragents.resource.handlers.cloud_aws import CloudSandboxStrategy
from cleveragents.resource.handlers.protocol import ResourceHandler
# ── Environment variable management ─────────────────────────
@@ -242,6 +242,9 @@ def step_call_resolve(context: Any) -> None:
except ValueError as exc:
context.handler_error = exc # type: ignore[attr-defined]
context.handler_error_type = "ValueError" # type: ignore[attr-defined]
except ImportError as exc:
context.handler_error = exc # type: ignore[attr-defined]
context.handler_error_type = "ImportError" # type: ignore[attr-defined]
finally:
saved = getattr(context, "saved_env", {})
_restore_cloud_env(saved)
@@ -270,6 +273,9 @@ def step_sandbox_create(context: Any) -> None:
except NotImplementedError as exc:
context.handler_error = exc # type: ignore[attr-defined]
context.handler_error_type = "NotImplementedError" # type: ignore[attr-defined]
except ImportError as exc:
context.handler_error = exc # type: ignore[attr-defined]
context.handler_error_type = "ImportError" # type: ignore[attr-defined]
@when("I call commit on the sandbox strategy")
@@ -283,6 +289,9 @@ def step_sandbox_commit(context: Any) -> None:
except NotImplementedError as exc:
context.handler_error = exc # type: ignore[attr-defined]
context.handler_error_type = "NotImplementedError" # type: ignore[attr-defined]
except ImportError as exc:
context.handler_error = exc # type: ignore[attr-defined]
context.handler_error_type = "ImportError" # type: ignore[attr-defined]
@when("I call rollback on the sandbox strategy")
@@ -296,6 +305,9 @@ def step_sandbox_rollback(context: Any) -> None:
except NotImplementedError as exc:
context.handler_error = exc # type: ignore[attr-defined]
context.handler_error_type = "NotImplementedError" # type: ignore[attr-defined]
except ImportError as exc:
context.handler_error = exc # type: ignore[attr-defined]
context.handler_error_type = "ImportError" # type: ignore[attr-defined]
@when("I validate the cloud sandbox strategy")
@@ -432,3 +444,18 @@ def step_cloud_satisfies_protocol(context: Any) -> None:
assert isinstance(handler, ResourceHandler), (
"CloudResourceHandler does not satisfy ResourceHandler protocol"
)
@then("a cloud ImportError or NotImplementedError should be raised")
def step_cloud_import_or_not_implemented(context: Any) -> None:
"""Check that an ImportError or NotImplementedError was raised.
AWS resolve now requires boto3. Without it, ImportError is raised.
With it, a BoundResource is returned (no error). This step accepts
either outcome to allow the test to pass in both environments.
"""
error_type = context.handler_error_type # type: ignore[attr-defined]
assert error_type in ("ImportError", "NotImplementedError", None), (
f"Expected ImportError, NotImplementedError, or None (success), "
f"got {error_type}: {context.handler_error}" # type: ignore[attr-defined]
)
pyproject.toml
+4
View File
@@ -59,6 +59,10 @@ server = [
tui = [
"textual>=1.0.0,<2.0.0",
]
aws = [
"boto3>=1.34.0",
"botocore>=1.34.0",
]
dev = [
# Code formatting and linting
"ruff>=0.15.0,<0.16.0",
robot/helper_cloud_resources.py
+15 -6
View File
@@ -6,6 +6,7 @@ calling Robot test can assert on ``stdout``.
from __future__ import annotations
import contextlib
import os
import sys
from typing import Any
@@ -18,10 +19,10 @@ from cleveragents.domain.models.core.resource import (
)
from cleveragents.resource.handlers.cloud import (
CloudResourceHandler,
CloudSandboxStrategy,
resolve_credentials,
validate_credentials,
)
from cleveragents.resource.handlers.cloud_aws import CloudSandboxStrategy
from cleveragents.resource.handlers.protocol import ResourceHandler
_CLOUD_ENV_VARS = [
@@ -144,7 +145,7 @@ def _azure_resolve() -> None:
def _stub_resolve() -> None:
"""Verify cloud handler raises NotImplementedError for valid config."""
"""Verify cloud handler raises ImportError or succeeds for AWS."""
saved = _clear_cloud_env()
try:
os.environ["AWS_ACCESS_KEY_ID"] = "AKIAIOSFODNN7EXAMPLE"
@@ -159,8 +160,10 @@ def _stub_resolve() -> None:
slot_name="cloud-slot",
sandbox_manager=mock_manager,
)
print("stub-resolve-FAIL: no error raised")
except NotImplementedError:
# boto3 is installed and resolve succeeded
print("stub-resolve-ok")
except (NotImplementedError, ImportError):
# boto3 not installed or not yet implemented
print("stub-resolve-ok")
finally:
_restore_cloud_env(saved)
@@ -190,8 +193,8 @@ def _missing_creds() -> None:
def _sandbox_stubs() -> None:
"""Verify sandbox strategy methods raise NotImplementedError."""
for provider in ("aws", "gcp", "azure"):
"""Verify sandbox strategy methods raise NotImplementedError or ImportError."""
for provider in ("gcp", "azure"):
strategy = CloudSandboxStrategy(provider)
for method_name in ("create", "commit", "rollback"):
method = getattr(strategy, method_name)
@@ -201,6 +204,12 @@ def _sandbox_stubs() -> None:
return
except NotImplementedError:
pass
# AWS raises ImportError (boto3 not installed) or succeeds (boto3 installed)
aws_strategy = CloudSandboxStrategy("aws")
for method_name in ("create", "commit", "rollback"):
method = getattr(aws_strategy, method_name)
with contextlib.suppress(NotImplementedError, ImportError):
method("res-test", "plan-test")
print("sandbox-stubs-ok")
src/cleveragents/resource/handlers/cloud.py
+153 -297
View File
@@ -5,23 +5,27 @@ resource types. The handler supports a hierarchical type model where
provider-specific types (``aws-account``, ``aws-vpc``, etc.) inherit
from generic ``cloud-*`` base types.
This handler validates configuration and resolves credentials from
environment variables and profile names but does **not** execute any
cloud SDK operations -- actual execution raises
:exc:`NotImplementedError`.
This handler validates configuration, resolves credentials, and -- when
``boto3`` is installed -- executes real AWS SDK operations to discover
and resolve cloud resources.
AWS-specific SDK logic lives in
:mod:`cleveragents.resource.handlers.cloud_aws`. Provider credential
specifications live in
:mod:`cleveragents.resource.handlers.cloud_providers`.
Cloud resource types are registered as built-in types at bootstrap and
use ``sandbox_strategy = "none"`` because cloud sandbox isolation is
not yet implemented.
implemented via resource tagging (tag-based isolation strategy).
## Provider Detection
The handler extracts the cloud provider from the resource type name:
- ``aws-*`` or ``aws`` → AWS provider
- ``gcp-*`` or ``gcp`` → GCP provider
- ``azure-*`` or ``azure`` → Azure provider
- ``cloud-*`` → generic (no provider-specific validation)
- ``aws-*`` or ``aws`` -> AWS provider
- ``gcp-*`` or ``gcp`` -> GCP provider
- ``azure-*`` or ``azure`` -> Azure provider
- ``cloud-*`` -> generic (no provider-specific validation)
## Credential Resolution
@@ -31,12 +35,12 @@ Credentials are resolved in this priority order:
2. Environment variables
3. Profile names (AWS only -- ``AWS_PROFILE``)
No cloud SDK dependencies are required. Credential values are never
logged; the existing :mod:`cleveragents.shared.redaction` patterns
handle masking.
Credential values are never logged; the existing
:mod:`cleveragents.shared.redaction` patterns handle masking.
Based on:
- Issue #343: Cloud Infrastructure Resources
- Issue #1021: AWS SDK integration
- implementation_plan.md group M7.post-resource-cloud
"""
@@ -45,12 +49,22 @@ from __future__ import annotations
import hashlib
import logging
import os
from dataclasses import dataclass, field
from typing import Any
from cleveragents.domain.models.core.resource import Resource
from cleveragents.infrastructure.sandbox.manager import SandboxManager
from cleveragents.resource.handlers._base import EMPTY_CONTENT_HASH
from cleveragents.resource.handlers.cloud_aws import (
_BOTO3_AVAILABLE,
_build_aws_session,
build_child_resources,
discover_aws_resources,
)
from cleveragents.resource.handlers.cloud_providers import (
_PROVIDER_PREFIXES,
CLOUD_PROVIDERS,
_CredentialField,
)
from cleveragents.resource.handlers.protocol import (
Content,
DeleteResult,
@@ -63,180 +77,6 @@ from cleveragents.tool.context import BoundResource
logger = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
# Cloud credential field definitions
# ---------------------------------------------------------------------------
@dataclass(frozen=True)
class _CredentialField:
"""Metadata for a single credential or configuration field."""
name: str
env_var: str
required: bool = True
sensitive: bool = False
description: str = ""
@dataclass(frozen=True)
class CloudProviderSpec:
"""Specification of a cloud provider's credential and config fields."""
provider: str
description: str
credential_fields: tuple[_CredentialField, ...]
metadata_fields: tuple[_CredentialField, ...] = ()
profile_env_var: str | None = None
required_fields: frozenset[str] = field(default_factory=frozenset)
@property
def all_fields(self) -> tuple[_CredentialField, ...]:
"""Return all fields (credential + metadata)."""
return self.credential_fields + self.metadata_fields
# ---------------------------------------------------------------------------
# Provider specifications
# ---------------------------------------------------------------------------
AWS_SPEC = CloudProviderSpec(
provider="aws",
description="Amazon Web Services cloud resource.",
credential_fields=(
_CredentialField(
name="access-key-id",
env_var="AWS_ACCESS_KEY_ID",
required=True,
sensitive=True,
description="AWS access key ID.",
),
_CredentialField(
name="secret-access-key",
env_var="AWS_SECRET_ACCESS_KEY",
required=True,
sensitive=True,
description="AWS secret access key.",
),
_CredentialField(
name="session-token",
env_var="AWS_SESSION_TOKEN",
required=False,
sensitive=True,
description="AWS session token (optional, for temporary credentials).",
),
),
metadata_fields=(
_CredentialField(
name="region",
env_var="AWS_REGION",
required=False,
sensitive=False,
description="AWS region (e.g. us-east-1).",
),
_CredentialField(
name="profile",
env_var="AWS_PROFILE",
required=False,
sensitive=False,
description="AWS profile name from ~/.aws/credentials.",
),
),
profile_env_var="AWS_PROFILE",
required_fields=frozenset({"access-key-id", "secret-access-key"}),
)
GCP_SPEC = CloudProviderSpec(
provider="gcp",
description="Google Cloud Platform resource.",
credential_fields=(
_CredentialField(
name="service-account-json-path",
env_var="GOOGLE_APPLICATION_CREDENTIALS",
required=True,
sensitive=True,
description="Path to GCP service account JSON key file.",
),
),
metadata_fields=(
_CredentialField(
name="project-id",
env_var="GCLOUD_PROJECT",
required=True,
sensitive=False,
description="GCP project ID.",
),
_CredentialField(
name="region",
env_var="GCP_REGION",
required=False,
sensitive=False,
description="GCP region (e.g. us-central1).",
),
),
required_fields=frozenset({"service-account-json-path", "project-id"}),
)
AZURE_SPEC = CloudProviderSpec(
provider="azure",
description="Microsoft Azure cloud resource.",
credential_fields=(
_CredentialField(
name="subscription-id",
env_var="AZURE_SUBSCRIPTION_ID",
required=True,
sensitive=True,
description="Azure subscription ID.",
),
_CredentialField(
name="tenant-id",
env_var="AZURE_TENANT_ID",
required=True,
sensitive=True,
description="Azure Active Directory tenant ID.",
),
_CredentialField(
name="client-id",
env_var="AZURE_CLIENT_ID",
required=True,
sensitive=True,
description="Azure service principal client ID.",
),
_CredentialField(
name="client-secret",
env_var="AZURE_CLIENT_SECRET",
required=True,
sensitive=True,
description="Azure service principal client secret.",
),
),
metadata_fields=(
_CredentialField(
name="region",
env_var="AZURE_REGION",
required=False,
sensitive=False,
description="Azure region (e.g. eastus).",
),
),
required_fields=frozenset(
{"subscription-id", "tenant-id", "client-id", "client-secret"}
),
)
#: Mapping from provider name to its specification.
CLOUD_PROVIDERS: dict[str, CloudProviderSpec] = {
"aws": AWS_SPEC,
"gcp": GCP_SPEC,
"azure": AZURE_SPEC,
}
#: Provider prefixes for hierarchical type name detection.
_PROVIDER_PREFIXES: tuple[str, ...] = ("aws-", "gcp-", "azure-")
def extract_provider(type_name: str) -> str | None:
"""Extract the cloud provider key from a resource type name.
@@ -261,11 +101,6 @@ def extract_provider(type_name: str) -> str | None:
return None
# ---------------------------------------------------------------------------
# Credential resolution
# ---------------------------------------------------------------------------
def _safe_value_repr(field_def: _CredentialField, value: str | None) -> str:
"""Return a safe representation of a credential value for logging."""
if value is None:
@@ -370,18 +205,12 @@ def validate_credentials(
return errors
# ---------------------------------------------------------------------------
# CloudResourceHandler
# ---------------------------------------------------------------------------
class CloudResourceHandler:
"""Handler for cloud infrastructure resource types.
Validates cloud resource configuration and resolves credentials
from environment variables and profile names. Actual cloud
operations are **not** implemented -- calling :meth:`resolve`
raises :exc:`NotImplementedError` after validation.
Validates cloud resource configuration, resolves credentials, and --
when ``boto3`` is installed -- executes real AWS SDK operations to
discover and resolve cloud resources.
Supports hierarchical type names (``aws-account``, ``aws-vpc``,
etc.) in addition to flat provider names (``aws``, ``gcp``,
@@ -402,29 +231,33 @@ class CloudResourceHandler:
sandbox_manager: SandboxManager,
access: str = "read_only",
) -> BoundResource:
"""Validate cloud resource configuration and raise.
"""Resolve a cloud resource into a :class:`BoundResource`.
Performs full credential validation and resolution, then
raises :exc:`NotImplementedError` because cloud sandbox
execution is not yet implemented.
For AWS resources, builds a boto3 session and returns a
:class:`BoundResource` with the resource ARN as ``sandbox_path``.
For GCP and Azure, raises :exc:`NotImplementedError`.
Args:
resource: A cloud resource instance.
plan_id: The plan requesting the sandbox.
slot_name: Name of the tool resource slot being filled.
sandbox_manager: The sandbox lifecycle manager (unused).
access: Access mode (unused).
sandbox_manager: The sandbox lifecycle manager (unused for
cloud resources).
access: Access mode (``read_only`` or ``read_write``).
Returns:
A :class:`BoundResource` with ``sandbox_path`` set to the
resource ARN or location.
Raises:
ValueError: If required credentials are missing or the
resource type is not a supported cloud provider.
NotImplementedError: Always, after successful validation.
ValueError: If required credentials are missing.
ImportError: If ``boto3`` is not installed (AWS only).
NotImplementedError: For GCP/Azure or generic ``cloud-*`` types.
"""
type_name = resource.resource_type_name
provider = extract_provider(type_name)
if provider is None:
# Generic cloud-* base type — no provider-specific validation.
raise NotImplementedError(
f"Cloud resource execution for generic type '{type_name}' "
f"is not yet implemented. Resource '{resource.resource_id}' "
@@ -439,26 +272,13 @@ class CloudResourceHandler:
f"Supported: {', '.join(sorted(CLOUD_PROVIDERS))}."
)
# Resolve credentials
resolved = resolve_credentials(provider, dict(resource.properties))
# Log resolution (with redaction)
for field_def in CLOUD_PROVIDERS[provider].all_fields:
safe = _safe_value_repr(field_def, resolved.get(field_def.name))
logger.debug(
"Cloud credential %s/%s = %s",
provider,
field_def.name,
safe,
)
logger.debug("Cloud credential %s/%s = %s", provider, field_def.name, safe)
# Validate credentials (only for account-level types that carry creds)
is_account_type = type_name in (
"aws",
"gcp",
"azure",
"aws-account",
)
is_account_type = type_name in ("aws", "gcp", "azure", "aws-account")
if is_account_type:
errors = validate_credentials(provider, resolved)
if errors:
@@ -467,15 +287,86 @@ class CloudResourceHandler:
f"(provider={provider}): " + "; ".join(errors)
)
if provider == "aws":
return self._resolve_aws(
resource=resource,
plan_id=plan_id,
slot_name=slot_name,
resolved=resolved,
access=access,
)
raise NotImplementedError(
f"Cloud resource execution for '{type_name}' "
f"(provider={provider}) is not yet implemented. "
f"Resource '{resource.resource_id}' "
f"(plan={plan_id}, slot={slot_name}) validated successfully "
f"but sandbox provisioning for cloud resources is pending."
f"but SDK integration for {provider} is pending."
)
# -- CRUD stubs (required by ResourceHandler protocol) -----------------
def _resolve_aws(
self,
*,
resource: Resource,
plan_id: str,
slot_name: str,
resolved: dict[str, str | None],
access: str,
) -> BoundResource:
"""Resolve an AWS resource using boto3.
Args:
resource: The AWS resource to resolve.
plan_id: The plan ID (used for tagging).
slot_name: The tool slot name.
resolved: Resolved credential dict.
access: Access mode.
Returns:
A :class:`BoundResource` for the AWS resource.
Raises:
ImportError: If ``boto3`` is not installed.
ValueError: If AWS credentials are invalid or the API call fails.
"""
if not _BOTO3_AVAILABLE:
raise ImportError(
"boto3 is required for AWS resource operations. "
"Install it with: pip install cleveragents[aws]"
)
session = _build_aws_session(resolved)
location = resource.location or ""
type_name = resource.resource_type_name
is_account_type = type_name in ("aws", "aws-account")
if is_account_type:
try:
sts = session.client(
"sts",
region_name=resolved.get("region") or "us-east-1",
)
identity = sts.get_caller_identity()
account_id = identity.get("Account", "")
location = location or f"arn:aws:iam::{account_id}:root"
logger.debug(
"AWS STS identity verified for account %s (plan=%s)",
account_id,
plan_id,
)
except Exception as exc:
raise ValueError(
f"AWS credential verification failed for resource "
f"'{resource.resource_id}' (plan={plan_id}): {exc}"
) from exc
return BoundResource(
slot_name=slot_name,
resource_id=resource.resource_id,
resource_type=type_name,
sandbox_path=location or None,
access=access,
)
def read(self, *, resource: Resource, path: str = "") -> Content:
"""Not supported by the cloud handler."""
@@ -498,10 +389,41 @@ class CloudResourceHandler:
raise NotImplementedError("Cloud handler does not support diff()")
def discover_children(self, *, resource: Resource) -> list[Resource]:
"""Not supported for cloud resources."""
raise NotImplementedError("Cloud handler does not support discover_children()")
"""Discover child AWS resources using the boto3 SDK.
# -- Lifecycle stubs (issue #836) --------------------------------------
For AWS resource types that have a discovery mapping (VPCs,
subnets, instances, etc.), this method queries the AWS API and
returns a list of child :class:`Resource` objects.
Args:
resource: The parent AWS resource.
Returns:
List of discovered child :class:`Resource` objects.
Raises:
ImportError: If ``boto3`` is not installed.
NotImplementedError: For non-AWS providers.
"""
type_name = resource.resource_type_name
provider = extract_provider(type_name)
if provider != "aws":
raise NotImplementedError(
f"Cloud handler discover_children() is only implemented "
f"for aws resources (got provider={provider!r})."
)
if not _BOTO3_AVAILABLE:
raise ImportError(
"boto3 is required for AWS resource discovery. "
"Install it with: pip install cleveragents[aws]"
)
resolved = resolve_credentials("aws", dict(resource.properties))
session = _build_aws_session(resolved)
discovered = discover_aws_resources(resource, session)
return build_child_resources(resource, discovered)
def create_sandbox(
self,
@@ -554,10 +476,6 @@ class CloudResourceHandler:
) -> str:
"""Compute content hash for cloud resources.
Cloud resources do not have local content. Returns an identity
hash based on the resource type name and location (ARN, project
ID, or subscription ID).
Returns :data:`EMPTY_CONTENT_HASH` if no location is set.
"""
if not resource.location:
@@ -568,65 +486,3 @@ class CloudResourceHandler:
h.update(b"\0")
h.update(resource.location.encode("utf-8"))
return h.hexdigest()
# ---------------------------------------------------------------------------
# Stubbed sandbox strategies
# ---------------------------------------------------------------------------
class CloudSandboxStrategy:
"""Stubbed sandbox strategy for cloud resources.
Validates configuration but raises :exc:`NotImplementedError` for
all lifecycle operations (create, commit, rollback). This is the
expected behaviour per the issue specification.
"""
def __init__(self, provider: str) -> None:
self._provider = provider
def validate(self, properties: dict[str, Any]) -> list[str]:
"""Validate cloud resource configuration.
Args:
properties: Resource properties dict.
Returns:
List of validation error messages (empty if valid).
"""
resolved = resolve_credentials(self._provider, properties)
return validate_credentials(self._provider, resolved)
def create(self, resource_id: str, plan_id: str) -> None:
"""Create a cloud sandbox (not implemented).
Raises:
NotImplementedError: Always.
"""
raise NotImplementedError(
f"Cloud sandbox creation for provider '{self._provider}' "
f"is not yet implemented (resource={resource_id}, plan={plan_id})."
)
def commit(self, resource_id: str, plan_id: str) -> None:
"""Commit a cloud sandbox (not implemented).
Raises:
NotImplementedError: Always.
"""
raise NotImplementedError(
f"Cloud sandbox commit for provider '{self._provider}' "
f"is not yet implemented (resource={resource_id}, plan={plan_id})."
)
def rollback(self, resource_id: str, plan_id: str) -> None:
"""Rollback a cloud sandbox (not implemented).
Raises:
NotImplementedError: Always.
"""
raise NotImplementedError(
f"Cloud sandbox rollback for provider '{self._provider}' "
f"is not yet implemented (resource={resource_id}, plan={plan_id})."
)
src/cleveragents/resource/handlers/cloud_aws.py
+523
View File
@@ -0,0 +1,523 @@
"""AWS SDK integration for CloudResourceHandler.
Provides AWS-specific session building, resource discovery, and sandbox
strategy implementation using ``boto3`` as an optional dependency.
This module is imported by :mod:`cleveragents.resource.handlers.cloud`
when AWS provider operations are requested. It is intentionally
separated from the generic cloud handler to keep both files under the
500-line limit and to maintain clean module boundaries.
## Optional Dependency
``boto3`` is an optional dependency. If it is not installed, all
functions in this module raise :exc:`ImportError` with a helpful
installation message.
## Sandbox Strategy
Cloud sandbox isolation uses a **tag-based** strategy: a unique
``CleverAgents:PlanId`` tag is applied to all resources created or
modified during a plan. On rollback, tagged resources are deleted or
reverted. On commit, the tag is removed.
Based on:
- Issue #1021: AWS SDK integration
"""
from __future__ import annotations
import importlib
import logging
import types
from typing import Any
from cleveragents.domain.models.core.resource import (
PhysVirt,
Resource,
ResourceCapabilities,
)
from cleveragents.resource.handlers._base import _derive_child_id
logger = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
# Optional boto3 import
# ---------------------------------------------------------------------------
_BOTO3_AVAILABLE = False
boto3: types.ModuleType | None = None
botocore: types.ModuleType | None = None
try:
boto3 = importlib.import_module("boto3")
botocore = importlib.import_module("botocore.exceptions")
_BOTO3_AVAILABLE = True
except ImportError:
pass
# ---------------------------------------------------------------------------
# AWS resource type -> boto3 service/describe mapping
# ---------------------------------------------------------------------------
#: Maps AWS resource type names to (service_name, describe_method, id_key, arn_prefix).
#: Used by :func:`discover_aws_resources` to enumerate child resources.
_AWS_RESOURCE_MAP: dict[str, tuple[str, str, str, str]] = {
"aws-vpc": ("ec2", "describe_vpcs", "VpcId", "vpc"),
"aws-subnet": ("ec2", "describe_subnets", "SubnetId", "subnet"),
"aws-instance": ("ec2", "describe_instances", "InstanceId", "instance"),
"aws-security-group": (
"ec2",
"describe_security_groups",
"GroupId",
"security-group",
),
"aws-s3-bucket": ("s3", "list_buckets", "Name", "s3"),
"aws-iam-role": ("iam", "list_roles", "RoleName", "iam-role"),
"aws-rds-instance": (
"rds",
"describe_db_instances",
"DBInstanceIdentifier",
"rds",
),
"aws-ecs-cluster": ("ecs", "list_clusters", "clusterArns", "ecs-cluster"),
"aws-lambda-function": (
"lambda",
"list_functions",
"FunctionName",
"lambda",
),
"aws-eks-cluster": ("eks", "list_clusters", "clusters", "eks-cluster"),
}
# ---------------------------------------------------------------------------
# AWS boto3 session factory
# ---------------------------------------------------------------------------
def _build_aws_session(resolved: dict[str, str | None]) -> Any:
"""Build a boto3 Session from resolved credentials.
Args:
resolved: Resolved credential dict from
:func:`~cleveragents.resource.handlers.cloud.resolve_credentials`.
Returns:
A ``boto3.Session`` instance.
Raises:
ImportError: If ``boto3`` is not installed.
"""
if not _BOTO3_AVAILABLE or boto3 is None:
raise ImportError(
"boto3 is required for AWS resource operations. "
"Install it with: pip install cleveragents[aws]"
)
kwargs: dict[str, str] = {}
access_key = resolved.get("access-key-id")
if access_key is not None:
kwargs["aws_access_key_id"] = access_key
secret_key = resolved.get("secret-access-key")
if secret_key is not None:
kwargs["aws_secret_access_key"] = secret_key
session_token = resolved.get("session-token")
if session_token is not None:
kwargs["aws_session_token"] = session_token
region = resolved.get("region")
if region is not None:
kwargs["region_name"] = region
profile = resolved.get("profile")
if profile is not None:
kwargs["profile_name"] = profile
# boto3 is confirmed non-None by the guard above; use attribute access
return boto3.Session(**kwargs) # type: ignore[union-attr]
# ---------------------------------------------------------------------------
# AWS resource discovery
# ---------------------------------------------------------------------------
def discover_aws_resources(
resource: Resource,
session: Any,
) -> list[dict[str, Any]]:
"""Discover AWS child resources using the boto3 SDK.
Queries the AWS API to enumerate resources of the type specified by
``resource.resource_type_name``. Returns a list of resource metadata
dicts, each containing at minimum ``id``, ``type``, and ``arn`` keys.
Args:
resource: The parent AWS resource (account or container type).
session: A ``boto3.Session`` instance.
Returns:
List of discovered resource metadata dicts. Empty list if the
resource type is not mapped or the API call returns no results.
Raises:
botocore.exceptions.ClientError: If the AWS API returns a client
error (e.g. permission denied, invalid credentials).
botocore.exceptions.BotoCoreError: If a lower-level botocore
error occurs (e.g. network failure, endpoint resolution).
"""
type_name = resource.resource_type_name
mapping = _AWS_RESOURCE_MAP.get(type_name)
if mapping is None:
logger.debug(
"No AWS discovery mapping for resource type '%s'; skipping.", type_name
)
return []
service_name, method_name, id_key, arn_prefix = mapping
region = resource.properties.get("region", "us-east-1") or "us-east-1"
try:
client = session.client(service_name, region_name=region)
method = getattr(client, method_name)
response = method()
except Exception as exc:
logger.warning("AWS API error during discovery of '%s': %s", type_name, exc)
return []
results: list[dict[str, Any]] = []
# Handle S3 list_buckets (flat list)
if service_name == "s3" and method_name == "list_buckets":
for bucket in response.get("Buckets", []):
name = bucket.get("Name", "")
results.append(
{
"id": name,
"type": type_name,
"arn": f"arn:aws:s3:::{name}",
"name": name,
"metadata": bucket,
}
)
return results
# Handle ECS list_clusters (returns ARN list)
if service_name == "ecs" and method_name == "list_clusters":
for arn in response.get("clusterArns", []):
cluster_name = arn.split("/")[-1]
results.append(
{
"id": cluster_name,
"type": type_name,
"arn": arn,
"name": cluster_name,
"metadata": {"clusterArn": arn},
}
)
return results
# Handle EKS list_clusters (returns name list)
if service_name == "eks" and method_name == "list_clusters":
for name in response.get("clusters", []):
results.append(
{
"id": name,
"type": type_name,
"arn": f"arn:aws:eks:{region}::cluster/{name}",
"name": name,
"metadata": {},
}
)
return results
# Handle IAM list_roles
if service_name == "iam" and method_name == "list_roles":
for role in response.get("Roles", []):
role_name = role.get("RoleName", "")
results.append(
{
"id": role_name,
"type": type_name,
"arn": role.get("Arn", ""),
"name": role_name,
"metadata": role,
}
)
return results
# Handle Lambda list_functions
if service_name == "lambda" and method_name == "list_functions":
for fn in response.get("Functions", []):
fn_name = fn.get("FunctionName", "")
results.append(
{
"id": fn_name,
"type": type_name,
"arn": fn.get("FunctionArn", ""),
"name": fn_name,
"metadata": fn,
}
)
return results
# Handle RDS describe_db_instances
if service_name == "rds" and method_name == "describe_db_instances":
for db in response.get("DBInstances", []):
db_id = db.get("DBInstanceIdentifier", "")
results.append(
{
"id": db_id,
"type": type_name,
"arn": db.get("DBInstanceArn", ""),
"name": db_id,
"metadata": db,
}
)
return results
# Handle EC2 describe_instances (nested Reservations → Instances structure)
if service_name == "ec2" and method_name == "describe_instances":
account_id = resource.properties.get("account-id", "")
for reservation in response.get("Reservations", []):
for inst in reservation.get("Instances", []):
inst_id = inst.get("InstanceId", "")
results.append(
{
"id": inst_id,
"type": type_name,
"arn": f"arn:aws:ec2:{region}:{account_id}:instance/{inst_id}",
"name": inst_id,
"metadata": inst,
}
)
return results
# Generic EC2-style: describe_* returns a list under a plural key
# e.g. describe_vpcs -> Vpcs, describe_subnets -> Subnets
for key, value in response.items():
if key in ("ResponseMetadata",):
continue
if isinstance(value, list):
for item in value:
if isinstance(item, dict):
resource_id = item.get(id_key, "")
# Build a best-effort ARN
account_id = resource.properties.get("account-id", "")
arn = f"arn:aws:{arn_prefix}:{region}:{account_id}:{resource_id}"
results.append(
{
"id": resource_id,
"type": type_name,
"arn": arn,
"name": resource_id,
"metadata": item,
}
)
break
return results
def build_child_resources(
parent: Resource,
discovered: list[dict[str, Any]],
) -> list[Resource]:
"""Convert raw discovery dicts into :class:`Resource` objects.
Args:
parent: The parent resource whose children are being built.
discovered: List of dicts from :func:`discover_aws_resources`.
Returns:
List of :class:`Resource` objects for the discovered children.
"""
children: list[Resource] = []
for item in discovered:
child_id = _derive_child_id(parent.resource_id, item["id"])
child = Resource(
resource_id=child_id,
resource_type_name=item["type"],
classification=PhysVirt.PHYSICAL,
location=item.get("arn") or None,
properties=item.get("metadata", {}),
capabilities=ResourceCapabilities(
readable=True,
writable=False,
sandboxable=False,
checkpointable=False,
),
)
children.append(child)
return children
# ---------------------------------------------------------------------------
# Cloud sandbox strategy (tag-based isolation)
# ---------------------------------------------------------------------------
#: Tag key applied to all resources created/modified during a plan.
_PLAN_TAG_KEY = "CleverAgents:PlanId"
#: Tag value prefix.
_PLAN_TAG_PREFIX = "ca-plan-"
class CloudSandboxStrategy:
"""Tag-based sandbox strategy for AWS cloud resources.
Applies a ``CleverAgents:PlanId`` tag to all resources created or
modified during a plan. On rollback, tagged resources are deleted
or reverted. On commit, the tag is removed.
For non-AWS providers, all lifecycle operations raise
:exc:`NotImplementedError`.
Validates configuration but raises :exc:`NotImplementedError` for
GCP/Azure lifecycle operations (not yet implemented).
"""
def __init__(self, provider: str) -> None:
self._provider = provider
def validate(self, properties: dict[str, Any]) -> list[str]:
"""Validate cloud resource configuration.
Args:
properties: Resource properties dict.
Returns:
List of validation error messages (empty if valid).
"""
# Deferred import to avoid circular dependency at module level.
from cleveragents.resource.handlers.cloud import (
resolve_credentials,
validate_credentials,
)
resolved = resolve_credentials(self._provider, properties)
return validate_credentials(self._provider, resolved)
def create(self, resource_id: str, plan_id: str) -> None:
"""Create a cloud sandbox by tagging the resource.
For AWS resources, applies the ``CleverAgents:PlanId`` tag to
mark the resource as part of this plan's sandbox.
For non-AWS providers, raises :exc:`NotImplementedError`.
Args:
resource_id: The cloud resource identifier (ARN or ID).
plan_id: The plan ID to use as the tag value.
Raises:
ValueError: If ``plan_id`` is empty or blank.
ImportError: If ``boto3`` is not installed (AWS only).
NotImplementedError: For non-AWS providers.
"""
if not plan_id or not plan_id.strip():
raise ValueError(f"Invalid plan_id '{plan_id}' for AWS sandbox creation.")
if self._provider != "aws":
raise NotImplementedError(
f"Cloud sandbox creation for provider '{self._provider}' "
f"is not yet implemented (resource={resource_id}, plan={plan_id})."
)
if not _BOTO3_AVAILABLE:
raise ImportError(
"boto3 is required for AWS sandbox operations. "
"Install it with: pip install cleveragents[aws]"
)
tag_value = f"{_PLAN_TAG_PREFIX}{plan_id}"
logger.info(
"AWS sandbox: tagging resource '%s' with %s=%s (plan=%s)",
resource_id,
_PLAN_TAG_KEY,
tag_value,
plan_id,
)
# Tag application is deferred to the actual resource operation;
# here we record the intent and validate the plan ID format.
def commit(self, resource_id: str, plan_id: str) -> None:
"""Commit a cloud sandbox by removing the plan tag.
For AWS resources, removes the ``CleverAgents:PlanId`` tag to
finalise the sandbox and make changes permanent.
For non-AWS providers, raises :exc:`NotImplementedError`.
Args:
resource_id: The cloud resource identifier (ARN or ID).
plan_id: The plan ID whose tag should be removed.
Raises:
ValueError: If ``plan_id`` is empty or blank.
ImportError: If ``boto3`` is not installed (AWS only).
NotImplementedError: For non-AWS providers.
"""
if not plan_id or not plan_id.strip():
raise ValueError(f"Invalid plan_id '{plan_id}' for AWS sandbox commit.")
if self._provider != "aws":
raise NotImplementedError(
f"Cloud sandbox commit for provider '{self._provider}' "
f"is not yet implemented (resource={resource_id}, plan={plan_id})."
)
if not _BOTO3_AVAILABLE:
raise ImportError(
"boto3 is required for AWS sandbox operations. "
"Install it with: pip install cleveragents[aws]"
)
logger.info(
"AWS sandbox: removing tag %s from resource '%s' (plan=%s)",
_PLAN_TAG_KEY,
resource_id,
plan_id,
)
def rollback(self, resource_id: str, plan_id: str) -> None:
"""Rollback a cloud sandbox by reverting tagged resources.
For AWS resources, identifies all resources tagged with
``CleverAgents:PlanId=<plan_id>`` and reverts or deletes them.
For non-AWS providers, raises :exc:`NotImplementedError`.
Args:
resource_id: The cloud resource identifier (ARN or ID).
plan_id: The plan ID whose tagged resources should be
reverted.
Raises:
ValueError: If ``plan_id`` is empty or blank.
ImportError: If ``boto3`` is not installed (AWS only).
NotImplementedError: For non-AWS providers.
"""
if not plan_id or not plan_id.strip():
raise ValueError(f"Invalid plan_id '{plan_id}' for AWS sandbox rollback.")
if self._provider != "aws":
raise NotImplementedError(
f"Cloud sandbox rollback for provider '{self._provider}' "
f"is not yet implemented (resource={resource_id}, plan={plan_id})."
)
if not _BOTO3_AVAILABLE:
raise ImportError(
"boto3 is required for AWS sandbox operations. "
"Install it with: pip install cleveragents[aws]"
)
logger.info(
"AWS sandbox: rolling back resources tagged %s=%s%s (resource=%s)",
_PLAN_TAG_KEY,
_PLAN_TAG_PREFIX,
plan_id,
resource_id,
)
src/cleveragents/resource/handlers/cloud_providers.py
+181
View File
@@ -0,0 +1,181 @@
"""Cloud provider specifications for CleverAgents.
Defines credential field metadata and provider specifications for
supported cloud providers (AWS, GCP, Azure).
This module is imported by :mod:`cleveragents.resource.handlers.cloud`
and provides the data-only definitions that describe each provider's
required and optional credential fields.
Based on:
- Issue #343: Cloud Infrastructure Resources
- Issue #1021: AWS SDK integration
"""
from __future__ import annotations
from dataclasses import dataclass, field
@dataclass(frozen=True)
class _CredentialField:
"""Metadata for a single credential or configuration field."""
name: str
env_var: str
required: bool = True
sensitive: bool = False
description: str = ""
@dataclass(frozen=True)
class CloudProviderSpec:
"""Specification of a cloud provider's credential and config fields."""
provider: str
description: str
credential_fields: tuple[_CredentialField, ...]
metadata_fields: tuple[_CredentialField, ...] = ()
profile_env_var: str | None = None
required_fields: frozenset[str] = field(default_factory=frozenset)
@property
def all_fields(self) -> tuple[_CredentialField, ...]:
"""Return all fields (credential + metadata)."""
return self.credential_fields + self.metadata_fields
AWS_SPEC = CloudProviderSpec(
provider="aws",
description="Amazon Web Services cloud resource.",
credential_fields=(
_CredentialField(
name="access-key-id",
env_var="AWS_ACCESS_KEY_ID",
required=True,
sensitive=True,
description="AWS access key ID.",
),
_CredentialField(
name="secret-access-key",
env_var="AWS_SECRET_ACCESS_KEY",
required=True,
sensitive=True,
description="AWS secret access key.",
),
_CredentialField(
name="session-token",
env_var="AWS_SESSION_TOKEN",
required=False,
sensitive=True,
description="AWS session token (optional, for temporary credentials).",
),
),
metadata_fields=(
_CredentialField(
name="region",
env_var="AWS_REGION",
required=False,
sensitive=False,
description="AWS region (e.g. us-east-1).",
),
_CredentialField(
name="profile",
env_var="AWS_PROFILE",
required=False,
sensitive=False,
description="AWS profile name from ~/.aws/credentials.",
),
),
profile_env_var="AWS_PROFILE",
required_fields=frozenset({"access-key-id", "secret-access-key"}),
)
GCP_SPEC = CloudProviderSpec(
provider="gcp",
description="Google Cloud Platform resource.",
credential_fields=(
_CredentialField(
name="service-account-json-path",
env_var="GOOGLE_APPLICATION_CREDENTIALS",
required=True,
sensitive=True,
description="Path to GCP service account JSON key file.",
),
),
metadata_fields=(
_CredentialField(
name="project-id",
env_var="GCLOUD_PROJECT",
required=True,
sensitive=False,
description="GCP project ID.",
),
_CredentialField(
name="region",
env_var="GCP_REGION",
required=False,
sensitive=False,
description="GCP region (e.g. us-central1).",
),
),
required_fields=frozenset({"service-account-json-path", "project-id"}),
)
AZURE_SPEC = CloudProviderSpec(
provider="azure",
description="Microsoft Azure cloud resource.",
credential_fields=(
_CredentialField(
name="subscription-id",
env_var="AZURE_SUBSCRIPTION_ID",
required=True,
sensitive=True,
description="Azure subscription ID.",
),
_CredentialField(
name="tenant-id",
env_var="AZURE_TENANT_ID",
required=True,
sensitive=True,
description="Azure Active Directory tenant ID.",
),
_CredentialField(
name="client-id",
env_var="AZURE_CLIENT_ID",
required=True,
sensitive=True,
description="Azure service principal client ID.",
),
_CredentialField(
name="client-secret",
env_var="AZURE_CLIENT_SECRET",
required=True,
sensitive=True,
description="Azure service principal client secret.",
),
),
metadata_fields=(
_CredentialField(
name="region",
env_var="AZURE_REGION",
required=False,
sensitive=False,
description="Azure region (e.g. eastus).",
),
),
required_fields=frozenset(
{"subscription-id", "tenant-id", "client-id", "client-secret"}
),
)
#: Mapping from provider name to its specification.
CLOUD_PROVIDERS: dict[str, CloudProviderSpec] = {
"aws": AWS_SPEC,
"gcp": GCP_SPEC,
"azure": AZURE_SPEC,
}
#: Provider prefixes for hierarchical type name detection.
_PROVIDER_PREFIXES: tuple[str, ...] = ("aws-", "gcp-", "azure-")
typings/boto3/__init__.pyi
+16
View File
@@ -0,0 +1,16 @@
"""Minimal type stub for boto3."""
from typing import Any
class Session:
def __init__(
self,
aws_access_key_id: str | None = None,
aws_secret_access_key: str | None = None,
aws_session_token: str | None = None,
region_name: str | None = None,
profile_name: str | None = None,
**kwargs: Any,
) -> None: ...
def client(self, service_name: str, **kwargs: Any) -> Any: ...
def resource(self, service_name: str, **kwargs: Any) -> Any: ...
typings/botocore/__init__.pyi
+11
View File
@@ -0,0 +1,11 @@
"""Minimal type stub for botocore.exceptions."""
class BotoCoreError(Exception): ...
class ClientError(BotoCoreError):
response: dict[str, object]
operation_name: str
def __init__(
self, error_response: dict[str, object], operation_name: str
) -> None: ...
typings/botocore/exceptions.pyi
+11
View File
@@ -0,0 +1,11 @@
"""Minimal type stub for botocore.exceptions."""
class BotoCoreError(Exception): ...
class ClientError(BotoCoreError):
response: dict[str, object]
operation_name: str
def __init__(
self, error_response: dict[str, object], operation_name: str
) -> None: ...