indymedia
/
epicyon
mirror of https://gitlab.com/bashrc2/epicyon
7
4
Fork 1
Code Issues 14 Releases Wiki Activity

flake8 style

merge-requests/30/head
Bob Mottram 2020-04-01 19:11:39 +00:00
parent ae5b9f579f
commit 65d3ace372
1 changed files with 260 additions and 237 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

495
announce.py
View File

@ -1,14 +1,12 @@
__filename__="announce.py" __filename__ = "announce.py"
__author__="Bob Mottram" __author__ = "Bob Mottram"
__license__="AGPL3+" __license__ = "AGPL3+"
__version__="1.1.0" __version__ = "1.1.0"
__maintainer__="Bob Mottram" __maintainer__ = "Bob Mottram"
__email__="bob@freedombone.net" __email__ = "bob@freedombone.net"
__status__="Production" __status__ = "Production"
import os import os
import time
import json
from pprint import pprint from pprint import pprint
from utils import removePostFromCache from utils import removePostFromCache
from utils import getStatusNumber from utils import getStatusNumber
@ -26,8 +24,9 @@ from session import postJson
from webfinger import webfingerHandle from webfinger import webfingerHandle
from auth import createBasicAuthHeader from auth import createBasicAuthHeader
def outboxAnnounce(recentPostsCache: {}, \
baseDir: str,messageJson: {},debug: bool) -> bool: def outboxAnnounce(recentPostsCache: {},
baseDir: str, messageJson: {}, debug: bool) -> bool:
""" Adds or removes announce entries from the shares collection """ Adds or removes announce entries from the shares collection
within a given post within a given post
""" """
@ -37,66 +36,67 @@ def outboxAnnounce(recentPostsCache: {}, \
return False return False
if not messageJson.get('object'): if not messageJson.get('object'):
return False return False
if messageJson['type']=='Announce': if messageJson['type'] == 'Announce':
if not isinstance(messageJson['object'], str): if not isinstance(messageJson['object'], str):
return False return False
nickname=getNicknameFromActor(messageJson['actor']) nickname = getNicknameFromActor(messageJson['actor'])
if not nickname: if not nickname:
print('WARN: no nickname found in '+messageJson['actor']) print('WARN: no nickname found in '+messageJson['actor'])
return False return False
domain,port=getDomainFromActor(messageJson['actor']) domain, port = getDomainFromActor(messageJson['actor'])
postFilename=locatePost(baseDir,nickname,domain,messageJson['object']) postFilename = locatePost(baseDir, nickname, domain,
messageJson['object'])
if postFilename: if postFilename:
updateAnnounceCollection(recentPostsCache,baseDir,postFilename, \ updateAnnounceCollection(recentPostsCache, baseDir, postFilename,
messageJson['actor'],domain,debug) messageJson['actor'], domain, debug)
return True return True
if messageJson['type']=='Undo': if messageJson['type'] == 'Undo':
if not isinstance(messageJson['object'], dict): if not isinstance(messageJson['object'], dict):
return False return False
if not messageJson['object'].get('type'): if not messageJson['object'].get('type'):
return False return False
if messageJson['object']['type']=='Announce': if messageJson['object']['type'] == 'Announce':
if not isinstance(messageJson['object']['object'], str): if not isinstance(messageJson['object']['object'], str):
return False return False
nickname=getNicknameFromActor(messageJson['actor']) nickname = getNicknameFromActor(messageJson['actor'])
if not nickname: if not nickname:
print('WARN: no nickname found in '+messageJson['actor']) print('WARN: no nickname found in ' + messageJson['actor'])
return False return False
domain,port=getDomainFromActor(messageJson['actor']) domain, port = getDomainFromActor(messageJson['actor'])
postFilename= \ postFilename = locatePost(baseDir, nickname, domain,
locatePost(baseDir,nickname,domain, \
messageJson['object']['object']) messageJson['object']['object'])
if postFilename: if postFilename:
undoAnnounceCollectionEntry(recentPostsCache, \ undoAnnounceCollectionEntry(recentPostsCache,
baseDir,postFilename, \ baseDir, postFilename,
messageJson['actor'], \ messageJson['actor'],
domain,debug) domain, debug)
return True return True
return False return False
def undoAnnounceCollectionEntry(recentPostsCache: {}, \
baseDir: str,postFilename: str, \ def undoAnnounceCollectionEntry(recentPostsCache: {},
actor: str,domain: str,debug: bool) -> None: baseDir: str, postFilename: str,
actor: str, domain: str, debug: bool) -> None:
"""Undoes an announce for a particular actor by removing it from """Undoes an announce for a particular actor by removing it from
the "shares" collection within a post. Note that the "shares" the "shares" collection within a post. Note that the "shares"
collection has no relation to shared items in shares.py. It's collection has no relation to shared items in shares.py. It's
shares of posts, not shares of physical objects. shares of posts, not shares of physical objects.
""" """
postJsonObject=loadJson(postFilename) postJsonObject = loadJson(postFilename)
if postJsonObject: if postJsonObject:
# remove any cached version of this announce so that the announce # remove any cached version of this announce so that the announce
# icon is changed # icon is changed
nickname=getNicknameFromActor(actor) nickname = getNicknameFromActor(actor)
cachedPostFilename= \ cachedPostFilename = getCachedPostFilename(baseDir, nickname, domain,
getCachedPostFilename(baseDir,nickname,domain,postJsonObject) postJsonObject)
if cachedPostFilename: if cachedPostFilename:
if os.path.isfile(cachedPostFilename): if os.path.isfile(cachedPostFilename):
os.remove(cachedPostFilename) os.remove(cachedPostFilename)
removePostFromCache(postJsonObject,recentPostsCache) removePostFromCache(postJsonObject, recentPostsCache)
if not postJsonObject.get('type'): if not postJsonObject.get('type'):
return return
if postJsonObject['type']!='Create': if postJsonObject['type'] != 'Create':
return return
if not postJsonObject.get('object'): if not postJsonObject.get('object'):
if debug: if debug:
@ -109,61 +109,65 @@ def undoAnnounceCollectionEntry(recentPostsCache: {}, \
return return
if not postJsonObject['object']['shares'].get('items'): if not postJsonObject['object']['shares'].get('items'):
return return
totalItems=0 totalItems = 0
if postJsonObject['object']['shares'].get('totalItems'): if postJsonObject['object']['shares'].get('totalItems'):
totalItems=postJsonObject['object']['shares']['totalItems'] totalItems = postJsonObject['object']['shares']['totalItems']
itemFound=False itemFound = False
for announceItem in postJsonObject['object']['shares']['items']: for announceItem in postJsonObject['object']['shares']['items']:
if announceItem.get('actor'): if announceItem.get('actor'):
if announceItem['actor']==actor: if announceItem['actor'] == actor:
if debug: if debug:
print('DEBUG: Announce was removed for '+actor) print('DEBUG: Announce was removed for ' + actor)
postJsonObject['object']['shares']['items'].remove(announceItem) anIt = announceItem
itemFound=True postJsonObject['object']['shares']['items'].remove(anIt)
itemFound = True
break break
if itemFound: if itemFound:
if totalItems==1: if totalItems == 1:
if debug: if debug:
print('DEBUG: shares (announcements) was removed from post') print('DEBUG: shares (announcements) ' +
'was removed from post')
del postJsonObject['object']['shares'] del postJsonObject['object']['shares']
else: else:
postJsonObject['object']['shares']['totalItems']= \ itlen = len(postJsonObject['object']['shares']['items'])
len(postJsonObject['object']['shares']['items']) postJsonObject['object']['shares']['totalItems'] = itlen
saveJson(postJsonObject,postFilename)
def updateAnnounceCollection(recentPostsCache: {}, \ saveJson(postJsonObject, postFilename)
baseDir: str,postFilename: str, \
actor: str,domain: str,debug: bool) -> None:
def updateAnnounceCollection(recentPostsCache: {},
baseDir: str, postFilename: str,
actor: str, domain: str, debug: bool) -> None:
"""Updates the announcements collection within a post """Updates the announcements collection within a post
Confusingly this is known as "shares", but isn't the Confusingly this is known as "shares", but isn't the
same as shared items within shares.py same as shared items within shares.py
It's shares of posts, not shares of physical objects. It's shares of posts, not shares of physical objects.
""" """
postJsonObject=loadJson(postFilename) postJsonObject = loadJson(postFilename)
if postJsonObject: if postJsonObject:
# remove any cached version of this announce so that the announce # remove any cached version of this announce so that the announce
# icon is changed # icon is changed
nickname=getNicknameFromActor(actor) nickname = getNicknameFromActor(actor)
cachedPostFilename= \ cachedPostFilename = getCachedPostFilename(baseDir, nickname, domain,
getCachedPostFilename(baseDir,nickname,domain,postJsonObject) postJsonObject)
if cachedPostFilename: if cachedPostFilename:
if os.path.isfile(cachedPostFilename): if os.path.isfile(cachedPostFilename):
os.remove(cachedPostFilename) os.remove(cachedPostFilename)
removePostFromCache(postJsonObject,recentPostsCache) removePostFromCache(postJsonObject, recentPostsCache)
if not postJsonObject.get('object'): if not postJsonObject.get('object'):
if debug: if debug:
pprint(postJsonObject) pprint(postJsonObject)
print('DEBUG: post '+announceUrl+' has no object') print('DEBUG: post ' + postFilename + ' has no object')
return return
if not isinstance(postJsonObject['object'], dict): if not isinstance(postJsonObject['object'], dict):
return return
postUrl=postJsonObject['id'].replace('/activity','')+'/shares' postUrl = postJsonObject['id'].replace('/activity', '') + '/shares'
if not postJsonObject['object'].get('shares'): if not postJsonObject['object'].get('shares'):
if debug: if debug:
print('DEBUG: Adding initial shares (announcements) to '+ \ print('DEBUG: Adding initial shares (announcements) to ' +
postUrl) postUrl)
announcementsJson={ announcementsJson = {
"@context": "https://www.w3.org/ns/activitystreams", "@context": "https://www.w3.org/ns/activitystreams",
'id': postUrl, 'id': postUrl,
'type': 'Collection', 'type': 'Collection',
@ -173,30 +177,33 @@ def updateAnnounceCollection(recentPostsCache: {}, \
'actor': actor 'actor': actor
}] }]
} }
postJsonObject['object']['shares']=announcementsJson postJsonObject['object']['shares'] = announcementsJson
else: else:
if postJsonObject['object']['shares'].get('items'): if postJsonObject['object']['shares'].get('items'):
for announceItem in postJsonObject['object']['shares']['items']: sharesItems = postJsonObject['object']['shares']['items']
for announceItem in sharesItems:
if announceItem.get('actor'): if announceItem.get('actor'):
if announceItem['actor']==actor: if announceItem['actor'] == actor:
return return
newAnnounce={ newAnnounce = {
'type': 'Announce', 'type': 'Announce',
'actor': actor 'actor': actor
} }
postJsonObject['object']['shares']['items'].append(newAnnounce) postJsonObject['object']['shares']['items'].append(newAnnounce)
postJsonObject['object']['shares']['totalItems']= \ itlen = len(postJsonObject['object']['shares']['items'])
len(postJsonObject['object']['shares']['items']) postJsonObject['object']['shares']['totalItems'] = itlen
else: else:
if debug: if debug:
print('DEBUG: shares (announcements) section of post has no items list') print('DEBUG: shares (announcements) section of post ' +
'has no items list')
if debug: if debug:
print('DEBUG: saving post with shares (announcements) added') print('DEBUG: saving post with shares (announcements) added')
pprint(postJsonObject) pprint(postJsonObject)
saveJson(postJsonObject,postFilename) saveJson(postJsonObject, postFilename)
def announcedByPerson(postJsonObject: {},nickname: str,domain: str) -> bool:
def announcedByPerson(postJsonObject: {}, nickname: str, domain: str) -> bool:
"""Returns True if the given post is announced by the given person """Returns True if the given post is announced by the given person
""" """
if not postJsonObject.get('object'): if not postJsonObject.get('object'):
@ -206,44 +213,47 @@ def announcedByPerson(postJsonObject: {},nickname: str,domain: str) -> bool:
# not to be confused with shared items # not to be confused with shared items
if not postJsonObject['object'].get('shares'): if not postJsonObject['object'].get('shares'):
return False return False
actorMatch=domain+'/users/'+nickname actorMatch = domain + '/users/' + nickname
for item in postJsonObject['object']['shares']['items']: for item in postJsonObject['object']['shares']['items']:
if item['actor'].endswith(actorMatch): if item['actor'].endswith(actorMatch):
return True return True
return False return False
def createAnnounce(session,baseDir: str,federationList: [], \
nickname: str,domain: str,port: int, \ def createAnnounce(session, baseDir: str, federationList: [],
toUrl: str,ccUrl: str,httpPrefix: str, \ nickname: str, domain: str, port: int,
objectUrl: str,saveToFile: bool, \ toUrl: str, ccUrl: str, httpPrefix: str,
clientToServer: bool, \ objectUrl: str, saveToFile: bool,
sendThreads: [],postLog: [], \ clientToServer: bool,
personCache: {},cachedWebfingers: {}, \ sendThreads: [], postLog: [],
debug: bool,projectVersion: str) -> {}: personCache: {}, cachedWebfingers: {},
debug: bool, projectVersion: str) -> {}:
"""Creates an announce message """Creates an announce message
Typically toUrl will be https://www.w3.org/ns/activitystreams#Public Typically toUrl will be https://www.w3.org/ns/activitystreams#Public
and ccUrl might be a specific person favorited or repeated and the and ccUrl might be a specific person favorited or repeated and the
followers url objectUrl is typically the url of the message, followers url objectUrl is typically the url of the message,
corresponding to url or atomUri in createPostBase corresponding to url or atomUri in createPostBase
""" """
if not urlPermitted(objectUrl,federationList,"inbox:write"): if not urlPermitted(objectUrl, federationList, "inbox:write"):
return None return None
if ':' in domain: if ':' in domain:
domain=domain.split(':')[0] domain = domain.split(':')[0]
fullDomain=domain fullDomain = domain
if port: if port:
if port!=80 and port!=443: if port != 80 and port != 443:
if ':' not in domain: if ':' not in domain:
fullDomain=domain+':'+str(port) fullDomain = domain + ':' + str(port)
statusNumber,published=getStatusNumber() statusNumber, published = getStatusNumber()
newAnnounceId= \ newAnnounceId = httpPrefix + '://' + fullDomain + \
httpPrefix+'://'+fullDomain+'/users/'+nickname+'/statuses/'+statusNumber '/users/' + nickname + '/statuses/' + statusNumber
newAnnounce={ atomUriStr = httpPrefix + '://' + fullDomain + '/users/' + nickname + \
'/statuses/' + statusNumber
newAnnounce = {
"@context": "https://www.w3.org/ns/activitystreams", "@context": "https://www.w3.org/ns/activitystreams",
'actor': httpPrefix+'://'+fullDomain+'/users/'+nickname, 'actor': httpPrefix+'://'+fullDomain+'/users/'+nickname,
'atomUri': httpPrefix+'://'+fullDomain+'/users/'+nickname+'/statuses/'+statusNumber, 'atomUri': atomUriStr,
'cc': [], 'cc': [],
'id': newAnnounceId+'/activity', 'id': newAnnounceId+'/activity',
'object': objectUrl, 'object': objectUrl,
@ -252,89 +262,93 @@ def createAnnounce(session,baseDir: str,federationList: [], \
'type': 'Announce' 'type': 'Announce'
} }
if ccUrl: if ccUrl:
if len(ccUrl)>0: if len(ccUrl) > 0:
newAnnounce['cc']=[ccUrl] newAnnounce['cc'] = [ccUrl]
if saveToFile: if saveToFile:
outboxDir=createOutboxDir(nickname,domain,baseDir) outboxDir = createOutboxDir(nickname, domain, baseDir)
filename=outboxDir+'/'+newAnnounceId.replace('/','#')+'.json' filename = outboxDir + '/' + newAnnounceId.replace('/', '#') + '.json'
saveJson(newAnnounce,filename) saveJson(newAnnounce, filename)
announceNickname=None announceNickname = None
announceDomain=None announceDomain = None
announcePort=None announcePort = None
if '/users/' in objectUrl or \ if '/users/' in objectUrl or \
'/channel/' in objectUrl or \ '/channel/' in objectUrl or \
'/profile/' in objectUrl: '/profile/' in objectUrl:
announceNickname=getNicknameFromActor(objectUrl) announceNickname = getNicknameFromActor(objectUrl)
announceDomain,announcePort=getDomainFromActor(objectUrl) announceDomain, announcePort = getDomainFromActor(objectUrl)
if announceNickname and announceDomain: if announceNickname and announceDomain:
sendSignedJson(newAnnounce,session,baseDir, \ sendSignedJson(newAnnounce, session, baseDir,
nickname,domain,port, \ nickname, domain, port,
announceNickname,announceDomain,announcePort,None, \ announceNickname, announceDomain, announcePort, None,
httpPrefix,True,clientToServer,federationList, \ httpPrefix, True, clientToServer, federationList,
sendThreads,postLog,cachedWebfingers,personCache, \ sendThreads, postLog, cachedWebfingers, personCache,
debug,projectVersion) debug, projectVersion)
return newAnnounce return newAnnounce
def announcePublic(session,baseDir: str,federationList: [], \
nickname: str,domain: str,port: int,httpPrefix: str, \ def announcePublic(session, baseDir: str, federationList: [],
objectUrl: str,clientToServer: bool, \ nickname: str, domain: str, port: int, httpPrefix: str,
sendThreads: [],postLog: [], \ objectUrl: str, clientToServer: bool,
personCache: {},cachedWebfingers: {}, \ sendThreads: [], postLog: [],
debug: bool,projectVersion: str) -> {}: personCache: {}, cachedWebfingers: {},
debug: bool, projectVersion: str) -> {}:
"""Makes a public announcement """Makes a public announcement
""" """
fromDomain=domain fromDomain = domain
if port: if port:
if port!=80 and port!=443: if port != 80 and port != 443:
if ':' not in domain: if ':' not in domain:
fromDomain=domain+':'+str(port) fromDomain = domain + ':' + str(port)
toUrl='https://www.w3.org/ns/activitystreams#Public' toUrl = 'https://www.w3.org/ns/activitystreams#Public'
ccUrl=httpPrefix+'://'+fromDomain+'/users/'+nickname+'/followers' ccUrl = httpPrefix + '://' + fromDomain + '/users/' + nickname + \
return createAnnounce(session,baseDir,federationList, \ '/followers'
nickname,domain,port, \ return createAnnounce(session, baseDir, federationList,
toUrl,ccUrl,httpPrefix, \ nickname, domain, port,
objectUrl,True,clientToServer, \ toUrl, ccUrl, httpPrefix,
sendThreads,postLog, \ objectUrl, True, clientToServer,
personCache,cachedWebfingers, \ sendThreads, postLog,
debug,projectVersion) personCache, cachedWebfingers,
debug, projectVersion)
def repeatPost(session,baseDir: str,federationList: [], \
nickname: str, domain: str, port: int, httpPrefix: str, \ def repeatPost(session, baseDir: str, federationList: [],
announceNickname: str, announceDomain: str, \ nickname: str, domain: str, port: int, httpPrefix: str,
announcePort: int, announceHttpsPrefix: str, \ announceNickname: str, announceDomain: str,
announceStatusNumber: int,clientToServer: bool, \ announcePort: int, announceHttpsPrefix: str,
sendThreads: [],postLog: [], \ announceStatusNumber: int, clientToServer: bool,
personCache: {},cachedWebfingers: {}, \ sendThreads: [], postLog: [],
debug: bool,projectVersion: str) -> {}: personCache: {}, cachedWebfingers: {},
debug: bool, projectVersion: str) -> {}:
"""Repeats a given status post """Repeats a given status post
""" """
announcedDomain=announceDomain announcedDomain = announceDomain
if announcePort: if announcePort:
if announcePort!=80 and announcePort!=443: if announcePort != 80 and announcePort != 443:
if ':' not in announcedDomain: if ':' not in announcedDomain:
announcedDomain=announcedDomain+':'+str(announcePort) announcedDomain = announcedDomain + ':' + str(announcePort)
objectUrl=announceHttpsPrefix+'://'+announcedDomain+'/users/'+ \ objectUrl = announceHttpsPrefix + '://' + announcedDomain + '/users/' + \
announceNickname+'/statuses/'+str(announceStatusNumber) announceNickname + '/statuses/' + str(announceStatusNumber)
return announcePublic(session,baseDir,federationList, \ return announcePublic(session, baseDir, federationList,
nickname,domain,port,httpPrefix, \ nickname, domain, port, httpPrefix,
objectUrl,clientToServer, \ objectUrl, clientToServer,
sendThreads,postLog, \ sendThreads, postLog,
personCache,cachedWebfingers, \ personCache, cachedWebfingers,
debug,projectVersion) debug, projectVersion)
def undoAnnounce(session,baseDir: str,federationList: [], \
nickname: str,domain: str,port: int, \ def undoAnnounce(session, baseDir: str, federationList: [],
toUrl: str,ccUrl: str,httpPrefix: str, \ nickname: str, domain: str, port: int,
objectUrl: str,saveToFile: bool, \ toUrl: str, ccUrl: str, httpPrefix: str,
clientToServer: bool, \ objectUrl: str, saveToFile: bool,
sendThreads: [],postLog: [], \ clientToServer: bool,
personCache: {},cachedWebfingers: {}, \ sendThreads: [], postLog: [],
personCache: {}, cachedWebfingers: {},
debug: bool) -> {}: debug: bool) -> {}:
"""Undoes an announce message """Undoes an announce message
Typically toUrl will be https://www.w3.org/ns/activitystreams#Public Typically toUrl will be https://www.w3.org/ns/activitystreams#Public
@ -342,18 +356,18 @@ def undoAnnounce(session,baseDir: str,federationList: [], \
objectUrl is typically the url of the message which was repeated, objectUrl is typically the url of the message which was repeated,
corresponding to url or atomUri in createPostBase corresponding to url or atomUri in createPostBase
""" """
if not urlPermitted(objectUrl,federationList,"inbox:write"): if not urlPermitted(objectUrl, federationList, "inbox:write"):
return None return None
if ':' in domain: if ':' in domain:
domain=domain.split(':')[0] domain = domain.split(':')[0]
fullDomain=domain fullDomain = domain
if port: if port:
if port!=80 and port!=443: if port != 80 and port != 443:
if ':' not in domain: if ':' not in domain:
fullDomain=domain+':'+str(port) fullDomain = domain + ':' + str(port)
newUndoAnnounce={ newUndoAnnounce = {
"@context": "https://www.w3.org/ns/activitystreams", "@context": "https://www.w3.org/ns/activitystreams",
'actor': httpPrefix+'://'+fullDomain+'/users/'+nickname, 'actor': httpPrefix+'://'+fullDomain+'/users/'+nickname,
'type': 'Undo', 'type': 'Undo',
@ -368,106 +382,109 @@ def undoAnnounce(session,baseDir: str,federationList: [], \
} }
} }
if ccUrl: if ccUrl:
if len(ccUrl)>0: if len(ccUrl) > 0:
newUndoAnnounce['object']['cc']=[ccUrl] newUndoAnnounce['object']['cc'] = [ccUrl]
announceNickname=None announceNickname = None
announceDomain=None announceDomain = None
announcePort=None announcePort = None
if '/users/' in objectUrl or \ if '/users/' in objectUrl or \
'/channel/' in objectUrl or \ '/channel/' in objectUrl or \
'/profile/' in objectUrl: '/profile/' in objectUrl:
announceNickname=getNicknameFromActor(objectUrl) announceNickname = getNicknameFromActor(objectUrl)
announceDomain,announcePort=getDomainFromActor(objectUrl) announceDomain, announcePort = getDomainFromActor(objectUrl)
if announceNickname and announceDomain: if announceNickname and announceDomain:
sendSignedJson(newUndoAnnounce,session,baseDir, \ sendSignedJson(newUndoAnnounce, session, baseDir,
nickname,domain,port, \ nickname, domain, port,
announceNickname,announceDomain,announcePort, \ announceNickname, announceDomain, announcePort,
'https://www.w3.org/ns/activitystreams#Public', \ 'https://www.w3.org/ns/activitystreams#Public',
httpPrefix,True,clientToServer,federationList, \ httpPrefix, True, clientToServer, federationList,
sendThreads,postLog,cachedWebfingers,personCache,debug) sendThreads, postLog, cachedWebfingers,
personCache, debug)
return newUndoAnnounce return newUndoAnnounce
def undoAnnouncePublic(session,baseDir: str,federationList: [], \
nickname: str,domain: str,port: int,httpPrefix: str, \ def undoAnnouncePublic(session, baseDir: str, federationList: [],
objectUrl: str,clientToServer: bool, \ nickname: str, domain: str, port: int, httpPrefix: str,
sendThreads: [],postLog: [], \ objectUrl: str, clientToServer: bool,
personCache: {},cachedWebfingers: {}, \ sendThreads: [], postLog: [],
personCache: {}, cachedWebfingers: {},
debug: bool) -> {}: debug: bool) -> {}:
"""Undoes a public announcement """Undoes a public announcement
""" """
fromDomain=domain fromDomain = domain
if port: if port:
if port!=80 and port!=443: if port != 80 and port != 443:
if ':' not in domain: if ':' not in domain:
fromDomain=domain+':'+str(port) fromDomain = domain + ':' + str(port)
toUrl='https://www.w3.org/ns/activitystreams#Public' toUrl = 'https://www.w3.org/ns/activitystreams#Public'
ccUrl=httpPrefix+'://'+fromDomain+'/users/'+nickname+'/followers' ccUrl = httpPrefix + '://' + fromDomain + '/users/' + nickname + \
return undoAnnounce(session,baseDir,federationList, \ '/followers'
nickname,domain,port, \ return undoAnnounce(session, baseDir, federationList,
toUrl,ccUrl,httpPrefix, \ nickname, domain, port,
objectUrl,True,clientToServer, \ toUrl, ccUrl, httpPrefix,
sendThreads,postLog, \ objectUrl, True, clientToServer,
personCache,cachedWebfingers, \ sendThreads, postLog,
personCache, cachedWebfingers,
debug) debug)
def undoRepeatPost(session,baseDir: str,federationList: [], \
nickname: str,domain: str,port: int,httpPrefix: str, \ def undoRepeatPost(session, baseDir: str, federationList: [],
announceNickname: str,announceDomain: str, \ nickname: str, domain: str, port: int, httpPrefix: str,
announcePort: int,announceHttpsPrefix: str, \ announceNickname: str, announceDomain: str,
announceStatusNumber: int,clientToServer: bool, \ announcePort: int, announceHttpsPrefix: str,
sendThreads: [],postLog: [], \ announceStatusNumber: int, clientToServer: bool,
personCache: {},cachedWebfingers: {}, \ sendThreads: [], postLog: [],
personCache: {}, cachedWebfingers: {},
debug: bool) -> {}: debug: bool) -> {}:
"""Undoes a status post repeat """Undoes a status post repeat
""" """
announcedDomain=announceDomain announcedDomain = announceDomain
if announcePort: if announcePort:
if announcePort!=80 and announcePort!=443: if announcePort != 80 and announcePort != 443:
if ':' not in announcedDomain: if ':' not in announcedDomain:
announcedDomain=announcedDomain+':'+str(announcePort) announcedDomain = announcedDomain + ':' + str(announcePort)
objectUrl=announceHttpsPrefix+'://'+announcedDomain+'/users/'+ \ objectUrl = announceHttpsPrefix + '://' + announcedDomain + '/users/' + \
announceNickname+'/statuses/'+str(announceStatusNumber) announceNickname + '/statuses/' + str(announceStatusNumber)
return undoAnnouncePublic(session,baseDir,federationList, \ return undoAnnouncePublic(session, baseDir, federationList,
nickname,domain,port,httpPrefix, \ nickname, domain, port, httpPrefix,
objectUrl,clientToServer, \ objectUrl, clientToServer,
sendThreads,postLog, \ sendThreads, postLog,
personCache,cachedWebfingers, \ personCache, cachedWebfingers,
debug) debug)
def sendAnnounceViaServer(baseDir: str,session, \
fromNickname: str,password: str, def sendAnnounceViaServer(baseDir: str, session,
fromDomain: str,fromPort: int, \ fromNickname: str, password: str,
httpPrefix: str,repeatObjectUrl: str, \ fromDomain: str, fromPort: int,
cachedWebfingers: {},personCache: {}, \ httpPrefix: str, repeatObjectUrl: str,
debug: bool,projectVersion: str) -> {}: cachedWebfingers: {}, personCache: {},
debug: bool, projectVersion: str) -> {}:
"""Creates an announce message via c2s """Creates an announce message via c2s
""" """
if not session: if not session:
print('WARN: No session for sendAnnounceViaServer') print('WARN: No session for sendAnnounceViaServer')
return 6 return 6
withDigest=True fromDomainFull = fromDomain
fromDomainFull=fromDomain
if fromPort: if fromPort:
if fromPort!=80 and fromPort!=443: if fromPort != 80 and fromPort != 443:
if ':' not in fromDomain: if ':' not in fromDomain:
fromDomainFull=fromDomain+':'+str(fromPort) fromDomainFull = fromDomain + ':' + str(fromPort)
toUrl='https://www.w3.org/ns/activitystreams#Public' toUrl = 'https://www.w3.org/ns/activitystreams#Public'
ccUrl=httpPrefix+'://'+fromDomainFull+'/users/'+fromNickname+'/followers' ccUrl = httpPrefix + '://' + fromDomainFull + '/users/' + fromNickname + \
'/followers'
statusNumber,published=getStatusNumber() statusNumber, published = getStatusNumber()
newAnnounceId= \ newAnnounceId = httpPrefix + '://' + fromDomainFull + '/users/' + \
httpPrefix+'://'+fromDomainFull+'/users/'+ \ fromNickname + '/statuses/' + statusNumber
fromNickname+'/statuses/'+statusNumber newAnnounceJson = {
newAnnounceJson={
"@context": "https://www.w3.org/ns/activitystreams", "@context": "https://www.w3.org/ns/activitystreams",
'actor': httpPrefix+'://'+fromDomainFull+'/users/'+fromNickname, 'actor': httpPrefix+'://'+fromDomainFull+'/users/'+fromNickname,
'atomUri': newAnnounceId, 'atomUri': newAnnounceId,
@ -479,42 +496,48 @@ def sendAnnounceViaServer(baseDir: str,session, \
'type': 'Announce' 'type': 'Announce'
} }
handle=httpPrefix+'://'+fromDomainFull+'/@'+fromNickname handle = httpPrefix + '://' + fromDomainFull + '/@' + fromNickname
# lookup the inbox for the To handle # lookup the inbox for the To handle
wfRequest= \ wfRequest = webfingerHandle(session, handle, httpPrefix,
webfingerHandle(session,handle,httpPrefix,cachedWebfingers, \ cachedWebfingers,
fromDomain,projectVersion) fromDomain, projectVersion)
if not wfRequest: if not wfRequest:
if debug: if debug:
print('DEBUG: announce webfinger failed for '+handle) print('DEBUG: announce webfinger failed for ' + handle)
return 1 return 1
postToBox='outbox' postToBox = 'outbox'
# get the actor inbox for the To handle # get the actor inbox for the To handle
inboxUrl,pubKeyId,pubKey,fromPersonId,sharedInbox,capabilityAcquisition,avatarUrl,displayName= \ (inboxUrl, pubKeyId, pubKey, fromPersonId,
getPersonBox(baseDir,session,wfRequest,personCache, \ sharedInbox, capabilityAcquisition,
projectVersion,httpPrefix,fromNickname,fromDomain,postToBox) avatarUrl, displayName) = getPersonBox(baseDir, session, wfRequest,
personCache,
projectVersion, httpPrefix,
fromNickname, fromDomain,
postToBox)
if not inboxUrl: if not inboxUrl:
if debug: if debug:
print('DEBUG: No '+postToBox+' was found for '+handle) print('DEBUG: No ' + postToBox + ' was found for ' + handle)
return 3 return 3
if not fromPersonId: if not fromPersonId:
if debug: if debug:
print('DEBUG: No actor was found for '+handle) print('DEBUG: No actor was found for ' + handle)
return 4 return 4
authHeader=createBasicAuthHeader(fromNickname,password) authHeader = createBasicAuthHeader(fromNickname, password)
headers={ headers = {
'host': fromDomain, \ 'host': fromDomain,
'Content-type': 'application/json', \ 'Content-type': 'application/json',
'Authorization': authHeader 'Authorization': authHeader
} }
postResult= \ postResult = postJson(session, newAnnounceJson, [], inboxUrl,
postJson(session,newAnnounceJson,[],inboxUrl,headers,"inbox:write") headers, "inbox:write")
if not postResult:
print('WARN: Announce not posted')
if debug: if debug:
print('DEBUG: c2s POST announce success') print('DEBUG: c2s POST announce success')