def add_bot_backend(request, user_profile, full_name=REQ(), short_name=REQ(),
                    default_sending_stream_name=REQ('default_sending_stream', default=None),
                    default_events_register_stream_name=REQ('default_events_register_stream', default=None),
                    default_all_public_streams=REQ(validator=check_bool, default=None)):
    # type: (HttpRequest, UserProfile, text_type, text_type, Optional[text_type], Optional[text_type], Optional[bool]) -> HttpResponse
    short_name += "-bot"
    email = short_name + "@" + user_profile.realm.domain
    form = CreateUserForm({'full_name': full_name, 'email': email})
    if not form.is_valid():
        # We validate client-side as well
        return json_error(_('Bad name or username'))

    try:
        get_user_profile_by_email(email)
        return json_error(_("Username already in use"))
    except UserProfile.DoesNotExist:
        pass

    if len(request.FILES) == 0:
        avatar_source = UserProfile.AVATAR_FROM_GRAVATAR
    elif len(request.FILES) != 1:
        return json_error(_("You may only upload one file at a time"))
    else:
        user_file = list(request.FILES.values())[0]
        upload_avatar_image(user_file, user_profile, email)
        avatar_source = UserProfile.AVATAR_FROM_USER

    default_sending_stream = None
    if default_sending_stream_name is not None:
        default_sending_stream = stream_or_none(default_sending_stream_name, user_profile.realm)
    if default_sending_stream and not default_sending_stream.is_public() and not \
        subscribed_to_stream(user_profile, default_sending_stream):
        return json_error(_('Insufficient permission'))

    default_events_register_stream = None
    if default_events_register_stream_name is not None:
        default_events_register_stream = stream_or_none(default_events_register_stream_name,
                                                        user_profile.realm)
    if default_events_register_stream and not default_events_register_stream.is_public() and not \
        subscribed_to_stream(user_profile, default_events_register_stream):
        return json_error(_('Insufficient permission'))


    bot_profile = do_create_user(email=email, password='',
                                 realm=user_profile.realm, full_name=full_name,
                                 short_name=short_name, active=True,
                                 bot_type=UserProfile.DEFAULT_BOT,
                                 bot_owner=user_profile,
                                 avatar_source=avatar_source,
                                 default_sending_stream=default_sending_stream,
                                 default_events_register_stream=default_events_register_stream,
                                 default_all_public_streams=default_all_public_streams)
    json_result = dict(
            api_key=bot_profile.api_key,
            avatar_url=avatar_url(bot_profile),
            default_sending_stream=get_stream_name(bot_profile.default_sending_stream),
            default_events_register_stream=get_stream_name(bot_profile.default_events_register_stream),
            default_all_public_streams=bot_profile.default_all_public_streams,
    )
    return json_success(json_result)

def api_dev_fetch_api_key(request: HttpRequest, username: str=REQ()) -> HttpResponse:
    """This function allows logging in without a password on the Zulip
    mobile apps when connecting to a Zulip development environment.  It
    requires DevAuthBackend to be included in settings.AUTHENTICATION_BACKENDS.
    """
    if not dev_auth_enabled() or settings.PRODUCTION:
        return json_error(_("Dev environment not enabled."))

    # Django invokes authenticate methods by matching arguments, and this
    # authentication flow will not invoke LDAP authentication because of
    # this condition of Django so no need to check if LDAP backend is
    # enabled.
    validate_login_email(username)

    subdomain = get_subdomain(request)
    realm = get_realm(subdomain)

    return_data = {}  # type: Dict[str, bool]
    user_profile = authenticate(dev_auth_username=username,
                                realm=realm,
                                return_data=return_data)
    if return_data.get("inactive_realm"):
        return json_error(_("Your realm has been deactivated."),
                          data={"reason": "realm deactivated"}, status=403)
    if return_data.get("inactive_user"):
        return json_error(_("Your account has been disabled."),
                          data={"reason": "user disable"}, status=403)
    if user_profile is None:
        return json_error(_("This user is not registered."),
                          data={"reason": "unregistered"}, status=403)
    do_login(request, user_profile)
    return json_success({"api_key": user_profile.api_key, "email": user_profile.email})

