httppeer: add support for httppostargs when we're sending a file

This is probably only used in the 'unbundle' command, but the code
ended up being cleaner to make it generic and treat *all* httppostargs
with a non-args request body as though they were file-like in
nature. It also means we get test coverage more or less for free. A
previous version of this change didn't use io.BytesIO, and it was a
lot more complicated.

This also fixes a server-side bug, so anyone using httppostargs should
update all of their servers to this revision or later *before* this
gets to their clients, otherwise servers will hang trying to over-read
the POST body.

Differential Revision: https://phab.mercurial-scm.org/D231

mercurial/hgweb/protocol.py

@@ -1,198 +1,201
 #
 # Copyright 21 May 2005 - (c) 2005 Jake Edge <[e-mail]>
 # Copyright 2005-2007 Matt Mackall <[e-mail]>
 #
 # This software may be used and distributed according to the terms of the
 # GNU General Public License version 2 or any later version.
 
 from __future__ import absolute_import
 
 import cgi
 import struct
 
 from .common import (
     HTTP_OK,
 )
 
 from .. import (
     util,
     wireproto,
 )
 stringio = util.stringio
 
 urlerr = util.urlerr
 urlreq = util.urlreq
 
 HGTYPE = 'application/mercurial-0.1'
 HGTYPE2 = 'application/mercurial-0.2'
 HGERRTYPE = 'application/hg-error'
 
 def decodevaluefromheaders(req, headerprefix):
     """Decode a long value from multiple HTTP request headers."""
     chunks = []
     i = 1
     while True:
         v = req.env.get('HTTP_%s_%d' % (
             headerprefix.upper().replace('-', '_'), i))
         if v is None:
             break
         chunks.append(v)
         i += 1
 
     return ''.join(chunks)
 
 class webproto(wireproto.abstractserverproto):
     def __init__(self, req, ui):
         self.req = req
         self.response = ''
         self.ui = ui
         self.name = 'http'
 
     def getargs(self, args):
         knownargs = self._args()
         data = {}
         keys = args.split()
         for k in keys:
             if k == '*':
                 star = {}
                 for key in knownargs.keys():
                     if key != 'cmd' and key not in keys:
                         star[key] = knownargs[key][0]
                 data['*'] = star
             else:
                 data[k] = knownargs[k][0]
         return [data[k] for k in keys]
     def _args(self):
         args = self.req.form.copy()
         postlen = int(self.req.env.get('HTTP_X_HGARGS_POST', 0))
         if postlen:
             args.update(cgi.parse_qs(
                 self.req.read(postlen), keep_blank_values=True))
             return args
 
         argvalue = decodevaluefromheaders(self.req, 'X-HgArg')
         args.update(cgi.parse_qs(argvalue, keep_blank_values=True))
         return args
     def getfile(self, fp):
         length = int(self.req.env['CONTENT_LENGTH'])
