Merge pull request #46 from anaconda/redact-info

RFC: redact user-agent aau tokens in conda info

anaconda_anon_usage/patch.py

@@ -2,6 +2,7 @@
 # needed to deploy the additional anonmyous user data. It pulls
 # the token management functions themselves from the api module.
 
+import re
 import sys
 
 from conda.auxlib.decorators import memoizedproperty
@@ -30,17 +31,39 @@ def _new_check_prefix(prefix, json=False):
     Context._old_check_prefix(prefix, json)
 
 
+def _new_get_main_info_str(info_dict):
+    ua = info_dict["user_agent"]
+    if " aau/" in ua:
+        ua_redact = re.sub(r" ([a-z]/)([^ ]+)", r" \1.", ua)
+        info_dict["user_agent"] = ua_redact
+    return Context._old_get_main_info_str(info_dict)
+
+
 def _patch_check_prefix():
+    if hasattr(Context, "_old_check_prefix"):
+        return
     _debug("Applying anaconda_anon_usage cli.install patch")
+
     from conda.cli import install as cli_install
 
     Context._old_check_prefix = cli_install.check_prefix
     cli_install.check_prefix = _new_check_prefix
     context._aau_initialized = True
 
 
+def _patch_conda_info():
+    if hasattr(Context, "_old_get_main_info_str"):
+        return
+    _debug("Applying anaconda_anon_usage conda info patch")
+
+    from conda.cli import main_info
+
+    Context._old_get_main_info_str = main_info.get_main_info_str
+    main_info.get_main_info_str = _new_get_main_info_str
+
+
 def main(plugin=False):
-    if hasattr(context, "_aau_initialized"):
+    if getattr(context, "_aau_initialized", None) is not None:
         _debug("anaconda_anon_usage already active")
         return False
     _debug("Applying anaconda_anon_usage context patch")
@@ -69,6 +92,7 @@ def main(plugin=False):
     if plugin:
         # The pre-command plugin avoids the circular import
         # of conda.cli.install, so we can apply the patch now
+        _patch_conda_info()
         _patch_check_prefix()
     else:
         # We need to delay further. Schedule the patch for the
@@ -77,6 +101,7 @@ def main(plugin=False):
         _old__init__ = context.__init__
 
         def _new_init(*args, **kwargs):
+            _patch_conda_info()
             _patch_check_prefix()
             context.__init__ = _old__init__
             _old__init__(*args, **kwargs)

tests/conftest.py

@@ -2,7 +2,7 @@
 from os.path import join
 
 import pytest
-from conda.base.context import Context
+from conda.base.context import Context, context
 
 from anaconda_anon_usage import tokens, utils
 
@@ -32,28 +32,29 @@ def client_token_string_cache_cleanup(request):
 def reset_patch(request):
     def _resetter():
         from conda.cli import install as cli_install
-
-        for container, name, old_value in (
-            (Context, "user_agent", Context._old_user_agent),
-            (Context, "_aau_initialized", None),
-            (Context, "anaconda_anon_usage", None),
-            (
-                Context,
-                "parameter_names",
-                tuple(
-                    filter(
-                        lambda name: name != "anaconda_anon_usage",
-                        Context.parameter_names,
-                    )
-                ),
-            ),
-            (Context, "checked_prefix", None),
-            (cli_install, "check_prefix", getattr(Context, "_old_check_prefix", None)),
-        ):
-            if hasattr(container, name):
-                delattr(container, name)
-            if old_value is None:
-                continue
-            setattr(container, name, old_value)
+        from conda.cli import main_info
+
+        for k in ("___new_user_agent", "__user_agent", "anaconda_anon_usage"):
+            context._cache_.pop(k, None)
+        context._aau_initialized = None
+        if hasattr(Context, "anaconda_anon_usage"):
+            delattr(Context, "anaconda_anon_usage")
+        if hasattr(Context, "checked_prefix"):
+            delattr(Context, "checked_prefix")
+        Context.parameter_names = tuple(
+            k for k in Context.parameter_names if k != "anaconda_anon_usage"
+        )
+        orig_check_prefix = getattr(Context, "_old_check_prefix", None)
+        if orig_check_prefix is not None:
+            cli_install.check_prefix = orig_check_prefix
+            delattr(Context, "_old_check_prefix")
+        orig_user_agent = getattr(Context, "_old_user_agent", None)
+        if orig_user_agent is not None:
+            Context.user_agent = orig_user_agent
+            delattr(Context, "_old_user_agent")
+        orig_get_main_info_str = getattr(main_info, "_old_get_main_info_str", None)
+        if orig_get_main_info_str is not None:
+            main_info.get_main_info_str = orig_get_main_info_str
+            delattr(Context, "_old_get_main_info_str")
 
     request.addfinalizer(_resetter)

tests/unit/test_patch.py

@@ -1,4 +1,4 @@
-from conda.base.context import Context, context
+from conda.base.context import context
 
 from anaconda_anon_usage import patch
 
@@ -18,6 +18,26 @@ def test_user_agent_no_token(monkeypatch):
 
 
 def test_main_already_patched():
-    Context._aau_initialized = True
+    response = patch.main(plugin=True)
+    assert response
     response = patch.main(plugin=True)
     assert not response
+
+
+def test_main_info():
+    patch.main(plugin=True)
+    tokens = dict(t.split("/", 1) for t in context.user_agent.split(" "))
+    tokens["c"] = tokens["e"] = tokens["s"] = "."
+    from conda.cli import main_info
+
+    info_dict = main_info.get_info_dict(False)
+    assert info_dict["user_agent"] == context.user_agent
+    info_str = main_info.get_main_info_str(info_dict)
+    ua_strs = [
+        x.strip().split(" : ", 1)[-1]
+        for x in info_str.splitlines()
+        if x.lstrip().startswith("user-agent : ")
+    ]
+    assert ua_strs
+    token2 = dict(t.split("/", 1) for t in ua_strs[0].split(" "))
+    assert token2 == tokens