( )⚙ D2869 wireproto: add request IDs to frames

This is an archive of the discontinued Mercurial Phabricator instance.

Differential D2869

wireproto: add request IDs to frames
ClosedPublic

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

Details

Reviewers
durin42
Group Reviewers
hg-reviewers
Commits
rHG2ec1fb9de638: wireproto: add request IDs to frames
Summary

One of my primary goals with the new wire protocol is to make
operations faster and enable both client and server-side
operations to scale to multiple CPU cores.

One of the ways we make server interactions faster is by reducing
the number of round trips to that server.

With the existing wire protocol, the "batch" command facilitates
executing multiple commands from a single request payload. The way
it works is the requests for multiple commands are serialized. The
server executes those commands sequentially then serializes all
their results. As an optimization for reducing round trips, this
is very effective. The technical implementation, however, is pretty
bad and suffers from a number of deficiencies. For example, it
creates a new place where authorization to run a command must be
checked. (The lack of this checking in older Mercurial releases
was CVE-2018-1000132.)

The principles behind the "batch" command are sound. However, the
execution is not. Therefore, I want to ditch "batch" in the
new wire protocol and have protocol level support for issuing
multiple requests in a single round trip.

This commit introduces support in the frame-based wire protocol to
facilitate this. We do this by adding a "request ID" to each frame.
If a server sees frames associated with different "request IDs," it
handles them as separate requests. All of this happening possibly
as part of the same message from client to server (the same request
body in the case of HTTP).

We /could/ model the exchange the way pipelined HTTP requests do,
where the server processes requests in order they are issued and
received. But this artifically constrains scalability. A better
model is to allow multi-requests to be executed concurrently and
for responses to be sent and handled concurrently. So the
specification explicitly allows this. There is some work to be done
around specifying dependencies between multi-requests. We take
the easy road for now and punt on this problem, declaring that
if order is important, clients must not issue the request until
responses to dependent requests have been received.

This commit focuses on the boilerplate of implementing the request
ID. The server reactor still can't manage multiple, in-flight
request IDs. This will be addressed in a subsequent commit.

Because the wire semantics have changed, we bump the version of the
media type.

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 7145.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 rHG2ec1fb9de638: wireproto: add request IDs to frames (authored by indygreg). · Explain WhyMar 21 2018, 9:40 PM
This revision was automatically updated to reflect the committed changes.

Revision Contents
Changeset List

Diff 7233

mercurial/debugcommands.py

Specifying Unified Frame-Based Protocol Frames Specifying Unified Frame-Based Protocol Frames
---------------------------------------------- ----------------------------------------------
It is possible to emit a *Unified Frame-Based Protocol* by using special It is possible to emit a *Unified Frame-Based Protocol* by using special
syntax. syntax.
A frame is composed as a type, flags, and payload. These can be parsed A frame is composed as a type, flags, and payload. These can be parsed
from a string of the form ``<type> <flags> <payload>``. That is, 3 from a string of the form ``<requestid> <type> <flags> <payload>``. That is,
space-delimited strings. 4 space-delimited strings.
``payload`` is the simplest: it is evaluated as a Python byte string ``payload`` is the simplest: it is evaluated as a Python byte string
literal. literal.
``requestid`` is an integer defining the request identifier.
``type`` can be an integer value for the frame type or the string name ``type`` can be an integer value for the frame type or the string name
of the type. The strings are defined in ``wireprotoframing.py``. e.g. of the type. The strings are defined in ``wireprotoframing.py``. e.g.
``command-name``. ``command-name``.
``flags`` is a ``|`` delimited list of flag components. Each component ``flags`` is a ``|`` delimited list of flag components. Each component
(and there can be just one) can be an integer or a flag name for the (and there can be just one) can be an integer or a flag name for the
specified frame type. Values are resolved to integers and then bitwise specified frame type. Values are resolved to integers and then bitwise
OR'd together. OR'd together.

mercurial/help/internals/wireprotocol.txt

(works similarly on HTTP, SSH, etc). (works similarly on HTTP, SSH, etc).
To operate the protocol, a bi-directional, half-duplex pipe supporting To operate the protocol, a bi-directional, half-duplex pipe supporting
ordered sends and receives is required. That is, each peer has one pipe ordered sends and receives is required. That is, each peer has one pipe
for sending data and another for receiving. for sending data and another for receiving.
The protocol is request-response based: the client issues requests to The protocol is request-response based: the client issues requests to
the server, which issues replies to those requests. Server-initiated the server, which issues replies to those requests. Server-initiated
messaging is not supported. messaging is not currently supported, but this specification carves
out room to implement it.
All data is read and written in atomic units called *frames*. These All data is read and written in atomic units called *frames*. These
are conceptually similar to TCP packets. Higher-level functionality are conceptually similar to TCP packets. Higher-level functionality
is built on the exchange and processing of frames. is built on the exchange and processing of frames.
Frames begin with a 4 octet header followed by a variable length All frames are associated with a numbered request. Frames can thus
be logically grouped by their request ID.
Frames begin with a 6 octet header followed by a variable length
payload:: payload::
+-----------------------------------------------+ +-----------------------------------------------+
| Length (24) | | Length (24) |
+-----------+-----------------------------------+ +---------------------------------+-------------+
| Type (4) | | Request ID (16) |
+-----------+ +----------+-----------+----------+
| Flags (4) | | Type (4) | Flags (4) |
+===========+===================================================| +==========+===========+========================================|
| Frame Payload (0...) ... | Frame Payload (0...) ...
+---------------------------------------------------------------+ +---------------------------------------------------------------+
The length of the frame payload is expressed as an unsigned 24 bit The length of the frame payload is expressed as an unsigned 24 bit
little endian integer. Values larger than 65535 MUST NOT be used unless little endian integer. Values larger than 65535 MUST NOT be used unless
given permission by the server as part of the negotiated capabilities given permission by the server as part of the negotiated capabilities
during the handshake. The frame header is not part of the advertised during the handshake. The frame header is not part of the advertised
frame length. frame length.
The 16-bit ``Request ID`` field denotes the integer request identifier,
stored as an unsigned little endian integer. Odd numbered requests are
client-initiated. Even numbered requests are server-initiated. This
refers to where the *request* was initiated - not where the *frame* was
initiated, so servers will send frames with odd ``Request ID`` in
response to client-initiated requests. Implementations are advised to
start ordering request identifiers at ``1`` and ``0``, increment by
``2``, and wrap around if all available numbers have been exhausted.
The 4-bit ``Type`` field denotes the type of message being sent. The 4-bit ``Type`` field denotes the type of message being sent.
The 4-bit ``Flags`` field defines special, per-type attributes for The 4-bit ``Flags`` field defines special, per-type attributes for
the frame. the frame.
The sections below define the frame types and their behavior. The sections below define the frame types and their behavior.
Command Request (``0x01``) Command Request (``0x01``)
Issuing Commands Issuing Commands
---------------- ----------------
A client can request that a remote run a command by sending it A client can request that a remote run a command by sending it
frames defining that command. This logical stream is composed of frames defining that command. This logical stream is composed of
1 ``Command Request`` frame, 0 or more ``Command Argument`` frames, 1 ``Command Request`` frame, 0 or more ``Command Argument`` frames,
and 0 or more ``Command Data`` frames. and 0 or more ``Command Data`` frames.
All frames composing a single command request MUST be associated with
the same ``Request ID``.
Clients MAY send additional command requests without waiting on the
response to a previous command request. If they do so, they MUST ensure
that the ``Request ID`` field of outbound frames does not conflict
with that of an active ``Request ID`` whose response has not yet been
fully received.
Servers MAY respond to commands in a different order than they were
sent over the wire. Clients MUST be prepared to deal with this. Servers
also MAY start executing commands in a different order than they were
received, or MAY execute multiple commands concurrently.
If there is a dependency between commands or a race condition between
commands executing (e.g. a read-only command that depends on the results
of a command that mutates the repository), then clients MUST NOT send
frames issuing a command until a response to all dependent commands has
been received.
TODO think about whether we should express dependencies between commands
to avoid roundtrip latency.
Argument frames are the recommended mechanism for transferring fixed Argument frames are the recommended mechanism for transferring fixed
sets of parameters to a command. Data frames are appropriate for sets of parameters to a command. Data frames are appropriate for
transferring variable data. A similar comparison would be to HTTP: transferring variable data. A similar comparison would be to HTTP:
argument frames are headers and the message body is data frames. argument frames are headers and the message body is data frames.
It is recommended for servers to delay the dispatch of a command It is recommended for servers to delay the dispatch of a command
until all argument frames for that command have been received. Servers until all argument frames for that command have been received. Servers
MAY impose limits on the maximum argument size. MAY impose limits on the maximum argument size.

