Preparing for the next release

CHANGELOG.md

@@ -5,6 +5,10 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ## Unreleased
+### Added
+- Fixed bug in parsing command line options for bridge
+- Added support for getting the client IP address from a proxy header (e.g. `X-Forwarded-For`)
+- Cleaned up port handling, so URLs with default ports don't have the ports specified
 
 ## [0.0.8] - 2024-10-24
 ### Added

bridge/Cargo.toml

@@ -3,7 +3,7 @@
 
 [package]
 name = "op-bridge"
-version = "0.0.1"
+version = "0.0.9"
 description = "An example of an OpenPortal user portal to OpenPortal bridge"
 edition = "2021"
 license = "MIT"

bridge/src/main.rs

@@ -40,6 +40,7 @@ async fn main() -> Result<()> {
         Some("127.0.0.1".to_owned()),
         Some(8044),
         None,
+        None,
         Some("http://localhost:3000".to_owned()),
         Some("127.0.0.1".to_owned()),
         Some(3000),

cluster/Cargo.toml

@@ -3,7 +3,7 @@
 
 [package]
 name = "op-cluster"
-version = "0.0.1"
+version = "0.0.9"
 description = "An example of an OpenPortal HPC cluster platform agent"
 edition = "2021"
 license = "MIT"

cluster/src/main.rs

@@ -35,6 +35,7 @@ async fn main() -> Result<()> {
         Some("127.0.0.1".to_owned()),
         Some(8045),
         None,
+        None,
         Some(AgentType::Platform),
     );
 

docs/cmdline/cluster/Cargo.toml

@@ -3,7 +3,7 @@
 
 [package]
 name = "example-cluster"
-version = "0.0.1"
+version = "0.0.9"
 description = "templemeads command line example - cluster agent"
 edition = "2021"
 license = "MIT"

docs/cmdline/cluster/src/main.rs

@@ -25,6 +25,7 @@ async fn main() -> Result<()> {
         Some("127.0.0.1".to_owned()),
         Some(8091),
         None,
+        None,
         Some(AgentType::Instance),
     );
 

docs/cmdline/portal/Cargo.toml

@@ -3,7 +3,7 @@
 
 [package]
 name = "example-portal"
-version = "0.0.1"
+version = "0.0.9"
 description = "templemeads command line example - portal agent"
 edition = "2021"
 license = "MIT"

docs/cmdline/portal/src/main.rs

@@ -21,6 +21,7 @@ async fn main() -> Result<()> {
         Some("127.0.0.1".to_owned()),
         Some(8090),
         None,
+        None,
         Some(AgentType::Portal),
     );
 

docs/echo/Cargo.toml

@@ -3,7 +3,7 @@
 
 [package]
 name = "example-echo"
-version = "0.0.1"
+version = "0.0.9"
 description = "Paddington Echo Service example"
 edition = "2021"
 license = "MIT"

docs/echo/src/main.rs

@@ -187,6 +187,7 @@ async fn run_client(invitation: &Path) -> Result<(), Error> {
         "127.0.0.1",
         &6502,
         &None,
+        &None,
     )?;
 
     // now give the invitation to connect to the server to the client
@@ -262,7 +263,7 @@ async fn run_server(
     invitation: &Path,
 ) -> Result<(), Error> {
     // create the echo-server service
-    let mut service = ServiceConfig::new("echo-server", url, ip, port, &None)?;
+    let mut service = ServiceConfig::new("echo-server", url, ip, port, &None, &None)?;
 
     let invite = service.add_client("echo-client", range)?;
 

docs/job/Cargo.toml

@@ -3,7 +3,7 @@
 
 [package]
 name = "example-job"
-version = "0.0.1"
+version = "0.0.9"
 description = "templemeads job example"
 edition = "2021"
 license = "MIT"

docs/job/src/main.rs

@@ -205,6 +205,7 @@ async fn run_cluster(invitation: &Path) -> Result<(), Error> {
         "127.0.0.1",
         &6502,
         &None,
+        &None,
     )?;
 
     // now give the invitation to connect to the server to the client
@@ -260,7 +261,7 @@ async fn run_portal(
     invitation: &Path,
 ) -> Result<(), Error> {
     // create a paddington service configuration for the portal agent
-    let mut service = ServiceConfig::new("portal", url, ip, port, &None)?;
+    let mut service = ServiceConfig::new("portal", url, ip, port, &None, &None)?;
 
     // add the cluster to the portal, returning an invitation
     let invite = service.add_client("cluster", range)?;

filesystem/Cargo.toml

@@ -3,7 +3,7 @@
 
 [package]
 name = "op-filesystem"
-version = "0.0.1"
+version = "0.0.9"
 description = "Agent that interfaces OpenPortal with a filesystem"
 edition = "2021"
 license = "MIT"

filesystem/src/main.rs

@@ -40,6 +40,7 @@ async fn main() -> Result<()> {
         Some("127.0.0.1".to_owned()),
         Some(8047),
         None,
+        None,
         Some(AgentType::Filesystem),
     );
 

freeipa/Cargo.toml

@@ -3,7 +3,7 @@
 
 [package]
 name = "op-freeipa"
-version = "0.0.1"
+version = "0.0.9"
 description = "Agent that interfaces OpenPortal with FreeIPA"
 edition = "2021"
 license = "MIT"

freeipa/src/main.rs

@@ -45,6 +45,7 @@ async fn main() -> Result<()> {
         Some("127.0.0.1".to_owned()),
         Some(8046),
         None,
+        None,
         Some(AgentType::Account),
     );
 

paddington/Cargo.toml

@@ -3,7 +3,7 @@
 
 [package]
 name = "paddington"
-version = "0.0.1"
+version = "0.0.9"
 description = "A library for implementing the OpenPortal communication protocol"
 edition = "2021"
 license = "MIT"
@@ -32,9 +32,10 @@ serde_json = "1.0.120"
 serde_with = { version="3.9.0", features = ["hex"] }
 thiserror = "1.0.63"
 tokio = { version = "1.0", features = ["full", "tracing"] }
-tokio-tungstenite = { version = "0.23.1", features = ["rustls-tls-native-roots"] }
+tokio-tungstenite = { version = "0.24.0", features = ["rustls-tls-native-roots"] }
 toml = "0.8.16"
 tracing = "0.1.40"
+tungstenite = "0.24.0"
 url = {version="2.5.2", features=["serde"]}
 
 [lints.rust]

paddington/src/config.rs

@@ -74,6 +74,7 @@ pub struct Defaults {
     ip: String,
     port: u16,
     healthcheck_port: Option<u16>,
+    proxy_header: Option<String>,
 }
 
 impl Defaults {
@@ -84,6 +85,7 @@ impl Defaults {
         ip: Option<String>,
         port: Option<u16>,
         healthcheck_port: Option<u16>,
+        proxy_header: Option<String>,
     ) -> Self {
         let config_file = config_file.unwrap_or(
             dirs::config_local_dir()
@@ -102,6 +104,7 @@ impl Defaults {
             ip: ip.unwrap_or("127.0.0.1".to_owned()),
             port: port.unwrap_or(8042),
             healthcheck_port,
+            proxy_header,
         }
     }
 
@@ -439,6 +442,7 @@ pub struct ServiceConfig {
     ip: IpAddr,
     port: u16,
     heathcheck_port: Option<u16>,
+    proxy_header: Option<String>,
 
     servers: Vec<ServerConfig>,
     clients: Vec<ClientConfig>,
@@ -452,6 +456,7 @@ impl ServiceConfig {
         ip: &str,
         port: &u16,
         healthcheck_port: &Option<u16>,
+        proxy_header: &Option<String>,
     ) -> Result<Self, Error> {
         Ok(ServiceConfig {
             name: name.to_string(),
@@ -461,6 +466,7 @@ impl ServiceConfig {
                 .with_context(|| format!("Could not parse IP address: {}", ip))?,
             port: *port,
             heathcheck_port: *healthcheck_port,
+            proxy_header: proxy_header.clone(),
             servers: Vec::new(),
             clients: Vec::new(),
             encryption: None,
@@ -531,6 +537,10 @@ impl ServiceConfig {
         self.heathcheck_port
     }
 
+    pub fn proxy_header(&self) -> Option<String> {
+        self.proxy_header.clone()
+    }
+
     pub fn add_client(&mut self, name: &str, ip: &str) -> Result<Invite, Error> {
         let ip = IpOrRange::new(ip)
             .with_context(|| format!("Could not parse into an IP address or IP range: {}", ip))?;
@@ -612,6 +622,7 @@ impl ServiceConfig {
         ip: IpAddr,
         port: u16,
         healthcheck_port: &Option<u16>,
+        proxy_header: &Option<String>,
     ) -> Result<ServiceConfig, Error> {
         // see if this config_dir exists - return an error if it does
         let config_file = path::absolute(config_file).with_context(|| {
@@ -625,7 +636,14 @@ impl ServiceConfig {
             return Err(Error::NotExists(config_file.to_string_lossy().to_string()));
         }
 
-        let config = ServiceConfig::new(&name, &url, &ip.to_string(), &port, healthcheck_port)?;
+        let config = ServiceConfig::new(
+            &name,
+            &url,
+            &ip.to_string(),
+            &port,
+            healthcheck_port,
+            proxy_header,
+        )?;
         save::<ServiceConfig>(config.clone(), &config_file)?;
 
         // check we can read the config and return it
@@ -688,17 +706,29 @@ mod tests {
 
     #[test]
     fn test_invitations() {
-        let mut primary =
-            ServiceConfig::new("primary", "http://localhost", "127.0.0.1", &5544, &None)
-                .unwrap_or_else(|e| {
-                    unreachable!("Cannot create service config: {}", e);
-                });
-
-        let mut secondary =
-            ServiceConfig::new("secondary", "http://localhost", "127.0.0.1", &5545, &None)
-                .unwrap_or_else(|e| {
-                    unreachable!("Cannot create service config: {}", e);
-                });
+        let mut primary = ServiceConfig::new(
+            "primary",
+            "http://localhost",
+            "127.0.0.1",
+            &5544,
+            &None,
+            &None,
+        )
+        .unwrap_or_else(|e| {
+            unreachable!("Cannot create service config: {}", e);
+        });
+
+        let mut secondary = ServiceConfig::new(
+            "secondary",
+            "http://localhost",
+            "127.0.0.1",
+            &5545,
+            &None,
+            &None,
+        )
+        .unwrap_or_else(|e| {
+            unreachable!("Cannot create service config: {}", e);
+        });
 
         // introduce the secondary to the primary
         let invite = primary

paddington/src/connection.rs

@@ -9,12 +9,15 @@ use futures_channel::mpsc::{unbounded, UnboundedSender};
 use futures_util::{future, pin_mut, stream::TryStreamExt};
 use secrecy::ExposeSecret;
 use serde::{de::DeserializeOwned, Serialize};
+use std::sync::Arc;
 use tokio::net::TcpStream;
 use tokio::sync::Mutex as TokioMutex;
 use tokio_tungstenite::connect_async;
 use tokio_tungstenite::tungstenite::protocol::Message as TokioMessage;
-
-use std::sync::Arc;
+use tungstenite::handshake::server::{
+    ErrorResponse as HandshakeErrorResponse, Request as HandshakeRequest,
+    Response as HandshakeResponse,
+};
 
 use crate::command::Command;
 use crate::config::{ClientConfig, PeerConfig, ServiceConfig};
@@ -358,10 +361,48 @@ impl Connection {
         // we now know we are the only ones handling the connection,
         // and are safe to update the keys etc.
 
-        let addr: std::net::SocketAddr = stream
+        let mut addr: std::net::SocketAddr = stream
             .peer_addr()
             .with_context(|| "Error getting the peer address. Ensure the connection is open.")?;
 
+        let proxy_header = self.config.proxy_header();
+        let mut proxy_client = None;
+
+        let process_headers = |request: &HandshakeRequest,
+                               response: HandshakeResponse|
+         -> Result<HandshakeResponse, HandshakeErrorResponse> {
+            if let Some(proxy_header) = proxy_header {
+                if let Some(value) = request
+                    .headers()
+                    .get(proxy_header)
+                    .and_then(|value| value.to_str().ok())
+                {
+                    proxy_client = Some(value.to_string());
+                }
+            }
+
+            Ok(response)
+        };
+
+        let ws_stream = tokio_tungstenite::accept_hdr_async(stream, process_headers)
+            .await
+            .with_context(|| {
+                format!(
+                    "Error accepting WebSocket connection from: {}. Closing connection.",
+                    addr
+                )
+            })?;
+
+        if let Some(proxy_client) = proxy_client {
+            tracing::info!("Proxy client: {:?}", proxy_client);
+            addr = proxy_client.parse().with_context(|| {
+                format!(
+                    "Error parsing proxy client address: {}. Closing connection.",
+                    proxy_client
+                )
+            })?;
+        }
+
         tracing::info!("Accepted connection from peer: {}", addr);
 
         let clients: Vec<ClientConfig> = self
@@ -379,15 +420,6 @@ impl Connection {
             ));
         }
 
-        let ws_stream = tokio_tungstenite::accept_async(stream)
-            .await
-            .with_context(|| {
-                format!(
-                    "Error accepting WebSocket connection from: {}. Closing connection.",
-                    addr
-                )
-            })?;
-
         // Split the WebSocket stream into incoming and outgoing parts
         let (mut outgoing, mut incoming) = ws_stream.split();