0d7efbc0-b94b-416d-be6b-612c09993f6b

Committed 14 Feb 2025 08:09PM CUT coverage: 85.092% (-0.03%) from 85.117%

Build # 0d7efbc0-b94b-416d-be6b-612c09993f6b

Build Type

Pull #5376

circleci

Committed by

jwhitlock

Commit Message

Markdown lint

Pull Request Pull Request #5376: MPP-4054: ADR 0005: Document the metrics backend changes

Run Details

2434 of 3561 branches covered (68.35%)

Branch coverage included in aggregate %.

17001 of 19279 relevant lines covered (88.18%)

9.92 hits per line

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

96.58

/privaterelay/utils.py

from __future__ import annotations

import json
import logging
import random
from collections.abc import Callable
from decimal import Decimal
from functools import cache, wraps
from pathlib import Path
from string import ascii_uppercase
from typing import TYPE_CHECKING, ParamSpec, TypedDict, TypeVar, cast

from django.conf import settings
from django.contrib.auth.models import AbstractBaseUser
from django.http import Http404, HttpRequest
from django.utils.translation.trans_real import parse_accept_lang_header

from waffle import get_waffle_flag_model
from waffle.models import logger as waffle_logger
from waffle.utils import get_cache as get_waffle_cache
from waffle.utils import get_setting as get_waffle_setting

from .plans import (
    CountryStr,
    LanguageStr,
    PeriodStr,
    PlanCountryLangMapping,
    get_premium_country_language_mapping,
)

if TYPE_CHECKING:
    from .glean_interface import RelayGleanLogger

info_logger = logging.getLogger("eventsinfo")


class CountryInfo(TypedDict):
    country_code: str
    countries: list[CountryStr]
    available_in_country: bool
    plan_country_lang_mapping: PlanCountryLangMapping


def get_countries_info_from_request_and_mapping(
    request: HttpRequest, mapping: PlanCountryLangMapping
) -> CountryInfo:
    country_code = _get_cc_from_request(request)
    countries = sorted(mapping.keys())
    available_in_country = country_code in countries
    return {
        "country_code": country_code,
        "countries": countries,
        "available_in_country": available_in_country,
        "plan_country_lang_mapping": mapping,
    }


def get_countries_info_from_lang_and_mapping(
    accept_lang: str, mapping: PlanCountryLangMapping
) -> CountryInfo:
    country_code = _get_cc_from_lang(accept_lang)
    countries = sorted(mapping.keys())
    available_in_country = country_code in countries
    return {
        "country_code": country_code,
        "countries": countries,
        "available_in_country": available_in_country,
        "plan_country_lang_mapping": mapping,
    }


def get_subplat_upgrade_link_by_language(
    accept_language: str, period: PeriodStr = "yearly"
) -> str:
    country_str = guess_country_from_accept_lang(accept_language)
    country = cast(CountryStr, country_str)
    language_str = accept_language.split("-")[0].lower()
    language = cast(LanguageStr, language_str)
    country_lang_mapping = get_premium_country_language_mapping()
    country_details = country_lang_mapping.get(country, country_lang_mapping["US"])
    if language in country_details:
        plan = country_details[language][period]
    else:
        first_key = list(country_details.keys())[0]
        plan = country_details[first_key][period]
    return (
        f"{settings.FXA_BASE_ORIGIN}/subscriptions/products/"
        f"{settings.PERIODICAL_PREMIUM_PROD_ID}?plan={plan['id']}"
    )


def _get_cc_from_request(request: HttpRequest) -> str:
    """Determine the user's region / country code."""

    log_data: dict[str, str] = {}
    cdn_region = None
    region = None
    if "X-Client-Region" in request.headers:
        cdn_region = region = request.headers["X-Client-Region"].upper()
        log_data["cdn_region"] = cdn_region
        log_data["region_method"] = "cdn"

    accept_language_region = None
    if "Accept-Language" in request.headers:
        log_data["accept_lang"] = request.headers["Accept-Language"]
        accept_language_region = _get_cc_from_lang(request.headers["Accept-Language"])
        log_data["accept_lang_region"] = accept_language_region
        if region is None:
            region = accept_language_region
            log_data["region_method"] = "accept_lang"

    if region is None:
        region = "US"
        log_data["region_method"] = "fallback"
    log_data["region"] = region

    # MPP-3284: Log details of region selection. Only log once per request, since some
    # endpoints, like /api/v1/runtime_data, call this multiple times.
    if not getattr(request, "_logged_region_details", False):
        setattr(request, "_logged_region_details", True)
        info_logger.info("region_details", extra=log_data)

    return region


def _get_cc_from_lang(accept_lang: str) -> str:
    try:
        return guess_country_from_accept_lang(accept_lang)
    except AcceptLanguageError:
        return ""


