云上自动化创建账户实践

import boto3
import time
from botocore.exceptions import ClientError
def delete_default_vpc(currentregion):
    print("Default VPC deletion in progress in {}".format(currentregion))
    ec2_client = get_client('ec2', currentregion)
    vpc_response = ec2_client.describe_vpcs()
    default_vpcid = None
    for i in range(0,len(vpc_response['Vpcs'])):
        if((vpc_response['Vpcs'][i]['IsDefault']) == True):
            print("Default VPC is", vpc_response['Vpcs'][i]['VpcId'])
            default_vpcid = vpc_response['Vpcs'][i]['VpcId']
    if not default_vpcid:
        print("Default VPC was removed, skipping...")
        return

    subnet_response = ec2_client.describe_subnets()
    subnet_delete_response = []
    default_subnets = []
    for i in range(0,len(subnet_response['Subnets'])):
        if(subnet_response['Subnets'][i]['VpcId'] == default_vpcid):
            default_subnets.append(subnet_response['Subnets'][i]['SubnetId'])
    for i in range(0,len(default_subnets)):
        subnet_delete_response.append(ec2_client.delete_subnet(SubnetId=default_subnets[i],DryRun=False))
    print("Default Subnets" + currentregion + "Deleted.")
    igw_response = ec2_client.describe_internet_gateways()
    for i in range(0,len(igw_response['InternetGateways'])):
        for j in range(0,len(igw_response['InternetGateways'][i]['Attachments'])):
            if(igw_response['InternetGateways'][i]['Attachments'][j]['VpcId'] == default_vpcid):
                default_igw = igw_response['InternetGateways'][i]['InternetGatewayId']
    print(default_igw)
    detach_default_igw_response = ec2_client.detach_internet_gateway(InternetGatewayId=default_igw,VpcId=default_vpcid,DryRun=False)
    delete_internet_gateway_response = ec2_client.delete_internet_gateway(InternetGatewayId=default_igw)
    print("Default IGW " + currentregion + "Deleted.")
    time.sleep(10)
    delete_vpc_response = ec2_client.delete_vpc(VpcId=default_vpcid,DryRun=False)
    print("Deleted Default VPC in {}".format(currentregion))
    return delete_vpc_response
def get_client(service, region):
  client = boto3.client(service, region_name = region)
  return client
def lambda_handler(event, context):
    ec2_client = boto3.client('ec2')
    # delete default vpc in every region
    regions = []
    regions_response = ec2_client.describe_regions()

    for i in range(0,len(regions_response['Regions'])):
        regions.append(regions_response['Regions'][i]['RegionName'])
    for r in regions:
        try:
            delete_vpc_response = delete_default_vpc(r)
        except ClientError as e:
            print("An error occured while deleting Default VPC in {}. Error: {}".format(r,e))

{
  "version": "2.0",
  "app_name": "quickstart",
  "api_gateway_endpoint_type": "REGIONAL",
  "stages": {
    "dev": {
      "api_gateway_stage": "api"
    }
  }
}

from chalice import Chalice, AuthResponse
app = Chalice(app_name='demoauth1')
@app.authorizer()
def demo_auth(auth_request):
    token = auth_request.token
    # This is just for demo purposes as shown in the API Gateway docs.
    # Normally you'd call an oauth provider, validate the
    # jwt token, etc.
    # In this example, the token is treated as the status for demo
    # purposes.
    if token == 'allow':
        return AuthResponse(routes=['/'], principal_id='user')
    else:
        # By specifying an empty list of routes,
        # we're saying this user is not authorized
        # for any URLs, which will result in an
        # Unauthorized response.
        return AuthResponse(routes=[], principal_id='user')


@app.route('/', authorizer=demo_auth)
def index():
    return {'context': app.current_request.context}

@app.authorizer()
def jwt_auth(auth_request):
    token = auth_request.token
    keycloak_endpoint, keycloak_realm = get_keycloak_url_and_realm()
    cert_url = "{0}realms/{1}/protocol/openid-connect/certs".format(keycloak_endpoint,keycloak_realm)
    options = {"verify_signature": True, "verify_aud": False, "exp": True}
    principal_id = 'NoneAuthUser'
    try:
        jwks_client = PyJWKClient(cert_url)
        signing_key = jwks_client.get_signing_key_from_jwt(token)
        user_info = jwt.decode(token, signing_key.key, algorithms=["RS256"],options=options)
        # Add auth logic 
        if 'keycloak-test' in user_info['resource_access']:
                principal_id = user_info ['sub']
                context = {'user_role': user_role,'user_email': user_email}
                return AuthResponse(routes=['*'], principal_id=principal_id,context=context)
    except Exception as e:
        logger.exception("Received an exception")
    return AuthResponse(routes=[], principal_id=principal_id)

{
    "id":"s3",
    "checkitems":[
    {
    "identifier":"BlockPublicAcls",
    "items":"Block Public Acls - Enabled",
    "settings":"S3 Block Public Access"
    },
    {
    "identifier":"BlockPublicPolicy",
    "items":"Block Public Policy - Enabled",
    "settings":"S3 Block Public Access"
    },
    {
    "identifier":"IgnorePublicAcls",
    "items":"Ignore Public Acls - Enabled",
    "settings":"S3 Block Public Access"
    },
    {
    "identifier":"RestrictPublicBuckets",
    "items":"Restrict Public Buckets - Enabled",
    "settings":"S3 Block Public Access"
    }
    ]
}

{
 "id": "s3",
 "source": {
  "publicAccessBlockConfig": {
   "BlockPublicAcls": true,
   "BlockPublicPolicy": true,
   "IgnorePublicAcls": true,
   "RestrictPublicBuckets": true
  }
 }
}