This is an archive of the discontinued Mercurial Phabricator instance.

Differential D2870

wireproto: support for receiving multiple requests
ClosedPublic

Authored by indygreg on Mar 14 2018, 9:10 PM.

Details

Reviewers
durin42
Group Reviewers
hg-reviewers
Commits
rHGc5e9c3b47366: wireproto: support for receiving multiple requests
Summary

Now that we have request IDs on each frame and a specification
that allows multiple requests to be issued simultaneously,
possibly interleaved, let's teach the server to deal with that.

Instead of tracking the state for *the* active command request,
we instead track the state of each receiving command by its
request ID. The multiple states in our state machine for processing
each command's state has been collapsed into a single state for
"receiving commands."

Tests have been added so our branch coverage covers all meaningful
branches.

However, we did lose some logical coverage. The implementation
of this new feature opens up the door to a server having partial
command requests when end of input is reached. We will probably
want a mechanism to deal with partial requests. For now, I've
tracked that as a known issue in the class docstring. I've
also noted an abuse vector that becomes a little bit easier to
exploit with this feature.

Diff Detail

Repository
rHG Mercurial
Lint
Automatic diff as part of commit; lint not applicable.
Unit
Automatic diff as part of commit; unit tests not applicable.

Event Timeline

indygreg created this revision.Mar 14 2018, 9:10 PM
indygreg updated this revision to Diff 7146.Mar 19 2018, 7:59 PM
durin42 accepted this revision.Mar 21 2018, 9:32 PM
This revision is now accepted and ready to land.Mar 21 2018, 9:32 PM
Closed by commit rHGc5e9c3b47366: wireproto: support for receiving multiple requests (authored by indygreg). · Explain WhyMar 21 2018, 9:41 PM
This revision was automatically updated to reflect the committed changes.

Revision Contents
Changeset List

Diff 7234

mercurial/wireprotoframing.py