def api_fetch_api_key(request, username=REQ(), password=REQ()):
    # type: (HttpRequest, str, str) -> HttpResponse
    return_data = {} # type: Dict[str, bool]
    if username == "google-oauth2-token":
        user_profile = authenticate(google_oauth2_token=password,
                                    realm_subdomain=get_subdomain(request),
                                    return_data=return_data)
    else:
        user_profile = authenticate(username=username,
                                    password=password,
                                    realm_subdomain=get_subdomain(request),
                                    return_data=return_data)
    if return_data.get("inactive_user") == True:
        return json_error(_("Your account has been disabled."),
                          data={"reason": "user disable"}, status=403)
    if return_data.get("inactive_realm") == True:
        return json_error(_("Your realm has been deactivated."),
                          data={"reason": "realm deactivated"}, status=403)
    if return_data.get("password_auth_disabled") == True:
        return json_error(_("Password auth is disabled in your team."),
                          data={"reason": "password auth disabled"}, status=403)
    if user_profile is None:
        if return_data.get("valid_attestation") == True:
            # We can leak that the user is unregistered iff they present a valid authentication string for the user.
            return json_error(_("This user is not registered; do so from a browser."),
                              data={"reason": "unregistered"}, status=403)
        return json_error(_("Your username or password is incorrect."),
                          data={"reason": "incorrect_creds"}, status=403)
    return json_success({"api_key": user_profile.api_key, "email": user_profile.email})

def get_presence_backend(request: HttpRequest, user_profile: UserProfile,
                         email: Text) -> HttpResponse:
    try:
        target = get_user(email, user_profile.realm)
    except UserProfile.DoesNotExist:
        return json_error(_('No such user'))
    if not target.is_active:
        return json_error(_('No such user'))
    if target.is_bot:
        return json_error(_('Presence is not supported for bot users.'))

    presence_dict = UserPresence.get_status_dict_by_user(target)
    if len(presence_dict) == 0:
        return json_error(_('No presence data for %s' % (target.email,)))

    # For initial version, we just include the status and timestamp keys
    result = dict(presence=presence_dict[target.email])
    aggregated_info = result['presence']['aggregated']
    aggr_status_duration = datetime_to_timestamp(timezone_now()) - aggregated_info['timestamp']
    if aggr_status_duration > settings.OFFLINE_THRESHOLD_SECS:
        aggregated_info['status'] = 'offline'
    for val in result['presence'].values():
        val.pop('client', None)
        val.pop('pushable', None)
    return json_success(result)

def api_jira_webhook(request, user_profile, client,
                     payload=REQ(argument_type='body'),
                     stream=REQ(default='jira')):
    # type: (HttpRequest, UserProfile, Client, Dict[str, Any], Text) -> HttpResponse

    event = payload.get('webhookEvent')
    if event == 'jira:issue_created':
        subject = get_issue_subject(payload)
        content = handle_created_issue_event(payload)
    elif event == 'jira:issue_deleted':
        subject = get_issue_subject(payload)
        content = handle_deleted_issue_event(payload)
    elif event == 'jira:issue_updated':
        subject = get_issue_subject(payload)
        content = handle_updated_issue_event(payload, user_profile)
    elif event in IGNORED_EVENTS:
        return json_success()
    else:
        if event is None:
            if not settings.TEST_SUITE:
                message = "Got JIRA event with None event type: {}".format(payload)
                logging.warning(message)
            return json_error(_("Event is not given by JIRA"))
        else:
            if not settings.TEST_SUITE:
                logging.warning("Got JIRA event type we don't support: {}".format(event))
            return json_error(_("Got JIRA event type we don't support: {}".format(event)))

    check_send_message(user_profile, client, "stream", [stream], subject, content)
    return json_success()

