Using selenium on NodeJS

 

In this post we will review how to use selenium on NodeJS to automate browser usage. This is usually used for automatic site testing. We will make some changes which are specifically useful for such scenario.

First, lets install the chormedriver from this link. Make sure to download the zip file which match you chrome browser version, unzip it, and move the chormedriver to a folder which is in the path, for example:

sudo mv chromedriver /usr/local/bin/

Now for the project: create the package.json file:

{
  "name": "demo",
  "version": "1.0.0",
  "description": "",
  "author": "",
  "license": "ISC",
  "scripts": {
    "demo": "node main.js"
  },
  "dependencies": {
    "chrome-modheader": "^1.0.6",
    "selenium-webdriver": "^4.0.0-beta.3"
  }
}

We use the selenium driver, and add the chrome-modheader, which allows us to set headers on the requests. Setting header can be used for A/B testing, and for setting the XFF header to simulate a source IP.

The general code structure is as follows:

const {Builder, until, By, logging} = require('selenium-webdriver')
const chrome = require('selenium-webdriver/chrome')
const {getExtension, getAddHeaderUrl} = require('chrome-modheader')

main()

async function main() {
  // our code here
}

We start our code by starting a browser, and setting a random IP in the XFF header.

const preferences = new logging.Preferences()
preferences.setLevel(logging.Type.BROWSER, logging.Level.ALL)

const options = new chrome.Options()
options.setLoggingPrefs(preferences)
options.addArguments('--ignore-certificate-errors')
options.addArguments('--no-sandbox')
options.addExtensions(getExtension())


const driver = await new Builder()
  .forBrowser('chrome')
  .setChromeOptions(options)
  .build()

function getRandomIpSection() {
  return Math.ceil(Math.random() * 256)
}

const ip = `${getRandomIpSection()}.${getRandomIpSection()}.${getRandomIpSection()}.${getRandomIpSection()}`
console.log(`random IP is ${ip}`)
await driver.get(getAddHeaderUrl('X-Forwarded-For', ip))

Now we open our site, and cleanup cookies for a fresh session:

await driver.get('http://my.site.com/')
await driver.manage().deleteAllCookies()

We can locate elements by CSS and by Xpath:

const BUTTON_SELECTOR = '#features a.read-more:first-of-type'
const button = await driver.wait(until.elementLocated(By.css(BUTTON_SELECTOR)), 10000)

const TEXT_SELECTOR = '//div/span[contains(@style,\'color: green\')]'
const text = await driver.wait(until.elementLocated(By.xpath(TEXT_SELECTOR)), 10000)

We can click on items. In this case we click on the element that we have just located.

button.click()

We can run our own javacript code in the page:

await driver.executeScript(`
  window.enableMyDebugLog=true
`)

We can sleep, waiting for something:

await driver.sleep(2000)

and we can scan the console logs:

const logs = await driver.manage().logs().get(logging.Type.BROWSER)
for (const log of logs) {
  const message = log.message
  if (message.includes("my-log-data")) {
    console.log(message)
  }
}

Eventually to close the browser, use quit:

await driver.quit()

go-redis using SCAN command in a Redis Cluster

  

In this post we will present how to use redis SCAN command in a cluster environment and go-redis library.

The go-redis library automatically handled a single key commands such as GET, SET. It recognizes the location of each slot on a relevant master, and address the master (or slave) that hold the specific key.

However, the SCAN command is a multi-keys related, and hence the go-redis does not handle it.

The way to handle it is using the ForEachMaster command, and run the SCAN command on each master, and finally aggregate the result. 

Additional item to handle is to maintain a cursor per each master, and detect end of data in all of the masters.

The cursor per master structure is listed below:

type cursorData struct {
   locations map[string]uint64
   endOfData map[string]bool
}

func (d *cursorData) EndOfData() bool {
   for _, end := range d.endOfData {
      if !end {
         return false
      }
   }
   return true
}

Next we can use the ForEachMaster to run SCAN:

func Scan(
   client *redis.ClusterClient,
   cursor redisclients.CursorInterface,
   match string,
   count int64,
) ([]string, redisclients.CursorInterface) {
   var cursorPerMaster *cursorData
   if cursor == nil {
      cursorPerMaster = &cursorData{
         locations: make(map[string]uint64),
         endOfData: make(map[string]bool),
      }
   } else {
      var ok bool
      cursorPerMaster, ok = (cursor).(*cursorData)
      if !ok {
         panic("conversion failed")
      }
   }

   allKeys := make([]string, 0)
   mutex := sync.Mutex{}

   err := client.ForEachMaster(context.Background(), func(ctx context.Context, master *redis.Client) error {
      key := master.String()

      mutex.Lock()
      alreadyDone := cursorPerMaster.endOfData[key]
      mutex.Unlock()

      if alreadyDone {
         return nil
      }

      mutex.Lock()
      masterCursor := cursorPerMaster.locations[key]
      mutex.Unlock()

      cmd := master.Scan(ctx, masterCursor, match, count)
      err := cmd.Err()
      if err != nil {
         return err
      }

      keys, nextCursor, err := cmd.Result()
      if err != nil {
         return err
      }

      mutex.Lock()
      allKeys = append(allKeys, keys...)
      cursorPerMaster.locations[key] = nextCursor
      cursorPerMaster.endOfData[key] = nextCursor == 0
      mutex.Unlock()

      return nil
   })

   if err != nil {
      panic(err)
   }

   return allKeys, cursorPerMaster
}

We should hide out implementation using an interface:

type CursorInterface interface {
   EndOfData() bool
}

An example for using this API is:

firstTime := true
var cursor redisclients.CursorInterface
for {
   var keys []string
   count := int64(10000)
   match := "*"
   if firstTime {
      firstTime = false
      keys, cursor = Scan(client, nil, match, count)
   } else {
      keys, cursor = Scan(client, cursor, match, count)
   }
   fmt.Print(keys)
   if cursor.EndOfData() {
      return 
   }
}

In case you want a simpler usage, for debug and test environment only, checkout the KEYS implementation in this post.

go-redis using KEYS command in a Redis Cluster

 

In this post we will present how to use redis KEYS command in a cluster environment and go-redis library.

The go-redis library automatically handled a single key commands such as GET, SET. It recognizes the location of each slot on a relevant master, and address the master (or slave) that hold the specific key.

However, the KEYS command is a multi-keys related, and hence the go-redis does not handle it.

The way to handle it is using the ForEachMaster command, and run the KEYS command on each master, and finally aggregate the result. An example for this is listed below.

func ClusterKeys(client *redis.ClusterClient, pattern string) []string {
   allKeys := make([]string, 0)
   mutex := sync.Mutex{}
   err := client.ForEachMaster(context.Background(), func(ctx context.Context, master *redis.Client) error {
      cmd := master.Keys(ctx, pattern)
      err := cmd.Err()
      if err != nil {
         return err
      }

      value, err := cmd.Result()
      if err != nil {
         return err
      }
      mutex.Lock()
      allKeys = append(allKeys, value...)
      mutex.Unlock()
      return nil
   })

   if err != nil {
      panic(err)
   }

   return allKeys
}

Notice that usage of the redis KEYS command in a production environment is not recommended. You might want to use SCAN command instead. For more details see this post.

Reading from AWS Kinesis using GoLang

 

In this post we will review reading from AWS Kinesis using GoLang. 

AWS kinesis is widely used in AWS services, for example, it can be used to read real time logging records from CloudFront. See this blog for setup of a real time logging on CloudFront.

To access AWS services, we start with authentication to AWS. See this blog for alternatives for authentication method. Once authentication is configured, we can connect to the kinesis data stream. The record reading is done from a specific shard. In this example, we will select the first available shard.