+        # If httppostargs is used, we need to read Content-Length
+        # minus the amount that was consumed by args.
+        length -= int(self.req.env.get('HTTP_X_HGARGS_POST', 0))
         for s in util.filechunkiter(self.req, limit=length):
             fp.write(s)
     def redirect(self):
         self.oldio = self.ui.fout, self.ui.ferr
         self.ui.ferr = self.ui.fout = stringio()
     def restore(self):
         val = self.ui.fout.getvalue()
         self.ui.ferr, self.ui.fout = self.oldio
         return val
 
     def _client(self):
         return 'remote:%s:%s:%s' % (
             self.req.env.get('wsgi.url_scheme') or 'http',
             urlreq.quote(self.req.env.get('REMOTE_HOST', '')),
             urlreq.quote(self.req.env.get('REMOTE_USER', '')))
 
     def responsetype(self, v1compressible=False):
         """Determine the appropriate response type and compression settings.
 
         The ``v1compressible`` argument states whether the response with
         application/mercurial-0.1 media types should be zlib compressed.
 
         Returns a tuple of (mediatype, compengine, engineopts).
         """
         # For now, if it isn't compressible in the old world, it's never
         # compressible. We can change this to send uncompressed 0.2 payloads
         # later.
         if not v1compressible:
             return HGTYPE, None, None
 
         # Determine the response media type and compression engine based
         # on the request parameters.
         protocaps = decodevaluefromheaders(self.req, 'X-HgProto').split(' ')
 
         if '0.2' in protocaps:
             # Default as defined by wire protocol spec.
             compformats = ['zlib', 'none']
             for cap in protocaps:
                 if cap.startswith('comp='):
                     compformats = cap[5:].split(',')
                     break
 
             # Now find an agreed upon compression format.
             for engine in wireproto.supportedcompengines(self.ui, self,
                                                          util.SERVERROLE):
                 if engine.wireprotosupport().name in compformats:
                     opts = {}
                     level = self.ui.configint('server',
                                               '%slevel' % engine.name())
                     if level is not None:
                         opts['level'] = level
 
                     return HGTYPE2, engine, opts
 
             # No mutually supported compression format. Fall back to the
             # legacy protocol.
 
         # Don't allow untrusted settings because disabling compression or
         # setting a very high compression level could lead to flooding
         # the server's network or CPU.
         opts = {'level': self.ui.configint('server', 'zliblevel')}
         return HGTYPE, util.compengines['zlib'], opts
 
 def iscmd(cmd):
     return cmd in wireproto.commands
 
 def call(repo, req, cmd):
     p = webproto(req, repo.ui)
 
     def genversion2(gen, compress, engine, engineopts):
         # application/mercurial-0.2 always sends a payload header
         # identifying the compression engine.
         name = engine.wireprotosupport().name
         assert 0 < len(name) < 256
         yield struct.pack('B', len(name))
         yield name
 
         if compress:
             for chunk in engine.compressstream(gen, opts=engineopts):
                 yield chunk
         else:
             for chunk in gen:
                 yield chunk
 
     rsp = wireproto.dispatch(repo, p, cmd)
     if isinstance(rsp, str):
         req.respond(HTTP_OK, HGTYPE, body=rsp)
         return []
     elif isinstance(rsp, wireproto.streamres):
         if rsp.reader:
             gen = iter(lambda: rsp.reader.read(32768), '')
         else:
             gen = rsp.gen
 
         # This code for compression should not be streamres specific. It
         # is here because we only compress streamres at the moment.
         mediatype, engine, engineopts = p.responsetype(rsp.v1compressible)
 
         if mediatype == HGTYPE and rsp.v1compressible:
             gen = engine.compressstream(gen, engineopts)
         elif mediatype == HGTYPE2:
             gen = genversion2(gen, rsp.v1compressible, engine, engineopts)
 
         req.respond(HTTP_OK, mediatype)
         return gen
     elif isinstance(rsp, wireproto.pushres):
         val = p.restore()
         rsp = '%d\n%s' % (rsp.res, val)
         req.respond(HTTP_OK, HGTYPE, body=rsp)
         return []
     elif isinstance(rsp, wireproto.pusherr):
         # drain the incoming bundle
         req.drain()
         p.restore()
         rsp = '0\n%s\n' % rsp.res
         req.respond(HTTP_OK, HGTYPE, body=rsp)
         return []
     elif isinstance(rsp, wireproto.ooberror):
         rsp = rsp.message
         req.respond(HTTP_OK, HGERRTYPE, body=rsp)
         return []

mercurial/httppeer.py

@@ -1,422 +1,464
 # httppeer.py - HTTP repository proxy classes for mercurial
 #
 # Copyright 2005, 2006 Matt Mackall <[e-mail]>
 # Copyright 2006 Vadim Gelfer <[e-mail]>
 #
 # This software may be used and distributed according to the terms of the
 # GNU General Public License version 2 or any later version.
 
 from __future__ import absolute_import
 
 import errno
+import io
 import os
 import socket
 import struct
 import tempfile
 
 from .i18n import _
 from .node import nullid
 from . import (
     bundle2,
     error,
     httpconnection,
     pycompat,
     statichttprepo,
     url,
     util,
     wireproto,
 )
 
 httplib = util.httplib
 urlerr = util.urlerr
 urlreq = util.urlreq
 
 def encodevalueinheaders(value, header, limit):
     """Encode a string value into multiple HTTP headers.
 
     ``value`` will be encoded into 1 or more HTTP headers with the names
     ``header-<N>`` where ``<N>`` is an integer starting at 1. Each header
     name + value will be at most ``limit`` bytes long.
 
     Returns an iterable of 2-tuples consisting of header names and values.
     """
     fmt = header + '-%s'
     valuelen = limit - len(fmt % '000') - len(': \r\n')
     result = []
 
     n = 0
     for i in xrange(0, len(value), valuelen):
         n += 1
         result.append((fmt % str(n), value[i:i + valuelen]))
 
     return result
 
 def _wraphttpresponse(resp):
     """Wrap an HTTPResponse with common error handlers.
 
     This ensures that any I/O from any consumer raises the appropriate
     error and messaging.
     """
     origread = resp.read
 
     class readerproxy(resp.__class__):
         def read(self, size=None):
             try:
                 return origread(size)
             except httplib.IncompleteRead as e:
                 # e.expected is an integer if length known or None otherwise.
                 if e.expected:
                     msg = _('HTTP request error (incomplete response; '
                             'expected %d bytes got %d)') % (e.expected,
                                                            len(e.partial))
                 else:
                     msg = _('HTTP request error (incomplete response)')
 
                 raise error.PeerTransportError(
                     msg,
                     hint=_('this may be an intermittent network failure; '
                            'if the error persists, consider contacting the '
                            'network or server operator'))
             except httplib.HTTPException as e:
                 raise error.PeerTransportError(
                     _('HTTP request error (%s)') % e,
                     hint=_('this may be an intermittent network failure; '
                            'if the error persists, consider contacting the '
                            'network or server operator'))
 
     resp.__class__ = readerproxy
 
