Rely solely on anaconda_auth for a/ token (#189)

* require anaconda_auth for a/ token

* cleanup

Author: mcg1969

anaconda_anon_usage/tokens.py

@@ -5,6 +5,7 @@
 # child environments.
 
 import base64
+import datetime as dt
 import json
 import re
 import sys
@@ -208,7 +209,7 @@ def _jwt_to_token(s):
       exp: the expiration time
     """
     if not s:
-        return None, 0
+        return
     try:
         # The JWT should have three parts separated by periods
         parts = s.split(".")
@@ -223,64 +224,44 @@ def _jwt_to_token(s):
         # The payload should have a positive integer expiration
         exp = parts[1].get("exp")
         assert isinstance(exp, int) and exp > 0, "Invalid expiration"
+        now = dt.datetime.now(tz=dt.timezone.utc).timestamp()
+        if exp < now:
+            _debug("API key expired %ds ago", int(now - exp))
+            return
         # The subscriber should be a non-empty UUID string
         sub = parts[1].get("sub")
         assert sub, "Invalid subscriber"
         # This is an Anaconda requirement, not a JWT requirement
         sub = uuid.UUID(sub).bytes
         token = base64.urlsafe_b64encode(sub).decode("ascii").strip("=")
-        return token, exp
+        return token
     except Exception as exc:
         _debug("Unexpected %s parsing API key: %s", type(exc), exc)
-        return None, 0
 
 
 @cached
 def anaconda_auth_token():
-    """Retrieve Anaconda Cloud token from keyring.
-
-    Examines all entries under 'Anaconda Cloud' in the keyring and
-    selects the token with the latest expiration date. This handles
-    migration from 'anaconda.cloud' to 'anaconda.com' entries.
-
+    """Returns the base64-encoded uid corresponding to the logged
+    in Anaconda Cloud user, if one is present.
     Returns:
         str: Base64-encoded token, or None if no valid token found.
     """
-    env = environ.get("ANACONDA_AUTH_API_KEY")
-    if env:
-        _debug("ANACONDA_AUTH_API_KEY environment variable found")
-        token, _ = _jwt_to_token(env)
-        if token:
-            _debug("Retrieved Anaconda token from environment: %s", token)
-            return token
-        _debug("Could not retrieve API key from environment")
-    fpath = expanduser(join(ANACONDA_DIR, "keyring"))
-    data = _read_file(fpath, "anaconda keyring")
-    if not data:
-        return
     try:
-        data = json.loads(data)
+        from anaconda_auth.token import TokenInfo, TokenNotFoundError
+
+        _debug("Module anaconda_auth loaded")
+        tinfo = TokenInfo.load(domain="anaconda.com")
+        if tinfo.api_key:
+            token = _jwt_to_token(tinfo.api_key)
+            _debug("Retrieved Anaconda auth token: %s", token)
+            return token
+    except ImportError:
+        _debug("Module anaconda_auth not available")
+    except TokenNotFoundError:
+        pass
     except Exception as exc:
-        _debug("Unexpected JSON decoding error parsing keyring file: %s", exc)
-        return
-    if not data or not isinstance(data, dict):
-        _debug("Empty keyring")
-        return
-    token, exp = None, 0
-    for key, rec in (data.get("Anaconda Cloud") or {}).items():
-        try:
-            tdata = json.loads(base64.b64decode(rec))["api_key"]
-        except Exception as exc:
-            _debug("Unexpected error parsing keyring entry '%s': %s", key, exc)
-        t_token, t_exp = _jwt_to_token(tdata)
-        if t_exp > exp:
-            token = t_token
-            exp = t_exp
-    if token:
-        _debug("Retrieved Anaconda token from keyring: %s", token)
-    else:
-        _debug("No Anaconda keyring records found")
-    return token
+        _debug("Unexpected error retrieving token using anaconda_auth: %s", exc)
+    _debug("No Anaconda API token found")
 
 
 @cached

tests/conftest.py

@@ -1,8 +1,9 @@
 import base64
+import datetime as dt
 import json
 import tempfile
 import uuid
-from os import environ, mkdir
+from os import mkdir
 from os.path import dirname, join
 
 import pytest
@@ -11,57 +12,57 @@
 
 from anaconda_anon_usage import tokens, utils
 
+try:
+    from anaconda_auth.token import AnacondaKeyring
+except ImportError:
+    AnacondaKeyring = None
 
-def _jsond(rec, urlsafe=False):
-    return (
-        base64.urlsafe_b64encode(json.dumps(rec).encode("ascii"))
-        .decode("ascii")
-        .rstrip("=")
-    )
 
+def _jsond(rec, strip=True):
+    result = json.dumps(rec)
+    result = base64.urlsafe_b64encode(result.encode("ascii"))
+    result = result.decode("ascii")
+    if strip:
+        result = result.rstrip("=")
+    return result
 
-def _test_keyring():
+
+def _keyring_data():
     domains = ["random.domain", "anaconda.cloud", "anaconda.com"]
     drecs, exp = {}, 0
+    exp = int(dt.datetime.now(tz=dt.timezone.utc).timestamp()) + 7884000
     for dom in domains:
-        exp = exp + 123456
+        exp = exp - 1
         sub = str(uuid.uuid4())
         header = {"alg": "RS256", "typ": "JWT"}
         payload = {"exp": exp, "sub": sub}
         # Not a real signature but we just need it to be a base64-encoded blob
         signature = {"fake": dom}
         api_key = ".".join(map(_jsond, (header, payload, signature)))
         rec = {"domain": dom, "api_key": api_key, "repo_tokens": [], "version": 2}
-        drecs[dom] = _jsond(rec)
+        # anaconda_auth doesn't like it when we strip the padding
+        drecs[dom] = _jsond(rec, strip=False)
     result = {"Anaconda Cloud": drecs}
     return json.dumps(result), sub, api_key
 
 
 @pytest.fixture
-def aau_token_path():
-    old_dir, old_adir = tokens.CONFIG_DIR, tokens.ANACONDA_DIR
-    with tempfile.TemporaryDirectory() as tname:
-        tokens.CONFIG_DIR = tokens.ANACONDA_DIR = tname
-        yield join(tname, "aau_token")
-    tokens.CONFIG_DIR = old_dir
-    tokens.ANACONDA_DIR = old_adir
-
-
-@pytest.fixture()
-def anaconda_uid(aau_token_path):
-    kpath = join(dirname(aau_token_path), "keyring")
-    kstr, sub, _ = _test_keyring()
-    with open(kpath, "w") as fp:
+def api_key_sub(monkeypatch, aau_token_path):
+    kstr, sub, _ = _keyring_data()
+    keyring_path = join(dirname(aau_token_path), "keyring")
+    with open(keyring_path, "w") as fp:
         fp.write(kstr)
-    yield sub
+    return sub
 
 
-@pytest.fixture()
-def anaconda_uid_env():
-    _, sub, api_key = _test_keyring()
-    environ["ANACONDA_AUTH_API_KEY"] = api_key
-    yield sub
-    del environ["ANACONDA_AUTH_API_KEY"]
+@pytest.fixture
+def aau_token_path(monkeypatch, tmp_path):
+    monkeypatch.setattr(tokens, "CONFIG_DIR", str(tmp_path))
+    monkeypatch.setattr(tokens, "ANACONDA_DIR", str(tmp_path))
+    keyring_path = tmp_path / "keyring"
+    if AnacondaKeyring is not None:
+        monkeypatch.setattr(AnacondaKeyring, "keyring_path", keyring_path)
+    return str(tmp_path / "aau_token")
 
 
 def _system_token_path(npaths=1):
@@ -126,22 +127,9 @@ def two_dotted_org_tokens(aau_token_path):
         yield t1 + t2[:1]
 
 
-def _env_clear():
-    if "ANACONDA_AUTH_API_KEY" in environ:
-        del environ["ANACONDA_AUTH_API_KEY"]
-    if "ANACONDA_ANON_USAGE_ORG_TOKEN" in environ:
-        del environ["ANACONDA_ANON_USAGE_ORG_TOKEN"]
-    if "ANACONDA_ANON_USAGE_MACHINE_TOKEN" in environ:
-        del environ["ANACONDA_ANON_USAGE_MACHINE_TOKEN"]
-    if "ANACONDA_ANON_USAGE_INSTALLER_TOKEN" in environ:
-        del environ["ANACONDA_ANON_USAGE_INSTALLER_TOKEN"]
-
-
 @pytest.fixture(autouse=True)
 def client_token_string_cache_cleanup(request):
-    _env_clear()
     request.addfinalizer(utils._cache_clear)
-    request.addfinalizer(_env_clear)
 
 
 @pytest.fixture(autouse=True)

tests/unit/test_tokens.py

@@ -1,11 +1,17 @@
 import base64
 import re
 import uuid
-from os import environ
 from os.path import exists
 
+import pytest
+
 from anaconda_anon_usage import tokens, utils
 
+try:
+    import anaconda_auth
+except ImportError:
+    anaconda_auth = None
+
 
 def test_client_token(aau_token_path):
     assert not exists(aau_token_path)
@@ -150,26 +156,26 @@ def test_token_string_with_two_dotted_org_tokens(two_dotted_org_tokens):
     assert token_string.count(" o/") == 2
 
 
-def test_token_string_with_env_org_token(no_system_tokens):
+def test_token_string_with_env_org_token(monkeypatch, no_system_tokens):
     org_token_e = utils._random_token()
     mch_token_e = utils._random_token()
     ins_token_e = utils._random_token()
-    environ["ANACONDA_ANON_USAGE_ORG_TOKEN"] = org_token_e
-    environ["ANACONDA_ANON_USAGE_MACHINE_TOKEN"] = mch_token_e
-    environ["ANACONDA_ANON_USAGE_INSTALLER_TOKEN"] = ins_token_e
+    monkeypatch.setenv("ANACONDA_ANON_USAGE_ORG_TOKEN", org_token_e)
+    monkeypatch.setenv("ANACONDA_ANON_USAGE_MACHINE_TOKEN", mch_token_e)
+    monkeypatch.setenv("ANACONDA_ANON_USAGE_INSTALLER_TOKEN", ins_token_e)
     token_string = tokens.token_string()
     assert "o/" + org_token_e in token_string
     assert "m/" + mch_token_e in token_string
 
 
-def test_token_string_with_system_and_env(system_tokens):
+def test_token_string_with_system_and_env(monkeypatch, system_tokens):
     org_token, mch_token, ins_token = system_tokens
     org_token_e = utils._random_token()
     mch_token_e = utils._random_token()
     ins_token_e = utils._random_token()
-    environ["ANACONDA_ANON_USAGE_ORG_TOKEN"] = org_token_e
-    environ["ANACONDA_ANON_USAGE_MACHINE_TOKEN"] = mch_token_e
-    environ["ANACONDA_ANON_USAGE_INSTALLER_TOKEN"] = ins_token_e
+    monkeypatch.setenv("ANACONDA_ANON_USAGE_ORG_TOKEN", org_token_e)
+    monkeypatch.setenv("ANACONDA_ANON_USAGE_MACHINE_TOKEN", mch_token_e)
+    monkeypatch.setenv("ANACONDA_ANON_USAGE_INSTALLER_TOKEN", ins_token_e)
     token_string = tokens.token_string()
     assert "i/" + ins_token in token_string
     assert "i/" + ins_token_e in token_string
@@ -183,13 +189,13 @@ def test_token_string_with_system_and_env(system_tokens):
     assert token_string.count(" i/") == 2
 
 
-def test_token_string_with_invalid_tokens(no_system_tokens):
+def test_token_string_with_invalid_tokens(monkeypatch, no_system_tokens):
     org_token_e = "invalid token"
     mch_token_e = "superlongtokenthathasnobusinessbeinganactualtoken"
     ins_token_e = "fake installer"
-    environ["ANACONDA_ANON_USAGE_ORG_TOKEN"] = org_token_e
-    environ["ANACONDA_ANON_USAGE_MACHINE_TOKEN"] = mch_token_e
-    environ["ANACONDA_ANON_USAGE_MACHINE_TOKEN"] = ins_token_e
+    monkeypatch.setenv("ANACONDA_ANON_USAGE_ORG_TOKEN", org_token_e)
+    monkeypatch.setenv("ANACONDA_ANON_USAGE_MACHINE_TOKEN", mch_token_e)
+    monkeypatch.setenv("ANACONDA_ANON_USAGE_INSTALLER_TOKEN", ins_token_e)
     token_string = tokens.token_string()
     assert "o/" not in token_string
     assert "m/" not in token_string
@@ -249,19 +255,11 @@ def test_token_string_env_readonly(monkeypatch, no_system_tokens):
     assert "m/" not in token_string
 
 
-def test_anaconda_string_keyring(anaconda_uid):
-    token_string = tokens.token_string()
-    assert "a/" in token_string
-    expected = uuid.UUID(anaconda_uid).bytes
-    expected = base64.urlsafe_b64encode(expected).decode("ascii").rstrip("=")
-    aval = re.sub("^.*a/", "", token_string).split(" ", 1)[0]
-    assert aval == expected
-
-
-def test_anaconda_string_env(anaconda_uid_env):
+@pytest.mark.skipif(anaconda_auth is None, reason="Requires the anaconda_auth module")
+def test_keyring_in_module(api_key_sub):
     token_string = tokens.token_string()
     assert "a/" in token_string
-    expected = uuid.UUID(anaconda_uid_env).bytes
+    expected = uuid.UUID(api_key_sub).bytes
     expected = base64.urlsafe_b64encode(expected).decode("ascii").rstrip("=")
     aval = re.sub("^.*a/", "", token_string).split(" ", 1)[0]
     assert aval == expected