Use x forward header (#6)

* Use X-Forward-Header to authenticate against TerminusDB

* Fix formatting

Author: rrooij

.gitignore

@@ -1,2 +1,3 @@
 /target
-*~
+*~
+.env

docker-compose.yml

@@ -9,6 +9,8 @@ services:
       - 6363:6363
     environment:
       - TERMINUSDB_SERVER_PORT=6363
+      - TERMINUSDB_INSECURE_USER_HEADER=X-User-Forward
+      - TERMINUSDB_INSECURE_USER_HEADER_ENABLED=true
     volumes:
       - ./storage:/app/terminusdb/storage
   semantic_indexer:
@@ -17,6 +19,7 @@ services:
       - "8080:8080"
     environment:
       - TERMINUSDB_CONTENT_ENDPOINT=http://terminusdb:6363
+      - TERMINUSDB_USER_FORWARD_HEADER=X-User-Forward
       - OPENAI_KEY=${OPENAI_KEY}
     volumes:
       - ./vector_storage:/app/storage

src/main.rs

@@ -37,6 +37,8 @@ enum Commands {
         #[arg(short, long)]
         content_endpoint: Option<String>,
         #[arg(short, long)]
+        user_forward_header: Option<String>,
+        #[arg(short, long)]
         directory: String,
         #[arg(short, long, default_value_t = 8080)]
         port: u16,
@@ -110,18 +112,24 @@ fn content_endpoint_or_env(c: Option<String>) -> Option<String> {
     c.or_else(|| std::env::var("TERMINUSDB_CONTENT_ENDPOINT").ok())
 }
 
+fn user_forward_header_or_env(c: Option<String>) -> String {
+    c.unwrap_or_else(|| std::env::var("TERMINUSDB_USER_FORWARD_HEADER").unwrap())
+}
+
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
     let args = Args::parse();
     match args.command {
         Commands::Serve {
             content_endpoint,
+            user_forward_header,
             directory,
             port,
             size,
         } => {
             server::serve(
                 directory,
+                user_forward_header_or_env(user_forward_header),
                 port,
                 size,
                 content_endpoint_or_env(content_endpoint),

src/server.rs

@@ -256,6 +256,7 @@ pub struct QueryResult {
 
 pub struct Service {
     content_endpoint: Option<String>,
+    user_forward_header: String,
     path: PathBuf,
     vector_store: VectorStore,
     pending: Mutex<HashSet<String>>,
@@ -277,6 +278,7 @@ enum TerminusIndexOperationError {}
 
 async fn get_operations_from_content_endpoint(
     content_endpoint: String,
+    user_forward_header: String,
     domain: String,
     commit: String,
     previous: Option<String>,
@@ -287,7 +289,11 @@ async fn get_operations_from_content_endpoint(
     }
     let endpoint = format!("{}/{}", content_endpoint, &domain);
     let url = reqwest::Url::parse_with_params(&endpoint, &params).unwrap();
-    let res = reqwest::get(url)
+    let client = reqwest::Client::new();
+    let res = client
+        .get(url)
+        .header(user_forward_header, "admin")
+        .send()
         .await
         .unwrap()
         .bytes_stream()
@@ -369,10 +375,16 @@ impl Service {
         s
     }
 
-    fn new<P: Into<PathBuf>>(path: P, num_bufs: usize, content_endpoint: Option<String>) -> Self {
+    fn new<P: Into<PathBuf>>(
+        path: P,
+        user_forward_header: String,
+        num_bufs: usize,
+        content_endpoint: Option<String>,
+    ) -> Self {
         let path = path.into();
         Service {
             content_endpoint,
+            user_forward_header,
             path: path.clone(),
             vector_store: VectorStore::new(path, num_bufs),
             pending: Mutex::new(HashSet::new()),
@@ -416,6 +428,7 @@ impl Service {
                 if self.test_and_set_pending(index_id.clone()).await {
                     let opstream = get_operations_from_content_endpoint(
                         content_endpoint.to_string(),
+                        self.user_forward_header.clone(),
                         domain.clone(),
                         commit.clone(),
                         previous.clone(),
@@ -725,12 +738,18 @@ enum AssignIndexError {
 
 pub async fn serve<P: Into<PathBuf>>(
     directory: P,
+    user_forward_header: String,
     port: u16,
     num_bufs: usize,
     content_endpoint: Option<String>,
 ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
     let addr = SocketAddr::new(IpAddr::V6(Ipv6Addr::UNSPECIFIED), port);
-    let service = Arc::new(Service::new(directory, num_bufs, content_endpoint));
+    let service = Arc::new(Service::new(
+        directory,
+        user_forward_header,
+        num_bufs,
+        content_endpoint,
+    ));
     let make_svc = make_service_fn(move |_conn| {
         let s = service.clone();
         async {