84353759-c057-4f20-b282-724c34504dc9

Committed 26 Nov 2025 04:22PM UTC coverage: 89.192% (+0.4%) from 88.772%

Build # 84353759-c057-4f20-b282-724c34504dc9

Build Type

Pull #6049

circleci

Committed by

jwhitlock

Commit Message

Update TermsAcceptedUserViewTest for new errors

Pull Request Pull Request #6049: fix(relay): Create alternate bearer token auth for FxA (MPP-3505)

Run Details

3016 of 4041 branches covered (74.63%)

Branch coverage included in aggregate %.

1334 of 1349 new or added lines in 9 files covered. (98.89%)

6 existing lines in 2 files now uncovered.

19067 of 20718 relevant lines covered (92.03%)

11.09 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

99.5

/api/tests/authentication_tests.py

import re
from collections.abc import Iterator
from datetime import UTC, datetime, timedelta
from typing import Any
from unittest.mock import Mock, patch

from django.conf import settings
from django.contrib.auth.models import AnonymousUser, User
from django.core.cache import BaseCache

import pytest
import responses
from allauth.socialaccount.models import SocialAccount, SocialApp
from pytest_django.fixtures import SettingsWrapper
from requests import ReadTimeout
from rest_framework.exceptions import AuthenticationFailed
from rest_framework.test import APIRequestFactory

from ..authentication import (
    INTROSPECT_ERROR,
    INTROSPECT_TOKEN_URL,
    FxaIntrospectData,
    FxaTokenAuthentication,
    IntrospectAuthenticationFailed,
    IntrospectionError,
    IntrospectionResponse,
    IntrospectUnavailable,
    as_b64,
    get_cache_key,
    introspect_and_cache_token,
    introspect_token,
    load_introspection_result_from_cache,
)


def _create_fxa_introspect_response(
    active: bool = True,
    uid: str | None = "an-fxa-id",
    scope: str | None = None,
    expiration: bool = True,
    error: str | None = None,
) -> FxaIntrospectData:
    """Create a Mozilla Accounts introspection response"""
    if error:
        return {"error": error}
    data: FxaIntrospectData = {"active": active}
    if uid:
        data["sub"] = uid
    data["scope"] = f"profile {settings.RELAY_SCOPE}" if scope is None else scope
    if expiration:
        now_time = int(datetime.now().timestamp())
        exp_ms = (now_time + 60 * 60) * 1000  # Time in milliseconds
        data["exp"] = exp_ms
    return data


def _mock_fxa_introspect_response(
    status_code: int = 200,
    data: FxaIntrospectData | str | None = None,
    timeout: bool = False,
    exception: Exception | None = None,
) -> responses.BaseResponse:
    """Mock the response to a Mozilla Accounts introspection request"""
    if timeout:
        return responses.add(
            responses.POST, INTROSPECT_TOKEN_URL, body=ReadTimeout("FxA is slow today")
        )
    if exception:
        return responses.add(responses.POST, INTROSPECT_TOKEN_URL, body=exception)
    return responses.add(
        responses.POST,
        INTROSPECT_TOKEN_URL,
        status=status_code,
        json=data,
    )


class MockTimer:
    """Mocked version of codetiming.Timer."""

    def __init__(self, logger: Any) -> None:
        assert logger is None

    def __enter__(self) -> "MockTimer":
        return self

    def __exit__(self, *exc_info: Any) -> None:
        self.last = 0.5


@pytest.fixture(autouse=True)
def mock_timer() -> Iterator[Mock]:
    with patch("api.authentication.Timer", side_effect=MockTimer) as MockedTimer:
        yield MockedTimer


@pytest.fixture(autouse=True)
def auth_2025_settings(settings: SettingsWrapper) -> SettingsWrapper:
    settings.FXA_TOKEN_AUTH_VERSION = "2025"
    return settings


