comparison hgext/acl.py @ 43076:2372284d9457

formatting: blacken the codebase

This is using my patch to black
(https://github.com/psf/black/pull/826) so we don't un-wrap collection
literals.

Done with:

  hg files 'set:**.py - mercurial/thirdparty/** - "contrib/python-zstandard/**"' | xargs black -S

# skip-blame mass-reformatting only

# no-check-commit reformats foo_bar functions

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

43075:57875cf423c9

    match,
    pycompat,
    registrar,
    util,
)
from mercurial.utils import (
    procutil,
)

urlreq = util.urlreq

# Note for extension authors: ONLY specify testedwith = 'ships-with-hg-core' for
# extensions which SHIP WITH MERCURIAL. Non-mainline extensions should

configtable = {}
configitem = registrar.configitem(configtable)

# deprecated config: acl.config
configitem('acl', 'config',
    default=None,
)
configitem('acl.groups', '.*',
    default=None,
    generic=True,
)
configitem('acl.deny.branches', '.*',
    default=None,
    generic=True,
)
configitem('acl.allow.branches', '.*',
    default=None,
    generic=True,
)
configitem('acl.deny', '.*',
    default=None,
    generic=True,
)
configitem('acl.allow', '.*',
    default=None,
    generic=True,
)
configitem('acl', 'sources',
    default=lambda: ['serve'],
)

def _getusers(ui, group):

    # First, try to use group definition from section [acl.groups]
    hgrcusers = ui.configlist('acl.groups', group)

    try:
        return util.groupmembers(group)
    except KeyError:
        raise error.Abort(_("group '%s' is undefined") % group)

def _usermatch(ui, user, usersorgroups):

    if usersorgroups == '*':
        return True

        if ug.startswith('!'):
            # Test for excluded user or group. Format:
            # if ug is a user  name: !username
            # if ug is a group name: !@groupname
            ug = ug[1:]
            if (not ug.startswith('@') and user != ug
                or ug.startswith('@') and user not in _getusers(ui, ug[1:])):
                return True

        # Test for user or group. Format:
        # if ug is a user  name: username
        # if ug is a group name: @groupname
        elif (user == ug
              or ug.startswith('@') and user in _getusers(ui, ug[1:])):
            return True

    return False

def buildmatch(ui, repo, user, key):
    '''return tuple of (match function, list enabled).'''
    if not ui.has_section(key):
        ui.debug('acl: %s not enabled\n' % key)
        return None

    pats = [pat for pat, users in ui.configitems(key)
            if _usermatch(ui, user, users)]
    ui.debug('acl: %s enabled, %d entries for user %s\n' %
             (key, len(pats), user))

    # Branch-based ACL
    if not repo:
        if pats:
            # If there's an asterisk (meaning "any branch"), always return True;

    # Path-based ACL
    if pats:
        return match.match(repo.root, '', pats)
    return util.never

