feat: Provision to add Custom TLS

press/hooks.py

@@ -193,6 +193,7 @@
 		"press.press.doctype.press_webhook_log.press_webhook_log.clean_logs_older_than_24_hours",
 		"press.press.doctype.virtual_disk_snapshot.virtual_disk_snapshot.sync_all_snapshots_from_aws",
 		"press.press.doctype.payment_due_extension.payment_due_extension.remove_payment_due_extension",
+		"press.press.doctype.tls_certificate.tls_certificate.alert_custom_provider_tls_renewal",
 	],
 	"hourly": [
 		"press.press.doctype.site.backups.cleanup_local",

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,6 +12,27 @@ frappe.ui.form.on('TLS Certificate', {
 				});
 			});
 		}
+
+		frm.trigger('show_obtain_certificate');
+		frm.trigger('toggle_read_only');
+		frm.trigger('toggle_hidden');
+		frm.trigger('toggle_copy_private_key');
+	},
+
+	provider: function (frm) {
+		frm.trigger('show_obtain_certificate');
+		frm.trigger('toggle_read_only');
+		frm.trigger('toggle_hidden');
+		frm.trigger('toggle_copy_private_key');
+	},
+
+	wildcard: function (frm) {
+		frm.trigger('toggle_read_only');
+		frm.trigger('toggle_hidden');
+		frm.trigger('toggle_copy_private_key');
+	},
+
+	toggle_copy_private_key: function (frm) {
 		if (!frm.doc.wildcard) {
 			frm.add_custom_button('Copy Private Key', () => {
 				frappe.confirm(
@@ -30,6 +44,54 @@ frappe.ui.form.on('TLS Certificate', {
 					() => frappe.utils.copy_to_clipboard(frm.doc.private_key),
 				);
 			});
+		} else {
+			if (frm.doc.provider == "Let's Encrypt") {
+				console.log("Let's Encrypt");
+				frm.remove_custom_button('Copy Private Key');
+			}
+		}
+	},
+
+	show_obtain_certificate: function (frm) {
+		if (frm.doc.provider == "Let's Encrypt") {
+			frm.add_custom_button(__('Obtain Certificate'), () => {
+				frm.call({
+					method: 'obtain_certificate',
+					doc: frm.doc,
+					callback: (result) => frm.refresh(),
+				});
+			});
+		} else {
+			frm.remove_custom_button(__('Obtain Certificate'));
 		}
 	},
+
+	toggle_read_only: function (frm) {
+		let fields = [
+			'certificate',
+			'private_key',
+			'intermediate_chain',
+			'full_chain',
+			'issued_on',
+			'expires_on',
+			'team',
+		];
+		fields.forEach(function (field) {
+			frm.set_df_property(
+				field,
+				'read_only',
+				frm.doc.provider == "Let's Encrypt",
+			);
+			frm.refresh_field(field);
+		});
+	},
+
+	toggle_hidden: function (frm) {
+		frm.set_df_property(
+			'private_key',
+			'hidden',
+			frm.doc.provider == "Let's Encrypt",
+		);
+		frm.refresh_field('private_key');
+	},
 });

press/press/doctype/tls_certificate/tls_certificate.json

@@ -19,8 +19,8 @@
   "section_break_10",
   "decoded_certificate",
   "certificate",
-  "full_chain",
   "intermediate_chain",
+  "full_chain",
   "private_key",
   "section_break_cvcg",
   "error",
@@ -50,8 +50,7 @@
    "fieldname": "decoded_certificate",
    "fieldtype": "Code",
    "label": "Decoded Certificate",
-   "read_only": 1,
-   "read_only_depends_on": "eval: doc.provider === \"Let's Encrypt\""
+   "read_only": 1
   },
   {
    "fieldname": "status",
@@ -77,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.provider === 'Other'",
    "reqd": 1
   },
   {
@@ -103,20 +102,23 @@
    "hide_border": 1
   },
   {
+   "description": "Only Domain\u2019s Certificate",
    "fieldname": "certificate",
    "fieldtype": "Code",
    "label": "Certificate",
    "read_only": 1,
    "read_only_depends_on": "eval: doc.provider === \"Let's Encrypt\""
   },
   {
+   "devscription": "Certificate + Intermediate + Trust chain for non lets encrypt certificates",
    "fieldname": "full_chain",
    "fieldtype": "Code",
    "label": "Full Chain",
    "read_only": 1,
    "read_only_depends_on": "eval: doc.provider == \"Let's Encrypt\""
   },
   {
+   "description": "Chain certificate to establish trust (Intermediate + Trust chain for non lets encrypt certificates)",
    "fieldname": "intermediate_chain",
    "fieldtype": "Code",
    "label": "Intermediate Chain",
@@ -169,7 +171,7 @@
  "grid_page_length": 50,
  "index_web_pages_for_search": 1,
  "links": [],
- "modified": "2025-03-30 18:03:13.270000",
+ "modified": "2025-04-07 14:24:48.585403",
  "modified_by": "Administrator",
  "module": "Press",
  "name": "TLS Certificate",
@@ -207,4 +209,4 @@
  "sort_order": "DESC",
  "states": [],
  "track_changes": 1
-}
+}

