facett-ops/tests/pypi_upload.rs
//! **`cargo ops publish-pypi`, against a mock warehouse** — pure Rust, no network.
//!
//! A `TcpListener` on 127.0.0.1 plays PyPI: the classifier list, the JSON API's
//! "is this version up?" answer, and the legacy upload endpoint, recording every
//! request. Each test asserts on what ARRIVED at that server (the form fields, the
//! digests, the Basic auth, the order), not on what the uploader says it sent — and
//! the refusals are driven red by building the wheel PyPI would refuse.
//!
//! The token never appears in anything the uploader returns: one test makes the
//! server ECHO the Authorization header back in its error body, which is the
//! adversarial case for "never print the token".
use std::io::{BufRead, BufReader, Read, Write};
use std::net::TcpListener;
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};
use base64::Engine as _;
use blake2::digest::consts::U32;
use blake2::{Blake2b, Digest as _};
use facett_ops::pypi::{self, PublishOptions, Repository, Token};
// ── the mock warehouse ───────────────────────────────────────────────────────
#[derive(Debug, Clone)]
struct Seen {
method: String,
path: String,
headers: Vec<(String, String)>,
body: Vec<u8>,
}
impl Seen {
fn header(&self, k: &str) -> Option<&str> {
self.headers.iter().find(|(h, _)| h.eq_ignore_ascii_case(k)).map(|(_, v)| v.as_str())
}
}
#[derive(Clone)]
struct Behaviour {
/// `(name, version)` pairs the JSON API says already exist.
existing: Vec<(String, String)>,
/// The upload's status + reason, and whether to echo the Authorization header.
upload_status: (u16, &'static str),
echo_auth: bool,
/// Classifiers the list leaves OUT.
unknown: Vec<String>,
}
impl Default for Behaviour {
fn default() -> Self {
Behaviour { existing: Vec::new(), upload_status: (200, "OK"), echo_auth: false, unknown: Vec::new() }
}
}
const CLASSIFIERS: &[&str] = &[
"Development Status :: 4 - Beta",
"Programming Language :: Python :: 3",
"Programming Language :: Rust",
"Topic :: Scientific/Engineering :: Visualization",
];
struct Mock {
base: String,
seen: Arc<Mutex<Vec<Seen>>>,
}
fn mock(b: Behaviour) -> Mock {
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
let base = format!("http://{}", listener.local_addr().unwrap());
let seen: Arc<Mutex<Vec<Seen>>> = Arc::default();
let log = seen.clone();
std::thread::spawn(move || {
for stream in listener.incoming() {
let Ok(mut stream) = stream else { continue };
let mut r = BufReader::new(stream.try_clone().unwrap());
let mut line = String::new();
if r.read_line(&mut line).unwrap_or(0) == 0 {
continue;
}
let mut parts = line.split_whitespace();
let method = parts.next().unwrap_or("").to_string();
let path = parts.next().unwrap_or("").to_string();
let mut headers = Vec::new();
loop {
let mut h = String::new();
r.read_line(&mut h).unwrap();
let h = h.trim_end().to_string();
if h.is_empty() {
break;
}
if let Some((k, v)) = h.split_once(':') {
headers.push((k.trim().to_string(), v.trim().to_string()));
}
}
let len: usize = headers
.iter()
.find(|(k, _)| k.eq_ignore_ascii_case("content-length"))
.and_then(|(_, v)| v.parse().ok())
.unwrap_or(0);
let mut body = vec![0u8; len];
r.read_exact(&mut body).unwrap();
let s = Seen { method: method.clone(), path: path.clone(), headers, body };
log.lock().unwrap().push(s.clone());
let (status, reason, text): (u16, &str, String) = if method == "GET" && path.contains("list_classifiers") {
let mut all: Vec<String> = CLASSIFIERS.iter().map(|c| c.to_string()).filter(|c| !b.unknown.contains(c)).collect();
all.extend((0..150).map(|i| format!("Topic :: Filler :: {i}")));
(200, "OK", all.join("\n"))
} else if method == "GET" && path.starts_with("/pypi/") && path.ends_with("/json") {
let p: Vec<&str> = path.trim_matches('/').split('/').collect();
let hit = b.existing.iter().any(|(n, v)| p.get(1) == Some(&n.as_str()) && p.get(2) == Some(&v.as_str()));
if hit { (200, "OK", "{}".to_string()) } else { (404, "Not Found", "{\"message\": \"Not Found\"}".to_string()) }
} else if method == "POST" && path == "/legacy/" {
let echo = if b.echo_auth { format!("you sent: {}", s.header("authorization").unwrap_or("")) } else { String::new() };
(b.upload_status.0, b.upload_status.1, format!("{echo}\n{} {}", b.upload_status.0, b.upload_status.1).trim_start().to_string())
} else {
(404, "Not Found", String::new())
};
// A 3xx points somewhere the uploader must NOT go (it never follows one).
let location = if (300..400).contains(&status) { "Location: /elsewhere/\r\n" } else { "" };
let _ = write!(
stream,
"HTTP/1.1 {status} {reason}\r\n{location}Content-Length: {}\r\nContent-Type: text/plain\r\nConnection: close\r\n\r\n{text}",
text.len()
);
}
});
Mock { base, seen }
}
impl Mock {
fn repo(&self) -> Repository {
Repository::Custom {
upload_url: format!("{}/legacy/", self.base),
api_base: self.base.clone(),
token_env: "FACETT_OPS_TEST_TOKEN_UNSET".to_string(),
}
}
fn posts(&self) -> Vec<Seen> {
self.seen.lock().unwrap().iter().filter(|s| s.method == "POST").cloned().collect()
}
}
// ── fixtures: wheels and an sdist, as a build backend would write them ───────
const README: &str = "# facett\n\nThe components. Live demo: <https://nordisk.codeberg.page/facett/>\n";
fn metadata(name: &str, version: &str, extra: &str) -> String {
format!(
"Metadata-Version: 2.4\nName: {name}\nVersion: {version}\n\
Summary: facett's components in the browser\n\
Author: ignalina\nLicense-Expression: MIT\n\
Project-URL: Homepage, https://ignalina.dk\n\
Project-URL: Repository, https://git.gunnar.rs/nordisk/facett\n\
Classifier: Development Status :: 4 - Beta\n\
Classifier: Programming Language :: Rust\n\
Requires-Python: >=3.9\n\
Provides-Extra: streamlit\n\
Requires-Dist: streamlit<2,>=1.40; extra == \"streamlit\"\n\
{extra}Description-Content-Type: text/markdown\n\n{README}"
)
}
fn scratch(tag: &str) -> PathBuf {
let base = std::env::var("CARGO_TARGET_TMPDIR").map(PathBuf::from).unwrap_or_else(|_| std::env::temp_dir());
let d = base.join(format!("pypi-{tag}"));
let _ = std::fs::remove_dir_all(&d);
std::fs::create_dir_all(&d).unwrap();
d
}
fn wheel(dir: &Path, file: &str, md: &str) -> PathBuf {
let dist = file.split('-').take(2).collect::<Vec<_>>().join("-");
let p = dir.join(file);
let mut z = zip::ZipWriter::new(std::fs::File::create(&p).unwrap());
let o: zip::write::FileOptions<'_, '_, ()> = zip::write::FileOptions::default();
z.start_file("facett/__init__.py", o).unwrap();
z.write_all(b"x = 1\n").unwrap();
z.start_file(format!("{dist}.dist-info/METADATA"), o).unwrap();
z.write_all(md.as_bytes()).unwrap();
z.finish().unwrap();
p
}
fn sdist(dir: &Path, stem: &str, pkg_info: &str) -> PathBuf {
let p = dir.join(format!("{stem}.tar.gz"));
let gz = flate2::write::GzEncoder::new(std::fs::File::create(&p).unwrap(), flate2::Compression::default());
let mut t = tar::Builder::new(gz);
for (name, body) in [(format!("{stem}/PKG-INFO"), pkg_info.as_bytes()), (format!("{stem}/pyproject.toml"), b"[project]\n".as_slice())] {
let mut h = tar::Header::new_gnu();
h.set_size(body.len() as u64);
h.set_mode(0o644);
h.set_cksum();
t.append_data(&mut h, name, body).unwrap();
}
t.into_inner().unwrap().finish().unwrap();
p
}
fn good(tag: &str) -> PathBuf {
wheel(&scratch(tag), "facett-0.2.0-py3-none-any.whl", &metadata("facett", "0.2.0", ""))
}
fn problems(path: &Path) -> Vec<String> {
pypi::check_metadata(&pypi::read_dist(path).unwrap())
}
/// A tiny multipart reader: `(name, filename, bytes)` per part, in order.
fn parts(s: &Seen) -> Vec<(String, Option<String>, Vec<u8>)> {
let ct = s.header("content-type").unwrap();
let boundary = ct.split("boundary=").nth(1).unwrap();
let delim = format!("--{boundary}").into_bytes();
let body = &s.body;
let mut idx = Vec::new();
let mut i = 0;
while i + delim.len() <= body.len() {
if body[i..].starts_with(&delim) {
idx.push(i);
i += delim.len();
} else {
i += 1;
}
}
let mut out = Vec::new();
for w in idx.windows(2) {
let part = &body[w[0] + delim.len() + 2..w[1] - 2]; // skip "\r\n" after delim, before next
let split = part.windows(4).position(|x| x == b"\r\n\r\n").unwrap();
let head = String::from_utf8_lossy(&part[..split]).to_string();
let content = part[split + 4..].to_vec();
let name = head.split("name=\"").nth(1).unwrap().split('"').next().unwrap().to_string();
let filename = head.split("filename=\"").nth(1).map(|f| f.split('"').next().unwrap().to_string());
out.push((name, filename, content));
}
out
}
fn field(p: &[(String, Option<String>, Vec<u8>)], k: &str) -> Vec<String> {
p.iter().filter(|(n, _, _)| n == k).map(|(_, _, v)| String::from_utf8_lossy(v).into_owned()).collect()
}
// ── the offline checks (the `twine check` half) ──────────────────────────────
#[test]
fn metadata_reads_repeated_fields_and_the_body() {
let md = pypi::Metadata::parse(&metadata("facett", "0.2.0", ""));
assert_eq!(md.get("name"), Some("facett"));
assert_eq!(md.all("Classifier").len(), 2);
assert_eq!(md.all("Project-URL")[0], "Homepage, https://ignalina.dk");
assert!(md.description().unwrap().starts_with("# facett"));
}
#[test]
fn a_good_wheel_has_no_problems() {
assert_eq!(problems(&good("ok")), Vec::<String>::new());
}
#[test]
fn what_pypi_would_refuse_is_refused_by_name() {
let cases: Vec<(&str, String, &str)> = vec"), "RELATIVELY"),
("rst", metadata("facett", "0.2.0", "").replace("text/markdown", "text/x-rst"), "reStructuredText"),
("version", metadata("facett", "0.2.1", ""), "0.2.1"),
("name", metadata("facett-other", "0.2.0", ""), "facett-other"),
("local", metadata("facett", "0.2.0+dirty", ""), "PEP 440"),
("license", metadata("facett", "0.2.0", "Classifier: License :: OSI Approved :: MIT License\n"), "PEP 639"),
("url", metadata("facett", "0.2.0", "Project-URL: A label that is far longer than thirty-two characters, https://x.y\n"), "Project-URL"),
("pyreq", metadata("facett", "0.2.0", "").replace(">=3.9", "three point nine"), "Requires-Python"),
];
for (tag, md, want) in cases {
let w = wheel(&scratch(&format!("red-{tag}")), "facett-0.2.0-py3-none-any.whl", &md);
let bad = problems(&w);
assert!(bad.iter().any(|b| b.contains(want)), "{tag}: expected a problem naming `{want}`, got {bad:?}");
}
}
#[test]
fn an_sdist_is_read_from_its_pkg_info() {
let d = scratch("sdist");
let p = sdist(&d, "facett_demo-0.2.0", &metadata("facett-demo", "0.2.0", ""));
let f = pypi::read_dist(&p).unwrap();
assert_eq!(f.filetype, "sdist");
assert_eq!(f.pyversion, "source");
assert_eq!(f.metadata.get("Name"), Some("facett-demo"));
assert_eq!(pypi::check_metadata(&f), Vec::<String>::new());
}
#[test]
fn the_token_is_never_formatted() {
let t = Token::new("pypi-SECRET-123".into()).unwrap();
assert_eq!(format!("{t:?}"), "Token(***)");
assert!(Token::new(" ".into()).is_err(), "an empty token must be refused, not sent");
let b64 = base64::engine::general_purpose::STANDARD.encode("__token__:pypi-SECRET-123");
let red = pypi::redact(&format!("raw pypi-SECRET-123 and Basic {b64}"), &t);
assert!(!red.contains("SECRET") && !red.contains(&b64), "{red}");
}
#[test]
fn the_default_index_is_testpypi_and_the_real_one_is_named() {
let o = pypi::parse_args(&["--dry-run".into(), "a.whl".into()]).unwrap();
assert_eq!(o.repository, Repository::TestPypi);
assert!(o.dry_run);
let o = pypi::parse_args(&["--repository".into(), "pypi".into()]).unwrap();
assert_eq!(o.repository, Repository::Pypi);
assert_eq!(Repository::Pypi.upload_url(), "https://upload.pypi.org/legacy/");
assert_eq!(Repository::TestPypi.upload_url(), "https://test.pypi.org/legacy/");
assert_eq!(Repository::Pypi.token_env(), "PYPI_API_TOKEN");
assert_eq!(Repository::TestPypi.token_env(), "TEST_PYPI_API_TOKEN");
assert!(pypi::parse_args(&["--repository".into(), "pipy".into()]).is_err());
}
// ── against the mock warehouse ──────────────────────────────────────────────
#[test]
fn a_dry_run_checks_the_index_and_sends_nothing() {
let m = mock(Behaviour::default());
let opts = PublishOptions { repository: m.repo(), dry_run: true, files: vec![good("dry")] };
let report = pypi::publish_with_token(&opts, None).expect("dry run must pass without a token");
assert!(m.posts().is_empty(), "a dry run POSTed");
let gets: Vec<String> = m.seen.lock().unwrap().iter().map(|s| s.path.clone()).collect();
assert!(gets.iter().any(|p| p.contains("list_classifiers")), "{gets:?}");
assert!(gets.iter().any(|p| p == "/pypi/facett/0.2.0/json"), "{gets:?}");
let text = report.join("\n");
let f = pypi::read_dist(&opts.files[0]).unwrap();
assert!(text.contains("DRY RUN") && text.contains(&f.sha256_hex()) && text.contains(&f.blake2_256_hex()), "{text}");
assert!(text.contains(":action") && text.contains("file_upload"), "{text}");
}
#[test]
fn an_upload_posts_the_warehouse_form_in_order() {
let m = mock(Behaviour::default());
let d = scratch("up");
let a = wheel(&d, "facett-0.2.0-py3-none-any.whl", &metadata("facett", "0.2.0", ""));
let b = wheel(&d, "facett_demo-0.2.0-py3-none-any.whl", &metadata("facett-demo", "0.2.0", "Requires-Dist: facett[streamlit]<0.3,>=0.2.0\n"));
let s = sdist(&d, "facett_demo-0.2.0", &metadata("facett-demo", "0.2.0", ""));
let opts = PublishOptions { repository: m.repo(), dry_run: false, files: vec![a.clone(), b.clone(), s.clone()] };
pypi::publish_with_token(&opts, Some(Token::new("pypi-SECRET-abc".into()).unwrap())).expect("upload");
let posts = m.posts();
assert_eq!(posts.len(), 3);
for (post, file) in posts.iter().zip([&a, &b, &s]) {
assert_eq!(post.path, "/legacy/");
let want_auth = format!("Basic {}", base64::engine::general_purpose::STANDARD.encode("__token__:pypi-SECRET-abc"));
assert_eq!(post.header("authorization"), Some(want_auth.as_str()), "HTTP Basic, user __token__");
let p = parts(post);
let bytes = std::fs::read(file).unwrap();
assert_eq!(field(&p, ":action"), ["file_upload"]);
assert_eq!(field(&p, "protocol_version"), ["1"]);
assert_eq!(field(&p, "sha256_digest"), [facett_ops::pip::sha256_hex(&bytes)]);
let mut h = Blake2b::<U32>::new();
h.update(&bytes);
let b2: String = h.finalize().iter().map(|x| format!("{x:02x}")).collect();
assert_eq!(field(&p, "blake2_256_digest"), [b2], "blake2b-256, computed independently here");
assert_eq!(field(&p, "classifiers").len(), 2, "repeated metadata fields are repeated form fields");
assert_eq!(field(&p, "metadata_version"), ["2.4"]);
assert_eq!(field(&p, "license_expression"), ["MIT"]);
assert_eq!(field(&p, "description_content_type"), ["text/markdown"]);
assert!(field(&p, "description")[0].starts_with("# facett"));
let content = p.iter().find(|(n, _, _)| n == "content").unwrap();
assert_eq!(content.1.as_deref(), Some(file.file_name().unwrap().to_str().unwrap()));
assert_eq!(content.2, bytes, "the file arrives byte for byte");
}
assert_eq!(field(&parts(&posts[0]), "name"), ["facett"], "the components go FIRST");
assert_eq!(field(&parts(&posts[0]), "filetype"), ["bdist_wheel"]);
assert_eq!(field(&parts(&posts[0]), "pyversion"), ["py3"]);
assert_eq!(field(&parts(&posts[1]), "requires_dist").len(), 2);
assert_eq!(field(&parts(&posts[2]), "filetype"), ["sdist"]);
assert_eq!(field(&parts(&posts[2]), "pyversion"), ["source"]);
}
#[test]
fn a_version_already_on_the_index_is_refused_before_anything_is_sent() {
let m = mock(Behaviour { existing: vec![("facett".into(), "0.2.0".into())], ..Behaviour::default() });
let opts = PublishOptions { repository: m.repo(), dry_run: false, files: vec![good("exists")] };
let err = pypi::publish_with_token(&opts, Some(Token::new("pypi-x".into()).unwrap())).unwrap_err();
assert!(err.contains("ALREADY"), "{err}");
assert!(m.posts().is_empty(), "it uploaded anyway");
}
#[test]
fn an_unknown_classifier_is_refused_before_anything_is_sent() {
let m = mock(Behaviour { unknown: vec!["Programming Language :: Rust".into()], ..Behaviour::default() });
let opts = PublishOptions { repository: m.repo(), dry_run: true, files: vec![good("classifier")] };
let err = pypi::publish_with_token(&opts, None).unwrap_err();
assert!(err.contains("Programming Language :: Rust"), "{err}");
}
#[test]
fn a_refusal_never_carries_the_token_even_when_the_server_echoes_it() {
let m = mock(Behaviour { upload_status: (403, "Invalid or non-existent authentication information"), echo_auth: true, ..Behaviour::default() });
let opts = PublishOptions { repository: m.repo(), dry_run: false, files: vec![good("403")] };
let secret = "pypi-AgEIcHlwaS5vcmcSECRET";
let err = pypi::publish_with_token(&opts, Some(Token::new(secret.into()).unwrap())).unwrap_err();
assert_eq!(m.posts().len(), 1, "the upload was attempted");
let b64 = base64::engine::general_purpose::STANDARD.encode(format!("__token__:{secret}"));
assert!(err.contains("403"), "{err}");
assert!(!err.contains(secret) && !err.contains(&b64), "the token leaked into the error: {err}");
assert!(err.contains("***"), "the echoed header must be visibly redacted: {err}");
}
#[test]
fn a_file_that_already_exists_says_bump_the_version() {
let m = mock(Behaviour { upload_status: (400, "File already exists"), ..Behaviour::default() });
let opts = PublishOptions { repository: m.repo(), dry_run: false, files: vec![good("400")] };
let err = pypi::publish_with_token(&opts, Some(Token::new("pypi-y".into()).unwrap())).unwrap_err();
assert!(err.contains("bump the version"), "{err}");
}
// ── ureq 2 -> 3 (2026-10-03, finish-bump): what the ureq 2 client did, pinned ──
#[test]
fn a_redirect_is_reported_and_never_followed() {
// ureq 2 ran with `redirects(0)`: a 3xx to the upload is an error that names it,
// and nothing is sent to where it points (a followed POST arrives as a GET).
let m = mock(Behaviour { upload_status: (301, "Moved Permanently"), ..Behaviour::default() });
let opts = PublishOptions { repository: m.repo(), dry_run: false, files: vec![good("301")] };
let err = pypi::publish_with_token(&opts, Some(Token::new("pypi-r".into()).unwrap())).unwrap_err();
assert!(err.contains("redirected (301 → /elsewhere/)"), "{err}");
assert_eq!(m.posts().len(), 1, "exactly the one POST");
assert!(!m.seen.lock().unwrap().iter().any(|s| s.path == "/elsewhere/"), "the redirect was followed");
}
#[test]
fn credentials_in_a_repository_url_never_reach_an_error() {
// A `Custom` index URL may carry `user:password@`. The server's refusal and
// ureq's own error text (a refused connection) both name the URL — redacted.
let m = mock(Behaviour { upload_status: (500, "Internal Server Error"), ..Behaviour::default() });
let base = m.base.replacen("http://", "http://alice:pw-URL-SECRET@", 1);
let repo = Repository::Custom {
upload_url: format!("{base}/legacy/"),
api_base: base.clone(),
token_env: "FACETT_OPS_TEST_TOKEN_UNSET".to_string(),
};
let opts = PublishOptions { repository: repo, dry_run: false, files: vec![good("userinfo")] };
let err = pypi::publish_with_token(&opts, Some(Token::new("pypi-u".into()).unwrap())).unwrap_err();
assert!(err.contains("500") && err.contains("http://***@127.0.0.1:"), "{err}");
assert!(!err.contains("pw-URL-SECRET") && !err.contains("alice"), "{err}");
let closed = TcpListener::bind("127.0.0.1:0").unwrap();
let port = closed.local_addr().unwrap().port();
drop(closed);
let dead = Repository::Custom {
upload_url: format!("http://alice:pw-URL-SECRET@127.0.0.1:{port}/legacy/"),
api_base: format!("http://alice:pw-URL-SECRET@127.0.0.1:{port}"),
token_env: "FACETT_OPS_TEST_TOKEN_UNSET".to_string(),
};
let err = pypi::version_on_index(&dead, "facett", "0.2.0").unwrap_err();
assert!(err.contains("***@127.0.0.1"), "{err}");
assert!(!err.contains("pw-URL-SECRET") && !err.contains("alice"), "{err}");
let err = pypi::known_classifiers(&dead).unwrap_err();
assert!(!err.contains("pw-URL-SECRET") && !err.contains("alice"), "{err}");
assert_eq!(
pypi::redact_url("GET https://u:p@x.org/a?b=`c` and http://y.org/ and `ftp://k@z.net`"),
"GET https://***@x.org/a?b=`c` and http://y.org/ and `ftp://***@z.net`"
);
}
#[test]
fn pypirc_gives_the_section_token_and_refuses_a_password_login() {
let d = scratch("pypirc");
let rc = d.join(".pypirc");
std::fs::write(
&rc,
"[distutils]\nindex-servers =\n pypi\n\n[pypi]\n# the owner's\nusername = __token__\npassword = pypi-FROM-RC\n\n[other]\nusername = alice\npassword = hunter2\n",
)
.unwrap();
let t = pypi::read_pypirc(&rc, "pypi").unwrap().expect("[pypi] has a token");
assert_eq!(format!("{t:?}"), "Token(***)");
let b64 = base64::engine::general_purpose::STANDARD.encode("__token__:pypi-FROM-RC");
assert_eq!(pypi::redact(&format!("x {b64} y"), &t), "x *** y", "it is the [pypi] password, as __token__");
assert!(pypi::read_pypirc(&rc, "testpypi").unwrap().is_none(), "no [testpypi] section");
let err = pypi::read_pypirc(&rc, "other").unwrap_err();
assert!(err.contains("__token__") && !err.contains("hunter2"), "{err}");
assert!(pypi::read_pypirc(&d.join("absent"), "pypi").unwrap().is_none());
assert_eq!(Repository::Pypi.pypirc_section(), "pypi");
assert_eq!(Repository::TestPypi.pypirc_section(), "testpypi");
}
// ── the name gate: no personal name, mail domain or home path is ever uploaded ──
//
// The planted words are assembled at run time (`["ric", "kard"].concat()`) so this
// file itself stays clean under the same grep the gate runs.
fn planted(parts: &[&str]) -> String {
parts.concat()
}
fn wheel_with(dir: &Path, file: &str, md: &str, extra: &[(&str, Vec<u8>)]) -> PathBuf {
let dist = file.split('-').take(2).collect::<Vec<_>>().join("-");
let p = dir.join(file);
let mut z = zip::ZipWriter::new(std::fs::File::create(&p).unwrap());
let o: zip::write::FileOptions<'_, '_, ()> =
zip::write::FileOptions::default().compression_method(zip::CompressionMethod::Deflated);
for (name, body) in extra {
z.start_file(*name, o).unwrap();
z.write_all(body).unwrap();
}
z.start_file(format!("{dist}.dist-info/METADATA"), o).unwrap();
z.write_all(md.as_bytes()).unwrap();
z.finish().unwrap();
p
}
fn sdist_with(dir: &Path, stem: &str, pkg_info: &str, extra: &[(String, Vec<u8>)]) -> PathBuf {
let p = dir.join(format!("{stem}.tar.gz"));
let gz = flate2::write::GzEncoder::new(std::fs::File::create(&p).unwrap(), flate2::Compression::default());
let mut t = tar::Builder::new(gz);
let mut members = vec![(format!("{stem}/PKG-INFO"), pkg_info.as_bytes().to_vec())];
members.extend(extra.iter().cloned());
for (name, body) in members {
let mut h = tar::Header::new_gnu();
h.set_size(body.len() as u64);
h.set_mode(0o644);
h.set_cksum();
t.append_data(&mut h, name, body.as_slice()).unwrap();
}
t.into_inner().unwrap().finish().unwrap();
p
}
/// A fake wasm module with `word` at byte `at`, the way a string literal sits in a
/// data segment: binary on both sides.
fn wasm_with(word: &str, at: usize) -> Vec<u8> {
let mut b = b"\0asm\x01\0\0\0".to_vec();
b.resize(at, 0xA5);
b.extend_from_slice(word.as_bytes());
b.extend_from_slice(&[0, 0xFF, 0x13, 0x37]);
b
}
#[test]
fn a_name_compiled_into_a_wasm_member_is_named_with_member_and_offset() {
let name = planted(&["Ric", "kard"]);
let w = wheel_with(
&scratch("gate-wasm"),
"facett-0.2.0-py3-none-any.whl",
&metadata("facett", "0.2.0", ""),
&[("facett/assets/facet_wrapped_bg.wasm", wasm_with(&name, 4096))],
);
let f = pypi::read_dist(&w).unwrap();
let hits = pypi::name_hits(&f).unwrap();
assert_eq!(hits.len(), 1, "{hits:?}");
assert_eq!(hits[0].member, "facett/assets/facet_wrapped_bg.wasm");
assert_eq!(hits[0].offset, 4096, "the offset is in the DECOMPRESSED member (it is deflated in the zip)");
let err = pypi::name_gate(&f).unwrap_err();
assert!(err.contains("facett/assets/facet_wrapped_bg.wasm @ byte 4096"), "{err}");
assert!(pypi::check_wheel_offline(&w).is_err(), "publish-pip's own check must refuse it too");
}
#[test]
fn every_gate_word_is_caught_case_insensitively_in_text_and_in_member_names() {
let words = [
planted(&["RICK", "ARD"]),
planted(&["Lun", "din"]),
planted(&["hen", "RIK"]),
planted(&["alice@X14", ".SE"]),
planted(&["/ho", "me/alice/git/facett"]),
planted(&["bob@u", "bm.se"]),
planted(&["\"U", "BM\""]),
];
for (i, word) in words.iter().enumerate() {
let w = wheel_with(
&scratch(&format!("gate-word-{i}")),
"facett-0.2.0-py3-none-any.whl",
&metadata("facett", "0.2.0", ""),
&[("facett/__init__.py", format!("AUTHOR = '{word}'\n").into_bytes())],
);
let hits = pypi::name_hits(&pypi::read_dist(&w).unwrap()).unwrap();
assert_eq!(hits.len(), 1, "`{word}`: {hits:?}");
assert_eq!(hits[0].member, "facett/__init__.py");
let content = format!("AUTHOR = '{word}'\n");
let at = &content.as_bytes()[hits[0].offset..hits[0].offset + hits[0].word.len()];
assert!(at.eq_ignore_ascii_case(hits[0].word.as_bytes()), "`{word}`: the offset {} does not point at `{}`", hits[0].offset, hits[0].word);
}
// In a member's PATH, not its bytes.
let path = planted(&["facett/assets/hen", "rik.json"]);
let w = wheel_with(&scratch("gate-path"), "facett-0.2.0-py3-none-any.whl", &metadata("facett", "0.2.0", ""), &[(path.as_str(), b"{}".to_vec())]);
let hits = pypi::name_hits(&pypi::read_dist(&w).unwrap()).unwrap();
assert!(hits.iter().any(|h| h.member.ends_with("(member name)") && h.offset == 14), "{hits:?}");
}
#[test]
fn submit_is_not_a_name_but_the_mail_domain_as_a_token_is() {
let clean = b"wgpu::Queue::submit on_submitted_work_done SubmitCountRequest __wbg_submit_1290d".to_vec();
assert!(pypi::scan_for_names(&clean).is_empty(), "the mail-domain token inside `submit` must not trip the gate");
assert_eq!(pypi::scan_for_names(planted(&["mail: x@u", "bm.se"]).as_bytes()).len(), 1);
assert_eq!(pypi::scan_for_names(planted(&["(U", "BM)"]).as_bytes()).len(), 1);
}
#[test]
fn an_sdist_member_and_a_gzipped_member_inside_it_are_read_through() {
let d = scratch("gate-sdist");
let home = planted(&["/ho", "me/bob/src"]);
let mut gz = flate2::write::GzEncoder::new(Vec::new(), flate2::Compression::default());
gz.write_all(format!("path = {home}\n").as_bytes()).unwrap();
let s = sdist_with(
&d,
"facett_demo-0.2.0",
&metadata("facett-demo", "0.2.0", ""),
&[
("facett_demo-0.2.0/facett_demo/app.py".to_string(), format!("# see {home}\n").into_bytes()),
("facett_demo-0.2.0/facett_demo/data.json.gz".to_string(), gz.finish().unwrap()),
],
);
let hits = pypi::name_hits(&pypi::read_dist(&s).unwrap()).unwrap();
assert!(hits.iter().any(|h| h.member == "facett_demo-0.2.0/facett_demo/app.py" && h.offset == 6), "{hits:?}");
assert!(hits.iter().any(|h| h.member == "facett_demo-0.2.0/facett_demo/data.json.gz (gunzipped)" && h.offset == 7), "{hits:?}");
}
#[test]
fn a_planted_name_stops_the_dry_run_and_the_upload_before_the_network() {
let name = planted(&["lun", "din"]);
for dry_run in [true, false] {
let m = mock(Behaviour::default());
let d = scratch(&format!("gate-publish-{dry_run}"));
let clean = wheel(&d, "facett-0.2.0-py3-none-any.whl", &metadata("facett", "0.2.0", ""));
let dirty = wheel_with(
&d,
"facett_demo-0.2.0-py3-none-any.whl",
&metadata("facett-demo", "0.2.0", ""),
&[("facett_demo/demo_payloads.json", format!("{{\"owner\": \"{name}\"}}").into_bytes())],
);
let opts = PublishOptions { repository: m.repo(), dry_run, files: vec![clean, dirty] };
let err = pypi::publish_with_token(&opts, Some(Token::new("pypi-z".into()).unwrap())).unwrap_err();
assert!(err.contains("personal names in 1 file(s)"), "{err}");
assert!(err.contains("facett_demo-0.2.0-py3-none-any.whl: facett_demo/demo_payloads.json @ byte 11"), "{err}");
assert!(m.seen.lock().unwrap().is_empty(), "dry_run={dry_run}: the gate must refuse before ANY request (not even the classifier list)");
}
// And a clean set passes it, saying so.
let m = mock(Behaviour::default());
let opts = PublishOptions { repository: m.repo(), dry_run: true, files: vec![good("gate-clean")] };
let report = pypi::publish_with_token(&opts, None).unwrap().join("\n");
assert!(report.contains("name gate: 1 file(s)") && report.contains("0 hits"), "{report}");
}
#[test]
fn the_owner_credit_on_an_ownership_line_passes_and_the_name_anywhere_else_refuses() {
let credit = pypi::OWNERSHIP_CREDIT;
// On an ownership line: exempt, with each marker.
for line in [
format!("Copyright (c) 2026 {credit}\n"),
format!("# SPDX-FileCopyrightText: 2026 {credit}\n"),
format!("\u{a9} 2026 {credit}\n"),
] {
assert!(pypi::scan_for_names(line.as_bytes()).is_empty(), "{line}");
}
// The same credit on a line WITHOUT a marker: refused.
assert!(!pypi::scan_for_names(format!("author = {credit}\n").as_bytes()).is_empty());
// The name alone on the ownership line, outside the exact credit: refused.
let first = planted(&["rick", "ard"]);
assert!(!pypi::scan_for_names(format!("Copyright (c) 2026 {first}\n").as_bytes()).is_empty());
// A second, unrelated hit on the same ownership line is still a hit.
let home = planted(&["/ho", "me/"]);
let hits = pypi::scan_for_names(format!("Copyright (c) 2026 {credit} {home}x\n").as_bytes());
assert_eq!(hits.len(), 1, "{hits:?}");
// And in a packaged wheel: LICENSE with the credit passes the gate; a payload with the name refuses.
let d = scratch("gate-owner-credit");
let ok = wheel_with(
&d,
"facett-0.2.1-py3-none-any.whl",
&metadata("facett", "0.2.1", ""),
&[("facett-0.2.1.dist-info/licenses/LICENSE", format!("MIT License\n\nCopyright (c) 2026 {credit}\n").into_bytes())],
);
assert!(pypi::name_hits(&pypi::read_dist(&ok).unwrap()).unwrap().is_empty());
let bad = wheel_with(
&d,
"facett_demo-0.2.1-py3-none-any.whl",
&metadata("facett-demo", "0.2.1", ""),
&[("facett_demo/demo_payloads.json", format!("{{\"owner\": \"{credit}\"}}").into_bytes())],
);
assert!(!pypi::name_hits(&pypi::read_dist(&bad).unwrap()).unwrap().is_empty());
}