def ensureenabled(ui):
    """make sure the extension is enabled when used as hook

    When acl is used through hooks, the extension is never formally loaded and
    enabled. This has some side effect, for example the config declaration is

    if 'acl' in ui._knownconfig:
        return
    ui.setconfig('extensions', 'acl', '', source='internal')
    extensions.loadall(ui, ['acl'])

def hook(ui, repo, hooktype, node=None, source=None, **kwargs):

    ensureenabled(ui)

    if hooktype not in ['pretxnchangegroup', 'pretxncommit', 'prepushkey']:
        raise error.Abort(
            _('config error - hook type "%s" cannot stop '
              'incoming changesets, commits, nor bookmarks') % hooktype)
    if (hooktype == 'pretxnchangegroup' and
        source not in ui.configlist('acl', 'sources')):
        ui.debug('acl: changes have source "%s" - skipping\n' % source)
        return

    user = None
    if source == 'serve' and r'url' in kwargs:

    if hooktype == 'prepushkey':
        _pkhook(ui, repo, hooktype, node, source, user, **kwargs)
    else:
        _txnhook(ui, repo, hooktype, node, source, user, **kwargs)

def _pkhook(ui, repo, hooktype, node, source, user, **kwargs):
    if kwargs[r'namespace'] == 'bookmarks':
        bookmark = kwargs[r'key']
        ctx = kwargs[r'new']
        allowbookmarks = buildmatch(ui, None, user, 'acl.allow.bookmarks')
        denybookmarks = buildmatch(ui, None, user, 'acl.deny.bookmarks')

        if denybookmarks and denybookmarks(bookmark):
            raise error.Abort(_('acl: user "%s" denied on bookmark "%s"'
                               ' (changeset "%s")')
                               % (user, bookmark, ctx))
        if allowbookmarks and not allowbookmarks(bookmark):
            raise error.Abort(_('acl: user "%s" not allowed on bookmark "%s"'
                               ' (changeset "%s")')
                               % (user, bookmark, ctx))
        ui.debug('acl: bookmark access granted: "%s" on bookmark "%s"\n'
                 % (ctx, bookmark))

def _txnhook(ui, repo, hooktype, node, source, user, **kwargs):
    # deprecated config: acl.config
    cfg = ui.config('acl', 'config')
    if cfg:
        ui.readconfig(cfg, sections=['acl.groups', 'acl.allow.branches',
            'acl.deny.branches', 'acl.allow', 'acl.deny'])

    allowbranches = buildmatch(ui, None, user, 'acl.allow.branches')
    denybranches = buildmatch(ui, None, user, 'acl.deny.branches')
    allow = buildmatch(ui, repo, user, 'acl.allow')
    deny = buildmatch(ui, repo, user, 'acl.deny')

    for rev in pycompat.xrange(repo[node].rev(), len(repo)):
        ctx = repo[rev]
        branch = ctx.branch()
        if denybranches and denybranches(branch):
            raise error.Abort(_('acl: user "%s" denied on branch "%s"'
                               ' (changeset "%s")')
                               % (user, branch, ctx))
        if allowbranches and not allowbranches(branch):
            raise error.Abort(_('acl: user "%s" not allowed on branch "%s"'
                               ' (changeset "%s")')
                               % (user, branch, ctx))
        ui.debug('acl: branch access granted: "%s" on branch "%s"\n'
        % (ctx, branch))

        for f in ctx.files():
            if deny and deny(f):
                raise error.Abort(_('acl: user "%s" denied on "%s"'
                ' (changeset "%s")') % (user, f, ctx))
            if allow and not allow(f):
                raise error.Abort(_('acl: user "%s" not allowed on "%s"'
                ' (changeset "%s")') % (user, f, ctx))
        ui.debug('acl: path access granted: "%s"\n' % ctx)

43076:2372284d9457

    match,
    pycompat,
    registrar,
    util,
)
from mercurial.utils import procutil

urlreq = util.urlreq

# Note for extension authors: ONLY specify testedwith = 'ships-with-hg-core' for
# extensions which SHIP WITH MERCURIAL. Non-mainline extensions should

configtable = {}
configitem = registrar.configitem(configtable)

# deprecated config: acl.config
configitem(
    'acl', 'config', default=None,
)
configitem(
    'acl.groups', '.*', default=None, generic=True,
)
configitem(
    'acl.deny.branches', '.*', default=None, generic=True,
)
configitem(
    'acl.allow.branches', '.*', default=None, generic=True,
)
configitem(
    'acl.deny', '.*', default=None, generic=True,
)
configitem(
    'acl.allow', '.*', default=None, generic=True,
)
configitem(
    'acl', 'sources', default=lambda: ['serve'],
)


def _getusers(ui, group):

    # First, try to use group definition from section [acl.groups]
    hgrcusers = ui.configlist('acl.groups', group)

    try:
        return util.groupmembers(group)
    except KeyError:
        raise error.Abort(_("group '%s' is undefined") % group)


def _usermatch(ui, user, usersorgroups):

    if usersorgroups == '*':
        return True

        if ug.startswith('!'):
            # Test for excluded user or group. Format:
            # if ug is a user  name: !username
            # if ug is a group name: !@groupname
            ug = ug[1:]
            if (
                not ug.startswith('@')
                and user != ug
                or ug.startswith('@')
                and user not in _getusers(ui, ug[1:])
            ):
                return True

        # Test for user or group. Format:
        # if ug is a user  name: username
        # if ug is a group name: @groupname
        elif user == ug or ug.startswith('@') and user in _getusers(ui, ug[1:]):
            return True

    return False


def buildmatch(ui, repo, user, key):
    '''return tuple of (match function, list enabled).'''
    if not ui.has_section(key):
        ui.debug('acl: %s not enabled\n' % key)
        return None

    pats = [
        pat for pat, users in ui.configitems(key) if _usermatch(ui, user, users)
    ]
    ui.debug(
        'acl: %s enabled, %d entries for user %s\n' % (key, len(pats), user)
    )

    # Branch-based ACL
    if not repo:
        if pats:
            # If there's an asterisk (meaning "any branch"), always return True;

    # Path-based ACL
    if pats:
        return match.match(repo.root, '', pats)
    return util.never


def ensureenabled(ui):
    """make sure the extension is enabled when used as hook

    When acl is used through hooks, the extension is never formally loaded and
    enabled. This has some side effect, for example the config declaration is

    if 'acl' in ui._knownconfig:
        return
    ui.setconfig('extensions', 'acl', '', source='internal')
    extensions.loadall(ui, ['acl'])