def api_zapier_webhook(request: HttpRequest, user_profile: UserProfile,
                       payload: Dict[str, Any]=REQ(argument_type='body')) -> HttpResponse:
    if payload.get('type') == 'auth':
        # The bot's details are used by our Zapier app to format a connection
        # label for users to be able to distinguish between different Zulip
        # bots and API keys in their UI
        return json_success({
            'full_name': user_profile.full_name,
            'email': user_profile.email,
            'id': user_profile.id
        })

    topic = payload.get('topic')
    content = payload.get('content')

    if topic is None:
        topic = payload.get('subject')  # Backwards-compatibility
        if topic is None:
            return json_error(_("Topic can't be empty"))

    if content is None:
        return json_error(_("Content can't be empty"))

    check_send_webhook_message(request, user_profile, topic, content)
    return json_success()

def json_invite_users(request, user_profile, invitee_emails_raw=REQ("invitee_emails")):
    # type: (HttpRequest, UserProfile, str) -> HttpResponse
    if not invitee_emails_raw:
        return json_error(_("You must specify at least one email address."))

    invitee_emails = get_invitee_emails_set(invitee_emails_raw)

    stream_names = request.POST.getlist('stream')
    if not stream_names:
        return json_error(_("You must specify at least one stream for invitees to join."))

    # We unconditionally sub you to the notifications stream if it
    # exists and is public.
    notifications_stream = user_profile.realm.notifications_stream
    if notifications_stream and not notifications_stream.invite_only:
        stream_names.append(notifications_stream.name)

    streams = [] # type: List[Stream]
    for stream_name in stream_names:
        stream = get_stream(stream_name, user_profile.realm)
        if stream is None:
            return json_error(_("Stream does not exist: %s. No invites were sent.") % (stream_name,))
        streams.append(stream)

    ret_error, error_data = do_invite_users(user_profile, invitee_emails, streams)

    if ret_error is not None:
        return json_error(data=error_data, msg=ret_error)
    else:
        return json_success()

def create_user_backend(request, user_profile, email=REQ(), password=REQ(),
                        full_name=REQ(), short_name=REQ()):
    # type: (HttpRequest, UserProfile, text_type, text_type, text_type, text_type) -> HttpResponse
    form = CreateUserForm({'full_name': full_name, 'email': email})
    if not form.is_valid():
        return json_error(_('Bad name or username'))

    # Check that the new user's email address belongs to the admin's realm
    # (Since this is an admin API, we don't require the user to have been
    # invited first.)
    realm = user_profile.realm
    if not email_allowed_for_realm(email, user_profile.realm):
        return json_error(_("Email '%(email)s' does not belong to domain '%(domain)s'") %
                          {'email': email, 'domain': realm.domain})

    try:
        user_profile = get_user_profile_by_email(email)
        return json_error(
            _("Email '%s' already in use") % (email,),
            dict(api_key=user_profile.api_key)
        )
    except UserProfile.DoesNotExist:
        pass

    user_profile = do_create_user(email, password, realm, full_name, short_name)
    return json_success(dict(api_key=user_profile.api_key))

def update_user_backend(request, user_profile, email,
                        full_name=REQ(default="", validator=check_string),
                        is_admin=REQ(default=None, validator=check_bool)):
    # type: (HttpRequest, UserProfile, text_type, Optional[text_type], Optional[bool]) -> HttpResponse
    try:
        target = get_user_profile_by_email(email)
    except UserProfile.DoesNotExist:
        return json_error(_('No such user'))

    if not user_profile.can_admin_user(target):
        return json_error(_('Insufficient permission'))

    if is_admin is not None:
        if not is_admin and check_last_admin(user_profile):
            return json_error(_('Cannot remove the only organization administrator'))
        do_change_is_admin(target, is_admin)

    if (full_name is not None and target.full_name != full_name and
            full_name.strip() != ""):
        # We don't respect `name_changes_disabled` here because the request
        # is on behalf of the administrator.
        new_full_name = full_name.strip()
        if len(new_full_name) > UserProfile.MAX_NAME_LENGTH:
            return json_error(_("Name too long!"))
        do_change_full_name(target, new_full_name)

    return json_success()

