11464425854

Committed 22 Oct 2024 04:24PM UTC coverage: 85.01% (-0.1%) from 85.128%

Build # 11464425854

Build Type

Pull #1393

github

Committed by

web-flow

Commit Message

Merge 73759436d into 2905c81bd

Pull Request Pull Request #1393: introduce DownloadTicket for files

Run Details

2945 of 3465 branches covered (84.99%)

Branch coverage included in aggregate %.

13 of 22 new or added lines in 2 files covered. (59.09%)

70 existing lines in 5 files now uncovered.

11743 of 13813 relevant lines covered (85.01%)

0.85 hits per line

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

56.19

/ephios/plugins/federation/views/api.py

from urllib.parse import urljoin

import django_filters
import requests
from django.conf import settings
from django.core.exceptions import MultipleObjectsReturned
from django.shortcuts import redirect
from django.urls import reverse
from django.views import View
from oauth2_provider.contrib.rest_framework import TokenHasScope
from oauthlib.oauth2 import WebApplicationClient
from requests_oauthlib import OAuth2Session
from rest_framework.exceptions import PermissionDenied
from rest_framework.generics import CreateAPIView, DestroyAPIView, ListAPIView
from rest_framework.permissions import AllowAny

from ephios.core.models import Event, Qualification
from ephios.plugins.federation.models import FederatedGuest, FederatedUser
from ephios.plugins.federation.serializers import (
    FederatedGuestCreateSerializer,
    SharedEventSerializer,
)


class RedeemInviteCodeView(CreateAPIView):
    """
    API view that accepts an InviteCode and creates a FederatedGuest (to start sharing events with that instance).
    """

    serializer_class = FederatedGuestCreateSerializer
    queryset = FederatedGuest.objects.all()
    authentication_classes = []
    permission_classes = [AllowAny]


class FederatedGuestDeleteView(DestroyAPIView):
    """
    API view that deletes a FederatedGuest (to stop sharing events with that instance).
    """

    queryset = FederatedGuest.objects.all()
    permission_classes = [TokenHasScope]
    required_scopes = []

    def get_object(self):
        try:
            # request.auth is an auth token, federatedguest is the reverse relation
            return self.request.auth.federatedguest
        except FederatedGuest.DoesNotExist as exc:
            raise PermissionDenied from exc


class SharedEventListView(ListAPIView):
    """
    API view that lists all events that are shared with the instance corresponding to the access token.
    """

    serializer_class = SharedEventSerializer
    permission_classes = [TokenHasScope]
    required_scopes = []
    filter_backends = [django_filters.rest_framework.DjangoFilterBackend]
    filterset_fields = {
        "shifts__start_time": ["gte"],
        "shifts__end_time": ["gte"],
        "type": ["exact"],
    }

    def get_queryset(self):
        try:
            # request.auth is an auth token, federatedguest is the reverse relation
            guest = self.request.auth.federatedguest
        except FederatedGuest.DoesNotExist as exc:
            raise PermissionDenied from exc
        return Event.objects.filter(federatedeventshare__shared_with=guest)

    def get_serializer_context(self):
        context = super().get_serializer_context()
        context["federated_guest"] = self.request.auth.federatedguest
        return context