# Map a primary language to the most probable country
# Top country derived from CLDR42 Supplemental Data, Language-Territory Information
# with the exception of Spanish (es), which is mapped to Spain (es) instead of
# Mexico (mx), which has the most speakers, but usually specifies es-MX.
_PRIMARY_LANGUAGE_TO_COUNTRY = {
    "ace": "ID",  # # Acehnese -> Indonesia
    "ach": "UG",  # Acholi -> Uganda
    "af": "ZA",  # Afrikaans -> South Africa
    "an": "ES",  # Aragonese -> Spain
    "ar": "EG",  # Arabic -> Egypt
    "arn": "CL",  # Mapudungun -> Chile
    "as": "IN",  # Assamese -> India
    "ast": "ES",  # Asturian -> Spain
    "az": "AZ",  # Azerbaijani -> Azerbaijan
    "be": "BY",  # Belerusian -> Belarus
    "bg": "BG",  # Bulgarian -> Bulgaria
    "bn": "BD",  # Bengali -> Bangladesh
    "bo": "CN",  # Tibetan -> China
    "br": "FR",  # Breton -> France
    "brx": "IN",  # Bodo -> India
    "bs": "BA",  # Bosnian -> Bosnia and Herzegovina
    "ca": "FR",  # Catalan -> France
    "cak": "MX",  # Kaqchikel -> Mexico
    "ckb": "IQ",  # Central Kurdish -> Iraq
    "cs": "CZ",  # Czech -> Czech Republic
    "cv": "RU",  # Chuvash -> Russia
    "cy": "GB",  # Welsh -> United Kingdom
    "da": "DK",  # Danish -> Denmark
    "de": "DE",  # German -> Germany
    "dsb": "DE",  # Lower Sorbian -> Germany
    "el": "GR",  # Greek -> Greece
    "en": "US",  # English -> Canada
    "eo": "SM",  # Esperanto -> San Marino
    "es": "ES",  # Spanish -> Spain (instead of Mexico, top by population)
    "et": "EE",  # Estonian -> Estonia
    "eu": "ES",  # Basque -> Spain
    "fa": "IR",  # Persian -> Iran
    "ff": "SN",  # Fulah -> Senegal
    "fi": "FI",  # Finnish -> Finland
    "fr": "FR",  # French -> France
    "frp": "FR",  # Arpitan -> France
    "fur": "IT",  # Friulian -> Italy
    "fy": "NL",  # Frisian -> Netherlands
    "ga": "IE",  # Irish -> Ireland
    "gd": "GB",  # Scottish Gaelic -> United Kingdom
    "gl": "ES",  # Galician -> Spain
    "gn": "PY",  # Guarani -> Paraguay
    "gu": "IN",  # Gujarati -> India
    "gv": "IM",  # Manx -> Isle of Man
    "he": "IL",  # Hebrew -> Israel
    "hi": "IN",  # Hindi -> India
    "hr": "HR",  # Croatian -> Croatia
    "hsb": "DE",  # Upper Sorbian -> Germany
    "hu": "HU",  # Hungarian -> Hungary
    "hy": "AM",  # Armenian -> Armenia
    "hye": "AM",  # Armenian Eastern Classic Orthography -> Armenia
    "ia": "FR",  # Interlingua -> France
    "id": "ID",  # Indonesian -> Indonesia
    "ilo": "PH",  # Iloko -> Philippines
    "is": "IS",  # Icelandic -> Iceland
    "it": "IT",  # Italian -> Italy
    "ixl": "MX",  # Ixil -> Mexico
    "ja": "JP",  # Japanese -> Japan
    "jiv": "MX",  # Shuar -> Mexico
    "ka": "GE",  # Georgian -> Georgia
    "kab": "DZ",  # Kayble -> Algeria
    "kk": "KZ",  # Kazakh -> Kazakhstan
    "km": "KH",  # Khmer -> Cambodia
    "kn": "IN",  # Kannada -> India
    "ko": "KR",  # Korean -> South Korea
    "ks": "IN",  # Kashmiri -> India
    "lb": "LU",  # Luxembourgish -> Luxembourg
    "lg": "UG",  # Luganda -> Uganda
    "lij": "IT",  # Ligurian -> Italy
    "lo": "LA",  # Lao -> Laos
    "lt": "LT",  # Lithuanian -> Lithuania
    "ltg": "LV",  # Latgalian -> Latvia
    "lus": "US",  # Mizo -> United States
    "lv": "LV",  # Latvian -> Latvia
    "mai": "IN",  # Maithili -> India
    "meh": "MX",  # Mixteco Yucuhiti -> Mexico
    "mix": "MX",  # Mixtepec Mixtec -> Mexico
    "mk": "MK",  # Macedonian -> North Macedonia
    "ml": "IN",  # Malayalam -> India
    "mr": "IN",  # Marathi -> India
    "ms": "MY",  # Malay -> Malaysia
    "my": "MM",  # Burmese -> Myanmar
    "nb": "NO",  # Norwegian Bokmål -> Norway
    "ne": "NP",  # Nepali -> Nepal
    "nl": "NL",  # Dutch -> Netherlands
    "nn": "NO",  # Norwegian Nynorsk -> Norway
    "oc": "FR",  # Occitan -> France
    "or": "IN",  # Odia -> India
    "pa": "IN",  # Punjabi -> India
    "pl": "PL",  # Polish -> Poland
    "ppl": "MX",  # Náhuat Pipil -> Mexico
    "pt": "BR",  # Portuguese -> Brazil
    "quc": "GT",  # K'iche' -> Guatemala
    "rm": "CH",  # Romansh -> Switzerland
    "ro": "RO",  # Romanian -> Romania
    "ru": "RU",  # Russian -> Russia
    "sat": "IN",  # Santali (Ol Chiki) -> India
    "sc": "IT",  # Sardinian -> Italy
    "scn": "IT",  # Sicilian -> Italy
    "sco": "GB",  # Scots -> United Kingdom
    "si": "LK",  # Sinhala -> Sri Lanka
    "sk": "SK",  # Slovak -> Slovakia
    "skr": "PK",  # Saraiki -> Pakistan
    "sl": "SI",  # Slovenian -> Slovenia
    "son": "ML",  # Songhay -> Mali
    "sq": "AL",  # Albanian -> Albania
    "sr": "RS",  # Serbian -> Serbia
    "sv": "SE",  # Swedish -> Sweeden
    "sw": "TZ",  # Swahili -> Tanzania
    "szl": "PL",  # Silesian -> Poland
    "ta": "IN",  # Tamil -> India
    "te": "IN",  # Telugu -> India
    "tg": "TJ",  # Tajik -> Tajikistan
    "th": "TH",  # Thai -> Thailand
    "tl": "PH",  # Tagalog -> Philippines
    "tr": "TR",  # Turkish or Crimean Tatar -> Turkey
    "trs": "MX",  # Triqui -> Mexico
    "uk": "UA",  # Ukrainian -> Ukraine
    "ur": "PK",  # Urdu -> Pakistan
    "uz": "UZ",  # Uzbek -> Uzbekistan
    "vi": "VN",  # Vietnamese -> Vietnam
    "wo": "SN",  # Wolof -> Senegal
    "xcl": "AM",  # Armenian Classic -> Armenia
    "xh": "ZA",  # Xhosa -> South Africa
    "zam": "MX",  # Miahuatlán Zapotec -> Mexico
    "zh": "CN",  # Chinese -> China
}

