Class method for checking if urls are blocked

This commit is contained in:
Mouse Reeve 2021-04-10 11:38:57 -07:00
parent 0caeb3ac33
commit 1903812b1d
3 changed files with 17 additions and 15 deletions

View file

@ -219,6 +219,12 @@ def dict_from_mappings(data, mappings):
def get_data(url, params=None):
""" wrapper for request.get """
# check if the url is blocked
if models.FederatedServer.is_blocked(url):
raise ConnectorException(
"Attempting to load data from blocked url: {:s}".format(url)
)
try:
resp = requests.get(
url,

View file

@ -1,4 +1,5 @@
""" connections to external ActivityPub servers """
from urllib.parse import urlparse
from django.db import models
from .base_model import BookWyrmModel
@ -38,3 +39,10 @@ class FederatedServer(BookWyrmModel):
# TODO: only reactivate users as appropriate
self.user_set.update(is_active=True)
@classmethod
def is_blocked(cls, url):
""" look up if a domain is blocked """
url = urlparse(url)
domain = url.netloc
return cls.objects.filter(server_name=domain, status="blocked").exists()

View file

@ -1,7 +1,7 @@
""" incoming activities """
import json
import re
from urllib.parse import urldefrag, urlparse
from urllib.parse import urldefrag
from django.http import HttpResponse, HttpResponseNotFound
from django.http import HttpResponseBadRequest, HttpResponseForbidden
@ -71,11 +71,7 @@ def is_blocked_user_agent(request):
if not user_agent:
return False
url = re.search(r"https?://{:s}/?".format(regex.domain), user_agent).group()
domain = urlparse(url).netloc
if not domain:
# idk, we'll try again later with the actor
return False
return is_blocked(domain)
return models.FederatedServer.is_blocked(url)
def is_blocked_activity(activity_json):
@ -84,15 +80,7 @@ def is_blocked_activity(activity_json):
if not actor:
# well I guess it's not even a valid activity so who knows
return False
url = urlparse(actor)
return is_blocked(url.netloc)
def is_blocked(domain):
""" is this domain blocked? """
return models.FederatedServer.objects.filter(
server_name=domain, status="blocked"
).exists()
return models.FederatedServer.is_blocked(actor)
@app.task