fix MPP-3802: stop ignoring bandit security checks

api/authentication.py

@@ -29,7 +29,11 @@ def get_cache_key(token):
 
 def introspect_token(token: str) -> dict[str, Any]:
     try:
-        fxa_resp = requests.post(INTROSPECT_TOKEN_URL, json={"token": token})
+        fxa_resp = requests.post(
+            INTROSPECT_TOKEN_URL,
+            json={"token": token},
+            timeout=settings.FXA_REQUESTS_TIMEOUT_SECONDS,
+        )
     except Exception as exc:
         logger.error(
             "Could not introspect token with FXA.",

api/exceptions.py

@@ -58,13 +58,17 @@ def __init__(
         self, detail: _APIExceptionInput = None, code: str | None = None
     ) -> None:
         """Check that derived classes have set the required data."""
-        assert isinstance(self.default_code, str)
-        assert isinstance(self.status_code, int)
+        if not isinstance(self.default_code, str):
+            raise TypeError("default_code must be type str")
+        if not isinstance(self.status_code, int):
+            raise TypeError("self.status_code must be type int")
         if hasattr(self, "default_detail_template"):
             context = self.error_context()
-            assert context
+            if not context:
+                raise ValueError("error_context is required")
             self.default_detail = self.default_detail_template.format(**context)
-        assert isinstance(self.default_detail, str)
+        if not isinstance(self.default_detail, str):
+            raise TypeError("self.default_detail must be type str")
         super().__init__(detail, code)
 
     def error_context(self) -> ErrorContextType:
@@ -76,7 +80,8 @@ def error_data(self) -> ErrorData:
 
         # For RelayAPIException classes, this is the default_code and is a string
         error_code = self.get_codes()
-        assert isinstance(error_code, str)
+        if not isinstance(error_code, str):
+            raise TypeError("error_code must be type str")
 
         # Build the Fluent error ID
         ftl_id_sub = "api-error-"

api/serializers/emails.py

@@ -12,7 +12,8 @@ class PremiumValidatorsMixin:
     def validate_block_list_emails(self, value):
         if not value:
             return value
-        assert hasattr(self, "context")
+        if not hasattr(self, "context"):
+            raise AttributeError("self must have attribute context")
         user = self.context["request"].user
         prefetch_related_objects([user], "socialaccount_set", "profile")
         if not user.profile.has_premium:

api/views/__init__.py

@@ -12,8 +12,10 @@ class SaveToRequestUser:
     """ModelViewSet mixin for creating object for the authenticated user."""
 
     def perform_create(self, serializer):
-        assert hasattr(self, "request")
-        assert hasattr(self.request, "user")
+        if not hasattr(self, "request"):
+            raise AttributeError("self must have request attribute.")
+        if not hasattr(self.request, "user"):
+            raise AttributeError("self.request must have user attribute.")
         serializer.save(user=self.request.user)
 
 

api/views/emails.py

@@ -74,15 +74,17 @@ class Meta:
 class AddressViewSet(Generic[_Address], SaveToRequestUser, ModelViewSet):
     def perform_create(self, serializer: BaseSerializer[_Address]) -> None:
         super().perform_create(serializer)
-        assert serializer.instance
+        if not serializer.instance:
+            raise ValueError("serializer.instance must be truthy value.")
         glean_logger().log_email_mask_created(
             request=self.request,
             mask=serializer.instance,
             created_by_api=True,
         )
 
     def perform_update(self, serializer: BaseSerializer[_Address]) -> None:
-        assert serializer.instance is not None
+        if not serializer.instance:
+            raise ValueError("serializer.instance must be truthy value.")
         old_description = serializer.instance.description
         super().perform_update(serializer)
         new_description = serializer.instance.description
@@ -200,10 +202,13 @@ def first_forwarded_email(request):
         return Response(f"{mask} does not exist for user.", status=HTTP_404_NOT_FOUND)
     profile = user.profile
     app_config = apps.get_app_config("emails")
-    assert isinstance(app_config, EmailsConfig)
+    if not isinstance(app_config, EmailsConfig):
+        raise TypeError("app_config must be type EmailsConfig")
     ses_client = app_config.ses_client
-    assert ses_client
-    assert settings.RELAY_FROM_ADDRESS
+    if not ses_client:
+        raise ValueError("ses_client must be truthy value.")
+    if not settings.RELAY_FROM_ADDRESS:
+        raise ValueError("settings.RELAY_FROM_ADDRESS must have a value.")
     with django_ftl.override(profile.language):
         translated_subject = ftl_bundle.format("forwarded-email-hero-header")
     first_forwarded_email_html = render_to_string(

api/views/phones.py

@@ -1316,17 +1316,24 @@ class RelaySMSException(Exception):
 
     def __init__(self, critical=False, *args, **kwargs):
         self.critical = critical
-        assert (
+        if not (
             self.default_detail is not None and self.default_detail_template is None
-        ) or (self.default_detail is None and self.default_detail_template is not None)
+        ) and not (
+            self.default_detail is None and self.default_detail_template is not None
+        ):
+            raise ValueError(
+                "One and only one of default_detail or "
+                "default_detail_template must be None."
+            )
         super().__init__(*args, **kwargs)
 
     @property
     def detail(self):
         if self.default_detail:
             return self.default_detail
         else:
-            assert self.default_detail_template is not None
+            if self.default_detail_template is None:
+                raise ValueError("self.default_detail_template must not be None.")
             return self.default_detail_template.format(**self.error_context())
 
     def get_codes(self):
@@ -1438,14 +1445,16 @@ def _prepare_sms_reply(
     if match and not match.contacts and match.match_type == "full":
         raise FullNumberMatchesNoSenders(full_number=match.detected)
     if match and len(match.contacts) > 1:
-        assert match.match_type == "short"
+        if not match.match_type == "short":
+            raise ValueError("match.match_type must be 'short'.")
         raise MultipleNumberMatches(short_prefix=match.detected)
 
     # Determine the destination number
     destination_number: str | None = None
     if match:
         # Use the sender matched by the prefix
-        assert len(match.contacts) == 1
+        if not len(match.contacts) == 1:
+            raise ValueError("len(match.contacts) must be 1.")
         destination_number = match.contacts[0].inbound_number
     else:
         # No prefix, default to last sender if any

api/views/privaterelay.py

@@ -330,7 +330,9 @@ def terms_accepted_user(request):
     except SocialAccount.DoesNotExist:
         # User does not exist, create a new Relay user
         fxa_profile_resp = requests.get(
-            FXA_PROFILE_URL, headers={"Authorization": f"Bearer {token}"}
+            FXA_PROFILE_URL,
+            headers={"Authorization": f"Bearer {token}"},
+            timeout=settings.FXA_REQUESTS_TIMEOUT_SECONDS,
         )
         if not (fxa_profile_resp.ok and fxa_profile_resp.content):
             logger.error(

emails/apps.py

@@ -65,7 +65,8 @@ def ready(self):
 
 def emails_config() -> EmailsConfig:
     emails_config = apps.get_app_config("emails")
-    assert isinstance(emails_config, EmailsConfig)
+    if not isinstance(emails_config, EmailsConfig):
+        raise TypeError("emails_config must be type EmailsConfig")
     return emails_config
 
 

emails/management/command_from_django_settings.py

@@ -37,7 +37,8 @@ def create_parser(self, prog_name, subcommand, **kwargs):
         * Add the Django settings and their values to the command help
         * Override the verbosity from an environment variable
         """
-        assert self.settings_to_locals
+        if not self.settings_to_locals:
+            raise ValueError("self.settings_to_locals must be truthy value.")
         epilog_lines = [
             (
                 "Parameters are read from Django settings and the related environment"
@@ -65,15 +66,18 @@ def create_parser(self, prog_name, subcommand, **kwargs):
         epilog = "\n".join(epilog_lines)
 
         parser = super().create_parser(prog_name, subcommand, epilog=epilog, **kwargs)
-        assert parser.formatter_class == DjangoHelpFormatter
+        if parser.formatter_class != DjangoHelpFormatter:
+            raise TypeError("parser.formatter_class must be DjangoHelpFormatter")
         parser.formatter_class = RawDescriptionDjangoHelpFormatter
-        assert verbosity_override is not None
+        if verbosity_override is None:
+            raise ValueError("verbosity_override must not be None.")
         parser.set_defaults(verbosity=verbosity_override)
         return parser
 
     def init_from_settings(self, verbosity):
         """Initialize local variables from settings"""
-        assert self.settings_to_locals
+        if not self.settings_to_locals:
+            raise ValueError("self.settings_to_locals must be truthy value.")
         for setting_key, local_name, help_str, validator in self.settings_to_locals:
             value = getattr(settings, setting_key)
             if not validator(value):

emails/management/commands/process_emails_from_sqs.py

@@ -152,8 +152,10 @@ def init_locals(self):
 
     def create_client(self):
         """Create the SQS client."""
-        assert self.aws_region
-        assert self.sqs_url
+        if not self.aws_region:
+            raise ValueError("self.aws_region must be truthy value.")
+        if not self.sqs_url:
+            raise ValueError("self.sqs_url must be truthy value.")
         sqs_client = boto3.resource("sqs", region_name=self.aws_region)
         return sqs_client.Queue(self.sqs_url)
 

emails/management/commands/send_welcome_emails.py

@@ -40,10 +40,13 @@ def _ses_message_props(data: str) -> ContentTypeDef:
 def send_welcome_email(profile: Profile) -> None:
     user = profile.user
     app_config = apps.get_app_config("emails")
-    assert isinstance(app_config, EmailsConfig)
+    if not isinstance(app_config, EmailsConfig):
+        raise TypeError("app_config must be type EmailsConfig")
     ses_client = app_config.ses_client
-    assert ses_client
-    assert settings.RELAY_FROM_ADDRESS
+    if not ses_client:
+        raise ValueError("ses_client must be truthy value")
+    if not settings.RELAY_FROM_ADDRESS:
+        raise ValueError("settings.RELAY_FROM_ADDRESS must be truthy value.")
     with django_ftl.override(profile.language):
         translated_subject = ftl_bundle.format("first-time-user-email-welcome")
     try:

emails/models.py

@@ -266,13 +266,16 @@ def next_email_try(self) -> datetime:
             return datetime.now(UTC)
 
         if bounce_type == "soft":
-            assert self.last_soft_bounce
+            if not self.last_soft_bounce:
+                raise ValueError("self.last_soft_bounce must be truthy value.")
             return self.last_soft_bounce + timedelta(
                 days=settings.SOFT_BOUNCE_ALLOWED_DAYS
             )
 
-        assert bounce_type == "hard"
-        assert self.last_hard_bounce
+        if bounce_type != "hard":
+            raise ValueError("bounce_type must be either 'soft' or 'hard'")
+        if not self.last_hard_bounce:
+            raise ValueError("self.last_hard_bounce must be truthy value.")
         return self.last_hard_bounce + timedelta(days=settings.HARD_BOUNCE_ALLOWED_DAYS)
 
     @property
@@ -293,7 +296,8 @@ def fxa(self) -> SocialAccount | None:
         # Note: we are NOT using .filter() here because it invalidates
         # any profile instances that were queried with prefetch_related, which
         # we use in at least the profile view to minimize queries
-        assert hasattr(self.user, "socialaccount_set")
+        if not hasattr(self.user, "socialaccount_set"):
+            raise AttributeError("self.user must have socialaccount_set attribute")
         for sa in self.user.socialaccount_set.all():
             if sa.provider == "fxa":
                 return sa
@@ -310,7 +314,8 @@ def display_name(self) -> str | None:
 
     @property
     def custom_domain(self) -> str:
-        assert self.subdomain
+        if not self.subdomain:
+            raise ValueError("self.subdomain must be truthy value.")
         return f"@{self.subdomain}.{settings.MOZMAIL_DOMAIN}"
 
     @property
@@ -844,7 +849,8 @@ def full_address(self) -> str:
 
     @property
     def metrics_id(self) -> str:
-        assert self.id
+        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}"
@@ -1002,7 +1008,8 @@ def make_domain_address(
             # DomainAlias will be a feature
             address = address_default()
             # Only check for bad words if randomly generated
-        assert isinstance(address, str)
+        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(
@@ -1052,7 +1059,8 @@ def full_address(self) -> str:
 
     @property
     def metrics_id(self) -> str:
-        assert self.id
+        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}"
@@ -1083,7 +1091,8 @@ def owner_has_premium(self):
 
     def increment_num_replied(self):
         address = self.relay_address or self.domain_address
-        assert 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"])

emails/utils.py

@@ -85,7 +85,7 @@ def get_trackers(level):
 
 def download_trackers(repo_url, category="Email"):
     # email tracker lists from shavar-prod-list as per agreed use under license:
-    resp = requests.get(repo_url)
+    resp = requests.get(repo_url, timeout=10)
     json_resp = resp.json()
     formatted_trackers = json_resp["categories"][category]
     trackers = []
@@ -198,7 +198,8 @@ def _get_hero_img_src(lang_code):
     if major_lang in avail_l10n_image_codes:
         img_locale = major_lang
 
-    assert settings.SITE_ORIGIN
+    if not settings.SITE_ORIGIN:
+        raise ValueError("settings.SITE_ORIGIN must have a value")
     return (
         settings.SITE_ORIGIN
         + f"/static/images/email-images/first-time-user/hero-image-{img_locale}.png"
@@ -229,8 +230,11 @@ def ses_send_raw_email(
     destination_address: str,
     message: EmailMessage,
 ) -> SendRawEmailResponseTypeDef:
-    assert (client := ses_client()) is not None
-    assert settings.AWS_SES_CONFIGSET
+    client = ses_client()
+    if client is None:
+        raise ValueError("client must have a value")
+    if not settings.AWS_SES_CONFIGSET:
+        raise ValueError("settings.AWS_SES_CONFIGSET must have a value")
 
     data = message.as_string()
     try:
@@ -393,7 +397,9 @@ def _get_bucket_and_key_from_s3_json(message_json):
 @time_if_enabled("s3_get_message_content")
 def get_message_content_from_s3(bucket, object_key):
     if bucket and object_key:
-        assert (client := s3_client()) is not None
+        client = s3_client()
+        if client is None:
+            raise ValueError("client must not be None")
         streamed_s3_object = client.get_object(Bucket=bucket, Key=object_key).get(
             "Body"
         )
@@ -405,7 +411,9 @@ def remove_message_from_s3(bucket, object_key):
     if bucket is None or object_key is None:
         return False
     try:
-        assert (client := s3_client()) is not None
+        client = s3_client()
+        if client is None:
+            raise ValueError("client must not be None")
         response = client.delete_object(Bucket=bucket, Key=object_key)
         return response.get("DeleteMarker")
     except ClientError as e: