diff --git a/press/press/doctype/tls_certificate/tls_certificate.js b/press/press/doctype/tls_certificate/tls_certificate.js index 658ab77047a..8e35c1fe32f 100644 --- a/press/press/doctype/tls_certificate/tls_certificate.js +++ b/press/press/doctype/tls_certificate/tls_certificate.js @@ -3,13 +3,6 @@ frappe.ui.form.on('TLS Certificate', { refresh: function (frm) { - frm.add_custom_button(__('Obtain Certificate'), () => { - frm.call({ - method: 'obtain_certificate', - doc: frm.doc, - callback: (result) => frm.refresh(), - }); - }); if (frm.doc.wildcard) { frm.add_custom_button(__('Trigger Callback'), () => { frm.call({ @@ -19,5 +12,39 @@ frappe.ui.form.on('TLS Certificate', { }); }); } + + frm.trigger('show_obtain_certificate'); + frm.trigger('toggle_read_only'); + }, + + custom: function (frm) { + frm.trigger('show_obtain_certificate'); + frm.trigger('toggle_read_only'); + }, + show_obtain_certificate: function (frm) { + if (!frm.doc.custom) { + frm.add_custom_button(__('Obtain Certificate'), () => { + frm.call({ + method: 'obtain_certificate', + doc: frm.doc, + callback: (result) => frm.refresh(), + }); + }); + } + }, + + toggle_read_only: function (frm) { + let fields = [ + 'certificate', + 'private_key', + 'intermediate_chain', + 'full_chain', + 'issued_on', + 'expires_on', + ]; + fields.forEach(function (field) { + frm.set_df_property(field, 'read_only', !frm.doc.custom); + frm.refresh_field(field); + }); }, }); diff --git a/press/press/doctype/tls_certificate/tls_certificate.json b/press/press/doctype/tls_certificate/tls_certificate.json index b9a8339788e..2c03f701ecb 100644 --- a/press/press/doctype/tls_certificate/tls_certificate.json +++ b/press/press/doctype/tls_certificate/tls_certificate.json @@ -11,6 +11,7 @@ "column_break_3", "rsa_key_size", "wildcard", + "custom", "section_break_6", "issued_on", "column_break_8", @@ -18,12 +19,13 @@ "section_break_10", "decoded_certificate", "certificate", - "full_chain", "intermediate_chain", + "full_chain", "private_key", "section_break_cvcg", - "error", - "retry_count" + "retry_count", + "error_tab", + "error" ], "fields": [ { @@ -74,7 +76,7 @@ "fieldtype": "Select", "label": "RSA Key Size", "options": "2048\n3072\n4096", - "read_only_depends_on": "eval: doc.wildcard", + "read_only_depends_on": "eval: doc.wildcard && doc.custom", "reqd": 1 }, { @@ -147,11 +149,22 @@ "fieldtype": "Int", "label": "Retry Count", "read_only": 1 + }, + { + "default": "0", + "fieldname": "custom", + "fieldtype": "Check", + "label": "Custom" + }, + { + "fieldname": "error_tab", + "fieldtype": "Tab Break", + "label": "Error" } ], "index_web_pages_for_search": 1, "links": [], - "modified": "2024-06-10 10:43:33.881463", + "modified": "2024-08-14 18:40:35.252344", "modified_by": "Administrator", "module": "Press", "name": "TLS Certificate", diff --git a/press/press/doctype/tls_certificate/tls_certificate.py b/press/press/doctype/tls_certificate/tls_certificate.py index 9c60468d3a9..5c978000e7d 100644 --- a/press/press/doctype/tls_certificate/tls_certificate.py +++ b/press/press/doctype/tls_certificate/tls_certificate.py @@ -29,6 +29,7 @@ class TLSCertificate(Document): from frappe.types import DF certificate: DF.Code | None + custom: DF.Check decoded_certificate: DF.Code | None domain: DF.Data error: DF.Code | None @@ -50,6 +51,10 @@ def autoname(self): else: self.name = self.domain + def validate(self): + self.configure_full_chain() + self.validate_custom_certificate() + def after_insert(self): self.obtain_certificate() @@ -60,8 +65,21 @@ def on_update(self): if self.has_value_changed("rsa_key_size"): self.obtain_certificate() + self._trigger_callbacks_for_custom_certificate() + + def _trigger_callbacks_for_custom_certificate(self): + if self.custom and ( + self.has_value_changed("certificate") or self.has_value_changed("private_key") + ): + frappe.enqueue_doc( + self.doctype, self.name, "_trigger_callbacks", enqueue_after_commit=True + ) + @frappe.whitelist() def obtain_certificate(self): + if self.custom: + return + (user, session_data, team,) = ( frappe.session.user, frappe.session.data, @@ -165,6 +183,11 @@ def trigger_self_hosted_server_callback(self): except Exception: pass + def configure_full_chain(self): + if self.custom and not self.full_chain: + if self.certificate and self.intermediate_chain: + self.full_chain = f"{self.certificate}\n{self.intermediate_chain}" + def _extract_certificate_details(self): x509 = OpenSSL.crypto.load_certificate(OpenSSL.crypto.FILETYPE_PEM, self.certificate) self.decoded_certificate = OpenSSL.crypto.dump_certificate( @@ -173,6 +196,64 @@ def _extract_certificate_details(self): self.issued_on = datetime.strptime(x509.get_notBefore().decode(), "%Y%m%d%H%M%SZ") self.expires_on = datetime.strptime(x509.get_notAfter().decode(), "%Y%m%d%H%M%SZ") + def _get_private_key_object(self): + try: + return OpenSSL.crypto.load_privatekey(OpenSSL.crypto.FILETYPE_PEM, self.private_key) + except OpenSSL.crypto.Error as e: + log_error("TLS Private Key Exception", certificate=self.name) + raise e + + def _get_certificate_object(self): + certificate = self._get_certificate() + try: + return OpenSSL.crypto.load_certificate(OpenSSL.crypto.FILETYPE_PEM, certificate) + except OpenSSL.crypto.Error as e: + log_error("Custom TLS Certificate Exception", certificate=self.name) + raise e + + def _validate_key_certificate_association(self): + context = OpenSSL.SSL.Context(OpenSSL.SSL.TLSv1_METHOD) + context.use_privatekey(self._get_private_key_object()) + context.use_certificate(self._get_certificate_object()) + + try: + context.check_privatekey() + except OpenSSL.SSL.Error: + log_error("TLS Key Certificate Association Exception", certificate=self.name) + frappe.throw("Private Key and Certificate do not match") + + def _get_certificate(self): + if self.full_chain: + return self.full_chain + + return self.certificate + + def validate_custom_certificate(self): + if not self.custom: + return + + if not self.certificate or not self.private_key: + return + + try: + self._validate_key_certificate_association() + self._extract_certificate_details() + + self.status = "Active" + self.retry_count = 0 + self.error = None + except Exception as e: + self.error = repr(e) + self.status = "Failure" + + def _trigger_callbacks(self): + self.trigger_site_domain_callback() + self.trigger_self_hosted_server_callback() + + if self.wildcard: + self.trigger_server_tls_setup_callback() + self._update_secondary_wildcard_domains() + get_permission_query_conditions = get_permission_query_conditions_for_doctype( "TLS Certificate" @@ -190,6 +271,7 @@ def renew_tls_certificates(): "status": ("in", ("Active", "Failure")), "expires_on": ("<", frappe.utils.add_days(None, 25)), "retry_count": ("<", 5), + "custom": 0, }, ignore_ifnull=True, order_by="expires_on ASC, status DESC", # Oldest first, then prefer failures. @@ -198,6 +280,7 @@ def renew_tls_certificates(): for certificate in pending: if tls_renewal_queue_size and (renewals_attempted >= tls_renewal_queue_size): break + site = frappe.db.get_value( "Site Domain", {"tls_certificate": certificate.name}, "site" )