mj-saunders
/
epicyon
forked from indymedia/epicyon
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

flake8 format

main
Bob Mottram 2020-04-04 11:28:58 +01:00
parent 3334cc0466
commit d55fe8bb59
1 changed files with 142 additions and 119 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

261
roles.py
View File

@ -1,14 +1,12 @@
__filename__="roles.py" __filename__ = "roles.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 json
import os import os
import time
from webfinger import webfingerHandle from webfinger import webfingerHandle
from auth import createBasicAuthHeader from auth import createBasicAuthHeader
from posts import getPersonBox from posts import getPersonBox
@ -18,109 +16,118 @@ from utils import getDomainFromActor
from utils import loadJson from utils import loadJson
from utils import saveJson from utils import saveJson
def clearModeratorStatus(baseDir: str) -> None: def clearModeratorStatus(baseDir: str) -> None:
"""Removes moderator status from all accounts """Removes moderator status from all accounts
This could be slow if there are many users, but only happens This could be slow if there are many users, but only happens
rarely when moderators are appointed or removed rarely when moderators are appointed or removed
""" """
directory=os.fsencode(baseDir+'/accounts/') directory = os.fsencode(baseDir + '/accounts/')
for f in os.scandir(directory): for f in os.scandir(directory):
f=f.name f = f.name
filename=os.fsdecode(f) filename = os.fsdecode(f)
if filename.endswith(".json") and '@' in filename: if filename.endswith(".json") and '@' in filename:
filename=os.path.join(baseDir+'/accounts/', filename) filename = os.path.join(baseDir + '/accounts/', filename)
if '"moderator"' in open(filename).read(): if '"moderator"' in open(filename).read():
actorJson=loadJson(filename) actorJson = loadJson(filename)
if actorJson: if actorJson:
if actorJson['roles'].get('instance'): if actorJson['roles'].get('instance'):
if 'moderator' in actorJson['roles']['instance']: if 'moderator' in actorJson['roles']['instance']:
actorJson['roles']['instance'].remove('moderator') actorJson['roles']['instance'].remove('moderator')
saveJson(actorJson,filename) saveJson(actorJson, filename)
def addModerator(baseDir: str,nickname: str,domain: str) -> None:
def addModerator(baseDir: str, nickname: str, domain: str) -> None:
"""Adds a moderator nickname to the file """Adds a moderator nickname to the file
""" """
if ':' in domain: if ':' in domain:
domain=domain.split(':')[0] domain = domain.split(':')[0]
moderatorsFile=baseDir+'/accounts/moderators.txt' moderatorsFile = baseDir + '/accounts/moderators.txt'
if os.path.isfile(moderatorsFile): if os.path.isfile(moderatorsFile):
# is this nickname already in the file? # is this nickname already in the file?
with open(moderatorsFile, "r") as f: with open(moderatorsFile, "r") as f:
lines=f.readlines() lines = f.readlines()
for moderator in lines: for moderator in lines:
moderator=moderator.strip('\n') moderator = moderator.strip('\n')
if line==nickname: if moderator == nickname:
return return
lines.append(nickname) lines.append(nickname)
with open(moderatorsFile, "w") as f: with open(moderatorsFile, "w") as f:
for moderator in lines: for moderator in lines:
moderator=moderator.strip('\n') moderator = moderator.strip('\n')
if len(moderator)>1: if len(moderator) > 1:
if os.path.isdir(baseDir+'/accounts/'+moderator+'@'+domain): if os.path.isdir(baseDir + '/accounts/' +
f.write(moderator+'\n') moderator + '@' + domain):
f.write(moderator + '\n')
else: else:
with open(moderatorsFile, "w+") as f: with open(moderatorsFile, "w+") as f:
if os.path.isdir(baseDir+'/accounts/'+nickname+'@'+domain): if os.path.isdir(baseDir + '/accounts/' +
f.write(nickname+'\n') nickname + '@' + domain):
f.write(nickname + '\n')
def removeModerator(baseDir: str,nickname: str):
def removeModerator(baseDir: str, nickname: str):
"""Removes a moderator nickname from the file """Removes a moderator nickname from the file
""" """
moderatorsFile=baseDir+'/accounts/moderators.txt' moderatorsFile = baseDir + '/accounts/moderators.txt'
if not os.path.isfile(moderatorsFile): if not os.path.isfile(moderatorsFile):
return return
with open(moderatorsFile, "r") as f: with open(moderatorsFile, "r") as f:
lines=f.readlines() lines = f.readlines()
with open(moderatorsFile, "w") as f: with open(moderatorsFile, "w") as f:
for moderator in lines: for moderator in lines:
moderator=moderator.strip('\n') moderator = moderator.strip('\n')
if len(moderator)>1 and moderator!=nickname: if len(moderator) > 1 and moderator != nickname:
f.write(moderator+'\n') f.write(moderator + '\n')
def setRole(baseDir: str,nickname: str,domain: str, \
project: str,role: str) -> bool: def setRole(baseDir: str, nickname: str, domain: str,
project: str, role: str) -> bool:
"""Set a person's role within a project """Set a person's role within a project
Setting the role to an empty string or None will remove it Setting the role to an empty string or None will remove it
""" """
# avoid giant strings # avoid giant strings
if len(role)>128 or len(project)>128: if len(role) > 128 or len(project) > 128:
return False return False
actorFilename=baseDir+'/accounts/'+nickname+'@'+domain+'.json' actorFilename = baseDir + '/accounts/' + \
nickname + '@' + domain + '.json'
if not os.path.isfile(actorFilename): if not os.path.isfile(actorFilename):
return False return False
actorJson=loadJson(actorFilename) actorJson = loadJson(actorFilename)
if actorJson: if actorJson:
if role: if role:
# add the role # add the role
if project=='instance' and 'role'=='moderator': if project == 'instance' and 'role' == 'moderator':
addModerator(baseDir,nickname,domain) addModerator(baseDir, nickname, domain)
if actorJson['roles'].get(project): if actorJson['roles'].get(project):
if role not in actorJson['roles'][project]: if role not in actorJson['roles'][project]:
actorJson['roles'][project].append(role) actorJson['roles'][project].append(role)
else: else:
actorJson['roles'][project]=[role] actorJson['roles'][project] = [role]
else: else:
# remove the role # remove the role
if project=='instance': if project == 'instance':
removeModerator(baseDir,nickname) removeModerator(baseDir, nickname)
if actorJson['roles'].get(project): if actorJson['roles'].get(project):
actorJson['roles'][project].remove(role) actorJson['roles'][project].remove(role)
# if the project contains no roles then remove it # if the project contains no roles then remove it
if len(actorJson['roles'][project])==0: if len(actorJson['roles'][project]) == 0:
del actorJson['roles'][project] del actorJson['roles'][project]
saveJson(actorJson,actorFilename) saveJson(actorJson, actorFilename)
return True return True
def getRoles(baseDir: str,nickname: str,domain: str, \
def getRoles(baseDir: str, nickname: str, domain: str,
project: str) -> []: project: str) -> []:
"""Returns the roles for a given person on a given project """Returns the roles for a given person on a given project
""" """
actorFilename=baseDir+'/accounts/'+nickname+'@'+domain+'.json' actorFilename = baseDir + '/accounts/' + \
nickname + '@' + domain + '.json'
if not os.path.isfile(actorFilename): if not os.path.isfile(actorFilename):
return False return False
actorJson=loadJson(actorFilename) actorJson = loadJson(actorFilename)
if actorJson: if actorJson:
if not actorJson.get('roles'): if not actorJson.get('roles'):
return None return None
@ -129,12 +136,14 @@ def getRoles(baseDir: str,nickname: str,domain: str, \
return actorJson['roles'][project] return actorJson['roles'][project]
return None return None
def outboxDelegate(baseDir: str,authenticatedNickname: str,messageJson: {},debug: bool) -> bool:
def outboxDelegate(baseDir: str, authenticatedNickname: str,
messageJson: {}, debug: bool) -> bool:
"""Handles receiving a delegation request """Handles receiving a delegation request
""" """
if not messageJson.get('type'): if not messageJson.get('type'):
return False return False
if not messageJson['type']=='Delegate': if not messageJson['type'] == 'Delegate':
return False return False
if not messageJson.get('object'): if not messageJson.get('object'):
return False return False
@ -142,7 +151,7 @@ def outboxDelegate(baseDir: str,authenticatedNickname: str,messageJson: {},debug
return False return False
if not messageJson['object'].get('type'): if not messageJson['object'].get('type'):
return False return False
if not messageJson['object']['type']=='Role': if not messageJson['object']['type'] == 'Role':
return False return False
if not messageJson['object'].get('object'): if not messageJson['object'].get('object'):
return False return False
@ -154,68 +163,70 @@ def outboxDelegate(baseDir: str,authenticatedNickname: str,messageJson: {},debug
print('WARN: No ; separator between project and role') print('WARN: No ; separator between project and role')
return False return False
delegatorNickname=getNicknameFromActor(messageJson['actor']) delegatorNickname = getNicknameFromActor(messageJson['actor'])
if delegatorNickname!=authenticatedNickname: if delegatorNickname != authenticatedNickname:
return return
domain,port=getDomainFromActor(messageJson['actor']) domain, port = getDomainFromActor(messageJson['actor'])
project=messageJson['object']['object'].split(';')[0].strip() project = messageJson['object']['object'].split(';')[0].strip()
# instance delegators can delagate to other projects # instance delegators can delagate to other projects
# than their own # than their own
canDelegate=False canDelegate = False
delegatorRoles=getRoles(baseDir,delegatorNickname, \ delegatorRoles = getRoles(baseDir, delegatorNickname,
domain,'instance') domain, 'instance')
if delegatorRoles: if delegatorRoles:
if 'delegator' in delegatorRoles: if 'delegator' in delegatorRoles:
canDelegate=True canDelegate = True
if canDelegate==False: if not canDelegate:
canDelegate=True canDelegate = True
# non-instance delegators can only delegate within their project # non-instance delegators can only delegate within their project
delegatorRoles=getRoles(baseDir,delegatorNickname, \ delegatorRoles = getRoles(baseDir, delegatorNickname,
domain,project) domain, project)
if delegatorRoles: if delegatorRoles:
if 'delegator' not in delegatorRoles: if 'delegator' not in delegatorRoles:
return False return False
else: else:
return False return False
if canDelegate==False: if not canDelegate:
return False return False
nickname=getNicknameFromActor(messageJson['object']['actor']) nickname = getNicknameFromActor(messageJson['object']['actor'])
if not nickname: if not nickname:
print('WARN: unable to find nickname in '+messageJson['object']['actor']) print('WARN: unable to find nickname in ' +
messageJson['object']['actor'])
return False return False
domainFull=domain role = \
if port: messageJson['object']['object'].split(';')[1].strip().lower()
if port!=80 and port!=443:
if ':' not in domain:
domainFull=domain+':'+str(port)
role=messageJson['object']['object'].split(';')[1].strip().lower()
if not role: if not role:
setRole(baseDir,nickname,domain,project,None) setRole(baseDir, nickname, domain, project, None)
return True return True
# what roles is this person already assigned to? # what roles is this person already assigned to?
existingRoles=getRoles(baseDir,nickname,domain,project) existingRoles = getRoles(baseDir, nickname, domain, project)
if existingRoles: if existingRoles:
if role in existingRoles: if role in existingRoles:
if debug: if debug:
print(nickname+'@'+domain+' is already assigned to the role '+role+' within the project '+project) print(nickname + '@' + domain +
' is already assigned to the role ' +
role + ' within the project ' + project)
return False return False
setRole(baseDir,nickname,domain,project,role) setRole(baseDir, nickname, domain, project, role)
if debug: if debug:
print(nickname+'@'+domain+' assigned to the role '+role+' within the project '+project) print(nickname + '@' + domain +
' assigned to the role ' + role +
' within the project ' + project)
return True return True
def sendRoleViaServer(baseDir: str,session, \
delegatorNickname: str,password: str, \ def sendRoleViaServer(baseDir: str, session,
delegatorDomain: str,delegatorPort: int, \ delegatorNickname: str, password: str,
httpPrefix: str,nickname: str, \ delegatorDomain: str, delegatorPort: int,
project: str,role: str, \ httpPrefix: str, nickname: str,
cachedWebfingers: {},personCache: {}, \ project: str, role: str,
debug: bool,projectVersion: str) -> {}: cachedWebfingers: {}, personCache: {},
debug: bool, projectVersion: str) -> {}:
"""A delegator creates a role for a person via c2s """A delegator creates a role for a person via c2s
Setting role to an empty string or None removes the role Setting role to an empty string or None removes the role
""" """
@ -223,28 +234,34 @@ def sendRoleViaServer(baseDir: str,session, \
print('WARN: No session for sendRoleViaServer') print('WARN: No session for sendRoleViaServer')
return 6 return 6
delegatorDomainFull=delegatorDomain delegatorDomainFull = delegatorDomain
if fromPort: if delegatorPort:
if fromPort!=80 and fromPort!=443: if delegatorPort != 80 and delegatorPort != 443:
if ':' not in delegatorDomain: if ':' not in delegatorDomain:
delegatorDomainFull=delegatorDomain+':'+str(fromPort) delegatorDomainFull = \
delegatorDomain + ':' + str(delegatorPort)
toUrl= \ toUrl = \
httpPrefix+'://'+delegatorDomainFull+'/users/'+nickname httpPrefix + '://' + delegatorDomainFull + '/users/' + nickname
ccUrl= \ ccUrl = \
httpPrefix+'://'+delegatorDomainFull+'/users/'+ \ httpPrefix + '://' + delegatorDomainFull + '/users/' + \
delegatorNickname+'/followers' delegatorNickname + '/followers'
if role: if role:
roleStr=project.lower()+';'+role.lower() roleStr = project.lower() + ';' + role.lower()
else: else:
roleStr=project.lower()+';' roleStr = project.lower() + ';'
newRoleJson={ actor = \
httpPrefix + '://' + delegatorDomainFull + \
'/users/' + delegatorNickname
delegateActor = \
httpPrefix + '://' + delegatorDomainFull + '/users/' + nickname
newRoleJson = {
'type': 'Delegate', 'type': 'Delegate',
'actor': httpPrefix+'://'+delegatorDomainFull+'/users/'+delegatorNickname, 'actor': actor,
'object': { 'object': {
'type': 'Role', 'type': 'Role',
'actor': httpPrefix+'://'+delegatorDomainFull+'/users/'+nickname, 'actor': delegateActor,
'object': roleStr, 'object': roleStr,
'to': [toUrl], 'to': [toUrl],
'cc': [ccUrl] 'cc': [ccUrl]
@ -253,46 +270,52 @@ def sendRoleViaServer(baseDir: str,session, \
'cc': [ccUrl] 'cc': [ccUrl]
} }
handle=httpPrefix+'://'+delegatorDomainFull+'/@'+delegatorNickname handle = \
httpPrefix + '://' + delegatorDomainFull + '/@' + delegatorNickname
# lookup the inbox for the To handle # lookup the inbox for the To handle
wfRequest=webfingerHandle(session,handle,httpPrefix,cachedWebfingers, \ wfRequest = webfingerHandle(session, handle, httpPrefix,
delegatorDomain,projectVersion) cachedWebfingers,
delegatorDomain, 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,
getPersonBox(baseDir,session,wfRequest,personCache, \ fromPersonId, sharedInbox,
projectVersion,httpPrefix, \ capabilityAcquisition,
delegatorNickname,delegatorDomain,postToBox) avatarUrl, displayName) = getPersonBox(baseDir, session,
wfRequest, personCache,
projectVersion, httpPrefix,
delegatorNickname,
delegatorDomain, 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(delegatorNickname,password) authHeader = createBasicAuthHeader(delegatorNickname, password)
headers={ headers = {
'host': delegatorDomain, \ 'host': delegatorDomain,
'Content-type': 'application/json', \ 'Content-type': 'application/json',
'Authorization': authHeader 'Authorization': authHeader
} }
postResult= \ postResult = \
postJson(session,newRoleJson,[],inboxUrl,headers,"inbox:write") postJson(session, newRoleJson, [], inboxUrl, headers, "inbox:write")
#if not postResult: if not postResult:
# if debug: if debug:
# print('DEBUG: POST announce failed for c2s to '+inboxUrl) print('DEBUG: POST announce failed for c2s to '+inboxUrl)
# return 5 # return 5
if debug: if debug:
print('DEBUG: c2s POST role success') print('DEBUG: c2s POST role success')