Handle mamba-style channel token masking

conda_lock/conda_solver.py

@@ -177,9 +177,13 @@ def normalize_url(url: str) -> str:
             if url.startswith(candidate1):
                 url = url.replace(candidate1, channel.url, 1)
 
-            candidate2 = channel.env_replaced_url()
+            candidate2 = channel.mamba_token_replaced_url()
             if url.startswith(candidate2):
                 url = url.replace(candidate2, channel.url, 1)
+
+            candidate3 = channel.env_replaced_url()
+            if url.startswith(candidate3):
+                url = url.replace(candidate3, channel.url, 1)
         return url
 
     # extract dependencies from package plan

conda_lock/interfaces/vendored_conda.py

@@ -1,5 +1,9 @@
 from conda_lock._vendor.conda.common.toposort import toposort
+from conda_lock._vendor.conda.common.url import (
+    mask_anaconda_token,
+    split_anaconda_token,
+)
 from conda_lock._vendor.conda.models.match_spec import MatchSpec
 
 
-__all__ = ["toposort", "MatchSpec"]
+__all__ = ["toposort", "MatchSpec", "mask_anaconda_token", "split_anaconda_token"]

conda_lock/invoke_conda.py

@@ -8,6 +8,7 @@
 import tempfile
 import threading
 
+from logging import getLogger
 from typing import IO, Dict, Iterator, List, Optional, Sequence, Union
 
 from ensureconda.api import determine_micromamba_version, ensureconda
@@ -16,6 +17,8 @@
 from conda_lock.models.channel import Channel
 
 
+logger = getLogger(__name__)
+
 PathLike = Union[str, pathlib.Path]
 
 CONDA_PKGS_DIRS: Optional[str] = None
@@ -106,6 +109,7 @@ def _invoke_conda(
         common_args.extend(shlex.split(conda_flags))
 
     cmd = [str(arg) for arg in [conda, *command_args, *common_args, *post_args]]
+    logger.debug(f"Invoking command: {shlex.join(cmd)}")
 
     with subprocess.Popen(
         cmd,

conda_lock/models/channel.py

@@ -41,6 +41,11 @@
 
 from pydantic import BaseModel, ConfigDict, Field
 
+from conda_lock._vendor.conda.common.url import (
+    mask_anaconda_token,
+    split_anaconda_token,
+)
+
 
 logger = logging.getLogger(__name__)
 token_pattern = re.compile(r"(.*)(/t/\$?\{?[a-zA-Z0-9-_]*\}?)(/.*)")
@@ -84,12 +89,13 @@ def conda_token_replaced_url(self) -> str:
         """Handle conda's token replacement in the output URL."""
         # TODO: pass in env vars maybe?
         expanded_url = expandvars(self.url)
-        if token_pattern.match(expanded_url):
-            replaced = token_pattern.sub(r"\1\3", expanded_url, 1)
-            p = urlparse(replaced)
-            replaced = urlunparse(p._replace(path="/t/<TOKEN>" + p.path))
-            return replaced
-        return expanded_url
+        return mask_anaconda_token(expanded_url)
+
+    def mamba_token_replaced_url(self) -> str:
+        """Handle mamba's token replacement in the output URL."""
+        expanded_url = expandvars(self.url)
+        _, token = split_anaconda_token(expanded_url)
+        return expanded_url.replace(token, "*****", 1) if token else expanded_url
 
 
 def _detect_used_env_var(

tests/test_conda_lock.py

@@ -2666,10 +2666,8 @@ def test_private_lock(
             [conda_exe, "--version"], stdout=subprocess.PIPE, encoding="utf8"
         )
         logging.info("using micromamba version %s", res.stdout)
-        pytest.xfail("micromamba doesn't support our quetz server urls properly")
     from ensureconda.resolve import platform_subdir
 
-    monkeypatch.setenv("QUETZ_API_KEY", quetz_server.api_key)
     monkeypatch.chdir(tmp_path)
 
     content = yaml.safe_dump(
@@ -2693,15 +2691,21 @@ def test_private_lock(
                 conda_exe,
             ],
             catch_exceptions=False,
+            env=dict(os.environ, QUETZ_API_KEY=quetz_server.api_key),
         )
         assert result.exit_code == 0
 
-    def run_install():
+    def run_install(with_env: bool) -> Result:
         with capsys.disabled():
             runner = CliRunner(mix_stderr=False)
             env_name = uuid.uuid4().hex
             env_prefix = tmp_path / env_name
 
+            if with_env:
+                env = dict(os.environ, QUETZ_API_KEY=quetz_server.api_key)
+            else:
+                env = dict(os.environ)
+
             result: Result = runner.invoke(
                 main,
                 [
@@ -2712,18 +2716,19 @@ def run_install():
                     str(env_prefix),
                     str(tmp_path / "conda-lock.yml"),
                 ],
-                catch_exceptions=True,
+                catch_exceptions=False,
+                env=env,
             )
 
         print(result.stdout, file=sys.stdout)
         print(result.stderr, file=sys.stderr)
-        assert result.exit_code == 0
+        return result
 
-    run_install()
+    result = run_install(with_env=True)
+    assert result.exit_code == 0
 
-    monkeypatch.delenv("QUETZ_API_KEY")
     with pytest.raises(MissingEnvVarError):
-        run_install()
+        run_install(with_env=False)
 
 
 def test_lookup_sources():