make.py

import datetime
import json
import pathlib
import subprocess
import sys

import troposphere
from troposphere import awslambda, events, iam, secretsmanager

DEBUG = True
LAMBDAS_DIR = pathlib.Path("lambdas")
LAMBDA_MAX_DURATION_SEC = 60 * 15
LAMBDA_RUNTIME = "python3.9"
LAMBDA_ASSUME_ROLE_POLICY = {
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "",
            "Effect": "Allow",
            "Principal": {"Service": "lambda.amazonaws.com"},
            "Action": "sts:AssumeRole",
        },
    ],
}


def make_lambda_environment(**kwargs):
    return awslambda.Environment(
        Variables={
            "LOG_LEVEL": "DEBUG" if DEBUG else "INFO",
            "POWERTOOLS_LOGGER_LOG_EVENT": "1" if DEBUG else "0",
            **kwargs,
        }
    )


def get_metadata():
    """build-time metadata for inclusion in template"""
    return {
        "git_revision": subprocess.run(("git", "rev-parse", "HEAD"), capture_output=True, text=True).stdout.rstrip(),
        "git_repository": subprocess.run(
            ("git", "rev-parse", "--show-toplevel"), capture_output=True, text=True
        ).stdout.rstrip(),
        "make_time": str(datetime.datetime.now(datetime.timezone.utc)),
    }


def make_layer(cft: troposphere.Template):
    return awslambda.LayerVersion(
        "Layer",
        template=cft,
        Content=awslambda.Content(
            S3Bucket=troposphere.Sub("quilt-lambda-${AWS::Region}"),
            S3Key="benchling-packager/benchling-packager-layer.4bcb4369305e6dca4ec2cec50d2891ad138adfc1f3833293d32a999bd1295770.zip",  # noqa
        ),
    )


def make_template(*, metadata: dict) -> troposphere.Template:
    description = "Automatically create a dedicated Quilt package for every Benchling notebook"
    cft = troposphere.Template(Description=description)
    troposphere.Output(
        "TemplateBuildMetadata",
        template=cft,
        Description="Metadata generated by the Quilt build system.",
        Value=json.dumps(metadata),
    )

    bus_name = troposphere.Parameter(
        "BenchlingEventBusName",
        template=cft,
        Type="String",
        AllowedPattern=r"^aws\.partner(/[\.\-_A-Za-z0-9]+){2,}$",
        Description=(
            "Name of event bus where Benchling events are emitted, " + "e.g aws.partner/benchling.com/tenant/app-name"
        ),
    )
    benchling_tenant = troposphere.Parameter(
        "BenchlingTenant",
        template=cft,
        Type="String",
        AllowedPattern=r"^[^/]+$",
        Description="Benchling tenant name, i.e. $BenchlingTenant in " + "https://$BenchlingTenant.benchling.com",
    )
    benchling_client_id = troposphere.Parameter(
        "BenchlingClientId",
        template=cft,
        Type="String",
        MinLength=1,
        Description="Client ID of Benchling app",
    )
    quilt_domain = troposphere.Parameter(
        "QuiltWebHost",
        template=cft,
        Type="String",
        AllowedPattern=r"^[^/]+$",
        Description="Hostname for your Quilt catalog, e.g. quilt.your-company.com",
    )
    dst_bucket = troposphere.Parameter(
        "DestinationBucket",
        template=cft,
        Type="String",
        AllowedPattern=r"^[\.\-a-z0-9]{3,63}$",
        Description="The name of S3 bucket where packages will be created",
    )
    pkg_prefix = troposphere.Parameter(
        "PackageNamePrefix",
        template=cft,
        Type="String",
        Default="benchling/",
        AllowedPattern=r".+/.*$",
        Description=(
            "Prefix for package names i.e. package names will be"
            " $PackageNamePrefix$ExperimentDisplayID,"
            " must contain, but not start with '/'"
        ),
    )

    make_layer(cft)
    benchling_client_secret = secretsmanager.Secret(
        "BenchlingClientSecret",
        template=cft,
    )
    role = iam.Role(
        "LambdaRole",
        template=cft,
        AssumeRolePolicyDocument=LAMBDA_ASSUME_ROLE_POLICY,
        ManagedPolicyArns=[
            "arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole",
        ],
        Policies=[
            iam.Policy(
                PolicyName="root",
                PolicyDocument={
                    "Version": "2012-10-17",
                    "Statement": [
                        {
                            "Effect": "Allow",
                            "Action": "secretsmanager:GetSecretValue",
                            "Resource": benchling_client_secret.ref(),
                        },
                        {
                            "Effect": "Allow",
                            "Action": [
                                "s3:GetObject",
                                "s3:GetObjectTagging",
                                "s3:GetObjectVersion",
                                "s3:GetObjectVersionTagging",
                                "s3:ListBucket",
                                "s3:ListBucketVersions",
                                "s3:PutObject",
                                "s3:PutObjectTagging",
                            ],
                            "Resource": [
                                troposphere.Sub("arn:aws:s3:::${DestinationBucket}"),
                                troposphere.Sub("arn:aws:s3:::${DestinationBucket}/*"),
                            ],
                        },
                    ],
                },
            ),
        ],
    )
    function = awslambda.Function(
        "Lambda",
        template=cft,
        Role=role.get_att("Arn"),
        Runtime=LAMBDA_RUNTIME,
        Timeout=LAMBDA_MAX_DURATION_SEC,
        Layers=[troposphere.Ref("Layer")],
        Environment=make_lambda_environment(
            BENCHLING_TENANT=benchling_tenant.ref(),
            BENCHLING_CLIENT_ID=benchling_client_id.ref(),
            BENCHLING_CLIENT_SECRET_ARN=benchling_client_secret.ref(),
            DST_BUCKET=dst_bucket.ref(),
            PKG_PREFIX=pkg_prefix.ref(),
            QUILT_CATALOG_DOMAIN=quilt_domain.ref(),
        ),
        Handler="index.lambda_handler",
        Code=awslambda.Code(ZipFile=(LAMBDAS_DIR / "main.py").read_text()),
        ReservedConcurrentExecutions=1,  # FIXME
        MemorySize=512,
    )
    rule = events.Rule(
        "EventBusRule",
        template=cft,
        EventBusName=bus_name.ref(),
        EventPattern={
            "source": [bus_name.ref()],
            "detail-type": ["v2.entry.created", "v2.entry.updated.fields"],
        },
        Targets=[
            events.Target(
                Id=function.title,
                Arn=function.get_att("Arn"),
            )
        ],
    )
    awslambda.Permission(
        "LambdaPermission",
        template=cft,
        FunctionName=function.get_att("Arn"),
        Action="lambda:InvokeFunction",
        Principal="events.amazonaws.com",
        SourceArn=rule.get_att("Arn"),
    )

    return cft


if __name__ == "__main__":
    cft = make_template(
        metadata=get_metadata(),
    )
    # sort_keys=False is a lucky way to get the Description with the
    # copyright on top; it could break (but not catastrophically)
    sys.stdout.write(cft.to_yaml(sort_keys=False))