epicyon/capabilities.py

106 lines
3.6 KiB
Python
Raw Normal View History

__filename__ = "capabilities.py"
__author__ = "Bob Mottram"
__license__ = "AGPL3+"
__version__ = "0.0.1"
__maintainer__ = "Bob Mottram"
__email__ = "bob@freedombone.net"
__status__ = "Production"
2019-07-06 09:07:24 +00:00
import os
2019-07-07 11:53:32 +00:00
import datetime
import time
import json
import commentjson
from auth import createPassword
2019-07-07 11:53:32 +00:00
def capabilitiesMakeDirs(baseDir: str):
if not os.path.isdir(baseDir+'/ocap'):
os.mkdir(baseDir+'/ocap')
# for capabilities accepted by this instance
if not os.path.isdir(baseDir+'/ocap/accept'):
os.mkdir(baseDir+'/ocap/accept')
# for capabilities granted to this instance
if not os.path.isdir(baseDir+'/ocap/granted'):
os.mkdir(baseDir+'/ocap/granted')
def capabilitiesRequest(baseDir: str,httpPrefix: str,domain: str, \
requestedActor: str, \
requestedCaps=["inbox:write","objects:read"]) -> {}:
2019-07-06 09:07:24 +00:00
# This is sent to the capabilities endpoint /caps/new
# which could be instance wide or for a particular person
2019-07-07 11:53:32 +00:00
# This could also be added to a follow activity
capabilitiesMakeDirs(baseDir)
ocapId=createPassword(32)
ocapRequest = {
"id": httpPrefix+"://"+requestedDomain+"/caps/request/"+ocapId,
"type": "Request",
2019-07-06 10:38:48 +00:00
"capability": requestedCaps,
2019-07-06 09:15:40 +00:00
"actor": requestedActor
2019-07-06 09:07:24 +00:00
}
2019-07-07 11:53:32 +00:00
return ocapRequest
2019-07-06 09:07:24 +00:00
2019-07-07 11:53:32 +00:00
def capabilitiesAccept(baseDir: str,httpPrefix: str,nickname: str,domain: str, \
acceptedActor: str, saveToFile: bool, \
acceptedCaps=["inbox:write","objects:read"]) -> {}:
2019-07-06 09:07:24 +00:00
# This gets returned to capabilities requester
2019-07-07 11:53:32 +00:00
# This could also be added to a follow Accept activity
# reject excessively long actors
if len(acceptedActor)>256:
return None
# make directories to store capabilities
capabilitiesMakeDirs(baseDir)
filename=baseDir+'/ocap/accept/'+acceptedActor.replace('/','#')+'.json'
ocapAccept=None
# if the capability already exists then load it from file
if os.path.isfile(filename):
with open(filename, 'r') as fp:
ocapAccept=commentjson.load(fp)
# otherwise create a new capability
if not ocapAccept:
ocapId=createPassword(32)
ocapAccept = {
"id": httpPrefix+"://"+domain+"/caps/"+ocapId,
"type": "Capability",
"capability": acceptedCaps,
"scope": acceptedActor,
"actor": httpPrefix+"://"+domain
}
if nickname:
ocapAccept['actor']=httpPrefix+"://"+domain+'/users/'+nickname
2019-07-06 09:07:24 +00:00
2019-07-07 11:53:32 +00:00
if saveToFile:
with open(filename, 'w') as fp:
commentjson.dump(ocapAccept, fp, indent=4, sort_keys=False)
return ocapAccept
def capabilitiesGrantedSave(baseDir :str,ocap: {}) -> bool:
"""A capabilities accept is received, so stor it for
reference when sending to the actor
"""
if not ocap.get('actor'):
return False
filename=baseDir+'/ocap/granted/'+ocap['actor'].replace('/','#')+'.json'
with open(filename, 'w') as fp:
commentjson.dump(ocap, fp, indent=4, sort_keys=False)
return True
def isCapable(actor: str,ocapGranted: {},capability: str) -> bool:
2019-07-06 09:07:24 +00:00
# is the given actor capable of using the current resource?
2019-07-07 11:53:32 +00:00
for id,ocap in ocapGranted.items():
if ocap['scope'] in actor:
if capability in ocap['capability']:
return True
return False
def isCapableId(id: str,ocapGranted: {},capability: str) -> bool:
# is the given id capable of using the current resource?
if ocapGranted.get(id):
if ocapGranted['id']['scope'] in actor:
if capability in ocapGranted['id']['capability']:
2019-07-06 10:33:57 +00:00
return True
2019-07-06 09:07:24 +00:00
return False