def create_user_backend(request: HttpRequest, user_profile: UserProfile,
                        email: str=REQ(), password: str=REQ(), full_name_raw: str=REQ("full_name"),
                        short_name: str=REQ()) -> HttpResponse:
    full_name = check_full_name(full_name_raw)
    form = CreateUserForm({'full_name': full_name, 'email': email})
    if not form.is_valid():
        return json_error(_('Bad name or username'))

    # Check that the new user's email address belongs to the admin's realm
    # (Since this is an admin API, we don't require the user to have been
    # invited first.)
    realm = user_profile.realm
    try:
        email_allowed_for_realm(email, user_profile.realm)
    except DomainNotAllowedForRealmError:
        return json_error(_("Email '%(email)s' not allowed in this organization") %
                          {'email': email})
    except DisposableEmailError:
        return json_error(_("Disposable email addresses are not allowed in this organization"))
    except EmailContainsPlusError:
        return json_error(_("Email addresses containing + are not allowed."))

    try:
        get_user_by_delivery_email(email, user_profile.realm)
        return json_error(_("Email '%s' already in use") % (email,))
    except UserProfile.DoesNotExist:
        pass

    do_create_user(email, password, realm, full_name, short_name)
    return json_success()

        def _wrapped_func_arguments(request, *args, **kwargs):
            # type: (HttpRequest, *Any, **Any) -> HttpResponse
            # First try block attempts to get the credentials we need to do authentication
            try:
                # Grab the base64-encoded authentication string, decode it, and split it into
                # the email and API key
                auth_type, credentials = request.META['HTTP_AUTHORIZATION'].split()
                # case insensitive per RFC 1945
                if auth_type.lower() != "basic":
                    return json_error(_("Only Basic authentication is supported."))
                role, api_key = base64.b64decode(force_bytes(credentials)).decode('utf-8').split(":")
            except ValueError:
                json_error(_("Invalid authorization header for basic auth"))
            except KeyError:
                return json_unauthorized("Missing authorization header for basic auth")

            # Now we try to do authentication or die
            try:
                # Could be a UserProfile or a Deployment
                profile = validate_api_key(role, api_key, is_webhook)
            except JsonableError as e:
                return json_unauthorized(e.error)
            request.user = profile
            process_client(request, profile)
            if isinstance(profile, UserProfile):
                request._email = profile.email
            else:
                assert isinstance(profile, Deployment) # type: ignore # https://github.com/python/mypy/issues/1720#issuecomment-228596830
                request._email = "deployment:" + role
                profile.rate_limits = ""
            # Apply rate limiting
            return rate_limit()(view_func)(request, profile, *args, **kwargs)

def api_stash_webhook(request, user_profile, stream=REQ(default='')):
    try:
        payload = ujson.loads(request.body)
    except ValueError:
        return json_error("Malformed JSON input")

    # We don't get who did the push, or we'd try to report that.
    try:
        repo_name = payload["repository"]["name"]
        project_name = payload["repository"]["project"]["name"]
        branch_name = payload["refChanges"][0]["refId"].split("/")[-1]
        commit_entries = payload["changesets"]["values"]
        commits = [(entry["toCommit"]["displayId"],
                    entry["toCommit"]["message"].split("\n")[0]) for \
                       entry in commit_entries]
        head_ref = commit_entries[-1]["toCommit"]["displayId"]
    except KeyError as e:
        return json_error("Missing key %s in JSON" % (e.message,))

    try:
        stream = request.GET['stream']
    except (AttributeError, KeyError):
        stream = 'commits'

    subject = "%s/%s: %s" % (project_name, repo_name, branch_name)

    content = "`%s` was pushed to **%s** in **%s/%s** with:\n\n" % (
        head_ref, branch_name, project_name, repo_name)
    content += "\n".join("* `%s`: %s" % (
            commit[0], commit[1]) for commit in commits)

    check_send_message(user_profile, get_client("ZulipStashWebhook"), "stream",
                       [stream], subject, content)
    return json_success()

 def process_exception(self, request, exception):
     if hasattr(exception, 'to_json_error_msg') and callable(exception.to_json_error_msg):
         return json_error(exception.to_json_error_msg())
     if request.error_format == "JSON":
         logging.error(traceback.format_exc())
         return json_error("Internal server error", status=500)
     return None

