76b0d84c-52e7-4179-8157-f3582acddb0b

Committed 05 Jun 2024 09:15PM UTC coverage: 84.718% (+0.02%) from 84.701%

Build # 76b0d84c-52e7-4179-8157-f3582acddb0b

Build Type

push

circleci

Committed by

jwhitlock

Commit Message

Remove unneeded transaction=True

Run Details

3713 of 4846 branches covered (76.62%)

Branch coverage included in aggregate %.

14 of 14 new or added lines in 2 files covered. (100.0%)

10 existing lines in 1 file now uncovered.

15113 of 17376 relevant lines covered (86.98%)

10.65 hits per line

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

96.19

/emails/models.py

from __future__ import annotations

import logging
import random
import re
import string
from collections.abc import Iterable
from datetime import UTC, datetime
from hashlib import sha256
from typing import Literal, cast

from django.conf import settings
from django.contrib.auth.models import User
from django.core.validators import MinLengthValidator
from django.db import models, transaction
from django.db.models.base import ModelBase

from api.exceptions import ErrorContextType, RelayAPIException
from privaterelay.utils import flag_is_active_in_task

from .utils import get_domains_from_settings, incr_if_enabled
from .validators import has_bad_words, is_blocklisted

logger = logging.getLogger("events")


DOMAIN_CHOICES = [(1, "RELAY_FIREFOX_DOMAIN"), (2, "MOZMAIL_DOMAIN")]


# This historical function is referenced in migration
# 0029_profile_add_deleted_metric_and_changeserver_storage_default
def default_server_storage():
    return True


def default_domain_numerical():
    domains = get_domains_from_settings()
    domain = domains["MOZMAIL_DOMAIN"]
    return get_domain_numerical(domain)


def address_hash(address, subdomain=None, domain=None):
    if not domain:
        domain = get_domains_from_settings()["MOZMAIL_DOMAIN"]
    if subdomain:
        return sha256(f"{address}@{subdomain}.{domain}".encode()).hexdigest()
    if domain == settings.RELAY_FIREFOX_DOMAIN:
        return sha256(f"{address}".encode()).hexdigest()
    return sha256(f"{address}@{domain}".encode()).hexdigest()


def address_default():
    return "".join(
        random.choices(  # noqa: S311 (standard pseudo-random generator used)
            string.ascii_lowercase + string.digits, k=9
        )
    )


def get_domain_numerical(domain_address):
    # get domain name from the address
    domains = get_domains_from_settings()
    domains_keys = list(domains.keys())
    domains_values = list(domains.values())
    domain_name = domains_keys[domains_values.index(domain_address)]
    # get domain numerical value from domain name
    choices = dict(DOMAIN_CHOICES)
    choices_keys = list(choices.keys())
    choices_values = list(choices.values())
    return choices_keys[choices_values.index(domain_name)]


def hash_subdomain(subdomain, domain=settings.MOZMAIL_DOMAIN):
    return sha256(f"{subdomain}.{domain}".encode()).hexdigest()


class RegisteredSubdomain(models.Model):
    subdomain_hash = models.CharField(max_length=64, db_index=True, unique=True)
    registered_at = models.DateTimeField(auto_now_add=True)

    def __str__(self):
        return self.subdomain_hash


class CannotMakeAddressException(RelayAPIException):
    """Base exception for RelayAddress or DomainAddress creation failure."""


class AccountIsPausedException(CannotMakeAddressException):
    default_code = "account_is_paused"
    default_detail = "Your account is on pause."
    status_code = 403


class AccountIsInactiveException(CannotMakeAddressException):
    default_code = "account_is_inactive"
    default_detail = "Your account is not active."
    status_code = 403


class RelayAddrFreeTierLimitException(CannotMakeAddressException):
    default_code = "free_tier_limit"
    default_detail_template = (
        "You’ve used all {free_tier_limit} email masks included with your free account."
        " You can reuse an existing mask, but using a unique mask for each account is"
        " the most secure option."
    )
    status_code = 403

    def __init__(self, free_tier_limit: int | None = None):
        self.free_tier_limit = free_tier_limit or settings.MAX_NUM_FREE_ALIASES
        super().__init__()

    def error_context(self) -> ErrorContextType:
        return {"free_tier_limit": self.free_tier_limit}


class DomainAddrFreeTierException(CannotMakeAddressException):
    default_code = "free_tier_no_subdomain_masks"
    default_detail = (
        "Your free account does not include custom subdomains for masks."
        " To create custom masks, upgrade to Relay Premium."
    )
    status_code = 403


class DomainAddrNeedSubdomainException(CannotMakeAddressException):
    default_code = "need_subdomain"
    default_detail = "Please select a subdomain before creating a custom email address."
    status_code = 400


class DomainAddrUpdateException(CannotMakeAddressException):
    """Exception raised when attempting to edit an existing domain address field."""

    default_code = "address_not_editable"
    default_detail = "You cannot edit an existing domain address field."
    status_code = 400


class DomainAddrUnavailableException(CannotMakeAddressException):
    default_code = "address_unavailable"
    default_detail_template = (
        "“{unavailable_address}” could not be created."
        " Please try again with a different mask name."
    )
    status_code = 400

    def __init__(self, unavailable_address: str):
        self.unavailable_address = unavailable_address
        super().__init__()

    def error_context(self) -> ErrorContextType:
        return {"unavailable_address": self.unavailable_address}


