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/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/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..59ff4dae0 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,57 @@ 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, +} + +// 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 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.") + } + + if len(cfg.Metrics) > 0 || cfg.Period == 0 { + return nil + } + + if !validPeriod(cfg.Period) { + return newWireError(errValidation, "Period must be 10, 20, 30 or a multiple of 60") + } + + 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 +} + // 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/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..843412985 --- /dev/null +++ b/server/aws/cloudwatch/json_protocol.go @@ -0,0 +1,311 @@ +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") + 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 +// 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. +// 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("{}") + } + + 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 + } + + if _, err := dec.Token(); !errors.Is(err, io.EOF) { + return nil, errTrailingJSON + } + + 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: + return jsonNumberToCBOR(t) + } + + return v, 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 + } + + 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 +// 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..1a9f45f89 --- /dev/null +++ b/server/aws/cloudwatch/json_protocol_test.go @@ -0,0 +1,448 @@ +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: "trailing data", op: "DescribeAlarms", body: `{"AlarmNames":[]} {"x":1}`, + 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..c594c6869 --- /dev/null +++ b/server/aws/cloudwatch/json_transcode_test.go @@ -0,0 +1,82 @@ +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: 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"` + Value float64 `cbor:"Value"` + Whole float64 `cbor:"Whole"` + Start *time.Time `cbor:"Start"` + End time.Time `cbor:"End"` + } + + 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) + } + + 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"`, `{"a":1} {"b":2}`, `{"a":1}x`, `{"a":1e999}`} { + 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/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/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/validation_parity_test.go b/server/aws/cloudwatch/validation_parity_test.go new file mode 100644 index 000000000..a77e151a3 --- /dev/null +++ b/server/aws/cloudwatch/validation_parity_test.go @@ -0,0 +1,262 @@ +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"}, + {"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"}, + {"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"}, + {"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{ + 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/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) + }) + } +} 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