epicyon/httpsig.py

472 lines
18 KiB
Python
Raw Normal View History

__filename__ = "httpsig.py"
2020-04-03 12:05:30 +00:00
__author__ = "Bob Mottram"
__credits__ = ['lamia']
__license__ = "AGPL3+"
2021-01-26 10:07:42 +00:00
__version__ = "1.2.0"
2020-04-03 12:05:30 +00:00
__maintainer__ = "Bob Mottram"
2021-09-10 16:14:50 +00:00
__email__ = "bob@libreserver.org"
2020-04-03 12:05:30 +00:00
__status__ = "Production"
2021-06-15 15:08:12 +00:00
__module_group__ = "Security"
2019-06-28 18:55:29 +00:00
2019-08-15 22:33:42 +00:00
# see https://tools.ietf.org/html/draft-cavage-http-signatures-06
2021-02-21 22:51:08 +00:00
#
# This might change in future
# see https://tools.ietf.org/html/draft-ietf-httpbis-message-signatures-01
2019-08-15 22:33:42 +00:00
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives.serialization import load_pem_private_key
from cryptography.hazmat.primitives.serialization import load_pem_public_key
from cryptography.hazmat.primitives.asymmetric import padding
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.asymmetric import utils as hazutils
2019-06-28 18:55:29 +00:00
import base64
2019-08-15 09:08:18 +00:00
from time import gmtime, strftime
2019-08-23 11:20:20 +00:00
import datetime
2020-12-16 11:38:40 +00:00
from utils import getFullDomain
from utils import getSHA256
2021-09-08 10:05:45 +00:00
from utils import getSHA512
2021-08-14 11:13:39 +00:00
from utils import localActorUrl
def messageContentDigest(messageBodyJsonStr: str) -> str:
2020-04-03 12:05:30 +00:00
msg = messageBodyJsonStr.encode('utf-8')
hashResult = getSHA256(msg)
return base64.b64encode(hashResult).decode('utf-8')
2020-04-03 12:05:30 +00:00
2020-04-03 12:05:30 +00:00
def signPostHeaders(dateStr: str, privateKeyPem: str,
nickname: str,
domain: str, port: int,
toDomain: str, toPort: int,
path: str,
httpPrefix: str,
2021-09-15 10:44:44 +00:00
messageBodyJsonStr: str,
contentType: str) -> str:
2019-06-28 18:55:29 +00:00
"""Returns a raw signature string that can be plugged into a header and
used to verify the authenticity of an HTTP transmission.
"""
2020-12-16 11:38:40 +00:00
domain = getFullDomain(domain, port)
2019-07-01 09:31:02 +00:00
2020-12-16 11:38:40 +00:00
toDomain = getFullDomain(toDomain, toPort)
2019-08-16 13:47:01 +00:00
if not dateStr:
2020-04-03 12:05:30 +00:00
dateStr = strftime("%a, %d %b %Y %H:%M:%S %Z", gmtime())
2021-09-08 10:05:45 +00:00
if nickname != domain and nickname.lower() != 'actor':
keyID = localActorUrl(httpPrefix, nickname, domain)
2021-08-31 21:02:58 +00:00
else:
# instance actor
2021-09-08 10:05:45 +00:00
keyID = httpPrefix + '://' + domain + '/actor'
keyID += '#main-key'
if not messageBodyJsonStr:
2020-04-03 12:05:30 +00:00
headers = {
2021-09-01 18:46:28 +00:00
'(request-target)': f'get {path}',
2020-04-03 12:05:30 +00:00
'host': toDomain,
'date': dateStr,
2021-09-15 10:44:44 +00:00
'accept': contentType
2020-04-03 12:05:30 +00:00
}
2019-06-28 18:55:29 +00:00
else:
2020-04-03 12:05:30 +00:00
bodyDigest = messageContentDigest(messageBodyJsonStr)
contentLength = len(messageBodyJsonStr)
headers = {
'(request-target)': f'post {path}',
'host': toDomain,
'date': dateStr,
'digest': f'SHA-256={bodyDigest}',
'content-type': 'application/activity+json',
'content-length': str(contentLength)
}
key = load_pem_private_key(privateKeyPem.encode('utf-8'),
None, backend=default_backend())
2020-04-03 12:05:30 +00:00
# headers.update({
# '(request-target)': f'post {path}',
# })
2019-06-28 18:55:29 +00:00
# build a digest for signing
2020-04-03 12:05:30 +00:00
signedHeaderKeys = headers.keys()
signedHeaderText = ''
2019-06-28 18:55:29 +00:00
for headerKey in signedHeaderKeys:
signedHeaderText += f'{headerKey}: {headers[headerKey]}\n'
2021-09-08 10:05:45 +00:00
# strip the trailing linefeed
signedHeaderText = signedHeaderText.rstrip('\n')
# signedHeaderText.encode('ascii') matches
headerDigest = getSHA256(signedHeaderText.encode('ascii'))
# print('headerDigest2: ' + str(headerDigest))
2019-06-28 18:55:29 +00:00
# Sign the digest
rawSignature = key.sign(headerDigest,
padding.PKCS1v15(),
hazutils.Prehashed(hashes.SHA256()))
2020-04-03 12:05:30 +00:00
signature = base64.b64encode(rawSignature).decode('ascii')
2019-06-28 18:55:29 +00:00
# Put it into a valid HTTP signature format
2020-04-03 12:05:30 +00:00
signatureDict = {
2019-06-28 18:55:29 +00:00
'keyId': keyID,
'algorithm': 'rsa-sha256',
'headers': ' '.join(signedHeaderKeys),
'signature': signature
}
2020-04-03 12:05:30 +00:00
signatureHeader = ','.join(
2019-06-28 18:55:29 +00:00
[f'{k}="{v}"' for k, v in signatureDict.items()])
return signatureHeader
2020-04-03 12:05:30 +00:00
def signPostHeadersNew(dateStr: str, privateKeyPem: str,
nickname: str,
domain: str, port: int,
toDomain: str, toPort: int,
path: str,
httpPrefix: str,
messageBodyJsonStr: str,
algorithm: str) -> (str, str):
"""Returns a raw signature strings that can be plugged into a header
as "Signature-Input" and "Signature"
used to verify the authenticity of an HTTP transmission.
See https://tools.ietf.org/html/draft-ietf-httpbis-message-signatures-01
"""
domain = getFullDomain(domain, port)
toDomain = getFullDomain(toDomain, toPort)
timeFormat = "%a, %d %b %Y %H:%M:%S %Z"
if not dateStr:
currTime = gmtime()
dateStr = strftime(timeFormat, currTime)
else:
currTime = datetime.datetime.strptime(dateStr, timeFormat)
2021-04-04 21:30:26 +00:00
secondsSinceEpoch = \
int((currTime - datetime.datetime(1970, 1, 1)).total_seconds())
2021-08-14 11:13:39 +00:00
keyID = localActorUrl(httpPrefix, nickname, domain) + '#main-key'
if not messageBodyJsonStr:
headers = {
'*request-target': f'post {path}',
'*created': str(secondsSinceEpoch),
'host': toDomain,
'date': dateStr,
'content-type': 'application/json'
}
else:
bodyDigest = messageContentDigest(messageBodyJsonStr)
contentLength = len(messageBodyJsonStr)
headers = {
'*request-target': f'post {path}',
'*created': str(secondsSinceEpoch),
'host': toDomain,
'date': dateStr,
'digest': f'SHA-256={bodyDigest}',
'content-type': 'application/activity+json',
'content-length': str(contentLength)
}
key = load_pem_private_key(privateKeyPem.encode('utf-8'),
None, backend=default_backend())
# build a digest for signing
signedHeaderKeys = headers.keys()
signedHeaderText = ''
for headerKey in signedHeaderKeys:
signedHeaderText += f'{headerKey}: {headers[headerKey]}\n'
signedHeaderText = signedHeaderText.strip()
# Sign the digest. Potentially other signing algorithms can be added here.
signature = ''
2021-09-08 10:05:45 +00:00
if algorithm == 'rsa-sha512':
headerDigest = getSHA512(signedHeaderText.encode('ascii'))
rawSignature = key.sign(headerDigest,
padding.PKCS1v15(),
hazutils.Prehashed(hashes.SHA512()))
signature = base64.b64encode(rawSignature).decode('ascii')
else:
2021-11-18 14:04:45 +00:00
# default rsa-sha256
2021-09-08 10:05:45 +00:00
headerDigest = getSHA256(signedHeaderText.encode('ascii'))
rawSignature = key.sign(headerDigest,
padding.PKCS1v15(),
hazutils.Prehashed(hashes.SHA256()))
signature = base64.b64encode(rawSignature).decode('ascii')
sigKey = 'sig1'
# Put it into a valid HTTP signature format
signatureInputDict = {
'keyId': keyID,
}
signatureIndexHeader = '; '.join(
[f'{k}="{v}"' for k, v in signatureInputDict.items()])
signatureIndexHeader += '; alg=hs2019'
signatureIndexHeader += '; created=' + str(secondsSinceEpoch)
signatureIndexHeader += \
'; ' + sigKey + '=(' + ', '.join(signedHeaderKeys) + ')'
signatureDict = {
sigKey: signature
}
signatureHeader = '; '.join(
[f'{k}=:{v}:' for k, v in signatureDict.items()])
2021-04-04 21:30:26 +00:00
return signatureIndexHeader, signatureHeader
def createSignedHeader(dateStr: str, privateKeyPem: str, nickname: str,
2020-04-03 12:05:30 +00:00
domain: str, port: int,
toDomain: str, toPort: int,
path: str, httpPrefix: str, withDigest: bool,
2021-09-14 22:03:26 +00:00
messageBodyJsonStr: str,
contentType: str) -> {}:
2019-08-16 13:47:01 +00:00
"""Note that the domain is the destination, not the sender
"""
2020-12-16 11:42:11 +00:00
headerDomain = getFullDomain(toDomain, toPort)
2019-07-01 09:31:02 +00:00
2021-09-15 10:44:44 +00:00
# if no date is given then create one
if not dateStr:
dateStr = strftime("%a, %d %b %Y %H:%M:%S %Z", gmtime())
2021-09-15 10:44:44 +00:00
# Content-Type or Accept header
2021-09-14 22:03:26 +00:00
if not contentType:
contentType = 'application/activity+json'
2021-09-15 10:44:44 +00:00
2021-09-14 22:03:26 +00:00
if not withDigest:
2020-04-03 12:05:30 +00:00
headers = {
2021-09-01 20:57:54 +00:00
'(request-target)': f'get {path}',
'host': headerDomain,
'date': dateStr,
'accept': contentType
2020-03-22 20:36:19 +00:00
}
2020-04-03 12:05:30 +00:00
signatureHeader = \
signPostHeaders(dateStr, privateKeyPem, nickname,
domain, port, toDomain, toPort,
2021-09-15 10:44:44 +00:00
path, httpPrefix, None, contentType)
2019-07-01 09:31:02 +00:00
else:
2020-04-03 12:05:30 +00:00
bodyDigest = messageContentDigest(messageBodyJsonStr)
contentLength = len(messageBodyJsonStr)
headers = {
2020-03-22 20:36:19 +00:00
'(request-target)': f'post {path}',
'host': headerDomain,
'date': dateStr,
'digest': f'SHA-256={bodyDigest}',
'content-length': str(contentLength),
'content-type': contentType
}
2020-04-03 12:05:30 +00:00
signatureHeader = \
signPostHeaders(dateStr, privateKeyPem, nickname,
domain, port,
toDomain, toPort,
2021-09-15 10:44:44 +00:00
path, httpPrefix, messageBodyJsonStr,
contentType)
2020-04-03 12:05:30 +00:00
headers['signature'] = signatureHeader
2019-07-01 09:31:02 +00:00
return headers
2020-04-03 12:05:30 +00:00
def _verifyRecentSignature(signedDateStr: str) -> bool:
2019-08-23 11:31:46 +00:00
"""Checks whether the given time taken from the header is within
12 hours of the current time
"""
2020-04-03 12:05:30 +00:00
currDate = datetime.datetime.utcnow()
dateFormat = "%a, %d %b %Y %H:%M:%S %Z"
signedDate = datetime.datetime.strptime(signedDateStr, dateFormat)
timeDiffSec = (currDate - signedDate).seconds
2019-08-23 11:39:16 +00:00
# 12 hours tollerance
if timeDiffSec > 43200:
2020-04-03 12:05:30 +00:00
print('WARN: Header signed too long ago: ' + signedDateStr)
print(str(timeDiffSec / (60 * 60)) + ' hours')
return False
if timeDiffSec < 0:
2020-04-03 12:05:30 +00:00
print('WARN: Header signed in the future! ' + signedDateStr)
print(str(timeDiffSec / (60 * 60)) + ' hours')
2019-08-23 11:30:37 +00:00
return False
return True
2020-04-03 12:05:30 +00:00
def verifyPostHeaders(httpPrefix: str, publicKeyPem: str, headers: dict,
path: str, GETmethod: bool,
messageBodyDigest: str,
messageBodyJsonStr: str, debug: bool,
2021-06-20 11:28:35 +00:00
noRecencyCheck: bool = False) -> bool:
2019-06-28 18:55:29 +00:00
"""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
2019-07-01 09:31:02 +00:00
messageBodyJsonStr - the received request body (used for digest)
2019-06-28 18:55:29 +00:00
"""
2019-08-23 11:20:20 +00:00
2019-06-28 18:55:29 +00:00
if GETmethod:
2020-04-03 12:05:30 +00:00
method = 'GET'
2019-06-28 18:55:29 +00:00
else:
2020-04-03 12:05:30 +00:00
method = 'POST'
2019-11-12 15:03:17 +00:00
if debug:
2020-04-03 12:05:30 +00:00
print('DEBUG: verifyPostHeaders ' + method)
2021-03-14 15:00:43 +00:00
print('verifyPostHeaders publicKeyPem: ' + str(publicKeyPem))
print('verifyPostHeaders headers: ' + str(headers))
2021-03-14 15:25:49 +00:00
print('verifyPostHeaders messageBodyJsonStr: ' +
str(messageBodyJsonStr))
2020-03-22 21:16:02 +00:00
pubkey = load_pem_public_key(publicKeyPem.encode('utf-8'),
backend=default_backend())
2019-06-28 18:55:29 +00:00
# Build a dictionary of the signature values
if headers.get('Signature-Input'):
signatureHeader = headers['Signature-Input']
fieldSep2 = ','
# split the signature input into separate fields
signatureDict = {
k.strip(): v.strip()
for k, v in [i.split('=', 1) for i in signatureHeader.split(';')]
}
requestTargetKey = None
requestTargetStr = None
for k, v in signatureDict.items():
if v.startswith('('):
requestTargetKey = k
requestTargetStr = v[1:-1]
break
if not requestTargetKey:
return False
signatureDict[requestTargetKey] = requestTargetStr
else:
requestTargetKey = 'headers'
signatureHeader = headers['signature']
fieldSep2 = ' '
# split the signature input into separate fields
signatureDict = {
k: v[1:-1]
for k, v in [i.split('=', 1) for i in signatureHeader.split(',')]
}
2019-06-28 18:55:29 +00:00
if debug:
print('signatureDict: ' + str(signatureDict))
2019-06-28 18:55:29 +00:00
# Unpack the signed headers and set values based on current headers and
# body (if a digest was included)
2020-04-03 12:05:30 +00:00
signedHeaderList = []
algorithm = 'rsa-sha256'
for signedHeader in signatureDict[requestTargetKey].split(fieldSep2):
signedHeader = signedHeader.strip()
2019-11-12 15:03:17 +00:00
if debug:
2020-04-03 12:05:30 +00:00
print('DEBUG: verifyPostHeaders signedHeader=' + signedHeader)
2019-06-28 18:55:29 +00:00
if signedHeader == '(request-target)':
# original Mastodon http signature
2020-04-03 12:05:30 +00:00
appendStr = f'(request-target): {method.lower()} {path}'
signedHeaderList.append(appendStr)
elif '*request-target' in signedHeader:
# https://tools.ietf.org/html/
# draft-ietf-httpbis-message-signatures-01
appendStr = f'*request-target: {method.lower()} {path}'
# remove ()
# if appendStr.startswith('('):
# appendStr = appendStr.split('(')[1]
# if ')' in appendStr:
# appendStr = appendStr.split(')')[0]
signedHeaderList.append(appendStr)
elif signedHeader == 'algorithm':
if headers.get(signedHeader):
algorithm = headers[signedHeader]
2019-06-28 18:55:29 +00:00
elif signedHeader == 'digest':
if messageBodyDigest:
2020-04-03 12:05:30 +00:00
bodyDigest = messageBodyDigest
else:
2020-04-03 12:05:30 +00:00
bodyDigest = messageContentDigest(messageBodyJsonStr)
2019-06-28 18:55:29 +00:00
signedHeaderList.append(f'digest: SHA-256={bodyDigest}')
2019-11-12 18:48:29 +00:00
elif signedHeader == 'content-length':
2019-11-12 19:20:55 +00:00
if headers.get(signedHeader):
2020-04-03 12:05:30 +00:00
appendStr = f'content-length: {headers[signedHeader]}'
signedHeaderList.append(appendStr)
2021-09-01 14:22:11 +00:00
elif headers.get('Content-Length'):
contentLength = headers['Content-Length']
signedHeaderList.append(f'content-length: {contentLength}')
elif headers.get('Content-length'):
contentLength = headers['Content-length']
appendStr = f'content-length: {contentLength}'
signedHeaderList.append(appendStr)
2019-11-12 17:16:34 +00:00
else:
2021-09-01 14:22:11 +00:00
if debug:
print('DEBUG: verifyPostHeaders ' + signedHeader +
' not found in ' + str(headers))
2019-06-28 18:55:29 +00:00
else:
2019-08-15 21:34:25 +00:00
if headers.get(signedHeader):
if signedHeader == 'date' and not noRecencyCheck:
if not _verifyRecentSignature(headers[signedHeader]):
2019-11-12 15:03:17 +00:00
if debug:
2020-04-03 12:05:30 +00:00
print('DEBUG: ' +
'verifyPostHeaders date is not recent ' +
headers[signedHeader])
2019-08-23 11:30:37 +00:00
return False
signedHeaderList.append(
f'{signedHeader}: {headers[signedHeader]}')
2019-08-15 21:34:25 +00:00
else:
2021-03-14 11:53:13 +00:00
if '-' in signedHeader:
2021-03-14 12:09:56 +00:00
# capitalise with dashes
# my-header becomes My-Header
2021-03-14 11:53:13 +00:00
headerParts = signedHeader.split('-')
2021-03-14 12:09:56 +00:00
signedHeaderCap = None
2021-03-14 11:53:13 +00:00
for part in headerParts:
if signedHeaderCap:
2021-03-14 12:09:56 +00:00
signedHeaderCap += '-' + part.capitalize()
else:
signedHeaderCap = part.capitalize()
2021-03-14 11:53:13 +00:00
else:
2021-03-14 12:09:56 +00:00
# header becomes Header
2021-03-14 11:53:13 +00:00
signedHeaderCap = signedHeader.capitalize()
2021-03-14 12:09:56 +00:00
if debug:
2021-03-14 15:00:43 +00:00
print('signedHeaderCap: ' + signedHeaderCap)
2021-03-14 12:09:56 +00:00
# if this is the date header then check it is recent
2020-04-03 12:05:30 +00:00
if signedHeaderCap == 'Date':
if not _verifyRecentSignature(headers[signedHeaderCap]):
2019-11-12 15:03:17 +00:00
if debug:
2020-04-03 12:05:30 +00:00
print('DEBUG: ' +
'verifyPostHeaders date is not recent ' +
headers[signedHeader])
2019-08-23 11:30:37 +00:00
return False
2021-03-14 12:09:56 +00:00
# add the capitalised header
2019-08-15 21:34:25 +00:00
if headers.get(signedHeaderCap):
signedHeaderList.append(
f'{signedHeader}: {headers[signedHeaderCap]}')
2021-03-14 18:34:30 +00:00
elif '-' in signedHeader:
# my-header becomes My-header
2021-03-14 18:29:10 +00:00
signedHeaderCap = signedHeader.capitalize()
if headers.get(signedHeaderCap):
signedHeaderList.append(
f'{signedHeader}: {headers[signedHeaderCap]}')
2019-06-28 18:55:29 +00:00
# Now we have our header data digest
2020-04-03 12:05:30 +00:00
signedHeaderText = '\n'.join(signedHeaderList)
if debug:
print('signedHeaderText:\n' + signedHeaderText + 'END')
2019-06-28 18:55:29 +00:00
# Get the signature, verify with public key, return result
signature = None
if headers.get('Signature-Input') and headers.get('Signature'):
# https://tools.ietf.org/html/
# draft-ietf-httpbis-message-signatures-01
headersSig = headers['Signature']
# remove sig1=:
if requestTargetKey + '=:' in headersSig:
headersSig = headersSig.split(requestTargetKey + '=:')[1]
headersSig = headersSig[:len(headersSig)-1]
signature = base64.b64decode(headersSig)
else:
# Original Mastodon signature
signature = base64.b64decode(signatureDict['signature'])
2021-09-15 10:44:44 +00:00
if debug:
print('signature: ' + algorithm + ' ' +
signatureDict['signature'])
2019-06-28 18:55:29 +00:00
# If extra signing algorithms need to be added then do it here
if algorithm == 'rsa-sha256':
headerDigest = getSHA256(signedHeaderText.encode('ascii'))
paddingStr = padding.PKCS1v15()
alg = hazutils.Prehashed(hashes.SHA256())
elif algorithm == 'rsa-sha512':
headerDigest = getSHA512(signedHeaderText.encode('ascii'))
paddingStr = padding.PKCS1v15()
alg = hazutils.Prehashed(hashes.SHA512())
else:
print('Unknown http signature algorithm: ' + algorithm)
paddingStr = padding.PKCS1v15()
alg = hazutils.Prehashed(hashes.SHA256())
headerDigest = ''
2019-06-28 18:55:29 +00:00
try:
pubkey.verify(signature, headerDigest, paddingStr, alg)
2019-06-28 18:55:29 +00:00
return True
except BaseException:
2019-11-12 15:03:17 +00:00
if debug:
2021-10-29 18:48:15 +00:00
print('EX: verifyPostHeaders pkcs1_15 verify failure')
return False