Skip to content

Commit c143659

Browse files
mikedorfmanpre-commit-ci[bot]
mikedorfman
and
pre-commit-ci[bot]
authored
mikedorfman/CUMULUS-4542 (#4258)
* First pass at an aws-api-proxy task * Added a readme Updated schema generators * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci * Updated Changelog * Updated old granule-invalidator references * Added partial failure compatibility * Added package_uv to bin root from a different feature branch * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci * Updates based on PR feedback * Updated ajv from 6.12.3 to 8.18.0 to test dependency vulnerability fix * Updated readonly->readOnly in schemas after strict-mode check * Updates per PR feedback * Reverting config validation to use a static list of literals with a bit comment disclaimer * Fixed broken output validation * Reverted ajv version bump --------- Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>
1 parent 6fada0a commit c143659

17 files changed

Lines changed: 2118 additions & 0 deletions

File tree

‎CHANGELOG.md‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -32,6 +32,10 @@ The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/).
3232
- Added python code for CnmResponse task adapted from https://github.com/podaac/cumulus-cnm-response-task
3333
- **CUMULUS-4395**
3434
- Added supporting Terraform for the CnmResponse task that allows it to be included in the Cumulus terraform zipfile and deployed with Cumulus.
35+
- **CUMULUS-4498**
36+
- Added `states:StartExecution` action to the `<prefix>-steprole` IAM role.
37+
- **CUMULUS-4542**
38+
- Created the `aws-api-proxy` coreified task, which provides the functionality to post a list of CNM messages to a specified SNS topic.
3539
- **CUMULUS-4517**
3640
- Added the `@cumulus/db/s3search` module to enable Cumulus record search via S3-backed tables.
3741
The S3Search subclasses inherit from search/BaseSearch, allowing them to reuse existing query

‎tasks/aws-api-proxy/README.md‎