def setup_fxa_introspect(
    status_code: int = 200,
    no_body: bool = False,
    text_body: str | None = None,
    active: bool = True,
    uid: str | None = "an-fxa-id",
    scope: str | None = None,
    expiration: bool = True,
    error: str | None = None,
    timeout: bool = False,
    exception: Exception | None = None,
) -> tuple[responses.BaseResponse, FxaIntrospectData | None]:
    """
    Mock a Mozilla Accounts introspection response. Return both the
    request-level response mock (to check how often the request was made)
    and the mocked response body.
    """
    data: FxaIntrospectData | None = None
    if no_body:
        mock_response = _mock_fxa_introspect_response(status_code)
    elif text_body:
        mock_response = _mock_fxa_introspect_response(status_code, text_body)
    elif timeout:
        mock_response = _mock_fxa_introspect_response(timeout=True)
    elif exception:
        mock_response = _mock_fxa_introspect_response(exception=exception)
    else:
        data = _create_fxa_introspect_response(
            active=active, uid=uid, scope=scope, expiration=expiration, error=error
        )
        mock_response = _mock_fxa_introspect_response(status_code, data)
    return mock_response, data


def future_exp() -> int:
    """Create an exp value (timestamp in ms) in the future"""
    future = datetime.now() + timedelta(days=1)
    return int(future.timestamp()) * 1000


_INTROSPECTION_RESPONSE_VALUEERROR_TEST_CASES = {
    "active_missing": ({}, "active should be true"),
    "active_false": ({"active": False}, "active should be true"),
    "sub_missing": ({"active": True}, "sub (FxA ID) should be set"),
    "sub_not_str": ({"active": True, "sub": 1}, "sub (FxA ID) should be set"),
    "sub_empty": ({"active": True, "sub": ""}, "sub (FxA ID) should be set"),
    "exp_not_int": (
        {"active": True, "sub": "s", "exp": "123"},
        "exp (Expiration timestamp in milliseconds) should be int",
    ),
    "missing_scope": (
        {"active": True, "sub": "s", "exp": 123},
        f"scope should include {settings.RELAY_SCOPE!r}",
    ),
    "wrong_scope": (
        {"active": True, "sub": "s", "exp": 123, "scope": "foo"},
        f"scope should include {settings.RELAY_SCOPE!r}",
    ),
}


@pytest.mark.parametrize(
    "data,message",
    _INTROSPECTION_RESPONSE_VALUEERROR_TEST_CASES.values(),
    ids=_INTROSPECTION_RESPONSE_VALUEERROR_TEST_CASES.keys(),
)
def test_introspection_response_init_bad_data_raises_value_error(
    data: FxaIntrospectData, message: str
) -> None:
    with pytest.raises(ValueError, match=re.escape(message)):
        IntrospectionResponse("token", data)


def test_introspection_response_with_expiration() -> None:
    data = _create_fxa_introspect_response()
    response = IntrospectionResponse("token", data)
    assert repr(response) == (
        "IntrospectionResponse(token='token',"
        " data={'active': True,"
        " 'sub': 'an-fxa-id',"
        f" 'scope': 'profile {settings.RELAY_SCOPE}',"
        f" 'exp': {data['exp']}"
        "},"
        " from_cache=False,"
        " request_s=None)"
    )
    assert 3530 < response.cache_timeout <= 3600  # about 60 minutes
    assert not response.is_expired


def test_introspection_response_repr_from_cache() -> None:
    data = _create_fxa_introspect_response(uid="other-fxa-id")
    response = IntrospectionResponse("token", data, from_cache=True)
    assert repr(response) == (
        "IntrospectionResponse(token='token',"
        " data={'active': True,"
        " 'sub': 'other-fxa-id',"
        f" 'scope': 'profile {settings.RELAY_SCOPE}',"
        f" 'exp': {data['exp']}"
        "},"
        " from_cache=True,"
        " request_s=None)"
    )
    assert 3530 < response.cache_timeout <= 3600  # about 60 minutes
    assert not response.is_expired


