Improve logging. Improve ending streams. Improve atsocat

node/atsocat.js

@@ -2,6 +2,7 @@ const net = require('net')
 const crypto = require('crypto')
 const duplex = require('/runtime/attest-duplex.js')
 const { endStream } = duplex
+const minimist = require('minimist')
 
 // similar to linux socat cmd
 
@@ -12,30 +13,34 @@ function onError(err) {
 
 async function main() {
   console.log(`main`)
-  const args = process.argv.slice(2)
-  const [listen, url] = [parseInt(args[0]), args[1]]
-  const testFn = (PCR2, userData) => Promise.resolve(PCR2)
+  const args = minimist(process.argv.slice(2))
+  const [listen, url] = [parseInt(args._[0]), args._[1]]
+
+  const testFn = async (PCR2, userData) => {
+    const total = crypto.createHash('sha256').update(PCR2.join('')).digest('hex')
+    if (args.total === undefined) { return [PCR2, total] }
+    if (args.total !== total) { throw new Error(`PCRs do not match (${args.total}) (${total})`) }
+    return [PCR2, total]
+  }
+
   const ok = await duplex.connect(url, testFn)
-  let [encrypt, decrypt, PCR] = ok
+  let [encrypt, decrypt, attestData] = ok
+  let [PCR, total] = attestData
   encrypt.destroy()
   decrypt.destroy()
 
   let [PCR0, PCR1, PCR2] = PCR
   console.log(`PCR0 ${PCR0}`)
   console.log(`PCR1 ${PCR1}`)
   console.log(`PCR2 ${PCR2}`)
-
-  const total = crypto.createHash('sha256')
-    .update(PCR.join(''))
-    .digest('hex')
   console.log(`TOTAL ${total}`)
 
   const tcpServer = net.createServer(async (client) => {
     console.log(`client connected`)
     const ok = await duplex.connect(url, testFn)
     console.log(`server connected`)
-    if (PCR.join('') !== ok[2].join('')) { onError(`PCR changed`) }
-    [encrypt, decrypt, PCR] = ok
+    if (PCR.join('') !== ok[2][0].join('')) { onError(`PCR changed`) }
+    [encrypt, decrypt, attestData] = ok
 
     const close = (err='') => {
       err = typeof err === 'boolean' ? '' : err
@@ -49,16 +54,16 @@ async function main() {
     encrypt.on('error', close)
     decrypt.on('error', close)
 
-    client.on('close', close)
-    encrypt.on('close', close)
-    decrypt.on('close', close)
+    client.once('close', close)
+    encrypt.once('close', close)
+    decrypt.once('close', close)
 
     client.pipe(encrypt)
     decrypt.pipe(client)
   })
 
   tcpServer.on('error', (err) => onError(new Error(`tcpServer error ${err.message}`)))
-  tcpServer.on('close', () => onError(new Error('tcpServer closed')))
+  tcpServer.once('close', () => onError(new Error('tcpServer closed')))
   tcpServer.listen(listen, '0.0.0.0')
 }
 

node/attest-duplex.js

@@ -37,17 +37,25 @@ function writeStream(stream, data) {
   return write
 }
 
-function endStream(stream) {
-  const [timer, timedout] = timeout(netTimeout)
+function endStream(stream, delay=netTimeout) {
+  if (stream.destroyed) { return }
+  const [timer, timedout] = timeout(delay)
   const end = () => {
     clearTimeout(timer)
     stream.destroy()
+    stream.removeListener('error', end)
+    stream.removeListener('close', end)
   }
   timedout.catch(end)
   stream.once('error', end)
+  if (stream.closed) { return }
   stream.once('close', end)
-  if (typeof stream.end === 'function') { return stream.end() }
-  stream.close()
+  try {
+    if (typeof stream.end === 'function') { return stream.end() }
+    stream.close()
+  } catch (err) {
+    stream.destroy()
+  }
 }
 
 const wellKnown = '/.well-known/lockhost'
@@ -79,7 +87,7 @@ function sendHello(url, nonce, envelope='tcp') {
     }
 
     conn.on('error', onErr)
-    conn.on('close', () => onErr(new Error('hello = close')))
+    conn.once('close', () => onErr(new Error('hello = close')))
     req = conn.request({ ':path': `${path}/hello?${params.toString()}` })
     req.on('error', onErr)
 
@@ -99,7 +107,7 @@ function sendHello(url, nonce, envelope='tcp') {
     req.on('end', () => {
       if (status !== 200) {
         res({ status })
-        conn.close()
+        endStream(conn)
         return
       }
       body = Buffer.concat(body).toString('utf8')
@@ -109,7 +117,7 @@ function sendHello(url, nonce, envelope='tcp') {
       } catch (err) {
         rej(new Error('hello = reply not json'))
       }
-      conn.close()
+      endStream(conn)
     })
 
     req.end()
@@ -167,7 +175,7 @@ async function connect(url, testFn) {
     }
 
     conn.on('error', onErr)
-    conn.on('close', () => onErr(new Error('session = close')))
+    conn.once('close', () => onErr(new Error('session = close')))
     req = conn.request({ ':method': 'POST', ':path': `${path}/session`, 'cookie': `sessionlh=${cookie}` })
     req.on('error', onErr)
 
@@ -192,9 +200,9 @@ async function connect(url, testFn) {
         endStream(conn)
       }
 
-      encrypt.on('close', close)
-      decrypt.on('close', close)
-      conn.on('close', close)
+      encrypt.once('close', close)
+      decrypt.once('close', close)
+      conn.once('close', close)
 
       encrypt.pipe(req)
       req.pipe(decrypt)
@@ -232,10 +240,10 @@ async function client(url, testFn, log, userData=null) {
       pack.on('error', close)
       unpack.on('error', close)
 
-      encrypt.on('close', close)
-      decrypt.on('close', close)
-      pack.on('close', close)
-      unpack.on('close', close)
+      encrypt.once('close', close)
+      decrypt.once('close', close)
+      pack.once('close', close)
+      unpack.once('close', close)
 
       // client is inside Nitro Enclave
       const attestEnclave = (nonce) => {
@@ -270,7 +278,7 @@ async function client(url, testFn, log, userData=null) {
 }
 
 module.exports = {
-  urlToHostAndPath,
+  timeout, urlToHostAndPath,
   writeStream, endStream,
   sendHello, startState,
   connect, client,

node/fetch.js

@@ -1,19 +1,7 @@
+const { timeout } = require('/runtime/attest-duplex.js')
 const noop = () => {}
 
-// use less timers by group 100ms
-const timeout = (ms) => {
-  let timer = null
-  const timedout = new Promise((res, rej) => {
-    const now = Date.now()
-    let next = now + ms
-    next = next - (next % 100)
-    next = (100 + next) - now
-    timer = setTimeout(rej, next, null)
-  })
-  return [timer, timedout]
-}
-
-// simple wrapper for timeouts
+// simple wrapper with timeouts
 module.exports = function fetchWithTimeout(request, timeoutms=10_000) {
   if (typeof request === 'string') { request = new Request(request) }
   if (!(request instanceof Request)) { return Promise.reject(new Error('fetch accepts url or instance of Request')) }

node/host.js

@@ -2,6 +2,7 @@ const net = require('net')
 const http = require('http')
 const crypto = require('crypto')
 const openVSock = require('./vsock.js')
+const { timeout, endStream } = require('/runtime/attest-duplex.js')
 
 const netTimeout = 10_000
 const vsockTimeout = 5_000
@@ -19,19 +20,6 @@ function onError(err) {
 
 const noop = () => {}
 
-// use less timers by group 100ms
-const timeout = (ms) => {
-  let timer = null
-  const timedout = new Promise((res, rej) => {
-    const now = Date.now()
-    let next = now + ms
-    next = next - (next % 100)
-    next = (100 + next) - now
-    timer = setTimeout(rej, next, null)
-  })
-  return [timer, timedout]
-}
-
 function write(stream, data) {
   const isNet = stream instanceof net.Socket
   const timeoutMs = isNet ? netTimeout : vsockTimeout
@@ -50,16 +38,8 @@ function write(stream, data) {
 
 function end(stream) {
   const isNet = stream instanceof net.Socket
-  const timeoutMs = isNet ? netTimeout : vsockTimeout
-  const [timer, timedout] = timeout(timeoutMs)
-  const end = () => {
-    clearTimeout(timer)
-    stream.destroy()
-  }
-  timedout.catch(end)
-  stream.once('error', end)
-  stream.once('close', end)
-  stream.end()
+  const delayMs = isNet ? netTimeout : vsockTimeout
+  endStream(stream, delayMs)
 }
 
 function connectToRemoteTcp(ip, port) {
@@ -69,9 +49,9 @@ function connectToRemoteTcp(ip, port) {
   const connect = new Promise((res, rej) => {
     timedout.catch((err) => rej(new Error(`connect ${info} timeout`)))
     conn.on('error', (err) => rej(new Error(`connect ${info} error ${err.message}`)))
-    conn.on('connectionAttemptFailed', () => rej(new Error(`connect ${info} failed`)))
-    conn.on('connectionAttemptTimeout', () => rej(new Error(`connect ${info} timeout`)))
-    conn.on('close', () => rej(new Error(`connect ${info} close`)))
+    conn.once('connectionAttemptFailed', () => rej(new Error(`connect ${info} failed`)))
+    conn.once('connectionAttemptTimeout', () => rej(new Error(`connect ${info} timeout`)))
+    conn.once('close', () => rej(new Error(`connect ${info} close`)))
     conn.connect(port, ip, () => res(conn))
   }).catch((err) => {
     conn.destroy()
@@ -105,7 +85,7 @@ async function onVSockData(obj) {
       }
 
       server.on('error', cleanup)
-      server.on('close', cleanup)
+      server.once('close', cleanup)
 
       // fwd data to runtime to enclave
       server.on('data', (data) => {
@@ -192,7 +172,7 @@ async function accept(client, port) {
   }
 
   client.on('error', cleanup)
-  client.on('close', cleanup)
+  client.once('close', cleanup)
 
   // fwd data to runtime to enclave
   client.on('data', (data) => {
@@ -208,7 +188,7 @@ function tcpServer(port) {
   const tcpServer = net.createServer(wrap)
   return new Promise((res, rej) => {
     tcpServer.on('error', (err) => onError(new Error(`tcpServer ${port} error ${err.message}`)))
-    tcpServer.on('close', () => onError(new Error(`tcpServer ${port} closed`)))
+    tcpServer.once('close', () => onError(new Error(`tcpServer ${port} closed`)))
     tcpServer.listen(port, '0.0.0.0', res)
   })
 }
@@ -237,7 +217,7 @@ function readBody(request) {
     request.setEncoding('utf8')
     request.on('error', rej)
     request.on('data', (chunk) => str += chunk)
-    request.on('end', () => res(str))
+    request.once('end', () => res(str))
   })
   read.catch(noop).finally(() => clearTimeout(timer))
   return read