wireproto.py
866 lines
| 34.4 KiB
| text/x-python
|
PythonLexer
/ mercurial / wireproto.py
|
r11581 | # wireproto.py - generic wire protocol support functions | ||
# | ||||
# Copyright 2005-2010 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. | ||||
|
r25993 | from __future__ import absolute_import | ||
import os | ||||
import tempfile | ||||
|
r11581 | |||
|
r25993 | from .i18n import _ | ||
from .node import ( | ||||
hex, | ||||
|
r32260 | nullid, | ||
|
r25993 | ) | ||
from . import ( | ||||
bundle2, | ||||
changegroup as changegroupmod, | ||||
|
r34099 | discovery, | ||
|
r25993 | encoding, | ||
error, | ||||
exchange, | ||||
pushkey as pushkeymod, | ||||
|
r30924 | pycompat, | ||
|
r26443 | streamclone, | ||
|
r25993 | util, | ||
|
r36090 | wireprototypes, | ||
|
r25993 | ) | ||
|
r20903 | |||
|
r37102 | from .utils import ( | ||
|
r37137 | procutil, | ||
|
r37102 | stringutil, | ||
) | ||||
|
r28883 | urlerr = util.urlerr | ||
urlreq = util.urlreq | ||||
|
r30909 | bundle2requiredmain = _('incompatible Mercurial client; bundle2 required') | ||
bundle2requiredhint = _('see https://www.mercurial-scm.org/wiki/' | ||||
'IncompatibleClient') | ||||
bundle2required = '%s\n(%s)\n' % (bundle2requiredmain, bundle2requiredhint) | ||||
|
r27246 | |||
|
r37411 | def clientcompressionsupport(proto): | ||
"""Returns a list of compression methods supported by the client. | ||||
Returns a list of the compression methods supported by the client | ||||
according to the protocol capabilities. If no such capability has | ||||
been announced, fallback to the default of zlib and uncompressed. | ||||
""" | ||||
for cap in proto.getprotocaps(): | ||||
if cap.startswith('comp='): | ||||
return cap[5:].split(',') | ||||
return ['zlib', 'none'] | ||||
|
r20902 | # wire protocol command can either return a string or one of these classes. | ||
|
r15017 | |||
|
r29590 | def getdispatchrepo(repo, proto, command): | ||
"""Obtain the repo used for processing wire protocol commands. | ||||
The intent of this function is to serve as a monkeypatch point for | ||||
extensions that need commands to operate on different repo views under | ||||
specialized circumstances. | ||||
""" | ||||
return repo.filtered('served') | ||||
|
r11581 | def dispatch(repo, proto, command): | ||
|
r29590 | repo = getdispatchrepo(repo, proto, command) | ||
|
r37311 | |||
transportversion = wireprototypes.TRANSPORTS[proto.name]['version'] | ||||
commandtable = commandsv2 if transportversion == 2 else commands | ||||
func, spec = commandtable[command] | ||||
|
r11581 | args = proto.getargs(spec) | ||
|
r37503 | |||
# Version 1 protocols define arguments as a list. Version 2 uses a dict. | ||||
if isinstance(args, list): | ||||
return func(repo, proto, *args) | ||||
elif isinstance(args, dict): | ||||
return func(repo, proto, **args) | ||||
else: | ||||
raise error.ProgrammingError('unexpected type returned from ' | ||||
'proto.getargs(): %s' % type(args)) | ||||
|
r11581 | |||
|
r13721 | def options(cmd, keys, others): | ||
opts = {} | ||||
for k in keys: | ||||
if k in others: | ||||
opts[k] = others[k] | ||||
del others[k] | ||||
if others: | ||||
|
r37137 | procutil.stderr.write("warning: %s ignored unexpected arguments %s\n" | ||
% (cmd, ",".join(others))) | ||||
|
r13721 | return opts | ||
|
r27633 | def bundle1allowed(repo, action): | ||
"""Whether a bundle1 operation is allowed from the server. | ||||
Priority is: | ||||
1. server.bundle1gd.<action> (if generaldelta active) | ||||
2. server.bundle1.<action> | ||||
3. server.bundle1gd (if generaldelta active) | ||||
4. server.bundle1 | ||||
""" | ||||
ui = repo.ui | ||||
gd = 'generaldelta' in repo.requirements | ||||
if gd: | ||||
|
r34614 | v = ui.configbool('server', 'bundle1gd.%s' % action) | ||
|
r27633 | if v is not None: | ||
return v | ||||
|
r34614 | v = ui.configbool('server', 'bundle1.%s' % action) | ||
|
r27246 | if v is not None: | ||
return v | ||||
|
r27633 | if gd: | ||
r33217 | v = ui.configbool('server', 'bundle1gd') | |||
|
r27633 | if v is not None: | ||
return v | ||||
r33216 | return ui.configbool('server', 'bundle1') | |||
|
r27246 | |||
|
r36088 | def supportedcompengines(ui, role): | ||
|
r30762 | """Obtain the list of supported compression engines for a request.""" | ||
assert role in (util.CLIENTROLE, util.SERVERROLE) | ||||
compengines = util.compengines.supportedwireengines(role) | ||||
# Allow config to override default list and ordering. | ||||
if role == util.SERVERROLE: | ||||
configengines = ui.configlist('server', 'compressionengines') | ||||
config = 'server.compressionengines' | ||||
else: | ||||
# This is currently implemented mainly to facilitate testing. In most | ||||
# cases, the server should be in charge of choosing a compression engine | ||||
# because a server has the most to lose from a sub-optimal choice. (e.g. | ||||
# CPU DoS due to an expensive engine or a network DoS due to poor | ||||
# compression ratio). | ||||
configengines = ui.configlist('experimental', | ||||
'clientcompressionengines') | ||||
config = 'experimental.clientcompressionengines' | ||||
# No explicit config. Filter out the ones that aren't supposed to be | ||||
# advertised and return default ordering. | ||||
if not configengines: | ||||
attr = 'serverpriority' if role == util.SERVERROLE else 'clientpriority' | ||||
return [e for e in compengines | ||||
if getattr(e.wireprotosupport(), attr) > 0] | ||||
# If compression engines are listed in the config, assume there is a good | ||||
# reason for it (like server operators wanting to achieve specific | ||||
# performance characteristics). So fail fast if the config references | ||||
# unusable compression engines. | ||||
validnames = set(e.name() for e in compengines) | ||||
invalidnames = set(e for e in configengines if e not in validnames) | ||||
if invalidnames: | ||||
raise error.Abort(_('invalid compression engine defined in %s: %s') % | ||||
(config, ', '.join(sorted(invalidnames)))) | ||||
compengines = [e for e in compengines if e.name() in configengines] | ||||
compengines = sorted(compengines, | ||||
key=lambda e: configengines.index(e.name())) | ||||
if not compengines: | ||||
raise error.Abort(_('%s config option does not specify any known ' | ||||
'compression engines') % config, | ||||
hint=_('usable compression engines: %s') % | ||||
', '.sorted(validnames)) | ||||
return compengines | ||||
|
r35999 | class commandentry(object): | ||
"""Represents a declared wire protocol command.""" | ||||
|
r36818 | def __init__(self, func, args='', transports=None, | ||
permission='push'): | ||||
|
r35999 | self.func = func | ||
self.args = args | ||||
|
r36627 | self.transports = transports or set() | ||
|
r36818 | self.permission = permission | ||
|
r35999 | |||
def _merge(self, func, args): | ||||
"""Merge this instance with an incoming 2-tuple. | ||||
This is called when a caller using the old 2-tuple API attempts | ||||
to replace an instance. The incoming values are merged with | ||||
data not captured by the 2-tuple and a new instance containing | ||||
the union of the two objects is returned. | ||||
""" | ||||
|
r36818 | return commandentry(func, args=args, transports=set(self.transports), | ||
permission=self.permission) | ||||
|
r35999 | |||
# Old code treats instances as 2-tuples. So expose that interface. | ||||
def __iter__(self): | ||||
yield self.func | ||||
yield self.args | ||||
def __getitem__(self, i): | ||||
if i == 0: | ||||
return self.func | ||||
elif i == 1: | ||||
return self.args | ||||
else: | ||||
raise IndexError('can only access elements 0 and 1') | ||||
class commanddict(dict): | ||||
"""Container for registered wire protocol commands. | ||||
It behaves like a dict. But __setitem__ is overwritten to allow silent | ||||
coercion of values from 2-tuples for API compatibility. | ||||
""" | ||||
def __setitem__(self, k, v): | ||||
if isinstance(v, commandentry): | ||||
pass | ||||
# Cast 2-tuples to commandentry instances. | ||||
elif isinstance(v, tuple): | ||||
if len(v) != 2: | ||||
raise ValueError('command tuples must have exactly 2 elements') | ||||
# It is common for extensions to wrap wire protocol commands via | ||||
# e.g. ``wireproto.commands[x] = (newfn, args)``. Because callers | ||||
# doing this aren't aware of the new API that uses objects to store | ||||
# command entries, we automatically merge old state with new. | ||||
if k in self: | ||||
v = self[k]._merge(v[0], v[1]) | ||||
else: | ||||
|
r36627 | # Use default values from @wireprotocommand. | ||
v = commandentry(v[0], args=v[1], | ||||
|
r36818 | transports=set(wireprototypes.TRANSPORTS), | ||
permission='push') | ||||
|
r35999 | else: | ||
raise ValueError('command entries must be commandentry instances ' | ||||
'or 2-tuples') | ||||
return super(commanddict, self).__setitem__(k, v) | ||||
|
r36000 | def commandavailable(self, command, proto): | ||
"""Determine if a command is available for the requested protocol.""" | ||||
|
r36627 | assert proto.name in wireprototypes.TRANSPORTS | ||
entry = self.get(command) | ||||
if not entry: | ||||
return False | ||||
if proto.name not in entry.transports: | ||||
return False | ||||
return True | ||||
# Constants specifying which transports a wire protocol command should be | ||||
# available on. For use with @wireprotocommand. | ||||
POLICY_V1_ONLY = 'v1-only' | ||||
POLICY_V2_ONLY = 'v2-only' | ||||
|
r36000 | |||
|
r37311 | # For version 1 transports. | ||
|
r35999 | commands = commanddict() | ||
|
r20906 | |||
|
r37311 | # For version 2 transports. | ||
commandsv2 = commanddict() | ||||
|
r37559 | def wireprotocommand(name, args=None, transportpolicy=POLICY_V1_ONLY, | ||
|
r36818 | permission='push'): | ||
|
r35998 | """Decorator to declare a wire protocol command. | ||
``name`` is the name of the wire protocol command being provided. | ||||
|
r37553 | ``args`` defines the named arguments accepted by the command. It is | ||
ideally a dict mapping argument names to their types. For backwards | ||||
compatibility, it can be a space-delimited list of argument names. For | ||||
version 1 transports, ``*`` denotes a special value that says to accept | ||||
all named arguments. | ||||
|
r36627 | |||
``transportpolicy`` is a POLICY_* constant denoting which transports | ||||
this wire protocol command should be exposed to. By default, commands | ||||
are exposed to all wire protocol transports. | ||||
|
r36818 | |||
``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. | ||||
|
r35998 | """ | ||
|
r37559 | if transportpolicy == POLICY_V1_ONLY: | ||
|
r36627 | transports = {k for k, v in wireprototypes.TRANSPORTS.items() | ||
if v['version'] == 1} | ||||
|
r37559 | transportversion = 1 | ||
|
r36627 | elif transportpolicy == POLICY_V2_ONLY: | ||
transports = {k for k, v in wireprototypes.TRANSPORTS.items() | ||||
if v['version'] == 2} | ||||
|
r37559 | transportversion = 2 | ||
|
r36627 | else: | ||
|
r36858 | raise error.ProgrammingError('invalid transport policy value: %s' % | ||
transportpolicy) | ||||
|
r36627 | |||
|
r37071 | # Because SSHv2 is a mirror of SSHv1, we allow "batch" commands through to | ||
# SSHv2. | ||||
# TODO undo this hack when SSH is using the unified frame protocol. | ||||
if name == b'batch': | ||||
transports.add(wireprototypes.SSHV2) | ||||
|
r36818 | if permission not in ('push', 'pull'): | ||
|
r36858 | raise error.ProgrammingError('invalid wire protocol permission; ' | ||
'got %s; expected "push" or "pull"' % | ||||
permission) | ||||
|
r36818 | |||
|
r37559 | if transportversion == 1: | ||
if args is None: | ||||
args = '' | ||||
|
r37553 | |||
|
r37559 | if not isinstance(args, bytes): | ||
raise error.ProgrammingError('arguments for version 1 commands ' | ||||
'must be declared as bytes') | ||||
elif transportversion == 2: | ||||
if args is None: | ||||
args = {} | ||||
if not isinstance(args, dict): | ||||
raise error.ProgrammingError('arguments for version 2 commands ' | ||||
'must be declared as dicts') | ||||
|
r37553 | |||
|
r20906 | def register(func): | ||
|
r37559 | if transportversion == 1: | ||
|
r37311 | if name in commands: | ||
raise error.ProgrammingError('%s command already registered ' | ||||
'for version 1' % name) | ||||
commands[name] = commandentry(func, args=args, | ||||
transports=transports, | ||||
permission=permission) | ||||
|
r37559 | elif transportversion == 2: | ||
|
r37311 | if name in commandsv2: | ||
raise error.ProgrammingError('%s command already registered ' | ||||
'for version 2' % name) | ||||
|
r37553 | |||
|
r37559 | commandsv2[name] = commandentry(func, args=args, | ||
|
r37311 | transports=transports, | ||
permission=permission) | ||||
|
r37559 | else: | ||
raise error.ProgrammingError('unhandled transport version: %d' % | ||||
transportversion) | ||||
|
r37311 | |||
|
r20906 | return func | ||
return register | ||||
|
r36773 | # TODO define a more appropriate permissions type to use for this. | ||
|
r37071 | @wireprotocommand('batch', 'cmds *', permission='pull', | ||
transportpolicy=POLICY_V1_ONLY) | ||||
|
r14622 | def batch(repo, proto, cmds, others): | ||
|
r37630 | unescapearg = wireprototypes.unescapebatcharg | ||
|
r18382 | repo = repo.filtered("served") | ||
|
r14622 | res = [] | ||
for pair in cmds.split(';'): | ||||
op, args = pair.split(' ', 1) | ||||
vals = {} | ||||
for a in args.split(','): | ||||
if a: | ||||
n, v = a.split('=') | ||||
|
r29734 | vals[unescapearg(n)] = unescapearg(v) | ||
|
r14622 | func, spec = commands[op] | ||
|
r36773 | |||
|
r36819 | # Validate that client has permissions to perform this command. | ||
perm = commands[op].permission | ||||
assert perm in ('push', 'pull') | ||||
proto.checkperm(perm) | ||||
|
r36773 | |||
|
r14622 | if spec: | ||
keys = spec.split() | ||||
data = {} | ||||
for k in keys: | ||||
if k == '*': | ||||
star = {} | ||||
for key in vals.keys(): | ||||
if key not in keys: | ||||
star[key] = vals[key] | ||||
data['*'] = star | ||||
else: | ||||
data[k] = vals[k] | ||||
result = func(repo, proto, *[data[k] for k in keys]) | ||||
else: | ||||
result = func(repo, proto) | ||||
|
r37309 | if isinstance(result, wireprototypes.ooberror): | ||
|
r15017 | return result | ||
|
r36091 | |||
# For now, all batchable commands must return bytesresponse or | ||||
# raw bytes (for backwards compatibility). | ||||
|
r37309 | assert isinstance(result, (wireprototypes.bytesresponse, bytes)) | ||
if isinstance(result, wireprototypes.bytesresponse): | ||||
|
r36091 | result = result.data | ||
|
r37630 | res.append(wireprototypes.escapebatcharg(result)) | ||
|
r36091 | |||
|
r37309 | return wireprototypes.bytesresponse(';'.join(res)) | ||
|
r14622 | |||
|
r36818 | @wireprotocommand('between', 'pairs', transportpolicy=POLICY_V1_ONLY, | ||
permission='pull') | ||||
|
r11583 | def between(repo, proto, pairs): | ||
|
r37630 | pairs = [wireprototypes.decodelist(p, '-') for p in pairs.split(" ")] | ||
|
r11581 | r = [] | ||
for b in repo.between(pairs): | ||||
|
r37630 | r.append(wireprototypes.encodelist(b) + "\n") | ||
|
r36091 | |||
|
r37309 | return wireprototypes.bytesresponse(''.join(r)) | ||
|
r11581 | |||
|
r37506 | @wireprotocommand('branchmap', permission='pull', | ||
transportpolicy=POLICY_V1_ONLY) | ||||
|
r11583 | def branchmap(repo, proto): | ||
|
r18281 | branchmap = repo.branchmap() | ||
|
r11581 | heads = [] | ||
for branch, nodes in branchmap.iteritems(): | ||||
|
r28883 | branchname = urlreq.quote(encoding.fromlocal(branch)) | ||
|
r37630 | branchnodes = wireprototypes.encodelist(nodes) | ||
|
r11597 | heads.append('%s %s' % (branchname, branchnodes)) | ||
|
r36091 | |||
|
r37309 | return wireprototypes.bytesresponse('\n'.join(heads)) | ||
|
r11581 | |||
|
r36818 | @wireprotocommand('branches', 'nodes', transportpolicy=POLICY_V1_ONLY, | ||
permission='pull') | ||||
|
r11583 | def branches(repo, proto, nodes): | ||
|
r37630 | nodes = wireprototypes.decodelist(nodes) | ||
|
r11581 | r = [] | ||
for b in repo.branches(nodes): | ||||
|
r37630 | r.append(wireprototypes.encodelist(b) + "\n") | ||
|
r36091 | |||
|
r37309 | return wireprototypes.bytesresponse(''.join(r)) | ||
|
r11581 | |||
|
r37554 | @wireprotocommand('clonebundles', '', permission='pull', | ||
transportpolicy=POLICY_V1_ONLY) | ||||
|
r26857 | def clonebundles(repo, proto): | ||
"""Server command for returning info for available bundles to seed clones. | ||||
Clients will parse this response and determine what bundle to fetch. | ||||
Extensions may wrap this command to filter or dynamically emit data | ||||
depending on the request. e.g. you could advertise URLs for the closest | ||||
data center given the client's IP address. | ||||
""" | ||||
|
r37309 | return wireprototypes.bytesresponse( | ||
repo.vfs.tryread('clonebundles.manifest')) | ||||
|
r20774 | |||
|
r36630 | wireprotocaps = ['lookup', 'branchmap', 'pushkey', | ||
|
r37071 | 'known', 'getbundle', 'unbundlehash'] | ||
|
r20775 | |||
def _capabilities(repo, proto): | ||||
"""return a list of capabilities for a repo | ||||
This function exists to allow extensions to easily wrap capabilities | ||||
computation | ||||
- returns a lists: easy to alter | ||||
- change done here will be propagated to both `capabilities` and `hello` | ||||
|
r21024 | command without any other action needed. | ||
|
r20775 | """ | ||
|
r20774 | # copy to prevent modification of the global list | ||
caps = list(wireprotocaps) | ||||
|
r36630 | |||
# Command of same name as capability isn't exposed to version 1 of | ||||
# transports. So conditionally add it. | ||||
if commands.commandavailable('changegroupsubset', proto): | ||||
caps.append('changegroupsubset') | ||||
|
r32744 | if streamclone.allowservergeneration(repo): | ||
r33222 | if repo.ui.configbool('server', 'preferuncompressed'): | |||
|
r16361 | caps.append('stream-preferred') | ||
|
r12296 | requiredformats = repo.requirements & repo.supportedformats | ||
# if our local revlogs are just revlogv1, add 'stream' cap | ||||
|
r32291 | if not requiredformats - {'revlogv1'}: | ||
|
r12296 | caps.append('stream') | ||
# otherwise, add 'streamreqs' detailing our local revlog format | ||||
else: | ||||
|
r26911 | caps.append('streamreqs=%s' % ','.join(sorted(requiredformats))) | ||
|
r33499 | if repo.ui.configbool('experimental', 'bundle2-advertise'): | ||
|
r35801 | capsblob = bundle2.encodecaps(bundle2.getrepocaps(repo, role='server')) | ||
|
r28883 | caps.append('bundle2=' + urlreq.quote(capsblob)) | ||
|
r28666 | caps.append('unbundle=%s' % ','.join(bundle2.bundlepriority)) | ||
|
r30563 | |||
|
r36631 | return proto.addcapabilities(repo, caps) | ||
|
r20775 | |||
|
r21024 | # If you are writing an extension and consider wrapping this function. Wrap | ||
|
r20775 | # `_capabilities` instead. | ||
|
r37551 | @wireprotocommand('capabilities', permission='pull', | ||
transportpolicy=POLICY_V1_ONLY) | ||||
|
r20775 | def capabilities(repo, proto): | ||
|
r37431 | caps = _capabilities(repo, proto) | ||
return wireprototypes.bytesresponse(' '.join(sorted(caps))) | ||||
|
r11594 | |||
|
r36818 | @wireprotocommand('changegroup', 'roots', transportpolicy=POLICY_V1_ONLY, | ||
permission='pull') | ||||
|
r11584 | def changegroup(repo, proto, roots): | ||
|
r37630 | nodes = wireprototypes.decodelist(roots) | ||
|
r34102 | outgoing = discovery.outgoing(repo, missingroots=nodes, | ||
missingheads=repo.heads()) | ||||
cg = changegroupmod.makechangegroup(repo, outgoing, '01', 'serve') | ||||
|
r35723 | gen = iter(lambda: cg.read(32768), '') | ||
|
r37309 | return wireprototypes.streamres(gen=gen) | ||
|
r11584 | |||
|
r36629 | @wireprotocommand('changegroupsubset', 'bases heads', | ||
|
r36818 | transportpolicy=POLICY_V1_ONLY, | ||
permission='pull') | ||||
|
r11584 | def changegroupsubset(repo, proto, bases, heads): | ||
|
r37630 | bases = wireprototypes.decodelist(bases) | ||
heads = wireprototypes.decodelist(heads) | ||||
|
r34099 | outgoing = discovery.outgoing(repo, missingroots=bases, | ||
missingheads=heads) | ||||
cg = changegroupmod.makechangegroup(repo, outgoing, '01', 'serve') | ||||
|
r35723 | gen = iter(lambda: cg.read(32768), '') | ||
|
r37309 | return wireprototypes.streamres(gen=gen) | ||
|
r11584 | |||
|
r36818 | @wireprotocommand('debugwireargs', 'one two *', | ||
|
r37508 | permission='pull', transportpolicy=POLICY_V1_ONLY) | ||
|
r13721 | def debugwireargs(repo, proto, one, two, others): | ||
# only accept optional args from the known set | ||||
opts = options('debugwireargs', ['three', 'four'], others) | ||||
|
r37309 | return wireprototypes.bytesresponse(repo.debugwireargs( | ||
one, two, **pycompat.strkwargs(opts))) | ||||
|
r13720 | |||
|
r37516 | def find_pullbundle(repo, proto, opts, clheads, heads, common): | ||
"""Return a file object for the first matching pullbundle. | ||||
Pullbundles are specified in .hg/pullbundles.manifest similar to | ||||
clonebundles. | ||||
For each entry, the bundle specification is checked for compatibility: | ||||
- Client features vs the BUNDLESPEC. | ||||
- Revisions shared with the clients vs base revisions of the bundle. | ||||
A bundle can be applied only if all its base revisions are known by | ||||
the client. | ||||
- At least one leaf of the bundle's DAG is missing on the client. | ||||
- Every leaf of the bundle's DAG is part of node set the client wants. | ||||
E.g. do not send a bundle of all changes if the client wants only | ||||
one specific branch of many. | ||||
""" | ||||
def decodehexstring(s): | ||||
return set([h.decode('hex') for h in s.split(';')]) | ||||
manifest = repo.vfs.tryread('pullbundles.manifest') | ||||
if not manifest: | ||||
return None | ||||
res = exchange.parseclonebundlesmanifest(repo, manifest) | ||||
res = exchange.filterclonebundleentries(repo, res) | ||||
if not res: | ||||
return None | ||||
cl = repo.changelog | ||||
heads_anc = cl.ancestors([cl.rev(rev) for rev in heads], inclusive=True) | ||||
common_anc = cl.ancestors([cl.rev(rev) for rev in common], inclusive=True) | ||||
compformats = clientcompressionsupport(proto) | ||||
for entry in res: | ||||
if 'COMPRESSION' in entry and entry['COMPRESSION'] not in compformats: | ||||
continue | ||||
# No test yet for VERSION, since V2 is supported by any client | ||||
# that advertises partial pulls | ||||
if 'heads' in entry: | ||||
try: | ||||
bundle_heads = decodehexstring(entry['heads']) | ||||
except TypeError: | ||||
# Bad heads entry | ||||
continue | ||||
if bundle_heads.issubset(common): | ||||
continue # Nothing new | ||||
if all(cl.rev(rev) in common_anc for rev in bundle_heads): | ||||
continue # Still nothing new | ||||
if any(cl.rev(rev) not in heads_anc and | ||||
cl.rev(rev) not in common_anc for rev in bundle_heads): | ||||
continue | ||||
if 'bases' in entry: | ||||
try: | ||||
bundle_bases = decodehexstring(entry['bases']) | ||||
except TypeError: | ||||
# Bad bases entry | ||||
continue | ||||
if not all(cl.rev(rev) in common_anc for rev in bundle_bases): | ||||
continue | ||||
path = entry['URL'] | ||||
repo.ui.debug('sending pullbundle "%s"\n' % path) | ||||
try: | ||||
return repo.vfs.open(path) | ||||
except IOError: | ||||
repo.ui.debug('pullbundle "%s" not accessible\n' % path) | ||||
continue | ||||
return None | ||||
|
r37557 | @wireprotocommand('getbundle', '*', permission='pull', | ||
transportpolicy=POLICY_V1_ONLY) | ||||
|
r13741 | def getbundle(repo, proto, others): | ||
|
r37631 | opts = options('getbundle', wireprototypes.GETBUNDLE_ARGUMENTS.keys(), | ||
others) | ||||
|
r13741 | for k, v in opts.iteritems(): | ||
|
r37631 | keytype = wireprototypes.GETBUNDLE_ARGUMENTS[k] | ||
|
r21646 | if keytype == 'nodes': | ||
|
r37630 | opts[k] = wireprototypes.decodelist(v) | ||
|
r21646 | elif keytype == 'csv': | ||
|
r25403 | opts[k] = list(v.split(',')) | ||
elif keytype == 'scsv': | ||||
|
r19201 | opts[k] = set(v.split(',')) | ||
|
r21988 | elif keytype == 'boolean': | ||
|
r26686 | # Client should serialize False as '0', which is a non-empty string | ||
# so it evaluates as a True bool. | ||||
if v == '0': | ||||
opts[k] = False | ||||
else: | ||||
opts[k] = bool(v) | ||||
|
r21646 | elif keytype != 'plain': | ||
raise KeyError('unknown getbundle option type %s' | ||||
% keytype) | ||||
|
r27246 | |||
|
r27633 | if not bundle1allowed(repo, 'pull'): | ||
|
r27246 | if not exchange.bundle2requested(opts.get('bundlecaps')): | ||
|
r36241 | if proto.name == 'http-v1': | ||
|
r37309 | return wireprototypes.ooberror(bundle2required) | ||
|
r30912 | raise error.Abort(bundle2requiredmain, | ||
hint=bundle2requiredhint) | ||||
|
r27246 | |||
|
r35805 | prefercompressed = True | ||
|
r35800 | |||
|
r30914 | try: | ||
|
r37516 | clheads = set(repo.changelog.heads()) | ||
heads = set(opts.get('heads', set())) | ||||
common = set(opts.get('common', set())) | ||||
common.discard(nullid) | ||||
if (repo.ui.configbool('server', 'pullbundle') and | ||||
'partial-pull' in proto.getprotocaps()): | ||||
# Check if a pre-built bundle covers this request. | ||||
bundle = find_pullbundle(repo, proto, opts, clheads, heads, common) | ||||
if bundle: | ||||
return wireprototypes.streamres(gen=util.filechunkiter(bundle), | ||||
prefer_uncompressed=True) | ||||
r33220 | if repo.ui.configbool('server', 'disablefullbundle'): | |||
|
r32260 | # Check to see if this is a full clone. | ||
|
r35778 | changegroup = opts.get('cg', True) | ||
if changegroup and not common and clheads == heads: | ||||
|
r32260 | raise error.Abort( | ||
_('server has pull-based clones disabled'), | ||||
hint=_('remove --pull if specified or upgrade Mercurial')) | ||||
|
r35803 | info, chunks = exchange.getbundlechunks(repo, 'serve', | ||
**pycompat.strkwargs(opts)) | ||||
|
r35805 | prefercompressed = info.get('prefercompressed', True) | ||
|
r30914 | except error.Abort as exc: | ||
# cleanly forward Abort error to the client | ||||
if not exchange.bundle2requested(opts.get('bundlecaps')): | ||||
|
r36241 | if proto.name == 'http-v1': | ||
|
r37309 | return wireprototypes.ooberror(pycompat.bytestr(exc) + '\n') | ||
|
r30914 | raise # cannot do better for bundle1 + ssh | ||
# bundle2 request expect a bundle2 reply | ||||
bundler = bundle2.bundle20(repo.ui) | ||||
|
r36272 | manargs = [('message', pycompat.bytestr(exc))] | ||
|
r30914 | advargs = [] | ||
if exc.hint is not None: | ||||
advargs.append(('hint', exc.hint)) | ||||
bundler.addpart(bundle2.bundlepart('error:abort', | ||||
manargs, advargs)) | ||||
|
r35800 | chunks = bundler.getchunks() | ||
|
r35805 | prefercompressed = False | ||
|
r35800 | |||
|
r37309 | return wireprototypes.streamres( | ||
gen=chunks, prefer_uncompressed=not prefercompressed) | ||||
|
r13741 | |||
|
r37503 | @wireprotocommand('heads', permission='pull', transportpolicy=POLICY_V1_ONLY) | ||
|
r11583 | def heads(repo, proto): | ||
|
r18281 | h = repo.heads() | ||
|
r37630 | return wireprototypes.bytesresponse(wireprototypes.encodelist(h) + '\n') | ||
|
r11581 | |||
|
r37507 | @wireprotocommand('hello', permission='pull', transportpolicy=POLICY_V1_ONLY) | ||
|
r11594 | def hello(repo, proto): | ||
|
r36239 | """Called as part of SSH handshake to obtain server info. | ||
Returns a list of lines describing interesting things about the | ||||
server, in an RFC822-like format. | ||||
|
r11594 | |||
|
r36239 | Currently, the only one defined is ``capabilities``, which consists of a | ||
line of space separated tokens describing server abilities: | ||||
capabilities: <token0> <token1> <token2> | ||||
""" | ||||
|
r36091 | caps = capabilities(repo, proto).data | ||
|
r37309 | return wireprototypes.bytesresponse('capabilities: %s\n' % caps) | ||
|
r11594 | |||
|
r37505 | @wireprotocommand('listkeys', 'namespace', permission='pull', | ||
transportpolicy=POLICY_V1_ONLY) | ||||
|
r11583 | def listkeys(repo, proto, namespace): | ||
|
r36546 | d = sorted(repo.listkeys(encoding.tolocal(namespace)).items()) | ||
|
r37309 | return wireprototypes.bytesresponse(pushkeymod.encodekeys(d)) | ||
|
r11581 | |||
|
r37556 | @wireprotocommand('lookup', 'key', permission='pull', | ||
transportpolicy=POLICY_V1_ONLY) | ||||
|
r11583 | def lookup(repo, proto, key): | ||
|
r11581 | try: | ||
|
r15925 | k = encoding.tolocal(key) | ||
|
r37371 | n = repo.lookup(k) | ||
r = hex(n) | ||||
|
r11581 | success = 1 | ||
|
r25660 | except Exception as inst: | ||
|
r37102 | r = stringutil.forcebytestr(inst) | ||
|
r11581 | success = 0 | ||
|
r37309 | return wireprototypes.bytesresponse('%d %s\n' % (success, r)) | ||
|
r11581 | |||
|
r37504 | @wireprotocommand('known', 'nodes *', permission='pull', | ||
transportpolicy=POLICY_V1_ONLY) | ||||
|
r14436 | def known(repo, proto, nodes, others): | ||
|
r37630 | v = ''.join(b and '1' or '0' | ||
for b in repo.known(wireprototypes.decodelist(nodes))) | ||||
|
r37309 | return wireprototypes.bytesresponse(v) | ||
|
r13723 | |||
|
r37411 | @wireprotocommand('protocaps', 'caps', permission='pull', | ||
transportpolicy=POLICY_V1_ONLY) | ||||
def protocaps(repo, proto, caps): | ||||
if proto.name == wireprototypes.SSHV1: | ||||
proto._protocaps = set(caps.split(' ')) | ||||
return wireprototypes.bytesresponse('OK') | ||||
|
r37555 | @wireprotocommand('pushkey', 'namespace key old new', permission='push', | ||
transportpolicy=POLICY_V1_ONLY) | ||||
|
r11583 | def pushkey(repo, proto, namespace, key, old, new): | ||
|
r13050 | # compatibility with pre-1.8 clients which were accidentally | ||
# sending raw binary nodes rather than utf-8-encoded hex | ||||
|
r37102 | if len(new) == 20 and stringutil.escapestr(new) != new: | ||
|
r13050 | # looks like it could be a binary node | ||
try: | ||||
|
r14064 | new.decode('utf-8') | ||
|
r13050 | new = encoding.tolocal(new) # but cleanly decodes as UTF-8 | ||
except UnicodeDecodeError: | ||||
pass # binary, leave unmodified | ||||
else: | ||||
new = encoding.tolocal(new) # normal path | ||||
|
r36083 | with proto.mayberedirectstdio() as output: | ||
|
r35997 | r = repo.pushkey(encoding.tolocal(namespace), encoding.tolocal(key), | ||
encoding.tolocal(old), new) or False | ||||
|
r17793 | |||
|
r36083 | output = output.getvalue() if output else '' | ||
|
r37309 | return wireprototypes.bytesresponse('%d\n%s' % (int(r), output)) | ||
|
r11581 | |||
|
r37552 | @wireprotocommand('stream_out', permission='pull', | ||
transportpolicy=POLICY_V1_ONLY) | ||||
|
r11585 | def stream(repo, proto): | ||
|
r11627 | '''If the server supports streaming clone, it advertises the "stream" | ||
capability with a value representing the version and flags of the repo | ||||
it is serving. Client checks to see if it understands the format. | ||||
''' | ||||
|
r37309 | return wireprototypes.streamreslegacy( | ||
streamclone.generatev1wireproto(repo)) | ||||
|
r11585 | |||
|
r37557 | @wireprotocommand('unbundle', 'heads', permission='push', | ||
transportpolicy=POLICY_V1_ONLY) | ||||
|
r11593 | def unbundle(repo, proto, heads): | ||
|
r37630 | their_heads = wireprototypes.decodelist(heads) | ||
|
r11593 | |||
|
r36084 | with proto.mayberedirectstdio() as output: | ||
|
r11593 | try: | ||
|
r36084 | exchange.check_heads(repo, their_heads, 'preparing changes') | ||
|
r37432 | cleanup = lambda: None | ||
try: | ||||
payload = proto.getpayload() | ||||
if repo.ui.configbool('server', 'streamunbundle'): | ||||
def cleanup(): | ||||
# Ensure that the full payload is consumed, so | ||||
# that the connection doesn't contain trailing garbage. | ||||
for p in payload: | ||||
pass | ||||
fp = util.chunkbuffer(payload) | ||||
else: | ||||
# write bundle data to temporary file as it can be big | ||||
fp, tempname = None, None | ||||
def cleanup(): | ||||
if fp: | ||||
fp.close() | ||||
if tempname: | ||||
os.unlink(tempname) | ||||
fd, tempname = tempfile.mkstemp(prefix='hg-unbundle-') | ||||
repo.ui.debug('redirecting incoming bundle to %s\n' % | ||||
tempname) | ||||
fp = os.fdopen(fd, pycompat.sysstr('wb+')) | ||||
r = 0 | ||||
for p in payload: | ||||
fp.write(p) | ||||
fp.seek(0) | ||||
|
r27246 | |||
|
r36084 | gen = exchange.readbundle(repo.ui, fp, None) | ||
if (isinstance(gen, changegroupmod.cg1unpacker) | ||||
and not bundle1allowed(repo, 'push')): | ||||
|
r36241 | if proto.name == 'http-v1': | ||
|
r36084 | # need to special case http because stderr do not get to | ||
# the http client on failed push so we need to abuse | ||||
# some other error type to make sure the message get to | ||||
# the user. | ||||
|
r37309 | return wireprototypes.ooberror(bundle2required) | ||
|
r36084 | raise error.Abort(bundle2requiredmain, | ||
hint=bundle2requiredhint) | ||||
|
r24796 | |||
|
r36084 | r = exchange.unbundle(repo, gen, their_heads, 'serve', | ||
|
r36086 | proto.client()) | ||
|
r36084 | if util.safehasattr(r, 'addpart'): | ||
# The return looks streamable, we are in the bundle2 case | ||||
# and should return a stream. | ||||
|
r37309 | return wireprototypes.streamreslegacy(gen=r.getchunks()) | ||
return wireprototypes.pushres( | ||||
r, output.getvalue() if output else '') | ||||
|
r36084 | |||
finally: | ||||
|
r37432 | cleanup() | ||
|
r36084 | |||
except (error.BundleValueError, error.Abort, error.PushRaced) as exc: | ||||
# handle non-bundle2 case first | ||||
if not getattr(exc, 'duringunbundle2', False): | ||||
try: | ||||
|
r25493 | raise | ||
|
r36084 | except error.Abort: | ||
|
r37137 | # The old code we moved used procutil.stderr directly. | ||
|
r36084 | # We did not change it to minimise code change. | ||
# This need to be moved to something proper. | ||||
# Feel free to do it. | ||||
|
r37137 | procutil.stderr.write("abort: %s\n" % exc) | ||
|
r36084 | if exc.hint is not None: | ||
|
r37137 | procutil.stderr.write("(%s)\n" % exc.hint) | ||
procutil.stderr.flush() | ||||
|
r37309 | return wireprototypes.pushres( | ||
0, output.getvalue() if output else '') | ||||
|
r36084 | except error.PushRaced: | ||
|
r37309 | return wireprototypes.pusherr( | ||
pycompat.bytestr(exc), | ||||
output.getvalue() if output else '') | ||||
|
r36084 | |||
bundler = bundle2.bundle20(repo.ui) | ||||
for out in getattr(exc, '_bundle2salvagedoutput', ()): | ||||
bundler.addpart(out) | ||||
try: | ||||
try: | ||||
raise | ||||
except error.PushkeyFailed as exc: | ||||
# check client caps | ||||
remotecaps = getattr(exc, '_replycaps', None) | ||||
if (remotecaps is not None | ||||
and 'pushkey' not in remotecaps.get('error', ())): | ||||
# no support remote side, fallback to Abort handler. | ||||
raise | ||||
part = bundler.newpart('error:pushkey') | ||||
part.addparam('in-reply-to', exc.partid) | ||||
if exc.namespace is not None: | ||||
part.addparam('namespace', exc.namespace, | ||||
mandatory=False) | ||||
if exc.key is not None: | ||||
part.addparam('key', exc.key, mandatory=False) | ||||
if exc.new is not None: | ||||
part.addparam('new', exc.new, mandatory=False) | ||||
if exc.old is not None: | ||||
part.addparam('old', exc.old, mandatory=False) | ||||
if exc.ret is not None: | ||||
part.addparam('ret', exc.ret, mandatory=False) | ||||
except error.BundleValueError as exc: | ||||
errpart = bundler.newpart('error:unsupportedcontent') | ||||
if exc.parttype is not None: | ||||
errpart.addparam('parttype', exc.parttype) | ||||
if exc.params: | ||||
errpart.addparam('params', '\0'.join(exc.params)) | ||||
except error.Abort as exc: | ||||
|
r37102 | manargs = [('message', stringutil.forcebytestr(exc))] | ||
|
r36084 | advargs = [] | ||
if exc.hint is not None: | ||||
advargs.append(('hint', exc.hint)) | ||||
bundler.addpart(bundle2.bundlepart('error:abort', | ||||
manargs, advargs)) | ||||
except error.PushRaced as exc: | ||||
|
r36332 | bundler.newpart('error:pushraced', | ||
|
r37102 | [('message', stringutil.forcebytestr(exc))]) | ||
|
r37309 | return wireprototypes.streamreslegacy(gen=bundler.getchunks()) | ||