def test_introspection_response_repr_with_request_s() -> None:
    data = _create_fxa_introspect_response()
    response = IntrospectionResponse("token", data, request_s=0.23)
    assert repr(response) == (
        "IntrospectionResponse(token='token',"
        " data={'active': True,"
        " 'sub': 'an-fxa-id',"
        f" 'scope': 'profile {settings.RELAY_SCOPE}',"
        f" 'exp': {data['exp']}"
        "},"
        " from_cache=False,"
        " request_s=0.23)"
    )


def test_introspection_response_fxa_id() -> None:
    data = _create_fxa_introspect_response(uid="the-fxa-id")
    response = IntrospectionResponse("token", data)
    assert response.fxa_id == "the-fxa-id"


def test_introspection_response_equality() -> None:
    data = _create_fxa_introspect_response()
    response = IntrospectionResponse("token", data)
    assert response == IntrospectionResponse("token", data)
    assert response != IntrospectionError("token", "NotJson", data=data)
    assert response != IntrospectionResponse("token", data, from_cache=True)
    other_sub = data.copy()
    other_sub["sub"] = "other-fxa-id"
    assert response != IntrospectionResponse("token", other_sub)


@pytest.mark.parametrize("from_cache", (True, False))
@pytest.mark.parametrize("request_s", (1.0, None))
def test_introspection_response_as_cache_value(
    from_cache: bool, request_s: None | float
) -> None:
    data = _create_fxa_introspect_response()
    response = IntrospectionResponse(
        "token", data, from_cache=from_cache, request_s=request_s
    )
    # from_cache and request_s is not in cached value
    assert response.as_cache_value() == {"data": data}


@pytest.mark.parametrize("from_cache", (True, False))
def test_introspection_response_save_to_cache(from_cache: bool) -> None:
    data = _create_fxa_introspect_response()
    response = IntrospectionResponse("token", data, from_cache=from_cache)
    mock_cache = Mock(spec_set=["set"])
    response.save_to_cache(mock_cache, "the-key", 60)
    mock_cache.set.assert_called_once_with(get_cache_key("the-key"), {"data": data}, 60)


_INTROSPECTION_ERROR_REPR_TEST_CASES: dict[str, tuple[IntrospectionError, str]] = {
    "Timeout": (
        IntrospectionError("token", "Timeout"),
        "IntrospectionError(token='token', error='Timeout', error_args=[],"
        " status_code=None, data=None, from_cache=False, request_s=None)",
    ),
    "FailedRequest": (
        IntrospectionError(
            "token2",
            "FailedRequest",
            error_args=["requests.ConnectionError", "Accounts Rebooting"],
        ),
        "IntrospectionError(token='token2', error='FailedRequest',"
        " error_args=['requests.ConnectionError', 'Accounts Rebooting'],"
        " status_code=None, data=None, from_cache=False, request_s=None)",
    ),
    "NotJson": (
        IntrospectionError("token3", "NotJson", error_args=[""], status_code=200),
        "IntrospectionError(token='token3', error='NotJson', error_args=[''],"
        " status_code=200, data=None, from_cache=False, request_s=None)",
    ),
    "NotAuthorized": (
        IntrospectionError("token4", "NotAuthorized", status_code=401),
        "IntrospectionError(token='token4', error='NotAuthorized', error_args=[],"
        " status_code=401, data=None, from_cache=False, request_s=None)",
    ),
    "NotActive": (
        IntrospectionError(
            "token5",
            "NotActive",
            status_code=200,
            data={"active": False},
            from_cache=True,
        ),
        "IntrospectionError(token='token5', error='NotActive', error_args=[],"
        " status_code=200, data={'active': False}, from_cache=True, request_s=None)",
    ),
    "NoSubject": (
        IntrospectionError(
            "token6", "NoSubject", status_code=200, data={"active": True}, request_s=0.3
        ),
        "IntrospectionError(token='token6', error='NoSubject', error_args=[],"
        " status_code=200, data={'active': True}, from_cache=False, request_s=0.3)",
    ),
}


