comparison mercurial/wireprotov2server.py @ 43077:687b865b95ad

formatting: byteify all mercurial/ and hgext/ string literals

Done with

    python3.7 contrib/byteify-strings.py -i $(hg files 'set:mercurial/**.py - mercurial/thirdparty/** + hgext/**.py - hgext/fsmonitor/pywatchman/** - mercurial/__init__.py')
    black -l 80 -t py33 -S $(hg files 'set:**.py - mercurial/thirdparty/** - "contrib/python-zstandard/**" - hgext/fsmonitor/pywatchman/**')

# skip-blame mass-reformatting only

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

43076:2372284d9457

    # Root URL does nothing meaningful... yet.
    if not urlparts:
        res.status = b'200 OK'
        res.headers[b'Content-Type'] = b'text/plain'
        res.setbodybytes(_('HTTP version 2 API handler'))
        return

    if len(urlparts) == 1:
        res.status = b'404 Not Found'
        res.headers[b'Content-Type'] = b'text/plain'
        res.setbodybytes(
            _('do not know how to process %s\n') % req.dispatchpath
        )
        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'
        res.setbodybytes(_('unknown permission: %s') % permission)
        return

    if req.method != 'POST':
        res.status = b'405 Method Not Allowed'
        res.headers[b'Allow'] = b'POST'
        res.setbodybytes(_('commands require POST requests'))
        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:
        checkperm(rctx, req, 'pull' if permission == b'ro' else 'push')
    except hgwebcommon.ErrorResponse as e:
        res.status = hgwebcommon.statusmessage(e.code, pycompat.bytestr(e))
        for k, v in e.headers:
            res.headers[k] = v
        res.setbodybytes('permission denied')
        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.
    extracommands = {'multirequest'}

    if command not in COMMANDS and command not in extracommands:
        res.status = b'404 Not Found'
        res.headers[b'Content-Type'] = b'text/plain'
        res.setbodybytes(_('unknown wire protocol command: %s\n') % command)
        return

    repo = rctx.repo
    ui = repo.ui

        not COMMANDS.commandavailable(command, proto)
        and command not in extracommands
    ):
        res.status = b'404 Not Found'
        res.headers[b'Content-Type'] = b'text/plain'
        res.setbodybytes(_('invalid wire protocol command: %s') % command)
        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'
        res.setbodybytes(
            _('client MUST specify Accept header with value: %s\n')
            % FRAMINGTYPE
        )
        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'
        res.setbodybytes(
            _('client MUST send Content-Type header with ' 'value: %s\n')
            % FRAMINGTYPE
        )
        return

    _processhttpv2request(ui, repo, req, res, permission, command, proto)

    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.
    if not ui.configbool('experimental', 'web.api.debugreflect'):
        res.status = b'404 Not Found'
        res.headers[b'Content-Type'] = b'text/plain'
        res.setbodybytes(_('debugreflect service not available'))
        return

    # We assume we have a unified framing protocol request body.

    reactor = wireprotoframing.serverreactor(ui)

        action, meta = reactor.onframerecv(frame)
        states.append(templatefilters.json((action, meta)))

    action, meta = reactor.oninputeof()
    meta['action'] = action
    states.append(templatefilters.json(meta))

    res.status = b'200 OK'
    res.headers[b'Content-Type'] = b'text/plain'
    res.setbodybytes(b'\n'.join(states))

        if not frame:
            break

        action, meta = reactor.onframerecv(frame)

        if action == 'wantframe':
            # Need more data before we can do anything.
            continue
        elif action == 'runcommand':
            # 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()

            if sentoutput:
                return

            seencommand = True

        elif action == 'error':
            # TODO define proper error mechanism.
            res.status = b'200 OK'
            res.headers[b'Content-Type'] = b'text/plain'
            res.setbodybytes(meta['message'] + b'\n')
            return
        else:
            raise error.ProgrammingError(
                'unhandled action from frame processor: %s' % action
            )

    action, meta = reactor.oninputeof()
    if action == 'sendframes':
        # 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
        res.setbodygen(meta['framegen'])
    elif action == 'noop':
        pass
    else:
        raise error.ProgrammingError(
            'unhandled action from frame processor: %s' % action
        )


