SSM Document CDK

This library provides a code-based utility for implementing SSM Documents. The SSM Document objects can be used to print YAML/JSON documents and to mimic document processing locally.

This library abstracts SSM Documents at a high level, with each step as well as the document itself being objects. The properties needed to build these objects correlate to the settings that apply to them, making them simple to make. This library can be used to test your document locally before deploying it to SSM.

Since the library is written in JSII, it can be exported to other languages that support JSII (Java, Python).

This is what you'd use if you wanted to:

1. The ability to test without deploying resources or executing an actual SSM on AWS.
2. Reusability of steps between documents by reusing existing items
3. Create logical higher-level groupings of reusable groups of steps ("Patterns")
4. Simple to use interface for writing documents
5. Import existing documents from a file (or string) and mimic them locally to test them.

Usage

Document Creation

Typescript usage (Execute AWS API Step)... The below creates the AutomationDocument in an AWS CDK stack.

# Example automatically generated from non-compiling source. May contain errors.
from ..automation_document import AutomationDocument


class HelloWorld(Stack):
    def __init__(self, app, id):
        super().__init__(app, id)

        # Create AutomationDocument
        my_doc = AutomationDocument(self, "MyDoc",
            document_format=DocumentFormat.JSON,
            document_name="MyDoc",
            doc_inputs=[Input.of_type_string("MyInput", default_value="a")]
        )

        # Define your steps...
        my_doc.add_step(PauseStep(self, "MyPauseStep",
            name="MyPauseStep",
            explicit_next_step=StepRef.from_name("step1")
        ))

        my_doc.add_step(ExecuteScriptStep(self, "MyExecuteStep",
            name="step1",
            language=ScriptLanguage.python(PythonVersion.VERSION_3_11, "my_func"),
            code=ScriptCode.from_file(resolve("test/test_file.py")),
            # OR ScriptCode.inline("def my_func(args, context):\n  return {'MyReturn': args['MyInput'] + '-suffix'}\n"),
            outputs=[{
                "output_type": DataTypeEnum.STRING,
                "name": "MyFuncOut",
                "selector": "$.Payload.MyReturn"
            }],
            on_failure=OnFailure.abort(),
            input_payload={"MyInput": StringVariable.of("MyInput")}
        ))

Document JSON/YAML Export as YAML/JSON

You can deploy the above document using CDK. To print the above document object as a JSON (or YAML), do the following:

# Example automatically generated from non-compiling source. May contain errors.
my_doc_json = my_doc.print()

Document Simulation

To run the document object in simulation mode, use the below. Simulation mode does NOT hit the SSM API, rather it mimics the execution that will happen in an SSM execution. The run happens locally and allows you to mock the calls to external services (AWS APIs for example) or to invoke those services using your local credentials.

# Example automatically generated from non-compiling source. May contain errors.
from ..simulation import Simulation


my_doc_json = Simulation.of_automation(my_doc).simulate(MyInput="FooBar")

Command Documents

Below is an example of how to use the library to create Command documents. Simulation for command documents is not yet supported for all command plugins. You can use a Docker image/container as a playground for testing the Command document execution for the supported plugins.

In this example there is a complete CDK stack. Notice that the CommandDocument is saved as a field so that it can be tested from the test code.

# Example automatically generated from non-compiling source. May contain errors.
class HelloCdkStack(Stack):
    def __init__(self, scope, id, props=None):
        super().__init__(scope, id, props)
        self.my_command_doc = CommandDocument(self, "MyCommandDoc",
            doc_inputs=[Input.of_type_string("FirstCommand", default_value="a")]
        )
        run_script_step = RunShellScriptStep(self, "MyShellScript",
            run_command=[
                StringVariable.of("FirstCommand"),
                HardCodedString.of("mkdir asdf")
            ]
        )
        self.my_command_doc.add_step(run_script_step)

Below is an example of how you would run a simulation against the above CommandDocument.

Currently, bash must be available in the container or the executions against the docker will not succeed.

# Example automatically generated from non-compiling source. May contain errors.
test("Test command doc", () => {
      const app = new cdk.App();
      const stack = new HelloCdk.HelloCdkStack(app, 'MyTestStack');
      // 1. $ docker pull amazonlinux
      // 2. $ docker run -di amazonlinux
      const simulation = Simulation.ofCommand(stack.myCommandDoc, {
        simulationPlatform: Platform.LINUX,
        environment: DockerEnvironment.fromContainer('MY_CONTAINER_ID')
      });
      simulation.simulate({FirstCommand: 'mkdir foobar'})
      // 3. The document should run the first command (create 'foobar') and create file 'asdf'
      // 4. $ docker exec -it <container name> bash
      // 5. Ensure that 'asdf' and 'foobar' were written to /tmp
    })

