Add persistant sessions

Cargo.toml

@@ -8,8 +8,9 @@ edition = "2018"
 cookie = "0.13"
 futures = "0.3"
 log = "0.4"
+nanoid = "0.3"
 pretty_env_logger = "0.4"
-sessions = { version = "0.0.2", features = ["fs-store", "nanoid", "tokio"] }
+sessions = { version = "0.0.2", features = ["fs-store", "tokio"] }
 time = "0.2"
 tokio = { version = "0.2", features = ["macros"] }
 warp = "0.2"

src/main.rs

@@ -1,7 +1,10 @@
 use std::collections::HashMap;
+use std::sync::Arc;
 
 use cookie::{Cookie, SameSite};
-use log::{info, trace};
+use futures::TryFutureExt;
+use log::{debug, error, info, trace};
+use sessions::{Session, SessionStatus, Storable};
 use warp::{Filter, Rejection, Reply};
 use warp::http::{StatusCode, Uri};
 
@@ -10,25 +13,30 @@ const SESSION_HEADER: &'static str = "Spectrum-Session";
 #[tokio::main]
 async fn main() {
   pretty_env_logger::init();
-  let mut store = sessions::FilesystemStore::new("data/sessions.json".into());
+  let store = Arc::new(sessions::FilesystemStore::new("data".into()));
 
   // GET /
   let root = warp::path::end()
     .and(warp::get())
-    .and(with_session(&mut store))
+    .map(|| info!("GET /")).untuple_one()
-    .map(|session| format!("Hello user #{}", session));
+    .and(with_session(store.clone(), SessionPolicy::Existing))
+    .map(|session: Session| format!("Hello, {} (user #{})!",
+                                    session.get::<String>(SESSION_NAME).unwrap().unwrap(),
+                                    session.id().unwrap()));
 
   // GET/POST /user name={name}
   let namechange = warp::path!("user")
     .and(warp::get()
-         .and(with_session(&mut store))
+         .map(|| info!("GET /user")).untuple_one()
+         .and(with_session(store.clone(), SessionPolicy::AllowNew))
+         // .recover(|err| todo!("prevent redir loop")) // reply::with_status(..)
          .and(warp::fs::file("./static/index.html"))
          .map(|_session, file| file)
-         // .recover(|err| todo!("prevent redir loop")) // reply::with_status(..)
          .or(warp::post()
+             .and(with_session(store.clone(), SessionPolicy::AllowNew))
              .and(warp::body::content_length_limit(1024 * 16))
              .and(warp::body::form())
-             .and_then(|form: HashMap<String, String>| async move {
+             .and_then(|session: Session, form: HashMap<String, String>| async move {
                let name = match form.get("name") {
                  None => return Err(warp::reject::custom(BadName)),
                  Some(name) if name.is_empty() ||
@@ -36,6 +44,9 @@ async fn main() {
                  Some(name) => name
                };
                info!("POST /user as \"{}\"", name);
+
+               let _old = session.set(SESSION_NAME, name.clone());
+               session.save().await.map_err(|e| warp::reject::custom(ServerError(Box::new(e))))?;
                Ok(warp::reply::with_header(StatusCode::SEE_OTHER,
                                            warp::http::header::LOCATION,
                                            "/"))
@@ -67,42 +78,81 @@ where From: 'static,
   })
 }
 
-fn with_session(_store: &mut impl sessions::Storable) -> impl Filter<Extract = (String,),
+#[derive(Debug, Clone, Copy)]
-                                                                    Error = Rejection> + Clone {
+enum SessionPolicy {
-  warp::cookie::cookie(SESSION_HEADER)
+  Existing,
-    .map(|session| {
+  AllowNew
-      info!("Found session: {}", session);
-      session
-    })
-    .or_else(clarify_error::<_, warp::reject::MissingCookie, NoSession>) // Has cookies, but not session cookie
-    .or_else(clarify_error::<_, warp::reject::InvalidHeader, NoSession>) // No cookies at all
 }
 
-#[derive(Debug)]
+const SESSION_NAME: &str = "name";
+
+fn with_session(store: Arc<impl Storable>,
+                policy: SessionPolicy) -> impl Filter<Extract = (Session,),
+                                                      Error = Rejection> + Clone {
+  warp::cookie::cookie(SESSION_HEADER)
+    .and_then(move |session_id: String| {
+      trace!("Looking up session: {}", session_id);
+      let store = store.clone();  // pending async reqs allowed to outlive returned Filter
+      async move {
+        let session = store.get(&session_id).await?;
+        session.set_id(session_id)?;
+        Ok(session)
+      }
+    }.map_err(|err: std::io::Error| warp::reject::custom(ServerError(Box::new(err)))))
+    .or_else(clarify_error::<_, warp::reject::MissingCookie, NoSession>)  // Has cookies, but not session cookie
+    .or_else(clarify_error::<_, warp::reject::InvalidHeader, NoSession>)  // No cookies at all
+    .and_then(move |session: Session| async move {
+      match (session.status(), policy) {
+        (Ok(SessionStatus::Existed), _ )
+          // Session exists, but name might not be set yet
+          if session.get::<String>(SESSION_NAME)
+              .map_or(false, |name| name.filter(|name| !name.is_empty()).is_some()) => Ok(session),
+        (Ok(SessionStatus::Existed), _ )                      => Err(warp::reject::custom(BadName)),
+        (Ok(SessionStatus::Created), SessionPolicy::AllowNew) => Ok(session),
+        (Ok(SessionStatus::Created), _)                       => Err(warp::reject::custom(NoSession)),
+        (Err(e), _) => Err(warp::reject::custom(ServerError(Box::new(e)))),
+        _           => Err(warp::reject::custom(NoSession))
+      }
+    })
+}
+
+#[derive(Debug, Default)]
 struct BadName;
 impl warp::reject::Reject for BadName {}
 
-async fn handle_reject(err: Rejection) -> Result<impl Reply, Rejection> {
+async fn handle_reject(err: Rejection) -> Result<Box<dyn Reply>, Rejection> {
-  match err.find() {
+  if let Some(BadName) = err.find() {
-    Some(BadName) => Ok(warp::reply::with_status(warp::reply(), StatusCode::BAD_REQUEST)),
+    Ok(Box::new(warp::reply::with_header(StatusCode::SEE_OTHER,
-    _ => Err(err)
+                                         warp::http::header::LOCATION,
-  }
+                                         "/user")))
+  } else if let Some(ServerError(e)) = err.find() {
+    error!("Server Error: {:?}", e);
+    Ok(Box::new(warp::reply::with_status("jrd done messed up X-(", StatusCode::INTERNAL_SERVER_ERROR)))
+  } else { Err(err) }
 }
 
+#[derive(Debug)]
+struct ServerError(Box<dyn std::error::Error + Send + Sync>);
+impl warp::reject::Reject for ServerError {}
+
 #[derive(Debug, Default)]
 struct NoSession;
 impl warp::reject::Reject for NoSession {}
 
 async fn handle_no_session(err: Rejection) -> Result<impl Reply, Rejection> {
   match err.find() {
-    Some(NoSession) => Ok(warp::reply::with_header(warp::redirect::temporary(Uri::from_static("/user")),
+    Some(NoSession) => {
+      debug!("No Session");
+      let id = nanoid::nanoid!(32);
+      Ok(warp::reply::with_header(warp::redirect::temporary(Uri::from_static("/user")),
                                   "Set-Cookie",
-                                                   Cookie::build(SESSION_HEADER, 5.to_string())
+                                  Cookie::build(SESSION_HEADER, id)
                                     .max_age(time::Duration::seconds(60 * 60 * 24 * 365))
                                     .max_age(time::Duration::seconds(10))
                                     .same_site(SameSite::Lax)
                                     .finish()
-                                                     .to_string())),
+                                    .to_string()))
+      }
     _ => Err(err)
   }
 }