wireprotov2server.py
1613 lines
| 48.5 KiB
| text/x-python
|
PythonLexer
/ mercurial / wireprotov2server.py
|
r37563 | # Copyright 21 May 2005 - (c) 2005 Jake Edge <jake@edge2.net> | ||
|
r47575 | # Copyright 2005-2007 Olivia Mackall <olivia@selenic.com> | ||
|
r37563 | # | ||
# 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 | ||||
|
r40214 | import collections | ||
|
r37563 | import contextlib | ||
from .i18n import _ | ||||
|
r47771 | from .node import hex | ||
|
r37563 | from . import ( | ||
|
r39666 | discovery, | ||
|
r37564 | encoding, | ||
|
r37563 | error, | ||
|
r40214 | match as matchmod, | ||
|
r39836 | narrowspec, | ||
|
r37563 | pycompat, | ||
|
r40365 | streamclone, | ||
|
r41448 | templatefilters, | ||
|
r40365 | util, | ||
|
r37563 | wireprotoframing, | ||
wireprototypes, | ||||
) | ||||
|
r43346 | from .interfaces import util as interfaceutil | ||
|
r37828 | from .utils import ( | ||
|
r40057 | cborutil, | ||
|
r44517 | hashutil, | ||
|
r39869 | stringutil, | ||
|
r37828 | ) | ||
|
r37563 | |||
|
r40168 | FRAMINGTYPE = b'application/mercurial-exp-framing-0006' | ||
|
r37563 | |||
|
r37662 | HTTP_WIREPROTO_V2 = wireprototypes.HTTP_WIREPROTO_V2 | ||
|
r37563 | |||
|
r37802 | COMMANDS = wireprototypes.commanddict() | ||
|
r40057 | # Value inserted into cache key computation function. Change the value to | ||
# force new cache keys for every command request. This should be done when | ||||
# there is a change to how caching works, etc. | ||||
GLOBAL_CACHE_VERSION = 1 | ||||
|
r43346 | |||
|
r37563 | def handlehttpv2request(rctx, req, res, checkperm, urlparts): | ||
from .hgweb import common as hgwebcommon | ||||
# URL space looks like: <permissions>/<command>, where <permission> can | ||||
# be ``ro`` or ``rw`` to signal read-only or read-write, respectively. | ||||
# Root URL does nothing meaningful... yet. | ||||
if not urlparts: | ||||
res.status = b'200 OK' | ||||
res.headers[b'Content-Type'] = b'text/plain' | ||||
|
r43347 | res.setbodybytes(_(b'HTTP version 2 API handler')) | ||
|
r37563 | return | ||
if len(urlparts) == 1: | ||||
res.status = b'404 Not Found' | ||||
res.headers[b'Content-Type'] = b'text/plain' | ||||
|
r43346 | res.setbodybytes( | ||
|
r43347 | _(b'do not know how to process %s\n') % req.dispatchpath | ||
|
r43346 | ) | ||
|
r37563 | return | ||
permission, command = urlparts[0:2] | ||||
if permission not in (b'ro', b'rw'): | ||||
res.status = b'404 Not Found' | ||||
res.headers[b'Content-Type'] = b'text/plain' | ||||
|
r43347 | res.setbodybytes(_(b'unknown permission: %s') % permission) | ||
|
r37563 | return | ||
|
r43347 | if req.method != b'POST': | ||
|
r37563 | res.status = b'405 Method Not Allowed' | ||
res.headers[b'Allow'] = b'POST' | ||||
|
r43347 | res.setbodybytes(_(b'commands require POST requests')) | ||
|
r37563 | return | ||
# At some point we'll want to use our own API instead of recycling the | ||||
# behavior of version 1 of the wire protocol... | ||||
# TODO return reasonable responses - not responses that overload the | ||||
# HTTP status line message for error reporting. | ||||
try: | ||||
|
r43347 | checkperm(rctx, req, b'pull' if permission == b'ro' else b'push') | ||
|
r37563 | except hgwebcommon.ErrorResponse as e: | ||
|
r47524 | res.status = hgwebcommon.statusmessage( | ||
e.code, stringutil.forcebytestr(e) | ||||
) | ||||
|
r37563 | for k, v in e.headers: | ||
res.headers[k] = v | ||||
|
r43347 | res.setbodybytes(b'permission denied') | ||
|
r37563 | return | ||
# We have a special endpoint to reflect the request back at the client. | ||||
if command == b'debugreflect': | ||||
_processhttpv2reflectrequest(rctx.repo.ui, rctx.repo, req, res) | ||||
return | ||||
# Extra commands that we handle that aren't really wire protocol | ||||
# commands. Think extra hard before making this hackery available to | ||||
# extension. | ||||
|
r43347 | extracommands = {b'multirequest'} | ||
|
r37563 | |||
|
r37802 | if command not in COMMANDS and command not in extracommands: | ||
|
r37563 | res.status = b'404 Not Found' | ||
res.headers[b'Content-Type'] = b'text/plain' | ||||
|
r43347 | res.setbodybytes(_(b'unknown wire protocol command: %s\n') % command) | ||
|
r37563 | return | ||
repo = rctx.repo | ||||
ui = repo.ui | ||||
proto = httpv2protocolhandler(req, ui) | ||||
|
r43346 | if ( | ||
not COMMANDS.commandavailable(command, proto) | ||||
and command not in extracommands | ||||
): | ||||
|
r37563 | res.status = b'404 Not Found' | ||
res.headers[b'Content-Type'] = b'text/plain' | ||||
|
r43347 | res.setbodybytes(_(b'invalid wire protocol command: %s') % command) | ||
|
r37563 | return | ||
# TODO consider cases where proxies may add additional Accept headers. | ||||
if req.headers.get(b'Accept') != FRAMINGTYPE: | ||||
res.status = b'406 Not Acceptable' | ||||
res.headers[b'Content-Type'] = b'text/plain' | ||||
|
r43346 | res.setbodybytes( | ||
|
r43347 | _(b'client MUST specify Accept header with value: %s\n') | ||
|
r43346 | % FRAMINGTYPE | ||
) | ||||
|
r37563 | return | ||
if req.headers.get(b'Content-Type') != FRAMINGTYPE: | ||||
res.status = b'415 Unsupported Media Type' | ||||
# TODO we should send a response with appropriate media type, | ||||
# since client does Accept it. | ||||
res.headers[b'Content-Type'] = b'text/plain' | ||||
|
r43346 | res.setbodybytes( | ||
|
r43387 | _(b'client MUST send Content-Type header with value: %s\n') | ||
|
r43346 | % FRAMINGTYPE | ||
) | ||||
|
r37563 | return | ||
_processhttpv2request(ui, repo, req, res, permission, command, proto) | ||||
|
r43346 | |||
|
r37563 | def _processhttpv2reflectrequest(ui, repo, req, res): | ||
"""Reads unified frame protocol request and dumps out state to client. | ||||
This special endpoint can be used to help debug the wire protocol. | ||||
Instead of routing the request through the normal dispatch mechanism, | ||||
we instead read all frames, decode them, and feed them into our state | ||||
tracker. We then dump the log of all that activity back out to the | ||||
client. | ||||
""" | ||||
# Reflection APIs have a history of being abused, accidentally disclosing | ||||
# sensitive data, etc. So we have a config knob. | ||||
|
r43347 | if not ui.configbool(b'experimental', b'web.api.debugreflect'): | ||
|
r37563 | res.status = b'404 Not Found' | ||
res.headers[b'Content-Type'] = b'text/plain' | ||||
|
r43347 | res.setbodybytes(_(b'debugreflect service not available')) | ||
|
r37563 | return | ||
# We assume we have a unified framing protocol request body. | ||||
|
r40165 | reactor = wireprotoframing.serverreactor(ui) | ||
|
r37563 | states = [] | ||
while True: | ||||
frame = wireprotoframing.readframe(req.bodyfh) | ||||
if not frame: | ||||
states.append(b'received: <no frame>') | ||||
break | ||||
|
r43346 | states.append( | ||
b'received: %d %d %d %s' | ||||
% (frame.typeid, frame.flags, frame.requestid, frame.payload) | ||||
) | ||||
|
r37563 | |||
action, meta = reactor.onframerecv(frame) | ||||
|
r41448 | states.append(templatefilters.json((action, meta))) | ||
|
r37563 | |||
action, meta = reactor.oninputeof() | ||||
|
r43347 | meta[b'action'] = action | ||
|
r41448 | states.append(templatefilters.json(meta)) | ||
|
r37563 | |||
res.status = b'200 OK' | ||||
res.headers[b'Content-Type'] = b'text/plain' | ||||
res.setbodybytes(b'\n'.join(states)) | ||||
|
r43346 | |||
|
r37563 | def _processhttpv2request(ui, repo, req, res, authedperm, reqcommand, proto): | ||
"""Post-validation handler for HTTPv2 requests. | ||||
Called when the HTTP request contains unified frame-based protocol | ||||
frames for evaluation. | ||||
""" | ||||
# TODO Some HTTP clients are full duplex and can receive data before | ||||
# the entire request is transmitted. Figure out a way to indicate support | ||||
# for that so we can opt into full duplex mode. | ||||
|
r40165 | reactor = wireprotoframing.serverreactor(ui, deferoutput=True) | ||
|
r37563 | seencommand = False | ||
|
r40173 | outstream = None | ||
|
r37563 | |||
while True: | ||||
frame = wireprotoframing.readframe(req.bodyfh) | ||||
if not frame: | ||||
break | ||||
action, meta = reactor.onframerecv(frame) | ||||
|
r43347 | if action == b'wantframe': | ||
|
r37563 | # Need more data before we can do anything. | ||
continue | ||||
|
r43347 | elif action == b'runcommand': | ||
|
r40173 | # Defer creating output stream because we need to wait for | ||
# protocol settings frames so proper encoding can be applied. | ||||
if not outstream: | ||||
outstream = reactor.makeoutputstream() | ||||
|
r43346 | sentoutput = _httpv2runcommand( | ||
ui, | ||||
repo, | ||||
req, | ||||
res, | ||||
authedperm, | ||||
reqcommand, | ||||
reactor, | ||||
outstream, | ||||
meta, | ||||
issubsequent=seencommand, | ||||
) | ||||
|
r37563 | |||
if sentoutput: | ||||
return | ||||
seencommand = True | ||||
|
r43347 | elif action == b'error': | ||
|
r37563 | # TODO define proper error mechanism. | ||
res.status = b'200 OK' | ||||
res.headers[b'Content-Type'] = b'text/plain' | ||||
|
r43347 | res.setbodybytes(meta[b'message'] + b'\n') | ||
|
r37563 | return | ||
else: | ||||
raise error.ProgrammingError( | ||||
|
r43347 | b'unhandled action from frame processor: %s' % action | ||
|
r43346 | ) | ||
|
r37563 | |||
action, meta = reactor.oninputeof() | ||||
|
r43347 | if action == b'sendframes': | ||
|
r37563 | # We assume we haven't started sending the response yet. If we're | ||
# wrong, the response type will raise an exception. | ||||
res.status = b'200 OK' | ||||
res.headers[b'Content-Type'] = FRAMINGTYPE | ||||
|
r43347 | res.setbodygen(meta[b'framegen']) | ||
elif action == b'noop': | ||||
|
r37563 | pass | ||
else: | ||||
|
r43346 | raise error.ProgrammingError( | ||
|
r43347 | b'unhandled action from frame processor: %s' % action | ||
|
r43346 | ) | ||
|
r37563 | |||
|
r43346 | def _httpv2runcommand( | ||
ui, | ||||
repo, | ||||
req, | ||||
res, | ||||
authedperm, | ||||
reqcommand, | ||||
reactor, | ||||
outstream, | ||||
command, | ||||
issubsequent, | ||||
): | ||||
|
r37563 | """Dispatch a wire protocol command made from HTTPv2 requests. | ||
The authenticated permission (``authedperm``) along with the original | ||||
command from the URL (``reqcommand``) are passed in. | ||||
""" | ||||
# We already validated that the session has permissions to perform the | ||||
# actions in ``authedperm``. In the unified frame protocol, the canonical | ||||
# command to run is expressed in a frame. However, the URL also requested | ||||
# to run a specific command. We need to be careful that the command we | ||||
# run doesn't have permissions requirements greater than what was granted | ||||
# by ``authedperm``. | ||||
# | ||||
# Our rule for this is we only allow one command per HTTP request and | ||||
# that command must match the command in the URL. However, we make | ||||
# an exception for the ``multirequest`` URL. This URL is allowed to | ||||
# execute multiple commands. We double check permissions of each command | ||||
# as it is invoked to ensure there is no privilege escalation. | ||||
# TODO consider allowing multiple commands to regular command URLs | ||||
# iff each command is the same. | ||||
|
r43347 | proto = httpv2protocolhandler(req, ui, args=command[b'args']) | ||
|
r37563 | |||
if reqcommand == b'multirequest': | ||||
|
r43347 | if not COMMANDS.commandavailable(command[b'command'], proto): | ||
|
r37563 | # TODO proper error mechanism | ||
res.status = b'200 OK' | ||||
res.headers[b'Content-Type'] = b'text/plain' | ||||
|
r43346 | res.setbodybytes( | ||
|
r43347 | _(b'wire protocol command not available: %s') | ||
% command[b'command'] | ||||
|
r43346 | ) | ||
|
r37563 | return True | ||
# TODO don't use assert here, since it may be elided by -O. | ||||
assert authedperm in (b'ro', b'rw') | ||||
|
r43347 | wirecommand = COMMANDS[command[b'command']] | ||
assert wirecommand.permission in (b'push', b'pull') | ||||
|
r37563 | |||
|
r43347 | if authedperm == b'ro' and wirecommand.permission != b'pull': | ||
|
r37563 | # TODO proper error mechanism | ||
res.status = b'403 Forbidden' | ||||
res.headers[b'Content-Type'] = b'text/plain' | ||||
|
r43346 | res.setbodybytes( | ||
|
r43387 | _(b'insufficient permissions to execute command: %s') | ||
|
r43347 | % command[b'command'] | ||
|
r43346 | ) | ||
|
r37563 | return True | ||
# TODO should we also call checkperm() here? Maybe not if we're going | ||||
# to overhaul that API. The granted scope from the URL check should | ||||
# be good enough. | ||||
else: | ||||
# Don't allow multiple commands outside of ``multirequest`` URL. | ||||
if issubsequent: | ||||
# TODO proper error mechanism | ||||
res.status = b'200 OK' | ||||
res.headers[b'Content-Type'] = b'text/plain' | ||||
|
r43346 | res.setbodybytes( | ||
|
r43387 | _(b'multiple commands cannot be issued to this URL') | ||
|
r43346 | ) | ||
|
r37563 | return True | ||
|
r43347 | if reqcommand != command[b'command']: | ||
|
r37563 | # TODO define proper error mechanism | ||
res.status = b'200 OK' | ||||
res.headers[b'Content-Type'] = b'text/plain' | ||||
|
r43347 | res.setbodybytes(_(b'command in frame must match command in URL')) | ||
|
r37563 | return True | ||
res.status = b'200 OK' | ||||
res.headers[b'Content-Type'] = FRAMINGTYPE | ||||
|
r39595 | try: | ||
|
r43347 | objs = dispatch(repo, proto, command[b'command'], command[b'redirect']) | ||
|
r39595 | |||
action, meta = reactor.oncommandresponsereadyobjects( | ||||
|
r43347 | outstream, command[b'requestid'], objs | ||
|
r43346 | ) | ||
|
r39595 | |||
|
r39835 | except error.WireprotoCommandError as e: | ||
action, meta = reactor.oncommanderror( | ||||
|
r43347 | outstream, command[b'requestid'], e.message, e.messageargs | ||
|
r43346 | ) | ||
|
r39835 | |||
|
r39595 | except Exception as e: | ||
|
r37744 | action, meta = reactor.onservererror( | ||
|
r43346 | outstream, | ||
|
r43347 | command[b'requestid'], | ||
_(b'exception when invoking command: %s') | ||||
|
r43346 | % stringutil.forcebytestr(e), | ||
) | ||||
|
r37563 | |||
|
r43347 | if action == b'sendframes': | ||
res.setbodygen(meta[b'framegen']) | ||||
|
r37563 | return True | ||
|
r43347 | elif action == b'noop': | ||
|
r37563 | return False | ||
else: | ||||
|
r43346 | raise error.ProgrammingError( | ||
|
r43347 | b'unhandled event from reactor: %s' % action | ||
|
r43346 | ) | ||
|
r37563 | |||
|
r37800 | def getdispatchrepo(repo, proto, command): | ||
|
r43347 | viewconfig = repo.ui.config(b'server', b'view') | ||
|
r42006 | return repo.filtered(viewconfig) | ||
|
r37800 | |||
|
r43346 | |||
|
r40061 | def dispatch(repo, proto, command, redirect): | ||
|
r40057 | """Run a wire protocol command. | ||
Returns an iterable of objects that will be sent to the client. | ||||
""" | ||||
|
r37800 | repo = getdispatchrepo(repo, proto, command) | ||
|
r40057 | entry = COMMANDS[command] | ||
func = entry.func | ||||
spec = entry.args | ||||
|
r37800 | args = proto.getargs(spec) | ||
|
r40057 | # There is some duplicate boilerplate code here for calling the command and | ||
# emitting objects. It is either that or a lot of indented code that looks | ||||
# like a pyramid (since there are a lot of code paths that result in not | ||||
# using the cacher). | ||||
callcommand = lambda: func(repo, proto, **pycompat.strkwargs(args)) | ||||
# Request is not cacheable. Don't bother instantiating a cacher. | ||||
if not entry.cachekeyfn: | ||||
for o in callcommand(): | ||||
yield o | ||||
return | ||||
|
r40061 | if redirect: | ||
redirecttargets = redirect[b'targets'] | ||||
redirecthashes = redirect[b'hashes'] | ||||
else: | ||||
redirecttargets = [] | ||||
redirecthashes = [] | ||||
|
r43346 | cacher = makeresponsecacher( | ||
repo, | ||||
proto, | ||||
command, | ||||
args, | ||||
cborutil.streamencode, | ||||
redirecttargets=redirecttargets, | ||||
redirecthashes=redirecthashes, | ||||
) | ||||
|
r40057 | |||
# But we have no cacher. Do default handling. | ||||
if not cacher: | ||||
for o in callcommand(): | ||||
yield o | ||||
return | ||||
with cacher: | ||||
|
r43346 | cachekey = entry.cachekeyfn( | ||
repo, proto, cacher, **pycompat.strkwargs(args) | ||||
) | ||||
|
r40057 | |||
# No cache key or the cacher doesn't like it. Do default handling. | ||||
if cachekey is None or not cacher.setcachekey(cachekey): | ||||
for o in callcommand(): | ||||
yield o | ||||
return | ||||
# Serve it from the cache, if possible. | ||||
cached = cacher.lookup() | ||||
if cached: | ||||
|
r43347 | for o in cached[b'objs']: | ||
|
r40057 | yield o | ||
return | ||||
# Else call the command and feed its output into the cacher, allowing | ||||
# the cacher to buffer/mutate objects as it desires. | ||||
for o in callcommand(): | ||||
for o in cacher.onobject(o): | ||||
yield o | ||||
for o in cacher.onfinished(): | ||||
yield o | ||||
|
r37800 | |||
|
r43346 | |||
|
r37828 | @interfaceutil.implementer(wireprototypes.baseprotocolhandler) | ||
|
r37563 | class httpv2protocolhandler(object): | ||
def __init__(self, req, ui, args=None): | ||||
self._req = req | ||||
self._ui = ui | ||||
self._args = args | ||||
@property | ||||
def name(self): | ||||
|
r37662 | return HTTP_WIREPROTO_V2 | ||
|
r37563 | |||
def getargs(self, args): | ||||
|
r39835 | # First look for args that were passed but aren't registered on this | ||
# command. | ||||
extra = set(self._args) - set(args) | ||||
if extra: | ||||
raise error.WireprotoCommandError( | ||||
|
r43347 | b'unsupported argument to command: %s' | ||
% b', '.join(sorted(extra)) | ||||
|
r43346 | ) | ||
|
r39835 | |||
# And look for required arguments that are missing. | ||||
|
r43347 | missing = {a for a in args if args[a][b'required']} - set(self._args) | ||
|
r39835 | |||
if missing: | ||||
raise error.WireprotoCommandError( | ||||
|
r43347 | b'missing required arguments: %s' % b', '.join(sorted(missing)) | ||
|
r43346 | ) | ||
|
r39835 | |||
# Now derive the arguments to pass to the command, taking into | ||||
# account the arguments specified by the client. | ||||
|
r37563 | data = {} | ||
|
r39835 | for k, meta in sorted(args.items()): | ||
# This argument wasn't passed by the client. | ||||
if k not in self._args: | ||||
|
r43347 | data[k] = meta[b'default']() | ||
|
r39835 | continue | ||
v = self._args[k] | ||||
# Sets may be expressed as lists. Silently normalize. | ||||
|
r43347 | if meta[b'type'] == b'set' and isinstance(v, list): | ||
|
r39835 | v = set(v) | ||
# TODO consider more/stronger type validation. | ||||
data[k] = v | ||||
|
r37563 | |||
return data | ||||
def getprotocaps(self): | ||||
# Protocol capabilities are currently not implemented for HTTP V2. | ||||
return set() | ||||
def getpayload(self): | ||||
raise NotImplementedError | ||||
@contextlib.contextmanager | ||||
def mayberedirectstdio(self): | ||||
raise NotImplementedError | ||||
def client(self): | ||||
raise NotImplementedError | ||||
def addcapabilities(self, repo, caps): | ||||
return caps | ||||
def checkperm(self, perm): | ||||
raise NotImplementedError | ||||
|
r37564 | |||
|
r43346 | |||
|
r37575 | def httpv2apidescriptor(req, repo): | ||
proto = httpv2protocolhandler(req, repo.ui) | ||||
return _capabilitiesv2(repo, proto) | ||||
|
r43346 | |||
|
r37564 | def _capabilitiesv2(repo, proto): | ||
"""Obtain the set of capabilities for version 2 transports. | ||||
These capabilities are distinct from the capabilities for version 1 | ||||
transports. | ||||
""" | ||||
caps = { | ||||
|
r43347 | b'commands': {}, | ||
b'framingmediatypes': [FRAMINGTYPE], | ||||
b'pathfilterprefixes': set(narrowspec.VALID_PREFIXES), | ||||
|
r37564 | } | ||
|
r37802 | for command, entry in COMMANDS.items(): | ||
|
r39837 | args = {} | ||
for arg, meta in entry.args.items(): | ||||
args[arg] = { | ||||
# TODO should this be a normalized type using CBOR's | ||||
# terminology? | ||||
|
r43347 | b'type': meta[b'type'], | ||
b'required': meta[b'required'], | ||||
|
r39837 | } | ||
|
r43347 | if not meta[b'required']: | ||
args[arg][b'default'] = meta[b'default']() | ||||
|
r39835 | |||
|
r43347 | if meta[b'validvalues']: | ||
args[arg][b'validvalues'] = meta[b'validvalues'] | ||||
|
r39838 | |||
|
r40365 | # TODO this type of check should be defined in a per-command callback. | ||
|
r43346 | if ( | ||
command == b'rawstorefiledata' | ||||
and not streamclone.allowservergeneration(repo) | ||||
): | ||||
|
r40365 | continue | ||
|
r43347 | caps[b'commands'][command] = { | ||
b'args': args, | ||||
b'permissions': [entry.permission], | ||||
|
r37564 | } | ||
|
r40208 | if entry.extracapabilitiesfn: | ||
extracaps = entry.extracapabilitiesfn(repo, proto) | ||||
|
r43347 | caps[b'commands'][command].update(extracaps) | ||
|
r40208 | |||
|
r43347 | caps[b'rawrepoformats'] = sorted(repo.requirements & repo.supportedformats) | ||
|
r37675 | |||
|
r40059 | targets = getadvertisedredirecttargets(repo, proto) | ||
if targets: | ||||
caps[b'redirect'] = { | ||||
b'targets': [], | ||||
b'hashes': [b'sha256', b'sha1'], | ||||
} | ||||
for target in targets: | ||||
entry = { | ||||
|
r43347 | b'name': target[b'name'], | ||
b'protocol': target[b'protocol'], | ||||
b'uris': target[b'uris'], | ||||
|
r40059 | } | ||
|
r43347 | for key in (b'snirequired', b'tlsversions'): | ||
|
r40059 | if key in target: | ||
entry[key] = target[key] | ||||
caps[b'redirect'][b'targets'].append(entry) | ||||
|
r37564 | return proto.addcapabilities(repo, caps) | ||
|
r43346 | |||
|
r40059 | def getadvertisedredirecttargets(repo, proto): | ||
"""Obtain a list of content redirect targets. | ||||
Returns a list containing potential redirect targets that will be | ||||
advertised in capabilities data. Each dict MUST have the following | ||||
keys: | ||||
name | ||||
The name of this redirect target. This is the identifier clients use | ||||
to refer to a target. It is transferred as part of every command | ||||
request. | ||||
protocol | ||||
Network protocol used by this target. Typically this is the string | ||||
in front of the ``://`` in a URL. e.g. ``https``. | ||||
uris | ||||
List of representative URIs for this target. Clients can use the | ||||
URIs to test parsing for compatibility or for ordering preference | ||||
for which target to use. | ||||
The following optional keys are recognized: | ||||
snirequired | ||||
Bool indicating if Server Name Indication (SNI) is required to | ||||
connect to this target. | ||||
tlsversions | ||||
List of bytes indicating which TLS versions are supported by this | ||||
target. | ||||
By default, clients reflect the target order advertised by servers | ||||
and servers will use the first client-advertised target when picking | ||||
a redirect target. So targets should be advertised in the order the | ||||
server prefers they be used. | ||||
""" | ||||
return [] | ||||
|
r43346 | |||
def wireprotocommand( | ||||
name, | ||||
args=None, | ||||
|
r43347 | permission=b'push', | ||
|
r43346 | cachekeyfn=None, | ||
extracapabilitiesfn=None, | ||||
): | ||||
|
r37798 | """Decorator to declare a wire protocol command. | ||
``name`` is the name of the wire protocol command being provided. | ||||
|
r39835 | ``args`` is a dict defining arguments accepted by the command. Keys are | ||
the argument name. Values are dicts with the following keys: | ||||
``type`` | ||||
The argument data type. Must be one of the following string | ||||
literals: ``bytes``, ``int``, ``list``, ``dict``, ``set``, | ||||
or ``bool``. | ||||
``default`` | ||||
A callable returning the default value for this argument. If not | ||||
specified, ``None`` will be the default value. | ||||
``example`` | ||||
An example value for this argument. | ||||
|
r37798 | |||
|
r39838 | ``validvalues`` | ||
Set of recognized values for this argument. | ||||
|
r37798 | ``permission`` defines the permission type needed to run this command. | ||
Can be ``push`` or ``pull``. These roughly map to read-write and read-only, | ||||
respectively. Default is to assume command requires ``push`` permissions | ||||
because otherwise commands not declaring their permissions could modify | ||||
a repository that is supposed to be read-only. | ||||
|
r39595 | |||
|
r40057 | ``cachekeyfn`` defines an optional callable that can derive the | ||
cache key for this request. | ||||
|
r40208 | ``extracapabilitiesfn`` defines an optional callable that defines extra | ||
command capabilities/parameters that are advertised next to the command | ||||
in the capabilities data structure describing the server. The callable | ||||
receives as arguments the repository and protocol objects. It returns | ||||
a dict of extra fields to add to the command descriptor. | ||||
|
r39595 | Wire protocol commands are generators of objects to be serialized and | ||
sent to the client. | ||||
If a command raises an uncaught exception, this will be translated into | ||||
a command error. | ||||
|
r40057 | |||
All commands can opt in to being cacheable by defining a function | ||||
(``cachekeyfn``) that is called to derive a cache key. This function | ||||
receives the same arguments as the command itself plus a ``cacher`` | ||||
argument containing the active cacher for the request and returns a bytes | ||||
containing the key in a cache the response to this command may be cached | ||||
under. | ||||
|
r37798 | """ | ||
|
r43346 | transports = { | ||
|
r43347 | k for k, v in wireprototypes.TRANSPORTS.items() if v[b'version'] == 2 | ||
|
r43346 | } | ||
|
r37798 | |||
|
r43347 | if permission not in (b'push', b'pull'): | ||
|
r43346 | raise error.ProgrammingError( | ||
|
r43347 | b'invalid wire protocol permission; ' | ||
b'got %s; expected "push" or "pull"' % permission | ||||
|
r43346 | ) | ||
|
r37798 | |||
if args is None: | ||||
args = {} | ||||
if not isinstance(args, dict): | ||||
|
r43346 | raise error.ProgrammingError( | ||
|
r43387 | b'arguments for version 2 commands must be declared as dicts' | ||
|
r43346 | ) | ||
|
r37798 | |||
|
r39835 | for arg, meta in args.items(): | ||
|
r43347 | if arg == b'*': | ||
|
r43346 | raise error.ProgrammingError( | ||
|
r43387 | b'* argument name not allowed on version 2 commands' | ||
|
r43346 | ) | ||
|
r39835 | |||
if not isinstance(meta, dict): | ||||
|
r43346 | raise error.ProgrammingError( | ||
|
r43347 | b'arguments for version 2 commands ' | ||
b'must declare metadata as a dict' | ||||
|
r43346 | ) | ||
|
r39835 | |||
|
r43347 | if b'type' not in meta: | ||
|
r43346 | raise error.ProgrammingError( | ||
|
r43347 | b'%s argument for command %s does not ' | ||
b'declare type field' % (arg, name) | ||||
|
r43346 | ) | ||
|
r39835 | |||
|
r43347 | if meta[b'type'] not in ( | ||
b'bytes', | ||||
b'int', | ||||
b'list', | ||||
b'dict', | ||||
b'set', | ||||
b'bool', | ||||
): | ||||
|
r43346 | raise error.ProgrammingError( | ||
|
r43347 | b'%s argument for command %s has ' | ||
b'illegal type: %s' % (arg, name, meta[b'type']) | ||||
|
r43346 | ) | ||
|
r39835 | |||
|
r43347 | if b'example' not in meta: | ||
|
r43346 | raise error.ProgrammingError( | ||
|
r43347 | b'%s argument for command %s does not ' | ||
b'declare example field' % (arg, name) | ||||
|
r43346 | ) | ||
|
r39835 | |||
|
r43347 | meta[b'required'] = b'default' not in meta | ||
|
r39835 | |||
|
r43347 | meta.setdefault(b'default', lambda: None) | ||
meta.setdefault(b'validvalues', None) | ||||
|
r39835 | |||
|
r37564 | def register(func): | ||
|
r37802 | if name in COMMANDS: | ||
|
r43346 | raise error.ProgrammingError( | ||
|
r43387 | b'%s command already registered for version 2' % name | ||
|
r43346 | ) | ||
|
r37798 | |||
|
r37802 | COMMANDS[name] = wireprototypes.commandentry( | ||
|
r43346 | func, | ||
args=args, | ||||
transports=transports, | ||||
permission=permission, | ||||
cachekeyfn=cachekeyfn, | ||||
extracapabilitiesfn=extracapabilitiesfn, | ||||
) | ||||
|
r37798 | |||
return func | ||||
|
r37564 | |||
return register | ||||
|
r43346 | |||
|
r40057 | def makecommandcachekeyfn(command, localversion=None, allargs=False): | ||
"""Construct a cache key derivation function with common features. | ||||
By default, the cache key is a hash of: | ||||
* The command name. | ||||
* A global cache version number. | ||||
* A local cache version number (passed via ``localversion``). | ||||
* All the arguments passed to the command. | ||||
* The media type used. | ||||
* Wire protocol version string. | ||||
* The repository path. | ||||
""" | ||||
if not allargs: | ||||
|
r43347 | raise error.ProgrammingError( | ||
b'only allargs=True is currently supported' | ||||
) | ||||
|
r40057 | |||
if localversion is None: | ||||
|
r43347 | raise error.ProgrammingError(b'must set localversion argument value') | ||
|
r40057 | |||
def cachekeyfn(repo, proto, cacher, **args): | ||||
spec = COMMANDS[command] | ||||
# Commands that mutate the repo can not be cached. | ||||
|
r43347 | if spec.permission == b'push': | ||
|
r40057 | return None | ||
# TODO config option to disable caching. | ||||
# Our key derivation strategy is to construct a data structure | ||||
# holding everything that could influence cacheability and to hash | ||||
# the CBOR representation of that. Using CBOR seems like it might | ||||
# be overkill. However, simpler hashing mechanisms are prone to | ||||
# duplicate input issues. e.g. if you just concatenate two values, | ||||
# "foo"+"bar" is identical to "fo"+"obar". Using CBOR provides | ||||
# "padding" between values and prevents these problems. | ||||
# Seed the hash with various data. | ||||
state = { | ||||
# To invalidate all cache keys. | ||||
b'globalversion': GLOBAL_CACHE_VERSION, | ||||
# More granular cache key invalidation. | ||||
b'localversion': localversion, | ||||
# Cache keys are segmented by command. | ||||
|
r41422 | b'command': command, | ||
|
r40057 | # Throw in the media type and API version strings so changes | ||
# to exchange semantics invalid cache. | ||||
b'mediatype': FRAMINGTYPE, | ||||
b'version': HTTP_WIREPROTO_V2, | ||||
# So same requests for different repos don't share cache keys. | ||||
b'repo': repo.root, | ||||
} | ||||
# The arguments passed to us will have already been normalized. | ||||
# Default values will be set, etc. This is important because it | ||||
# means that it doesn't matter if clients send an explicit argument | ||||
# or rely on the default value: it will all normalize to the same | ||||
# set of arguments on the server and therefore the same cache key. | ||||
# | ||||
# Arguments by their very nature must support being encoded to CBOR. | ||||
# And the CBOR encoder is deterministic. So we hash the arguments | ||||
# by feeding the CBOR of their representation into the hasher. | ||||
if allargs: | ||||
state[b'args'] = pycompat.byteskwargs(args) | ||||
cacher.adjustcachekeystate(state) | ||||
|
r44517 | hasher = hashutil.sha1() | ||
|
r40057 | for chunk in cborutil.streamencode(state): | ||
hasher.update(chunk) | ||||
return pycompat.sysbytes(hasher.hexdigest()) | ||||
return cachekeyfn | ||||
|
r43346 | |||
def makeresponsecacher( | ||||
repo, proto, command, args, objencoderfn, redirecttargets, redirecthashes | ||||
): | ||||
|
r40057 | """Construct a cacher for a cacheable command. | ||
Returns an ``iwireprotocolcommandcacher`` instance. | ||||
Extensions can monkeypatch this function to provide custom caching | ||||
backends. | ||||
""" | ||||
return None | ||||
|
r43346 | |||
|
r40212 | def resolvenodes(repo, revisions): | ||
"""Resolve nodes from a revisions specifier data structure.""" | ||||
cl = repo.changelog | ||||
clhasnode = cl.hasnode | ||||
seen = set() | ||||
nodes = [] | ||||
if not isinstance(revisions, list): | ||||
|
r43346 | raise error.WireprotoCommandError( | ||
|
r43387 | b'revisions must be defined as an array' | ||
|
r43346 | ) | ||
|
r40212 | |||
for spec in revisions: | ||||
if b'type' not in spec: | ||||
raise error.WireprotoCommandError( | ||||
|
r43347 | b'type key not present in revision specifier' | ||
|
r43346 | ) | ||
|
r40212 | |||
typ = spec[b'type'] | ||||
if typ == b'changesetexplicit': | ||||
if b'nodes' not in spec: | ||||
raise error.WireprotoCommandError( | ||||
|
r43347 | b'nodes key not present in changesetexplicit revision ' | ||
b'specifier' | ||||
|
r43346 | ) | ||
|
r40212 | |||
for node in spec[b'nodes']: | ||||
if node not in seen: | ||||
nodes.append(node) | ||||
seen.add(node) | ||||
elif typ == b'changesetexplicitdepth': | ||||
for key in (b'nodes', b'depth'): | ||||
if key not in spec: | ||||
raise error.WireprotoCommandError( | ||||
|
r43347 | b'%s key not present in changesetexplicitdepth revision ' | ||
b'specifier', | ||||
|
r43346 | (key,), | ||
) | ||||
|
r40212 | |||
|
r43346 | for rev in repo.revs( | ||
b'ancestors(%ln, %s)', spec[b'nodes'], spec[b'depth'] - 1 | ||||
): | ||||
|
r40212 | node = cl.node(rev) | ||
if node not in seen: | ||||
nodes.append(node) | ||||
seen.add(node) | ||||
elif typ == b'changesetdagrange': | ||||
for key in (b'roots', b'heads'): | ||||
if key not in spec: | ||||
raise error.WireprotoCommandError( | ||||
|
r43347 | b'%s key not present in changesetdagrange revision ' | ||
b'specifier', | ||||
|
r43346 | (key,), | ||
) | ||||
|
r40212 | |||
if not spec[b'heads']: | ||||
raise error.WireprotoCommandError( | ||||
|
r43347 | b'heads key in changesetdagrange cannot be empty' | ||
|
r43346 | ) | ||
|
r40212 | |||
if spec[b'roots']: | ||||
common = [n for n in spec[b'roots'] if clhasnode(n)] | ||||
else: | ||||
|
r47771 | common = [repo.nullid] | ||
|
r40212 | |||
for n in discovery.outgoing(repo, common, spec[b'heads']).missing: | ||||
if n not in seen: | ||||
nodes.append(n) | ||||
seen.add(n) | ||||
else: | ||||
raise error.WireprotoCommandError( | ||||
|
r43347 | b'unknown revision specifier type: %s', (typ,) | ||
|
r43346 | ) | ||
|
r40212 | |||
return nodes | ||||
|
r43346 | |||
|
r43347 | @wireprotocommand(b'branchmap', permission=b'pull') | ||
|
r37564 | def branchmapv2(repo, proto): | ||
|
r43376 | yield { | ||
encoding.fromlocal(k): v | ||||
for k, v in pycompat.iteritems(repo.branchmap()) | ||||
} | ||||
|
r43346 | |||
|
r37564 | |||
|
r43347 | @wireprotocommand(b'capabilities', permission=b'pull') | ||
|
r37564 | def capabilitiesv2(repo, proto): | ||
|
r39595 | yield _capabilitiesv2(repo, proto) | ||
|
r37564 | |||
|
r43346 | |||
|
r39835 | @wireprotocommand( | ||
|
r43347 | b'changesetdata', | ||
|
r39835 | args={ | ||
|
r43347 | b'revisions': { | ||
b'type': b'list', | ||||
b'example': [ | ||||
|
r46554 | { | ||
b'type': b'changesetexplicit', | ||||
b'nodes': [b'abcdef...'], | ||||
} | ||||
|
r43346 | ], | ||
|
r39840 | }, | ||
|
r43347 | b'fields': { | ||
b'type': b'set', | ||||
b'default': set, | ||||
b'example': {b'parents', b'revision'}, | ||||
b'validvalues': {b'bookmarks', b'parents', b'phase', b'revision'}, | ||||
|
r39835 | }, | ||
}, | ||||
|
r43347 | permission=b'pull', | ||
|
r43346 | ) | ||
|
r40212 | def changesetdata(repo, proto, revisions, fields): | ||
|
r39672 | # TODO look for unknown fields and abort when they can't be serviced. | ||
|
r39838 | # This could probably be validated by dispatcher using validvalues. | ||
|
r39672 | |||
|
r39666 | cl = repo.changelog | ||
|
r40212 | outgoing = resolvenodes(repo, revisions) | ||
|
r39668 | publishing = repo.publishing() | ||
|
r39666 | |||
if outgoing: | ||||
|
r43347 | repo.hook(b'preoutgoing', throw=True, source=b'serve') | ||
|
r39666 | |||
yield { | ||||
b'totalitems': len(outgoing), | ||||
} | ||||
|
r39668 | # The phases of nodes already transferred to the client may have changed | ||
# since the client last requested data. We send phase-only records | ||||
# for these revisions, if requested. | ||||
|
r40211 | # TODO actually do this. We'll probably want to emit phase heads | ||
# in the ancestry set of the outgoing revisions. This will ensure | ||||
# that phase updates within that set are seen. | ||||
if b'phase' in fields: | ||||
pass | ||||
|
r39668 | |||
|
r39670 | nodebookmarks = {} | ||
for mark, node in repo._bookmarks.items(): | ||||
nodebookmarks.setdefault(node, set()).add(mark) | ||||
|
r39666 | # It is already topologically sorted by revision number. | ||
for node in outgoing: | ||||
d = { | ||||
b'node': node, | ||||
} | ||||
if b'parents' in fields: | ||||
d[b'parents'] = cl.parents(node) | ||||
|
r39668 | if b'phase' in fields: | ||
if publishing: | ||||
d[b'phase'] = b'public' | ||||
else: | ||||
ctx = repo[node] | ||||
d[b'phase'] = ctx.phasestr() | ||||
|
r39670 | if b'bookmarks' in fields and node in nodebookmarks: | ||
d[b'bookmarks'] = sorted(nodebookmarks[node]) | ||||
del nodebookmarks[node] | ||||
|
r39839 | followingmeta = [] | ||
followingdata = [] | ||||
|
r39666 | |||
if b'revision' in fields: | ||||
|
r46679 | revisiondata = cl.revision(node) | ||
|
r39839 | followingmeta.append((b'revision', len(revisiondata))) | ||
followingdata.append(revisiondata) | ||||
|
r39666 | |||
|
r39672 | # TODO make it possible for extensions to wrap a function or register | ||
# a handler to service custom fields. | ||||
|
r39839 | if followingmeta: | ||
d[b'fieldsfollowing'] = followingmeta | ||||
|
r39666 | yield d | ||
|
r39839 | for extra in followingdata: | ||
yield extra | ||||
|
r39666 | |||
|
r39670 | # If requested, send bookmarks from nodes that didn't have revision | ||
# data sent so receiver is aware of any bookmark updates. | ||||
if b'bookmarks' in fields: | ||||
|
r43376 | for node, marks in sorted(pycompat.iteritems(nodebookmarks)): | ||
|
r39670 | yield { | ||
b'node': node, | ||||
b'bookmarks': sorted(marks), | ||||
} | ||||
|
r43346 | |||
|
r39675 | class FileAccessError(Exception): | ||
"""Represents an error accessing a specific file.""" | ||||
def __init__(self, path, msg, args): | ||||
self.path = path | ||||
self.msg = msg | ||||
self.args = args | ||||
|
r43346 | |||
|
r39675 | def getfilestore(repo, proto, path): | ||
"""Obtain a file storage object for use with wire protocol. | ||||
Exists as a standalone function so extensions can monkeypatch to add | ||||
access control. | ||||
""" | ||||
# This seems to work even if the file doesn't exist. So catch | ||||
# "empty" files and return an error. | ||||
fl = repo.file(path) | ||||
if not len(fl): | ||||
|
r43347 | raise FileAccessError(path, b'unknown file: %s', (path,)) | ||
|
r39675 | |||
return fl | ||||
|
r43346 | |||
|
r40959 | def emitfilerevisions(repo, path, revisions, linknodes, fields): | ||
|
r40213 | for revision in revisions: | ||
d = { | ||||
b'node': revision.node, | ||||
} | ||||
if b'parents' in fields: | ||||
d[b'parents'] = [revision.p1node, revision.p2node] | ||||
|
r40427 | if b'linknode' in fields: | ||
|
r40959 | d[b'linknode'] = linknodes[revision.node] | ||
|
r40427 | |||
|
r40213 | followingmeta = [] | ||
followingdata = [] | ||||
if b'revision' in fields: | ||||
if revision.revision is not None: | ||||
followingmeta.append((b'revision', len(revision.revision))) | ||||
followingdata.append(revision.revision) | ||||
else: | ||||
d[b'deltabasenode'] = revision.basenode | ||||
followingmeta.append((b'delta', len(revision.delta))) | ||||
followingdata.append(revision.delta) | ||||
if followingmeta: | ||||
d[b'fieldsfollowing'] = followingmeta | ||||
yield d | ||||
for extra in followingdata: | ||||
yield extra | ||||
|
r43346 | |||
|
r40214 | def makefilematcher(repo, pathfilter): | ||
"""Construct a matcher from a path filter dict.""" | ||||
# Validate values. | ||||
if pathfilter: | ||||
for key in (b'include', b'exclude'): | ||||
for pattern in pathfilter.get(key, []): | ||||
if not pattern.startswith((b'path:', b'rootfilesin:')): | ||||
raise error.WireprotoCommandError( | ||||
|
r43347 | b'%s pattern must begin with `path:` or `rootfilesin:`; ' | ||
b'got %s', | ||||
|
r43346 | (key, pattern), | ||
) | ||||
|
r40214 | |||
if pathfilter: | ||||
|
r43346 | matcher = matchmod.match( | ||
repo.root, | ||||
b'', | ||||
include=pathfilter.get(b'include', []), | ||||
exclude=pathfilter.get(b'exclude', []), | ||||
) | ||||
|
r40214 | else: | ||
matcher = matchmod.match(repo.root, b'') | ||||
# Requested patterns could include files not in the local store. So | ||||
# filter those out. | ||||
|
r40685 | return repo.narrowmatch(matcher) | ||
|
r40214 | |||
|
r43346 | |||
|
r39835 | @wireprotocommand( | ||
|
r43347 | b'filedata', | ||
|
r39835 | args={ | ||
|
r43347 | b'haveparents': { | ||
b'type': b'bool', | ||||
b'default': lambda: False, | ||||
b'example': True, | ||||
|
r39835 | }, | ||
|
r46554 | b'nodes': { | ||
b'type': b'list', | ||||
b'example': [b'0123456...'], | ||||
}, | ||||
|
r43347 | b'fields': { | ||
b'type': b'set', | ||||
b'default': set, | ||||
b'example': {b'parents', b'revision'}, | ||||
b'validvalues': {b'parents', b'revision', b'linknode'}, | ||||
|
r39835 | }, | ||
|
r46554 | b'path': { | ||
b'type': b'bytes', | ||||
b'example': b'foo.txt', | ||||
}, | ||||
|
r39835 | }, | ||
|
r43347 | permission=b'pull', | ||
|
r40057 | # TODO censoring a file revision won't invalidate the cache. | ||
# Figure out a way to take censoring into account when deriving | ||||
# the cache key. | ||||
|
r43347 | cachekeyfn=makecommandcachekeyfn(b'filedata', 1, allargs=True), | ||
|
r43346 | ) | ||
|
r39835 | def filedata(repo, proto, haveparents, nodes, fields, path): | ||
|
r40214 | # TODO this API allows access to file revisions that are attached to | ||
# secret changesets. filesdata does not have this problem. Maybe this | ||||
# API should be deleted? | ||||
|
r39675 | try: | ||
# Extensions may wish to access the protocol handler. | ||||
store = getfilestore(repo, proto, path) | ||||
except FileAccessError as e: | ||||
raise error.WireprotoCommandError(e.msg, e.args) | ||||
|
r40959 | clnode = repo.changelog.node | ||
linknodes = {} | ||||
|
r39675 | # Validate requested nodes. | ||
for node in nodes: | ||||
try: | ||||
store.rev(node) | ||||
except error.LookupError: | ||||
|
r43346 | raise error.WireprotoCommandError( | ||
|
r43347 | b'unknown file node: %s', (hex(node),) | ||
|
r43346 | ) | ||
|
r39675 | |||
|
r40959 | # TODO by creating the filectx against a specific file revision | ||
# instead of changeset, linkrev() is always used. This is wrong for | ||||
# cases where linkrev() may refer to a hidden changeset. But since this | ||||
# API doesn't know anything about changesets, we're not sure how to | ||||
# disambiguate the linknode. Perhaps we should delete this API? | ||||
fctx = repo.filectx(path, fileid=node) | ||||
linknodes[node] = clnode(fctx.introrev()) | ||||
|
r43346 | revisions = store.emitrevisions( | ||
nodes, | ||||
revisiondata=b'revision' in fields, | ||||
assumehaveparentrevisions=haveparents, | ||||
) | ||||
|
r39675 | |||
yield { | ||||
|
r39900 | b'totalitems': len(nodes), | ||
|
r39675 | } | ||
|
r40959 | for o in emitfilerevisions(repo, path, revisions, linknodes, fields): | ||
|
r40213 | yield o | ||
|
r39675 | |||
|
r43346 | |||
|
r40214 | def filesdatacapabilities(repo, proto): | ||
batchsize = repo.ui.configint( | ||||
|
r43346 | b'experimental', b'server.filesdata.recommended-batch-size' | ||
) | ||||
|
r40214 | return { | ||
b'recommendedbatchsize': batchsize, | ||||
} | ||||
|
r43346 | |||
|
r40214 | @wireprotocommand( | ||
|
r43347 | b'filesdata', | ||
|
r40214 | args={ | ||
|
r43347 | b'haveparents': { | ||
b'type': b'bool', | ||||
b'default': lambda: False, | ||||
b'example': True, | ||||
|
r40214 | }, | ||
|
r43347 | b'fields': { | ||
b'type': b'set', | ||||
b'default': set, | ||||
b'example': {b'parents', b'revision'}, | ||||
b'validvalues': { | ||||
|
r43346 | b'firstchangeset', | ||
b'linknode', | ||||
b'parents', | ||||
b'revision', | ||||
}, | ||||
|
r40214 | }, | ||
|
r43347 | b'pathfilter': { | ||
b'type': b'dict', | ||||
b'default': lambda: None, | ||||
b'example': {b'include': [b'path:tests']}, | ||||
|
r40214 | }, | ||
|
r43347 | b'revisions': { | ||
b'type': b'list', | ||||
b'example': [ | ||||
|
r46554 | { | ||
b'type': b'changesetexplicit', | ||||
b'nodes': [b'abcdef...'], | ||||
} | ||||
|
r43346 | ], | ||
|
r40214 | }, | ||
}, | ||||
|
r43347 | permission=b'pull', | ||
|
r40214 | # TODO censoring a file revision won't invalidate the cache. | ||
# Figure out a way to take censoring into account when deriving | ||||
# the cache key. | ||||
|
r43347 | cachekeyfn=makecommandcachekeyfn(b'filesdata', 1, allargs=True), | ||
|
r43346 | extracapabilitiesfn=filesdatacapabilities, | ||
) | ||||
|
r40214 | def filesdata(repo, proto, haveparents, fields, pathfilter, revisions): | ||
# TODO This should operate on a repo that exposes obsolete changesets. There | ||||
# is a race between a client making a push that obsoletes a changeset and | ||||
# another client fetching files data for that changeset. If a client has a | ||||
# changeset, it should probably be allowed to access files data for that | ||||
# changeset. | ||||
outgoing = resolvenodes(repo, revisions) | ||||
filematcher = makefilematcher(repo, pathfilter) | ||||
|
r40959 | # path -> {fnode: linknode} | ||
fnodes = collections.defaultdict(dict) | ||||
|
r40214 | |||
|
r40960 | # We collect the set of relevant file revisions by iterating the changeset | ||
# revisions and either walking the set of files recorded in the changeset | ||||
# or by walking the manifest at that revision. There is probably room for a | ||||
# storage-level API to request this data, as it can be expensive to compute | ||||
# and would benefit from caching or alternate storage from what revlogs | ||||
# provide. | ||||
|
r40214 | for node in outgoing: | ||
ctx = repo[node] | ||||
|
r40960 | mctx = ctx.manifestctx() | ||
md = mctx.read() | ||||
|
r40214 | |||
|
r40960 | if haveparents: | ||
checkpaths = ctx.files() | ||||
else: | ||||
checkpaths = md.keys() | ||||
|
r40214 | |||
|
r40960 | for path in checkpaths: | ||
fnode = md[path] | ||||
|
r40214 | |||
|
r40960 | if path in fnodes and fnode in fnodes[path]: | ||
continue | ||||
|
r40214 | |||
|
r40960 | if not filematcher(path): | ||
continue | ||||
fnodes[path].setdefault(fnode, node) | ||||
|
r40214 | |||
yield { | ||||
b'totalpaths': len(fnodes), | ||||
|
r43346 | b'totalitems': sum(len(v) for v in fnodes.values()), | ||
|
r40214 | } | ||
for path, filenodes in sorted(fnodes.items()): | ||||
try: | ||||
store = getfilestore(repo, proto, path) | ||||
except FileAccessError as e: | ||||
raise error.WireprotoCommandError(e.msg, e.args) | ||||
yield { | ||||
b'path': path, | ||||
b'totalitems': len(filenodes), | ||||
} | ||||
|
r43346 | revisions = store.emitrevisions( | ||
filenodes.keys(), | ||||
revisiondata=b'revision' in fields, | ||||
assumehaveparentrevisions=haveparents, | ||||
) | ||||
|
r40214 | |||
|
r40959 | for o in emitfilerevisions(repo, path, revisions, filenodes, fields): | ||
|
r40214 | yield o | ||
|
r43346 | |||
|
r39835 | @wireprotocommand( | ||
|
r43347 | b'heads', | ||
|
r39835 | args={ | ||
|
r43347 | b'publiconly': { | ||
b'type': b'bool', | ||||
b'default': lambda: False, | ||||
b'example': False, | ||||
|
r39835 | }, | ||
}, | ||||
|
r43347 | permission=b'pull', | ||
|
r43346 | ) | ||
|
r39835 | def headsv2(repo, proto, publiconly): | ||
|
r37564 | if publiconly: | ||
|
r43347 | repo = repo.filtered(b'immutable') | ||
|
r37564 | |||
|
r39595 | yield repo.heads() | ||
|
r37564 | |||
|
r43346 | |||
|
r39835 | @wireprotocommand( | ||
|
r43347 | b'known', | ||
|
r39835 | args={ | ||
|
r43347 | b'nodes': { | ||
b'type': b'list', | ||||
b'default': list, | ||||
b'example': [b'deadbeef'], | ||||
}, | ||||
|
r39835 | }, | ||
|
r43347 | permission=b'pull', | ||
|
r43346 | ) | ||
|
r39835 | def knownv2(repo, proto, nodes): | ||
|
r37564 | result = b''.join(b'1' if n else b'0' for n in repo.known(nodes)) | ||
|
r39595 | yield result | ||
|
r37564 | |||
|
r43346 | |||
|
r39835 | @wireprotocommand( | ||
|
r43347 | b'listkeys', | ||
|
r46554 | args={ | ||
b'namespace': { | ||||
b'type': b'bytes', | ||||
b'example': b'ns', | ||||
}, | ||||
}, | ||||
|
r43347 | permission=b'pull', | ||
|
r43346 | ) | ||
|
r39835 | def listkeysv2(repo, proto, namespace): | ||
|
r37564 | keys = repo.listkeys(encoding.tolocal(namespace)) | ||
|
r43346 | keys = { | ||
encoding.fromlocal(k): encoding.fromlocal(v) | ||||
|
r43376 | for k, v in pycompat.iteritems(keys) | ||
|
r43346 | } | ||
|
r37564 | |||
|
r39595 | yield keys | ||
|
r37564 | |||
|
r43346 | |||
|
r39835 | @wireprotocommand( | ||
|
r43347 | b'lookup', | ||
|
r46554 | args={ | ||
b'key': { | ||||
b'type': b'bytes', | ||||
b'example': b'foo', | ||||
}, | ||||
}, | ||||
|
r43347 | permission=b'pull', | ||
|
r43346 | ) | ||
|
r37564 | def lookupv2(repo, proto, key): | ||
key = encoding.tolocal(key) | ||||
# TODO handle exception. | ||||
node = repo.lookup(key) | ||||
|
r39595 | yield node | ||
|
r37564 | |||
|
r43346 | |||
|
r40208 | def manifestdatacapabilities(repo, proto): | ||
batchsize = repo.ui.configint( | ||||
|
r43346 | b'experimental', b'server.manifestdata.recommended-batch-size' | ||
) | ||||
|
r40208 | |||
return { | ||||
b'recommendedbatchsize': batchsize, | ||||
} | ||||
|
r43346 | |||
|
r39835 | @wireprotocommand( | ||
|
r43347 | b'manifestdata', | ||
|
r39835 | args={ | ||
|
r46554 | b'nodes': { | ||
b'type': b'list', | ||||
b'example': [b'0123456...'], | ||||
}, | ||||
|
r43347 | b'haveparents': { | ||
b'type': b'bool', | ||||
b'default': lambda: False, | ||||
b'example': True, | ||||
|
r39835 | }, | ||
|
r43347 | b'fields': { | ||
b'type': b'set', | ||||
b'default': set, | ||||
b'example': {b'parents', b'revision'}, | ||||
b'validvalues': {b'parents', b'revision'}, | ||||
|
r39835 | }, | ||
|
r46554 | b'tree': { | ||
b'type': b'bytes', | ||||
b'example': b'', | ||||
}, | ||||
|
r39835 | }, | ||
|
r43347 | permission=b'pull', | ||
cachekeyfn=makecommandcachekeyfn(b'manifestdata', 1, allargs=True), | ||||
|
r43346 | extracapabilitiesfn=manifestdatacapabilities, | ||
) | ||||
|
r39835 | def manifestdata(repo, proto, haveparents, nodes, fields, tree): | ||
|
r39673 | store = repo.manifestlog.getstorage(tree) | ||
# Validate the node is known and abort on unknown revisions. | ||||
for node in nodes: | ||||
try: | ||||
store.rev(node) | ||||
except error.LookupError: | ||||
|
r43347 | raise error.WireprotoCommandError(b'unknown node: %s', (node,)) | ||
|
r39673 | |||
|
r43346 | revisions = store.emitrevisions( | ||
nodes, | ||||
revisiondata=b'revision' in fields, | ||||
assumehaveparentrevisions=haveparents, | ||||
) | ||||
|
r39673 | |||
yield { | ||||
|
r39900 | b'totalitems': len(nodes), | ||
|
r39673 | } | ||
|
r39900 | for revision in revisions: | ||
|
r39673 | d = { | ||
|
r39900 | b'node': revision.node, | ||
|
r39673 | } | ||
if b'parents' in fields: | ||||
|
r39900 | d[b'parents'] = [revision.p1node, revision.p2node] | ||
|
r39673 | |||
|
r39839 | followingmeta = [] | ||
followingdata = [] | ||||
|
r39673 | if b'revision' in fields: | ||
|
r39900 | if revision.revision is not None: | ||
followingmeta.append((b'revision', len(revision.revision))) | ||||
followingdata.append(revision.revision) | ||||
|
r39673 | else: | ||
|
r39900 | d[b'deltabasenode'] = revision.basenode | ||
followingmeta.append((b'delta', len(revision.delta))) | ||||
followingdata.append(revision.delta) | ||||
|
r39839 | |||
if followingmeta: | ||||
d[b'fieldsfollowing'] = followingmeta | ||||
|
r39673 | |||
yield d | ||||
|
r39839 | for extra in followingdata: | ||
yield extra | ||||
|
r39673 | |||
|
r43346 | |||
|
r39835 | @wireprotocommand( | ||
|
r43347 | b'pushkey', | ||
|
r39835 | args={ | ||
|
r46554 | b'namespace': { | ||
b'type': b'bytes', | ||||
b'example': b'ns', | ||||
}, | ||||
b'key': { | ||||
b'type': b'bytes', | ||||
b'example': b'key', | ||||
}, | ||||
b'old': { | ||||
b'type': b'bytes', | ||||
b'example': b'old', | ||||
}, | ||||
b'new': { | ||||
b'type': b'bytes', | ||||
b'example': b'new', | ||||
}, | ||||
|
r39835 | }, | ||
|
r43347 | permission=b'push', | ||
|
r43346 | ) | ||
|
r37564 | def pushkeyv2(repo, proto, namespace, key, old, new): | ||
# TODO handle ui output redirection | ||||
|
r43346 | yield repo.pushkey( | ||
encoding.tolocal(namespace), | ||||
encoding.tolocal(key), | ||||
encoding.tolocal(old), | ||||
encoding.tolocal(new), | ||||
) | ||||
|
r40365 | |||
@wireprotocommand( | ||||
|
r43347 | b'rawstorefiledata', | ||
|
r40365 | args={ | ||
|
r43347 | b'files': { | ||
b'type': b'list', | ||||
b'example': [b'changelog', b'manifestlog'], | ||||
}, | ||||
b'pathfilter': { | ||||
b'type': b'list', | ||||
b'default': lambda: None, | ||||
b'example': {b'include': [b'path:tests']}, | ||||
|
r40365 | }, | ||
}, | ||||
|
r43347 | permission=b'pull', | ||
|
r43346 | ) | ||
|
r40365 | def rawstorefiledata(repo, proto, files, pathfilter): | ||
if not streamclone.allowservergeneration(repo): | ||||
raise error.WireprotoCommandError(b'stream clone is disabled') | ||||
# TODO support dynamically advertising what store files "sets" are | ||||
# available. For now, we support changelog, manifestlog, and files. | ||||
files = set(files) | ||||
allowedfiles = {b'changelog', b'manifestlog'} | ||||
unsupported = files - allowedfiles | ||||
if unsupported: | ||||
|
r43346 | raise error.WireprotoCommandError( | ||
b'unknown file type: %s', (b', '.join(sorted(unsupported)),) | ||||
) | ||||
|
r40365 | |||
with repo.lock(): | ||||
topfiles = list(repo.store.topfiles()) | ||||
sendfiles = [] | ||||
totalsize = 0 | ||||
# TODO this is a bunch of storage layer interface abstractions because | ||||
# it assumes revlogs. | ||||
r47657 | for rl_type, name, encodedname, size in topfiles: | |||
# XXX use the `rl_type` for that | ||||
|
r40365 | if b'changelog' in files and name.startswith(b'00changelog'): | ||
pass | ||||
elif b'manifestlog' in files and name.startswith(b'00manifest'): | ||||
pass | ||||
else: | ||||
continue | ||||
sendfiles.append((b'store', name, size)) | ||||
totalsize += size | ||||
yield { | ||||
b'filecount': len(sendfiles), | ||||
b'totalsize': totalsize, | ||||
} | ||||
for location, name, size in sendfiles: | ||||
yield { | ||||
b'location': location, | ||||
b'path': name, | ||||
b'size': size, | ||||
} | ||||
# We have to use a closure for this to ensure the context manager is | ||||
# closed only after sending the final chunk. | ||||
def getfiledata(): | ||||
|
r43347 | with repo.svfs(name, b'rb', auditpath=False) as fh: | ||
|
r40365 | for chunk in util.filechunkiter(fh, limit=size): | ||
yield chunk | ||||
|
r43346 | yield wireprototypes.indefinitebytestringresponse(getfiledata()) | ||