Add support for conditional HTTP requests.

Cargo.toml

@@ -12,6 +12,7 @@ readme = "README.md"
 
 [dependencies]
 arc-swap        = "1.0"
+bytes           = "1"
 chrono          = "0.4.31"
 clap            = { version = "3.0", features = [ "cargo" ] }
 crossbeam-utils = "0.8.4"
@@ -29,7 +30,7 @@ rustls-webpki   = "0.101.6"
 serde           = { version = "1.0", features = ["derive"] }
 serde_json      = "1.0"
 slab            = "0.4.2"
-tokio           = { version = "1.6", features = ["io-util", "macros", "net", "rt", "rt-multi-thread", "sync", "time"]}
+tokio           = { version = "1.6", features = ["fs", "io-util", "macros", "net", "rt", "rt-multi-thread", "sync", "time"]}
 tokio-rustls    = "0.24.1"
 tokio-stream    = { version = "0.1", features = ["net"] }
 toml            = "0.8.2"

src/lib.rs

@@ -33,3 +33,4 @@ pub mod metrics;
 pub mod payload;
 pub mod targets;
 pub mod units;
+pub mod utils;

src/manager.rs

@@ -5,7 +5,7 @@ use std::collections::HashMap;
 use std::sync::Arc;
 use log::error;
 use serde::Deserialize;
-use reqwest::blocking::Client as HttpClient;
+use reqwest::Client as HttpClient;
 use tokio::runtime::Runtime;
 use crate::{http, metrics};
 use crate::comms::{Gate, GateAgent, Link};

src/targets/http.rs

@@ -3,15 +3,20 @@
 use std::convert::Infallible;
 use std::sync::Arc;
 use arc_swap::ArcSwap;
+use chrono::{DateTime, Utc};
 use futures::stream;
-use hyper::{Body, Method, Request, Response};
+use hyper::{Body, Method, Request, Response, StatusCode};
+use hyper::http::response;
 use log::debug;
+use rpki::rtr::State;
 use serde::Deserialize;
 use crate::payload;
 use crate::comms::Link;
 use crate::formats::output;
 use crate::log::ExitError;
 use crate::manager::Component;
+use crate::utils::http::EtagsIter;
+use crate::utils::http::{format_http_date, parse_http_date};
 
 
 //------------ Target --------------------------------------------------------
@@ -43,30 +48,45 @@ impl Target {
                     return None
                 }
 
-                if let Some(update) = http_source.set() {
-                    Some(
-                        Response::builder()
-                        .header("Content-Type", format.content_type())
-                        .body(Body::wrap_stream(stream::iter(
-                            format.stream(update)
-                            .map(Result::<_, Infallible>::Ok)
-                        )))
-                        .unwrap()
-                    )
+                let update = http_source.data();
+                let update = match update.as_ref() {
+                    Some(update) => update,
+                    None => {
+                        return Some(
+                            Response::builder()
+                            .status(503)
+                            .header("Content-Type", "text/plain")
+                            .body(
+                                "Initial validation ongoing. \
+                                 Please wait.".into()
+                            )
+                            .unwrap()
+                        )
+                    }
+                };
+
+                if update.is_not_modified(request) {
+                    return Some(update.not_modified())
                 }
-                else {
-                    Some(
+
+                Some(
+                    update.header(
                         Response::builder()
-                        .status(503)
-                        .header("Content-Type", "text/plain")
-                        .body("Initial validation ongoing. Please wait.".into())
-                        .unwrap()
+                    ).header(
+                        "Content-Type", format.content_type()
                     )
-                }
+                    .body(Body::wrap_stream(stream::iter(
+                        format.stream(update.set.clone())
+                        .map(Result::<_, Infallible>::Ok)
+                    )))
+                    .unwrap()
+                )
             }
         );
         component.register_http_resource(processor.clone());
 
+        let mut state = State::new();
+
         loop {
             debug!("Target {}: link status: {}",
                     component.name(), unit.get_status()
@@ -76,33 +96,106 @@ impl Target {
                     "Target {}: Got update ({} entries)",
                     component.name(), update.set().len()
                 );
-                source.update(update);
+                source.update(SourceData::new(update, &mut state));
             }
         }
     }
 }
 
 
-
 //------------ Source --------------------------------------------------------
 
 /// The date source for an HTTP target.
 #[derive(Clone, Default)]
 struct Source {
     /// The current set of RTR data.
-    data: Arc<ArcSwap<Option<payload::Set>>>
+    data: Arc<ArcSwap<Option<SourceData>>>
 }
 
 impl Source {
     /// Updates the data source from the given update.
-    fn update(&self, update: payload::Update) {
-        self.data.store(Some(update.set().clone()).into())
+    fn update(&self, data: SourceData) {
+        self.data.store(Some(data).into())
     }
 
-    /// Returns the current payload set.
-    fn set(&self) -> Option<payload::Set> {
-        self.data.load().as_ref().as_ref().cloned()
+    /// Returns the current payload data.
+    fn data(&self) -> Arc<Option<SourceData>> {
+        self.data.load_full()
     }
 }
 
 
+//------------ SourceData ----------------------------------------------------
+
+/// The data held by a data source.
+///
+struct SourceData {
+    set: payload::Set,
+    etag: String,
+    created: DateTime<Utc>,
+}
+
+impl SourceData {
+    fn new(update: payload::Update, state: &mut State) -> Self {
+        let etag = format!("\"{:x}-{}\"", state.session(), state.serial());
+        state.inc();
+        Self {
+            set: update.set().clone(),
+            etag,
+            created: Utc::now(),
+        }
+    }
+
+    /// Returns whether 304 Not Modified response should be retured.
+    fn is_not_modified(&self, req: &Request<Body>) -> bool {
+        // First, check If-None-Match.
+        for value in req.headers().get_all("If-None-Match").iter() {
+            // Skip ill-formatted values. By being lazy here we may falsely
+            // return a full response, so this should be fine.
+            let value = match value.to_str() {
+                Ok(value) => value,
+                Err(_) => continue
+            };
+            let value = value.trim();
+            if value == "*" {
+                return true
+            }
+            for tag in EtagsIter::new(value) {
+                if tag.trim() == self.etag {
+                    return true
+                }
+            }
+        }
+
+        // Now, the If-Modified-Since header.
+        if let Some(value) = req.headers().get("If-Modified-Since") {
+            let value = match value.to_str() {
+                Ok(value) => value,
+                Err(_) => return false,
+            };
+            if let Some(date) = parse_http_date(value) {
+                if date >= self.created {
+                    return true
+                }
+            }
+        }
+
+        false
+    }
+
+    fn not_modified(&self) -> Response<Body> {
+        self.header(
+            response::Builder::new().status(
+                StatusCode::NOT_MODIFIED
+            )
+        ).body(Body::empty()).expect("broken HTTP response builder")
+    }
+
+    fn header(&self, builder: response::Builder) -> response::Builder {
+        builder.header(
+            "ETag", &self.etag
+        ).header(
+            "Last-Modified", format_http_date(self.created)
+        )
+    }
+}