This is an archive of the discontinued Mercurial Phabricator instance.

Differential D3243

httppeer: support protocol upgrade
ClosedPublic

Authored by indygreg on Apr 10 2018, 9:28 PM.

Details

Reviewers
durin42
Group Reviewers
hg-reviewers
Commits
rHG8a73132214a3: httppeer: support protocol upgrade
Summary

With the new handshake defined and in place on the server, we can
now implement it on the client.

The HTTP handshake mechanism has been taught to add headers advertising
its support for the new capabilities response. Response handling
has been adjusted to allow CBOR responses through. And makepeer()
has been taught to instantiate a mutually supported peer.

The HTTPv2 peer class doesn't implement the full peer interface. So
HTTPv2 is not yet usable as a peer.

Like the server side, we support registering handlers for
different API services. This allows extensions to easily implement
API services and peers. A practical use case for this is to
provide a previous implementation of the experimental version 2
wire protocol to a future version of Mercurial. We know there will
be BC breaks after 4.6 ships. But someone could take the peer and
server code from 4.6, drop it in an extension, and allow its use
indefinitely.

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.Apr 10 2018, 9:28 PM
durin42 accepted this revision.Apr 11 2018, 12:54 PM
This revision is now accepted and ready to land.Apr 11 2018, 12:54 PM
Closed by commit rHG8a73132214a3: httppeer: support protocol upgrade (authored by indygreg). · Explain WhyApr 11 2018, 1:01 PM
This revision was automatically updated to reflect the committed changes.

Revision Contents
Changeset List

Diff 8004

mercurial/configitems.py

default=dynamicdefault, default=dynamicdefault,
) )
coreconfigitem('experimental', 'graphstyle.grandparent', coreconfigitem('experimental', 'graphstyle.grandparent',
default=dynamicdefault, default=dynamicdefault,
) )
coreconfigitem('experimental', 'hook-track-tags', coreconfigitem('experimental', 'hook-track-tags',
default=False, default=False,
) )
coreconfigitem('experimental', 'httppeer.advertise-v2',
default=False,
)
coreconfigitem('experimental', 'httppostargs', coreconfigitem('experimental', 'httppostargs',
default=False, default=False,
) )
coreconfigitem('experimental', 'mergedriver', coreconfigitem('experimental', 'mergedriver',
default=None, default=None,
) )
coreconfigitem('experimental', 'obsmarkers-exchange-debug', coreconfigitem('experimental', 'obsmarkers-exchange-debug',
default=False, default=False,

mercurial/debugcommands.py

templater, templater,
treediscovery, treediscovery,
upgrade, upgrade,
url as urlmod, url as urlmod,
util, util,
vfs as vfsmod, vfs as vfsmod,
wireprotoframing, wireprotoframing,
wireprotoserver, wireprotoserver,
wireprototypes,
) )
from .utils import ( from .utils import (
dateutil, dateutil,
procutil, procutil,
stringutil, stringutil,
) )
release = lockmod.release release = lockmod.release
# have near complete control over what's sent on the wire. # have near complete control over what's sent on the wire.
if opts['peer'] == 'raw': if opts['peer'] == 'raw':
openerargs[r'sendaccept'] = False openerargs[r'sendaccept'] = False
opener = urlmod.opener(ui, authinfo, **openerargs) opener = urlmod.opener(ui, authinfo, **openerargs)
if opts['peer'] == 'http2': if opts['peer'] == 'http2':
ui.write(_('creating http peer for wire protocol version 2\n')) ui.write(_('creating http peer for wire protocol version 2\n'))
peer = httppeer.httpv2peer(ui, path, opener) peer = httppeer.httpv2peer(
ui, path, 'api/%s' % wireprototypes.HTTPV2,
opener, httppeer.urlreq.request, {})
elif opts['peer'] == 'raw': elif opts['peer'] == 'raw':
ui.write(_('using raw connection to peer\n')) ui.write(_('using raw connection to peer\n'))
peer = None peer = None
elif opts['peer']: elif opts['peer']:
raise error.Abort(_('--peer %s not supported with HTTP peers') % raise error.Abort(_('--peer %s not supported with HTTP peers') %
opts['peer']) opts['peer'])
else: else:
peer = httppeer.makepeer(ui, path, opener=opener) peer = httppeer.makepeer(ui, path, opener=opener)

mercurial/httppeer.py

error, error,
httpconnection, httpconnection,
pycompat, pycompat,
statichttprepo, statichttprepo,
url as urlmod, url as urlmod,
util, util,
wireproto, wireproto,
wireprotoframing, wireprotoframing,
wireprototypes,
wireprotov2server, wireprotov2server,
) )
httplib = util.httplib httplib = util.httplib
urlerr = util.urlerr urlerr = util.urlerr
urlreq = util.urlreq urlreq = util.urlreq
def encodevalueinheaders(value, header, limit): def encodevalueinheaders(value, header, limit):
dbg(line % ' finished in %.4f seconds (%s)' dbg(line % ' finished in %.4f seconds (%s)'
% (util.timer() - start, res.code)) % (util.timer() - start, res.code))
# Insert error handlers for common I/O failures. # Insert error handlers for common I/O failures.
_wraphttpresponse(res) _wraphttpresponse(res)
return res return res
def parsev1commandresponse(ui, baseurl, requrl, qs, resp, compressible): def parsev1commandresponse(ui, baseurl, requrl, qs, resp, compressible,
allowcbor=False):
# record the url we got redirected to # record the url we got redirected to
respurl = pycompat.bytesurl(resp.geturl()) respurl = pycompat.bytesurl(resp.geturl())
if respurl.endswith(qs): if respurl.endswith(qs):
respurl = respurl[:-len(qs)] respurl = respurl[:-len(qs)]
if baseurl.rstrip('/') != respurl.rstrip('/'): if baseurl.rstrip('/') != respurl.rstrip('/'):
if not ui.quiet: if not ui.quiet:
ui.warn(_('real URL is %s\n') % respurl) ui.warn(_('real URL is %s\n') % respurl)
if not proto.startswith('application/mercurial-'): if not proto.startswith('application/mercurial-'):
ui.debug("requested URL: '%s'\n" % util.hidepassword(requrl)) ui.debug("requested URL: '%s'\n" % util.hidepassword(requrl))
raise error.RepoError( raise error.RepoError(
_("'%s' does not appear to be an hg repository:\n" _("'%s' does not appear to be an hg repository:\n"
"---%%<--- (%s)\n%s\n---%%<---\n") "---%%<--- (%s)\n%s\n---%%<---\n")
% (safeurl, proto or 'no content-type', resp.read(1024))) % (safeurl, proto or 'no content-type', resp.read(1024)))
try: try:
version = proto.split('-', 1)[1] subtype = proto.split('-', 1)[1]
version_info = tuple([int(n) for n in version.split('.')])
# Unless we end up supporting CBOR in the legacy wire protocol,
# this should ONLY be encountered for the initial capabilities
# request during handshake.
if subtype == 'cbor':
if allowcbor:
return respurl, proto, resp
else:
raise error.RepoError(_('unexpected CBOR response from '
'server'))
version_info = tuple([int(n) for n in subtype.split('.')])
except ValueError: except ValueError:
raise error.RepoError(_("'%s' sent a broken Content-Type " raise error.RepoError(_("'%s' sent a broken Content-Type "
"header (%s)") % (safeurl, proto)) "header (%s)") % (safeurl, proto))
# TODO consider switching to a decompression reader that uses # TODO consider switching to a decompression reader that uses
# generators. # generators.
if version_info == (0, 1): if version_info == (0, 1):
if compressible: if compressible:
resp = util.compengines['zlib'].decompressorreader(resp) resp = util.compengines['zlib'].decompressorreader(resp)
elif version_info == (0, 2): elif version_info == (0, 2):
# application/mercurial-0.2 always identifies the compression # application/mercurial-0.2 always identifies the compression
# engine in the payload header. # engine in the payload header.
elen = struct.unpack('B', resp.read(1))[0] elen = struct.unpack('B', resp.read(1))[0]
ename = resp.read(elen) ename = resp.read(elen)
engine = util.compengines.forwiretype(ename) engine = util.compengines.forwiretype(ename)
resp = engine.decompressorreader(resp) resp = engine.decompressorreader(resp)
else: else:
raise error.RepoError(_("'%s' uses newer protocol %s") % raise error.RepoError(_("'%s' uses newer protocol %s") %
(safeurl, version)) (safeurl, subtype))
return respurl, resp return respurl, proto, resp
class httppeer(wireproto.wirepeer): class httppeer(wireproto.wirepeer):
def __init__(self, ui, path, url, opener, requestbuilder, caps): def __init__(self, ui, path, url, opener, requestbuilder, caps):
self.ui = ui self.ui = ui
self._path = path self._path = path
self._url = url self._url = url
self._caps = caps self._caps = caps
self._urlopener = opener self._urlopener = opener
args = pycompat.byteskwargs(args) args = pycompat.byteskwargs(args)
req, cu, qs = makev1commandrequest(self.ui, self._requestbuilder, req, cu, qs = makev1commandrequest(self.ui, self._requestbuilder,
self._caps, self.capable, self._caps, self.capable,
self._url, cmd, args) self._url, cmd, args)
resp = sendrequest(self.ui, self._urlopener, req) resp = sendrequest(self.ui, self._urlopener, req)
self._url, resp = parsev1commandresponse(self.ui, self._url, cu, qs, self._url, ct, resp = parsev1commandresponse(self.ui, self._url, cu, qs,
resp, _compressible) resp, _compressible)
return resp return resp
def _call(self, cmd, **args): def _call(self, cmd, **args):
fp = self._callstream(cmd, **args) fp = self._callstream(cmd, **args)
try: try:
return fp.read() return fp.read()
finally: finally:
def _callcompressable(self, cmd, **args): def _callcompressable(self, cmd, **args):
return self._callstream(cmd, _compressible=True, **args) return self._callstream(cmd, _compressible=True, **args)
def _abort(self, exception): def _abort(self, exception):
raise exception raise exception
# TODO implement interface for version 2 peers # TODO implement interface for version 2 peers
class httpv2peer(object): class httpv2peer(object):
def __init__(self, ui, repourl, opener): def __init__(self, ui, repourl, apipath, opener, requestbuilder,
apidescriptor):
self.ui = ui self.ui = ui
if repourl.endswith('/'): if repourl.endswith('/'):
repourl = repourl[:-1] repourl = repourl[:-1]
self.url = repourl self.url = repourl
self._apipath = apipath
self._opener = opener self._opener = opener
# This is an its own attribute to facilitate extensions overriding self._requestbuilder = requestbuilder
# the default type. self._descriptor = apidescriptor
self._requestbuilder = urlreq.request
def close(self): def close(self):
pass pass
# TODO require to be part of a batched primitive, use futures. # TODO require to be part of a batched primitive, use futures.
def _call(self, name, **args): def _call(self, name, **args):
"""Call a wire protocol command with arguments.""" """Call a wire protocol command with arguments."""
raise error.ProgrammingError('unknown permission type: %s' % raise error.ProgrammingError('unknown permission type: %s' %
permission) permission)
permission = { permission = {
'push': 'rw', 'push': 'rw',
'pull': 'ro', 'pull': 'ro',
}[permission] }[permission]
url = '%s/api/%s/%s/%s' % (self.url, wireprotov2server.HTTPV2, url = '%s/%s/%s/%s' % (self.url, self._apipath, permission, name)
permission, name)
# TODO this should be part of a generic peer for the frame-based # TODO this should be part of a generic peer for the frame-based
# protocol. # protocol.
reactor = wireprotoframing.clientreactor(hasmultiplesend=False, reactor = wireprotoframing.clientreactor(hasmultiplesend=False,
buffersends=True) buffersends=True)
request, action, meta = reactor.callcommand(name, args) request, action, meta = reactor.callcommand(name, args)
assert action == 'noop' assert action == 'noop'
results.append(decoder.decode()) results.append(decoder.decode())
else: else:
results.append(meta['data']) results.append(meta['data'])
else: else:
error.ProgrammingError('unhandled action: %s' % action) error.ProgrammingError('unhandled action: %s' % action)
return results return results
# Registry of API service names to metadata about peers that handle it.
#
# The following keys are meaningful:
#
# init
# Callable receiving (ui, repourl, servicepath, opener, requestbuilder,
# apidescriptor) to create a peer.
#
# priority
# Integer priority for the service. If we could choose from multiple
# services, we choose the one with the highest priority.
API_PEERS = {
wireprototypes.HTTPV2: {
'init': httpv2peer,
'priority': 50,
},
}
def performhandshake(ui, url, opener, requestbuilder): def performhandshake(ui, url, opener, requestbuilder):
# The handshake is a request to the capabilities command. # The handshake is a request to the capabilities command.
caps = None caps = None
def capable(x): def capable(x):
raise error.ProgrammingError('should not be called') raise error.ProgrammingError('should not be called')
args = {}
# The client advertises support for newer protocols by adding an
# X-HgUpgrade-* header with a list of supported APIs and an
# X-HgProto-* header advertising which serializing formats it supports.
# We only support the HTTP version 2 transport and CBOR responses for
# now.
advertisev2 = ui.configbool('experimental', 'httppeer.advertise-v2')
if advertisev2:
args['headers'] = {
r'X-HgProto-1': r'cbor',
}
args['headers'].update(
encodevalueinheaders(' '.join(sorted(API_PEERS)),
'X-HgUpgrade',
# We don't know the header limit this early.
# So make it small.
1024))
req, requrl, qs = makev1commandrequest(ui, requestbuilder, caps, req, requrl, qs = makev1commandrequest(ui, requestbuilder, caps,
capable, url, 'capabilities', capable, url, 'capabilities',
{}) args)
resp = sendrequest(ui, opener, req) resp = sendrequest(ui, opener, req)
respurl, resp = parsev1commandresponse(ui, url, requrl, qs, resp, respurl, ct, resp = parsev1commandresponse(ui, url, requrl, qs, resp,
compressible=False) compressible=False,
allowcbor=advertisev2)
try: try:
rawcaps = resp.read() rawdata = resp.read()
finally: finally:
resp.close() resp.close()
return respurl, set(rawcaps.split()) if not ct.startswith('application/mercurial-'):
raise error.ProgrammingError('unexpected content-type: %s' % ct)
if advertisev2:
if ct == 'application/mercurial-cbor':
try:
info = cbor.loads(rawdata)
except cbor.CBORDecodeError:
raise error.Abort(_('error decoding CBOR from remote server'),
hint=_('try again and consider contacting '
'the server operator'))
# We got a legacy response. That's fine.
elif ct in ('application/mercurial-0.1', 'application/mercurial-0.2'):
info = {
'v1capabilities': set(rawdata.split())
}
else:
raise error.RepoError(
_('unexpected response type from server: %s') % ct)
else:
info = {
'v1capabilities': set(rawdata.split())
}
return respurl, info
def makepeer(ui, path, opener=None, requestbuilder=urlreq.request): def makepeer(ui, path, opener=None, requestbuilder=urlreq.request):
"""Construct an appropriate HTTP peer instance. """Construct an appropriate HTTP peer instance.
``opener`` is an ``url.opener`` that should be used to establish ``opener`` is an ``url.opener`` that should be used to establish
connections, perform HTTP requests. connections, perform HTTP requests.
``requestbuilder`` is the type used for constructing HTTP requests. ``requestbuilder`` is the type used for constructing HTTP requests.
It exists as an argument so extensions can override the default. It exists as an argument so extensions can override the default.
""" """
u = util.url(path) u = util.url(path)
if u.query or u.fragment: if u.query or u.fragment:
raise error.Abort(_('unsupported URL component: "%s"') % raise error.Abort(_('unsupported URL component: "%s"') %
(u.query or u.fragment)) (u.query or u.fragment))
# urllib cannot handle URLs with embedded user or passwd. # urllib cannot handle URLs with embedded user or passwd.
url, authinfo = u.authinfo() url, authinfo = u.authinfo()
ui.debug('using %s\n' % url) ui.debug('using %s\n' % url)
opener = opener or urlmod.opener(ui, authinfo) opener = opener or urlmod.opener(ui, authinfo)
respurl, caps = performhandshake(ui, url, opener, requestbuilder) respurl, info = performhandshake(ui, url, opener, requestbuilder)
return httppeer(ui, path, respurl, opener, requestbuilder, caps) # Given the intersection of APIs that both we and the server support,
# sort by their advertised priority and pick the first one.
#
# TODO consider making this request-based and interface driven. For
# example, the caller could say "I want a peer that does X." It's quite
# possible that not all peers would do that. Since we know the service
# capabilities, we could filter out services not meeting the
# requirements. Possibly by consulting the interfaces defined by the
# peer type.
apipeerchoices = set(info.get('apis', {}).keys()) & set(API_PEERS.keys())
preferredchoices = sorted(apipeerchoices,
key=lambda x: API_PEERS[x]['priority'],
reverse=True)
for service in preferredchoices:
apipath = '%s/%s' % (info['apibase'].rstrip('/'), service)
return API_PEERS[service]['init'](ui, respurl, apipath, opener,
requestbuilder,
info['apis'][service])
# Failed to construct an API peer. Fall back to legacy.
return httppeer(ui, path, respurl, opener, requestbuilder,
info['v1capabilities'])
def instance(ui, path, create): def instance(ui, path, create):
if create: if create:
raise error.Abort(_('cannot create new http repository')) raise error.Abort(_('cannot create new http repository'))
try: try:
if path.startswith('https:') and not urlmod.has_https: if path.startswith('https:') and not urlmod.has_https:
raise error.Abort(_('Python support for SSL and HTTPS ' raise error.Abort(_('Python support for SSL and HTTPS '
'is not installed')) 'is not installed'))

