AWS Lambda调用EC2实例私有REST API方法咨询（含堡垒机场景）

Got it, let's break down your problem and fix it step by step. You've already nailed getting the private IPs of EC2 instances attached to your ELB, but the next two hurdles are: calling the REST API properly in Lambda, and routing that traffic through your bastion host to reach the private subnet instances. Here's how to tackle both:

1. Replace subprocess.call with a native Python HTTP client #

Using subprocess to run curl is clunky in Lambda—you're relying on the runtime having curl installed, and it's hard to debug errors. Instead, use the requests library, which is the standard for Python HTTP calls and works seamlessly in Lambda.

2. Set up an SSH tunnel via your bastion host #

Since your EC2 instances only have private IPs, Lambda can't reach them directly (unless it's in the same VPC with proper security group rules). The solution is to create an SSH tunnel through your bastion host to forward traffic from Lambda to the private instances. We'll use the sshtunnel library to handle this in code.

Prep work first #

• Make sure your bastion host's security group allows SSH access (port 22) from Lambda's IP range (or from the VPC CIDR if Lambda is in the same VPC).
• Store your bastion's SSH private key in AWS Secrets Manager (never hardcode it!)—we'll fetch it securely in the Lambda function.

3. Full optimized code #

import boto3
import requests
from sshtunnel import SSHTunnelForwarder
from botocore.exceptions import ClientError

def get_bastion_private_key():
    # Fetch bastion SSH key from Secrets Manager (replace with your secret name)
    secrets_client = boto3.client('secretsmanager')
    try:
        response = secrets_client.get_secret_value(SecretId='bastion-ssh-private-key')
        return response['SecretString']
    except ClientError as e:
        print(f"Failed to fetch bastion key: {str(e)}")
        raise

def get_elb_instance_ips(elb_name):
    # Optimize: Initialize boto3 clients once instead of looping
    elb_client = boto3.client('elb')
    ec2_resource = boto3.resource('ec2')

    # Get target ELB details
    elb_response = elb_client.describe_load_balancers(LoadBalancerNames=[elb_name])
    if not elb_response['LoadBalancerDescriptions']:
        print(f"No ELB found with name: {elb_name}")
        return []

    # Extract instance IDs and fetch their private IPs in bulk
    instance_ids = [inst['InstanceId'] for inst in elb_response['LoadBalancerDescriptions'][0]['Instances']]
    instances = ec2_resource.instances.filter(InstanceIds=instance_ids)
    return [inst.private_ip_address for inst in instances]

def call_private_api_through_bastion(instance_ips, bastion_host, bastion_user, bastion_key):
    # Create SSH tunnel: forward local ports to each EC2 instance's HTTPS port (443)
    with SSHTunnelForwarder(
        (bastion_host, 22),
        ssh_username=bastion_user,
        ssh_private_key=bastion_key,
        remote_bind_addresses=[(ip, 443) for ip in instance_ips]
    ) as tunnel:
        # Loop through each instance and call the API via the forwarded local port
        for idx, ip in enumerate(instance_ips):
            local_port = tunnel.local_bind_ports[idx]
            api_url = f"https://localhost:{local_port}/voice/diag"
            
            try:
                # Skip cert verification if your private instances use self-signed certs
                response = requests.get(api_url, headers={'cache-control': 'no-cache'}, verify=False)
                print(f"Success calling {ip}: Status {response.status_code}")
                print(f"Response: {response.text[:200]}...")  # Print first 200 chars to avoid clutter
            except requests.exceptions.RequestException as e:
                print(f"Failed calling {ip}: {str(e)}")

def lambda_handler(event, context):
    print("Received DynamoDB stream event:", event)
    
    # Pull configs from Lambda environment variables (better than hardcoding!)
    elb_name = 'xxxx-xxx-xxx-xxx-2-BlueELB'
    bastion_public_ip = 'your-bastion-public-ip'
    bastion_ssh_user = 'ec2-user'  # Common default for Amazon Linux, adjust for your OS

    # Get bastion key and instance IPs
    bastion_key = get_bastion_private_key()
    instance_ips = get_elb_instance_ips(elb_name)
    
    if not instance_ips:
        return "No instances found attached to the ELB"
    
    # Run API calls through bastion
    call_private_api_through_bastion(instance_ips, bastion_public_ip, bastion_ssh_user, bastion_key)
    
    return "API calls completed successfully"

4. Lambda Environment Setup #

• Add dependencies as a Lambda Layer: Lambda doesn't have requests or sshtunnel pre-installed. Create a layer with these libraries:

  # Create a temp directory for the layer
  mkdir -p python/lib/python3.6/site-packages
  # Install dependencies matching Lambda's Python 3.6 runtime
  pip install requests sshtunnel -t python/lib/python3.6/site-packages
  # Zip the layer
  zip -r lambda-api-deps.zip python/
  

  Upload this zip to Lambda as a new layer, then attach it to your function.
• IAM Permissions: Add these to your Lambda's IAM role:
  • secretsmanager:GetSecretValue (to fetch the bastion key)
  • VPC permissions (if Lambda is in a VPC): ec2:CreateNetworkInterface, ec2:DescribeNetworkInterfaces, ec2:DeleteNetworkInterface
• VPC Configuration: If your bastion is in a VPC, put Lambda in the same VPC's private subnets for more secure access (no need to route through the public internet).

5. Quick Tips #

• Move all hardcoded values (ELB name, bastion IP) to Lambda environment variables so you can update them without redeploying code.
• If your API uses HTTP instead of HTTPS, change the port to 80 and remove verify=False.
• Add retries for API calls using requests.adapters.HTTPAdapter if you expect occasional network blips.

内容的提问来源于stack exchange，提问作者user93726