class FederationOAuthView(View):
    """
    View that handles the OAuth2 flow for federated users from another instance.
    """

    def get(self, request, *args, **kwargs):
        try:
            guest_pk = self.request.session.get("federation_guest_pk", kwargs.get("guest"))
            self.guest = FederatedGuest.objects.get(pk=guest_pk)
        except (KeyError, FederatedGuest.DoesNotExist, MultipleObjectsReturned) as exc:
            raise PermissionDenied from exc
        if "error" in request.GET.keys():
            return redirect(
                urljoin(
                    urljoin(self.guest.url, reverse("federation:external_event_list")),
                    "?error=oauth_error",
                )
            )
        elif "code" in request.GET.keys():
            self._oauth_callback()
            try:
                return redirect(
                    "federation:event_detail",
                    pk=self.request.session["federation_event"],
                    guest=guest_pk,
                )
            except KeyError:
                return redirect(
                    urljoin(
                        urljoin(self.guest.url, reverse("federation:external_event_list")),
                        "?error=event_error",
                    )
                )
        else:
            return redirect(self._get_authorization_url())

    def _get_authorization_url(self):
        oauth_client = WebApplicationClient(client_id=self.guest.client_id)
        oauth = OAuth2Session(
            client=oauth_client,
            redirect_uri=urljoin(settings.GET_SITE_URL(), reverse("federation:oauth_callback")),
            scope=["ME_READ"],
        )
        verifier = oauth_client.create_code_verifier(64)
        self.request.session["code_verifier"] = verifier
        self.request.session["federation_event"] = self.kwargs["pk"]
        self.request.session["federation_guest_pk"] = self.guest.pk
        challenge = oauth_client.create_code_challenge(verifier, "S256")
        authorization_url, _ = oauth.authorization_url(
            urljoin(self.guest.url, "api/oauth/authorize/"),
            code_challenge=challenge,
            code_challenge_method="S256",
        )
        return authorization_url

    def _oauth_callback(self):
        oauth_client = WebApplicationClient(
            client_id=self.guest.client_id, code_verifier=self.request.session["code_verifier"]
        )
        oauth = OAuth2Session(client=oauth_client)
        token = oauth.fetch_token(
            urljoin(self.guest.url, "api/oauth/token/"),
            authorization_response=urljoin(settings.GET_SITE_URL(), self.request.get_full_path()),
            client_secret=self.guest.client_secret,
            code_verifier=self.request.session["code_verifier"],
        )
        self.request.session["federation_access_token"] = token["access_token"]
        self.request.session.set_expiry(token["expires_in"])
        user_data = requests.get(
            urljoin(self.guest.url, "api/users/me/"),
            headers={"Authorization": f"Bearer {token['access_token']}"},
            timeout=5,
        )
        try:
            user = FederatedUser.objects.get(
                federated_instance=self.guest, email=user_data.json()["email"]
            )
        except FederatedUser.DoesNotExist:
            user = self._create_user(user_data.json())
        self.request.session["federated_user"] = user.pk

    def _create_user(self, user_data):
        user = FederatedUser.objects.create(
            federated_instance=self.guest,
            email=user_data["email"],
            display_name=user_data["display_name"],
            date_of_birth=user_data["date_of_birth"],
        )
        for qualification in user_data["qualifications"]:
            try:
                # Note that we assign the qualification on the host instance without further checks.
                # This may lead to incorrect qualifications if the inclusions for the qualification
                # are defined differently on the guest instance. We are accepting this as it should
                # not happen with the pre-defined qualifications as we are displaying a warning if
                # the user adapt these and custom qualifications will have different uuids anyway.
                user.qualifications.add(Qualification.objects.get(uuid=qualification["uuid"]))
            except Qualification.DoesNotExist:
                for included_qualification in qualification["includes"]:
                    try:
                        user.qualifications.add(
                            Qualification.objects.get(uuid=included_qualification)
                        )
                    except Qualification.DoesNotExist:
                        continue
        return user

