codekingpro/portable-devtools

1"""2    flask_security.cli3    ~~~~~~~~~~~~~~~~~~4 5    Command Line Interface for managing accounts and roles.6 7    :copyright: (c) 2016 by CERN.8    :copyright: (c) 2019-2022 by J. Christopher Wagner9    :license: MIT, see LICENSE for more details.10"""11 12import functools13 14import click15from flask import current_app16from werkzeug.local import LocalProxy17from .quart_compat import get_quart_status18 19from .changeable import admin_change_password20from .forms import build_form21from .utils import (22    lookup_identity,23    get_identity_attributes,24    get_identity_attribute,25    hash_password,26)27 28if get_quart_status():  # pragma: no cover29    import quart.cli30 31    # quart cli doesn't provide the with_appcontext function32    def with_appcontext(f):33        """Wraps a callback so that it's guaranteed to be executed with the34        script's application context.  If callbacks are registered directly35        to the ``app.cli`` object then they are wrapped with this function36        by default unless it's disabled.37        """38 39        @click.pass_context40        def decorator(__ctx, *args, **kwargs):41            with __ctx.ensure_object(quart.cli.ScriptInfo).load_app().app_context():42                return __ctx.invoke(f, *args, **kwargs)43 44        return functools.update_wrapper(decorator, f)45 46else:47    import flask.cli48 49    with_appcontext = flask.cli.with_appcontext50 51 52_security = LocalProxy(lambda: current_app.extensions["security"])53_datastore = LocalProxy(lambda: current_app.extensions["security"].datastore)54 55 56def commit(fn):57    """Decorator to commit changes in datastore."""58 59    @functools.wraps(fn)60    def wrapper(*args, **kwargs):61        fn(*args, **kwargs)62        _datastore.commit()63 64    return wrapper65 66 67def fix_errors(form_errors):68    # Form errors might have lazy text which normally would be processed by69    # render_template70    errors = {}71    for k, v in form_errors.items():72        errors[k] = [str(e) for e in v]73    return errors74 75 76@click.group()77def users():78    """User commands.79 80    For commands that require a USER - pass in any identity attribute.81    """82 83 84@click.group()85def roles():86    """Role commands."""87 88 89@users.command(90    "create",91    short_help=(92        "Create a new user with one or more attributes using the syntax:"93        " attr:value. If attr isn't set 'email' is presumed."94        " Identity attribute values will be validated using the configured"95        " confirm_register_form;"96        " however, any ADDITIONAL attribute:value pairs will be sent to"97        " datastore.create_user"98    ),99)100@click.argument(101    "attributes",102    nargs=-1,103)104@click.password_option()105@click.option("-a", "--active", default=False, is_flag=True)106@with_appcontext107@commit108def users_create(attributes, password, active):109    """Create a user."""110    kwargs = {}111 112    identity_attributes = get_identity_attributes()113    for attrarg in attributes:114        # If given identity is an identity_attribute - do a bit of pre-validating115        # to provide nicer errors.116        attr = "email"117        if ":" in attrarg:118            attr, attrarg = attrarg.split(":")119        if attr in identity_attributes:120            details = get_identity_attribute(attr)121            idata = details["mapper"](attrarg)122            if not idata:123                raise click.UsageError(124                    f"Attr {attr} with value {attrarg} wasn't accepted by mapper"125                )126 127        kwargs[attr] = attrarg128    kwargs.update(**{"password": password})129 130    form = build_form("confirm_register_form", meta={"csrf": False}, **kwargs)131 132    if form.validate():133        # We don't use the form directly to provide values so that this CLI can actually134        # set any usermodel attribute. We do grab email and password from the form135        # so that we get any normalization results.136        kwargs["password"] = hash_password(form.password.data)137        kwargs["active"] = active138        # echo normalized email...139        if "email" in kwargs:140            kwargs["email"] = form.email.data141        _datastore.create_user(**kwargs)142        click.secho("User created successfully.", fg="green")143        kwargs["password"] = "****"144        click.echo(kwargs)145    else:146        raise click.UsageError(f"Error creating user. {fix_errors(form.errors)}")147 148 149@roles.command("create")150@click.argument("name")151@click.option("-d", "--description", default=None)152@click.option("-p", "--permissions", help="A comma separated list")153@with_appcontext154@commit155def roles_create(**kwargs):156    """Create a role."""157 158    # For some reason Click puts arguments in kwargs - even if they weren't specified.159    if "permissions" in kwargs and not kwargs["permissions"]:160        del kwargs["permissions"]161    if "permissions" in kwargs and not hasattr(_datastore.role_model, "permissions"):162        raise click.UsageError("Role model does not support permissions")163    _datastore.create_role(**kwargs)164    click.secho('Role "%(name)s" created successfully.' % kwargs, fg="green")165 166 167@roles.command("add")168@click.argument("user")169@click.argument("role")170@with_appcontext171@commit172def roles_add(user, role):173    """Add role to user.174 175    USER is identity as defined by SECURITY_USER_IDENTITY_ATTRIBUTES.176    """177    user_obj = lookup_identity(user)178    if user_obj is None:179        raise click.UsageError("User not found.")180 181    role = _datastore._prepare_role_modify_args(role)182    if role is None:183        raise click.UsageError("Cannot find role.")184    if _datastore.add_role_to_user(user_obj, role):185        click.secho(186            f'Role "{role.name}" added to user "{user}" successfully.',187            fg="green",188        )189    else:190        raise click.UsageError("Cannot add role to user.")191 192 193@roles.command("remove")194@click.argument("user")195@click.argument("role")196@with_appcontext197@commit198def roles_remove(user, role):199    """Remove role from user.200 201    USER is identity as defined by SECURITY_USER_IDENTITY_ATTRIBUTES.202    """203    user_obj = lookup_identity(user)204    if user_obj is None:205        raise click.UsageError("User not found.")206 207    role = _datastore._prepare_role_modify_args(role)208    if role is None:209        raise click.UsageError("Cannot find role.")210    if _datastore.remove_role_from_user(user_obj, role):211        click.secho(212            f'Role "{role.name}" removed from user "{user}" successfully.',213            fg="green",214        )215    else:216        raise click.UsageError("Cannot remove role from user.")217 218 219@roles.command("add_permissions")220@click.argument("role")221@click.argument("permissions")222@with_appcontext223@commit224def roles_add_permissions(role, permissions):225    """Add permissions to role.226 227    Role is an existing role name.228    Permissions are a comma separated list.229    """230    role = _datastore._prepare_role_modify_args(role)231    if role is None:232        raise click.UsageError("Cannot find role.")233    permlist = [s.strip() for s in permissions.split(",")]234    if _datastore.add_permissions_to_role(role, permlist):235        click.secho(236            f'Permission(s) "{permissions}" added to role "{role.name}" successfully.',237            fg="green",238        )239    else:  # pragma: no cover240        raise click.UsageError("Cannot add permission(s) to role.")241 242 243@roles.command("remove_permissions")244@click.argument("role")245@click.argument("permissions")246@with_appcontext247@commit248def roles_remove_permissions(role, permissions):249    """Remove permissions from role.250 251    Role is an existing role name.252    Permissions are a comma separated list.253    """254    role = _datastore._prepare_role_modify_args(role)255    if role is None:256        raise click.UsageError("Cannot find role.")257    permlist = [s.strip() for s in permissions.split(",")]258    if _datastore.remove_permissions_from_role(role, permlist):259        click.secho(260            f'Permission(s) "{permissions}" removed from role'261            f' "{role.name}" successfully.',262            fg="green",263        )264    else:  # pragma: no cover265        raise click.UsageError("Cannot remove permission(s) from role.")266 267 268@users.command("activate")269@click.argument("user")270@with_appcontext271@commit272def users_activate(user):273    """Activate a user.274 275    USER is identity as defined by SECURITY_USER_IDENTITY_ATTRIBUTES.276    """277    user_obj = lookup_identity(user)278    if user_obj is None:279        raise click.UsageError("User not found.")280    if _datastore.activate_user(user_obj):281        click.secho(f'User "{user}" has been activated.', fg="green")282    else:283        click.secho(f'User "{user}" was already activated.', fg="yellow")284 285 286@users.command("deactivate")287@click.argument("user")288@with_appcontext289@commit290def users_deactivate(user):291    """Deactivate a user.292 293    USER is identity as defined by SECURITY_USER_IDENTITY_ATTRIBUTES.294    """295    user_obj = lookup_identity(user)296    if user_obj is None:297        raise click.UsageError("User not found.")298    if _datastore.deactivate_user(user_obj):299        click.secho(f'User "{user}" has been deactivated.', fg="green")300    else:301        click.secho(f'User "{user}" was already deactivated.', fg="yellow")302 303 304@users.command("reset_access")305@click.argument("user")306@with_appcontext307@commit308def users_reset_access(user):309    """Reset all authentication credentials for user.310    This includes sessions, authentication tokens, two-factor311    and unified sign in secrets. The user's password is not affected.312 313    USER is identity as defined by SECURITY_USER_IDENTITY_ATTRIBUTES.314 315    """316    user_obj = lookup_identity(user)317    if user_obj is None:318        raise click.UsageError("User not found.")319    _datastore.reset_user_access(user_obj)320    click.secho(321        f'User "{user}" authentication credentials have been reset.', fg="green"322    )323 324 325@users.command("change_password")326@click.argument("user")327@click.password_option()328@with_appcontext329@commit330def users_change_password(user, password):331    """332    Administratively change a user's password.333    All the user's sessions will be immediately invalidated.334    You will have to inform the user via an out of band mechanism335    what their new password is.336 337    USER is identity as defined by SECURITY_USER_IDENTITY_ATTRIBUTES.338 339    """340    user_obj = lookup_identity(user)341    if user_obj is None:342        raise click.UsageError("User not found.")343 344    kwargs = {"password": password, "password_confirm": password}345    form = build_form("reset_password_form", meta={"csrf": False}, **kwargs)346    form.user = user_obj347 348    if form.validate():349        # validation will normalize password350        admin_change_password(user_obj, form.password.data, notify=False)351    else:352        raise click.UsageError(f"Error changing password. {fix_errors(form.errors)}")353