feat: provision to handle custom tls certificates

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);
+		});
 	},
 });

press/press/doctype/tls_certificate/tls_certificate.json

@@ -11,19 +11,21 @@
   "column_break_3",
   "rsa_key_size",
   "wildcard",
+  "custom",
   "section_break_6",
   "issued_on",
   "column_break_8",
   "expires_on",
   "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",

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"
 		)