8354d07c-7eab-4972-926d-a2104e534166

import logging
from typing import Mapping, Optional

from django.conf import settings
from django.contrib.auth.models import User
from django.db import IntegrityError

from rest_framework.exceptions import APIException
from rest_framework.response import Response
from rest_framework.views import exception_handler
from rest_framework.serializers import ValidationError

from django_filters import rest_framework as filters
from drf_yasg.utils import swagger_auto_schema
from drf_yasg.views import get_schema_view
from drf_yasg import openapi
from waffle import get_waffle_flag_model
from waffle.models import Switch, Sample
from rest_framework import (
    decorators,
    permissions,
    response,
    status,
    viewsets,
    exceptions,
)
from emails.utils import incr_if_enabled

from privaterelay.utils import (
    get_countries_info_from_request_and_mapping,
)

from emails.models import (
    CannotMakeAddressException,
    DomainAddress,
    Profile,
    RelayAddress,
)


from ..exceptions import ConflictError, RelayAPIException
from ..permissions import IsOwner, CanManageFlags
from ..serializers import (
    DomainAddressSerializer,
    ProfileSerializer,
    RelayAddressSerializer,
    UserSerializer,
    FlagSerializer,
    WebcompatIssueSerializer,
)

from privaterelay.ftl_bundles import main as ftl_bundle

info_logger = logging.getLogger("eventsinfo")
schema_view = get_schema_view(
    openapi.Info(
        title="Relay API",
        default_version="v1",
        description="API endpints for Relay back-end",
        contact=openapi.Contact(email="lcrouch+relayapi@mozilla.com"),
    ),
    public=settings.DEBUG,
    permission_classes=[permissions.AllowAny],
)


class SaveToRequestUser:
    def perform_create(self, serializer):
        serializer.save(user=self.request.user)


class RelayAddressFilter(filters.FilterSet):
    used_on = filters.CharFilter(field_name="used_on", lookup_expr="icontains")

    class Meta:
        model = RelayAddress
        fields = [
            "enabled",
            "description",
            "generated_for",
            "block_list_emails",
            "used_on",
            # read-only
            "id",
            "address",
            "domain",
            "created_at",
            "last_modified_at",
            "last_used_at",
            "num_forwarded",
            "num_blocked",
            "num_spam",
        ]


class RelayAddressViewSet(SaveToRequestUser, viewsets.ModelViewSet):
    serializer_class = RelayAddressSerializer
    permission_classes = [permissions.IsAuthenticated, IsOwner]
    filterset_class = RelayAddressFilter

    def get_queryset(self):
        return RelayAddress.objects.filter(user=self.request.user)


class DomainAddressFilter(filters.FilterSet):
    used_on = filters.CharFilter(field_name="used_on", lookup_expr="icontains")

    class Meta:
        model = DomainAddress
        fields = [
            "enabled",
            "description",
            "block_list_emails",
            "used_on",
            # read-only
            "id",
            "address",
            "domain",
            "created_at",
            "last_modified_at",
            "last_used_at",
            "num_forwarded",
            "num_blocked",
            "num_spam",
        ]


class DomainAddressViewSet(SaveToRequestUser, viewsets.ModelViewSet):
    serializer_class = DomainAddressSerializer
    permission_classes = [permissions.IsAuthenticated, IsOwner]
    filterset_class = DomainAddressFilter

    def get_queryset(self):
        return DomainAddress.objects.filter(user=self.request.user)

    def perform_create(self, serializer):
        try:
            serializer.save(user=self.request.user)
        except IntegrityError as e:
            domain_address = DomainAddress.objects.filter(
                user=self.request.user, address=serializer.validated_data.get("address")
            ).first()
            raise ConflictError(
                {"id": domain_address.id, "full_address": domain_address.full_address}
            )


class ProfileViewSet(viewsets.ModelViewSet):
    serializer_class = ProfileSerializer
    permission_classes = [permissions.IsAuthenticated, IsOwner]
    http_method_names = ["get", "post", "head", "put", "patch"]

    def get_queryset(self):
        return Profile.objects.filter(user=self.request.user)


class UserViewSet(viewsets.ModelViewSet):
    serializer_class = UserSerializer
    permission_classes = [permissions.IsAuthenticated, IsOwner]
    http_method_names = ["get", "head"]

    def get_queryset(self):
        return User.objects.filter(id=self.request.user.id)


@decorators.api_view()
@decorators.permission_classes([permissions.AllowAny])
def runtime_data(request):
    flags = get_waffle_flag_model().get_all()
    flag_values = [(f.name, f.is_active(request)) for f in flags]
    switches = Switch.get_all()
    switch_values = [(s.name, s.is_active()) for s in switches]
    samples = Sample.get_all()
    sample_values = [(s.name, s.is_active()) for s in samples]
    return response.Response(
        {
            "FXA_ORIGIN": settings.FXA_BASE_ORIGIN,
            "PERIODICAL_PREMIUM_PRODUCT_ID": settings.PERIODICAL_PREMIUM_PROD_ID,
            "GOOGLE_ANALYTICS_ID": settings.GOOGLE_ANALYTICS_ID,
            "BUNDLE_PRODUCT_ID": settings.BUNDLE_PROD_ID,
            "PHONE_PRODUCT_ID": settings.PHONE_PROD_ID,
            "PERIODICAL_PREMIUM_PLANS": get_countries_info_from_request_and_mapping(
                request, settings.PERIODICAL_PREMIUM_PLAN_COUNTRY_LANG_MAPPING
            ),
            "PHONE_PLANS": get_countries_info_from_request_and_mapping(
                request, settings.PHONE_PLAN_COUNTRY_LANG_MAPPING
            ),
            "BUNDLE_PLANS": get_countries_info_from_request_and_mapping(
                request, settings.BUNDLE_PLAN_COUNTRY_LANG_MAPPING
            ),
            "BASKET_ORIGIN": settings.BASKET_ORIGIN,
            "WAFFLE_FLAGS": flag_values,
            "WAFFLE_SWITCHES": switch_values,
            "WAFFLE_SAMPLES": sample_values,
            "MAX_MINUTES_TO_VERIFY_REAL_PHONE": settings.MAX_MINUTES_TO_VERIFY_REAL_PHONE,
        }
    )


class FlagFilter(filters.FilterSet):
    class Meta:
        model = get_waffle_flag_model()
        fields = [
            "name",
            "everyone",
            # "users",
            # read-only
            "id",
        ]


class FlagViewSet(viewsets.ModelViewSet):
    serializer_class = FlagSerializer
    permission_classes = [permissions.IsAuthenticated, CanManageFlags]
    filterset_class = FlagFilter
    http_method_names = ["get", "post", "head", "patch"]

    def get_queryset(self):
        flags = get_waffle_flag_model().objects
        return flags


@swagger_auto_schema(methods=["post"], request_body=WebcompatIssueSerializer)
@decorators.api_view(["POST"])
@decorators.permission_classes([permissions.IsAuthenticated])
def report_webcompat_issue(request):
    serializer = WebcompatIssueSerializer(data=request.data)
    if serializer.is_valid():
        info_logger.info("webcompat_issue", extra=serializer.data)
        incr_if_enabled("webcompat_issue", 1)
        for k, v in serializer.data.items():
            if v and k != "issue_on_domain":
                incr_if_enabled(f"webcompat_issue_{k}", 1)
        return response.Response(status=status.HTTP_201_CREATED)
    return response.Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)


def relay_exception_handler(exc: Exception, context: Mapping) -> Optional[Response]:
    """
    Add error information to response data.

    When the error is a RelayAPIException, these additional fields may be present and
    the information will be translated if an Accept-Language header is added to the request:

    error_code - A string identifying the error, for client-side translation
    error_context - Additional data needed for client-side translation
    """

    response = exception_handler(exc, context)

    if response and isinstance(exc, RelayAPIException):
        error_codes = exc.get_codes()
        error_context = exc.error_context()
        if isinstance(error_codes, str):
            response.data["error_code"] = error_codes

            # Build Fluent error ID
            ftl_id_sub = "api-error-"
            ftl_id_error = error_codes.replace("_", "-")
            ftl_id = ftl_id_sub + ftl_id_error

            # Replace default message with Fluent string
            response.data["detail"] = ftl_bundle.format(ftl_id, error_context)

        if error_context:
            response.data["error_context"] = error_context

        response.data["error_code"] = error_codes

    return response

1	import logging	1✔
2	from typing import Mapping, Optional	1✔
3
4	from django.conf import settings	1✔
5	from django.contrib.auth.models import User	1✔
6	from django.db import IntegrityError	1✔
7
8	from rest_framework.exceptions import APIException	1✔
9	from rest_framework.response import Response	1✔
10	from rest_framework.views import exception_handler	1✔
11	from rest_framework.serializers import ValidationError	1✔
12
13	from django_filters import rest_framework as filters	1✔
14	from drf_yasg.utils import swagger_auto_schema	1✔
15	from drf_yasg.views import get_schema_view	1✔
16	from drf_yasg import openapi	1✔
17	from waffle import get_waffle_flag_model	1✔
18	from waffle.models import Switch, Sample	1✔
19	from rest_framework import (	1✔
20	decorators,
21	permissions,
22	response,
23	status,
24	viewsets,
25	exceptions,
26	)
27	from emails.utils import incr_if_enabled	1✔
28
29	from privaterelay.utils import (	1✔
30	get_countries_info_from_request_and_mapping,
31	)
32
33	from emails.models import (	1✔
34	CannotMakeAddressException,
35	DomainAddress,
36	Profile,
37	RelayAddress,
38	)
39
40
41	from ..exceptions import ConflictError, RelayAPIException	1✔
42	from ..permissions import IsOwner, CanManageFlags	1✔
43	from ..serializers import (	1✔
44	DomainAddressSerializer,
45	ProfileSerializer,
46	RelayAddressSerializer,
47	UserSerializer,
48	FlagSerializer,
49	WebcompatIssueSerializer,
50	)
51
52	from privaterelay.ftl_bundles import main as ftl_bundle	1✔
53
54	info_logger = logging.getLogger("eventsinfo")	1✔
55	schema_view = get_schema_view(	1✔
56	openapi.Info(
57	title="Relay API",
58	default_version="v1",
59	description="API endpints for Relay back-end",
60	contact=openapi.Contact(email="lcrouch+relayapi@mozilla.com"),
61	),
62	public=settings.DEBUG,
63	permission_classes=[permissions.AllowAny],
64	)
65
66
67	class SaveToRequestUser:	1✔
68	def perform_create(self, serializer):	1✔
69	serializer.save(user=self.request.user)	1✔
70
71
72	class RelayAddressFilter(filters.FilterSet):	1✔
73	used_on = filters.CharFilter(field_name="used_on", lookup_expr="icontains")	1✔
74
75	class Meta:	1✔
76	model = RelayAddress	1✔
77	fields = [	1✔
78	"enabled",
79	"description",
80	"generated_for",
81	"block_list_emails",
82	"used_on",
83	# read-only
84	"id",
85	"address",
86	"domain",
87	"created_at",
88	"last_modified_at",
89	"last_used_at",
90	"num_forwarded",
91	"num_blocked",
92	"num_spam",
93	]
94
95
96	class RelayAddressViewSet(SaveToRequestUser, viewsets.ModelViewSet):	1✔
97	serializer_class = RelayAddressSerializer	1✔
98	permission_classes = [permissions.IsAuthenticated, IsOwner]	1✔
99	filterset_class = RelayAddressFilter	1✔
100
101	def get_queryset(self):	1✔
102	return RelayAddress.objects.filter(user=self.request.user)	1✔
103
104
105	class DomainAddressFilter(filters.FilterSet):	1✔
106	used_on = filters.CharFilter(field_name="used_on", lookup_expr="icontains")	1✔
107
108	class Meta:	1✔
109	model = DomainAddress	1✔
110	fields = [	1✔
111	"enabled",
112	"description",
113	"block_list_emails",
114	"used_on",
115	# read-only
116	"id",
117	"address",
118	"domain",
119	"created_at",
120	"last_modified_at",
121	"last_used_at",
122	"num_forwarded",
123	"num_blocked",
124	"num_spam",
125	]
126
127
128	class DomainAddressViewSet(SaveToRequestUser, viewsets.ModelViewSet):	1✔
129	serializer_class = DomainAddressSerializer	1✔
130	permission_classes = [permissions.IsAuthenticated, IsOwner]	1✔
131	filterset_class = DomainAddressFilter	1✔
132
133	def get_queryset(self):	1✔
134	return DomainAddress.objects.filter(user=self.request.user)	1✔
135
136	def perform_create(self, serializer):	1✔
137	try:	1✔
138	serializer.save(user=self.request.user)	1✔
139	except IntegrityError as e:	1✔
140	domain_address = DomainAddress.objects.filter(	×
141	user=self.request.user, address=serializer.validated_data.get("address")
142	).first()
143	raise ConflictError(	×
144	{"id": domain_address.id, "full_address": domain_address.full_address}
145	)
146
147
148	class ProfileViewSet(viewsets.ModelViewSet):	1✔
149	serializer_class = ProfileSerializer	1✔
150	permission_classes = [permissions.IsAuthenticated, IsOwner]	1✔
151	http_method_names = ["get", "post", "head", "put", "patch"]	1✔
152
153	def get_queryset(self):	1✔
154	return Profile.objects.filter(user=self.request.user)	1✔
155
156
157	class UserViewSet(viewsets.ModelViewSet):	1✔
158	serializer_class = UserSerializer	1✔
159	permission_classes = [permissions.IsAuthenticated, IsOwner]	1✔
160	http_method_names = ["get", "head"]	1✔
161
162	def get_queryset(self):	1✔
163	return User.objects.filter(id=self.request.user.id)	1✔
164
165
166	@decorators.api_view()	1✔
167	@decorators.permission_classes([permissions.AllowAny])	1✔
168	def runtime_data(request):	1✔
169	flags = get_waffle_flag_model().get_all()	1✔
170	flag_values = [(f.name, f.is_active(request)) for f in flags]	1✔
171	switches = Switch.get_all()	1✔
172	switch_values = [(s.name, s.is_active()) for s in switches]	1✔
173	samples = Sample.get_all()	1✔
174	sample_values = [(s.name, s.is_active()) for s in samples]	1✔
175	return response.Response(	1✔
176	{
177	"FXA_ORIGIN": settings.FXA_BASE_ORIGIN,
178	"PERIODICAL_PREMIUM_PRODUCT_ID": settings.PERIODICAL_PREMIUM_PROD_ID,
179	"GOOGLE_ANALYTICS_ID": settings.GOOGLE_ANALYTICS_ID,
180	"BUNDLE_PRODUCT_ID": settings.BUNDLE_PROD_ID,
181	"PHONE_PRODUCT_ID": settings.PHONE_PROD_ID,
182	"PERIODICAL_PREMIUM_PLANS": get_countries_info_from_request_and_mapping(
183	request, settings.PERIODICAL_PREMIUM_PLAN_COUNTRY_LANG_MAPPING
184	),
185	"PHONE_PLANS": get_countries_info_from_request_and_mapping(
186	request, settings.PHONE_PLAN_COUNTRY_LANG_MAPPING
187	),
188	"BUNDLE_PLANS": get_countries_info_from_request_and_mapping(
189	request, settings.BUNDLE_PLAN_COUNTRY_LANG_MAPPING
190	),
191	"BASKET_ORIGIN": settings.BASKET_ORIGIN,
192	"WAFFLE_FLAGS": flag_values,
193	"WAFFLE_SWITCHES": switch_values,
194	"WAFFLE_SAMPLES": sample_values,
195	"MAX_MINUTES_TO_VERIFY_REAL_PHONE": settings.MAX_MINUTES_TO_VERIFY_REAL_PHONE,
196	}
197	)
198
199
200	class FlagFilter(filters.FilterSet):	1✔
201	class Meta:	1✔
202	model = get_waffle_flag_model()	1✔
203	fields = [	1✔
204	"name",
205	"everyone",
206	# "users",
207	# read-only
208	"id",
209	]
210
211
212	class FlagViewSet(viewsets.ModelViewSet):	1✔
213	serializer_class = FlagSerializer	1✔
214	permission_classes = [permissions.IsAuthenticated, CanManageFlags]	1✔
215	filterset_class = FlagFilter	1✔
216	http_method_names = ["get", "post", "head", "patch"]	1✔
217
218	def get_queryset(self):	1✔
219	flags = get_waffle_flag_model().objects	1✔
220	return flags	1✔
221
222
223	@swagger_auto_schema(methods=["post"], request_body=WebcompatIssueSerializer)	1✔
224	@decorators.api_view(["POST"])	1✔
225	@decorators.permission_classes([permissions.IsAuthenticated])	1✔
226	def report_webcompat_issue(request):	1✔
227	serializer = WebcompatIssueSerializer(data=request.data)	×
228	if serializer.is_valid():	×
229	info_logger.info("webcompat_issue", extra=serializer.data)	×
230	incr_if_enabled("webcompat_issue", 1)	×
231	for k, v in serializer.data.items():	×
232	if v and k != "issue_on_domain":	×
233	incr_if_enabled(f"webcompat_issue_{k}", 1)	×
234	return response.Response(status=status.HTTP_201_CREATED)	×
235	return response.Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)	×
236
237
238	def relay_exception_handler(exc: Exception, context: Mapping) -> Optional[Response]:	1✔
239	"""
240	Add error information to response data.
241
242	When the error is a RelayAPIException, these additional fields may be present and
243	the information will be translated if an Accept-Language header is added to the request:
244
245	error_code - A string identifying the error, for client-side translation
246	error_context - Additional data needed for client-side translation
247	"""
248
249	response = exception_handler(exc, context)	1✔
250
251	if response and isinstance(exc, RelayAPIException):	1✔
252	error_codes = exc.get_codes()	1✔
253	error_context = exc.error_context()	1✔
254	if isinstance(error_codes, str):	1!
255	response.data["error_code"] = error_codes	1✔
256
257	# Build Fluent error ID
258	ftl_id_sub = "api-error-"	1✔
259	ftl_id_error = error_codes.replace("_", "-")	1✔
260	ftl_id = ftl_id_sub + ftl_id_error	1✔
261
262	# Replace default message with Fluent string
263	response.data["detail"] = ftl_bundle.format(ftl_id, error_context)	1✔
264
265	if error_context:	1✔
266	response.data["error_context"] = error_context	1✔
267
268	response.data["error_code"] = error_codes	1✔
269
270	return response	1✔

mozilla / fx-private-relay / 8354d07c-7eab-4972-926d-a2104e534166

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