def create_realm_custom_profile_field(request: HttpRequest,
                                      user_profile: UserProfile, name: str=REQ(),
                                      hint: str=REQ(default=''),
                                      field_data: ProfileFieldData=REQ(default={},
                                                                       converter=ujson.loads),
                                      field_type: int=REQ(validator=check_int)) -> HttpResponse:
    if not name.strip():
        return json_error(_("Name cannot be blank."))

    error = hint_validator('hint', hint)
    if error:
        return json_error(error)

    field_types = [i[0] for i in CustomProfileField.FIELD_TYPE_CHOICES]
    if field_type not in field_types:
        return json_error(_("Invalid field type."))

    error = validate_field_data(field_data)
    if error:
        return json_error(error)

    try:
        field = try_add_realm_custom_profile_field(
            realm=user_profile.realm,
            name=name,
            field_data=field_data,
            field_type=field_type,
            hint=hint,
        )
        return json_success({'id': field.id})
    except IntegrityError:
        return json_error(_("A field with that name already exists."))

def update_user_custom_profile_data(
        request: HttpRequest,
        user_profile: UserProfile,
        data: List[Dict[str, Union[int, str]]]=REQ(validator=check_list(
            check_dict([('id', check_int)])))) -> HttpResponse:
    for item in data:
        field_id = item['id']
        try:
            field = CustomProfileField.objects.get(id=field_id)
        except CustomProfileField.DoesNotExist:
            return json_error(_('Field id {id} not found.').format(id=field_id))

        validators = CustomProfileField.FIELD_VALIDATORS
        extended_validators = CustomProfileField.EXTENDED_FIELD_VALIDATORS
        field_type = field.field_type
        value = item['value']
        var_name = '{}'.format(field.name)
        if field_type in validators:
            validator = validators[field_type]
            result = validator(var_name, value)
        else:
            # Check extended validators.
            extended_validator = extended_validators[field_type]
            field_data = field.field_data
            result = extended_validator(var_name, field_data, value)

        if result is not None:
            return json_error(result)

    do_update_user_custom_profile_data(user_profile, data)
    # We need to call this explicitly otherwise constraints are not check
    return json_success()

def invite_users_backend(request: HttpRequest, user_profile: UserProfile,
                         invitee_emails_raw: str=REQ("invitee_emails"),
                         invite_as_admin: Optional[bool]=REQ(validator=check_bool, default=False),
                         ) -> HttpResponse:

    if user_profile.realm.invite_by_admins_only and not user_profile.is_realm_admin:
        return json_error(_("Must be an organization administrator"))
    if invite_as_admin and not user_profile.is_realm_admin:
        return json_error(_("Must be an organization administrator"))
    if not invitee_emails_raw:
        return json_error(_("You must specify at least one email address."))

    invitee_emails = get_invitee_emails_set(invitee_emails_raw)

    stream_names = request.POST.getlist('stream')
    if not stream_names:
        return json_error(_("You must specify at least one stream for invitees to join."))

    # We unconditionally sub you to the notifications stream if it
    # exists and is public.
    notifications_stream = user_profile.realm.notifications_stream  # type: Optional[Stream]
    if notifications_stream and not notifications_stream.invite_only:
        stream_names.append(notifications_stream.name)

    streams = []  # type: List[Stream]
    for stream_name in stream_names:
        try:
            (stream, recipient, sub) = access_stream_by_name(user_profile, stream_name)
        except JsonableError:
            return json_error(_("Stream does not exist: %s. No invites were sent.") % (stream_name,))
        streams.append(stream)

    do_invite_users(user_profile, invitee_emails, streams, invite_as_admin)
    return json_success()

