epicyon/inbox.py

291 lines
11 KiB
Python
Raw Normal View History

2019-06-28 21:59:54 +00:00
__filename__ = "inbox.py"
__author__ = "Bob Mottram"
__license__ = "AGPL3+"
__version__ = "0.0.1"
__maintainer__ = "Bob Mottram"
__email__ = "bob@freedombone.net"
__status__ = "Production"
import json
import os
2019-06-29 10:08:59 +00:00
import datetime
2019-07-04 12:23:53 +00:00
import time
import json
import commentjson
from shutil import copyfile
2019-07-02 10:39:55 +00:00
from utils import urlPermitted
2019-07-04 10:02:56 +00:00
from utils import createInboxQueueDir
2019-07-06 13:49:25 +00:00
from utils import getStatusNumber
2019-07-04 12:23:53 +00:00
from httpsig import verifyPostHeaders
from session import createSession
2019-07-04 19:34:28 +00:00
from session import getJson
2019-07-04 12:23:53 +00:00
from follow import receiveFollowRequest
2019-07-04 14:36:29 +00:00
from pprint import pprint
2019-07-04 19:34:28 +00:00
from cache import getPersonFromCache
2019-07-04 20:25:19 +00:00
from cache import storePersonInCache
2019-07-06 15:17:21 +00:00
from acceptreject import receiveAcceptReject
2019-07-04 19:34:28 +00:00
def getPersonPubKey(session,personUrl: str,personCache: {},debug: bool) -> str:
if not personUrl:
return None
personUrl=personUrl.replace('#main-key','')
personJson = getPersonFromCache(personUrl,personCache)
if not personJson:
if debug:
print('DEBUG: Obtaining public key for '+personUrl)
2019-07-04 20:25:19 +00:00
asHeader = {'Accept': 'application/ld+json; profile="https://www.w3.org/ns/activitystreams"'}
2019-07-04 19:34:28 +00:00
personJson = getJson(session,personUrl,asHeader,None)
if not personJson:
return None
pubKey=None
if personJson.get('publicKey'):
if personJson['publicKey'].get('publicKeyPem'):
pubKey=personJson['publicKey']['publicKeyPem']
else:
if personJson.get('publicKeyPem'):
pubKey=personJson['publicKeyPem']
if not pubKey:
if debug:
print('DEBUG: Public key not found for '+personUrl)
storePersonInCache(personUrl,personJson,personCache)
return pubKey
2019-06-28 21:59:54 +00:00
2019-07-02 15:07:27 +00:00
def inboxMessageHasParams(messageJson: {}) -> bool:
"""Checks whether an incoming message contains expected parameters
"""
2019-07-06 13:49:25 +00:00
expectedParams=['type','actor','object']
2019-07-02 15:07:27 +00:00
for param in expectedParams:
if not messageJson.get(param):
return False
2019-07-06 13:49:25 +00:00
if not messageJson.get('to'):
allowedWithoutToParam=['Follow','Request','Capability']
if messageJson['type'] not in allowedWithoutToParam:
return False
2019-07-02 15:07:27 +00:00
return True
2019-07-06 10:33:57 +00:00
def inboxPermittedMessage(domain: str,messageJson: {},federationList: [],capsList: []) -> bool:
2019-06-28 21:59:54 +00:00
""" check that we are receiving from a permitted domain
"""
testParam='actor'
if not messageJson.get(testParam):
return False
actor=messageJson[testParam]
# always allow the local domain
2019-07-01 11:48:54 +00:00
if domain in actor:
2019-06-28 21:59:54 +00:00
return True
2019-07-06 10:33:57 +00:00
if not urlPermitted(actor,federationList,capsList,"inbox:write"):
2019-06-28 21:59:54 +00:00
return False
2019-07-06 13:49:25 +00:00
if messageJson['type']!='Follow':
if messageJson.get('object'):
if messageJson['object'].get('inReplyTo'):
inReplyTo=messageJson['object']['inReplyTo']
if not urlPermitted(inReplyTo,federationList,capsList):
return False
2019-06-28 21:59:54 +00:00
return True
2019-06-29 10:08:59 +00:00
2019-07-02 20:54:22 +00:00
def validPublishedDate(published) -> bool:
2019-06-29 10:08:59 +00:00
currTime=datetime.datetime.utcnow()
pubDate=datetime.datetime.strptime(published,"%Y-%m-%dT%H:%M:%SZ")
daysSincePublished = (currTime - pubTime).days
if daysSincePublished>30:
return False
return True
2019-07-04 10:02:56 +00:00
2019-07-06 13:49:25 +00:00
def savePostToInboxQueue(baseDir: str,httpPrefix: str,nickname: str, domain: str,postJson: {},host: str,headers: str,postPath: str,debug: bool) -> str:
2019-07-04 10:02:56 +00:00
"""Saves the give json to the inbox queue for the person
keyId specifies the actor sending the post
"""
if ':' in domain:
domain=domain.split(':')[0]
2019-07-06 13:49:25 +00:00
if postJson.get('id'):
postId=postJson['id'].replace('/activity','')
else:
statusNumber,published = getStatusNumber()
postId=httpPrefix+'://'+domain+'/users/'+nickname+'/statuses/'+statusNumber
2019-07-04 10:09:27 +00:00
currTime=datetime.datetime.utcnow()
published=currTime.strftime("%Y-%m-%dT%H:%M:%SZ")
2019-07-05 11:27:18 +00:00
inboxQueueDir=createInboxQueueDir(nickname,domain,baseDir)
handle=nickname+'@'+domain
destination=baseDir+'/accounts/'+handle+'/inbox/'+postId.replace('/','#')+'.json'
if os.path.isfile(destination):
2019-07-06 13:49:25 +00:00
if debug:
print('DEBUG: inbox item already exists')
return None
filename=inboxQueueDir+'/'+postId.replace('/','#')+'.json'
sharedInboxItem=False
if nickname=='sharedinbox':
sharedInboxItem=True
2019-07-04 14:36:29 +00:00
newQueueItem = {
'sharedInbox': sharedInboxItem,
2019-07-04 10:09:27 +00:00
'published': published,
2019-07-04 20:25:19 +00:00
'host': host,
2019-07-04 12:23:53 +00:00
'headers': headers,
2019-07-05 22:13:20 +00:00
'path': postPath,
'post': postJson,
'filename': filename,
'destination': destination
2019-07-04 10:02:56 +00:00
}
2019-07-06 13:49:25 +00:00
if debug:
print('Inbox queue item created')
pprint(newQueueItem)
2019-07-04 10:02:56 +00:00
with open(filename, 'w') as fp:
commentjson.dump(newQueueItem, fp, indent=4, sort_keys=False)
return filename
2019-07-04 12:23:53 +00:00
2019-07-06 13:49:25 +00:00
def runInboxQueue(baseDir: str,httpPrefix: str,sendThreads: [],postLog: [],cachedWebfingers: {},personCache: {},queue: [],domain: str,port: int,useTor: bool,federationList: [],capsList: [],debug: bool) -> None:
2019-07-04 12:23:53 +00:00
"""Processes received items and moves them to
the appropriate directories
"""
currSessionTime=int(time.time())
sessionLastUpdate=currSessionTime
session=createSession(domain,port,useTor)
if debug:
print('DEBUG: Inbox queue running')
while True:
if len(queue)>0:
currSessionTime=int(time.time())
if currSessionTime-sessionLastUpdate>1200:
session=createSession(domain,port,useTor)
sessionLastUpdate=currSessionTime
# oldest item first
queue.sort()
queueFilename=queue[0]
if not os.path.isfile(queueFilename):
if debug:
print("DEBUG: queue item rejected becase it has no file: "+queueFilename)
queue.pop(0)
continue
# Load the queue json
with open(queueFilename, 'r') as fp:
queueJson=commentjson.load(fp)
2019-07-04 19:34:28 +00:00
# Try a few times to obtain the public key
2019-07-04 12:23:53 +00:00
pubKey=None
keyId=None
2019-07-04 17:31:41 +00:00
for tries in range(8):
2019-07-04 14:36:29 +00:00
keyId=None
signatureParams=queueJson['headers'].split(',')
for signatureItem in signatureParams:
if signatureItem.startswith('keyId='):
if '"' in signatureItem:
keyId=signatureItem.split('"')[1]
break
if not keyId:
if debug:
print('DEBUG: No keyId in signature: '+queueJson['headers']['signature'])
os.remove(queueFilename)
queue.pop(0)
continue
pubKey=getPersonPubKey(session,keyId,personCache,debug)
2019-07-04 17:31:41 +00:00
if pubKey:
print('DEBUG: public key: '+str(pubKey))
break
if debug:
print('DEBUG: Retry '+str(tries+1)+' obtaining public key for '+keyId)
time.sleep(5)
2019-07-04 12:23:53 +00:00
if not pubKey:
if debug:
2019-07-04 17:31:41 +00:00
print('DEBUG: public key could not be obtained from '+keyId)
2019-07-04 12:23:53 +00:00
os.remove(queueFilename)
queue.pop(0)
continue
# check the signature
2019-07-04 20:25:19 +00:00
verifyHeaders={
'host': queueJson['host'],
'signature': queueJson['headers']
2019-07-05 22:13:20 +00:00
}
2019-07-04 12:23:53 +00:00
if not verifyPostHeaders(httpPrefix, \
2019-07-04 20:25:19 +00:00
pubKey, verifyHeaders, \
2019-07-05 22:13:20 +00:00
queueJson['path'], False, \
2019-07-04 20:25:19 +00:00
json.dumps(queueJson['post'])):
2019-07-04 12:23:53 +00:00
if debug:
print('DEBUG: Header signature check failed')
os.remove(queueFilename)
queue.pop(0)
continue
2019-07-04 17:31:41 +00:00
if debug:
print('DEBUG: Signature check success')
2019-07-05 18:57:19 +00:00
if receiveFollowRequest(session, \
baseDir,httpPrefix,port, \
sendThreads,postLog, \
cachedWebfingers,
personCache,
2019-07-04 20:25:19 +00:00
queueJson['post'], \
2019-07-06 13:49:25 +00:00
federationList,capsList, \
debug):
2019-07-04 12:23:53 +00:00
if debug:
2019-07-04 17:31:41 +00:00
print('DEBUG: Follow accepted from '+keyId)
2019-07-05 18:57:19 +00:00
os.remove(queueFilename)
queue.pop(0)
continue
2019-07-06 15:17:21 +00:00
if receiveAcceptReject(session, \
2019-07-06 19:24:52 +00:00
baseDir,httpPrefix,domain,port, \
2019-07-06 15:17:21 +00:00
sendThreads,postLog, \
cachedWebfingers,
personCache,
queueJson['post'], \
federationList,capsList, \
debug):
if debug:
print('DEBUG: Accept/Reject received from '+keyId)
os.remove(queueFilename)
queue.pop(0)
continue
2019-07-04 12:23:53 +00:00
if debug:
print('DEBUG: Queue post accepted')
2019-07-04 17:31:41 +00:00
if queueJson['sharedInbox']:
if '/users/' in keyId:
# Who is this from? Use the actor from the keyId where we obtained the public key
fromList=keyId.replace('https://','').replace('http://','').replace('dat://','').replace('#main-key','').split('/users/')
fromNickname=fromList[1]
fromDomain=fromList[0]
# get the followers of the sender
followList=getFollowersOfPerson(baseDir,fromNickname,fromDomain)
for followerHandle in followList:
followerDir=baseDir+'/accounts/'+followerHandle
if os.path.isdir(followerDir):
if not os.path.isdir(followerDir+'/inbox'):
os.mkdir(followerDir+'/inbox')
postId=queueJson['post']['id'].replace('/activity','')
destination=followerDir+'/inbox/'+postId.replace('/','#')+'.json'
if os.path.isfile(destination):
# post already exists in this person's inbox
continue
# We could do this in a more storage space efficient way
# by linking to the inbox of sharedinbox@domain
# However, this allows for easy deletion by individuals
# without affecting any other people
copyfile(queueFilename, destination)
# copy to followers
# remove item from shared inbox
os.remove(queueFilename)
else:
# move to the destination inbox
os.rename(queueFilename,queueJson['destination'])
2019-07-04 12:23:53 +00:00
queue.pop(0)
time.sleep(2)