relay: catch 202s
[relay.git/.git] / relay / actor.py
1 import aiohttp
2 import aiohttp.web
3 import asyncio
4 import logging
5 import uuid
6 import re
7 import urllib.parse
8 import simplejson as json
9 import cgi
10 from Crypto.PublicKey import RSA
11 from .database import DATABASE
12 from .http_debug import http_debug
13
14 from cachetools import LFUCache
15
16
17 # generate actor keys if not present
18 if "actorKeys" not in DATABASE:
19     logging.info("No actor keys present, generating 4096-bit RSA keypair.")
20
21     privkey = RSA.generate(4096)
22     pubkey = privkey.publickey()
23
24     DATABASE["actorKeys"] = {
25         "publicKey": pubkey.exportKey('PEM'),
26         "privateKey": privkey.exportKey('PEM')
27     }
28
29
30 PRIVKEY = RSA.importKey(DATABASE["actorKeys"]["privateKey"])
31 PUBKEY = PRIVKEY.publickey()
32
33
34 from . import app, CONFIG
35 from .remote_actor import fetch_actor
36
37
38 AP_CONFIG = CONFIG.get('ap', {'host': 'localhost','blocked_instances':[]})
39 CACHE_SIZE = CONFIG.get('cache-size', 16384)
40
41
42 CACHE = LFUCache(CACHE_SIZE)
43
44
45 async def actor(request):
46     data = {
47         "@context": "https://www.w3.org/ns/activitystreams",
48         "endpoints": {
49             "sharedInbox": "https://{}/inbox".format(request.host)
50         },
51         "followers": "https://{}/followers".format(request.host),
52         "following": "https://{}/following".format(request.host),
53         "inbox": "https://{}/inbox".format(request.host),
54         "name": "ActivityRelay",
55         "type": "Application",
56         "id": "https://{}/actor".format(request.host),
57         "publicKey": {
58             "id": "https://{}/actor#main-key".format(request.host),
59             "owner": "https://{}/actor".format(request.host),
60             "publicKeyPem": DATABASE["actorKeys"]["publicKey"]
61         },
62         "summary": "ActivityRelay bot",
63         "preferredUsername": "relay",
64         "url": "https://{}/actor".format(request.host)
65     }
66     return aiohttp.web.json_response(data)
67
68
69 app.router.add_get('/actor', actor)
70
71
72 from .http_signatures import sign_headers
73
74
75 get_actor_inbox = lambda actor: actor.get('endpoints', {}).get('sharedInbox', actor['inbox'])
76
77
78 async def push_message_to_actor(actor, message, our_key_id):
79     inbox = get_actor_inbox(actor)
80
81     url = urllib.parse.urlsplit(inbox)
82
83     # XXX: Digest
84     data = json.dumps(message)
85     headers = {
86         '(request-target)': 'post {}'.format(url.path),
87         'Content-Length': str(len(data)),
88         'Content-Type': 'application/activity+json',
89         'User-Agent': 'ActivityRelay'
90     }
91     headers['signature'] = sign_headers(headers, PRIVKEY, our_key_id)
92     headers.pop('(request-target)')
93
94     logging.debug('%r >> %r', inbox, message)
95
96     async with aiohttp.ClientSession(trace_configs=[http_debug()]) as session:
97         async with session.post(inbox, data=data, headers=headers) as resp:
98             if resp.status == 202:
99                 return
100             resp_payload = await resp.text()
101             logging.debug('%r >> resp %r', inbox, resp_payload)
102
103
104 async def follow_remote_actor(actor_uri):
105     logging.info('following: %r', actor_uri)
106
107     actor = await fetch_actor(actor_uri)
108
109     message = {
110         "@context": "https://www.w3.org/ns/activitystreams",
111         "type": "Follow",
112         "to": [actor['id']],
113         "object": actor['id'],
114         "id": "https://{}/activities/{}".format(AP_CONFIG['host'], uuid.uuid4()),
115         "actor": "https://{}/actor".format(AP_CONFIG['host'])
116     }
117     await push_message_to_actor(actor, message, "https://{}/actor#main-key".format(AP_CONFIG['host']))
118
119
120 async def unfollow_remote_actor(actor_uri):
121     logging.info('unfollowing: %r', actor_uri)
122
123     actor = await fetch_actor(actor_uri)
124
125     message = {
126         "@context": "https://www.w3.org/ns/activitystreams",
127         "type": "Undo",
128         "to": [actor['id']],
129         "object": {
130              "type": "Follow",
131              "object": actor_uri,
132              "actor": actor['id'],
133              "id": "https://{}/activities/{}".format(AP_CONFIG['host'], uuid.uuid4())
134         },
135         "id": "https://{}/activities/{}".format(AP_CONFIG['host'], uuid.uuid4()),
136         "actor": "https://{}/actor".format(AP_CONFIG['host'])
137     }
138     await push_message_to_actor(actor, message, "https://{}/actor#main-key".format(AP_CONFIG['host']))
139
140
141 tag_re = re.compile(r'(<!--.*?-->|<[^>]*>)')
142 def strip_html(data):
143     no_tags = tag_re.sub('', data)
144     return cgi.escape(no_tags)
145
146
147 def distill_inboxes(actor):
148     global DATABASE
149
150     inbox = get_actor_inbox(actor)
151     targets = [target for target in DATABASE.get('relay-list', []) if target != inbox]
152
153     assert inbox not in targets
154
155     return targets
156
157
158 def distill_object_id(activity):
159     logging.debug('>> determining object ID for %r', activity['object'])
160     obj = activity['object']
161
162     if isinstance(obj, str):
163         return obj
164
165     return obj['id']
166
167
168 async def handle_relay(actor, data, request):
169     global CACHE
170
171     object_id = distill_object_id(data)
172
173     if object_id in CACHE:
174         logging.debug('>> already relayed %r as %r', object_id, CACHE[object_id])
175         return
176
177     # don't relay mastodon announces -- causes LRP fake direction issues
178     if data['type'] == 'Announce' and len(data.get('cc', [])) > 0:
179         return
180
181     activity_id = "https://{}/activities/{}".format(request.host, uuid.uuid4())
182
183     message = {
184         "@context": "https://www.w3.org/ns/activitystreams",
185         "type": "Announce",
186         "to": ["https://{}/actor/followers".format(request.host)],
187         "actor": "https://{}/actor".format(request.host),
188         "object": object_id,
189         "id": activity_id
190     }
191
192     logging.debug('>> relay: %r', message)
193
194     inboxes = distill_inboxes(actor)
195
196     futures = [push_message_to_actor({'inbox': inbox}, message, 'https://{}/actor#main-key'.format(request.host)) for inbox in inboxes]
197     asyncio.ensure_future(asyncio.gather(*futures))
198
199     CACHE[object_id] = activity_id
200
201
202 async def handle_follow(actor, data, request):
203     global DATABASE
204
205     following = DATABASE.get('relay-list', [])
206     inbox = get_actor_inbox(actor)
207
208     if urllib.parse.urlsplit(inbox).hostname in AP_CONFIG['blocked_instances']:
209         return
210
211     if inbox not in following:
212         following += [inbox]
213         DATABASE['relay-list'] = following
214
215         if data['object'].endswith('/actor'):
216             asyncio.ensure_future(follow_remote_actor(actor['id']))
217
218     message = {
219         "@context": "https://www.w3.org/ns/activitystreams",
220         "type": "Accept",
221         "to": [actor["id"]],
222         "actor": "https://{}/actor".format(request.host),
223
224         # this is wrong per litepub, but mastodon < 2.4 is not compliant with that profile.
225         "object": {
226              "type": "Follow",
227              "id": data["id"],
228              "object": "https://{}/actor".format(request.host),
229              "actor": actor["id"]
230         },
231
232         "id": "https://{}/activities/{}".format(request.host, uuid.uuid4()),
233     }
234
235     asyncio.ensure_future(push_message_to_actor(actor, message, 'https://{}/actor#main-key'.format(request.host)))
236
237
238 async def handle_undo(actor, data, request):
239     global DATABASE
240
241     child = data['object']
242     if child['type'] == 'Follow':
243         following = DATABASE.get('relay-list', [])
244
245         inbox = get_actor_inbox(actor)
246
247         if inbox in following:
248             following.remove(inbox)
249             DATABASE['relay-list'] = following
250
251         if child['object'].endswith('/actor'):
252             await unfollow_remote_actor(actor['id'])
253
254
255 processors = {
256     'Announce': handle_relay,
257     'Create': handle_relay,
258     'Follow': handle_follow,
259     'Undo': handle_undo
260 }
261
262
263 async def inbox(request):
264     data = await request.json()
265
266     if 'actor' not in data or not request['validated']:
267         raise aiohttp.web.HTTPUnauthorized(body='access denied', content_type='text/plain')
268
269     actor = await fetch_actor(data["actor"])
270     actor_uri = 'https://{}/actor'.format(request.host)
271
272     logging.debug(">> payload %r", data)
273
274     processor = processors.get(data['type'], None)
275     if processor:
276         await processor(actor, data, request)
277
278     return aiohttp.web.Response(body=b'{}', content_type='application/activity+json')
279
280
281 app.router.add_post('/inbox', inbox)