wireprotoserver.py
458 lines
| 14.9 KiB
| text/x-python
|
PythonLexer
/ mercurial / wireprotoserver.py
|
r35874 | # Copyright 21 May 2005 - (c) 2005 Jake Edge <jake@edge2.net> | ||
# Copyright 2005-2007 Matt Mackall <mpm@selenic.com> | ||||
# | ||||
# 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 | ||||
|
r35890 | import abc | ||
|
r36083 | import contextlib | ||
|
r35874 | import struct | ||
|
r35877 | import sys | ||
|
r35874 | |||
|
r35877 | from .i18n import _ | ||
|
r35874 | from . import ( | ||
|
r35877 | encoding, | ||
|
r35874 | error, | ||
|
r35877 | hook, | ||
|
r35874 | pycompat, | ||
util, | ||||
wireproto, | ||||
|
r36090 | wireprototypes, | ||
|
r35874 | ) | ||
stringio = util.stringio | ||||
urlerr = util.urlerr | ||||
urlreq = util.urlreq | ||||
|
r35876 | HTTP_OK = 200 | ||
|
r35874 | HGTYPE = 'application/mercurial-0.1' | ||
HGTYPE2 = 'application/mercurial-0.2' | ||||
HGERRTYPE = 'application/hg-error' | ||||
|
r35994 | # Names of the SSH protocol implementations. | ||
SSHV1 = 'ssh-v1' | ||||
# This is advertised over the wire. Incremental the counter at the end | ||||
# to reflect BC breakages. | ||||
SSHV2 = 'exp-ssh-v2-0001' | ||||
|
r36006 | class baseprotocolhandler(object): | ||
"""Abstract base class for wire protocol handlers. | ||||
|
r35878 | |||
|
r36006 | A wire protocol handler serves as an interface between protocol command | ||
handlers and the wire protocol transport layer. Protocol handlers provide | ||||
methods to read command arguments, redirect stdio for the duration of | ||||
the request, handle response types, etc. | ||||
|
r35878 | """ | ||
|
r35890 | __metaclass__ = abc.ABCMeta | ||
|
r35891 | @abc.abstractproperty | ||
def name(self): | ||||
"""The name of the protocol implementation. | ||||
Used for uniquely identifying the transport type. | ||||
""" | ||||
|
r35890 | @abc.abstractmethod | ||
|
r35878 | def getargs(self, args): | ||
"""return the value for arguments in <args> | ||||
returns a list of values (same order as <args>)""" | ||||
|
r35890 | @abc.abstractmethod | ||
|
r36087 | def forwardpayload(self, fp): | ||
"""Read the raw payload and forward to a file. | ||||
|
r35878 | |||
|
r36087 | The payload is read in full before the function returns. | ||
|
r35878 | """ | ||
|
r35890 | @abc.abstractmethod | ||
|
r36083 | def mayberedirectstdio(self): | ||
"""Context manager to possibly redirect stdio. | ||||
The context manager yields a file-object like object that receives | ||||
stdout and stderr output when the context manager is active. Or it | ||||
yields ``None`` if no I/O redirection occurs. | ||||
The intent of this context manager is to capture stdio output | ||||
so it may be sent in the response. Some transports support streaming | ||||
stdio to the client in real time. For these transports, stdio output | ||||
won't be captured. | ||||
""" | ||||
|
r36086 | @abc.abstractmethod | ||
def client(self): | ||||
"""Returns a string representation of this client (as bytes).""" | ||||
|
r35874 | def decodevaluefromheaders(req, headerprefix): | ||
"""Decode a long value from multiple HTTP request headers. | ||||
Returns the value as a bytes, not a str. | ||||
""" | ||||
chunks = [] | ||||
i = 1 | ||||
prefix = headerprefix.upper().replace(r'-', r'_') | ||||
while True: | ||||
v = req.env.get(r'HTTP_%s_%d' % (prefix, i)) | ||||
if v is None: | ||||
break | ||||
chunks.append(pycompat.bytesurl(v)) | ||||
i += 1 | ||||
return ''.join(chunks) | ||||
|
r36006 | class webproto(baseprotocolhandler): | ||
|
r35874 | def __init__(self, req, ui): | ||
|
r35884 | self._req = req | ||
self._ui = ui | ||||
|
r35891 | |||
@property | ||||
def name(self): | ||||
return 'http' | ||||
|
r35874 | |||
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] | ||||
|
r35881 | |||
|
r35874 | def _args(self): | ||
|
r35918 | args = util.rapply(pycompat.bytesurl, self._req.form.copy()) | ||
|
r35884 | postlen = int(self._req.env.get(r'HTTP_X_HGARGS_POST', 0)) | ||
|
r35874 | if postlen: | ||
|
r36094 | args.update(urlreq.parseqs( | ||
|
r35884 | self._req.read(postlen), keep_blank_values=True)) | ||
|
r35874 | return args | ||
|
r35884 | argvalue = decodevaluefromheaders(self._req, r'X-HgArg') | ||
|
r36094 | args.update(urlreq.parseqs(argvalue, keep_blank_values=True)) | ||
|
r35874 | return args | ||
|
r35881 | |||
|
r36087 | def forwardpayload(self, fp): | ||
|
r35884 | length = int(self._req.env[r'CONTENT_LENGTH']) | ||
|
r35874 | # If httppostargs is used, we need to read Content-Length | ||
# minus the amount that was consumed by args. | ||||
|
r35884 | length -= int(self._req.env.get(r'HTTP_X_HGARGS_POST', 0)) | ||
for s in util.filechunkiter(self._req, limit=length): | ||||
|
r35874 | fp.write(s) | ||
|
r35881 | |||
|
r36083 | @contextlib.contextmanager | ||
def mayberedirectstdio(self): | ||||
oldout = self._ui.fout | ||||
olderr = self._ui.ferr | ||||
out = util.stringio() | ||||
try: | ||||
self._ui.fout = out | ||||
self._ui.ferr = out | ||||
yield out | ||||
finally: | ||||
self._ui.fout = oldout | ||||
self._ui.ferr = olderr | ||||
|
r36086 | def client(self): | ||
|
r35874 | return 'remote:%s:%s:%s' % ( | ||
|
r35884 | 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', ''))) | ||||
|
r35874 | |||
def iscmd(cmd): | ||||
return cmd in wireproto.commands | ||||
|
r36002 | def parsehttprequest(repo, req, query): | ||
"""Parse the HTTP request for a wire protocol request. | ||||
If the current request appears to be a wire protocol request, this | ||||
function returns a dict with details about that request, including | ||||
an ``abstractprotocolserver`` instance suitable for handling the | ||||
request. Otherwise, ``None`` is returned. | ||||
``req`` is a ``wsgirequest`` instance. | ||||
""" | ||||
# 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 | ||||
# request. | ||||
if r'cmd' not in req.form: | ||||
return None | ||||
cmd = pycompat.sysbytes(req.form[r'cmd'][0]) | ||||
# The "cmd" request parameter is used by both the wire protocol and hgweb. | ||||
# 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 | ||||
# route it to a protocol handler. This is better than routing possible | ||||
# wire protocol requests to hgweb because it prevents hgweb from using | ||||
# known wire protocol commands and it is less confusing for machine | ||||
# clients. | ||||
if cmd not in wireproto.commands: | ||||
return None | ||||
|
r35882 | proto = webproto(req, repo.ui) | ||
|
r35874 | |||
|
r36002 | return { | ||
'cmd': cmd, | ||||
'proto': proto, | ||||
'dispatch': lambda: _callhttp(repo, req, proto, cmd), | ||||
|
r36004 | 'handleerror': lambda ex: _handlehttperror(ex, req, cmd), | ||
|
r36002 | } | ||
|
r36089 | def _httpresponsetype(ui, req, prefer_uncompressed): | ||
"""Determine the appropriate response type and compression settings. | ||||
Returns a tuple of (mediatype, compengine, engineopts). | ||||
""" | ||||
# Determine the response media type and compression engine based | ||||
# on the request parameters. | ||||
protocaps = decodevaluefromheaders(req, r'X-HgProto').split(' ') | ||||
if '0.2' in protocaps: | ||||
# All clients are expected to support uncompressed data. | ||||
if prefer_uncompressed: | ||||
return HGTYPE2, util._noopengine(), {} | ||||
# 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(ui, util.SERVERROLE): | ||||
if engine.wireprotosupport().name in compformats: | ||||
opts = {} | ||||
level = 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': ui.configint('server', 'zliblevel')} | ||||
return HGTYPE, util.compengines['zlib'], opts | ||||
|
r36002 | def _callhttp(repo, req, proto, cmd): | ||
|
r35874 | def genversion2(gen, 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 | ||||
for chunk in gen: | ||||
yield chunk | ||||
|
r35882 | rsp = wireproto.dispatch(repo, proto, cmd) | ||
|
r36000 | |||
if not wireproto.commands.commandavailable(cmd, proto): | ||||
req.respond(HTTP_OK, HGERRTYPE, | ||||
body=_('requested wire protocol command is not available ' | ||||
'over HTTP')) | ||||
return [] | ||||
|
r35874 | if isinstance(rsp, bytes): | ||
req.respond(HTTP_OK, HGTYPE, body=rsp) | ||||
return [] | ||||
|
r36091 | elif isinstance(rsp, wireprototypes.bytesresponse): | ||
req.respond(HTTP_OK, HGTYPE, body=rsp.data) | ||||
return [] | ||||
|
r36090 | elif isinstance(rsp, wireprototypes.streamreslegacy): | ||
|
r35874 | gen = rsp.gen | ||
req.respond(HTTP_OK, HGTYPE) | ||||
return gen | ||||
|
r36090 | elif isinstance(rsp, wireprototypes.streamres): | ||
|
r35874 | gen = rsp.gen | ||
# This code for compression should not be streamres specific. It | ||||
# is here because we only compress streamres at the moment. | ||||
|
r36089 | mediatype, engine, engineopts = _httpresponsetype( | ||
repo.ui, req, rsp.prefer_uncompressed) | ||||
|
r35874 | gen = engine.compressstream(gen, engineopts) | ||
if mediatype == HGTYPE2: | ||||
gen = genversion2(gen, engine, engineopts) | ||||
req.respond(HTTP_OK, mediatype) | ||||
return gen | ||||
|
r36090 | elif isinstance(rsp, wireprototypes.pushres): | ||
|
r36084 | rsp = '%d\n%s' % (rsp.res, rsp.output) | ||
|
r35874 | req.respond(HTTP_OK, HGTYPE, body=rsp) | ||
return [] | ||||
|
r36090 | elif isinstance(rsp, wireprototypes.pusherr): | ||
|
r36005 | # This is the httplib workaround documented in _handlehttperror(). | ||
|
r35874 | req.drain() | ||
|
r36005 | |||
|
r35874 | rsp = '0\n%s\n' % rsp.res | ||
req.respond(HTTP_OK, HGTYPE, body=rsp) | ||||
return [] | ||||
|
r36090 | elif isinstance(rsp, wireprototypes.ooberror): | ||
|
r35874 | rsp = rsp.message | ||
req.respond(HTTP_OK, HGERRTYPE, body=rsp) | ||||
return [] | ||||
raise error.ProgrammingError('hgweb.protocol internal failure', rsp) | ||||
|
r35877 | |||
|
r36004 | def _handlehttperror(e, req, cmd): | ||
"""Called when an ErrorResponse is raised during HTTP request processing.""" | ||||
|
r36005 | |||
# Clients using Python's httplib are stateful: the HTTP client | ||||
# won't process an HTTP response until all request data is | ||||
# sent to the server. The intent of this code is to ensure | ||||
# we always read HTTP request data from the client, thus | ||||
# ensuring httplib transitions to a state that allows it to read | ||||
# the HTTP response. In other words, it helps prevent deadlocks | ||||
# on clients using httplib. | ||||
if (req.env[r'REQUEST_METHOD'] == r'POST' and | ||||
# But not if Expect: 100-continue is being used. | ||||
|
r36004 | (req.env.get('HTTP_EXPECT', | ||
'').lower() != '100-continue') or | ||||
|
r36005 | # Or the non-httplib HTTP library is being advertised by | ||
# the client. | ||||
|
r36004 | req.env.get('X-HgHttp2', '')): | ||
req.drain() | ||||
else: | ||||
req.headers.append((r'Connection', r'Close')) | ||||
|
r36005 | # TODO This response body assumes the failed command was | ||
# "unbundle." That assumption is not always valid. | ||||
|
r36004 | req.respond(e, HGTYPE, body='0\n%s\n' % e) | ||
return '' | ||||
|
r36081 | def _sshv1respondbytes(fout, value): | ||
"""Send a bytes response for protocol version 1.""" | ||||
fout.write('%d\n' % len(value)) | ||||
fout.write(value) | ||||
fout.flush() | ||||
def _sshv1respondstream(fout, source): | ||||
write = fout.write | ||||
for chunk in source.gen: | ||||
write(chunk) | ||||
fout.flush() | ||||
def _sshv1respondooberror(fout, ferr, rsp): | ||||
ferr.write(b'%s\n-\n' % rsp) | ||||
ferr.flush() | ||||
fout.write(b'\n') | ||||
fout.flush() | ||||
|
r36082 | class sshv1protocolhandler(baseprotocolhandler): | ||
"""Handler for requests services via version 1 of SSH protocol.""" | ||||
def __init__(self, ui, fin, fout): | ||||
|
r35888 | self._ui = ui | ||
|
r36082 | self._fin = fin | ||
self._fout = fout | ||||
|
r35877 | |||
|
r35891 | @property | ||
def name(self): | ||||
|
r36092 | return SSHV1 | ||
|
r35891 | |||
|
r35877 | def getargs(self, args): | ||
data = {} | ||||
keys = args.split() | ||||
for n in xrange(len(keys)): | ||||
|
r35888 | argline = self._fin.readline()[:-1] | ||
|
r35877 | arg, l = argline.split() | ||
if arg not in keys: | ||||
raise error.Abort(_("unexpected parameter %r") % arg) | ||||
if arg == '*': | ||||
star = {} | ||||
for k in xrange(int(l)): | ||||
|
r35888 | argline = self._fin.readline()[:-1] | ||
|
r35877 | arg, l = argline.split() | ||
|
r35888 | val = self._fin.read(int(l)) | ||
|
r35877 | star[arg] = val | ||
data['*'] = star | ||||
else: | ||||
|
r35888 | val = self._fin.read(int(l)) | ||
|
r35877 | data[arg] = val | ||
return [data[k] for k in keys] | ||||
|
r36087 | def forwardpayload(self, fpout): | ||
# The file is in the form: | ||||
# | ||||
# <chunk size>\n<chunk> | ||||
# ... | ||||
# 0\n | ||||
|
r36081 | _sshv1respondbytes(self._fout, b'') | ||
|
r35888 | count = int(self._fin.readline()) | ||
|
r35877 | while count: | ||
|
r35888 | fpout.write(self._fin.read(count)) | ||
count = int(self._fin.readline()) | ||||
|
r35877 | |||
|
r36083 | @contextlib.contextmanager | ||
def mayberedirectstdio(self): | ||||
yield None | ||||
|
r36086 | def client(self): | ||
|
r36082 | client = encoding.environ.get('SSH_CLIENT', '').split(' ', 1)[0] | ||
return 'remote:ssh:' + client | ||||
class sshserver(object): | ||||
def __init__(self, ui, repo): | ||||
self._ui = ui | ||||
self._repo = repo | ||||
self._fin = ui.fin | ||||
self._fout = ui.fout | ||||
hook.redirect(True) | ||||
ui.fout = repo.ui.fout = ui.ferr | ||||
# Prevent insertion/deletion of CRs | ||||
util.setbinary(self._fin) | ||||
util.setbinary(self._fout) | ||||
self._proto = sshv1protocolhandler(self._ui, self._fin, self._fout) | ||||
|
r35877 | def serve_forever(self): | ||
|
r35886 | while self.serve_one(): | ||
pass | ||||
|
r35877 | sys.exit(0) | ||
def serve_one(self): | ||||
|
r35888 | cmd = self._fin.readline()[:-1] | ||
|
r36082 | if cmd and wireproto.commands.commandavailable(cmd, self._proto): | ||
rsp = wireproto.dispatch(self._repo, self._proto, cmd) | ||||
|
r36081 | |||
if isinstance(rsp, bytes): | ||||
_sshv1respondbytes(self._fout, rsp) | ||||
|
r36091 | elif isinstance(rsp, wireprototypes.bytesresponse): | ||
_sshv1respondbytes(self._fout, rsp.data) | ||||
|
r36090 | elif isinstance(rsp, wireprototypes.streamres): | ||
|
r36081 | _sshv1respondstream(self._fout, rsp) | ||
|
r36090 | elif isinstance(rsp, wireprototypes.streamreslegacy): | ||
|
r36081 | _sshv1respondstream(self._fout, rsp) | ||
|
r36090 | elif isinstance(rsp, wireprototypes.pushres): | ||
|
r36081 | _sshv1respondbytes(self._fout, b'') | ||
_sshv1respondbytes(self._fout, bytes(rsp.res)) | ||||
|
r36090 | elif isinstance(rsp, wireprototypes.pusherr): | ||
|
r36081 | _sshv1respondbytes(self._fout, rsp.res) | ||
|
r36090 | elif isinstance(rsp, wireprototypes.ooberror): | ||
|
r36081 | _sshv1respondooberror(self._fout, self._ui.ferr, rsp.message) | ||
else: | ||||
raise error.ProgrammingError('unhandled response type from ' | ||||
'wire protocol command: %s' % rsp) | ||||
|
r35877 | elif cmd: | ||
|
r36081 | _sshv1respondbytes(self._fout, b'') | ||
|
r35877 | return cmd != '' | ||