mercurial/wireprotoframing.py

import struct import struct
from .i18n import _ from .i18n import _
from . import ( from . import (
error, error,
util, util,
) )
FRAME_HEADER_SIZE = 4 FRAME_HEADER_SIZE = 6
DEFAULT_MAX_FRAME_SIZE = 32768 DEFAULT_MAX_FRAME_SIZE = 32768
FRAME_TYPE_COMMAND_NAME = 0x01 FRAME_TYPE_COMMAND_NAME = 0x01
FRAME_TYPE_COMMAND_ARGUMENT = 0x02 FRAME_TYPE_COMMAND_ARGUMENT = 0x02
FRAME_TYPE_COMMAND_DATA = 0x03 FRAME_TYPE_COMMAND_DATA = 0x03
FRAME_TYPE_BYTES_RESPONSE = 0x04 FRAME_TYPE_BYTES_RESPONSE = 0x04
FRAME_TYPE_ERROR_RESPONSE = 0x05 FRAME_TYPE_ERROR_RESPONSE = 0x05
FRAME_TYPE_COMMAND_ARGUMENT: FLAGS_COMMAND_ARGUMENT, FRAME_TYPE_COMMAND_ARGUMENT: FLAGS_COMMAND_ARGUMENT,
FRAME_TYPE_COMMAND_DATA: FLAGS_COMMAND_DATA, FRAME_TYPE_COMMAND_DATA: FLAGS_COMMAND_DATA,
FRAME_TYPE_BYTES_RESPONSE: FLAGS_BYTES_RESPONSE, FRAME_TYPE_BYTES_RESPONSE: FLAGS_BYTES_RESPONSE,
FRAME_TYPE_ERROR_RESPONSE: FLAGS_ERROR_RESPONSE, FRAME_TYPE_ERROR_RESPONSE: FLAGS_ERROR_RESPONSE,
} }
ARGUMENT_FRAME_HEADER = struct.Struct(r'<HH') ARGUMENT_FRAME_HEADER = struct.Struct(r'<HH')
def makeframe(frametype, frameflags, payload): def makeframe(requestid, frametype, frameflags, payload):
"""Assemble a frame into a byte array.""" """Assemble a frame into a byte array."""
# TODO assert size of payload. # TODO assert size of payload.
frame = bytearray(FRAME_HEADER_SIZE + len(payload)) frame = bytearray(FRAME_HEADER_SIZE + len(payload))
# 24 bits length
# 16 bits request id
# 4 bits type
# 4 bits flags
l = struct.pack(r'<I', len(payload)) l = struct.pack(r'<I', len(payload))
frame[0:3] = l[0:3] frame[0:3] = l[0:3]
frame[3] = (frametype << 4) | frameflags struct.pack_into(r'<H', frame, 3, requestid)
frame[4:] = payload frame[5] = (frametype << 4) | frameflags
frame[6:] = payload
return frame return frame
def makeframefromhumanstring(s): def makeframefromhumanstring(s):
"""Given a string of the form: <type> <flags> <payload>, creates a frame. """Create a frame from a human readable string
Strings have the form:
<request-id> <type> <flags> <payload>
This can be used by user-facing applications and tests for creating This can be used by user-facing applications and tests for creating
frames easily without having to type out a bunch of constants. frames easily without having to type out a bunch of constants.
Request ID is an integer.
Frame type and flags can be specified by integer or named constant. Frame type and flags can be specified by integer or named constant.
Flags can be delimited by `|` to bitwise OR them together. Flags can be delimited by `|` to bitwise OR them together.
""" """
frametype, frameflags, payload = s.split(b' ', 2) requestid, frametype, frameflags, payload = s.split(b' ', 3)
requestid = int(requestid)
if frametype in FRAME_TYPES: if frametype in FRAME_TYPES:
frametype = FRAME_TYPES[frametype] frametype = FRAME_TYPES[frametype]
else: else:
frametype = int(frametype) frametype = int(frametype)
finalflags = 0 finalflags = 0
validflags = FRAME_TYPE_FLAGS[frametype] validflags = FRAME_TYPE_FLAGS[frametype]
for flag in frameflags.split(b'|'): for flag in frameflags.split(b'|'):
if flag in validflags: if flag in validflags:
finalflags |= validflags[flag] finalflags |= validflags[flag]
else: else:
finalflags |= int(flag) finalflags |= int(flag)
payload = util.unescapestr(payload) payload = util.unescapestr(payload)
return makeframe(frametype, finalflags, payload) return makeframe(requestid, frametype, finalflags, payload)
def parseheader(data): def parseheader(data):
"""Parse a unified framing protocol frame header from a buffer. """Parse a unified framing protocol frame header from a buffer.
The header is expected to be in the buffer at offset 0 and the The header is expected to be in the buffer at offset 0 and the
buffer is expected to be large enough to hold a full header. buffer is expected to be large enough to hold a full header.
""" """
# 24 bits payload length (little endian) # 24 bits payload length (little endian)
# 4 bits frame type # 4 bits frame type
# 4 bits frame flags # 4 bits frame flags
# ... payload # ... payload
framelength = data[0] + 256 * data[1] + 16384 * data[2] framelength = data[0] + 256 * data[1] + 16384 * data[2]
typeflags = data[3] requestid = struct.unpack_from(r'<H', data, 3)[0]
typeflags = data[5]
frametype = (typeflags & 0xf0) >> 4 frametype = (typeflags & 0xf0) >> 4
frameflags = typeflags & 0x0f frameflags = typeflags & 0x0f
return frametype, frameflags, framelength return requestid, frametype, frameflags, framelength
def readframe(fh): def readframe(fh):
"""Read a unified framing protocol frame from a file object. """Read a unified framing protocol frame from a file object.
Returns a 3-tuple of (type, flags, payload) for the decoded frame or Returns a 3-tuple of (type, flags, payload) for the decoded frame or
None if no frame is available. May raise if a malformed frame is None if no frame is available. May raise if a malformed frame is
seen. seen.
""" """
header = bytearray(FRAME_HEADER_SIZE) header = bytearray(FRAME_HEADER_SIZE)
readcount = fh.readinto(header) readcount = fh.readinto(header)
if readcount == 0: if readcount == 0:
return None return None
if readcount != FRAME_HEADER_SIZE: if readcount != FRAME_HEADER_SIZE:
raise error.Abort(_('received incomplete frame: got %d bytes: %s') % raise error.Abort(_('received incomplete frame: got %d bytes: %s') %
(readcount, header)) (readcount, header))
frametype, frameflags, framelength = parseheader(header) requestid, frametype, frameflags, framelength = parseheader(header)
payload = fh.read(framelength) payload = fh.read(framelength)
if len(payload) != framelength: if len(payload) != framelength:
raise error.Abort(_('frame length error: expected %d; got %d') % raise error.Abort(_('frame length error: expected %d; got %d') %
(framelength, len(payload))) (framelength, len(payload)))
return frametype, frameflags, payload return requestid, frametype, frameflags, payload
def createcommandframes(cmd, args, datafh=None): def createcommandframes(requestid, cmd, args, datafh=None):
"""Create frames necessary to transmit a request to run a command. """Create frames necessary to transmit a request to run a command.
This is a generator of bytearrays. Each item represents a frame This is a generator of bytearrays. Each item represents a frame
ready to be sent over the wire to a peer. ready to be sent over the wire to a peer.
""" """
flags = 0 flags = 0
if args: if args:
flags |= FLAG_COMMAND_NAME_HAVE_ARGS flags |= FLAG_COMMAND_NAME_HAVE_ARGS
if datafh: if datafh:
flags |= FLAG_COMMAND_NAME_HAVE_DATA flags |= FLAG_COMMAND_NAME_HAVE_DATA
if not flags: if not flags:
flags |= FLAG_COMMAND_NAME_EOS flags |= FLAG_COMMAND_NAME_EOS
yield makeframe(FRAME_TYPE_COMMAND_NAME, flags, cmd) yield makeframe(requestid, FRAME_TYPE_COMMAND_NAME, flags, cmd)
for i, k in enumerate(sorted(args)): for i, k in enumerate(sorted(args)):
v = args[k] v = args[k]
last = i == len(args) - 1 last = i == len(args) - 1
# TODO handle splitting of argument values across frames. # TODO handle splitting of argument values across frames.
payload = bytearray(ARGUMENT_FRAME_HEADER.size + len(k) + len(v)) payload = bytearray(ARGUMENT_FRAME_HEADER.size + len(k) + len(v))
offset = 0 offset = 0
ARGUMENT_FRAME_HEADER.pack_into(payload, offset, len(k), len(v)) ARGUMENT_FRAME_HEADER.pack_into(payload, offset, len(k), len(v))
offset += ARGUMENT_FRAME_HEADER.size offset += ARGUMENT_FRAME_HEADER.size
payload[offset:offset + len(k)] = k payload[offset:offset + len(k)] = k
offset += len(k) offset += len(k)
payload[offset:offset + len(v)] = v payload[offset:offset + len(v)] = v
flags = FLAG_COMMAND_ARGUMENT_EOA if last else 0 flags = FLAG_COMMAND_ARGUMENT_EOA if last else 0
yield makeframe(FRAME_TYPE_COMMAND_ARGUMENT, flags, payload) yield makeframe(requestid, FRAME_TYPE_COMMAND_ARGUMENT, flags, payload)
if datafh: if datafh:
while True: while True:
data = datafh.read(DEFAULT_MAX_FRAME_SIZE) data = datafh.read(DEFAULT_MAX_FRAME_SIZE)
done = False done = False
if len(data) == DEFAULT_MAX_FRAME_SIZE: if len(data) == DEFAULT_MAX_FRAME_SIZE:
flags = FLAG_COMMAND_DATA_CONTINUATION flags = FLAG_COMMAND_DATA_CONTINUATION
else: else:
flags = FLAG_COMMAND_DATA_EOS flags = FLAG_COMMAND_DATA_EOS
assert datafh.read(1) == b'' assert datafh.read(1) == b''
done = True done = True
yield makeframe(FRAME_TYPE_COMMAND_DATA, flags, data) yield makeframe(requestid, FRAME_TYPE_COMMAND_DATA, flags, data)
if done: if done:
break break
def createbytesresponseframesfrombytes(data, def createbytesresponseframesfrombytes(requestid, data,
maxframesize=DEFAULT_MAX_FRAME_SIZE): maxframesize=DEFAULT_MAX_FRAME_SIZE):
"""Create a raw frame to send a bytes response from static bytes input. """Create a raw frame to send a bytes response from static bytes input.
Returns a generator of bytearrays. Returns a generator of bytearrays.
""" """
# Simple case of a single frame. # Simple case of a single frame.
if len(data) <= maxframesize: if len(data) <= maxframesize:
yield makeframe(FRAME_TYPE_BYTES_RESPONSE, yield makeframe(requestid, FRAME_TYPE_BYTES_RESPONSE,
FLAG_BYTES_RESPONSE_EOS, data) FLAG_BYTES_RESPONSE_EOS, data)
return return
offset = 0 offset = 0
while True: while True:
chunk = data[offset:offset + maxframesize] chunk = data[offset:offset + maxframesize]
offset += len(chunk) offset += len(chunk)
done = offset == len(data) done = offset == len(data)
if done: if done:
flags = FLAG_BYTES_RESPONSE_EOS flags = FLAG_BYTES_RESPONSE_EOS
else: else:
flags = FLAG_BYTES_RESPONSE_CONTINUATION flags = FLAG_BYTES_RESPONSE_CONTINUATION
yield makeframe(FRAME_TYPE_BYTES_RESPONSE, flags, chunk) yield makeframe(requestid, FRAME_TYPE_BYTES_RESPONSE, flags, chunk)
if done: if done:
break break
def createerrorframe(msg, protocol=False, application=False): def createerrorframe(requestid, msg, protocol=False, application=False):
# TODO properly handle frame size limits. # TODO properly handle frame size limits.
assert len(msg) <= DEFAULT_MAX_FRAME_SIZE assert len(msg) <= DEFAULT_MAX_FRAME_SIZE
flags = 0 flags = 0
if protocol: if protocol:
flags |= FLAG_ERROR_RESPONSE_PROTOCOL flags |= FLAG_ERROR_RESPONSE_PROTOCOL
if application: if application:
flags |= FLAG_ERROR_RESPONSE_APPLICATION flags |= FLAG_ERROR_RESPONSE_APPLICATION
yield makeframe(FRAME_TYPE_ERROR_RESPONSE, flags, msg) yield makeframe(requestid, FRAME_TYPE_ERROR_RESPONSE, flags, msg)
class serverreactor(object): class serverreactor(object):
"""Holds state of a server handling frame-based protocol requests. """Holds state of a server handling frame-based protocol requests.
This class is the "brain" of the unified frame-based protocol server This class is the "brain" of the unified frame-based protocol server
component. While the protocol is stateless from the perspective of component. While the protocol is stateless from the perspective of
requests/commands, something needs to track which frames have been requests/commands, something needs to track which frames have been
received, what frames to expect, etc. This class is that thing. received, what frames to expect, etc. This class is that thing.
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
self._activecommand = None self._activecommand = None
self._activeargs = None self._activeargs = None
self._activedata = None self._activedata = None
self._expectingargs = None self._expectingargs = None
self._expectingdata = None self._expectingdata = None
self._activeargname = None self._activeargname = None
self._activeargchunks = None self._activeargchunks = None
def onframerecv(self, 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-args': self._onframereceivingargs,
'command-receiving-data': self._onframereceivingdata, '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(frametype, frameflags, payload) return meth(requestid, frametype, frameflags, payload)
def onbytesresponseready(self, data): def onbytesresponseready(self, requestid, data):
"""Signal that a bytes response is ready to be sent to the client. """Signal that a bytes response is ready to be sent to the client.
The raw bytes response is passed as an argument. The raw bytes response is passed as an argument.
""" """
framegen = createbytesresponseframesfrombytes(data) framegen = createbytesresponseframesfrombytes(requestid, data)
if self._deferoutput: if self._deferoutput:
self._bufferedframegens.append(framegen) self._bufferedframegens.append(framegen)
return 'noop', {} return 'noop', {}
else: else:
return 'sendframes', { return 'sendframes', {
'framegen': framegen, 'framegen': framegen,
} }
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, msg): def onapplicationerror(self, requestid, msg):
return 'sendframes', { return 'sendframes', {
'framegen': createerrorframe(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):
return 'runcommand', { return 'runcommand', {
'requestid': self._activerequestid,
'command': self._activecommand, 'command': self._activecommand,
'args': self._activeargs, 'args': self._activeargs,
'data': self._activedata.getvalue() if self._activedata else None, 'data': self._activedata.getvalue() if self._activedata else None,
} }
def _makewantframeresult(self): def _makewantframeresult(self):
return 'wantframe', { return 'wantframe', {
'state': self._state, 'state': self._state,
} }
def _onframeidle(self, 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
self._activecommand = payload self._activecommand = payload
self._activeargs = {} self._activeargs = {}
self._activedata = None self._activedata = None
if frameflags & FLAG_COMMAND_NAME_EOS: if frameflags & FLAG_COMMAND_NAME_EOS:
return self._makeruncommandresult() return self._makeruncommandresult()
self._expectingargs = bool(frameflags & FLAG_COMMAND_NAME_HAVE_ARGS) self._expectingargs = bool(frameflags & FLAG_COMMAND_NAME_HAVE_ARGS)
self._expectingdata = bool(frameflags & FLAG_COMMAND_NAME_HAVE_DATA) self._expectingdata = bool(frameflags & FLAG_COMMAND_NAME_HAVE_DATA)
if self._expectingargs: if self._expectingargs:
self._state = 'command-receiving-args' self._state = 'command-receiving-args'
return self._makewantframeresult() return self._makewantframeresult()
elif self._expectingdata: elif self._expectingdata:
self._activedata = util.bytesio() self._activedata = util.bytesio()
self._state = 'command-receiving-data' 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, frametype, frameflags, payload): def _onframereceivingargs(self, requestid, frametype, frameflags, payload):
if frametype != FRAME_TYPE_COMMAND_ARGUMENT: if frametype != FRAME_TYPE_COMMAND_ARGUMENT:
self._state = 'errored' self._state = 'errored'
return self._makeerrorresult(_('expected command argument ' return self._makeerrorresult(_('expected command argument '
'frame; got %d') % frametype) 'frame; got %d') % frametype)
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
# buffer data frames. # buffer data frames.
return self._makewantframeresult() return self._makewantframeresult()
else: else:
self._state = 'waiting' self._state = 'waiting'
return self._makeruncommandresult() return self._makeruncommandresult()
else: else:
return self._makewantframeresult() return self._makewantframeresult()
def _onframereceivingdata(self, frametype, frameflags, payload): def _onframereceivingdata(self, requestid, frametype, frameflags, payload):
if frametype != FRAME_TYPE_COMMAND_DATA: if frametype != FRAME_TYPE_COMMAND_DATA:
self._state = 'errored' self._state = 'errored'
return self._makeerrorresult(_('expected command data frame; ' return self._makeerrorresult(_('expected command data frame; '
'got %d') % frametype) 'got %d') % frametype)
# TODO support streaming data instead of buffering it. # TODO support streaming data instead of buffering it.
self._activedata.write(payload) self._activedata.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) self._activedata.seek(0)
self._state = 'idle' self._state = 'idle'
return self._makeruncommandresult() 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, frametype, frameflags, payload): def _onframeerrored(self, requestid, frametype, frameflags, payload):
return self._makeerrorresult(_('server already errored')) return self._makeerrorresult(_('server already errored'))