def _httpv2runcommand(
    ui,

    # 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.

    proto = httpv2protocolhandler(req, ui, args=command['args'])

    if reqcommand == b'multirequest':
        if not COMMANDS.commandavailable(command['command'], proto):
            # TODO proper error mechanism
            res.status = b'200 OK'
            res.headers[b'Content-Type'] = b'text/plain'
            res.setbodybytes(
                _('wire protocol command not available: %s')
                % command['command']
            )
            return True

        # TODO don't use assert here, since it may be elided by -O.
        assert authedperm in (b'ro', b'rw')
        wirecommand = COMMANDS[command['command']]
        assert wirecommand.permission in ('push', 'pull')

        if authedperm == b'ro' and wirecommand.permission != 'pull':
            # TODO proper error mechanism
            res.status = b'403 Forbidden'
            res.headers[b'Content-Type'] = b'text/plain'
            res.setbodybytes(
                _('insufficient permissions to execute ' 'command: %s')
                % command['command']
            )
            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

        if issubsequent:
            # TODO proper error mechanism
            res.status = b'200 OK'
            res.headers[b'Content-Type'] = b'text/plain'
            res.setbodybytes(
                _('multiple commands cannot be issued to this ' 'URL')
            )
            return True

        if reqcommand != command['command']:
            # TODO define proper error mechanism
            res.status = b'200 OK'
            res.headers[b'Content-Type'] = b'text/plain'
            res.setbodybytes(_('command in frame must match command in URL'))
            return True

    res.status = b'200 OK'
    res.headers[b'Content-Type'] = FRAMINGTYPE

    try:
        objs = dispatch(repo, proto, command['command'], command['redirect'])

        action, meta = reactor.oncommandresponsereadyobjects(
            outstream, command['requestid'], objs
        )

    except error.WireprotoCommandError as e:
        action, meta = reactor.oncommanderror(
            outstream, command['requestid'], e.message, e.messageargs
        )

    except Exception as e:
        action, meta = reactor.onservererror(
            outstream,
            command['requestid'],
            _('exception when invoking command: %s')
            % stringutil.forcebytestr(e),
        )

    if action == 'sendframes':
        res.setbodygen(meta['framegen'])
        return True
    elif action == 'noop':
        return False
    else:
        raise error.ProgrammingError(
            'unhandled event from reactor: %s' % action
        )


def getdispatchrepo(repo, proto, command):
    viewconfig = repo.ui.config('server', 'view')
    return repo.filtered(viewconfig)


def dispatch(repo, proto, command, redirect):
    """Run a wire protocol command.

        # Serve it from the cache, if possible.
        cached = cacher.lookup()

        if cached:
            for o in cached['objs']:
                yield o
            return

        # Else call the command and feed its output into the cacher, allowing
        # the cacher to buffer/mutate objects as it desires.

        # 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(
                'unsupported argument to command: %s' % ', '.join(sorted(extra))
            )

        # And look for required arguments that are missing.
        missing = {a for a in args if args[a]['required']} - set(self._args)

        if missing:
            raise error.WireprotoCommandError(
                'missing required arguments: %s' % ', '.join(sorted(missing))
            )

        # Now derive the arguments to pass to the command, taking into
        # account the arguments specified by the client.
        data = {}
        for k, meta in sorted(args.items()):
            # This argument wasn't passed by the client.
            if k not in self._args:
                data[k] = meta['default']()
                continue

            v = self._args[k]

            # Sets may be expressed as lists. Silently normalize.
            if meta['type'] == 'set' and isinstance(v, list):
                v = set(v)

            # TODO consider more/stronger type validation.

            data[k] = v

    These capabilities are distinct from the capabilities for version 1
    transports.
    """
    caps = {
        'commands': {},
        'framingmediatypes': [FRAMINGTYPE],
        'pathfilterprefixes': set(narrowspec.VALID_PREFIXES),
    }

    for command, entry in COMMANDS.items():
        args = {}

        for arg, meta in entry.args.items():
            args[arg] = {
                # TODO should this be a normalized type using CBOR's
                # terminology?
                b'type': meta['type'],
                b'required': meta['required'],
            }

            if not meta['required']:
                args[arg][b'default'] = meta['default']()

            if meta['validvalues']:
                args[arg][b'validvalues'] = meta['validvalues']

        # TODO this type of check should be defined in a per-command callback.
        if (
            command == b'rawstorefiledata'
            and not streamclone.allowservergeneration(repo)
        ):
            continue

        caps['commands'][command] = {
            'args': args,
            'permissions': [entry.permission],
        }

        if entry.extracapabilitiesfn:
            extracaps = entry.extracapabilitiesfn(repo, proto)
            caps['commands'][command].update(extracaps)

    caps['rawrepoformats'] = sorted(repo.requirements & repo.supportedformats)

    targets = getadvertisedredirecttargets(repo, proto)
    if targets:
        caps[b'redirect'] = {
            b'targets': [],
            b'hashes': [b'sha256', b'sha1'],
        }

        for target in targets:
            entry = {
                b'name': target['name'],
                b'protocol': target['protocol'],
                b'uris': target['uris'],
            }

            for key in ('snirequired', 'tlsversions'):
                if key in target:
                    entry[key] = target[key]

            caps[b'redirect'][b'targets'].append(entry)

def wireprotocommand(
    name,
    args=None,
    permission='push',
    cachekeyfn=None,
    extracapabilitiesfn=None,
):
    """Decorator to declare a wire protocol command.

    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.
    """
    transports = {
        k for k, v in wireprototypes.TRANSPORTS.items() if v['version'] == 2
    }

    if permission not in ('push', 'pull'):
        raise error.ProgrammingError(
            'invalid wire protocol permission; '
            'got %s; expected "push" or "pull"' % permission
        )

    if args is None:
        args = {}

    if not isinstance(args, dict):
        raise error.ProgrammingError(
            'arguments for version 2 commands ' 'must be declared as dicts'
        )

    for arg, meta in args.items():
        if arg == '*':
            raise error.ProgrammingError(
                '* argument name not allowed on ' 'version 2 commands'
            )

        if not isinstance(meta, dict):
            raise error.ProgrammingError(
                'arguments for version 2 commands '
                'must declare metadata as a dict'
            )

        if 'type' not in meta:
            raise error.ProgrammingError(
                '%s argument for command %s does not '
                'declare type field' % (arg, name)
            )

        if meta['type'] not in ('bytes', 'int', 'list', 'dict', 'set', 'bool'):
            raise error.ProgrammingError(
                '%s argument for command %s has '
                'illegal type: %s' % (arg, name, meta['type'])
            )

        if 'example' not in meta:
            raise error.ProgrammingError(
                '%s argument for command %s does not '
                'declare example field' % (arg, name)
            )

        meta['required'] = 'default' not in meta

        meta.setdefault('default', lambda: None)
        meta.setdefault('validvalues', None)

    def register(func):
        if name in COMMANDS:
            raise error.ProgrammingError(
                '%s command already registered ' 'for version 2' % name
            )

        COMMANDS[name] = wireprototypes.commandentry(
            func,
            args=args,

    * The media type used.
    * Wire protocol version string.
    * The repository path.
    """
    if not allargs:
        raise error.ProgrammingError('only allargs=True is currently supported')

    if localversion is None:
        raise error.ProgrammingError('must set localversion argument value')

    def cachekeyfn(repo, proto, cacher, **args):
        spec = COMMANDS[command]

        # Commands that mutate the repo can not be cached.
        if spec.permission == 'push':
            return None

        # TODO config option to disable caching.

        # Our key derivation strategy is to construct a data structure

    seen = set()
    nodes = []

    if not isinstance(revisions, list):
        raise error.WireprotoCommandError(
            'revisions must be defined as an ' 'array'
        )

    for spec in revisions:
        if b'type' not in spec:
            raise error.WireprotoCommandError(
                'type key not present in revision specifier'
            )

        typ = spec[b'type']

        if typ == b'changesetexplicit':
            if b'nodes' not in spec:
                raise error.WireprotoCommandError(
                    'nodes key not present in changesetexplicit revision '
                    'specifier'
                )

            for node in spec[b'nodes']:
                if node not in seen:
                    nodes.append(node)

        elif typ == b'changesetexplicitdepth':
            for key in (b'nodes', b'depth'):
                if key not in spec:
                    raise error.WireprotoCommandError(
                        '%s key not present in changesetexplicitdepth revision '
                        'specifier',
                        (key,),
                    )

            for rev in repo.revs(
                b'ancestors(%ln, %s)', spec[b'nodes'], spec[b'depth'] - 1

        elif typ == b'changesetdagrange':
            for key in (b'roots', b'heads'):
                if key not in spec:
                    raise error.WireprotoCommandError(
                        '%s key not present in changesetdagrange revision '
                        'specifier',
                        (key,),
                    )

            if not spec[b'heads']:
                raise error.WireprotoCommandError(
                    'heads key in changesetdagrange cannot be empty'
                )

            if spec[b'roots']:
                common = [n for n in spec[b'roots'] if clhasnode(n)]
            else:

                    nodes.append(n)
                    seen.add(n)

        else:
            raise error.WireprotoCommandError(
                'unknown revision specifier type: %s', (typ,)
            )

    return nodes


@wireprotocommand('branchmap', permission='pull')
def branchmapv2(repo, proto):
    yield {encoding.fromlocal(k): v for k, v in repo.branchmap().iteritems()}


@wireprotocommand('capabilities', permission='pull')
def capabilitiesv2(repo, proto):
    yield _capabilitiesv2(repo, proto)


@wireprotocommand(
    'changesetdata',
    args={
        'revisions': {
            'type': 'list',
            'example': [
                {b'type': b'changesetexplicit', b'nodes': [b'abcdef...'],}
            ],
        },
        'fields': {
            'type': 'set',
            'default': set,
            'example': {b'parents', b'revision'},
            'validvalues': {b'bookmarks', b'parents', b'phase', b'revision'},
        },
    },
    permission='pull',
)
def changesetdata(repo, proto, revisions, fields):
    # TODO look for unknown fields and abort when they can't be serviced.
    # This could probably be validated by dispatcher using validvalues.

    cl = repo.changelog
    outgoing = resolvenodes(repo, revisions)
    publishing = repo.publishing()

    if outgoing:
        repo.hook('preoutgoing', throw=True, source='serve')

    yield {
        b'totalitems': len(outgoing),
    }

    # 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):
        raise FileAccessError(path, 'unknown file: %s', (path,))

    return fl


def emitfilerevisions(repo, path, revisions, linknodes, fields):

    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(
                        '%s pattern must begin with `path:` or `rootfilesin:`; '
                        'got %s',
                        (key, pattern),
                    )

    if pathfilter:
        matcher = matchmod.match(

    # filter those out.
    return repo.narrowmatch(matcher)


@wireprotocommand(
    'filedata',
    args={
        'haveparents': {
            'type': 'bool',
            'default': lambda: False,
            'example': True,
        },
        'nodes': {'type': 'list', 'example': [b'0123456...'],},
        'fields': {
            'type': 'set',
            'default': set,
            'example': {b'parents', b'revision'},
            'validvalues': {b'parents', b'revision', b'linknode'},
        },
        'path': {'type': 'bytes', 'example': b'foo.txt',},
    },
    permission='pull',
    # TODO censoring a file revision won't invalidate the cache.
    # Figure out a way to take censoring into account when deriving
    # the cache key.
    cachekeyfn=makecommandcachekeyfn('filedata', 1, allargs=True),
)
def filedata(repo, proto, haveparents, nodes, fields, path):
    # 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?

    for node in nodes:
        try:
            store.rev(node)
        except error.LookupError:
            raise error.WireprotoCommandError(
                'unknown file node: %s', (hex(node),)
            )

        # 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

        b'recommendedbatchsize': batchsize,
    }


@wireprotocommand(
    'filesdata',
    args={
        'haveparents': {
            'type': 'bool',
            'default': lambda: False,
            'example': True,
        },
        'fields': {
            'type': 'set',
            'default': set,
            'example': {b'parents', b'revision'},
            'validvalues': {
                b'firstchangeset',
                b'linknode',
                b'parents',
                b'revision',
            },
        },
        'pathfilter': {
            'type': 'dict',
            'default': lambda: None,
            'example': {b'include': [b'path:tests']},
        },
        'revisions': {
            'type': 'list',
            'example': [
                {b'type': b'changesetexplicit', b'nodes': [b'abcdef...'],}
            ],
        },
    },
    permission='pull',
    # TODO censoring a file revision won't invalidate the cache.
    # Figure out a way to take censoring into account when deriving
    # the cache key.
    cachekeyfn=makecommandcachekeyfn('filesdata', 1, allargs=True),
    extracapabilitiesfn=filesdatacapabilities,
)
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

        for o in emitfilerevisions(repo, path, revisions, filenodes, fields):
            yield o


@wireprotocommand(
    'heads',
    args={
        'publiconly': {
            'type': 'bool',
            'default': lambda: False,
            'example': False,
        },
    },
    permission='pull',
)
def headsv2(repo, proto, publiconly):
    if publiconly:
        repo = repo.filtered('immutable')

    yield repo.heads()


@wireprotocommand(
    'known',
    args={
        'nodes': {'type': 'list', 'default': list, 'example': [b'deadbeef'],},
    },
    permission='pull',
)
def knownv2(repo, proto, nodes):
    result = b''.join(b'1' if n else b'0' for n in repo.known(nodes))
    yield result


@wireprotocommand(
    'listkeys',
    args={'namespace': {'type': 'bytes', 'example': b'ns',},},
    permission='pull',
)
def listkeysv2(repo, proto, namespace):
    keys = repo.listkeys(encoding.tolocal(namespace))
    keys = {
        encoding.fromlocal(k): encoding.fromlocal(v)

    yield keys


@wireprotocommand(
    'lookup',
    args={'key': {'type': 'bytes', 'example': b'foo',},},
    permission='pull',
)
def lookupv2(repo, proto, key):
    key = encoding.tolocal(key)

    # TODO handle exception.

        b'recommendedbatchsize': batchsize,
    }


@wireprotocommand(
    'manifestdata',
    args={
        'nodes': {'type': 'list', 'example': [b'0123456...'],},
        'haveparents': {
            'type': 'bool',
            'default': lambda: False,
            'example': True,
        },
        'fields': {
            'type': 'set',
            'default': set,
            'example': {b'parents', b'revision'},
            'validvalues': {b'parents', b'revision'},
        },
        'tree': {'type': 'bytes', 'example': b'',},
    },
    permission='pull',
    cachekeyfn=makecommandcachekeyfn('manifestdata', 1, allargs=True),
    extracapabilitiesfn=manifestdatacapabilities,
)
def manifestdata(repo, proto, haveparents, nodes, fields, tree):
    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:
            raise error.WireprotoCommandError('unknown node: %s', (node,))

    revisions = store.emitrevisions(
        nodes,
        revisiondata=b'revision' in fields,
        assumehaveparentrevisions=haveparents,

        for extra in followingdata:
            yield extra


@wireprotocommand(
    'pushkey',
    args={
        'namespace': {'type': 'bytes', 'example': b'ns',},
        'key': {'type': 'bytes', 'example': b'key',},
        'old': {'type': 'bytes', 'example': b'old',},
        'new': {'type': 'bytes', 'example': 'new',},
    },
    permission='push',
)
def pushkeyv2(repo, proto, namespace, key, old, new):
    # TODO handle ui output redirection
    yield repo.pushkey(
        encoding.tolocal(namespace),

        encoding.tolocal(new),
    )


@wireprotocommand(
    'rawstorefiledata',
    args={
        'files': {'type': 'list', 'example': [b'changelog', b'manifestlog'],},
        'pathfilter': {
            'type': 'list',
            'default': lambda: None,
            'example': {b'include': [b'path:tests']},
        },
    },
    permission='pull',
)
def rawstorefiledata(repo, proto, files, pathfilter):
    if not streamclone.allowservergeneration(repo):
        raise error.WireprotoCommandError(b'stream clone is disabled')

        }

        # We have to use a closure for this to ensure the context manager is
        # closed only after sending the final chunk.
        def getfiledata():
            with repo.svfs(name, 'rb', auditpath=False) as fh:
                for chunk in util.filechunkiter(fh, limit=size):
                    yield chunk

        yield wireprototypes.indefinitebytestringresponse(getfiledata())

43077:687b865b95ad

    # Root URL does nothing meaningful... yet.
    if not urlparts:
        res.status = b'200 OK'
        res.headers[b'Content-Type'] = b'text/plain'
        res.setbodybytes(_(b'HTTP version 2 API handler'))
        return

    if len(urlparts) == 1:
        res.status = b'404 Not Found'
        res.headers[b'Content-Type'] = b'text/plain'
        res.setbodybytes(
            _(b'do not know how to process %s\n') % req.dispatchpath
        )
        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'
        res.setbodybytes(_(b'unknown permission: %s') % permission)
        return

    if req.method != b'POST':
        res.status = b'405 Method Not Allowed'
        res.headers[b'Allow'] = b'POST'
        res.setbodybytes(_(b'commands require POST requests'))
        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:
        checkperm(rctx, req, b'pull' if permission == b'ro' else b'push')
    except hgwebcommon.ErrorResponse as e:
        res.status = hgwebcommon.statusmessage(e.code, pycompat.bytestr(e))
        for k, v in e.headers:
            res.headers[k] = v
        res.setbodybytes(b'permission denied')
        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.
    extracommands = {b'multirequest'}

    if command not in COMMANDS and command not in extracommands:
        res.status = b'404 Not Found'
        res.headers[b'Content-Type'] = b'text/plain'
        res.setbodybytes(_(b'unknown wire protocol command: %s\n') % command)
        return

    repo = rctx.repo
    ui = repo.ui

        not COMMANDS.commandavailable(command, proto)
        and command not in extracommands
    ):
        res.status = b'404 Not Found'
        res.headers[b'Content-Type'] = b'text/plain'
        res.setbodybytes(_(b'invalid wire protocol command: %s') % command)
        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'
        res.setbodybytes(
            _(b'client MUST specify Accept header with value: %s\n')
            % FRAMINGTYPE
        )
        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'
        res.setbodybytes(
            _(b'client MUST send Content-Type header with ' b'value: %s\n')
            % FRAMINGTYPE
        )
        return

    _processhttpv2request(ui, repo, req, res, permission, command, proto)

    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.
    if not ui.configbool(b'experimental', b'web.api.debugreflect'):
        res.status = b'404 Not Found'
        res.headers[b'Content-Type'] = b'text/plain'
        res.setbodybytes(_(b'debugreflect service not available'))
        return

    # We assume we have a unified framing protocol request body.

    reactor = wireprotoframing.serverreactor(ui)

        action, meta = reactor.onframerecv(frame)
        states.append(templatefilters.json((action, meta)))

    action, meta = reactor.oninputeof()
    meta[b'action'] = action
    states.append(templatefilters.json(meta))

    res.status = b'200 OK'
    res.headers[b'Content-Type'] = b'text/plain'
    res.setbodybytes(b'\n'.join(states))

        if not frame:
            break

        action, meta = reactor.onframerecv(frame)

        if action == b'wantframe':
            # Need more data before we can do anything.
            continue
        elif action == b'runcommand':
            # 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()

            if sentoutput:
                return

            seencommand = True

        elif action == b'error':
            # TODO define proper error mechanism.
            res.status = b'200 OK'
            res.headers[b'Content-Type'] = b'text/plain'
            res.setbodybytes(meta[b'message'] + b'\n')
            return
        else:
            raise error.ProgrammingError(
                b'unhandled action from frame processor: %s' % action
            )

    action, meta = reactor.oninputeof()
    if action == b'sendframes':
        # 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
        res.setbodygen(meta[b'framegen'])
    elif action == b'noop':
        pass
    else:
        raise error.ProgrammingError(
            b'unhandled action from frame processor: %s' % action
        )


def _httpv2runcommand(
    ui,

    # 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.

    proto = httpv2protocolhandler(req, ui, args=command[b'args'])

    if reqcommand == b'multirequest':
        if not COMMANDS.commandavailable(command[b'command'], proto):
            # TODO proper error mechanism
            res.status = b'200 OK'
            res.headers[b'Content-Type'] = b'text/plain'
            res.setbodybytes(
                _(b'wire protocol command not available: %s')
                % command[b'command']
            )
            return True

        # TODO don't use assert here, since it may be elided by -O.
        assert authedperm in (b'ro', b'rw')
        wirecommand = COMMANDS[command[b'command']]
        assert wirecommand.permission in (b'push', b'pull')

        if authedperm == b'ro' and wirecommand.permission != b'pull':
            # TODO proper error mechanism
            res.status = b'403 Forbidden'
            res.headers[b'Content-Type'] = b'text/plain'
            res.setbodybytes(
                _(b'insufficient permissions to execute ' b'command: %s')
                % command[b'command']
            )
            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

        if issubsequent:
            # TODO proper error mechanism
            res.status = b'200 OK'
            res.headers[b'Content-Type'] = b'text/plain'
            res.setbodybytes(
                _(b'multiple commands cannot be issued to this ' b'URL')
            )
            return True

        if reqcommand != command[b'command']:
            # TODO define proper error mechanism
            res.status = b'200 OK'
            res.headers[b'Content-Type'] = b'text/plain'
            res.setbodybytes(_(b'command in frame must match command in URL'))
            return True

    res.status = b'200 OK'
    res.headers[b'Content-Type'] = FRAMINGTYPE

    try:
        objs = dispatch(repo, proto, command[b'command'], command[b'redirect'])

        action, meta = reactor.oncommandresponsereadyobjects(
            outstream, command[b'requestid'], objs
        )

    except error.WireprotoCommandError as e:
        action, meta = reactor.oncommanderror(
            outstream, command[b'requestid'], e.message, e.messageargs
        )

    except Exception as e:
        action, meta = reactor.onservererror(
            outstream,
            command[b'requestid'],
            _(b'exception when invoking command: %s')
            % stringutil.forcebytestr(e),
        )

    if action == b'sendframes':
        res.setbodygen(meta[b'framegen'])
        return True
    elif action == b'noop':
        return False
    else:
        raise error.ProgrammingError(
            b'unhandled event from reactor: %s' % action
        )


def getdispatchrepo(repo, proto, command):
    viewconfig = repo.ui.config(b'server', b'view')
    return repo.filtered(viewconfig)


def dispatch(repo, proto, command, redirect):
    """Run a wire protocol command.

        # Serve it from the cache, if possible.
        cached = cacher.lookup()

        if cached:
            for o in cached[b'objs']:
                yield o
            return

        # Else call the command and feed its output into the cacher, allowing
        # the cacher to buffer/mutate objects as it desires.

        # 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(
                b'unsupported argument to command: %s'
                % b', '.join(sorted(extra))
            )

        # And look for required arguments that are missing.
        missing = {a for a in args if args[a][b'required']} - set(self._args)

        if missing:
            raise error.WireprotoCommandError(
                b'missing required arguments: %s' % b', '.join(sorted(missing))
            )

        # Now derive the arguments to pass to the command, taking into
        # account the arguments specified by the client.
        data = {}
        for k, meta in sorted(args.items()):
            # This argument wasn't passed by the client.
            if k not in self._args:
                data[k] = meta[b'default']()
                continue

            v = self._args[k]

            # Sets may be expressed as lists. Silently normalize.
            if meta[b'type'] == b'set' and isinstance(v, list):
                v = set(v)

            # TODO consider more/stronger type validation.

            data[k] = v

    These capabilities are distinct from the capabilities for version 1
    transports.
    """
    caps = {
        b'commands': {},
        b'framingmediatypes': [FRAMINGTYPE],
        b'pathfilterprefixes': set(narrowspec.VALID_PREFIXES),
    }

    for command, entry in COMMANDS.items():
        args = {}

        for arg, meta in entry.args.items():
            args[arg] = {
                # TODO should this be a normalized type using CBOR's
                # terminology?
                b'type': meta[b'type'],
                b'required': meta[b'required'],
            }

            if not meta[b'required']:
                args[arg][b'default'] = meta[b'default']()

            if meta[b'validvalues']:
                args[arg][b'validvalues'] = meta[b'validvalues']

        # TODO this type of check should be defined in a per-command callback.
        if (
            command == b'rawstorefiledata'
            and not streamclone.allowservergeneration(repo)
        ):
            continue

        caps[b'commands'][command] = {
            b'args': args,
            b'permissions': [entry.permission],
        }

        if entry.extracapabilitiesfn:
            extracaps = entry.extracapabilitiesfn(repo, proto)
            caps[b'commands'][command].update(extracaps)

    caps[b'rawrepoformats'] = sorted(repo.requirements & repo.supportedformats)

    targets = getadvertisedredirecttargets(repo, proto)
    if targets:
        caps[b'redirect'] = {
            b'targets': [],
            b'hashes': [b'sha256', b'sha1'],
        }

        for target in targets:
            entry = {
                b'name': target[b'name'],
                b'protocol': target[b'protocol'],
                b'uris': target[b'uris'],
            }

            for key in (b'snirequired', b'tlsversions'):
                if key in target:
                    entry[key] = target[key]

            caps[b'redirect'][b'targets'].append(entry)

def wireprotocommand(
    name,
    args=None,
    permission=b'push',
    cachekeyfn=None,
    extracapabilitiesfn=None,
):
    """Decorator to declare a wire protocol command.

    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.
    """
    transports = {
        k for k, v in wireprototypes.TRANSPORTS.items() if v[b'version'] == 2
    }

    if permission not in (b'push', b'pull'):
        raise error.ProgrammingError(
            b'invalid wire protocol permission; '
            b'got %s; expected "push" or "pull"' % permission
        )

    if args is None:
        args = {}

    if not isinstance(args, dict):
        raise error.ProgrammingError(
            b'arguments for version 2 commands ' b'must be declared as dicts'
        )

    for arg, meta in args.items():
        if arg == b'*':
            raise error.ProgrammingError(
                b'* argument name not allowed on ' b'version 2 commands'
            )

        if not isinstance(meta, dict):
            raise error.ProgrammingError(
                b'arguments for version 2 commands '
                b'must declare metadata as a dict'
            )

        if b'type' not in meta:
            raise error.ProgrammingError(
                b'%s argument for command %s does not '
                b'declare type field' % (arg, name)
            )

        if meta[b'type'] not in (
            b'bytes',
            b'int',
            b'list',
            b'dict',
            b'set',
            b'bool',
        ):
            raise error.ProgrammingError(
                b'%s argument for command %s has '
                b'illegal type: %s' % (arg, name, meta[b'type'])
            )

        if b'example' not in meta:
            raise error.ProgrammingError(
                b'%s argument for command %s does not '
                b'declare example field' % (arg, name)
            )

        meta[b'required'] = b'default' not in meta

        meta.setdefault(b'default', lambda: None)
        meta.setdefault(b'validvalues', None)

    def register(func):
        if name in COMMANDS:
            raise error.ProgrammingError(
                b'%s command already registered ' b'for version 2' % name
            )

        COMMANDS[name] = wireprototypes.commandentry(
            func,
            args=args,

    * The media type used.
    * Wire protocol version string.
    * The repository path.
    """
    if not allargs:
        raise error.ProgrammingError(
            b'only allargs=True is currently supported'
        )

    if localversion is None:
        raise error.ProgrammingError(b'must set localversion argument value')

    def cachekeyfn(repo, proto, cacher, **args):
        spec = COMMANDS[command]

        # Commands that mutate the repo can not be cached.
        if spec.permission == b'push':
            return None

        # TODO config option to disable caching.

        # Our key derivation strategy is to construct a data structure

    seen = set()
    nodes = []

    if not isinstance(revisions, list):
        raise error.WireprotoCommandError(
            b'revisions must be defined as an ' b'array'
        )

    for spec in revisions:
        if b'type' not in spec:
            raise error.WireprotoCommandError(
                b'type key not present in revision specifier'
            )

        typ = spec[b'type']

        if typ == b'changesetexplicit':
            if b'nodes' not in spec:
                raise error.WireprotoCommandError(
                    b'nodes key not present in changesetexplicit revision '
                    b'specifier'
                )

            for node in spec[b'nodes']:
                if node not in seen:
                    nodes.append(node)

        elif typ == b'changesetexplicitdepth':
            for key in (b'nodes', b'depth'):
                if key not in spec:
                    raise error.WireprotoCommandError(
                        b'%s key not present in changesetexplicitdepth revision '
                        b'specifier',
                        (key,),
                    )

            for rev in repo.revs(
                b'ancestors(%ln, %s)', spec[b'nodes'], spec[b'depth'] - 1

        elif typ == b'changesetdagrange':
            for key in (b'roots', b'heads'):
                if key not in spec:
                    raise error.WireprotoCommandError(
                        b'%s key not present in changesetdagrange revision '
                        b'specifier',
                        (key,),
                    )

            if not spec[b'heads']:
                raise error.WireprotoCommandError(
                    b'heads key in changesetdagrange cannot be empty'
                )

            if spec[b'roots']:
                common = [n for n in spec[b'roots'] if clhasnode(n)]
            else:

                    nodes.append(n)
                    seen.add(n)

        else:
            raise error.WireprotoCommandError(
                b'unknown revision specifier type: %s', (typ,)
            )

    return nodes


@wireprotocommand(b'branchmap', permission=b'pull')
def branchmapv2(repo, proto):
    yield {encoding.fromlocal(k): v for k, v in repo.branchmap().iteritems()}


@wireprotocommand(b'capabilities', permission=b'pull')
def capabilitiesv2(repo, proto):
    yield _capabilitiesv2(repo, proto)


@wireprotocommand(
    b'changesetdata',
    args={
        b'revisions': {
            b'type': b'list',
            b'example': [
                {b'type': b'changesetexplicit', b'nodes': [b'abcdef...'],}
            ],
        },
        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'},
        },
    },
    permission=b'pull',
)
def changesetdata(repo, proto, revisions, fields):
    # TODO look for unknown fields and abort when they can't be serviced.
    # This could probably be validated by dispatcher using validvalues.

    cl = repo.changelog
    outgoing = resolvenodes(repo, revisions)
    publishing = repo.publishing()

    if outgoing:
        repo.hook(b'preoutgoing', throw=True, source=b'serve')

    yield {
        b'totalitems': len(outgoing),
    }

    # 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):
        raise FileAccessError(path, b'unknown file: %s', (path,))

    return fl


def emitfilerevisions(repo, path, revisions, linknodes, fields):

    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(
                        b'%s pattern must begin with `path:` or `rootfilesin:`; '
                        b'got %s',
                        (key, pattern),
                    )

    if pathfilter:
        matcher = matchmod.match(

    # filter those out.
    return repo.narrowmatch(matcher)


@wireprotocommand(
    b'filedata',
    args={
        b'haveparents': {
            b'type': b'bool',
            b'default': lambda: False,
            b'example': True,
        },
        b'nodes': {b'type': b'list', b'example': [b'0123456...'],},
        b'fields': {
            b'type': b'set',
            b'default': set,
            b'example': {b'parents', b'revision'},
            b'validvalues': {b'parents', b'revision', b'linknode'},
        },
        b'path': {b'type': b'bytes', b'example': b'foo.txt',},
    },
    permission=b'pull',
    # TODO censoring a file revision won't invalidate the cache.
    # Figure out a way to take censoring into account when deriving
    # the cache key.
    cachekeyfn=makecommandcachekeyfn(b'filedata', 1, allargs=True),
)
def filedata(repo, proto, haveparents, nodes, fields, path):
    # 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?

    for node in nodes:
        try:
            store.rev(node)
        except error.LookupError:
            raise error.WireprotoCommandError(
                b'unknown file node: %s', (hex(node),)
            )

        # 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

        b'recommendedbatchsize': batchsize,
    }


@wireprotocommand(
    b'filesdata',
    args={
        b'haveparents': {
            b'type': b'bool',
            b'default': lambda: False,
            b'example': True,
        },
        b'fields': {
            b'type': b'set',
            b'default': set,
            b'example': {b'parents', b'revision'},
            b'validvalues': {
                b'firstchangeset',
                b'linknode',
                b'parents',
                b'revision',
            },
        },
        b'pathfilter': {
            b'type': b'dict',
            b'default': lambda: None,
            b'example': {b'include': [b'path:tests']},
        },
        b'revisions': {
            b'type': b'list',
            b'example': [
                {b'type': b'changesetexplicit', b'nodes': [b'abcdef...'],}
            ],
        },
    },
    permission=b'pull',
    # TODO censoring a file revision won't invalidate the cache.
    # Figure out a way to take censoring into account when deriving
    # the cache key.
    cachekeyfn=makecommandcachekeyfn(b'filesdata', 1, allargs=True),
    extracapabilitiesfn=filesdatacapabilities,
)
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

        for o in emitfilerevisions(repo, path, revisions, filenodes, fields):
            yield o


@wireprotocommand(
    b'heads',
    args={
        b'publiconly': {
            b'type': b'bool',
            b'default': lambda: False,
            b'example': False,
        },
    },
    permission=b'pull',
)
def headsv2(repo, proto, publiconly):
    if publiconly:
        repo = repo.filtered(b'immutable')

    yield repo.heads()


@wireprotocommand(
    b'known',
    args={
        b'nodes': {
            b'type': b'list',
            b'default': list,
            b'example': [b'deadbeef'],
        },
    },
    permission=b'pull',
)
def knownv2(repo, proto, nodes):
    result = b''.join(b'1' if n else b'0' for n in repo.known(nodes))
    yield result


@wireprotocommand(
    b'listkeys',
    args={b'namespace': {b'type': b'bytes', b'example': b'ns',},},
    permission=b'pull',
)
def listkeysv2(repo, proto, namespace):
    keys = repo.listkeys(encoding.tolocal(namespace))
    keys = {
        encoding.fromlocal(k): encoding.fromlocal(v)

    yield keys


@wireprotocommand(
    b'lookup',
    args={b'key': {b'type': b'bytes', b'example': b'foo',},},
    permission=b'pull',
)
def lookupv2(repo, proto, key):
    key = encoding.tolocal(key)

    # TODO handle exception.

        b'recommendedbatchsize': batchsize,
    }


@wireprotocommand(
    b'manifestdata',
    args={
        b'nodes': {b'type': b'list', b'example': [b'0123456...'],},
        b'haveparents': {
            b'type': b'bool',
            b'default': lambda: False,
            b'example': True,
        },
        b'fields': {
            b'type': b'set',
            b'default': set,
            b'example': {b'parents', b'revision'},
            b'validvalues': {b'parents', b'revision'},
        },
        b'tree': {b'type': b'bytes', b'example': b'',},
    },
    permission=b'pull',
    cachekeyfn=makecommandcachekeyfn(b'manifestdata', 1, allargs=True),
    extracapabilitiesfn=manifestdatacapabilities,
)
def manifestdata(repo, proto, haveparents, nodes, fields, tree):
    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:
            raise error.WireprotoCommandError(b'unknown node: %s', (node,))

    revisions = store.emitrevisions(
        nodes,
        revisiondata=b'revision' in fields,
        assumehaveparentrevisions=haveparents,

        for extra in followingdata:
            yield extra


@wireprotocommand(
    b'pushkey',
    args={
        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',},
    },
    permission=b'push',
)
def pushkeyv2(repo, proto, namespace, key, old, new):
    # TODO handle ui output redirection
    yield repo.pushkey(
        encoding.tolocal(namespace),

        encoding.tolocal(new),
    )


@wireprotocommand(
    b'rawstorefiledata',
    args={
        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']},
        },
    },
    permission=b'pull',
)
def rawstorefiledata(repo, proto, files, pathfilter):
    if not streamclone.allowservergeneration(repo):
        raise error.WireprotoCommandError(b'stream clone is disabled')

        }

        # We have to use a closure for this to ensure the context manager is
        # closed only after sending the final chunk.
        def getfiledata():
            with repo.svfs(name, b'rb', auditpath=False) as fh:
                for chunk in util.filechunkiter(fh, limit=size):
                    yield chunk

        yield wireprototypes.indefinitebytestringresponse(getfiledata())