This is an archive of the discontinued Mercurial Phabricator instance.

Differential D2775

hgweb: create dedicated type for WSGI responses
ClosedPublic

Authored by indygreg on Mar 9 2018, 11:53 PM.

Details

Reviewers
durin42
Group Reviewers
hg-reviewers
Commits
rHGa88d68dc3ee8: hgweb: create dedicated type for WSGI responses
Summary

We have refactored the request side of WSGI processing into a dedicated
type. Now let's do the same thing for the response side.

We invent a `wsgiresponse` type. It takes an instance of a
request (for consulation) and the WSGI application's "start_response"
handler.

The type basically allows setting the HTTP status line, response
headers, and the response body.

The WSGI application calls sendresponse() to start sending output.
Output is emitted as a generator to be fed through the WSGI application.
According to PEP 3333, this is the preferred way for output to be
transmitted. (Our legacy `wsgirequest` exposed a write() to send
data. We do not wish to support this API because it isn't recommended
by PEP 3333.)

The wire protocol code has been ported to use the new API.

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 9 2018, 11:53 PM
indygreg edited the summary of this revision. (Show Details)Mar 10 2018, 3:03 PM
indygreg updated this revision to Diff 6837.
durin42 accepted this revision.Mar 12 2018, 5:02 PM
This revision is now accepted and ready to land.Mar 12 2018, 5:02 PM
Closed by commit rHGa88d68dc3ee8: hgweb: create dedicated type for WSGI responses (authored by indygreg). · Explain WhyMar 12 2018, 5:32 PM
This revision was automatically updated to reflect the committed changes.

Revision Contents
Changeset List

Diff 6914

mercurial/hgweb/hgweb_mod.py

with self._obtainrepo() as repo: with self._obtainrepo() as repo:
profile = repo.ui.configbool('profiling', 'enabled') profile = repo.ui.configbool('profiling', 'enabled')
with profiling.profile(repo.ui, enabled=profile): with profiling.profile(repo.ui, enabled=profile):
for r in self._runwsgi(wsgireq, repo): for r in self._runwsgi(wsgireq, repo):
yield r yield r
def _runwsgi(self, wsgireq, repo): def _runwsgi(self, wsgireq, repo):
req = wsgireq.req req = wsgireq.req
res = wsgireq.res
rctx = requestcontext(self, repo) rctx = requestcontext(self, repo)
# This state is global across all threads. # This state is global across all threads.
encoding.encoding = rctx.config('web', 'encoding') encoding.encoding = rctx.config('web', 'encoding')
rctx.repo.ui.environ = wsgireq.env rctx.repo.ui.environ = wsgireq.env
if rctx.csp: if rctx.csp:
# hgwebdir may have added CSP header. Since we generate our own, # hgwebdir may have added CSP header. Since we generate our own,
# replace it. # replace it.
wsgireq.headers = [h for h in wsgireq.headers wsgireq.headers = [h for h in wsgireq.headers
if h[0] != 'Content-Security-Policy'] if h[0] != 'Content-Security-Policy']
wsgireq.headers.append(('Content-Security-Policy', rctx.csp)) wsgireq.headers.append(('Content-Security-Policy', rctx.csp))
res.headers['Content-Security-Policy'] = rctx.csp
handled, res = wireprotoserver.handlewsgirequest( handled = wireprotoserver.handlewsgirequest(
rctx, wsgireq, req, self.check_perm) rctx, wsgireq, req, res, self.check_perm)
if handled: if handled:
return res return res.sendresponse()
if req.havepathinfo: if req.havepathinfo:
query = req.dispatchpath query = req.dispatchpath
else: else:
query = req.querystring.partition('&')[0].partition(';')[0] query = req.querystring.partition('&')[0].partition(';')[0]
# translate user-visible url structure to internal structure # translate user-visible url structure to internal structure

mercurial/hgweb/request.py

HTTP_NOT_MODIFIED, HTTP_NOT_MODIFIED,
statusmessage, statusmessage,
) )
from ..thirdparty import ( from ..thirdparty import (
attr, attr,
) )
from .. import ( from .. import (
error,
pycompat, pycompat,
util, util,
) )
@attr.s(frozen=True) @attr.s(frozen=True)
class parsedrequest(object): class parsedrequest(object):
"""Represents a parsed WSGI request. """Represents a parsed WSGI request.
dispatchparts=dispatchparts, dispatchpath=dispatchpath, dispatchparts=dispatchparts, dispatchpath=dispatchpath,
havepathinfo='PATH_INFO' in env, havepathinfo='PATH_INFO' in env,
querystring=querystring, querystring=querystring,
querystringlist=querystringlist, querystringlist=querystringlist,
querystringdict=querystringdict, querystringdict=querystringdict,
headers=headers, headers=headers,
bodyfh=bodyfh) bodyfh=bodyfh)
class wsgiresponse(object):
"""Represents a response to a WSGI request.
A response consists of a status line, headers, and a body.
Consumers must populate the ``status`` and ``headers`` fields and
make a call to a ``setbody*()`` method before the response can be
issued.
When it is time to start sending the response over the wire,
``sendresponse()`` is called. It handles emitting the header portion
of the response message. It then yields chunks of body data to be
written to the peer. Typically, the WSGI application itself calls
and returns the value from ``sendresponse()``.
"""
def __init__(self, req, startresponse):
"""Create an empty response tied to a specific request.
``req`` is a ``parsedrequest``. ``startresponse`` is the
``start_response`` function passed to the WSGI application.
"""
self._req = req
self._startresponse = startresponse
self.status = None
self.headers = wsgiheaders.Headers([])
self._bodybytes = None
self._bodygen = None
self._started = False
def setbodybytes(self, b):
"""Define the response body as static bytes."""
if self._bodybytes is not None or self._bodygen is not None:
raise error.ProgrammingError('cannot define body multiple times')
self._bodybytes = b
self.headers['Content-Length'] = '%d' % len(b)
def setbodygen(self, gen):
"""Define the response body as a generator of bytes."""
if self._bodybytes is not None or self._bodygen is not None:
raise error.ProgrammingError('cannot define body multiple times')
self._bodygen = gen
def sendresponse(self):
"""Send the generated response to the client.
Before this is called, ``status`` must be set and one of
``setbodybytes()`` or ``setbodygen()`` must be called.
Calling this method multiple times is not allowed.
"""
if self._started:
raise error.ProgrammingError('sendresponse() called multiple times')
self._started = True
if not self.status:
raise error.ProgrammingError('status line not defined')
if self._bodybytes is None and self._bodygen is None:
raise error.ProgrammingError('response body not defined')
# Various HTTP clients (notably httplib) won't read the HTTP response
# until the HTTP request has been sent in full. If servers (us) send a
# response before the HTTP request has been fully sent, the connection
# may deadlock because neither end is reading.
#
# We work around this by "draining" the request data before
# sending any response in some conditions.
drain = False
close = False
# If the client sent Expect: 100-continue, we assume it is smart enough
# to deal with the server sending a response before reading the request.
# (httplib doesn't do this.)
if self._req.headers.get('Expect', '').lower() == '100-continue':
pass
# Only tend to request methods that have bodies. Strictly speaking,
# we should sniff for a body. But this is fine for our existing
# WSGI applications.
elif self._req.method not in ('POST', 'PUT'):
pass
else:
# If we don't know how much data to read, there's no guarantee
# that we can drain the request responsibly. The WSGI
# specification only says that servers *should* ensure the
# input stream doesn't overrun the actual request. So there's
# no guarantee that reading until EOF won't corrupt the stream
# state.
if not isinstance(self._req.bodyfh, util.cappedreader):
close = True
else:
# We /could/ only drain certain HTTP response codes. But 200 and
# non-200 wire protocol responses both require draining. Since
# we have a capped reader in place for all situations where we
# drain, it is safe to read from that stream. We'll either do
# a drain or no-op if we're already at EOF.
drain = True
if close:
self.headers['Connection'] = 'Close'
if drain:
assert isinstance(self._req.bodyfh, util.cappedreader)
while True:
chunk = self._req.bodyfh.read(32768)
if not chunk:
break
self._startresponse(pycompat.sysstr(self.status), self.headers.items())
if self._bodybytes:
yield self._bodybytes
elif self._bodygen:
for chunk in self._bodygen:
yield chunk
else:
error.ProgrammingError('do not know how to send body')
class wsgirequest(object): class wsgirequest(object):
"""Higher-level API for a WSGI request. """Higher-level API for a WSGI request.
WSGI applications are invoked with 2 arguments. They are used to WSGI applications are invoked with 2 arguments. They are used to
instantiate instances of this class, which provides higher-level APIs instantiate instances of this class, which provides higher-level APIs
for obtaining request parameters, writing HTTP output, etc. for obtaining request parameters, writing HTTP output, etc.
""" """
def __init__(self, wsgienv, start_response): def __init__(self, wsgienv, start_response):
self.err = wsgienv[r'wsgi.errors'] self.err = wsgienv[r'wsgi.errors']
self.threaded = wsgienv[r'wsgi.multithread'] self.threaded = wsgienv[r'wsgi.multithread']
self.multiprocess = wsgienv[r'wsgi.multiprocess'] self.multiprocess = wsgienv[r'wsgi.multiprocess']
self.run_once = wsgienv[r'wsgi.run_once'] self.run_once = wsgienv[r'wsgi.run_once']
self.env = wsgienv self.env = wsgienv
self.req = parserequestfromenv(wsgienv, inp) self.req = parserequestfromenv(wsgienv, inp)
self.form = self.req.querystringdict self.form = self.req.querystringdict
self.res = wsgiresponse(self.req, start_response)
self._start_response = start_response self._start_response = start_response
self.server_write = None self.server_write = None
self.headers = [] self.headers = []
def respond(self, status, type, filename=None, body=None): def respond(self, status, type, filename=None, body=None):
if not isinstance(type, str): if not isinstance(type, str):
type = pycompat.sysstr(type) type = pycompat.sysstr(type)
if self._start_response is not None: if self._start_response is not None:

mercurial/wireprotoserver.py

# This method exists mostly so that extensions like remotefilelog can # This method exists mostly so that extensions like remotefilelog can
# disable a kludgey legacy method only over http. As of early 2018, # disable a kludgey legacy method only over http. As of early 2018,
# there are no other known users, so with any luck we can discard this # there are no other known users, so with any luck we can discard this
# hook if remotefilelog becomes a first-party extension. # hook if remotefilelog becomes a first-party extension.
def iscmd(cmd): def iscmd(cmd):
return cmd in wireproto.commands return cmd in wireproto.commands
def handlewsgirequest(rctx, wsgireq, req, checkperm): def handlewsgirequest(rctx, wsgireq, req, res, checkperm):
"""Possibly process a wire protocol request. """Possibly process a wire protocol request.
If the current request is a wire protocol request, the request is If the current request is a wire protocol request, the request is
processed by this function. processed by this function.
``wsgireq`` is a ``wsgirequest`` instance. ``wsgireq`` is a ``wsgirequest`` instance.
``req`` is a ``parsedrequest`` instance. ``req`` is a ``parsedrequest`` instance.
``res`` is a ``wsgiresponse`` instance.
Returns a 2-tuple of (bool, response) where the 1st element indicates Returns a bool indicating if the request was serviced. If set, the caller
whether the request was handled and the 2nd element is a return should stop processing the request, as a response has already been issued.
value for a WSGI application (often a generator of bytes).
""" """
# Avoid cycle involving hg module. # Avoid cycle involving hg module.
from .hgweb import common as hgwebcommon from .hgweb import common as hgwebcommon
repo = rctx.repo repo = rctx.repo
# HTTP version 1 wire protocol requests are denoted by a "cmd" query # HTTP version 1 wire protocol requests are denoted by a "cmd" query
# string parameter. If it isn't present, this isn't a wire protocol # string parameter. If it isn't present, this isn't a wire protocol
# request. # request.
if 'cmd' not in req.querystringdict: if 'cmd' not in req.querystringdict:
return False, None return False
cmd = req.querystringdict['cmd'][0] cmd = req.querystringdict['cmd'][0]
# The "cmd" request parameter is used by both the wire protocol and hgweb. # The "cmd" request parameter is used by both the wire protocol and hgweb.
# While not all wire protocol commands are available for all transports, # While not all wire protocol commands are available for all transports,
# if we see a "cmd" value that resembles a known wire protocol command, we # if we see a "cmd" value that resembles a known wire protocol command, we
# route it to a protocol handler. This is better than routing possible # route it to a protocol handler. This is better than routing possible
# wire protocol requests to hgweb because it prevents hgweb from using # wire protocol requests to hgweb because it prevents hgweb from using
# known wire protocol commands and it is less confusing for machine # known wire protocol commands and it is less confusing for machine
# clients. # clients.
if not iscmd(cmd): if not iscmd(cmd):
return False, None return False
# The "cmd" query string argument is only valid on the root path of the # The "cmd" query string argument is only valid on the root path of the
# repo. e.g. ``/?cmd=foo``, ``/repo?cmd=foo``. URL paths within the repo # repo. e.g. ``/?cmd=foo``, ``/repo?cmd=foo``. URL paths within the repo
# like ``/blah?cmd=foo`` are not allowed. So don't recognize the request # like ``/blah?cmd=foo`` are not allowed. So don't recognize the request
# in this case. We send an HTTP 404 for backwards compatibility reasons. # in this case. We send an HTTP 404 for backwards compatibility reasons.
if req.dispatchpath: if req.dispatchpath:
res = _handlehttperror( res.status = hgwebcommon.statusmessage(404)
hgwebcommon.ErrorResponse(hgwebcommon.HTTP_NOT_FOUND), wsgireq, res.headers['Content-Type'] = HGTYPE
req) # TODO This is not a good response to issue for this request. This
# is mostly for BC for now.
return True, res res.setbodybytes('0\n%s\n' % b'Not Found')
return True
proto = httpv1protocolhandler(wsgireq, req, repo.ui, proto = httpv1protocolhandler(wsgireq, req, repo.ui,
lambda perm: checkperm(rctx, wsgireq, perm)) lambda perm: checkperm(rctx, wsgireq, perm))
# The permissions checker should be the only thing that can raise an # The permissions checker should be the only thing that can raise an
# ErrorResponse. It is kind of a layer violation to catch an hgweb # ErrorResponse. It is kind of a layer violation to catch an hgweb
# exception here. So consider refactoring into a exception type that # exception here. So consider refactoring into a exception type that
# is associated with the wire protocol. # is associated with the wire protocol.
try: try:
res = _callhttp(repo, wsgireq, req, proto, cmd) _callhttp(repo, wsgireq, req, res, proto, cmd)
except hgwebcommon.ErrorResponse as e: except hgwebcommon.ErrorResponse as e:
res = _handlehttperror(e, wsgireq, req) for k, v in e.headers:
res.headers[k] = v
res.status = hgwebcommon.statusmessage(e.code, pycompat.bytestr(e))
# TODO This response body assumes the failed command was
# "unbundle." That assumption is not always valid.
res.setbodybytes('0\n%s\n' % pycompat.bytestr(e))
return True, res return True
def _httpresponsetype(ui, req, prefer_uncompressed): def _httpresponsetype(ui, req, prefer_uncompressed):
"""Determine the appropriate response type and compression settings. """Determine the appropriate response type and compression settings.
Returns a tuple of (mediatype, compengine, engineopts). Returns a tuple of (mediatype, compengine, engineopts).
""" """
# Determine the response media type and compression engine based # Determine the response media type and compression engine based
# on the request parameters. # on the request parameters.
# legacy protocol. # legacy protocol.
# Don't allow untrusted settings because disabling compression or # Don't allow untrusted settings because disabling compression or
# setting a very high compression level could lead to flooding # setting a very high compression level could lead to flooding
# the server's network or CPU. # the server's network or CPU.
opts = {'level': ui.configint('server', 'zliblevel')} opts = {'level': ui.configint('server', 'zliblevel')}
return HGTYPE, util.compengines['zlib'], opts return HGTYPE, util.compengines['zlib'], opts
def _callhttp(repo, wsgireq, req, proto, cmd): def _callhttp(repo, wsgireq, req, res, proto, cmd):
# Avoid cycle involving hg module.
from .hgweb import common as hgwebcommon
def genversion2(gen, engine, engineopts): def genversion2(gen, engine, engineopts):
# application/mercurial-0.2 always sends a payload header # application/mercurial-0.2 always sends a payload header
# identifying the compression engine. # identifying the compression engine.
name = engine.wireprotosupport().name name = engine.wireprotosupport().name
assert 0 < len(name) < 256 assert 0 < len(name) < 256
yield struct.pack('B', len(name)) yield struct.pack('B', len(name))
yield name yield name
for chunk in gen: for chunk in gen:
yield chunk yield chunk
def setresponse(code, contenttype, bodybytes=None, bodygen=None):
if code == HTTP_OK:
res.status = '200 Script output follows'
else:
res.status = hgwebcommon.statusmessage(code)
res.headers['Content-Type'] = contenttype
if bodybytes is not None:
res.setbodybytes(bodybytes)
if bodygen is not None:
res.setbodygen(bodygen)
if not wireproto.commands.commandavailable(cmd, proto): if not wireproto.commands.commandavailable(cmd, proto):
wsgireq.respond(HTTP_OK, HGERRTYPE, setresponse(HTTP_OK, HGERRTYPE,
body=_('requested wire protocol command is not ' _('requested wire protocol command is not available over '
'available over HTTP')) 'HTTP'))
return [] return
proto.checkperm(wireproto.commands[cmd].permission) proto.checkperm(wireproto.commands[cmd].permission)
rsp = wireproto.dispatch(repo, proto, cmd) rsp = wireproto.dispatch(repo, proto, cmd)
if isinstance(rsp, bytes): if isinstance(rsp, bytes):
wsgireq.respond(HTTP_OK, HGTYPE, body=rsp) setresponse(HTTP_OK, HGTYPE, bodybytes=rsp)
return []
elif isinstance(rsp, wireprototypes.bytesresponse): elif isinstance(rsp, wireprototypes.bytesresponse):
wsgireq.respond(HTTP_OK, HGTYPE, body=rsp.data) setresponse(HTTP_OK, HGTYPE, bodybytes=rsp.data)
return []
elif isinstance(rsp, wireprototypes.streamreslegacy): elif isinstance(rsp, wireprototypes.streamreslegacy):
gen = rsp.gen setresponse(HTTP_OK, HGTYPE, bodygen=rsp.gen)
wsgireq.respond(HTTP_OK, HGTYPE)
return gen
elif isinstance(rsp, wireprototypes.streamres): elif isinstance(rsp, wireprototypes.streamres):
gen = rsp.gen gen = rsp.gen
# This code for compression should not be streamres specific. It # This code for compression should not be streamres specific. It
# is here because we only compress streamres at the moment. # is here because we only compress streamres at the moment.
mediatype, engine, engineopts = _httpresponsetype( mediatype, engine, engineopts = _httpresponsetype(
repo.ui, req, rsp.prefer_uncompressed) repo.ui, req, rsp.prefer_uncompressed)
gen = engine.compressstream(gen, engineopts) gen = engine.compressstream(gen, engineopts)
if mediatype == HGTYPE2: if mediatype == HGTYPE2:
gen = genversion2(gen, engine, engineopts) gen = genversion2(gen, engine, engineopts)
wsgireq.respond(HTTP_OK, mediatype) setresponse(HTTP_OK, mediatype, bodygen=gen)
return gen
elif isinstance(rsp, wireprototypes.pushres): elif isinstance(rsp, wireprototypes.pushres):
rsp = '%d\n%s' % (rsp.res, rsp.output) rsp = '%d\n%s' % (rsp.res, rsp.output)
wsgireq.respond(HTTP_OK, HGTYPE, body=rsp) setresponse(HTTP_OK, HGTYPE, bodybytes=rsp)
return []
elif isinstance(rsp, wireprototypes.pusherr): elif isinstance(rsp, wireprototypes.pusherr):
rsp = '0\n%s\n' % rsp.res rsp = '0\n%s\n' % rsp.res
wsgireq.respond(HTTP_OK, HGTYPE, body=rsp) res.drain = True
return [] setresponse(HTTP_OK, HGTYPE, bodybytes=rsp)
elif isinstance(rsp, wireprototypes.ooberror): elif isinstance(rsp, wireprototypes.ooberror):
rsp = rsp.message setresponse(HTTP_OK, HGERRTYPE, bodybytes=rsp.message)
wsgireq.respond(HTTP_OK, HGERRTYPE, body=rsp) else:
return []
raise error.ProgrammingError('hgweb.protocol internal failure', rsp) raise error.ProgrammingError('hgweb.protocol internal failure', rsp)
def _handlehttperror(e, wsgireq, req):
"""Called when an ErrorResponse is raised during HTTP request processing."""
# TODO This response body assumes the failed command was
# "unbundle." That assumption is not always valid.
wsgireq.respond(e, HGTYPE, body='0\n%s\n' % pycompat.bytestr(e))
return ''
def _sshv1respondbytes(fout, value): def _sshv1respondbytes(fout, value):
"""Send a bytes response for protocol version 1.""" """Send a bytes response for protocol version 1."""
fout.write('%d\n' % len(value)) fout.write('%d\n' % len(value))
fout.write(value) fout.write(value)
fout.flush() fout.flush()
def _sshv1respondstream(fout, source): def _sshv1respondstream(fout, source):
write = fout.write write = fout.write