| 19 | |
| 20 | |
| 21 | class InternalCryptoCaller(CryptoCaller): |
| 22 | def fail(self, msg: str) -> None: |
| 23 | raise InternalError(msg) |
| 24 | |
| 25 | def password_hash(self, password: str, salt_password: str) -> str: |
| 26 | salt = salt_password.encode() if salt_password else bcrypt.gensalt() |
| 27 | return bcrypt.hashpw(password.encode(), salt).decode() |
| 28 | |
| 29 | def verify_password(self, password: str, hashed_password: str) -> bool: |
| 30 | _password = password.encode() |
| 31 | _hashed_password = hashed_password.encode() |
| 32 | try: |
| 33 | ok = bcrypt.checkpw(_password, _hashed_password) |
| 34 | except ValueError as err: |
| 35 | self.fail(str(err)) |
| 36 | return ok |
| 37 | |
| 38 | def create_private_key(self) -> str: |
| 39 | pkey = crypto.PKey() |
| 40 | pkey.generate_key(crypto.TYPE_RSA, 2048) |
| 41 | return crypto.dump_privatekey(crypto.FILETYPE_PEM, pkey).decode() |
| 42 | |
| 43 | def create_self_signed_cert( |
| 44 | self, dname: Dict[str, str], pkey: str |
| 45 | ) -> str: |
| 46 | _pkey = crypto.load_privatekey(crypto.FILETYPE_PEM, pkey) |
| 47 | |
| 48 | # Create a "subject" object |
| 49 | with warnings.catch_warnings(): |
| 50 | warnings.simplefilter("ignore") |
| 51 | req = crypto.X509Req() |
| 52 | subj = req.get_subject() |
| 53 | |
| 54 | # populate the subject with the dname settings |
| 55 | for k, v in dname.items(): |
| 56 | setattr(subj, k, v) |
| 57 | |
| 58 | # create a self-signed cert |
| 59 | cert = crypto.X509() |
| 60 | cert.set_subject(req.get_subject()) |
| 61 | cert.set_serial_number(int(uuid4())) |
| 62 | cert.gmtime_adj_notBefore(0) |
| 63 | cert.gmtime_adj_notAfter(10 * 365 * 24 * 60 * 60) # 10 years |
| 64 | cert.set_issuer(cert.get_subject()) |
| 65 | cert.set_pubkey(_pkey) |
| 66 | cert.sign(_pkey, 'sha512') |
| 67 | return crypto.dump_certificate(crypto.FILETYPE_PEM, cert).decode() |
| 68 | |
| 69 | def _load_cert(self, crt: Union[str, bytes]) -> Any: |
| 70 | crt_buffer = crt.encode() if isinstance(crt, str) else crt |
| 71 | try: |
| 72 | cert = crypto.load_certificate(crypto.FILETYPE_PEM, crt_buffer) |
| 73 | except (ValueError, crypto.Error) as e: |
| 74 | self.fail('Invalid certificate: %s' % str(e)) |
| 75 | return cert |
| 76 | |
| 77 | def _issuer_info(self, cert: Any) -> Tuple[str, str]: |
| 78 | components = cert.get_issuer().get_components() |