Coverage for pki_tools/types/crypto_parser.py: 95%

1import abc 

2import re 

3from collections.abc import Callable 

4from enum import Enum 

5from typing import TypeVar 

6 

7import yaml 

8from cryptography.hazmat.primitives import serialization 

9from loguru import logger 

10from pydantic import BaseModel 

11 

12from pki_tools.exceptions import ( 

13 LoadError, 

14 MissingInit, 

15) 

16 

17CryptoObject = TypeVar("CryptoObject") 

18 

19 

20class CryptoParser(BaseModel, abc.ABC): 

21 """ 

22 CryptoParser is an abstract class used by all the types 

23 parsing cryptography objects into [pki_tools][pki_tools.types.certificate] 

24 pydantic classes. 

25 """ 

26 

27 _x509_obj: CryptoObject 

28 

29 def __init__(self, **kwargs): 

30 super().__init__(**kwargs) 

31 

32 if "_x509_obj" not in kwargs: 

33 try: 

34 self._x509_obj = self._to_cryptography() 

35 except MissingInit: 

36 logger.trace("Can't create crypto object before init") 

37 

38 @classmethod 

39 @abc.abstractmethod 

40 def from_cryptography( 

41 cls: type["CryptoParser"], crypto_obj: CryptoObject 

42 ) -> "CryptoParser": 

43 """ 

44 Parses a cryptography x509 object into a 

45 [CryptoParser][pki_tools.types.crypto_parser.CryptoParser] 

46 

47 Arguments: 

48 crypto_obj: The cryptography object 

49 

50 Returns: 

51 CryptoParser object 

52 """ 

53 

54 @abc.abstractmethod 

55 def _to_cryptography(self) -> CryptoObject: 

56 """ 

57 Creates a x509 cryptography object from this class 

58 

59 Returns: A x509 cryptography object 

60 """ 

61 

62 @abc.abstractmethod 

63 def _string_dict(self) -> dict[str, str]: 

64 """ 

65 Creates a dict representation of the object 

66 

67 Returns: A dict containing all the keys in the CryptoParser 

68 """ 

69 

70 def __str__(self): 

71 return yaml.safe_dump( 

72 self._string_dict(), 

73 indent=2, 

74 default_flow_style=False, 

75 explicit_start=False, 

76 default_style="", 

77 ) 

78 

79 

80class Encoding(Enum): 

81 """ 

82 Describes the encoding used for writing/reading 

83 [InitCryptoParser][pki_tools.types.crypto_parser.InitCryptoParser] 

84 objects. 

85 

86 Attributes: 

87 PEM: PEM string format 

88 DER: DER binary format 

89 """ 

90 

91 PEM = "pem_string" 

92 DER = "der_bytes" 

93 

94 

95class HelperFunc(BaseModel): 

96 func: Callable 

97 kwargs: dict | None = {} 

98 

99 

100class CryptoConfig(BaseModel): 

101 load_pem: HelperFunc 

102 load_der: HelperFunc 

103 pem_regexp: re.Pattern 

104 

105 

106class InitCryptoParser(CryptoParser, abc.ABC): 

107 """ 

108 Extends the CryptoParser into an object that requires initialization 

109 before it can be used (while created as a 

110 [pki_tools][pki_tools.types.certificate] object and not loaded from 

111 cryptography). This can, for example, be a Certificate that needs 

112 to be signed with a KeyPair containing the private key. 

113 

114 Attempt to e.g. dumping a certificate to a PEM string without using the 

115 sign (init) function first will result in a 

116 [MissingInit][pki_tools.exceptions.MissingInit] exception. 

117 

118 Classes implementing the `InitCryptoParser` will also automatically get 

119 functions to load and write the objects with the supported 

120 [Encoding][pki_tools.types.crypto_parser.Encoding] as well as 

121 writing/reading the encoded content to/from files. 

122 """ 

123 

124 _init_func: str = "sign" 

125 

126 @classmethod 

127 def from_pem_string( 

128 cls: type["InitCryptoParser"], pem: str 

129 ) -> "InitCryptoParser": 

130 """ 

131 Loads the object from a PEM string 

132 

133 Arguments: 

134 pem: The PEM encoded object in string format 

135 

136 Returns: 

137 A created object from the PEM 

138 """ 

