Automatically replace Trusted Publishing Tokens

.coveragerc

@@ -13,6 +13,7 @@ exclude_lines =
 
     # exclude typing.TYPE_CHECKING
     if TYPE_CHECKING:
+    if t.TYPE_CHECKING:
 
 [html]
 show_contexts = True

changelog/1246.feature.rst

@@ -0,0 +1,7 @@
+Automatically refresh short-lived PyPI token in long running Trusted Publishing
+uploads.
+
+In the event that a trusted publishing upload job is taking longer than the
+validity period of a trusted publishing token (15 minutes at the time of this
+writing), *and* we are already 10 minutes into that validity period, we will
+begin to attempt to replace the token on each subsequent request.

tests/test_auth.py

@@ -1,9 +1,13 @@
+import base64
 import getpass
 import logging
 import platform
 import re
+import time
+import typing as t
 
 import pytest
+import requests.auth
 
 from twine import auth
 from twine import exceptions
@@ -299,3 +303,154 @@ def test_warns_for_empty_password(
 )
 def test_keyring_module():
     assert auth.keyring is not None
+
+
+def test_resolver_authenticator_config_authentication(config):
+    config.update(username="username", password="password")
+    res = auth.Resolver(config, auth.CredentialInput())
+    assert isinstance(res.authenticator, requests.auth.HTTPBasicAuth)
+
+
+def test_resolver_authenticator_credential_input_authentication(config):
+    res = auth.Resolver(config, auth.CredentialInput("username", "password"))
+    assert isinstance(res.authenticator, requests.auth.HTTPBasicAuth)
+
+
+def test_resolver_authenticator_trusted_publishing_authentication(config):
+    res = auth.Resolver(
+        config, auth.CredentialInput(username="__token__", password="skip-stdin")
+    )
+    res._tp_token = auth.TrustedPublishingToken(
+        success=True,
+        token="fake-tp-token",
+    )
+    assert isinstance(res.authenticator, auth.TrustedPublishingAuthenticator)
+
+
+class MockResponse:
+    def __init__(self, status_code: int, json: t.Any) -> None:
+        self.status_code = status_code
+        self._json = json
+
+    def json(self, *args, **kwargs) -> t.Any:
+        return self._json
+
+    def raise_for_status(self) -> None:
+        if 400 <= self.status_code:
+            raise requests.exceptions.HTTPError()
+
+    def ok(self) -> bool:
+        return self.status_code == 200
+
+
+class MockSession:
+    def __init__(
+        self,
+        get_response_list: t.List[MockResponse],
+        post_response_list: t.List[MockResponse],
+    ) -> None:
+        self.post_counter = self.get_counter = 0
+        self.get_response_list = get_response_list
+        self.post_response_list = post_response_list
+
+    def get(self, url: str, **kwargs) -> MockResponse:
+        response = self.get_response_list[self.get_counter]
+        self.get_counter += 1
+        return response
+
+    def post(self, url: str, **kwargs) -> MockResponse:
+        response = self.post_response_list[self.post_counter]
+        self.post_counter += 1
+        return response
+
+
+def test_trusted_publish_authenticator_refreshes_token(monkeypatch, config):
+    def make_session():
+        return MockSession(
+            get_response_list=[
+                MockResponse(status_code=200, json={"audience": "fake-aud"})
+            ],
+            post_response_list=[
+                MockResponse(
+                    status_code=200,
+                    json={
+                        "success": True,
+                        "token": "new-token",
+                        "expires": int(time.time()) + 900,
+                    },
+                ),
+            ],
+        )
+
+    def detect_credential(*args, **kwargs) -> str:
+        return "fake-oidc-token"
+
+    config.update({"repository": utils.TEST_REPOSITORY})
+    res = auth.Resolver(config, auth.CredentialInput(username="__token__"))
+    res._tp_token = auth.TrustedPublishingToken(
+        success=True,
+        token="expiring-tp-token",
+    )
+    res._expires = int(time.time()) + 4 * 60
+    monkeypatch.setattr(auth, "detect_credential", detect_credential)
+    monkeypatch.setattr(auth.utils, "make_requests_session", make_session)
+    authenticator = auth.TrustedPublishingAuthenticator(resolver=res)
+    prepped_req = requests.models.PreparedRequest()
+    prepped_req.prepare_headers({})
+    request = authenticator(prepped_req)
+    assert (
+        request.headers["Authorization"]
+        == f"Basic {base64.b64encode(b'__token__:new-token').decode()}"
+    )
+
+
+def test_trusted_publish_authenticator_reuses_token(monkeypatch, config):
+    def make_session():
+        return MockSession(
+            get_response_list=[
+                MockResponse(status_code=200, json={"audience": "fake-aud"})
+            ],
+            post_response_list=[
+                MockResponse(
+                    status_code=200,
+                    json={
+                        "success": True,
+                        "token": "new-token",
+                        "expires": int(time.time()) + 900,
+                    },
+                ),
+            ],
+        )
+
+    def detect_credential(*args, **kwargs) -> str:
+        return "fake-oidc-token"
+
+    config.update({"repository": utils.TEST_REPOSITORY})
+    res = auth.Resolver(config, auth.CredentialInput(username="__token__"))
+    res._tp_token = auth.TrustedPublishingToken(
+        success=True,
+        token="valid-tp-token",
+    )
+    res._expires = int(time.time()) + 900
+    monkeypatch.setattr(auth, "detect_credential", detect_credential)
+    monkeypatch.setattr(auth.utils, "make_requests_session", make_session)
+    authenticator = auth.TrustedPublishingAuthenticator(resolver=res)
+    prepped_req = requests.models.PreparedRequest()
+    prepped_req.prepare_headers({})
+    request = authenticator(prepped_req)
+    assert (
+        request.headers["Authorization"]
+        == f"Basic {base64.b64encode(b'__token__:valid-tp-token').decode()}"
+    )
+
+
+def test_inability_to_make_token_raises_error():
+    class MockResolver:
+        def make_trusted_publishing_token(self) -> None:
+            return None
+
+    authenticator = auth.TrustedPublishingAuthenticator(
+        resolver=MockResolver(),
+    )
+    with pytest.raises(exceptions.TrustedPublishingFailure):
+        authenticator(None)

twine/__init__.py

@@ -39,6 +39,7 @@
     import importlib_metadata
 
 metadata = importlib_metadata.metadata("twine")
+assert metadata is not None  # nosec: B101
 
 
 __title__ = metadata["name"]