mercurial/wireprotoserver.py

urlerr = util.urlerr urlerr = util.urlerr
urlreq = util.urlreq urlreq = util.urlreq
HTTP_OK = 200 HTTP_OK = 200
HGTYPE = 'application/mercurial-0.1' HGTYPE = 'application/mercurial-0.1'
HGTYPE2 = 'application/mercurial-0.2' HGTYPE2 = 'application/mercurial-0.2'
HGERRTYPE = 'application/hg-error' HGERRTYPE = 'application/hg-error'
FRAMINGTYPE = b'application/mercurial-exp-framing-0001' FRAMINGTYPE = b'application/mercurial-exp-framing-0002'
HTTPV2 = wireprototypes.HTTPV2 HTTPV2 = wireprototypes.HTTPV2
SSHV1 = wireprototypes.SSHV1 SSHV1 = wireprototypes.SSHV1
SSHV2 = wireprototypes.SSHV2 SSHV2 = wireprototypes.SSHV2
def decodevaluefromheaders(req, headerprefix): def decodevaluefromheaders(req, headerprefix):
"""Decode a long value from multiple HTTP request headers. """Decode a long value from multiple HTTP request headers.
while True: while True:
frame = wireprotoframing.readframe(req.bodyfh) frame = wireprotoframing.readframe(req.bodyfh)
if not frame: if not frame:
states.append(b'received: <no frame>') states.append(b'received: <no frame>')
break break
frametype, frameflags, payload = frame requestid, frametype, frameflags, payload = frame
states.append(b'received: %d %d %s' % (frametype, frameflags, payload)) states.append(b'received: %d %d %d %s' % (frametype, frameflags,
requestid, payload))
action, meta = reactor.onframerecv(frametype, frameflags, payload) action, meta = reactor.onframerecv(requestid, frametype, frameflags,
payload)
states.append(json.dumps((action, meta), sort_keys=True, states.append(json.dumps((action, meta), sort_keys=True,
separators=(', ', ': '))) separators=(', ', ': ')))
action, meta = reactor.oninputeof() action, meta = reactor.oninputeof()
meta['action'] = action meta['action'] = action
states.append(json.dumps(meta, sort_keys=True, separators=(', ',': '))) states.append(json.dumps(meta, sort_keys=True, separators=(', ',': ')))
res.status = b'200 OK' res.status = b'200 OK'
assert wirecommand.permission in ('push', 'pull') assert wirecommand.permission in ('push', 'pull')
rsp = wireproto.dispatch(repo, proto, command['command']) rsp = wireproto.dispatch(repo, proto, command['command'])
res.status = b'200 OK' res.status = b'200 OK'
res.headers[b'Content-Type'] = FRAMINGTYPE res.headers[b'Content-Type'] = FRAMINGTYPE
if isinstance(rsp, wireprototypes.bytesresponse): if isinstance(rsp, wireprototypes.bytesresponse):
action, meta = reactor.onbytesresponseready(rsp.data) action, meta = reactor.onbytesresponseready(command['requestid'],
rsp.data)
else: else:
action, meta = reactor.onapplicationerror( action, meta = reactor.onapplicationerror(
_('unhandled response type from wire proto command')) _('unhandled response type from wire proto command'))
if action == 'sendframes': if action == 'sendframes':
res.setbodygen(meta['framegen']) res.setbodygen(meta['framegen'])
elif action == 'noop': elif action == 'noop':
pass pass

