zulip/zerver/tests/test_decorators.py

# -*- coding: utf-8 -*-
import mock

from typing import Any, Dict, Iterable, List, Optional, Text, Tuple
from django.test import TestCase
from django.http import HttpResponse, HttpRequest
from django.test.client import RequestFactory
from django.conf import settings

from zerver.lib.actions import do_deactivate_realm, do_deactivate_user, \
    do_reactivate_user, do_reactivate_realm
from zerver.lib.initial_password import initial_password
from zerver.lib.test_helpers import (
    HostRequestMock,
)
from zerver.lib.test_classes import (
    ZulipTestCase,
    WebhookTestCase,
)
from zerver.lib.response import json_response
from zerver.lib.request import \
    REQ, has_request_variables, RequestVariableMissingError, \
    RequestVariableConversionError, JsonableError
from zerver.decorator import (
    api_key_only_webhook_view,
    authenticated_json_post_view, authenticated_json_view,
    authenticate_notify,
    get_client_name, internal_notify_view, is_local_addr,
    rate_limit, validate_api_key, logged_in_and_active,
    return_success_on_head_request
)
from zerver.lib.validator import (
    check_string, check_dict, check_dict_only, check_bool, check_float, check_int, check_list, Validator,
    check_variable_type, equals, check_none_or, check_url,
)
from zerver.models import \
    get_realm, get_user, UserProfile, Client, Realm

import ujson

