| 191 | |
| 192 | |
| 193 | class CertStore: |
| 194 | |
| 195 | def __init__(self, fpath: str): |
| 196 | self._store_dir = fpath |
| 197 | if not os.path.exists(self._store_dir): |
| 198 | os.makedirs(self._store_dir) |
| 199 | self._creds_by_name = {} |
| 200 | |
| 201 | @property |
| 202 | def path(self) -> str: |
| 203 | return self._store_dir |
| 204 | |
| 205 | def save(self, creds: Credentials, name: str = None, |
| 206 | chain: List[Credentials] = None, |
| 207 | single_file: bool = False) -> None: |
| 208 | name = name if name is not None else creds.name |
| 209 | cert_file = self.get_cert_file(name=name, key_type=creds.key_type) |
| 210 | pkey_file = self.get_pkey_file(name=name, key_type=creds.key_type) |
| 211 | if single_file: |
| 212 | pkey_file = None |
| 213 | with open(cert_file, "wb") as fd: |
| 214 | fd.write(creds.cert_pem) |
| 215 | if chain: |
| 216 | for c in chain: |
| 217 | fd.write(c.cert_pem) |
| 218 | if pkey_file is None: |
| 219 | fd.write(creds.pkey_pem) |
| 220 | if pkey_file is not None: |
| 221 | with open(pkey_file, "wb") as fd: |
| 222 | fd.write(creds.pkey_pem) |
| 223 | creds.set_files(cert_file, pkey_file) |
| 224 | self._add_credentials(name, creds) |
| 225 | |
| 226 | def save_chain(self, creds: Credentials, infix: str, with_root=False): |
| 227 | name = creds.name |
| 228 | chain = [creds] |
| 229 | while creds.issuer is not None: |
| 230 | creds = creds.issuer |
| 231 | chain.append(creds) |
| 232 | if not with_root and len(chain) > 1: |
| 233 | chain = chain[:-1] |
| 234 | chain_file = os.path.join(self._store_dir, f'{name}-{infix}.pem') |
| 235 | with open(chain_file, "wb") as fd: |
| 236 | for c in chain: |
| 237 | fd.write(c.cert_pem) |
| 238 | |
| 239 | def _add_credentials(self, name: str, creds: Credentials): |
| 240 | if name not in self._creds_by_name: |
| 241 | self._creds_by_name[name] = [] |
| 242 | self._creds_by_name[name].append(creds) |
| 243 | |
| 244 | def get_credentials_for_name(self, name) -> List[Credentials]: |
| 245 | return self._creds_by_name[name] if name in self._creds_by_name else [] |
| 246 | |
| 247 | def get_cert_file(self, name: str, key_type=None) -> str: |
| 248 | key_infix = ".{0}".format(key_type) if key_type is not None else "" |
| 249 | return os.path.join(self._store_dir, f'{name}{key_infix}.cert.pem') |
| 250 |
no outgoing calls
no test coverage detected