class DomainAddrDuplicateException(CannotMakeAddressException):
    default_code = "duplicate_address"
    default_detail_template = (
        "“{duplicate_address}” already exists."
        " Please try again with a different mask name."
    )
    status_code = 409

    def __init__(self, duplicate_address: str):
        self.duplicate_address = duplicate_address
        super().__init__()

    def error_context(self) -> ErrorContextType:
        return {"duplicate_address": self.duplicate_address}


class RelayAddress(models.Model):
    user = models.ForeignKey(User, on_delete=models.CASCADE)
    address = models.CharField(max_length=64, default=address_default, unique=True)
    domain = models.PositiveSmallIntegerField(
        choices=DOMAIN_CHOICES, default=default_domain_numerical
    )
    enabled = models.BooleanField(default=True)
    description = models.CharField(max_length=64, blank=True)
    created_at = models.DateTimeField(auto_now_add=True, db_index=True)
    last_modified_at = models.DateTimeField(auto_now=True, db_index=True)
    last_used_at = models.DateTimeField(blank=True, null=True)
    num_forwarded = models.PositiveIntegerField(default=0)
    num_blocked = models.PositiveIntegerField(default=0)
    num_level_one_trackers_blocked = models.PositiveIntegerField(default=0, null=True)
    num_replied = models.PositiveIntegerField(default=0)
    num_spam = models.PositiveIntegerField(default=0)
    generated_for = models.CharField(max_length=255, blank=True)
    block_list_emails = models.BooleanField(default=False)
    used_on = models.TextField(default=None, blank=True, null=True)

    class Meta:
        indexes = [
            # Find when a user first used the add-on
            models.Index(
                name="idx_ra_created_by_addon",
                fields=["user"],
                condition=~models.Q(generated_for__exact=""),
                include=["created_at"],
            ),
        ]

    def __str__(self):
        return self.address

    def delete(self, *args, **kwargs):
        # TODO: create hard bounce receipt rule in AWS for the address
        deleted_address = DeletedAddress.objects.create(
            address_hash=address_hash(self.address, domain=self.domain_value),
            num_forwarded=self.num_forwarded,
            num_blocked=self.num_blocked,
            num_replied=self.num_replied,
            num_spam=self.num_spam,
        )
        deleted_address.save()
        profile = self.user.profile
        profile.refresh_from_db()
        profile.address_last_deleted = datetime.now(UTC)
        profile.num_address_deleted += 1
        profile.num_email_forwarded_in_deleted_address += self.num_forwarded
        profile.num_email_blocked_in_deleted_address += self.num_blocked
        profile.num_level_one_trackers_blocked_in_deleted_address = (
            profile.num_level_one_trackers_blocked_in_deleted_address or 0
        ) + (self.num_level_one_trackers_blocked or 0)
        profile.num_email_replied_in_deleted_address += self.num_replied
        profile.num_email_spam_in_deleted_address += self.num_spam
        profile.num_deleted_relay_addresses += 1
        profile.last_engagement = datetime.now(UTC)
        profile.save()
        return super().delete(*args, **kwargs)

    def save(
        self,
        force_insert: bool | tuple[ModelBase, ...] = False,
        force_update: bool = False,
        using: str | None = None,
        update_fields: Iterable[str] | None = None,
    ) -> None:
        from privaterelay.models import Profile

        if self._state.adding:
            with transaction.atomic():
                locked_profile = Profile.objects.select_for_update().get(user=self.user)
                check_user_can_make_another_address(locked_profile.user)
                while True:
                    address_is_allowed = not is_blocklisted(self.address)
                    address_is_valid = valid_address(self.address, self.domain_value)
                    if address_is_valid and address_is_allowed:
                        break
                    self.address = address_default()
                locked_profile.update_abuse_metric(address_created=True)
                locked_profile.last_engagement = datetime.now(UTC)
                locked_profile.save()
        if (not self.user.profile.server_storage) and any(
            (self.description, self.generated_for, self.used_on)
        ):
            self.description = ""
            self.generated_for = ""
            self.used_on = ""
            if update_fields is not None:
                update_fields = {"description", "generated_for", "used_on"}.union(
                    update_fields
                )
        if not self.user.profile.has_premium and self.block_list_emails:
            self.block_list_emails = False
            if update_fields is not None:
                update_fields = {"block_list_emails"}.union(update_fields)
        super().save(
            force_insert=force_insert,
            force_update=force_update,
            using=using,
            update_fields=update_fields,
        )

    @property
    def domain_value(self) -> str:
        domain = cast(
            Literal["RELAY_FIREFOX_DOMAIN", "MOZMAIL_DOMAIN"], self.get_domain_display()
        )
        return get_domains_from_settings()[domain]

    @property
    def full_address(self) -> str:
        return f"{self.address}@{self.domain_value}"

    @property
    def metrics_id(self) -> str:
        if not self.id:
            raise ValueError("self.id must be truthy value.")
        # Prefix with 'R' for RelayAddress, since there may be a DomainAddress with the
        # same row ID
        return f"R{self.id}"


