DataNotification.md

Data Notification

Table of Contents

• Introduction
• Workflow steps
  • Get topics available to subscribe
  • Subscribing to a topic
    • Hash-based Message Authentication Code (HMAC) Subscription
    • Google Service Account (GSA) Subscription
  • Get a Subscription by ID
  • Delete a Subscription by ID
  • Handle notifications
  • Update secret for a Subscription
• Version info endpoint
• Current Limitations
• Support for Collaboration Context

Introduction

The OSDU notification system allows for interested consumers to subscribe to data and metadata changes using a publish/subscriber pattern.

A typical workflow using notification is:

• Consumer finds a "topic" that they want to keep up to date with any changes in OSDU.
• Consumer creates Push endpoint, that is used to receive notifications on the interested topic.
• Consumer creates a Subscription in OSDU Notification System, and proves the ownership of the Push endpoint.
• Consumer starts to receive notifications for that topic and processes the message to synchronize with the OSDU state.
• Consumer periodically rotates the "secret" used for subscription.

The topics below describe these steps/APIs in detail that allow consumers to create such integrated workflows using OSDU Notification.

Back to Table of Contents

Steps

Get topics available to subscribe

Consumer uses Data notification "topics" API to view the list of supported notification topics and corresponding sample messages.

GET api/register/v1/topics

curl

curl --request GET \
  --url 'https://register-svc.osdu.com/api/register/v1/topics' \
  --header 'Authorization:  Bearer <JWT>' \
  --header 'Content-Type: application/json' \
  --header 'data-partition-id: common' \

A sample output is shown below. Please note the "name" of the topic. This is required to create a Subscription for a topic you are interested in.

Sample response

[
  {
	"name": "recordstopic",
    "description": "This notification is sent whenever a new record or record version is created, updated or deleted in storage. 'previousVersionKind' is noted upon 'kind' update. Record deletion is noted as a soft 'deletionType'. Record purge is noted as a hard 'deletionType'.",
    "state": "ACTIVE",
    "example": [
      {
        "id": "osdu:abc:123",
        "kind": "osdu:petrel:regularheightfieldsurface:1.0.0",
        "op": "create"
      },
      {
          "id": "osdu:abc:345",
          "kind": "osdu:petrel:regularheightfieldsurface:1.0.1",
          "op": "update",
          "previousVersionKind": "osdu:petrel:regularheightfieldsurface:1.0.0"
      },
      {
          "id": "osdu:abc:567",
          "kind": "osdu:petrel:regularheightfieldsurface:1.0.0",
          "op": "delete",
          "deletionType": "soft"
      },
      {
          "id": "osdu:abc:789",
          "kind": "osdu:petrel:regularheightfieldsurface:1.0.0",
          "op": "delete",
          "deletionType": "hard"
      }
    ]
  },
  {
    "name": "schemachangedtopic",
    "description": "This notification is sent whenever a new schema is created or updated via schema-service.",
    "state": "ACTIVE",
    "example": [
      {
        "kind": "osdu:wks:wellbore:1.0.0",
        "op": "update"
      },
      {
        "kind": "osdu:wks:wellbore:2.0.0",
        "op": "create"
      }
    ]
  },
  {
    "name": "statuschangedtopic",
    "description": "Every Service/Stage would publish their respective status changed information in this topic.",
    "state": "ACTIVE",
    "example": [
      {
        "kind": "status",
        "properties": {
          "correlationId": "12345",
          "recordId": "osdu:file:3479d828-a47d-4e13-a1f8-9791a19e1a7e",
          "recordIdVersion": "1610537924768407",
          "stage": "STORAGE_SYNC",
          "status": "FAILED",
          "message": "acl is not valid",        
          "errorCode ": 400,
          "timestamp ": 1622118996000
        }
      },
      {
        "kind": "dataSetDetails",
        "properties": {
          "correlationId": "12345",
          "dataSetId": "12345",
          "dataSetIdVersion": "1",
          "dataSetType": "FILE",
          "recordCount": 10,
           "timestamp ": 1622118996000
        }
      }
    ]
  }
]

Back to Table of Contents

Subscribing to a topic

The consumer uses the data notification Subscription API to create a Subscription for the topic of interest. A subscription id is returned in the response that can be used to get the subscription details again or delete the subscription later.

