Instrument databricks auth for UCODE_DEBUG diagnostics (#80)

* Instrument databricks auth for UCODE_DEBUG diagnostics

Adds opt-in debug logging to diagnose cases where `databricks auth token`
returns no token even though the CLI itself is authenticated. When
`UCODE_DEBUG=1`, `has_valid_databricks_auth` and `get_databricks_token`
now dump (once per process) the CLI version, the scrubbed profiles list,
the scrubbed `~/.databrickscfg`, and the set of `DATABRICKS_*` env vars
visible to the subprocess, plus the rc/stderr of each `auth token` call.
A `_format_subprocess_result` helper suppresses stdout on success so the
access token never lands in `~/.ucode/debug.log`.

Also removes the dead `databricks auth login --no-browser` retry block
in `get_databricks_token`: with `capture_output=True` the device-code URL
is never visible to the user, so the retry can never complete a fresh
auth — it only obscures the original failure.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* Restore get_databricks_token retry and add debug coverage for reauth path

The previous commit removed the silent re-auth retry on the grounds that
--no-browser can never complete a fresh auth. That's true, but the retry
also handles OAuth token refresh for sessions that just need a silent
refresh — which the e2e TestGeminiAuthRecovery test exercises.

Restores the _fetch / auth login --no-browser / _fetch retry loop, adds
debug logging around the reauth attempt (triggered/rc/stderr), and
restores the unit test that covers the retry path.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* Update uv.lock registry source to pypi-proxy.cloud.databricks.com

Reflects the current internal PyPI proxy (go/pypi-registry-access).
Download URLs remain files.pythonhosted.org so external users are unaffected.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: rohita5l

src/ucode/databricks.py

@@ -3,6 +3,7 @@
 
 from __future__ import annotations
 
+import functools
 import json
 import logging
 import logging.handlers
@@ -12,6 +13,7 @@
 import shlex
 import shutil
 import subprocess
+from pathlib import Path
 from typing import cast
 from urllib import error as urllib_error
 from urllib import request as urllib_request
@@ -88,6 +90,105 @@ def _debug(label: str, detail: str) -> None:
         logger.debug("%s: %s", label, detail)
 
 
+_SECRET_KEY_PATTERN = re.compile(
+    r"(token|secret|password|bearer|api_key|apikey)", re.IGNORECASE
+)
+
+
+def _format_subprocess_result(
+    result: subprocess.CompletedProcess[str],
+) -> str:
+    """Format a CompletedProcess for the debug log without leaking tokens.
+
+    On success, stdout is suppressed (it often contains the access token).
+    On failure, stdout/stderr are included truncated."""
+    stderr = (result.stderr or "").strip()[:500]
+    if result.returncode == 0:
+        return f"rc=0 stderr={stderr!r}"
+    stdout = (result.stdout or "").strip()[:500]
+    return f"rc={result.returncode} stdout={stdout!r} stderr={stderr!r}"
+
+
+def _scrub_databrickscfg(text: str) -> str:
+    """Redact value of any INI key that looks secret-bearing."""
+    out: list[str] = []
+    for line in text.splitlines():
+        stripped = line.lstrip()
+        if "=" in stripped and not stripped.startswith(("#", ";")):
+            key = stripped.split("=", 1)[0].strip()
+            if _SECRET_KEY_PATTERN.search(key):
+                indent = line[: len(line) - len(stripped)]
+                out.append(f"{indent}{key} = <redacted>")
+                continue
+        out.append(line)
+    return "\n".join(out)
+
+
+def _scrub_json(value: object) -> object:
+    if isinstance(value, dict):
+        return {
+            k: ("<redacted>" if _SECRET_KEY_PATTERN.search(k) else _scrub_json(v))
+            for k, v in value.items()
+        }
+    if isinstance(value, list):
+        return [_scrub_json(v) for v in value]
+    return value
+
+
+@functools.cache
+def _log_auth_diagnostics() -> None:
+    """Dump CLI version, profiles, and ~/.databrickscfg (scrubbed) to the debug log.
+
+    No-op unless UCODE_DEBUG=1; cached so it runs at most once per process."""
+    if not _debug_enabled():
+        return
+
+    try:
+        version_result = subprocess.run(
+            ["databricks", "--version"],
+            check=False,
+            capture_output=True,
+            text=True,
+            timeout=10,
+        )
+        version = (version_result.stdout or version_result.stderr or "").strip()
+        _debug("databricks --version", version[:200])
+    except (OSError, subprocess.TimeoutExpired) as exc:
+        _debug("databricks --version", f"exception: {type(exc).__name__}: {exc}")
+
+    try:
+        profiles_result = subprocess.run(
+            ["databricks", "auth", "profiles", "--output", "json"],
+            check=False,
+            capture_output=True,
+            text=True,
+            timeout=10,
+        )
+        _debug(
+            "databricks auth profiles",
+            f"rc={profiles_result.returncode} "
+            f"stderr={(profiles_result.stderr or '').strip()[:300]!r}",
+        )
+        if profiles_result.returncode == 0 and profiles_result.stdout:
+            try:
+                payload = json.loads(profiles_result.stdout)
+                _debug("profiles json", json.dumps(_scrub_json(payload))[:2000])
+            except json.JSONDecodeError as exc:
+                _debug("profiles json", f"decode error: {exc}")
+    except (OSError, subprocess.TimeoutExpired) as exc:
+        _debug("databricks auth profiles", f"exception: {type(exc).__name__}: {exc}")
+
+    cfg_path = Path(os.environ.get("DATABRICKS_CONFIG_FILE") or "~/.databrickscfg").expanduser()
+    try:
+        if cfg_path.is_file():
+            raw = cfg_path.read_text(encoding="utf-8", errors="replace")
+            _debug(f"databrickscfg ({cfg_path})", _scrub_databrickscfg(raw)[:4000])
+        else:
+            _debug(f"databrickscfg ({cfg_path})", "not present")
+    except OSError as exc:
+        _debug(f"databrickscfg ({cfg_path})", f"read error: {exc}")
+
+
 def _http_get_json(
     url: str, token: str, *, timeout: int = 10
 ) -> tuple[dict | list | None, str | None]:
@@ -230,6 +331,7 @@ def install_databricks_cli() -> None:
 
 
 def has_valid_databricks_auth(workspace: str) -> bool:
+    _log_auth_diagnostics()
     try:
         env = build_databricks_cli_env(workspace)
         result = run(
@@ -240,11 +342,16 @@ def has_valid_databricks_auth(workspace: str) -> bool:
             env=env,
             timeout=15,
         )
+        _debug(
+            "has_valid_databricks_auth",
+            _format_subprocess_result(result),
+        )
         if result.returncode != 0:
             return False
         data = json.loads(result.stdout or "{}")
         return bool(data.get("access_token"))
-    except (json.JSONDecodeError, OSError, subprocess.TimeoutExpired):
+    except (json.JSONDecodeError, OSError, subprocess.TimeoutExpired) as exc:
+        _debug("has_valid_databricks_auth", f"exception: {type(exc).__name__}: {exc}")
         return False
 
 
@@ -312,36 +419,54 @@ def ensure_databricks_auth(workspace: str) -> None:
 
 
 def get_databricks_token(workspace: str, *, force_refresh: bool = False) -> str:
+    _log_auth_diagnostics()
     env = build_databricks_cli_env(workspace)
     cmd = ["databricks", "auth", "token", "--host", workspace, "--output", "json"]
     if force_refresh:
         cmd.append("--force-refresh")
 
+    _debug(
+        "get_databricks_token.env",
+        "set="
+        + ",".join(
+            sorted(
+                k for k in env if k.startswith("DATABRICKS_") or k in {"BUNDLE_PROFILE"}
+            )
+        ),
+    )
+
     def _fetch() -> str:
         try:
             result = run(
                 cmd,
+                check=False,
                 capture_output=True,
                 text=True,
                 env=env,
                 timeout=15,
             )
-            return json.loads(result.stdout or "{}").get("access_token", "")
-        except (subprocess.CalledProcessError, subprocess.TimeoutExpired, json.JSONDecodeError):
-            return ""
+            _debug("auth token", _format_subprocess_result(result))
+            if result.returncode == 0:
+                return json.loads(result.stdout or "{}").get("access_token", "")
+        except (subprocess.TimeoutExpired, json.JSONDecodeError) as exc:
+            _debug("auth token", f"exception: {type(exc).__name__}: {exc}")
+        return ""
 
     token = _fetch()
     if not token:
         # Session may have expired — attempt non-interactive re-auth and retry once.
+        _debug("auth token", "empty on first fetch; attempting auth login --no-browser")
         try:
-            run(
+            reauth = run(
                 ["databricks", "auth", "login", "--host", workspace, "--no-browser"],
                 capture_output=True,
+                text=True,
                 env=env,
                 timeout=30,
             )
-        except (subprocess.CalledProcessError, subprocess.TimeoutExpired):
-            pass
+            _debug("auth login", _format_subprocess_result(reauth))
+        except (subprocess.CalledProcessError, subprocess.TimeoutExpired) as exc:
+            _debug("auth login", f"exception: {type(exc).__name__}: {exc}")
         token = _fetch()
 
     if not token:

tests/test_databricks.py

@@ -11,7 +11,10 @@
 import ucode.databricks as db_mod
 from ucode.databricks import (
     AI_GATEWAY_V2_DOCS_URL,
+    _format_subprocess_result,
     _parse_databricks_cli_version,
+    _scrub_databrickscfg,
+    _scrub_json,
     build_auth_shell_command,
     build_databricks_cli_env,
     build_opencode_base_urls,
@@ -146,6 +149,102 @@ def test_prefers_databricks_bearer(self, tmp_path):
         assert result.stdout.strip() == "bearer-token"
 
 
+class TestFormatSubprocessResult:
+    def test_suppresses_stdout_on_success(self):
+        result = subprocess.CompletedProcess(
+            args=["databricks", "auth", "token"],
+            returncode=0,
+            stdout='{"access_token": "dapi-secret-do-not-leak", "token_type": "Bearer"}',
+            stderr="",
+        )
+        formatted = _format_subprocess_result(result)
+        assert "dapi-secret-do-not-leak" not in formatted
+        assert "rc=0" in formatted
+
+    def test_includes_stdout_on_failure(self):
+        result = subprocess.CompletedProcess(
+            args=["databricks", "auth", "token"],
+            returncode=1,
+            stdout="useful diagnostic output",
+            stderr="error: no matching profile",
+        )
+        formatted = _format_subprocess_result(result)
+        assert "rc=1" in formatted
+        assert "useful diagnostic output" in formatted
+        assert "no matching profile" in formatted
+
+
+class TestScrubDatabrickscfg:
+    def test_redacts_token_value(self):
+        text = "[DEFAULT]\nhost = https://example.databricks.com\ntoken = dapi-secret\n"
+        scrubbed = _scrub_databrickscfg(text)
+        assert "dapi-secret" not in scrubbed
+        assert "token = <redacted>" in scrubbed
+        assert "host = https://example.databricks.com" in scrubbed
+
+    def test_redacts_various_secret_keys(self):
+        text = (
+            "[p]\n"
+            "client_secret = secret-val-1\n"
+            "bearer_token = secret-val-2\n"
+            "api_key = secret-val-3\n"
+            "password = secret-val-4\n"
+            "auth_type = oauth-u2m\n"
+        )
+        scrubbed = _scrub_databrickscfg(text)
+        for secret in ("secret-val-1", "secret-val-2", "secret-val-3", "secret-val-4"):
+            assert secret not in scrubbed
+        assert "auth_type = oauth-u2m" in scrubbed
+
+    def test_preserves_comments_and_sections(self):
+        text = "# comment\n[DEFAULT]\nhost = https://x\n; another comment with token = leak\n"
+        scrubbed = _scrub_databrickscfg(text)
+        assert "# comment" in scrubbed
+        assert "[DEFAULT]" in scrubbed
+        assert "; another comment with token = leak" in scrubbed
+
+    def test_key_matching_is_case_insensitive(self):
+        text = "[p]\nTOKEN = upper\nAccess_Token = mixed\n"
+        scrubbed = _scrub_databrickscfg(text)
+        assert "upper" not in scrubbed
+        assert "mixed" not in scrubbed
+
+
+class TestScrubJson:
+    def test_redacts_secret_keys(self):
+        payload = {
+            "access_token": "dapi-secret",
+            "host": "https://example.databricks.com",
+        }
+        scrubbed = _scrub_json(payload)
+        assert isinstance(scrubbed, dict)
+        assert scrubbed["access_token"] == "<redacted>"
+        assert scrubbed["host"] == "https://example.databricks.com"
+
+    def test_recurses_into_nested_structures(self):
+        payload = {
+            "profiles": [
+                {"name": "DEFAULT", "client_secret": "abc"},
+                {"name": "other", "password": "pw"},
+            ]
+        }
+        scrubbed = _scrub_json(payload)
+        assert scrubbed == {
+            "profiles": [
+                {"name": "DEFAULT", "client_secret": "<redacted>"},
+                {"name": "other", "password": "<redacted>"},
+            ]
+        }
+
+    def test_passes_through_scalars_and_non_secret_keys(self):
+        assert _scrub_json("plain") == "plain"
+        assert _scrub_json(42) == 42
+        assert _scrub_json({"host": "x", "auth_type": "pat"}) == {
+            "host": "x",
+            "auth_type": "pat",
+        }
+
+
 class TestGetDatabricksToken:
     def _fake_databricks(self, tmp_path, script: str) -> dict:
         fake = tmp_path / "databricks"