uv-publish 0.0.45

This is an internal component crate of uv
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127

//! Services that implement PyPI's Trusted Publishing interfaces.

use reqwest_middleware::ClientWithMiddleware;
use tracing::{debug, trace};
use url::Url;
use uv_client::BaseClient;
use uv_redacted::DisplaySafeUrl;

use crate::trusted_publishing::{
    Audience, MintTokenRequest, PublishToken, TrustedPublishingError, TrustedPublishingService,
    decode_oidc_token,
};

pub(crate) struct PyPIPublishingService<'a> {
    client: &'a ClientWithMiddleware,
    registry: &'a DisplaySafeUrl,
}

impl<'a> PyPIPublishingService<'a> {
    pub(crate) fn new(registry: &'a DisplaySafeUrl, client: &'a BaseClient) -> Self {
        Self {
            client: client.for_host(registry).raw_client(),
            registry,
        }
    }
}

impl TrustedPublishingService for PyPIPublishingService<'_> {
    fn client(&self) -> &ClientWithMiddleware {
        self.client
    }

    async fn audience(&self) -> Result<String, super::TrustedPublishingError> {
        // `pypa/gh-action-pypi-publish` uses `netloc` (RFC 1808), which is deprecated for authority
        // (RFC 3986).
        // Prefer HTTPS for OIDC discovery; allow HTTP only in test builds
        let scheme: &str = if cfg!(feature = "test") {
            self.registry.scheme()
        } else {
            "https"
        };
        let audience_url = DisplaySafeUrl::parse(&format!(
            "{}://{}/_/oidc/audience",
            scheme,
            self.registry.authority()
        ))?;
        debug!("Querying the trusted publishing audience from {audience_url}");
        let response = self
            .client
            .get(Url::from(audience_url.clone()))
            .send()
            .await
            .map_err(|err| TrustedPublishingError::ReqwestMiddleware(audience_url.clone(), err))?;
        let audience = response
            .error_for_status()
            .map_err(|err| TrustedPublishingError::Reqwest(audience_url.clone(), err))?
            .json::<Audience>()
            .await
            .map_err(|err| TrustedPublishingError::Reqwest(audience_url.clone(), err))?;
        trace!("The audience is `{}`", &audience.audience);
        Ok(audience.audience)
    }

    async fn exchange_token(
        &self,
        oidc_token: ambient_id::IdToken,
    ) -> Result<super::TrustedPublishingToken, super::TrustedPublishingError> {
        // Prefer HTTPS for OIDC minting; allow HTTP only in test builds
        let scheme: &str = if cfg!(feature = "test") {
            self.registry.scheme()
        } else {
            "https"
        };
        let mint_token_url = DisplaySafeUrl::parse(&format!(
            "{}://{}/_/oidc/mint-token",
            scheme,
            self.registry.authority()
        ))?;
        debug!("Querying the trusted publishing upload token from {mint_token_url}");
        let mint_token_payload = MintTokenRequest {
            token: oidc_token.reveal().to_string(),
        };
        let response = self
            .client
            .post(Url::from(mint_token_url.clone()))
            .json(&mint_token_payload)
            .send()
            .await
            .map_err(|err| {
                TrustedPublishingError::ReqwestMiddleware(mint_token_url.clone(), err)
            })?;

        // reqwest's implementation of `.json()` also goes through `.bytes()`
        let status = response.status();
        let body = response
            .bytes()
            .await
            .map_err(|err| TrustedPublishingError::Reqwest(mint_token_url.clone(), err))?;

        if status.is_success() {
            let publish_token: PublishToken = serde_json::from_slice(&body)?;
            Ok(publish_token.token)
        } else {
            match decode_oidc_token(oidc_token.reveal()) {
                Some(claims) => {
                    // An error here means that something is misconfigured, e.g. a typo in the PyPI
                    // configuration, so we're showing the body and the JWT claims for more context, see
                    // https://docs.pypi.org/trusted-publishers/troubleshooting/#token-minting
                    // for what the body can mean.
                    Err(TrustedPublishingError::TokenRejected(
                        status,
                        String::from_utf8_lossy(&body).to_string(),
                        claims,
                    ))
                }
                None => {
                    // This is not a user configuration error, the OIDC token should always have a valid
                    // format.
                    Err(TrustedPublishingError::InvalidOidcToken(
                        status,
                        String::from_utf8_lossy(&body).to_string(),
                    ))
                }
            }
        }
    }
}