def hook(ui, repo, hooktype, node=None, source=None, **kwargs):

    ensureenabled(ui)

    if hooktype not in ['pretxnchangegroup', 'pretxncommit', 'prepushkey']:
        raise error.Abort(
            _(
                'config error - hook type "%s" cannot stop '
                'incoming changesets, commits, nor bookmarks'
            )
            % hooktype
        )
    if hooktype == 'pretxnchangegroup' and source not in ui.configlist(
        'acl', 'sources'
    ):
        ui.debug('acl: changes have source "%s" - skipping\n' % source)
        return

    user = None
    if source == 'serve' and r'url' in kwargs:

    if hooktype == 'prepushkey':
        _pkhook(ui, repo, hooktype, node, source, user, **kwargs)
    else:
        _txnhook(ui, repo, hooktype, node, source, user, **kwargs)


def _pkhook(ui, repo, hooktype, node, source, user, **kwargs):
    if kwargs[r'namespace'] == 'bookmarks':
        bookmark = kwargs[r'key']
        ctx = kwargs[r'new']
        allowbookmarks = buildmatch(ui, None, user, 'acl.allow.bookmarks')
        denybookmarks = buildmatch(ui, None, user, 'acl.deny.bookmarks')

        if denybookmarks and denybookmarks(bookmark):
            raise error.Abort(
                _('acl: user "%s" denied on bookmark "%s"' ' (changeset "%s")')
                % (user, bookmark, ctx)
            )
        if allowbookmarks and not allowbookmarks(bookmark):
            raise error.Abort(
                _(
                    'acl: user "%s" not allowed on bookmark "%s"'
                    ' (changeset "%s")'
                )
                % (user, bookmark, ctx)
            )
        ui.debug(
            'acl: bookmark access granted: "%s" on bookmark "%s"\n'
            % (ctx, bookmark)
        )


def _txnhook(ui, repo, hooktype, node, source, user, **kwargs):
    # deprecated config: acl.config
    cfg = ui.config('acl', 'config')
    if cfg:
        ui.readconfig(
            cfg,
            sections=[
                'acl.groups',
                'acl.allow.branches',
                'acl.deny.branches',
                'acl.allow',
                'acl.deny',
            ],
        )

    allowbranches = buildmatch(ui, None, user, 'acl.allow.branches')
    denybranches = buildmatch(ui, None, user, 'acl.deny.branches')
    allow = buildmatch(ui, repo, user, 'acl.allow')
    deny = buildmatch(ui, repo, user, 'acl.deny')

    for rev in pycompat.xrange(repo[node].rev(), len(repo)):
        ctx = repo[rev]
        branch = ctx.branch()
        if denybranches and denybranches(branch):
            raise error.Abort(
                _('acl: user "%s" denied on branch "%s"' ' (changeset "%s")')
                % (user, branch, ctx)
            )
        if allowbranches and not allowbranches(branch):
            raise error.Abort(
                _(
                    'acl: user "%s" not allowed on branch "%s"'
                    ' (changeset "%s")'
                )
                % (user, branch, ctx)
            )
        ui.debug(
            'acl: branch access granted: "%s" on branch "%s"\n' % (ctx, branch)
        )

        for f in ctx.files():
            if deny and deny(f):
                raise error.Abort(
                    _('acl: user "%s" denied on "%s"' ' (changeset "%s")')
                    % (user, f, ctx)
                )
            if allow and not allow(f):
                raise error.Abort(
                    _('acl: user "%s" not allowed on "%s"' ' (changeset "%s")')
                    % (user, f, ctx)
                )
        ui.debug('acl: path access granted: "%s"\n' % ctx)