• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

mozilla / fx-private-relay / 20fdad42-28a5-47cf-a496-b03bf8e9bb6b

09 May 2024 06:22PM CUT coverage: 84.08% (-0.6%) from 84.64%
20fdad42-28a5-47cf-a496-b03bf8e9bb6b

push

circleci

web-flow
Merge pull request #4684 from mozilla/enable-flak8-bandit-checks-mpp-3802

fix MPP-3802: stop ignoring bandit security checks

3602 of 4734 branches covered (76.09%)

Branch coverage included in aggregate %.

74 of 158 new or added lines in 24 files covered. (46.84%)

4 existing lines in 4 files now uncovered.

14687 of 17018 relevant lines covered (86.3%)

10.86 hits per line

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

98.58
/privaterelay/glean_interface.py
1
"""Relay interface to EventsServerEventLogger generated by glean_parser."""
2

3
from __future__ import annotations
1✔
4

5
from datetime import datetime
1✔
6
from logging import getLogger
1✔
7
from typing import Any, Literal, NamedTuple
1✔
8

9
from django.conf import settings
1✔
10
from django.contrib.auth.models import User
1✔
11
from django.http import HttpRequest
1✔
12

13
from ipware import get_client_ip
1✔
14

15
from emails.models import DomainAddress, RelayAddress
1✔
16

17
from .glean.server_events import GLEAN_EVENT_MOZLOG_TYPE, EventsServerEventLogger
1✔
18
from .types import RELAY_CHANNEL_NAME
1✔
19

20
# Enumerate the mask setting that caused an email to not be forwarded.
21
EmailBlockedReason = Literal[
1✔
22
    "block_all",  # The mask is set to block all mail
23
    "block_promotional",  # The mask is set to block promotional / list mail
24
]
25

26

27
def _opt_dt_to_glean(value: datetime | None) -> int:
1✔
28
    """Convert an optional datetime to an integer timestamp."""
29
    return -1 if value is None else int(value.timestamp())
1✔
30

31

32
def _opt_str_to_glean(value: str | None) -> str:
1✔
33
    """Convert an optional string to a (possibly empty) string."""
34
    return "" if value is None else value
1✔
35

36

37
class RequestData(NamedTuple):
1✔
38
    """Extract and store data from the request."""
39

40
    user_agent: str | None = None
1✔
41
    ip_address: str | None = None
1✔
42

43
    @classmethod
1✔
44
    def from_request(cls, request: HttpRequest) -> RequestData:
1✔
45
        user_agent = request.headers.get("user-agent", None)
1✔
46
        client_ip, is_routable = get_client_ip(request)
1✔
47
        ip_address = client_ip if (client_ip and is_routable) else None
1✔
48
        return cls(user_agent=user_agent, ip_address=ip_address)
1✔
49

50

51
class UserData(NamedTuple):
1✔
52
    """Extract and store data from a Relay user."""
53

54
    metrics_enabled: bool
1✔
55
    fxa_id: str | None = None
1✔
56
    n_random_masks: int = 0
1✔
57
    n_domain_masks: int = 0
1✔
58
    n_deleted_random_masks: int = 0
1✔
59
    n_deleted_domain_masks: int = 0
1✔
60
    date_joined_relay: datetime | None = None
1✔
61
    date_joined_premium: datetime | None = None
1✔
62
    premium_status: str = ""
1✔
63
    has_extension: bool = False
1✔
64
    date_got_extension: datetime | None = None
1✔
65

66
    @classmethod
1✔
67
    def from_user(cls, user: User) -> UserData:
1✔
68
        metrics_enabled = user.profile.metrics_enabled
1✔
69
        if not metrics_enabled:
1✔
70
            return cls(metrics_enabled=False)
1✔
71

72
        fxa_id = user.profile.fxa.uid if user.profile.fxa else None
1✔
73
        n_random_masks = user.relayaddress_set.count()
1✔
74
        n_domain_masks = user.domainaddress_set.count()
1✔
75
        n_deleted_random_masks = user.profile.num_deleted_relay_addresses
1✔
76
        n_deleted_domain_masks = user.profile.num_deleted_domain_addresses
1✔
77
        date_joined_relay = user.date_joined
1✔
78
        if user.profile.has_premium:
1✔
79
            if user.profile.has_phone:
1✔
80
                date_joined_premium = user.profile.date_subscribed_phone
1✔
81
            else:
82
                date_joined_premium = user.profile.date_subscribed
1✔
83
        else:
84
            date_joined_premium = None
1✔
85
        premium_status = user.profile.metrics_premium_status
1✔
86
        try:
1✔
87
            earliest_mask = (
1✔
88
                user.relayaddress_set.exclude(generated_for__exact="")
89
                .only("id", "created_at")
90
                .earliest("created_at")
91
            )
92
        except RelayAddress.DoesNotExist:
1✔
93
            has_extension = False
1✔
94
            date_got_extension = None
1✔
95
        else:
96
            has_extension = True
1✔
97
            date_got_extension = earliest_mask.created_at
1✔
98

99
        return cls(
1✔
100
            metrics_enabled=True,
101
            fxa_id=fxa_id,
102
            n_random_masks=n_random_masks,
103
            n_domain_masks=n_domain_masks,
104
            n_deleted_random_masks=n_deleted_random_masks,
105
            n_deleted_domain_masks=n_deleted_domain_masks,
106
            date_joined_relay=date_joined_relay,
107
            date_joined_premium=date_joined_premium,
108
            premium_status=premium_status,
109
            has_extension=has_extension,
110
            date_got_extension=date_got_extension,
111
        )
112

113

114
class EmailMaskData(NamedTuple):
1✔
115
    """Extract and store data from a Relay email mask."""
116

117
    is_random_mask: bool
1✔
118
    has_website: bool
1✔
119

120
    @classmethod
1✔
121
    def from_mask(cls, mask: RelayAddress | DomainAddress) -> EmailMaskData:
1✔
122
        if isinstance(mask, RelayAddress):
1✔
123
            is_random_mask = True
1✔
124
            has_website = bool(mask.generated_for)
1✔
125
        else:
126
            is_random_mask = False
1✔
127
            has_website = False
1✔
128
        return EmailMaskData(is_random_mask=is_random_mask, has_website=has_website)
1✔
129

130

131
class RelayGleanLogger(EventsServerEventLogger):
1✔
132
    """Extend the generated EventsServerEventLogger for Relay usage."""
133

134
    def __init__(
1✔
135
        self,
136
        application_id: str,
137
        app_display_version: str,
138
        channel: RELAY_CHANNEL_NAME,
139
    ):
140
        if not settings.GLEAN_EVENT_MOZLOG_TYPE == GLEAN_EVENT_MOZLOG_TYPE:
1!
NEW
141
            raise ValueError(
×
142
                "settings.GLEAN_EVENT_MOZLOG_TYPE must equal GLEAN_EVENT_MOZLOG_TYPE"
143
            )
144
        self._logger = getLogger(GLEAN_EVENT_MOZLOG_TYPE)
1✔
145
        super().__init__(
1✔
146
            application_id=application_id,
147
            app_display_version=app_display_version,
148
            channel=channel,
149
        )
150

151
    def emit_record(self, now: datetime, ping: dict[str, Any]) -> None:
1✔
152
        """Emit record as a log instead of a print()"""
153
        self._logger.info(GLEAN_EVENT_MOZLOG_TYPE, extra=ping)
1✔
154

155
    def log_email_mask_created(
1✔
156
        self,
157
        *,
158
        request: HttpRequest | None = None,
159
        mask: RelayAddress | DomainAddress,
160
        created_by_api: bool,
161
    ) -> None:
162
        """Log that a Relay email mask was created."""
163
        user_data = UserData.from_user(mask.user)
1✔
164
        if not user_data.metrics_enabled:
1✔
165
            return
1✔
166
        request_data = RequestData.from_request(request) if request else RequestData()
1✔
167
        mask_data = EmailMaskData.from_mask(mask)
1✔
168
        self.record_email_mask_created(
1✔
169
            user_agent=_opt_str_to_glean(request_data.user_agent),
170
            ip_address=_opt_str_to_glean(request_data.ip_address),
171
            fxa_id=_opt_str_to_glean(user_data.fxa_id),
172
            platform="",
173
            n_random_masks=user_data.n_random_masks,
174
            n_domain_masks=user_data.n_domain_masks,
175
            n_deleted_random_masks=user_data.n_deleted_random_masks,
176
            n_deleted_domain_masks=user_data.n_deleted_domain_masks,
177
            date_joined_relay=_opt_dt_to_glean(user_data.date_joined_relay),
178
            premium_status=user_data.premium_status,
179
            date_joined_premium=_opt_dt_to_glean(user_data.date_joined_premium),
180
            has_extension=user_data.has_extension,
181
            date_got_extension=_opt_dt_to_glean(user_data.date_got_extension),
182
            is_random_mask=mask_data.is_random_mask,
183
            created_by_api=created_by_api,
184
            has_website=mask_data.has_website,
185
        )
186

187
    def log_email_mask_label_updated(
1✔
188
        self,
189
        *,
190
        request: HttpRequest,
191
        mask: RelayAddress | DomainAddress,
192
    ) -> None:
193
        """Log that a Relay email mask's label was changed."""
194
        user_data = UserData.from_user(mask.user)
1✔
195
        if not user_data.metrics_enabled:
1✔
196
            return
1✔
197
        request_data = RequestData.from_request(request)
1✔
198
        mask_data = EmailMaskData.from_mask(mask)
1✔
199
        self.record_email_mask_label_updated(
1✔
200
            user_agent=_opt_str_to_glean(request_data.user_agent),
201
            ip_address=_opt_str_to_glean(request_data.ip_address),
202
            fxa_id=_opt_str_to_glean(user_data.fxa_id),
203
            platform="",
204
            n_random_masks=user_data.n_random_masks,
205
            n_domain_masks=user_data.n_domain_masks,
206
            n_deleted_random_masks=user_data.n_deleted_random_masks,
207
            n_deleted_domain_masks=user_data.n_deleted_domain_masks,
208
            date_joined_relay=_opt_dt_to_glean(user_data.date_joined_relay),
209
            premium_status=user_data.premium_status,
210
            date_joined_premium=_opt_dt_to_glean(user_data.date_joined_premium),
211
            has_extension=user_data.has_extension,
212
            date_got_extension=_opt_dt_to_glean(user_data.date_got_extension),
213
            is_random_mask=mask_data.is_random_mask,
214
        )
215

216
    def log_email_mask_deleted(
1✔
217
        self,
218
        *,
219
        request: HttpRequest,
220
        user: User,
221
        is_random_mask: bool,
222
    ) -> None:
223
        """Log that a Relay email mask was deleted."""
224
        user_data = UserData.from_user(user)
1✔
225
        if not user_data.metrics_enabled:
1✔
226
            return
1✔
227
        request_data = RequestData.from_request(request)
1✔
228
        self.record_email_mask_deleted(
1✔
229
            user_agent=_opt_str_to_glean(request_data.user_agent),
230
            ip_address=_opt_str_to_glean(request_data.ip_address),
231
            fxa_id=_opt_str_to_glean(user_data.fxa_id),
232
            platform="",
233
            n_random_masks=user_data.n_random_masks,
234
            n_domain_masks=user_data.n_domain_masks,
235
            n_deleted_random_masks=user_data.n_deleted_random_masks,
236
            n_deleted_domain_masks=user_data.n_deleted_domain_masks,
237
            date_joined_relay=_opt_dt_to_glean(user_data.date_joined_relay),
238
            premium_status=user_data.premium_status,
239
            date_joined_premium=_opt_dt_to_glean(user_data.date_joined_premium),
240
            has_extension=user_data.has_extension,
241
            date_got_extension=_opt_dt_to_glean(user_data.date_got_extension),
242
            is_random_mask=is_random_mask,
243
        )
244

245
    def log_email_forwarded(
1✔
246
        self,
247
        *,
248
        mask: RelayAddress | DomainAddress,
249
        is_reply: bool = False,
250
    ) -> None:
251
        """Log that an email was forwarded."""
252
        user_data = UserData.from_user(mask.user)
1✔
253
        if not user_data.metrics_enabled:
1✔
254
            return
1✔
255
        request_data = RequestData()
1✔
256
        mask_data = EmailMaskData.from_mask(mask)
1✔
257
        self.record_email_forwarded(
1✔
258
            user_agent=_opt_str_to_glean(request_data.user_agent),
259
            ip_address=_opt_str_to_glean(request_data.ip_address),
260
            fxa_id=_opt_str_to_glean(user_data.fxa_id),
261
            platform="",
262
            n_random_masks=user_data.n_random_masks,
263
            n_domain_masks=user_data.n_domain_masks,
264
            n_deleted_random_masks=user_data.n_deleted_random_masks,
265
            n_deleted_domain_masks=user_data.n_deleted_domain_masks,
266
            date_joined_relay=_opt_dt_to_glean(user_data.date_joined_relay),
267
            premium_status=user_data.premium_status,
268
            date_joined_premium=_opt_dt_to_glean(user_data.date_joined_premium),
269
            has_extension=user_data.has_extension,
270
            date_got_extension=_opt_dt_to_glean(user_data.date_got_extension),
271
            is_random_mask=mask_data.is_random_mask,
272
            is_reply=is_reply,
273
        )
274

275
    def log_email_blocked(
1✔
276
        self,
277
        *,
278
        mask: RelayAddress | DomainAddress,
279
        reason: EmailBlockedReason,
280
        is_reply: bool = False,
281
    ) -> None:
282
        """Log that an email was not forwarded."""
283
        user_data = UserData.from_user(mask.user)
1✔
284
        if not user_data.metrics_enabled:
1✔
285
            return
1✔
286
        request_data = RequestData()
1✔
287
        mask_data = EmailMaskData.from_mask(mask)
1✔
288
        self.record_email_blocked(
1✔
289
            user_agent=_opt_str_to_glean(request_data.user_agent),
290
            ip_address=_opt_str_to_glean(request_data.ip_address),
291
            fxa_id=_opt_str_to_glean(user_data.fxa_id),
292
            platform="",
293
            n_random_masks=user_data.n_random_masks,
294
            n_domain_masks=user_data.n_domain_masks,
295
            n_deleted_random_masks=user_data.n_deleted_random_masks,
296
            n_deleted_domain_masks=user_data.n_deleted_domain_masks,
297
            date_joined_relay=_opt_dt_to_glean(user_data.date_joined_relay),
298
            premium_status=user_data.premium_status,
299
            date_joined_premium=_opt_dt_to_glean(user_data.date_joined_premium),
300
            has_extension=user_data.has_extension,
301
            date_got_extension=_opt_dt_to_glean(user_data.date_got_extension),
302
            is_random_mask=mask_data.is_random_mask,
303
            is_reply=is_reply,
304
            reason=reason,
305
        )
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2025 Coveralls, Inc