# Special cases for language tags
_LANGUAGE_TAG_TO_COUNTRY_OVERRIDE = {
    # Would be Catalan in Valencian script -> France
    # Change to Valencian -> Spain
    ("ca", "VALENCIA"): "ES",
    # Spanish in UN region 419 (Latin America and Carribean)
    # Pick Mexico, which has highest number of Spanish speakers
    ("es", "419"): "MX",
    # Would be Galician (Greenland) -> Greenland
    # Change to Galician (Galicia region of Spain) -> Spain
    ("gl", "GL"): "ES",
}


class AcceptLanguageError(ValueError):
    """There was an issue processing the Accept-Language header."""

    def __init__(self, message, accept_lang):
        super().__init__(message)
        self.accept_lang = accept_lang


def guess_country_from_accept_lang(accept_lang: str) -> str:
    """
    Guess the user's country from the Accept-Language header

    Return is a 2-letter ISO 3166 country code

    If an issue is detected, a AcceptLanguageError is raised.

    The header may come directly from a web request, or may be the header
    captured by Mozilla Accounts (FxA) at signup.

    Even with all this logic and special casing, it is still more accurate to
    use a GeoIP lookup or a country code provided by the infrastructure.

    See RFC 9110, "HTTP Semantics", section 12.5.4, "Accept-Language"
    See RFC 5646, "Tags for Identifying Languages", and examples in Appendix A
    """
    lang_q_pairs = parse_accept_lang_header(accept_lang.strip())
    if not lang_q_pairs:
        raise AcceptLanguageError("Invalid Accept-Language string", accept_lang)
    top_lang_tag = lang_q_pairs[0][0]

    subtags = top_lang_tag.split("-")
    lang = subtags[0].lower()
    if lang == "i":
        raise AcceptLanguageError("Irregular language tag", accept_lang)
    if lang == "x":
        raise AcceptLanguageError("Private-use language tag", accept_lang)
    if lang == "*":
        raise AcceptLanguageError("Wildcard language tag", accept_lang)
    if len(lang) < 2:
        raise AcceptLanguageError("Invalid one-character primary language", accept_lang)
    if len(lang) == 3 and lang[0] == "q" and lang[1] <= "t":
        raise AcceptLanguageError(
            "Private-use language tag (RFC 5646 2.2.1)", accept_lang
        )

    for maybe_region_raw in subtags[1:]:
        maybe_region = maybe_region_raw.upper()

        # Look for a special case
        if override := _LANGUAGE_TAG_TO_COUNTRY_OVERRIDE.get((lang, maybe_region)):
            return override

        if len(maybe_region) <= 1:
            # One-character extension or empty, stop processing
            break
        if (
            len(maybe_region) == 2
            and all(c in ascii_uppercase for c in maybe_region)
            and
            # RFC 5646 2.2.4 "Region Subtag" point 6, reserved subtags
            maybe_region != "AA"
            and maybe_region != "ZZ"
            and maybe_region[0] != "X"
            and (maybe_region[0] != "Q" or maybe_region[1] < "M")
        ):
            # Subtag is a non-private ISO 3166 country code
            return maybe_region

        # Subtag is probably a script, like "Hans" in "zh-Hans-CN"
        # Loop to the next subtag, which might be a ISO 3166 country code

    # Guess the country from a simple language tag
    try:
        return _PRIMARY_LANGUAGE_TO_COUNTRY[lang]
    except KeyError:
        raise AcceptLanguageError("Unknown langauge", accept_lang)


# Generics for defining function decorators
# https://mypy.readthedocs.io/en/stable/generics.html#declaring-decorators
_Params = ParamSpec("_Params")
_RetVal = TypeVar("_RetVal")


def enable_or_404(
    check_function: Callable[[], bool],
    message: str = "This conditional view is disabled.",
) -> Callable[[Callable[_Params, _RetVal]], Callable[_Params, _RetVal]]:
    """
    Returns decorator that enables a view if a check function passes,
    otherwise returns a 404.

    Usage:

        def percent_1():
           import random
           return random.randint(1, 100) == 1

        @enable_if(percent_1)
        def lucky_view(request):
            #  1 in 100 chance of getting here
            # 99 in 100 chance of 404
    """

    def decorator(func: Callable[_Params, _RetVal]) -> Callable[_Params, _RetVal]:
        @wraps(func)
        def inner(*args: _Params.args, **kwargs: _Params.kwargs) -> _RetVal:
            if check_function():
                return func(*args, **kwargs)
            else:
                raise Http404(message)  # Display a message with DEBUG=True

        return inner

    return decorator