+class _multifile(object):
+    def __init__(self, *fileobjs):
+        for f in fileobjs:
+            if not util.safehasattr(f, 'length'):
+                raise ValueError(
+                    '_multifile only supports file objects that '
+                    'have a length but this one does not:', type(f), f)
+        self._fileobjs = fileobjs
+        self._index = 0
+
+    @property
+    def length(self):
+        return sum(f.length for f in self._fileobjs)
+
+    def read(self, amt=None):
+        if amt <= 0:
+            return ''.join(f.read() for f in self._fileobjs)
+        parts = []
+        while amt and self._index < len(self._fileobjs):
+            parts.append(self._fileobjs[self._index].read(amt))
+            got = len(parts[-1])
+            if got < amt:
+                self._index += 1
+            amt -= got
+        return ''.join(parts)
+
+    def seek(self, offset, whence=os.SEEK_SET):
+        if whence != os.SEEK_SET:
+            raise NotImplementedError(
+                '_multifile does not support anything other'
+                ' than os.SEEK_SET for whence on seek()')
+        if offset != 0:
+            raise NotImplementedError(
+                '_multifile only supports seeking to start, but that '
+                'could be fixed if you need it')
+        for f in self._fileobjs:
+            f.seek(0)
+        self._index = 0
+
 class httppeer(wireproto.wirepeer):
     def __init__(self, ui, path):
         self._path = path
         self._caps = None
         self._urlopener = None
         self._requestbuilder = None
         u = util.url(path)
         if u.query or u.fragment:
             raise error.Abort(_('unsupported URL component: "%s"') %
                              (u.query or u.fragment))
 
         # urllib cannot handle URLs with embedded user or passwd
         self._url, authinfo = u.authinfo()
 
         self._ui = ui
         ui.debug('using %s\n' % self._url)
 
         self._urlopener = url.opener(ui, authinfo)
         self._requestbuilder = urlreq.request
 
     def __del__(self):
         urlopener = getattr(self, '_urlopener', None)
         if urlopener:
             for h in urlopener.handlers:
                 h.close()
                 getattr(h, "close_all", lambda : None)()
 
     # Begin of _basepeer interface.
 
     @util.propertycache
     def ui(self):
         return self._ui
 
     def url(self):
         return self._path
 
     def local(self):
         return None
 
     def peer(self):
         return self
 
     def canpush(self):
         return True
 
     def close(self):
         pass
 
     # End of _basepeer interface.
 
     # Begin of _basewirepeer interface.
 
     def capabilities(self):
         if self._caps is None:
             try:
                 self._fetchcaps()
             except error.RepoError:
                 self._caps = set()
             self.ui.debug('capabilities: %s\n' %
                           (' '.join(self._caps or ['none'])))
         return self._caps
 
     # End of _basewirepeer interface.
 
     # look up capabilities only when needed
 
     def _fetchcaps(self):
         self._caps = set(self._call('capabilities').split())
 
     def _callstream(self, cmd, _compressible=False, **args):
         if cmd == 'pushkey':
             args['data'] = ''
         data = args.pop('data', None)
         headers = args.pop('headers', {})
 
         self.ui.debug("sending %s command\n" % cmd)
         q = [('cmd', cmd)]
         headersize = 0
         varyheaders = []
         # Important: don't use self.capable() here or else you end up
         # with infinite recursion when trying to look up capabilities
         # for the first time.
         postargsok = self._caps is not None and 'httppostargs' in self._caps
