Coverage for pki_tools/types/csr.py: 70%

1import re 

2 

3import yaml 

4from cryptography import x509 

5 

6from pki_tools.exceptions import MissingInit 

7from pki_tools.types.crypto_parser import ( 

8 CryptoConfig, 

9 HelperFunc, 

10 InitCryptoParser, 

11) 

12from pki_tools.types.extensions import Extensions 

13from pki_tools.types.key_pair import ( 

14 CryptoKeyPair, 

15 CryptoPrivateKey, 

16 CryptoPublicKey, 

17 Ed448PrivateKey, 

18 Ed25519PrivateKey, 

19) 

20from pki_tools.types.name import Name 

21from pki_tools.types.signature_algorithm import ( 

22 SignatureAlgorithm, 

23) 

24from pki_tools.types.utils import _byte_to_hex 

25 

26PEM_CSR_REGEX = re.compile( 

27 r"\s*-+BEGIN CERTIFICATE REQUEST-+[\w+/\s=]*-+END CERTIFICATE REQUEST-+\s*" 

28) 

29 

30 

31class CertificateSigningRequest(InitCryptoParser): 

32 """ 

33 Represents a certificate signing request (CSR). 

34 

35 Attributes: 

36 subject: The subject of the CSR. 

37 public_key: Public key associated with the CSR. 

38 extensions: Extensions associated with the CSR. 

39 attributes: Attributes of the CSR. 

40 signature_algorithm: Signature algorithm used while signing the CSR. 

41 

42 --8<-- "docs/examples/csr.md" 

43 """ 

44 

45 subject: Name 

46 

47 public_key: CryptoPublicKey | None = None 

48 extensions: Extensions | None = None 

49 attributes: dict[str, bytes] | None = None 

50 signature_algorithm: SignatureAlgorithm | None = None 

51 

52 _private_key: CryptoPrivateKey | None 

53 

54 @classmethod 

55 def from_cryptography( 

56 cls: type["CertificateSigningRequest"], 

57 crypto_csr: x509.CertificateSigningRequest, 

58 ) -> "CertificateSigningRequest": 

59 """ 

60 Create a CertificateSigningRequest object from a cryptography 

61 CertificateSigningRequest. 

62 

63 Args: 

64 crypto_csr: Cryptography CertificateSigningRequest. 

65 

66 Returns: 

67 Instance of CertificateSigningRequest. 

68 

69 --8<-- "docs/examples/csr_from_cryptography.md" 

70 """ 

71 attributes = {} 

72 for att in crypto_csr.attributes: 

73 attributes[att.oid.dotted_string] = att.value 

74 

75 signature_algorithm = None 

76 if crypto_csr.signature_hash_algorithm: 

77 signature_algorithm = SignatureAlgorithm.from_cryptography( 

78 crypto_csr.signature_hash_algorithm, 

79 crypto_csr.signature_algorithm_parameters, 

80 ) 

81 

82 ret = cls( 

83 subject=Name.from_cryptography(crypto_csr.subject), 

84 extensions=Extensions.from_cryptography(crypto_csr.extensions), 

85 signature_algorithm=signature_algorithm, 

86 signature_value=_byte_to_hex(crypto_csr.signature), 

87 public_key=CryptoPublicKey.from_cryptography( 

88 crypto_csr.public_key() 

89 ), 

90 attributes=attributes, 

91 _x509_obj=crypto_csr, 

92 ) 

93 ret._x509_obj = crypto_csr 

94 return ret 

95 

96 @property 

97 def tbs_bytes(self) -> bytes: 

98 """ 

99 Get the bytes to be signed of the CSR. 

100 

101 Returns: 

102 TBS bytes of the CSR. 

103 """ 

104 return self._crypto_object.tbs_certrequest_bytes 

105 

106 def sign( 

107 self, 

108 key_pair: CryptoKeyPair, 

109 signature_algorithm: SignatureAlgorithm | None = None, 

110 ): 

111 """ 

112 Sign the CSR with the provided key pair and signature algorithm. 

113 

114 Args: 

115 key_pair: Key pair with the private key to use while signing the 

116 CSR 

117 signature_algorithm: Signature algorithm to use for signing. 

118 """ 

119 self._private_key = key_pair.private_key 

120 self.signature_algorithm = signature_algorithm 

121 self._x509_obj = self._to_cryptography() 

122 

123 def _string_dict(self): 

124 ret = { 

125 "Certificate Signing Request": { 

126 "Subject": str(self.subject), 

127 "Public Key": self.public_key._string_dict(), 

128 } 

129 } 

130 if self.extensions: 

131 ret["Extensions"] = self.extensions._string_dict() 

132 if self.signature_algorithm is not None: 

133 signature_alg = self.signature_algorithm.algorithm.name.value 

134 ret["Signature Algorithm"] = signature_alg 

135 if self.attributes is not None: 

136 attributes = [] 

137 

138 for k in self.attributes: 

139 val = self.attributes[k] 

140 if k == "1.2.840.113549.1.9.14": 

141 val = _byte_to_hex(val) 

142 elif k == "1.2.840.113549.1.9.7": 

143 val = val.decode() 

144 attributes.append(f"{k}: {val!s}") 

145 

146 if attributes: 

147 ret["Certificate Signing Request"]["Attributes"] = attributes 

148 return ret 

149 

150 def _to_cryptography(self) -> x509.CertificateSigningRequest: 

151 if not hasattr(self, "_private_key"): 

152 raise MissingInit( 

153 f"Please use CertificateSigningRequest." 

154 f"{self._init_func} function" 

155 ) 

156 

157 crypto_key = self._private_key._to_cryptography() 

158 

159 builder = x509.CertificateSigningRequestBuilder() 

160 builder = builder.subject_name(self.subject._to_cryptography()) 

161 

162 if self.extensions is not None: 

163 for extension in self.extensions: 

164 builder = builder.add_extension( 

165 extension._to_cryptography(), extension.critical 

166 ) 

167 

168 if self.attributes is not None: 

169 for attribute_oid, value in self.attributes.items(): 

170 oid = x509.ObjectIdentifier(attribute_oid) 

171 builder = builder.add_attribute(oid, value) 

172 

173 if isinstance(self._private_key, (Ed448PrivateKey, Ed25519PrivateKey)): 

174 alg = None 

175 else: 

176 alg = self.signature_algorithm.algorithm._to_cryptography() 

177 

178 return builder.sign(crypto_key, alg) 

179 

180 def __str__(self) -> str: 

181 return yaml.safe_dump( 

182 self._string_dict(), 

183 indent=2, 

184 default_flow_style=False, 

185 explicit_start=True, 

186 default_style="", 

187 ) 

188 

189 @classmethod 

190 def _crypto_config(cls) -> CryptoConfig: 

191 return CryptoConfig( 

192 load_pem=HelperFunc(func=x509.load_pem_x509_csr), 

193 load_der=HelperFunc(func=x509.load_der_x509_csr), 

194 pem_regexp=PEM_CSR_REGEX, 

195 )