diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index a1706a9c..6d41dbd0 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -726,6 +726,7 @@ jobs: docker pull mirror.gcr.io/library/redis:5.0 docker pull mirror.gcr.io/sonatype/nexus3:3.70.1 docker pull mirror.gcr.io/verdaccio/verdaccio:6.11.1 + docker pull mirror.gcr.io/library/nginx:1.27.5-alpine - name: Run e2e tests (forge-e2e feature) id: tests diff --git a/README.md b/README.md index 67737ee4..82004593 100644 --- a/README.md +++ b/README.md @@ -240,8 +240,8 @@ The guides below are also a site, with navigation and search: built into the server: design, operation, and writing one. - **[Importing repositories](./docs/import.md)** — copy a hosted repository of Nexus, Artifactory, Verdaccio, a registry or another Forge (NuGet, npm, - Docker) into Forge through the format's own protocol: dry run, conflicts, resuming, and - the cutover. + Docker, raw) into Forge through the format's own protocol: dry run, + conflicts, resuming, and the cutover. Run `cargo xtask help` for the complete command list. diff --git a/crates/db/src/import_dao.rs b/crates/db/src/import_dao.rs index b5652f16..5fa6c8b6 100644 --- a/crates/db/src/import_dao.rs +++ b/crates/db/src/import_dao.rs @@ -25,9 +25,9 @@ use uuid::Uuid; pub struct RepositoryImport { pub id: i64, pub repository_id: i64, - /// The source: a NuGet service index or a registry. For an import stored - /// before 2.2, the Nexus server that `source_repository` names a - /// repository of. + /// The source: a NuGet service index, a registry, or a raw repository. + /// For an import stored before 2.2, the Nexus server that + /// `source_repository` names a repository of. pub url: String, /// The Nexus repository of an import stored before 2.2; `None` since. pub source_repository: Option, @@ -47,8 +47,9 @@ pub struct RepositoryImport { pub state: String, /// `scan` or `copy`. pub phase: String, - /// What the scan walks: `search`, `catalog`, `all` (npm's `/-/all`) or - /// `names`. + /// What the scan walks: `search`, `catalog`, `all` (npm's `/-/all`), + /// `assets` (Nexus' asset listing), `directories` (another Forge's raw + /// listing) or `names`. pub scan_listing: String, #[serde(skip)] pub cursor: Option, @@ -82,7 +83,8 @@ pub struct NewImport<'a> { pub on_conflict: &'a str, pub concurrency: i32, pub dry_run: bool, - /// Where the scan starts: `search`, `catalog`, `all` or `names`. + /// Where the scan starts: `search`, `catalog`, `all`, `assets`, + /// `directories` or `names`. pub scan_listing: &'a str, pub source_server: Option<&'a str>, pub source_total: Option, @@ -444,8 +446,9 @@ pub struct ImportItem { #[serde(skip)] pub import_id: i64, /// `component` (a version, a Docker tag), `manifest` (a Docker referrer - /// held by digest) or `tag` (an npm dist-tag: `name` the package, - /// `version` the tag). + /// held by digest), `tag` (an npm dist-tag: `name` the package, + /// `version` the tag) or `file` (a raw file: `name` its path, `version` + /// empty). pub kind: String, pub name: String, pub version: String, diff --git a/crates/db/src/raw_dao.rs b/crates/db/src/raw_dao.rs index c8053975..8a2068af 100644 --- a/crates/db/src/raw_dao.rs +++ b/crates/db/src/raw_dao.rs @@ -270,6 +270,58 @@ pub async fn store( ))) } +/// When the file at `path` last changed (`updated_at`), locked `FOR UPDATE` +/// in `tx`: what an import that replaces it keeps its new upload time after. +pub async fn lock_updated_at( + tx: &mut Transaction<'_, Postgres>, + repository_id: i64, + path: &str, +) -> sqlx::Result>> { + sqlx::query_scalar( + "SELECT updated_at FROM raw_file WHERE repository_id = $1 AND path = $2 FOR UPDATE", + ) + .bind(repository_id) + .bind(path) + .fetch_optional(&mut **tx) + .await +} + +/// Give file `id`, which an import just stored, the times its source names: +/// `uploaded_at` (when the source stored the content) becomes its upload +/// time, which `Last-Modified` and retention's age rules read, and +/// `downloaded_at` its last download. +/// +/// A file the import `replaced` keeps an upload time after `replaced`, the +/// one of the content it replaced: a client that cached that content and +/// asks `If-Modified-Since` must not be told it is current. +pub async fn set_imported_times( + tx: &mut Transaction<'_, Postgres>, + id: i64, + uploaded_at: Option>, + downloaded_at: Option>, + replaced: Option>, +) -> sqlx::Result<()> { + sqlx::query( + r#" + UPDATE raw_file + SET created_at = CASE WHEN $4::timestamptz IS NULL + THEN coalesce($2, created_at) ELSE created_at END, + updated_at = CASE WHEN $2::timestamptz IS NULL THEN updated_at + WHEN $4::timestamptz IS NULL THEN $2 + ELSE greatest($2, $4 + interval '1 microsecond') END, + last_downloaded_at = coalesce($3, last_downloaded_at) + WHERE id = $1 + "#, + ) + .bind(id) + .bind(uploaded_at) + .bind(downloaded_at) + .bind(replaced) + .execute(&mut **tx) + .await?; + Ok(()) +} + /// Delete the file at `path`, releasing its blob reference. Returns whether /// there was one. pub async fn delete( diff --git a/crates/e2e/tests/raw_import.rs b/crates/e2e/tests/raw_import.rs new file mode 100644 index 00000000..52d7e673 --- /dev/null +++ b/crates/e2e/tests/raw_import.rs @@ -0,0 +1,677 @@ +//! E2E: imports of raw repositories from real sources: +//! +//! * a Nexus Repository 3 OSS container (`sonatype/nexus3`): a raw-hosted +//! repository read through Nexus' asset listing (its REST API), with +//! nested directories, a file name with a space and a non-ASCII +//! character, more files than a page of the listing holds and a path +//! Forge refuses. Every file arrives with its bytes, content type and +//! creation time (`blobCreated`), its `.sha256` sidecar in Forge is +//! Nexus' SHA-256, a file Nexus served has Nexus' last download, and its +//! uploader is the item's. A dry run reports the files, their size and +//! the refused path and writes nothing; a second run copies only the new +//! file, a changed one is a conflict, and `overwrite` replaces it; +//! * another Forge: a raw repository read through its JSON listing, with +//! more files in a directory than a page of it holds; upload times and +//! last downloads carry over, and a second run downloads nothing; +//! * nginx, a web server without a listing: named paths, a file checked +//! against its `.sha256` sidecar, one whose sidecar does not match, and +//! one the server does not have. +//! +//! The conflict policies, resuming a walk and the preflight's refusals are +//! covered against a mock source in `crates/server/tests/raw_import.rs`. +//! Gated behind `forge-e2e`; Nexus takes a minute or two to start. + +#![cfg(feature = "forge-e2e")] + +use std::collections::BTreeMap; +use std::time::Duration; + +use chrono::{DateTime, Utc}; +use forge_e2e::{StackBuilder, TestServer}; +use forge_server::import::ImportWorker; +use serde_json::{json, Value}; +use sha2::Digest; +use testcontainers::core::{IntoContainerPort, WaitFor}; +use testcontainers::runners::AsyncRunner; +use testcontainers::{GenericImage, ImageExt}; + +const NEXUS_IMAGE: &str = "mirror.gcr.io/sonatype/nexus3"; +const NEXUS_TAG: &str = "3.70.1"; +/// With `NEXUS_SECURITY_RANDOMPASSWORD=false`, Nexus' admin password. +const ADMIN: (&str, &str) = ("admin", "admin123"); + +const NGINX_IMAGE: &str = "mirror.gcr.io/library/nginx"; +const NGINX_TAG: &str = "1.27.5-alpine"; + +/// More files than a page of Nexus' asset listing holds. +const BULK_FILES: usize = 25; + +/// More files in one directory than a page of Forge's listing holds (the +/// import asks for 100). +const FORGE_BULK: usize = 105; + +fn sha256(bytes: &[u8]) -> String { + hex::encode(sha2::Sha256::digest(bytes)) +} + +/// `base` with `path` appended, each segment percent-encoded. +fn url_of(base: &str, path: &str) -> String { + let mut url = url::Url::parse(&format!("{}/", base.trim_end_matches('/'))).unwrap(); + { + let mut segments = url.path_segments_mut().unwrap(); + segments.pop_if_empty(); + segments.extend(path.split('/')); + } + url.to_string() +} + +fn time(s: &str) -> DateTime { + DateTime::parse_from_rfc3339(s).unwrap().with_timezone(&Utc) +} + +fn assert_recent(t: Option>) { + let t = t.expect("a time"); + assert!( + (Utc::now() - t).num_seconds().abs() < 300, + "{t} is not about now" + ); +} + +// ── Nexus ─────────────────────────────────────────────────────────────────── + +/// A Nexus with its REST API and repositories at `url`. +struct Nexus { + url: String, + client: reqwest::Client, +} + +impl Nexus { + async fn start() -> (Self, testcontainers::ContainerAsync) { + let container = GenericImage::new(NEXUS_IMAGE, NEXUS_TAG) + .with_exposed_port(8081.tcp()) + .with_wait_for(WaitFor::message_on_stdout("Started Sonatype Nexus")) + .with_env_var("NEXUS_SECURITY_RANDOMPASSWORD", "false") + .with_env_var( + "INSTALL4J_ADD_VM_PARAMS", + "-Xms512m -Xmx1g -XX:MaxDirectMemorySize=1g", + ) + .with_startup_timeout(Duration::from_secs(600)) + .start() + .await + .expect("start nexus (is Docker running?)"); + let host = container.get_host().await.unwrap(); + let port = container.get_host_port_ipv4(8081).await.unwrap(); + let nexus = Nexus { + url: format!("http://{host}:{port}"), + client: reqwest::Client::new(), + }; + for _ in 0..300 { + let up = nexus + .client + .get(format!("{}/service/rest/v1/status/writable", nexus.url)) + .send() + .await + .is_ok_and(|r| r.status() == 200); + if up { + return (nexus, container); + } + tokio::time::sleep(Duration::from_secs(1)).await; + } + panic!("nexus never became writable"); + } + + fn request(&self, method: reqwest::Method, url: String) -> reqwest::RequestBuilder { + self.client + .request(method, url) + .basic_auth(ADMIN.0, Some(ADMIN.1)) + } + + async fn create_raw_hosted(&self, name: &str) { + let body = json!({ + "name": name, "online": true, + "storage": {"blobStoreName": "default", "strictContentTypeValidation": false, "writePolicy": "allow"}, + }); + let resp = self + .request( + reqwest::Method::POST, + format!("{}/service/rest/v1/repositories/raw/hosted", self.url), + ) + .json(&body) + .send() + .await + .unwrap(); + assert_eq!(resp.status(), 201, "{}", resp.text().await.unwrap()); + } + + fn repository(&self, name: &str) -> String { + format!("{}/repository/{name}", self.url) + } + + /// PUT a file into the raw repository `repo`; returns the status. + async fn put(&self, repo: &str, path: &str, bytes: &[u8]) -> reqwest::StatusCode { + self.request(reqwest::Method::PUT, url_of(&self.repository(repo), path)) + .body(bytes.to_vec()) + .send() + .await + .unwrap() + .status() + } + + async fn get(&self, repo: &str, path: &str) -> Vec { + let resp = self + .request(reqwest::Method::GET, url_of(&self.repository(repo), path)) + .send() + .await + .unwrap(); + assert_eq!(resp.status(), 200, "GET {path}"); + resp.bytes().await.unwrap().to_vec() + } + + /// Every asset of `repo` by path, and how many pages the listing took. + async fn assets(&self, repo: &str) -> (BTreeMap, usize) { + let mut assets = BTreeMap::new(); + let mut token: Option = None; + let mut pages = 0; + loop { + let mut url = format!("{}/service/rest/v1/assets?repository={repo}", self.url); + if let Some(token) = &token { + url.push_str(&format!("&continuationToken={token}")); + } + let page: Value = self + .request(reqwest::Method::GET, url) + .send() + .await + .unwrap() + .json() + .await + .unwrap(); + pages += 1; + for asset in page["items"].as_array().unwrap() { + let path = asset["path"].as_str().unwrap(); + assets.insert( + path.strip_prefix('/').unwrap_or(path).to_string(), + asset.clone(), + ); + } + match page["continuationToken"].as_str() { + Some(next) => token = Some(next.to_string()), + None => return (assets, pages), + } + } + } +} + +// ── Forge ─────────────────────────────────────────────────────────────────── + +async fn admin( + srv: &TestServer, + method: reqwest::Method, + path: &str, + body: Option, +) -> reqwest::Response { + let (u, p) = srv.admin_basic(); + let mut req = srv + .client() + .request(method, format!("{}{path}", srv.base_url)) + .basic_auth(u, Some(p)); + if let Some(body) = body { + req = req.json(&body); + } + req.send().await.unwrap() +} + +async fn raw_hosted(srv: &TestServer, name: &str) -> i64 { + let resp = admin( + srv, + reqwest::Method::POST, + "/admin/repositories", + Some(json!({"name": name, "kind": "hosted", "format": "raw"})), + ) + .await; + assert_eq!(resp.status(), 201, "{}", resp.text().await.unwrap()); + resp.json::().await.unwrap()["id"].as_i64().unwrap() +} + +/// Start an import into `repo` and run it to its end. +async fn import(srv: &TestServer, repo: &str, body: Value) -> Value { + let resp = admin( + srv, + reqwest::Method::POST, + &format!("/admin/repositories/{repo}/imports"), + Some(body), + ) + .await; + assert_eq!(resp.status(), 201, "{}", resp.text().await.unwrap()); + let id = resp.json::().await.unwrap()["id"].as_i64().unwrap(); + for _ in 0..1200 { + ImportWorker::new(srv.state.clone()).tick().await.unwrap(); + let view: Value = admin( + srv, + reqwest::Method::GET, + &format!("/admin/repositories/{repo}/imports/{id}"), + None, + ) + .await + .json() + .await + .unwrap(); + if !matches!(view["state"].as_str(), Some("queued" | "running")) { + return view; + } + tokio::time::sleep(Duration::from_millis(100)).await; + } + panic!("import into {repo} did not end"); +} + +async fn items(srv: &TestServer, repo: &str, id: &Value, state: &str) -> Vec { + admin( + srv, + reqwest::Method::GET, + &format!( + "/admin/repositories/{repo}/imports/{}/items?state={state}&limit=1000", + id.as_i64().unwrap() + ), + None, + ) + .await + .json() + .await + .unwrap() +} + +/// GET `path` of Forge's `repo`: its `Content-Type` and bytes. +async fn fetch(srv: &TestServer, repo: &str, path: &str) -> (String, Vec) { + let (u, p) = srv.admin_basic(); + let resp = srv + .client() + .get(url_of(&format!("{}/repository/{repo}", srv.base_url), path)) + .basic_auth(u, Some(p)) + .send() + .await + .unwrap(); + assert_eq!(resp.status(), 200, "GET {repo}/{path}"); + let content_type = resp.headers()["content-type"].to_str().unwrap().to_string(); + (content_type, resp.bytes().await.unwrap().to_vec()) +} + +async fn put(srv: &TestServer, repo: &str, path: &str, bytes: &[u8]) { + let (u, p) = srv.admin_basic(); + let resp = srv + .client() + .put(url_of(&format!("{}/repository/{repo}", srv.base_url), path)) + .basic_auth(u, Some(p)) + .body(bytes.to_vec()) + .send() + .await + .unwrap(); + assert_eq!(resp.status(), 201, "PUT {path}"); +} + +/// A file Forge holds: `(updated_at, last_downloaded_at, content_type)`. +async fn stored( + srv: &TestServer, + repo_id: i64, + path: &str, +) -> (DateTime, Option>, String) { + sqlx::query_as( + "SELECT updated_at, last_downloaded_at, content_type FROM raw_file WHERE repository_id = $1 AND path = $2", + ) + .bind(repo_id) + .bind(path) + .fetch_optional(&srv.pool) + .await + .unwrap() + .unwrap_or_else(|| panic!("Forge has no {path}")) +} + +async fn file_count(srv: &TestServer, repo_id: i64) -> i64 { + sqlx::query_scalar("SELECT count(*) FROM raw_file WHERE repository_id = $1") + .bind(repo_id) + .fetch_one(&srv.pool) + .await + .unwrap() +} + +// ── Tests ─────────────────────────────────────────────────────────────────── + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn raw_import_from_a_real_nexus() { + let (nexus, _container) = Nexus::start().await; + nexus.create_raw_hosted("raw-hosted").await; + let mut files: BTreeMap> = BTreeMap::new(); + files.insert("docs/readme.txt".into(), b"read me\n".to_vec()); + files.insert("docs/guides/a b \u{fc}.txt".into(), b"a guide".to_vec()); + files.insert("deep/a/b/c/data.json".into(), b"{\"a\": 1}".to_vec()); + files.insert("bin/tool-1.0.tgz".into(), b"\x1f\x8b tool".to_vec()); + for i in 0..BULK_FILES { + files.insert( + format!("bulk/file-{i:02}.bin"), + format!("bulk file {i}").into_bytes(), + ); + } + for (path, bytes) in &files { + assert_eq!(nexus.put("raw-hosted", path, bytes).await, 201, "{path}"); + } + // A segment Forge refuses (over 255 bytes), which Nexus stores. + let refused = format!("odd/{}.txt", "x".repeat(300)); + assert_eq!(nexus.put("raw-hosted", &refused, b"odd").await, 201); + // Nexus serves one file before the import: its last download is set. + nexus.get("raw-hosted", "docs/readme.txt").await; + let mut assets = BTreeMap::new(); + for _ in 0..60 { + let (listed, pages) = nexus.assets("raw-hosted").await; + assert!(pages > 1, "the listing took {pages} page"); + eprintln!( + "Nexus' asset listing: {} assets in {pages} pages", + listed.len() + ); + assets = listed; + if assets["docs/readme.txt"]["lastDownloaded"].is_string() { + break; + } + tokio::time::sleep(Duration::from_millis(500)).await; + } + assert_eq!(assets.len(), files.len() + 1); + assert!(assets["docs/readme.txt"]["lastDownloaded"].is_string()); + + let srv = StackBuilder::default().start().await; + let repo_id = raw_hosted(&srv, "raw-copy").await; + let credentials = |url: String| json!({"url": url, "username": ADMIN.0, "password": ADMIN.1}); + + // A dry run: the files, their size and the refused path; nothing is + // written. + let mut dry = credentials(nexus.repository("raw-hosted")); + dry["dry_run"] = json!(true); + let done = import(&srv, "raw-copy", dry).await; + assert_eq!(done["state"], "completed", "{done}"); + assert_eq!(done["listing"], "assets", "{done}"); + assert!( + done["source_server"] + .as_str() + .is_some_and(|s| s.starts_with("Nexus/")), + "{done}" + ); + assert_eq!(done["progress"]["planned"], files.len(), "{done}"); + assert_eq!(done["progress"]["failed"], 1, "{done}"); + let total: usize = files.values().map(Vec::len).sum(); + assert_eq!(done["progress"]["planned_size"], total, "{done}"); + assert_eq!(file_count(&srv, repo_id).await, 0); + + // The import. + let done = import( + &srv, + "raw-copy", + credentials(nexus.repository("raw-hosted")), + ) + .await; + assert_eq!(done["state"], "completed", "{done}"); + assert_eq!(done["progress"]["imported"], files.len(), "{done}"); + assert_eq!(done["progress"]["bytes_copied"], total, "{done}"); + let failed = items(&srv, "raw-copy", &done["id"], "failed").await; + assert_eq!(failed.len(), 1, "{failed:?}"); + assert_eq!(failed[0]["name"], refused.as_str()); + assert!( + failed[0]["detail"] + .as_str() + .is_some_and(|d| d.contains("longer than 255 bytes")), + "{failed:?}" + ); + let imported = items(&srv, "raw-copy", &done["id"], "imported").await; + assert!( + imported.iter().all(|i| i["uploader"] == "admin"), + "{imported:?}" + ); + + // Times first (a download here moves the last download), then bytes, + // content types and sidecars. + for path in files.keys() { + let asset = &assets[path]; + let (updated, downloaded, _) = stored(&srv, repo_id, path).await; + assert_eq!( + updated, + time(asset["blobCreated"].as_str().unwrap()), + "{path}" + ); + match asset["lastDownloaded"].as_str() { + Some(at) => assert_eq!(downloaded, Some(time(at)), "{path}"), + None => assert_recent(downloaded), + } + } + for (path, bytes) in &files { + let asset = &assets[path]; + let (content_type, served) = fetch(&srv, "raw-copy", path).await; + assert_eq!(&served, bytes, "{path}"); + let nexus_type = asset["contentType"].as_str().unwrap(); + if nexus_type != "application/octet-stream" { + assert_eq!(content_type, nexus_type, "{path}"); + } + let (_, sidecar) = fetch(&srv, "raw-copy", &format!("{path}.sha256")).await; + let nexus_sha256 = asset["checksum"]["sha256"].as_str().unwrap(); + assert!( + String::from_utf8(sidecar) + .unwrap() + .starts_with(&nexus_sha256.to_lowercase()), + "{path}" + ); + assert_eq!(nexus_sha256.to_lowercase(), sha256(bytes), "{path}"); + } + + // A second run copies only the new file; the changed one is a + // conflict, decided from Nexus' SHA-256. + assert_eq!(nexus.put("raw-hosted", "docs/new.txt", b"new").await, 201); + let rebuilt = b"\x1f\x8b tool, rebuilt".to_vec(); + assert_eq!( + nexus.put("raw-hosted", "bin/tool-1.0.tgz", &rebuilt).await, + 201 + ); + let done = import( + &srv, + "raw-copy", + credentials(nexus.repository("raw-hosted")), + ) + .await; + assert_eq!(done["state"], "completed", "{done}"); + assert_eq!(done["progress"]["imported"], 1, "{done}"); + assert_eq!(done["progress"]["conflict"], 1, "{done}"); + assert_eq!(done["progress"]["identical"], files.len() - 1, "{done}"); + assert_eq!(done["progress"]["failed"], 1, "{done}"); + assert_eq!(done["progress"]["bytes_copied"], 3, "{done}"); + assert_eq!( + fetch(&srv, "raw-copy", "bin/tool-1.0.tgz").await.1, + files["bin/tool-1.0.tgz"] + ); + + // `overwrite`, in the request form of a Nexus server and repository. + let mut legacy = credentials(nexus.url.clone()); + legacy["repository"] = json!("raw-hosted"); + legacy["on_conflict"] = json!("overwrite"); + let done = import(&srv, "raw-copy", legacy).await; + assert_eq!(done["state"], "completed", "{done}"); + assert_eq!(done["progress"]["replaced"], 1, "{done}"); + assert_eq!(done["progress"]["identical"], files.len(), "{done}"); + assert_eq!(fetch(&srv, "raw-copy", "bin/tool-1.0.tgz").await.1, rebuilt); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn raw_import_from_another_forge() { + let srv = StackBuilder::default().start().await; + let source_id = raw_hosted(&srv, "raw-source").await; + let copy_id = raw_hosted(&srv, "raw-forge-copy").await; + let mut files: BTreeMap> = BTreeMap::new(); + files.insert("README.txt".into(), b"read me".to_vec()); + files.insert("docs/a b \u{fc}.txt".into(), b"a guide".to_vec()); + files.insert("deep/a/b/c/data.json".into(), b"{\"a\": 1}".to_vec()); + for i in 0..FORGE_BULK { + files.insert( + format!("bulk/file-{i:03}.bin"), + format!("bulk file {i}").into_bytes(), + ); + } + files.insert("bulk/sub/last.txt".into(), b"after the bulk".to_vec()); + for (path, bytes) in &files { + put(&srv, "raw-source", path, bytes).await; + } + // A download in the source: its last download carries over. + fetch(&srv, "raw-source", "README.txt").await; + let mut source_downloaded = None; + for _ in 0..50 { + source_downloaded = stored(&srv, source_id, "README.txt").await.1; + if source_downloaded.is_some() { + break; + } + tokio::time::sleep(Duration::from_millis(100)).await; + } + assert!(source_downloaded.is_some()); + + // The source's times as the import finds them (its downloads from the + // source are downloads too). + let mut source_times = BTreeMap::new(); + for path in files.keys() { + source_times.insert(path.clone(), stored(&srv, source_id, path).await); + } + + let (user, token) = srv.admin_basic(); + let body = json!({ + "url": format!("{}/repository/raw-source", srv.base_url), + "username": user, + "password": token, + }); + let done = import(&srv, "raw-forge-copy", body.clone()).await; + assert_eq!(done["state"], "completed", "{done}"); + assert_eq!(done["listing"], "directories", "{done}"); + assert_eq!(done["progress"]["imported"], files.len(), "{done}"); + assert_eq!(done["progress"]["failed"], 0, "{done}"); + + for (path, source) in &source_times { + let copy = stored(&srv, copy_id, path).await; + assert_eq!(copy.0, source.0, "{path}: the upload time"); + assert_eq!(copy.2, source.2, "{path}: the content type"); + match source.1 { + Some(at) => assert_eq!(copy.1, Some(at), "{path}"), + None => assert_recent(copy.1), + } + } + for (path, bytes) in &files { + assert_eq!( + &fetch(&srv, "raw-forge-copy", path).await.1, + bytes, + "{path}" + ); + } + + // Forge lists each file's SHA-256: a second run downloads nothing. + let done = import(&srv, "raw-forge-copy", body).await; + assert_eq!(done["state"], "completed", "{done}"); + assert_eq!(done["progress"]["identical"], files.len(), "{done}"); + assert_eq!(done["progress"]["bytes_copied"], 0, "{done}"); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn raw_import_of_named_paths_from_a_web_server() { + let root = "/usr/share/nginx/html/releases/1.0"; + let tool = b"\x1f\x8b the tool".to_vec(); + let notes = b"release notes".to_vec(); + let bad = b"bad bytes".to_vec(); + let container = GenericImage::new(NGINX_IMAGE, NGINX_TAG) + .with_exposed_port(80.tcp()) + .with_wait_for(WaitFor::Nothing) + .with_copy_to(format!("{root}/tool.tgz"), tool.clone()) + .with_copy_to( + format!("{root}/tool.tgz.sha256"), + format!("{} tool.tgz\n", sha256(&tool)).into_bytes(), + ) + .with_copy_to(format!("{root}/notes en \u{fc}.txt"), notes.clone()) + .with_copy_to(format!("{root}/bad.bin"), bad.clone()) + .with_copy_to( + format!("{root}/bad.bin.sha256"), + sha256(b"other bytes").into_bytes(), + ) + // The files' times, which nginx serves as `Last-Modified`. + .with_cmd(vec![ + "sh".to_string(), + "-c".to_string(), + format!("touch -d '2024-05-06 07:08:09' {root}/* && exec nginx -g 'daemon off;'"), + ]) + .start() + .await + .expect("start nginx (is Docker running?)"); + let host = container.get_host().await.unwrap(); + let port = container.get_host_port_ipv4(80).await.unwrap(); + let url = format!("http://{host}:{port}"); + let client = reqwest::Client::new(); + for _ in 0..100 { + let up = client + .get(format!("{url}/releases/1.0/tool.tgz")) + .send() + .await + .is_ok_and(|r| r.status() == 200); + if up { + break; + } + tokio::time::sleep(Duration::from_millis(200)).await; + } + + let srv = StackBuilder::default().start().await; + let repo_id = raw_hosted(&srv, "raw-named").await; + // nginx serves no listing: the import names its paths. + let resp = admin( + &srv, + reqwest::Method::POST, + "/admin/repositories/raw-named/imports", + Some(json!({"url": url})), + ) + .await; + assert_eq!(resp.status(), 400); + let body = json!({ + "url": url, + "names": [ + "releases/1.0/tool.tgz", + "releases/1.0/notes en \u{fc}.txt", + "releases/1.0/bad.bin", + "releases/1.0/missing.txt", + ], + }); + let done = import(&srv, "raw-named", body.clone()).await; + assert_eq!(done["state"], "completed", "{done}"); + assert_eq!(done["listing"], "names", "{done}"); + assert_eq!(done["progress"]["imported"], 2, "{done}"); + assert_eq!(done["progress"]["failed"], 2, "{done}"); + let failed = items(&srv, "raw-named", &done["id"], "failed").await; + let detail = |path: &str| { + failed + .iter() + .find(|i| i["name"] == path) + .unwrap_or_else(|| panic!("{path} in {failed:?}"))["detail"] + .as_str() + .unwrap() + .to_string() + }; + assert!( + detail("releases/1.0/bad.bin").contains("its .sha256 sidecar"), + "{failed:?}" + ); + assert_eq!( + detail("releases/1.0/missing.txt"), + "the source has no file at this path" + ); + + for (path, bytes, nginx_type) in [ + ("releases/1.0/tool.tgz", &tool, "application/octet-stream"), + ("releases/1.0/notes en \u{fc}.txt", ¬es, "text/plain"), + ] { + let (updated, downloaded, _) = stored(&srv, repo_id, path).await; + assert_eq!(updated, time("2024-05-06T07:08:09Z"), "{path}"); + assert_recent(downloaded); + let (content_type, served) = fetch(&srv, "raw-named", path).await; + assert_eq!(&served, bytes, "{path}"); + if nginx_type != "application/octet-stream" { + assert_eq!(content_type, nginx_type, "{path}"); + } + } + + // A second run: the file with a sidecar is identical without a download, + // the one without is downloaded and compared. + let done = import(&srv, "raw-named", body).await; + assert_eq!(done["progress"]["identical"], 2, "{done}"); + assert_eq!(done["progress"]["bytes_copied"], notes.len(), "{done}"); +} diff --git a/crates/server/src/admin/imports.rs b/crates/server/src/admin/imports.rs index 454dd0bd..4162ca67 100644 --- a/crates/server/src/admin/imports.rs +++ b/crates/server/src/admin/imports.rs @@ -1,6 +1,6 @@ //! Imports into a hosted repository from another server ([`crate::import`]): //! a NuGet feed, an npm registry or a registry API, read through the -//! format's own protocol. **Admin-only.** +//! format's own protocol, or a raw repository. **Admin-only.** //! //! * `GET /admin/repositories/{name}/imports` — the repository's imports, //! newest first ([`ImportView`]). @@ -8,9 +8,11 @@ //! ([`CreateImport`]). The source is checked first (the **preflight**): //! a feed's service index resolves and its search answers (with how many //! packages it finds), an npm registry lists its packages (its search, -//! or `/-/all`), or a registry answers its API and serves a catalog; an -//! import that names its package ids, npm packages or scopes, or images -//! needs no search or catalog. A failed check is a 400 (502 when the source is +//! or `/-/all`), a registry answers its API and serves a catalog, or a +//! raw repository serves a listing (another Forge's JSON listing, or +//! Nexus' asset listing); an import that names its package ids, npm +//! packages or scopes, images or paths needs no search, catalog or +//! listing. A failed check is a 400 (502 when the source is //! unreachable or failing) saying what is wrong; nothing is stored. A //! repository runs one import at a time (409). `dry_run` decides every //! item without writing a package: what would be copied, and the @@ -43,14 +45,16 @@ use forge_core::auth::principal::Principal; use forge_core::net::ssrf::validate_url; use forge_core::repo::proxy::Credentials; use forge_db::import_dao::{self, NewImport, RepositoryImport}; -use forge_db::model::{RepoKind, Repository}; +use forge_db::model::{DeploymentPolicy, RepoKind, Repository}; use forge_db::{repository_dao, secret_dao}; +use forge_plugin::api::files; use serde::{Deserialize, Serialize}; use super::db_error; use crate::images::import::ImportSource; use crate::import::feed::Feed; use crate::import::npm::NpmRegistry; +use crate::import::raw::RawSource; use crate::import::{ nexus_source_url, password_secret, source_url, Listing, Protocol, SourceError, FORMATS, MAX_NAMES, NPM_PAGE, SEARCH_PAGE, @@ -97,15 +101,17 @@ fn is_unique_violation(e: &sqlx::Error) -> bool { pub struct CreateImport { /// The source: a NuGet feed's service index /// (`https://nexus.example.com/repository/nuget-hosted/index.json`), an - /// npm registry (`https://nexus.example.com/repository/npm-hosted`), or a - /// registry (`https://registry.example.com`). With `repository`, a Nexus - /// server instead (the request before imports read the format's - /// protocol). + /// npm registry (`https://nexus.example.com/repository/npm-hosted`), a + /// registry (`https://registry.example.com`), or a raw repository + /// (`https://nexus.example.com/repository/raw-hosted`, a directory of a + /// web server). With `repository`, a Nexus server instead (the request + /// before imports read the format's protocol). pub url: String, /// The Nexus repository at `url` to read (with `url` naming the Nexus /// server): its service index, `/repository//index.json`, its - /// npm registry, `/repository/`, or its registry API, - /// `registry_url` or `/repository/`. + /// npm registry, `/repository/`, its registry API, + /// `registry_url` or `/repository/`, or (raw) the repository, + /// `/repository/`. #[serde(default)] pub repository: Option, /// With `repository`: the Nexus Docker repository's connector, e.g. @@ -114,7 +120,9 @@ pub struct CreateImport { pub registry_url: Option, /// The package ids or images to copy, instead of every package the /// feed's search finds (every image the registry's catalog lists). npm: - /// package names, or scopes (`@acme`: every package of the scope). + /// package names, or scopes (`@acme`: every package of the scope). Raw: + /// the paths of the files to copy, below `url`, instead of every file + /// the source's listing names. #[serde(default)] pub names: Option>, /// The source's credentials (both, or neither for anonymous access). @@ -211,8 +219,9 @@ pub struct ImportView { /// preflight, or as many as the scan walked, whichever is more. npm: what /// the registry's search reports, or how many `/-/all` lists. pub source_total: Option, - /// What the scan walks: `search`, `catalog`, `all` (npm's `/-/all`) or - /// `names`. + /// What the scan walks: `search`, `catalog`, `all` (npm's `/-/all`), + /// `assets` (Nexus' asset listing), `directories` (another Forge's raw + /// listing) or `names`. pub listing: String, pub last_error: Option, pub created_by: Option, @@ -362,9 +371,13 @@ pub async fn create( } let images = state.formats.stores_images(&repo.format); let protocol = Protocol::of(&repo.format, images); + let raw = protocol == Protocol::Raw; let Some(format) = state.formats.get(&repo.format) else { return bad_request(format!("the {} format is not loaded", repo.format)); }; + if raw && repo.deployment_policy() == DeploymentPolicy::ReadOnly { + return bad_request("the repository is read-only: an import stores no file in it"); + } // The request itself. let url = body.url.trim().trim_end_matches('/').to_string(); @@ -407,6 +420,27 @@ pub async fn create( if name.is_empty() { continue; } + if raw { + // A path below the source's URL, as the file is stored + // in Forge. + if name.ends_with('/') { + return bad_request(format!( + "names: {name:?} is a directory; name files, or leave names out to copy every file the source lists" + )); + } + match files::normalize(&name) { + Ok(path) => name = path, + Err(e) => { + return bad_request(format!( + "names: Forge refuses the path {name:?}: {e}" + )) + } + } + if seen.insert(name.clone()) { + kept.push(name); + } + continue; + } // npm: a scope stands for every package of it. let scope = (protocol == Protocol::Npm) .then(|| forge_format_npm::import::parse_scope(&name)) @@ -447,7 +481,33 @@ pub async fn create( // The preflight: does the source speak the format's protocol, and list // what to copy? let mut listing = Listing::of(images, names.is_some()); - let (server, total) = if protocol == Protocol::Npm { + let (server, total) = if raw { + let mut from = RawSource::new( + state.upstream.clone(), + &source, + username.clone(), + password.clone(), + ); + match &names { + Some(names) => { + if let Err(e) = from.probe_named(&names[0]).await { + return preflight_error(&format!("the source at {source}"), e); + } + } + None => match from.probe().await { + Ok(Some(found)) => listing = found, + Ok(None) => { + return bad_request(format!( + "the source at {source} lists no files: it serves neither a Forge raw \ + repository's JSON listing nor, as a Nexus repository \ + (/repository/), Nexus' asset listing; name the paths to import" + )) + } + Err(e) => return preflight_error(&format!("the source at {source}"), e), + }, + } + (from.server().map(str::to_string), None) + } else if protocol == Protocol::Npm { let mut registry = NpmRegistry::new( state.upstream.clone(), &source, diff --git a/crates/server/src/files/store.rs b/crates/server/src/files/store.rs index c1e20229..5298b1f0 100644 --- a/crates/server/src/files/store.rs +++ b/crates/server/src/files/store.rs @@ -24,7 +24,7 @@ use forge_core::repo::{PackageMutation, PublishAuthorizationContext, WriteError} use forge_db::model::{DeploymentPolicy, RepoKind, Repository}; use forge_db::raw_dao::{self, NewRawFile, Redeploy, StoreOutcome}; use forge_plugin::api::files; -use forge_storage::{CasError, StoredBlob}; +use forge_storage::{CasError, StagedBlob, StoredBlob}; use futures_util::{Stream, TryStreamExt}; use md5::Md5; use serde::Serialize; @@ -48,6 +48,9 @@ pub enum Via { /// The web UI's endpoints: a signed-in session (whose CSRF token the /// principal middleware checked) or a token. Browser, + /// An import ([`crate::import`]), running as the administrator who + /// started it. + Import, } /// Uploaded content, stored in the CAS (its reference still to be taken), @@ -93,8 +96,16 @@ pub enum Content { Staged(Ingested), } -/// Stage `body` (at most `limit` bytes) and store it content-addressed. -pub async fn ingest(files: &Files<'_>, body: S, limit: u64) -> Result +/// Received content, staged but not yet stored, with the digests the CAS +/// does not keep: what an import checks before it stores it. +pub struct Staged { + pub blob: StagedBlob, + pub sha1: String, + pub md5: String, +} + +/// Stage `body` (at most `limit` bytes), hashing it on the way. +pub async fn stage(files: &Files<'_>, body: S, limit: u64) -> Result where S: Stream> + Send + Unpin, E: Into>, @@ -108,21 +119,40 @@ where d.sha1.update(chunk); d.md5.update(chunk); }); - let staged = files + let blob = files .state .cas .stage(tokio_util::io::StreamReader::new(body), limit) .await .map_err(cas_error)?; let digests = std::mem::take(&mut *digests.lock().unwrap_or_else(PoisonError::into_inner)); - let stored = files.state.cas.commit(staged).await.map_err(cas_error)?; - Ok(Ingested { - stored, + Ok(Staged { + blob, sha1: hex::encode(digests.sha1.finalize()), md5: hex::encode(digests.md5.finalize()), }) } +/// Stage `body` (at most `limit` bytes) and store it content-addressed. +pub async fn ingest(files: &Files<'_>, body: S, limit: u64) -> Result +where + S: Stream> + Send + Unpin, + E: Into>, +{ + let staged = stage(files, body, limit).await?; + let stored = files + .state + .cas + .commit(staged.blob) + .await + .map_err(cas_error)?; + Ok(Ingested { + stored, + sha1: staged.sha1, + md5: staged.md5, + }) +} + /// An ingest failure as the answer to the request. pub fn cas_error(error: CasError) -> FileError { match error { @@ -197,6 +227,7 @@ pub async fn refusal( } ) || matches!(files.principal, Principal::Workload { .. }) } + Via::Import => true, }; if !credential { return Some(FileError::CredentialRequired); @@ -551,6 +582,181 @@ pub async fn remove( Ok(true) } +// ── Imports ───────────────────────────────────────────────────────────────── + +/// A file an import ([`crate::import`]) copies from another server. +pub struct ImportedFile<'a> { + pub path: &'a str, + /// The content type the source serves it with, kept as an upload's + /// declared type is ([`content_type`]). + pub declared: Option<&'a str>, + /// Replace a file the path holds with other bytes (otherwise it is a + /// [`Imported::Conflict`]), even where the repository refuses + /// redeploys. + pub overwrite: bool, + /// When the source stored the content: the file's upload time. + pub uploaded_at: Option>, + /// When the source last served it: the file's last download. + pub downloaded_at: Option>, + /// Who uploaded it to the source, for the audit trail. + pub uploader: Option<&'a str>, + /// Where it came from (the source's URL), for the audit trail. + pub source: &'a str, +} + +/// What an import did with one file. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Imported { + /// The path was free; the file is stored. + Created, + /// The path held other bytes; they were replaced. + Replaced, + /// The path holds these bytes already; nothing was written. + Identical, + /// The path holds other bytes and was left alone. + Conflict, +} + +/// Why an import may not store a file at `path` of hosted `repo`, as the +/// administrator it runs as (`files.principal`): the repository is not +/// hosted or is read-only, or the administrator's package selectors leave +/// the path out. A dry run asks this too. +pub async fn import_refusal(files: &Files<'_>, repo: &Repository, path: &str) -> Option { + if let Err(e) = files::validate_path(path) { + return Some(FileError::BadRequest(e.to_string())); + } + if let Some(refused) = refusal(files, repo, Level::Write, Via::Import).await { + return Some(refused); + } + let write = files + .state + .authz + .package_filter_for(&files.principal, repo, Level::Write) + .await; + (!write.allows(path)).then(|| FileError::Forbidden(format!("you may not upload to {path}"))) +} + +/// Store a file an import downloaded (`staged`, checked against what the +/// source lists) in hosted `repo`, through the same rules as an upload: the +/// path rules, the selectors of the administrator the import runs as, the +/// read-only policy, deduplication in the content-addressed store, and the +/// audit trail. A path that holds the same bytes is +/// [`Imported::Identical`]; one with other bytes is replaced only with +/// [`ImportedFile::overwrite`] (and the delete level on the path), else it +/// is an [`Imported::Conflict`]. A stored file takes the source's times. +pub async fn import( + files: &Files<'_>, + repo: &Repository, + staged: Staged, + file: &ImportedFile<'_>, +) -> Result { + let path = file.path; + if let Some(refused) = import_refusal(files, repo, path).await { + return Err(refused); + } + let authz = &files.state.authz; + let may_replace = authz.can_delete(&files.principal, repo).await + && authz + .package_filter_for(&files.principal, repo, Level::Delete) + .await + .allows(path); + // Nothing reaches the object store for a path that keeps what it holds. + if let Some(existing) = raw_dao::find(&files.state.pool, repo.id, path).await? { + if existing.blob_sha256 == staged.blob.sha256 { + return Ok(Imported::Identical); + } + if !file.overwrite { + return Ok(Imported::Conflict); + } + if !may_replace { + return Err(FileError::Forbidden(format!( + "{path} exists; replacing a file needs the delete permission" + ))); + } + } + let stored = files + .state + .cas + .commit(staged.blob) + .await + .map_err(cas_error)?; + let content_type = content_type(file.declared, path); + let new = NewRawFile { + path, + sha256: &stored.sha256, + size: i64::try_from(stored.size).map_err(|_| FileError::TooLarge(stored.size))?, + content_type: &content_type, + sha512_base64: &stored.sha512_base64, + sha1: &staged.sha1, + md5: &staged.md5, + upstream_etag: None, + upstream_last_modified: None, + checked: false, + }; + let redeploy = if file.overwrite && may_replace { + Redeploy::Allow + } else { + Redeploy::Deny + }; + let grace = files.state.config.outbox.grace(); + let mut tx = forge_db::begin(&files.state.pool).await?; + let previous = raw_dao::lock_updated_at(&mut tx, repo.id, path).await?; + let imported = + match raw_dao::store(&mut tx, repo.id, &new, stored.ingest_token, redeploy, grace).await? { + StoreOutcome::Created(id) => { + raw_dao::set_imported_times( + &mut tx, + id, + file.uploaded_at, + file.downloaded_at, + None, + ) + .await?; + Imported::Created + } + StoreOutcome::Replaced(id) => { + raw_dao::set_imported_times( + &mut tx, + id, + file.uploaded_at, + file.downloaded_at, + previous, + ) + .await?; + Imported::Replaced + } + // Stored meanwhile, by an upload or another import. + StoreOutcome::Unchanged(_) => Imported::Identical, + StoreOutcome::Exists => Imported::Conflict, + }; + if matches!(imported, Imported::Created | Imported::Replaced) { + record( + files, + repo, + PackageMutation::Publish, + path, + &stored.sha256, + imported == Imported::Replaced, + &mut tx, + ) + .await?; + } + tx.commit().await?; + tracing::info!( + target: "forge::audit", + action = "import", + repo = %repo.name, + path, + sha256 = %stored.sha256, + outcome = ?imported, + principal = %files.principal.audit_label(), + source = %file.source, + uploader = file.uploader.unwrap_or(""), + "file imported" + ); + Ok(imported) +} + #[cfg(test)] mod tests { use super::*; diff --git a/crates/server/src/import/mod.rs b/crates/server/src/import/mod.rs index 12286bef..a51f0233 100644 --- a/crates/server/src/import/mod.rs +++ b/crates/server/src/import/mod.rs @@ -1,8 +1,8 @@ //! Importing a repository from another server into a Forge hosted -//! repository of the same format (NuGet, npm, Docker), one repository at a -//! time. The source is read through the format's own protocol, so it may be -//! any server that speaks it: Nexus, Artifactory, ProGet, BaGet, Verdaccio, -//! Harbor, a `distribution` registry or another Forge. +//! repository of the same format (NuGet, npm, Docker, raw), one repository +//! at a time. The source is read through the format's own protocol, so it +//! may be any server that speaks it: Nexus, Artifactory, ProGet, BaGet, +//! Verdaccio, Harbor, a `distribution` registry or another Forge. //! //! * **NuGet:** a feed's service index (`…/index.json`, [`feed::Feed`]): //! its search names the package ids, its package base address lists each @@ -18,13 +18,17 @@ //! * **Docker:** a registry's API ([`ImportSource`]): its catalog //! (`/v2/_catalog`) names the images, their tag lists the tags, and the //! referrers API the signatures and SBOMs of each copied manifest. +//! * **Raw:** a raw repository ([`raw::RawSource`]), which has no protocol +//! beyond a `GET` of a path: Nexus' asset listing, or another Forge's JSON +//! listing, names its files with their checksums and times. //! -//! An import can name its package ids or images instead (`names`): a feed's -//! search may miss some (Nexus stops at 10,000 packages; a package whose -//! versions are all unlisted is in no search result), Docker Hub, GHCR and -//! ECR serve no catalog, and an npm registry may serve neither a search +//! An import can name its package ids, images or paths instead (`names`): a +//! feed's search may miss some (Nexus stops at 10,000 packages; a package +//! whose versions are all unlisted is in no search result), Docker Hub, GHCR +//! and ECR serve no catalog, an npm registry may serve neither a search //! that lists every package nor `/-/all` (an npm import names scopes too, -//! `@acme`). +//! `@acme`), and a plain web server lists no files (a named file is checked +//! against its `.sha256` sidecar, where it has one). //! //! An administrator starts an import on a hosted repository //! ([`crate::admin::imports`]); this worker runs it, on one instance at a @@ -60,6 +64,15 @@ //! is no publish time). The protocols carry no last download, so an //! imported version counts as downloaded at its import, and no push //! time for a tag, so a tag is created at its import; +//! * a raw file is downloaded from the import's URL and its path, +//! verified against its size and the checksums the source lists, and +//! stored through the file store ([`crate::files::store::import`]): +//! the path rules, the administrator's selectors, the read-only policy, +//! CAS dedup and the audit trail. Its upload time and last download are +//! the source's (Nexus' `blobCreated` and `lastDownloaded`, another +//! Forge's `last_modified` and `last_downloaded`, a web server's +//! `Last-Modified`), its last download the import's where the source +//! names none. A path Forge refuses fails as its item; //! * a version Forge holds with other bytes is a conflict: skipped and //! reported (`skip`), replaced (`overwrite`), or it fails the import //! (`fail`); @@ -87,6 +100,7 @@ pub mod feed; pub mod npm; +pub mod raw; use std::collections::HashMap; use std::panic::AssertUnwindSafe; @@ -94,6 +108,7 @@ use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use std::sync::Arc; use std::time::Duration; +use base64::Engine as _; use chrono::{DateTime, Datelike, Utc}; use forge_core::auth::principal::{AuthMethod, Principal}; use forge_core::authz::{Level, PackageFilter}; @@ -103,7 +118,8 @@ use forge_core::repo::{ImportOptions, ImportOutcome, RepoKind, WriteError}; use forge_crypto::PlaintextSecret; use forge_db::import_dao::{self, ImportItem, NewItem, RepositoryImport}; use forge_db::model::Repository; -use forge_db::{component_dao, repository_dao, secret_dao, tag_dao, user_dao}; +use forge_db::{component_dao, raw_dao, repository_dao, secret_dao, tag_dao, user_dao}; +use forge_plugin::api::files; use forge_plugin::api::images::ImageError; use forge_plugin::api::VersionMetadata; use forge_storage::StagedBlob; @@ -114,15 +130,18 @@ use serde::{Deserialize, Serialize}; use tokio::task::JoinHandle; use uuid::Uuid; +use crate::files::store::{self as file_store, Imported, ImportedFile}; +use crate::files::Files; use crate::images::import::{ImageImportOptions, ImageOutcome, ImportSource}; use crate::images::{Failure, Registry}; use crate::state::AppState; use feed::Feed; use npm::NpmRegistry; +use raw::RawSource; -/// The formats an import copies: NuGet and npm packages, and Docker images -/// into the image store. -pub const FORMATS: &[&str] = &["nuget", "npm", "docker"]; +/// The formats an import copies: NuGet and npm packages, Docker images into +/// the image store, and raw files into the file store. +pub const FORMATS: &[&str] = &["nuget", "npm", "docker", "raw"]; /// How often an idle instance looks for an import to run. const POLL_INTERVAL: Duration = Duration::from_secs(5); @@ -216,7 +235,11 @@ pub enum Listing { Catalog, /// An npm registry's legacy listing of every package, `/-/all`. All, - /// The package ids or images the import names. + /// Raw: Nexus' asset listing (`/service/rest/v1/assets`). + Assets, + /// Raw: another Forge's JSON listing, a directory at a time. + Directories, + /// The package ids, images or paths the import names. Names, } @@ -235,6 +258,8 @@ impl Listing { Listing::Search => "search", Listing::Catalog => "catalog", Listing::All => "all", + Listing::Assets => "assets", + Listing::Directories => "directories", Listing::Names => "names", } } @@ -243,6 +268,8 @@ impl Listing { match s { "catalog" => Listing::Catalog, "all" => Listing::All, + "assets" => Listing::Assets, + "directories" => Listing::Directories, "names" => Listing::Names, _ => Listing::Search, } @@ -258,6 +285,8 @@ pub enum Protocol { Npm, /// A registry API (Docker / OCI images). Images, + /// A raw repository: its files at their paths, and a listing of them. + Raw, } impl Protocol { @@ -267,6 +296,7 @@ impl Protocol { match format { _ if images => Protocol::Images, "npm" => Protocol::Npm, + "raw" => Protocol::Raw, _ => Protocol::NuGet, } } @@ -274,9 +304,9 @@ impl Protocol { /// Where in a Nexus server a repository is: `/repository/`. /// Nexus serves a NuGet repository's service index below it, an npm -/// repository's registry at it, and a Docker repository's registry API at -/// it (for reads, on its own port), unless the repository has a connector -/// of its own (`registry_url`). +/// repository's registry and a raw repository's files at it, and a Docker +/// repository's registry API at it (for reads, on its own port), unless the +/// repository has a connector of its own (`registry_url`). pub fn nexus_source_url( url: &str, repository: &str, @@ -286,7 +316,7 @@ pub fn nexus_source_url( let path = format!("{}/repository/{repository}", url.trim_end_matches('/')); match (protocol, registry_url) { (Protocol::Images, Some(registry_url)) => registry_url.to_string(), - (Protocol::Images | Protocol::Npm, _) => path, + (Protocol::Images | Protocol::Npm | Protocol::Raw, _) => path, (Protocol::NuGet, _) => format!("{path}/index.json"), } } @@ -435,6 +465,8 @@ enum Target { registry: Registry, source: ImportSource, }, + /// Raw: from a raw repository, through the file store. + Files { source: Box }, } /// What a run works with. @@ -452,7 +484,8 @@ struct Ctx { /// `source`). An item an import stored before reads as none of it. #[derive(Debug, Default, Serialize, Deserialize)] struct ItemSource { - /// The package's SHA-512, base64, where the feed lists it. + /// The package's SHA-512, base64, where the feed lists it (raw: where + /// Nexus lists it). #[serde(default, skip_serializing_if = "Option::is_none")] sha512: Option, /// npm: `dist.integrity` (an SRI string) and `dist.shasum` (hex SHA-1), @@ -470,31 +503,69 @@ struct ItemSource { /// npm, a `tag` item: the version the dist-tag points at. #[serde(default, skip_serializing_if = "Option::is_none")] target: Option, + /// Raw: the file's SHA-256 and SHA-1 (hex), where the source lists them. + #[serde(default, skip_serializing_if = "Option::is_none")] + sha256: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + sha1: Option, + /// Raw: the SHA-256 is the one its `.sha256` sidecar holds. + #[serde(default, skip_serializing_if = "std::ops::Not::not")] + sidecar: bool, + /// Raw: the content type the source lists. + #[serde(default, skip_serializing_if = "Option::is_none")] + content_type: Option, + /// Raw: why the item fails without being copied (a path Forge + /// refuses). + #[serde(default, skip_serializing_if = "Option::is_none")] + error: Option, + /// Raw: a named file the source did not have when it was scanned. + #[serde(default, skip_serializing_if = "std::ops::Not::not")] + missing: bool, } -/// A version or tag the scan found. +/// A version, tag or file the scan found. struct Found { - /// `component`, or `tag` (an npm dist-tag). + /// `component`, `tag` (an npm dist-tag) or `file` (raw). kind: &'static str, name: String, - /// Empty for a named package or image the source does not have. + /// Empty for a named package or image the source does not have, and + /// for a file. version: String, + /// The size the source lists (raw). + size: Option, published_at: Option>, + /// When the source last served it (raw). + downloaded_at: Option>, + /// Who uploaded it to the source (raw, from Nexus). + uploader: Option, source: ItemSource, } impl Found { - /// A named package or image the source does not have: an item that - /// fails, so the import reports it. - fn missing(name: &str) -> Found { + /// A package version or image tag. + fn component( + name: &str, + version: String, + published_at: Option>, + source: ItemSource, + ) -> Found { Found { kind: "component", name: name.to_string(), - version: String::new(), - published_at: None, - source: ItemSource::default(), + version, + size: None, + published_at, + downloaded_at: None, + uploader: None, + source, } } + + /// A named package or image the source does not have: an item that + /// fails, so the import reports it. + fn missing(name: &str) -> Found { + Found::component(name, String::new(), None, ItemSource::default()) + } } /// One page of the scan: what it found, and the cursor of the next page @@ -573,7 +644,7 @@ impl Run<'_> { .ok_or_else(|| End::Failed("the repository was deleted".into()))?; if repo.kind() != RepoKind::Hosted || !FORMATS.contains(&repo.format.as_str()) { return Err(End::Failed(format!( - "imports go to a hosted NuGet, npm or Docker repository; {} is a {} {} repository", + "imports go to a hosted NuGet, npm, Docker or raw repository; {} is a {} {} repository", repo.name, repo.format, repo.kind ))); } @@ -609,7 +680,16 @@ impl Run<'_> { let images = self.state.formats.stores_images(&repo.format); let protocol = Protocol::of(&repo.format, images); let url = source_url(import, protocol); - let target = if images { + let target = if protocol == Protocol::Raw { + Target::Files { + source: Box::new(RawSource::new( + self.state.upstream.clone(), + &url, + import.username.clone(), + password, + )), + } + } else if images { let credentials = match (&import.username, password) { (Some(username), Some(password)) => { Some(Credentials::for_url(&url, username.clone(), password)) @@ -750,10 +830,10 @@ impl Run<'_> { kind: found.kind, name: &found.name, version: &found.version, - size: None, + size: found.size, published_at: found.published_at, - downloaded_at: None, - uploader: None, + downloaded_at: found.downloaded_at, + uploader: found.uploader.as_deref(), source, }) .collect(); @@ -916,6 +996,50 @@ impl Run<'_> { walked: None, }) } + (Target::Files { source }, Listing::Assets) => { + let page = source.assets(cursor).await?; + if page.next.is_some() && page.next.as_deref() == cursor { + return Err(SourceError::Invalid(format!( + "Nexus' asset listing answers the continuationToken {} again", + cursor.unwrap_or_default() + ))); + } + Ok(Page { + found: page.files.into_iter().map(listed_file).collect(), + next: page.next, + walked: None, + }) + } + (Target::Files { source }, Listing::Directories) => { + let walk = raw::Walk::parse(cursor)?; + let page = source.directory(&walk.dir, walk.after.as_deref()).await?; + let step = walk.step(page); + if step.next.as_ref() == Some(&walk) { + return Err(SourceError::Invalid(format!( + "the listing of {:?} answers the same page after {:?} again", + walk.dir, + walk.after.unwrap_or_default() + ))); + } + let mut found: Vec = step.files.into_iter().map(listed_file).collect(); + found.extend(step.refused.into_iter().map(|(dir, reason)| { + let error = format!( + "Forge refuses the directory ({reason}): nothing below it is copied" + ); + file_item( + &dir, + ItemSource { + error: Some(error), + ..ItemSource::default() + }, + ) + })); + Ok(Page { + found, + next: step.next.map(|walk| walk.cursor()), + walked: None, + }) + } _ => Err(SourceError::Invalid(format!( "a {} repository's import does not walk a {}", ctx.repo.format, @@ -946,9 +1070,10 @@ impl Run<'_> { Ok(pages.into_iter().flatten().collect()) } - /// The versions of the package id `name`, or the tags of the image - /// `name`; for npm, the versions and dist-tags of the package `name`, or - /// of every package of the scope `name` (`@acme`). + /// The versions of the package id `name`, the tags of the image `name`, + /// or the file at the path `name`; for npm, the versions and dist-tags + /// of the package `name`, or of every package of the scope `name` + /// (`@acme`). async fn expand_one( &self, ctx: &Ctx, @@ -959,6 +1084,7 @@ impl Run<'_> { match &ctx.target { Target::Packages { feed, .. } => package_versions(feed, name, named).await, Target::Images { registry, source } => image_tags(registry, source, name, named).await, + Target::Files { source } => Ok(vec![named_file(source, name).await?]), Target::Npm { registry, .. } => { let Some(scope) = forge_format_npm::import::parse_scope(name) else { return npm_package(registry, name, named).await; @@ -1019,7 +1145,7 @@ impl Run<'_> { async fn copy(&self, ctx: &Ctx) -> Result<(), End> { let failures = AtomicUsize::new(0); let concurrency = self.import.concurrency.clamp(1, 16) as usize; - for kind in ["component", "manifest", "tag"] { + for kind in ["component", "manifest", "tag", "file"] { let mut after = 0; loop { self.check().await?; @@ -1080,47 +1206,50 @@ impl Run<'_> { return Ok(Recorded::NotStarted); } let pool = &self.state.pool; - let copied = if item.version.is_empty() { - Err(match ctx.target { - Target::Packages { .. } => "the source has no package of this id".to_string(), - Target::Npm { .. } - if forge_format_npm::import::parse_scope(&item.name).is_some() => - { - "the source has no package of this scope".to_string() - } - Target::Npm { .. } => "the source has no package of this name".to_string(), - Target::Images { .. } => "the source has no tag of this image".to_string(), - }) - } else { - let source: ItemSource = - serde_json::from_value(item.source.clone()).unwrap_or_default(); - match &ctx.target { - Target::Packages { feed, writer } => { - self.copy_package(ctx, PackageSource::Feed(feed), writer, &item, &source) - .await - } - Target::Npm { filter, .. } if item.kind == "tag" => { - self.copy_dist_tag(ctx, filter, &item, &source).await - } - Target::Npm { - registry, - writer, - filter, - } => { - let copied = self - .copy_package(ctx, PackageSource::Npm(registry), writer, &item, &source) - .await; - match copied { - Ok(copied) => { - self.carry_deprecation(ctx, filter, &item, &source, copied) - .await - } - Err(e) => Err(e), + let source: ItemSource = serde_json::from_value(item.source.clone()).unwrap_or_default(); + let copied = match &ctx.target { + // A file's item has no version. + Target::Files { source } => self.copy_file(ctx, source, &item).await, + Target::Packages { .. } if item.version.is_empty() => { + Err("the source has no package of this id".to_string()) + } + Target::Npm { .. } + if item.version.is_empty() + && forge_format_npm::import::parse_scope(&item.name).is_some() => + { + Err("the source has no package of this scope".to_string()) + } + Target::Npm { .. } if item.version.is_empty() => { + Err("the source has no package of this name".to_string()) + } + Target::Images { .. } if item.version.is_empty() => { + Err("the source has no tag of this image".to_string()) + } + Target::Packages { feed, writer } => { + self.copy_package(ctx, PackageSource::Feed(feed), writer, &item, &source) + .await + } + Target::Npm { filter, .. } if item.kind == "tag" => { + self.copy_dist_tag(ctx, filter, &item, &source).await + } + Target::Npm { + registry, + writer, + filter, + } => { + let copied = self + .copy_package(ctx, PackageSource::Npm(registry), writer, &item, &source) + .await; + match copied { + Ok(copied) => { + self.carry_deprecation(ctx, filter, &item, &source, copied) + .await } + Err(e) => Err(e), } - Target::Images { registry, source } => { - self.copy_image(ctx, registry, source, &item).await - } + } + Target::Images { registry, source } => { + self.copy_image(ctx, registry, source, &item).await } }; match copied { @@ -1143,9 +1272,8 @@ impl Run<'_> { .await?; if state == "conflict" && self.import.on_conflict == "fail" { return Ok(Recorded::Conflict(format!( - "{} {}: {}", - item.name, - item.version, + "{}: {}", + item_label(&item), detail.unwrap_or_default() ))); } @@ -1154,10 +1282,7 @@ impl Run<'_> { Err(error) => { tracing::warn!(import = self.import.id, name = %item.name, version = %item.version, %error, "import item failed"); import_dao::finish_item(pool, item.id, "failed", Some(&error), 0).await?; - Ok(Recorded::Failed(format!( - "{} {}: {error}", - item.name, item.version - ))) + Ok(Recorded::Failed(format!("{}: {error}", item_label(&item)))) } } } @@ -1458,6 +1583,120 @@ impl Run<'_> { }) } + /// Copy a raw file through the file store. A file Forge holds with the + /// SHA-256 the source lists is `identical` without a download, and one + /// it holds with another SHA-256 a conflict (unless the import + /// overwrites); a dry run decides from the listing alone. Otherwise the + /// file is downloaded from the import's URL and its path, checked + /// against its size and every checksum listed, and stored. + async fn copy_file( + &self, + ctx: &Ctx, + source: &RawSource, + item: &ImportItem, + ) -> Result { + let listed: ItemSource = serde_json::from_value(item.source.clone()).unwrap_or_default(); + // A path Forge refuses, or a file the source does not have, says + // nothing of the source: it fails as its item and does not count + // towards the failures in a row that stop an import. + if let Some(error) = listed.error { + return Ok(copied("failed", Some(error), 0)); + } + let path = item.name.as_str(); + let files = Files { + state: self.state, + principal: ctx.principal.clone(), + }; + if let Some(refused) = file_store::import_refusal(&files, &ctx.repo, path).await { + return Err(refused.message()); + } + if let Some(size) = item.size.filter(|&size| size as u64 > ctx.limit) { + return Err(format!( + "the file is {size} bytes, more than the repository's upload limit of {} bytes", + ctx.limit + )); + } + let overwrite = self.import.on_conflict == "overwrite"; + let stored = raw_dao::find(&self.state.pool, ctx.repo.id, path) + .await + .map_err(|e| format!("database error: {e}"))?; + if let (Some(stored), Some(expected)) = (&stored, &listed.sha256) { + if stored.blob_sha256 == *expected { + return Ok(copied("identical", None, 0)); + } + if !overwrite { + return Ok(file_conflict()); + } + } + if self.import.dry_run { + return Ok(match (&stored, &listed.sha256) { + _ if listed.missing => copied("failed", Some(NO_FILE.into()), 0), + (None, _) => copied("planned", None, 0), + (Some(_), None) => copied( + "planned", + Some("Forge has this file; the source lists no SHA-256 to compare".into()), + 0, + ), + (Some(_), Some(_)) => copied( + "planned", + Some("would replace the file Forge has with other bytes".into()), + 0, + ), + }); + } + + let url = source.file_url(path); + let Some(response) = source.download(path).await.map_err(|e| e.to_string())? else { + return Ok(copied("failed", Some(NO_FILE.into()), 0)); + }; + if response.content_length().is_some_and(|l| l > ctx.limit) { + return Err(WriteError::TooLarge { limit: ctx.limit }.to_string()); + } + let declared = listed.content_type.clone().or_else(|| { + response + .headers() + .get(reqwest::header::CONTENT_TYPE) + .and_then(|v| v.to_str().ok()) + .map(str::to_string) + }); + let body = response.bytes_stream().map_err(std::io::Error::other); + let staged = file_store::stage(&files, body, ctx.limit) + .await + .map_err(|e| e.message())?; + verify_file(&staged, item.size, &listed, &url)?; + let bytes = staged.blob.size; + let imported = file_store::import( + &files, + &ctx.repo, + staged, + &ImportedFile { + path, + declared: declared.as_deref(), + overwrite, + uploaded_at: item.published_at, + downloaded_at: Some(item.downloaded_at.unwrap_or_else(Utc::now)), + uploader: item.uploader.as_deref(), + source: &ctx.source, + }, + ) + .await + .map_err(|e| e.message())?; + Ok(match imported { + Imported::Created => copied("imported", None, bytes), + Imported::Replaced => copied( + "replaced", + Some("replaced the file Forge had with other bytes".into()), + bytes, + ), + Imported::Identical => copied("identical", None, bytes), + Imported::Conflict => copied( + "conflict", + Some("Forge has other bytes at this path".into()), + bytes, + ), + }) + } + /// Copy a Docker tag (`component`) or referrer (`manifest`) through the /// registry, and record the referrers the source lists for it. async fn copy_image( @@ -1570,6 +1809,103 @@ impl Run<'_> { } } +/// The file at the named `path`: what a `HEAD` says of it, and the SHA-256 +/// of its `.sha256` sidecar, where the source serves one. A file the source +/// does not have is found as one that fails. +async fn named_file(source: &RawSource, path: &str) -> Result { + let Some(head) = source.head(path).await? else { + return Ok(file_item( + path, + ItemSource { + missing: true, + ..ItemSource::default() + }, + )); + }; + let sha256 = source.sidecar(path).await?; + Ok(listed_file(raw::Listed { + path: path.to_string(), + size: head.size, + content_type: head.content_type, + sidecar: sha256.is_some(), + sha256, + created: head.last_modified, + ..raw::Listed::default() + })) +} + +/// A file a raw source lists, as an item: a path Forge refuses fails as its +/// item, with the reason; a time at the epoch (a file unpacked without its +/// times) is none; Nexus' SHA-512 is kept in base64, as the store keeps it. +fn listed_file(listed: raw::Listed) -> Found { + let error = files::validate_path(&listed.path).err().map(|e| { + let reason = e.to_string(); + format!( + "Forge refuses the path: {}", + reason.trim_start_matches("invalid path: ") + ) + }); + let after_epoch = |t: &DateTime| t.timestamp() > 86_400; + Found { + kind: "file", + name: item_path(&listed.path), + version: String::new(), + size: listed.size.and_then(|s| i64::try_from(s).ok()), + published_at: listed.created.filter(after_epoch), + downloaded_at: listed.downloaded.filter(after_epoch), + uploader: listed.uploader, + source: ItemSource { + sha512: listed + .sha512 + .and_then(|hex| hex::decode(hex).ok()) + .map(|bytes| base64::engine::general_purpose::STANDARD.encode(bytes)), + sha256: listed.sha256, + sha1: listed.sha1, + sidecar: listed.sidecar, + content_type: listed.content_type, + error, + ..ItemSource::default() + }, + } +} + +/// A file (or directory) of a raw source the source lists nothing of: one +/// that fails (its `error`), or a named file it did not have (`missing`). +fn file_item(path: &str, source: ItemSource) -> Found { + Found { + kind: "file", + name: item_path(path), + version: String::new(), + size: None, + published_at: None, + downloaded_at: None, + uploader: None, + source, + } +} + +/// A source's path as an item names it: as it is, unless Forge refuses it +/// (and the item fails): then its control characters escaped (a database +/// text holds no NUL) and cut to the longest path Forge stores. +fn item_path(path: &str) -> String { + if files::validate_path(path).is_ok() { + return path.to_string(); + } + let mut out = String::new(); + for c in path.chars() { + if out.len() >= files::MAX_PATH_BYTES { + out.push('…'); + break; + } + if c.is_control() { + out.extend(c.escape_default()); + } else { + out.push(c); + } + } + out +} + /// Every version the feed lists of the package `id`, with what its /// registration says of each. A package the feed does not know is left /// out, or found as missing when the import `named` it. @@ -1600,16 +1936,15 @@ async fn package_versions(feed: &Feed, id: &str, named: bool) -> Result = Vec::with_capacity(package.releases.len()); for release in &package.releases { - found.push(Found { - kind: "component", - name: package.name.clone(), - version: release.version.clone(), - published_at: release.published, - source: ItemSource { + found.push(Found::component( + &package.name, + release.version.clone(), + release.published, + ItemSource { sha512: release.sha512(), integrity: release.integrity.clone(), shasum: release.shasum.clone(), tarball: registry.tarball_url(&package.name, &release.file), deprecated: release.deprecated.clone(), - target: None, + ..ItemSource::default() }, - }); + )); } for (tag, version) in &package.dist_tags { found.push(Found { kind: "tag", - name: package.name.clone(), - version: tag.clone(), - published_at: None, - source: ItemSource { - target: Some(version.clone()), - ..ItemSource::default() - }, + ..Found::component( + &package.name, + tag.clone(), + None, + ItemSource { + target: Some(version.clone()), + ..ItemSource::default() + }, + ) }); } Ok(found) @@ -1687,13 +2023,11 @@ async fn image_tags( }; let more = page.more && !page.tags.is_empty(); last = page.tags.last().cloned().or(last); - found.extend(page.tags.into_iter().map(|tag| Found { - kind: "component", - name: image.to_string(), - version: tag, - published_at: None, - source: ItemSource::default(), - })); + found.extend( + page.tags + .into_iter() + .map(|tag| Found::component(image, tag, None, ItemSource::default())), + ); if !more { break; } @@ -1729,6 +2063,67 @@ fn conflict_item() -> Copied { ) } +/// An item as an error names it: a package version or image tag, or a +/// file's path. +fn item_label(item: &ImportItem) -> String { + if item.version.is_empty() { + item.name.clone() + } else { + format!("{} {}", item.name, item.version) + } +} + +/// Why a file the source does not have fails. +const NO_FILE: &str = "the source has no file at this path"; + +fn file_conflict() -> Copied { + copied( + "conflict", + Some("Forge has other bytes at this path".into()), + 0, + ) +} + +/// Check a downloaded file against the size and every checksum the source +/// lists for it (Nexus: SHA-512, SHA-256 and SHA-1; another Forge: SHA-256; +/// a named file: its `.sha256` sidecar). +fn verify_file( + staged: &file_store::Staged, + size: Option, + listed: &ItemSource, + url: &str, +) -> Result<(), String> { + let blob = &staged.blob; + if let Some(size) = size.filter(|&size| u64::try_from(size).ok() != Some(blob.size)) { + return Err(format!( + "the download of {url} is {} bytes; the source lists {size}", + blob.size + )); + } + let checks = [ + ( + "SHA-512", + listed.sha512.as_deref(), + blob.sha512_base64.as_str(), + ), + ("SHA-256", listed.sha256.as_deref(), blob.sha256.as_str()), + ("SHA-1", listed.sha1.as_deref(), staged.sha1.as_str()), + ]; + for (name, expected, actual) in checks { + if expected.is_some_and(|expected| expected != actual) { + let whose = if name == "SHA-256" && listed.sidecar { + "its .sha256 sidecar holds" + } else { + "the source lists for it" + }; + return Err(format!( + "the download of {url} does not match the {name} {whose}" + )); + } + } + Ok(()) +} + /// Check a staged download against the SHA-512 the source (a `lister`, the /// feed or the registry) lists for it, or, for an npm tarball listed without /// one, against its `dist.integrity` and `dist.shasum`. @@ -1897,6 +2292,16 @@ mod tests { assert_eq!(Protocol::of("npm", false), Protocol::Npm); assert_eq!(Protocol::of("nuget", false), Protocol::NuGet); assert_eq!(Protocol::of("docker", true), Protocol::Images); + assert_eq!(Protocol::of("raw", false), Protocol::Raw); + assert_eq!( + nexus_source_url( + "https://nexus.example.com/", + "raw-hosted", + None, + Protocol::Raw + ), + "https://nexus.example.com/repository/raw-hosted" + ); } #[test] @@ -1908,12 +2313,78 @@ mod tests { Listing::Search, Listing::Catalog, Listing::All, + Listing::Assets, + Listing::Directories, Listing::Names, ] { assert_eq!(Listing::parse(listing.as_str()), listing); } } + #[test] + fn a_listed_file_keeps_what_the_source_says_and_fails_on_a_path_forge_refuses() { + let sha512 = "ab".repeat(64); + let created: DateTime = "2024-05-06T07:08:09.123Z".parse().unwrap(); + let found = listed_file(raw::Listed { + path: "docs/a bü.txt".into(), + size: Some(12), + content_type: Some("text/plain".into()), + sha512: Some(sha512.clone()), + sha256: Some("cd".repeat(32)), + created: Some(created), + downloaded: Some(DateTime::UNIX_EPOCH), + uploader: Some("admin".into()), + ..raw::Listed::default() + }); + assert_eq!(found.kind, "file"); + assert_eq!(found.name, "docs/a bü.txt"); + assert_eq!(found.version, ""); + assert_eq!(found.size, Some(12)); + assert_eq!(found.published_at, Some(created)); + // A time at the epoch is none. + assert_eq!(found.downloaded_at, None); + assert_eq!(found.uploader.as_deref(), Some("admin")); + assert_eq!(found.source.error, None); + // Nexus' hex SHA-512 is kept in base64, as the store keeps it. + assert_eq!( + found.source.sha512.as_deref(), + Some( + base64::engine::general_purpose::STANDARD + .encode(hex::decode(&sha512).unwrap()) + .as_str() + ) + ); + let stored = serde_json::to_value(&found.source).unwrap(); + assert_eq!(stored["content_type"], "text/plain"); + assert!(stored.get("missing").is_none() && stored.get("sidecar").is_none()); + + for (path, reason) in [ + ("a//b", "empty segment"), + ("a/../b", "`..` segments"), + ("a\\b", "backslashes"), + ("a/b\u{7}c", "control characters"), + (&format!("a/{}", "x".repeat(256)), "longer than 255 bytes"), + (&"abc/".repeat(300), "longer than 1024 bytes"), + ] { + let found = listed_file(raw::Listed { + path: path.to_string(), + ..raw::Listed::default() + }); + let error = found.source.error.unwrap_or_default(); + assert!(error.starts_with("Forge refuses the path: "), "{error}"); + assert!(error.contains(reason), "{path}: {error}"); + } + } + + #[test] + fn a_refused_path_is_named_printably() { + assert_eq!(item_path("a/b c"), "a/b c"); + assert_eq!(item_path("a/b\0c\n"), "a/b\\u{0}c\\n"); + let long = item_path(&"ü".repeat(2000)); + assert!(long.ends_with('…')); + assert!(long.len() <= files::MAX_PATH_BYTES + 4); + } + #[test] fn an_item_stored_before_reads_as_no_checksum() { let legacy = serde_json::json!({ diff --git a/crates/server/src/import/raw.rs b/crates/server/src/import/raw.rs new file mode 100644 index 00000000..ab93ac49 --- /dev/null +++ b/crates/server/src/import/raw.rs @@ -0,0 +1,885 @@ +//! A raw repository as an import's source. Raw has no protocol beyond a +//! `GET` of a path and no standard listing, so the import reads one of +//! these: +//! +//! * **Nexus' asset listing**, `GET /service/rest/v1/assets?repository=`, +//! paged with `continuationToken` ([`RawSource::assets`]): each file's +//! path, size, content type, SHA-1, SHA-256 and SHA-512, when its blob was +//! created, when it was last downloaded, and who uploaded it. It is the +//! only listing of a Nexus raw repository with checksums (its HTML browse +//! pages carry none), so for raw the import reads Nexus' REST API. The +//! source is the repository, `/repository/`. +//! * **Another Forge's listing**, `GET /` as JSON, paged with +//! `after` ([`RawSource::directory`]): each file's size, content type, +//! SHA-256, upload time and last download, one directory at a time +//! ([`Walk`]). +//! * **The paths the import names**, for a server with no listing (a plain +//! web server): a `HEAD` says the file's size, type and `Last-Modified` +//! ([`RawSource::head`]), and its `.sha256` sidecar, where the server +//! serves one, its SHA-256 ([`RawSource::sidecar`]). +//! +//! A file is downloaded from the import's URL and its path +//! ([`RawSource::file_url`]), never from Nexus' `downloadUrl`, which carries +//! the base URL Nexus is configured with. Requests go through the shared +//! [`UpstreamHttp`] client (timeouts, size caps, no cloud-metadata targets), +//! with the credentials sent to the source's origin only. + +use chrono::{DateTime, Utc}; +use forge_core::net::upstream::{Origin, RequestClass, SpooledBody, UpstreamAuth, UpstreamHttp}; +use forge_crypto::{ExposeSecret, PlaintextSecret}; +use forge_plugin::api::files; +use reqwest::StatusCode; +use serde::de::{Deserializer, MapAccess, Visitor}; +use serde::{Deserialize, Serialize}; +use serde_json::{Map, Value as JsonValue}; + +use super::{Listing, SourceError}; + +/// Entries a page of another Forge's listing is asked for (its default). +pub const DIRECTORY_PAGE: u32 = 100; + +/// A file as the source lists it. +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct Listed { + /// Its path below the source's URL, as the source names it. + pub path: String, + pub size: Option, + pub content_type: Option, + /// Lowercase hex digests. + pub sha1: Option, + pub sha256: Option, + pub sha512: Option, + /// The SHA-256 is the one its `.sha256` sidecar holds. + pub sidecar: bool, + /// When the source stored the content: Nexus' `blobCreated` (else its + /// `lastModified`), another Forge's `last_modified`, a web server's + /// `Last-Modified`. + pub created: Option>, + /// When the source last served it: Nexus' `lastDownloaded`, another + /// Forge's `last_downloaded`. + pub downloaded: Option>, + /// Who uploaded it (Nexus). + pub uploader: Option, +} + +/// One page of Nexus' asset listing. +#[derive(Debug, Default)] +pub struct AssetPage { + pub files: Vec, + /// The `continuationToken` of the next page; `None` on the last. + pub next: Option, +} + +/// One entry of another Forge's listing. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum Entry { + /// A file, its `path` the directory's and its name. + File(Listed), + /// A subdirectory: its name, without the trailing `/`. + Directory(String), +} + +/// One page of a directory of another Forge's listing. +#[derive(Debug, Default)] +pub struct DirectoryPage { + pub entries: Vec, + /// The `after` of the next page; `None` on the last. + pub next: Option, +} + +/// Where a walk of another Forge's listing is: in directory `dir` (`""` for +/// the root, else ending in `/`), after the entry `after` (a subdirectory's +/// name ending in `/`). It is the scan's cursor, so a walk resumes where it +/// stopped without a stack: once a directory is listed to its end, the walk +/// goes on in its parent after it. +#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct Walk { + pub dir: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub after: Option, +} + +/// What one step of a [`Walk`] found. +#[derive(Debug, Default, PartialEq, Eq)] +pub struct Step { + /// The files, up to the page's first subdirectory. + pub files: Vec, + /// Subdirectories whose names Forge refuses (`a\b/`): not walked, and + /// reported as failed items. + pub refused: Vec<(String, String)>, + /// Where the walk goes on; `None` once it is done. + pub next: Option, +} + +impl Walk { + /// The walk's cursor, as the import stores it. + pub fn cursor(&self) -> String { + serde_json::to_string(self).unwrap_or_default() + } + + /// The walk a stored cursor names; the root's start without one. + pub fn parse(cursor: Option<&str>) -> Result { + match cursor { + None => Ok(Walk::default()), + Some(cursor) => serde_json::from_str(cursor).map_err(|e| { + SourceError::Invalid(format!("the import's cursor {cursor:?} is unreadable: {e}")) + }), + } + } + + /// The next step after `page`, the page of `self.dir` after + /// `self.after`. The files up to the page's first subdirectory are + /// taken, and the walk goes down into it; the rest of the page is read + /// again once that subdirectory is walked. A page without a + /// subdirectory goes on to the next page, or, at the directory's end, + /// back up to the parent after it. + pub fn step(&self, page: DirectoryPage) -> Step { + let mut step = Step::default(); + for entry in page.entries { + match entry { + Entry::File(file) => step.files.push(file), + Entry::Directory(name) => { + let dir = format!("{}{name}/", self.dir); + match files::validate_segment(&name) + .and_then(|()| files::validate_directory(&dir)) + { + Ok(()) => { + step.next = Some(Walk { dir, after: None }); + return step; + } + // Not walked: Forge could store nothing below it. + Err(e) => step.refused.push((dir, e.to_string())), + } + } + } + } + step.next = match page.next { + Some(after) => Some(Walk { + dir: self.dir.clone(), + after: Some(after), + }), + None => self.up(), + }; + step + } + + /// The walk in the parent directory, after this one; `None` at the root. + fn up(&self) -> Option { + let dir = self.dir.strip_suffix('/')?; + let (parent, name) = match dir.rsplit_once('/') { + Some((parent, name)) => (format!("{parent}/"), name), + None => (String::new(), dir), + }; + Some(Walk { + dir: parent, + after: Some(format!("{name}/")), + }) + } +} + +/// What a `HEAD` of a named file says. +#[derive(Debug, Default)] +pub struct Head { + pub size: Option, + pub content_type: Option, + pub last_modified: Option>, +} + +/// A raw repository at `url` (no trailing `/`). +pub struct RawSource { + http: UpstreamHttp, + url: String, + username: Option, + password: Option, + /// The `Server` header the source answered the preflight with. + server: Option, +} + +impl RawSource { + pub fn new( + http: UpstreamHttp, + url: &str, + username: Option, + password: Option, + ) -> Self { + RawSource { + http, + url: url.trim_end_matches('/').to_string(), + username, + password, + server: None, + } + } + + /// The `Server` header the source answered [`Self::probe`] with. + pub fn server(&self) -> Option<&str> { + self.server.as_deref() + } + + /// The Basic credentials, for the source's origin only. + fn auth(&self) -> Option> { + match (&self.username, &self.password) { + (Some(username), Some(password)) => Some(UpstreamAuth { + origin: Origin::of(&self.url), + username, + password: password.expose_secret(), + bearer: false, + }), + _ => None, + } + } + + /// Where the source serves the file at `path`, each segment + /// percent-encoded. + pub fn file_url(&self, path: &str) -> String { + format!("{}/{}", self.url, files::encode(path)) + } + + /// The Nexus server and repository the URL names, when it has the shape + /// of a Nexus repository's: `/repository/`. + pub fn nexus(&self) -> Option<(&str, &str)> { + let (base, name) = self.url.rsplit_once("/repository/")?; + let valid = !name.is_empty() + && name + .bytes() + .all(|b| b.is_ascii_alphanumeric() || b"._-".contains(&b)); + (valid && base.contains("://")).then_some((base, name)) + } + + fn assets_url(&self, cursor: Option<&str>) -> Option { + let (base, repository) = self.nexus()?; + let mut query = url::form_urlencoded::Serializer::new(String::new()); + query.append_pair("repository", repository); + if let Some(cursor) = cursor { + query.append_pair("continuationToken", cursor); + } + Some(format!("{base}/service/rest/v1/assets?{}", query.finish())) + } + + fn directory_url(&self, dir: &str, after: Option<&str>, limit: u32) -> String { + let mut query = url::form_urlencoded::Serializer::new(String::new()); + query.append_pair("format", "json"); + query.append_pair("limit", &limit.to_string()); + if let Some(after) = after { + query.append_pair("after", after); + } + format!("{}/{}?{}", self.url, files::encode(dir), query.finish()) + } + + /// GET `url`: the answer, or `None` for a status that says the resource + /// is not there (or is no document of this kind). A refusal of the + /// credentials, a server error and a failed transfer are errors. + async fn get_document( + &self, + url: &str, + accept: &str, + ) -> Result, Vec)>, SourceError> { + let mut headers = reqwest::header::HeaderMap::new(); + if let Ok(accept) = reqwest::header::HeaderValue::from_str(accept) { + headers.insert(reqwest::header::ACCEPT, accept); + } + let response = self + .http + .send( + reqwest::Method::GET, + url, + RequestClass::Metadata, + headers, + self.auth().as_ref(), + ) + .await + .map_err(|e| SourceError::of(e, url))?; + let status = response.status(); + let server = header(&response, reqwest::header::SERVER); + if !status.is_success() { + return match status { + StatusCode::UNAUTHORIZED | StatusCode::FORBIDDEN => { + Err(SourceError::status(status, url)) + } + s if s.is_server_error() || s == StatusCode::TOO_MANY_REQUESTS => { + Err(SourceError::status(status, url)) + } + _ => Ok(None), + }; + } + let config = self.http.config(); + let body = SpooledBody::read( + response, + url, + config.max_metadata_bytes, + config.spool_memory_bytes, + ) + .await + .map_err(|e| SourceError::of(e, url))?; + let bytes = tokio::task::spawn_blocking(move || -> std::io::Result> { + let mut out = Vec::new(); + std::io::Read::read_to_end(&mut body.into_reader()?, &mut out)?; + Ok(out) + }) + .await + .map_err(|e| SourceError::Invalid(e.to_string()))? + .map_err(|e| SourceError::Unreachable(format!("reading {url}: {e}")))?; + Ok(Some((server, bytes))) + } + + /// The preflight: which listing names the source's files. Another + /// Forge's (its root as JSON) is tried first, then, for a URL of a + /// Nexus repository's shape, Nexus' asset listing. `None` when the + /// source serves neither: the import names its paths. + pub async fn probe(&mut self) -> Result, SourceError> { + let url = self.directory_url("", None, 1); + if let Some((server, body)) = self.get_document(&url, "application/json").await? { + self.server = server; + if parse_directory(&body).is_ok() { + return Ok(Some(Listing::Directories)); + } + } + let Some(url) = self.assets_url(None) else { + return Ok(None); + }; + if let Some((server, body)) = self.get_document(&url, "application/json").await? { + if parse_assets(&body).is_ok() { + self.server = server; + return Ok(Some(Listing::Assets)); + } + } + Ok(None) + } + + /// The preflight of an import that names its paths: the first named + /// file is asked for, so wrong credentials or an unreachable server are + /// reported before the import is stored. A file it does not have is no + /// error (it fails as its item). + pub async fn probe_named(&mut self, path: &str) -> Result<(), SourceError> { + let url = self.file_url(path); + let response = self + .http + .send( + reqwest::Method::HEAD, + &url, + RequestClass::Metadata, + reqwest::header::HeaderMap::new(), + self.auth().as_ref(), + ) + .await + .map_err(|e| SourceError::of(e, &url))?; + self.server = header(&response, reqwest::header::SERVER); + let status = response.status(); + if matches!(status, StatusCode::UNAUTHORIZED | StatusCode::FORBIDDEN) + || status.is_server_error() + { + return Err(SourceError::status(status, &url)); + } + Ok(()) + } + + /// The page of Nexus' asset listing after `cursor`. + pub async fn assets(&self, cursor: Option<&str>) -> Result { + let url = self.assets_url(cursor).ok_or_else(|| { + SourceError::Invalid(format!( + "{} is no Nexus repository's URL (/repository/)", + self.url + )) + })?; + let (_, body) = self + .get_document(&url, "application/json") + .await? + .ok_or_else(|| { + SourceError::Invalid(format!("Nexus serves no asset listing at {url}")) + })?; + parse_assets(&body).map_err(|e| SourceError::Invalid(format!("{url}: {e}"))) + } + + /// The page of the directory `dir` (`""` or ending in `/`) of another + /// Forge's listing after the entry `after`. + pub async fn directory( + &self, + dir: &str, + after: Option<&str>, + ) -> Result { + let url = self.directory_url(dir, after, DIRECTORY_PAGE); + let Some((_, body)) = self.get_document(&url, "application/json").await? else { + // Emptied since it was listed in its parent. + return Ok(DirectoryPage::default()); + }; + let mut page = + parse_directory(&body).map_err(|e| SourceError::Invalid(format!("{url}: {e}")))?; + for entry in &mut page.entries { + if let Entry::File(file) = entry { + file.path = format!("{dir}{}", file.path); + } + } + Ok(page) + } + + /// What a `HEAD` of the file at `path` says; `None` when the source has + /// no file there. A server that does not answer `HEAD` says nothing: the + /// download tells. + pub async fn head(&self, path: &str) -> Result, SourceError> { + let url = self.file_url(path); + let response = self + .http + .send( + reqwest::Method::HEAD, + &url, + RequestClass::Metadata, + reqwest::header::HeaderMap::new(), + self.auth().as_ref(), + ) + .await + .map_err(|e| SourceError::of(e, &url))?; + let status = response.status(); + match status { + StatusCode::NOT_FOUND | StatusCode::GONE => return Ok(None), + StatusCode::METHOD_NOT_ALLOWED | StatusCode::NOT_IMPLEMENTED => { + return Ok(Some(Head::default())) + } + s if !s.is_success() => return Err(SourceError::status(s, &url)), + _ => {} + } + Ok(Some(Head { + size: header(&response, reqwest::header::CONTENT_LENGTH).and_then(|v| v.parse().ok()), + content_type: header(&response, reqwest::header::CONTENT_TYPE), + last_modified: header(&response, reqwest::header::LAST_MODIFIED) + .and_then(|v| DateTime::parse_from_rfc2822(&v).ok()) + .map(|t| t.with_timezone(&Utc)), + })) + } + + /// The SHA-256 the `.sha256` sidecar of the file at `path` holds, where + /// the source serves one (``, or ` ` as `sha256sum` + /// prints it). A sidecar that holds no SHA-256 is none. + pub async fn sidecar(&self, path: &str) -> Result, SourceError> { + let url = self.file_url(&format!("{path}.sha256")); + let document = match self.get_document(&url, "text/plain, */*").await { + Ok(document) => document, + // A server that keeps its sidecars from these credentials has + // none for the import. + Err(SourceError::Refused(_)) => None, + Err(e) => return Err(e), + }; + Ok(document.and_then(|(_, body)| parse_sidecar(&body))) + } + + /// GET the file at `path` (streamed by the caller); `None` when the + /// source has no file there. + pub async fn download(&self, path: &str) -> Result, SourceError> { + let url = self.file_url(path); + let response = self + .http + .get(&url, RequestClass::Download, self.auth().as_ref()) + .await + .map_err(|e| SourceError::of(e, &url))?; + match response.status() { + s if s.is_success() => Ok(Some(response)), + StatusCode::NOT_FOUND | StatusCode::GONE => Ok(None), + s => Err(SourceError::status(s, &url)), + } + } +} + +fn header(response: &reqwest::Response, name: reqwest::header::HeaderName) -> Option { + response + .headers() + .get(name) + .and_then(|v| v.to_str().ok()) + .map(str::trim) + .filter(|v| !v.is_empty()) + .map(str::to_string) +} + +/// A lowercase hex digest of `len` digits, if `s` is one. +fn hex_digest(s: &str, len: usize) -> Option { + (s.len() == len && s.bytes().all(|b| b.is_ascii_hexdigit())).then(|| s.to_ascii_lowercase()) +} + +/// The SHA-256 of a `.sha256` sidecar: its first word, 64 hex digits. +pub fn parse_sidecar(body: &[u8]) -> Option { + let text = std::str::from_utf8(body.get(..body.len().min(4096))?).ok()?; + hex_digest(text.split_whitespace().next()?, 64) +} + +fn rfc3339(value: Option<&JsonValue>) -> Option> { + value + .and_then(JsonValue::as_str) + .and_then(|s| DateTime::parse_from_rfc3339(s).ok()) + .map(|t| t.with_timezone(&Utc)) +} + +fn text(value: Option<&JsonValue>) -> Option { + value + .and_then(JsonValue::as_str) + .map(str::trim) + .filter(|s| !s.is_empty()) + .map(str::to_string) +} + +/// A page of Nexus' asset listing (`{"items": [...], "continuationToken"}`). +pub fn parse_assets(body: &[u8]) -> Result { + #[derive(Deserialize)] + struct Raw { + items: Vec, + #[serde(rename = "continuationToken", default)] + continuation_token: Option, + } + let raw: Raw = serde_json::from_slice(body).map_err(|e| e.to_string())?; + let files = raw + .items + .into_iter() + .filter_map(|Lenient(asset)| { + // Nexus names a raw asset's path without a leading `/` (older + // releases with one). + let path = asset.get("path").and_then(JsonValue::as_str)?; + let path = path.strip_prefix('/').unwrap_or(path).to_string(); + let checksum = |key: &str| { + asset + .get("checksum") + .and_then(|c| c.get(key)) + .and_then(JsonValue::as_str) + .and_then(|s| hex_digest(s, s.len())) + .filter(|s| !s.is_empty()) + }; + Some(Listed { + path, + size: asset.get("fileSize").and_then(JsonValue::as_u64), + content_type: text(asset.get("contentType")), + sha1: checksum("sha1").filter(|s| s.len() == 40), + sha256: checksum("sha256").filter(|s| s.len() == 64), + sha512: checksum("sha512").filter(|s| s.len() == 128), + sidecar: false, + created: rfc3339(asset.get("blobCreated")).or(rfc3339(asset.get("lastModified"))), + downloaded: rfc3339(asset.get("lastDownloaded")), + uploader: text(asset.get("uploader")), + }) + }) + .collect(); + Ok(AssetPage { + files, + next: raw.continuation_token.filter(|t| !t.is_empty()), + }) +} + +/// A page of another Forge's listing (`{"entries": [...], "next"}`); a +/// file's `path` is its name, the caller prefixes its directory. +pub fn parse_directory(body: &[u8]) -> Result { + #[derive(Deserialize)] + struct Raw { + entries: Vec>, + #[serde(default)] + next: Option, + } + let raw: Raw = serde_json::from_slice(body).map_err(|e| e.to_string())?; + let mut entries = Vec::with_capacity(raw.entries.len()); + for entry in raw.entries { + let Some(name) = entry.get("name").and_then(JsonValue::as_str) else { + continue; + }; + let directory = entry.get("type").and_then(JsonValue::as_str) == Some("directory"); + if directory { + entries.push(Entry::Directory( + name.strip_suffix('/').unwrap_or(name).to_string(), + )); + continue; + } + entries.push(Entry::File(Listed { + path: name.to_string(), + size: entry.get("size").and_then(JsonValue::as_u64), + content_type: text(entry.get("content_type")), + sha256: entry + .get("sha256") + .and_then(JsonValue::as_str) + .and_then(|s| hex_digest(s, 64)), + created: rfc3339(entry.get("last_modified")), + downloaded: rfc3339(entry.get("last_downloaded")), + ..Listed::default() + })); + } + Ok(DirectoryPage { + entries, + next: raw.next.filter(|n| !n.is_empty()), + }) +} + +/// A JSON object whose repeated keys keep their first non-null value. Nexus +/// 3.70 writes `"blobCreated": null` and then `"blobCreated": "