tests/test-http-api-httpv2.t

$ HTTPV2=exp-http-v2-0001 $ HTTPV2=exp-http-v2-0001
$ MEDIATYPE=application/mercurial-exp-framing-0001 $ MEDIATYPE=application/mercurial-exp-framing-0002
$ send() { $ send() {
> hg --verbose debugwireproto --peer raw http://$LOCALIP:$HGPORT/ > hg --verbose debugwireproto --peer raw http://$LOCALIP:$HGPORT/
> } > }
$ cat > dummycommands.py << EOF $ cat > dummycommands.py << EOF
> from mercurial import wireprototypes, wireproto > from mercurial import wireprototypes, wireproto
> @wireproto.wireprotocommand('customreadonly', permission='pull') > @wireproto.wireprotocommand('customreadonly', permission='pull')
s> \r\n s> \r\n
s> makefile('rb', None) s> makefile('rb', None)
s> HTTP/1.1 406 Not Acceptable\r\n s> HTTP/1.1 406 Not Acceptable\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: 85\r\n s> Content-Length: 85\r\n
s> \r\n s> \r\n
s> client MUST specify Accept header with value: application/mercurial-exp-framing-0001\n s> client MUST specify Accept header with value: application/mercurial-exp-framing-0002\n
Bad Accept header results in 406 Bad Accept header results in 406
$ send << EOF $ send << EOF
> httprequest POST api/$HTTPV2/ro/customreadonly > httprequest POST api/$HTTPV2/ro/customreadonly
> accept: invalid > accept: invalid
> user-agent: test > user-agent: test
> EOF > EOF
using raw connection to peer using raw connection to peer
s> POST /api/exp-http-v2-0001/ro/customreadonly HTTP/1.1\r\n s> POST /api/exp-http-v2-0001/ro/customreadonly HTTP/1.1\r\n
s> Accept-Encoding: identity\r\n s> Accept-Encoding: identity\r\n
s> accept: invalid\r\n s> accept: invalid\r\n
s> user-agent: test\r\n s> user-agent: test\r\n
s> host: $LOCALIP:$HGPORT\r\n (glob) s> host: $LOCALIP:$HGPORT\r\n (glob)
s> \r\n s> \r\n
s> makefile('rb', None) s> makefile('rb', None)
s> HTTP/1.1 406 Not Acceptable\r\n s> HTTP/1.1 406 Not Acceptable\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: 85\r\n s> Content-Length: 85\r\n
s> \r\n s> \r\n
s> client MUST specify Accept header with value: application/mercurial-exp-framing-0001\n s> client MUST specify Accept header with value: application/mercurial-exp-framing-0002\n
Bad Content-Type header results in 415 Bad Content-Type header results in 415
$ send << EOF $ send << EOF
> httprequest POST api/$HTTPV2/ro/customreadonly > httprequest POST api/$HTTPV2/ro/customreadonly
> accept: $MEDIATYPE > accept: $MEDIATYPE
> user-agent: test > user-agent: test
> content-type: badmedia > content-type: badmedia
> EOF > EOF
using raw connection to peer using raw connection to peer
s> POST /api/exp-http-v2-0001/ro/customreadonly HTTP/1.1\r\n s> POST /api/exp-http-v2-0001/ro/customreadonly HTTP/1.1\r\n
s> Accept-Encoding: identity\r\n s> Accept-Encoding: identity\r\n
s> accept: application/mercurial-exp-framing-0001\r\n s> accept: application/mercurial-exp-framing-0002\r\n
s> content-type: badmedia\r\n s> content-type: badmedia\r\n
s> user-agent: test\r\n s> user-agent: test\r\n
s> host: $LOCALIP:$HGPORT\r\n (glob) s> host: $LOCALIP:$HGPORT\r\n (glob)
s> \r\n s> \r\n
s> makefile('rb', None) s> makefile('rb', None)
s> HTTP/1.1 415 Unsupported Media Type\r\n s> HTTP/1.1 415 Unsupported Media Type\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: 88\r\n s> Content-Length: 88\r\n
s> \r\n s> \r\n
s> client MUST send Content-Type header with value: application/mercurial-exp-framing-0001\n s> client MUST send Content-Type header with value: application/mercurial-exp-framing-0002\n
Request to read-only command works out of the box Request to read-only command works out of the box
$ send << EOF $ send << EOF
> httprequest POST api/$HTTPV2/ro/customreadonly > httprequest POST api/$HTTPV2/ro/customreadonly
> accept: $MEDIATYPE > accept: $MEDIATYPE
> content-type: $MEDIATYPE > content-type: $MEDIATYPE
> user-agent: test > user-agent: test
> frame command-name eos customreadonly > frame 1 command-name eos customreadonly
> EOF > EOF
using raw connection to peer using raw connection to peer
s> POST /api/exp-http-v2-0001/ro/customreadonly HTTP/1.1\r\n s> POST /api/exp-http-v2-0001/ro/customreadonly HTTP/1.1\r\n
s> Accept-Encoding: identity\r\n s> Accept-Encoding: identity\r\n
s> accept: application/mercurial-exp-framing-0001\r\n s> accept: application/mercurial-exp-framing-0002\r\n
s> content-type: application/mercurial-exp-framing-0001\r\n s> content-type: application/mercurial-exp-framing-0002\r\n
s> user-agent: test\r\n s> user-agent: test\r\n
s> content-length: 18\r\n s> *\r\n (glob)
s> host: $LOCALIP:$HGPORT\r\n (glob) s> host: $LOCALIP:$HGPORT\r\n (glob)
s> \r\n s> \r\n
s> \x0e\x00\x00\x11customreadonly s> \x0e\x00\x00\x01\x00\x11customreadonly
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: application/mercurial-exp-framing-0001\r\n s> Content-Type: application/mercurial-exp-framing-0002\r\n
s> Transfer-Encoding: chunked\r\n s> Transfer-Encoding: chunked\r\n
s> \r\n s> \r\n
s> 21\r\n s> 23\r\n
s> \x1d\x00\x00Bcustomreadonly bytes response s> \x1d\x00\x00\x01\x00Bcustomreadonly bytes response
s> \r\n s> \r\n
s> 0\r\n s> 0\r\n
s> \r\n s> \r\n
Request to read-write command fails because server is read-only by default Request to read-write command fails because server is read-only by default
GET to read-write request yields 405 GET to read-write request yields 405
Authorized request for valid read-write command works Authorized request for valid read-write command works
$ send << EOF $ send << EOF
> httprequest POST api/$HTTPV2/rw/customreadonly > httprequest POST api/$HTTPV2/rw/customreadonly
> user-agent: test > user-agent: test
> accept: $MEDIATYPE > accept: $MEDIATYPE
> content-type: $MEDIATYPE > content-type: $MEDIATYPE
> frame command-name eos customreadonly > frame 1 command-name eos customreadonly
> EOF > EOF
using raw connection to peer using raw connection to peer
s> POST /api/exp-http-v2-0001/rw/customreadonly HTTP/1.1\r\n s> POST /api/exp-http-v2-0001/rw/customreadonly HTTP/1.1\r\n
s> Accept-Encoding: identity\r\n s> Accept-Encoding: identity\r\n
s> accept: application/mercurial-exp-framing-0001\r\n s> accept: application/mercurial-exp-framing-0002\r\n
s> content-type: application/mercurial-exp-framing-0001\r\n s> content-type: application/mercurial-exp-framing-0002\r\n
s> user-agent: test\r\n s> user-agent: test\r\n
s> content-length: 18\r\n s> content-length: 20\r\n
s> host: $LOCALIP:$HGPORT\r\n (glob) s> host: $LOCALIP:$HGPORT\r\n (glob)
s> \r\n s> \r\n
s> \x0e\x00\x00\x11customreadonly s> \x0e\x00\x00\x01\x00\x11customreadonly
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: application/mercurial-exp-framing-0001\r\n s> Content-Type: application/mercurial-exp-framing-0002\r\n
s> Transfer-Encoding: chunked\r\n s> Transfer-Encoding: chunked\r\n
s> \r\n s> \r\n
s> 21\r\n s> 23\r\n
s> \x1d\x00\x00Bcustomreadonly bytes response s> \x1d\x00\x00\x01\x00Bcustomreadonly bytes response
s> \r\n s> \r\n
s> 0\r\n s> 0\r\n
s> \r\n s> \r\n
Authorized request for unknown command is rejected Authorized request for unknown command is rejected
$ send << EOF $ send << EOF
> httprequest POST api/$HTTPV2/rw/badcommand > httprequest POST api/$HTTPV2/rw/badcommand
> user-agent: test > user-agent: test
> accept: $MEDIATYPE > accept: $MEDIATYPE
> EOF > EOF
using raw connection to peer using raw connection to peer
s> POST /api/exp-http-v2-0001/rw/badcommand HTTP/1.1\r\n s> POST /api/exp-http-v2-0001/rw/badcommand HTTP/1.1\r\n
s> Accept-Encoding: identity\r\n s> Accept-Encoding: identity\r\n
s> accept: application/mercurial-exp-framing-0001\r\n s> accept: application/mercurial-exp-framing-0002\r\n
s> user-agent: test\r\n s> user-agent: test\r\n
s> host: $LOCALIP:$HGPORT\r\n (glob) s> host: $LOCALIP:$HGPORT\r\n (glob)
s> \r\n s> \r\n
s> makefile('rb', None) s> makefile('rb', None)
s> HTTP/1.1 404 Not Found\r\n s> HTTP/1.1 404 Not Found\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
Command frames can be reflected via debugreflect Command frames can be reflected via debugreflect
$ send << EOF $ send << EOF
> httprequest POST api/$HTTPV2/ro/debugreflect > httprequest POST api/$HTTPV2/ro/debugreflect
> accept: $MEDIATYPE > accept: $MEDIATYPE
> content-type: $MEDIATYPE > content-type: $MEDIATYPE
> user-agent: test > user-agent: test
> frame command-name have-args command1 > frame 1 command-name have-args command1
> frame command-argument 0 \x03\x00\x04\x00fooval1 > frame 1 command-argument 0 \x03\x00\x04\x00fooval1
> frame command-argument eoa \x04\x00\x03\x00bar1val > frame 1 command-argument eoa \x04\x00\x03\x00bar1val
> EOF > EOF
using raw connection to peer using raw connection to peer
s> POST /api/exp-http-v2-0001/ro/debugreflect HTTP/1.1\r\n s> POST /api/exp-http-v2-0001/ro/debugreflect HTTP/1.1\r\n
s> Accept-Encoding: identity\r\n s> Accept-Encoding: identity\r\n
s> accept: application/mercurial-exp-framing-0001\r\n s> accept: application/mercurial-exp-framing-0002\r\n
s> content-type: application/mercurial-exp-framing-0001\r\n s> content-type: application/mercurial-exp-framing-0002\r\n
s> user-agent: test\r\n s> user-agent: test\r\n
s> content-length: 42\r\n s> content-length: 48\r\n
s> host: $LOCALIP:$HGPORT\r\n (glob) s> host: $LOCALIP:$HGPORT\r\n (glob)
s> \r\n s> \r\n
s> \x08\x00\x00\x12command1\x0b\x00\x00 \x03\x00\x04\x00fooval1\x0b\x00\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: 310\r\n s> Content-Length: 332\r\n
s> \r\n s> \r\n
s> received: 1 2 command1\n s> received: 1 2 1 command1\n
s> ["wantframe", {"state": "command-receiving-args"}]\n s> ["wantframe", {"state": "command-receiving-args"}]\n
s> received: 2 0 \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-args"}]\n
s> received: 2 2 \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}]\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