of the command to run are given in the data structure. of the command to run are given in the data structure.
wantframe wantframe
Indicates that nothing of interest happened and the server is waiting on Indicates that nothing of interest happened and the server is waiting on
more frames from the client before anything interesting can be done. more frames from the client before anything interesting can be done.
noop noop
Indicates no additional action is required. Indicates no additional action is required.
Known Issues
------------
There are no limits to the number of partially received commands or their
size. A malicious client could stream command request data and exhaust the
server's memory.
Partially received commands are not acted upon when end of input is
reached. Should the server error if it receives a partial request?
Should the client send a message to abort a partially transmitted request
to facilitate graceful shutdown?
Active requests that haven't been responded to aren't tracked. This means
that if we receive a command and instruct its dispatch, another command
with its request ID can come in over the wire and there will be a race
between who responds to what.
""" """
def __init__(self, deferoutput=False): def __init__(self, deferoutput=False):
"""Construct a new server reactor. """Construct a new server reactor.
``deferoutput`` can be used to indicate that no output frames should be ``deferoutput`` can be used to indicate that no output frames should be
instructed to be sent until input has been exhausted. In this mode, instructed to be sent until input has been exhausted. In this mode,
events that would normally generate output frames (such as a command events that would normally generate output frames (such as a command
response being ready) will instead defer instructing the consumer to response being ready) will instead defer instructing the consumer to
send those frames. This is useful for half-duplex transports where the send those frames. This is useful for half-duplex transports where the
sender cannot receive until all data has been transmitted. sender cannot receive until all data has been transmitted.
""" """
self._deferoutput = deferoutput self._deferoutput = deferoutput
self._state = 'idle' self._state = 'idle'
self._bufferedframegens = [] self._bufferedframegens = []
self._activerequestid = None # request id -> dict of commands that are actively being received.
self._activecommand = None self._receivingcommands = {}
self._activeargs = None
self._activedata = None
self._expectingargs = None
self._expectingdata = None
self._activeargname = None
self._activeargchunks = None
def onframerecv(self, requestid, frametype, frameflags, payload): def onframerecv(self, requestid, frametype, frameflags, payload):
"""Process a frame that has been received off the wire. """Process a frame that has been received off the wire.
Returns a dict with an ``action`` key that details what action, Returns a dict with an ``action`` key that details what action,
if any, the consumer should take next. if any, the consumer should take next.
""" """
handlers = { handlers = {
'idle': self._onframeidle, 'idle': self._onframeidle,
'command-receiving-args': self._onframereceivingargs, 'command-receiving': self._onframecommandreceiving,
'command-receiving-data': self._onframereceivingdata,
'errored': self._onframeerrored, 'errored': self._onframeerrored,
} }
meth = handlers.get(self._state) meth = handlers.get(self._state)
if not meth: if not meth:
raise error.ProgrammingError('unhandled state: %s' % self._state) raise error.ProgrammingError('unhandled state: %s' % self._state)
return meth(requestid, frametype, frameflags, payload) return meth(requestid, frametype, frameflags, payload)
} }
def oninputeof(self): def oninputeof(self):
"""Signals that end of input has been received. """Signals that end of input has been received.
No more frames will be received. All pending activity should be No more frames will be received. All pending activity should be
completed. completed.
""" """
# TODO should we do anything about in-flight commands?
if not self._deferoutput or not self._bufferedframegens: if not self._deferoutput or not self._bufferedframegens:
return 'noop', {} return 'noop', {}
# If we buffered all our responses, emit those. # If we buffered all our responses, emit those.
def makegen(): def makegen():
for gen in self._bufferedframegens: for gen in self._bufferedframegens:
for frame in gen: for frame in gen:
yield frame yield frame
return 'sendframes', { return 'sendframes', {
'framegen': makegen(), 'framegen': makegen(),
} }
def onapplicationerror(self, requestid, msg): def onapplicationerror(self, requestid, msg):
return 'sendframes', { return 'sendframes', {
'framegen': createerrorframe(requestid, msg, application=True), 'framegen': createerrorframe(requestid, msg, application=True),
} }
def _makeerrorresult(self, msg): def _makeerrorresult(self, msg):
return 'error', { return 'error', {
'message': msg, 'message': msg,
} }
def _makeruncommandresult(self): def _makeruncommandresult(self, requestid):
entry = self._receivingcommands[requestid]
del self._receivingcommands[requestid]
if self._receivingcommands:
self._state = 'command-receiving'
else:
self._state = 'idle'
return 'runcommand', { return 'runcommand', {
'requestid': self._activerequestid, 'requestid': requestid,
'command': self._activecommand, 'command': entry['command'],
'args': self._activeargs, 'args': entry['args'],
'data': self._activedata.getvalue() if self._activedata else None, 'data': entry['data'].getvalue() if entry['data'] else None,
} }
def _makewantframeresult(self): def _makewantframeresult(self):
return 'wantframe', { return 'wantframe', {
'state': self._state, 'state': self._state,
} }
def _onframeidle(self, requestid, frametype, frameflags, payload): def _onframeidle(self, requestid, frametype, frameflags, payload):
# The only frame type that should be received in this state is a # The only frame type that should be received in this state is a
# command request. # command request.
if frametype != FRAME_TYPE_COMMAND_NAME: if frametype != FRAME_TYPE_COMMAND_NAME:
self._state = 'errored' self._state = 'errored'
return self._makeerrorresult( return self._makeerrorresult(
_('expected command frame; got %d') % frametype) _('expected command frame; got %d') % frametype)
self._activerequestid = requestid if requestid in self._receivingcommands:
self._activecommand = payload self._state = 'errored'
self._activeargs = {} return self._makeerrorresult(
self._activedata = None _('request with ID %d already received') % requestid)
if frameflags & FLAG_COMMAND_NAME_EOS: expectingargs = bool(frameflags & FLAG_COMMAND_NAME_HAVE_ARGS)
return self._makeruncommandresult() expectingdata = bool(frameflags & FLAG_COMMAND_NAME_HAVE_DATA)
self._expectingargs = bool(frameflags & FLAG_COMMAND_NAME_HAVE_ARGS) self._receivingcommands[requestid] = {
self._expectingdata = bool(frameflags & FLAG_COMMAND_NAME_HAVE_DATA) 'command': payload,
'args': {},
'data': None,
'expectingargs': expectingargs,
'expectingdata': expectingdata,
}
if self._expectingargs: if frameflags & FLAG_COMMAND_NAME_EOS:
self._state = 'command-receiving-args' return self._makeruncommandresult(requestid)
return self._makewantframeresult()
elif self._expectingdata: if expectingargs or expectingdata:
self._activedata = util.bytesio() self._state = 'command-receiving'
self._state = 'command-receiving-data'
return self._makewantframeresult() return self._makewantframeresult()
else: else:
self._state = 'errored' self._state = 'errored'
return self._makeerrorresult(_('missing frame flags on ' return self._makeerrorresult(_('missing frame flags on '
'command frame')) 'command frame'))
def _onframereceivingargs(self, requestid, frametype, frameflags, payload): def _onframecommandreceiving(self, requestid, frametype, frameflags,
if frametype != FRAME_TYPE_COMMAND_ARGUMENT: payload):
# It could be a new command request. Process it as such.
if frametype == FRAME_TYPE_COMMAND_NAME:
return self._onframeidle(requestid, frametype, frameflags, payload)
# All other frames should be related to a command that is currently
# receiving.
if requestid not in self._receivingcommands:
self._state = 'errored'
return self._makeerrorresult(
_('received frame for request that is not receiving: %d') %
requestid)
entry = self._receivingcommands[requestid]
if frametype == FRAME_TYPE_COMMAND_ARGUMENT:
if not entry['expectingargs']:
self._state = 'errored' self._state = 'errored'
return self._makeerrorresult(_('expected command argument ' return self._makeerrorresult(_(
'frame; got %d') % frametype) 'received command argument frame for request that is not '
'expecting arguments: %d') % requestid)
return self._handlecommandargsframe(requestid, entry, frametype,
frameflags, payload)
elif frametype == FRAME_TYPE_COMMAND_DATA:
if not entry['expectingdata']:
self._state = 'errored'
return self._makeerrorresult(_(
'received command data frame for request that is not '
'expecting data: %d') % requestid)
if entry['data'] is None:
entry['data'] = util.bytesio()
return self._handlecommanddataframe(requestid, entry, frametype,
frameflags, payload)
def _handlecommandargsframe(self, requestid, entry, frametype, frameflags,
payload):
# The frame and state of command should have already been validated.
assert frametype == FRAME_TYPE_COMMAND_ARGUMENT
offset = 0 offset = 0
namesize, valuesize = ARGUMENT_FRAME_HEADER.unpack_from(payload) namesize, valuesize = ARGUMENT_FRAME_HEADER.unpack_from(payload)
offset += ARGUMENT_FRAME_HEADER.size offset += ARGUMENT_FRAME_HEADER.size
# The argument name MUST fit inside the frame. # The argument name MUST fit inside the frame.
argname = bytes(payload[offset:offset + namesize]) argname = bytes(payload[offset:offset + namesize])
offset += namesize offset += namesize
if len(argname) != namesize: if len(argname) != namesize:
self._state = 'errored' self._state = 'errored'
return self._makeerrorresult(_('malformed argument frame: ' return self._makeerrorresult(_('malformed argument frame: '
'partial argument name')) 'partial argument name'))
argvalue = bytes(payload[offset:]) argvalue = bytes(payload[offset:])
# Argument value spans multiple frames. Record our active state # Argument value spans multiple frames. Record our active state
# and wait for the next frame. # and wait for the next frame.
if frameflags & FLAG_COMMAND_ARGUMENT_CONTINUATION: if frameflags & FLAG_COMMAND_ARGUMENT_CONTINUATION:
raise error.ProgrammingError('not yet implemented') raise error.ProgrammingError('not yet implemented')
self._activeargname = argname
self._activeargchunks = [argvalue]
self._state = 'command-arg-continuation'
return self._makewantframeresult()
# Common case: the argument value is completely contained in this # Common case: the argument value is completely contained in this
# frame. # frame.
if len(argvalue) != valuesize: if len(argvalue) != valuesize:
self._state = 'errored' self._state = 'errored'
return self._makeerrorresult(_('malformed argument frame: ' return self._makeerrorresult(_('malformed argument frame: '
'partial argument value')) 'partial argument value'))
self._activeargs[argname] = argvalue entry['args'][argname] = argvalue
if frameflags & FLAG_COMMAND_ARGUMENT_EOA: if frameflags & FLAG_COMMAND_ARGUMENT_EOA:
if self._expectingdata: if entry['expectingdata']:
self._state = 'command-receiving-data'
self._activedata = util.bytesio()
# TODO signal request to run a command once we don't # TODO signal request to run a command once we don't
# buffer data frames. # buffer data frames.
return self._makewantframeresult() return self._makewantframeresult()
else: else:
self._state = 'waiting' return self._makeruncommandresult(requestid)
return self._makeruncommandresult()
else: else:
return self._makewantframeresult() return self._makewantframeresult()
def _onframereceivingdata(self, requestid, frametype, frameflags, payload): def _handlecommanddataframe(self, requestid, entry, frametype, frameflags,
if frametype != FRAME_TYPE_COMMAND_DATA: payload):
self._state = 'errored' assert frametype == FRAME_TYPE_COMMAND_DATA
return self._makeerrorresult(_('expected command data frame; '
'got %d') % frametype)
# TODO support streaming data instead of buffering it. # TODO support streaming data instead of buffering it.
self._activedata.write(payload) entry['data'].write(payload)
if frameflags & FLAG_COMMAND_DATA_CONTINUATION: if frameflags & FLAG_COMMAND_DATA_CONTINUATION:
return self._makewantframeresult() return self._makewantframeresult()
elif frameflags & FLAG_COMMAND_DATA_EOS: elif frameflags & FLAG_COMMAND_DATA_EOS:
self._activedata.seek(0) entry['data'].seek(0)
self._state = 'idle' return self._makeruncommandresult(requestid)
return self._makeruncommandresult()
else: else:
self._state = 'errored' self._state = 'errored'
return self._makeerrorresult(_('command data frame without ' return self._makeerrorresult(_('command data frame without '
'flags')) 'flags'))
def _onframeerrored(self, requestid, frametype, frameflags, payload): def _onframeerrored(self, requestid, frametype, frameflags, payload):
return self._makeerrorresult(_('server already errored')) return self._makeerrorresult(_('server already errored'))