-        # TODO: support for httppostargs when data is a file-like
-        # object rather than a basestring
-        canmungedata = not data or isinstance(data, basestring)
-        if postargsok and canmungedata:
+        if postargsok and args:
             strargs = urlreq.urlencode(sorted(args.items()))
-            if strargs:
                 if not data:
                     data = strargs
-                elif isinstance(data, basestring):
-                    data = strargs + data
+            else:
+                if isinstance(data, basestring):
+                    i = io.BytesIO(data)
+                    i.length = len(data)
+                    data = i
+                argsio = io.BytesIO(strargs)
+                argsio.length = len(strargs)
+                data = _multifile(argsio, data)
                 headers['X-HgArgs-Post'] = len(strargs)
         else:
             if len(args) > 0:
                 httpheader = self.capable('httpheader')
                 if httpheader:
                     headersize = int(httpheader.split(',', 1)[0])
             if headersize > 0:
                 # The headers can typically carry more data than the URL.
                 encargs = urlreq.urlencode(sorted(args.items()))
                 for header, value in encodevalueinheaders(encargs, 'X-HgArg',
                                                           headersize):
                     headers[header] = value
                     varyheaders.append(header)
             else:
                 q += sorted(args.items())
         qs = '?%s' % urlreq.urlencode(q)
         cu = "%s%s" % (self._url, qs)
         size = 0
         if util.safehasattr(data, 'length'):
             size = data.length
         elif data is not None:
             size = len(data)
         if size and self.ui.configbool('ui', 'usehttp2'):
             headers['Expect'] = '100-Continue'
             headers['X-HgHttp2'] = '1'
         if data is not None and 'Content-Type' not in headers:
             headers['Content-Type'] = 'application/mercurial-0.1'
 
         # Tell the server we accept application/mercurial-0.2 and multiple
         # compression formats if the server is capable of emitting those
         # payloads.
         protoparams = []
 
         mediatypes = set()
         if self._caps is not None:
             mt = self.capable('httpmediatype')
             if mt:
                 protoparams.append('0.1')
                 mediatypes = set(mt.split(','))
 
         if '0.2tx' in mediatypes:
             protoparams.append('0.2')
 
         if '0.2tx' in mediatypes and self.capable('compression'):
             # We /could/ compare supported compression formats and prune
             # non-mutually supported or error if nothing is mutually supported.
             # For now, send the full list to the server and have it error.
             comps = [e.wireprotosupport().name for e in
                      util.compengines.supportedwireengines(util.CLIENTROLE)]
             protoparams.append('comp=%s' % ','.join(comps))
 
         if protoparams:
             protoheaders = encodevalueinheaders(' '.join(protoparams),
                                                 'X-HgProto',
                                                 headersize or 1024)
             for header, value in protoheaders:
                 headers[header] = value
                 varyheaders.append(header)
 
         if varyheaders:
             headers['Vary'] = ','.join(varyheaders)
 
         req = self._requestbuilder(cu, data, headers)
 
         if data is not None:
             self.ui.debug("sending %s bytes\n" % size)
             req.add_unredirected_header('Content-Length', '%d' % size)
         try:
             resp = self._urlopener.open(req)
         except urlerr.httperror as inst:
             if inst.code == 401:
                 raise error.Abort(_('authorization failed'))
             raise
         except httplib.HTTPException as inst:
             self.ui.debug('http error while sending %s command\n' % cmd)
             self.ui.traceback()
             raise IOError(None, inst)
 
         # Insert error handlers for common I/O failures.
         _wraphttpresponse(resp)
 
         # record the url we got redirected to
         resp_url = resp.geturl()
         if resp_url.endswith(qs):
             resp_url = resp_url[:-len(qs)]
         if self._url.rstrip('/') != resp_url.rstrip('/'):
             if not self.ui.quiet:
                 self.ui.warn(_('real URL is %s\n') % resp_url)
         self._url = resp_url
         try:
             proto = resp.getheader('content-type')
         except AttributeError:
             proto = resp.headers.get('content-type', '')
 
         safeurl = util.hidepassword(self._url)
         if proto.startswith('application/hg-error'):
             raise error.OutOfBandError(resp.read())
         # accept old "text/plain" and "application/hg-changegroup" for now
         if not (proto.startswith('application/mercurial-') or
                 (proto.startswith('text/plain')
                  and not resp.headers.get('content-length')) or
                 proto.startswith('application/hg-changegroup')):
             self.ui.debug("requested URL: '%s'\n" % util.hidepassword(cu))
             raise error.RepoError(
                 _("'%s' does not appear to be an hg repository:\n"
                   "---%%<--- (%s)\n%s\n---%%<---\n")
                 % (safeurl, proto or 'no content-type', resp.read(1024)))
 
         if proto.startswith('application/mercurial-'):
             try:
                 version = proto.split('-', 1)[1]
                 version_info = tuple([int(n) for n in version.split('.')])
             except ValueError:
                 raise error.RepoError(_("'%s' sent a broken Content-Type "
                                         "header (%s)") % (safeurl, proto))
 
             # TODO consider switching to a decompression reader that uses
             # generators.
             if version_info == (0, 1):
                 if _compressible:
                     return util.compengines['zlib'].decompressorreader(resp)
                 return resp
             elif version_info == (0, 2):
                 # application/mercurial-0.2 always identifies the compression
                 # engine in the payload header.
                 elen = struct.unpack('B', resp.read(1))[0]
                 ename = resp.read(elen)
                 engine = util.compengines.forwiretype(ename)
                 return engine.decompressorreader(resp)
             else:
                 raise error.RepoError(_("'%s' uses newer protocol %s") %
                                       (safeurl, version))
 
         if _compressible:
             return util.compengines['zlib'].decompressorreader(resp)
 
         return resp
 
     def _call(self, cmd, **args):
         fp = self._callstream(cmd, **args)
         try:
             return fp.read()
         finally:
             # if using keepalive, allow connection to be reused
             fp.close()
 
     def _callpush(self, cmd, cg, **args):
         # have to stream bundle to a temp file because we do not have
         # http 1.1 chunked transfer.
 
         types = self.capable('unbundle')
         try:
             types = types.split(',')
         except AttributeError:
             # servers older than d1b16a746db6 will send 'unbundle' as a
             # boolean capability. They only support headerless/uncompressed
             # bundles.
             types = [""]
         for x in types:
             if x in bundle2.bundletypes:
                 type = x
                 break
 
         tempname = bundle2.writebundle(self.ui, cg, None, type)
         fp = httpconnection.httpsendfile(self.ui, tempname, "rb")
         headers = {'Content-Type': 'application/mercurial-0.1'}
 
         try:
             r = self._call(cmd, data=fp, headers=headers, **args)
             vals = r.split('\n', 1)
             if len(vals) < 2:
                 raise error.ResponseError(_("unexpected response:"), r)
             return vals
         except socket.error as err:
             if err.args[0] in (errno.ECONNRESET, errno.EPIPE):
                 raise error.Abort(_('push failed: %s') % err.args[1])
             raise error.Abort(err.args[1])
         finally:
             fp.close()
             os.unlink(tempname)
 
     def _calltwowaystream(self, cmd, fp, **args):
         fh = None
         fp_ = None
         filename = None
         try:
             # dump bundle to disk
             fd, filename = tempfile.mkstemp(prefix="hg-bundle-", suffix=".hg")
             fh = os.fdopen(fd, pycompat.sysstr("wb"))
             d = fp.read(4096)
             while d:
                 fh.write(d)
                 d = fp.read(4096)
             fh.close()
             # start http push
             fp_ = httpconnection.httpsendfile(self.ui, filename, "rb")
             headers = {'Content-Type': 'application/mercurial-0.1'}
             return self._callstream(cmd, data=fp_, headers=headers, **args)
         finally:
             if fp_ is not None:
                 fp_.close()
             if fh is not None:
                 fh.close()
                 os.unlink(filename)
 
     def _callcompressable(self, cmd, **args):
         return self._callstream(cmd, _compressible=True, **args)
 
     def _abort(self, exception):
         raise exception
 
 class httpspeer(httppeer):
     def __init__(self, ui, path):
         if not url.has_https:
             raise error.Abort(_('Python support for SSL and HTTPS '
                                'is not installed'))
         httppeer.__init__(self, ui, path)
 
 def instance(ui, path, create):
     if create:
         raise error.Abort(_('cannot create new http repository'))
     try:
         if path.startswith('https:'):
             inst = httpspeer(ui, path)
         else:
             inst = httppeer(ui, path)
         try:
             # Try to do useful work when checking compatibility.
             # Usually saves a roundtrip since we want the caps anyway.
             inst._fetchcaps()
         except error.RepoError:
             # No luck, try older compatibility check.
             inst.between([(nullid, nullid)])
         return inst
     except error.RepoError as httpexception:
         try:
             r = statichttprepo.instance(ui, "static-" + path, create)
             ui.note(_('(falling back to static-http)\n'))
             return r
         except error.RepoError:
             raise httpexception # use the original http RepoError instead