FIX: PR #744 review round 2 (Sumit) — probe/audit hardening (4 Low)

- tls_connect_probe.py: _split_top_level now models a SINGLE-LEVEL braced value matching the
  production parser (connection_string_parser.py::_parse_braced_value) -- an inner `{` is a
  literal (not a nested open) and a `{` is only a brace-open at a value's START, so
  `Pwd={a{b};Encrypt=no` splits correctly and force_tls no longer emits a duplicate
  `encrypt` that mssql_python.connect would reject. (The production parser cannot be imported
  here: it pulls in mssql_python -> the native ddbc_bindings extension, which this standalone
  probe must stay importable / unit-testable without.)
- audit_bundled_binaries.py::_openssl_range_ok: the upper bound is valid only as an EXCLUSIVE
  `<` at numeric release 4.0(.0...) -- `<4`, `<4.0`, `<4.0.0`, `<4.0a0` pass; `<=4`, `<4.1`,
  `<4.0.1` (each admitting some openssl 4.x) now correctly FAIL.
- audit_bundled_binaries.py: an unknown `linux-*` subdir with no _SUBDIR_MACHINE mapping now
  FAILS CLOSED instead of proceeding with expected_machine=None and silently skipping the ELF
  architecture gate.
- _conda_pkg.py::iter_payload_members: a `.conda` missing its pkg-*.tar.zst now RAISES (was a
  bare `return` -> silent empty iteration); both audit scripts convert that to a violation.

The Medium (the job-level TLS secret never reaching the posix `bash:` step) is already
resolved: the live Encrypt=yes gate was removed from the build pipeline entirely in the
previous commit (it moves to the release pipeline, where the secret is always present so the
gate is unconditional).

+7 unit tests (119 pass). black + flake8 clean; mypy clean on the scripts.

Author: jahnvi480

conda/tls_connect_probe.py

@@ -90,37 +90,53 @@ def describe(exc):
 
 
 def _split_top_level(conn):
