mirror of
https://github.com/stablyai/orca.git
synced 2026-10-22 08:03:37 +00:00
fix(ssh): bound received bytes, not just the declared stream size
The marker gate only checked what the relay claimed. handleChunk then appended every chunk with no ceiling, ACKed each one so the relay's credit window kept reopening, and re-armed the inactivity timer — so a stream that under-declared totalBytes and never sent responseEnd was still reassembled without bound. Reject at the offending chunk instead, mirroring ssh-filesystem-stream-reader's per-chunk length check. Also replaces a vacuous assertion in the at-the-cap test: chunk handlers are subscribed before the sentinel is read, so their presence held even when the gate rejected.
This commit is contained in:
@@ -71,7 +71,7 @@ describe('requestGitStreamable reassembly cap', () => {
|
||||
|
||||
it('admits a stream declared exactly at the cap', async () => {
|
||||
const payload = Buffer.from(JSON.stringify({ kind: 'text' }), 'utf-8')
|
||||
const { mux, chunkHandlers } = createFakeMux({
|
||||
const { mux, notify } = createFakeMux({
|
||||
streamId: 8,
|
||||
totalBytes: payload.length,
|
||||
chunkCount: 1
|
||||
@@ -85,8 +85,51 @@ describe('requestGitStreamable reassembly cap', () => {
|
||||
await Promise.resolve()
|
||||
await Promise.resolve()
|
||||
|
||||
// The marker passed the gate: a chunk subscriber is live and nothing rejected.
|
||||
expect(chunkHandlers).toHaveLength(1)
|
||||
// The marker passed the gate: the stream was not cancelled and nothing rejected.
|
||||
expect(notify).not.toHaveBeenCalledWith('git.cancelResponseStream', expect.anything())
|
||||
expect(settled).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
// Why: the marker gate only bounds the declared size; a relay that declares small and
|
||||
// then streams forever (never sending responseEnd) must be cut off mid-flight.
|
||||
it('rejects a relay that streams past its declared size', async () => {
|
||||
const chunk = Buffer.from('a'.repeat(8), 'utf-8')
|
||||
const { mux, chunkHandlers, notify } = createFakeMux({
|
||||
streamId: 9,
|
||||
totalBytes: chunk.length,
|
||||
chunkCount: 1
|
||||
})
|
||||
|
||||
const result = requestGitStreamable(mux, 'git.diff', { cwd: '/repo' })
|
||||
await Promise.resolve()
|
||||
await Promise.resolve()
|
||||
|
||||
const feed = chunkHandlers[0]
|
||||
feed({ streamId: 9, seq: 0, data: chunk.toString('base64') })
|
||||
feed({ streamId: 9, seq: 1, data: chunk.toString('base64') })
|
||||
|
||||
await expect(result).rejects.toThrow(/sent more chunks than declared/)
|
||||
expect(notify).toHaveBeenCalledWith('git.cancelResponseStream', { streamId: 9 })
|
||||
})
|
||||
|
||||
it('rejects a chunk that overruns the declared byte total', async () => {
|
||||
const { mux, chunkHandlers, notify } = createFakeMux({
|
||||
streamId: 10,
|
||||
totalBytes: 4,
|
||||
chunkCount: 2
|
||||
})
|
||||
|
||||
const result = requestGitStreamable(mux, 'git.diff', { cwd: '/repo' })
|
||||
await Promise.resolve()
|
||||
await Promise.resolve()
|
||||
|
||||
chunkHandlers[0]({
|
||||
streamId: 10,
|
||||
seq: 0,
|
||||
data: Buffer.from('abcdefgh', 'utf-8').toString('base64')
|
||||
})
|
||||
|
||||
await expect(result).rejects.toThrow(/overran declared size: 8\/4 bytes/)
|
||||
expect(notify).toHaveBeenCalledWith('git.cancelResponseStream', { streamId: 10 })
|
||||
})
|
||||
})
|
||||
|
||||
@@ -156,6 +156,25 @@ export function requestGitStreamable(
|
||||
return
|
||||
}
|
||||
const decoded = Buffer.from(data, 'base64')
|
||||
// Why: without this the declared size is advisory — a relay that never sends
|
||||
// responseEnd could stream past it forever and OOM the main process. The relay
|
||||
// derives both figures from the same buffer, so an honest stream cannot trip it.
|
||||
if (expectedSeq >= chunkCount) {
|
||||
fail(
|
||||
new GitResponseStreamError(
|
||||
`Git stream ${streamIdRef.current} sent more chunks than declared: ${chunkCount}`
|
||||
)
|
||||
)
|
||||
return
|
||||
}
|
||||
if (receivedBytes + decoded.length > totalBytes) {
|
||||
fail(
|
||||
new GitResponseStreamError(
|
||||
`Git stream ${streamIdRef.current} overran declared size: ${receivedBytes + decoded.length}/${totalBytes} bytes`
|
||||
)
|
||||
)
|
||||
return
|
||||
}
|
||||
parts.push(decoded)
|
||||
receivedBytes += decoded.length
|
||||
expectedSeq += 1
|
||||
|
||||
Reference in New Issue
Block a user