def enable_if_setting(
    setting_name: str,
    message_fmt: str = "This view is disabled because {setting_name} is False",
) -> Callable[[Callable[_Params, _RetVal]], Callable[_Params, _RetVal]]:
    """
    Returns decorator that enables a view if a setting is truthy, otherwise
    returns a 404.

    Usage:

        @enable_if_setting("DEBUG")
        def debug_only_view(request):
            # DEBUG == True

    Or in URLS:

        path(
            "developer_info",
            enable_if_setting("DEBUG")(debug_only_view)
        ),
        name="developer-info",
    ),

    """

    def setting_is_truthy() -> bool:
        return bool(getattr(settings, setting_name))

    return enable_or_404(
        setting_is_truthy, message_fmt.format(setting_name=setting_name)
    )


def flag_is_active_in_task(flag_name: str, user: AbstractBaseUser | None) -> bool:
    """
    Test if a flag is active in a task (not in a web request).

    This mirrors AbstractBaseFlag.is_active, replicating these checks:
    * Logs missing flags, if configured
    * Creates missing flags, if configured
    * Returns default for missing flags
    * Checks flag.everyone
    * Checks flag.users and flag.groups, if a user is passed
    * Returns random results for flag.percent

    It does not include:
    * Overriding a flag with a query parameter
    * Persisting a flag in a cookie (includes percent flags)
    * Language-specific overrides (could be added later)
    * Read-only mode for percent flags

    When using this function, use the @override_flag decorator in tests, rather
    than manually creating flags in the database.
    """
    flag = get_waffle_flag_model().get(flag_name)
    if not flag.pk:
        log_level = get_waffle_setting("LOG_MISSING_FLAGS")
        if log_level:
            waffle_logger.log(log_level, "Flag %s not found", flag_name)
        if get_waffle_setting("CREATE_MISSING_FLAGS"):
            flag, _created = get_waffle_flag_model().objects.get_or_create(
                name=flag_name,
                defaults={"everyone": get_waffle_setting("FLAG_DEFAULT")},
            )
            cache = get_waffle_cache()
            cache.set(flag._cache_key(flag.name), flag)

        return bool(get_waffle_setting("FLAG_DEFAULT"))

    # Removed - check for override as request query parameter

    if flag.everyone:
        return True
    elif flag.everyone is False:
        return False

    # Removed - check for testing override in request query or cookie
    # Removed - check for language-specific override

    if user is not None:
        active_for_user = flag.is_active_for_user(user)
        if active_for_user is not None:
            return bool(active_for_user)

    if flag.percent and flag.percent > 0:
        # Removed - check for waffles attribute of request
        # Removed - check for cookie setting for flag
        # Removed - check for read-only mode

        if Decimal(str(random.uniform(0, 100))) <= flag.percent:  # noqa: S311
            # Removed - setting the flag for future checks
            return True

    return False


class VersionInfo(TypedDict):
    source: str
    version: str
    commit: str
    build: str


@cache
def get_version_info(base_dir: str | Path | None = None) -> VersionInfo:
    """Return version information written by build process."""
    if base_dir is None:
        base_path = Path(settings.BASE_DIR)
    else:
        base_path = Path(base_dir)
    version_json_path = base_path / "version.json"
    info = {}
    if version_json_path.exists():
        with version_json_path.open() as version_file:
            try:
                info = json.load(version_file)
            except ValueError:
                pass
            if not hasattr(info, "get"):
                info = {}
    version_info = VersionInfo(
        source=info.get("source", "https://github.com/mozilla/fx-private-relay"),
        version=info.get("version", "unknown"),
        commit=info.get("commit", "unknown"),
        build=info.get("build", "not built"),
    )
    return version_info


@cache
def glean_logger() -> RelayGleanLogger:
    from .glean_interface import RelayGleanLogger

    version_info = get_version_info()
    return RelayGleanLogger(
        application_id="relay-backend",
        app_display_version=version_info["version"],
        channel=settings.RELAY_CHANNEL,
    )