POST /api/register/v1/subscription/

To subscribe to a topic, consumers must have a "https" endpoint supporting both "GET" and "POST" methods. "GET" is used as a challenge endpoint when creating (or updating) a subscription to validate that consumer owns this endpoint. "POST" is used for pushing the notifications to consumers.

The challenge is performed only when creating a subscription or when Updating secret for a Subscription

Below are the details of the two types of Subscriptions and the challenge process:

Hash-based Message Authentication Code (HMAC) Subscription using a "secret" string

curl

  curl --request POST \
    --url 'https://register-svc.osdu.com/api/register/v1/subscription \
    -header 'Authorization: Bearer <JWT>' \
    -header 'Content-Type: application/json' \
    -header 'data-partition-id: common' \
    -data '{
  	"name": "testSubscription",
  	"description": "Description",
  	"topic": "records-changed",
  	"pushEndpoint": "<DomainEndpoint>",
  	"secret": {
  		"secretType": "HMAC",
  		"value": "testSecret"
  	}
  }'

Before creating an HMAC Subscription, the consumer needs to make sure that "GET" is supported on the endpoint being registered with OSDU Notification and the endpoint accepts query parameters named "crc" & "hmac". OSDU Notification will send a "GET" request on this endpoint with a random crc, and expects a response hash generated using the crc & the secret value (i.e. "testSecret" in the example above).

In addition, consumers may also want to validate the hmac field, which is the signature that will be used when a message is pushed to this endpoint. The signature verification must be used in the push endpoint implementation before processing the messages, to ensure that the message is coming from OSDU Notification.

Note: Secret value may not contain any special characters (only alphanumeric characters) and the number of characters must be even.

Sample API definition for setting up the challange end point

...
paths:
   ...
   ...
  '/consumer':
    get:
      summary: Notification "get" API
      consumes:
        - application/json
      produces:
        - application/json
      parameters:
        - in: query
          name: crc
          type: string
          required: true
        - in: query
          name: hmac
          type: string
          required: true
      responses:
        '200':
          description: OK
          schema:
            $ref: '#/definitions/challengeResponse'

    ...
    ...

definitions:
  challengeResponse:
    type: object
    required:
      - schema
    properties:
      responseHash:
        type: string    

Sample Java code to generate the hmac signature, validate it and send a response with hash

...
import javax.crypto.Mac;
import javax.crypto.spec.SecretKeySpec;
import javax.xml.bind.DatatypeConverter;
....

@GET
@Path(<DomainEndpoint>)
public Response challenge(@QueryParam("crc") @NotBlank String crc, 
                          @QueryParam("hmac") @NotBlank String hmac) {
    // Use the secret you send to the subscriber registration create request
    // Hint: Secret string can be stored as configuration for the service    
    String secret = <getSecretString()>
    verifyHmacSignature(hmac, secret);
    
    String response = getResponseHash( secret + crc);
    return Response.status(HttpStatus.SC_OK).entity(Collections.singletonMap("responseHash", response)).build();
}

private String getResponseHash(String input) {
    String response = Hashing.sha256()
            .hashString(input, StandardCharsets.UTF_8)
            .toString();
    return Base64.getEncoder().encodeToString(response.getBytes());
}

private static final String HMAC_SHA_256 = "HmacSHA256";
private static final String DATA_FORMAT = "{\"expireMillisecond\": \"%s\",\"hashMechanism\": \"hmacSHA256\",\"endpointUrl\": \"%s\",\"nonce\": \"%s\"}";
private static final String HMAC_SHA_256 = "HmacSHA256";
private static final String DATA_FORMAT = "{\"expireMillisecond\": \"%s\",\"hashMechanism\": \"hmacSHA256\",\"endpointUrl\": \"%s\",\"nonce\": \"%s\"}";
private static final String NOTIFICATION_SERVICE = "de-notification-service";
private static final long EXPIRE_DURATION = 30000L;

