Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/acme2certifier/acme_srv/helpers/crypto.py: 67%

1# -*- coding: utf-8 -*- 

2"""Cryptographic operations for acme2certifier""" 

3 

4import hashlib 

5import json 

6import logging 

7import re 

8import secrets 

9from string import digits, ascii_letters 

10from typing import Dict, Tuple 

11from jwcrypto import jwk, jws 

12from .encoding import b64decode_pad, b64_encode 

13from .validation import dkeys_lower 

14 

15 

16def decode_deserialize(logger: logging.Logger, string: str) -> Dict: 

17 """decode and deserialize string""" 

18 logger.debug("Helper.decode_deserialize()") 

19 # b64 decode 

20 string_decode = b64decode_pad(logger, string) 

21 # deserialize if b64 decoding was successful 

22 if string_decode and string_decode != "ERR: b64 decoding error": 

23 try: 

24 string_decode = json.loads(string_decode) 

25 except ValueError: 

26 string_decode = "ERR: Json decoding error" 

27 

28 return string_decode 

29 

30 

31def decode_message( 

32 logger: logging.Logger, message: str 

33) -> Tuple[str, str, Dict[str, str], Dict[str, str], str]: 

34 """decode jwstoken and return header, payload and signature""" 

35 logger.debug("Helper.decode_message()") 

36 jwstoken = jws.JWS() 

37 result = False 

38 error = None 

39 try: 

40 jwstoken.deserialize(message) 

41 protected = json.loads(jwstoken.objects["protected"]) 

42 if bool(jwstoken.objects["payload"]): 

43 payload = json.loads(jwstoken.objects["payload"]) 

44 else: 

45 payload = {} 

46 signature = jwstoken.objects["signature"] 

47 result = True 

48 except Exception as err: 

49 logger.error("Error during message decoding %s", err) 

50 error = str(err) 

51 protected = {} 

52 payload = {} 

53 signature = None 

54 

55 if payload: 

56 payload = dkeys_lower(payload) 

57 return (result, error, protected, payload, signature) 

58 

59 

60def generate_random_string(logger: logging.Logger, length: int) -> str: 

61 """generate cryptographically secure random string to be used as name""" 

62 logger.debug("Helper.generate_random_string()") 

63 char_set = digits + ascii_letters 

64 return "".join(secrets.choice(char_set) for _ in range(length)) 

65 

66 

67def jwk_thumbprint_get(logger: logging.Logger, pub_key: Dict[str, str]) -> str: 

68 """get thumbprint""" 

69 logger.debug("Helper.jwk_thumbprint_get()") 

70 if pub_key: 

71 try: 

72 jwkey = jwk.JWK(**pub_key) 

73 thumbprint = jwkey.thumbprint() 

74 except Exception as err: 

75 logger.error("Could not get the JWKEY thumbprint from public key: %s", err) 

76 thumbprint = None 

77 else: 

78 thumbprint = None 

79 

80 logger.debug("Helper.jwk_thumbprint_get() ended with: %s", thumbprint) 

81 return thumbprint 

82 

83 

84def sha256_hash(logger: logging.Logger, string: str) -> str: 

85 """hash string""" 

86 logger.debug("Helper.sha256_hash()") 

87 result = hashlib.sha256(string.encode("utf-8")).digest() 

88 logger.debug( 

89 "Helper.sha256_hash() ended with %s (base64-encoded)", 

90 b64_encode(logger, result), 

91 ) 

92 return result 

93 

94 

95def sha256_hash_hex(logger: logging.Logger, string: str) -> str: 

96 """hash string""" 

97 logger.debug("Helper.sha256_hash_hex()") 

98 result = hashlib.sha256(string.encode("utf-8")).hexdigest() 

99 logger.debug("Helper.sha256_hash_hex() ended with %s", result) 

100 return result 

101 

102 

103def signature_check( 

104 logger: logging.Logger, message: str, pub_key: str, json_: bool = False 

105) -> Tuple[bool, str]: 

106 """check JWS""" 

107 logger.debug("Helper.signature_check(%s)", json_) 

108 

109 result = False 

110 error = None 

111 

112 if pub_key: 

113 # load key 

114 try: 

115 if json_: 

116 logger.debug("Helper.signature_check(): load key from json") 

117 jwkey = jwk.JWK.from_json(pub_key) 

118 else: 

119 logger.debug("Helper.signature_check(): load plain json") 

120 jwkey = jwk.JWK(**pub_key) 

121 except Exception as err: 

122 logger.error("Loading of public key failed %s", err) 

123 jwkey = None 

124 result = False 

125 error = str(err) 

126 

127 # verify signature 

128 if jwkey: 

129 jwstoken = jws.JWS() 

130 jwstoken.deserialize(message) 

131 try: 

132 jwstoken.verify(jwkey) 

133 result = True 

134 except Exception as err: 

135 logger.error("Message verification failed %s", err) 

136 error = str(err) 

137 else: 

138 logger.error("No jwkey extracted") 

139 else: 

140 logger.error("No pubkey specified.") 

141 error = "No key specified." 

142 

143 logger.debug("Helper.signature_check() ended with: %s, %s", result, error) 

144 return (result, error) 

145 

146 

147def string_sanitize(logger: logging.Logger, unsafe_str: str) -> str: 

148 """sanitize string""" 

149 logger.debug("Helper.string_sanitize()") 

150 allowed_range = set(range(32, 127)) 

151 safe_str = "" 

152 for char in unsafe_str: 

153 cp_ = ord(char) 

154 if cp_ in allowed_range: 

155 safe_str += char 

156 elif cp_ == 9: 

157 safe_str += " " * 4 

158 return re.sub(r"\s+", " ", safe_str)