tests/test-http-protocol.t

$ . $TESTDIR/wireprotohelpers.sh
$ cat >> $HGRCPATH << EOF $ cat >> $HGRCPATH << EOF
> [web] > [web]
> push_ssl = false > push_ssl = false
> allow_push = * > allow_push = *
> EOF > EOF
$ hg init server $ hg init server
$ cd server $ cd server
s> Date: $HTTP_DATE$\r\n s> Date: $HTTP_DATE$\r\n
s> Content-Type: application/mercurial-0.1\r\n s> Content-Type: application/mercurial-0.1\r\n
s> Content-Length: 30\r\n s> Content-Length: 30\r\n
s> \r\n s> \r\n
s> bookmarks\t\n s> bookmarks\t\n
s> namespaces\t\n s> namespaces\t\n
s> phases\t s> phases\t
Client with HTTPv2 enabled advertises that and gets old capabilities response from old server
$ hg --config experimental.httppeer.advertise-v2=true --verbose debugwireproto http://$LOCALIP:$HGPORT << EOF
> command heads
> EOF
s> GET /?cmd=capabilities HTTP/1.1\r\n
s> Accept-Encoding: identity\r\n
s> vary: X-HgProto-1,X-HgUpgrade-1\r\n
s> x-hgproto-1: cbor\r\n
s> x-hgupgrade-1: exp-http-v2-0001\r\n
s> accept: application/mercurial-0.1\r\n
s> host: $LOCALIP:$HGPORT\r\n (glob)
s> user-agent: Mercurial debugwireproto\r\n
s> \r\n
s> makefile('rb', None)
s> HTTP/1.1 200 Script output follows\r\n
s> Server: testing stub value\r\n
s> Date: $HTTP_DATE$\r\n
s> Content-Type: application/mercurial-0.1\r\n
s> Content-Length: 458\r\n
s> \r\n
s> batch branchmap $USUAL_BUNDLE2_CAPS_SERVER$ changegroupsubset compression=$BUNDLE2_COMPRESSIONS$ getbundle httpheader=1024 httpmediatype=0.1rx,0.1tx,0.2tx known lookup pushkey streamreqs=generaldelta,revlogv1 unbundle=HG10GZ,HG10BZ,HG10UN unbundlehash
sending heads command
s> GET /?cmd=heads HTTP/1.1\r\n
s> Accept-Encoding: identity\r\n
s> vary: X-HgProto-1\r\n
s> x-hgproto-1: 0.1 0.2 comp=$USUAL_COMPRESSIONS$ partial-pull\r\n
s> accept: application/mercurial-0.1\r\n
s> host: $LOCALIP:$HGPORT\r\n (glob)
s> user-agent: Mercurial debugwireproto\r\n
s> \r\n
s> makefile('rb', None)
s> HTTP/1.1 200 Script output follows\r\n
s> Server: testing stub value\r\n
s> Date: $HTTP_DATE$\r\n
s> Content-Type: application/mercurial-0.1\r\n
s> Content-Length: 41\r\n
s> \r\n
s> 0000000000000000000000000000000000000000\n
response: b'0000000000000000000000000000000000000000\n'
$ killdaemons.py
$ enablehttpv2 empty
$ hg -R empty serve -p $HGPORT -d --pid-file hg.pid
$ cat hg.pid > $DAEMON_PIDS
Client with HTTPv2 enabled automatically upgrades if the server supports it
$ hg --config experimental.httppeer.advertise-v2=true --verbose debugwireproto http://$LOCALIP:$HGPORT << EOF
> command heads
> EOF
s> GET /?cmd=capabilities HTTP/1.1\r\n
s> Accept-Encoding: identity\r\n
s> vary: X-HgProto-1,X-HgUpgrade-1\r\n
s> x-hgproto-1: cbor\r\n
s> x-hgupgrade-1: exp-http-v2-0001\r\n
s> accept: application/mercurial-0.1\r\n
s> host: $LOCALIP:$HGPORT\r\n (glob)
s> user-agent: Mercurial debugwireproto\r\n
s> \r\n
s> makefile('rb', None)
s> HTTP/1.1 200 OK\r\n
s> Server: testing stub value\r\n
s> Date: $HTTP_DATE$\r\n
s> Content-Type: application/mercurial-cbor\r\n
s> Content-Length: 879\r\n
s> \r\n
s> \xa3Dapis\xa1Pexp-http-v2-0001\xa2Hcommands\xa7Eheads\xa2Dargs\xa1Jpubliconly\xf4Kpermissions\x81DpullEknown\xa2Dargs\xa1Enodes\x81HdeadbeefKpermissions\x81DpullFlookup\xa2Dargs\xa1CkeyCfooKpermissions\x81DpullGpushkey\xa2Dargs\xa4CkeyCkeyCnewCnewColdColdInamespaceBnsKpermissions\x81DpushHlistkeys\xa2Dargs\xa1InamespaceBnsKpermissions\x81DpullIbranchmap\xa2Dargs\xa0Kpermissions\x81DpullLcapabilities\xa2Dargs\xa0Kpermissions\x81DpullKcompression\x82\xa1DnameDzstd\xa1DnameDzlibGapibaseDapi/Nv1capabilitiesY\x01\xcabatch branchmap $USUAL_BUNDLE2_CAPS_SERVER$ changegroupsubset compression=$BUNDLE2_COMPRESSIONS$ getbundle httpheader=1024 httpmediatype=0.1rx,0.1tx,0.2tx known lookup pushkey streamreqs=generaldelta,revlogv1 unbundle=HG10GZ,HG10BZ,HG10UN unbundlehash
sending heads command
s> POST /api/exp-http-v2-0001/ro/heads HTTP/1.1\r\n
s> Accept-Encoding: identity\r\n
s> accept: application/mercurial-exp-framing-0003\r\n
s> content-type: application/mercurial-exp-framing-0003\r\n
s> content-length: 20\r\n
s> host: $LOCALIP:$HGPORT\r\n (glob)
s> user-agent: Mercurial debugwireproto\r\n
s> \r\n
s> \x0c\x00\x00\x01\x00\x01\x01\x11\xa1DnameEheads
s> makefile('rb', None)
s> HTTP/1.1 200 OK\r\n
s> Server: testing stub value\r\n
s> Date: $HTTP_DATE$\r\n
s> Content-Type: application/mercurial-exp-framing-0003\r\n
s> Transfer-Encoding: chunked\r\n
s> \r\n
s> 1e\r\n
s> \x16\x00\x00\x01\x00\x02\x01F
s> \x81T\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00
s> \r\n
received frame(size=22; request=1; stream=2; streamflags=stream-begin; type=bytes-response; flags=eos|cbor)
s> 0\r\n
s> \r\n
response: [[b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00']]
$ killdaemons.py $ killdaemons.py