@pytest.mark.parametrize(
    "err,expected",
    _INTROSPECTION_ERROR_REPR_TEST_CASES.values(),
    ids=_INTROSPECTION_ERROR_REPR_TEST_CASES.keys(),
)
def test_introspection_error_repr(err: IntrospectionError, expected: str) -> None:
    assert repr(err) == expected


@pytest.mark.parametrize("from_cache", (True, False))
def test_introspection_error_as_cache_value_no_optional_params(
    from_cache: bool,
) -> None:
    error = IntrospectionError("token", "Timeout", from_cache=from_cache)
    # from_cache is not in cached value
    assert error.as_cache_value() == {"error": "Timeout"}


def test_introspection_error_save_to_cache_no_optional_params() -> None:
    error = IntrospectionError("token", "Timeout")
    mock_cache = Mock(spec_set=["set"])
    error.save_to_cache(mock_cache, "cache-key", 60)
    mock_cache.set.assert_called_once_with(
        get_cache_key("cache-key"), error.as_cache_value(), 60
    )


def test_introspection_error_as_cache_value_all_optional_params() -> None:
    error = IntrospectionError(
        "token",
        "NotOK",
        error_args=["something"],
        status_code=401,
        data={"error": "crazy stuff"},
        from_cache=True,
    )
    assert error.as_cache_value() == {
        "error": "NotOK",
        "status_code": 401,
        "data": {"error": "crazy stuff"},
        "error_args": ["something"],
    }


def test_introspection_error_save_to_cache_all_optional_params() -> None:
    error = IntrospectionError(
        "token",
        "NotOK",
        error_args=["something"],
        status_code=401,
        data={"error": "crazy stuff"},
        from_cache=True,
    )
    mock_cache = Mock(spec_set=["set"])
    error.save_to_cache(mock_cache, "cache-key", 60)
    mock_cache.set.assert_called_once_with(
        get_cache_key("cache-key"), error.as_cache_value(), 60
    )


def test_introspection_error_eq() -> None:
    err = IntrospectionError("token", "NotActive")
    assert err == IntrospectionError("token", "NotActive")
    assert err != IntrospectionError("other_token", "NotActive")
    assert err != IntrospectionError("token", "NotActive", status_code=200)
    assert err != IntrospectionError("token", "NotActive", data={})
    assert err != IntrospectionError("token", "NotActive", error_args=["an arg"])
    assert err != IntrospectionError("token", "NotActive", from_cache=True)
    assert err != IntrospectAuthenticationFailed(err)


def test_introspection_error_raises_exception_401() -> None:
    error = IntrospectionError(
        "token", "NotActive", status_code=401, data={"active": False}
    )
    with pytest.raises(IntrospectAuthenticationFailed) as exc_info:
        error.raise_exception("METHOD", "path")
    exception = exc_info.value
    assert exception.status_code == 401
    assert str(exception.detail) == "Incorrect authentication credentials."
    assert exception.args == (error,)


def test_introspection_error_raises_exception_503() -> None:
    error = IntrospectionError("token", "Timeout")
    with pytest.raises(IntrospectUnavailable) as exc_info:
        error.raise_exception("METHOD", "path")
    exception = exc_info.value
    assert exception.status_code == 503
    expected_detail = "Introspection temporarily unavailable, try again later."
    assert str(exception.detail) == expected_detail
    assert exception.args == (error,)


@responses.activate
def test_introspect_token_success_returns_introspection_response() -> None:
    mock_response, fxa_data = setup_fxa_introspect()
    assert fxa_data is not None

    fxa_resp = introspect_token("the-token")
    assert fxa_resp == IntrospectionResponse("the-token", fxa_data, request_s=0.5)
    assert mock_response.call_count == 1


