Avoid heap corruption with NULLs in TVPs (#1450)

src/params.cpp

@@ -534,9 +534,15 @@ static void FreeInfos(ParamInfo* a, Py_ssize_t count)
     PyMem_Free(a);
 }
 
-static bool GetNullInfo(Cursor* cur, Py_ssize_t index, ParamInfo& info)
+static bool GetNullInfo(Cursor* cur, Py_ssize_t index, ParamInfo& info, bool isTVP)
 {
-    if (!GetParamType(cur, index, info.ParameterType))
+    // GetParamType won't work for TVP columns, so we fall back on SQL_VARCHAR.
+    if (isTVP)
+    {
+        if (info.ParameterType == SQL_UNKNOWN_TYPE)
+            info.ParameterType = SQL_VARCHAR;
+    }
+    else if (!GetParamType(cur, index, info.ParameterType))
         return false;
 
     info.ValueType     = SQL_C_DEFAULT;
@@ -1023,7 +1029,7 @@ bool GetParameterInfo(Cursor* cur, Py_ssize_t index, PyObject* param, ParamInfo&
     // Populates `info`.
 
     if (param == Py_None)
-        return GetNullInfo(cur, index, info);
+        return GetNullInfo(cur, index, info, isTVP);
 
     if (param == null_binary)
         return GetNullBinaryInfo(cur, index, info);

tests/sqlserver_test.py

@@ -1,5 +1,6 @@
 #!/usr/bin/python
 
+import gc
 import os
 import re
 import uuid
@@ -1617,6 +1618,53 @@ def test_tvp_diffschema(cursor: pyodbc.Cursor):
     _test_tvp(cursor, True)
 
 
+def _test_tvp_with_nulls_cleanup(cursor: pyodbc.Cursor, procname: str, typename: str):
+    """Leave the forest as pristine as you found it."""
+
+    cursor.execute(f"""\
+        IF OBJECT_ID(N'dbo.{procname}', N'P') IS NOT NULL
+        DROP PROCEDURE dbo.{procname};
+    """)
+    cursor.execute(f"""
+        IF TYPE_ID(N'dbo.{typename}') IS NOT NULL
+            DROP TYPE dbo.{typename};
+    """)
+
+
+@pytest.mark.skipif(SQLSERVER_YEAR < 2008, reason="TVP not supported until 2008")
+@pytest.mark.skipif(IS_FREEDTS, reason="FreeTDS does not support TVP")
+def test_tvp_with_nulls(cursor: pyodbc.Cursor):
+    """Make sure NULL values in a TVP don't crash the interpreter."""
+
+    # Start with a clean slate.
+    typename = "typeTestNullsInTVP"
+    procname = "spTestNullsInTVP"
+    _test_tvp_with_nulls_cleanup(cursor, procname, typename)
+
+    # Create the custom type and stored procedure.
+    ncols = 100
+    cols = ", ".join([f"col_{c:03d} DECIMAL(36,20)" for c in range(1, ncols+1)])
+    cursor.execute(f"CREATE TYPE dbo.{typename} AS TABLE ({cols})")
+    cursor.execute(f"""\
+        CREATE PROCEDURE dbo.{procname}
+            @data dbo.{typename} READONLY
+        AS
+        BEGIN
+            RETURN 0;
+        END;
+    """)
+    cursor.commit()
+
+    # Invoke the stored procedure.
+    tvp: list[list] = [[3.14159] * ncols, [None] * ncols]
+    cursor.execute(f"EXEC [dbo].{procname} @data=?", [tvp])
+    gc.collect()
+
+    # Be a good digital citizen.
+    _test_tvp_with_nulls_cleanup(cursor, procname, typename)
+    cursor.commit()
+
+
 @pytest.mark.skipif(SQLSERVER_YEAR < 2000, reason='sql_variant not supported until 2000')
 def test_sql_variant(cursor: pyodbc.Cursor):
     """