Update missing content API

.gitignore

@@ -1,2 +1,3 @@
+/content
 /target
 *.swp

crates/api/Cargo.toml

@@ -11,3 +11,4 @@ warg-crypto = { workspace = true }
 serde = { workspace = true }
 serde_with = { workspace = true }
 thiserror = { workspace = true }
+itertools = { workspace = true }

crates/api/src/v1/package.rs

@@ -21,6 +21,26 @@ pub enum ContentSource {
     },
 }
 
+/// Represents the supported kinds of content upload endpoints.
+#[derive(Clone, Debug, Serialize, Deserialize)]
+#[serde(tag = "type", rename_all = "camelCase")]
+pub enum UploadEndpoint {
+    /// Content may be uploaded via HTTP POST to the given URL.
+    /// If the endpoint responds with "201 Created" and a Location header, that
+    /// header's value will be the content source.
+    HttpPost {
+        /// The URL to POST content to.
+        url: String,
+    },
+}
+
+/// Information about missing content.
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct MissingContent {
+    /// Upload endpoint(s) that may be used to provide missing content.
+    pub upload: Vec<UploadEndpoint>,
+}
+
 /// Represents a request to publish a record to a package log.
 #[derive(Serialize, Deserialize)]
 #[serde(rename = "camelCase")]
@@ -45,13 +65,13 @@ pub struct PackageRecord {
 }
 
 impl PackageRecord {
-    /// Gets the missing content digests of the record.
-    pub fn missing_content(&self) -> &[AnyHash] {
+    /// Gets the missing content of the record.
+    pub fn missing_content(&self) -> impl Iterator<Item = (&AnyHash, &MissingContent)> {
         match &self.state {
             PackageRecordState::Sourcing {
                 missing_content, ..
-            } => missing_content,
-            _ => &[],
+            } => itertools::Either::Left(missing_content.iter()),
+            _ => itertools::Either::Right(std::iter::empty()),
         }
     }
 }
