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

Fix http signature with port number

master
Bob Mottram 2019-07-01 10:31:02 +01:00
parent b0a85e6b4e
commit d9d5ce94dc
3 changed files with 53 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

32
httpsig.py
View File

@ -7,7 +7,6 @@ __maintainer__ = "Bob Mottram"
__email__ = "bob@freedombone.net"
__status__ = "Production"
from person import createPerson
from Crypto.PublicKey import RSA
from Crypto.Hash import SHA256
#from Crypto.Signature import PKCS1_v1_5
@ -16,14 +15,18 @@ from requests.auth import AuthBase
import base64
import json
def signPostHeaders(privateKeyPem: str, username: str, domain: str, 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.
"""
prefix='https'
if not https:
prefix='http'
keyID = prefix+'://'+domain+'/'+username+'#main-key'
if port!=80 and port!=443:
domain=domain+':'+str(port)
keyID = prefix+'://'+domain+'/users/'+username+'/main-key'
if not messageBodyJson:
headers = {'host': domain}
else:
@ -56,14 +59,31 @@ def signPostHeaders(privateKeyPem: str, username: str, domain: str, path: str, h
[f'{k}="{v}"' for k, v in signatureDict.items()])
return signatureHeader
def verifyPostHeaders(https: bool, publicKeyPem: str, headers: dict, path: str, GETmethod: bool, messageBodyJson: str) -> bool:
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:
headerDomain=headerDomain+':'+str(port)
if not withDigest:
headers = {'host': headerDomain}
else:
messageBodyJsonStr=json.dumps(messageBodyJson)
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)
headers['signature'] = signatureHeader
return headers
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
headers - should be a dictionary of request headers
path - the relative url that was requested from this site
GETmethod - GET or POST
messageBodyJson - the received request body (used for digest)
messageBodyJsonStr - the received request body (used for digest)
"""
if GETmethod:
method='GET'
@ -90,7 +110,7 @@ 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(messageBodyJson.encode()).digest())
bodyDigest = base64.b64encode(SHA256.new(messageBodyJsonStr.encode()).digest())
signedHeaderList.append(f'digest: SHA-256={bodyDigest}')
else:
signedHeaderList.append(

11
posts.py
View File

@ -24,6 +24,7 @@ from random import randint
from session import getJson
from session import postJson
from webfinger import webfingerHandle
from httpsig import createSignedHeader
try:
from BeautifulSoup import BeautifulSoup
except ImportError:
@ -342,6 +343,8 @@ def sendPost(session,baseDir,username: str, domain: str, port: int, toUsername:
if not https:
prefix='http'
withDigest=True
if toPort!=80 and toPort!=443:
toDomain=toDomain+':'+str(toPort)
@ -369,14 +372,14 @@ def sendPost(session,baseDir,username: str, domain: str, port: int, toUsername:
return 5
# construct the http header
signatureHeader = signPostHeaders(privateKeyPem, username, domain, '/inbox', https, postJsonObject)
signatureHeader['Content-type'] = 'application/json'
signatureHeaderJson = createSignedHeader(privateKeyPem, username, domain, port, '/inbox', https, withDigest, postJsonObject)
signatureHeaderJson['Content-type'] = 'application/json'
print("*************signatureHeaderJson "+str(signatureHeaderJson))
# 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,signatureHeader.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

32
tests.py
View File

@ -34,31 +34,37 @@ def testHttpsigBase(withDigest):
username='socrates'
domain='argumentative.social'
https=True
port=80
port=5576
baseDir=os.getcwd()
privateKeyPem,publicKeyPem,person,wfEndpoint=createPerson(baseDir,username,domain,port,https,False)
messageBodyJson = '{"a key": "a value", "another key": "A string"}'
messageBodyJsonStr = '{"a key": "a value", "another key": "A string"}'
headersDomain=domain
if port!=80 and port !=443:
headersDomain=domain+':'+str(port)
if not withDigest:
headers = {'host': domain}
headers = {'host': headersDomain}
else:
bodyDigest = base64.b64encode(SHA256.new(messageBodyJson.encode()).digest())
headers = {'host': domain, 'digest': f'SHA-256={bodyDigest}'}
bodyDigest = base64.b64encode(SHA256.new(messageBodyJsonStr.encode()).digest())
headers = {'host': headersDomain, 'digest': f'SHA-256={bodyDigest}'}
path='/inbox'
signatureHeader = signPostHeaders(privateKeyPem, username, domain, path, https, None)
signatureHeader = signPostHeaders(privateKeyPem, username, domain, port, path, https, None)
headers['signature'] = signatureHeader
assert verifyPostHeaders(https, publicKeyPem, headers, '/inbox' ,False, messageBodyJson)
assert verifyPostHeaders(https, publicKeyPem, headers, '/parambulator/inbox', False , messageBodyJson) == False
assert verifyPostHeaders(https, publicKeyPem, headers, '/inbox', True, messageBodyJson) == False
assert verifyPostHeaders(https, publicKeyPem, headers, '/inbox' ,False, messageBodyJsonStr)
assert verifyPostHeaders(https, publicKeyPem, headers, '/parambulator/inbox', False , messageBodyJsonStr) == False
assert verifyPostHeaders(https, publicKeyPem, headers, '/inbox', True, messageBodyJsonStr) == False
if not withDigest:
# fake domain
headers = {'host': 'bogon.domain'}
else:
# correct domain but fake message
messageBodyJson = '{"a key": "a value", "another key": "Fake GNUs"}'
bodyDigest = base64.b64encode(SHA256.new(messageBodyJson.encode()).digest())
messageBodyJsonStr = '{"a key": "a value", "another key": "Fake GNUs"}'
bodyDigest = base64.b64encode(SHA256.new(messageBodyJsonStr.encode()).digest())
headers = {'host': domain, 'digest': f'SHA-256={bodyDigest}'}
headers['signature'] = signatureHeader
assert verifyPostHeaders(https, publicKeyPem, headers, '/inbox', True, messageBodyJson) == False
assert verifyPostHeaders(https, publicKeyPem, headers, '/inbox', True, messageBodyJsonStr) == False
def testHttpsig():
testHttpsigBase(False)
@ -179,7 +185,7 @@ def testPostMessageBetweenServers():
sendResult = sendPost(sessionAlice,aliceDir,'alice', '127.0.0.1', alicePort, 'bob', '127.0.0.1', bobPort, '', https, 'Why is a mouse when it spins?', False, True, federationList, aliceSendThreads, alicePostLog, inReplyTo, inReplyToAtomUri, subject)
print('sendResult: '+str(sendResult))
time.sleep(3)
time.sleep(5)
# stop the servers
thrAlice.kill()