michael/pyfedi
1
0
Fork 0
mirror of https://codeberg.org/rimu/pyfedi synced 2025-01-23 11:26:56 -08:00
Code Issues Projects Releases Packages Wiki Activity Actions

Share remote_object_to_json()

Browse source
This commit is contained in:
freamon 2025-01-21 03:04:20 +00:00
parent 4eb9259b0e
commit 2803054767
2 changed files with 2 additions and 73 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

35
app/activitypub/util.py
View file

@ -2181,7 +2181,6 @@ def can_delete(user_ap_id, post):
return can_edit(user_ap_id, post) return can_edit(user_ap_id, post)
# TODO: import this into community/util for backfilling, instead of having 2 copies, and - also - call it from resolve_remote_post()
def remote_object_to_json(uri): def remote_object_to_json(uri):
try: try:
object_request = get_request(uri, headers={'Accept': 'application/activity+json'}) object_request = get_request(uri, headers={'Accept': 'application/activity+json'})
@ -2233,39 +2232,7 @@ def resolve_remote_post(uri: str, community, announce_id, store_ap_json, nodebb=
actor_domain = None actor_domain = None
actor = None actor = None
try: post_data = remote_object_to_json(uri)
object_request = get_request(uri, headers={'Accept': 'application/activity+json'})
except httpx.HTTPError:
time.sleep(3)
try:
object_request = get_request(uri, headers={'Accept': 'application/activity+json'})
except httpx.HTTPError:
return None
if object_request.status_code == 200:
try:
post_data = object_request.json()
except:
object_request.close()
return None
object_request.close()
elif object_request.status_code == 401:
try:
site = Site.query.get(1)
object_request = signed_get_request(uri, site.private_key, f"https://{current_app.config['SERVER_NAME']}/actor#main-key")
except httpx.HTTPError:
time.sleep(3)
try:
object_request = signed_get_request(uri, site.private_key, f"https://{current_app.config['SERVER_NAME']}/actor#main-key")
except httpx.HTTPError:
return None
try:
post_data = object_request.json()
except:
object_request.close()
return None
object_request.close()
else:
return None
# find the author. Make sure their domain matches the site hosting it to mitigate impersonation attempts # find the author. Make sure their domain matches the site hosting it to mitigate impersonation attempts
if 'attributedTo' in post_data: if 'attributedTo' in post_data:

40
app/community/util.py
View file

@ -12,7 +12,7 @@ from pillow_heif import register_heif_opener
from app import db, cache, celery from app import db, cache, celery
from app.activitypub.signature import post_request, default_context, signed_get_request from app.activitypub.signature import post_request, default_context, signed_get_request
from app.activitypub.util import find_actor_or_create, actor_json_to_model, ensure_domains_match, \ from app.activitypub.util import find_actor_or_create, actor_json_to_model, ensure_domains_match, \
find_hashtag_or_create, create_post find_hashtag_or_create, create_post, remote_object_to_json
from app.constants import POST_TYPE_ARTICLE, POST_TYPE_LINK, POST_TYPE_IMAGE, POST_TYPE_VIDEO, NOTIF_POST, \ from app.constants import POST_TYPE_ARTICLE, POST_TYPE_LINK, POST_TYPE_IMAGE, POST_TYPE_VIDEO, NOTIF_POST, \
POST_TYPE_POLL POST_TYPE_POLL
from app.models import Community, File, BannedInstances, PostReply, Post, utcnow, CommunityMember, Site, \ from app.models import Community, File, BannedInstances, PostReply, Post, utcnow, CommunityMember, Site, \
@ -79,44 +79,6 @@ def search_for_community(address: str):
return None return None
def remote_object_to_json(uri):
try:
object_request = get_request(uri, headers={'Accept': 'application/activity+json'})
except httpx.HTTPError:
time.sleep(3)
try:
object_request = get_request(uri, headers={'Accept': 'application/activity+json'})
except httpx.HTTPError:
return None
if object_request.status_code == 200:
try:
object = object_request.json()
return object
except:
object_request.close()
return None
object_request.close()
elif object_request.status_code == 401:
try:
site = Site.query.get(1)
object_request = signed_get_request(uri, site.private_key, f"https://{current_app.config['SERVER_NAME']}/actor#main-key")
except httpx.HTTPError:
time.sleep(3)
try:
object_request = signed_get_request(uri, site.private_key, f"https://{current_app.config['SERVER_NAME']}/actor#main-key")
except httpx.HTTPError:
return None
try:
object = object_request.json()
return object
except:
object_request.close()
return None
object_request.close()
else:
return None
@celery.task @celery.task
def retrieve_mods_and_backfill(community_id: int, server, name, community_json=None): def retrieve_mods_and_backfill(community_id: int, server, name, community_json=None):
with current_app.app_context(): with current_app.app_context():