tests/test-http-api-httpv2.t

s> host: $LOCALIP:$HGPORT\r\n (glob) s> host: $LOCALIP:$HGPORT\r\n (glob)
s> \r\n s> \r\n
s> \x08\x00\x00\x01\x00\x12command1\x0b\x00\x00\x01\x00 \x03\x00\x04\x00fooval1\x0b\x00\x00\x01\x00"\x04\x00\x03\x00bar1val s> \x08\x00\x00\x01\x00\x12command1\x0b\x00\x00\x01\x00 \x03\x00\x04\x00fooval1\x0b\x00\x00\x01\x00"\x04\x00\x03\x00bar1val
s> makefile('rb', None) s> makefile('rb', None)
s> HTTP/1.1 200 OK\r\n s> HTTP/1.1 200 OK\r\n
s> Server: testing stub value\r\n s> Server: testing stub value\r\n
s> Date: $HTTP_DATE$\r\n s> Date: $HTTP_DATE$\r\n
s> Content-Type: text/plain\r\n s> Content-Type: text/plain\r\n
s> Content-Length: 332\r\n s> Content-Length: 322\r\n
s> \r\n s> \r\n
s> received: 1 2 1 command1\n s> received: 1 2 1 command1\n
s> ["wantframe", {"state": "command-receiving-args"}]\n s> ["wantframe", {"state": "command-receiving"}]\n
s> received: 2 0 1 \x03\x00\x04\x00fooval1\n s> received: 2 0 1 \x03\x00\x04\x00fooval1\n
s> ["wantframe", {"state": "command-receiving-args"}]\n s> ["wantframe", {"state": "command-receiving"}]\n
s> received: 2 2 1 \x04\x00\x03\x00bar1val\n s> received: 2 2 1 \x04\x00\x03\x00bar1val\n
s> ["runcommand", {"args": {"bar1": "val", "foo": "val1"}, "command": "command1", "data": null, "requestid": 1}]\n s> ["runcommand", {"args": {"bar1": "val", "foo": "val1"}, "command": "command1", "data": null, "requestid": 1}]\n
s> received: <no frame>\n s> received: <no frame>\n
s> {"action": "noop"} s> {"action": "noop"}
$ cat error.log $ cat error.log