class DecoratorTestCase(TestCase):
    def __init__(self, *args, **kwargs):
        # type: (*Any, **Any) -> None
        # This method should be removed when we migrate to version 3 of Python
        import six
        if six.PY2:
            self.assertRaisesRegex = self.assertRaisesRegexp
        super(TestCase, self).__init__(*args, **kwargs)

    def test_get_client_name(self):
        # type: () -> None
        class Request(object):
            def __init__(self, GET, POST, META):
                # type: (Dict[str, str], Dict[str, str], Dict[str, str]) -> None
                self.GET = GET
                self.POST = POST
                self.META = META

        req = Request(
            GET=dict(),
            POST=dict(),
            META=dict(),
        )

        self.assertEqual(get_client_name(req, is_json_view=True), 'website')
        self.assertEqual(get_client_name(req, is_json_view=False), 'Unspecified')

        req = Request(
            GET=dict(),
            POST=dict(),
            META=dict(HTTP_USER_AGENT='Mozilla/bla bla bla'),
        )

        self.assertEqual(get_client_name(req, is_json_view=True), 'website')
        self.assertEqual(get_client_name(req, is_json_view=False), 'Mozilla')

        req = Request(
            GET=dict(),
            POST=dict(),
            META=dict(HTTP_USER_AGENT='ZulipDesktop/bla bla bla'),
        )

        self.assertEqual(get_client_name(req, is_json_view=True), 'ZulipDesktop')
        self.assertEqual(get_client_name(req, is_json_view=False), 'ZulipDesktop')

        req = Request(
            GET=dict(client='fancy phone'),
            POST=dict(),
            META=dict(),
        )

        self.assertEqual(get_client_name(req, is_json_view=True), 'fancy phone')
        self.assertEqual(get_client_name(req, is_json_view=False), 'fancy phone')

    def test_REQ_converter(self):
        # type: () -> None

        def my_converter(data):
            # type: (str) -> List[str]
            lst = ujson.loads(data)
            if not isinstance(lst, list):
                raise ValueError('not a list')
            if 13 in lst:
                raise JsonableError('13 is an unlucky number!')
            return lst

        @has_request_variables
        def get_total(request, numbers=REQ(converter=my_converter)):
            # type: (HttpRequest, Iterable[int]) -> int
            return sum(numbers)

        class Request(object):
            GET = {}  # type: Dict[str, str]
            POST = {}  # type: Dict[str, str]

        request = Request()

        with self.assertRaises(RequestVariableMissingError):
            get_total(request)

        request.POST['numbers'] = 'bad_value'
        with self.assertRaises(RequestVariableConversionError) as cm:
            get_total(request)
        self.assertEqual(str(cm.exception), "Bad value for 'numbers': bad_value")

        request.POST['numbers'] = ujson.dumps('{fun: unfun}')
        with self.assertRaises(JsonableError) as cm:
            get_total(request)
        self.assertEqual(str(cm.exception), 'Bad value for \'numbers\': "{fun: unfun}"')

        request.POST['numbers'] = ujson.dumps([2, 3, 5, 8, 13, 21])
        with self.assertRaises(JsonableError) as cm:
            get_total(request)
        self.assertEqual(str(cm.exception), "13 is an unlucky number!")

        request.POST['numbers'] = ujson.dumps([1, 2, 3, 4, 5, 6])
        result = get_total(request)
        self.assertEqual(result, 21)

    def test_REQ_converter_and_validator_invalid(self):
        # type: () -> None
        with self.assertRaisesRegex(AssertionError, "converter and validator are mutually exclusive"):
            @has_request_variables
            def get_total(request, numbers=REQ(validator=check_list(check_int),
                                               converter=lambda: None)):
                # type: (HttpRequest, Iterable[int]) -> int
                return sum(numbers)  # nocoverage -- isn't intended to be run

    def test_REQ_validator(self):
        # type: () -> None

        @has_request_variables
        def get_total(request, numbers=REQ(validator=check_list(check_int))):
            # type: (HttpRequest, Iterable[int]) -> int
            return sum(numbers)

        class Request(object):
            GET = {}  # type: Dict[str, str]
            POST = {}  # type: Dict[str, str]

        request = Request()

        with self.assertRaises(RequestVariableMissingError):
            get_total(request)

        request.POST['numbers'] = 'bad_value'
        with self.assertRaises(JsonableError) as cm:
            get_total(request)
        self.assertEqual(str(cm.exception), 'argument "numbers" is not valid json.')

        request.POST['numbers'] = ujson.dumps([1, 2, "what?", 4, 5, 6])
        with self.assertRaises(JsonableError) as cm:
            get_total(request)
        self.assertEqual(str(cm.exception), 'numbers[2] is not an integer')

        request.POST['numbers'] = ujson.dumps([1, 2, 3, 4, 5, 6])
        result = get_total(request)
        self.assertEqual(result, 21)

    def test_REQ_argument_type(self):
        # type: () -> None
        @has_request_variables
        def get_payload(request, payload=REQ(argument_type='body')):
            # type: (HttpRequest, Dict[str, Dict]) -> Dict[str, Dict]
            return payload

        class MockRequest(object):
            body = {}  # type: Any

        request = MockRequest()

        request.body = 'notjson'
        with self.assertRaises(JsonableError) as cm:
            get_payload(request)
        self.assertEqual(str(cm.exception), 'Malformed JSON')

        request.body = '{"a": "b"}'
        self.assertEqual(get_payload(request), {'a': 'b'})

        # Test we properly handle an invalid argument_type.
        with self.assertRaises(Exception) as cm:
            @has_request_variables
            def test(request, payload=REQ(argument_type="invalid")):
                # type: (HttpRequest, Dict[str, Dict]) -> None
                pass  # nocoverage # this function isn't meant to be called
            test(request)

    def test_api_key_only_webhook_view(self):
        # type: () -> None
        @api_key_only_webhook_view('ClientName')
        def my_webhook(request, user_profile):
            # type: (HttpRequest, UserProfile) -> Text
            return user_profile.email

        class Request(HostRequestMock):
            GET = {}  # type: Dict[str, str]
            POST = {}  # type: Dict[str, str]
            COOKIES = {}  # type: Dict[str, str]
            META = {'PATH_INFO': ''}

        webhook_bot_email = 'webhook-bot@zulip.com'
        webhook_bot_realm = get_realm('zulip')
        webhook_bot = get_user(webhook_bot_email, webhook_bot_realm)
        webhook_bot_api_key = webhook_bot.api_key

        request = Request()  # type: Any
        request.host = settings.EXTERNAL_HOST

        request.POST['api_key'] = 'not_existing_api_key'
        with self.assertRaisesRegex(JsonableError, "Invalid API key"):
            my_webhook(request)

        # Start a valid request here
        request.POST['api_key'] = webhook_bot_api_key

        with self.settings(REALMS_HAVE_SUBDOMAINS=True):
            with mock.patch('logging.warning') as mock_warning:
                with self.assertRaisesRegex(JsonableError,
                                            "Account is not associated with this subdomain"):
                    api_result = my_webhook(request)

                mock_warning.assert_called_with(
                    "User {} attempted to access webhook API on wrong "
                    "subdomain {}".format(webhook_bot_email, ''))

            with mock.patch('logging.warning') as mock_warning:
                with self.assertRaisesRegex(JsonableError,
                                            "Account is not associated with this subdomain"):
                    request.host = "acme." + settings.EXTERNAL_HOST
                    api_result = my_webhook(request)

                mock_warning.assert_called_with(
                    "User {} attempted to access webhook API on wrong "
                    "subdomain {}".format(webhook_bot_email, 'acme'))

        with self.settings(RATE_LIMITING=True):
            with mock.patch('zerver.decorator.rate_limit_user') as rate_limit_mock:
                api_result = my_webhook(request)

        # Verify rate limiting was attempted.
        self.assertTrue(rate_limit_mock.called)

        # Verify decorator set the magic _email field used by some of our back end logging.
        self.assertEqual(request._email, webhook_bot_email)

        # Verify the main purpose of the decorator, which is that it passed in the
        # user_profile to my_webhook, allowing it return the correct
        # email for the bot (despite the API caller only knowing the API key).
        self.assertEqual(api_result, webhook_bot_email)

        # Now deactivate the user
        webhook_bot.is_active = False
        webhook_bot.save()
        with self.assertRaisesRegex(JsonableError, "Account not active"):
            my_webhook(request)

        # Reactive the user, but deactivate their realm.
        webhook_bot.is_active = True
        webhook_bot.save()
        webhook_bot.realm.deactivated = True
        webhook_bot.realm.save()
        with self.assertRaisesRegex(JsonableError, "Realm for account has been deactivated"):
            my_webhook(request)


