__filename__ = "httpsig.py" __author__ = "Bob Mottram" __credits__ = ['lamia'] __license__ = "AGPL3+" __version__ = "1.2.0" __maintainer__ = "Bob Mottram" __email__ = "bob@libreserver.org" __status__ = "Production" __module_group__ = "Security" # see https://tools.ietf.org/html/draft-cavage-http-signatures-06 # # This might change in future # see https://tools.ietf.org/html/draft-ietf-httpbis-message-signatures 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 import base64 from time import gmtime, strftime import datetime from utils import get_full_domain from utils import get_sha_256 from utils import get_sha_512 from utils import local_actor_url def message_content_digest(message_body_json_str: str, digest_algorithm: str) -> str: """Returns the digest for the message body """ msg = message_body_json_str.encode('utf-8') if digest_algorithm in ('rsa-sha512', 'rsa-pss-sha512'): hash_result = get_sha_512(msg) else: hash_result = get_sha_256(msg) return base64.b64encode(hash_result).decode('utf-8') def get_digest_prefix(digest_algorithm: str) -> str: """Returns the prefix for the message body digest """ if digest_algorithm in ('rsa-sha512', 'rsa-pss-sha512'): return 'SHA-512' return 'SHA-256' def get_digest_algorithm_from_headers(http_headers: {}) -> str: """Returns the digest algorithm from http headers """ 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'): return 'rsa-sha512' return 'rsa-sha256' def sign_post_headers(dateStr: str, private_key_pem: str, 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: """Returns a raw signature string that can be plugged into a header and used to verify the authenticity of an HTTP transmission. """ domain = get_full_domain(domain, port) to_domain = get_full_domain(to_domain, to_port) if not dateStr: dateStr = strftime("%a, %d %b %Y %H:%M:%S %Z", gmtime()) if nickname != domain and nickname.lower() != 'actor': key_id = local_actor_url(http_prefix, nickname, domain) else: # instance actor key_id = http_prefix + '://' + domain + '/actor' key_id += '#main-key' if not message_body_json_str: headers = { '(request-target)': f'get {path}', 'host': to_domain, 'date': dateStr, 'accept': content_type } else: 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}', 'host': to_domain, 'date': dateStr, 'digest': f'{digest_prefix}={body_digest}', 'content-type': 'application/activity+json', 'content-length': str(content_length) } key = load_pem_private_key(private_key_pem.encode('utf-8'), None, backend=default_backend()) # headers.update({ # '(request-target)': f'post {path}', # }) # build a digest for signing 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' # strip the trailing linefeed signed_header_text = signed_header_text.rstrip('\n') # signed_header_text.encode('ascii') matches header_digest = get_sha_256(signed_header_text.encode('ascii')) # print('header_digest2: ' + str(header_digest)) # Sign the digest raw_signature = key.sign(header_digest, padding.PKCS1v15(), hazutils.Prehashed(hashes.SHA256())) signature = base64.b64encode(raw_signature).decode('ascii') # Put it into a valid HTTP signature format signature_dict = { 'keyId': key_id, 'algorithm': algorithm, 'headers': ' '.join(signed_header_keys), 'signature': signature } signature_header = ','.join( [f'{k}="{v}"' for k, v in signature_dict.items()]) return signature_header def sign_post_headers_new(dateStr: str, private_key_pem: str, nickname: str, domain: str, port: int, to_domain: str, to_port: int, path: str, http_prefix: str, message_body_json_str: str, algorithm: str, digest_algorithm: str, 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 """ domain = get_full_domain(domain, port) to_domain = get_full_domain(to_domain, to_port) time_format = "%a, %d %b %Y %H:%M:%S %Z" if not dateStr: curr_time = gmtime() dateStr = strftime(time_format, curr_time) else: curr_time = datetime.datetime.strptime(dateStr, time_format) seconds_since_epoch = \ int((curr_time - datetime.datetime(1970, 1, 1)).total_seconds()) key_id = local_actor_url(http_prefix, nickname, domain) + '#main-key' if not message_body_json_str: headers = { '@request-target': f'get {path}', '@created': str(seconds_since_epoch), 'host': to_domain, 'date': dateStr } else: 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}', '@created': str(seconds_since_epoch), 'host': to_domain, 'date': dateStr, 'digest': f'{digest_prefix}={body_digest}', 'content-type': 'application/activity+json', 'content-length': str(content_length) } key = load_pem_private_key(private_key_pem.encode('utf-8'), None, backend=default_backend()) # build a digest for signing 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: 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 = '' if algorithm == 'rsa-sha512': 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') else: # default rsa-sha256 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') sig_key = 'sig1' # Put it into a valid HTTP signature format signature_input_dict = { 'keyId': key_id, } 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 } signature_header = '; '.join( [f'{k}=:{v}:' for k, v in signature_dict.items()]) return signature_index_header, signature_header def create_signed_header(dateStr: str, private_key_pem: str, nickname: str, domain: str, port: int, to_domain: str, to_port: int, path: str, http_prefix: str, withDigest: bool, message_body_json_str: str, content_type: str) -> {}: """Note that the domain is the destination, not the sender """ algorithm = 'rsa-sha256' digest_algorithm = 'rsa-sha256' header_domain = get_full_domain(to_domain, to_port) # if no date is given then create one if not dateStr: dateStr = strftime("%a, %d %b %Y %H:%M:%S %Z", gmtime()) # Content-Type or Accept header if not content_type: content_type = 'application/activity+json' if not withDigest: headers = { '(request-target)': f'get {path}', 'host': header_domain, 'date': dateStr, 'accept': content_type } signature_header = \ sign_post_headers(dateStr, private_key_pem, nickname, domain, port, to_domain, to_port, path, http_prefix, None, content_type, algorithm, None) else: 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}', 'host': header_domain, 'date': dateStr, 'digest': f'{digest_prefix}={body_digest}', 'content-length': str(content_length), 'content-type': content_type } signature_header = \ sign_post_headers(dateStr, private_key_pem, nickname, domain, port, to_domain, to_port, path, http_prefix, message_body_json_str, content_type, algorithm, digest_algorithm) headers['signature'] = signature_header return headers def _verify_recent_signature(signed_date_str: str) -> bool: """Checks whether the given time taken from the header is within 12 hours of the current time """ curr_date = datetime.datetime.utcnow() date_format = "%a, %d %b %Y %H:%M:%S %Z" signed_date = datetime.datetime.strptime(signed_date_str, date_format) time_diff_sec = (curr_date - signed_date).seconds # 12 hours tollerance if time_diff_sec > 43200: print('WARN: Header signed too long ago: ' + signed_date_str) print(str(time_diff_sec / (60 * 60)) + ' hours') return False if time_diff_sec < 0: print('WARN: Header signed in the future! ' + signed_date_str) print(str(time_diff_sec / (60 * 60)) + ' hours') return False return True def verify_post_headers(http_prefix: str, publicKeyPem: str, headers: dict, path: str, GETmethod: bool, messageBodyDigest: str, message_body_json_str: str, debug: bool, noRecencyCheck: bool = False) -> 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 message_body_json_str - the received request body (used for digest) """ if GETmethod: method = 'GET' else: method = 'POST' if debug: print('DEBUG: verify_post_headers ' + method) print('verify_post_headers publicKeyPem: ' + str(publicKeyPem)) print('verify_post_headers headers: ' + str(headers)) print('verify_post_headers message_body_json_str: ' + str(message_body_json_str)) pubkey = load_pem_public_key(publicKeyPem.encode('utf-8'), backend=default_backend()) # Build a dictionary of the signature values if headers.get('Signature-Input') or headers.get('signature-input'): if headers.get('Signature-Input'): signature_header = headers['Signature-Input'] else: signature_header = headers['signature-input'] field_sep2 = ',' # split the signature input into separate fields signature_dict = { k.strip(): v.strip() for k, v in [i.split('=', 1) for i in signature_header.split(';')] } 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 signature_dict[request_target_key] = request_target_str else: request_target_key = 'headers' signature_header = headers['signature'] field_sep2 = ' ' # split the signature input into separate fields signature_dict = { k: v[1:-1] for k, v in [i.split('=', 1) for i in signature_header.split(',')] } if debug: print('signature_dict: ' + str(signature_dict)) # Unpack the signed headers and set values based on current headers and # body (if a digest was included) signed_header_list = [] algorithm = 'rsa-sha256' digest_algorithm = 'rsa-sha256' for signed_header in signature_dict[request_target_key].split(field_sep2): signed_header = signed_header.strip() if debug: print('DEBUG: verify_post_headers signed_header=' + signed_header) if signed_header == '(request-target)': # original Mastodon http signature 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 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] if debug: print('http signature algorithm: ' + algorithm) elif signed_header == 'digest': if messageBodyDigest: body_digest = messageBodyDigest else: 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) elif headers.get('Content-Length'): content_length = headers['Content-Length'] signed_header_list.append(f'content-length: {content_length}') elif headers.get('Content-length'): content_length = headers['Content-length'] append_str = f'content-length: {content_length}' signed_header_list.append(append_str) else: if debug: print('DEBUG: verify_post_headers ' + signed_header + ' not found in ' + str(headers)) else: if headers.get(signed_header): if signed_header == 'date' and not noRecencyCheck: if not _verify_recent_signature(headers[signed_header]): if debug: print('DEBUG: ' + 'verify_post_headers date is not recent ' + headers[signed_header]) return False signed_header_list.append( f'{signed_header}: {headers[signed_header]}') else: if '-' in signed_header: # capitalise with dashes # my-header becomes My-Header header_parts = signed_header.split('-') signed_header_cap = None for part in header_parts: if signed_header_cap: signed_header_cap += '-' + part.capitalize() else: signed_header_cap = part.capitalize() else: # header becomes Header signed_header_cap = signed_header.capitalize() if debug: print('signed_header_cap: ' + signed_header_cap) # if this is the date header then check it is recent if signed_header_cap == 'Date': signed_hdr_cap = headers[signed_header_cap] if not _verify_recent_signature(signed_hdr_cap): if debug: print('DEBUG: ' + 'verify_post_headers date is not recent ' + headers[signed_header]) return False # add the capitalised header if headers.get(signed_header_cap): signed_header_list.append( f'{signed_header}: {headers[signed_header_cap]}') elif '-' in signed_header: # my-header becomes My-header signed_header_cap = signed_header.capitalize() if headers.get(signed_header_cap): signed_header_list.append( f'{signed_header}: {headers[signed_header_cap]}') # Now we have our header data digest signed_header_text = '\n'.join(signed_header_list) if debug: print('\nverify_post_headers signed_header_text:\n' + signed_header_text + '\nEND\n') # 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'): headers_sig = headers['Signature'] else: headers_sig = headers['signature'] # remove sig1=: 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 headers_sig = signature_dict['signature'] signature = base64.b64decode(headers_sig) if debug: print('signature: ' + algorithm + ' ' + headers_sig) # log unusual signing algorithms if signature_dict.get('alg'): print('http signature algorithm: ' + signature_dict['alg']) # If extra signing algorithms need to be added then do it here if not signature_dict.get('alg'): alg = hazutils.Prehashed(hashes.SHA256()) elif (signature_dict['alg'] == 'rsa-sha256' or signature_dict['alg'] == 'rsa-v1_5-sha256' or signature_dict['alg'] == 'hs2019'): alg = hazutils.Prehashed(hashes.SHA256()) elif (signature_dict['alg'] == 'rsa-sha512' or signature_dict['alg'] == 'rsa-pss-sha512'): alg = hazutils.Prehashed(hashes.SHA512()) else: alg = hazutils.Prehashed(hashes.SHA256()) 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: print('Unknown http digest algorithm: ' + digest_algorithm) header_digest = '' padding_str = padding.PKCS1v15() try: pubkey.verify(signature, header_digest, padding_str, alg) return True except BaseException: if debug: print('EX: verify_post_headers pkcs1_15 verify failure') return False