private void verifyHmacSignature(String hmac, String secret) throws Exception {
    if (Strings.isNullOrEmpty(hmac)) {
        throw new Exception(MISSING_HMAC_SIGNATURE);
    }
    if (Strings.isNullOrEmpty(secret)) {
        throw new Exception(MISSING_SECRET_VALUE);
    }
    String[] tokens = hmac.split("\\.");
    if (tokens.length != 2) {
        throw new Exception(INVALID_SIGNATURE);
    }
    byte[] dataBytes = Base64.getDecoder().decode(tokens[0]);
    String requestSignature = tokens[1];

    String data = new String(dataBytes, StandardCharsets.UTF_8);
    HmacData hmacData = new Gson().fromJson(data, HmacData.class);
    String url = hmacData.getEndpointUrl();
    String nonce = hmacData.getNonce();
    String expireTime = hmacData.getExpireMillisecond();
    if (Strings.isNullOrEmpty(url) || Strings.isNullOrEmpty(nonce) || Strings.isNullOrEmpty(expireTime)) {
        throw new Exception(MISSING_ATTRIBUTES_IN_SIGNATURE);
    }
    String newSignature = getSignedSignature(url, secret, expireTime, nonce);
    if (!requestSignature.equalsIgnoreCase(newSignature)) {
        throw new Exception(INVALID_SIGNATURE);
    }
}

private String getSignedSignature(String url, String secret, String expireTime, String nonce) throws Exception {
    if (Strings.isNullOrEmpty(url) || Strings.isNullOrEmpty(secret) || !StringUtils.isNumeric(expireTime)) {
        throw new Exception(ERROR_GENERATING_SIGNATURE);
    }
    final long expiry = Long.parseLong(expireTime);
    if (System.currentTimeMillis() > expiry) {
        throw new Exception(SIGNATURE_EXPIRED);
    }
    String timeStamp = String.valueOf(expiry - EXPIRE_DURATION);
    String data = String.format(DATA_FORMAT, expireTime, url, nonce);
    try {
        final byte[] signature = getSignature(secret, nonce, timeStamp, data);
        return DatatypeConverter.printHexBinary(signature).toLowerCase();
    } catch (Exception ex) {
        throw new Exception(ERROR_GENERATING_SIGNATURE, ex);
    }
}

private byte[] getSignature(String secret, String nonce, String timeStamp, String data) throws Exception {
    final byte[] secretBytes = DatatypeConverter.parseHexBinary(secret);
    final byte[] nonceBytes = DatatypeConverter.parseHexBinary(nonce);
    final byte[] encryptedNonce = computeHmacSha256(nonceBytes, secretBytes);
    final byte[] encryptedTimestamp = computeHmacSha256(timeStamp, encryptedNonce);
    final byte[] signedKey = computeHmacSha256(NOTIFICATION_SERVICE, encryptedTimestamp);
    final byte[] signature = computeHmacSha256(data, signedKey);
    return signature;
}

private byte[] computeHmacSha256(final String data, final byte[] key) throws Exception {
    final Mac mac = Mac.getInstance(HMAC_SHA_256);
    mac.init(new SecretKeySpec(key, HMAC_SHA_256));
    return mac.doFinal(data.getBytes(StandardCharsets.UTF_8));
}

private byte[] computeHmacSha256(final byte[] data, final byte[] key) throws Exception {
    final Mac mac = Mac.getInstance(HMAC_SHA_256);
    mac.init(new SecretKeySpec(key, HMAC_SHA_256));
    return mac.doFinal(data);
}

Google Service Account (GSA) Subscription using audience & service account key

curl

  curl --request POST \
    --url 'https://register-svc.osdu.com/api/register/v1/subscriber \
    -header 'Authorization: Bearer <JWT>' \
    -header 'Content-Type: application/json' \
    -header 'data-partition-id: common' \
    -data '{
  	"name": "testSubscription",
  	"description": "Description",
  	"topic": "records-changed",
  	"pushEndpoint": "<DomainEndpoint>",
  	"secret": {
  		"secretType": "GSA",
  		"value": {
  			"audience":"<audience>",
  			"key":"<service account key file contents>"
  		}
  	}
  }'

Before creating a GSA Subscription, the consumer needs to make sure that "GET" is supported on the endpoint being registered with OSDU Notification and it accepts a query parameter named "crc". OSDU Notification will send a "GET" request on this endpoint with a random crc, and expects a response hash generated using crc & the private_key_id field from the Service account used for subscription.

In addition, consumers may also want to validate the google id token provided as "authorization" header, which will be generated using the audience & key provided. The google id token must be used in the push endpoint implementation before processing the messages, to ensure that the message is coming from OSDU Notification.