class RateLimitTestCase(TestCase):
    def errors_disallowed(self):
        # type: () -> mock
        # Due to what is probably a hack in rate_limit(),
        # some tests will give a false positive (or succeed
        # for the wrong reason), unless we complain
        # about logging errors.  There might be a more elegant way
        # make logging errors fail than what I'm doing here.
        class TestLoggingErrorException(Exception):
            pass
        return mock.patch('logging.error', side_effect=TestLoggingErrorException)

    def test_internal_local_clients_skip_rate_limiting(self):
        # type: () -> None
        class Client(object):
            name = 'internal'

        class Request(object):
            client = Client()
            META = {'REMOTE_ADDR': '127.0.0.1'}

        req = Request()

        def f(req):
            # type: (Any) -> str
            return 'some value'

        f = rate_limit()(f)

        with self.settings(RATE_LIMITING=True):
            with mock.patch('zerver.decorator.rate_limit_user') as rate_limit_mock:
                with self.errors_disallowed():
                    self.assertEqual(f(req), 'some value')

        self.assertFalse(rate_limit_mock.called)

    def test_debug_clients_skip_rate_limiting(self):
        # type: () -> None
        class Client(object):
            name = 'internal'

        class Request(object):
            client = Client()
            META = {'REMOTE_ADDR': '3.3.3.3'}

        req = Request()

        def f(req):
            # type: (Any) -> str
            return 'some value'

        f = rate_limit()(f)

        with self.settings(RATE_LIMITING=True):
            with mock.patch('zerver.decorator.rate_limit_user') as rate_limit_mock:
                with self.errors_disallowed():
                    with self.settings(DEBUG_RATE_LIMITING=True):
                        self.assertEqual(f(req), 'some value')

        self.assertFalse(rate_limit_mock.called)

    def test_rate_limit_setting_of_false_bypasses_rate_limiting(self):
        # type: () -> None
        class Client(object):
            name = 'external'

        class Request(object):
            client = Client()
            META = {'REMOTE_ADDR': '3.3.3.3'}
            user = 'stub'  # any non-None value here exercises the correct code path

        req = Request()

        def f(req):
            # type: (Any) -> str
            return 'some value'

        f = rate_limit()(f)

        with self.settings(RATE_LIMITING=False):
            with mock.patch('zerver.decorator.rate_limit_user') as rate_limit_mock:
                with self.errors_disallowed():
                    self.assertEqual(f(req), 'some value')

        self.assertFalse(rate_limit_mock.called)

    def test_rate_limiting_happens_in_normal_case(self):
        # type: () -> None
        class Client(object):
            name = 'external'

        class Request(object):
            client = Client()
            META = {'REMOTE_ADDR': '3.3.3.3'}
            user = 'stub'  # any non-None value here exercises the correct code path

        req = Request()

        def f(req):
            # type: (Any) -> str
            return 'some value'

        f = rate_limit()(f)

        with self.settings(RATE_LIMITING=True):
            with mock.patch('zerver.decorator.rate_limit_user') as rate_limit_mock:
                with self.errors_disallowed():
                    self.assertEqual(f(req), 'some value')

        self.assertTrue(rate_limit_mock.called)

