"""Relay interface to EventsServerEventLogger generated by glean_parser.""" from __future__ import annotations from datetime import datetime from logging import getLogger from typing import Any, Literal, NamedTuple from django.conf import settings from django.contrib.auth.models import User from django.http import HttpRequest from ipware import get_client_ip from emails.models import DomainAddress, RelayAddress from .glean.server_events import GLEAN_EVENT_MOZLOG_TYPE, EventsServerEventLogger from .types import RELAY_CHANNEL_NAME # Enumerate the mask setting that caused an email to not be forwarded. EmailBlockedReason = Literal[ "block_all", # The mask is set to block all mail "block_promotional", # The mask is set to block promotional / list mail ] def _opt_dt_to_glean(value: datetime | None) -> int: """Convert an optional datetime to an integer timestamp.""" if value == datetime.min: return -2 # datetime was not checked if value is None: return -1 # datetime does not exist return int(value.timestamp()) def _opt_str_to_glean(value: str | None) -> str: """Convert an optional string to a (possibly empty) string.""" return "" if value is None else value class RequestData(NamedTuple): """Extract and store data from the request.""" user_agent: str | None = None ip_address: str | None = None @classmethod def from_request(cls, request: HttpRequest) -> RequestData: user_agent = request.headers.get("user-agent", None) client_ip, is_routable = get_client_ip(request) ip_address = client_ip if (client_ip and is_routable) else None return cls(user_agent=user_agent, ip_address=ip_address) class UserData(NamedTuple): """Extract and store data from a Relay user.""" metrics_enabled: bool fxa_id: str | None = None n_random_masks: int = 0 n_domain_masks: int = 0 n_deleted_random_masks: int = 0 n_deleted_domain_masks: int = 0 date_joined_relay: datetime | None = None date_joined_premium: datetime | None = None premium_status: str = "" has_extension: bool = False date_got_extension: datetime | None = None @classmethod def from_user(cls, user: User) -> UserData: metrics_enabled = user.profile.metrics_enabled if not metrics_enabled: return cls(metrics_enabled=False) fxa_id = user.profile.metrics_fxa_id or None n_random_masks = user.relayaddress_set.count() n_domain_masks = user.domainaddress_set.count() n_deleted_random_masks = user.profile.num_deleted_relay_addresses n_deleted_domain_masks = user.profile.num_deleted_domain_addresses date_joined_relay = user.date_joined if user.profile.has_premium: if user.profile.has_phone: date_joined_premium = user.profile.date_subscribed_phone else: date_joined_premium = user.profile.date_subscribed else: date_joined_premium = None premium_status = user.profile.metrics_premium_status # Until more accurate date_got_extension is calculated (MPP-3765) # do not check for when the user got extension has_extension = False date_got_extension = datetime.min return cls( metrics_enabled=True, fxa_id=fxa_id, n_random_masks=n_random_masks, n_domain_masks=n_domain_masks, n_deleted_random_masks=n_deleted_random_masks, n_deleted_domain_masks=n_deleted_domain_masks, date_joined_relay=date_joined_relay, date_joined_premium=date_joined_premium, premium_status=premium_status, has_extension=has_extension, date_got_extension=date_got_extension, ) class EmailMaskData(NamedTuple): """Extract and store data from a Relay email mask.""" is_random_mask: bool has_website: bool @classmethod def from_mask(cls, mask: RelayAddress | DomainAddress) -> EmailMaskData: if isinstance(mask, RelayAddress): is_random_mask = True has_website = bool(mask.generated_for) else: is_random_mask = False has_website = False return EmailMaskData(is_random_mask=is_random_mask, has_website=has_website) class RelayGleanLogger(EventsServerEventLogger): """Extend the generated EventsServerEventLogger for Relay usage.""" def __init__( self, application_id: str, app_display_version: str, channel: RELAY_CHANNEL_NAME, ): if not settings.GLEAN_EVENT_MOZLOG_TYPE == GLEAN_EVENT_MOZLOG_TYPE: raise ValueError( "settings.GLEAN_EVENT_MOZLOG_TYPE must equal GLEAN_EVENT_MOZLOG_TYPE" ) self._logger = getLogger(GLEAN_EVENT_MOZLOG_TYPE) super().__init__( application_id=application_id, app_display_version=app_display_version, channel=channel, ) def emit_record(self, now: datetime, ping: dict[str, Any]) -> None: """Emit record as a log instead of a print()""" self._logger.info(GLEAN_EVENT_MOZLOG_TYPE, extra=ping) def log_email_mask_created( self, *, request: HttpRequest | None = None, mask: RelayAddress | DomainAddress, created_by_api: bool, ) -> None: """Log that a Relay email mask was created.""" user_data = UserData.from_user(mask.user) if not user_data.metrics_enabled: return request_data = RequestData.from_request(request) if request else RequestData() mask_data = EmailMaskData.from_mask(mask) self.record_email_mask_created( user_agent=_opt_str_to_glean(request_data.user_agent), ip_address=_opt_str_to_glean(request_data.ip_address), fxa_id=_opt_str_to_glean(user_data.fxa_id), platform="", n_random_masks=user_data.n_random_masks, n_domain_masks=user_data.n_domain_masks, n_deleted_random_masks=user_data.n_deleted_random_masks, n_deleted_domain_masks=user_data.n_deleted_domain_masks, date_joined_relay=_opt_dt_to_glean(user_data.date_joined_relay), premium_status=user_data.premium_status, date_joined_premium=_opt_dt_to_glean(user_data.date_joined_premium), has_extension=user_data.has_extension, date_got_extension=_opt_dt_to_glean(user_data.date_got_extension), is_random_mask=mask_data.is_random_mask, created_by_api=created_by_api, has_website=mask_data.has_website, ) def log_email_mask_label_updated( self, *, request: HttpRequest, mask: RelayAddress | DomainAddress, ) -> None: """Log that a Relay email mask's label was changed.""" user_data = UserData.from_user(mask.user) if not user_data.metrics_enabled: return request_data = RequestData.from_request(request) mask_data = EmailMaskData.from_mask(mask) self.record_email_mask_label_updated( user_agent=_opt_str_to_glean(request_data.user_agent), ip_address=_opt_str_to_glean(request_data.ip_address), fxa_id=_opt_str_to_glean(user_data.fxa_id), platform="", n_random_masks=user_data.n_random_masks, n_domain_masks=user_data.n_domain_masks, n_deleted_random_masks=user_data.n_deleted_random_masks, n_deleted_domain_masks=user_data.n_deleted_domain_masks, date_joined_relay=_opt_dt_to_glean(user_data.date_joined_relay), premium_status=user_data.premium_status, date_joined_premium=_opt_dt_to_glean(user_data.date_joined_premium), has_extension=user_data.has_extension, date_got_extension=_opt_dt_to_glean(user_data.date_got_extension), is_random_mask=mask_data.is_random_mask, ) def log_email_mask_deleted( self, *, request: HttpRequest, user: User, is_random_mask: bool, ) -> None: """Log that a Relay email mask was deleted.""" user_data = UserData.from_user(user) if not user_data.metrics_enabled: return request_data = RequestData.from_request(request) self.record_email_mask_deleted( user_agent=_opt_str_to_glean(request_data.user_agent), ip_address=_opt_str_to_glean(request_data.ip_address), fxa_id=_opt_str_to_glean(user_data.fxa_id), platform="", n_random_masks=user_data.n_random_masks, n_domain_masks=user_data.n_domain_masks, n_deleted_random_masks=user_data.n_deleted_random_masks, n_deleted_domain_masks=user_data.n_deleted_domain_masks, date_joined_relay=_opt_dt_to_glean(user_data.date_joined_relay), premium_status=user_data.premium_status, date_joined_premium=_opt_dt_to_glean(user_data.date_joined_premium), has_extension=user_data.has_extension, date_got_extension=_opt_dt_to_glean(user_data.date_got_extension), is_random_mask=is_random_mask, ) def log_email_forwarded( self, *, mask: RelayAddress | DomainAddress, is_reply: bool = False, ) -> None: """Log that an email was forwarded.""" user_data = UserData.from_user(mask.user) if not user_data.metrics_enabled: return request_data = RequestData() mask_data = EmailMaskData.from_mask(mask) self.record_email_forwarded( user_agent=_opt_str_to_glean(request_data.user_agent), ip_address=_opt_str_to_glean(request_data.ip_address), fxa_id=_opt_str_to_glean(user_data.fxa_id), platform="", n_random_masks=user_data.n_random_masks, n_domain_masks=user_data.n_domain_masks, n_deleted_random_masks=user_data.n_deleted_random_masks, n_deleted_domain_masks=user_data.n_deleted_domain_masks, date_joined_relay=_opt_dt_to_glean(user_data.date_joined_relay), premium_status=user_data.premium_status, date_joined_premium=_opt_dt_to_glean(user_data.date_joined_premium), has_extension=user_data.has_extension, date_got_extension=_opt_dt_to_glean(user_data.date_got_extension), is_random_mask=mask_data.is_random_mask, is_reply=is_reply, ) def log_email_blocked( self, *, mask: RelayAddress | DomainAddress, reason: EmailBlockedReason, is_reply: bool = False, ) -> None: """Log that an email was not forwarded.""" user_data = UserData.from_user(mask.user) if not user_data.metrics_enabled: return request_data = RequestData() mask_data = EmailMaskData.from_mask(mask) self.record_email_blocked( user_agent=_opt_str_to_glean(request_data.user_agent), ip_address=_opt_str_to_glean(request_data.ip_address), fxa_id=_opt_str_to_glean(user_data.fxa_id), platform="", n_random_masks=user_data.n_random_masks, n_domain_masks=user_data.n_domain_masks, n_deleted_random_masks=user_data.n_deleted_random_masks, n_deleted_domain_masks=user_data.n_deleted_domain_masks, date_joined_relay=_opt_dt_to_glean(user_data.date_joined_relay), premium_status=user_data.premium_status, date_joined_premium=_opt_dt_to_glean(user_data.date_joined_premium), has_extension=user_data.has_extension, date_got_extension=_opt_dt_to_glean(user_data.date_got_extension), is_random_mask=mask_data.is_random_mask, is_reply=is_reply, reason=reason, ) def log_api_accessed(self, request: HttpRequest) -> None: """Log that any Relay API endpoint was accessed.""" if not request.user or not request.user.is_authenticated: return request_data = RequestData.from_request(request) user_data = UserData.from_user(request.user) self.record_api_accessed( user_agent=_opt_str_to_glean(request_data.user_agent), ip_address=_opt_str_to_glean(request_data.ip_address), endpoint=request.path, method=_opt_str_to_glean(request.method), fxa_id=_opt_str_to_glean(user_data.fxa_id), ) def log_text_received( self, *, user: User, ) -> None: """Log that a text message was received.""" user_data = UserData.from_user(user) if not user_data.metrics_enabled: return request_data = RequestData() self.record_phone_text_received( user_agent=_opt_str_to_glean(request_data.user_agent), ip_address=_opt_str_to_glean(request_data.ip_address), fxa_id=_opt_str_to_glean(user_data.fxa_id), ) def log_call_received( self, *, user: User, ) -> None: """Log that a phone call was received.""" user_data = UserData.from_user(user) if not user_data.metrics_enabled: return request_data = RequestData() self.record_phone_call_received( user_agent=_opt_str_to_glean(request_data.user_agent), ip_address=_opt_str_to_glean(request_data.ip_address), fxa_id=_opt_str_to_glean(user_data.fxa_id), )