1	from urllib.parse import urljoin	1✔
2
3	import django_filters	1✔
4	import requests	1✔
5	from django.conf import settings	1✔
6	from django.core.exceptions import MultipleObjectsReturned	1✔
7	from django.shortcuts import redirect	1✔
8	from django.urls import reverse	1✔
9	from django.views import View	1✔
10	from oauth2_provider.contrib.rest_framework import TokenHasScope	1✔
11	from oauthlib.oauth2 import WebApplicationClient	1✔
12	from requests_oauthlib import OAuth2Session	1✔
13	from rest_framework.exceptions import PermissionDenied	1✔
14	from rest_framework.generics import CreateAPIView, DestroyAPIView, ListAPIView	1✔
15	from rest_framework.permissions import AllowAny	1✔
16
17	from ephios.core.models import Event, Qualification	1✔
18	from ephios.plugins.federation.models import FederatedGuest, FederatedUser	1✔
19	from ephios.plugins.federation.serializers import (	1✔
20	FederatedGuestCreateSerializer,
21	SharedEventSerializer,
22	)
23
24
25	class RedeemInviteCodeView(CreateAPIView):	1✔
26	"""
27	API view that accepts an InviteCode and creates a FederatedGuest (to start sharing events with that instance).
28	"""
29
30	serializer_class = FederatedGuestCreateSerializer	1✔
31	queryset = FederatedGuest.objects.all()	1✔
32	authentication_classes = []	1✔
33	permission_classes = [AllowAny]	1✔
34
35
36	class FederatedGuestDeleteView(DestroyAPIView):	1✔
37	"""
38	API view that deletes a FederatedGuest (to stop sharing events with that instance).
39	"""
40
41	queryset = FederatedGuest.objects.all()	1✔
42	permission_classes = [TokenHasScope]	1✔
43	required_scopes = []	1✔
44
45	def get_object(self):	1✔
46	try:	×
47	# request.auth is an auth token, federatedguest is the reverse relation
48	return self.request.auth.federatedguest	×
49	except FederatedGuest.DoesNotExist as exc:	×
50	raise PermissionDenied from exc	×
51
52
53	class SharedEventListView(ListAPIView):	1✔
54	"""
55	API view that lists all events that are shared with the instance corresponding to the access token.
56	"""
57
58	serializer_class = SharedEventSerializer	1✔
59	permission_classes = [TokenHasScope]	1✔
60	required_scopes = []	1✔
61	filter_backends = [django_filters.rest_framework.DjangoFilterBackend]	1✔
62	filterset_fields = {	1✔
63	"shifts__start_time": ["gte"],
64	"shifts__end_time": ["gte"],
65	"type": ["exact"],
66	}
67
68	def get_queryset(self):	1✔
69	try:	1✔
70	# request.auth is an auth token, federatedguest is the reverse relation
71	guest = self.request.auth.federatedguest	1✔
72	except FederatedGuest.DoesNotExist as exc:	×
73	raise PermissionDenied from exc	×
74	return Event.objects.filter(federatedeventshare__shared_with=guest)	1✔
75
76	def get_serializer_context(self):	1✔
77	context = super().get_serializer_context()	1✔
78	context["federated_guest"] = self.request.auth.federatedguest	1✔
79	return context	1✔
80
81
82	class FederationOAuthView(View):	1✔
83	"""
84	View that handles the OAuth2 flow for federated users from another instance.
85	"""
86
87	def get(self, request, args, *kwargs):	1✔
UNCOV 88	try:	×
89	guest_pk = self.request.session.get("federation_guest_pk", kwargs.get("guest"))	×
90	self.guest = FederatedGuest.objects.get(pk=guest_pk)	×
91	except (KeyError, FederatedGuest.DoesNotExist, MultipleObjectsReturned) as exc:	×
92	raise PermissionDenied from exc	×
93	if "error" in request.GET.keys():	×
94	return redirect(	×
95	urljoin(
96	urljoin(self.guest.url, reverse("federation:external_event_list")),
97	"?error=oauth_error",
98	)
99	)
UNCOV 100	elif "code" in request.GET.keys():	×
UNCOV 101	self._oauth_callback()	×
102	try:	×
103	return redirect(	×
104	"federation:event_detail",
105	pk=self.request.session["federation_event"],
106	guest=guest_pk,
107	)
108	except KeyError:	×
109	return redirect(	×
110	urljoin(
111	urljoin(self.guest.url, reverse("federation:external_event_list")),
112	"?error=event_error",
113	)
114	)
115	else:
UNCOV 116	return redirect(self._get_authorization_url())	×
117
118	def _get_authorization_url(self):	1✔
UNCOV 119	oauth_client = WebApplicationClient(client_id=self.guest.client_id)	×
UNCOV 120	oauth = OAuth2Session(	×
121	client=oauth_client,
122	redirect_uri=urljoin(settings.GET_SITE_URL(), reverse("federation:oauth_callback")),
123	scope=["ME_READ"],
124	)
125	verifier = oauth_client.create_code_verifier(64)	×
UNCOV 126	self.request.session["code_verifier"] = verifier	×
UNCOV 127	self.request.session["federation_event"] = self.kwargs["pk"]	×
UNCOV 128	self.request.session["federation_guest_pk"] = self.guest.pk	×
UNCOV 129	challenge = oauth_client.create_code_challenge(verifier, "S256")	×
UNCOV 130	authorization_url, _ = oauth.authorization_url(	×
131	urljoin(self.guest.url, "api/oauth/authorize/"),
132	code_challenge=challenge,
133	code_challenge_method="S256",
134	)
UNCOV 135	return authorization_url	×
136
137	def _oauth_callback(self):	1✔
138	oauth_client = WebApplicationClient(	×
139	client_id=self.guest.client_id, code_verifier=self.request.session["code_verifier"]
140	)
UNCOV 141	oauth = OAuth2Session(client=oauth_client)	×
142	token = oauth.fetch_token(	×
143	urljoin(self.guest.url, "api/oauth/token/"),
144	authorization_response=urljoin(settings.GET_SITE_URL(), self.request.get_full_path()),
145	client_secret=self.guest.client_secret,
146	code_verifier=self.request.session["code_verifier"],
147	)
UNCOV 148	self.request.session["federation_access_token"] = token["access_token"]	×
UNCOV 149	self.request.session.set_expiry(token["expires_in"])	×
UNCOV 150	user_data = requests.get(	×
151	urljoin(self.guest.url, "api/users/me/"),
152	headers={"Authorization": f"Bearer {token['access_token']}"},
153	timeout=5,
154	)
UNCOV 155	try:	×
UNCOV 156	user = FederatedUser.objects.get(	×
157	federated_instance=self.guest, email=user_data.json()["email"]
158	)
UNCOV 159	except FederatedUser.DoesNotExist:	×
UNCOV 160	user = self._create_user(user_data.json())	×
UNCOV 161	self.request.session["federated_user"] = user.pk	×
162
163	def _create_user(self, user_data):	1✔
164	user = FederatedUser.objects.create(	1✔
165	federated_instance=self.guest,
166	email=user_data["email"],
167	display_name=user_data["display_name"],
168	date_of_birth=user_data["date_of_birth"],
169	)
170	for qualification in user_data["qualifications"]:	1✔
171	try:	1✔
172	# Note that we assign the qualification on the host instance without further checks.
173	# This may lead to incorrect qualifications if the inclusions for the qualification
174	# are defined differently on the guest instance. We are accepting this as it should
175	# not happen with the pre-defined qualifications as we are displaying a warning if
176	# the user adapt these and custom qualifications will have different uuids anyway.
177	user.qualifications.add(Qualification.objects.get(uuid=qualification["uuid"]))	1✔
178	except Qualification.DoesNotExist:	1✔
179	for included_qualification in qualification["includes"]:	1✔
180	try:	1✔
181	user.qualifications.add(	1✔
182	Qualification.objects.get(uuid=included_qualification)
183	)
UNCOV 184	except Qualification.DoesNotExist:	×
UNCOV 185	continue	×
186	return user	1✔

ephios-dev / ephios / 11464425854

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