class ValidatorTestCase(TestCase):
    def test_check_string(self):
        # type: () -> None
        x = "hello"  # type: Any
        self.assertEqual(check_string('x', x), None)

        x = 4
        self.assertEqual(check_string('x', x), 'x is not a string')

    def test_check_bool(self):
        # type: () -> None
        x = True  # type: Any
        self.assertEqual(check_bool('x', x), None)

        x = 4
        self.assertEqual(check_bool('x', x), 'x is not a boolean')

    def test_check_int(self):
        # type: () -> None
        x = 5  # type: Any
        self.assertEqual(check_int('x', x), None)

        x = [{}]
        self.assertEqual(check_int('x', x), 'x is not an integer')

    def test_check_float(self):
        # type: () -> None
        x = 5.5  # type: Any
        self.assertEqual(check_float('x', x), None)

        x = 5
        self.assertEqual(check_float('x', x), 'x is not a float')

        x = [{}]
        self.assertEqual(check_float('x', x), 'x is not a float')

    def test_check_list(self):
        # type: () -> None
        x = 999  # type: Any
        error = check_list(check_string)('x', x)
        self.assertEqual(error, 'x is not a list')

        x = ["hello", 5]
        error = check_list(check_string)('x', x)
        self.assertEqual(error, 'x[1] is not a string')

        x = [["yo"], ["hello", "goodbye", 5]]
        error = check_list(check_list(check_string))('x', x)
        self.assertEqual(error, 'x[1][2] is not a string')

        x = ["hello", "goodbye", "hello again"]
        error = check_list(check_string, length=2)('x', x)
        self.assertEqual(error, 'x should have exactly 2 items')

    def test_check_dict(self):
        # type: () -> None
        keys = [
            ('names', check_list(check_string)),
            ('city', check_string),
        ]  # type: List[Tuple[str, Validator]]

        x = {
            'names': ['alice', 'bob'],
            'city': 'Boston',
        }  # type: Any
        error = check_dict(keys)('x', x)
        self.assertEqual(error, None)

        x = 999
        error = check_dict(keys)('x', x)
        self.assertEqual(error, 'x is not a dict')

        x = {}
        error = check_dict(keys)('x', x)
        self.assertEqual(error, 'names key is missing from x')

        x = {
            'names': ['alice', 'bob', {}]
        }
        error = check_dict(keys)('x', x)
        self.assertEqual(error, 'x["names"][2] is not a string')

        x = {
            'names': ['alice', 'bob'],
            'city': 5
        }
        error = check_dict(keys)('x', x)
        self.assertEqual(error, 'x["city"] is not a string')

        # test dict_only
        x = {
            'names': ['alice', 'bob'],
            'city': 'Boston',
        }
        error = check_dict_only(keys)('x', x)
        self.assertEqual(error, None)

        x = {
            'names': ['alice', 'bob'],
            'city': 'Boston',
            'state': 'Massachusetts',
        }
        error = check_dict_only(keys)('x', x)
        self.assertEqual(error, 'Unexpected arguments: state')

    def test_encapsulation(self):
        # type: () -> None
        # There might be situations where we want deep
        # validation, but the error message should be customized.
        # This is an example.
        def check_person(val):
            # type: (Any) -> Optional[str]
            error = check_dict([
                ('name', check_string),
                ('age', check_int),
            ])('_', val)
            if error:
                return 'This is not a valid person'
            return None

        person = {'name': 'King Lear', 'age': 42}
        self.assertEqual(check_person(person), None)

        nonperson = 'misconfigured data'
        self.assertEqual(check_person(nonperson), 'This is not a valid person')

    def test_check_variable_type(self):
        # type: () -> None
        x = 5  # type: Any
        self.assertEqual(check_variable_type([check_string, check_int])('x', x), None)

        x = 'x'
        self.assertEqual(check_variable_type([check_string, check_int])('x', x), None)

        x = [{}]
        self.assertEqual(check_variable_type([check_string, check_int])('x', x), 'x is not an allowed_type')

    def test_equals(self):
        # type: () -> None
        x = 5  # type: Any
        self.assertEqual(equals(5)('x', x), None)
        self.assertEqual(equals(6)('x', x), 'x != 6 (5 is wrong)')

    def test_check_none_or(self):
        # type: () -> None
        x = 5  # type: Any
        self.assertEqual(check_none_or(check_int)('x', x), None)
        x = None
        self.assertEqual(check_none_or(check_int)('x', x), None)
        x = 'x'
        self.assertEqual(check_none_or(check_int)('x', x), 'x is not an integer')

    def test_check_url(self):
        # type: () -> None
        url = "http://127.0.0.1:5002/"  # type: Any
        check_url('url', url)

        url = "http://zulip-bots.example.com/"
        check_url('url', url)

        url = "http://127.0.0"
        with self.assertRaises(JsonableError):
            check_url('url', url)

class DeactivatedRealmTest(ZulipTestCase):
    def test_send_deactivated_realm(self):
        # type: () -> None
        """
        rest_dispatch rejects requests in a deactivated realm, both /json and api

        """
        realm = get_realm("zulip")
        do_deactivate_realm(get_realm("zulip"))

        result = self.client_post("/json/messages", {"type": "private",
                                                     "content": "Test message",
                                                     "client": "test suite",
                                                     "to": self.example_email("othello")})
        self.assert_json_error_contains(result, "Not logged in", status_code=401)

        # Even if a logged-in session was leaked, it still wouldn't work
        realm.deactivated = False
        realm.save()
        self.login(self.example_email("hamlet"))
        realm.deactivated = True
        realm.save()

        result = self.client_post("/json/messages", {"type": "private",
                                                     "content": "Test message",
                                                     "client": "test suite",
                                                     "to": self.example_email("othello")})
        self.assert_json_error_contains(result, "has been deactivated", status_code=400)

        result = self.client_post("/api/v1/messages", {"type": "private",
                                                       "content": "Test message",
                                                       "client": "test suite",
                                                       "to": self.example_email("othello")},
                                  **self.api_auth(self.example_email("hamlet")))
        self.assert_json_error_contains(result, "has been deactivated", status_code=401)

    def test_fetch_api_key_deactivated_realm(self):
        # type: () -> None
        """
        authenticated_json_view views fail in a deactivated realm

        """
        realm = get_realm("zulip")
        user_profile = self.example_user('hamlet')
        email = user_profile.email
        test_password = "abcd1234"
        user_profile.set_password(test_password)

        self.login(email)
        realm.deactivated = True
        realm.save()
        result = self.client_post("/json/fetch_api_key", {"password": test_password})
        self.assert_json_error_contains(result, "has been deactivated", status_code=400)

    def test_login_deactivated_realm(self):
        # type: () -> None
        """
        logging in fails in a deactivated realm

        """
        do_deactivate_realm(get_realm("zulip"))
        result = self.login_with_return(self.example_email("hamlet"))
        self.assert_in_response("has been deactivated", result)

    def test_webhook_deactivated_realm(self):
        # type: () -> None
        """
        Using a webhook while in a deactivated realm fails

        """
        do_deactivate_realm(get_realm("zulip"))
        email = self.example_email("hamlet")
        api_key = self.get_api_key(email)
        url = "/api/v1/external/jira?api_key=%s&stream=jira_custom" % (api_key,)
        data = self.fixture_data('jira', "created_v2")
        result = self.client_post(url, data,
                                  content_type="application/json")
        self.assert_json_error_contains(result, "has been deactivated", status_code=400)

class LoginRequiredTest(ZulipTestCase):
    def test_login_required(self):
        # type: () -> None
        """
        Verifies the zulip_login_required decorator blocks deactivated users.
        """
        user_profile = self.example_user('hamlet')
        email = user_profile.email

        # Verify fails if logged-out
        result = self.client_get('/accounts/accept_terms/')
        self.assertEqual(result.status_code, 302)

        # Verify succeeds once logged-in
        self.login(email)
        result = self.client_get('/accounts/accept_terms/')
        self.assert_in_response("I agree to the", result)

        # Verify fails if user deactivated (with session still valid)
        user_profile.is_active = False
        user_profile.save()
        result = self.client_get('/accounts/accept_terms/')
        self.assertEqual(result.status_code, 302)

        # Verify succeeds if user reactivated
        do_reactivate_user(user_profile)
        self.login(email)
        result = self.client_get('/accounts/accept_terms/')
        self.assert_in_response("I agree to the", result)

        # Verify fails if realm deactivated
        user_profile.realm.deactivated = True
        user_profile.realm.save()
        result = self.client_get('/accounts/accept_terms/')
        self.assertEqual(result.status_code, 302)

class FetchAPIKeyTest(ZulipTestCase):
    def test_fetch_api_key_success(self):
        # type: () -> None
        email = self.example_email("cordelia")

        self.login(email)
        result = self.client_post("/json/fetch_api_key", {"password": initial_password(email)})
        self.assert_json_success(result)

    def test_fetch_api_key_wrong_password(self):
        # type: () -> None
        email = self.example_email("cordelia")

        self.login(email)
        result = self.client_post("/json/fetch_api_key", {"password": "wrong_password"})
        self.assert_json_error_contains(result, "password is incorrect")

class InactiveUserTest(ZulipTestCase):
    def test_send_deactivated_user(self):
        # type: () -> None
        """
        rest_dispatch rejects requests from deactivated users, both /json and api

        """
        user_profile = self.example_user('hamlet')
        email = user_profile.email
        self.login(email)
        do_deactivate_user(user_profile)

        result = self.client_post("/json/messages", {"type": "private",
                                                     "content": "Test message",
                                                     "client": "test suite",
                                                     "to": self.example_email("othello")})
        self.assert_json_error_contains(result, "Not logged in", status_code=401)

        # Even if a logged-in session was leaked, it still wouldn't work
        do_reactivate_user(user_profile)
        self.login(email)
        user_profile.is_active = False
        user_profile.save()

        result = self.client_post("/json/messages", {"type": "private",
                                                     "content": "Test message",
                                                     "client": "test suite",
                                                     "to": self.example_email("othello")})
        self.assert_json_error_contains(result, "Account not active", status_code=400)

        result = self.client_post("/api/v1/messages", {"type": "private",
                                                       "content": "Test message",
                                                       "client": "test suite",
                                                       "to": self.example_email("othello")},
                                  **self.api_auth(self.example_email("hamlet")))
        self.assert_json_error_contains(result, "Account not active", status_code=401)

    def test_fetch_api_key_deactivated_user(self):
        # type: () -> None
        """
        authenticated_json_view views fail with a deactivated user

        """
        user_profile = self.example_user('hamlet')
        email = user_profile.email
        test_password = "abcd1234"
        user_profile.set_password(test_password)
        user_profile.save()

        self.login(email, password=test_password)
        user_profile.is_active = False
        user_profile.save()
        result = self.client_post("/json/fetch_api_key", {"password": test_password})
        self.assert_json_error_contains(result, "Account not active", status_code=400)

    def test_login_deactivated_user(self):
        # type: () -> None
        """
        logging in fails with an inactive user

        """
        user_profile = self.example_user('hamlet')
        do_deactivate_user(user_profile)

        result = self.login_with_return(self.example_email("hamlet"))
        self.assert_in_response("Please enter a correct email and password", result)

    def test_webhook_deactivated_user(self):
        # type: () -> None
        """
        Deactivated users can't use webhooks

        """
        user_profile = self.example_user('hamlet')
        email = user_profile.email
        do_deactivate_user(user_profile)

        api_key = self.get_api_key(email)
        url = "/api/v1/external/jira?api_key=%s&stream=jira_custom" % (api_key,)
        data = self.fixture_data('jira', "created_v2")
        result = self.client_post(url, data,
                                  content_type="application/json")
        self.assert_json_error_contains(result, "Account not active", status_code=400)


