feat: Remove JWT parsing and adjust tests

Author: delilahw

src/artifacts-helper/codespaces_artifacts_helper_keyring/src/codespaces_artifacts_helper_keyring/artifacts_helper_credential_provider.py

@@ -5,22 +5,12 @@
 import os
 import shutil
 import subprocess
-from dataclasses import dataclass
 from pathlib import Path
 from typing import Optional, Union
 
-import jwt
 import requests
 
 
-@dataclass
-class Credentials:
-    """A set of credentials, consisting of a username and password."""
-
-    username: str
-    password: str
-
-
 class ArtifactsHelperCredentialProviderError(RuntimeError):
     """Generic error for ArtifactsHelperCredentialProvider."""
 
@@ -77,26 +67,23 @@ def auth_helper_installed(cls, auth_helper_path: Union[os.PathLike, str]) -> boo
         """Check whether the authentication helper is installed and executable."""
         return cls.resolve_auth_helper_path(auth_helper_path) is not None
 
-    def get_credentials(self, url: str) -> Optional[Credentials]:
-        """Get credentials for the given URL.
+    def get_token(self, url: str) -> Optional[str]:
+        """Get an access token for the given URL.
 
         Args:
             url: The URL to retrieve credentials for.
 
         Returns:
-            The credentials for the URL, or `None` if no credentials could be retrieved.
+            The token for the URL, or `None` if no credentials could be retrieved.
         """
         # Public feed short circuit: return nothing if not getting credentials for the
         # upload endpoint (which always requires auth) and the endpoint is public (can
         # authenticate without credentials).
         if not self._is_upload_endpoint(url) and self._can_authenticate(url, None):
             return None
 
-        jwt_str = self._get_jwt_from_helper()
-        if not jwt_str:
-            return None
-
-        return self._get_credentials_from_jwt(jwt_str)
+        token = self._get_token_from_helper()
+        return token if token else None
 
     @staticmethod
     def _is_upload_endpoint(url) -> bool:
@@ -107,7 +94,7 @@ def _can_authenticate(self, url, auth) -> bool:
         response = requests.get(url, auth=auth, timeout=self.timeout)
         return response.status_code < 500 and response.status_code not in (401, 403)
 
-    def _get_jwt_from_helper(self) -> str:
+    def _get_token_from_helper(self) -> str:
         if self.auth_tool_path is None:
             raise ArtifactsHelperCredentialProviderError(
                 "Failed to get credentials: No authentication tool found"
@@ -140,17 +127,3 @@ def _get_jwt_from_helper(self) -> str:
                 f"Failed to get credentials: Process {self.auth_tool_path} timed out "
                 f"after {self.timeout} seconds"
             ) from e
-
-    def _get_credentials_from_jwt(self, jwt_str: str) -> Credentials:
-        try:
-            decoded = jwt.decode(
-                jwt_str, verify=False, options={"verify_signature": False}
-            )
-            return Credentials(
-                username=decoded.get("unique_name", decoded.get("upn", None)),
-                password=jwt_str,
-            )
-        except jwt.PyJWTError as e:
-            raise ArtifactsHelperCredentialProviderError(
-                f"Failed to decode JWT: {e}"
-            ) from e

src/artifacts-helper/codespaces_artifacts_helper_keyring/src/codespaces_artifacts_helper_keyring/keyring_backend.py

@@ -21,6 +21,8 @@ class CodespacesArtifactsHelperKeyringBackend(KeyringBackend):
         "pkgs.vsts.me",
     )
 
+    DEFAULT_USERNAME = "codespaces"
+
     _PROVIDER: Type[ArtifactsHelperCredentialProvider] = (
         ArtifactsHelperCredentialProvider
     )
@@ -43,10 +45,10 @@ def get_credential(
             return None
 
         provider = self._PROVIDER(auth_helper_path=self.AUTH_HELPER_PATH)
-        creds = provider.get_credentials(service)
-        if creds is None:
+        token = provider.get_token(service)
+        if not token:
             return None
-        return SimpleCredential(creds.username or username, creds.password)
+        return SimpleCredential(username or self.DEFAULT_USERNAME, token)
 
     def _is_supported_netloc(self, service) -> bool:
         try:

src/artifacts-helper/codespaces_artifacts_helper_keyring/tests/test_artifacts_helper_credential_provider.py

@@ -14,7 +14,6 @@
 class TestArtifactsHelperWrapper(unittest.TestCase):
     SUPPORTED_HOST = "https://pkgs.dev.azure.com/"
     TEST_JWT = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJ1cG4iOiJ1cG5AY29udG9zby5jb20iLCJ1bmlxdWVfbmFtZSI6Im5hbWVAY29udG9zby5jb20ifQ.srKYrr5B0i29XERHsvE6mqZpLBzyyrX-gUKe9OHZODw"  # noqa: E501
-    TEST_JWT_USERNAME = "name@contoso.com"
 
     def __init__(self, *args, **kwargs) -> None:
         super().__init__(*args, **kwargs)
@@ -61,48 +60,37 @@ def test_auth_helper_installed_and_executable(self):
         )
         assert ArtifactsHelperCredentialProvider.auth_helper_installed(self.script_path)
 
-    def test_get_jwt_from_helper(self):
-        raw_jwt_value = "raw_jwt_here._-azAZ09"
-        self.write_script(f"print('{raw_jwt_value}')")
+    def test_get_token_from_helper(self):
+        raw_token_value = "raw_token_here._-azAZ09"
+        self.write_script(f"print('{raw_token_value}')")
         provider = ArtifactsHelperCredentialProvider(self.script_path)
-        assert provider._get_jwt_from_helper() == raw_jwt_value.strip()
+        assert provider._get_token_from_helper() == raw_token_value.strip()
 
-    def test_get_jwt_from_helper_not_installed(self):
+    def test_get_token_from_helper_not_installed(self):
         provider = ArtifactsHelperCredentialProvider()
         with pytest.raises(
             ArtifactsHelperCredentialProviderError, match="No authentication tool found"
         ):
-            provider._get_jwt_from_helper()
+            provider._get_token_from_helper()
 
-    def test_get_credentials_from_jwt(self):
-        provider = ArtifactsHelperCredentialProvider()
-        creds = provider._get_credentials_from_jwt(self.TEST_JWT)
-        assert creds.username == self.TEST_JWT_USERNAME
-        assert creds.password == self.TEST_JWT
-
-    def test_get_credentials(self):
+    def test_get_token(self):
         self.write_script(f"print('{self.TEST_JWT}')")
         provider = ArtifactsHelperCredentialProvider(self.script_path)
-        creds = provider.get_credentials(self.SUPPORTED_HOST)
-        assert creds.username == self.TEST_JWT_USERNAME
-        assert creds.password == self.TEST_JWT
+        assert provider.get_token(self.SUPPORTED_HOST) == self.TEST_JWT
 
-    def test_get_credentials_invalid_jwt(self):
-        self.write_script("print('invalid jwt')")
+    def test_get_token_invalid_token(self):
+        self.write_script(r"print('\n\n\n')")
         provider = ArtifactsHelperCredentialProvider(self.script_path)
-        with pytest.raises(
-            ArtifactsHelperCredentialProviderError, match="Failed to decode JWT:"
-        ):
-            provider.get_credentials(self.SUPPORTED_HOST)
+        assert provider.get_token(self.SUPPORTED_HOST) is None
 
-    def test_get_crendentials_helper_non_zero_exit(self):
+    def test_get_token_helper_non_zero_exit(self):
         self.write_script("exit(1)")
         provider = ArtifactsHelperCredentialProvider(self.script_path)
         with pytest.raises(
             ArtifactsHelperCredentialProviderError,
             match=f"Process .*{self.script_name}.* exited with code 1",
         ):
-            provider.get_credentials(self.SUPPORTED_HOST)
+            provider.get_token(self.SUPPORTED_HOST)
 
 
 def set_path_executable(path: Union[os.PathLike, str]):

src/artifacts-helper/codespaces_artifacts_helper_keyring/tests/test_keyring_backend.py

@@ -7,16 +7,15 @@
     ArtifactsHelperCredentialProvider,
     CodespacesArtifactsHelperKeyringBackend,
 )
-from keyring.credentials import SimpleCredential
 
 # Shouldn't be accessed by tests, but needs to be able
 # to get past the quick check.
 SUPPORTED_HOST = "https://pkgs.dev.azure.com/"
 
 
 class FakeProvider(ArtifactsHelperCredentialProvider):
-    def get_credentials(self, service):
-        return SimpleCredential("user" + service[-4:], "pass" + service[-4:])
+    def get_token(self, service):
+        return "pass" + service[-4:]
 
     @staticmethod
     def auth_helper_installed(auth_tool_path):
@@ -83,9 +82,15 @@ def test_get_credential_unsupported_host(only_backend):
     assert keyring.get_credential("https://example.com", None) is None
 
 
-def test_get_credential(only_backend, fake_provider):
+def test_get_credential_default_username(only_backend, fake_provider):
+    creds = keyring.get_credential(SUPPORTED_HOST + "1234", "user12345678")
+    assert creds.username == "user12345678"
+    assert creds.password == "pass1234"
+
+
+def test_get_credential_with_username(only_backend, fake_provider):
     creds = keyring.get_credential(SUPPORTED_HOST + "1234", None)
-    assert creds.username == "user1234"
+    assert creds.username == "codespaces"
     assert creds.password == "pass1234"
 
 
@@ -128,7 +133,7 @@ def test_delete_password_fallback(passwords, fake_provider):
 
 def test_cannot_delete_password(passwords, fake_provider):
     # Ensure we are getting good credentials
-    creds = keyring.get_credential(SUPPORTED_HOST + "1234", None)
+    creds = keyring.get_credential(SUPPORTED_HOST + "1234", "user1234")
     assert creds.username == "user1234"
     assert creds.password == "pass1234"