Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion .github/workflows/ci.yml
Original file line number Diff line number Diff line change
Expand Up @@ -43,5 +43,6 @@ jobs:
- name: Build distribution
shell: bash
run: |
python -m pip install --upgrade pip build
python -m pip install --upgrade pip build==1.5.0
python -m build
python scripts/verify_distribution.py dist
3 changes: 2 additions & 1 deletion .github/workflows/publish.yml
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@ jobs:
- name: Build distribution
shell: bash
run: |
python -m pip install --upgrade pip build
python -m pip install --upgrade pip build==1.5.0
python -m build
python scripts/verify_distribution.py dist
- uses: pypa/gh-action-pypi-publish@cef221092ed1bacb1cc03d23a2d87d1d172e277b # release/v1
2 changes: 1 addition & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
[build-system]
requires = ["hatchling"]
requires = ["hatchling==1.31.0"]
build-backend = "hatchling.build"

[project]
Expand Down
109 changes: 109 additions & 0 deletions scripts/verify_distribution.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,109 @@
#!/usr/bin/env python3

"""Validate release archives before handing them to a package registry."""

from __future__ import annotations

import argparse
import email.policy
import re
import tarfile
import zipfile
from dataclasses import dataclass
from email.parser import BytesParser
from pathlib import Path

MAX_METADATA_BYTES = 1024 * 1024
MAX_METADATA_VERSION = (2, 4)
METADATA_VERSION_PATTERN = re.compile(r"^(\d+)\.(\d+)$")


@dataclass(frozen=True)
class DistributionMetadata:
name: str
version: str
metadata_version: tuple[int, int]


def parse_metadata(payload: bytes, source: str) -> DistributionMetadata:
if len(payload) > MAX_METADATA_BYTES:
raise ValueError(f"{source} metadata exceeds the 1 MiB limit")
document = BytesParser(policy=email.policy.compat32).parsebytes(payload)
raw_metadata_version = document.get("Metadata-Version", "")
match = METADATA_VERSION_PATTERN.fullmatch(raw_metadata_version)
if match is None:
raise ValueError(f"{source} has an invalid Metadata-Version")
metadata_version = (int(match.group(1)), int(match.group(2)))
if metadata_version > MAX_METADATA_VERSION:
maximum = ".".join(str(part) for part in MAX_METADATA_VERSION)
raise ValueError(
f"{source} uses unsupported Metadata-Version "
f"{raw_metadata_version}; maximum is {maximum}"
)
name = document.get("Name", "").strip()
version = document.get("Version", "").strip()
if not name or not version:
raise ValueError(f"{source} metadata must contain Name and Version")
return DistributionMetadata(name, version, metadata_version)


def read_wheel_metadata(path: Path) -> DistributionMetadata:
with zipfile.ZipFile(path) as archive:
candidates = [
name for name in archive.namelist() if name.endswith(".dist-info/METADATA")
]
if len(candidates) != 1:
raise ValueError(f"{path.name} must contain exactly one METADATA file")
info = archive.getinfo(candidates[0])
if info.file_size > MAX_METADATA_BYTES:
raise ValueError(f"{path.name} metadata exceeds the 1 MiB limit")
return parse_metadata(archive.read(info), path.name)


def read_sdist_metadata(path: Path) -> DistributionMetadata:
with tarfile.open(path, mode="r:gz") as archive:
candidates = [
member
for member in archive.getmembers()
if member.isfile() and member.name.endswith("/PKG-INFO")
]
if len(candidates) != 1:
raise ValueError(f"{path.name} must contain exactly one PKG-INFO file")
member = candidates[0]
if member.size > MAX_METADATA_BYTES:
raise ValueError(f"{path.name} metadata exceeds the 1 MiB limit")
extracted = archive.extractfile(member)
if extracted is None:
raise ValueError(f"{path.name} PKG-INFO could not be read")
return parse_metadata(extracted.read(MAX_METADATA_BYTES + 1), path.name)


def verify_distribution(directory: Path) -> DistributionMetadata:
wheels = sorted(directory.glob("*.whl"))
sdists = sorted(directory.glob("*.tar.gz"))
if len(wheels) != 1 or len(sdists) != 1:
raise ValueError("distribution directory must contain one wheel and one sdist")
wheel_metadata = read_wheel_metadata(wheels[0])
sdist_metadata = read_sdist_metadata(sdists[0])
if (wheel_metadata.name, wheel_metadata.version) != (
sdist_metadata.name,
sdist_metadata.version,
):
raise ValueError("wheel and sdist Name/Version metadata do not match")
return wheel_metadata


def arguments() -> argparse.Namespace:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("directory", type=Path)
return parser.parse_args()


def main() -> None:
metadata = verify_distribution(arguments().directory)
metadata_version = ".".join(str(part) for part in metadata.metadata_version)
print(f"verified {metadata.name} {metadata.version} (metadata {metadata_version})")


if __name__ == "__main__":
main()
68 changes: 68 additions & 0 deletions tests/unit/test_distribution_verifier.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,68 @@
from __future__ import annotations

import io
import subprocess
import sys
import tarfile
import zipfile
from pathlib import Path


def metadata(version: str, package_version: str = "1.2.3") -> bytes:
return (
f"Metadata-Version: {version}\n"
"Name: rstreamlabs-rstream\n"
f"Version: {package_version}\n\n"
).encode()


def write_wheel(directory: Path, payload: bytes) -> None:
path = directory / "rstreamlabs_rstream-1.2.3-py3-none-any.whl"
with zipfile.ZipFile(path, mode="w") as archive:
archive.writestr("rstreamlabs_rstream-1.2.3.dist-info/METADATA", payload)


def write_sdist(directory: Path, payload: bytes) -> None:
path = directory / "rstreamlabs_rstream-1.2.3.tar.gz"
with tarfile.open(path, mode="w:gz") as archive:
info = tarfile.TarInfo("rstreamlabs_rstream-1.2.3/PKG-INFO")
info.size = len(payload)
archive.addfile(info, io.BytesIO(payload))


def verify(directory: Path) -> subprocess.CompletedProcess[str]:
script = Path(__file__).parents[2] / "scripts" / "verify_distribution.py"
return subprocess.run(
[sys.executable, str(script), str(directory)],
check=False,
capture_output=True,
text=True,
)


def test_accepts_matching_metadata_2_4(tmp_path: Path) -> None:
payload = metadata("2.4")
write_wheel(tmp_path, payload)
write_sdist(tmp_path, payload)
result = verify(tmp_path)
assert result.returncode == 0
assert result.stdout.strip() == (
"verified rstreamlabs-rstream 1.2.3 (metadata 2.4)"
)


def test_rejects_metadata_newer_than_publisher_support(tmp_path: Path) -> None:
payload = metadata("2.5")
write_wheel(tmp_path, payload)
write_sdist(tmp_path, payload)
result = verify(tmp_path)
assert result.returncode != 0
assert "unsupported Metadata-Version 2.5" in result.stderr


def test_rejects_wheel_and_sdist_version_mismatch(tmp_path: Path) -> None:
write_wheel(tmp_path, metadata("2.4"))
write_sdist(tmp_path, metadata("2.4", package_version="1.2.4"))
result = verify(tmp_path)
assert result.returncode != 0
assert "Name/Version metadata do not match" in result.stderr