agent/connector/world/src/pip.rs
//! PyPI/pip half of the `world` connector: turn a Python dependency manifest
//! into `RemoteAsset`s the agent can pull through into a znippy archive.
//!
//! Two front doors, both yielding `Vec<PipPackage>`: `parse_requirements_txt`
//! (pinned `name==version`, files empty — extras and env-markers stripped) and
//! `parse_uv_lockfile` (registry packages only, sdist+wheel URLs already inline).
//! Requirements packages are then filled in by `resolve_with_index` against a
//! `PypiJsonClient` (the `/pypi/{name}/{ver}/json` API), which
//! only hits the network for packages whose `files` are still empty — so lockfile
//! entries skip the API entirely. `pip_packages_to_assets` is the final hop into
//! `transfer_to_znippy`, laying each file at `packages/{normalized_name}/{filename}`.
//!
//! Gotcha: `normalize_name` (PEP 503: lowercase, `_`/`.` → `-`) governs only the
//! archive *directory*; the filename is preserved verbatim, so `Pillow` lands
//! under `packages/pillow/Pillow-10.0.0.tar.gz`. The PyPI JSON body is capped at
//! `MAX_PYPI_JSON_BYTES` (M6) so a hostile index can't OOM the agent.
use serde::Deserialize;
use traits::RemoteAsset;
// ─── Error type ────────────────────────────────────────────────────────────────
#[derive(Debug)]
pub enum PipError {
Io(std::io::Error),
Parse {
line: usize,
msg: String,
},
PypiApi {
name: String,
version: String,
source: Box<dyn std::error::Error + Send + Sync>,
},
}
impl std::fmt::Display for PipError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::Io(e) => write!(f, "I/O error: {}", e),
Self::Parse { line, msg } => write!(f, "parse error at line {}: {}", line, msg),
Self::PypiApi {
name,
version,
source,
} => {
write!(f, "PyPI API error for {}@{}: {}", name, version, source)
}
}
}
}
impl std::error::Error for PipError {
fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
match self {
Self::Io(e) => Some(e),
Self::Parse { .. } => None,
Self::PypiApi { source, .. } => Some(source.as_ref()),
}
}
}
impl From<std::io::Error> for PipError {
fn from(e: std::io::Error) -> Self {
Self::Io(e)
}
}
// ─── Data types ────────────────────────────────────────────────────────────────
#[derive(Debug, Clone)]
pub struct PipPackage {
pub name: String,
pub version: String,
pub files: Vec<PipFile>,
}
#[derive(Debug, Clone)]
pub struct PipFile {
pub filename: String,
pub url: String,
pub hash: Option<String>,
}
impl PipFile {
/// Archive path within a znippy archive: `packages/{normalized_name}/{filename}`.
pub fn archive_path(&self, package_name: &str) -> String {
format!(
"packages/{}/{}",
normalize_name(package_name),
self.filename
)
}
}
// ─── Name normalization (PEP 503) ──────────────────────────────────────────────
fn normalize_name(s: &str) -> String {
s.to_lowercase().replace(['_', '.'], "-")
}
// ─── Serde structs for PyPI JSON API (private) ─────────────────────────────────
#[derive(Deserialize)]
struct PypiResponse {
urls: Vec<PypiUrl>,
}
#[derive(Deserialize)]
struct PypiUrl {
filename: String,
url: String,
digests: PypiDigests,
}
#[derive(Deserialize)]
struct PypiDigests {
sha256: Option<String>,
}
/// Cap on the PyPI JSON metadata body. A package's `/json` response is small;
/// this bounds memory against a malicious/compromised index streaming an
/// unbounded body (M6).
const MAX_PYPI_JSON_BYTES: u64 = 256 * 1024 * 1024;
// ─── Concrete PyPI JSON client ─────────────────────────────────────────────────
pub struct PypiJsonClient {
client: reqwest::Client,
base_url: String,
}
impl PypiJsonClient {
pub fn new() -> Self {
Self::with_base_url("https://pypi.org")
}
pub fn with_base_url(base_url: impl Into<String>) -> Self {
let client = tls_roots(reqwest::Client::builder())
.user_agent("holger-agent/0.5.0")
.build()
.unwrap_or_else(|_| {
tls_roots(reqwest::Client::builder())
.build()
.expect("reqwest client")
});
Self {
client,
base_url: base_url.into(),
}
}
}
impl Default for PypiJsonClient {
fn default() -> Self {
Self::new()
}
}
impl PypiJsonClient {
/// Fetches list of files for a package+version from the PyPI JSON API.
pub async fn fetch_files(&self, name: &str, version: &str) -> Result<Vec<PipFile>, PipError> {
let url = format!("{}/pypi/{}/{}/json", self.base_url, name, version);
let resp = self
.client
.get(&url)
.send()
.await
.map_err(|e| PipError::PypiApi {
name: name.to_string(),
version: version.to_string(),
source: Box::new(e),
})?;
if !resp.status().is_success() {
let status = resp.status();
return Err(PipError::PypiApi {
name: name.to_string(),
version: version.to_string(),
source: Box::new(std::io::Error::other(format!("HTTP {}", status))),
});
}
let mkerr = |e: Box<dyn std::error::Error + Send + Sync>| PipError::PypiApi {
name: name.to_string(),
version: version.to_string(),
source: e,
};
// M6: cap the body so a malicious index cannot OOM the agent. Check
// Content-Length up front, then enforce while streaming.
if let Some(len) = resp.content_length() {
if len > MAX_PYPI_JSON_BYTES {
return Err(mkerr(Box::new(std::io::Error::other(format!(
"PyPI JSON body {len} bytes exceeds cap"
)))));
}
}
let mut resp = resp;
let mut body: Vec<u8> = Vec::new();
while let Some(chunk) = resp.chunk().await.map_err(|e| mkerr(Box::new(e)))? {
if body.len() as u64 + chunk.len() as u64 > MAX_PYPI_JSON_BYTES {
return Err(mkerr(Box::new(std::io::Error::other(
"PyPI JSON body exceeded cap",
))));
}
body.extend_from_slice(&chunk);
}
let pypi: PypiResponse = serde_json::from_slice(&body).map_err(|e| PipError::PypiApi {
name: name.to_string(),
version: version.to_string(),
source: Box::new(e),
})?;
let files = pypi
.urls
.into_iter()
.map(|u| PipFile {
filename: u.filename,
url: u.url,
hash: u.digests.sha256.map(|h| format!("sha256:{}", h)),
})
.collect();
Ok(files)
}
}
// ─── Serde structs for uv.lock (private) ───────────────────────────────────────
#[derive(Deserialize)]
struct UvLock {
#[allow(dead_code)]
version: Option<u32>,
#[serde(default)]
package: Vec<UvPackage>,
}
#[derive(Deserialize)]
struct UvPackage {
name: String,
version: String,
source: Option<UvSource>,
sdist: Option<UvFile>,
#[serde(default)]
wheels: Vec<UvFile>,
}
#[derive(Deserialize)]
struct UvSource {
registry: Option<String>,
}
#[derive(Deserialize)]
struct UvFile {
url: String,
hash: Option<String>,
}
// ─── Parse functions ───────────────────────────────────────────────────────────
/// Parses a requirements.txt with pinned versions (`name==version`).
/// Returns packages with empty `files` — call `resolve_with_index` to populate.
pub fn parse_requirements_txt(
path: impl AsRef<std::path::Path>,
) -> Result<Vec<PipPackage>, PipError> {
let content = std::fs::read_to_string(path.as_ref())?;
let mut packages = Vec::new();
for (idx, raw_line) in content.lines().enumerate() {
let line_num = idx + 1;
// Strip inline comments and surrounding whitespace
let line = raw_line.split('#').next().unwrap_or("").trim();
if line.is_empty() {
continue;
}
// Skip option lines (-r, -e, --index-url, etc.)
if line.starts_with('-') {
continue;
}
// Strip environment markers (everything after the first ';')
let line = line.split(';').next().unwrap_or("").trim();
// Parse name==version
let (name_part, version) = line.split_once("==").ok_or_else(|| PipError::Parse {
line: line_num,
msg: format!("expected 'name==version', got: {}", line),
})?;
let name_part = name_part.trim();
let version = version.trim().to_string();
// Strip extras (e.g. "requests[security]" → "requests")
let name = if let Some(bracket_pos) = name_part.find('[') {
name_part[..bracket_pos].trim()
} else {
name_part
}
.to_string();
if name.is_empty() {
return Err(PipError::Parse {
line: line_num,
msg: "empty package name".to_string(),
});
}
packages.push(PipPackage {
name,
version,
files: Vec::new(),
});
}
Ok(packages)
}
/// Parses a uv.lock file and returns packages with resolved download URLs.
pub fn parse_uv_lockfile(path: impl AsRef<std::path::Path>) -> Result<Vec<PipPackage>, PipError> {
let content = std::fs::read_to_string(path.as_ref())?;
let lock: UvLock = toml::from_str(&content).map_err(|e| PipError::Parse {
line: 0,
msg: e.to_string(),
})?;
let packages = lock
.package
.into_iter()
.filter(|pkg| {
pkg.source
.as_ref()
.and_then(|s| s.registry.as_deref())
.map(|r| r.starts_with("https://pypi.org"))
.unwrap_or(false)
})
.map(|pkg| {
let mut files: Vec<PipFile> = Vec::new();
if let Some(sdist) = pkg.sdist {
let filename = sdist
.url
.split('/')
.next_back()
.unwrap_or(&sdist.url)
.to_string();
files.push(PipFile {
filename,
url: sdist.url,
hash: sdist.hash,
});
}
for wheel in pkg.wheels {
let filename = wheel
.url
.split('/')
.next_back()
.unwrap_or(&wheel.url)
.to_string();
files.push(PipFile {
filename,
url: wheel.url,
hash: wheel.hash,
});
}
PipPackage {
name: pkg.name,
version: pkg.version,
files,
}
})
.collect();
Ok(packages)
}
/// Populates `files` for packages using the given PyPI index.
pub async fn resolve_with_index(
packages: Vec<PipPackage>,
index: &PypiJsonClient,
) -> Result<Vec<PipPackage>, PipError> {
let mut resolved = Vec::with_capacity(packages.len());
for pkg in packages {
if pkg.files.is_empty() {
let files = index.fetch_files(&pkg.name, &pkg.version).await?;
resolved.push(PipPackage {
name: pkg.name,
version: pkg.version,
files,
});
} else {
resolved.push(pkg);
}
}
Ok(resolved)
}
/// Converts resolved pip packages to generic assets for transfer_to_znippy.
pub fn pip_packages_to_assets(packages: &[PipPackage]) -> Vec<RemoteAsset> {
packages
.iter()
.flat_map(|pkg| {
pkg.files.iter().map(|file| RemoteAsset {
path: file.archive_path(&pkg.name),
download_url: file.url.clone(),
content_type: Some("application/octet-stream".to_string()),
size: None,
// PyPI spells it `sha256:<hex>`; the verifier strips the prefix.
sha256: file.hash.clone(),
// PyPI publishes no change token; the digest above is the proof.
change_token: None,
})
})
.collect()
}
// ─── Tests ─────────────────────────────────────────────────────────────────────
/// reqwest 0.13 with reqwest 0.12 `rustls-tls` behaviour (2026-10-02): the ring
/// provider and the bundled Mozilla roots (webpki), not 0.13's aws-lc-rs + OS
/// trust store. Roots added later on the builder are merged in, as in 0.12.
#[allow(dead_code)]
fn tls_roots(b: reqwest::ClientBuilder) -> reqwest::ClientBuilder {
let _ = rustls::crypto::ring::default_provider().install_default();
b.tls_certs_only(
webpki_root_certs::TLS_SERVER_ROOT_CERTS
.iter()
.filter_map(|c| reqwest::Certificate::from_der(c).ok()),
)
}
#[cfg(test)]
mod tests {
use super::*;
#[cfg(feature = "testmatrix")]
use traits::test_support::fstatus;
#[test]
fn test_normalize_name_basic() {
assert_eq!(normalize_name("requests"), "requests");
assert_eq!(normalize_name("Pillow"), "pillow");
assert_eq!(normalize_name("python_dotenv"), "python-dotenv");
assert_eq!(normalize_name("apache.kafka"), "apache-kafka");
#[cfg(feature = "testmatrix")]
fstatus(
"world-connector",
"pip_normalize_name_basic",
normalize_name("Pillow") == "pillow"
&& normalize_name("apache.kafka") == "apache-kafka",
&format!(
"Pillow->{} apache.kafka->{}",
normalize_name("Pillow"),
normalize_name("apache.kafka")
),
);
}
#[test]
fn test_normalize_name_edge_cases() {
assert_eq!(normalize_name(""), "");
assert_eq!(normalize_name("My_Package.Name"), "my-package-name");
assert_eq!(normalize_name("UPPER_CASE"), "upper-case");
}
#[test]
fn test_archive_path() {
let file = PipFile {
filename: "requests-2.31.0.tar.gz".to_string(),
url: "https://example.com/requests-2.31.0.tar.gz".to_string(),
hash: None,
};
let p = file.archive_path("requests");
assert_eq!(p, "packages/requests/requests-2.31.0.tar.gz");
#[cfg(feature = "testmatrix")]
fstatus(
"world-connector",
"pip_archive_path",
p == "packages/requests/requests-2.31.0.tar.gz",
&format!("archive_path = {p}"),
);
}
#[test]
fn test_archive_path_normalization() {
let file = PipFile {
filename: "Pillow-10.0.0.tar.gz".to_string(),
url: "https://example.com/Pillow-10.0.0.tar.gz".to_string(),
hash: None,
};
assert_eq!(
file.archive_path("Pillow"),
"packages/pillow/Pillow-10.0.0.tar.gz"
);
}
#[test]
fn test_parse_requirements_txt() {
let content = "# This is a comment\n\
requests==2.31.0\n\
pillow==10.0.0 # inline comment\n\
Flask==3.0.0\n\
requests[security]==2.31.0\n\
django==4.2; python_version >= \"3.8\"\n";
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("requirements.txt");
std::fs::write(&path, content).unwrap();
let packages = parse_requirements_txt(&path).unwrap();
assert_eq!(packages.len(), 5);
assert_eq!(packages[0].name, "requests");
assert_eq!(packages[0].version, "2.31.0");
assert_eq!(packages[1].name, "pillow");
assert_eq!(packages[2].name, "Flask");
assert_eq!(packages[3].name, "requests"); // extras stripped
assert_eq!(packages[4].name, "django"); // markers stripped
assert_eq!(packages[4].version, "4.2");
for pkg in &packages {
assert!(pkg.files.is_empty());
}
#[cfg(feature = "testmatrix")]
fstatus(
"world-connector",
"pip_parse_requirements_txt",
packages.len() == 5
&& packages[3].name == "requests"
&& packages[4].name == "django"
&& packages[4].version == "4.2",
&format!(
"5 reqs parsed; extras stripped ({}), markers stripped ({} {})",
packages[3].name, packages[4].name, packages[4].version
),
);
}
#[test]
fn test_parse_requirements_txt_invalid() {
let content = "invalid-line-without-version\n";
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("requirements-invalid.txt");
std::fs::write(&path, content).unwrap();
let result = parse_requirements_txt(&path);
assert!(result.is_err());
let parse_line_1 = matches!(&result, Err(PipError::Parse { line, .. }) if *line == 1);
assert!(
parse_line_1,
"expected PipError::Parse at line 1, got {result:?}"
);
#[cfg(feature = "testmatrix")]
fstatus(
"world-connector",
"pip_parse_requirements_invalid",
parse_line_1,
&format!("invalid requirement -> PipError::Parse at line 1 = {parse_line_1}"),
);
}
#[test]
fn test_parse_uv_lockfile() {
let content = r#"
version = 1
[[package]]
name = "requests"
version = "2.31.0"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/req/requests-2.31.0.tar.gz", hash = "sha256:abc123", size = 12345 }
wheels = [
{ url = "https://files.pythonhosted.org/packages/req/requests-2.31.0-py3-none-any.whl", hash = "sha256:def456", size = 67890 },
]
[[package]]
name = "certifi"
version = "2023.7.22"
source = { registry = "https://pypi.org/simple" }
[[package]]
name = "local-pkg"
version = "1.0.0"
source = { path = "/local/path" }
"#;
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("uv.lock");
std::fs::write(&path, content).unwrap();
let packages = parse_uv_lockfile(&path).unwrap();
// local-pkg should be filtered out (no registry source)
assert_eq!(packages.len(), 2);
let requests = &packages[0];
assert_eq!(requests.name, "requests");
assert_eq!(requests.version, "2.31.0");
assert_eq!(requests.files.len(), 2);
assert_eq!(requests.files[0].filename, "requests-2.31.0.tar.gz");
assert_eq!(requests.files[0].hash, Some("sha256:abc123".to_string()));
assert_eq!(
requests.files[1].filename,
"requests-2.31.0-py3-none-any.whl"
);
assert_eq!(requests.files[1].hash, Some("sha256:def456".to_string()));
// certifi has no sdist or wheels
let certifi = &packages[1];
assert_eq!(certifi.name, "certifi");
assert!(certifi.files.is_empty());
#[cfg(feature = "testmatrix")]
fstatus(
"world-connector",
"pip_parse_uv_lockfile",
packages.len() == 2
&& requests.files.len() == 2
&& requests.files[0].hash == Some("sha256:abc123".to_string())
&& certifi.files.is_empty(),
&format!(
"local-pkg filtered (kept {}), requests has {} files w/ hashes",
packages.len(),
requests.files.len()
),
);
}
}