def check_user_can_make_another_address(user: User) -> None:
    if not user.is_active:
        raise AccountIsInactiveException()

    if user.profile.is_flagged:
        raise AccountIsPausedException()
    # MPP-3021: return early for premium users to avoid at_max_free_aliases DB query
    if user.profile.has_premium:
        return
    if user.profile.at_max_free_aliases:
        raise RelayAddrFreeTierLimitException()


def valid_address_pattern(address):
    #   can't start or end with a hyphen
    #   must be 1-63 lowercase alphanumeric characters and/or hyphens
    valid_address_pattern = re.compile("^(?![-.])[a-z0-9-.]{1,63}(?<![-.])$")
    return valid_address_pattern.match(address) is not None


def valid_address(address: str, domain: str, subdomain: str | None = None) -> bool:
    address_pattern_valid = valid_address_pattern(address)
    address_contains_badword = has_bad_words(address)
    address_already_deleted = 0
    if not subdomain or flag_is_active_in_task(
        "custom_domain_management_redesign", None
    ):
        address_already_deleted = DeletedAddress.objects.filter(
            address_hash=address_hash(address, domain=domain, subdomain=subdomain)
        ).count()
    if (
        address_already_deleted > 0
        or address_contains_badword
        or not address_pattern_valid
    ):
        return False
    return True


class DeletedAddress(models.Model):
    address_hash = models.CharField(max_length=64, db_index=True)
    num_forwarded = models.PositiveIntegerField(default=0)
    num_blocked = models.PositiveIntegerField(default=0)
    num_replied = models.PositiveIntegerField(default=0)
    num_spam = models.PositiveIntegerField(default=0)

    def __str__(self):
        return self.address_hash


def check_user_can_make_domain_address(user: User) -> None:
    if not user.profile.has_premium:
        raise DomainAddrFreeTierException()

    if not user.profile.subdomain:
        raise DomainAddrNeedSubdomainException()

    if not user.profile.user.is_active:
        raise AccountIsInactiveException()

    if user.profile.is_flagged:
        raise AccountIsPausedException()


class DomainAddress(models.Model):
    user = models.ForeignKey(User, on_delete=models.CASCADE)
    address = models.CharField(
        max_length=64, validators=[MinLengthValidator(limit_value=1)]
    )
    enabled = models.BooleanField(default=True)
    description = models.CharField(max_length=64, blank=True)
    domain = models.PositiveSmallIntegerField(choices=DOMAIN_CHOICES, default=2)
    created_at = models.DateTimeField(auto_now_add=True, db_index=True)
    first_emailed_at = models.DateTimeField(null=True, db_index=True)
    last_modified_at = models.DateTimeField(auto_now=True, db_index=True)
    last_used_at = models.DateTimeField(blank=True, null=True)
    num_forwarded = models.PositiveIntegerField(default=0)
    num_blocked = models.PositiveIntegerField(default=0)
    num_level_one_trackers_blocked = models.PositiveIntegerField(default=0, null=True)
    num_replied = models.PositiveIntegerField(default=0)
    num_spam = models.PositiveIntegerField(default=0)
    block_list_emails = models.BooleanField(default=False)
    used_on = models.TextField(default=None, blank=True, null=True)

    class Meta:
        unique_together = ["user", "address"]

    def __str__(self):
        return self.address

    def save(
        self,
        force_insert: bool | tuple[ModelBase, ...] = False,
        force_update: bool = False,
        using: str | None = None,
        update_fields: Iterable[str] | None = None,
    ) -> None:
        user_profile = self.user.profile
        if self._state.adding:
            check_user_can_make_domain_address(self.user)
            domain_address_valid = valid_address(
                self.address, self.domain_value, user_profile.subdomain
            )
            if not domain_address_valid:
                if self.first_emailed_at:
                    incr_if_enabled("domainaddress.create_via_email_fail")
                raise DomainAddrUnavailableException(unavailable_address=self.address)

            if DomainAddress.objects.filter(
                user=self.user, address=self.address
            ).exists():
                raise DomainAddrDuplicateException(duplicate_address=self.address)

            user_profile.update_abuse_metric(address_created=True)
            user_profile.last_engagement = datetime.now(UTC)
            user_profile.save(update_fields=["last_engagement"])
            incr_if_enabled("domainaddress.create")
            if self.first_emailed_at:
                incr_if_enabled("domainaddress.create_via_email")
        else:
            # The model is in an update state, do not allow 'address' field updates
            existing_instance = DomainAddress.objects.get(id=self.id)
            if existing_instance.address != self.address:
                raise DomainAddrUpdateException()

        if not user_profile.has_premium and self.block_list_emails:
            self.block_list_emails = False
            if update_fields:
                update_fields = {"block_list_emails"}.union(update_fields)
        if (not user_profile.server_storage) and (self.description or self.used_on):
            self.description = ""
            self.used_on = ""
            if update_fields:
                update_fields = {"description", "used_on"}.union(update_fields)
        super().save(
            force_insert=force_insert,
            force_update=force_update,
            using=using,
            update_fields=update_fields,
        )

    @staticmethod
    def make_domain_address(
        user: User, address: str | None = None, made_via_email: bool = False
    ) -> DomainAddress:
        check_user_can_make_domain_address(user)

        if not address:
            # FIXME: if the alias is randomly generated and has bad words
            # we should retry like make_relay_address does
            # not fixing this now because not sure randomly generated
            # DomainAlias will be a feature
            address = address_default()
            # Only check for bad words if randomly generated
        if not isinstance(address, str):
            raise TypeError("address must be type str")

        first_emailed_at = datetime.now(UTC) if made_via_email else None
        domain_address = DomainAddress.objects.create(
            user=user, address=address, first_emailed_at=first_emailed_at
        )
        return domain_address

    def delete(self, *args, **kwargs):
        # TODO: create hard bounce receipt rule in AWS for the address
        deleted_address = DeletedAddress.objects.create(
            address_hash=address_hash(
                self.address, self.user.profile.subdomain, self.domain_value
            ),
            num_forwarded=self.num_forwarded,
            num_blocked=self.num_blocked,
            num_replied=self.num_replied,
            num_spam=self.num_spam,
        )
        deleted_address.save()
        profile = self.user.profile
        profile.address_last_deleted = datetime.now(UTC)
        profile.num_address_deleted += 1
        profile.num_email_forwarded_in_deleted_address += self.num_forwarded
        profile.num_email_blocked_in_deleted_address += self.num_blocked
        profile.num_level_one_trackers_blocked_in_deleted_address = (
            profile.num_level_one_trackers_blocked_in_deleted_address or 0
        ) + (self.num_level_one_trackers_blocked or 0)
        profile.num_email_replied_in_deleted_address += self.num_replied
        profile.num_email_spam_in_deleted_address += self.num_spam
        profile.num_deleted_domain_addresses += 1
        profile.last_engagement = datetime.now(UTC)
        profile.save()
        return super().delete(*args, **kwargs)

    @property
    def domain_value(self) -> str:
        domain = cast(
            Literal["RELAY_FIREFOX_DOMAIN", "MOZMAIL_DOMAIN"], self.get_domain_display()
        )
        return get_domains_from_settings()[domain]

    @property
    def full_address(self) -> str:
        return f"{self.address}@{self.user.profile.subdomain}.{self.domain_value}"

    @property
    def metrics_id(self) -> str:
        if not self.id:
            raise ValueError("self.id must be truthy value.")
        # Prefix with 'D' for DomainAddress, since there may be a RelayAddress with the
        # same row ID
        return f"D{self.id}"