@responses.activate
def test_introspect_token_no_expiration_uses_grace_period() -> None:
    mock_response, fxa_data = setup_fxa_introspect(expiration=False)
    assert fxa_data is not None

    fxa_resp = introspect_token("the-token")
    assert isinstance(fxa_resp, IntrospectionResponse)
    assert fxa_resp.token == "the-token"
    assert "exp" in fxa_resp.data
    assert mock_response.call_count == 1


# Test cases for introspect_token() that return an IntrospectionError
# Tuple is:
# - arguments to setup_fxa_introspect
# - the IntrospectionError error
# - other keyword parameters to IntrospectionError.
#   The special keyword parameter {"data": _SETUP_FXA_DATA} means to use
#   the mocked FxA Introspect body.
_SETUP_FXA_DATA = object()
_SETUP_FXA_DATA_B64 = object()
_INTROSPECT_TOKEN_FAILURE_TEST_CASES: list[
    tuple[dict[str, Any], INTROSPECT_ERROR, dict[str, Any]]
] = [
    ({"timeout": True}, "Timeout", {}),
    (
        {"exception": Exception("An Exception")},
        "FailedRequest",
        {"error_args": ["Exception", "An Exception"]},
    ),
    ({"no_body": True}, "NotJson", {"status_code": 200, "error_args": [as_b64("")]}),
    (
        {"text_body": '[{"active": false}]'},
        "NotJsonDict",
        {
            "status_code": 200,
            "error_args": [as_b64('[{"active": false}]')],
        },
    ),
    (
        {"status_code": 401, "active": False},
        "NotAuthorized",
        {"status_code": 401, "error_args": [_SETUP_FXA_DATA_B64]},
    ),
    # Attempting to continue on non-200 error
    # ({"status_code": 500}, "NotOK", {"status_code": 500, "data": _SETUP_FXA_DATA}),
    ({"active": False}, "NotActive", {"status_code": 200, "data": _SETUP_FXA_DATA}),
    ({"uid": None}, "NoSubject", {"status_code": 200, "data": _SETUP_FXA_DATA}),
    ({"scope": "foo"}, "MissingScope", {"status_code": 200, "data": _SETUP_FXA_DATA}),
]


@pytest.mark.parametrize(
    "setup_args,error,error_params",
    _INTROSPECT_TOKEN_FAILURE_TEST_CASES,
    ids=[case[1] for case in _INTROSPECT_TOKEN_FAILURE_TEST_CASES],
)
@responses.activate
def test_introspect_token_error_returns_introspection_error(
    setup_args: dict[str, Any], error: INTROSPECT_ERROR, error_params: dict[str, Any]
) -> None:
    mock_response, fxa_data = setup_fxa_introspect(**setup_args)
    params = error_params.copy()
    if error_params.get("data") is _SETUP_FXA_DATA:
        assert fxa_data is not None
        params["data"] = fxa_data
    if error_params.get("error_args", [None])[0] is _SETUP_FXA_DATA_B64:
        assert fxa_data is not None
        params["error_args"] = [as_b64(fxa_data)]
    expected_resp = IntrospectionError("err-token", error, request_s=0.5, **params)

    fxa_resp = introspect_token("err-token")
    assert fxa_resp == expected_resp
    assert mock_response.call_count == 1


def test_load_introspection_result_from_cache_expired_token() -> None:
    fxa_data = _create_fxa_introspect_response()
    expired = datetime.now() - timedelta(days=1)
    fxa_data["exp"] = int(expired.timestamp()) * 1000
    cache = Mock(spec_set=["get"])
    cache.get.return_value = {"data": fxa_data}
    token = "cached_token"

    response = load_introspection_result_from_cache(cache, token)
    assert response == IntrospectionError(
        token,
        "TokenExpired",
        error_args=[response.error_args[0]],
        data=fxa_data,
        from_cache=True,
    )
    # Error arg is how many seconds it is expired
    one_day_ago = -1 * 24 * 60 * 60
    assert one_day_ago + 5 > int(response.error_args[0]) >= one_day_ago
    cache.get.assert_called_once_with(get_cache_key(token))


