diff --git a/ddpui/core/dbtfunctions.py b/ddpui/core/dbtfunctions.py
index eba511b70..e4a09998f 100644
--- a/ddpui/core/dbtfunctions.py
+++ b/ddpui/core/dbtfunctions.py
@@ -41,10 +41,15 @@ def map_airbyte_destination_spec_to_dbtcli_profile(
         if mode:
             conn_info["sslmode"] = mode
 
-        if ca_certificate and dbt_project_params.org_project_dir:
-            file_path = os.path.join(dbt_project_params.org_project_dir, "sslrootcert.pem")
-            with open(file_path, "w", encoding="utf-8") as file:
-                file.write(ca_certificate)
-            conn_info["sslrootcert"] = file_path
+        if ca_certificate:
+            if not dbt_project_params or not dbt_project_params.org_project_dir:
+                raise Exception(
+                    "org_project_dir is required to save the ca_certificate for dbt ssl connections"
+                )
+
+            conn_info["sslrootcert_content"] = ca_certificate
+            conn_info["sslrootcert"] = os.path.join(
+                dbt_project_params.org_project_dir, "sslrootcert.pem"
+            )
 
     return conn_info
diff --git a/ddpui/ddpdbt/schema.py b/ddpui/ddpdbt/schema.py
index de92e02bf..f9e0cd62e 100644
--- a/ddpui/ddpdbt/schema.py
+++ b/ddpui/ddpdbt/schema.py
@@ -1,4 +1,4 @@
-from typing import Union
+from typing import Union, Optional
 from ninja import Schema
 from pathlib import Path
 
diff --git a/ddpui/tests/core/test_dbtfunctions.py b/ddpui/tests/core/test_dbtfunctions.py
index f59945cda..cc390b3c4 100644
--- a/ddpui/tests/core/test_dbtfunctions.py
+++ b/ddpui/tests/core/test_dbtfunctions.py
@@ -1,3 +1,5 @@
+import os
+
 from ddpui.ddpdbt.schema import DbtProjectParams
 from ddpui.core.dbtfunctions import map_airbyte_destination_spec_to_dbtcli_profile
 
@@ -63,7 +65,7 @@ def test_map_airbyte_destination_spec_to_dbtcli_profile_success_tunnel_params(tm
 
 
 def test_map_airbyte_destination_spec_to_dbtcli_profile_success_ssl_params(tmpdir):
-    """Tests all the success cases"""
+    """Tests ssl params are stored in conn_info for runtime cert writing"""
     dbt_project_params = DbtProjectParams(
         org_project_dir=str(tmpdir),
         dbt_env_dir="/path/to/dbt_venv",
@@ -78,7 +80,39 @@ def test_map_airbyte_destination_spec_to_dbtcli_profile_success_ssl_params(tmpdi
 
     conn_info = {"ssl_mode": {"mode": "verify-ca", "ca_certificate": "ca_certificate"}}
     res = map_airbyte_destination_spec_to_dbtcli_profile(conn_info, dbt_project_params)
-    assert res["sslmode"] == conn_info["ssl_mode"]["mode"]
+    assert res["sslmode"] == "verify-ca"
     assert res["sslrootcert"] == f"{tmpdir}/sslrootcert.pem"
-    with open(f"{tmpdir}/sslrootcert.pem") as file:
-        assert file.read() == conn_info["ssl_mode"]["ca_certificate"]
+    assert res["sslrootcert_content"] == "ca_certificate"
+    # cert should NOT be written to disk at setup time
+    assert not os.path.exists(f"{tmpdir}/sslrootcert.pem")
+
+
+def test_map_airbyte_destination_spec_to_dbtcli_profile_ssl_mode_only(tmpdir):
+    """Tests ssl_mode with mode but no ca_certificate"""
+    dbt_project_params = DbtProjectParams(
+        org_project_dir=str(tmpdir),
+        dbt_env_dir="/path/to/dbt_venv",
+        dbt_repo_dir="/path/to/dbt_repo",
+        project_dir="/path/to/project_dir",
+        target="target",
+        dbt_binary="dbt_binary",
+        venv_binary="path/to/venv/bin",
+        clients_base_dir="/path/to/clients_base",
+        project_dir_relative="org/dbtrepo",
+    )
+
+    conn_info = {"ssl_mode": {"mode": "require"}}
+    res = map_airbyte_destination_spec_to_dbtcli_profile(conn_info, dbt_project_params)
+    assert res["sslmode"] == "require"
+    assert "sslrootcert" not in res
+    assert "sslrootcert_content" not in res
+
+
+def test_map_airbyte_destination_spec_to_dbtcli_profile_ssl_no_org_project_dir():
+    """Tests ssl with ca_certificate but no dbt_project_params raises"""
+    conn_info = {"ssl_mode": {"mode": "verify-ca", "ca_certificate": "ca_certificate"}}
+    try:
+        map_airbyte_destination_spec_to_dbtcli_profile(conn_info, None)
+        assert False, "should have raised"
+    except Exception as e:
+        assert "org_project_dir is required" in str(e)