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

Tidying

master
Bob Mottram 2019-07-02 21:54:22 +01:00
parent 8276f24468
commit 306f9edf46
10 changed files with 135 additions and 86 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
announce.py
View File

@ -12,7 +12,10 @@ from utils import getStatusNumber
from utils import createOutboxDir
from utils import urlPermitted
def createAnnounce(baseDir: str,federationList: [],username: str, domain: str, port: int,toUrl: str, ccUrl: str, https: bool, objectUrl: str, saveToFile: bool) -> {}:
def createAnnounce(baseDir: str,federationList: [], \
username: str, domain: str, port: int, \
toUrl: str, ccUrl: str, https: bool, \
objectUrl: str, saveToFile: bool) -> {}:
"""Creates an announce message
Typically toUrl will be https://www.w3.org/ns/activitystreams#Public
and ccUrl might be a specific person favorited or repeated and the followers url
@ -52,7 +55,9 @@ def createAnnounce(baseDir: str,federationList: [],username: str, domain: str, p
commentjson.dump(newAnnounce, fp, indent=4, sort_keys=False)
return newAnnounce
def announcePublic(baseDir: str,federationList: [],username: str, domain: str, port: int, https: bool, objectUrl: str, saveToFile: bool) -> {}:
def announcePublic(baseDir: str,federationList: [], \
username: str, domain: str, port: int, https: bool, \
objectUrl: str, saveToFile: bool) -> {}:
"""Makes a public announcement
"""
prefix='https'
@ -65,9 +70,14 @@ def announcePublic(baseDir: str,federationList: [],username: str, domain: str, p
toUrl = 'https://www.w3.org/ns/activitystreams#Public'
ccUrl = prefix + '://'+fromDomain+'/users/'+username+'/followers'
return createAnnounce(baseDir,username, domain, port,toUrl, ccUrl, https, objectUrl, saveToFile)
return createAnnounce(baseDir,username, domain, port, \
toUrl, ccUrl, https, objectUrl, saveToFile)
def repeatPost(baseDir: str,federationList: [],username: str, domain: str, port: int, https: bool, announceUsername: str, announceDomain: str, announcePort: int, announceStatusNumber: int, announceHttps: bool, saveToFile: bool) -> {}:
def repeatPost(baseDir: str,federationList: [], \
username: str, domain: str, port: int, https: bool, \
announceUsername: str, announceDomain: str, \
announcePort: int, announceHttps: bool, \
announceStatusNumber: int, saveToFile: bool) -> {}:
"""Repeats a given status post
"""
prefix='https'
@ -78,7 +88,8 @@ def repeatPost(baseDir: str,federationList: [],username: str, domain: str, port:
if announcePort!=80 and announcePort!=443:
announcedDomain=announcedDomain+':'+str(announcePort)
objectUrl = prefix + '://'+announcedDomain+'/users/'+announceUsername+'/statuses/'+str(announceStatusNumber)
objectUrl = prefix + '://'+announcedDomain+'/users/'+ \
announceUsername+'/statuses/'+str(announceStatusNumber)
return announcePublic(baseDir,username, domain, port, https, objectUrl, saveToFile)

31
daemon.py
View File

@ -116,32 +116,40 @@ class PubServer(BaseHTTPRequestHandler):
return
print('############### _webfinger end')
# get outbox feed for a person
outboxFeed=personOutboxJson(self.server.baseDir,self.server.domain,self.server.port,self.path,self.server.https,maxPostsInFeed)
outboxFeed=personOutboxJson(self.server.baseDir,self.server.domain, \
self.server.port,self.path, \
self.server.https,maxPostsInFeed)
if outboxFeed:
self._set_headers('application/json')
self.wfile.write(json.dumps(outboxFeed).encode('utf-8'))
self.server.GETbusy=False
return
following=getFollowingFeed(self.server.baseDir,self.server.domain,self.server.port,self.path,self.server.https,followsPerPage)
following=getFollowingFeed(self.server.baseDir,self.server.domain, \
self.server.port,self.path, \
self.server.https,followsPerPage)
if following:
self._set_headers('application/json')
self.wfile.write(json.dumps(following).encode('utf-8'))
self.server.GETbusy=False
return
followers=getFollowingFeed(self.server.baseDir,self.server.domain,self.server.port,self.path,self.server.https,followsPerPage,'followers')
followers=getFollowingFeed(self.server.baseDir,self.server.domain, \
self.server.port,self.path, \
self.server.https,followsPerPage,'followers')
if followers:
self._set_headers('application/json')
self.wfile.write(json.dumps(followers).encode('utf-8'))
self.server.GETbusy=False
return
# look up a person
getPerson = personLookup(self.server.domain,self.path,self.server.baseDir)
getPerson = personLookup(self.server.domain,self.path, \
self.server.baseDir)
if getPerson:
self._set_headers('application/json')
self.wfile.write(json.dumps(getPerson).encode('utf-8'))
self.server.GETbusy=False
return
personKey = personKeyLookup(self.server.domain,self.path,self.server.baseDir)
personKey = personKeyLookup(self.server.domain,self.path, \
self.server.baseDir)
if personKey:
self._set_headers('text/html; charset=utf-8')
self.wfile.write(personKey.encode('utf-8'))
@ -228,13 +236,16 @@ class PubServer(BaseHTTPRequestHandler):
currSessionTime=int(time.time())
if currSessionTime-self.server.sessionLastUpdate>1200:
self.server.sessionLastUpdate=currSessionTime
self.server.session = createSession(self.server.domain,self.server.port,self.server.useTor)
self.server.session = \
createSession(self.server.domain,self.server.port, \
self.server.useTor)
print('**************** POST started new session')
print('**************** POST get actor url from '+self.server.baseDir)
personUrl=messageJson['actor']
print('**************** POST get public key of '+personUrl+' from '+self.server.baseDir)
pubKey=getPersonPubKey(self.server.session,personUrl,self.server.personCache)
pubKey=getPersonPubKey(self.server.session,personUrl, \
self.server.personCache)
if not pubKey:
print('**************** POST no sender public key')
self.send_response(401)
@ -242,14 +253,16 @@ class PubServer(BaseHTTPRequestHandler):
self.server.POSTbusy=False
return
print('**************** POST check signature')
if not verifyPostHeaders(self.server.https, pubKey, self.headers, '/inbox' ,False, json.dumps(messageJson)):
if not verifyPostHeaders(self.server.https, pubKey, self.headers, \
'/inbox' ,False, json.dumps(messageJson)):
print('**************** POST signature verification failed')
self.send_response(401)
self.end_headers()
self.server.POSTbusy=False
return
print('**************** POST valid')
if receiveFollowRequest(self.server.baseDir,messageJson,self.server.federationList):
if receiveFollowRequest(self.server.baseDir,messageJson, \
self.server.federationList):
self.send_response(200)
self.end_headers()
self.server.POSTbusy=False

6
epicyon.py
View File

@ -85,10 +85,10 @@ setBio(baseDir,username,domain,'Some personal info')
#pprint(outboxJson)
#testPostMessageBetweenServers()
runDaemon(domain,port,https,federationList,useTor)
#runDaemon(domain,port,https,federationList,useTor)
#testHttpsig()
sys.exit()
#sys.exit()
#pprint(person)
#print('\n')
@ -110,5 +110,5 @@ wfResult = json.dumps(wfRequest, indent=4, sort_keys=True)
maxMentions=10
maxEmoji=10
maxAttachments=5
userPosts = getPosts(session,personUrl,10,maxMentions,maxEmoji,maxAttachments,federationList,personCache)
userPosts = getPosts(session,personUrl,30,maxMentions,maxEmoji,maxAttachments,federationList,personCache)
#print(str(userPosts))

5
follow.py
View File

@ -237,8 +237,8 @@ def receiveFollowRequest(baseDir: str,messageJson: {},federationList: []) -> boo
def sendFollowRequest(baseDir: str,username: str,domain: str,port: int,https: bool, \
followUsername: str,followDomain: str,followPort: bool,followHttps: bool, \
federationList: []):
"""Sends a follow request
federationList: []) -> {}:
"""Gets the json object for sending a follow request
"""
if not domainPermitted(followDomain,federationList):
return None
@ -268,3 +268,4 @@ def sendFollowRequest(baseDir: str,username: str,domain: str,port: int,https: bo
if ccUrl:
if len(ccUrl)>0:
newFollow['cc']=ccUrl
return newFollow

24
httpsig.py
View File

@ -15,7 +15,9 @@ from requests.auth import AuthBase
import base64
import json
def signPostHeaders(privateKeyPem: str, username: str, domain: str, port: int,path: str, https: bool, messageBodyJson: {}) -> str:
def signPostHeaders(privateKeyPem: str, username: str, domain: str, \
port: int,path: str, \
https: bool, messageBodyJson: {}) -> str:
"""Returns a raw signature string that can be plugged into a header and
used to verify the authenticity of an HTTP transmission.
"""
@ -30,7 +32,8 @@ def signPostHeaders(privateKeyPem: str, username: str, domain: str, port: int,pa
if not messageBodyJson:
headers = {'host': domain}
else:
bodyDigest = base64.b64encode(SHA256.new(messageBodyJson.encode()).digest())
bodyDigest = \
base64.b64encode(SHA256.new(messageBodyJson.encode()).digest())
headers = {'host': domain, 'digest': f'SHA-256={bodyDigest}'}
privateKeyPem = RSA.import_key(privateKeyPem)
headers.update({
@ -59,7 +62,9 @@ def signPostHeaders(privateKeyPem: str, username: str, domain: str, port: int,pa
[f'{k}="{v}"' for k, v in signatureDict.items()])
return signatureHeader
def createSignedHeader(privateKeyPem: str,username: str,domain: str,port: int,path: str,https: bool,withDigest: bool,messageBodyJson: {}) -> {}:
def createSignedHeader(privateKeyPem: str,username: str,domain: str,port: int, \
path: str,https: bool,withDigest: bool, \
messageBodyJson: {}) -> {}:
headerDomain=domain
if port!=80 and port!=443:
@ -69,15 +74,19 @@ def createSignedHeader(privateKeyPem: str,username: str,domain: str,port: int,pa
headers = {'host': headerDomain}
else:
messageBodyJsonStr=json.dumps(messageBodyJson)
bodyDigest = base64.b64encode(SHA256.new(messageBodyJsonStr.encode()).digest())
bodyDigest = \
base64.b64encode(SHA256.new(messageBodyJsonStr.encode()).digest())
headers = {'host': headerDomain, 'digest': f'SHA-256={bodyDigest}'}
path='/inbox'
signatureHeader = signPostHeaders(privateKeyPem, username, domain, port, path, https, None)
signatureHeader = signPostHeaders(privateKeyPem, username, domain, port, \
path, https, None)
headers['signature'] = signatureHeader
headers['Content-type'] = 'application/json'
return headers
def verifyPostHeaders(https: bool, publicKeyPem: str, headers: dict, path: str, GETmethod: bool, messageBodyJsonStr: str) -> bool:
def verifyPostHeaders(https: bool, publicKeyPem: str, headers: dict, \
path: str, GETmethod: bool, \
messageBodyJsonStr: str) -> bool:
"""Returns true or false depending on if the key that we plugged in here
validates against the headers, method, and path.
publicKeyPem - the public key from an rsa key pair
@ -111,7 +120,8 @@ def verifyPostHeaders(https: bool, publicKeyPem: str, headers: dict, path: str,
signedHeaderList.append(
f'(request-target): {method.lower()} {path}')
elif signedHeader == 'digest':
bodyDigest = base64.b64encode(SHA256.new(messageBodyJsonStr.encode()).digest())
bodyDigest = \
base64.b64encode(SHA256.new(messageBodyJsonStr.encode()).digest())
signedHeaderList.append(f'digest: SHA-256={bodyDigest}')
else:
signedHeaderList.append(

36
inbox.py
View File

@ -42,44 +42,10 @@ def inboxPermittedMessage(domain: str,messageJson: {},federationList: []) -> boo
return True
def receivePublicMessage(message: {}) -> bool:
print("TODO")
def validPublishedDate(published):
def validPublishedDate(published) -> bool:
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
def receiveMessage(message: {},baseDir: str):
if not message.get('type'):
return
if message['type']!='Create':
return
if not message.get('published'):
return
# is the message too old?
if not validPublishedDate(message['published']):
return
if not message.get('to'):
return
if not message.get('id'):
return
for recipient in message['to']:
if recipient.endswith('/activitystreams#Public'):
receivePublicMessage(message)
continue
username=''
domain=''
messageId=message['id'].replace('/','_')
handle=username.lower()+'@'+domain.lower()
if not os.path.isdir(baseDir+'/accounts/'+handle):
os.mkdir(baseDir+'/accounts/'+handle)
if not os.path.isdir(baseDir+'/accounts/'+handle+'/inbox'):
os.mkdir(baseDir+'/accounts/'+handle+'/inbox')
filename=baseDir+'/accounts/'+handle+'/inbox/'+messageId+'.json'
with open(filename, 'w') as fp:
commentjson.dump(personJson, fp, indent=4, sort_keys=False)

8
like.py
View File

@ -10,7 +10,8 @@ import json
import commentjson
from utils import urlPermitted
def like(baseDir: str,federationList: [],username: str,domain: str,port: int,toUrl: str,ccUrl: str,https: bool,objectUrl: str,saveToFile: bool) -> {}:
def like(baseDir: str,federationList: [],username: str,domain: str,port: int, \
toUrl: str,ccUrl: str,https: bool,objectUrl: str,saveToFile: bool) -> {}:
"""Creates a like
Typically toUrl will be a followers collection
and ccUrl might be a specific person whose post was liked
@ -42,7 +43,10 @@ def like(baseDir: str,federationList: [],username: str,domain: str,port: int,toU
# TODO update likes collection
return newLike
def likePost(baseDir: str,federationList: [],username: str, domain: str, port: int, https: bool, likeUsername: str, likeDomain: str, likePort: int, likeStatusNumber: int, likeHttps: bool,saveToFile: bool) -> {}:
def likePost(baseDir: str,federationList: [], \
username: str, domain: str, port: int, https: bool, \n
likeUsername: str, likeDomain: str, likePort: int, likeHttps: bool, \n
likeStatusNumber: int,saveToFile: bool) -> {}:
"""Likes a given status post
"""
prefix='https'

20
person.py
View File

@ -21,7 +21,8 @@ def generateRSAKey() -> (str,str):
publicKeyPem = key.publickey().exportKey("PEM").decode("utf-8")
return privateKeyPem,publicKeyPem
def createPerson(baseDir: str,username: str,domain: str,port: int,https: bool, saveToFile: bool) -> (str,str,{},{}):
def createPerson(baseDir: str,username: str,domain: str,port: int, \
https: bool, saveToFile: bool) -> (str,str,{},{}):
"""Returns the private key, public key, actor and webfinger endpoint
"""
prefix='https'
@ -29,7 +30,8 @@ def createPerson(baseDir: str,username: str,domain: str,port: int,https: bool, s
prefix='http'
privateKeyPem,publicKeyPem=generateRSAKey()
webfingerEndpoint=createWebfingerEndpoint(username,domain,port,https,publicKeyPem)
webfingerEndpoint= \
createWebfingerEndpoint(username,domain,port,https,publicKeyPem)
if saveToFile:
storeWebfingerEndpoint(username,domain,baseDir,webfingerEndpoint)
@ -140,7 +142,10 @@ def personKeyLookup(domain: str,path: str,baseDir: str) -> str:
def personLookup(domain: str,path: str,baseDir: str) -> {}:
"""Lookup the person for an given username
"""
notPersonLookup=['/inbox','/outbox','/outboxarchive','/followers','/following','/featured','.png','.jpg','.gif','.mpv','#main-key','/main-key']
notPersonLookup=['/inbox','/outbox','/outboxarchive', \
'/followers','/following','/featured', \
'.png','.jpg','.gif','.mpv', \
'#main-key','/main-key']
for ending in notPersonLookup:
if path.endswith(ending):
return None
@ -164,7 +169,8 @@ def personLookup(domain: str,path: str,baseDir: str) -> {}:
personJson=commentjson.load(fp)
return personJson
def personOutboxJson(baseDir: str,domain: str,port: int,path: str,https: bool,noOfItems: int) -> []:
def personOutboxJson(baseDir: str,domain: str,port: int,path: str, \
https: bool,noOfItems: int) -> []:
"""Obtain the outbox feed for the given person
"""
if not '/outbox' in path:
@ -198,9 +204,11 @@ def personOutboxJson(baseDir: str,domain: str,port: int,path: str,https: bool,no
return None
if not validUsername(username):
return None
return createOutbox(baseDir,username,domain,port,https,noOfItems,headerOnly,pageNumber)
return createOutbox(baseDir,username,domain,port,https, \
noOfItems,headerOnly,pageNumber)
def setPreferredUsername(baseDir: str,username: str, domain: str, preferredName: str) -> bool:
def setPreferredUsername(baseDir: str,username: str, domain: str, \
preferredName: str) -> bool:
if len(preferredName)>32:
return False
handle=username.lower()+'@'+domain.lower()

58
posts.py
View File

@ -118,7 +118,9 @@ def getPersonPubKey(session,personUrl: str,personCache: {}) -> str:
storePersonInCache(personUrl,personJson,personCache)
return pubKey
def getPosts(session,outboxUrl: str,maxPosts: int,maxMentions: int,maxEmoji: int,maxAttachments: int,federationList: [],personCache: {}) -> {}:
def getPosts(session,outboxUrl: str,maxPosts: int,maxMentions: int, \
maxEmoji: int,maxAttachments: int,federationList: [], \
personCache: {}) -> {}:
personPosts={}
if not outboxUrl:
return personPosts
@ -234,7 +236,10 @@ def deleteAllPosts(username: str, domain: str,baseDir: str) -> None:
except Exception as e:
print(e)
def createPostBase(baseDir: str,username: str, domain: str, port: int,toUrl: str, ccUrl: str, https: bool, content: str, followersOnly: bool, saveToFile: bool, inReplyTo=None, inReplyToAtomUri=None, subject=None) -> {}:
def createPostBase(baseDir: str,username: str, domain: str, port: int, \
toUrl: str, ccUrl: str, https: bool, content: str, \
followersOnly: bool, saveToFile: bool, \
inReplyTo=None, inReplyToAtomUri=None, subject=None) -> {}:
"""Creates a message
"""
prefix='https'
@ -308,21 +313,29 @@ def createPostBase(baseDir: str,username: str, domain: str, port: int,toUrl: str
commentjson.dump(newPost, fp, indent=4, sort_keys=False)
return newPost
def createPublicPost(baseDir: str,username: str, domain: str, port: int,https: bool, content: str, followersOnly: bool, saveToFile: bool, inReplyTo=None, inReplyToAtomUri=None, subject=None) -> {}:
def createPublicPost(baseDir: str,username: str, domain: str, port: int,https: bool, \
content: str, followersOnly: bool, saveToFile: bool, \
inReplyTo=None, inReplyToAtomUri=None, subject=None) -> {}:
"""Public post to the outbox
"""
prefix='https'
if not https:
prefix='http'
return createPostBase(baseDir,username, domain, port,'https://www.w3.org/ns/activitystreams#Public', prefix+'://'+domain+'/users/'+username+'/followers', https, content, followersOnly, saveToFile, inReplyTo, inReplyToAtomUri, subject)
return createPostBase(baseDir,username, domain, port, \
'https://www.w3.org/ns/activitystreams#Public', \
prefix+'://'+domain+'/users/'+username+'/followers', \
https, content, followersOnly, saveToFile, \
inReplyTo, inReplyToAtomUri, subject)
def threadSendPost(session,postJsonObject: {},federationList: [],inboxUrl: str,baseDir: str,signatureHeaderJson: {},postLog: []) -> None:
def threadSendPost(session,postJsonObject: {},federationList: [],inboxUrl: str, \
baseDir: str,signatureHeaderJson: {},postLog: []) -> None:
"""Sends a post with exponential backoff
"""
tries=0
backoffTime=60
for attempt in range(20):
postResult = postJson(session,postJsonObject,federationList,inboxUrl,signatureHeaderJson)
postResult = postJson(session,postJsonObject,federationList, \
inboxUrl,signatureHeaderJson)
if postResult:
postLog.append(postJsonObject['published']+' '+postResult+'\n')
# keep the length of the log finite
@ -339,7 +352,12 @@ def threadSendPost(session,postJsonObject: {},federationList: [],inboxUrl: str,b
time.sleep(backoffTime)
backoffTime *= 2
def sendPost(session,baseDir: str,username: str, domain: str, port: int, toUsername: str, toDomain: str, toPort: int, cc: str, https: bool, content: str, followersOnly: bool, saveToFile: bool, federationList: [], sendThreads: [], postLog: [], cachedWebfingers: {},personCache: {},inReplyTo=None, inReplyToAtomUri=None, subject=None) -> int:
def sendPost(session,baseDir: str,username: str, domain: str, port: int, \
toUsername: str, toDomain: str, toPort: int, cc: str, \
https: bool, content: str, followersOnly: bool, \
saveToFile: bool, federationList: [], sendThreads: [], \
postLog: [], cachedWebfingers: {},personCache: {}, \
inReplyTo=None, inReplyToAtomUri=None, subject=None) -> int:
"""Post to another inbox
"""
prefix='https'
@ -359,7 +377,8 @@ def sendPost(session,baseDir: str,username: str, domain: str, port: int, toUsern
return 1
# get the actor inbox for the To handle
inboxUrl,pubKeyId,pubKey,toPersonId = getPersonBox(session,wfRequest,personCache,'inbox')
inboxUrl,pubKeyId,pubKey,toPersonId = \
getPersonBox(session,wfRequest,personCache,'inbox')
if not inboxUrl:
return 2
if not pubKey:
@ -367,7 +386,11 @@ def sendPost(session,baseDir: str,username: str, domain: str, port: int, toUsern
if not toPersonId:
return 4
postJsonObject=createPostBase(baseDir,username,domain,port,toPersonId,cc,https,content,followersOnly,saveToFile,inReplyTo,inReplyToAtomUri,subject)
postJsonObject=createPostBase(baseDir,username,domain,port, \
toPersonId,cc,https,content, \
followersOnly,saveToFile, \
inReplyTo,inReplyToAtomUri, \
subject)
# get the senders private key
privateKeyPem=getPersonKey(username,domain,baseDir,'private')
@ -375,18 +398,26 @@ def sendPost(session,baseDir: str,username: str, domain: str, port: int, toUsern
return 5
# construct the http header
signatureHeaderJson = createSignedHeader(privateKeyPem, username, domain, port, '/inbox', https, withDigest, postJsonObject)
signatureHeaderJson = \
createSignedHeader(privateKeyPem, username, domain, port, \
'/inbox', https, withDigest, postJsonObject)
# Keep the number of threads being used small
while len(sendThreads)>10:
sendThreads[0].kill()
sendThreads.pop(0)
thr = threadWithTrace(target=threadSendPost,args=(session,postJsonObject.copy(),federationList,inboxUrl,baseDir,signatureHeaderJson.copy(),postLog),daemon=True)
thr = threadWithTrace(target=threadSendPost,args=(session, \
postJsonObject.copy(), \
federationList, \
inboxUrl,baseDir, \
signatureHeaderJson.copy(), \
postLog),daemon=True)
sendThreads.append(thr)
thr.start()
return 0
def createOutbox(baseDir: str,username: str,domain: str,port: int,https: bool,itemsPerPage: int,headerOnly: bool,pageNumber=None) -> {}:
def createOutbox(baseDir: str,username: str,domain: str,port: int,https: bool, \
itemsPerPage: int,headerOnly: bool,pageNumber=None) -> {}:
"""Constructs the outbox feed
"""
prefix='https'
@ -484,7 +515,8 @@ def createOutbox(baseDir: str,username: str,domain: str,port: int,https: bool,it
return outboxHeader
return outboxItems
def archivePosts(username: str,domain: str,baseDir: str,maxPostsInOutbox=256) -> None:
def archivePosts(username: str,domain: str,baseDir: str, \
maxPostsInOutbox=256) -> None:
"""Retain a maximum number of posts within the outbox
Move any others to an archive directory
"""

12
webfinger.py
View File

@ -21,10 +21,12 @@ def parseHandle(handle: str) -> (str,str):
if '.' not in handle:
return None, None
if '/@' in handle:
domain, username = handle.replace('https://','').replace('http://','').split('/@')
domain, username = \
handle.replace('https://','').replace('http://','').split('/@')
else:
if '/users/' in handle:
domain, username = handle.replace('https://','').replace('http://','').split('/users/')
domain, username = \
handle.replace('https://','').replace('http://','').split('/users/')
else:
if '@' in handle:
username, domain = handle.split('@')
@ -67,7 +69,8 @@ def generateMagicKey(publicKeyPem) -> str:
pubexp = base64.urlsafe_b64encode(number.long_to_bytes(privkey.e)).decode("utf-8")
return f"data:application/magic-public-key,RSA.{mod}.{pubexp}"
def storeWebfingerEndpoint(username: str,domain: str,baseDir: str,wfJson: {}) -> bool:
def storeWebfingerEndpoint(username: str,domain: str,baseDir: str, \
wfJson: {}) -> bool:
"""Stores webfinger endpoint for a user to a file
"""
handle=username+'@'+domain
@ -79,7 +82,8 @@ def storeWebfingerEndpoint(username: str,domain: str,baseDir: str,wfJson: {}) ->
commentjson.dump(wfJson, fp, indent=4, sort_keys=False)
return True
def createWebfingerEndpoint(username: str,domain: str,port: int,https: bool,publicKeyPem) -> {}:
def createWebfingerEndpoint(username: str,domain: str,port: int, \
https: bool,publicKeyPem) -> {}:
"""Creates a webfinger endpoint for a user
"""
prefix='https'