def test_load_introspection_result_from_cache_introspection_response() -> None:
    fxa_data = _create_fxa_introspect_response()
    cache = Mock(spec_set=["get"])
    cache.get.return_value = {"data": fxa_data}
    token = "cached_token"

    response = load_introspection_result_from_cache(cache, token)
    assert isinstance(response, IntrospectionResponse)
    assert response == IntrospectionResponse(token, fxa_data, from_cache=True)
    cache.get.assert_called_once_with(get_cache_key(token))


def test_load_introspection_result_from_cache_introspection_error_no_args() -> None:
    cache = Mock(spec_set=["get"])
    cache.get.return_value = {"error": "Timeout"}
    token = "cached_token"

    error = load_introspection_result_from_cache(cache, token)
    assert isinstance(error, IntrospectionError)
    assert error == IntrospectionError(token, "Timeout", from_cache=True)
    cache.get.assert_called_once_with(get_cache_key(token))


def test_load_introspection_result_from_cache_introspection_error_all_args() -> None:
    cache = Mock(spec_set=["get"])
    cache.get.return_value = {
        "error": "NotOK",
        "status_code": 401,
        "data": {"error": "crazy stuff"},
        "error_args": ["something"],
    }
    token = "cached_error_token"

    error = load_introspection_result_from_cache(cache, token)
    assert isinstance(error, IntrospectionError)
    assert error == IntrospectionError(
        token,
        "NotOK",
        error_args=["something"],
        status_code=401,
        data={"error": "crazy stuff"},
        from_cache=True,
    )
    cache.get.assert_called_once_with(get_cache_key(token))


def test_load_introspection_result_from_cache_introspection_bad_value() -> None:
    cache = Mock(spec_set=["get"])
    cache.get.return_value = "Not a dictionary"
    token = "uncached_token"

    assert load_introspection_result_from_cache(cache, token) is None
    cache.get.assert_called_once_with(get_cache_key(token))


@responses.activate
def test_introspect_and_cache_token_mocked_success_is_cached(cache: BaseCache) -> None:
    user_token = "user-123"
    cache_key = get_cache_key(user_token)
    fxa_id = "fxa-id-for-user-123"
    mock_response, fxa_data = setup_fxa_introspect(uid=fxa_id)
    assert fxa_data is not None
    assert cache.get(cache_key) is None

    # get FxA ID for the first time
    fxa_resp = introspect_and_cache_token(user_token)
    assert isinstance(fxa_resp, IntrospectionResponse)
    assert fxa_resp.fxa_id == fxa_id
    assert not fxa_resp.from_cache
    assert mock_response.call_count == 1
    assert cache.get(cache_key) == fxa_resp.as_cache_value()

    # now check that the 2nd call did NOT make another fxa request
    fxa_resp2 = introspect_and_cache_token(user_token)
    assert isinstance(fxa_resp, IntrospectionResponse)
    assert fxa_resp2.from_cache
    assert mock_response.call_count == 1


@responses.activate
def test_introspect_and_cache_token_mocked_success_with_use_cache_false(
    cache: BaseCache,
) -> None:
    user_token = "user-123"
    cache_key = get_cache_key(user_token)
    fxa_id = "fxa-id-for-user-123"
    mock_response, fxa_data = setup_fxa_introspect(uid=fxa_id)
    assert fxa_data is not None
    cache.set(cache_key, "An invalid cache value that is not read")

    # skip cache, call introspect API, set cache to new data
    fxa_resp = introspect_and_cache_token(user_token, read_from_cache=False)
    assert isinstance(fxa_resp, IntrospectionResponse)
    assert fxa_resp.fxa_id == fxa_id
    assert not fxa_resp.from_cache
    assert mock_response.call_count == 1
    assert cache.get(cache_key) == fxa_resp.as_cache_value()


