-
-
Notifications
You must be signed in to change notification settings - Fork 3.6k
/
signals.py
103 lines (80 loc) · 3.24 KB
/
signals.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
# -*- coding: utf-8 -*-
"""Signal handling for core app."""
from __future__ import absolute_import
import logging
from corsheaders import signals
from django.conf import settings
from django.db.models.signals import pre_delete
from django.dispatch import Signal
from django.db.models import Q, Count
from django.dispatch import receiver
from future.backports.urllib.parse import urlparse
from readthedocs.oauth.models import RemoteOrganization
from readthedocs.projects.models import Project, Domain
log = logging.getLogger(__name__)
WHITELIST_URLS = [
'/api/v2/footer_html',
'/api/v2/search',
'/api/v2/docsearch',
'/api/v2/sustainability',
]
webhook_github = Signal(providing_args=['project', 'data', 'event'])
webhook_gitlab = Signal(providing_args=['project', 'data', 'event'])
webhook_bitbucket = Signal(providing_args=['project', 'data', 'event'])
def decide_if_cors(sender, request, **kwargs): # pylint: disable=unused-argument
"""
Decide whether a request should be given CORS access.
This checks that:
* The URL is whitelisted against our CORS-allowed domains
* The Domain exists in our database, and belongs to the project being queried.
Returns True when a request should be given CORS access.
"""
if 'HTTP_ORIGIN' not in request.META:
return False
host = urlparse(request.META['HTTP_ORIGIN']).netloc.split(':')[0]
# Don't do domain checking for this API for now
if request.path_info.startswith('/api/v2/sustainability'):
return True
valid_url = False
for url in WHITELIST_URLS:
if request.path_info.startswith(url):
valid_url = True
break
if valid_url:
project_slug = request.GET.get('project', None)
try:
project = Project.objects.get(slug=project_slug)
except Project.DoesNotExist:
log.warning(
'Invalid project passed to domain. [%s:%s]',
project_slug,
host,
)
return False
domain = Domain.objects.filter(
Q(domain__icontains=host),
Q(project=project) | Q(project__subprojects__child=project)
)
if domain.exists():
return True
return False
@receiver(pre_delete, sender=settings.AUTH_USER_MODEL)
def delete_projects_and_organizations(sender, instance, *args, **kwargs):
# Here we count the owner list from the projects that the user own
# Then exclude the projects where there are more than one owner
# Add annotate before filter
# https://github.com/rtfd/readthedocs.org/pull/4577
# https://docs.djangoproject.com/en/2.1/topics/db/aggregation/#order-of-annotate-and-filter-clauses # noqa
projects = (
Project.objects.annotate(num_users=Count('users'))
.filter(users=instance.id).exclude(num_users__gt=1)
)
# Here we count the users list from the organization that the user belong
# Then exclude the organizations where there are more than one user
oauth_organizations = (
RemoteOrganization.objects.annotate(num_users=Count('users'))
.filter(users=instance.id).exclude(num_users__gt=1)
)
projects.delete()
oauth_organizations.delete()
signals.check_request_enabled.connect(decide_if_cors)