class TestValidateApiKey(ZulipTestCase):
    def setUp(self):
        # type: () -> None
        zulip_realm = get_realm('zulip')
        self.webhook_bot = get_user('webhook-bot@zulip.com', zulip_realm)
        self.default_bot = get_user('default-bot@zulip.com', zulip_realm)

    def test_validate_api_key_if_profile_does_not_exist(self):
        # type: () -> None
        with self.assertRaises(JsonableError):
            validate_api_key(HostRequestMock(), 'email@doesnotexist.com', 'api_key')

    def test_validate_api_key_if_api_key_does_not_match_profile_api_key(self):
        # type: () -> None
        with self.assertRaises(JsonableError):
            validate_api_key(HostRequestMock(), self.webhook_bot.email, 'not_32_length')

        with self.assertRaises(JsonableError):
            validate_api_key(HostRequestMock(), self.webhook_bot.email, self.default_bot.api_key)

    def test_validate_api_key_if_profile_is_not_active(self):
        # type: () -> None
        self._change_is_active_field(self.default_bot, False)
        with self.assertRaises(JsonableError):
            validate_api_key(HostRequestMock(), self.default_bot.email, self.default_bot.api_key)
        self._change_is_active_field(self.default_bot, True)

    def test_validate_api_key_if_profile_is_incoming_webhook_and_is_webhook_is_unset(self):
        # type: () -> None
        with self.assertRaises(JsonableError):
            validate_api_key(HostRequestMock(), self.webhook_bot.email, self.webhook_bot.api_key)

    def test_validate_api_key_if_profile_is_incoming_webhook_and_is_webhook_is_set(self):
        # type: () -> None
        profile = validate_api_key(HostRequestMock(), self.webhook_bot.email, self.webhook_bot.api_key, is_webhook=True)
        self.assertEqual(profile.pk, self.webhook_bot.pk)

    def test_valid_api_key_if_user_is_on_wrong_subdomain(self):
        # type: () -> None
        with self.settings(REALMS_HAVE_SUBDOMAINS=True):
            with self.settings(RUNNING_INSIDE_TORNADO=False):
                with mock.patch('logging.warning') as mock_warning:
                    with self.assertRaisesRegex(JsonableError,
                                                "Account is not associated with this subdomain"):
                        validate_api_key(HostRequestMock(host=settings.EXTERNAL_HOST),
                                         self.default_bot.email,
                                         self.default_bot.api_key)

                    mock_warning.assert_called_with(
                        "User {} attempted to access API on wrong "
                        "subdomain {}".format(self.default_bot.email, ''))

                with mock.patch('logging.warning') as mock_warning:
                    with self.assertRaisesRegex(JsonableError,
                                                "Account is not associated with this subdomain"):
                        validate_api_key(HostRequestMock(host='acme.' + settings.EXTERNAL_HOST),
                                         self.default_bot.email,
                                         self.default_bot.api_key)

                    mock_warning.assert_called_with(
                        "User {} attempted to access API on wrong "
                        "subdomain {}".format(self.default_bot.email, 'acme'))

    def _change_is_active_field(self, profile, value):
        # type: (UserProfile, bool) -> None
        profile.is_active = value
        profile.save()

class TestInternalNotifyView(TestCase):
    BORING_RESULT = 'boring'

    class Request(object):
        def __init__(self, POST, META):
            # type: (Dict, Dict) -> None
            self.POST = POST
            self.META = META
            self.method = 'POST'

    def internal_notify(self, is_tornado, req):
        # type: (bool, HttpRequest) -> HttpResponse
        boring_view = lambda req: self.BORING_RESULT
        return internal_notify_view(is_tornado)(boring_view)(req)

    def test_valid_internal_requests(self):
        # type: () -> None
        secret = 'random'
        req = self.Request(
            POST=dict(secret=secret),
            META=dict(REMOTE_ADDR='127.0.0.1'),
        )

        with self.settings(SHARED_SECRET=secret):
            self.assertTrue(authenticate_notify(req))
            self.assertEqual(self.internal_notify(False, req), self.BORING_RESULT)
            self.assertEqual(req._email, 'internal')

            with self.assertRaises(RuntimeError):
                self.internal_notify(True, req)

        req._tornado_handler = 'set'
        with self.settings(SHARED_SECRET=secret):
            self.assertTrue(authenticate_notify(req))
            self.assertEqual(self.internal_notify(True, req), self.BORING_RESULT)
            self.assertEqual(req._email, 'internal')

            with self.assertRaises(RuntimeError):
                self.internal_notify(False, req)

    def test_internal_requests_with_broken_secret(self):
        # type: () -> None
        secret = 'random'
        req = self.Request(
            POST=dict(secret=secret),
            META=dict(REMOTE_ADDR='127.0.0.1'),
        )

        with self.settings(SHARED_SECRET='broken'):
            self.assertFalse(authenticate_notify(req))
            self.assertEqual(self.internal_notify(True, req).status_code, 403)

    def test_external_requests(self):
        # type: () -> None
        secret = 'random'
        req = self.Request(
            POST=dict(secret=secret),
            META=dict(REMOTE_ADDR='3.3.3.3'),
        )

        with self.settings(SHARED_SECRET=secret):
            self.assertFalse(authenticate_notify(req))
            self.assertEqual(self.internal_notify(True, req).status_code, 403)

    def test_is_local_address(self):
        # type: () -> None
        self.assertTrue(is_local_addr('127.0.0.1'))
        self.assertTrue(is_local_addr('::1'))
        self.assertFalse(is_local_addr('42.43.44.45'))

class TestHumanUsersOnlyDecorator(ZulipTestCase):
    def test_human_only_endpoints(self):
        # type: () -> None
        post_endpoints = [
            "/api/v1/users/me/presence",
        ]
        for endpoint in post_endpoints:
            result = self.client_post(endpoint, **self.api_auth('default-bot@zulip.com'))
            self.assert_json_error(result, "This endpoint does not accept bot requests.")

        patch_endpoints = [
            "/api/v1/settings/display",
            "/api/v1/settings/notifications",
            "/api/v1/settings/ui",
        ]
        for endpoint in patch_endpoints:
            result = self.client_patch(endpoint, **self.api_auth('default-bot@zulip.com'))
            self.assert_json_error(result, "This endpoint does not accept bot requests.")

class TestAuthenticatedJsonPostViewDecorator(ZulipTestCase):
    def test_authenticated_json_post_view_if_everything_is_correct(self):
        # type: () -> None
        user_email = self.example_email('hamlet')
        user_realm = get_realm('zulip')
        self._login(user_email, user_realm)
        response = self._do_test(user_email)
        self.assertEqual(response.status_code, 200)

    def test_authenticated_json_post_view_if_subdomain_is_invalid(self):
        # type: () -> None
        user_email = self.example_email('hamlet')
        user_realm = get_realm('zulip')
        self._login(user_email, user_realm)
        with self.settings(REALMS_HAVE_SUBDOMAINS=True):
            with mock.patch('logging.warning') as mock_warning, \
                    mock.patch('zerver.decorator.get_subdomain', return_value=''):
                self.assert_json_error_contains(self._do_test(user_email),
                                                "Account is not associated with this "
                                                "subdomain")
                mock_warning.assert_called_with(
                    "User {} attempted to access JSON API on wrong "
                    "subdomain {}".format(user_email, ''))

            with mock.patch('logging.warning') as mock_warning, \
                    mock.patch('zerver.decorator.get_subdomain', return_value='acme'):
                self.assert_json_error_contains(self._do_test(user_email),
                                                "Account is not associated with this "
                                                "subdomain")
                mock_warning.assert_called_with(
                    "User {} attempted to access JSON API on wrong "
                    "subdomain {}".format(user_email, 'acme'))

    def test_authenticated_json_post_view_if_user_is_incoming_webhook(self):
        # type: () -> None
        user_email = 'webhook-bot@zulip.com'
        user_realm = get_realm('zulip')
        self._login(user_email, user_realm, password="test")  # we set a password because user is a bot
        self.assert_json_error_contains(self._do_test(user_email), "Webhook bots can only access webhooks")

    def test_authenticated_json_post_view_if_user_is_not_active(self):
        # type: () -> None
        user_email = self.example_email('hamlet')
        user_realm = get_realm('zulip')
        self._login(user_email, user_realm, password="test")
        # Get user_profile after _login so that we have the latest data.
        user_profile = get_user(user_email, user_realm)
        # we deactivate user manually because do_deactivate_user removes user session
        user_profile.is_active = False
        user_profile.save()
        self.assert_json_error_contains(self._do_test(user_email), "Account not active")
        do_reactivate_user(user_profile)

    def test_authenticated_json_post_view_if_user_realm_is_deactivated(self):
        # type: () -> None
        user_email = self.example_email('hamlet')
        user_realm = get_realm('zulip')
        user_profile = get_user(user_email, user_realm)
        self._login(user_email, user_realm)
        # we deactivate user's realm manually because do_deactivate_user removes user session
        user_profile.realm.deactivated = True
        user_profile.realm.save()
        self.assert_json_error_contains(self._do_test(user_email), "Realm for account has been deactivated")
        do_reactivate_realm(user_profile.realm)

    def _do_test(self, user_email):
        # type: (Text) -> HttpResponse
        data = {"status": '"started"'}
        return self.client_post(r'/json/tutorial_status', data)

    def _login(self, user_email, user_realm, password=None):
        # type: (Text, Realm, str) -> None
        if password:
            user_profile = get_user(user_email, user_realm)
            user_profile.set_password(password)
            user_profile.save()
        self.login(user_email, password)