@responses.activate
def test_introspect_and_cache_token_mocked_error_is_cached(cache: BaseCache) -> None:
    user_token = "user-123"
    cache_key = get_cache_key(user_token)
    mock_response, fxa_data = setup_fxa_introspect(timeout=True)
    assert fxa_data is None
    assert cache.get(cache_key) is None

    # Timeout for the first time
    expected_error = IntrospectionError(user_token, "Timeout", request_s=0.5)
    result = introspect_and_cache_token(user_token)
    assert result == expected_error
    assert mock_response.call_count == 1
    assert cache.get(cache_key) == expected_error.as_cache_value()

    # now check that the 2nd call did NOT make another fxa request
    result = introspect_and_cache_token(user_token)
    assert result == IntrospectionError(user_token, "Timeout", from_cache=True)
    assert mock_response.call_count == 1


def test_fxa_token_authentication_no_auth_header_skips() -> None:
    req = APIRequestFactory().get("/api/endpoint")
    assert FxaTokenAuthentication().authenticate(req) is None


def test_fxa_token_authentication_not_bearer_token_auth_header_skips() -> None:
    headers = {"Authorization": "unexpected 123"}
    req = APIRequestFactory().get("/api/endpoint", headers=headers)
    assert FxaTokenAuthentication().authenticate(req) is None


def test_fxa_token_authentication_incomplete_bearer_token_raises_auth_fail() -> None:
    headers = {"Authorization": "Bearer "}
    req = APIRequestFactory().get("/api/endpoint", headers=headers)
    with pytest.raises(
        AuthenticationFailed, match=r"Invalid token header\. No credentials provided\."
    ):
        FxaTokenAuthentication().authenticate(req)


@responses.activate
def test_fxa_token_authentication_known_relay_user_returns_user(
    free_user: User,
    fxa_social_app: SocialApp,
    cache: BaseCache,
) -> None:
    fxa_id = "some-fxa-id"
    token = "bearer-token"
    SocialAccount.objects.create(provider="fxa", uid=fxa_id, user=free_user)
    mock_response, fxa_data = setup_fxa_introspect(uid=fxa_id)
    assert fxa_data is not None
    headers = {"Authorization": f"Bearer {token}"}
    req = APIRequestFactory().get("/api/endpoint", headers=headers)
    expected_resp = IntrospectionResponse(token, fxa_data, request_s=0.5)

    assert FxaTokenAuthentication().authenticate(req) == (free_user, expected_resp)
    assert cache.get(get_cache_key(token)) == expected_resp.as_cache_value()


@responses.activate
def test_fxa_token_authentication_unknown_token_raises_auth_fail(
    cache: BaseCache,
) -> None:
    mock_response, fxa_data = setup_fxa_introspect(status_code=401, active=False)
    assert fxa_data is not None
    token = "bearer-token"
    headers = {"Authorization": f"Bearer {token}"}
    req = APIRequestFactory().get("/api/endpoint", headers=headers)
    expected_error = IntrospectionError(
        token,
        "NotAuthorized",
        status_code=401,
        error_args=[as_b64(fxa_data)],
        request_s=0.5,
    )

    with pytest.raises(
        AuthenticationFailed, match=r"Incorrect authentication credentials\."
    ) as exc_info:
        FxaTokenAuthentication().authenticate(req)
    assert exc_info.value.args[0] == expected_error
    assert cache.get(get_cache_key(token)) == expected_error.as_cache_value()


@responses.activate
@pytest.mark.django_db
def test_fxa_token_authentication_not_yet_relay_user_returns_anon_user(
    cache: BaseCache,
) -> None:
    mock_response, fxa_data = setup_fxa_introspect()
    assert fxa_data is not None
    token = "bearer-token"
    headers = {"Authorization": f"Bearer {token}"}
    req = APIRequestFactory().get("/api/endpoint", headers=headers)
    introspect_response = IntrospectionResponse(token, fxa_data, request_s=0.5)
    user_and_auth = FxaTokenAuthentication().authenticate(req)
    assert user_and_auth is not None
    user, auth = user_and_auth
    assert user == AnonymousUser()
    assert not user.is_authenticated
    assert auth == introspect_response


