blob: 0831acf754dd73116733bf8dd48b5f2cdb8e8167 [file]
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership. The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
import json
import os
import urllib.request
from contextlib import contextmanager
REGISTRY_URL = "https://crates.io"
USER_AGENT = "apache-reqsign-release/1.0 (https://github.com/apache/reqsign)"
def request_json(url: str, *, headers=None, body=None):
request = urllib.request.Request(
url,
data=json.dumps(body).encode() if body is not None else None,
headers={
"User-Agent": USER_AGENT,
"Content-Type": "application/json",
**(headers or {}),
},
)
with urllib.request.urlopen(request, timeout=30) as response:
return json.load(response)
@contextmanager
def temporary_trusted_publishing_token():
request_url = os.environ["ACTIONS_ID_TOKEN_REQUEST_URL"]
separator = "&" if "?" in request_url else "?"
jwt = request_json(
f"{request_url}{separator}audience=crates.io",
headers={
"Authorization": f"Bearer {os.environ['ACTIONS_ID_TOKEN_REQUEST_TOKEN']}"
},
)["value"]
assert jwt, "GitHub returned an empty OIDC token"
print(f"::add-mask::{jwt}", flush=True)
token = request_json(
f"{REGISTRY_URL}/api/v1/trusted_publishing/tokens", body={"jwt": jwt}
)["token"]
assert token, "crates.io returned an empty Trusted Publishing token"
print(f"::add-mask::{token}", flush=True)
try:
yield token
finally:
request = urllib.request.Request(
f"{REGISTRY_URL}/api/v1/trusted_publishing/tokens",
headers={"Authorization": f"Bearer {token}", "User-Agent": USER_AGENT},
method="DELETE",
)
with urllib.request.urlopen(request, timeout=30):
pass