diff --git a/services/api/api/iron-proxy.base.yaml b/services/api/api/iron-proxy.base.yaml index 00a8eb382..be9b8edb7 100644 --- a/services/api/api/iron-proxy.base.yaml +++ b/services/api/api/iron-proxy.base.yaml @@ -78,6 +78,15 @@ transforms: - "x-as-user-email" - "/^x-codex-.*$/" - "/^x-openai-.*$/" + # AWS SigV4 request headers (x-amz-target, x-amz-date, + # x-amz-content-sha256, x-amz-security-token) for the aws_auth transform. + - "/^x-amz-.*$/" + # AWS SDK headers that the SDK signer folds into the SigV4 signed-headers + # set (amz-sdk-request, amz-sdk-invocation-id, and x-amzn-query-mode for + # CloudWatch's query-JSON protocol). aws_auth signs them, so they must + # survive egress filtering or AWS rejects with InvalidSignatureException. + - "/^amz-sdk-.*$/" + - "/^x-amzn-.*$/" - "/^x-[a-z0-9-]*(api-key|apikey|secret|token|auth|key)$/" log: diff --git a/services/api/api/proxy_config.py b/services/api/api/proxy_config.py index adfe9307f..2e5a9032c 100644 --- a/services/api/api/proxy_config.py +++ b/services/api/api/proxy_config.py @@ -12,6 +12,9 @@ minting OAuth2 access tokens for the declared grant. - ``hmac_sign`` transforms — one per unique ``HmacSignSecret`` signing scheme, HMAC-signing each request and injecting the configured headers. +- ``aws_auth`` transforms — one per unique ``AwsAuthSecret`` credential set, + re-signing AWS SigV4 requests (the tool signs with placeholder credentials) + with real keys resolved from the secret source. - top-level ``postgres:`` — one listener per ``PgDsnSecret`` on sequential ports starting at 5432, ordered by name. """ @@ -26,6 +29,7 @@ import yaml from api.tool_manager import ( + AwsAuthSecret, BrokeredTokenSecret, GcpAuthSecret, GitHubAppTokenSecret, @@ -53,7 +57,7 @@ def load_base_config() -> str: PG_LISTEN_PORT_BASE = 5432 _MANAGED_TRANSFORMS: frozenset[str] = frozenset( - {"secrets", "gcp_auth", "oauth_token", "hmac_sign"} + {"secrets", "gcp_auth", "oauth_token", "hmac_sign", "aws_auth"} ) # Iron-proxy ``source`` schema for resolving secret values. ``env`` reads the @@ -459,6 +463,51 @@ def _build_hmac_sign_transforms( return transforms +def _build_aws_auth_transforms( + secrets: list[SecretDef], +) -> list[dict[str, Any]]: + """``aws_auth`` transforms: one per unique credential set + scope. + + Entries that share the same credential refs, session token, and + allowed regions/services are merged — their host rules are unioned so the + same signing config covers every upstream that opted in. iron-proxy re-signs + requests the tool signed with placeholder credentials, drawing the real keys + from the resolved secret sources. + """ + by_scheme: dict[ + tuple[str, str, str | None, tuple[str, ...], tuple[str, ...]], + set[str], + ] = {} + for secret in secrets: + if not isinstance(secret, AwsAuthSecret): + continue + key = ( + secret.access_key_id_ref, + secret.secret_access_key_ref, + secret.session_token_ref, + secret.allowed_regions, + secret.allowed_services, + ) + by_scheme.setdefault(key, set()).update(secret.hosts) + + transforms: list[dict[str, Any]] = [] + for key in sorted(by_scheme, key=lambda k: (k[0], k[1])): + access_key_id_ref, secret_access_key_ref, session_token_ref, regions, services = key + config: dict[str, Any] = { + "access_key_id": _build_source(access_key_id_ref), + "secret_access_key": _build_source(secret_access_key_ref), + } + if session_token_ref: + config["session_token"] = _build_source(session_token_ref) + if regions: + config["allowed_regions"] = list(regions) + if services: + config["allowed_services"] = list(services) + config["rules"] = [{"host": h} for h in sorted(by_scheme[key])] + transforms.append({"name": "aws_auth", "config": config}) + return transforms + + def _build_postgres_listeners( secrets: list[SecretDef], pg_listen_ports: dict[str, int], @@ -524,6 +573,7 @@ def render_proxy_yaml( if oauth_token is not None: new_transforms.append(oauth_token) new_transforms.extend(_build_hmac_sign_transforms(secrets)) + new_transforms.extend(_build_aws_auth_transforms(secrets)) if new_transforms: for index, transform in enumerate(transforms): if (transform or {}).get("name") == "header_allowlist": diff --git a/services/api/api/sandbox/kubernetes.py b/services/api/api/sandbox/kubernetes.py index 6d4b51042..1c65f6527 100644 --- a/services/api/api/sandbox/kubernetes.py +++ b/services/api/api/sandbox/kubernetes.py @@ -522,6 +522,24 @@ def _build_tool_server_container( env.append({"name": name, "value": dsn}) _apply_tool_server_extra_env(env, no_proxy) + # AWS region for the cloudwatch tool (non-secret). The tool signs with + # placeholder credentials and iron-proxy re-signs with the real keys, so no + # AWS credentials belong in this process — only the region, which boto3 + # needs to pick the endpoint host and signing scope. Optional: the tool + # defaults to us-east-1 when unset. + env.append( + { + "name": "AWS_REGION", + "valueFrom": { + "secretKeyRef": { + "name": secret_name, + "key": _secret_env_key("AWS_REGION"), + "optional": True, + } + }, + } + ) + volume_mounts: list[dict[str, Any]] = [ { "name": "firewall-ca", diff --git a/services/api/api/tool_manager.py b/services/api/api/tool_manager.py index 81b7d6f33..d86fc027a 100644 --- a/services/api/api/tool_manager.py +++ b/services/api/api/tool_manager.py @@ -307,6 +307,31 @@ class HmacSignSecret: allow_chunked_body: bool = False +@dataclass(frozen=True) +class AwsAuthSecret: + """AWS SigV4 re-signing handled by iron-proxy's ``aws_auth`` transform. + + The tool's AWS SDK signs each request with throwaway *placeholder* + credentials; iron-proxy reads the region and service from the inbound + signature's credential scope, strips that signature, and re-signs with the + real credentials resolved from ``access_key_id``/``secret_access_key`` (and + optional ``session_token``). The real keys never reach the sandbox — this is + the SigV4 analogue of the ``secrets`` transform's placeholder swap. + + ``allowed_services``/``allowed_regions`` scope which AWS services/regions the + proxy will sign for; ``hosts`` becomes the iron-proxy ``rules``. Credential + refs resolve like every other secret (env var or 1Password item). + """ + + name: str + hosts: tuple[str, ...] + access_key_id_ref: str + secret_access_key_ref: str + session_token_ref: str | None = None + allowed_regions: tuple[str, ...] = () + allowed_services: tuple[str, ...] = () + + SecretDef = ( HttpSecret | GcpAuthSecret @@ -315,6 +340,7 @@ class HmacSignSecret: | BrokeredTokenSecret | GitHubAppTokenSecret | HmacSignSecret + | AwsAuthSecret ) @@ -1040,6 +1066,59 @@ def _parse_secret(entry: Any, *, default_hosts: tuple[str, ...] = ()) -> SecretD timestamp_format=timestamp_format, allow_chunked_body=allow_chunked_body, ) + if secret_type == "aws_auth": + hosts = entry.get("hosts", []) + if ( + not isinstance(hosts, list) + or not hosts + or not all(isinstance(h, str) and h for h in hosts) + ): + raise ValueError( + f"aws_auth entry {name!r} 'hosts' must be a non-empty array " + f"of non-empty strings" + ) + access_key_id_ref = entry.get("access_key_id") + if not isinstance(access_key_id_ref, str) or not access_key_id_ref: + raise ValueError( + f"aws_auth entry {name!r} requires a non-empty 'access_key_id'" + ) + secret_access_key_ref = entry.get("secret_access_key") + if not isinstance(secret_access_key_ref, str) or not secret_access_key_ref: + raise ValueError( + f"aws_auth entry {name!r} requires a non-empty 'secret_access_key'" + ) + session_token_ref = entry.get("session_token") + if session_token_ref is not None and ( + not isinstance(session_token_ref, str) or not session_token_ref + ): + raise ValueError( + f"aws_auth entry {name!r} 'session_token' must be a non-empty string" + ) + allowed_regions = entry.get("allowed_regions", []) + if not isinstance(allowed_regions, list) or not all( + isinstance(r, str) and r for r in allowed_regions + ): + raise ValueError( + f"aws_auth entry {name!r} 'allowed_regions' must be an array of " + f"non-empty strings" + ) + allowed_services = entry.get("allowed_services", []) + if not isinstance(allowed_services, list) or not all( + isinstance(s, str) and s for s in allowed_services + ): + raise ValueError( + f"aws_auth entry {name!r} 'allowed_services' must be an array of " + f"non-empty strings" + ) + return AwsAuthSecret( + name=name, + hosts=tuple(hosts), + access_key_id_ref=access_key_id_ref, + secret_access_key_ref=secret_access_key_ref, + session_token_ref=session_token_ref, + allowed_regions=tuple(allowed_regions), + allowed_services=tuple(allowed_services), + ) raise ValueError(f"unknown secret type {secret_type!r}") @@ -1065,10 +1144,11 @@ async def _resolve_secrets(secrets: list[SecretDef]) -> dict[str, str]: ``ToolContext`` — the tool gets back the ``replacer`` token, which iron-proxy swaps for the real credential at the network boundary. Inject-mode HTTP secrets are applied entirely by iron-proxy and never reach the tool. - ``GcpAuthSecret``, ``OAuthTokenSecret`` and ``PgDsnSecret`` are likewise not - exposed via context: gcp_auth and oauth_token are minted and injected on the - wire by iron-proxy, and pg_dsn reaches the tool as an environment variable - set on the sandbox by the kubernetes backend. + ``GcpAuthSecret``, ``OAuthTokenSecret``, ``AwsAuthSecret`` and ``PgDsnSecret`` + are likewise not exposed via context: gcp_auth, oauth_token and aws_auth are + minted/re-signed and injected on the wire by iron-proxy (the tool signs AWS + requests with placeholder credentials), and pg_dsn reaches the tool as an + environment variable set on the sandbox by the kubernetes backend. """ return {s.name: s.replacer for s in secrets if _is_replace_secret(s)} diff --git a/services/api/tests/test_proxy_config.py b/services/api/tests/test_proxy_config.py index 4afa9dcd5..3cfdd8d3a 100644 --- a/services/api/tests/test_proxy_config.py +++ b/services/api/tests/test_proxy_config.py @@ -12,6 +12,7 @@ ) from api.tool_manager import ( DEFAULT_MATCH_HEADERS, + AwsAuthSecret, GcpAuthSecret, GitHubAppTokenSecret, HmacHeader, @@ -1838,3 +1839,100 @@ def test_render_github_app_token_uses_broker_credential_id( "rules": [{"host": "api.github.com"}, {"host": "github.com"}], } ] + + +# ── aws_auth parser ────────────────────────────────────────────────────────── + + +def _aws_entry(**overrides): + entry = { + "type": "aws_auth", + "name": "cloudwatch", + "access_key_id": "AWS_ACCESS_KEY_ID", + "secret_access_key": "AWS_SECRET_ACCESS_KEY", + "hosts": ["logs.*.amazonaws.com", "monitoring.*.amazonaws.com"], + "allowed_services": ["logs", "monitoring"], + } + entry.update(overrides) + return entry + + +def test_parser_typed_aws_auth_full_example() -> None: + secret = _parse_secret(_aws_entry()) + assert isinstance(secret, AwsAuthSecret) + assert secret.access_key_id_ref == "AWS_ACCESS_KEY_ID" + assert secret.secret_access_key_ref == "AWS_SECRET_ACCESS_KEY" + assert secret.session_token_ref is None + assert secret.hosts == ("logs.*.amazonaws.com", "monitoring.*.amazonaws.com") + assert secret.allowed_services == ("logs", "monitoring") + assert secret.allowed_regions == () + + +def test_parser_aws_auth_accepts_session_token_and_regions() -> None: + secret = _parse_secret( + _aws_entry(session_token="AWS_SESSION_TOKEN", allowed_regions=["us-east-1"]) + ) + assert isinstance(secret, AwsAuthSecret) + assert secret.session_token_ref == "AWS_SESSION_TOKEN" + assert secret.allowed_regions == ("us-east-1",) + + +def test_parser_aws_auth_requires_access_key_id() -> None: + entry = _aws_entry() + del entry["access_key_id"] + with pytest.raises(ValueError, match="requires a non-empty 'access_key_id'"): + _parse_secret(entry) + + +def test_parser_aws_auth_requires_secret_access_key() -> None: + entry = _aws_entry() + del entry["secret_access_key"] + with pytest.raises(ValueError, match="requires a non-empty 'secret_access_key'"): + _parse_secret(entry) + + +def test_parser_aws_auth_requires_hosts() -> None: + with pytest.raises(ValueError, match="'hosts' must be a non-empty array"): + _parse_secret(_aws_entry(hosts=[])) + + +# ── aws_auth renderer ──────────────────────────────────────────────────────── + + +def test_render_emits_aws_auth_transform() -> None: + secrets = [ + AwsAuthSecret( + name="cloudwatch", + hosts=("logs.*.amazonaws.com", "monitoring.*.amazonaws.com"), + access_key_id_ref="AWS_ACCESS_KEY_ID", + secret_access_key_ref="AWS_SECRET_ACCESS_KEY", + allowed_services=("logs", "monitoring"), + ) + ] + cfg = yaml.safe_load(render_proxy_yaml(secrets)) + aws = next(t for t in cfg["transforms"] if t["name"] == "aws_auth") + assert aws["config"]["access_key_id"] == {"type": "env", "var": "AWS_ACCESS_KEY_ID"} + assert aws["config"]["secret_access_key"] == { + "type": "env", + "var": "AWS_SECRET_ACCESS_KEY", + } + assert aws["config"]["allowed_services"] == ["logs", "monitoring"] + assert "session_token" not in aws["config"] + assert {r["host"] for r in aws["config"]["rules"]} == { + "logs.*.amazonaws.com", + "monitoring.*.amazonaws.com", + } + + +def test_render_merges_aws_auth_hosts_for_shared_credentials() -> None: + secrets = [ + AwsAuthSecret("a", ("logs.*.amazonaws.com",), "AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY"), + AwsAuthSecret("b", ("monitoring.*.amazonaws.com",), "AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY"), + ] + cfg = yaml.safe_load(render_proxy_yaml(secrets)) + blocks = [t for t in cfg["transforms"] if t["name"] == "aws_auth"] + assert len(blocks) == 1 + assert {r["host"] for r in blocks[0]["config"]["rules"]} == { + "logs.*.amazonaws.com", + "monitoring.*.amazonaws.com", + } diff --git a/services/api/tests/test_sandbox_kubernetes_backend.py b/services/api/tests/test_sandbox_kubernetes_backend.py index 8df9199d7..2e9d530c7 100644 --- a/services/api/tests/test_sandbox_kubernetes_backend.py +++ b/services/api/tests/test_sandbox_kubernetes_backend.py @@ -1203,6 +1203,36 @@ def test_proxy_iron_env_injects_broker_when_url_set( } +def test_tool_server_container_exposes_aws_region_not_credentials( + monkeypatch: pytest.MonkeyPatch, +) -> None: + from api.sandbox.kubernetes import _build_tool_server_container + + monkeypatch.setenv("KUBERNETES_TOOL_SERVER_IMAGE", "centaur-api:latest") + monkeypatch.setenv("KUBERNETES_SECRET_ENV_NAME", "centaur-infra-env") + + container = _build_tool_server_container( + thread_key="slack:T:C:1.0", + container_name="tool-server", + firewall_host="fw", + api_url="http://api:8000", + overlay_mount=None, + database_url="postgresql://centaur:centaur@127.0.0.1:15432/centaur", + ) + by_name = {e["name"]: e for e in container["env"]} + + # Region (non-secret) is exposed, optional so sandboxes start without it. + assert by_name["AWS_REGION"]["valueFrom"]["secretKeyRef"] == { + "name": "centaur-infra-env", + "key": "AWS_REGION", + "optional": True, + } + # Credentials must NOT be here — iron-proxy's aws_auth transform holds them + # and re-signs on the wire; they never enter the tool process. + assert "AWS_ACCESS_KEY_ID" not in by_name + assert "AWS_SECRET_ACCESS_KEY" not in by_name + + @pytest.mark.asyncio async def test_per_sandbox_proxy_uses_bootstrap_secret_for_onepassword( monkeypatch: pytest.MonkeyPatch, diff --git a/services/iron-proxy/Dockerfile b/services/iron-proxy/Dockerfile index 80bb2fff1..fce9ae1c6 100644 --- a/services/iron-proxy/Dockerfile +++ b/services/iron-proxy/Dockerfile @@ -1,4 +1,4 @@ -FROM ironsh/iron-proxy:0.42.0-rc.3@sha256:d62792cbc1121e0277cb6bf76df53bf6d1050748cbe625a9ac3039dff1cee88f +FROM ironsh/iron-proxy:0.42.0-rc.4@sha256:9989c5fde4ba3891e53942f08d2de33e690bf2734e5e451210fd845beb992e6f USER root RUN apk add --no-cache curl jq diff --git a/services/iron-proxy/iron-proxy.yaml b/services/iron-proxy/iron-proxy.yaml index 00a8eb382..be9b8edb7 100644 --- a/services/iron-proxy/iron-proxy.yaml +++ b/services/iron-proxy/iron-proxy.yaml @@ -78,6 +78,15 @@ transforms: - "x-as-user-email" - "/^x-codex-.*$/" - "/^x-openai-.*$/" + # AWS SigV4 request headers (x-amz-target, x-amz-date, + # x-amz-content-sha256, x-amz-security-token) for the aws_auth transform. + - "/^x-amz-.*$/" + # AWS SDK headers that the SDK signer folds into the SigV4 signed-headers + # set (amz-sdk-request, amz-sdk-invocation-id, and x-amzn-query-mode for + # CloudWatch's query-JSON protocol). aws_auth signs them, so they must + # survive egress filtering or AWS rejects with InvalidSignatureException. + - "/^amz-sdk-.*$/" + - "/^x-amzn-.*$/" - "/^x-[a-z0-9-]*(api-key|apikey|secret|token|auth|key)$/" log: diff --git a/tools/infra/cloudwatch/.env.example b/tools/infra/cloudwatch/.env.example new file mode 100644 index 000000000..ad555dc1b --- /dev/null +++ b/tools/infra/cloudwatch/.env.example @@ -0,0 +1,16 @@ +# The real AWS credentials are resolved by iron-proxy (via the secrets backend +# / 1Password), NOT by this tool — boto3 signs with placeholders and iron-proxy +# re-signs on the wire. Store a read-only CloudWatch IAM user's keys under the +# names below in your secrets backend, the same way as other tool credentials. +# Scope the IAM policy tightly — read-only CloudWatch Logs + Metrics only, e.g.: +# logs:DescribeLogGroups, logs:FilterLogEvents, logs:GetLogEvents, +# logs:StartQuery, logs:GetQueryResults, logs:StopQuery, +# cloudwatch:ListMetrics, cloudwatch:GetMetricData, +# cloudwatch:DescribeAlarms, cloudwatch:DescribeAlarmHistory +AWS_ACCESS_KEY_ID=your-access-key-id +AWS_SECRET_ACCESS_KEY=your-secret-access-key + +# Region the log groups / metrics live in. NOT a secret — this is the only AWS +# value the tool process itself reads (to pick the endpoint + signing scope). +# Defaults to us-east-1 if unset. +AWS_REGION=us-east-1 diff --git a/tools/infra/cloudwatch/client.py b/tools/infra/cloudwatch/client.py new file mode 100644 index 000000000..5ede3ff9b --- /dev/null +++ b/tools/infra/cloudwatch/client.py @@ -0,0 +1,376 @@ +"""AWS CloudWatch client for read-only logs, metrics, and alarms. + +Mirrors the useful read-only surface of the AWS CloudWatch MCP server using +boto3: browse log groups, tail/filter log events, run CloudWatch Logs Insights +queries, list metrics and pull metric data, and inspect alarms. + +AWS auth rides iron-proxy's ``aws_auth`` transform (declared in pyproject.toml): +boto3 signs each request with throwaway *placeholder* credentials, and iron-proxy +reads the region/service from the signature scope, strips it, and re-signs with +the real read-only IAM keys it resolves from the secrets backend. The real keys +never enter this process — the SigV4 analogue of the ``secrets`` placeholder +swap. Only the region is a real value: boto3 needs it to pick the endpoint host +and credential scope, and it isn't a secret. +""" + +from __future__ import annotations + +import os +from datetime import UTC, datetime, timedelta +from typing import Any + +_DEFAULT_REGION = "us-east-1" + +# boto3 must sign with *some* credentials; iron-proxy's aws_auth transform +# discards this signature and re-signs with the real keys, so the value is +# irrelevant beyond being non-empty. +_PLACEHOLDER_CREDENTIAL = "iron-proxy-resigns-this" + + +class CloudWatchClient: + """Read-only client for CloudWatch Logs and Metrics (boto3, SigV4). + + boto3 signs with placeholder credentials; iron-proxy's ``aws_auth`` transform + re-signs with the real read-only IAM keys (resolved from the secrets backend), + so credentials never reach this process. The region comes from ``AWS_REGION`` + (a non-secret), defaulting to ``us-east-1``. boto3 clients are built lazily on + first use so tool discovery never needs network access. + """ + + def __init__(self, region: str | None = None): + self._region = region + self.__logs: Any = None + self.__cw: Any = None + + # -- boto3 plumbing (lazy) ---------------------------------------------- + + @property + def region(self) -> str: + # Region is non-secret config, read straight from the env (not secret(), + # whose server-mode StubBackend returns the key name as a placeholder + # rather than the default). Defaults to us-east-1 when unset. + return self._region or os.getenv("AWS_REGION") or _DEFAULT_REGION # noqa: TID251 + + def _session(self) -> Any: + import boto3 # lazy: keeps import cheap and tests boto3-free + + # Placeholder credentials — iron-proxy re-signs on the wire. Passed + # explicitly so boto3 never reaches for IMDS / ambient AWS config. + return boto3.session.Session( + aws_access_key_id=_PLACEHOLDER_CREDENTIAL, + aws_secret_access_key=_PLACEHOLDER_CREDENTIAL, + region_name=self.region, + ) + + def _logs(self) -> Any: + if self.__logs is None: + self.__logs = self._session().client("logs") + return self.__logs + + def _cw(self) -> Any: + if self.__cw is None: + self.__cw = self._session().client("cloudwatch") + return self.__cw + + @staticmethod + def _call(fn: Any, **kwargs: Any) -> dict: + """Invoke a boto3 call, dropping None args and normalizing errors.""" + clean = {k: v for k, v in kwargs.items() if v is not None} + try: + return fn(**clean) + except Exception as exc: # botocore.ClientError et al. + raise RuntimeError(f"CloudWatch API error: {exc}") from exc + + # -- Logs: groups & events ---------------------------------------------- + + def list_log_groups( + self, + name_prefix: str | None = None, + limit: int = 50, + ) -> list[dict]: + """List CloudWatch log groups, optionally filtered by name prefix. + + Use this to discover log group names for filter_log_events / start_query. + + Args: + name_prefix: Only return groups whose name starts with this string. + limit: Max groups to return (1-50). + """ + resp = self._call( + self._logs().describe_log_groups, + logGroupNamePrefix=name_prefix, + limit=max(1, min(limit, 50)), + ) + return _clean(resp.get("logGroups", [])) + + def filter_log_events( + self, + log_group_name: str, + filter_pattern: str | None = None, + start_time: str | None = None, + end_time: str | None = None, + limit: int = 100, + ) -> dict: + """Search log events in a group within a time window. + + The workhorse for grepping logs. ``filter_pattern`` uses CloudWatch Logs + filter syntax (e.g. 'ERROR', '"timeout"', '{ $.level = "error" }'). + + Args: + log_group_name: Exact log group name (see list_log_groups). + filter_pattern: CloudWatch Logs filter pattern. Omit to return all events. + start_time: ISO-8601 timestamp or epoch (s/ms). Defaults to 1h before end. + end_time: ISO-8601 timestamp or epoch (s/ms). Defaults to now. + limit: Max events to return (1-10000). + """ + start_ms, end_ms = _resolve_window_ms(start_time, end_time) + resp = self._call( + self._logs().filter_log_events, + logGroupName=log_group_name, + filterPattern=filter_pattern, + startTime=start_ms, + endTime=end_ms, + limit=max(1, min(limit, 10000)), + ) + return { + "events": _clean(resp.get("events", [])), + "searched_log_streams": _clean(resp.get("searchedLogStreams", [])), + } + + # -- Logs Insights ------------------------------------------------------- + + def start_query( + self, + log_group_names: list[str] | str, + query_string: str, + start_time: str | None = None, + end_time: str | None = None, + limit: int = 100, + ) -> dict: + """Start a CloudWatch Logs Insights query. Returns a query_id to poll. + + Logs Insights is asynchronous: call this to start, then poll + get_query_results with the returned query_id until status is Complete. + + Args: + log_group_names: One name or a list of log group names to query. + query_string: Logs Insights query, e.g. + 'fields @timestamp, @message | filter @message like /ERROR/ | sort @timestamp desc'. + start_time: ISO-8601 timestamp or epoch (s/ms). Defaults to 1h before end. + end_time: ISO-8601 timestamp or epoch (s/ms). Defaults to now. + limit: Max rows the query may return (1-10000). + """ + names = [log_group_names] if isinstance(log_group_names, str) else list(log_group_names) + start_ms, end_ms = _resolve_window_ms(start_time, end_time) + resp = self._call( + self._logs().start_query, + logGroupNames=names, + queryString=query_string, + startTime=start_ms // 1000, # Insights wants epoch seconds + endTime=end_ms // 1000, + limit=max(1, min(limit, 10000)), + ) + return _clean(resp) + + def get_query_results(self, query_id: str) -> dict: + """Get results/status for a Logs Insights query started with start_query. + + Status is one of Scheduled, Running, Complete, Failed, Cancelled, Timeout. + Poll until status is Complete (or terminal) before trusting the results. + + Args: + query_id: The query_id returned by start_query. + """ + resp = self._call(self._logs().get_query_results, queryId=query_id) + return _clean(resp) + + def stop_query(self, query_id: str) -> dict: + """Stop a running Logs Insights query. + + Args: + query_id: The query_id returned by start_query. + """ + return _clean(self._call(self._logs().stop_query, queryId=query_id)) + + # -- Metrics ------------------------------------------------------------- + + def list_metrics( + self, + namespace: str | None = None, + metric_name: str | None = None, + limit: int = 100, + ) -> list[dict]: + """List available metrics, optionally filtered by namespace/name. + + Use to discover the namespace, metric name, and dimensions to pass to + get_metric_data. + + Args: + namespace: e.g. 'AWS/EC2', 'AWS/Lambda', or a custom namespace. + metric_name: e.g. 'CPUUtilization', 'Errors'. + limit: Max metrics to return (results are truncated client-side). + """ + resp = self._call( + self._cw().list_metrics, + Namespace=namespace, + MetricName=metric_name, + ) + return _clean(resp.get("Metrics", []))[: max(1, limit)] + + def get_metric_data( + self, + namespace: str, + metric_name: str, + dimensions: dict[str, str] | None = None, + stat: str = "Average", + period: int = 300, + start_time: str | None = None, + end_time: str | None = None, + ) -> dict: + """Fetch time-series data points for a single metric. + + Args: + namespace: Metric namespace, e.g. 'AWS/Lambda'. + metric_name: Metric name, e.g. 'Errors'. + dimensions: Dimension name→value map, e.g. {'FunctionName': 'my-fn'}. + stat: Statistic — Average, Sum, Minimum, Maximum, SampleCount, or p99 etc. + period: Granularity in seconds (must be a multiple of 60). + start_time: ISO-8601 timestamp or epoch (s/ms). Defaults to 1h before end. + end_time: ISO-8601 timestamp or epoch (s/ms). Defaults to now. + """ + start_dt, end_dt = _resolve_window_dt(start_time, end_time) + dims = [{"Name": k, "Value": v} for k, v in (dimensions or {}).items()] + resp = self._call( + self._cw().get_metric_data, + MetricDataQueries=[ + { + "Id": "m1", + "MetricStat": { + "Metric": { + "Namespace": namespace, + "MetricName": metric_name, + "Dimensions": dims, + }, + "Period": period, + "Stat": stat, + }, + "ReturnData": True, + } + ], + StartTime=start_dt, + EndTime=end_dt, + ) + return _clean(resp.get("MetricDataResults", [])) + + # -- Alarms -------------------------------------------------------------- + + def describe_alarms( + self, + state_value: str | None = None, + alarm_name_prefix: str | None = None, + limit: int = 50, + ) -> list[dict]: + """List metric alarms, optionally filtered by state and name prefix. + + Pass state_value='ALARM' to see only currently-firing alarms. + + Args: + state_value: 'OK', 'ALARM', or 'INSUFFICIENT_DATA'. + alarm_name_prefix: Only alarms whose name starts with this string. + limit: Max alarms to return (1-100). + """ + resp = self._call( + self._cw().describe_alarms, + StateValue=state_value, + AlarmNamePrefix=alarm_name_prefix, + MaxRecords=max(1, min(limit, 100)), + ) + return _clean(resp.get("MetricAlarms", [])) + + def get_alarm_history( + self, + alarm_name: str | None = None, + limit: int = 50, + ) -> list[dict]: + """Get state-change history for an alarm (or all alarms). + + Args: + alarm_name: Restrict to one alarm. Omit for history across all alarms. + limit: Max history items to return (1-100). + """ + resp = self._call( + self._cw().describe_alarm_history, + AlarmName=alarm_name, + MaxRecords=max(1, min(limit, 100)), + ) + return _clean(resp.get("AlarmHistoryItems", [])) + + # -- Lifecycle ----------------------------------------------------------- + + def close(self): + self.__logs = None + self.__cw = None + + def __enter__(self): + return self + + def __exit__(self, *args): + self.close() + + +# -- Helpers ----------------------------------------------------------------- + + +def _to_epoch_ms(value: str | int | float | None) -> int | None: + """Coerce an ISO-8601 string or epoch (seconds or millis) to epoch millis.""" + if value is None: + return None + if isinstance(value, (int, float)): + # Heuristic: values past ~2001 in seconds are < 1e12; millis are larger. + return int(value if value > 1_000_000_000_000 else value * 1000) + text = str(value).strip().replace("Z", "+00:00") + dt = datetime.fromisoformat(text) + if dt.tzinfo is None: + dt = dt.replace(tzinfo=UTC) + return int(dt.timestamp() * 1000) + + +def _resolve_window_ms(start: str | None, end: str | None) -> tuple[int, int]: + """Resolve a (start, end) window to epoch millis, defaulting to the last hour.""" + end_ms = _to_epoch_ms(end) + if end_ms is None: + end_ms = int(datetime.now(UTC).timestamp() * 1000) + start_ms = _to_epoch_ms(start) + if start_ms is None: + start_ms = end_ms - int(timedelta(hours=1).total_seconds() * 1000) + return start_ms, end_ms + + +def _resolve_window_dt(start: str | None, end: str | None) -> tuple[datetime, datetime]: + """Resolve a (start, end) window to tz-aware datetimes (CloudWatch metrics API).""" + start_ms, end_ms = _resolve_window_ms(start, end) + return ( + datetime.fromtimestamp(start_ms / 1000, tz=UTC), + datetime.fromtimestamp(end_ms / 1000, tz=UTC), + ) + + +def _clean(obj: Any) -> Any: + """Make a boto3 response JSON-serializable. + + Converts datetimes to ISO-8601, decodes bytes, and strips the boilerplate + ``ResponseMetadata`` envelope so it doesn't bloat the agent's context. + """ + if isinstance(obj, dict): + return {k: _clean(v) for k, v in obj.items() if k != "ResponseMetadata"} + if isinstance(obj, (list, tuple)): + return [_clean(v) for v in obj] + if isinstance(obj, datetime): + return obj.isoformat() + if isinstance(obj, (bytes, bytearray)): + return bytes(obj).decode("utf-8", "replace") + return obj + + +def _client() -> CloudWatchClient: + return CloudWatchClient() diff --git a/tools/infra/cloudwatch/pyproject.toml b/tools/infra/cloudwatch/pyproject.toml new file mode 100644 index 000000000..b09ed8bc6 --- /dev/null +++ b/tools/infra/cloudwatch/pyproject.toml @@ -0,0 +1,32 @@ +[project] +name = "cloudwatch" +description = "AWS CloudWatch — Logs Insights queries, log events, metrics, and alarms (read-only)" +version = "0.1.0" +requires-python = ">=3.11" +dependencies = [ + "boto3>=1.34", +] + +[build-system] +requires = ["hatchling"] +build-backend = "hatchling.build" + + +# AWS SigV4 rides iron-proxy's `aws_auth` transform: boto3 signs each request +# with throwaway placeholder credentials, and iron-proxy re-signs it with the +# real read-only IAM keys it resolves from `access_key_id`/`secret_access_key` +# (stored in the secrets backend like any other tool credential). The real keys +# never enter the tool process — same placeholder-swap model as the `secrets` +# transform, just for SigV4. Scope the IAM user to read-only CloudWatch. +[tool.centaur] +module = "client.py" +hosts = ["logs.*.amazonaws.com", "monitoring.*.amazonaws.com"] +secrets = [ + { type = "aws_auth", name = "cloudwatch", access_key_id = "AWS_ACCESS_KEY_ID", secret_access_key = "AWS_SECRET_ACCESS_KEY", hosts = [ + "logs.*.amazonaws.com", + "monitoring.*.amazonaws.com", + ], allowed_services = [ + "logs", + "monitoring", + ] }, +] diff --git a/tools/infra/cloudwatch/test_client.py b/tools/infra/cloudwatch/test_client.py new file mode 100644 index 000000000..3be0b2dd9 --- /dev/null +++ b/tools/infra/cloudwatch/test_client.py @@ -0,0 +1,164 @@ +from __future__ import annotations + +import importlib.util +from datetime import UTC, datetime +from pathlib import Path +from typing import Any + +spec = importlib.util.spec_from_file_location( + "cloudwatch_client", Path(__file__).with_name("client.py") +) +assert spec and spec.loader +module = importlib.util.module_from_spec(spec) +spec.loader.exec_module(module) +CloudWatchClient = module.CloudWatchClient + + +class FakeBotoClient: + """Records boto3 calls and returns canned responses, no network or boto3.""" + + def __init__(self, responses: dict[str, Any] | None = None) -> None: + self.calls: list[dict[str, Any]] = [] + self.responses = responses or {} + + def __getattr__(self, name: str): + def _method(**kwargs: Any) -> Any: + self.calls.append({"op": name, "kwargs": kwargs}) + return self.responses.get(name, {}) + + return _method + + +class RecordingCloudWatchClient(CloudWatchClient): + """Swaps boto3 logs/cloudwatch clients for recording fakes.""" + + def __init__(self, **responses: Any) -> None: + super().__init__(region="us-west-2") + self.logs = FakeBotoClient(responses) + self.cw = FakeBotoClient(responses) + + def _logs(self) -> Any: + return self.logs + + def _cw(self) -> Any: + return self.cw + + +def test_list_log_groups_clamps_limit_and_drops_none() -> None: + client = RecordingCloudWatchClient(describe_log_groups={"logGroups": [{"logGroupName": "/x"}]}) + + out = client.list_log_groups(limit=999) + + assert out == [{"logGroupName": "/x"}] + call = client.logs.calls[-1] + assert call["op"] == "describe_log_groups" + assert call["kwargs"] == {"limit": 50} # clamped, name_prefix=None dropped + + +def test_filter_log_events_defaults_to_last_hour() -> None: + client = RecordingCloudWatchClient() + + client.filter_log_events("/aws/lambda/fn", end_time="2026-05-28T12:00:00Z") + + kwargs = client.logs.calls[-1]["kwargs"] + assert kwargs["logGroupName"] == "/aws/lambda/fn" + assert kwargs["endTime"] == int(datetime(2026, 5, 28, 12, tzinfo=UTC).timestamp() * 1000) + assert kwargs["startTime"] == kwargs["endTime"] - 3_600_000 + assert "filterPattern" not in kwargs # None dropped + + +def test_filter_log_events_passes_pattern_and_clamps_limit() -> None: + client = RecordingCloudWatchClient() + + client.filter_log_events("/g", filter_pattern="ERROR", limit=99999) + + kwargs = client.logs.calls[-1]["kwargs"] + assert kwargs["filterPattern"] == "ERROR" + assert kwargs["limit"] == 10000 + + +def test_start_query_normalizes_names_and_uses_epoch_seconds() -> None: + client = RecordingCloudWatchClient(start_query={"queryId": "q-1"}) + + out = client.start_query( + "/only-one", + "fields @message", + start_time="2026-05-28T11:00:00Z", + end_time="2026-05-28T12:00:00Z", + ) + + assert out == {"queryId": "q-1"} + kwargs = client.logs.calls[-1]["kwargs"] + assert kwargs["logGroupNames"] == ["/only-one"] + assert kwargs["startTime"] == int(datetime(2026, 5, 28, 11, tzinfo=UTC).timestamp()) + assert kwargs["endTime"] == int(datetime(2026, 5, 28, 12, tzinfo=UTC).timestamp()) + + +def test_get_metric_data_builds_single_query() -> None: + client = RecordingCloudWatchClient(get_metric_data={"MetricDataResults": [{"Id": "m1"}]}) + + out = client.get_metric_data( + "AWS/Lambda", + "Errors", + dimensions={"FunctionName": "fn"}, + stat="Sum", + period=60, + start_time="2026-05-28T11:00:00Z", + end_time="2026-05-28T12:00:00Z", + ) + + assert out == [{"Id": "m1"}] + kwargs = client.cw.calls[-1]["kwargs"] + q = kwargs["MetricDataQueries"][0] + assert q["MetricStat"]["Metric"]["Namespace"] == "AWS/Lambda" + assert q["MetricStat"]["Metric"]["Dimensions"] == [{"Name": "FunctionName", "Value": "fn"}] + assert q["MetricStat"]["Stat"] == "Sum" + assert q["MetricStat"]["Period"] == 60 + assert kwargs["StartTime"] == datetime(2026, 5, 28, 11, tzinfo=UTC) + assert kwargs["EndTime"] == datetime(2026, 5, 28, 12, tzinfo=UTC) + + +def test_describe_alarms_filters_active() -> None: + client = RecordingCloudWatchClient(describe_alarms={"MetricAlarms": [{"AlarmName": "a"}]}) + + out = client.describe_alarms(state_value="ALARM") + + assert out == [{"AlarmName": "a"}] + kwargs = client.cw.calls[-1]["kwargs"] + assert kwargs["StateValue"] == "ALARM" + assert kwargs["MaxRecords"] == 50 + assert "AlarmNamePrefix" not in kwargs + + +def test_clean_strips_metadata_and_serializes_datetimes() -> None: + cleaned = module._clean( + { + "ResponseMetadata": {"RequestId": "abc"}, + "MetricAlarms": [{"StateUpdatedTimestamp": datetime(2026, 5, 28, tzinfo=UTC)}], + } + ) + + assert "ResponseMetadata" not in cleaned + assert cleaned["MetricAlarms"][0]["StateUpdatedTimestamp"] == "2026-05-28T00:00:00+00:00" + + +def test_to_epoch_ms_handles_seconds_and_millis() -> None: + assert module._to_epoch_ms(1_700_000_000) == 1_700_000_000_000 # seconds → ms + assert module._to_epoch_ms(1_700_000_000_000) == 1_700_000_000_000 # already ms + assert module._to_epoch_ms(None) is None + + +def test_api_errors_are_wrapped() -> None: + client = RecordingCloudWatchClient() + + def boom(**_: Any): + raise ValueError("AccessDenied") + + client.logs.describe_log_groups = boom # type: ignore[assignment] + + try: + client.list_log_groups() + except RuntimeError as exc: + assert "CloudWatch API error" in str(exc) + else: + raise AssertionError("expected RuntimeError")