139 cfg = cls._crypto_config() 

140 

141 if not isinstance(pem, str): 

142 logger.bind(pem=pem).debug("PEM parameter is not a string") 

143 raise LoadError() 

144 

145 pem = re.sub(r"\n\s*", "\n", pem) 

146 

147 if not re.match(cfg.pem_regexp, pem): 

148 logger.bind(pem=pem).debug("PEM parameter has invalid format") 

149 raise LoadError() 

150 

151 func = cfg.load_pem.func 

152 kwargs = cfg.load_pem.kwargs 

153 try: 

154 crypto_obj = func(pem.encode(), **kwargs) 

155 except (ValueError, TypeError) as e: 

156 logger.bind(error=str(e)).debug( 

157 "Cryptography failed to load PEM object" 

158 ) 

159 raise LoadError() from e 

160 return cls.from_cryptography(crypto_obj) 

161 

162 @classmethod 

163 def from_der_bytes( 

164 cls: type["InitCryptoParser"], der: bytes 

165 ) -> "InitCryptoParser": 

166 """ 

167 Loads the object from DER bytes 

168 

169 Arguments: 

170 der: The DER encoded object 

171 

172 Returns: 

173 A created object from the DER bytes 

174 """ 

175 cfg = cls._crypto_config() 

176 

177 func = cfg.load_der.func 

178 kwargs = cfg.load_der.kwargs 

179 

180 try: 

181 crypto_obj = func(der, **kwargs) 

182 except (ValueError, TypeError) as e: 

183 logger.bind(error=str(e)).debug( 

184 "Cryptography failed to load DER object" 

185 ) 

186 raise LoadError() from e 

187 return cls.from_cryptography(crypto_obj) 

188 

189 @classmethod 

190 def from_file( 

191 cls, file_path: str, encoding: Encoding = Encoding.PEM 

192 ) -> "InitCryptoParser": 

193 """ 

194 Reads a file containing one PEM into the object 

195 

196 Arguments: 

197 file_path: The path to the file (can be relative the caller or 

198 absolute) 

199 encoding: The encoding to use for the dumped data 

200 Returns: 

201 The Certificate loaded from the specified file 

202 """ 

203 mode = "r" 

204 if encoding == Encoding.DER: 

205 mode = "rb" 

206 

207 with open(file_path, mode) as f: 

208 data = f.read() 

209 

210 return getattr(cls, f"from_{encoding.value}")(data) 

211 

212 @property 

213 def der_bytes(self) -> bytes: 

214 """ 

215 Returns the DER bytes of the object 

216 

217 Returns: 

218 The DER bytes. 

219 """ 

220 return self._crypto_object.public_bytes( 

221 encoding=serialization.Encoding.DER 

222 ) 

223 

224 @property 

225 def pem_bytes(self) -> bytes: 

226 """ 

227 Returns the PEM bytes of the object 

228 

229 Returns: 

230 The PEM bytes. 

231 """ 

232 return self._crypto_object.public_bytes( 

233 encoding=serialization.Encoding.PEM 

234 ) 

235 

236 @property 

237 def pem_string(self) -> str: 

238 """ 

239 Returns: 

240 PEM decoded into a string 

241 """ 

242 return self.pem_bytes.decode() 

243 

244 def to_file( 

245 self, file_path: str, encoding: Encoding = Encoding.PEM 

246 ) -> None: 

247 """ 

248 Saves the object with specified encoding to the specified file, 

249 creating it if it doesn't exist. 

250 

251 Args: 

252 file_path: The path to the file (can be relative the caller or 

253 absolute) 

254 encoding: The encoding to use for the dumped data 

255 """ 

256 mode = "w" 

257 if encoding == Encoding.DER: 

258 mode = "wb" 

259 with open(file_path, mode) as f: 

260 f.write(getattr(self, encoding.value)) 

261 

262 @classmethod 

263 @abc.abstractmethod 

264 def _crypto_config(cls) -> CryptoConfig: 

265 pass 

266 

267 @property 

268 def _crypto_object(self) -> CryptoObject: 

269 if not hasattr(self, "_x509_obj") or self._x509_obj is None: 

270 init_func = f"{self.__class__.__name__}.{self._init_func}" 

271 raise MissingInit(f"Please use the {init_func} first") 

272 

273 return self._x509_obj