Sample API definition for setting up the challange end point

...
paths:
   ...
   ...
  '/consumer':
    get:
      summary: Notification "get" API
      consumes:
        - application/json
      produces:
        - application/json
      parameters:
        - in: query
          name: crc
          type: string
          required: true
      responses:
        '200':
          description: OK
          schema:
            $ref: '#/definitions/challengeResponse'
    ...
    ...

definitions:
  challengeResponse:
    type: object
    required:
      - schema
    properties:
      responseHash:
        type: string    

Sample Java code to validate the google id token and send a response with hash

...

import com.google.api.client.googleapis.auth.oauth2.GoogleIdToken;
import com.google.api.client.googleapis.auth.oauth2.GoogleIdTokenVerifier;
import com.google.api.client.http.javanet.NetHttpTransport;
import com.google.api.client.json.jackson2.JacksonFactory;

...

@GET
@Path(<DomainEndpoint>)
public Response test(@QueryParam("crc") @NotBlank String crc) {
    if(!verifyToken())
        return Response.status(HttpStatus.SC_BAD_REQUEST).entity("Authorization signature validation Failed").build();

    // PrivateKeyId can either be stored as configuration for the service or can get from the service account key file.
    String secret = <getPrivateKeyId()>;
    return getResponse(crc, secret);
}

private boolean verifyToken() {
    try {
        GoogleIdTokenVerifier verifier = new GoogleIdTokenVerifier.Builder(new NetHttpTransport(), JacksonFactory.getDefaultInstance())
                .setAudience(Collections.singletonList(<getGoogleAudiences()>))
                .build();
        GoogleIdToken idToken = verifier.verify(<getAuthorizationHeader()>); 
        return idToken != null;
    } catch (Exception ex) {
        return false;
    }
}

private Response getResponse(String crc, String secretString) {
    String response = secretString + crc;
    response = Hashing.sha256()
            .hashString(response, StandardCharsets.UTF_8)
            .toString();
    response = Base64.getEncoder().encodeToString(response.getBytes());
    ChallengeResponse cr = new ChallengeResponse();
    cr.responseHash = response;

    return Response
            .status(HttpStatus.SC_OK)
            .entity(cr)
            .build();
}

Back to Table of Contents

Get a Subscription by ID

Consumers use this API to get the subscription details for the Subscription with the given id.

GET /api/register/v1/subscription/{id}

curl

curl --request GET \
   --url 'https://register-svc.osdu.com/api/register/v1/subscription/{id} \
  --header 'Authorization: Bearer <JWT>' \
  --header 'data-partition-id: common'

Back to Table of Contents

Delete a Subscription by ID

Consumers use this API to delete a Subscription with the given subscription id.

DELETE /api/register/v1/subscription/{id}

curl

curl --request DELETE\
  --url 'https://register-svc.osdu.com/api/register/v1/subscription/<id>' \
  --header 'authorization: Bearer <JWT>' \
  --header 'content-type: application/json' \
  --header 'data-partition-id: common'

Back to Table of Contents

Handling notifications

Consumers will start receiving messages for the topics that they have subscribed for.

Message Contents

The sample message for record change notification looks like this:

[ 
  {"id":"record_id_1","kind":"kind1","op":"create","recordUpdated":"false"},
  {"id":"record_id_1","kind":"kind1","op":"create","recordUpdated":"true"},
  {"id":"record_id_2","kind":"kind2","op":"delete","deletionType":"soft"},
  {"id":"record_id_3","kind":"kind2","op":"delete","deletionType":"hard"}
  ...
]

Please note that each message can contain a maximum of 50 record updates and can have updates for multiple kinds and operations. If there are more than 50 record updates, subscribers will receive multiple messages.

Possible values of operation types (i.e. "op" field in above example) are as follows:

• create
• delete
• create_schema

Notification handler endpoint - HMAC Secret type

Endpoints with HMAC Subscriptions must accept a parameter named "hmac", which has a signature, as described above. Consumer should make sure to validate this signature before processing messages in request body.

A simple API definition to receive notifications

...
  '/consumer':
    post:
      summary: Receive notification
      description: "Receives push notification from OSDU"
      consumes:
        - application/json
      produces:
        - application/json
      parameters:
        - in: query
          name: hmac
          type: string
          required: true
      responses:
        '200':
          description: OK
