package storage // Copyright 2017 Microsoft Corporation // // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. // You may obtain a copy of the License at // // http://www.apache.org/licenses/LICENSE-2.0 // // Unless required by applicable law or agreed to in writing, software // distributed under the License is distributed on an "AS IS" BASIS, // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. // See the License for the specific language governing permissions and // limitations under the License. import ( "bytes" "encoding/base64" "encoding/json" "errors" "fmt" "io/ioutil" "net/http" "net/url" "strconv" "strings" "time" "github.com/satori/go.uuid" ) // Annotating as secure for gas scanning /* #nosec */ const ( partitionKeyNode = "PartitionKey" rowKeyNode = "RowKey" etagErrorTemplate = "Etag didn't match: %v" ) var ( errEmptyPayload = errors.New("Empty payload is not a valid metadata level for this operation") errNilPreviousResult = errors.New("The previous results page is nil") errNilNextLink = errors.New("There are no more pages in this query results") ) // Entity represents an entity inside an Azure table. type Entity struct { Table *Table PartitionKey string RowKey string TimeStamp time.Time OdataMetadata string OdataType string OdataID string OdataEtag string OdataEditLink string Properties map[string]interface{} } // GetEntityReference returns an Entity object with the specified // partition key and row key. func (t *Table) GetEntityReference(partitionKey, rowKey string) *Entity { return &Entity{ PartitionKey: partitionKey, RowKey: rowKey, Table: t, } } // EntityOptions includes options for entity operations. type EntityOptions struct { Timeout uint RequestID string `header:"x-ms-client-request-id"` } // GetEntityOptions includes options for a get entity operation type GetEntityOptions struct { Select []string RequestID string `header:"x-ms-client-request-id"` } // Get gets the referenced entity. Which properties to get can be // specified using the select option. // See: // https://docs.microsoft.com/en-us/rest/api/storageservices/fileservices/query-entities // https://docs.microsoft.com/en-us/rest/api/storageservices/fileservices/querying-tables-and-entities func (e *Entity) Get(timeout uint, ml MetadataLevel, options *GetEntityOptions) error { if ml == EmptyPayload { return errEmptyPayload } // RowKey and PartitionKey could be lost if not included in the query // As those are the entity identifiers, it is best if they are not lost rk := e.RowKey pk := e.PartitionKey query := url.Values{ "timeout": {strconv.FormatUint(uint64(timeout), 10)}, } headers := e.Table.tsc.client.getStandardHeaders() headers[headerAccept] = string(ml) if options != nil { if len(options.Select) > 0 { query.Add("$select", strings.Join(options.Select, ",")) } headers = mergeHeaders(headers, headersFromStruct(*options)) } uri := e.Table.tsc.client.getEndpoint(tableServiceName, e.buildPath(), query) resp, err := e.Table.tsc.client.exec(http.MethodGet, uri, headers, nil, e.Table.tsc.auth) if err != nil { return err } defer drainRespBody(resp) if err = checkRespCode(resp, []int{http.StatusOK}); err != nil { return err } respBody, err := ioutil.ReadAll(resp.Body) if err != nil { return err } err = json.Unmarshal(respBody, e) if err != nil { return err } e.PartitionKey = pk e.RowKey = rk return nil } // Insert inserts the referenced entity in its table. // The function fails if there is an entity with the same // PartitionKey and RowKey in the table. // ml determines the level of detail of metadata in the operation response, // or no data at all. // See: https://docs.microsoft.com/rest/api/storageservices/fileservices/insert-entity func (e *Entity) Insert(ml MetadataLevel, options *EntityOptions) error { query, headers := options.getParameters() headers = mergeHeaders(headers, e.Table.tsc.client.getStandardHeaders()) body, err := json.Marshal(e) if err != nil { return err } headers = addBodyRelatedHeaders(headers, len(body)) headers = addReturnContentHeaders(headers, ml) uri := e.Table.tsc.client.getEndpoint(tableServiceName, e.Table.buildPath(), query) resp, err := e.Table.tsc.client.exec(http.MethodPost, uri, headers, bytes.NewReader(body), e.Table.tsc.auth) if err != nil { return err } defer drainRespBody(resp) if ml != EmptyPayload { if err = checkRespCode(resp, []int{http.StatusCreated}); err != nil { return err } data, err := ioutil.ReadAll(resp.Body) if err != nil { return err } if err = e.UnmarshalJSON(data); err != nil { return err } } else { if err = checkRespCode(resp, []int{http.StatusNoContent}); err != nil { return err } } return nil } // Update updates the contents of an entity. The function fails if there is no entity // with the same PartitionKey and RowKey in the table or if the ETag is different // than the one in Azure. // See: https://docs.microsoft.com/en-us/rest/api/storageservices/fileservices/update-entity2 func (e *Entity) Update(force bool, options *EntityOptions) error { return e.updateMerge(force, http.MethodPut, options) } // Merge merges the contents of entity specified with PartitionKey and RowKey // with the content specified in Properties. // The function fails if there is no entity with the same PartitionKey and // RowKey in the table or if the ETag is different than the one in Azure. // Read more: https://docs.microsoft.com/en-us/rest/api/storageservices/fileservices/merge-entity func (e *Entity) Merge(force bool, options *EntityOptions) error { return e.updateMerge(force, "MERGE", options) } // Delete deletes the entity. // The function fails if there is no entity with the same PartitionKey and // RowKey in the table or if the ETag is different than the one in Azure. // See: https://docs.microsoft.com/en-us/rest/api/storageservices/fileservices/delete-entity1 func (e *Entity) Delete(force bool, options *EntityOptions) error { query, headers := options.getParameters() headers = mergeHeaders(headers, e.Table.tsc.client.getStandardHeaders()) headers = addIfMatchHeader(headers, force, e.OdataEtag) headers = addReturnContentHeaders(headers, EmptyPayload) uri := e.Table.tsc.client.getEndpoint(tableServiceName, e.buildPath(), query) resp, err := e.Table.tsc.client.exec(http.MethodDelete, uri, headers, nil, e.Table.tsc.auth) if err != nil { if resp.StatusCode == http.StatusPreconditionFailed { return fmt.Errorf(etagErrorTemplate, err) } return err } defer drainRespBody(resp) if err = checkRespCode(resp, []int{http.StatusNoContent}); err != nil { return err } return e.updateTimestamp(resp.Header) } // InsertOrReplace inserts an entity or replaces the existing one. // Read more: https://docs.microsoft.com/rest/api/storageservices/fileservices/insert-or-replace-entity func (e *Entity) InsertOrReplace(options *EntityOptions) error { return e.insertOr(http.MethodPut, options) } // InsertOrMerge inserts an entity or merges the existing one. // Read more: https://docs.microsoft.com/en-us/rest/api/storageservices/fileservices/insert-or-merge-entity func (e *Entity) InsertOrMerge(options *EntityOptions) error { return e.insertOr("MERGE", options) } func (e *Entity) buildPath() string { return fmt.Sprintf("%s(PartitionKey='%s', RowKey='%s')", e.Table.buildPath(), e.PartitionKey, e.RowKey) } // MarshalJSON is a custom marshaller for entity func (e *Entity) MarshalJSON() ([]byte, error) { completeMap := map[string]interface{}{} completeMap[partitionKeyNode] = e.PartitionKey completeMap[rowKeyNode] = e.RowKey for k, v := range e.Properties { typeKey := strings.Join([]string{k, OdataTypeSuffix}, "") switch t := v.(type) { case []byte: completeMap[typeKey] = OdataBinary completeMap[k] = t case time.Time: completeMap[typeKey] = OdataDateTime completeMap[k] = t.Format(time.RFC3339Nano) case uuid.UUID: completeMap[typeKey] = OdataGUID completeMap[k] = t.String() case int64: completeMap[typeKey] = OdataInt64 completeMap[k] = fmt.Sprintf("%v", v) default: completeMap[k] = v } if strings.HasSuffix(k, OdataTypeSuffix) { if !(completeMap[k] == OdataBinary || completeMap[k] == OdataDateTime || completeMap[k] == OdataGUID || completeMap[k] == OdataInt64) { return nil, fmt.Errorf("Odata.type annotation %v value is not valid", k) } valueKey := strings.TrimSuffix(k, OdataTypeSuffix) if _, ok := completeMap[valueKey]; !ok { return nil, fmt.Errorf("Odata.type annotation %v defined without value defined", k) } } } return json.Marshal(completeMap) } // UnmarshalJSON is a custom unmarshaller for entities func (e *Entity) UnmarshalJSON(data []byte) error { errorTemplate := "Deserializing error: %v" props := map[string]interface{}{} err := json.Unmarshal(data, &props) if err != nil { return err } // deselialize metadata e.OdataMetadata = stringFromMap(props, "odata.metadata") e.OdataType = stringFromMap(props, "odata.type") e.OdataID = stringFromMap(props, "odata.id") e.OdataEtag = stringFromMap(props, "odata.etag") e.OdataEditLink = stringFromMap(props, "odata.editLink") e.PartitionKey = stringFromMap(props, partitionKeyNode) e.RowKey = stringFromMap(props, rowKeyNode) // deserialize timestamp timeStamp, ok := props["Timestamp"] if ok { str, ok := timeStamp.(string) if !ok { return fmt.Errorf(errorTemplate, "Timestamp casting error") } t, err := time.Parse(time.RFC3339Nano, str) if err != nil { return fmt.Errorf(errorTemplate, err) } e.TimeStamp = t } delete(props, "Timestamp") delete(props, "Timestamp@odata.type") // deserialize entity (user defined fields) for k, v := range props { if strings.HasSuffix(k, OdataTypeSuffix) { valueKey := strings.TrimSuffix(k, OdataTypeSuffix) str, ok := props[valueKey].(string) if !ok { return fmt.Errorf(errorTemplate, fmt.Sprintf("%v casting error", v)) } switch v { case OdataBinary: props[valueKey], err = base64.StdEncoding.DecodeString(str) if err != nil { return fmt.Errorf(errorTemplate, err) } case OdataDateTime: t, err := time.Parse("2006-01-02T15:04:05Z", str) if err != nil { return fmt.Errorf(errorTemplate, err) } props[valueKey] = t case OdataGUID: props[valueKey] = uuid.FromStringOrNil(str) case OdataInt64: i, err := strconv.ParseInt(str, 10, 64) if err != nil { return fmt.Errorf(errorTemplate, err) } props[valueKey] = i default: return fmt.Errorf(errorTemplate, fmt.Sprintf("%v is not supported", v)) } delete(props, k) } } e.Properties = props return nil } func getAndDelete(props map[string]interface{}, key string) interface{} { if value, ok := props[key]; ok { delete(props, key) return value } return nil } func addIfMatchHeader(h map[string]string, force bool, etag string) map[string]string { if force { h[headerIfMatch] = "*" } else { h[headerIfMatch] = etag } return h } // updates Etag and timestamp func (e *Entity) updateEtagAndTimestamp(headers http.Header) error { e.OdataEtag = headers.Get(headerEtag) return e.updateTimestamp(headers) } func (e *Entity) updateTimestamp(headers http.Header) error { str := headers.Get(headerDate) t, err := time.Parse(time.RFC1123, str) if err != nil { return fmt.Errorf("Update timestamp error: %v", err) } e.TimeStamp = t return nil } func (e *Entity) insertOr(verb string, options *EntityOptions) error { query, headers := options.getParameters() headers = mergeHeaders(headers, e.Table.tsc.client.getStandardHeaders()) body, err := json.Marshal(e) if err != nil { return err } headers = addBodyRelatedHeaders(headers, len(body)) headers = addReturnContentHeaders(headers, EmptyPayload) uri := e.Table.tsc.client.getEndpoint(tableServiceName, e.buildPath(), query) resp, err := e.Table.tsc.client.exec(verb, uri, headers, bytes.NewReader(body), e.Table.tsc.auth) if err != nil { return err } defer drainRespBody(resp) if err = checkRespCode(resp, []int{http.StatusNoContent}); err != nil { return err } return e.updateEtagAndTimestamp(resp.Header) } func (e *Entity) updateMerge(force bool, verb string, options *EntityOptions) error { query, headers := options.getParameters() headers = mergeHeaders(headers, e.Table.tsc.client.getStandardHeaders()) body, err := json.Marshal(e) if err != nil { return err } headers = addBodyRelatedHeaders(headers, len(body)) headers = addIfMatchHeader(headers, force, e.OdataEtag) headers = addReturnContentHeaders(headers, EmptyPayload) uri := e.Table.tsc.client.getEndpoint(tableServiceName, e.buildPath(), query) resp, err := e.Table.tsc.client.exec(verb, uri, headers, bytes.NewReader(body), e.Table.tsc.auth) if err != nil { if resp.StatusCode == http.StatusPreconditionFailed { return fmt.Errorf(etagErrorTemplate, err) } return err } defer drainRespBody(resp) if err = checkRespCode(resp, []int{http.StatusNoContent}); err != nil { return err } return e.updateEtagAndTimestamp(resp.Header) } func stringFromMap(props map[string]interface{}, key string) string { value := getAndDelete(props, key) if value != nil { return value.(string) } return "" } func (options *EntityOptions) getParameters() (url.Values, map[string]string) { query := url.Values{} headers := map[string]string{} if options != nil { query = addTimeout(query, options.Timeout) headers = headersFromStruct(*options) } return query, headers }