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

http signature fixes

master
Bob Mottram 2019-08-15 22:34:25 +01:00
parent ed050d48f5
commit 977160eecf
5 changed files with 46 additions and 45 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
daemon.py
View File

@ -400,6 +400,7 @@ class PubServer(BaseHTTPRequestHandler):
messageJson,
self.headers['host'],
self.headers['signature'],
self.headers,
'/'+self.path.split('/')[-1],
self.server.debug)
if queueFilename:

47
httpsig.py
View File

@ -28,20 +28,21 @@ def signPostHeaders(privateKeyPem: str, nickname: str, domain: str, \
dateStr=strftime("%a, %d %b %Y %H:%M:%S %Z", gmtime())
keyID = httpPrefix+'://'+domain+'/users/'+nickname+'#main-key'
if not messageBodyJson:
headers = {'host': domain}
headers = {'(request-target)': f'post {path}','host': domain,'date': dateStr,'content-type': 'application/json'}
else:
bodyDigest = \
base64.b64encode(SHA256.new(messageBodyJson.encode()).digest())
headers = {'host': domain,'digest': f'SHA-256={bodyDigest}'}
headers = {'(request-target)': f'post {path}','host': domain,'date': dateStr,'digest': f'SHA-256={bodyDigest}','content-type': 'application/json'}
privateKeyPem = RSA.import_key(privateKeyPem)
headers.update({
'(request-target)': f'post {path}',
})
#headers.update({
# '(request-target)': f'post {path}',
#})
# build a digest for signing
signedHeaderKeys = headers.keys()
signedHeaderText = ''
for headerKey in signedHeaderKeys:
signedHeaderText += f'{headerKey}: {headers[headerKey]}\n'
#print(f'headerKey: {headerKey}: {headers[headerKey]}')
signedHeaderText = signedHeaderText.strip()
headerDigest = SHA256.new(signedHeaderText.encode('ascii'))
@ -53,13 +54,11 @@ def signPostHeaders(privateKeyPem: str, nickname: str, domain: str, \
signatureDict = {
'keyId': keyID,
'algorithm': 'rsa-sha256',
# 'date': dateStr,
'headers': ' '.join(signedHeaderKeys),
'signature': signature
}
signatureHeader = ','.join(
[f'{k}="{v}"' for k, v in signatureDict.items()])
# print('signatureHeader: '+str(signatureHeader))
return signatureHeader
def createSignedHeader(privateKeyPem: str,nickname: str,domain: str,port: int, \
@ -67,27 +66,27 @@ def createSignedHeader(privateKeyPem: str,nickname: str,domain: str,port: int, \
messageBodyJson: {}) -> {}:
headerDomain=domain
if port!=80 and port!=443:
headerDomain=headerDomain+':'+str(port)
if port:
if port!=80 and port!=443:
headerDomain=headerDomain+':'+str(port)
dateStr=strftime("%a, %d %b %Y %H:%M:%S %Z", gmtime())
path='/inbox'
if not withDigest:
headers = {'host': headerDomain}
headers = {'(request-target)': f'post {path}','host': headerDomain,'date': dateStr}
else:
messageBodyJsonStr=json.dumps(messageBodyJson)
bodyDigest = \
base64.b64encode(SHA256.new(messageBodyJsonStr.encode()).digest())
headers = {'host': headerDomain, 'digest': f'SHA-256={bodyDigest}'}
path='/inbox'
headers = {'(request-target)': f'post {path}','host': headerDomain,'date': dateStr,'digest': f'SHA-256={bodyDigest}','content-type': 'application/json'}
signatureHeader = signPostHeaders(privateKeyPem, nickname, domain, port, \
path, httpPrefix, None)
headers['date'] = dateStr
headers['signature'] = signatureHeader
headers['Content-type'] = 'application/json'
#print('******************************************************http headers: '+str(headers))
return headers
def verifyPostHeaders(httpPrefix: str, publicKeyPem: str, headers: dict, \
path: str, GETmethod: bool, \
def verifyPostHeaders(httpPrefix: str,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.
@ -109,6 +108,8 @@ def verifyPostHeaders(httpPrefix: str, publicKeyPem: str, headers: dict, \
k: v[1:-1]
for k, v in [i.split('=', 1) for i in signatureHeader.split(',')]
}
#print('signatureHeader: '+str(signatureHeader))
#print('signatureDict: '+str(signatureDict))
# Unpack the signed headers and set values based on current headers and
# body (if a digest was included)
@ -117,21 +118,19 @@ def verifyPostHeaders(httpPrefix: str, publicKeyPem: str, headers: dict, \
if signedHeader == '(request-target)':
signedHeaderList.append(
f'(request-target): {method.lower()} {path}')
elif signedHeader.lower() == 'content-type':
continue
elif signedHeader == 'date':
continue
elif signedHeader == 'digest':
bodyDigest = \
base64.b64encode(SHA256.new(messageBodyJsonStr.encode()).digest())
signedHeaderList.append(f'digest: SHA-256={bodyDigest}')
else:
try:
if headers.get(signedHeader):
signedHeaderList.append(
f'{signedHeader}: {headers[signedHeader]}')
except Exception as e:
print('http signature check failure')
print(e)
else:
signedHeaderCap=signedHeader.capitalize()
if headers.get(signedHeaderCap):
signedHeaderList.append(
f'{signedHeader}: {headers[signedHeaderCap]}')
# Now we have our header data digest
signedHeaderText = '\n'.join(signedHeaderList)

11
inbox.py
View File

@ -164,7 +164,7 @@ def validPublishedDate(published) -> bool:
return False
return True
def savePostToInboxQueue(baseDir: str,httpPrefix: str,nickname: str, domain: str,postJsonObject: {},host: str,headers: str,postPath: str,debug: bool) -> str:
def savePostToInboxQueue(baseDir: str,httpPrefix: str,nickname: str, domain: str,postJsonObject: {},host: str,headers: str,httpHeaders: str,postPath: str,debug: bool) -> str:
"""Saves the give json to the inbox queue for the person
keyId specifies the actor sending the post
"""
@ -225,6 +225,7 @@ def savePostToInboxQueue(baseDir: str,httpPrefix: str,nickname: str, domain: str
'published': published,
'host': host,
'headers': headers,
'httpHeaders': httpHeaders,
'path': postPath,
'post': postJsonObject,
'filename': filename,
@ -1135,15 +1136,11 @@ def runInboxQueue(projectVersion: str, \
continue
# check the signature
verifyHeaders={
'host': queueJson['host'],
'signature': queueJson['headers']
}
if debug:
print('DEBUG: checking http headers')
pprint(verifyHeaders)
pprint(queueJson['headers'])
if not verifyPostHeaders(httpPrefix, \
pubKey,verifyHeaders, \
pubKey,queueJson['headers'], \
queueJson['path'],False, \
json.dumps(queueJson['post'])):
if debug:

4
session.py
View File

@ -40,7 +40,7 @@ def getJson(session,url: str,headers: {},params: {}, \
print('WARN: no session specified for getJson')
session.cookies.clear()
try:
result=session.get(url, headers=sessionHeaders, params=sessionParams, allow_redirects=True)
result=session.get(url, headers=sessionHeaders, params=sessionParams)
return result.json()
except Exception as e:
print('ERROR: getJson failed')
@ -62,7 +62,7 @@ def postJson(session,postJsonObject: {},federationList: [],inboxUrl: str,headers
print('postJson: '+inboxUrl+' not permitted')
return None
postResult = session.post(url = inboxUrl, data = json.dumps(postJsonObject), headers=headers, allow_redirects=True)
postResult = session.post(url = inboxUrl, data = json.dumps(postJsonObject), headers=headers)
return postResult.text
def postImage(session,attachImageFilename: str,federationList: [],inboxUrl: str,headers: {},capability: str) -> str:

28
tests.py
View File

@ -82,7 +82,7 @@ def testHttpsigBase(withDigest):
privateKeyPem,publicKeyPem,person,wfEndpoint= \
createPerson(path,nickname,domain,port,httpPrefix,False,password)
assert privateKeyPem
messageBodyJsonStr = '{"a key": "a value", "another key": "A string","yet another key": "A string"}'
messageBodyJsonStr = '{"a key": "a value", "another key": "A string","yet another key": "Another string"}'
headersDomain=domain
if port!=80 and port !=443:
@ -90,39 +90,43 @@ def testHttpsigBase(withDigest):
dateStr=strftime("%a, %d %b %Y %H:%M:%S %Z", gmtime())
if not withDigest:
headers = {'host': headersDomain}
headers = {'host': headersDomain,'date': dateStr,'content-type': 'application/json'}
else:
bodyDigest = \
base64.b64encode(SHA256.new(messageBodyJsonStr.encode()).digest())
headers = {'host': headersDomain, 'date': dateStr, 'digest': f'SHA-256={bodyDigest}'}
headers = {'host': headersDomain,'date': dateStr,'digest': f'SHA-256={bodyDigest}','content-type': 'application/json'}
boxpath='/inbox'
signatureHeader = \
signPostHeaders(privateKeyPem, nickname, domain, port, boxpath, httpPrefix, None)
headers['signature'] = signatureHeader
assert verifyPostHeaders(httpPrefix, publicKeyPem, headers, \
'/inbox' ,False, messageBodyJsonStr)
'/inbox' ,False, \
messageBodyJsonStr)
assert verifyPostHeaders(httpPrefix, publicKeyPem, headers, \
'/parambulator/inbox', False , messageBodyJsonStr) == False
'/parambulator/inbox',False, \
messageBodyJsonStr) == False
assert verifyPostHeaders(httpPrefix, publicKeyPem, headers, \
'/inbox', True, messageBodyJsonStr) == False
'/inbox',True, \
messageBodyJsonStr) == False
if not withDigest:
# fake domain
headers = {'host': 'bogon.domain'}
headers = {'host': 'bogon.domain','date': dateStr,'content-type': 'application/json'}
else:
# correct domain but fake message
messageBodyJsonStr = '{"a key": "a value", "another key": "Fake GNUs", "yet another key": "Fake GNUs"}'
messageBodyJsonStr = '{"a key": "a value", "another key": "Fake GNUs", "yet another key": "More Fake GNUs"}'
bodyDigest = base64.b64encode(SHA256.new(messageBodyJsonStr.encode()).digest())
headers = {'host': domain, 'date': dateStr, 'digest': f'SHA-256={bodyDigest}'}
headers = {'host': domain,'date': dateStr,'digest': f'SHA-256={bodyDigest}','content-type': 'application/json'}
headers['signature'] = signatureHeader
assert verifyPostHeaders(httpPrefix, publicKeyPem, headers, \
'/inbox', True, messageBodyJsonStr) == False
assert verifyPostHeaders(httpPrefix,publicKeyPem,headers, \
'/inbox',True, \
messageBodyJsonStr) == False
os.chdir(baseDir)
shutil.rmtree(path)
def testHttpsig():
testHttpsigBase(False)
testHttpsigBase(True)
testHttpsigBase(False)
def testCache():
print('testCache')