spacebar/rust-nex
6
0
Fork 0
Code Issues Pull requests Projects Releases Packages 50 Wiki Activity Actions

feat: rnex works, all major roadblocks are gone, the rnex rewrite is now in working state

Browse source
This commit is contained in:
DJMrTV 2025-05-07 21:52:05 +02:00
commit 5e726fc9b0
12 changed files with 494 additions and 667 deletions
Download patch file Download diff file Expand all files Collapse all files

178
src/main.rs
View file

@ -7,33 +7,37 @@
//! also the first and only current usage of rnex, expect this and rnex to be split into seperate
//! repos soon.
use crate::rmc::protocols::auth::RemoteAuth;
use crate::rmc::protocols::auth::RawAuthInfo;
use crate::rmc::protocols::auth::RawAuth;
use crate::nex::account::Account;
use crate::nex::auth_handler::{AuthHandler, RemoteAuthClientProtocol};
use crate::prudp::packet::VirtualPort;
use crate::prudp::router::Router;
use crate::prudp::secure::Secure;
use crate::prudp::sockaddr::PRUDPSockAddr;
use crate::prudp::socket::Unsecure;
use chrono::{Local, SecondsFormat};
use log::{error, info};
use once_cell::sync::Lazy;
use simplelog::{
ColorChoice, CombinedLogger, Config, LevelFilter, TermLogger, TerminalMode, WriteLogger,
};
use std::fs::File;
use std::net::{Ipv4Addr, SocketAddrV4};
use std::{env, fs};
use std::marker::PhantomData;
use std::ops::{BitAnd, BitOr};
use std::str::FromStr;
use macros::rmc_struct;
use crate::prudp::unsecure::Unsecure;
use crate::rmc::protocols::auth::Auth;
use crate::rmc::protocols::auth::RawAuth;
use crate::rmc::protocols::auth::RawAuthInfo;
use crate::rmc::protocols::auth::RemoteAuth;
use crate::rmc::protocols::{new_rmc_gateway_connection, OnlyRemote};
use crate::rmc::response::ErrorCode;
use crate::rmc::structures::any::Any;
use crate::rmc::structures::connection_data::ConnectionData;
use crate::rmc::structures::qresult::QResult;
use chrono::{Local, SecondsFormat};
use log::{error, info};
use macros::rmc_struct;
use once_cell::sync::Lazy;
use simplelog::{
ColorChoice, CombinedLogger, Config, LevelFilter, TermLogger, TerminalMode, WriteLogger,
};
use std::fs::File;
use std::marker::PhantomData;
use std::net::{Ipv4Addr, SocketAddrV4};
use std::ops::{BitAnd, BitOr};
use std::str::FromStr;
use std::time::Duration;
use std::{env, fs};
use tokio::task::JoinHandle;
mod endianness;
mod prudp;
@ -43,9 +47,9 @@ pub mod rmc;
mod grpc;
mod kerberos;
mod nex;
mod web;
mod versions;
mod result;
mod versions;
mod web;
static KERBEROS_SERVER_PASSWORD: Lazy<String> = Lazy::new(|| {
env::var("AUTH_SERVER_PASSWORD")
@ -258,86 +262,112 @@ async fn start_secure_server() -> SecureServer{
}
}*/
async fn start_auth() -> JoinHandle<()> {
tokio::spawn(async {
let (router_secure, _) = Router::new(SocketAddrV4::new(*OWN_IP_PRIVATE, *AUTH_SERVER_PORT))
.await
.expect("unable to start router");
impl Auth for AuthClient{
async fn login(&self, name: String) -> Result<(), ErrorCode> {
todo!()
}
let mut socket_secure = router_secure
.add_socket(VirtualPort::new(1, 10), Unsecure("6f599f81"))
.await
.expect("unable to add socket");
async fn login_ex(&self, name: String, extra_data: Any) -> Result<(QResult, u32, Vec<u8>, ConnectionData, String), ErrorCode> {
todo!()
}
// let conn = socket_secure.connect(auth_sockaddr).await.unwrap();
async fn request_ticket(&self, source_pid: u32, destination_pid: u32) -> Result<(QResult, Vec<u8>), ErrorCode> {
todo!()
}
loop {
let Some(conn) = socket_secure.accept().await else {
error!("server crashed");
return;
};
async fn get_pid(&self, username: String) -> Result<u32, ErrorCode> {
todo!()
}
info!("new connected user!");
async fn get_name(&self, pid: u32) -> Result<String, ErrorCode> {
todo!()
}
let _ = new_rmc_gateway_connection(conn, |_| AuthHandler {
destination_server_acct: &SECURE_SERVER_ACCOUNT,
build_name: "branch:origin/project/wup-agmj build:3_8_15_2004_0",
station_url: &SECURE_STATION_URL,
});
}
})
}
#[rmc_struct(AuthClientProtocol)]
struct AuthClient {
async fn start_secure() -> JoinHandle<()> {
tokio::spawn(async {
let (router_secure, _) =
Router::new(SocketAddrV4::new(*OWN_IP_PRIVATE, *SECURE_SERVER_PORT))
.await
.expect("unable to start router");
let mut socket_secure = router_secure
.add_socket(
VirtualPort::new(1, 10),
Secure("6f599f81", &SECURE_SERVER_ACCOUNT),
)
.await
.expect("unable to add socket");
// let conn = socket_secure.connect(auth_sockaddr).await.unwrap();
loop {
let Some(conn) = socket_secure.accept().await else {
error!("server crashed");
return;
};
info!("new connected user on secure :D!");
let _ = new_rmc_gateway_connection(conn, |_| AuthHandler {
destination_server_acct: &SECURE_SERVER_ACCOUNT,
build_name: "branch:origin/project/wup-agmj build:3_8_15_2004_0",
station_url: &SECURE_STATION_URL,
});
}
})
}
define_rmc_proto!(
proto AuthClientProtocol{
Auth
}
);
async fn start_test() {
let addr = SocketAddrV4::new(*OWN_IP_PRIVATE, *AUTH_SERVER_PORT);
let virt_addr = VirtualPort::new(1, 10);
let prudp_addr = PRUDPSockAddr::new(addr, virt_addr);
async fn start_servers() {
//let auth_ip = SocketAddrV4::from_str("157.90.13.221:30039").unwrap();
let auth_ip = SocketAddrV4::from_str("31.220.75.208:10000").unwrap();
let auth_port = VirtualPort::new(1, 10);
let auth_sockaddr = PRUDPSockAddr::new(auth_ip, auth_port);
let (router_secure, _) = Router::new(SocketAddrV4::new(*OWN_IP_PRIVATE, *AUTH_SERVER_PORT))
let (router_test, _) = Router::new(SocketAddrV4::new(*OWN_IP_PRIVATE, 26969))
.await
.expect("unable to start router");
let mut socket_secure = router_secure
let mut socket_secure = router_test
.add_socket(VirtualPort::new(1, 10), Unsecure("6f599f81"))
.await
.expect("unable to add socket");
// let conn = socket_secure.connect(auth_sockaddr).await.unwrap();
let conn = socket_secure.connect(prudp_addr).await.unwrap();
loop {
let Some(conn) = socket_secure.accept().await else {
error!("server crashed");
return;
};
let remote =
new_rmc_gateway_connection(conn, |r| OnlyRemote::<RemoteAuthClientProtocol>::new(r));
info!("new connected user!");
let v = remote
.login_ex("1469690705".to_string(), Any::default())
.await
.unwrap();
let _ = new_rmc_gateway_connection(conn, |_| AuthClient {}); //OnlyRemote::<RemoteAuthClientProtocol>::new
println!("got it");
}
}
/*
async fn start_servers() {
#[cfg(feature = "auth")]
let auth_server = start_auth_server().await;
let auth_server = start_auth().await;
#[cfg(feature = "secure")]
let secure_server = start_secure_server().await;
let web_server = web::start_web().await;
let secure_server = start_secure().await;
//let web_server = web::start_web().await;
//tokio::time::sleep(Duration::from_secs(1)).await;
//start_test().await;
#[cfg(feature = "auth")]
auth_server.join_handle.await.expect("auth server crashed");
auth_server.await.expect("auth server crashed");
#[cfg(feature = "secure")]
secure_server.join_handle.await.expect("auth server crashed");
web_server.await.expect("webserver crashed");*/
}
secure_server.await.expect("auth server crashed");
//web_server.await.expect("webserver crashed");
}

167
src/nex/auth_handler.rs Normal file
View file

@ -0,0 +1,167 @@
use crate::grpc::account;
use crate::kerberos::{derive_key, KerberosDateTime, Ticket};
use crate::nex::account::Account;
use crate::rmc::protocols::auth::{Auth, RawAuth, RawAuthInfo, RemoteAuth};
use crate::rmc::response::ErrorCode;
use crate::rmc::response::ErrorCode::Core_Unknown;
use crate::rmc::structures::any::Any;
use crate::rmc::structures::connection_data::ConnectionData;
use crate::rmc::structures::qresult::QResult;
use crate::rmc::structures::RmcSerialize;
use crate::{define_rmc_proto, kerberos, rmc};
use macros::rmc_struct;
define_rmc_proto!(
proto AuthClientProtocol{
Auth
}
);
#[rmc_struct(AuthClientProtocol)]
pub struct AuthHandler {
pub destination_server_acct: &'static Account,
pub build_name: &'static str,
pub station_url: &'static str,
}
pub fn generate_ticket(
source_act_login_data: (u32, [u8; 16]),
dest_act_login_data: (u32, [u8; 16]),
) -> Box<[u8]> {
let source_key = derive_key(source_act_login_data.0, source_act_login_data.1);
let dest_key = derive_key(dest_act_login_data.0, dest_act_login_data.1);
let internal_data = kerberos::TicketInternalData::new(source_act_login_data.0);
let encrypted_inner = internal_data.encrypt(dest_key);
let encrypted_session_ticket = Ticket {
pid: dest_act_login_data.0,
session_key: internal_data.session_key,
}
.encrypt(source_key, &encrypted_inner);
encrypted_session_ticket
}
async fn get_login_data_by_pid(pid: u32) -> Option<(u32, [u8; 16])> {
let Ok(mut client) = account::Client::new().await else {
return None;
};
let Ok(passwd) = client.get_nex_password(pid).await else {
return None;
};
Some((pid, passwd))
}
impl Auth for AuthHandler {
async fn login(&self, name: String) -> Result<(), ErrorCode> {
todo!()
}
async fn login_ex(
&self,
name: String,
extra_data: Any,
) -> Result<(QResult, u32, Vec<u8>, ConnectionData, String), ErrorCode> {
let Ok(pid) = name.parse() else {
return Err(ErrorCode::Core_InvalidArgument);
};
let Ok(mut client) = account::Client::new().await else {
return Err(ErrorCode::Core_Exception);
};
let Ok(passwd) = client.get_nex_password(pid).await else {
return Err(ErrorCode::Core_Exception);
};
let source_login_data = (pid, passwd);
let destination_login_data = self.destination_server_acct.get_login_data();
let ticket = generate_ticket(source_login_data, destination_login_data);
let result = QResult::success(Core_Unknown);
let connection_data = ConnectionData {
station_url: self.station_url.to_string(),
special_station_url: "".to_string(),
//date_time: KerberosDateTime::new(1,1,1,1,1,1),
date_time: KerberosDateTime::now(),
special_protocols: Vec::new(),
};
Ok((
result,
source_login_data.0,
ticket.into(),
connection_data,
self.build_name.to_owned(),
))
}
async fn request_ticket(
&self,
source_pid: u32,
destination_pid: u32,
) -> Result<(QResult, Vec<u8>), ErrorCode> {
let Some(source_login_data) = get_login_data_by_pid(source_pid).await else {
return Err(ErrorCode::Core_Exception);
};
let desgination_login_data = if destination_pid == self.destination_server_acct.pid {
self.destination_server_acct.get_login_data()
} else {
let Some(login) = get_login_data_by_pid(destination_pid).await else {
return Err(ErrorCode::Core_Exception);
};
login
};
let result = QResult::success(Core_Unknown);
let ticket = generate_ticket(source_login_data, desgination_login_data);
Ok((result, ticket.into()))
}
async fn get_pid(&self, username: String) -> Result<u32, ErrorCode> {
Err(ErrorCode::Core_Exception)
}
async fn get_name(&self, pid: u32) -> Result<String, ErrorCode> {
Err(ErrorCode::Core_Exception)
}
}
#[cfg(test)]
mod test {
use crate::rmc::structures::connection_data::ConnectionData;
use crate::rmc::structures::qresult::QResult;
use crate::rmc::structures::RmcSerialize;
use crate::rmc::response::RMCResponse;
use std::io::Cursor;
#[test]
fn test() {
let stuff = hex::decode("200100000a0106000000028000000100010051b3995774000000a6321c7f78847c1c5e9fb825eb26bd91841f1a40d92fc694159666119cb13527f1463ac48ad42a63e6613ede67041554b1770978112e6f1f3e177a2bfc75933216dbe38f70133a1eb28e2ae32a4b5c4b0c3e3efd4c02907992e259b257270b57a9dbe7792f4721b07f8fafb9e32d50f2555c616a015c0000004b007072756470733a2f5049443d323b7369643d313b73747265616d3d31303b747970653d323b616464726573733d322e3234332e39352e3131333b706f72743d31303030313b4349443d3100000000000100002c153ba51f00000033006272616e63683a6f726967696e2f70726f6a6563742f7775702d61676d6a206275696c643a335f385f31355f323030345f3000").unwrap();
let stuff = RMCResponse::new(&mut Cursor::new(stuff)).unwrap();
let crate::rmc::response::RMCResponseResult::Success { call_id, method_id, data: stuff} = stuff.response_result else {
panic!()
};
// let stuff = hex::decode("0100010051B399577400000085F1736FCFBE93660275A3FE36FED6C2EFC57222AC99A9219CF54170A415B02DF1463AC48AD42A6307813FDE67041554B177097832ED000F892D9551A09F88E9CB0388DC1BC9527CC7384556A3287B2A349ABBF7E34A5A3EC14C2287CC7F78DA616BC3B03A035347FBD2E9A505C8EF42447CD809015F0000004E007072756470733A2F73747265616D3D31303B747970653D323B616464726573733D3139322E3136382E3137382E3132303B706F72743D31303030313B4349443D313B5049443D323B7369643D310000000000010000CDF53AA51F00000033006272616E63683A6F726967696E2F70726F6A6563742F7775702D61676D6A206275696C643A335F385F31355F323030345F3000").unwrap();
// let stuff = hex::decode("0100010051b399577400000037d3d4814d2b16dd546c94a75d32637b45f856b5abe73cf26cfaa235c5f2c1cef1463ac48ad42a637d873fde67041554b177097880cfa7e10bb810eaf686bfb0a0cf3d65b1f476ebc046d0855327986f557dca14fbb8594883c186b863f2206f22baa0309dbcc81da2f883cb2cdc12628ec7fced015c0000004b007072756470733a2f5049443d323b7369643d313b73747265616d3d31303b747970653d323b616464726573733d322e3234332e39352e3131333b706f72743d31303030313b4349443d310000000000010000b7f33aa51f00000033006272616e63683a6f726967696e2f70726f6a6563742f7775702d61676d6a206275696c643a335f385f31355f323030345f3000").unwrap();
let data = <(QResult, u32, Vec<u8>, ConnectionData, String) as RmcSerialize>::deserialize(
&mut Cursor::new(stuff),
).unwrap();
println!("data: {:?}", data);
}
}

3
src/nex/mod.rs
View file

@ -1 +1,2 @@
pub mod account;
pub mod account;
pub mod auth_handler;

5
src/prudp/mod.rs
View file

@ -3,5 +3,6 @@ pub mod router;
pub mod socket;
mod auth_module;
pub mod sockaddr;
//pub mod secure;
pub mod station_url;
pub mod station_url;
pub mod secure;
pub mod unsecure;

3
src/prudp/router.rs
View file

@ -50,8 +50,7 @@ impl Router {
trace!("got valid prudp packet from someone({}): \n{:?}", addr, packet);
let connection = packet.source_sockaddr(addr);
println!("data from {:?}", connection);
let endpoints = self.endpoints.read().await;

94
src/prudp/secure.rs
View file

@ -4,10 +4,13 @@ use log::error;
use rc4::cipher::StreamCipherCoreWrapper;
use rc4::{KeyInit, Rc4, Rc4Core, StreamCipher};
use rc4::consts::U16;
use typenum::U5;
use crate::endianness::{IS_BIG_ENDIAN, ReadExtensions};
use crate::kerberos::{derive_key, TicketInternalData};
use crate::nex::account::Account;
use crate::prudp::socket::EncryptionPair;
use crate::prudp::packet::PRUDPPacket;
use crate::prudp::socket::{CryptoHandler, CryptoHandlerConnectionInstance, EncryptionPair};
use crate::prudp::unsecure::UnsecureInstance;
use crate::rmc::structures::RmcSerialize;
pub fn read_secure_connection_data(data: &[u8], act: &Account) -> Option<([u8; 32], u32, u32)>{
@ -97,4 +100,93 @@ pub fn generate_secure_encryption_pairs(mut session_key: [u8; 32], count: u8) ->
}
vec
}
pub struct Secure(pub &'static str, pub &'static Account);
pub struct SecureInstance {
access_key: &'static str,
session_key: [u8; 32],
streams: Vec<EncryptionPair<Rc4<U32>>>,
self_signature: [u8; 16],
remote_signature: [u8; 16],
pid: u32,
}
impl CryptoHandler for Secure {
type CryptoConnectionInstance = SecureInstance;
fn instantiate(
&self,
remote_signature: [u8; 16],
self_signature: [u8; 16],
payload: &[u8],
substream_count: u8,
) -> Option<(Vec<u8>, Self::CryptoConnectionInstance)> {
let (session_key, pid, check_value) = read_secure_connection_data(payload, &self.1)?;
let check_value_response = check_value + 1;
let data = bytemuck::bytes_of(&check_value_response);
let mut response = Vec::new();
data.serialize(&mut response).ok()?;
let encryption_pairs = generate_secure_encryption_pairs(session_key, substream_count);
Some((
response,
SecureInstance {
pid,
streams: encryption_pairs,
session_key,
access_key: self.0,
remote_signature,
self_signature,
},
))
}
fn sign_pre_handshake(&self, packet: &mut PRUDPPacket) {
packet.set_sizes();
packet.calculate_and_assign_signature(self.0, None, None);
}
}
impl CryptoHandlerConnectionInstance for SecureInstance {
type Encryption = Rc4<U5>;
fn decrypt_incoming(&mut self, substream: u8, data: &mut [u8]) {
if let Some(crypt_pair) = self.streams.get_mut(substream as usize){
crypt_pair.recv.apply_keystream(data);
}
}
fn encrypt_outgoing(&mut self, substream: u8, data: &mut [u8]) {
if let Some(crypt_pair) = self.streams.get_mut(substream as usize){
crypt_pair.send.apply_keystream(data);
}
}
fn get_user_id(&self) -> u32 {
self.pid
}
fn sign_connect(&self, packet: &mut PRUDPPacket) {
packet.set_sizes();
packet.calculate_and_assign_signature(self.access_key, None, Some(self.self_signature));
}
fn sign_packet(&self, packet: &mut PRUDPPacket) {
packet.set_sizes();
packet.calculate_and_assign_signature(self.access_key, Some(self.session_key), Some(self.self_signature));
}
fn verify_packet(&self, packet: &PRUDPPacket) -> bool {
true
}
}

576
src/prudp/socket.rs
View file

@ -27,6 +27,7 @@ use tokio::net::UdpSocket;
use tokio::sync::mpsc::{channel, Receiver, Sender};
use tokio::sync::{Mutex, RwLock};
use tokio_stream::Stream;
use crate::nex::account::Account;
// due to the way this is designed crashing the router thread causes deadlock, sorry ;-;
// (maybe i will fix that some day)
@ -39,473 +40,13 @@ pub struct EncryptionPair<T: StreamCipher + Send> {
}
impl<T: StreamCipher + Send> EncryptionPair<T> {
fn init_both<F: Fn() -> T>(func: F) -> Self {
pub fn init_both<F: Fn() -> T>(func: F) -> Self {
Self {
recv: func(),
send: func(),
}
}
}
/*
pub async fn process_packet(
self: &Arc<Self>,
client_address: PRUDPSockAddr,
packet: &PRUDPPacket,
) {
let conn = self.connections.read().await;
if !conn.contains_key(&client_address) {
drop(conn);
let mut conn = self.connections.write().await;
//only insert if we STILL dont have the connection preventing double insertion
if !conn.contains_key(&client_address) {
conn.insert(
client_address,
(
Arc::new(Mutex::new(ConnectionData {
sock_addr: client_address,
id: random(),
signature: [0; 16],
server_signature: [0; 16],
active_connection_data: None,
})),
Arc::new(Mutex::new(())),
),
);
}
drop(conn);
} else {
drop(conn);
}
let connections = self.connections.read().await;
let Some(conn) = connections.get(&client_address) else {
error!("connection is still not present after making sure connection is present, giving up.");
return;
};
let conn = conn.clone();
// dont keep holding the connections list unnescesarily
drop(connections);
let mutual_exclusion_packet_handeling_mtx = conn.1.lock().await;
let mut connection = conn.0.lock().await;
if (packet.header.types_and_flags.get_flags() & ACK) != 0 {
//todo: handle acknowledgements and resending packets propperly
println!("got ack");
return;
}
if (packet.header.types_and_flags.get_flags() & MULTI_ACK) != 0 {
println!("got multi ack");
return;
}
match packet.header.types_and_flags.get_types() {
SYN => {
println!("got syn");
// reset heartbeat?
let mut response_packet = packet.base_response_packet();
response_packet.header.types_and_flags.set_types(SYN);
response_packet.header.types_and_flags.set_flag(ACK);
response_packet.header.types_and_flags.set_flag(HAS_SIZE);
connection.signature = client_address.calculate_connection_signature();
response_packet
.options
.push(ConnectionSignature(connection.signature));
for options in &packet.options {
match options {
SupportedFunctions(functions) => response_packet
.options
.push(SupportedFunctions(*functions & 0x04)),
MaximumSubstreamId(max_substream) => response_packet
.options
.push(MaximumSubstreamId(*max_substream)),
_ => { /* ??? */ }
}
}
response_packet.set_sizes();
response_packet.calculate_and_assign_signature(self.access_key, None, None);
let mut vec = Vec::new();
response_packet
.write_to(&mut vec)
.expect("somehow failed to convert backet to bytes");
self.socket
.send_to(&vec, client_address.regular_socket_addr)
.await
.expect("failed to send data back");
}
CONNECT => {
println!("got connect");
let Some(MaximumSubstreamId(max_substream)) = packet
.options
.iter()
.find(|v| matches!(v, MaximumSubstreamId(_)))
else {
return;
};
let Some((response_data, encryption_pairs, active_secure_connection_data)) =
(self.on_connect_handler)(packet.clone(), *max_substream).await
else {
error!("invalid connection request");
return;
};
connection.active_connection_data = Some(ActiveConnectionData {
encryption_pairs,
reliable_client_queue: VecDeque::new(),
reliable_client_counter: 2,
reliable_server_counter: 1,
server_session_id: packet.header.session_id,
active_secure_connection_data,
connection_id: random(),
});
let mut response_packet = packet.base_response_packet();
response_packet.payload = response_data;
response_packet.header.types_and_flags.set_types(CONNECT);
response_packet.header.types_and_flags.set_flag(ACK);
response_packet.header.types_and_flags.set_flag(HAS_SIZE);
// todo: (or not) sliding windows and stuff
response_packet.header.session_id = packet.header.session_id;
response_packet.header.sequence_id = 1;
response_packet
.options
.push(ConnectionSignature(Default::default()));
//let mut init_seq_id = 0;
for option in &packet.options {
match option {
MaximumSubstreamId(max_substream) => response_packet
.options
.push(MaximumSubstreamId(*max_substream)),
SupportedFunctions(funcs) => {
response_packet.options.push(SupportedFunctions(*funcs))
}
ConnectionSignature(sig) => connection.server_signature = *sig,
PacketOption::InitialSequenceId(_id) => {
//init_seq_id = *id;
}
_ => { /* ? */ }
}
}
// Splatoon doesnt use compression so we arent gonna compress unless i at some point
// want to implement some server which requires it
// No encryption here for the same reason
// todo: implement something to do secure servers
if connection.server_signature == <[u8; 16] as Default>::default() {
error!("didn't get connection signature from client")
}
response_packet.set_sizes();
response_packet.calculate_and_assign_signature(
self.access_key,
None,
Some(connection.server_signature),
);
let mut vec = Vec::new();
response_packet
.write_to(&mut vec)
.expect("somehow failed to convert backet to bytes");
self.socket
.send_to(&vec, client_address.regular_socket_addr)
.await
.expect("failed to send data back");
}
DATA => {
if (packet.header.types_and_flags.get_flags() & RELIABLE) != 0 {
let Some(active_connection) = connection.active_connection_data.as_mut() else {
error!("got data packet on non active connection!");
return;
};
match active_connection
.reliable_client_queue
.binary_search_by_key(&packet.header.sequence_id, |p| p.header.sequence_id)
{
Ok(_) => warn!("recieved packet twice"),
Err(position) => active_connection
.reliable_client_queue
.insert(position, packet.clone()),
}
if (packet.header.types_and_flags.get_flags() & NEED_ACK) != 0 {
let mut ack = packet.base_acknowledgement_packet();
ack.header.session_id = active_connection.server_session_id;
ack.set_sizes();
let potential_session_key = connection
.active_connection_data
.as_ref()
.unwrap()
.active_secure_connection_data
.as_ref()
.map(|s| s.session_key);
ack.calculate_and_assign_signature(
self.access_key,
potential_session_key,
Some(connection.server_signature),
);
let mut vec = Vec::new();
ack.write_to(&mut vec)
.expect("somehow failed to convert backet to bytes");
self.socket
.send_to(&vec, client_address.regular_socket_addr)
.await
.expect("failed to send data back");
}
drop(connection);
while let Some(mut packet) = {
let mut locked = conn.0.lock().await;
let packet = locked
.active_connection_data
.as_mut()
.map(|a| {
a.reliable_client_queue
.front()
.is_some_and(|v| {
v.header.sequence_id == a.reliable_client_counter
})
.then(|| a.reliable_client_queue.pop_front())
})
.flatten()
.flatten();
drop(locked);
packet
} {
if packet.options.iter().any(|v| match v {
PacketOption::FragmentId(f) => *f != 0,
_ => false,
}) {
error!("fragmented packets are unsupported right now")
}
let mut locked = conn.0.lock().await;
let active_connection = locked.active_connection_data.as_mut()
.expect("we litterally just recieved a packet which requires the connection to be active, failing this should be impossible");
active_connection.reliable_client_counter = active_connection
.reliable_client_counter
.overflowing_add(1)
.0;
let Some(stream) = active_connection
.encryption_pairs
.get_mut(packet.header.substream_id as usize)
.map(|e| &mut e.recv)
else {
return;
};
stream.apply_keystream(&mut packet.payload);
drop(locked);
// we cant divert this off to another thread we HAVE to process it now to keep order
(self.on_data_handler)(packet, self.clone(), conn.0.clone()).await;
// ignored for now
}
} else {
error!("unreliable packets are unimplemented");
unimplemented!()
}
//info!("{:?}", packet);
}
PING => {
let ConnectionData {
active_connection_data,
server_signature,
..
} = &mut *connection;
info!("got ping");
if (packet.header.types_and_flags.get_flags() & NEED_ACK) != 0 {
let Some(active_connection) = active_connection_data.as_mut() else {
error!("got data packet on non active connection!");
return;
};
let mut ack = packet.base_acknowledgement_packet();
ack.header.session_id = active_connection.server_session_id;
ack.set_sizes();
let potential_session_key = active_connection
.active_secure_connection_data
.as_ref()
.map(|s| s.session_key);
ack.calculate_and_assign_signature(
self.access_key,
potential_session_key,
Some(*server_signature),
);
let mut vec = Vec::new();
ack.write_to(&mut vec)
.expect("somehow failed to convert backet to bytes");
self.socket
.send_to(&vec, client_address.regular_socket_addr)
.await
.expect("failed to send data back");
}
}
DISCONNECT => {
println!("got disconnect");
let Some(active_connection) = &connection.active_connection_data else {
return;
};
let mut ack = packet.base_acknowledgement_packet();
ack.header.session_id = active_connection.server_session_id;
ack.set_sizes();
let potential_session_key = active_connection
.active_secure_connection_data
.as_ref()
.map(|s| s.session_key);
ack.calculate_and_assign_signature(
self.access_key,
potential_session_key,
Some(connection.server_signature),
);
let mut vec = Vec::new();
ack.write_to(&mut vec)
.expect("somehow failed to convert backet to bytes");
self.socket
.send_to(&vec, client_address.regular_socket_addr)
.await
.expect("failed to send data back");
self.socket
.send_to(&vec, client_address.regular_socket_addr)
.await
.expect("failed to send data back");
self.socket
.send_to(&vec, client_address.regular_socket_addr)
.await
.expect("failed to send data back");
}
_ => error!(
"unimplemented packet type: {}",
packet.header.types_and_flags.get_types()
),
}
drop(mutual_exclusion_packet_handeling_mtx)
}*/
/*
impl ConnectionData {
pub async fn finish_and_send_packet_to(
&mut self,
socket: &SocketData,
mut packet: PRUDPPacket,
) {
let mut web = WEB_DATA.lock().await;
web.data.push((
self.sock_addr.regular_socket_addr,
Outgoing(hex::encode(&packet.payload)),
));
drop(web);
if (packet.header.types_and_flags.get_flags() & RELIABLE) != 0 {
let Some(active_connection) = self.active_connection_data.as_mut() else {
error!("tried to send a secure packet to an inactive connection");
return;
};
packet.header.sequence_id = active_connection.reliable_server_counter;
active_connection.reliable_server_counter += 1;
let Some(encryption) = active_connection
.encryption_pairs
.get_mut(packet.header.substream_id as usize)
.map(|e| &mut e.send)
else {
return;
};
encryption.apply_keystream(&mut packet.payload);
}
packet.header.session_id = self
.active_connection_data
.as_ref()
.map(|v| v.server_session_id)
.unwrap_or_default();
packet.header.source_port = socket.virtual_port;
packet.header.destination_port = self.sock_addr.virtual_port;
packet.set_sizes();
let potential_session_key = self
.active_connection_data
.as_ref()
.unwrap()
.active_secure_connection_data
.as_ref()
.map(|s| s.session_key);
packet.calculate_and_assign_signature(
socket.access_key,
potential_session_key,
Some(self.server_signature),
);
let mut vec = Vec::new();
packet
.write_to(&mut vec)
.expect("somehow failed to convert backet to bytes");
if let Err(e) = socket
.socket
.send_to(&vec, self.sock_addr.regular_socket_addr)
.await
{
error!("unable to send packet to destination: {}", e);
}
}
}*/
pub struct NewEncryptionPair<E: StreamCipher> {
pub send: E,
@ -538,9 +79,11 @@ impl<E: CryptoHandlerConnectionInstance> Deref for InternalConnection<E>{
impl<E: CryptoHandlerConnectionInstance> InternalConnection<E>{
fn next_server_count(&mut self) -> u16{
let prev_val = self.reliable_server_counter;
let (val, _) = self.reliable_server_counter.overflowing_add(1);
self.reliable_server_counter = val;
val
println!("{}", prev_val);
prev_val
}
}
@ -645,14 +188,14 @@ impl<T: CryptoHandlerConnectionInstance> AnyInternalConnection for InternalConne
self.crypto_handler_instance.sign_packet(&mut packet);
packet.set_sizes();
let mut vec = Vec::new();
packet
.write_to(&mut vec)
.expect("somehow failed to convert backet to bytes");
println!("{}", hex::encode(&vec));
self.socket
.send_to(&vec, self.socket_addr.regular_socket_addr)
.await
@ -670,8 +213,6 @@ impl<T: CryptoHandler> InternalSocket<T> {
.write_to(&mut vec)
.expect("somehow failed to convert backet to bytes");
println!("sent out: {}", hex::encode(&vec));
self.socket
.send_to(&vec, dest.regular_socket_addr)
.await
@ -824,12 +365,15 @@ impl<T: CryptoHandler> InternalSocket<T> {
let session_id = packet.header.session_id;
let (return_data, crypto) = self.crypto_handler.instantiate(
let Some((return_data, crypto)) = self.crypto_handler.instantiate(
remote_signature,
*own_signature,
&packet.payload,
1 + *max_substream,
);
) else {
error!("someone attempted to connect with invalid data");
return;
};
let mut response = packet.base_response_packet();
response.header.types_and_flags.set_types(CONNECT);
@ -895,12 +439,6 @@ impl<T: CryptoHandler> InternalSocket<T> {
mem::swap(&mut data, &mut packet.payload);
conn.data_sender.send(data).await.expect("socket died");
if packet.header.types_and_flags.get_flags() & NEED_ACK == 0{
return;
}
let mut response = packet.base_acknowledgement_packet();
response.header.types_and_flags.set_flag(HAS_SIZE | ACK);
response.header.session_id = conn.session_id;
@ -908,6 +446,10 @@ impl<T: CryptoHandler> InternalSocket<T> {
conn.crypto_handler_instance.sign_packet(&mut response);
self.send_packet_unbuffered(address, response).await;
conn.data_sender.send(data).await.ok();
}
async fn handle_ping(&self, address: PRUDPSockAddr, packet: PRUDPPacket){
@ -993,6 +535,8 @@ impl<T: CryptoHandler> AnyInternalSocket for InternalSocket<T> {
SYN => self.handle_syn(address, packet).await,
CONNECT => self.handle_connect(address, packet).await,
DATA => self.handle_data(address, packet).await,
DISCONNECT => self.handle_disconnect(address, packet).await,
PING => self.handle_ping(address, packet).await,
_ => {
error!(
"unimplemented packet type: {}",
@ -1068,7 +612,7 @@ impl<T: CryptoHandler> AnyInternalSocket for InternalSocket<T> {
return None;
};
let (_, crypt) = self.crypto_handler.instantiate(remote_signature, *own_signature, &[], 1);
let (_, crypt) = self.crypto_handler.instantiate(remote_signature, *own_signature, &[], 1)?;
//todo: make this work for secure servers as well
self.create_connection(crypt, address, 0).await;
@ -1130,7 +674,7 @@ pub trait CryptoHandler: Send + Sync + 'static {
own_signature: [u8; 16],
_: &[u8],
substream_count: u8,
) -> (Vec<u8>, Self::CryptoConnectionInstance);
) -> Option<(Vec<u8>, Self::CryptoConnectionInstance)>;
fn sign_pre_handshake(&self, packet: &mut PRUDPPacket);
}
@ -1162,6 +706,7 @@ impl ExternalConnection{
impl SendingConnection{
pub async fn send(&self, data: Vec<u8>) -> Option<()> {
println!("{}", hex::encode(&data));
let internal = self.inernal.upgrade()?;
let mut internal = internal.lock().await;
@ -1169,81 +714,4 @@ impl SendingConnection{
internal.send_data_packet(data).await;
Some(())
}
}
pub struct Unsecure(pub &'static str);
pub struct UnsecureInstance {
key: &'static str,
streams: Vec<EncryptionPair<Rc4<U5>>>,
self_signature: [u8; 16],
remote_signature: [u8; 16],
}
// my hand was forced to use lazy so that we can guarantee this code
// only runs once and so that i can put it here as a "constant" (for performance and readability)
// since for some reason rust crypto doesn't have any const time key initialization
static DEFAULT_KEY: Lazy<Key<U5>> = Lazy::new(|| Key::from(*b"CD&ML"));
impl CryptoHandler for Unsecure {
type CryptoConnectionInstance = UnsecureInstance;
fn instantiate(
&self,
remote_signature: [u8; 16],
self_signature: [u8; 16],
_: &[u8],
substream_count: u8,
) -> (Vec<u8>, Self::CryptoConnectionInstance) {
(
Vec::new(),
UnsecureInstance {
streams: (0..substream_count)
.map(|_| EncryptionPair::init_both(|| Rc4::new(&DEFAULT_KEY)))
.collect(),
key: self.0,
remote_signature,
self_signature,
},
)
}
fn sign_pre_handshake(&self, packet: &mut PRUDPPacket) {
packet.set_sizes();
packet.calculate_and_assign_signature(self.0, None, None);
}
}
impl CryptoHandlerConnectionInstance for UnsecureInstance {
type Encryption = Rc4<U5>;
fn decrypt_incoming(&mut self, substream: u8, data: &mut [u8]) {
if let Some(crypt_pair) = self.streams.get_mut(substream as usize){
crypt_pair.recv.apply_keystream(data);
}
}
fn encrypt_outgoing(&mut self, substream: u8, data: &mut [u8]) {
if let Some(crypt_pair) = self.streams.get_mut(substream as usize){
crypt_pair.send.apply_keystream(data);
}
}
fn get_user_id(&self) -> u32 {
0
}
fn sign_connect(&self, packet: &mut PRUDPPacket) {
packet.set_sizes();
packet.calculate_and_assign_signature(self.key, None, Some(self.self_signature));
}
fn sign_packet(&self, packet: &mut PRUDPPacket) {
packet.set_sizes();
packet.calculate_and_assign_signature(self.key, None, Some(self.self_signature));
}
fn verify_packet(&self, packet: &PRUDPPacket) -> bool {
true
}
}
}

84
src/prudp/unsecure.rs Normal file
View file

@ -0,0 +1,84 @@
use once_cell::sync::Lazy;
use rc4::{Key, KeyInit, Rc4, StreamCipher};
use typenum::U5;
use crate::prudp::packet::PRUDPPacket;
use crate::prudp::socket::{CryptoHandler, CryptoHandlerConnectionInstance, EncryptionPair};
pub struct Unsecure(pub &'static str);
pub struct UnsecureInstance {
key: &'static str,
streams: Vec<EncryptionPair<Rc4<U5>>>,
self_signature: [u8; 16],
remote_signature: [u8; 16],
}
// my hand was forced to use lazy so that we can guarantee this code
// only runs once and so that i can put it here as a "constant" (for performance and readability)
// since for some reason rust crypto doesn't have any const time key initialization
static DEFAULT_KEY: Lazy<Key<U5>> = Lazy::new(|| Key::from(*b"CD&ML"));
impl CryptoHandler for Unsecure {
type CryptoConnectionInstance = UnsecureInstance;
fn instantiate(
&self,
remote_signature: [u8; 16],
self_signature: [u8; 16],
_: &[u8],
substream_count: u8,
) -> Option<(Vec<u8>, Self::CryptoConnectionInstance)> {
Some((
Vec::new(),
UnsecureInstance {
streams: (0..substream_count)
.map(|_| EncryptionPair::init_both(|| Rc4::new(&DEFAULT_KEY)))
.collect(),
key: self.0,
remote_signature,
self_signature,
},
))
}
fn sign_pre_handshake(&self, packet: &mut PRUDPPacket) {
packet.set_sizes();
packet.calculate_and_assign_signature(self.0, None, None);
}
}
impl CryptoHandlerConnectionInstance for UnsecureInstance {
type Encryption = Rc4<U5>;
fn decrypt_incoming(&mut self, substream: u8, data: &mut [u8]) {
if let Some(crypt_pair) = self.streams.get_mut(substream as usize){
crypt_pair.recv.apply_keystream(data);
}
}
fn encrypt_outgoing(&mut self, substream: u8, data: &mut [u8]) {
if let Some(crypt_pair) = self.streams.get_mut(substream as usize){
crypt_pair.send.apply_keystream(data);
}
}
fn get_user_id(&self) -> u32 {
0
}
fn sign_connect(&self, packet: &mut PRUDPPacket) {
packet.set_sizes();
packet.calculate_and_assign_signature(self.key, None, Some(self.self_signature));
}
fn sign_packet(&self, packet: &mut PRUDPPacket) {
packet.set_sizes();
packet.calculate_and_assign_signature(self.key, None, Some(self.self_signature));
}
fn verify_packet(&self, packet: &PRUDPPacket) -> bool {
true
}
}

12
src/rmc/protocols/mod.rs
View file

@ -34,7 +34,7 @@ pub enum RemoteCallError {
ServerError(ErrorCode),
#[error("Connection broke")]
ConnectionBroke,
#[error("Error reading response data")]
#[error("Error reading response data: {0}")]
InvalidResponse(#[from] structures::Error),
}
@ -148,7 +148,7 @@ macro_rules! define_rmc_proto {
$($protocol:path),*
}) => {
paste::paste!{
trait [<Local $name>]: std::any::Any $( + [<Raw $protocol>] + $protocol)* {
pub trait [<Local $name>]: std::any::Any $( + [<Raw $protocol>] + $protocol)* {
async fn rmc_call(&self, remote_response_connection: &crate::prudp::socket::SendingConnection, protocol_id: u16, method_id: u32, call_id: u32, rest: Vec<u8>){
match protocol_id{
$(
@ -159,7 +159,7 @@ macro_rules! define_rmc_proto {
}
}
struct [<Remote $name>](crate::rmc::protocols::RmcConnection);
pub struct [<Remote $name>](crate::rmc::protocols::RmcConnection);
impl crate::rmc::protocols::RemoteInstantiatable for [<Remote $name>]{
fn new(conn: crate::rmc::protocols::RmcConnection) -> Self{
@ -235,7 +235,7 @@ async fn handle_incoming<T: RmcCallable + Send + Sync + 'static>(
return
};
if proto_id & 0x80 == 0{
if (proto_id & 0x80) == 0{
let Some(response) = RMCResponse::new(&mut Cursor::new(v)).display_err_or_some() else {
error!("ending rmc gateway.");
return
@ -260,9 +260,11 @@ async fn handle_incoming<T: RmcCallable + Send + Sync + 'static>(
rest_of_data
} = message;
info!("got rmc request, handeling it now...");
remote.rmc_call(&sending_conn, protocol_id, method_id, call_id, rest_of_data).await;
info!("got rmc request");
}
}
}

16
src/rmc/response.rs
View file

@ -36,7 +36,7 @@ pub struct RMCResponse {
impl RMCResponse {
pub fn new(stream: &mut (impl Seek + Read)) -> io::Result<Self>{
// ignore the size for now this will only be used for checking
let _: u32 = stream.read_struct(IS_BIG_ENDIAN)?;
let size: u32 = stream.read_struct(IS_BIG_ENDIAN)?;
let protocol_id: u8 = stream.read_struct(IS_BIG_ENDIAN)?;
@ -54,9 +54,9 @@ impl RMCResponse {
let method_id: u32 = stream.read_struct(IS_BIG_ENDIAN)?;
let method_id = method_id & (!0x8000);
let mut data: Vec<u8> = Vec::new();
let mut data: Vec<u8> = vec![0u8; (size - 2 - 4 - 4) as _];
stream.read_to_end(&mut data)?;
stream.read(&mut data)?;
RMCResponseResult::Success {
@ -154,15 +154,13 @@ pub async fn send_result(
method_id: u32,
call_id: u32,
) {
println!("{}", hex::encode(result.clone().unwrap()));
let response_result = match result {
Ok(v) => RMCResponseResult::Success {
call_id,
method_id,
data: {
let mut vec = Vec::new();
v.serialize(&mut vec).expect("serialization error");
vec
}
data: v
},
Err(e) =>
RMCResponseResult::Error {
@ -187,7 +185,7 @@ pub async fn send_response(connection: &SendingConnection, rmcresponse: RMCRespo
//taken from kinnays error list directly
#[allow(nonstandard_style)]
#[repr(u32)]
#[derive(Debug, EnumTryInto)]
#[derive(Debug, EnumTryInto, Clone, Copy)]
pub enum ErrorCode {
Core_Unknown = 0x00010001,
Core_NotImplemented = 0x00010002,

27
src/rmc/structures/connection_data.rs
View file

@ -1,30 +1,15 @@
use std::io::{Read, Write};
use bytemuck::bytes_of;
use macros::RmcSerialize;
use crate::kerberos::KerberosDateTime;
use crate::rmc::structures::{rmc_struct, RmcSerialize};
#[derive(Debug)]
pub struct ConnectionData<'a>{
pub station_url: &'a str,
#[derive(Debug, RmcSerialize)]
#[rmc_struct(1)]
pub struct ConnectionData{
pub station_url: String,
pub special_protocols: Vec<u8>,
pub special_station_url: &'a str,
pub special_station_url: String,
pub date_time: KerberosDateTime
}
impl<'a> RmcSerialize for ConnectionData<'a>{
fn serialize(&self, writer: &mut dyn Write) -> crate::rmc::structures::Result<()> {
rmc_struct::write_struct(writer, 1, |v|{
self.station_url.serialize(v).expect("unable to write station url");
self.special_protocols.serialize(v).expect("unable to write special protocols");
self.special_station_url.serialize(v).expect("unable to write special station url");
v.write_all(bytes_of(&self.date_time)).expect("unable to write date time");
Ok(())
})
}
fn deserialize(_reader: &mut dyn Read) -> crate::rmc::structures::Result<Self> {
todo!()
}
}

2
src/rmc/structures/rmc_struct.rs
View file

@ -17,7 +17,7 @@ pub fn write_struct(writer: &mut dyn Write, version: u8, pred: impl FnOnce(&mut
(pred)(&mut scratch_space)?;
let u32_size= scratch_space.len() as u32;
let u32_size = scratch_space.len() as u32;
writer.write_all(bytes_of(&u32_size))?;
writer.write_all(&scratch_space)?;