class Reply(models.Model):
    relay_address = models.ForeignKey(
        RelayAddress, on_delete=models.CASCADE, blank=True, null=True
    )
    domain_address = models.ForeignKey(
        DomainAddress, on_delete=models.CASCADE, blank=True, null=True
    )
    lookup = models.CharField(max_length=255, blank=False, db_index=True)
    encrypted_metadata = models.TextField(blank=False)
    created_at = models.DateField(auto_now_add=True, null=False, db_index=True)

    @property
    def address(self):
        return self.relay_address or self.domain_address

    @property
    def profile(self):
        return self.address.user.profile

    @property
    def owner_has_premium(self):
        return self.profile.has_premium

    def increment_num_replied(self):
        address = self.relay_address or self.domain_address
        if not address:
            raise ValueError("address must be truthy value")
        address.num_replied += 1
        address.last_used_at = datetime.now(UTC)
        address.save(update_fields=["num_replied", "last_used_at"])
        return address.num_replied

1	from __future__ import annotations	1✔
2
3	import logging	1✔
4	import random	1✔
5	import re	1✔
6	import string	1✔
7	from collections.abc import Iterable	1✔
8	from datetime import UTC, datetime	1✔
9	from hashlib import sha256	1✔
10	from typing import Literal, cast	1✔
11
12	from django.conf import settings	1✔
13	from django.contrib.auth.models import User	1✔
14	from django.core.validators import MinLengthValidator	1✔
15	from django.db import models, transaction	1✔
16	from django.db.models.base import ModelBase	1✔
17
18	from api.exceptions import ErrorContextType, RelayAPIException	1✔
19	from privaterelay.utils import flag_is_active_in_task	1✔
20
21	from .utils import get_domains_from_settings, incr_if_enabled	1✔
22	from .validators import has_bad_words, is_blocklisted	1✔
23
24	logger = logging.getLogger("events")	1✔
25
26
27	DOMAIN_CHOICES = [(1, "RELAY_FIREFOX_DOMAIN"), (2, "MOZMAIL_DOMAIN")]	1✔
28
29
30	# This historical function is referenced in migration
31	# 0029_profile_add_deleted_metric_and_changeserver_storage_default
32	def default_server_storage():	1✔
UNCOV 33	return True	×
34
35
36	def default_domain_numerical():	1✔
37	domains = get_domains_from_settings()	1✔
38	domain = domains["MOZMAIL_DOMAIN"]	1✔
39	return get_domain_numerical(domain)	1✔
40
41
42	def address_hash(address, subdomain=None, domain=None):	1✔
43	if not domain:	1✔
44	domain = get_domains_from_settings()["MOZMAIL_DOMAIN"]	1✔
45	if subdomain:	1✔
46	return sha256(f"{address}@{subdomain}.{domain}".encode()).hexdigest()	1✔
47	if domain == settings.RELAY_FIREFOX_DOMAIN:	1✔
48	return sha256(f"{address}".encode()).hexdigest()	1✔
49	return sha256(f"{address}@{domain}".encode()).hexdigest()	1✔
50
51
52	def address_default():	1✔
53	return "".join(	1✔
54	random.choices( # noqa: S311 (standard pseudo-random generator used)
55	string.ascii_lowercase + string.digits, k=9
56	)
57	)
58
59
60	def get_domain_numerical(domain_address):	1✔
61	# get domain name from the address
62	domains = get_domains_from_settings()	1✔
63	domains_keys = list(domains.keys())	1✔
64	domains_values = list(domains.values())	1✔
65	domain_name = domains_keys[domains_values.index(domain_address)]	1✔
66	# get domain numerical value from domain name
67	choices = dict(DOMAIN_CHOICES)	1✔
68	choices_keys = list(choices.keys())	1✔
69	choices_values = list(choices.values())	1✔
70	return choices_keys[choices_values.index(domain_name)]	1✔
71
72
73	def hash_subdomain(subdomain, domain=settings.MOZMAIL_DOMAIN):	1✔
74	return sha256(f"{subdomain}.{domain}".encode()).hexdigest()	1✔
75
76
77	class RegisteredSubdomain(models.Model):	1✔
78	subdomain_hash = models.CharField(max_length=64, db_index=True, unique=True)	1✔
79	registered_at = models.DateTimeField(auto_now_add=True)	1✔
80
81	def __str__(self):	1✔
UNCOV 82	return self.subdomain_hash	×
83
84
85	class CannotMakeAddressException(RelayAPIException):	1✔
86	"""Base exception for RelayAddress or DomainAddress creation failure."""
87
88
89	class AccountIsPausedException(CannotMakeAddressException):	1✔
90	default_code = "account_is_paused"	1✔
91	default_detail = "Your account is on pause."	1✔
92	status_code = 403	1✔
93
94
95	class AccountIsInactiveException(CannotMakeAddressException):	1✔
96	default_code = "account_is_inactive"	1✔
97	default_detail = "Your account is not active."	1✔
98	status_code = 403	1✔
99
100
101	class RelayAddrFreeTierLimitException(CannotMakeAddressException):	1✔
102	default_code = "free_tier_limit"	1✔
103	default_detail_template = (	1✔
104	"You’ve used all {free_tier_limit} email masks included with your free account."
105	" You can reuse an existing mask, but using a unique mask for each account is"
106	" the most secure option."
107	)
108	status_code = 403	1✔
109
110	def __init__(self, free_tier_limit: int \| None = None):	1✔
111	self.free_tier_limit = free_tier_limit or settings.MAX_NUM_FREE_ALIASES	1✔
112	super().__init__()	1✔
113
114	def error_context(self) -> ErrorContextType:	1✔
115	return {"free_tier_limit": self.free_tier_limit}	1✔
116
117
118	class DomainAddrFreeTierException(CannotMakeAddressException):	1✔
119	default_code = "free_tier_no_subdomain_masks"	1✔
120	default_detail = (	1✔
121	"Your free account does not include custom subdomains for masks."
122	" To create custom masks, upgrade to Relay Premium."
123	)
124	status_code = 403	1✔
125
126
127	class DomainAddrNeedSubdomainException(CannotMakeAddressException):	1✔
128	default_code = "need_subdomain"	1✔
129	default_detail = "Please select a subdomain before creating a custom email address."	1✔
130	status_code = 400	1✔
131
132
133	class DomainAddrUpdateException(CannotMakeAddressException):	1✔
134	"""Exception raised when attempting to edit an existing domain address field."""
135
136	default_code = "address_not_editable"	1✔
137	default_detail = "You cannot edit an existing domain address field."	1✔
138	status_code = 400	1✔
139
140
141	class DomainAddrUnavailableException(CannotMakeAddressException):	1✔
142	default_code = "address_unavailable"	1✔
143	default_detail_template = (	1✔
144	"“{unavailable_address}” could not be created."
145	" Please try again with a different mask name."
146	)
147	status_code = 400	1✔
148
149	def __init__(self, unavailable_address: str):	1✔
150	self.unavailable_address = unavailable_address	1✔
151	super().__init__()	1✔
152
153	def error_context(self) -> ErrorContextType:	1✔
154	return {"unavailable_address": self.unavailable_address}	1✔
155
156
157	class DomainAddrDuplicateException(CannotMakeAddressException):	1✔
158	default_code = "duplicate_address"	1✔
159	default_detail_template = (	1✔
160	"“{duplicate_address}” already exists."
161	" Please try again with a different mask name."
162	)
163	status_code = 409	1✔
164
165	def __init__(self, duplicate_address: str):	1✔
166	self.duplicate_address = duplicate_address	1✔
167	super().__init__()	1✔
168
169	def error_context(self) -> ErrorContextType:	1✔
170	return {"duplicate_address": self.duplicate_address}	1✔
171
172
173	class RelayAddress(models.Model):	1✔
174	user = models.ForeignKey(User, on_delete=models.CASCADE)	1✔
175	address = models.CharField(max_length=64, default=address_default, unique=True)	1✔
176	domain = models.PositiveSmallIntegerField(	1✔
177	choices=DOMAIN_CHOICES, default=default_domain_numerical
178	)
179	enabled = models.BooleanField(default=True)	1✔
180	description = models.CharField(max_length=64, blank=True)	1✔
181	created_at = models.DateTimeField(auto_now_add=True, db_index=True)	1✔
182	last_modified_at = models.DateTimeField(auto_now=True, db_index=True)	1✔
183	last_used_at = models.DateTimeField(blank=True, null=True)	1✔
184	num_forwarded = models.PositiveIntegerField(default=0)	1✔
185	num_blocked = models.PositiveIntegerField(default=0)	1✔
186	num_level_one_trackers_blocked = models.PositiveIntegerField(default=0, null=True)	1✔
187	num_replied = models.PositiveIntegerField(default=0)	1✔
188	num_spam = models.PositiveIntegerField(default=0)	1✔
189	generated_for = models.CharField(max_length=255, blank=True)	1✔
190	block_list_emails = models.BooleanField(default=False)	1✔
191	used_on = models.TextField(default=None, blank=True, null=True)	1✔
192
193	class Meta:	1✔
194	indexes = [	1✔
195	# Find when a user first used the add-on
196	models.Index(
197	name="idx_ra_created_by_addon",
198	fields=["user"],
199	condition=~models.Q(generated_for__exact=""),
200	include=["created_at"],
201	),
202	]
203
204	def __str__(self):	1✔
205	return self.address	1✔
206
207	def delete(self, args, *kwargs):	1✔
208	# TODO: create hard bounce receipt rule in AWS for the address
209	deleted_address = DeletedAddress.objects.create(	1✔
210	address_hash=address_hash(self.address, domain=self.domain_value),
211	num_forwarded=self.num_forwarded,
212	num_blocked=self.num_blocked,
213	num_replied=self.num_replied,
214	num_spam=self.num_spam,
215	)
216	deleted_address.save()	1✔
217	profile = self.user.profile	1✔
218	profile.refresh_from_db()	1✔
219	profile.address_last_deleted = datetime.now(UTC)	1✔
220	profile.num_address_deleted += 1	1✔
221	profile.num_email_forwarded_in_deleted_address += self.num_forwarded	1✔
222	profile.num_email_blocked_in_deleted_address += self.num_blocked	1✔
223	profile.num_level_one_trackers_blocked_in_deleted_address = (	1✔
224	profile.num_level_one_trackers_blocked_in_deleted_address or 0
225	) + (self.num_level_one_trackers_blocked or 0)
226	profile.num_email_replied_in_deleted_address += self.num_replied	1✔
227	profile.num_email_spam_in_deleted_address += self.num_spam	1✔
228	profile.num_deleted_relay_addresses += 1	1✔
229	profile.last_engagement = datetime.now(UTC)	1✔
230	profile.save()	1✔
231	return super().delete(args, *kwargs)	1✔
232
233	def save(	1✔
234	self,
235	force_insert: bool \| tuple[ModelBase, ...] = False,
236	force_update: bool = False,
237	using: str \| None = None,
238	update_fields: Iterable[str] \| None = None,
239	) -> None:
240	from privaterelay.models import Profile	1✔
241
242	if self._state.adding:	1✔
243	with transaction.atomic():	1✔
244	locked_profile = Profile.objects.select_for_update().get(user=self.user)	1✔
245	check_user_can_make_another_address(locked_profile.user)	1✔
246	while True:	1✔
247	address_is_allowed = not is_blocklisted(self.address)	1✔
248	address_is_valid = valid_address(self.address, self.domain_value)	1✔
249	if address_is_valid and address_is_allowed:	1✔
250	break	1✔
251	self.address = address_default()	1✔
252	locked_profile.update_abuse_metric(address_created=True)	1✔
253	locked_profile.last_engagement = datetime.now(UTC)	1✔
254	locked_profile.save()	1✔
255	if (not self.user.profile.server_storage) and any(	1✔
256	(self.description, self.generated_for, self.used_on)
257	):
258	self.description = ""	1✔
259	self.generated_for = ""	1✔
260	self.used_on = ""	1✔
261	if update_fields is not None:	1✔
262	update_fields = {"description", "generated_for", "used_on"}.union(	1✔
263	update_fields
264	)
265	if not self.user.profile.has_premium and self.block_list_emails:	1✔
266	self.block_list_emails = False	1✔
267	if update_fields is not None:	1✔
268	update_fields = {"block_list_emails"}.union(update_fields)	1✔
269	super().save(	1✔
270	force_insert=force_insert,
271	force_update=force_update,
272	using=using,
273	update_fields=update_fields,
274	)
275
276	@property	1✔
277	def domain_value(self) -> str:	1✔
278	domain = cast(	1✔
279	Literal["RELAY_FIREFOX_DOMAIN", "MOZMAIL_DOMAIN"], self.get_domain_display()
280	)
281	return get_domains_from_settings()[domain]	1✔
282
283	@property	1✔
284	def full_address(self) -> str:	1✔
285	return f"{self.address}@{self.domain_value}"	1✔
286
287	@property	1✔
288	def metrics_id(self) -> str:	1✔
289	if not self.id:	1!
UNCOV 290	raise ValueError("self.id must be truthy value.")	×
291	# Prefix with 'R' for RelayAddress, since there may be a DomainAddress with the
292	# same row ID
293	return f"R{self.id}"	1✔
294
295
296	def check_user_can_make_another_address(user: User) -> None:	1✔
297	if not user.is_active:	1✔
298	raise AccountIsInactiveException()	1✔
299
300	if user.profile.is_flagged:	1✔
301	raise AccountIsPausedException()	1✔
302	# MPP-3021: return early for premium users to avoid at_max_free_aliases DB query
303	if user.profile.has_premium:	1✔
304	return	1✔
305	if user.profile.at_max_free_aliases:	1✔
306	raise RelayAddrFreeTierLimitException()	1✔
307
308
309	def valid_address_pattern(address):	1✔
310	# can't start or end with a hyphen
311	# must be 1-63 lowercase alphanumeric characters and/or hyphens
312	valid_address_pattern = re.compile("^(?![-.])[a-z0-9-.]{1,63}(?<![-.])$")	1✔
313	return valid_address_pattern.match(address) is not None	1✔
314
315
316	def valid_address(address: str, domain: str, subdomain: str \| None = None) -> bool:	1✔
317	address_pattern_valid = valid_address_pattern(address)	1✔
318	address_contains_badword = has_bad_words(address)	1✔
319	address_already_deleted = 0	1✔
320	if not subdomain or flag_is_active_in_task(	1✔
321	"custom_domain_management_redesign", None
322	):
323	address_already_deleted = DeletedAddress.objects.filter(	1✔
324	address_hash=address_hash(address, domain=domain, subdomain=subdomain)
325	).count()
326	if (	1✔
327	address_already_deleted > 0
328	or address_contains_badword
329	or not address_pattern_valid
330	):
331	return False	1✔
332	return True	1✔
333
334
335	class DeletedAddress(models.Model):	1✔
336	address_hash = models.CharField(max_length=64, db_index=True)	1✔
337	num_forwarded = models.PositiveIntegerField(default=0)	1✔
338	num_blocked = models.PositiveIntegerField(default=0)	1✔
339	num_replied = models.PositiveIntegerField(default=0)	1✔
340	num_spam = models.PositiveIntegerField(default=0)	1✔
341
342	def __str__(self):	1✔
UNCOV 343	return self.address_hash	×
344
345
346	def check_user_can_make_domain_address(user: User) -> None:	1✔
347	if not user.profile.has_premium:	1✔
348	raise DomainAddrFreeTierException()	1✔
349
350	if not user.profile.subdomain:	1✔
351	raise DomainAddrNeedSubdomainException()	1✔
352
353	if not user.profile.user.is_active:	1!
UNCOV 354	raise AccountIsInactiveException()	×
355
356	if user.profile.is_flagged:	1✔
357	raise AccountIsPausedException()	1✔
358
359
360	class DomainAddress(models.Model):	1✔
361	user = models.ForeignKey(User, on_delete=models.CASCADE)	1✔
362	address = models.CharField(	1✔
363	max_length=64, validators=[MinLengthValidator(limit_value=1)]
364	)
365	enabled = models.BooleanField(default=True)	1✔
366	description = models.CharField(max_length=64, blank=True)	1✔
367	domain = models.PositiveSmallIntegerField(choices=DOMAIN_CHOICES, default=2)	1✔
368	created_at = models.DateTimeField(auto_now_add=True, db_index=True)	1✔
369	first_emailed_at = models.DateTimeField(null=True, db_index=True)	1✔
370	last_modified_at = models.DateTimeField(auto_now=True, db_index=True)	1✔
371	last_used_at = models.DateTimeField(blank=True, null=True)	1✔
372	num_forwarded = models.PositiveIntegerField(default=0)	1✔
373	num_blocked = models.PositiveIntegerField(default=0)	1✔
374	num_level_one_trackers_blocked = models.PositiveIntegerField(default=0, null=True)	1✔
375	num_replied = models.PositiveIntegerField(default=0)	1✔
376	num_spam = models.PositiveIntegerField(default=0)	1✔
377	block_list_emails = models.BooleanField(default=False)	1✔
378	used_on = models.TextField(default=None, blank=True, null=True)	1✔
379
380	class Meta:	1✔
381	unique_together = ["user", "address"]	1✔
382
383	def __str__(self):	1✔
UNCOV 384	return self.address	×
385
386	def save(	1✔
387	self,
388	force_insert: bool \| tuple[ModelBase, ...] = False,
389	force_update: bool = False,
390	using: str \| None = None,
391	update_fields: Iterable[str] \| None = None,
392	) -> None:
393	user_profile = self.user.profile	1✔
394	if self._state.adding:	1✔
395	check_user_can_make_domain_address(self.user)	1✔
396	domain_address_valid = valid_address(	1✔
397	self.address, self.domain_value, user_profile.subdomain
398	)
399	if not domain_address_valid:	1✔
400	if self.first_emailed_at:	1!
UNCOV 401	incr_if_enabled("domainaddress.create_via_email_fail")	×
402	raise DomainAddrUnavailableException(unavailable_address=self.address)	1✔
403
404	if DomainAddress.objects.filter(	1✔
405	user=self.user, address=self.address
406	).exists():
407	raise DomainAddrDuplicateException(duplicate_address=self.address)	1✔
408
409	user_profile.update_abuse_metric(address_created=True)	1✔
410	user_profile.last_engagement = datetime.now(UTC)	1✔
411	user_profile.save(update_fields=["last_engagement"])	1✔
412	incr_if_enabled("domainaddress.create")	1✔
413	if self.first_emailed_at:	1✔
414	incr_if_enabled("domainaddress.create_via_email")	1✔
415	else:
416	# The model is in an update state, do not allow 'address' field updates
417	existing_instance = DomainAddress.objects.get(id=self.id)	1✔
418	if existing_instance.address != self.address:	1✔
419	raise DomainAddrUpdateException()	1✔
420
421	if not user_profile.has_premium and self.block_list_emails:	1✔
422	self.block_list_emails = False	1✔
423	if update_fields:	1✔
424	update_fields = {"block_list_emails"}.union(update_fields)	1✔
425	if (not user_profile.server_storage) and (self.description or self.used_on):	1✔
426	self.description = ""	1✔
427	self.used_on = ""	1✔
428	if update_fields:	1✔
429	update_fields = {"description", "used_on"}.union(update_fields)	1✔
430	super().save(	1✔
431	force_insert=force_insert,
432	force_update=force_update,
433	using=using,
434	update_fields=update_fields,
435	)
436
437	@staticmethod	1✔
438	def make_domain_address(	1✔
439	user: User, address: str \| None = None, made_via_email: bool = False
440	) -> DomainAddress:
441	check_user_can_make_domain_address(user)	1✔
442
443	if not address:	1✔
444	# FIXME: if the alias is randomly generated and has bad words
445	# we should retry like make_relay_address does
446	# not fixing this now because not sure randomly generated
447	# DomainAlias will be a feature
448	address = address_default()	1✔
449	# Only check for bad words if randomly generated
450	if not isinstance(address, str):	1!
UNCOV 451	raise TypeError("address must be type str")	×
452
453	first_emailed_at = datetime.now(UTC) if made_via_email else None	1✔
454	domain_address = DomainAddress.objects.create(	1✔
455	user=user, address=address, first_emailed_at=first_emailed_at
456	)
457	return domain_address	1✔
458
459	def delete(self, args, *kwargs):	1✔
460	# TODO: create hard bounce receipt rule in AWS for the address
461	deleted_address = DeletedAddress.objects.create(	1✔
462	address_hash=address_hash(
463	self.address, self.user.profile.subdomain, self.domain_value
464	),
465	num_forwarded=self.num_forwarded,
466	num_blocked=self.num_blocked,
467	num_replied=self.num_replied,
468	num_spam=self.num_spam,
469	)
470	deleted_address.save()	1✔
471	profile = self.user.profile	1✔
472	profile.address_last_deleted = datetime.now(UTC)	1✔
473	profile.num_address_deleted += 1	1✔
474	profile.num_email_forwarded_in_deleted_address += self.num_forwarded	1✔
475	profile.num_email_blocked_in_deleted_address += self.num_blocked	1✔
476	profile.num_level_one_trackers_blocked_in_deleted_address = (	1✔
477	profile.num_level_one_trackers_blocked_in_deleted_address or 0
478	) + (self.num_level_one_trackers_blocked or 0)
479	profile.num_email_replied_in_deleted_address += self.num_replied	1✔
480	profile.num_email_spam_in_deleted_address += self.num_spam	1✔
481	profile.num_deleted_domain_addresses += 1	1✔
482	profile.last_engagement = datetime.now(UTC)	1✔
483	profile.save()	1✔
484	return super().delete(args, *kwargs)	1✔
485
486	@property	1✔
487	def domain_value(self) -> str:	1✔
488	domain = cast(	1✔
489	Literal["RELAY_FIREFOX_DOMAIN", "MOZMAIL_DOMAIN"], self.get_domain_display()
490	)
491	return get_domains_from_settings()[domain]	1✔
492
493	@property	1✔
494	def full_address(self) -> str:	1✔
495	return f"{self.address}@{self.user.profile.subdomain}.{self.domain_value}"	1✔
496
497	@property	1✔
498	def metrics_id(self) -> str:	1✔
499	if not self.id:	1!
UNCOV 500	raise ValueError("self.id must be truthy value.")	×
501	# Prefix with 'D' for DomainAddress, since there may be a RelayAddress with the
502	# same row ID
503	return f"D{self.id}"	1✔
504
505
506	class Reply(models.Model):	1✔
507	relay_address = models.ForeignKey(	1✔
508	RelayAddress, on_delete=models.CASCADE, blank=True, null=True
509	)
510	domain_address = models.ForeignKey(	1✔
511	DomainAddress, on_delete=models.CASCADE, blank=True, null=True
512	)
513	lookup = models.CharField(max_length=255, blank=False, db_index=True)	1✔
514	encrypted_metadata = models.TextField(blank=False)	1✔
515	created_at = models.DateField(auto_now_add=True, null=False, db_index=True)	1✔
516
517	@property	1✔
518	def address(self):	1✔
519	return self.relay_address or self.domain_address	1✔
520
521	@property	1✔
522	def profile(self):	1✔
523	return self.address.user.profile	1✔
524
525	@property	1✔
526	def owner_has_premium(self):	1✔
527	return self.profile.has_premium	1✔
528
529	def increment_num_replied(self):	1✔
530	address = self.relay_address or self.domain_address	1✔
531	if not address:	1!
UNCOV 532	raise ValueError("address must be truthy value")	×
533	address.num_replied += 1	1✔
534	address.last_used_at = datetime.now(UTC)	1✔
535	address.save(update_fields=["num_replied", "last_used_at"])	1✔
536	return address.num_replied	1✔

mozilla / fx-private-relay / 76b0d84c-52e7-4179-8157-f3582acddb0b

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