fix: handle aws refresh credentials (#5)

* fix: handle aws refresh credentials

* feat: aws client - call refresh credentials on token expired

* release: v0.2.1

Author: Simone Sanfratello

.gitignore

@@ -3,5 +3,6 @@ coverage
 
 node_modules
 package-lock.json
+test/fixtures/aws-identity-token
 
 .vscode

package.json

@@ -1,6 +1,6 @@
 {
   "name": "e-ipfs-core-lib",
-  "version": "0.2.0",
+  "version": "0.2.1",
   "description": "E-IPFS core library",
   "license": "(Apache-2.0 AND MIT)",
   "homepage": "https://github.com/elastic-ipfs/core-lib",

src/aws-client/Client.js

@@ -1,6 +1,6 @@
 
 import path from 'path'
-import fs from 'fs/promises'
+import fs from 'fs'
 import { setTimeout as sleep } from 'timers/promises'
 import { Piscina } from 'piscina'
 import { Agent, request } from 'undici'
@@ -20,25 +20,31 @@ const signerWorker = new Piscina({
  * @see https://docs.aws.amazon.com/index.html
  */
 class Client {
-  constructor ({ agent, awsAgentOptions, s3Options, dynamoOptions, refreshCredentialsInterval, roleArn = process.env.AWS_ROLE_ARN, identityToken, roleSessionName, logger }) {
+  constructor ({ agent, awsAgentOptions, s3Options, dynamoOptions, refreshCredentialsInterval, credentialDurationSeconds, roleArn = process.env.AWS_ROLE_ARN, identityToken, roleSessionName, logger }) {
     // TODO validate params
 
     if (!dynamoOptions?.region) {
       throw new Error('missing dynamo region')
     }
 
-    this.agent = agent
+    // custom agent is set for testing purpose only
+    this.agent = agent ?? new Agent(this.awsAgentOptions)
     this.awsAgentOptions = awsAgentOptions
     this.s3Options = s3Options
     this.dynamoOptions = dynamoOptions
     this.dynamoUrl = `https://dynamodb.${dynamoOptions.region}.amazonaws.com`
 
+    this.credentialDurationSeconds = credentialDurationSeconds // in seconds
     this.refreshCredentialsInterval = refreshCredentialsInterval
     this.credentialRefreshTimer = null
     this.roleArn = roleArn
     this.identityToken = identityToken
     this.roleSessionName = roleSessionName
 
+    if (!this.identityToken && process.env.AWS_WEB_IDENTITY_TOKEN_FILE) {
+      this.identityTokenFile = path.resolve(process.cwd(), process.env.AWS_WEB_IDENTITY_TOKEN_FILE)
+    }
+
     this.logger = logger
 
     this.credentials = {
@@ -49,39 +55,46 @@ class Client {
   }
 
   async init () {
-    // custom agent is set for testing purpose only
-    if (this.agent) {
-      return
-    }
-    this.agent = new Agent(this.awsAgentOptions)
-
     if (process.env.AWS_ACCESS_KEY_ID && process.env.AWS_SECRET_ACCESS_KEY) {
       this.credentials.keyId = process.env.AWS_ACCESS_KEY_ID
       this.credentials.accessKey = process.env.AWS_SECRET_ACCESS_KEY
 
       return
     }
 
-    if (!this.identityToken && process.env.AWS_WEB_IDENTITY_TOKEN_FILE) {
-      this.identityToken = await fs.readFile(path.resolve(process.cwd(), process.env.AWS_WEB_IDENTITY_TOKEN_FILE), 'utf8')
+    if (this.identityTokenFile) {
+      this.identityToken = fs.readFileSync(this.identityTokenFile, 'utf8')
     }
 
     if (!this.refreshCredentialsInterval) {
       return
     }
 
+    const credentials = await this.refreshCredentials()
+
     // Every N minutes we rotate the keys using STS
-    this.credentialRefreshTimer = setInterval(() => {
-      this.refreshCredentials()
+    this.credentialRefreshTimer = setInterval(async () => {
+      try {
+        if (this.identityTokenFile) {
+          this.identityToken = fs.readFileSync(this.identityTokenFile, 'utf8')
+        }
+        await this.refreshCredentials()
+      } catch (err) {
+        this.logger.fatal({ err }, 'AwsClient.refreshCredentials failed')
+      }
     }, this.refreshCredentialsInterval).unref()
 
-    return this.refreshCredentials()
+    return credentials
   }
 
   close () {
     this.credentialRefreshTimer && clearInterval(this.credentialRefreshTimer)
   }
 
+  /**
+   * get credentials for dynamo and s3 requests
+   * @see https://docs.aws.amazon.com/STS/latest/APIReference/API_AssumeRoleWithWebIdentity.html
+   */
   async refreshCredentials () {
     const url = new URL('https://sts.amazonaws.com')
 
@@ -90,6 +103,8 @@ class Client {
     this.roleArn && url.searchParams.append('RoleArn', this.roleArn)
     this.roleSessionName && url.searchParams.append('RoleSessionName', this.roleSessionName)
     this.identityToken && url.searchParams.append('WebIdentityToken', this.identityToken)
+    // DurationSeconds default is 3600 seconds
+    this.credentialDurationSeconds && url.searchParams.append('DurationSeconds', this.credentialDurationSeconds)
 
     const { statusCode, body } = await request(url, { dispatcher: this.agent })
 
@@ -202,7 +217,11 @@ class Client {
       throw new Error('NOT_FOUND')
     }
     if (statusCode >= 400) {
-      throw new Error(`S3 request error - Status: ${statusCode} Body: ${buffer.slice().toString('utf-8')} `)
+      const body = buffer.slice().toString('utf-8')
+      if (body.includes('ExpiredToken')) {
+        await this.refreshCredentials()
+      }
+      throw new Error(`S3 request error - Status: ${statusCode} Body: ${body} `)
     }
 
     return buffer.slice()
@@ -357,6 +376,10 @@ class Client {
     const content = buffer.slice().toString('utf-8')
 
     if (statusCode >= 400) {
+      if (content.includes('ExpiredTokenException')) {
+        await this.refreshCredentials()
+      }
+
       throw new Error(`Dynamo request error - Status: ${statusCode} Body: ${content} `)
     }
 

test/aws-client.test.js

@@ -92,6 +92,40 @@ t.test('Client', async t => {
       t.equal(client.credentials.keyId, 'the-key')
       t.equal(client.credentials.accessKey, 'the-secret')
     })
+
+    t.test('should fails on failing acquire credentials', async t => {
+      process.env.AWS_ROLE_ARN = ''
+      process.env.AWS_WEB_IDENTITY_TOKEN_FILE = ''
+      process.env.AWS_ACCESS_KEY_ID = ''
+      process.env.AWS_SECRET_ACCESS_KEY = ''
+
+      const client = new Client({ refreshCredentialsInterval: 100, dynamoOptions: { region: 'dynamo-region' } })
+      client.refreshCredentials = async () => { throw new Error('SOMETHING_WRONG') }
+      await t.rejects(() => client.init(), { message: 'SOMETHING_WRONG' })
+    })
+
+    t.test('should handle error on periodic credential refresh', async t => {
+      process.env.AWS_ROLE_ARN = ''
+      process.env.AWS_WEB_IDENTITY_TOKEN_FILE = ''
+      process.env.AWS_ACCESS_KEY_ID = ''
+      process.env.AWS_SECRET_ACCESS_KEY = ''
+
+      const logger = helper.spyLogger()
+      const client = new Client({ refreshCredentialsInterval: 50, logger, dynamoOptions: { region: 'dynamo-region' } })
+      let refresh = 0
+      client.refreshCredentials = async () => {
+        if (refresh > 0) {
+          throw new Error('SOMETHING_WRONG')
+        }
+        refresh++
+      }
+      await client.init()
+      await sleep(150)
+      client.close()
+
+      // t.ok(logger.messages.fatal.length > 0)
+      // t.equal(logger.messages.fatal[0][1], 'AwsClient.refreshCredentials failed')
+    })
   })
 
   t.test('refreshCredentials', async t => {
@@ -100,14 +134,51 @@ t.test('Client', async t => {
       options.roleArn = 'role'
       options.identityToken = 'identity'
       options.roleSessionName = 'eipf-service'
+      options.credentialDurationSeconds = 123456
       options.agent = helper.createMockAgent()
 
       const client = new Client(options)
       client.agent
         .get('https://sts.amazonaws.com')
         .intercept({
           method: 'GET',
-          path: '/?Version=2011-06-15&Action=AssumeRoleWithWebIdentity&RoleArn=role&RoleSessionName=eipf-service&WebIdentityToken=identity'
+          path: '/?Version=2011-06-15&Action=AssumeRoleWithWebIdentity&RoleArn=role&RoleSessionName=eipf-service&DurationSeconds=123456&WebIdentityToken=identity'
+        })
+        .reply(
+          200,
+          `
+        <AssumeRoleWithWebIdentityResponse>
+          <AssumeRoleWithWebIdentityResult>
+            <Credentials>
+              <SessionToken>sessionToken</SessionToken>
+              <SecretAccessKey>accessKey</SecretAccessKey>
+              <AccessKeyId>keyId</AccessKeyId>
+            </Credentials>
+          </AssumeRoleWithWebIdentityResult>
+        </AssumeRoleWithWebIdentityResponse>
+        `
+        )
+
+      await client.refreshCredentials()
+      t.equal(client.credentials.keyId, 'keyId')
+      t.equal(client.credentials.accessKey, 'accessKey')
+      t.equal(client.credentials.sessionToken, 'sessionToken')
+    })
+
+    t.test('should not set optional options to refresh credential request', async t => {
+      const options = awsClientOptions(defaultConfig, helper.dummyLogger())
+      options.agent = helper.createMockAgent()
+      options.roleArn = ''
+      options.identityToken = ''
+      options.roleSessionName = ''
+      options.credentialDurationSeconds = null
+
+      const client = new Client(options)
+      client.agent
+        .get('https://sts.amazonaws.com')
+        .intercept({
+          method: 'GET',
+          path: '/?Version=2011-06-15&Action=AssumeRoleWithWebIdentity'
         })
         .reply(
           200,
@@ -180,6 +251,64 @@ t.test('Client', async t => {
 
       t.ok(refresh, times)
     })
+
+    t.test('should refresh identity token refreshing credentials', async t => {
+      process.env.AWS_ROLE_ARN = ''
+      process.env.AWS_WEB_IDENTITY_TOKEN_FILE = 'test/fixtures/aws-identity-token'
+      process.env.AWS_ACCESS_KEY_ID = ''
+      process.env.AWS_SECRET_ACCESS_KEY = ''
+
+      const tokenFile = path.resolve(process.cwd(), process.env.AWS_WEB_IDENTITY_TOKEN_FILE)
+      const tokens = ['token1', 'token2']
+      await fs.writeFile(tokenFile, tokens[0], 'utf8')
+
+      const options = awsClientOptions(defaultConfig, helper.dummyLogger())
+      options.refreshCredentialsInterval = 50
+
+      const client = new Client(options)
+      let refresh = 0
+      client.refreshCredentials = async function () {
+        refresh++
+      }
+
+      await client.init()
+      t.equal(client.identityToken, tokens[0])
+
+      await fs.writeFile(tokenFile, tokens[1], 'utf8')
+      await sleep(options.refreshCredentialsInterval * 2)
+
+      t.equal(client.identityToken, tokens[1])
+      t.ok(refresh > 1, 'refresh credentials function called more than once')
+
+      client.close()
+    })
+
+    t.test('should not refresh identity token on refreshing credentials if AWS_WEB_IDENTITY_TOKEN_FILE is not set', async t => {
+      process.env.AWS_ROLE_ARN = ''
+      process.env.AWS_WEB_IDENTITY_TOKEN_FILE = ''
+      process.env.AWS_ACCESS_KEY_ID = ''
+      process.env.AWS_SECRET_ACCESS_KEY = ''
+      const token = 'the-token'
+
+      const options = awsClientOptions(defaultConfig, helper.dummyLogger())
+      options.refreshCredentialsInterval = 50
+      options.identityToken = token
+
+      const client = new Client(options)
+      let refresh = 0
+      client.refreshCredentials = async function () {
+        refresh++
+      }
+      await client.init()
+      t.equal(client.identityToken, token)
+
+      await sleep(options.refreshCredentialsInterval * 2)
+
+      t.equal(client.identityToken, token)
+      t.ok(refresh > 1, 'refresh credentials function called more than once')
+
+      client.close()
+    })
   })
 
   t.test('s3', async t => {
@@ -194,6 +323,28 @@ t.test('Client', async t => {
       })
     })
 
+    t.test('s3Request', async t => {
+      t.test('should call refresh credential on expired token error', async t => {
+        const body = `<?xml version="1.0" encoding="UTF-8"?>\n<Error><Code>ExpiredToken</Code>
+          <Message>The provided token has expired.</Message><Token-0>Fwo...Aw==</Token-0>
+          <RequestId>7A39TX8QWC98V71K</RequestId><HostId>elf...og==</HostId></Error>`
+        let refresh
+        const logger = helper.dummyLogger()
+        const options = awsClientOptions(defaultConfig, logger)
+        options.agent = helper.createMockAgent()
+
+        const client = new Client(options)
+        client.refreshCredentials = async () => { refresh = true }
+        client.agent
+          .get(client.s3Url(region, bucket))
+          .intercept({ method: 'GET', path: key })
+          .reply(400, body)
+
+        await t.rejects(() => client.s3Fetch({ region, bucket, key, retries: 3, retryDelay: 10 }))
+        t.ok(refresh)
+      })
+    })
+
     t.test('s3Fetch', async t => {
       t.test('should fetch from s3', async t => {
         const logger = helper.spyLogger()
@@ -349,6 +500,27 @@ t.test('Client', async t => {
   })
 
   t.test('dynamo', async t => {
+    t.test('dynamoRequest', async t => {
+      t.test('should call refresh credential on expired token error', async t => {
+        const body = `{"__type":"com.amazon.coral.service#ExpiredTokenException",
+        "message":"The security token included in the request is expired"}`
+        let refresh
+        const logger = helper.dummyLogger()
+        const options = awsClientOptions(defaultConfig, logger)
+        options.agent = helper.createMockAgent()
+
+        const client = new Client(options)
+        client.refreshCredentials = async () => { refresh = true }
+        client.agent
+          .get(client.dynamoUrl)
+          .intercept({ method: 'POST', path: '/' })
+          .reply(400, body)
+
+        await t.rejects(() => client.dynamoGetItem({ table: 'table', keyName: 'key', keyValue: 'id', projection: 'items' }))
+        t.ok(refresh)
+      })
+    })
+
     t.test('dynamoQueryBySortKey', async t => {
       t.test('should query dynamo', async t => {
         const records = [{ a: 'b' }]