tests/test-wireproto-serverreactor.py

"""Command argument frame when not running a command is an error.""" """Command argument frame when not running a command is an error."""
result = self._sendsingleframe(makereactor(), result = self._sendsingleframe(makereactor(),
b'1 command-argument 0 ignored') b'1 command-argument 0 ignored')
self.assertaction(result, 'error') self.assertaction(result, 'error')
self.assertEqual(result[1], { self.assertEqual(result[1], {
'message': b'expected command frame; got 2', 'message': b'expected command frame; got 2',
}) })
def testunexpectedcommandargumentreceiving(self):
"""Same as above but the command is receiving."""
results = list(sendframes(makereactor(), [
ffs(b'1 command-name have-data command'),
ffs(b'1 command-argument eoa ignored'),
]))
self.assertaction(results[1], 'error')
self.assertEqual(results[1][1], {
'message': b'received command argument frame for request that is '
b'not expecting arguments: 1',
})
def testunexpectedcommanddata(self): def testunexpectedcommanddata(self):
"""Command argument frame when not running a command is an error.""" """Command argument frame when not running a command is an error."""
result = self._sendsingleframe(makereactor(), result = self._sendsingleframe(makereactor(),
b'1 command-data 0 ignored') b'1 command-data 0 ignored')
self.assertaction(result, 'error') self.assertaction(result, 'error')
self.assertEqual(result[1], { self.assertEqual(result[1], {
'message': b'expected command frame; got 3', 'message': b'expected command frame; got 3',
}) })
def testunexpectedcommanddatareceiving(self):
"""Same as above except the command is receiving."""
results = list(sendframes(makereactor(), [
ffs(b'1 command-name have-args command'),
ffs(b'1 command-data eos ignored'),
]))
self.assertaction(results[1], 'error')
self.assertEqual(results[1][1], {
'message': b'received command data frame for request that is not '
b'expecting data: 1',
})
def testmissingcommandframeflags(self): def testmissingcommandframeflags(self):
"""Command name frame must have flags set.""" """Command name frame must have flags set."""
result = self._sendsingleframe(makereactor(), result = self._sendsingleframe(makereactor(),
b'1 command-name 0 command') b'1 command-name 0 command')
self.assertaction(result, 'error') self.assertaction(result, 'error')
self.assertEqual(result[1], { self.assertEqual(result[1], {
'message': b'missing frame flags on command frame', 'message': b'missing frame flags on command frame',
}) })
def testconflictingrequestid(self):
"""Multiple fully serviced commands with same request ID is allowed."""
results = list(sendframes(makereactor(), [
ffs(b'1 command-name eos command'),
ffs(b'1 command-name eos command'),
ffs(b'1 command-name eos command'),
]))
for i in range(3):
self.assertaction(results[i], 'runcommand')
self.assertEqual(results[i][1], {
'requestid': 1,
'command': b'command',
'args': {},
'data': None,
})
def testconflictingrequestid(self):
"""Request ID for new command matching in-flight command is illegal."""
results = list(sendframes(makereactor(), [
ffs(b'1 command-name have-args command'),
ffs(b'1 command-name eos command'),
]))
self.assertaction(results[0], 'wantframe')
self.assertaction(results[1], 'error')
self.assertEqual(results[1][1], {
'message': b'request with ID 1 already received',
})
def testinterleavedcommands(self):
results = list(sendframes(makereactor(), [
ffs(b'1 command-name have-args command1'),
ffs(b'3 command-name have-args command3'),
ffs(br'1 command-argument 0 \x03\x00\x03\x00foobar'),
ffs(br'3 command-argument 0 \x03\x00\x03\x00bizbaz'),
ffs(br'3 command-argument eoa \x03\x00\x03\x00keyval'),
ffs(br'1 command-argument eoa \x04\x00\x03\x00key1val'),
]))
self.assertEqual([t[0] for t in results], [
'wantframe',
'wantframe',
'wantframe',
'wantframe',
'runcommand',
'runcommand',
])
self.assertEqual(results[4][1], {
'requestid': 3,
'command': 'command3',
'args': {b'biz': b'baz', b'key': b'val'},
'data': None,
})
self.assertEqual(results[5][1], {
'requestid': 1,
'command': 'command1',
'args': {b'foo': b'bar', b'key1': b'val'},
'data': None,
})
def testmissingargumentframe(self): def testmissingargumentframe(self):
# This test attempts to test behavior when reactor has an incomplete
# command request waiting on argument data. But it doesn't handle that
# scenario yet. So this test does nothing of value.
frames = [ frames = [
ffs(b'1 command-name have-args command'), ffs(b'1 command-name have-args command'),
ffs(b'1 command-name 0 ignored'),
] ]
results = list(sendframes(makereactor(), frames)) results = list(sendframes(makereactor(), frames))
self.assertEqual(len(results), 2)
self.assertaction(results[0], 'wantframe') self.assertaction(results[0], 'wantframe')
self.assertaction(results[1], 'error')
self.assertEqual(results[1][1], {
'message': b'expected command argument frame; got 1',
})
def testincompleteargumentname(self): def testincompleteargumentname(self):
"""Argument frame with incomplete name.""" """Argument frame with incomplete name."""
frames = [ frames = [
ffs(b'1 command-name have-args command1'), ffs(b'1 command-name have-args command1'),
ffs(br'1 command-argument eoa \x04\x00\xde\xadfoo'), ffs(br'1 command-argument eoa \x04\x00\xde\xadfoo'),
] ]
self.assertEqual(len(results), 2) self.assertEqual(len(results), 2)
self.assertaction(results[0], 'wantframe') self.assertaction(results[0], 'wantframe')
self.assertaction(results[1], 'error') self.assertaction(results[1], 'error')
self.assertEqual(results[1][1], { self.assertEqual(results[1][1], {
'message': b'malformed argument frame: partial argument value', 'message': b'malformed argument frame: partial argument value',
}) })
def testmissingcommanddataframe(self): def testmissingcommanddataframe(self):
# The reactor doesn't currently handle partially received commands.
# So this test is failing to do anything with request 1.
frames = [ frames = [
ffs(b'1 command-name have-data command1'), ffs(b'1 command-name have-data command1'),
ffs(b'1 command-name eos command2'), ffs(b'3 command-name eos command2'),
] ]
results = list(sendframes(makereactor(), frames)) results = list(sendframes(makereactor(), frames))
self.assertEqual(len(results), 2) self.assertEqual(len(results), 2)
self.assertaction(results[0], 'wantframe') self.assertaction(results[0], 'wantframe')
self.assertaction(results[1], 'error') self.assertaction(results[1], 'runcommand')
self.assertEqual(results[1][1], {
'message': b'expected command data frame; got 1',
})
def testmissingcommanddataframeflags(self): def testmissingcommanddataframeflags(self):
frames = [ frames = [
ffs(b'1 command-name have-data command1'), ffs(b'1 command-name have-data command1'),
ffs(b'1 command-data 0 data'), ffs(b'1 command-data 0 data'),
] ]
results = list(sendframes(makereactor(), frames)) results = list(sendframes(makereactor(), frames))
self.assertEqual(len(results), 2) self.assertEqual(len(results), 2)
self.assertaction(results[0], 'wantframe') self.assertaction(results[0], 'wantframe')
self.assertaction(results[1], 'error') self.assertaction(results[1], 'error')
self.assertEqual(results[1][1], { self.assertEqual(results[1][1], {
'message': b'command data frame without flags', 'message': b'command data frame without flags',
}) })
def testframefornonreceivingrequest(self):
"""Receiving a frame for a command that is not receiving is illegal."""
results = list(sendframes(makereactor(), [
ffs(b'1 command-name eos command1'),
ffs(b'3 command-name have-data command3'),
ffs(b'1 command-argument eoa ignored'),
]))
self.assertaction(results[2], 'error')
self.assertEqual(results[2][1], {
'message': b'received frame for request that is not receiving: 1',
})
def testsimpleresponse(self): def testsimpleresponse(self):
"""Bytes response to command sends result frames.""" """Bytes response to command sends result frames."""
reactor = makereactor() reactor = makereactor()
list(sendcommandframes(reactor, 1, b'mycommand', {})) list(sendcommandframes(reactor, 1, b'mycommand', {}))
result = reactor.onbytesresponseready(1, b'response') result = reactor.onbytesresponseready(1, b'response')
self.assertaction(result, 'sendframes') self.assertaction(result, 'sendframes')
self.assertframesequal(result[1]['framegen'], [ self.assertframesequal(result[1]['framegen'], [