@responses.activate
def test_fxa_token_authentication_inactive_relay_user(
    free_user: User,
    fxa_social_app: SocialApp,
    cache: BaseCache,
) -> None:
    """TODO: Should this be an IsActive or other permission check?"""
    free_user.is_active = False
    free_user.save()
    fxa_id = "some-fxa-id"
    token = "bearer-token"
    SocialAccount.objects.create(provider="fxa", uid=fxa_id, user=free_user)
    mock_response, fxa_data = setup_fxa_introspect(uid=fxa_id)
    assert fxa_data is not None
    headers = {"Authorization": f"Bearer {token}"}
    req = APIRequestFactory().get("/api/endpoint", headers=headers)
    expected_resp = IntrospectionResponse(token, fxa_data, request_s=0.5)
    user_and_auth = FxaTokenAuthentication().authenticate(req)
    assert user_and_auth is not None
    user, auth = user_and_auth
    assert user == free_user
    assert user.is_active is False
    assert auth == expected_resp
    assert cache.get(get_cache_key(token)) == expected_resp.as_cache_value()


@pytest.mark.parametrize(
    "method,path",
    [
        ("POST", "/api/some_endpoint"),
        ("DELETE", "/api/some_endpoint"),
        ("PUT", "/api/some_endpoint"),
        ("DELETE", "/api/v1/relayaddresses/1234"),
        ("PUT", "/api/v1/relayaddresses/1234"),
    ],
)
def test_fxa_token_authentication_skip_cache(
    method: str, path: str, free_user: User, fxa_social_app: SocialApp
) -> None:
    """FxA introspect is always called (use_cache=False) for some methods."""
    fxa_id = "non-cached-id"
    SocialAccount.objects.create(provider="fxa", uid=fxa_id, user=free_user)
    token = "bearer-token"
    headers = {"Authorization": f"Bearer {token}"}
    req = getattr(APIRequestFactory(), method.lower())(path=path, headers=headers)
    auth = FxaTokenAuthentication()
    introspect_response = IntrospectionResponse(
        token, _create_fxa_introspect_response(uid=fxa_id)
    )

    with patch(
        "api.authentication.introspect_and_cache_token",
        return_value=introspect_response,
    ) as introspect:
        user_and_token = auth.authenticate(req)
    introspect.assert_called_once_with(token, False)
    assert user_and_token == (free_user, introspect_response)


@pytest.mark.parametrize(
    "method,path",
    [
        ("GET", "/api/some_endpoint"),
        ("HEAD", "/api/some_endpoint"),
        ("OPTIONS", "/api/some_endpoint"),
        ("POST", "/api/v1/relayaddresses/"),
    ],
)
def test_fxa_token_authentication_read_from_cache(
    method: str, path: str, free_user: User, fxa_social_app: SocialApp
) -> None:
    """Cached FxA introspect results are used (use_cache=True) for some methods."""
    fxa_id = "non-cached-id"
    SocialAccount.objects.create(provider="fxa", uid=fxa_id, user=free_user)
    token = "bearer-token"
    headers = {"Authorization": f"Bearer {token}"}
    req = getattr(APIRequestFactory(), method.lower())(path=path, headers=headers)
    auth = FxaTokenAuthentication()
    introspect_response = IntrospectionResponse(
        token, _create_fxa_introspect_response(uid=fxa_id), from_cache=True
    )

    with patch(
        "api.authentication.introspect_and_cache_token",
        return_value=introspect_response,
    ) as introspect:
        user_and_token = auth.authenticate(req)
    introspect.assert_called_once_with(token, True)
    assert user_and_token == (free_user, introspect_response)

mozilla / fx-private-relay / 84353759-c057-4f20-b282-724c34504dc9

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous