From dd95a3a9e72ec25b080b589e4a08a5e771809cde Mon Sep 17 00:00:00 2001 From: Nitin Kumar Date: Sun, 27 Sep 2026 17:05:56 +0530 Subject: [PATCH 1/3] fix(cloudwatch): serve the awsJson1_0 protocol used by botocore 1.43+ (CW-X12) Adds a JSON codec over the shared CBOR per-op cores and unwraps gzip request bodies on all three protocols. --- server/aws/cloudwatch/handler.go | 36 +- server/aws/cloudwatch/json_protocol.go | 293 ++++++++++++ server/aws/cloudwatch/json_protocol_test.go | 444 ++++++++++++++++++ server/aws/cloudwatch/json_transcode_test.go | 81 ++++ .../cloudwatch/request_compression_test.go | 176 +++++++ server/aws/cloudwatch/request_encoding.go | 46 ++ server/aws/cloudwatch_json_dispatch_test.go | 39 ++ 7 files changed, 1108 insertions(+), 7 deletions(-) create mode 100644 server/aws/cloudwatch/json_protocol.go create mode 100644 server/aws/cloudwatch/json_protocol_test.go create mode 100644 server/aws/cloudwatch/json_transcode_test.go create mode 100644 server/aws/cloudwatch/request_compression_test.go create mode 100644 server/aws/cloudwatch/request_encoding.go create mode 100644 server/aws/cloudwatch_json_dispatch_test.go diff --git a/server/aws/cloudwatch/handler.go b/server/aws/cloudwatch/handler.go index 238fe2057..58e835f9d 100644 --- a/server/aws/cloudwatch/handler.go +++ b/server/aws/cloudwatch/handler.go @@ -1,5 +1,5 @@ -// Package cloudwatch implements AWS CloudWatch's Smithy RPC-v2-CBOR protocol -// as a server.Handler. +// Package cloudwatch implements AWS CloudWatch as a server.Handler over the +// three protocols its clients speak: Smithy RPC-v2-CBOR, awsJson1_0 and query. // // Modern aws-sdk-go-v2 CloudWatch clients no longer use the AWS query protocol. // They send CBOR-encoded request bodies to URLs like @@ -9,7 +9,9 @@ // Content-Type: application/cbor // // This handler matches those requests, decodes CBOR, dispatches to the -// monitoring driver, and writes CBOR responses. +// monitoring driver, and writes CBOR responses. botocore 1.43+ sends +// awsJson1_0 instead (json_protocol.go), and the AWS CLI v2 and older SDKs +// send query (query.go). All three run through the same per-op cores. package cloudwatch import ( @@ -89,7 +91,8 @@ func (h *Handler) SetIPAMMetrics(ipam netdriver.IPAMMetrics) { h.ipam = ipam } -// Matches returns true for Smithy rpc-v2-cbor requests, and for classic +// Matches returns true for Smithy rpc-v2-cbor requests, for awsJson1_0 +// requests whose X-Amz-Target names the CloudWatch service, and for classic // query-protocol CloudWatch requests (used by the AWS CLI and older SDKs), // disambiguated from EC2 by the SigV4 "monitoring" credential scope. func (*Handler) Matches(r *http.Request) bool { @@ -97,11 +100,18 @@ func (*Handler) Matches(r *http.Request) bool { return true } - return isQueryRequest(r) + return isJSONRequest(r) || isQueryRequest(r) } // ServeHTTP parses the URL path for the operation name and dispatches. func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) { + decodeRequestBody(r) + + if r.Header.Get(protocolHeader) != protocolValue && isJSONRequest(r) { + h.serveJSON(w, r) + return + } + if isQueryRequest(r) { h.serveQuery(w, r) return @@ -203,8 +213,14 @@ func extractOperation(path string) string { return path[i+len(opMarker):] } -// writeCBORError writes an rpc-v2-cbor error response. +// writeCBORError writes an rpc-v2-cbor error response, or the awsJson1_0 +// error when w is the writer of a JSON request. func writeCBORError(w http.ResponseWriter, status int, errType, msg string) { + if jw, ok := w.(*jsonWriter); ok { + jw.writeError(status, errType, msg) + return + } + payload := map[string]any{ "__type": errType, "message": msg, @@ -232,8 +248,14 @@ func mustSmithyEncMode() cbor.EncMode { return mode } -// writeCBORResponse writes a successful rpc-v2-cbor response body. +// writeCBORResponse writes a successful rpc-v2-cbor response body, or the +// awsJson1_0 body when w is the writer of a JSON request. func writeCBORResponse(w http.ResponseWriter, payload any) { + if jw, ok := w.(*jsonWriter); ok { + jw.writeResult(payload) + return + } + body, err := smithyEncMode.Marshal(payload) if err != nil { writeCBORError(w, http.StatusInternalServerError, "InternalError", err.Error()) diff --git a/server/aws/cloudwatch/json_protocol.go b/server/aws/cloudwatch/json_protocol.go new file mode 100644 index 000000000..b41470285 --- /dev/null +++ b/server/aws/cloudwatch/json_protocol.go @@ -0,0 +1,293 @@ +package cloudwatch + +// CloudWatch over awsJson1_0. botocore 1.43+ (boto3 and the Python aws CLI) +// picks this protocol for CloudWatch: a POST with +// +// X-Amz-Target: GraniteServiceVersion20100801. +// Content-Type: application/x-amz-json-1.0 +// +// The JSON and CBOR wire shapes use the same member names. The two protocols +// only differ in how timestamps (epoch seconds numbers in JSON, CBOR tag 1) +// and blobs (base64 strings in JSON, CBOR byte strings) are written, and in +// the error envelope. So the JSON codec converts the request body to CBOR, +// runs the same per-op handler and core as rpc-v2-cbor, and converts the +// result back to JSON on the way out. + +import ( + "bytes" + "encoding/json" + "errors" + "fmt" + "io" + "math" + "net/http" + "reflect" + "strconv" + "strings" + "time" + + "github.com/fxamacker/cbor/v2" +) + +const ( + jsonTargetPrefix = "GraniteServiceVersion20100801." + jsonContentType = "application/x-amz-json-1.0" + jsonTargetHeader = "X-Amz-Target" + queryErrorHeader = "X-Amzn-Query-Error" + errTypeHeader = "X-Amzn-Errortype" + errSerialization = "SerializationException" + msPerSecond = 1e3 +) + +var errNotJSONObject = errors.New("request body must be a JSON object") + +// jsonErrorShapes maps the query error codes the cores return to the error +// shape names of the CloudWatch model. CloudWatch is awsQueryCompatible, so +// the shape name goes in __type and the query code rides in the +// X-Amzn-Query-Error header, which botocore uses as the error code. +// +//nolint:gochecknoglobals // read-only lookup table +var jsonErrorShapes = map[string]string{ + "InvalidParameterValue": "InvalidParameterValueException", + "InvalidParameterCombination": "InvalidParameterCombinationException", + "MissingParameter": "MissingRequiredParameterException", + "ValidationError": "ValidationException", + "InvalidFormat": "InvalidFormatFault", + "LimitExceeded": "LimitExceededFault", + "InternalServiceError": "InternalServiceFault", + "InvalidParameterInput": "DashboardInvalidInputError", +} + +// isJSONRequest reports whether r is a CloudWatch awsJson1_0 request. +func isJSONRequest(r *http.Request) bool { + return r.Method == http.MethodPost && jsonOperation(r) != "" +} + +// jsonOperation returns the operation named by the X-Amz-Target header, or "" +// when the target is not a CloudWatch one. +func jsonOperation(r *http.Request) string { + op, ok := strings.CutPrefix(r.Header.Get(jsonTargetHeader), jsonTargetPrefix) + if !ok { + return "" + } + + return op +} + +// serveJSON handles a CloudWatch awsJson1_0 request. +func (h *Handler) serveJSON(w http.ResponseWriter, r *http.Request) { + jw := &jsonWriter{w: w} + + r.Body = http.MaxBytesReader(w, r.Body, maxBodyBytes) + + raw, err := io.ReadAll(r.Body) + if err != nil { + jw.writeError(http.StatusBadRequest, errSerialization, err.Error()) + return + } + + body, err := jsonToCBOR(raw) + if err != nil { + jw.writeError(http.StatusBadRequest, errSerialization, err.Error()) + return + } + + h.dispatch(jw, r, jsonOperation(r), body) +} + +// jsonWriter is the http.ResponseWriter the per-op handlers get for a JSON +// request. writeCBORResponse and writeCBORError hand their payload to it, so +// every op answers in JSON without a JSON twin of its handler. +type jsonWriter struct { + w http.ResponseWriter +} + +func (j *jsonWriter) Header() http.Header { return j.w.Header() } +func (j *jsonWriter) Write(b []byte) (int, error) { return j.w.Write(b) } +func (j *jsonWriter) WriteHeader(status int) { j.w.WriteHeader(status) } + +// writeResult encodes a CBOR wire struct as a JSON response body. +func (j *jsonWriter) writeResult(payload any) { + body, err := cborPayloadToJSON(payload) + if err != nil { + j.writeError(http.StatusInternalServerError, "InternalError", err.Error()) + return + } + + j.w.Header().Set("Content-Type", jsonContentType) + j.w.WriteHeader(http.StatusOK) + _, _ = j.w.Write(body) +} + +// writeError writes an awsJson1_0 error: the model shape name in __type and +// X-Amzn-Errortype, and the query code in X-Amzn-Query-Error. +func (j *jsonWriter) writeError(status int, code, msg string) { + shape := code + if s, ok := jsonErrorShapes[code]; ok { + shape = s + } + + fault := "Sender" + if status >= http.StatusInternalServerError { + fault = "Receiver" + } + + body, _ := json.Marshal(map[string]string{"__type": shape, "message": msg}) + + hdr := j.w.Header() + hdr.Set("Content-Type", jsonContentType) + hdr.Set(errTypeHeader, shape) + hdr.Set(queryErrorHeader, code+";"+fault) + j.w.WriteHeader(status) + _, _ = j.w.Write(body) +} + +// jsonToCBOR re-encodes a JSON request body as CBOR for the shared decoders. +// Integer literals become CBOR integers so they decode into int fields, and +// timestamps (epoch seconds numbers) decode into time.Time as untagged +// numbers. An empty body is an empty input. +func jsonToCBOR(raw []byte) ([]byte, error) { + if len(bytes.TrimSpace(raw)) == 0 { + raw = []byte("{}") + } + + dec := json.NewDecoder(bytes.NewReader(raw)) + dec.UseNumber() + + var v any + if err := dec.Decode(&v); err != nil { + return nil, fmt.Errorf("invalid JSON: %w", err) + } + + if _, ok := v.(map[string]any); !ok { + return nil, errNotJSONObject + } + + conv, err := jsonNumbersToCBOR(v) + if err != nil { + return nil, err + } + + return cbor.Marshal(conv) +} + +func jsonNumbersToCBOR(v any) (any, error) { + switch t := v.(type) { + case map[string]any: + for k, e := range t { + c, err := jsonNumbersToCBOR(e) + if err != nil { + return nil, err + } + + t[k] = c + } + case []any: + for i, e := range t { + c, err := jsonNumbersToCBOR(e) + if err != nil { + return nil, err + } + + t[i] = c + } + case json.Number: + if i, err := strconv.ParseInt(t.String(), 10, 64); err == nil { + return i, nil + } + + f, err := t.Float64() + if err != nil { + return nil, fmt.Errorf("invalid number %q: %w", t, err) + } + + return f, nil + } + + return v, nil +} + +// cborDecMode decodes a CBOR response back to generic values, with string map +// keys and tag 1 timestamps as time.Time. +var cborDecMode = mustCBORDecMode() //nolint:gochecknoglobals // reused decoder + +func mustCBORDecMode() cbor.DecMode { + mode, err := cbor.DecOptions{DefaultMapType: reflect.TypeFor[map[string]any]()}.DecMode() + if err != nil { + panic(err) + } + + return mode +} + +// cborPayloadToJSON encodes a CBOR wire struct as JSON. It goes through the +// CBOR encoding so the cbor struct tags (names, omitempty) stay the single +// definition of the wire shape. +func cborPayloadToJSON(payload any) ([]byte, error) { + enc, err := smithyEncMode.Marshal(payload) + if err != nil { + return nil, err + } + + var v any + if err := cborDecMode.Unmarshal(enc, &v); err != nil { + return nil, err + } + + return json.Marshal(cborToJSONValue(v)) +} + +// cborToJSONValue converts decoded CBOR values to their awsJson1_0 form: +// timestamps as epoch seconds, byte strings as base64 (encoding/json does +// that for []byte), and null members dropped. +func cborToJSONValue(v any) any { + switch t := v.(type) { + case map[string]any: + out := make(map[string]any, len(t)) + + for k, e := range t { + if e == nil { + continue + } + + out[k] = cborToJSONValue(e) + } + + return out + case []any: + out := make([]any, len(t)) + for i, e := range t { + out[i] = cborToJSONValue(e) + } + + return out + case time.Time: + return epochSeconds(t) + case float64: + return jsonDouble(t) + } + + return v +} + +// jsonDouble keeps a double a double on the wire: 20 is written as 20.0, so +// botocore hands back a float like it does for real CloudWatch. CBOR keeps +// int and double members apart, so only double members get here. +func jsonDouble(f float64) any { + if math.IsNaN(f) || math.IsInf(f, 0) { + return f + } + + s := strconv.FormatFloat(f, 'f', -1, 64) + if !strings.ContainsAny(s, ".e") { + s += ".0" + } + + return json.Number(s) +} + +// epochSeconds renders t as the JSON number of seconds since the epoch, with +// millisecond precision like the AWS SDKs. +func epochSeconds(t time.Time) json.Number { + return json.Number(strconv.FormatFloat(float64(t.UnixMilli())/msPerSecond, 'f', -1, 64)) +} diff --git a/server/aws/cloudwatch/json_protocol_test.go b/server/aws/cloudwatch/json_protocol_test.go new file mode 100644 index 000000000..4aeeb7bd9 --- /dev/null +++ b/server/aws/cloudwatch/json_protocol_test.go @@ -0,0 +1,444 @@ +package cloudwatch_test + +import ( + "encoding/json" + "fmt" + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/stackshy/cloudemu/v2/config" + cwprovider "github.com/stackshy/cloudemu/v2/providers/aws/cloudwatch" + cwserver "github.com/stackshy/cloudemu/v2/server/aws/cloudwatch" +) + +// jsonTarget is the X-Amz-Target prefix botocore sends for CloudWatch over +// awsJson1_0. +const jsonTarget = "GraniteServiceVersion20100801." + +type jsonResp struct { + status int + header http.Header + body map[string]any + raw string +} + +func newJSONServer(t *testing.T) *httptest.Server { + t.Helper() + + ts := httptest.NewServer(cwserver.New(cwprovider.New(config.NewOptions()))) + t.Cleanup(ts.Close) + + return ts +} + +func jsonCall(t *testing.T, ts *httptest.Server, op, body string) jsonResp { + t.Helper() + + req, err := http.NewRequest(http.MethodPost, ts.URL+"/", strings.NewReader(body)) //nolint:noctx // test request + if err != nil { + t.Fatalf("new request: %v", err) + } + + req.Header.Set("Content-Type", "application/x-amz-json-1.0") + req.Header.Set("X-Amz-Target", jsonTarget+op) + req.Header.Set("Authorization", monitoringAuth) + + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatalf("%s: %v", op, err) + } + defer resp.Body.Close() + + raw, _ := io.ReadAll(resp.Body) + + out := jsonResp{status: resp.StatusCode, header: resp.Header, raw: string(raw)} + if err := json.Unmarshal(raw, &out.body); err != nil { + t.Fatalf("%s: response is not a JSON object: %v: %q", op, err, raw) + } + + return out +} + +func jsonList(t *testing.T, r jsonResp, key string) []any { + t.Helper() + + v, ok := r.body[key].([]any) + if !ok { + t.Fatalf("%s is not a JSON list: %s", key, r.raw) + } + + return v +} + +func TestJSONProtocolMatches(t *testing.T) { + h := cwserver.New(cwprovider.New(config.NewOptions())) + + mk := func(target string) *http.Request { + r := httptest.NewRequest(http.MethodPost, "/", strings.NewReader("{}")) + r.Header.Set("Content-Type", "application/x-amz-json-1.0") + r.Header.Set("X-Amz-Target", target) + + return r + } + + if !h.Matches(mk(jsonTarget + "DescribeAlarms")) { + t.Fatal("Matches must claim a GraniteServiceVersion20100801 target") + } + + for _, target := range []string{ + "DynamoDB_20120810.ListTables", + "Logs_20140328.DescribeLogGroups", + "GraniteServiceVersion20100801", + "GraniteServiceVersion20100801X.DescribeAlarms", + } { + if h.Matches(mk(target)) { + t.Fatalf("Matches must not claim target %q", target) + } + } +} + +// TestJSONProtocolEveryOp drives every CloudWatch operation served on query +// and CBOR through the awsJson1_0 codec, in lifecycle order. +func TestJSONProtocolEveryOp(t *testing.T) { + ts := newJSONServer(t) + now := time.Now().Unix() + alarmARN := "" + streamARN := "" + + steps := []struct { + op string + body func() string + check func(t *testing.T, r jsonResp) + }{ + {op: "PutMetricData", body: func() string { + return fmt.Sprintf(`{"Namespace":"App","MetricData":[{"MetricName":"Latency","Value":12.5,"Unit":"Milliseconds",`+ + `"Timestamp":%d,"Dimensions":[{"Name":"Host","Value":"a"}]},`+ + `{"MetricName":"Latency","Values":[1,2],"Counts":[1,1],"Timestamp":%d.5,"Dimensions":[{"Name":"Host","Value":"b"}]}]}`, + now, now) + }}, + {op: "GetMetricStatistics", body: func() string { + return fmt.Sprintf(`{"Namespace":"App","MetricName":"Latency","Dimensions":[{"Name":"Host","Value":"a"}],`+ + `"StartTime":%d,"EndTime":%d,"Period":60,"Statistics":["Sum","Maximum"]}`, now-600, now+600) + }, check: func(t *testing.T, r jsonResp) { + dps := jsonList(t, r, "Datapoints") + if len(dps) != 1 { + t.Fatalf("want 1 datapoint: %s", r.raw) + } + + dp := dps[0].(map[string]any) + if _, ok := dp["Timestamp"].(float64); !ok { + t.Fatalf("Timestamp must be epoch seconds: %s", r.raw) + } + + if dp["Sum"] != 12.5 { + t.Fatalf("Sum = %v: %s", dp["Sum"], r.raw) + } + }}, + {op: "GetMetricData", body: func() string { + return fmt.Sprintf(`{"StartTime":%d,"EndTime":%d,"MetricDataQueries":[{"Id":"m1","MetricStat":{"Metric":`+ + `{"Namespace":"App","MetricName":"Latency","Dimensions":[{"Name":"Host","Value":"a"}]},"Period":60,"Stat":"Sum"}}]}`, + now-600, now+600) + }, check: func(t *testing.T, r jsonResp) { + res := jsonList(t, r, "MetricDataResults") + if len(res) != 1 { + t.Fatalf("want 1 result: %s", r.raw) + } + + m := res[0].(map[string]any) + ts, ok := m["Timestamps"].([]any) + if !ok || len(ts) != 1 { + t.Fatalf("Timestamps: %s", r.raw) + } + + if _, ok := ts[0].(float64); !ok { + t.Fatalf("Timestamps must be epoch seconds: %s", r.raw) + } + }}, + {op: "ListMetrics", body: func() string { return `{"Namespace":"App"}` }, check: func(t *testing.T, r jsonResp) { + if n := len(jsonList(t, r, "Metrics")); n != 2 { + t.Fatalf("want 2 metrics, got %d: %s", n, r.raw) + } + }}, + {op: "PutMetricAlarm", body: func() string { + return `{"AlarmName":"high","Namespace":"App","MetricName":"Latency","ComparisonOperator":"GreaterThanThreshold",` + + `"Threshold":100,"Period":60,"EvaluationPeriods":1,"Statistic":"Average",` + + `"Dimensions":[{"Name":"Host","Value":"a"}],"Tags":[{"Key":"env","Value":"dev"}]}` + }}, + {op: "DescribeAlarms", body: func() string { return `{"AlarmNames":["high"]}` }, check: func(t *testing.T, r jsonResp) { + alarms := jsonList(t, r, "MetricAlarms") + if len(alarms) != 1 { + t.Fatalf("want 1 alarm: %s", r.raw) + } + + a := alarms[0].(map[string]any) + if a["Threshold"] != float64(100) || a["Period"] != float64(60) { + t.Fatalf("alarm fields: %s", r.raw) + } + + if _, ok := a["StateUpdatedTimestamp"].(float64); !ok { + t.Fatalf("StateUpdatedTimestamp must be epoch seconds: %s", r.raw) + } + + alarmARN, _ = a["AlarmArn"].(string) + }}, + {op: "DescribeAlarmsForMetric", body: func() string { + return `{"Namespace":"App","MetricName":"Latency","Dimensions":[{"Name":"Host","Value":"a"}]}` + }, check: func(t *testing.T, r jsonResp) { + if n := len(jsonList(t, r, "MetricAlarms")); n != 1 { + t.Fatalf("want 1 alarm: %s", r.raw) + } + }}, + {op: "SetAlarmState", body: func() string { + return `{"AlarmName":"high","StateValue":"ALARM","StateReason":"test"}` + }}, + {op: "DescribeAlarmHistory", body: func() string { return `{"AlarmName":"high"}` }, check: func(t *testing.T, r jsonResp) { + items := jsonList(t, r, "AlarmHistoryItems") + if len(items) == 0 { + t.Fatalf("want history: %s", r.raw) + } + + if _, ok := items[0].(map[string]any)["Timestamp"].(float64); !ok { + t.Fatalf("history Timestamp must be epoch seconds: %s", r.raw) + } + }}, + {op: "DisableAlarmActions", body: func() string { return `{"AlarmNames":["high"]}` }}, + {op: "EnableAlarmActions", body: func() string { return `{"AlarmNames":["high"]}` }}, + {op: "PutCompositeAlarm", body: func() string { + return `{"AlarmName":"combo","AlarmRule":"ALARM(\"high\")"}` + }}, + {op: "TagResource", body: func() string { + return fmt.Sprintf(`{"ResourceARN":%q,"Tags":[{"Key":"team","Value":"core"}]}`, alarmARN) + }}, + {op: "ListTagsForResource", body: func() string { + return fmt.Sprintf(`{"ResourceARN":%q}`, alarmARN) + }, check: func(t *testing.T, r jsonResp) { + if n := len(jsonList(t, r, "Tags")); n != 2 { + t.Fatalf("want 2 tags: %s", r.raw) + } + }}, + {op: "UntagResource", body: func() string { + return fmt.Sprintf(`{"ResourceARN":%q,"TagKeys":["team","env"]}`, alarmARN) + }}, + {op: "ListTagsForResource", body: func() string { + return fmt.Sprintf(`{"ResourceARN":%q}`, alarmARN) + }, check: func(t *testing.T, r jsonResp) { + if n := len(jsonList(t, r, "Tags")); n != 0 { + t.Fatalf("want an empty Tags list: %s", r.raw) + } + }}, + {op: "DeleteAlarms", body: func() string { return `{"AlarmNames":["combo","high"]}` }}, + {op: "DescribeAlarms", body: func() string { return `{}` }, check: func(t *testing.T, r jsonResp) { + if n := len(jsonList(t, r, "MetricAlarms")); n != 0 { + t.Fatalf("want an empty MetricAlarms list: %s", r.raw) + } + }}, + {op: "PutDashboard", body: func() string { + return `{"DashboardName":"main","DashboardBody":"{\"widgets\":[]}"}` + }, check: func(t *testing.T, r jsonResp) { + if n := len(jsonList(t, r, "DashboardValidationMessages")); n != 0 { + t.Fatalf("want an empty validation list: %s", r.raw) + } + }}, + {op: "GetDashboard", body: func() string { return `{"DashboardName":"main"}` }, check: func(t *testing.T, r jsonResp) { + if r.body["DashboardBody"] != `{"widgets":[]}` { + t.Fatalf("DashboardBody: %s", r.raw) + } + }}, + {op: "ListDashboards", body: func() string { return `{}` }, check: func(t *testing.T, r jsonResp) { + entries := jsonList(t, r, "DashboardEntries") + if len(entries) != 1 { + t.Fatalf("want 1 dashboard: %s", r.raw) + } + + if _, ok := entries[0].(map[string]any)["LastModified"].(float64); !ok { + t.Fatalf("LastModified must be epoch seconds: %s", r.raw) + } + }}, + {op: "DeleteDashboards", body: func() string { return `{"DashboardNames":["main"]}` }}, + {op: "PutMetricStream", body: func() string { + return `{"Name":"s1","FirehoseArn":"arn:aws:firehose:us-east-1:123456789012:deliverystream/f",` + + `"RoleArn":"arn:aws:iam::123456789012:role/r","OutputFormat":"json","IncludeFilters":[{"Namespace":"App"}]}` + }, check: func(t *testing.T, r jsonResp) { + streamARN, _ = r.body["Arn"].(string) + if streamARN == "" { + t.Fatalf("Arn: %s", r.raw) + } + }}, + {op: "GetMetricStream", body: func() string { return `{"Name":"s1"}` }, check: func(t *testing.T, r jsonResp) { + if r.body["State"] != "running" { + t.Fatalf("State: %s", r.raw) + } + + if _, ok := r.body["CreationDate"].(float64); !ok { + t.Fatalf("CreationDate must be epoch seconds: %s", r.raw) + } + }}, + {op: "StopMetricStreams", body: func() string { return `{"Names":["s1"]}` }}, + {op: "StartMetricStreams", body: func() string { return `{"Names":["s1"]}` }}, + {op: "ListMetricStreams", body: func() string { return `{}` }, check: func(t *testing.T, r jsonResp) { + if n := len(jsonList(t, r, "Entries")); n != 1 { + t.Fatalf("want 1 stream: %s", r.raw) + } + }}, + {op: "DeleteMetricStream", body: func() string { return `{"Name":"s1"}` }}, + {op: "PutAnomalyDetector", body: func() string { + return `{"SingleMetricAnomalyDetector":{"Namespace":"App","MetricName":"Latency","Stat":"Average"}}` + }}, + {op: "DescribeAnomalyDetectors", body: func() string { return `{}` }, check: func(t *testing.T, r jsonResp) { + if n := len(jsonList(t, r, "AnomalyDetectors")); n != 1 { + t.Fatalf("want 1 detector: %s", r.raw) + } + }}, + {op: "DeleteAnomalyDetector", body: func() string { + return `{"SingleMetricAnomalyDetector":{"Namespace":"App","MetricName":"Latency","Stat":"Average"}}` + }}, + } + + for i, s := range steps { + r := jsonCall(t, ts, s.op, s.body()) + if r.status != http.StatusOK { + t.Fatalf("step %d %s: status %d: %s", i, s.op, r.status, r.raw) + } + + if ct := r.header.Get("Content-Type"); ct != "application/x-amz-json-1.0" { + t.Fatalf("step %d %s: Content-Type = %q", i, s.op, ct) + } + + if s.check != nil { + s.check(t, r) + } + } +} + +func TestJSONProtocolListMetricsPaging(t *testing.T) { + ts := newJSONServer(t) + + var data []string + for i := range 520 { + data = append(data, fmt.Sprintf(`{"MetricName":"M%03d","Value":1}`, i)) + } + + for start := 0; start < len(data); start += 500 { + end := min(start+500, len(data)) + body := `{"Namespace":"Paged","MetricData":[` + strings.Join(data[start:end], ",") + `]}` + + if r := jsonCall(t, ts, "PutMetricData", body); r.status != http.StatusOK { + t.Fatalf("PutMetricData: %d %s", r.status, r.raw) + } + } + + first := jsonCall(t, ts, "ListMetrics", `{"Namespace":"Paged"}`) + token, _ := first.body["NextToken"].(string) + + if len(jsonList(t, first, "Metrics")) != 500 || token == "" { + t.Fatalf("first page: want 500 metrics and a NextToken: %d %v", len(jsonList(t, first, "Metrics")), first.body["NextToken"]) + } + + second := jsonCall(t, ts, "ListMetrics", fmt.Sprintf(`{"Namespace":"Paged","NextToken":%q}`, token)) + if len(jsonList(t, second, "Metrics")) != 20 || second.body["NextToken"] != nil { + t.Fatalf("second page: %s", second.raw) + } +} + +func TestJSONProtocolErrors(t *testing.T) { + ts := newJSONServer(t) + + tests := []struct { + name string + op string + body string + status int + errType string + queryError string + }{ + { + name: "dashboard not found", op: "GetDashboard", body: `{"DashboardName":"nope"}`, + status: http.StatusNotFound, errType: "ResourceNotFound", queryError: "ResourceNotFound;Sender", + }, + { + name: "invalid parameter value", op: "SetAlarmState", body: `{"AlarmName":"x","StateValue":"BOGUS","StateReason":"r"}`, + status: http.StatusBadRequest, errType: "ValidationException", queryError: "ValidationError;Sender", + }, + { + name: "missing parameter", op: "PutMetricAlarm", body: `{"AlarmName":"a"}`, + status: http.StatusBadRequest, + }, + { + name: "bad next token", op: "DescribeAlarms", body: `{"NextToken":"!!!"}`, + status: http.StatusBadRequest, errType: "InvalidNextToken", queryError: "InvalidNextToken;Sender", + }, + { + name: "list metrics bad next token", op: "ListMetrics", body: `{"NextToken":"!!!"}`, + status: http.StatusBadRequest, errType: "InvalidParameterValueException", queryError: "InvalidParameterValue;Sender", + }, + { + name: "anomaly detector not found", op: "DeleteAnomalyDetector", + body: `{"SingleMetricAnomalyDetector":{"Namespace":"App","MetricName":"Nope","Stat":"Average"}}`, + status: http.StatusNotFound, errType: "ResourceNotFoundException", queryError: "ResourceNotFoundException;Sender", + }, + { + name: "malformed json", op: "DescribeAlarms", body: `{"AlarmNames":`, + status: http.StatusBadRequest, errType: "SerializationException", + }, + { + name: "unknown operation", op: "NoSuchOperation", body: `{}`, + status: http.StatusBadRequest, errType: "UnknownOperationException", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + r := jsonCall(t, ts, tt.op, tt.body) + if r.status != tt.status { + t.Fatalf("status = %d, want %d: %s", r.status, tt.status, r.raw) + } + + errType, _ := r.body["__type"].(string) + if errType == "" || r.header.Get("X-Amzn-Errortype") != errType { + t.Fatalf("__type %q and X-Amzn-Errortype %q must match: %s", errType, r.header.Get("X-Amzn-Errortype"), r.raw) + } + + if tt.errType != "" && errType != tt.errType { + t.Fatalf("__type = %q, want %q", errType, tt.errType) + } + + if tt.queryError != "" && r.header.Get("X-Amzn-Query-Error") != tt.queryError { + t.Fatalf("X-Amzn-Query-Error = %q, want %q", r.header.Get("X-Amzn-Query-Error"), tt.queryError) + } + + if msg, _ := r.body["message"].(string); msg == "" { + t.Fatalf("message must be set: %s", r.raw) + } + }) + } +} + +// TestJSONProtocolEmptyCollections checks that list outputs come back as JSON +// lists, not null, when there is nothing to return. +func TestJSONProtocolEmptyCollections(t *testing.T) { + ts := newJSONServer(t) + + for op, key := range map[string]string{ + "ListMetrics": "Metrics", + "DescribeAlarms": "MetricAlarms", + "ListDashboards": "DashboardEntries", + } { + r := jsonCall(t, ts, op, `{}`) + if r.status != http.StatusOK { + t.Fatalf("%s: %d %s", op, r.status, r.raw) + } + + if n := len(jsonList(t, r, key)); n != 0 { + t.Fatalf("%s: want an empty %s: %s", op, key, r.raw) + } + } + + // An empty request body is the same as {}. + if r := jsonCall(t, ts, "ListMetrics", ""); r.status != http.StatusOK { + t.Fatalf("empty body: %d %s", r.status, r.raw) + } +} diff --git a/server/aws/cloudwatch/json_transcode_test.go b/server/aws/cloudwatch/json_transcode_test.go new file mode 100644 index 000000000..b3178c874 --- /dev/null +++ b/server/aws/cloudwatch/json_transcode_test.go @@ -0,0 +1,81 @@ +package cloudwatch + +import ( + "testing" + "time" + + "github.com/fxamacker/cbor/v2" +) + +// TestCBORPayloadToJSON covers the awsJson1_0 value rules on a response: +// timestamps as epoch seconds, blobs as base64, doubles kept as doubles, +// empty lists and maps kept, and null or omitted members left out. +func TestCBORPayloadToJSON(t *testing.T) { + type out struct { + When time.Time `cbor:"When"` + Image []byte `cbor:"MetricWidgetImage"` + Empty []string `cbor:"Empty"` + Stats map[string]float64 `cbor:"Stats"` + Skipped string `cbor:"Skipped,omitempty"` + Nil *float64 `cbor:"Nil"` + Sum float64 `cbor:"Sum"` + Period int `cbor:"Period"` + } + + got, err := cborPayloadToJSON(out{ + When: time.Unix(1700000000, 250*int64(time.Millisecond)), + Image: []byte("png-bytes"), + Empty: []string{}, + Stats: map[string]float64{}, + Sum: 20, + Period: 60, + }) + if err != nil { + t.Fatalf("cborPayloadToJSON: %v", err) + } + + want := `{"Empty":[],"MetricWidgetImage":"cG5nLWJ5dGVz","Period":60,"Stats":{},"Sum":20.0,"When":1700000000.25}` + if string(got) != want { + t.Fatalf("got %s\nwant %s", got, want) + } +} + +// TestJSONToCBOR covers the request rules: integer literals decode into int +// fields, fractional numbers into floats, epoch seconds into time.Time. +func TestJSONToCBOR(t *testing.T) { + type in struct { + Period int `cbor:"Period"` + Value float64 `cbor:"Value"` + Whole float64 `cbor:"Whole"` + Start *time.Time `cbor:"Start"` + End time.Time `cbor:"End"` + } + + body, err := jsonToCBOR([]byte(`{"Period":60,"Value":1.5,"Whole":3,"Start":1700000000,"End":1700000000.5}`)) + if err != nil { + t.Fatalf("jsonToCBOR: %v", err) + } + + var v in + if err := cbor.Unmarshal(body, &v); err != nil { + t.Fatalf("decode: %v", err) + } + + if v.Period != 60 || v.Value != 1.5 || v.Whole != 3 { + t.Fatalf("numbers: %+v", v) + } + + if v.Start == nil || v.Start.Unix() != 1700000000 || v.End.UnixMilli() != 1700000000500 { + t.Fatalf("timestamps: %+v", v) + } + + for _, bad := range []string{`[1]`, `{"a":`, `"x"`} { + if _, err := jsonToCBOR([]byte(bad)); err == nil { + t.Fatalf("jsonToCBOR(%s) must fail", bad) + } + } + + if _, err := jsonToCBOR(nil); err != nil { + t.Fatalf("empty body: %v", err) + } +} diff --git a/server/aws/cloudwatch/request_compression_test.go b/server/aws/cloudwatch/request_compression_test.go new file mode 100644 index 000000000..3d188bed7 --- /dev/null +++ b/server/aws/cloudwatch/request_compression_test.go @@ -0,0 +1,176 @@ +package cloudwatch_test + +import ( + "bytes" + "compress/gzip" + "fmt" + "io" + "net/http" + "net/url" + "strconv" + "strings" + "testing" + + "github.com/aws/aws-sdk-go-v2/aws" + awscw "github.com/aws/aws-sdk-go-v2/service/cloudwatch" + cwtypes "github.com/aws/aws-sdk-go-v2/service/cloudwatch/types" +) + +const compressedMetrics = 600 + +func gzipBytes(t *testing.T, b []byte) []byte { + t.Helper() + + var buf bytes.Buffer + + zw := gzip.NewWriter(&buf) + if _, err := zw.Write(b); err != nil { + t.Fatal(err) + } + + if err := zw.Close(); err != nil { + t.Fatal(err) + } + + return buf.Bytes() +} + +func postCompressed(t *testing.T, target string, headers map[string]string, body []byte) (int, string) { + t.Helper() + + req, err := http.NewRequest(http.MethodPost, target, bytes.NewReader(body)) //nolint:noctx // test request + if err != nil { + t.Fatal(err) + } + + for k, v := range headers { + req.Header.Set(k, v) + } + + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + + raw, _ := io.ReadAll(resp.Body) + + return resp.StatusCode, string(raw) +} + +func listedMetrics(t *testing.T, ts string, namespace string) int { + t.Helper() + + n := 0 + token := "" + + for { + body := fmt.Sprintf(`{"Namespace":%q,"NextToken":%q}`, namespace, token) + + status, raw := postCompressed(t, ts, map[string]string{ + "Content-Type": "application/x-amz-json-1.0", "X-Amz-Target": jsonTarget + "ListMetrics", + }, []byte(body)) + if status != http.StatusOK { + t.Fatalf("ListMetrics: %d %s", status, raw) + } + + n += strings.Count(raw, `"MetricName"`) + + i := strings.Index(raw, `"NextToken":"`) + if i < 0 { + return n + } + + rest := raw[i+len(`"NextToken":"`):] + token = rest[:strings.Index(rest, `"`)] + } +} + +// TestCompressedPutMetricData covers the gzip request body the SDKs send for +// a large PutMetricData, on all three protocols. +func TestCompressedPutMetricData(t *testing.T) { + ts := newJSONServer(t) + + t.Run("json", func(t *testing.T) { + var data []string + for i := range compressedMetrics { + data = append(data, fmt.Sprintf(`{"MetricName":"J%04d","Value":1}`, i)) + } + + body := gzipBytes(t, []byte(`{"Namespace":"GzJSON","MetricData":[`+strings.Join(data, ",")+`]}`)) + + status, raw := postCompressed(t, ts.URL, map[string]string{ + "Content-Type": "application/x-amz-json-1.0", "X-Amz-Target": jsonTarget + "PutMetricData", + "Content-Encoding": "gzip", + }, body) + if status != http.StatusOK { + t.Fatalf("PutMetricData: %d %s", status, raw) + } + + if n := listedMetrics(t, ts.URL, "GzJSON"); n != compressedMetrics { + t.Fatalf("listed %d metrics, want %d", n, compressedMetrics) + } + }) + + t.Run("query", func(t *testing.T) { + form := url.Values{"Action": {"PutMetricData"}, "Version": {"2010-08-01"}, "Namespace": {"GzQuery"}} + for i := range compressedMetrics { + p := "MetricData.member." + strconv.Itoa(i+1) + "." + form.Set(p+"MetricName", fmt.Sprintf("Q%04d", i)) + form.Set(p+"Value", "1") + } + + status, raw := postCompressed(t, ts.URL, map[string]string{ + "Content-Type": "application/x-www-form-urlencoded", "Authorization": monitoringAuth, + "Content-Encoding": "gzip", + }, gzipBytes(t, []byte(form.Encode()))) + if status != http.StatusOK { + t.Fatalf("PutMetricData: %d %s", status, raw) + } + + if n := listedMetrics(t, ts.URL, "GzQuery"); n != compressedMetrics { + t.Fatalf("listed %d metrics, want %d", n, compressedMetrics) + } + }) + + t.Run("cbor sdk", func(t *testing.T) { + client, ctx := newCWClient(t) + + data := make([]cwtypes.MetricDatum, 0, compressedMetrics) + for i := range compressedMetrics { + data = append(data, cwtypes.MetricDatum{MetricName: aws.String(fmt.Sprintf("C%04d", i)), Value: aws.Float64(1)}) + } + + if _, err := client.PutMetricData(ctx, &awscw.PutMetricDataInput{ + Namespace: aws.String("GzCBOR"), MetricData: data, + }); err != nil { + t.Fatalf("PutMetricData: %v", err) + } + + n := 0 + + p := awscw.NewListMetricsPaginator(client, &awscw.ListMetricsInput{Namespace: aws.String("GzCBOR")}) + for p.HasMorePages() { + page, err := p.NextPage(ctx) + if err != nil { + t.Fatalf("ListMetrics: %v", err) + } + + n += len(page.Metrics) + } + + if n != compressedMetrics { + t.Fatalf("listed %d metrics, want %d", n, compressedMetrics) + } + }) + + t.Run("bad gzip", func(t *testing.T) { + status, raw := postCompressed(t, ts.URL, map[string]string{ + "Content-Type": "application/x-amz-json-1.0", "X-Amz-Target": jsonTarget + "PutMetricData", + "Content-Encoding": "gzip", + }, []byte("not gzip")) + if status != http.StatusBadRequest || !strings.Contains(raw, "SerializationException") { + t.Fatalf("bad gzip: %d %s", status, raw) + } + }) +} diff --git a/server/aws/cloudwatch/request_encoding.go b/server/aws/cloudwatch/request_encoding.go new file mode 100644 index 000000000..910c8f563 --- /dev/null +++ b/server/aws/cloudwatch/request_encoding.go @@ -0,0 +1,46 @@ +package cloudwatch + +import ( + "compress/gzip" + "io" + "net/http" + "strings" +) + +// PutMetricData carries the requestCompression trait, so the SDKs gzip its +// body once it passes 10 KiB and send Content-Encoding: gzip. That happens +// on every protocol, so the body is unwrapped before any codec reads it. +func decodeRequestBody(r *http.Request) { + if !strings.EqualFold(strings.TrimSpace(r.Header.Get("Content-Encoding")), "gzip") { + return + } + + r.Body = &gzipBody{src: r.Body} + r.Header.Del("Content-Encoding") + r.ContentLength = -1 +} + +// gzipBody opens the gzip stream on the first Read, so a bad stream shows up +// as a read error in the codec that reads the body, which answers it in its +// own wire format. +type gzipBody struct { + src io.ReadCloser + zr *gzip.Reader +} + +func (g *gzipBody) Read(p []byte) (int, error) { + if g.zr == nil { + zr, err := gzip.NewReader(g.src) + if err != nil { + return 0, err + } + + g.zr = zr + } + + return g.zr.Read(p) +} + +func (g *gzipBody) Close() error { + return g.src.Close() +} diff --git a/server/aws/cloudwatch_json_dispatch_test.go b/server/aws/cloudwatch_json_dispatch_test.go new file mode 100644 index 000000000..eb3121deb --- /dev/null +++ b/server/aws/cloudwatch_json_dispatch_test.go @@ -0,0 +1,39 @@ +package aws_test + +import ( + "net/http" + "strings" + "testing" + + "github.com/stretchr/testify/assert" +) + +// TestCloudWatchJSONDispatch checks that the full AWS server routes a +// GraniteServiceVersion20100801 awsJson1_0 request to CloudWatch, and that +// CloudWatch does not take the requests of other awsJson services. +func TestCloudWatchJSONDispatch(t *testing.T) { + ts := fullAWSServer(t) + + tests := []struct { + name string + target string + want string + }{ + {name: "cloudwatch", target: "GraniteServiceVersion20100801.DescribeAlarms", want: `"MetricAlarms":[]`}, + {name: "dynamodb", target: "DynamoDB_20120810.ListTables", want: `"TableNames"`}, + {name: "cloudwatch logs", target: "Logs_20140328.DescribeLogGroups", want: `"logGroups"`}, + {name: "kinesis", target: "Kinesis_20131202.ListStreams", want: `"StreamNames"`}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + status, body := doRequest(t, ts, http.MethodPost, "/", "{}", map[string]string{ + "Content-Type": "application/x-amz-json-1.0", + "X-Amz-Target": tt.target, + }) + + assert.Equal(t, http.StatusOK, status, body) + assert.True(t, strings.Contains(body, tt.want), "want %s in %s", tt.want, body) + }) + } +} From bbacc67572743b958fb4c47c5838483abd4e218b Mon Sep 17 00:00:00 2001 From: Nitin Kumar Date: Sun, 27 Sep 2026 17:43:33 +0530 Subject: [PATCH 2/3] fix(cloudwatch): authz for the JSON target and request validation parity (CW-X13) Maps the Granite target to cloudwatch in the authz gate, tightens the JSON codec, and fixes tag, dimension, alarm period and statistic, dashboard body and metric stream error codes on every protocol. Adds AlarmConfigurationUpdatedTimestamp to MetricAlarm output. --- providers/aws/cloudwatch/cloudwatch.go | 5 +- server/aws/authzgate.go | 1 + server/aws/authzgate_cloudwatch_test.go | 110 ++++++++ .../aws/cloudwatch/core_alarm_statistics.go | 38 +++ server/aws/cloudwatch/core_alarms.go | 4 + server/aws/cloudwatch/core_metrics.go | 8 + server/aws/cloudwatch/dashboards.go | 17 +- server/aws/cloudwatch/json_protocol.go | 44 +++- server/aws/cloudwatch/json_protocol_test.go | 4 + server/aws/cloudwatch/json_transcode_test.go | 9 +- server/aws/cloudwatch/metric_data_ops.go | 90 ++----- server/aws/cloudwatch/metric_streams.go | 16 +- server/aws/cloudwatch/ops.go | 7 +- server/aws/cloudwatch/query.go | 7 +- server/aws/cloudwatch/query_dashboards.go | 2 +- server/aws/cloudwatch/query_metric_streams.go | 34 ++- .../cloudwatch/query_metric_streams_test.go | 9 +- .../aws/cloudwatch/validation_parity_test.go | 243 ++++++++++++++++++ services/monitoring/driver/driver.go | 3 + 19 files changed, 524 insertions(+), 127 deletions(-) create mode 100644 server/aws/authzgate_cloudwatch_test.go create mode 100644 server/aws/cloudwatch/validation_parity_test.go diff --git a/providers/aws/cloudwatch/cloudwatch.go b/providers/aws/cloudwatch/cloudwatch.go index f49f29fa5..79ac744a4 100644 --- a/providers/aws/cloudwatch/cloudwatch.go +++ b/providers/aws/cloudwatch/cloudwatch.go @@ -657,8 +657,9 @@ func toAlarmInfo(a *alarmData) driver.AlarmInfo { Metrics: metricmath.Clone(a.Metrics), ThresholdMetricID: a.ThresholdMetricID, - EvaluateLowSampleCountPercentile: a.EvaluateLowSampleCountPercentile, - EvaluationWindow: cloneWindow(a.EvaluationWindow), + AlarmConfigurationUpdatedTimestamp: a.ConfigUpdatedAt, + EvaluateLowSampleCountPercentile: a.EvaluateLowSampleCountPercentile, + EvaluationWindow: cloneWindow(a.EvaluationWindow), } } diff --git a/server/aws/authzgate.go b/server/aws/authzgate.go index 8abcd69de..cd6e68cf6 100644 --- a/server/aws/authzgate.go +++ b/server/aws/authzgate.go @@ -62,6 +62,7 @@ var jsonRPCServiceByTarget = map[string]string{ "Route53Resolver.": "route53resolver", "AWSEvents.": "events", "Logs_20140328.": "logs", + "GraniteServiceVersion20100801.": "cloudwatch", "SageMaker.": "sagemaker", "secretsmanager.": "secretsmanager", "KeyspacesService.": "cassandra", diff --git a/server/aws/authzgate_cloudwatch_test.go b/server/aws/authzgate_cloudwatch_test.go new file mode 100644 index 000000000..f16044ac6 --- /dev/null +++ b/server/aws/authzgate_cloudwatch_test.go @@ -0,0 +1,110 @@ +package aws + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/aws/aws-sdk-go-v2/aws" + v4 "github.com/aws/aws-sdk-go-v2/aws/signer/v4" + + cloudemu "github.com/stackshy/cloudemu/v2" + iamdriver "github.com/stackshy/cloudemu/v2/services/iam/driver" +) + +// cloudWatchJSONProbe answers any CloudWatch awsJson1_0 request with 200. +type cloudWatchJSONProbe struct{} + +func (cloudWatchJSONProbe) Matches(r *http.Request) bool { + return strings.HasPrefix(r.Header.Get("X-Amz-Target"), "GraniteServiceVersion20100801.") +} + +func (cloudWatchJSONProbe) ServeHTTP(w http.ResponseWriter, _ *http.Request) { + _, _ = io.WriteString(w, "{}") +} + +func signedCloudWatchJSONRequest(t *testing.T, url, op string, creds aws.Credentials) *http.Request { + t.Helper() + + body := `{}` + + req, err := http.NewRequestWithContext(context.Background(), http.MethodPost, url, strings.NewReader(body)) + if err != nil { + t.Fatalf("new request: %v", err) + } + + req.Header.Set("X-Amz-Target", "GraniteServiceVersion20100801."+op) + req.Header.Set("Content-Type", "application/x-amz-json-1.0") + + sum := sha256.Sum256([]byte(body)) + if err := v4.NewSigner().SignHTTP( + context.Background(), creds, req, hex.EncodeToString(sum[:]), "monitoring", "us-east-1", time.Now(), + ); err != nil { + t.Fatalf("sign: %v", err) + } + + return req +} + +// TestAuthzGateCloudWatchJSON checks that the authorization gate maps a +// GraniteServiceVersion20100801 target to the cloudwatch IAM service, so an +// allow on cloudwatch:* lets boto3's CloudWatch calls through and an explicit +// deny still blocks its action. +func TestAuthzGateCloudWatchJSON(t *testing.T) { + cloud := cloudemu.NewAWS() + ctx := context.Background() + + if _, err := cloud.IAM.CreateUser(ctx, iamdriver.UserConfig{Name: "cw"}); err != nil { + t.Fatalf("CreateUser: %v", err) + } + + doc := `{"Version":"2012-10-17","Statement":[` + + `{"Effect":"Allow","Action":"cloudwatch:*","Resource":"*"},` + + `{"Effect":"Deny","Action":"cloudwatch:DeleteAlarms","Resource":"*"}]}` + + pol, err := cloud.IAM.CreatePolicy(ctx, iamdriver.PolicyConfig{Name: "cwpol", PolicyDocument: doc}) + if err != nil { + t.Fatalf("CreatePolicy: %v", err) + } + + if err := cloud.IAM.AttachUserPolicy(ctx, "cw", pol.ARN); err != nil { + t.Fatalf("AttachUserPolicy: %v", err) + } + + ak, err := cloud.IAM.CreateAccessKey(ctx, iamdriver.AccessKeyConfig{UserName: "cw"}) + if err != nil { + t.Fatalf("CreateAccessKey: %v", err) + } + + srv := New(Drivers{IAM: cloud.IAM, AccountID: "123456789012", Region: "us-east-1", EnforceAuth: true}) + srv.Register(cloudWatchJSONProbe{}) + + ts := httptest.NewServer(srv) + defer ts.Close() + + creds := aws.Credentials{AccessKeyID: ak.AccessKeyID, SecretAccessKey: ak.SecretAccessKey} + + for op, want := range map[string]int{ + "DescribeAlarms": http.StatusOK, + "PutMetricData": http.StatusOK, + "DeleteAlarms": http.StatusForbidden, + } { + resp, err := http.DefaultClient.Do(signedCloudWatchJSONRequest(t, ts.URL+"/", op, creds)) + if err != nil { + t.Fatalf("%s: %v", op, err) + } + + raw, _ := io.ReadAll(resp.Body) + resp.Body.Close() + + if resp.StatusCode != want { + t.Fatalf("%s: status %d, want %d: %s", op, resp.StatusCode, want, raw) + } + } +} diff --git a/server/aws/cloudwatch/core_alarm_statistics.go b/server/aws/cloudwatch/core_alarm_statistics.go index e0cde2069..8ba1e8af7 100644 --- a/server/aws/cloudwatch/core_alarm_statistics.go +++ b/server/aws/cloudwatch/core_alarm_statistics.go @@ -30,6 +30,10 @@ func validateAlarmStatistic(cfg *mondriver.AlarmConfig) error { return newWireError(errInvalidParameterCombo, "Exactly one of Statistic or ExtendedStatistic must be specified.") case cfg.Stat == "" && cfg.ExtendedStatistic == "": return newWireError(errValidation, "Exactly one of Statistic or ExtendedStatistic must be specified.") + case cfg.Stat != "" && !validStatistics[cfg.Stat]: + return newWireError(errValidation, "1 validation error detected: Value '"+cfg.Stat+ + "' at 'statistic' failed to satisfy constraint: Member must satisfy enum value set: "+ + "[Maximum, SampleCount, Sum, Minimum, Average]") case cfg.ExtendedStatistic != "": if _, err := alarmeval.ParseExtendedStatistic(cfg.ExtendedStatistic); err != nil { return newWireError(errValidation, "The value "+cfg.ExtendedStatistic+ @@ -40,6 +44,40 @@ func validateAlarmStatistic(cfg *mondriver.AlarmConfig) error { return nil } +// validStatistics is the closed Statistic enum. +// +//nolint:gochecknoglobals // fixed lookup table for a closed enum. +var validStatistics = map[string]bool{ + statSampleCount: true, statAverage: true, statSum: true, statMinimum: true, statMaximum: true, +} + +// maxAlarmLookback is how far back, in seconds, EvaluationPeriods * Period +// may reach. +const maxAlarmLookback = 86400 + +// validateAlarmPeriods checks Period and DatapointsToAlarm. A single-metric +// alarm's Period is 10, 20, 30 or a multiple of 60, and the alarm cannot +// look back more than a day. +func validateAlarmPeriods(cfg *mondriver.AlarmConfig) error { + if cfg.DatapointsToAlarm > 0 && cfg.EvaluationPeriods > 0 && cfg.DatapointsToAlarm > cfg.EvaluationPeriods { + return newWireError(errValidation, "DatapointsToAlarm must be less than or equal to EvaluationPeriods.") + } + + if len(cfg.Metrics) > 0 || cfg.Period == 0 { + return nil + } + + switch p := cfg.Period; { + case !validPeriod(p): + return newWireError(errValidation, "Period must be 10, 20, 30 or a multiple of 60") + case p*max(cfg.EvaluationPeriods, 1) > maxAlarmLookback: + return newWireError(errValidation, "Metrics cannot be checked across more than a day "+ + "(EvaluationPeriods * Period must be <= 86400)") + } + + return nil +} + // validateLowSample checks EvaluateLowSampleCountPercentile. func validateLowSample(v string) error { switch v { diff --git a/server/aws/cloudwatch/core_alarms.go b/server/aws/cloudwatch/core_alarms.go index 34118cccd..cfc095d9a 100644 --- a/server/aws/cloudwatch/core_alarms.go +++ b/server/aws/cloudwatch/core_alarms.go @@ -65,6 +65,10 @@ func (h *Handler) putMetricAlarmCore( return err } + if err := validateAlarmPeriods(cfg); err != nil { + return err + } + if err := validateLowSample(cfg.EvaluateLowSampleCountPercentile); err != nil { return err } diff --git a/server/aws/cloudwatch/core_metrics.go b/server/aws/cloudwatch/core_metrics.go index e0a876a86..5c3d69227 100644 --- a/server/aws/cloudwatch/core_metrics.go +++ b/server/aws/cloudwatch/core_metrics.go @@ -481,6 +481,9 @@ func setStat(dp *datapoint, stat requestedStat, value float64) { } } +// maxDatumDimensions is the most dimensions one metric can have. +const maxDatumDimensions = 30 + // putMetricDataCore validates every datum and then stores them. One bad // datum rejects the whole request, so nothing is stored. func (h *Handler) putMetricDataCore(ctx context.Context, in *putMetricDataInput) error { @@ -489,6 +492,11 @@ func (h *Handler) putMetricDataCore(ctx context.Context, in *putMetricDataInput) for i := range in.MetricData { d := &in.MetricData[i] + if len(d.Dimensions) > maxDatumDimensions { + return newWireError(errInvalidParameterValue, "The collection MetricData.member."+strconv.Itoa(i+1)+ + ".Dimensions must not have a size greater than "+strconv.Itoa(maxDatumDimensions)+".") + } + if d.Unit != "" && !alarmeval.ValidUnit(d.Unit) { return newWireError(errInvalidParameterValue, "The parameter MetricData.member."+strconv.Itoa(i+1)+ ".Unit must be a value in the set ["+strings.Join(alarmeval.Units(), ", ")+"]") diff --git a/server/aws/cloudwatch/dashboards.go b/server/aws/cloudwatch/dashboards.go index c010393dd..099f9da71 100644 --- a/server/aws/cloudwatch/dashboards.go +++ b/server/aws/cloudwatch/dashboards.go @@ -8,6 +8,7 @@ package cloudwatch import ( "context" + "encoding/json" "net/http" "time" @@ -24,6 +25,20 @@ type dashboardStore interface { DeleteDashboards(ctx context.Context, names []string) error } +// errInvalidParameterInput is the query code of DashboardInvalidInputError. +const errInvalidParameterInput = "InvalidParameterInput" + +// putDashboardCore stores a dashboard after checking that its body is a JSON +// object, which is what PutDashboard rejects with InvalidParameterInput. +func putDashboardCore(ctx context.Context, store dashboardStore, name, body string) error { + var doc map[string]any + if err := json.Unmarshal([]byte(body), &doc); err != nil || doc == nil { + return newWireError(errInvalidParameterInput, "The field DashboardBody must be a valid JSON object") + } + + return store.PutDashboard(ctx, name, body) +} + type putDashboardInput struct { DashboardName string `cbor:"DashboardName"` DashboardBody string `cbor:"DashboardBody"` @@ -53,7 +68,7 @@ func (h *Handler) putDashboard(w http.ResponseWriter, r *http.Request, body []by return } - if err := store.PutDashboard(r.Context(), in.DashboardName, in.DashboardBody); err != nil { + if err := putDashboardCore(r.Context(), store, in.DashboardName, in.DashboardBody); err != nil { writeDriverErr(w, err) return } diff --git a/server/aws/cloudwatch/json_protocol.go b/server/aws/cloudwatch/json_protocol.go index b41470285..843412985 100644 --- a/server/aws/cloudwatch/json_protocol.go +++ b/server/aws/cloudwatch/json_protocol.go @@ -39,7 +39,10 @@ const ( msPerSecond = 1e3 ) -var errNotJSONObject = errors.New("request body must be a JSON object") +var ( + errNotJSONObject = errors.New("request body must be a JSON object") + errTrailingJSON = errors.New("invalid JSON: unexpected data after the top-level object") +) // jsonErrorShapes maps the query error codes the cores return to the error // shape names of the CloudWatch model. CloudWatch is awsQueryCompatible, so @@ -143,9 +146,10 @@ func (j *jsonWriter) writeError(status int, code, msg string) { } // jsonToCBOR re-encodes a JSON request body as CBOR for the shared decoders. -// Integer literals become CBOR integers so they decode into int fields, and -// timestamps (epoch seconds numbers) decode into time.Time as untagged -// numbers. An empty body is an empty input. +// Integral numbers (60, 60.0, 6e1) become CBOR integers so they decode into +// int fields as well as float ones, and timestamps (epoch seconds numbers) +// decode into time.Time as untagged numbers. An empty body is an empty input, +// and anything after the JSON object is rejected. func jsonToCBOR(raw []byte) ([]byte, error) { if len(bytes.TrimSpace(raw)) == 0 { raw = []byte("{}") @@ -163,6 +167,10 @@ func jsonToCBOR(raw []byte) ([]byte, error) { return nil, errNotJSONObject } + if _, err := dec.Token(); !errors.Is(err, io.EOF) { + return nil, errTrailingJSON + } + conv, err := jsonNumbersToCBOR(v) if err != nil { return nil, err @@ -192,19 +200,29 @@ func jsonNumbersToCBOR(v any) (any, error) { t[i] = c } case json.Number: - if i, err := strconv.ParseInt(t.String(), 10, 64); err == nil { - return i, nil - } + return jsonNumberToCBOR(t) + } - f, err := t.Float64() - if err != nil { - return nil, fmt.Errorf("invalid number %q: %w", t, err) - } + return v, nil +} - return f, nil +// jsonNumberToCBOR returns an int64 for an integral number in int64 range and +// a float64 otherwise. +func jsonNumberToCBOR(n json.Number) (any, error) { + if i, err := strconv.ParseInt(n.String(), 10, 64); err == nil { + return i, nil } - return v, nil + f, err := n.Float64() + if err != nil { + return nil, fmt.Errorf("invalid number %q: %w", n, err) + } + + if f == math.Trunc(f) && f >= math.MinInt64 && f < math.MaxInt64 { + return int64(f), nil + } + + return f, nil } // cborDecMode decodes a CBOR response back to generic values, with string map diff --git a/server/aws/cloudwatch/json_protocol_test.go b/server/aws/cloudwatch/json_protocol_test.go index 4aeeb7bd9..1a9f45f89 100644 --- a/server/aws/cloudwatch/json_protocol_test.go +++ b/server/aws/cloudwatch/json_protocol_test.go @@ -384,6 +384,10 @@ func TestJSONProtocolErrors(t *testing.T) { name: "malformed json", op: "DescribeAlarms", body: `{"AlarmNames":`, status: http.StatusBadRequest, errType: "SerializationException", }, + { + name: "trailing data", op: "DescribeAlarms", body: `{"AlarmNames":[]} {"x":1}`, + status: http.StatusBadRequest, errType: "SerializationException", + }, { name: "unknown operation", op: "NoSuchOperation", body: `{}`, status: http.StatusBadRequest, errType: "UnknownOperationException", diff --git a/server/aws/cloudwatch/json_transcode_test.go b/server/aws/cloudwatch/json_transcode_test.go index b3178c874..c594c6869 100644 --- a/server/aws/cloudwatch/json_transcode_test.go +++ b/server/aws/cloudwatch/json_transcode_test.go @@ -40,8 +40,9 @@ func TestCBORPayloadToJSON(t *testing.T) { } } -// TestJSONToCBOR covers the request rules: integer literals decode into int -// fields, fractional numbers into floats, epoch seconds into time.Time. +// TestJSONToCBOR covers the request rules: integral numbers (60.0 too) decode +// into int fields, fractional numbers into floats, epoch seconds into +// time.Time, and trailing data is rejected. func TestJSONToCBOR(t *testing.T) { type in struct { Period int `cbor:"Period"` @@ -51,7 +52,7 @@ func TestJSONToCBOR(t *testing.T) { End time.Time `cbor:"End"` } - body, err := jsonToCBOR([]byte(`{"Period":60,"Value":1.5,"Whole":3,"Start":1700000000,"End":1700000000.5}`)) + body, err := jsonToCBOR([]byte(`{"Period":60.0,"Value":1.5,"Whole":3,"Start":1700000000,"End":1700000000.5}`)) if err != nil { t.Fatalf("jsonToCBOR: %v", err) } @@ -69,7 +70,7 @@ func TestJSONToCBOR(t *testing.T) { t.Fatalf("timestamps: %+v", v) } - for _, bad := range []string{`[1]`, `{"a":`, `"x"`} { + for _, bad := range []string{`[1]`, `{"a":`, `"x"`, `{"a":1} {"b":2}`, `{"a":1}x`, `{"a":1e999}`} { if _, err := jsonToCBOR([]byte(bad)); err == nil { t.Fatalf("jsonToCBOR(%s) must fail", bad) } diff --git a/server/aws/cloudwatch/metric_data_ops.go b/server/aws/cloudwatch/metric_data_ops.go index d377c420e..676571030 100644 --- a/server/aws/cloudwatch/metric_data_ops.go +++ b/server/aws/cloudwatch/metric_data_ops.go @@ -9,6 +9,7 @@ package cloudwatch import ( "context" "encoding/json" + "errors" "net/http" "strings" "time" @@ -344,31 +345,8 @@ func (h *Handler) tagResource(w http.ResponseWriter, r *http.Request, body []byt return } - if name, ok := metricStreamNameFromARN(in.ResourceARN); ok { - tagger, ok := h.monitoring.(metricStreamTagger) - if !ok { - writeCBORError(w, http.StatusBadRequest, "UnknownOperationException", "tagging not supported") - return - } - - if err := tagger.AddMetricStreamTags(r.Context(), name, tagsToMap(in.Tags)); err != nil { - writeDriverErr(w, err) - return - } - - writeCBORResponse(w, struct{}{}) - - return - } - - tagger, ok := h.monitoring.(alarmTagger) - if !ok { - writeCBORError(w, http.StatusBadRequest, "UnknownOperationException", "tagging not supported") - return - } - - if err := tagger.AddAlarmTags(r.Context(), alarmNameFromARN(in.ResourceARN), tagsToMap(in.Tags)); err != nil { - writeDriverErr(w, err) + if err := h.addResourceTagsByARN(r.Context(), in.ResourceARN, tagsToMap(in.Tags)); err != nil { + writeTagRouteErr(w, err) return } @@ -382,31 +360,8 @@ func (h *Handler) untagResource(w http.ResponseWriter, r *http.Request, body []b return } - if name, ok := metricStreamNameFromARN(in.ResourceARN); ok { - tagger, ok := h.monitoring.(metricStreamTagger) - if !ok { - writeCBORError(w, http.StatusBadRequest, "UnknownOperationException", "tagging not supported") - return - } - - if err := tagger.RemoveMetricStreamTags(r.Context(), name, in.TagKeys); err != nil { - writeDriverErr(w, err) - return - } - - writeCBORResponse(w, struct{}{}) - - return - } - - tagger, ok := h.monitoring.(alarmTagger) - if !ok { - writeCBORError(w, http.StatusBadRequest, "UnknownOperationException", "tagging not supported") - return - } - - if err := tagger.RemoveAlarmTags(r.Context(), alarmNameFromARN(in.ResourceARN), in.TagKeys); err != nil { - writeDriverErr(w, err) + if err := h.removeResourceTagsByARN(r.Context(), in.ResourceARN, in.TagKeys); err != nil { + writeTagRouteErr(w, err) return } @@ -420,37 +375,24 @@ func (h *Handler) listTagsForResource(w http.ResponseWriter, r *http.Request, bo return } - if name, ok := metricStreamNameFromARN(in.ResourceARN); ok { - tagger, ok := h.monitoring.(metricStreamTagger) - if !ok { - writeCBORError(w, http.StatusBadRequest, "UnknownOperationException", "tagging not supported") - return - } - - tags, err := tagger.MetricStreamTags(r.Context(), name) - if err != nil { - writeDriverErr(w, err) - return - } - - writeCBORResponse(w, listTagsForResourceOutput{Tags: mapToTags(tags)}) - + tags, err := h.resourceTagsByARN(r.Context(), in.ResourceARN) + if err != nil { + writeTagRouteErr(w, err) return } - tagger, ok := h.monitoring.(alarmTagger) - if !ok { - writeCBORError(w, http.StatusBadRequest, "UnknownOperationException", "tagging not supported") - return - } + writeCBORResponse(w, listTagsForResourceOutput{Tags: mapToTags(tags)}) +} - tags, err := tagger.AlarmTags(r.Context(), alarmNameFromARN(in.ResourceARN)) - if err != nil { - writeDriverErr(w, err) +// writeTagRouteErr is the rpc-v2-cbor and JSON counterpart of +// writeTagRouteQueryErr. +func writeTagRouteErr(w http.ResponseWriter, err error) { + if errors.Is(err, errTaggingUnsupported) { + writeCBORError(w, http.StatusBadRequest, "UnknownOperationException", err.Error()) return } - writeCBORResponse(w, listTagsForResourceOutput{Tags: mapToTags(tags)}) + writeDriverErr(w, tagRouteErr(err)) } // alarmNameFromARN extracts the alarm name from a CloudWatch alarm ARN of the diff --git a/server/aws/cloudwatch/metric_streams.go b/server/aws/cloudwatch/metric_streams.go index a1b1aa221..70b831d8e 100644 --- a/server/aws/cloudwatch/metric_streams.go +++ b/server/aws/cloudwatch/metric_streams.go @@ -308,20 +308,18 @@ func (h *Handler) setMetricStreamsRunning(w http.ResponseWriter, r *http.Request writeCBORResponse(w, struct{}{}) } -// writeMetricStreamDriverErr maps a metric-stream driver error to the real -// CloudWatch error shape names these operations document: ResourceNotFoundException -// (GetMetricStream) and InvalidParameterValueException (PutMetricStream). These -// carry the "Exception" suffix that the shared writeDriverErr's shorter -// names (used by the older alarm operations) drop. The exact name matters: an -// SDK/Terraform delete-waiter matches on the deserialized error code, and a -// mismatched name looks like an unexpected failure rather than a signal that -// the resource is gone. +// writeMetricStreamDriverErr maps a metric-stream driver error to the error +// codes these operations return: ResourceNotFoundException (GetMetricStream) +// and InvalidParameterValue (PutMetricStream). The code is the shape's query +// error code, so ResourceNotFoundException keeps its suffix while +// InvalidParameterValueException goes on the wire as InvalidParameterValue. +// The exact code matters: an SDK/Terraform delete-waiter matches on it. func writeMetricStreamDriverErr(w http.ResponseWriter, err error) { switch { case cerrors.IsNotFound(err): writeCBORError(w, http.StatusNotFound, "ResourceNotFoundException", err.Error()) case cerrors.IsInvalidArgument(err): - writeCBORError(w, http.StatusBadRequest, "InvalidParameterValueException", err.Error()) + writeCBORError(w, http.StatusBadRequest, errInvalidParameterValue, cerrors.Message(err)) default: writeDriverErr(w, err) } diff --git a/server/aws/cloudwatch/ops.go b/server/aws/cloudwatch/ops.go index 5bdbb4b76..84321b055 100644 --- a/server/aws/cloudwatch/ops.go +++ b/server/aws/cloudwatch/ops.go @@ -270,8 +270,9 @@ type metricAlarmCBR struct { Metrics []metricDataQueryCBR `cbor:"Metrics,omitempty"` ThresholdMetricID string `cbor:"ThresholdMetricId,omitempty"` - EvaluateLowSampleCountPercentile string `cbor:"EvaluateLowSampleCountPercentile,omitempty"` - EvaluationWindow *evaluationWindowCBR `cbor:"EvaluationWindow,omitempty"` + EvaluateLowSampleCountPercentile string `cbor:"EvaluateLowSampleCountPercentile,omitempty"` + EvaluationWindow *evaluationWindowCBR `cbor:"EvaluationWindow,omitempty"` + AlarmConfigurationUpdatedTimestamp *time.Time `cbor:"AlarmConfigurationUpdatedTimestamp,omitempty"` } type describeAlarmsOutput struct { @@ -382,6 +383,8 @@ func toMetricAlarmCBR(a *mondriver.AlarmInfo) metricAlarmCBR { m.StateTransitionedTimestamp = &ts } + m.AlarmConfigurationUpdatedTimestamp = optTime(a.AlarmConfigurationUpdatedTimestamp) + return m } diff --git a/server/aws/cloudwatch/query.go b/server/aws/cloudwatch/query.go index 32f0afe39..ebfb16ab0 100644 --- a/server/aws/cloudwatch/query.go +++ b/server/aws/cloudwatch/query.go @@ -395,6 +395,8 @@ func toAlarmMemberXML(a *mondriver.AlarmInfo) alarmMemberXML { m.StateTransitionedTimestamp = a.StateTransitionedTimestamp.UTC().Format(time.RFC3339) } + m.AlarmConfigurationUpdatedTimestamp = xmlTime(optTime(a.AlarmConfigurationUpdatedTimestamp)) + return m } @@ -744,8 +746,9 @@ type alarmMemberXML struct { Metrics []metricDataQueryXML `xml:"Metrics>member,omitempty"` ThresholdMetricID string `xml:"ThresholdMetricId,omitempty"` - EvaluateLowSampleCountPercentile string `xml:"EvaluateLowSampleCountPercentile,omitempty"` - EvaluationWindow *evaluationWindowXML `xml:"EvaluationWindow,omitempty"` + EvaluateLowSampleCountPercentile string `xml:"EvaluateLowSampleCountPercentile,omitempty"` + EvaluationWindow *evaluationWindowXML `xml:"EvaluationWindow,omitempty"` + AlarmConfigurationUpdatedTimestamp string `xml:"AlarmConfigurationUpdatedTimestamp,omitempty"` } type compositeAlarmMemberXML struct { diff --git a/server/aws/cloudwatch/query_dashboards.go b/server/aws/cloudwatch/query_dashboards.go index 3bb9b0002..43869b42f 100644 --- a/server/aws/cloudwatch/query_dashboards.go +++ b/server/aws/cloudwatch/query_dashboards.go @@ -20,7 +20,7 @@ func (h *Handler) queryPutDashboard(w http.ResponseWriter, r *http.Request) { return } - if err := store.PutDashboard(r.Context(), r.Form.Get("DashboardName"), r.Form.Get("DashboardBody")); err != nil { + if err := putDashboardCore(r.Context(), store, r.Form.Get("DashboardName"), r.Form.Get("DashboardBody")); err != nil { writeQueryDriverErr(w, err) return } diff --git a/server/aws/cloudwatch/query_metric_streams.go b/server/aws/cloudwatch/query_metric_streams.go index 249ab74c8..b6eba892d 100644 --- a/server/aws/cloudwatch/query_metric_streams.go +++ b/server/aws/cloudwatch/query_metric_streams.go @@ -22,15 +22,13 @@ import ( // writeMetricStreamQueryDriverErr is the query-protocol counterpart of // writeMetricStreamDriverErr (see metric_streams.go): it maps a metric-stream -// driver error to CloudWatch's real ResourceNotFoundException / -// InvalidParameterValueException error codes rather than the shorter names -// the shared writeQueryDriverErr uses for the older alarm operations. +// driver error to ResourceNotFoundException / InvalidParameterValue. func writeMetricStreamQueryDriverErr(w http.ResponseWriter, err error) { switch { case cerrors.IsNotFound(err): writeQueryError(w, http.StatusNotFound, "ResourceNotFoundException", err.Error()) case cerrors.IsInvalidArgument(err): - writeQueryError(w, http.StatusBadRequest, "InvalidParameterValueException", err.Error()) + writeQueryError(w, http.StatusBadRequest, errInvalidParameterValue, cerrors.Message(err)) default: writeQueryDriverErr(w, err) } @@ -200,7 +198,7 @@ func (h *Handler) queryTagResource(w http.ResponseWriter, r *http.Request) { arn := r.Form.Get("ResourceARN") if err := h.addResourceTagsByARN(r.Context(), arn, queryTagPairs(r, "Tags.member.")); err != nil { - writeTagRouteQueryErr(w, arn, err) + writeTagRouteQueryErr(w, err) return } @@ -211,7 +209,7 @@ func (h *Handler) queryUntagResource(w http.ResponseWriter, r *http.Request) { arn := r.Form.Get("ResourceARN") if err := h.removeResourceTagsByARN(r.Context(), arn, queryStringList(r, "TagKeys.member.")); err != nil { - writeTagRouteQueryErr(w, arn, err) + writeTagRouteQueryErr(w, err) return } @@ -223,7 +221,7 @@ func (h *Handler) queryListTagsForResource(w http.ResponseWriter, r *http.Reques tags, err := h.resourceTagsByARN(r.Context(), arn) if err != nil { - writeTagRouteQueryErr(w, arn, err) + writeTagRouteQueryErr(w, err) return } @@ -291,20 +289,28 @@ func (h *Handler) resourceTagsByARN(ctx context.Context, arn string) (map[string // writeTagRouteQueryErr writes the query-protocol response for a tag-routing // error: an unsupported-capability error becomes InvalidAction, and any other -// error is mapped by the metric-stream or alarm driver-error mapper depending -// on which resource kind arn routed to. -func writeTagRouteQueryErr(w http.ResponseWriter, arn string, err error) { +// error is mapped by tagRouteErr. +func writeTagRouteQueryErr(w http.ResponseWriter, err error) { if errors.Is(err, errTaggingUnsupported) { writeQueryError(w, http.StatusBadRequest, "InvalidAction", err.Error()) return } - if _, ok := metricStreamNameFromARN(arn); ok { - writeMetricStreamQueryDriverErr(w, err) - return + writeQueryDriverErr(w, tagRouteErr(err)) +} + +// tagRouteErr maps a tagging error to the codes TagResource, UntagResource +// and ListTagsForResource document: a missing alarm or metric stream is a +// 404 ResourceNotFoundException, and a bad value is InvalidParameterValue. +func tagRouteErr(err error) error { + switch { + case cerrors.IsNotFound(err): + return newNotFoundError(errResourceNotFoundException, cerrors.Message(err)) + case cerrors.IsInvalidArgument(err): + return newWireError(errInvalidParameterValue, cerrors.Message(err)) } - writeQueryDriverErr(w, err) + return err } // ---- form parsing helpers ---- diff --git a/server/aws/cloudwatch/query_metric_streams_test.go b/server/aws/cloudwatch/query_metric_streams_test.go index 878fc551d..c31d0ce25 100644 --- a/server/aws/cloudwatch/query_metric_streams_test.go +++ b/server/aws/cloudwatch/query_metric_streams_test.go @@ -209,9 +209,8 @@ func TestQueryMetricStreamLifecycle(t *testing.T) { } // TestQueryPutMetricStreamValidation confirms the query-protocol path rejects -// IncludeFilters and ExcludeFilters supplied together with the real -// InvalidParameterValueException error name (not the shorter -// InvalidParameterValue the older alarm operations return). +// IncludeFilters and ExcludeFilters supplied together with InvalidParameterValue, +// the query error code of the InvalidParameterValueException shape. func TestQueryPutMetricStreamValidation(t *testing.T) { h := cwserver.New(cwprovider.New(config.NewOptions())) ts := httptest.NewServer(h) @@ -246,7 +245,7 @@ func TestQueryPutMetricStreamValidation(t *testing.T) { if code != http.StatusBadRequest { t.Fatalf("PutMetricStream with both filters: code=%d, want 400, body=%s", code, body) } - if !strings.Contains(body, "InvalidParameterValueException") { - t.Fatalf("PutMetricStream with both filters: want InvalidParameterValueException, body=%s", body) + if !strings.Contains(body, "InvalidParameterValue") { + t.Fatalf("PutMetricStream with both filters: want InvalidParameterValue, body=%s", body) } } diff --git a/server/aws/cloudwatch/validation_parity_test.go b/server/aws/cloudwatch/validation_parity_test.go new file mode 100644 index 000000000..1050e9b00 --- /dev/null +++ b/server/aws/cloudwatch/validation_parity_test.go @@ -0,0 +1,243 @@ +package cloudwatch_test + +import ( + "bytes" + "encoding/json" + "fmt" + "io" + "net/http" + "net/http/httptest" + "net/url" + "regexp" + "sort" + "strconv" + "strings" + "testing" + + "github.com/fxamacker/cbor/v2" +) + +// wireProtocol sends one operation with a generic input over one CloudWatch +// protocol and returns the HTTP status, the error code a client sees ("" on +// success) and the raw body. +type wireProtocol struct { + name string + call func(t *testing.T, ts *httptest.Server, op string, in map[string]any) (int, string, string) +} + +var queryCodePattern = regexp.MustCompile(`([^<]+)`) + +func doCW(t *testing.T, req *http.Request) (int, http.Header, []byte) { + t.Helper() + + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + + raw, _ := io.ReadAll(resp.Body) + + return resp.StatusCode, resp.Header, raw +} + +// wireProtocols covers query (the AWS CLI v2), rpc-v2-cbor (the Go SDK) and +// awsJson1_0 (botocore 1.43+). +func wireProtocols() []wireProtocol { + return []wireProtocol{ + {name: "query", call: func(t *testing.T, ts *httptest.Server, op string, in map[string]any) (int, string, string) { + t.Helper() + + form := url.Values{"Action": {op}, "Version": {"2010-08-01"}} + flattenQuery(form, "", in) + + req, _ := http.NewRequest(http.MethodPost, ts.URL+"/", strings.NewReader(form.Encode())) //nolint:noctx // test request + req.Header.Set("Content-Type", "application/x-www-form-urlencoded") + req.Header.Set("Authorization", monitoringAuth) + + status, _, raw := doCW(t, req) + code := "" + + if m := queryCodePattern.FindSubmatch(raw); status != http.StatusOK && m != nil { + code = string(m[1]) + } + + return status, code, string(raw) + }}, + {name: "cbor", call: func(t *testing.T, ts *httptest.Server, op string, in map[string]any) (int, string, string) { + t.Helper() + + body, err := cbor.Marshal(in) + if err != nil { + t.Fatal(err) + } + + req, _ := http.NewRequest(http.MethodPost, //nolint:noctx // test request + ts.URL+"/service/GraniteServiceVersion20100801/operation/"+op, bytes.NewReader(body)) + req.Header.Set("Smithy-Protocol", "rpc-v2-cbor") + req.Header.Set("Content-Type", "application/cbor") + + status, _, raw := doCW(t, req) + + var out map[string]any + _ = cbor.Unmarshal(raw, &out) + + code, _ := out["__type"].(string) + if status == http.StatusOK { + code = "" + } + + return status, code, fmt.Sprint(out) + }}, + {name: "json", call: func(t *testing.T, ts *httptest.Server, op string, in map[string]any) (int, string, string) { + t.Helper() + + body, _ := json.Marshal(in) + + req, _ := http.NewRequest(http.MethodPost, ts.URL+"/", bytes.NewReader(body)) //nolint:noctx // test request + req.Header.Set("Content-Type", "application/x-amz-json-1.0") + req.Header.Set("X-Amz-Target", jsonTarget+op) + + status, hdr, raw := doCW(t, req) + + // botocore takes the code from X-Amzn-Query-Error on this + // awsQueryCompatible service. + code, _, _ := strings.Cut(hdr.Get("X-Amzn-Query-Error"), ";") + + return status, code, string(raw) + }}, + } +} + +// flattenQuery writes in as query-protocol parameters: nested members as +// "A.B" and list items as "A.member.N". +func flattenQuery(form url.Values, prefix string, in any) { + join := func(k string) string { + if prefix == "" { + return k + } + + return prefix + "." + k + } + + switch v := in.(type) { + case map[string]any: + keys := make([]string, 0, len(v)) + for k := range v { + keys = append(keys, k) + } + + sort.Strings(keys) + + for _, k := range keys { + flattenQuery(form, join(k), v[k]) + } + case []any: + for i, e := range v { + flattenQuery(form, prefix+".member."+strconv.Itoa(i+1), e) + } + default: + form.Set(prefix, fmt.Sprint(v)) + } +} + +func alarmInput(overrides map[string]any) map[string]any { + in := map[string]any{ + "AlarmName": "a1", "Namespace": "App", "MetricName": "Latency", + "ComparisonOperator": "GreaterThanThreshold", "Threshold": 1.5, + "Period": 60, "EvaluationPeriods": 2, "Statistic": "Average", + } + + for k, v := range overrides { + in[k] = v + } + + return in +} + +// TestValidationParity checks the request errors that each protocol must +// answer with the same code and status (tracker row CW-X13). +func TestValidationParity(t *testing.T) { + var dims []any + for i := range 31 { + dims = append(dims, map[string]any{"Name": "d" + strconv.Itoa(i), "Value": "v"}) + } + + missingARN := "arn:aws:cloudwatch:us-east-1:123456789012:alarm:missing" + + tests := []struct { + name string + op string + in map[string]any + status int + code string + }{ + {"tag missing alarm", "TagResource", + map[string]any{"ResourceARN": missingARN, "Tags": []any{map[string]any{"Key": "k", "Value": "v"}}}, + http.StatusNotFound, "ResourceNotFoundException"}, + {"untag missing alarm", "UntagResource", + map[string]any{"ResourceARN": missingARN, "TagKeys": []any{"k"}}, http.StatusNotFound, "ResourceNotFoundException"}, + {"list tags missing alarm", "ListTagsForResource", + map[string]any{"ResourceARN": missingARN}, http.StatusNotFound, "ResourceNotFoundException"}, + {"list tags missing stream", "ListTagsForResource", + map[string]any{"ResourceARN": "arn:aws:cloudwatch:us-east-1:123456789012:metric-stream/missing"}, + http.StatusNotFound, "ResourceNotFoundException"}, + {"31 dimensions", "PutMetricData", map[string]any{"Namespace": "App", "MetricData": []any{ + map[string]any{"MetricName": "m", "Value": 1, "Dimensions": dims}, + }}, http.StatusBadRequest, "InvalidParameterValue"}, + {"period 61", "PutMetricAlarm", alarmInput(map[string]any{"Period": 61}), http.StatusBadRequest, "ValidationError"}, + {"period over a day", "PutMetricAlarm", alarmInput(map[string]any{"Period": 3600, "EvaluationPeriods": 25}), + http.StatusBadRequest, "ValidationError"}, + {"bad statistic", "PutMetricAlarm", alarmInput(map[string]any{"Statistic": "Median"}), + http.StatusBadRequest, "ValidationError"}, + {"datapoints over periods", "PutMetricAlarm", alarmInput(map[string]any{"DatapointsToAlarm": 3}), + http.StatusBadRequest, "ValidationError"}, + {"dashboard body not json", "PutDashboard", map[string]any{"DashboardName": "d", "DashboardBody": "{not json"}, + http.StatusBadRequest, "InvalidParameterInput"}, + {"dashboard body not an object", "PutDashboard", map[string]any{"DashboardName": "d", "DashboardBody": `"x"`}, + http.StatusBadRequest, "InvalidParameterInput"}, + {"stream include and exclude", "PutMetricStream", map[string]any{ + "Name": "s", "FirehoseArn": "arn:aws:firehose:us-east-1:123456789012:deliverystream/f", + "RoleArn": "arn:aws:iam::123456789012:role/r", "OutputFormat": "json", + "IncludeFilters": []any{map[string]any{"Namespace": "A"}}, + "ExcludeFilters": []any{map[string]any{"Namespace": "B"}}, + }, http.StatusBadRequest, "InvalidParameterValue"}, + {"period 30 and 20 datapoints ok", "PutMetricAlarm", + alarmInput(map[string]any{"Period": 30, "EvaluationPeriods": 20, "DatapointsToAlarm": 20}), http.StatusOK, ""}, + {"30 dimensions ok", "PutMetricData", map[string]any{"Namespace": "App", "MetricData": []any{ + map[string]any{"MetricName": "m", "Value": 1, "Dimensions": dims[:30]}, + }}, http.StatusOK, ""}, + } + + for _, p := range wireProtocols() { + t.Run(p.name, func(t *testing.T) { + ts := newJSONServer(t) + + for _, tt := range tests { + status, code, raw := p.call(t, ts, tt.op, tt.in) + if status != tt.status || code != tt.code { + t.Errorf("%s: got %d %q, want %d %q: %s", tt.name, status, code, tt.status, tt.code, raw) + } + } + }) + } +} + +// TestAlarmConfigurationUpdatedTimestamp checks DescribeAlarms returns the +// time the alarm configuration was last put, on every protocol. +func TestAlarmConfigurationUpdatedTimestamp(t *testing.T) { + for _, p := range wireProtocols() { + t.Run(p.name, func(t *testing.T) { + ts := newJSONServer(t) + + if status, _, raw := p.call(t, ts, "PutMetricAlarm", alarmInput(nil)); status != http.StatusOK { + t.Fatalf("PutMetricAlarm: %d %s", status, raw) + } + + status, _, raw := p.call(t, ts, "DescribeAlarms", map[string]any{"AlarmNames": []any{"a1"}}) + if status != http.StatusOK || !strings.Contains(raw, "AlarmConfigurationUpdatedTimestamp") { + t.Fatalf("DescribeAlarms: %d, want AlarmConfigurationUpdatedTimestamp: %s", status, raw) + } + }) + } +} diff --git a/services/monitoring/driver/driver.go b/services/monitoring/driver/driver.go index 4a29f60ea..da61bf376 100644 --- a/services/monitoring/driver/driver.go +++ b/services/monitoring/driver/driver.go @@ -161,6 +161,9 @@ type AlarmInfo struct { Tags map[string]string // StateTransitionedTimestamp is when State last changed. StateTransitionedTimestamp time.Time + // AlarmConfigurationUpdatedTimestamp is when the configuration was last + // put. Populated by AWS CloudWatch; zero for the others. + AlarmConfigurationUpdatedTimestamp time.Time // Metrics and ThresholdMetricID echo a metric-math alarm's query list. Metrics []MetricDataQuery ThresholdMetricID string From 46a5a142adb201f6985ac975cb976e802df04347 Mon Sep 17 00:00:00 2001 From: Nitin Kumar Date: Sun, 27 Sep 2026 18:28:49 +0530 Subject: [PATCH 3/3] fix(cloudwatch): week lookback cap for hourly alarm periods, ARN check on metric streams --- providers/aws/cloudwatch/metric_streams.go | 18 ++++++++++ .../aws/cloudwatch/core_alarm_statistics.go | 33 ++++++++++++++----- .../aws/cloudwatch/validation_parity_test.go | 21 +++++++++++- 3 files changed, 63 insertions(+), 9 deletions(-) diff --git a/providers/aws/cloudwatch/metric_streams.go b/providers/aws/cloudwatch/metric_streams.go index 771582d4e..ae72a788b 100644 --- a/providers/aws/cloudwatch/metric_streams.go +++ b/providers/aws/cloudwatch/metric_streams.go @@ -9,6 +9,7 @@ package cloudwatch import ( "context" "sort" + "strings" "time" "github.com/stackshy/cloudemu/v2/errors" @@ -111,6 +112,12 @@ func validateMetricStreamConfig(cfg *driver.MetricStreamConfig) error { return errors.Newf(errors.InvalidArgument, "RoleArn is required") } + for _, p := range [...]struct{ name, arn string }{{"FirehoseArn", cfg.FirehoseARN}, {"RoleArn", cfg.RoleARN}} { + if !validARN(p.arn) { + return errors.Newf(errors.InvalidArgument, "The value %s for parameter %s is not a valid ARN.", p.arn, p.name) + } + } + if !validMetricStreamOutputFormats[cfg.OutputFormat] { return errors.Newf(errors.InvalidArgument, "invalid OutputFormat %q", cfg.OutputFormat) } @@ -122,6 +129,17 @@ func validateMetricStreamConfig(cfg *driver.MetricStreamConfig) error { return nil } +// validARN reports whether s has the arn:partition:service:region:account:resource +// shape, with the partition, service and resource set. +func validARN(s string) bool { + parts := strings.SplitN(s, ":", arnParts) + + return len(parts) == arnParts && parts[0] == "arn" && parts[1] != "" && parts[2] != "" && parts[5] != "" +} + +// arnParts is the number of colon-separated fields in an ARN. +const arnParts = 6 + // GetMetricStream returns the named metric stream, or NotFound (the // CloudWatch ResourceNotFoundException) when it does not exist. func (m *Mock) GetMetricStream(_ context.Context, name string) (*driver.MetricStreamInfo, error) { diff --git a/server/aws/cloudwatch/core_alarm_statistics.go b/server/aws/cloudwatch/core_alarm_statistics.go index 8ba1e8af7..59ff4dae0 100644 --- a/server/aws/cloudwatch/core_alarm_statistics.go +++ b/server/aws/cloudwatch/core_alarm_statistics.go @@ -51,13 +51,17 @@ var validStatistics = map[string]bool{ statSampleCount: true, statAverage: true, statSum: true, statMinimum: true, statMaximum: true, } -// maxAlarmLookback is how far back, in seconds, EvaluationPeriods * Period -// may reach. -const maxAlarmLookback = 86400 +// How far back, in seconds, EvaluationPeriods * Period may reach: one day +// for a Period under an hour, seven days for a Period of an hour or more. +const ( + maxAlarmLookbackDay = 86400 + maxAlarmLookbackWeek = 604800 + hourPeriod = 3600 +) // validateAlarmPeriods checks Period and DatapointsToAlarm. A single-metric -// alarm's Period is 10, 20, 30 or a multiple of 60, and the alarm cannot -// look back more than a day. +// alarm's Period is 10, 20, 30 or a multiple of 60, and EvaluationPeriods * +// Period is capped at a day, or at a week when Period is an hour or more. func validateAlarmPeriods(cfg *mondriver.AlarmConfig) error { if cfg.DatapointsToAlarm > 0 && cfg.EvaluationPeriods > 0 && cfg.DatapointsToAlarm > cfg.EvaluationPeriods { return newWireError(errValidation, "DatapointsToAlarm must be less than or equal to EvaluationPeriods.") @@ -67,12 +71,25 @@ func validateAlarmPeriods(cfg *mondriver.AlarmConfig) error { return nil } - switch p := cfg.Period; { - case !validPeriod(p): + if !validPeriod(cfg.Period) { return newWireError(errValidation, "Period must be 10, 20, 30 or a multiple of 60") - case p*max(cfg.EvaluationPeriods, 1) > maxAlarmLookback: + } + + return validateAlarmLookback(cfg.Period, cfg.EvaluationPeriods) +} + +// validateAlarmLookback checks EvaluationPeriods * Period against the day or +// week cap that applies to period p. +func validateAlarmLookback(p, evaluationPeriods int) error { + lookback := p * max(evaluationPeriods, 1) + + switch { + case p < hourPeriod && lookback > maxAlarmLookbackDay: return newWireError(errValidation, "Metrics cannot be checked across more than a day "+ "(EvaluationPeriods * Period must be <= 86400)") + case p >= hourPeriod && lookback > maxAlarmLookbackWeek: + return newWireError(errValidation, "Metrics cannot be checked across more than a week "+ + "(EvaluationPeriods * Period must be <= 604800)") } return nil diff --git a/server/aws/cloudwatch/validation_parity_test.go b/server/aws/cloudwatch/validation_parity_test.go index 1050e9b00..a77e151a3 100644 --- a/server/aws/cloudwatch/validation_parity_test.go +++ b/server/aws/cloudwatch/validation_parity_test.go @@ -186,7 +186,19 @@ func TestValidationParity(t *testing.T) { map[string]any{"MetricName": "m", "Value": 1, "Dimensions": dims}, }}, http.StatusBadRequest, "InvalidParameterValue"}, {"period 61", "PutMetricAlarm", alarmInput(map[string]any{"Period": 61}), http.StatusBadRequest, "ValidationError"}, - {"period over a day", "PutMetricAlarm", alarmInput(map[string]any{"Period": 3600, "EvaluationPeriods": 25}), + {"sub-hour period over a day", "PutMetricAlarm", alarmInput(map[string]any{"Period": 1800, "EvaluationPeriods": 49}), + http.StatusBadRequest, "ValidationError"}, + {"sub-hour period at a day ok", "PutMetricAlarm", + alarmInput(map[string]any{"Period": 1800, "EvaluationPeriods": 48, "DatapointsToAlarm": 1}), http.StatusOK, ""}, + {"hour period over a week", "PutMetricAlarm", alarmInput(map[string]any{"Period": 3600, "EvaluationPeriods": 169}), + http.StatusBadRequest, "ValidationError"}, + {"hour period at a week ok", "PutMetricAlarm", + alarmInput(map[string]any{"Period": 3600, "EvaluationPeriods": 168, "DatapointsToAlarm": 1}), http.StatusOK, ""}, + {"6h period 8 times ok", "PutMetricAlarm", + alarmInput(map[string]any{"Period": 21600, "EvaluationPeriods": 8, "DatapointsToAlarm": 1}), http.StatusOK, ""}, + {"day period 7 times ok", "PutMetricAlarm", + alarmInput(map[string]any{"Period": 86400, "EvaluationPeriods": 7, "DatapointsToAlarm": 1}), http.StatusOK, ""}, + {"day period 8 times", "PutMetricAlarm", alarmInput(map[string]any{"Period": 86400, "EvaluationPeriods": 8}), http.StatusBadRequest, "ValidationError"}, {"bad statistic", "PutMetricAlarm", alarmInput(map[string]any{"Statistic": "Median"}), http.StatusBadRequest, "ValidationError"}, @@ -202,6 +214,13 @@ func TestValidationParity(t *testing.T) { "IncludeFilters": []any{map[string]any{"Namespace": "A"}}, "ExcludeFilters": []any{map[string]any{"Namespace": "B"}}, }, http.StatusBadRequest, "InvalidParameterValue"}, + {"stream bad firehose arn", "PutMetricStream", map[string]any{ + "Name": "s", "FirehoseArn": "bad", "RoleArn": "arn:aws:iam::123456789012:role/r", "OutputFormat": "json", + }, http.StatusBadRequest, "InvalidParameterValue"}, + {"stream bad role arn", "PutMetricStream", map[string]any{ + "Name": "s", "FirehoseArn": "arn:aws:firehose:us-east-1:123456789012:deliverystream/f", + "RoleArn": "arn:aws:iam::123456789012", "OutputFormat": "json", + }, http.StatusBadRequest, "InvalidParameterValue"}, {"period 30 and 20 datapoints ok", "PutMetricAlarm", alarmInput(map[string]any{"Period": 30, "EvaluationPeriods": 20, "DatapointsToAlarm": 20}), http.StatusOK, ""}, {"30 dimensions ok", "PutMetricData", map[string]any{"Namespace": "App", "MetricData": []any{