|
| 1 | +""" |
| 2 | +Firmitas |
| 3 | +Copyright (C) 2023-2024 Akashdeep Dhar |
| 4 | +
|
| 5 | +This program is free software: you can redistribute it and/or modify it under |
| 6 | +the terms of the GNU General Public License as published by the Free Software |
| 7 | +Foundation, either version 3 of the License, or (at your option) any later |
| 8 | +version. |
| 9 | +
|
| 10 | +This program is distributed in the hope that it will be useful, but WITHOUT |
| 11 | +ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS |
| 12 | +FOR A PARTICULAR PURPOSE. See the GNU General Public License for more |
| 13 | +details. |
| 14 | +
|
| 15 | +You should have received a copy of the GNU General Public License along with |
| 16 | +this program. If not, see <https://www.gnu.org/licenses/>. |
| 17 | +
|
| 18 | +Any Red Hat trademarks that are incorporated in the source code or |
| 19 | +documentation are not subject to the GNU General Public License and may only |
| 20 | +be used or replicated with the express permission of Red Hat, Inc. |
| 21 | +""" |
| 22 | + |
| 23 | + |
| 24 | +from pathlib import Path |
| 25 | + |
| 26 | +import yaml |
| 27 | +from cryptography import x509 |
| 28 | +from cryptography.hazmat.backends import default_backend |
| 29 | + |
| 30 | +from firmitas.base.maintool import readcert |
| 31 | +from firmitas.conf import logrdata, standard |
| 32 | + |
| 33 | + |
| 34 | +class Difference: |
| 35 | + def __init__(self): |
| 36 | + # Populate a list of certificate files from the device |
| 37 | + self.certlist_from_device = [] |
| 38 | + certloca = Path(standard.certloca) |
| 39 | + for item in certloca.iterdir(): |
| 40 | + if not item.is_file() or ".crt" not in item.name: |
| 41 | + continue |
| 42 | + self.certlist_from_device.append(item.name) |
| 43 | + |
| 44 | + # Populate a list of certificate files from the config |
| 45 | + self.certlist_from_config = [] |
| 46 | + self.certdict = yaml.safe_load(Path(standard.hostloca).read_text()) |
| 47 | + for item in self.certdict: |
| 48 | + self.certlist_from_config.append(self.certdict[item]["path"]) |
| 49 | + |
| 50 | + # Compute a list of certificate files to insert into the config |
| 51 | + self.to_insert = [ |
| 52 | + item for item in self.certlist_from_device if item not in self.certlist_from_config |
| 53 | + ] |
| 54 | + |
| 55 | + # Compute a list of certificate files to remove from the config |
| 56 | + self.to_remove = [ |
| 57 | + item for item in self.certlist_from_config if item not in self.certlist_from_device |
| 58 | + ] |
| 59 | + |
| 60 | + def insert(self): |
| 61 | + logrdata.logrobjc.warning( |
| 62 | + f"Inserting {len(self.to_insert)} certificate(s) that are now tracked" |
| 63 | + ) |
| 64 | + |
| 65 | + for item in self.to_insert: |
| 66 | + certname = item.replace(".crt", "") |
| 67 | + certpath = Path(standard.certloca, item) |
| 68 | + try: |
| 69 | + certobjc = x509.load_pem_x509_certificate(certpath.read_bytes(), default_backend()) |
| 70 | + readdata = readcert(certobjc) |
| 71 | + except ValueError: |
| 72 | + logrdata.logrobjc.error( |
| 73 | + f"[{certname}] The specified X.509-standard TLS certificate could not be read" |
| 74 | + ) |
| 75 | + else: |
| 76 | + logrdata.logrobjc.info( |
| 77 | + f"[{certname}] The specified X.509-standard TLS certificate was read successfully" # noqa : E501 |
| 78 | + ) |
| 79 | + self.certdict[certname] = { |
| 80 | + "path": item, |
| 81 | + "user": standard.username, |
| 82 | + "certstat": { |
| 83 | + "strtdate": readdata[0], |
| 84 | + "stopdate": readdata[1], |
| 85 | + "cstarted": readdata[2], |
| 86 | + "cstopped": readdata[3], |
| 87 | + "daystobt": readdata[4], |
| 88 | + "daystodd": readdata[5], |
| 89 | + "issuauth": readdata[6], |
| 90 | + "serialno": readdata[7], |
| 91 | + }, |
| 92 | + "notistat": { |
| 93 | + "done": False, |
| 94 | + "link": "", |
| 95 | + "time": "", |
| 96 | + } |
| 97 | + } |
| 98 | + |
| 99 | + def remove(self): |
| 100 | + logrdata.logrobjc.warning( |
| 101 | + f"Removing {len(self.to_remove)} certificate(s) that are not tracked" |
| 102 | + ) |
| 103 | + |
| 104 | + for item in self.to_remove: |
| 105 | + certname = item.replace(".crt", "") |
| 106 | + self.certdict.pop(certname) |
| 107 | + |
| 108 | + def action(self): |
| 109 | + # Do not act if the list is empty |
| 110 | + if len(self.to_insert) > 0: |
| 111 | + self.insert() |
| 112 | + |
| 113 | + # Do not act if the list is empty |
| 114 | + if len(self.to_remove) > 0: |
| 115 | + self.remove() |
| 116 | + |
| 117 | + with open(standard.hostloca, "w") as yamlfile: |
| 118 | + yaml.safe_dump(self.certdict, yamlfile) |
0 commit comments