Boto3 Wrapper Reference

Complete reference for the ASCEND Boto3 governance wrapper, which adds automatic risk classification and authorization to AWS SDK operations.

Installation

pip install ascend-boto3-wrapper

GovernedBoto3Session

The main class for creating governed AWS service clients.

Constructor

from ascend_boto3 import GovernedBoto3Session

session = GovernedBoto3Session(
    ascend_client: AscendClient,
    region_name: str = None,
    profile_name: str = None,
    aws_access_key_id: str = None,
    aws_secret_access_key: str = None,
    aws_session_token: str = None,
    fail_mode: str = "closed",
    log_all_operations: bool = True,
    skip_read_operations: bool = False,
    custom_risk_mappings: Dict[str, str] = None,
    excluded_operations: List[str] = None,
    metrics_callback: Callable = None
)

Parameters

Parameter	Type	Default	Description
ascend_client	AscendClient	Required	Initialized ASCEND client
region_name	str	None	AWS region (falls back to env/config)
profile_name	str	None	AWS profile name
aws_access_key_id	str	None	AWS access key ID
aws_secret_access_key	str	None	AWS secret access key
aws_session_token	str	None	AWS session token
fail_mode	str	"closed"	Behavior when ASCEND unavailable
log_all_operations	bool	True	Log all operations to ASCEND
skip_read_operations	bool	False	Skip governance for read operations
custom_risk_mappings	Dict	None	Override default risk classifications
excluded_operations	List	None	Operations to bypass governance
metrics_callback	Callable	None	Callback for metrics

Methods

client()

Create a governed service client.

def client(
    service_name: str,
    **kwargs
) -> GovernedServiceClient

Example:

s3 = session.client("s3")
dynamodb = session.client("dynamodb")
lambda_client = session.client("lambda")

resource()

Create a governed service resource.

def resource(
    service_name: str,
    **kwargs
) -> GovernedServiceResource

Example:

s3_resource = session.resource("s3")
bucket = s3_resource.Bucket("my-bucket")

---

Risk Classification

Default Risk Mappings

The wrapper includes default risk mappings for common AWS operations:

S3 Operations

Operation	Risk Level
ListBuckets, ListObjects, GetObject, HeadObject	Low
PutObject, CopyObject, CreateBucket	Medium
DeleteObject, DeleteBucket	High
DeleteBucketPolicy, PutBucketPolicy	Critical

DynamoDB Operations

Operation	Risk Level
GetItem, Query, Scan, DescribeTable	Low
PutItem, UpdateItem, CreateTable	Medium
DeleteItem	High
DeleteTable	Critical

EC2 Operations

Operation	Risk Level
DescribeInstances, DescribeSecurityGroups	Low
RunInstances, CreateSecurityGroup	Medium
TerminateInstances, StopInstances	High
DeleteSecurityGroup, ModifyVpcAttribute	Critical

Lambda Operations

Operation	Risk Level
ListFunctions, GetFunction	Low
Invoke	Medium
UpdateFunctionCode, UpdateFunctionConfiguration	High
DeleteFunction	Critical

IAM Operations

Operation	Risk Level
ListUsers, ListRoles, GetUser	Low
CreateUser, CreateRole	High
AttachUserPolicy, AttachRolePolicy	Critical
DeleteUser, DeleteRole	Critical

Custom Risk Mappings

Override default risk classifications:

session = GovernedBoto3Session(
    ascend_client=ascend,
    custom_risk_mappings={
        # Format: "service:Operation"
        "s3:DeleteObject": "critical",      # Upgrade from high
        "lambda:Invoke": "high",            # Upgrade from medium
        "dynamodb:GetItem": "medium"        # Upgrade from low
    }
)

---

GovernedServiceClient

Wrapper around boto3 service clients with governance.

Methods

All standard boto3 methods are available. Each method is intercepted for governance evaluation.

s3 = session.client("s3")

# All standard methods work
s3.list_buckets()
s3.put_object(Bucket="bucket", Key="key", Body="data")
s3.delete_object(Bucket="bucket", Key="key")

Governance Bypass

For operations that should bypass governance:

# Use the underlying client directly (not recommended)
s3._underlying_client.list_buckets()

# Or configure excluded operations
session = GovernedBoto3Session(
    ascend_client=ascend,
    excluded_operations=["sts:GetCallerIdentity", "s3:HeadBucket"]
)

---

Exceptions

GovernanceError

Raised when an operation is denied by ASCEND policies.

class GovernanceError(Exception):
    operation: str          # AWS operation name
    service: str            # AWS service name
    reason: str             # Denial reason
    risk_score: int         # Risk score (0-100)
    risk_level: str         # Risk level
    policy_violations: List[str]  # Violated policies
    action_id: str          # ASCEND action ID

Example:

from ascend_boto3 import GovernanceError

try:
    s3.delete_bucket(Bucket="production-bucket")