@@ -66,18 +86,22 @@ impl PackageRecord {
 #[allow(clippy::large_enum_variant)]
 pub enum PackageRecordState {
     /// The package record needs content sources.
+    #[serde(rename_all = "camelCase")]
     Sourcing {
         /// The digests of the missing content.
-        missing_content: Vec<AnyHash>,
+        missing_content: HashMap<AnyHash, MissingContent>,
     },
     /// The package record is processing.
+    #[serde(rename_all = "camelCase")]
     Processing,
     /// The package record is rejected.
+    #[serde(rename_all = "camelCase")]
     Rejected {
         /// The reason the record was rejected.
         reason: String,
     },
     /// The package record was successfully published to the log.
+    #[serde(rename_all = "camelCase")]
     Published {
         /// The envelope of the package record.
         record: ProtoEnvelopeBody,

crates/api/src/v1/paths.rs

@@ -1,6 +1,5 @@
 //! The paths of the Warg REST API.
 
-use warg_crypto::hash::AnyHash;
 use warg_protocol::registry::{LogId, RecordId};
 
 /// The path of the "fetch logs" API.
@@ -23,11 +22,6 @@ pub fn package_record(log_id: &LogId, record_id: &RecordId) -> String {
     format!("v1/package/{log_id}/record/{record_id}")
 }
 
-/// The path for a package record's content.
-pub fn package_record_content(log_id: &LogId, record_id: &RecordId, digest: &AnyHash) -> String {
-    format!("v1/package/{log_id}/record/{record_id}/content/{digest}")
-}
-
 /// The path for proving checkpoint consistency.
 pub fn prove_consistency() -> &'static str {
     "v1/proof/consistency"

crates/client/src/api.rs

@@ -1,12 +1,11 @@
 //! A module for Warg registry API clients.
 
-use anyhow::{anyhow, bail, Context, Result};
+use anyhow::{anyhow, Result};
 use bytes::Bytes;
 use futures_util::{Stream, TryStreamExt};
-use reqwest::{Body, IntoUrl, Response, StatusCode, Url};
+use reqwest::{Body, IntoUrl, Response, StatusCode};
 use serde::de::DeserializeOwned;
 use thiserror::Error;
-use url::Host;
 use warg_api::v1::{
     fetch::{FetchError, FetchLogsRequest, FetchLogsResponse},
     package::{
@@ -27,6 +26,8 @@ use warg_transparency::{
     map::MapProofBundle,
 };
 
+use crate::registry_url::RegistryUrl;
+
 /// Represents an error that occurred while communicating with the registry.
 #[derive(Debug, Error)]
 pub enum ClientError {
@@ -122,71 +123,28 @@ async fn into_result<T: DeserializeOwned, E: DeserializeOwned + Into<ClientError
 /// Represents a Warg API client for communicating with
 /// a Warg registry server.
 pub struct Client {
-    url: Url,
+    url: RegistryUrl,
     client: reqwest::Client,
 }
 
 impl Client {
     /// Creates a new API client with the given URL.
     pub fn new(url: impl IntoUrl) -> Result<Self> {
-        let url = Self::validate_url(url)?;
+        let url = RegistryUrl::new(url)?;
         Ok(Self {
             url,
             client: reqwest::Client::new(),
         })
     }
 
     /// Gets the URL of the API client.
-    pub fn url(&self) -> &str {
-        self.url.as_str()
-    }
-
-    /// Parses and validates the given URL.
-    ///
-    /// Returns the validated URL on success.
-    pub fn validate_url(url: impl IntoUrl) -> Result<Url> {
-        // Default to a HTTPS scheme if none is provided
-        let url: Url = if !url.as_str().contains("://") {
-            Url::parse(&format!("https://{url}", url = url.as_str()))
-                .context("failed to parse registry server URL")?
-        } else {
-            url.into_url()
-                .context("failed to parse registry server URL")?
-        };
-
-        match url.scheme() {
-            "https" => {}
-            "http" => {
-                // Only allow HTTP connections to loopback
-                match url
-                    .host()
-                    .ok_or_else(|| anyhow!("expected a host for URL `{url}`"))?
-                {
-                    Host::Domain(d) => {
-                        if d != "localhost" {
-                            bail!("an unsecured connection is not permitted to `{d}`");
-                        }
-                    }
-                    Host::Ipv4(ip) => {
-                        if !ip.is_loopback() {
-                            bail!("an unsecured connection is not permitted to address `{ip}`");
-                        }
-                    }
-                    Host::Ipv6(ip) => {
-                        if !ip.is_loopback() {
-                            bail!("an unsecured connection is not permitted to address `{ip}`");
-                        }
-                    }
-                }
-            }
-            _ => bail!("expected a HTTPS scheme for URL `{url}`"),
-        }
-        Ok(url)
+    pub fn url(&self) -> &RegistryUrl {
+        &self.url
     }
 
     /// Gets the latest checkpoint from the registry.
     pub async fn latest_checkpoint(&self) -> Result<SerdeEnvelope<MapCheckpoint>, ClientError> {
-        let url = self.url.join(paths::fetch_checkpoint()).unwrap();
+        let url = self.url.join(paths::fetch_checkpoint());
         tracing::debug!("getting latest checkpoint at `{url}`");
         into_result::<_, FetchError>(reqwest::get(url).await?).await
     }
@@ -196,7 +154,7 @@ impl Client {
         &self,
         request: FetchLogsRequest<'_>,
     ) -> Result<FetchLogsResponse, ClientError> {
-        let url = self.url.join(paths::fetch_logs()).unwrap();
+        let url = self.url.join(paths::fetch_logs());
         tracing::debug!("fetching logs at `{url}`");
 
         let response = self.client.post(url).json(&request).send().await?;
@@ -209,10 +167,7 @@ impl Client {
         log_id: &LogId,
         request: PublishRecordRequest<'_>,
     ) -> Result<PackageRecord, ClientError> {
-        let url = self
-            .url
-            .join(&paths::publish_package_record(log_id))
-            .unwrap();
+        let url = self.url.join(&paths::publish_package_record(log_id));
         tracing::debug!(
             "appending record to package `{id}` at `{url}`",
             id = request.id
@@ -228,10 +183,7 @@ impl Client {
         log_id: &LogId,
         record_id: &RecordId,
     ) -> Result<PackageRecord, ClientError> {
-        let url = self
-            .url
-            .join(&paths::package_record(log_id, record_id))
-            .unwrap();
+        let url = self.url.join(&paths::package_record(log_id, record_id));
         tracing::debug!("getting record `{record_id}` for package `{log_id}` at `{url}`");
 
         let response = reqwest::get(url).await?;
@@ -283,7 +235,7 @@ impl Client {
 
     /// Proves the inclusion of the given package log heads in the registry.
     pub async fn prove_inclusion(&self, request: InclusionRequest<'_>) -> Result<(), ClientError> {
-        let url = self.url.join(paths::prove_inclusion()).unwrap();
+        let url = self.url.join(paths::prove_inclusion());
         tracing::debug!("proving checkpoint inclusion at `{url}`");
 
         let response = into_result::<InclusionResponse, ProofError>(
@@ -303,7 +255,7 @@ impl Client {
         &self,
         request: ConsistencyRequest<'_>,
     ) -> Result<(), ClientError> {
-        let url = self.url.join(paths::prove_consistency()).unwrap();
+        let url = self.url.join(paths::prove_consistency());
         let response = into_result::<ConsistencyResponse, ProofError>(
             self.client.post(url).json(&request).send().await?,
         )
@@ -349,15 +301,12 @@ impl Client {
     /// Uploads package content to the registry.
     pub async fn upload_content(
         &self,
-        log_id: &LogId,
-        record_id: &RecordId,
-        digest: &AnyHash,
+        url: &str,
         content: impl Into<Body>,
     ) -> Result<String, ClientError> {
-        let url = self
-            .url
-            .join(&paths::package_record_content(log_id, record_id, digest))
-            .unwrap();
+        // Upload URLs may be relative to the registry URL.
+        let url = self.url.join(url);
+
         tracing::debug!("uploading content to `{url}`");
 
         let response = self.client.post(url).body(content).send().await?;

crates/client/src/config.rs

@@ -1,10 +1,9 @@
 //! Module for client configuration.
 
-use crate::{api, ClientError};
+use crate::{ClientError, RegistryUrl};
 use anyhow::{anyhow, Context, Result};
 use normpath::PathExt;
 use once_cell::sync::Lazy;
-use reqwest::Url;
 use serde::{Deserialize, Serialize};
 use std::{
     env::current_dir,
@@ -63,7 +62,7 @@ fn normalize_path(path: &Path) -> PathBuf {
 /// Paths used for storage
 pub struct StoragePaths {
     /// The registry URL relating to the storage paths.
-    pub url: Url,
+    pub registry_url: RegistryUrl,
     /// The path to the registry storage directory.
     pub registries_dir: PathBuf,
     /// The path to the content storage directory.
@@ -242,16 +241,16 @@ impl Config {
         &self,
         url: Option<&str>,
     ) -> Result<StoragePaths, ClientError> {
-        let url = api::Client::validate_url(
+        let registry_url = RegistryUrl::new(
             url.or(self.default_url.as_deref())
                 .ok_or(ClientError::NoDefaultUrl)?,
         )?;
 
-        let host = url.host().unwrap().to_string().to_ascii_lowercase();
-        let registries_dir = self.registries_dir()?.join(host);
+        let label = registry_url.safe_label();
+        let registries_dir = self.registries_dir()?.join(label);
         let content_dir = self.content_dir()?;
         Ok(StoragePaths {
-            url,
+            registry_url,
             registries_dir,
             content_dir,
         })