-    """Split an ODBC connection string on TOP-LEVEL ``;`` only.
-
-    An ODBC value wrapped in ``{...}`` may itself contain ``;`` (MS-ODBCSTR), so a
-    naive ``split(';')`` would shred braced values. Track brace depth and break only
-    at depth 0. Inside a braced value ``}}`` is an escaped literal ``}`` (MS-ODBCSTR),
-    NOT a close -- consume both and keep the depth so the value is not split early.
+    """Split an ODBC connection string on TOP-LEVEL ``;`` only, matching the production
+    parser's grammar (mssql_python/connection_string_parser.py, ``_parse_braced_value``).
+
+    A value is BRACED only when ``{`` is the first non-space char right after its ``=``;
+    inside a braced value everything is literal until a single closing ``}`` (with ``}}`` an
+    escaped literal ``}``), and an inner ``{`` is NOT a nested open -- braced values are
+    single-level. A ``;`` inside a braced value is part of the value, not a separator; a
+    ``{`` anywhere other than a value's start is a literal character.
+
+    (Reimplemented rather than importing the production parser: that module pulls in
+    ``mssql_python`` -> the native ``ddbc_bindings`` extension, which this standalone probe
+    must stay importable / unit-testable without.)
     """
     segments = []
-    buf = ""
-    depth = 0
     s = conn.strip()
+    n = len(s)
     i = 0
-    while i < len(s):
+    seg_start = 0
+    seen_eq = False  # have we passed the '=' that starts this segment's value?
+    while i < n:
         ch = s[i]
-        if ch == "{":
-            depth += 1
-            buf += ch
-        elif ch == "}":
-            if depth > 0 and i + 1 < len(s) and s[i + 1] == "}":
-                buf += "}}"  # escaped literal '}' inside a braced value; not a close
-                i += 2
-                continue
-            depth = max(0, depth - 1)
-            buf += ch
-        elif ch == ";" and depth == 0:
-            segments.append(buf)
-            buf = ""
-        else:
-            buf += ch
+        if ch == ";":
+            segments.append(s[seg_start:i])
+            i += 1
+            seg_start = i
+            seen_eq = False
+            continue
+        if ch == "=" and not seen_eq:
+            seen_eq = True
+            i += 1
+            # A braced value begins ONLY if '{' is the first non-space char after '='.
+            j = i
+            while j < n and s[j] in " \t":
+                j += 1
+            if j < n and s[j] == "{":
+                i = j + 1  # past the opening '{'
+                while i < n:
+                    if s[i] == "}":
+                        if i + 1 < n and s[i + 1] == "}":
+                            i += 2  # escaped literal '}' -- stay in the braced value
+                            continue
+                        i += 1  # single '}' closes the braced value
+                        break
+                    i += 1  # any other char (incl. an inner '{') is literal
+            continue
         i += 1
-    segments.append(buf)
+    segments.append(s[seg_start:])
     return segments
 
 

eng/scripts/_conda_pkg.py

@@ -40,7 +40,7 @@ def iter_payload_members(path: str):
                 None,
             )
             if pkg_name is None:
-                return
+                raise ValueError(f"{path}: no pkg-*.tar.zst payload found in .conda archive")
             blob = zstd_decompress(zf.read(pkg_name))
         with tarfile.open(fileobj=io.BytesIO(blob)) as tf:
             for m in tf.getmembers():

eng/scripts/assert_pe_machine.py

@@ -76,11 +76,16 @@ def audit_package(path: str) -> list[str]:
         print(f"  SKIP (no Windows PE payload): {base_name} [subdir={subdir or '?'}]")
         return []
 
+    try:
+        members = list(_iter_payload_members(path))
+    except ValueError as exc:  # malformed payload (e.g. .conda missing pkg-*.tar.zst)
+        return [f"{base_name}: unreadable/malformed package payload ({exc})."]
+
     errors: list[str] = []
     native_seen = 0
     binding_seen = 0
     driver_dll_seen = 0
-    for name, data in _iter_payload_members(path):
+    for name, data in members:
         if not name.lower().endswith(_NATIVE_SUFFIXES):
             continue
         native_seen += 1

eng/scripts/audit_bundled_binaries.py

@@ -229,12 +229,14 @@ def _dep_names(depends) -> set:
 
 
 def _openssl_range_ok(constraint: str) -> bool:
-    """True iff an openssl spec pins the Driver-18 ABI range: a ``>=3`` lower AND a ``<4`` upper.
-
-    Parses each comma-separated clause as ``(operator, version)`` instead of substring-
-    matching, so a loose spelling like ``>=3,<40`` (whose ``<40`` merely CONTAINS ``<4``)
-    is correctly REJECTED, while conda's canonical alpha upper bound ``<4.0a0`` is accepted.
-    A spec with no upper bound (bare ``>=3``) is rejected -- it would admit openssl 4.
+    """True iff an openssl spec pins the Driver-18 ABI range: a ``>=3`` lower AND an EXCLUSIVE
+    upper that admits NO openssl 4.x.
+
+    Parses each comma-separated clause as ``(operator, version)``. The upper bound is valid
+    ONLY as an exclusive ``<`` whose numeric release is ``4`` with trailing zeros -- ``<4``,
+    ``<4.0``, ``<4.0.0`` or a ``4.0`` pre-release like ``<4.0a0`` (conda's canonical bound).
+    ``<=4``, ``<4.1`` and ``<4.0.1`` each admit some 4.x build and are REJECTED, as is a spec
+    with no upper bound (bare ``>=3``) and a loose ``<40`` (whose ``<40`` is not ``<4``).
     """
     has_lower_3 = False
     has_upper_4 = False
@@ -243,10 +245,14 @@ def _openssl_range_ok(constraint: str) -> bool:
         if not m:
             continue
         op, ver = m.group(1), m.group(2)
-        major = int(ver.split(".")[0])
-        if op in (">=", ">", "==", "=") and major == 3:
+        parts = [int(p) for p in ver.split(".") if p != ""]
+        if not parts:
+            continue
+        if op in (">=", ">", "==", "=") and parts[0] == 3:
             has_lower_3 = True
-        elif op in ("<", "<=") and major == 4:
+        # Exclusive '<' only, at numeric release 4.0(.0...) -- a '4.0' pre-release such as
+        # '<4.0a0' captures as '4.0' here, so it qualifies; '<=4', '<4.1', '<4.0.1' do not.
+        elif op == "<" and parts[0] == 4 and all(p == 0 for p in parts[1:]):
             has_upper_4 = True
     return has_lower_3 and has_upper_4
 
@@ -264,8 +270,16 @@ def audit_package(path: str) -> list[str]:
         print(f"  SKIP (no Linux ELF payload): {base_name} [subdir={subdir or '?'}]")
         return []
 
-    # The conda subdir is the arch authority; every vendored ELF must match it.
+    # The conda subdir is the arch authority; every vendored ELF must match it. A linux
+    # subdir with no e_machine mapping (e.g. a future linux-ppc64le) FAILS CLOSED rather than
+    # silently skipping the architecture gate this audit exists to enforce.
     expected_machine = _SUBDIR_MACHINE.get(subdir)
+    if expected_machine is None:
+        return [
+            f"{base_name}: unrecognized Linux subdir '{subdir}' has no known ELF machine "
+            f"mapping -- add it to _SUBDIR_MACHINE so the arch gate can enforce it "
+            f"(refusing to skip the architecture check)."
+        ]
 
     errors: list[str] = []
 
@@ -297,7 +311,12 @@ def audit_package(path: str) -> list[str]:
     dirs_with_inst: set = set()
     vendored: list[str] = []
 
-    for name, data in _iter_payload_members(path):
+    try:
+        members = list(_iter_payload_members(path))
+    except ValueError as exc:  # malformed payload (e.g. .conda missing pkg-*.tar.zst)
+        return [f"{base_name}: unreadable/malformed package payload ({exc})."]
+
+    for name, data in members:
         base = posixpath.basename(name)
         norm = "/" + name
         member_dir = posixpath.dirname(name)
@@ -323,14 +342,13 @@ def audit_package(path: str) -> list[str]:
         # Architecture gate: the ELF machine MUST match the package's conda subdir, so
         # an x86_64 driver mislabeled under a linux-aarch64 package (which the emulated
         # leg's best-effort runtime probe would not catch) fails here.
-        if expected_machine is not None:
-            mach = elf_machine(data)
-            if mach != expected_machine:
-                errors.append(
-                    f"{name}: ELF machine {mach} ({_MACHINE_NAME.get(mach, 'unknown')}) does "
-                    f"not match the '{subdir}' package arch {expected_machine} "
-                    f"({_MACHINE_NAME[expected_machine]}) -- wrong-arch/mislabeled driver."
-                )
+        mach = elf_machine(data)
+        if mach != expected_machine:
+            errors.append(
+                f"{name}: ELF machine {mach} ({_MACHINE_NAME.get(mach, 'unknown')}) does "
+                f"not match the '{subdir}' package arch {expected_machine} "
+                f"({_MACHINE_NAME[expected_machine]}) -- wrong-arch/mislabeled driver."
+            )
 
         dyn = elf_dynamic(data)
         entries = _entries(effective_runpath(dyn))

tests/test_028_tls_connect_probe.py

@@ -215,6 +215,31 @@ def test_split_top_level_handles_escaped_braces():
     ]
 
 
+def test_split_top_level_inner_brace_is_literal():
+    """A braced value is SINGLE-LEVEL (matches _parse_braced_value): an inner '{' is a literal
+    char, so '{a{b}' ends at the first '}' and ';Encrypt=no' is a separate segment -- not
+    absorbed into Pwd (which would make force_tls emit a duplicate Encrypt)."""
+    probe = _load_probe()
+    assert probe._split_top_level("Pwd={a{b};Encrypt=no") == ["Pwd={a{b}", "Encrypt=no"]
+
+
+def test_split_top_level_mid_value_brace_is_literal():
+    """A '{' that is NOT the first char of a value is literal (the value is not braced), so
+    'Server=a{b' is a simple value and ';' still splits -- matching the production parser."""
+    probe = _load_probe()
+    assert probe._split_top_level("Server=a{b;Encrypt=no") == ["Server=a{b", "Encrypt=no"]
+
+
+def test_force_tls_braced_password_with_inner_brace():
+    """End-to-end for the inner-'{' case: the user's Encrypt=no is dropped and Encrypt=yes
+    appended exactly ONCE -- no duplicate 'encrypt' that mssql_python.connect would reject."""
+    probe = _load_probe()
+    out = probe.force_tls("Pwd={a{b};Encrypt=no").lower()
+    assert "encrypt=yes" in out
+    assert "encrypt=no" not in out
+    assert out.count("encrypt=") == 1
+
+
 def test_redact_masks_values_and_flags_bare_segments():
     """The debug line must never leak a value and must surface a no-value segment."""
     probe = _load_probe()

tests/test_029_bundled_binary_audit.py

@@ -294,6 +294,22 @@ def test_audit_passes_openssl_alpha_upper_bound(tmp_path):
     assert not any("range-pinned" in e for e in errors)
 
 
+def test_audit_fails_openssl_upper_admits_4x(tmp_path):
+    # '<=4', '<4.1', '<4.0.1' each admit some openssl 4.x -> must FAIL (looser than the pin).
+    for spec in ("openssl >=3,<=4", "openssl >=3,<4.1", "openssl >=3,<4.0.1"):
+        depends = ["python", "azure-identity", "krb5", "libtool", spec]
+        errors = audit.audit_package(_make_pkg(tmp_path, depends=depends))
+        assert any("range-pinned" in e for e in errors), spec
+
+
+def test_audit_passes_openssl_exclusive_4_variants(tmp_path):
+    # '<4', '<4.0', '<4.0.0' all exclude every openssl 4.x and are accepted.
+    for spec in ("openssl >=3,<4", "openssl >=3,<4.0", "openssl >=3,<4.0.0"):
+        depends = ["python", "azure-identity", "krb5", "libtool", spec]
+        errors = audit.audit_package(_make_pkg(tmp_path, depends=depends))
+        assert not any("range-pinned" in e for e in errors), spec
+
+
 def test_audit_fails_driver_missing_from_one_subdir(tmp_path):
     # debian_ubuntu is complete, but rhel ships only libodbcinst (driver dropped). A
     # package-global count would pass since debian_ubuntu supplies a driver; per-subdir
@@ -435,3 +451,10 @@ def test_elf_machine_reads_arch():
     assert audit.elf_machine(_make_elf64()) == 62
     assert audit.elf_machine(_make_elf64(machine=183)) == 183
     assert audit.elf_machine(b"not an elf") is None
+
+
+def test_audit_fails_unknown_linux_subdir(tmp_path):
+    # A linux subdir with no e_machine mapping (e.g. a future linux-ppc64le) must FAIL CLOSED,
+    # not silently skip the architecture gate this audit exists to enforce.
+    errors = audit.audit_package(_make_pkg(tmp_path, subdir="linux-ppc64le"))
+    assert any("unrecognized Linux subdir" in e for e in errors)

tests/test_030_pe_machine_assert.py

@@ -220,3 +220,27 @@ def test_non_windows_package_skipped(tmp_path):
         {"lib/python3.12/site-packages/mssql_python/_core.so": b"\x7fELF fake"},
     )
     assert ape.audit_package(p) == []
+
+
+@pytest.mark.skipif(not _zstd_available(), reason="no zstandard backend available")
+def test_win_conda_missing_pkg_payload_fails(tmp_path):
+    # A .conda with info/index.json but NO pkg-*.tar.zst payload must FAIL (not silently pass
+    # with zero members): iter_payload_members raises, and audit_package -> a violation.
+    name = "mssql-python-1.13.0-py312_0"
+    index = {
+        "name": "mssql-python",
+        "version": "1.13.0",
+        "build": "py312_0",
+        "subdir": "win-arm64",
+    }
+    info_buf = io.BytesIO()
+    with tarfile.open(fileobj=info_buf, mode="w") as tf:
+        idx = json.dumps(index).encode()
+        ti = tarfile.TarInfo("info/index.json")
+        ti.size = len(idx)
+        tf.addfile(ti, io.BytesIO(idx))
+    conda_path = tmp_path / f"{name}.conda"
+    with zipfile.ZipFile(conda_path, "w") as zf:
+        zf.writestr(f"info-{name}.tar.zst", _zstd_compress(info_buf.getvalue()))
+    errors = ape.audit_package(str(conda_path))
+    assert any("pkg-" in e for e in errors)