def webathena_kerberos_login(request, user_profile,
                             cred=REQ(default=None)):
    # type: (HttpRequest, UserProfile, Text) -> HttpResponse
    if cred is None:
        return json_error(_("Could not find Kerberos credential"))
    if not user_profile.realm.webathena_enabled:
        return json_error(_("Webathena login not enabled"))

    try:
        parsed_cred = ujson.loads(cred)
        user = parsed_cred["cname"]["nameString"][0]
        if user == "golem":
            # Hack for an mit.edu user whose Kerberos username doesn't
            # match what he zephyrs as
            user = "ctl"
        assert(user == user_profile.email.split("@")[0])
        ccache = make_ccache(parsed_cred)
    except Exception:
        return json_error(_("Invalid Kerberos cache"))

    # TODO: Send these data via (say) rabbitmq
    try:
        subprocess.check_call(["ssh", settings.PERSONAL_ZMIRROR_SERVER, "--",
                               "/home/zulip/zulip/bots/process_ccache",
                               force_str(user),
                               force_str(user_profile.api_key),
                               force_str(base64.b64encode(ccache))])
    except Exception:
        logging.exception("Error updating the user's ccache")
        return json_error(_("We were unable to setup mirroring for you"))

    return json_success()

def webathena_kerberos_login(request: HttpRequest, user_profile: UserProfile,
                             cred: Text=REQ(default=None)) -> HttpResponse:
    global kerberos_alter_egos
    if cred is None:
        return json_error(_("Could not find Kerberos credential"))
    if not user_profile.realm.webathena_enabled:
        return json_error(_("Webathena login not enabled"))

    try:
        parsed_cred = ujson.loads(cred)
        user = parsed_cred["cname"]["nameString"][0]
        if user in kerberos_alter_egos:
            user = kerberos_alter_egos[user]
        assert(user == user_profile.email.split("@")[0])
        ccache = make_ccache(parsed_cred)
    except Exception:
        return json_error(_("Invalid Kerberos cache"))

    # TODO: Send these data via (say) rabbitmq
    try:
        subprocess.check_call(["ssh", settings.PERSONAL_ZMIRROR_SERVER, "--",
                               "/home/zulip/python-zulip-api/zulip/integrations/zephyr/process_ccache",
                               force_str(user),
                               force_str(user_profile.api_key),
                               force_str(base64.b64encode(ccache))])
    except Exception:
        logging.exception("Error updating the user's ccache")
        return json_error(_("We were unable to setup mirroring for you"))

    return json_success()

def process_submessage(request: HttpRequest,
                       user_profile: UserProfile,
                       message_id: int=REQ(validator=check_int),
                       msg_type: str=REQ(),
                       content: str=REQ(),
                       ) -> HttpResponse:
    message, user_message = access_message(user_profile, message_id)

    if not settings.ALLOW_SUB_MESSAGES:  # nocoverage
        msg = 'Feature not enabled'
        return json_error(msg)

    try:
        data = ujson.loads(content)
    except Exception:
        return json_error(_("Invalid json for submessage"))

    do_add_submessage(
        sender_id=user_profile.id,
        message_id=message.id,
        msg_type=msg_type,
        content=content,
        data=data,
    )
    return json_success()

def json_invite_users(request, user_profile,
                      invitee_emails_raw=REQ("invitee_emails"),
                      body=REQ("custom_body", default=None)):
    # type: (HttpRequest, UserProfile, str, Optional[str]) -> HttpResponse
    if not invitee_emails_raw:
        return json_error(_("You must specify at least one email address."))
    if body == '':
        body = None

    invitee_emails = get_invitee_emails_set(invitee_emails_raw)

    stream_names = request.POST.getlist('stream')
    if not stream_names:
        return json_error(_("You must specify at least one stream for invitees to join."))

    # We unconditionally sub you to the notifications stream if it
    # exists and is public.
    notifications_stream = user_profile.realm.notifications_stream  # type: Optional[Stream]
    if notifications_stream and not notifications_stream.invite_only:
        stream_names.append(notifications_stream.name)

    streams = []  # type: List[Stream]
    for stream_name in stream_names:
        try:
            (stream, recipient, sub) = access_stream_by_name(user_profile, stream_name)
        except JsonableError:
            return json_error(_("Stream does not exist: %s. No invites were sent.") % (stream_name,))
        streams.append(stream)

    ret_error, error_data = do_invite_users(user_profile, invitee_emails, streams, body)

    if ret_error is not None:
        return json_error(data=error_data, msg=ret_error)
    else:
        return json_success()