...

A simple Java implementation of the endpoint

@POST
@Path(<ConsumerEndpoint>)
public Response processMessage(@NotBlank(message = "Request body can not be null") String data,
                               @NotBlank(message = "'hmac signature' can not be null or empty") @QueryParam("hmac") String hmac,
                               @HeaderParam("data-partition-id") String partitionId,
                               @HeaderParam("correlation-id") String correlationId) throws Exception {
    // Exception is thrown if not possible to verify Hmac signature
    verifyHmacSignature(hmac)

    ...
    // Get message from body
    
    // Process message
    
    ...
    
    return response
}

Notification handler Endpoint - GSA Secret type

Endpoints with GSA Subscriptions must accept a google id token as Authorization header and must validate this before processing messages.

A simple API definition to receive notifications

...
  '/consumer':
    post:
      summary: Receive notification
      description: "Receives push notification from OSDU"
      consumes:
        - application/json
      produces:
        - application/json
      responses:
        '200':
          description: OK
...

A simple implementation of the the endpoint will look like this

@POST
@Path(<ConsumerEndpoint>)
public Response processMessage(Object o) {
    if (!verifyToken(getAuthorizationHeader())) {
        return Response.status(HttpStatus.SC_BAD_REQUEST).entity("Authorization signature validation Failed").build();
    }

    ...
    // Get message from signature
    
    // Process message
    
    ...
    
    return response;
}

Responding to Notifications

The notification service expects a response with the code in the 200-299 range for successfully acknowledged messages. It expects such a response from the consumer endpoint within 30 seconds. If acknowledgement is not received in this time, the notification service will continue to call the endpoint for 5 days. The frequency of the message slows down if it consistently fails to receive a successful acknowledgement.

Back to Table of Contents

Update secret for a Subscription

Consumers might want to regularly update their secret for the Subscriptions to avoid security issues.

This can be done using the OSDU Notification update subscription API. Consumers must update the "GET" endpoint first to point to new secret, as the same verification will be performed again with new secret value.

The change in secret takes effect immediately.

PUT /api/register/v1/subscription/{id}/secret

curl

curl --request PUT \
  --url 'https://register-svc.osdu.com/api/register/v1/subscription/{id}/secret' \
  --header 'authorization: Bearer <JWT>' \
  --header 'content-type: application/json' \
  --header 'data-partition-id: common' \
  --data '{
            "secretType": "HMAC",
            "value": <newValue>
         }'

Back to Table of Contents

Version info endpoint

For deployment available public /info endpoint, which provides build and git related information.

Example response:

{
    "groupId": "org.opengroup.osdu",
    "artifactId": "storage-gcp",
    "version": "0.10.0-SNAPSHOT",
    "buildTime": "2021-07-09T14:29:51.584Z",
    "branch": "feature/GONRG-2681_Build_info",
    "commitId": "7777",
    "commitMessage": "Added copyright to version info properties file",
    "connectedOuterServices": [
      {
        "name": "elasticSearch",
        "version":"..."
      },
      {
        "name": "postgresSql",
        "version":"..."
      },
      {
        "name": "redis",
        "version":"..."
      }
    ]
}

This endpoint takes information from files, generated by spring-boot-maven-plugin, git-commit-id-plugin plugins. Need to specify paths for generated files to matching properties:

• version.info.buildPropertiesPath
• version.info.gitPropertiesPath

Back to table of contents

Current Limitations

• There is no filtering applied on messages (such as based on the kind etc.) at the moment in OSDU. All the messages will be pushed to consumers.

Back to Table of Contents

Support for Collaboration Context

Register service and Notification service are collaboration aware. For now, to enable collaboration context support you have to enable collaboration feature flag in the services. Refer to this Wiki for more details on how to do that. That means when the collaboration context feature flag is enabled the list of topics returned will have a new topic "recordstopic-v2" which will receive notifications when "x-collaboration" header is provided in the request.

curl for a collaboration context header provided request

curl --request GET \
  --url 'https://register-svc.osdu.com/api/register/v1/topics' \
  --header 'Authorization:  Bearer <JWT>' \
  --header 'Content-Type: application/json' \
  --header 'data-partition-id: common' \
  --header 'x-collaboration: id=<collaboration id>,application=<application name>' \