class TestAuthenticatedJsonViewDecorator(ZulipTestCase):
    def test_authenticated_json_view_if_subdomain_is_invalid(self):
        # type: () -> None
        user_email = self.example_email("hamlet")
        self.login(user_email)
        with self.settings(REALMS_HAVE_SUBDOMAINS=True):
            with mock.patch('logging.warning') as mock_warning, \
                    mock.patch('zerver.decorator.get_subdomain', return_value=''):
                self.assert_json_error_contains(self._do_test(str(user_email)),
                                                "Account is not associated with this "
                                                "subdomain")
                mock_warning.assert_called_with(
                    "User {} attempted to access JSON API on wrong "
                    "subdomain {}".format(user_email, ''))

            with mock.patch('logging.warning') as mock_warning, \
                    mock.patch('zerver.decorator.get_subdomain', return_value='acme'):
                self.assert_json_error_contains(self._do_test(str(user_email)),
                                                "Account is not associated with this "
                                                "subdomain")
                mock_warning.assert_called_with(
                    "User {} attempted to access JSON API on wrong "
                    "subdomain {}".format(user_email, 'acme'))

    def _do_test(self, user_email):
        # type: (str) -> HttpResponse
        data = {"status": '"started"'}
        return self.client_post(r'/json/tutorial_status', data)

class TestZulipLoginRequiredDecorator(ZulipTestCase):
    def test_zulip_login_required_if_subdomain_is_invalid(self):
        # type: () -> None
        user_email = self.example_email("hamlet")
        self.login(user_email)

        with self.settings(REALMS_HAVE_SUBDOMAINS=True):
            with mock.patch('zerver.decorator.get_subdomain', return_value='zulip'):
                result = self.client_get('/accounts/accept_terms/')
                self.assertEqual(result.status_code, 200)

            with mock.patch('zerver.decorator.get_subdomain', return_value=''):
                result = self.client_get('/accounts/accept_terms/')
                self.assertEqual(result.status_code, 302)

            with mock.patch('zerver.decorator.get_subdomain', return_value='acme'):
                result = self.client_get('/accounts/accept_terms/')
                self.assertEqual(result.status_code, 302)

class TestRequireServerAdminDecorator(ZulipTestCase):
    def test_require_server_admin_decorator(self):
        # type: () -> None
        user_email = self.example_email('hamlet')
        user_realm = get_realm('zulip')
        self.login(user_email)

        result = self.client_get('/activity')
        self.assertEqual(result.status_code, 302)

        user_profile = get_user(user_email, user_realm)
        user_profile.is_staff = True
        user_profile.save()

        result = self.client_get('/activity')
        self.assertEqual(result.status_code, 200)

class ReturnSuccessOnHeadRequestDecorator(ZulipTestCase):
    def test_return_success_on_head_request_returns_200_if_request_method_is_head(self):
        # type: () -> None
        class HeadRequest(object):
            method = 'HEAD'

        request = HeadRequest()

        @return_success_on_head_request
        def test_function(request):
            # type: (HttpRequest) -> HttpResponse
            return json_response(msg=u'from_test_function')  # nocoverage. isn't meant to be called

        response = test_function(request)
        self.assert_json_success(response)
        self.assertNotEqual(ujson.loads(response.content).get('msg'), u'from_test_function')

    def test_return_success_on_head_request_returns_normal_response_if_request_method_is_not_head(self):
            # type: () -> None
            class HeadRequest(object):
                method = 'POST'

            request = HeadRequest()

            @return_success_on_head_request
            def test_function(request):
                # type: (HttpRequest) -> HttpResponse
                return json_response(msg=u'from_test_function')

            response = test_function(request)
            self.assertEqual(ujson.loads(response.content).get('msg'), u'from_test_function')

class RestAPITest(ZulipTestCase):
    def test_method_not_allowed(self):
        # type: () -> None
        self.login(self.example_email("hamlet"))
        result = self.client_patch('/json/users')
        self.assertEqual(result.status_code, 405)
        self.assert_in_response('Method Not Allowed', result)

    def test_options_method(self):
        # type: () -> None
        self.login(self.example_email("hamlet"))
        result = self.client_options('/json/users')
        self.assertEqual(result.status_code, 204)
        self.assertEqual(str(result['Allow']), 'GET, POST')

        result = self.client_options('/json/streams/15')
        self.assertEqual(result.status_code, 204)
        self.assertEqual(str(result['Allow']), 'DELETE, PATCH')

    def test_http_accept_redirect(self):
        # type: () -> None
        result = self.client_get('/json/users',
                                 HTTP_ACCEPT='text/html')
        self.assertEqual(result.status_code, 302)
        self.assertTrue(result["Location"].endswith("/login/?next=/json/users"))