Lines changed: 147 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,147 @@
1+
2+
# @cumulus/aws-api-proxy-task
3+
4+
This task provides a Cumulus Message Adapter (CMA) wrapped proxy that invokes a
5+
pre-approved AWS API action via boto3. It is intended for use within a Cumulus
6+
workflow and enforces guardrails through schema validation and IAM permissions.
7+
Currently, the schema allows the SNS `publish` action, with support for multiple
8+
invocations in a single task execution via `iterate_by`.
9+
10+
## Usage
11+
12+
This lambda takes the following input and config objects, derived from workflow
13+
configuration using the
14+
[Cumulus Message Adapter](https://github.com/nasa/cumulus-message-adapter/blob/master/CONTRACT.md).
15+
The output from the task follows the CMA contract and provides the information
16+
detailed below.
17+
18+
### Configuration
19+
20+
| field name | type | default | required | values | description
21+
| ---------- | ---- | ------- | -------- | ------ | -----------
22+
| service | string | N/A | yes | `sns` | AWS service to invoke
23+
| action | string | N/A | yes | `publish` | Service action to invoke
24+
| parameters | object | N/A | yes | N/A | Parameters passed to the AWS SDK action
25+
| parameters.TopicArn | string | N/A | yes | N/A | SNS Topic ARN to publish to
26+
| parameters.Message | string | N/A | yes | N/A | Message string to publish
27+
| iterate_by | string | N/A | no | N/A | Name of a `parameters` field containing a list; each value produces a separate call
28+
| parameter_filters | array | [] | no | N/A | List of transformations to apply to parameter fields before invoking the API
29+
| parameter_filters[].name | string | N/A | yes | `json.dumps` | Filter to apply
30+
| parameter_filters[].field | string | N/A | yes | N/A | Parameter field name to transform
31+
32+
The following sample configuration publishes two messages to two topics by
33+
iterating over a list of ARNs:
34+
35+
```json
36+
{
37+
"service": "sns",
38+
"action": "publish",
39+
"iterate_by": "TopicArn",
40+
"parameters": {
41+
"TopicArn": [
42+
"arn:aws:sns:us-east-1:123456789012:ExampleTopicA",
43+
"arn:aws:sns:us-east-1:123456789012:ExampleTopicB"
44+
],
45+
"Message": "Message for topic"
46+
}
47+
}
48+
```
49+
50+
When `iterate_by` is set, the named field in `parameters` must be a list. The
51+
task will create one API call per list entry by substituting each value into the
52+
specified field. If `parameter_filters` are provided, each filter is applied to
53+
the named field across all generated parameter sets. The only supported filter
54+
is `json.dumps`, which serializes the field value as JSON.
55+
56+
### Input
57+
58+
The input payload for this task is expected to be empty. All invocation details
59+
are passed via the task configuration.
60+
61+
### Output
62+
63+
The task returns a list of boto3 responses (one per entry generated by
64+
`iterate_by`).
65+
Each response includes the `MessageId` for the SNS publish request. Example:
66+
67+
```json
68+
{
69+
"result_list": [
70+
{
71+
"MessageId": "11111111-1111-1111-1111-111111111111"
72+
},
73+
{
74+
"MessageId": "22222222-2222-2222-2222-222222222222"
75+
}
76+
]
77+
}
78+
```
79+
80+
### Example workflow configuration
81+
82+
This task can be used to send workflow notifications via SNS in a controlled
83+
and audited way. A sample task state could look like:
84+
85+
```json
86+
{
87+
"Publish Workflow Notification": {
88+
"Type": "Task",
89+
"Resource": "${aws_api_proxy_arn}",
90+
"Parameters": {
91+
"cma": {
92+
"event.$": "$",
93+
"task_config": {
94+
"service": "sns",
95+
"action": "publish",
96+
"parameter_filters": [
97+
{
98+
"name": "json.dumps",
99+
"field": "Message"
100+
}
101+
],
102+
"parameters": {
103+
"TopicArn": "${sns_publish_arn}",
104+
"Message.$": "$.meta.cnm"
105+
}
106+
}
107+
}
108+
},
109+
"ResultPath": null,
110+
"End": true
111+
}
112+
}
113+
```
114+
115+
## Architecture
116+
117+
```mermaid
118+
graph TB
119+
A["AWS Lambda Handler<br/>lambda_handler"] -->|calls| B["Lambda Adapter<br/>lambda_adapter"]
120+
B -->|invokes boto3 client| C["AWS SDK (boto3)"]
121+
C -->|executes action| D["AWS Service API<br/>(e.g., SNS publish)"]
122+
D -->|responses| E["CMA Output<br/>result_list"]
123+
```
124+
125+
### Internal Dependencies
126+
127+
This task relies on the Cumulus Message Adapter and requires an IAM role that
128+
permits the configured AWS service action (for example, `sns:Publish` to specific
129+
topics). Guardrails are enforced by schema validation and IAM permissions.
130+
131+
### External Dependencies
132+
133+
- boto3
134+
- https://github.com/nasa/cumulus-message-adapter
135+
- https://github.com/nasa/cumulus-message-adapter-python
136+
137+
## Contributing
138+
139+
To make a contribution, please [see our Cumulus contributing guidelines](https://github.com/nasa/cumulus/blob/master/CONTRIBUTING.md)
140+
and our documentation on [adding a task](https://nasa.github.io/cumulus/docs/adding-a-task)
141+
142+
## About Cumulus
143+
144+
Cumulus is a cloud-based data ingest, archive, distribution and management
145+
prototype for NASA's future Earth science data streams.
146+
147+
[Cumulus Documentation](https://nasa.github.io/cumulus)

‎tasks/aws-api-proxy/package.json‎

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,31 @@
1+
{
2+
"name": "@cumulus/aws-api-proxy-task",
3+
"private": true,
4+
"version": "21.2.0",
5+
"description": "AWS API proxy task",
6+
"main": "index.js",
7+
"homepage": "https://github.com/nasa/cumulus/tree/master/tasks/aws-api-proxy",
8+
"repository": {
9+
"type": "git",
10+
"url": "https://github.com/nasa/cumulus"
11+
},
12+
"scripts": {
13+
"test": "uv run pytest -q",
14+
"lint": "uv run ruff check . && uv run ruff format --diff . && uv run mypy src",
15+
"clean": "rm -rf dist && rm -rf .venv && mkdir dist",
16+
"build": "uv sync",
17+
"prepare": "npm run build",
18+
"package": "npm run clean && ../../bin/package_uv.sh 3.12 ${PWD}",
19+
"install-python-deps": "npm run build"
20+
},
21+
"publishConfig": {
22+
"access": "restricted"
23+
},
24+
"nyc": {
25+
"exclude": [
26+
"tests"
27+
]
28+
},
29+
"author": "Cumulus Authors",
30+
"license": "Apache-2.0"
31+
}

‎tasks/aws-api-proxy/pyproject.toml‎

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,32 @@
1+
[project]
2+
name = "aws-api-proxy"
3+
version = "0.1.0"
4+
description = "Makes specified AWS API calls wrapped the Cumulus Message Adapter"
5+
readme = "README.md"
6+
requires-python = ">=3.12"
7+
dependencies = [
8+
"cumulus-message-adapter>=2.0.5",
9+
"cumulus-message-adapter-python>=2.4.0",
10+
]
11+
12+
[project.urls]
13+
homepage = "https://github.com/nasa/cumulus/blob/master/tasks/aws-api-proxy"
14+
documentation = "https://github.com/nasa/cumulus/blob/master/tasks/aws-api-proxy/README.md"
15+
repository = "https://github.com/nasa/cumulus.git"
16+
17+
[dependency-groups]
18+
dev = [
19+
"mypy>=1.0.0",
20+
"ruff~=0.14.0",
21+
"pytest>=9.0.2",
22+
"pytest-cov>=7.0.0",
23+
"moto~=5.1",
24+
"pydantic>=2.10.4",
25+
]
26+
27+
[tool.pytest.ini_options]
28+
pythonpath = ["src"]
29+
30+
[project.scripts]
31+
generate-config-schema = "aws_api_proxy.schemas.config_schema:main"
32+
generate-output-schema = "aws_api_proxy.schemas.output_schema:main"

‎tasks/aws-api-proxy/src/aws_api_proxy/__init__.py‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,14 @@
1+
"""AWS API proxy task for Cumulus.
2+
3+
This module provides a CMA-wrapped way to call a specified boto3 client within AWS
4+
lambda. This may be called once or multiple times against a provided list. Guardrails
5+
are provided via configuration validation which specifies a predefined set of allowed
6+
services and actions in addition to a dedicated IAM role for this lambda.
7+
"""
8+
9+
import logging
10+
import os
11+
12+
from cumulus_logger import CumulusLogger
13+
14+
LOGGER = CumulusLogger(__name__, level=int(os.environ.get("LOGLEVEL", logging.DEBUG)))

‎tasks/aws-api-proxy/src/aws_api_proxy/aws_api_proxy.py‎

Lines changed: 89 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,89 @@
1+
"""AWS API proxy task for Cumulus.
2+
3+
This module provides a CMA-wrapped way to call a specified boto3 client within AWS
4+
lambda. This may be called once or multiple times against a provided list. Guardrails
5+
are provided via configuration validation which specifies a predefined set of allowed
6+
services and actions in addition to a dedicated IAM role for this lambda.
7+
"""
8+
9+
import asyncio
10+
import json
11+
from typing import Any
12+
13+
import boto3
14+
15+
from . import LOGGER
16+
17+
PARAMETER_FILTERS = {
18+
"json.dumps": json.dumps,
19+
}
20+
21+
22+
async def run_with_limit(method, parameters_list, max_concurrency=5):
23+
"""Run the given method with the provided parameters, limiting concurrency."""
24+
semaphore = asyncio.Semaphore(max_concurrency)
25+
26+
async def worker(parameters):
27+
async with semaphore:
28+
# Since boto3 is not async, run it in a thread
29+
# There is also aiobotocore which allows doing real async for AWS API calls.
30+
# However, I'm avoiding additional third-party libraries that aren't
31+
# officially supported/endorsed by AWS. If this were more complicated,
32+
# moving to aibotocore might make sense, but for now it's fairly
33+
# straight-forward to do it this way.
34+
return await asyncio.to_thread(method, **parameters)
35+
36+
results = await asyncio.gather(
37+
*(worker(parameters) for parameters in parameters_list), return_exceptions=True
38+
)
39+
return results
40+
41+
42+
def lambda_adapter(event: dict, _: Any) -> dict[str, Any]:
43+
"""Handle AWS API Proxy requests."""
44+
config = event.get("config", {})
45+
service = config.get("service")
46+
action = config.get("action")
47+
parameters = config.get("parameters")
48+
iterate_by = config.get("iterate_by")
49+
parameter_filters = config.get("parameter_filters", [])
50+
if iterate_by:
51+
iterate_by_parameters = parameters.get(iterate_by)
52+
if not isinstance(iterate_by_parameters, list):
53+
raise ValueError(
54+
f"iterate_by field '{iterate_by}' must be a list in parameters."
55+
)
56+
parameters_list = [
57+
{**parameters, iterate_by: value} for value in parameters[iterate_by]
58+
]
59+
else:
60+
parameters_list = [parameters]
61+
62+
for parameter_filter in parameter_filters:
63+
parameter_filter_name = parameter_filter.get("name")
64+
parameter_filter_field = parameter_filter.get("field")
65+
parameter_filter_func = PARAMETER_FILTERS.get(
66+
parameter_filter_name, lambda x: x
67+
)
68+
parameters_list = [
69+
{
70+
k: parameter_filter_func(v) if k == parameter_filter_field else v
71+
for k, v in parameters.items()
72+
}
73+
for parameters in parameters_list
74+
]
75+
76+
LOGGER.info(
77+
f"Received request to call AWS service {service} "
78+
f"with action {action} and parameters {parameters_list}"
79+
)
80+
client = boto3.client(service)
81+
method = getattr(client, action)
82+
83+
responses = asyncio.run(run_with_limit(method, parameters_list))
84+
LOGGER.info(
85+
f"Received response from AWS service {service} "
86+
f"with action {action}: {responses}"
87+
)
88+
89+
return {"result_list": responses}

0 commit comments

Comments
 (0)