1	from __future__ import annotations	1✔
2
3	import json	1✔
4	import logging	1✔
5	import random	1✔
6	from collections.abc import Callable	1✔
7	from decimal import Decimal	1✔
8	from functools import cache, wraps	1✔
9	from pathlib import Path	1✔
10	from string import ascii_uppercase	1✔
11	from typing import TYPE_CHECKING, ParamSpec, TypedDict, TypeVar, cast	1✔
12
13	from django.conf import settings	1✔
14	from django.contrib.auth.models import AbstractBaseUser	1✔
15	from django.http import Http404, HttpRequest	1✔
16	from django.utils.translation.trans_real import parse_accept_lang_header	1✔
17
18	from waffle import get_waffle_flag_model	1✔
19	from waffle.models import logger as waffle_logger	1✔
20	from waffle.utils import get_cache as get_waffle_cache	1✔
21	from waffle.utils import get_setting as get_waffle_setting	1✔
22
23	from .plans import (	1✔
24	CountryStr,
25	LanguageStr,
26	PeriodStr,
27	PlanCountryLangMapping,
28	get_premium_country_language_mapping,
29	)
30
31	if TYPE_CHECKING:
32	from .glean_interface import RelayGleanLogger
33
34	info_logger = logging.getLogger("eventsinfo")	1✔
35
36
37	class CountryInfo(TypedDict):	1✔
38	country_code: str	1✔
39	countries: list[CountryStr]	1✔
40	available_in_country: bool	1✔
41	plan_country_lang_mapping: PlanCountryLangMapping	1✔
42
43
44	def get_countries_info_from_request_and_mapping(	1✔
45	request: HttpRequest, mapping: PlanCountryLangMapping
46	) -> CountryInfo:
47	country_code = _get_cc_from_request(request)	1✔
48	countries = sorted(mapping.keys())	1✔
49	available_in_country = country_code in countries	1✔
50	return {	1✔
51	"country_code": country_code,
52	"countries": countries,
53	"available_in_country": available_in_country,
54	"plan_country_lang_mapping": mapping,
55	}
56
57
58	def get_countries_info_from_lang_and_mapping(	1✔
59	accept_lang: str, mapping: PlanCountryLangMapping
60	) -> CountryInfo:
61	country_code = _get_cc_from_lang(accept_lang)	1✔
62	countries = sorted(mapping.keys())	1✔
63	available_in_country = country_code in countries	1✔
64	return {	1✔
65	"country_code": country_code,
66	"countries": countries,
67	"available_in_country": available_in_country,
68	"plan_country_lang_mapping": mapping,
69	}
70
71
72	def get_subplat_upgrade_link_by_language(	1✔
73	accept_language: str, period: PeriodStr = "yearly"
74	) -> str:
75	country_str = guess_country_from_accept_lang(accept_language)	1✔
76	country = cast(CountryStr, country_str)	1✔
77	language_str = accept_language.split("-")[0].lower()	1✔
78	language = cast(LanguageStr, language_str)	1✔
79	country_lang_mapping = get_premium_country_language_mapping()	1✔
80	country_details = country_lang_mapping.get(country, country_lang_mapping["US"])	1✔
81	if language in country_details:	1!
82	plan = country_details[language][period]	×
83	else:
84	first_key = list(country_details.keys())[0]	1✔
85	plan = country_details[first_key][period]	1✔
86	return (	1✔
87	f"{settings.FXA_BASE_ORIGIN}/subscriptions/products/"
88	f"{settings.PERIODICAL_PREMIUM_PROD_ID}?plan={plan['id']}"
89	)
90
91
92	def _get_cc_from_request(request: HttpRequest) -> str:	1✔
93	"""Determine the user's region / country code."""
94
95	log_data: dict[str, str] = {}	1✔
96	cdn_region = None	1✔
97	region = None	1✔
98	if "X-Client-Region" in request.headers:	1✔
99	cdn_region = region = request.headers["X-Client-Region"].upper()	1✔
100	log_data["cdn_region"] = cdn_region	1✔
101	log_data["region_method"] = "cdn"	1✔
102
103	accept_language_region = None	1✔
104	if "Accept-Language" in request.headers:	1✔
105	log_data["accept_lang"] = request.headers["Accept-Language"]	1✔
106	accept_language_region = _get_cc_from_lang(request.headers["Accept-Language"])	1✔
107	log_data["accept_lang_region"] = accept_language_region	1✔
108	if region is None:	1✔
109	region = accept_language_region	1✔
110	log_data["region_method"] = "accept_lang"	1✔
111
112	if region is None:	1✔
113	region = "US"	1✔
114	log_data["region_method"] = "fallback"	1✔
115	log_data["region"] = region	1✔
116
117	# MPP-3284: Log details of region selection. Only log once per request, since some
118	# endpoints, like /api/v1/runtime_data, call this multiple times.
119	if not getattr(request, "_logged_region_details", False):	1✔
120	setattr(request, "_logged_region_details", True)	1✔
121	info_logger.info("region_details", extra=log_data)	1✔
122
123	return region	1✔
124
125
126	def _get_cc_from_lang(accept_lang: str) -> str:	1✔
127	try:	1✔
128	return guess_country_from_accept_lang(accept_lang)	1✔
129	except AcceptLanguageError:	1✔
130	return ""	1✔
131
132
133	# Map a primary language to the most probable country
134	# Top country derived from CLDR42 Supplemental Data, Language-Territory Information
135	# with the exception of Spanish (es), which is mapped to Spain (es) instead of
136	# Mexico (mx), which has the most speakers, but usually specifies es-MX.
137	_PRIMARY_LANGUAGE_TO_COUNTRY = {	1✔
138	"ace": "ID", # # Acehnese -> Indonesia
139	"ach": "UG", # Acholi -> Uganda
140	"af": "ZA", # Afrikaans -> South Africa
141	"an": "ES", # Aragonese -> Spain
142	"ar": "EG", # Arabic -> Egypt
143	"arn": "CL", # Mapudungun -> Chile
144	"as": "IN", # Assamese -> India
145	"ast": "ES", # Asturian -> Spain
146	"az": "AZ", # Azerbaijani -> Azerbaijan
147	"be": "BY", # Belerusian -> Belarus
148	"bg": "BG", # Bulgarian -> Bulgaria
149	"bn": "BD", # Bengali -> Bangladesh
150	"bo": "CN", # Tibetan -> China
151	"br": "FR", # Breton -> France
152	"brx": "IN", # Bodo -> India
153	"bs": "BA", # Bosnian -> Bosnia and Herzegovina
154	"ca": "FR", # Catalan -> France
155	"cak": "MX", # Kaqchikel -> Mexico
156	"ckb": "IQ", # Central Kurdish -> Iraq
157	"cs": "CZ", # Czech -> Czech Republic
158	"cv": "RU", # Chuvash -> Russia
159	"cy": "GB", # Welsh -> United Kingdom
160	"da": "DK", # Danish -> Denmark
161	"de": "DE", # German -> Germany
162	"dsb": "DE", # Lower Sorbian -> Germany
163	"el": "GR", # Greek -> Greece
164	"en": "US", # English -> Canada
165	"eo": "SM", # Esperanto -> San Marino
166	"es": "ES", # Spanish -> Spain (instead of Mexico, top by population)
167	"et": "EE", # Estonian -> Estonia
168	"eu": "ES", # Basque -> Spain
169	"fa": "IR", # Persian -> Iran
170	"ff": "SN", # Fulah -> Senegal
171	"fi": "FI", # Finnish -> Finland
172	"fr": "FR", # French -> France
173	"frp": "FR", # Arpitan -> France
174	"fur": "IT", # Friulian -> Italy
175	"fy": "NL", # Frisian -> Netherlands
176	"ga": "IE", # Irish -> Ireland
177	"gd": "GB", # Scottish Gaelic -> United Kingdom
178	"gl": "ES", # Galician -> Spain
179	"gn": "PY", # Guarani -> Paraguay
180	"gu": "IN", # Gujarati -> India
181	"gv": "IM", # Manx -> Isle of Man
182	"he": "IL", # Hebrew -> Israel
183	"hi": "IN", # Hindi -> India
184	"hr": "HR", # Croatian -> Croatia
185	"hsb": "DE", # Upper Sorbian -> Germany
186	"hu": "HU", # Hungarian -> Hungary
187	"hy": "AM", # Armenian -> Armenia
188	"hye": "AM", # Armenian Eastern Classic Orthography -> Armenia
189	"ia": "FR", # Interlingua -> France
190	"id": "ID", # Indonesian -> Indonesia
191	"ilo": "PH", # Iloko -> Philippines
192	"is": "IS", # Icelandic -> Iceland
193	"it": "IT", # Italian -> Italy
194	"ixl": "MX", # Ixil -> Mexico
195	"ja": "JP", # Japanese -> Japan
196	"jiv": "MX", # Shuar -> Mexico
197	"ka": "GE", # Georgian -> Georgia
198	"kab": "DZ", # Kayble -> Algeria
199	"kk": "KZ", # Kazakh -> Kazakhstan
200	"km": "KH", # Khmer -> Cambodia
201	"kn": "IN", # Kannada -> India
202	"ko": "KR", # Korean -> South Korea
203	"ks": "IN", # Kashmiri -> India
204	"lb": "LU", # Luxembourgish -> Luxembourg
205	"lg": "UG", # Luganda -> Uganda
206	"lij": "IT", # Ligurian -> Italy
207	"lo": "LA", # Lao -> Laos
208	"lt": "LT", # Lithuanian -> Lithuania
209	"ltg": "LV", # Latgalian -> Latvia
210	"lus": "US", # Mizo -> United States
211	"lv": "LV", # Latvian -> Latvia
212	"mai": "IN", # Maithili -> India
213	"meh": "MX", # Mixteco Yucuhiti -> Mexico
214	"mix": "MX", # Mixtepec Mixtec -> Mexico
215	"mk": "MK", # Macedonian -> North Macedonia
216	"ml": "IN", # Malayalam -> India
217	"mr": "IN", # Marathi -> India
218	"ms": "MY", # Malay -> Malaysia
219	"my": "MM", # Burmese -> Myanmar
220	"nb": "NO", # Norwegian Bokmål -> Norway
221	"ne": "NP", # Nepali -> Nepal
222	"nl": "NL", # Dutch -> Netherlands
223	"nn": "NO", # Norwegian Nynorsk -> Norway
224	"oc": "FR", # Occitan -> France
225	"or": "IN", # Odia -> India
226	"pa": "IN", # Punjabi -> India
227	"pl": "PL", # Polish -> Poland
228	"ppl": "MX", # Náhuat Pipil -> Mexico
229	"pt": "BR", # Portuguese -> Brazil
230	"quc": "GT", # K'iche' -> Guatemala
231	"rm": "CH", # Romansh -> Switzerland
232	"ro": "RO", # Romanian -> Romania
233	"ru": "RU", # Russian -> Russia
234	"sat": "IN", # Santali (Ol Chiki) -> India
235	"sc": "IT", # Sardinian -> Italy
236	"scn": "IT", # Sicilian -> Italy
237	"sco": "GB", # Scots -> United Kingdom
238	"si": "LK", # Sinhala -> Sri Lanka
239	"sk": "SK", # Slovak -> Slovakia
240	"skr": "PK", # Saraiki -> Pakistan
241	"sl": "SI", # Slovenian -> Slovenia
242	"son": "ML", # Songhay -> Mali
243	"sq": "AL", # Albanian -> Albania
244	"sr": "RS", # Serbian -> Serbia
245	"sv": "SE", # Swedish -> Sweeden
246	"sw": "TZ", # Swahili -> Tanzania
247	"szl": "PL", # Silesian -> Poland
248	"ta": "IN", # Tamil -> India
249	"te": "IN", # Telugu -> India
250	"tg": "TJ", # Tajik -> Tajikistan
251	"th": "TH", # Thai -> Thailand
252	"tl": "PH", # Tagalog -> Philippines
253	"tr": "TR", # Turkish or Crimean Tatar -> Turkey
254	"trs": "MX", # Triqui -> Mexico
255	"uk": "UA", # Ukrainian -> Ukraine
256	"ur": "PK", # Urdu -> Pakistan
257	"uz": "UZ", # Uzbek -> Uzbekistan
258	"vi": "VN", # Vietnamese -> Vietnam
259	"wo": "SN", # Wolof -> Senegal
260	"xcl": "AM", # Armenian Classic -> Armenia
261	"xh": "ZA", # Xhosa -> South Africa
262	"zam": "MX", # Miahuatlán Zapotec -> Mexico
263	"zh": "CN", # Chinese -> China
264	}
265
266	# Special cases for language tags
267	_LANGUAGE_TAG_TO_COUNTRY_OVERRIDE = {	1✔
268	# Would be Catalan in Valencian script -> France
269	# Change to Valencian -> Spain
270	("ca", "VALENCIA"): "ES",
271	# Spanish in UN region 419 (Latin America and Carribean)
272	# Pick Mexico, which has highest number of Spanish speakers
273	("es", "419"): "MX",
274	# Would be Galician (Greenland) -> Greenland
275	# Change to Galician (Galicia region of Spain) -> Spain
276	("gl", "GL"): "ES",
277	}
278
279
280	class AcceptLanguageError(ValueError):	1✔
281	"""There was an issue processing the Accept-Language header."""
282
283	def __init__(self, message, accept_lang):	1✔
284	super().__init__(message)	1✔
285	self.accept_lang = accept_lang	1✔
286
287
288	def guess_country_from_accept_lang(accept_lang: str) -> str:	1✔
289	"""
290	Guess the user's country from the Accept-Language header
291
292	Return is a 2-letter ISO 3166 country code
293
294	If an issue is detected, a AcceptLanguageError is raised.
295
296	The header may come directly from a web request, or may be the header
297	captured by Mozilla Accounts (FxA) at signup.
298
299	Even with all this logic and special casing, it is still more accurate to
300	use a GeoIP lookup or a country code provided by the infrastructure.
301
302	See RFC 9110, "HTTP Semantics", section 12.5.4, "Accept-Language"
303	See RFC 5646, "Tags for Identifying Languages", and examples in Appendix A
304	"""
305	lang_q_pairs = parse_accept_lang_header(accept_lang.strip())	1✔
306	if not lang_q_pairs:	1✔
307	raise AcceptLanguageError("Invalid Accept-Language string", accept_lang)	1✔
308	top_lang_tag = lang_q_pairs[0][0]	1✔
309
310	subtags = top_lang_tag.split("-")	1✔
311	lang = subtags[0].lower()	1✔
312	if lang == "i":	1✔
313	raise AcceptLanguageError("Irregular language tag", accept_lang)	1✔
314	if lang == "x":	1✔
315	raise AcceptLanguageError("Private-use language tag", accept_lang)	1✔
316	if lang == "*":	1✔
317	raise AcceptLanguageError("Wildcard language tag", accept_lang)	1✔
318	if len(lang) < 2:	1✔
319	raise AcceptLanguageError("Invalid one-character primary language", accept_lang)	1✔
320	if len(lang) == 3 and lang[0] == "q" and lang[1] <= "t":	1✔
321	raise AcceptLanguageError(	1✔
322	"Private-use language tag (RFC 5646 2.2.1)", accept_lang
323	)
324
325	for maybe_region_raw in subtags[1:]:	1✔
326	maybe_region = maybe_region_raw.upper()	1✔
327
328	# Look for a special case
329	if override := _LANGUAGE_TAG_TO_COUNTRY_OVERRIDE.get((lang, maybe_region)):	1✔
330	return override	1✔
331
332	if len(maybe_region) <= 1:	1✔
333	# One-character extension or empty, stop processing
334	break	1✔
335	if (	1✔
336	len(maybe_region) == 2
337	and all(c in ascii_uppercase for c in maybe_region)
338	and
339	# RFC 5646 2.2.4 "Region Subtag" point 6, reserved subtags
340	maybe_region != "AA"
341	and maybe_region != "ZZ"
342	and maybe_region[0] != "X"
343	and (maybe_region[0] != "Q" or maybe_region[1] < "M")
344	):
345	# Subtag is a non-private ISO 3166 country code
346	return maybe_region	1✔
347
348	# Subtag is probably a script, like "Hans" in "zh-Hans-CN"
349	# Loop to the next subtag, which might be a ISO 3166 country code
350
351	# Guess the country from a simple language tag
352	try:	1✔
353	return _PRIMARY_LANGUAGE_TO_COUNTRY[lang]	1✔
354	except KeyError:	1✔
355	raise AcceptLanguageError("Unknown langauge", accept_lang)	1✔
356
357
358	# Generics for defining function decorators
359	# https://mypy.readthedocs.io/en/stable/generics.html#declaring-decorators
360	_Params = ParamSpec("_Params")	1✔
361	_RetVal = TypeVar("_RetVal")	1✔
362
363
364	def enable_or_404(	1✔
365	check_function: Callable[[], bool],
366	message: str = "This conditional view is disabled.",
367	) -> Callable[[Callable[_Params, _RetVal]], Callable[_Params, _RetVal]]:
368	"""
369	Returns decorator that enables a view if a check function passes,
370	otherwise returns a 404.
371
372	Usage:
373
374	def percent_1():
375	import random
376	return random.randint(1, 100) == 1
377
378	@enable_if(percent_1)
379	def lucky_view(request):
380	# 1 in 100 chance of getting here
381	# 99 in 100 chance of 404
382	"""
383
384	def decorator(func: Callable[_Params, _RetVal]) -> Callable[_Params, _RetVal]:	1✔
385	@wraps(func)	1✔
386	def inner(args: _Params.args, *kwargs: _Params.kwargs) -> _RetVal:	1✔
387	if check_function():	×
388	return func(args, *kwargs)	×
389	else:
390	raise Http404(message) # Display a message with DEBUG=True	×
391
392	return inner	1✔
393
394	return decorator	1✔
395
396
397	def enable_if_setting(	1✔
398	setting_name: str,
399	message_fmt: str = "This view is disabled because {setting_name} is False",
400	) -> Callable[[Callable[_Params, _RetVal]], Callable[_Params, _RetVal]]:
401	"""
402	Returns decorator that enables a view if a setting is truthy, otherwise
403	returns a 404.
404
405	Usage:
406
407	@enable_if_setting("DEBUG")
408	def debug_only_view(request):
409	# DEBUG == True
410
411	Or in URLS:
412
413	path(
414	"developer_info",
415	enable_if_setting("DEBUG")(debug_only_view)
416	),
417	name="developer-info",
418	),
419
420	"""
421
422	def setting_is_truthy() -> bool:	1✔
423	return bool(getattr(settings, setting_name))	×
424
425	return enable_or_404(	1✔
426	setting_is_truthy, message_fmt.format(setting_name=setting_name)
427	)
428
429
430	def flag_is_active_in_task(flag_name: str, user: AbstractBaseUser \| None) -> bool:	1✔
431	"""
432	Test if a flag is active in a task (not in a web request).
433
434	This mirrors AbstractBaseFlag.is_active, replicating these checks:
435	* Logs missing flags, if configured
436	* Creates missing flags, if configured
437	* Returns default for missing flags
438	* Checks flag.everyone
439	* Checks flag.users and flag.groups, if a user is passed
440	* Returns random results for flag.percent
441
442	It does not include:
443	* Overriding a flag with a query parameter
444	* Persisting a flag in a cookie (includes percent flags)
445	* Language-specific overrides (could be added later)
446	* Read-only mode for percent flags
447
448	When using this function, use the @override_flag decorator in tests, rather
449	than manually creating flags in the database.
450	"""
451	flag = get_waffle_flag_model().get(flag_name)	1✔
452	if not flag.pk:	1✔
453	log_level = get_waffle_setting("LOG_MISSING_FLAGS")	1✔
454	if log_level:	1✔
455	waffle_logger.log(log_level, "Flag %s not found", flag_name)	1✔
456	if get_waffle_setting("CREATE_MISSING_FLAGS"):	1✔
457	flag, _created = get_waffle_flag_model().objects.get_or_create(	1✔
458	name=flag_name,
459	defaults={"everyone": get_waffle_setting("FLAG_DEFAULT")},
460	)
461	cache = get_waffle_cache()	1✔
462	cache.set(flag._cache_key(flag.name), flag)	1✔
463
464	return bool(get_waffle_setting("FLAG_DEFAULT"))	1✔
465
466	# Removed - check for override as request query parameter
467
468	if flag.everyone:	1✔
469	return True	1✔
470	elif flag.everyone is False:	1✔
471	return False	1✔
472
473	# Removed - check for testing override in request query or cookie
474	# Removed - check for language-specific override
475
476	if user is not None:	1✔
477	active_for_user = flag.is_active_for_user(user)	1✔
478	if active_for_user is not None:	1✔
479	return bool(active_for_user)	1✔
480
481	if flag.percent and flag.percent > 0:	1✔
482	# Removed - check for waffles attribute of request
483	# Removed - check for cookie setting for flag
484	# Removed - check for read-only mode
485
486	if Decimal(str(random.uniform(0, 100))) <= flag.percent: # noqa: S311	1✔
487	# Removed - setting the flag for future checks
488	return True	1✔
489
490	return False	1✔
491
492
493	class VersionInfo(TypedDict):	1✔
494	source: str	1✔
495	version: str	1✔
496	commit: str	1✔
497	build: str	1✔
498
499
500	@cache	1✔
501	def get_version_info(base_dir: str \| Path \| None = None) -> VersionInfo:	1✔
502	"""Return version information written by build process."""
503	if base_dir is None:	1✔
504	base_path = Path(settings.BASE_DIR)	1✔
505	else:
506	base_path = Path(base_dir)	1✔
507	version_json_path = base_path / "version.json"	1✔
508	info = {}	1✔
509	if version_json_path.exists():	1✔
510	with version_json_path.open() as version_file:	1✔
511	try:	1✔
512	info = json.load(version_file)	1✔
513	except ValueError:	1✔
514	pass	1✔
515	if not hasattr(info, "get"):	1✔
516	info = {}	1✔
517	version_info = VersionInfo(	1✔
518	source=info.get("source", "https://github.com/mozilla/fx-private-relay"),
519	version=info.get("version", "unknown"),
520	commit=info.get("commit", "unknown"),
521	build=info.get("build", "not built"),
522	)
523	return version_info	1✔
524
525
526	@cache	1✔
527	def glean_logger() -> RelayGleanLogger:	1✔
528	from .glean_interface import RelayGleanLogger	1✔
529
530	version_info = get_version_info()	1✔
531	return RelayGleanLogger(	1✔
532	application_id="relay-backend",
533	app_display_version=version_info["version"],
534	channel=settings.RELAY_CHANNEL,
535	)

mozilla / fx-private-relay / 0d7efbc0-b94b-416d-be6b-612c09993f6b

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