Tuesday, 3 October 2023

Using Mongoose with Amazon Document DB

 We live in 2023, and unless it is absolutely necessary, no one calls the DB layer directly. We use the concept of ORM for every modern language and database.

The same is correct for MondoDB, popular document database. There is a project called "Mongoose" that creates the abstract layer above the database. The technology has been there for some time, and we know that it works fine.

But what about Amazon Document DB? If you are new to Document DB, you should know that this is Amazon proprietary technology, which doesn't run the Mongo engine but is based on the technology and concepts developed by Amazon in-house. DocumentDB emulates the Mongo API, and basically each tool and driver that you can use with MongoDB you can also use with DocumentDB. There are still some differences, but all major features are supported.

In this post, we will check if Mongoose can work with DocumentDB.

The code and instructions are based on two other documents:

1. Getting started with Mongoose and MongoDB

2.  Connect to DocumentDB from EC2.

I started by creating the DocumentDB database according to link 2 above.

My cluster looks like this:


It is very important to make sure that the database security group has port 27017 set in the security group inbound rule.




The DocumentDB is deployed into VPC, and there is no public access. It doesn't mean that it cannot be accessed by the internet, but the easiest way to access it is by creating EC2 in the same VPC. I created the one that runs Amazon Linux 3.



You need to make sure that EC2 has the role that allows access to DocumentDB.




and the role looks like


Now we are ready to write the application.



I created a Node application according to Link 1 above.

Don't forget to modify the package.json file and add the module type



You also need global-bundle.pem file since we use TLS connection to DocumentDB.

You can get it by running:

wget https://truststore.pki.rds.amazonaws.com/global/global-bundle.pem

The content of the files:

db.js (note that the connection string is masked)
========================



import mongoose from "mongoose";

export default function connectDB() {
  const url = "mongodb://xxxaws.com:27017/?tls=true&tlsCAFile=global-bundle.pem&retryWrites=false";

  try {
    mongoose.connect(url, {
      useNewUrlParser: true,
      useUnifiedTopology: true,
      maxPoolSize: 30
    });
  } catch (err) {
    console.error(err.message);
    process.exit(1);
  }
  const dbConnection = mongoose.connection;
  dbConnection.once("open", (_) => {
    console.log(`Database connected: ${url}`);
  });

  dbConnection.on("error", (err) => {
    console.error(`connection error: ${err}`);
  });
  return;
}

Blog.js
======================
import mongoose from 'mongoose';
const { Schema, model } = mongoose;

const blogSchema = new Schema({
  title: String,
  slug: String,
  published: Boolean,
  author: String,
  content: String,
  tags: [String],
  createdAt: Date,
  updatedAt: Date,
  comments: [{
    user: String,
    content: String,
    votes: Number
  }]
});

const Blog = model('Blog', blogSchema);
export default Blog;

index.js

=========================

import express from "express";
import connectDB from "./config/db.js";
import Blog from './model/Blog.js';

const app = express();
const PORT = 3000;

app.use(express.json());
app.use(express.urlencoded({ extended: true }));

connectDB();


// Create a new blog post object
const article = new Blog({
  title: 'Using mongoose with Amazon Document DB',
  slug: 'learnfrommike',
  published: true,
  content: 'Testing mongoose with Amazon Document DB',
  tags: ['Mongoose', 'Amazon DocumentDB'],
});

// Insert the article in our DocumentDB database
await article.save();

const firstArticle = await Blog.findOne({});
console.log(firstArticle);

// You can modify this function to return the Article object instead of
// printing it to the console
app.get("/", (request, response) => { response.send({ message: "Hello from an Express API!" }); }); app.listen(PORT, () => { console.log(`Server running at http://localhost:${PORT}`); });


If we did everything correctly, we should see the content of
"Article" object, being returned from DocumentDB.





The result is good. So Mongoose is compatible with DocumentDB.
Enjoy!


Monday, 7 August 2023

Amazon Aurora audit data with activity streams

 The ability to store the activity that is executed in your database is a must-have in the modern technology landscape.

First of all, there are different sets of regulations and compliance requirements.

In addition, it can be used for internal analytics, in order to understand the pattern of usage or detect anomalies.

Security tools for application detection and prevention can rely on this data to identify and prevent security attacks on your application.

Amazon Aurora provides the ability to store the audit with just one click. All you need to do is create the cluster and choose the "Start Activity Stream" option.

And this is exactly what I did.


I already clicked the button this is why you see the stop option.

Once you click to enable the stream, the data flows to the Kinesis Data Stream that was provisioned for you automatically. All the data is encrypted with the KMS key. It is up tp you to decide what you want to do with the data.

Amazon documentation suggest the following architecture:


Which streams the data to S3 or external application. I decided to modify the architecture a little bit.


The data in the activity stream is encrypted. And it is fine to production use-case. But for this blog I want to see the actual activity in S3. To achieve this, I will use the transformation Lambda of Kinesis Firehose and use it to decode the message before it is stored in S3.

Please find below the Lambda code:

import base64
import json
import zlib
import aws_encryption_sdk
from aws_encryption_sdk import CommitmentPolicy
from aws_encryption_sdk.internal.crypto import WrappingKey
from aws_encryption_sdk.key_providers.raw import RawMasterKeyProvider
from aws_encryption_sdk.identifiers import WrappingAlgorithm, EncryptionKeyType
import boto3