return framing.serverreactor(deferoutput=deferoutput) return framing.serverreactor(deferoutput=deferoutput)
def sendframes(reactor, gen): def sendframes(reactor, gen):
"""Send a generator of frame bytearray to a reactor. """Send a generator of frame bytearray to a reactor.
Emits a generator of results from ``onframerecv()`` calls. Emits a generator of results from ``onframerecv()`` calls.
""" """
for frame in gen: for frame in gen:
frametype, frameflags, framelength = framing.parseheader(frame) rid, frametype, frameflags, framelength = framing.parseheader(frame)
payload = frame[framing.FRAME_HEADER_SIZE:] payload = frame[framing.FRAME_HEADER_SIZE:]
assert len(payload) == framelength assert len(payload) == framelength
yield reactor.onframerecv(frametype, frameflags, payload) yield reactor.onframerecv(rid, frametype, frameflags, payload)
def sendcommandframes(reactor, cmd, args, datafh=None): def sendcommandframes(reactor, rid, cmd, args, datafh=None):
"""Generate frames to run a command and send them to a reactor.""" """Generate frames to run a command and send them to a reactor."""
return sendframes(reactor, framing.createcommandframes(cmd, args, datafh)) return sendframes(reactor,
framing.createcommandframes(rid, cmd, args, datafh))
class FrameTests(unittest.TestCase): class FrameTests(unittest.TestCase):
def testdataexactframesize(self): def testdataexactframesize(self):
data = util.bytesio(b'x' * framing.DEFAULT_MAX_FRAME_SIZE) data = util.bytesio(b'x' * framing.DEFAULT_MAX_FRAME_SIZE)
frames = list(framing.createcommandframes(b'command', {}, data)) frames = list(framing.createcommandframes(1, b'command', {}, data))
self.assertEqual(frames, [ self.assertEqual(frames, [
ffs(b'command-name have-data command'), ffs(b'1 command-name have-data command'),
ffs(b'command-data continuation %s' % data.getvalue()), ffs(b'1 command-data continuation %s' % data.getvalue()),
ffs(b'command-data eos ') ffs(b'1 command-data eos ')
]) ])
def testdatamultipleframes(self): def testdatamultipleframes(self):
data = util.bytesio(b'x' * (framing.DEFAULT_MAX_FRAME_SIZE + 1)) data = util.bytesio(b'x' * (framing.DEFAULT_MAX_FRAME_SIZE + 1))
frames = list(framing.createcommandframes(b'command', {}, data)) frames = list(framing.createcommandframes(1, b'command', {}, data))
self.assertEqual(frames, [ self.assertEqual(frames, [
ffs(b'command-name have-data command'), ffs(b'1 command-name have-data command'),
ffs(b'command-data continuation %s' % ( ffs(b'1 command-data continuation %s' % (
b'x' * framing.DEFAULT_MAX_FRAME_SIZE)), b'x' * framing.DEFAULT_MAX_FRAME_SIZE)),
ffs(b'command-data eos x'), ffs(b'1 command-data eos x'),
]) ])
def testargsanddata(self): def testargsanddata(self):
data = util.bytesio(b'x' * 100) data = util.bytesio(b'x' * 100)
frames = list(framing.createcommandframes(b'command', { frames = list(framing.createcommandframes(1, b'command', {
b'key1': b'key1value', b'key1': b'key1value',
b'key2': b'key2value', b'key2': b'key2value',
b'key3': b'key3value', b'key3': b'key3value',
}, data)) }, data))
self.assertEqual(frames, [ self.assertEqual(frames, [
ffs(b'command-name have-args|have-data command'), ffs(b'1 command-name have-args|have-data command'),
ffs(br'command-argument 0 \x04\x00\x09\x00key1key1value'), ffs(br'1 command-argument 0 \x04\x00\x09\x00key1key1value'),
ffs(br'command-argument 0 \x04\x00\x09\x00key2key2value'), ffs(br'1 command-argument 0 \x04\x00\x09\x00key2key2value'),
ffs(br'command-argument eoa \x04\x00\x09\x00key3key3value'), ffs(br'1 command-argument eoa \x04\x00\x09\x00key3key3value'),
ffs(b'command-data eos %s' % data.getvalue()), ffs(b'1 command-data eos %s' % data.getvalue()),
]) ])
class ServerReactorTests(unittest.TestCase): class ServerReactorTests(unittest.TestCase):
def _sendsingleframe(self, reactor, s): def _sendsingleframe(self, reactor, s):
results = list(sendframes(reactor, [ffs(s)])) results = list(sendframes(reactor, [ffs(s)]))
self.assertEqual(len(results), 1) self.assertEqual(len(results), 1)
return results[0] return results[0]
def assertaction(self, res, expected): def assertaction(self, res, expected):
self.assertIsInstance(res, tuple) self.assertIsInstance(res, tuple)
self.assertEqual(len(res), 2) self.assertEqual(len(res), 2)
self.assertIsInstance(res[1], dict) self.assertIsInstance(res[1], dict)
self.assertEqual(res[0], expected) self.assertEqual(res[0], expected)
def assertframesequal(self, frames, framestrings): def assertframesequal(self, frames, framestrings):
expected = [ffs(s) for s in framestrings] expected = [ffs(s) for s in framestrings]
self.assertEqual(list(frames), expected) self.assertEqual(list(frames), expected)
def test1framecommand(self): def test1framecommand(self):
"""Receiving a command in a single frame yields request to run it.""" """Receiving a command in a single frame yields request to run it."""
reactor = makereactor() reactor = makereactor()
results = list(sendcommandframes(reactor, b'mycommand', {})) results = list(sendcommandframes(reactor, 1, b'mycommand', {}))
self.assertEqual(len(results), 1) self.assertEqual(len(results), 1)
self.assertaction(results[0], 'runcommand') self.assertaction(results[0], 'runcommand')
self.assertEqual(results[0][1], { self.assertEqual(results[0][1], {
'requestid': 1,
'command': b'mycommand', 'command': b'mycommand',
'args': {}, 'args': {},
'data': None, 'data': None,
}) })
result = reactor.oninputeof() result = reactor.oninputeof()
self.assertaction(result, 'noop') self.assertaction(result, 'noop')
def test1argument(self): def test1argument(self):
reactor = makereactor() reactor = makereactor()
results = list(sendcommandframes(reactor, b'mycommand', results = list(sendcommandframes(reactor, 41, b'mycommand',
{b'foo': b'bar'})) {b'foo': b'bar'}))
self.assertEqual(len(results), 2) self.assertEqual(len(results), 2)
self.assertaction(results[0], 'wantframe') self.assertaction(results[0], 'wantframe')
self.assertaction(results[1], 'runcommand') self.assertaction(results[1], 'runcommand')
self.assertEqual(results[1][1], { self.assertEqual(results[1][1], {
'requestid': 41,
'command': b'mycommand', 'command': b'mycommand',
'args': {b'foo': b'bar'}, 'args': {b'foo': b'bar'},
'data': None, 'data': None,
}) })
def testmultiarguments(self): def testmultiarguments(self):
reactor = makereactor() reactor = makereactor()
results = list(sendcommandframes(reactor, b'mycommand', results = list(sendcommandframes(reactor, 1, b'mycommand',
{b'foo': b'bar', b'biz': b'baz'})) {b'foo': b'bar', b'biz': b'baz'}))
self.assertEqual(len(results), 3) self.assertEqual(len(results), 3)
self.assertaction(results[0], 'wantframe') self.assertaction(results[0], 'wantframe')
self.assertaction(results[1], 'wantframe') self.assertaction(results[1], 'wantframe')
self.assertaction(results[2], 'runcommand') self.assertaction(results[2], 'runcommand')
self.assertEqual(results[2][1], { self.assertEqual(results[2][1], {
'requestid': 1,
'command': b'mycommand', 'command': b'mycommand',
'args': {b'foo': b'bar', b'biz': b'baz'}, 'args': {b'foo': b'bar', b'biz': b'baz'},
'data': None, 'data': None,
}) })
def testsimplecommanddata(self): def testsimplecommanddata(self):
reactor = makereactor() reactor = makereactor()
results = list(sendcommandframes(reactor, b'mycommand', {}, results = list(sendcommandframes(reactor, 1, b'mycommand', {},
util.bytesio(b'data!'))) util.bytesio(b'data!')))
self.assertEqual(len(results), 2) self.assertEqual(len(results), 2)
self.assertaction(results[0], 'wantframe') self.assertaction(results[0], 'wantframe')
self.assertaction(results[1], 'runcommand') self.assertaction(results[1], 'runcommand')
self.assertEqual(results[1][1], { self.assertEqual(results[1][1], {
'requestid': 1,
'command': b'mycommand', 'command': b'mycommand',
'args': {}, 'args': {},
'data': b'data!', 'data': b'data!',
}) })
def testmultipledataframes(self): def testmultipledataframes(self):
frames = [ frames = [
ffs(b'command-name have-data mycommand'), ffs(b'1 command-name have-data mycommand'),
ffs(b'command-data continuation data1'), ffs(b'1 command-data continuation data1'),
ffs(b'command-data continuation data2'), ffs(b'1 command-data continuation data2'),
ffs(b'command-data eos data3'), ffs(b'1 command-data eos data3'),
] ]
reactor = makereactor() reactor = makereactor()
results = list(sendframes(reactor, frames)) results = list(sendframes(reactor, frames))
self.assertEqual(len(results), 4) self.assertEqual(len(results), 4)
for i in range(3): for i in range(3):
self.assertaction(results[i], 'wantframe') self.assertaction(results[i], 'wantframe')
self.assertaction(results[3], 'runcommand') self.assertaction(results[3], 'runcommand')
self.assertEqual(results[3][1], { self.assertEqual(results[3][1], {
'requestid': 1,
'command': b'mycommand', 'command': b'mycommand',
'args': {}, 'args': {},
'data': b'data1data2data3', 'data': b'data1data2data3',
}) })
def testargumentanddata(self): def testargumentanddata(self):
frames = [ frames = [
ffs(b'command-name have-args|have-data command'), ffs(b'1 command-name have-args|have-data command'),
ffs(br'command-argument 0 \x03\x00\x03\x00keyval'), ffs(br'1 command-argument 0 \x03\x00\x03\x00keyval'),
ffs(br'command-argument eoa \x03\x00\x03\x00foobar'), ffs(br'1 command-argument eoa \x03\x00\x03\x00foobar'),
ffs(b'command-data continuation value1'), ffs(b'1 command-data continuation value1'),
ffs(b'command-data eos value2'), ffs(b'1 command-data eos value2'),
] ]
reactor = makereactor() reactor = makereactor()
results = list(sendframes(reactor, frames)) results = list(sendframes(reactor, frames))
self.assertaction(results[-1], 'runcommand') self.assertaction(results[-1], 'runcommand')
self.assertEqual(results[-1][1], { self.assertEqual(results[-1][1], {
'requestid': 1,
'command': b'command', 'command': b'command',
'args': { 'args': {
b'key': b'val', b'key': b'val',
b'foo': b'bar', b'foo': b'bar',
}, },
'data': b'value1value2', 'data': b'value1value2',
}) })
def testunexpectedcommandargument(self): def testunexpectedcommandargument(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'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 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'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 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'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 testmissingargumentframe(self): def testmissingargumentframe(self):
frames = [ frames = [
ffs(b'command-name have-args command'), ffs(b'1 command-name have-args command'),
ffs(b'command-name 0 ignored'), ffs(b'1 command-name 0 ignored'),
] ]
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'expected command argument frame; got 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'command-name have-args command1'), ffs(b'1 command-name have-args command1'),
ffs(br'command-argument eoa \x04\x00\xde\xadfoo'), ffs(br'1 command-argument eoa \x04\x00\xde\xadfoo'),
] ]
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'malformed argument frame: partial argument name', 'message': b'malformed argument frame: partial argument name',
}) })
def testincompleteargumentvalue(self): def testincompleteargumentvalue(self):
"""Argument frame with incomplete value.""" """Argument frame with incomplete value."""
frames = [ frames = [
ffs(b'command-name have-args command'), ffs(b'1 command-name have-args command'),
ffs(br'command-argument eoa \x03\x00\xaa\xaafoopartialvalue'), ffs(br'1 command-argument eoa \x03\x00\xaa\xaafoopartialvalue'),
] ]
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'malformed argument frame: partial argument value', 'message': b'malformed argument frame: partial argument value',
}) })
def testmissingcommanddataframe(self): def testmissingcommanddataframe(self):
frames = [ frames = [
ffs(b'command-name have-data command1'), ffs(b'1 command-name have-data command1'),
ffs(b'command-name eos command2'), ffs(b'1 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], 'error')
self.assertEqual(results[1][1], { self.assertEqual(results[1][1], {
'message': b'expected command data frame; got 1', 'message': b'expected command data frame; got 1',
}) })
def testmissingcommanddataframeflags(self): def testmissingcommanddataframeflags(self):
frames = [ frames = [
ffs(b'command-name have-data command1'), ffs(b'1 command-name have-data command1'),
ffs(b'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 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, b'mycommand', {})) list(sendcommandframes(reactor, 1, b'mycommand', {}))
result = reactor.onbytesresponseready(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'], [
b'bytes-response eos response', b'1 bytes-response eos response',
]) ])
def testmultiframeresponse(self): def testmultiframeresponse(self):
"""Bytes response spanning multiple frames is handled.""" """Bytes response spanning multiple frames is handled."""
first = b'x' * framing.DEFAULT_MAX_FRAME_SIZE first = b'x' * framing.DEFAULT_MAX_FRAME_SIZE
second = b'y' * 100 second = b'y' * 100
reactor = makereactor() reactor = makereactor()
list(sendcommandframes(reactor, b'mycommand', {})) list(sendcommandframes(reactor, 1, b'mycommand', {}))
result = reactor.onbytesresponseready(first + second) result = reactor.onbytesresponseready(1, first + second)
self.assertaction(result, 'sendframes') self.assertaction(result, 'sendframes')
self.assertframesequal(result[1]['framegen'], [ self.assertframesequal(result[1]['framegen'], [
b'bytes-response continuation %s' % first, b'1 bytes-response continuation %s' % first,
b'bytes-response eos %s' % second, b'1 bytes-response eos %s' % second,
]) ])
def testapplicationerror(self): def testapplicationerror(self):
reactor = makereactor() reactor = makereactor()
list(sendcommandframes(reactor, b'mycommand', {})) list(sendcommandframes(reactor, 1, b'mycommand', {}))
result = reactor.onapplicationerror(b'some message') result = reactor.onapplicationerror(1, b'some message')
self.assertaction(result, 'sendframes') self.assertaction(result, 'sendframes')
self.assertframesequal(result[1]['framegen'], [ self.assertframesequal(result[1]['framegen'], [
b'error-response application some message', b'1 error-response application some message',
]) ])
def test1commanddeferresponse(self): def test1commanddeferresponse(self):
"""Responses when in deferred output mode are delayed until EOF.""" """Responses when in deferred output mode are delayed until EOF."""
reactor = makereactor(deferoutput=True) reactor = makereactor(deferoutput=True)
results = list(sendcommandframes(reactor, b'mycommand', {})) results = list(sendcommandframes(reactor, 1, b'mycommand', {}))
self.assertEqual(len(results), 1) self.assertEqual(len(results), 1)
self.assertaction(results[0], 'runcommand') self.assertaction(results[0], 'runcommand')
result = reactor.onbytesresponseready(b'response') result = reactor.onbytesresponseready(1, b'response')
self.assertaction(result, 'noop') self.assertaction(result, 'noop')
result = reactor.oninputeof() result = reactor.oninputeof()
self.assertaction(result, 'sendframes') self.assertaction(result, 'sendframes')
self.assertframesequal(result[1]['framegen'], [ self.assertframesequal(result[1]['framegen'], [
b'bytes-response eos response', b'1 bytes-response eos response',
]) ])
def testmultiplecommanddeferresponse(self): def testmultiplecommanddeferresponse(self):
reactor = makereactor(deferoutput=True) reactor = makereactor(deferoutput=True)
list(sendcommandframes(reactor, b'command1', {})) list(sendcommandframes(reactor, 1, b'command1', {}))
list(sendcommandframes(reactor, b'command2', {})) list(sendcommandframes(reactor, 3, b'command2', {}))
result = reactor.onbytesresponseready(b'response1') result = reactor.onbytesresponseready(1, b'response1')
self.assertaction(result, 'noop') self.assertaction(result, 'noop')
result = reactor.onbytesresponseready(b'response2') result = reactor.onbytesresponseready(3, b'response2')
self.assertaction(result, 'noop') self.assertaction(result, 'noop')
result = reactor.oninputeof() result = reactor.oninputeof()
self.assertaction(result, 'sendframes') self.assertaction(result, 'sendframes')
self.assertframesequal(result[1]['framegen'], [ self.assertframesequal(result[1]['framegen'], [
b'bytes-response eos response1', b'1 bytes-response eos response1',
b'bytes-response eos response2' b'3 bytes-response eos response2'
])
def testrequestidtracking(self):
reactor = makereactor(deferoutput=True)
list(sendcommandframes(reactor, 1, b'command1', {}))
list(sendcommandframes(reactor, 3, b'command2', {}))
list(sendcommandframes(reactor, 5, b'command3', {}))
# Register results for commands out of order.
reactor.onbytesresponseready(3, b'response3')
reactor.onbytesresponseready(1, b'response1')
reactor.onbytesresponseready(5, b'response5')
result = reactor.oninputeof()
self.assertaction(result, 'sendframes')
self.assertframesequal(result[1]['framegen'], [
b'3 bytes-response eos response3',
b'1 bytes-response eos response1',
b'5 bytes-response eos response5',
]) ])
if __name__ == '__main__': if __name__ == '__main__':
import silenttestrunner import silenttestrunner
silenttestrunner.main(__name__) silenttestrunner.main(__name__)