import (
   "context"
   "fmt"
   "github.com/aws/aws-sdk-go-v2/aws"
   "github.com/aws/aws-sdk-go-v2/config"
   "github.com/aws/aws-sdk-go-v2/service/kinesis"
)

func consume() {
   awsConfig, err := config.LoadDefaultConfig(context.Background(), config.WithRegion("us-east-1"))
   if err != nil {
      panic(err)
   }

   client := kinesis.NewFromConfig(awsConfig)

   streamName := "my-kinesis-data-stream"
   describeInput := kinesis.DescribeStreamInput{
      StreamName: aws.String(streamName),
   }
   describeOutput, err := client.DescribeStream(context.Background(), &describeInput)
   if err != nil {
      panic(err)
   }
   shard := describeOutput.StreamDescription.Shards[0]

The reading from the stream is done using an infinite busy waiting loop, and an iterator. In this example, we start reading from the first record in the shard, and keep waiting for additional new records.

iteratorInput := kinesis.GetShardIteratorInput{
   ShardId:                shard.ShardId,
   ShardIteratorType:      "TRIM_HORIZON",
   StreamName:             aws.String(streamName),
   StartingSequenceNumber: nil,
   Timestamp:              nil,
}
iteratorOutput, err := client.GetShardIterator(context.Background(), &iteratorInput)
if err != nil {
   panic(err)
}

var iterator = *iteratorOutput.ShardIterator
for {
   getInput := kinesis.GetRecordsInput{
      ShardIterator: &iterator,
      Limit:         nil,
   }
   getOutput, err := client.GetRecords(context.Background(), &getInput)
   if err != nil {
      panic(err)
   }
   for _, record := range getOutput.Records {
      data := string(record.Data)
      fmt.Printf("data: %v\n", data)
   }

   iterator = *getOutput.NextShardIterator
}

CloudFront Real-Time Logging using CloudFormation

 

In this post we will review how to use a CloudFormation template to configure Real-Time Logging to a Kinesis data stream.

First we need to create the kinesis data stream:

KinesisDataStream:
  Type: AWS::Kinesis::Stream
  Properties:
    Name: my-kinesis-data-stream
    RetentionPeriodHours: 24
    ShardCount: 1

Next, we configure a IAM role with permission to write to the kinesis data stream:

RealTimeLogggingRole:
  Type: AWS::IAM::Role
  Properties:
    Tags:
      - Key: Name
        Value: my-real-time-logging-role
    Path: "/"
    AssumeRolePolicyDocument:
      Version: 2012-10-17
      Statement:
        - Effect: Allow
          Action: sts:AssumeRole
          Principal:
            Service: cloudfront.amazonaws.com
    Policies:
      - PolicyName: my-real-time-logging-policy
        PolicyDocument:
          Version: 2012-10-17
          Statement:
            - Effect: Allow
              Action:
                - kinesis:DescribeStreamSummary
                - kinesis:DescribeStream
                - kinesis:PutRecord
                - kinesis:PutRecords
              Resource:
                - !GetAtt KinesisDataStream.Arn

Now we can configure the real time logging to use the IAM role and the kinesis stream:

RealTimeLoggging:
  Type: AWS::CloudFront::RealtimeLogConfig
  Properties:
    Name: my-real-time-logging
    SamplingRate: 100
    Fields:
      - timestamp
      - c-ip
      - cs-host
      - cs-uri-stem
      - cs-headers
    EndPoints:
      - StreamType: Kinesis
        KinesisStreamConfig:
          RoleArn: !GetAtt RealTimeLogggingRole.Arn
          StreamArn: !GetAtt KinesisDataStream.Arn

The last thing to do, is to configure our CloudFront distribution to use this real time logging:

CloudFrontDistribution:
  Type: AWS::CloudFront::Distribution
  Properties:
    DistributionConfig:
      DefaultCacheBehavior:
        RealtimeLogConfigArn: !Ref RealTimeLoggging

Notice that the CloudFront distribution displayed here is only partial. For a full example of a CloudFront creation, see this post.

Throttle API calls in GO

 

In this post we will review how to throttle API calls to a library in GO.

We assume that the library does some updates upon a notification. In this case, the notification does not contain any data, it simply means: "Something changed, do your stuff". 

Now, we want the update to occur no more than once in a minute. It does not matter how many time it was notified, only that the library does its work no more than once in a minute.

To do this, we use an updates channel, and a GO routine. The throttle struct includes the channel for updates, as well as the configuration of throttle.

package throttle

import (
   "time"
)

type Handler func() error

type Throttle struct {
   interval time.Duration
   handler  Handler
   updates  chan bool
   lastRun  time.Time
}

func CreateThrottle(
   interval time.Duration,
   handler Handler,
) *Throttle {
   return &Throttle{
      interval: interval,
      handler:  handler,
      updates:  make(chan bool, 100),
      lastRun:  time.Now(),
   }
}

Next we include the API calls for the throttle:

func (t *Throttle) Start() {
   go t.channelLoop()
}

func (t *Throttle) Notify() {
   t.updates <- true
}

and last, we implement the GO routine to handle the updates. Notice that in case the notifications are too frequent, we schedule a later run of the API upon a timer.

func (t *Throttle) channelLoop() {
   var nextTimeout *time.Duration

   for {
      if nextTimeout == nil {
         _ = <-t.updates
      } else {
         select {
         case _ = <-t.updates:
         case <-time.After(*nextTimeout):
         }
      }

      passedTime := time.Now().Sub(t.lastRun)
      if passedTime < t.interval {
         timeout := t.interval - passedTime
         nextTimeout = &timeout
      } else {
         err := t.handler()
         if err != nil {
            panic(err)
         }
         t.lastRun = time.Now()
         nextTimeout = nil
      }
   }
}

That's it! 

The throttle API is ready. Let's see and example of usage:

throttling = CreateThrottle(time.Minute, myExecutor)
throttling.Start()

How To Fake Source IP XFF Header

 

Recently in one of our test sites, I had to fake my source IP, as I had to test the GUI response to multiple source IPs. I had to work using a valid browser, in my case Chrome.

The first thing I've tried is using IPFuck Chrome extension, but it failed. Chrome was aware that it is sending an additional header, and the site was blocking this behavior using the Access-Control-Allow-Headers option. 

The solution in my case was to add a NGINX reverse proxy to handle the header addition. I have setup a local NGINX to proxy the request to their original target.

The NGINX run script is using docker:

docker stop faker
docker rm faker
docker run --name faker --network host -v ${PWD}/empty:/docker-entrypoint.d -v ${PWD}/nginx.conf:/etc/nginx/nginx.conf nginx

And the folder of the script contains a folder named "empty", as well as nginx.conf file:

user  nginx;
worker_processes  1;

error_log  /dev/stdout debug;
pid        /var/run/nginx.pid;

events {
    worker_connections  1024;
}

http {
    include       /etc/nginx/mime.types;
    default_type  application/octet-stream;

    log_format  main  '$remote_addr - $remote_user [$time_local] "$request" '
                      '$status $body_bytes_sent "$http_referer" '
                      '"$http_user_agent" "$http_x_forwarded_for"';

    access_log  /dev/stdout;

    sendfile        on;
    keepalive_timeout  65;

   server {
      listen       8080;

      location / {
         resolver 10.221.1.47; 
         proxy_pass http://$http_host$uri$is_args$args;
           proxy_set_header Host $host;
           proxy_set_header X-Forwarded-For 52.14.41.49;
       }
   }
}

To make the browser use the NGINX reverse proxy, I had to setup it to use the proxy localhost:8080.

Final Note

Notice that this is working for HTTP sites. 

HTTPS sites should have additional configuration for the SSL support.