Patterns (High-Level Constructs)

In typical CDK style, you can assemble often used groups of steps into higher level Constructs.

Consider if you typically create AutomationDocuments that start with logging the time and end with logging the total time taken. You can create a high-level Automation Document and extend that when you implement an Automation.

See the TimedDocument class to see such implementation.

Or consider the case of multiple steps that are always run together such as rebooting and instance and waiting for it to be active.

The below example is copied from the RebootInstanceAndWait class:

# Example automatically generated from non-compiling source. May contain errors.
class RebootInstanceAndWait(CompositeAutomationStep):

    def __init__(self, scope, id, instance_id):
        super().__init__(scope, id)
        self.reboot = AwsApiStep(self, "RebootInstances",
            service=AwsService.EC2,
            pascal_case_api="RebootInstances",
            api_params={"InstanceIds": [instance_id]},
            outputs=[]
        )
        self.describe = WaitForResourceStep(self, "DescribeInstances",
            service=AwsService.EC2,
            pascal_case_api="DescribeInstances",
            api_params={"InstanceIds": [instance_id]},
            selector="$.Reservations[0].Instances[0].State.Name",
            desired_values=["running"]
        )

    def add_to_document(self, doc):
        doc.add_step(self.reboot)
        doc.add_step(self.describe)

Now, you can use RebootInstanceAndWait as a step in a document and the child steps will be included.

Existing Documents

Do you have an existing document that you want to convert to code and/or test locally using the simulation?

Import Existing Document

Here is an example of how you can import an existing document and then simulate it locally with mocked AWS resources:

# Example automatically generated from non-compiling source. May contain errors.
# Initialize Mocks
sleeper = MockSleep()
aws_invoker = MockAwsInvoker()
aws_invoker.when_then({"aws_api": "listBuckets", "aws_params": {}, "service": AwsService.S3}, Owner={"ID": "BUCKET_ID"})

# ======> Create document from file <=======
stack = Stack()
my_automation_doc = StringDocument.from_file(stack, "MyAutomationDoc", "test/myAutomation.json")

# Execute simulation
sim_output = Simulation.of_automation(my_automation_doc,
    sleep_hook=sleeper,
    aws_invoker=aws_invoker
).simulate()

# Assert simulation result
assert.deep_equal(aws_invoker.previous_invocations, [aws_api="listBuckets", aws_params={}, service=AwsService.S3
])
assert.deep_equal(sleeper.sleep_milli_invocations, [3000])
assert.deep_equal(sim_output.outputs["simulationSteps"], ["MySleep", "GetBucketId"])

Import Existing Steps

You can also grab a string step (or steps) and import them as CDK step constructs. This can be used to convert existing documents into CDK with each step defined separately. Doing so will allow you do modify steps and reuse them in other documents.

Here's a simple example of a sleep step copy and pasted from its original yaml:

# Example automatically generated from non-compiling source. May contain errors.
StringStep.from_yaml(self, """
        name: sleep
        action: aws:sleep
        inputs:
          Duration: PT0M
    """)

The above will return the CDK construct SleepStep.

Incident Manager

This library provides L2 constructs for IncidentResponse as follows:

# Example automatically generated from non-compiling source. May contain errors.
IncidentResponse(self, "MyIncidentResponsePlan",
    incident_template=IncidentTemplate.critical("EC2 Instance Utilization Impacted",
        summary="EC2 Instance Impacted"
    ),
    actions=[
        IncidentResponseAction.ssm_automation(my_automation_doc, ec2_cw_alarm_role,
            parameters={
                "IncidentRecordArn": StringVariable.of("INCIDENT_RECORD_ARN"),
                "InvolvedResources": StringVariable.of("INVOLVED_RESOURCES"),
                "AutomationAssumeRole": HardCodedString.of(ec2_cw_alarm_role.role_arn)
            }
        )
    ]
)

Notice how the myAutomationDoc is specified which is a reference to an AutomationDocument created using this library.

What is Planned?

This library currently contains AutomationDocument and CommandDocument steps. Simulation for AutomationDocuments is fully supported. Simulation for CommandDocuments is limited.

Stay tuned!

Related Projects

• https://github.com/udondan/cdk-ssm-document

Security

See CONTRIBUTING for more information.

License

This project is licensed under the Apache-2.0 License.