epicyon/httpsig.py

570 lines
23 KiB
Python
Raw Normal View History

__filename__ = "httpsig.py"
2020-04-03 12:05:30 +00:00
__author__ = "Bob Mottram"
__credits__ = ['lamia']
__license__ = "AGPL3+"
2023-01-21 23:03:30 +00:00
__version__ = "1.4.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
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
2021-12-26 12:45:03 +00:00
from utils import get_full_domain
2021-12-26 12:13:46 +00:00
from utils import get_sha_256
from utils import get_sha_512
2021-12-26 10:19:59 +00:00
from utils import local_actor_url
2023-11-20 22:27:58 +00:00
from utils import date_utcnow
from utils import date_epoch
from utils import date_from_string_format
2022-01-02 15:41:05 +00:00
def message_content_digest(message_body_json_str: str,
digest_algorithm: str) -> str:
2021-11-23 11:41:40 +00:00
"""Returns the digest for the message body
"""
2022-01-02 15:41:05 +00:00
msg = message_body_json_str.encode('utf-8')
if digest_algorithm in ('rsa-sha512', 'rsa-pss-sha512'):
hash_result = get_sha_512(msg)
2021-11-23 11:41:40 +00:00
else:
2022-01-02 15:41:05 +00:00
hash_result = get_sha_256(msg)
return base64.b64encode(hash_result).decode('utf-8')
2020-04-03 12:05:30 +00:00
2022-01-02 15:41:05 +00:00
def get_digest_prefix(digest_algorithm: str) -> str:
2021-11-23 11:41:40 +00:00
"""Returns the prefix for the message body digest
"""
2022-01-02 15:41:05 +00:00
if digest_algorithm in ('rsa-sha512', 'rsa-pss-sha512'):
2021-11-23 11:41:40 +00:00
return 'SHA-512'
return 'SHA-256'
2022-01-02 15:41:05 +00:00
def get_digest_algorithm_from_headers(http_headers: {}) -> str:
2021-11-23 11:41:40 +00:00
"""Returns the digest algorithm from http headers
"""
2022-01-02 15:41:05 +00:00
digest_str = None
if http_headers.get('digest'):
digest_str = http_headers['digest']
elif http_headers.get('Digest'):
digest_str = http_headers['Digest']
if digest_str:
if digest_str.startswith('SHA-512'):
2021-11-23 11:41:40 +00:00
return 'rsa-sha512'
return 'rsa-sha256'
2022-05-30 21:41:18 +00:00
def sign_post_headers(date_str: str, private_key_pem: str,
2022-01-02 15:41:05 +00:00
nickname: str, domain: str, port: int,
to_domain: str, to_port: int,
path: str, http_prefix: str,
message_body_json_str: str,
content_type: str, algorithm: str,
digest_algorithm: 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.
"""
2021-12-26 12:45:03 +00:00
domain = get_full_domain(domain, port)
2019-07-01 09:31:02 +00:00
2022-01-02 15:41:05 +00:00
to_domain = get_full_domain(to_domain, to_port)
2019-08-16 13:47:01 +00:00
2022-05-30 21:41:18 +00:00
if not date_str:
date_str = 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':
2022-01-02 15:41:05 +00:00
key_id = local_actor_url(http_prefix, nickname, domain)
2021-08-31 21:02:58 +00:00
else:
# instance actor
2022-01-02 15:41:05 +00:00
key_id = http_prefix + '://' + domain + '/actor'
key_id += '#main-key'
if not message_body_json_str:
2020-04-03 12:05:30 +00:00
headers = {
2021-09-01 18:46:28 +00:00
'(request-target)': f'get {path}',
2022-01-02 15:41:05 +00:00
'host': to_domain,
2022-05-30 21:41:18 +00:00
'date': date_str,
2021-12-26 15:32:00 +00:00
'accept': content_type
2020-04-03 12:05:30 +00:00
}
2019-06-28 18:55:29 +00:00
else:
2022-01-02 15:41:05 +00:00
body_digest = \
message_content_digest(message_body_json_str, digest_algorithm)
digest_prefix = get_digest_prefix(digest_algorithm)
content_length = len(message_body_json_str)
2020-04-03 12:05:30 +00:00
headers = {
'(request-target)': f'post {path}',
2022-01-02 15:41:05 +00:00
'host': to_domain,
2022-05-30 21:41:18 +00:00
'date': date_str,
2022-01-02 15:41:05 +00:00
'digest': f'{digest_prefix}={body_digest}',
2020-04-03 12:05:30 +00:00
'content-type': 'application/activity+json',
2022-01-02 15:41:05 +00:00
'content-length': str(content_length)
2020-04-03 12:05:30 +00:00
}
2022-01-02 15:41:05 +00:00
key = load_pem_private_key(private_key_pem.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
2022-01-02 15:41:05 +00:00
signed_header_keys = headers.keys()
signed_header_text = ''
for header_key in signed_header_keys:
signed_header_text += f'{header_key}: {headers[header_key]}\n'
2021-09-08 10:05:45 +00:00
# strip the trailing linefeed
2022-01-02 15:41:05 +00:00
signed_header_text = signed_header_text.rstrip('\n')
# signed_header_text.encode('ascii') matches
2023-01-18 10:18:48 +00:00
try:
sig_header_encoded = signed_header_text.encode('ascii')
except UnicodeEncodeError:
sig_header_encoded = signed_header_text
print('WARN: sign_post_headers unable to ascii encode ' +
signed_header_text)
header_digest = get_sha_256(sig_header_encoded)
2022-01-02 15:41:05 +00:00
# print('header_digest2: ' + str(header_digest))
2019-06-28 18:55:29 +00:00
# Sign the digest
2022-01-02 15:41:05 +00:00
raw_signature = key.sign(header_digest,
padding.PKCS1v15(),
hazutils.Prehashed(hashes.SHA256()))
signature = base64.b64encode(raw_signature).decode('ascii')
2019-06-28 18:55:29 +00:00
# Put it into a valid HTTP signature format
2022-01-02 15:41:05 +00:00
signature_dict = {
'keyId': key_id,
2021-11-23 11:41:40 +00:00
'algorithm': algorithm,
2022-01-02 15:41:05 +00:00
'headers': ' '.join(signed_header_keys),
2019-06-28 18:55:29 +00:00
'signature': signature
}
2022-01-02 15:41:05 +00:00
signature_header = ','.join(
[f'{k}="{v}"' for k, v in signature_dict.items()])
return signature_header
2019-06-28 18:55:29 +00:00
2020-04-03 12:05:30 +00:00
2022-05-30 21:41:18 +00:00
def sign_post_headers_new(date_str: str, private_key_pem: str,
2021-12-29 21:55:09 +00:00
nickname: str,
domain: str, port: int,
2022-01-02 15:41:05 +00:00
to_domain: str, to_port: int,
2021-12-29 21:55:09 +00:00
path: str,
http_prefix: str,
2022-01-02 15:41:05 +00:00
message_body_json_str: str,
algorithm: str, digest_algorithm: str,
2021-12-29 21:55:09 +00:00
debug: bool) -> (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
"""
2021-12-26 12:45:03 +00:00
domain = get_full_domain(domain, port)
2022-01-02 15:41:05 +00:00
to_domain = get_full_domain(to_domain, to_port)
2022-01-02 15:41:05 +00:00
time_format = "%a, %d %b %Y %H:%M:%S %Z"
2022-05-30 21:41:18 +00:00
if not date_str:
2021-12-26 13:17:46 +00:00
curr_time = gmtime()
date_str = strftime(time_format, curr_time)
else:
2023-11-20 22:27:58 +00:00
curr_time = date_from_string_format(date_str, [time_format])
2022-01-02 15:41:05 +00:00
seconds_since_epoch = \
2023-11-20 22:27:58 +00:00
int((curr_time - date_epoch()).total_seconds())
2022-01-02 15:41:05 +00:00
key_id = local_actor_url(http_prefix, nickname, domain) + '#main-key'
if not message_body_json_str:
headers = {
'@request-target': f'get {path}',
2022-01-02 15:41:05 +00:00
'@created': str(seconds_since_epoch),
'host': to_domain,
2022-05-30 21:41:18 +00:00
'date': date_str
}
else:
2022-01-02 15:41:05 +00:00
body_digest = message_content_digest(message_body_json_str,
digest_algorithm)
digest_prefix = get_digest_prefix(digest_algorithm)
content_length = len(message_body_json_str)
headers = {
'@request-target': f'post {path}',
2022-01-02 15:41:05 +00:00
'@created': str(seconds_since_epoch),
'host': to_domain,
2022-05-30 21:41:18 +00:00
'date': date_str,
2022-01-02 15:41:05 +00:00
'digest': f'{digest_prefix}={body_digest}',
'content-type': 'application/activity+json',
2022-01-02 15:41:05 +00:00
'content-length': str(content_length)
}
2022-01-02 15:41:05 +00:00
key = load_pem_private_key(private_key_pem.encode('utf-8'),
None, backend=default_backend())
# build a digest for signing
2022-01-02 15:41:05 +00:00
signed_header_keys = headers.keys()
signed_header_text = ''
for header_key in signed_header_keys:
signed_header_text += f'{header_key}: {headers[header_key]}\n'
signed_header_text = signed_header_text.strip()
if debug:
2022-01-02 15:41:05 +00:00
print('\nsign_post_headers_new signed_header_text:\n' +
signed_header_text + '\nEND\n')
# Sign the digest. Potentially other signing algorithms can be added here.
signature = ''
2021-09-08 10:05:45 +00:00
if algorithm == 'rsa-sha512':
2022-01-02 15:41:05 +00:00
header_digest = get_sha_512(signed_header_text.encode('ascii'))
raw_signature = key.sign(header_digest,
padding.PKCS1v15(),
hazutils.Prehashed(hashes.SHA512()))
signature = base64.b64encode(raw_signature).decode('ascii')
2021-09-08 10:05:45 +00:00
else:
2021-11-18 14:04:45 +00:00
# default rsa-sha256
2022-01-02 15:41:05 +00:00
header_digest = get_sha_256(signed_header_text.encode('ascii'))
raw_signature = key.sign(header_digest,
padding.PKCS1v15(),
hazutils.Prehashed(hashes.SHA256()))
signature = base64.b64encode(raw_signature).decode('ascii')
2022-01-02 15:41:05 +00:00
sig_key = 'sig1'
# Put it into a valid HTTP signature format
2022-01-02 15:41:05 +00:00
signature_input_dict = {
'keyId': key_id,
}
2022-01-02 15:41:05 +00:00
signature_index_header = '; '.join(
[f'{k}="{v}"' for k, v in signature_input_dict.items()])
signature_index_header += '; alg=hs2019'
signature_index_header += '; created=' + str(seconds_since_epoch)
signature_index_header += \
'; ' + sig_key + '=(' + ', '.join(signed_header_keys) + ')'
signature_dict = {
sig_key: signature
}
2022-01-02 15:41:05 +00:00
signature_header = '; '.join(
[f'{k}=:{v}:' for k, v in signature_dict.items()])
return signature_index_header, signature_header
2022-05-30 21:41:18 +00:00
def create_signed_header(date_str: str, private_key_pem: str, nickname: str,
2021-12-29 21:55:09 +00:00
domain: str, port: int,
2022-01-02 15:41:05 +00:00
to_domain: str, to_port: int,
2022-05-30 21:41:18 +00:00
path: str, http_prefix: str, with_digest: bool,
2022-01-02 15:41:05 +00:00
message_body_json_str: str,
2021-12-29 21:55:09 +00:00
content_type: str) -> {}:
2019-08-16 13:47:01 +00:00
"""Note that the domain is the destination, not the sender
"""
2021-11-23 11:41:40 +00:00
algorithm = 'rsa-sha256'
2022-01-02 15:41:05 +00:00
digest_algorithm = 'rsa-sha256'
header_domain = get_full_domain(to_domain, to_port)
2019-07-01 09:31:02 +00:00
2021-09-15 10:44:44 +00:00
# if no date is given then create one
2022-05-30 21:41:18 +00:00
if not date_str:
date_str = strftime("%a, %d %b %Y %H:%M:%S %Z", gmtime())
2021-09-15 10:44:44 +00:00
# Content-Type or Accept header
2021-12-26 15:32:00 +00:00
if not content_type:
content_type = 'application/activity+json'
2021-09-15 10:44:44 +00:00
2022-05-30 21:41:18 +00:00
if not with_digest:
2020-04-03 12:05:30 +00:00
headers = {
2021-09-01 20:57:54 +00:00
'(request-target)': f'get {path}',
2022-01-02 15:41:05 +00:00
'host': header_domain,
2022-05-30 21:41:18 +00:00
'date': date_str,
2021-12-26 15:32:00 +00:00
'accept': content_type
2020-03-22 20:36:19 +00:00
}
2022-01-02 15:41:05 +00:00
signature_header = \
2022-05-30 21:41:18 +00:00
sign_post_headers(date_str, private_key_pem, nickname,
2022-01-02 15:41:05 +00:00
domain, port, to_domain, to_port,
2021-12-29 21:55:09 +00:00
path, http_prefix, None, content_type,
algorithm, None)
2019-07-01 09:31:02 +00:00
else:
2022-01-02 15:41:05 +00:00
body_digest = message_content_digest(message_body_json_str,
digest_algorithm)
digest_prefix = get_digest_prefix(digest_algorithm)
content_length = len(message_body_json_str)
2020-04-03 12:05:30 +00:00
headers = {
2020-03-22 20:36:19 +00:00
'(request-target)': f'post {path}',
2022-01-02 15:41:05 +00:00
'host': header_domain,
2022-05-30 21:41:18 +00:00
'date': date_str,
2022-01-02 15:41:05 +00:00
'digest': f'{digest_prefix}={body_digest}',
'content-length': str(content_length),
2021-12-26 15:32:00 +00:00
'content-type': content_type
2020-03-22 20:36:19 +00:00
}
2022-01-02 15:41:05 +00:00
signature_header = \
2022-05-30 21:41:18 +00:00
sign_post_headers(date_str, private_key_pem, nickname,
2021-12-29 21:55:09 +00:00
domain, port,
2022-01-02 15:41:05 +00:00
to_domain, to_port,
path, http_prefix, message_body_json_str,
content_type, algorithm, digest_algorithm)
headers['signature'] = signature_header
2019-07-01 09:31:02 +00:00
return headers
2020-04-03 12:05:30 +00:00
2022-01-02 15:41:05 +00:00
def _verify_recent_signature(signed_date_str: 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
"""
2023-11-20 22:27:58 +00:00
curr_date = date_utcnow()
formats = ("%a, %d %b %Y %H:%M:%S %Z",
"%a, %d %b %Y %H:%M:%S %z")
2023-11-20 22:27:58 +00:00
signed_date = date_from_string_format(signed_date_str, formats)
if not signed_date:
return False
2023-11-20 22:27:58 +00:00
2023-02-21 13:26:17 +00:00
time_diff_sec = (curr_date - signed_date).total_seconds()
2019-08-23 11:39:16 +00:00
# 12 hours tollerance
2022-01-02 15:41:05 +00:00
if time_diff_sec > 43200:
2023-02-21 11:35:10 +00:00
print('WARN: Header signed too long ago: ' + signed_date_str + ' ' +
str(time_diff_sec / (60 * 60)) + ' hours')
return False
2023-02-21 13:26:17 +00:00
# allow clocks to be off by a few mins
if time_diff_sec < -480:
2023-02-21 11:35:10 +00:00
print('WARN: Header signed in the future! ' + signed_date_str + ' ' +
str(time_diff_sec / (60 * 60)) + ' hours')
2019-08-23 11:30:37 +00:00
return False
return True
2020-04-03 12:05:30 +00:00
2021-12-29 21:55:09 +00:00
def verify_post_headers(http_prefix: str,
2022-05-30 21:41:18 +00:00
public_key_pem: str, headers: dict,
path: str, get_method: bool,
message_body_digest: str,
2022-01-02 15:41:05 +00:00
message_body_json_str: str, debug: bool,
2022-05-30 21:41:18 +00:00
no_recency_check: 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.
2022-05-30 21:41:18 +00:00
public_key_pem - the public key from an rsa key pair
2019-06-28 18:55:29 +00:00
headers - should be a dictionary of request headers
path - the relative url that was requested from this site
2022-05-30 21:41:18 +00:00
get_method - GET or POST
2022-01-02 15:41:05 +00:00
message_body_json_str - the received request body (used for digest)
2019-06-28 18:55:29 +00:00
"""
2019-08-23 11:20:20 +00:00
2022-05-30 21:41:18 +00:00
if get_method:
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:
2021-12-29 21:55:09 +00:00
print('DEBUG: verify_post_headers ' + method)
2022-05-30 21:41:18 +00:00
print('verify_post_headers public_key_pem: ' + str(public_key_pem))
2021-12-29 21:55:09 +00:00
print('verify_post_headers headers: ' + str(headers))
2022-01-02 15:41:05 +00:00
print('verify_post_headers message_body_json_str: ' +
str(message_body_json_str))
2020-03-22 21:16:02 +00:00
2022-05-30 21:41:18 +00:00
pubkey = load_pem_public_key(public_key_pem.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') or headers.get('signature-input'):
if headers.get('Signature-Input'):
2022-01-02 15:41:05 +00:00
signature_header = headers['Signature-Input']
else:
2022-01-02 15:41:05 +00:00
signature_header = headers['signature-input']
field_sep2 = ','
# split the signature input into separate fields
2022-01-02 15:41:05 +00:00
signature_dict = {
k.strip(): v.strip()
2022-01-02 15:41:05 +00:00
for k, v in [i.split('=', 1) for i in signature_header.split(';')]
}
2022-01-02 15:41:05 +00:00
request_target_key = None
request_target_str = None
for key_str, value_str in signature_dict.items():
if value_str.startswith('('):
request_target_key = key_str
request_target_str = value_str[1:-1]
elif value_str.startswith('"'):
signature_dict[key_str] = value_str[1:-1]
if not request_target_key:
return False
2022-01-02 15:41:05 +00:00
signature_dict[request_target_key] = request_target_str
else:
2022-01-02 15:41:05 +00:00
request_target_key = 'headers'
signature_header = headers['signature']
field_sep2 = ' '
# split the signature input into separate fields
2022-01-02 15:41:05 +00:00
signature_dict = {
k: v[1:-1]
2022-01-02 15:41:05 +00:00
for k, v in [i.split('=', 1) for i in signature_header.split(',')]
}
2019-06-28 18:55:29 +00:00
if debug:
2022-01-02 15:41:05 +00:00
print('signature_dict: ' + str(signature_dict))
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)
2022-01-02 15:41:05 +00:00
signed_header_list = []
algorithm = 'rsa-sha256'
2022-01-02 15:41:05 +00:00
digest_algorithm = 'rsa-sha256'
for signed_header in signature_dict[request_target_key].split(field_sep2):
signed_header = signed_header.strip()
2019-11-12 15:03:17 +00:00
if debug:
2022-01-02 15:41:05 +00:00
print('DEBUG: verify_post_headers signed_header=' + signed_header)
if signed_header == '(request-target)':
# original Mastodon http signature
2022-01-02 15:41:05 +00:00
append_str = f'(request-target): {method.lower()} {path}'
signed_header_list.append(append_str)
elif '@request-target' in signed_header:
# https://tools.ietf.org/html/
# draft-ietf-httpbis-message-signatures
2022-01-02 15:41:05 +00:00
append_str = f'@request-target: {method.lower()} {path}'
signed_header_list.append(append_str)
elif '@created' in signed_header:
if signature_dict.get('created'):
created_str = str(signature_dict['created'])
append_str = f'@created: {created_str}'
signed_header_list.append(append_str)
elif '@expires' in signed_header:
if signature_dict.get('expires'):
expires_str = str(signature_dict['expires'])
append_str = f'@expires: {expires_str}'
signed_header_list.append(append_str)
elif '@method' in signed_header:
append_str = f'@expires: {method}'
signed_header_list.append(append_str)
elif '@scheme' in signed_header:
signed_header_list.append('@scheme: http')
elif '@authority' in signed_header:
authority_str = None
if signature_dict.get('authority'):
authority_str = str(signature_dict['authority'])
elif signature_dict.get('Authority'):
authority_str = str(signature_dict['Authority'])
if authority_str:
append_str = f'@authority: {authority_str}'
signed_header_list.append(append_str)
elif signed_header == 'algorithm':
if headers.get(signed_header):
algorithm = headers[signed_header]
2021-11-22 11:52:55 +00:00
if debug:
print('http signature algorithm: ' + algorithm)
2022-01-02 15:41:05 +00:00
elif signed_header == 'digest':
2022-05-30 21:41:18 +00:00
if message_body_digest:
body_digest = message_body_digest
else:
2022-01-02 15:41:05 +00:00
body_digest = \
message_content_digest(message_body_json_str,
digest_algorithm)
signed_header_list.append(f'digest: SHA-256={body_digest}')
elif signed_header == 'content-length':
if headers.get(signed_header):
append_str = f'content-length: {headers[signed_header]}'
signed_header_list.append(append_str)
2021-09-01 14:22:11 +00:00
elif headers.get('Content-Length'):
2022-01-02 15:41:05 +00:00
content_length = headers['Content-Length']
signed_header_list.append(f'content-length: {content_length}')
2021-09-01 14:22:11 +00:00
elif headers.get('Content-length'):
2022-01-02 15:41:05 +00:00
content_length = headers['Content-length']
append_str = f'content-length: {content_length}'
signed_header_list.append(append_str)
2019-11-12 17:16:34 +00:00
else:
2021-09-01 14:22:11 +00:00
if debug:
2022-01-02 15:41:05 +00:00
print('DEBUG: verify_post_headers ' + signed_header +
2021-09-01 14:22:11 +00:00
' not found in ' + str(headers))
2019-06-28 18:55:29 +00:00
else:
2022-01-02 15:41:05 +00:00
if headers.get(signed_header):
2022-05-30 21:41:18 +00:00
if signed_header == 'date' and not no_recency_check:
2022-01-02 15:41:05 +00:00
if not _verify_recent_signature(headers[signed_header]):
2019-11-12 15:03:17 +00:00
if debug:
2020-04-03 12:05:30 +00:00
print('DEBUG: ' +
2021-12-29 21:55:09 +00:00
'verify_post_headers date is not recent ' +
2022-01-02 15:41:05 +00:00
headers[signed_header])
2019-08-23 11:30:37 +00:00
return False
2022-01-02 15:41:05 +00:00
signed_header_list.append(
f'{signed_header}: {headers[signed_header]}')
2019-08-15 21:34:25 +00:00
else:
2022-01-02 15:41:05 +00:00
if '-' in signed_header:
2021-03-14 12:09:56 +00:00
# capitalise with dashes
# my-header becomes My-Header
2022-01-02 15:41:05 +00:00
header_parts = signed_header.split('-')
signed_header_cap = None
for part in header_parts:
if signed_header_cap:
signed_header_cap += '-' + part.capitalize()
2021-03-14 12:09:56 +00:00
else:
2022-01-02 15:41:05 +00:00
signed_header_cap = part.capitalize()
2021-03-14 11:53:13 +00:00
else:
2021-03-14 12:09:56 +00:00
# header becomes Header
2022-01-02 15:41:05 +00:00
signed_header_cap = signed_header.capitalize()
2021-03-14 12:09:56 +00:00
if debug:
2022-01-02 15:41:05 +00:00
print('signed_header_cap: ' + signed_header_cap)
2021-03-14 12:09:56 +00:00
# if this is the date header then check it is recent
2022-01-02 15:41:05 +00:00
if signed_header_cap == 'Date':
signed_hdr_cap = headers[signed_header_cap]
if not _verify_recent_signature(signed_hdr_cap):
2019-11-12 15:03:17 +00:00
if debug:
2020-04-03 12:05:30 +00:00
print('DEBUG: ' +
2021-12-29 21:55:09 +00:00
'verify_post_headers date is not recent ' +
2022-01-02 15:41:05 +00:00
headers[signed_header])
2019-08-23 11:30:37 +00:00
return False
2021-03-14 12:09:56 +00:00
# add the capitalised header
2022-01-02 15:41:05 +00:00
if headers.get(signed_header_cap):
signed_header_list.append(
f'{signed_header}: {headers[signed_header_cap]}')
elif '-' in signed_header:
2021-03-14 18:34:30 +00:00
# my-header becomes My-header
2022-01-02 15:41:05 +00:00
signed_header_cap = signed_header.capitalize()
if headers.get(signed_header_cap):
signed_header_list.append(
f'{signed_header}: {headers[signed_header_cap]}')
2019-06-28 18:55:29 +00:00
# Now we have our header data digest
2022-01-02 15:41:05 +00:00
signed_header_text = '\n'.join(signed_header_list)
if debug:
2022-01-02 15:41:05 +00:00
print('\nverify_post_headers signed_header_text:\n' +
signed_header_text + '\nEND\n')
2019-06-28 18:55:29 +00:00
# Get the signature, verify with public key, return result
if (headers.get('Signature-Input') and headers.get('Signature')) or \
(headers.get('signature-input') and headers.get('signature')):
# https://tools.ietf.org/html/
# draft-ietf-httpbis-message-signatures
if headers.get('Signature'):
2022-01-02 15:41:05 +00:00
headers_sig = headers['Signature']
else:
2022-01-02 15:41:05 +00:00
headers_sig = headers['signature']
# remove sig1=:
2022-01-02 15:41:05 +00:00
if request_target_key + '=:' in headers_sig:
headers_sig = headers_sig.split(request_target_key + '=:')[1]
headers_sig = headers_sig[:len(headers_sig)-1]
signature = base64.b64decode(headers_sig)
else:
# Original Mastodon signature
2022-01-02 15:41:05 +00:00
headers_sig = signature_dict['signature']
signature = base64.b64decode(headers_sig)
if debug:
2022-01-02 15:41:05 +00:00
print('signature: ' + algorithm + ' ' + headers_sig)
2019-06-28 18:55:29 +00:00
2021-11-22 11:52:55 +00:00
# log unusual signing algorithms
2022-01-02 15:41:05 +00:00
if signature_dict.get('alg'):
print('http signature algorithm: ' + signature_dict['alg'])
2021-11-22 11:52:55 +00:00
# If extra signing algorithms need to be added then do it here
2022-01-02 15:41:05 +00:00
if not signature_dict.get('alg'):
2021-11-22 11:52:55 +00:00
alg = hazutils.Prehashed(hashes.SHA256())
2022-01-02 15:41:05 +00:00
elif (signature_dict['alg'] == 'rsa-sha256' or
signature_dict['alg'] == 'rsa-v1_5-sha256' or
signature_dict['alg'] == 'hs2019'):
2021-11-22 11:52:55 +00:00
alg = hazutils.Prehashed(hashes.SHA256())
2022-01-02 15:41:05 +00:00
elif (signature_dict['alg'] == 'rsa-sha512' or
signature_dict['alg'] == 'rsa-pss-sha512'):
2021-11-22 11:52:55 +00:00
alg = hazutils.Prehashed(hashes.SHA512())
else:
alg = hazutils.Prehashed(hashes.SHA256())
2021-11-22 11:52:55 +00:00
2022-01-02 15:41:05 +00:00
if digest_algorithm == 'rsa-sha256':
header_digest = get_sha_256(signed_header_text.encode('ascii'))
elif digest_algorithm == 'rsa-sha512':
header_digest = get_sha_512(signed_header_text.encode('ascii'))
else:
2022-01-02 15:41:05 +00:00
print('Unknown http digest algorithm: ' + digest_algorithm)
header_digest = ''
padding_str = padding.PKCS1v15()
2019-06-28 18:55:29 +00:00
try:
2022-01-02 15:41:05 +00:00
pubkey.verify(signature, header_digest, padding_str, alg)
2019-06-28 18:55:29 +00:00
return True
except BaseException:
2019-11-12 15:03:17 +00:00
if debug:
2021-12-29 21:55:09 +00:00
print('EX: verify_post_headers pkcs1_15 verify failure')
return False