except GovernanceError as e:
    print(f"Operation denied: {e.service}:{e.operation}")
    print(f"Reason: {e.reason}")
    print(f"Risk score: {e.risk_score}")
    print(f"Violations: {e.policy_violations}")

PendingApprovalError

Raised when an operation requires human approval.

class PendingApprovalError(Exception):
    operation: str          # AWS operation name
    service: str            # AWS service name
    approval_id: str        # Approval request ID
    approvers: List[str]    # Required approvers
    action_id: str          # ASCEND action ID
    expires_at: str         # Approval expiration

Example:

from ascend_boto3 import PendingApprovalError

try:
    dynamodb.delete_table(TableName="production-users")
except PendingApprovalError as e:
    print(f"Approval required: {e.approval_id}")
    print(f"Required approvers: {e.approvers}")
    print(f"Expires: {e.expires_at}")

    # Store for later checking
    save_pending_operation(e.approval_id, e.action_id)

ServiceUnavailableError

Raised when ASCEND service is unavailable (in fail-closed mode).

class ServiceUnavailableError(Exception):
    operation: str          # AWS operation name
    service: str            # AWS service name
    fail_mode: str          # Current fail mode
    original_error: Exception  # Underlying error

Example:

from ascend_boto3 import ServiceUnavailableError

try:
    s3.put_object(Bucket="bucket", Key="key", Body="data")
except ServiceUnavailableError as e:
    if e.fail_mode == "open":
        # Operation was allowed despite unavailability
        print("Warning: Governance bypassed")
    else:
        # Operation was blocked
        print("Error: Governance service unavailable")
        print(f"Original error: {e.original_error}")

---

Context Manager

Use context managers for transactional governance:

from ascend_boto3 import governed_operation

# Single operation context
async with governed_operation(session, "s3", "batch_upload") as ctx:
    for file in files:
        s3.put_object(Bucket="bucket", Key=file.name, Body=file.content)

    # All operations logged under single action
    print(f"Action ID: {ctx.action_id}")

---

Async Support

For async operations using aioboto3:

from ascend_boto3 import AsyncGovernedBoto3Session
import asyncio

async def main():
    session = AsyncGovernedBoto3Session(
        ascend_client=ascend,
        region_name="us-east-1"
    )

    async with session.client("s3") as s3:
        response = await s3.list_buckets()
        print(response["Buckets"])

asyncio.run(main())

---

Metrics Callback

Register a callback for real-time metrics:

def metrics_callback(
    operation: str,
    service: str,
    risk_level: str,
    allowed: bool,
    duration_ms: int,
    action_id: str
):
    """Called after each governed operation."""
    # Send to monitoring system
    datadog.gauge("ascend.boto3.latency", duration_ms, tags=[
        f"operation:{operation}",
        f"service:{service}",
        f"risk:{risk_level}",
        f"allowed:{allowed}"
    ])

session = GovernedBoto3Session(
    ascend_client=ascend,
    metrics_callback=metrics_callback
)

---

Configuration Examples

Production Configuration

session = GovernedBoto3Session(
    ascend_client=ascend,
    region_name="us-east-1",
    fail_mode="closed",              # Block on ASCEND failure
    log_all_operations=True,         # Full audit trail
    skip_read_operations=False,      # Govern everything
    custom_risk_mappings={
        "s3:DeleteObject": "critical",
        "ec2:TerminateInstances": "critical"
    }
)

Development Configuration

session = GovernedBoto3Session(
    ascend_client=ascend,
    region_name="us-west-2",
    fail_mode="open",                # Allow on ASCEND failure
    log_all_operations=False,        # Reduce logging
    skip_read_operations=True,       # Only govern writes
    excluded_operations=[
        "sts:GetCallerIdentity"
    ]
)

Testing Configuration

# Mock session for tests
class MockGovernedBoto3Session:
    def __init__(self, allow_all: bool = True):
        self.allow_all = allow_all
        self.calls = []

    def client(self, service_name: str) -> "MockServiceClient":
        return MockServiceClient(service_name, self)

class MockServiceClient:
    def __init__(self, service: str, session: MockGovernedBoto3Session):
        self.service = service
        self.session = session

    def __getattr__(self, operation: str):
        def mock_operation(**kwargs):
            self.session.calls.append({
                "service": self.service,
                "operation": operation,
                "kwargs": kwargs
            })

            if not self.session.allow_all:
                raise GovernanceError(
                    operation=operation,
                    service=self.service,
                    reason="Test denial"
                )

            return {"test": True}

        return mock_operation

---

Supported Services

The wrapper supports all boto3 services. Services with default risk mappings include:

• S3 - Object storage
• DynamoDB - NoSQL database
• EC2 - Compute instances
• Lambda - Serverless functions
• IAM - Identity management
• RDS - Relational databases
• SNS - Notifications
• SQS - Message queues
• Secrets Manager - Secret storage
• KMS - Key management
• CloudWatch - Monitoring

For services without default mappings, operations default to "medium" risk.

---

See Also

• Quickstart Guide
• Lambda Authorizer
• Python SDK Reference