Merge pull request #27 from PocketRelay/dev

Merging qos fixes

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "pocket-relay"
-version = "0.5.3"
+version = "0.5.4"
 description = "Pocket Relay Server"
 readme = "README.md"
 keywords = ["EA", "PocketRelay", "MassEffect"]

Dockerfile

@@ -8,7 +8,7 @@ RUN apk add curl
 WORKDIR /app
 
 # Download server executable
-RUN curl -LJ -o pocket-relay-linux https://github.com/PocketRelay/Server/releases/download/v0.5.3/pocket-relay-linux
+RUN curl -LJ -o pocket-relay-linux https://github.com/PocketRelay/Server/releases/download/v0.5.4/pocket-relay-linux
 
 # Make the server executable
 RUN chmod +x ./pocket-relay-linux

src/main.rs

@@ -30,7 +30,7 @@ async fn main() {
     App::init(config).await;
 
     // Create the HTTP router
-    let router = routes::router().into_make_service();
+    let router = routes::router().into_make_service_with_connect_info::<SocketAddr>();
 
     // Create futures for server and shutdown signal
     let server_future = Server::bind(&addr).serve(router);

src/routes/qos.rs

@@ -34,20 +34,22 @@ pub struct QosQuery {
 ///
 /// `query` The query string from the client
 pub async fn qos(Query(query): Query<QosQuery>) -> Xml {
+    debug!("Recieved QOS query: (Port: {})", query.port);
+
     /// Port for the local Quality of Service server
     const QOS_PORT: Port = 42130;
-
-    debug!("Recieved QOS query: (Port: {})", query.port);
+    const IP: u32 = u32::from_be_bytes([127, 0, 0, 1]);
 
     let response = format!(
-        r"<qos> <numprobes>0</numprobes>
+        r#"<?xml version="1.0" encoding="UTF-8"?><qos> <numprobes>0</numprobes>
     <qosport>{}</qosport>
     <probesize>0</probesize>
     <qoshost>127.0.0.1</qoshost>
+    <qosip>{}</qosip>
     <requestid>1</requestid>
     <reqsecret>0</reqsecret>
-</qos>",
-        QOS_PORT
+</qos>"#,
+        QOS_PORT, IP
     );
     Xml(response)
 }

src/routes/server.rs

@@ -10,6 +10,7 @@ use crate::{
 };
 use axum::{
     body::Empty,
+    extract::ConnectInfo,
     http::{header, HeaderValue, StatusCode},
     response::{IntoResponse, Response},
     Json,
@@ -18,7 +19,10 @@ use blaze_pk::packet::PacketCodec;
 use interlink::service::Service;
 use log::{debug, error};
 use serde::{Deserialize, Serialize};
-use std::sync::atomic::{AtomicU32, Ordering};
+use std::{
+    net::SocketAddr,
+    sync::atomic::{AtomicU32, Ordering},
+};
 use tokio::{fs::read_to_string, io::split};
 use tokio_util::codec::{FramedRead, FramedWrite};
 
@@ -51,7 +55,12 @@ pub async fn server_details() -> Json<ServerDetails> {
 /// Handles upgrading connections from the Pocket Relay Client tool
 /// from HTTP over to the Blaze protocol for proxing the game traffic
 /// as blaze sessions using HTTP Upgrade
-pub async fn upgrade(upgrade: BlazeUpgrade) -> Response {
+pub async fn upgrade(
+    ConnectInfo(socket_addr): ConnectInfo<SocketAddr>,
+    upgrade: BlazeUpgrade,
+) -> Response {
+    // TODO: Socket address extraction for forwarded reverse proxy
+
     tokio::spawn(async move {
         let socket = match upgrade.upgrade().await {
             Ok(value) => value,
@@ -72,7 +81,7 @@ pub async fn upgrade(upgrade: BlazeUpgrade) -> Response {
             ctx.attach_stream(read, true);
             let writer = ctx.attach_sink(write);
 
-            Session::new(session_id, socket.host_target, writer)
+            Session::new(session_id, socket.host_target, writer, socket_addr)
         });
     });
 

src/session/mod.rs

@@ -28,7 +28,7 @@ use blaze_pk::{
 };
 use interlink::prelude::*;
 use log::{debug, error, log_enabled};
-use std::{fmt::Debug, io};
+use std::{fmt::Debug, io, net::SocketAddr};
 
 pub mod models;
 pub mod routes;
@@ -40,6 +40,9 @@ pub struct Session {
     /// Unique identifier for this session.
     id: SessionID,
 
+    /// Connection socket addr
+    addr: SocketAddr,
+
     /// Packet writer sink for the session
     writer: SinkLink<Packet>,
 
@@ -294,6 +297,22 @@ impl Handler<UpdateClientMessage> for Session {
     }
 }
 
+#[derive(Message)]
+#[msg(rtype = "SocketAddr")]
+pub struct GetSocketAddrMessage;
+
+impl Handler<GetSocketAddrMessage> for Session {
+    type Response = Mr<GetSocketAddrMessage>;
+
+    fn handle(
+        &mut self,
+        _msg: GetSocketAddrMessage,
+        _ctx: &mut ServiceContext<Self>,
+    ) -> Self::Response {
+        Mr(self.addr)
+    }
+}
+
 /// Creates a set session packet and sends it to all the
 /// provided session links
 #[derive(Message)]
@@ -424,12 +443,18 @@ impl Session {
     /// `id`             The unique session ID
     /// `values`         The networking TcpStream and address
     /// `message_sender` The message sender for session messages
-    pub fn new(id: SessionID, host_target: SessionHostTarget, writer: SinkLink<Packet>) -> Self {
+    pub fn new(
+        id: SessionID,
+        host_target: SessionHostTarget,
+        writer: SinkLink<Packet>,
+        addr: SocketAddr,
+    ) -> Self {
         Self {
             id,
             writer,
             data: SessionData::default(),
             host_target,
+            addr,
         }
     }
 

src/session/routes/user_sessions.rs

@@ -7,12 +7,14 @@ use crate::{
             errors::{ServerError, ServerResult},
             user_sessions::*,
         },
-        GetLookupMessage, HardwareFlagMessage, LookupResponse, NetworkInfoMessage, SessionLink,
-        SetPlayerMessage,
+        GetLookupMessage, GetSocketAddrMessage, HardwareFlagMessage, LookupResponse,
+        NetworkInfoMessage, SessionLink, SetPlayerMessage,
     },
     state::App,
+    utils::models::NetAddress,
 };
 use log::error;
+use std::net::SocketAddr;
 
 /// Attempts to lookup another authenticated session details
 ///
@@ -128,7 +130,20 @@ pub async fn handle_resume_session(
 ///     }
 /// }
 /// ```
-pub async fn handle_update_network(session: &mut SessionLink, req: UpdateNetworkRequest) {
+pub async fn handle_update_network(session: &mut SessionLink, mut req: UpdateNetworkRequest) {
+    let ext = &mut req.address.external;
+
+    // If address is missing
+    if ext.0 .0.is_unspecified() {
+        // Obtain socket address from session
+        if let Ok(SocketAddr::V4(addr)) = session.send(GetSocketAddrMessage).await {
+            let ip = addr.ip();
+            // Replace address with new address and port with same as local port
+            ext.0 = NetAddress(*ip);
+            ext.1 = req.address.internal.1;
+        }
+    }
+
     let _ = session
         .send(NetworkInfoMessage {
             groups: req.address,