enc_client = aws_encryption_sdk.EncryptionSDKClient(commitment_policy=CommitmentPolicy.REQUIRE_ENCRYPT_ALLOW_DECRYPT)

class MyRawMasterKeyProvider(RawMasterKeyProvider):
    provider_id = "BC"

    def __new__(cls, *args, **kwargs):
        obj = super(RawMasterKeyProvider, cls).__new__(cls)
        return obj

    def __init__(self, plain_key):
        RawMasterKeyProvider.__init__(self)
        self.wrapping_key = WrappingKey(wrapping_algorithm=WrappingAlgorithm.AES_256_GCM_IV12_TAG16_NO_PADDING,
                                        wrapping_key=plain_key, wrapping_key_type=EncryptionKeyType.SYMMETRIC)

    def _get_raw_key(self, key_id):
        return self.wrapping_key


def decrypt_payload(payload, data_key):
    my_key_provider = MyRawMasterKeyProvider(data_key)
    my_key_provider.add_master_key("DataKey")
    decrypted_plaintext, header = enc_client.decrypt(
        source=payload,
        materials_manager=aws_encryption_sdk.materials_managers.default.DefaultCryptoMaterialsManager(master_key_provider=my_key_provider))
    return decrypted_plaintext


def decrypt_decompress(payload, key):
    decrypted = decrypt_payload(payload, key)
    return zlib.decompress(decrypted, zlib.MAX_WBITS + 16)

def lambda_handler(event, context):
    output = []
    session = boto3.session.Session()
    kms = session.client('kms')
   
    RESOURCE_ID = 'cluster-XXXXXXXXXXXXX'     
    for record in event['records']:
        
        recdord_data_plain = base64.b64decode(record['data']).decode('utf-8')
        
        record_data = json.loads(recdord_data_plain)
        payload_decoded = base64.b64decode(record_data['databaseActivityEvents'])
        data_key_decoded = base64.b64decode(record_data['key'])
        data_key_decrypt_result = kms.decrypt(CiphertextBlob=data_key_decoded,
                                              EncryptionContext={'aws:rds:dbc-id': RESOURCE_ID})
        
        plaintext =  (decrypt_decompress(payload_decoded, data_key_decrypt_result['Plaintext']))
        
        events = json.loads(plaintext)
        
        ## Filtering logic. ## Removes heartbeat and rdsadmin events
        # if events['databaseActivityEventList']
        # if events['databaseActivityEventList'] is actually an empty array, then lines 67 and 68 wont happen and it will skip to 70 
        for dbEvent in events['databaseActivityEventList'][:]:
            if dbEvent['type'] == "heartbeat" or (dbEvent['dbUserName'] and dbEvent["dbUserName"] == "rdsadmin"):
                events['databaseActivityEventList'].remove(dbEvent)
        
        result = 'ProcessingFailed'
        
        if events['databaseActivityEventList']: # This is the same as len(events['databaseActivityEventList']) != 0 since an empty array is "FALSEY" in python
            result = 'Ok'
            print('Decrypted data bellow')
            print(json.dumps(events))        
   
        # Do custom processing on the payload here

        output_record = {
            'recordId': record['recordId'],
            'result': result,
            'data': base64.b64encode(json.dumps(events).encode('utf-8')).decode('utf-8')
        }
        output.append(output_record)

    print('Successfully processed {} records.'.format(len(event['records'])))

    return {'records': output}


The code is based on Amazon workshop for activity stream, I just modified the code to run inside Lambda. Also, note that you need a security package to run the code, which I also took from the workshop and uploaded as a lambda layer. You have to use Python 3.9 at least.

Next, use you favorite database client to run DDL or DML statement:

End here is the result that I downloaded from S3 and opened with Json Editor (I masked some data for the security reasons)

{
"type": "DatabaseActivityMonitoringRecord",
"clusterId": "cluster-XXXXXXXXXXXX",
"instanceId": "db-3TLYJL7WCFXYMCAXXXXXXXXA",
"databaseActivityEventList": [
{
"logTime": "2023-08-07 19:22:14.534688+00",
"statementId": 2,
"substatementId": 1,
"objectType": null,
"command": "CREATE ROLE",
"objectName": null,
"databaseName": "postgres",
"dbUserName": "postgres",
"remoteHost": "172.31.43.XXXX",
"remotePort": "33468",
"sessionId": "64d14447.5a3b",
"rowCount": null,
"commandText": "CREATE USER davide WITH PASSWORD <REDACTED>",
"paramList": [],
"pid": 23099,
"clientApplication": "psql",
"exitCode": null,
"class": "ROLE",
"serverVersion": "14.7.1",
"serverType": "PostgreSQL",
"serviceName": "Amazon Aurora PostgreSQL-Compatible edition",
"serverHost": "172.31.XXXXXXX",
"netProtocol": "TCP",
"dbProtocol": "Postgres 3.0",
"type": "record",
"startTime": "2023-08-07 19:22:14.533149+00",
"errorMessage": null
}
]
}






You need to be aware that once you enable the stream options there is a little overhead for CPU consumption.