press/press/doctype/tls_certificate/tls_certificate.py

@@ -13,6 +13,7 @@
 import frappe
 import OpenSSL
 from frappe.model.document import Document
+from frappe.query_builder.functions import Date
 
 from press.api.site import check_dns_cname_a
 from press.exceptions import (
@@ -61,6 +62,16 @@
 	def after_insert(self):
 		self.obtain_certificate()
 
+	def validate(self):
+		if self.provider == "Other":
+			if not self.team:
+				frappe.throw("Team is mandatory for custom TLS certificates.")
+
+			self.configure_full_chain()
+			self.validate_key_length()
+			self.validate_key_certificate_association()
+			self._extract_certificate_details()
+
 	def on_update(self):
 		if self.is_new():
 			return
@@ -72,6 +83,7 @@
 	def obtain_certificate(self):
 		if self.provider != "Let's Encrypt":
 			return
+
 		if self.retry_count >= RETRY_LIMIT:
 			frappe.throw("Retry limit exceeded. Please check the error and try again.", TLSRetryLimitExceeded)
 		(
@@ -83,6 +95,7 @@
 			frappe.session.data,
 			get_current_team(),
 		)
+
 		frappe.set_user(frappe.get_value("Team", team, "user"))
 		frappe.enqueue_doc(
 			self.doctype,
@@ -200,6 +213,50 @@
 		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 configure_full_chain(self):
+		if not self.full_chain:
+			self.full_chain = f"{self.certificate}\n{self.intermediate_chain}"
+
+	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):
+		try:
+			return OpenSSL.crypto.load_certificate(OpenSSL.crypto.FILETYPE_PEM, self.full_chain)
+		except OpenSSL.crypto.Error as e:
+			log_error("Custom TLS Certificate Exception", certificate=self.name)
+			raise e
+
+	def validate_key_length(self):
+		private_key = self._get_private_key_object()
+
+		if private_key.bits() != int(self.rsa_key_size):
+			frappe.throw(
+				f"Private key length does not match the selected RSA key size. Expected {self.rsa_key_size} bits, got {private_key.bits()} bits."
+			)
+
+	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()
+			self.status = "Active"
+			self.retry_count = 0
+			self.error = None
+		except OpenSSL.SSL.Error as e:
+			self.error = repr(e)
+			log_error("TLS Key Certificate Association Exception", certificate=self.name)
+			frappe.throw("Private Key and Certificate do not match")
+		finally:
+			if self.error:
+				self.status = "Failure"
+
 
 get_permission_query_conditions = get_permission_query_conditions_for_doctype("TLS Certificate")
 
@@ -264,7 +321,9 @@
 	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")
+
 		try:
 			if not should_renew(site, certificate):
 				continue
@@ -286,6 +345,35 @@
 			frappe.db.commit()
 
 
+def alert_custom_provider_tls_renewal():
+	seven_days = frappe.utils.add_days(None, 7).date()
+	fifteen_days = frappe.utils.add_days(None, 15).date()
+
+	tls_cert = frappe.qb.DocType("TLS Certificate")
+
+	# Notify team members 15 days and 7 days before expiry
+
+	query = (
+		frappe.qb.from_(tls_cert)
+		.select(tls_cert.name, tls_cert.domain, tls_cert.team, tls_cert.expires_on)
+		.where(tls_cert.status.isin(["Active", "Failure"]))
+		.where((Date(tls_cert.expires_on) == seven_days) | (Date(tls_cert.expires_on) == fifteen_days))
+		.where(tls_cert.provider == "Other")
+	)
+
+	pending = query.run(as_dict=True)
+
+	for certificate in pending:
+		if certificate.team:
+			notify_email = frappe.get_value("Team", certificate.team, "notify_email")
+
+			frappe.sendmail(
+				recipients=notify_email,
+				subject=f"TLS Certificate Renewal Required: {certificate.name}",
+				message=f"TLS Certificate {certificate.name} is due for renewal on {certificate.expires_on}. Please renew the certificate to avoid service disruption.",
+			)
+
+
 def update_server_tls_certifcate(server, certificate):
 	try:
 		proxysql_admin_password = None