A sample output is shown below when the collaboration context feature flag is set to true.

Sample response

[
    {
        "name": "recordstopic",
        "description": "This notification is sent whenever a new record or record version is created, updated or deleted in storage. 'previousVersionKind' is noted upon 'kind' update. Record deletion is noted as a soft 'deletionType'. Record purge is noted as a hard 'deletionType'.",
        "state": "ACTIVE",
        "example": [
            {
                "id": "osdu:abc:123",
                "kind": "osdu:petrel:regularheightfieldsurface:1.0.0",
                "op": "create"
            },
            {
                "id": "osdu:abc:345",
                "kind": "osdu:petrel:regularheightfieldsurface:1.0.1",
                "op": "update",
                "previousVersionKind": "osdu:petrel:regularheightfieldsurface:1.0.0"
            },
            {
                "id": "osdu:abc:567",
                "kind": "osdu:petrel:regularheightfieldsurface:1.0.0",
                "op": "delete",
                "deletionType": "soft"
            },
            {
                "id": "osdu:abc:789",
                "kind": "osdu:petrel:regularheightfieldsurface:1.0.0",
                "op": "delete",
                "deletionType": "hard"
            }
        ]
    },
    {
        "name": "schemachangedtopic",
        "description": "This notification is sent whenever a new schema is created or updated via schema-service.",
        "state": "ACTIVE",
        "example": [
            {
                "kind": "osdu:wks:wellbore:1.0.0",
                "op": "update"
            },
            {
                "kind": "osdu:wks:wellbore:2.0.0",
                "op": "create"
            }
        ]
    },
    {
        "name": "statuschangedtopic",
        "description": "Every Service/Stage would publish their respective status changed information in this topic.",
        "state": "ACTIVE",
        "example": [
            {
                "kind": "status",
                "properties": {
                    "correlationId": "12345",
                    "recordId": "osdu:file:3479d828-a47d-4e13-a1f8-9791a19e1a7e",
                    "recordIdVersion": "1610537924768407",
                    "stage": "STORAGE_SYNC",
                    "status": "FAILED",
                    "message": "acl is not valid",
                    "errorCode ": 400.0,
                    "timestamp ": 1.622118996E12
                }
            },
            {
                "kind": "dataSetDetails",
                "properties": {
                    "correlationId": "12345",
                    "dataSetId": "12345",
                    "dataSetIdVersion": "1",
                    "dataSetType": "FILE",
                    "recordCount": 10.0,
                    "timestamp ": 1.622118996E12
                }
            }
        ]
    },
    {
        "name": "recordstopic-v2",
        "description": "This notification is sent whenever a new record or record version is created, updated or deleted in storage for all collaboration and non-collaboration context changes. The collaboration context will be provided as part of the message properties if exist. 'previousVersionKind' is noted upon 'kind' update. Record deletion is noted as a soft 'deletionType'. Record purge is noted as a hard 'deletionType'.",
        "state": "ACTIVE",
        "example": {
            "message": {
                "data": [
                    {
                        "id": "osdu:abc:123",
                        "kind": "osdu:petrel:regularheightfieldsurface:1.0.0",
                        "op": "create"
                    }
                ],
                "account-id": "opendes",
                "data-partition-id": "opendes",
                "correlation-id": "4f1982a2-cbdf-438a-b5a1-e0c6239d46fc",
                "x-collaboration": "id=1e1c4e74-3b9b-4b17-a0d5-67766558ec65,application=Test App"
            }
        }
    }
]

When the feature flag is set to true and the consumer provides "x-collaboration" header in the request for creating, updating, and deleting a record. the message sent will contain the collaboration context header as an attribute.

Example of a message when the x-collaboration header is provided:

{
   "message": {
      "data": [
         {
            "id": "<message-id>",
            "version": "1617915304347525",
            "modifiedBy": "abc@xyz.com",
            "kind": "common:welldb:wellbore:1.0.0",
            "op": "create"
         }
      ],
      "account-id": "opendes",
      "data-partition-id": "opendes",
      "correlation-id": "<corrilation-id>",
      "x-collaboration": "id=<collaboration-id>,application=<app-name>"
   }
}