Sanae6/smo-server
1
0
Fork 0
Code Issues Pull requests Actions Packages Projects Releases Wiki Activity

proximity server beginnings

Browse source
This commit is contained in:
Aubrey 2024-12-17 18:29:22 -06:00
parent cf709fb43c
commit 1963549b45
No known key found for this signature in database
7 changed files with 1740 additions and 380 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
.gitignore vendored
View file

@ -1,2 +1,4 @@
/target
/.direnv
*.pem
*.p12

1459
Cargo.lock generated
View file

File diff suppressed because it is too large Load diff

9
Cargo.toml
View file

@ -4,11 +4,18 @@ version = "0.1.0"
edition = "2021"
[dependencies]
anyhow = "1.0.94"
anyhow = { version = "1.0.94", features = ["backtrace"] }
crc32fast = "1.4.2"
glam = "0.29.2"
heapless = "0.8.0"
newtype-enum = { git = "https://github.com/mgjm/newtype-enum", version = "0.1.0" }
tokio = { version = "1.42.0", features = ["rt", "macros", "net", "sync", "rt-multi-thread", "io-util", "io-std"] }
tracing = "0.1.41"
tracing-subscriber = { version = "0.3.19", features = ["env-filter"] }
uuid = { version = "1.11.0", features = ["v4"] }
wtransport = "0.5.0"
xtra = { version = "0.6.0", features = ["macros", "tokio"] }
zerocopy = { version = "0.8.13", features = ["derive"] }
[dev-dependencies]
roead = "1.0.0"

1
flake.nix
View file

@ -21,6 +21,7 @@
devShells.default = mkShell {
buildInputs = [
pkgs.fenix.stable.completeToolchain
cmake
pkg-config
openssl
];

92
src/main.rs
View file

@ -1,38 +1,104 @@
pub mod faker;
pub mod game;
pub mod packet;
pub mod player;
pub mod protocol;
pub mod server;
use std::{collections::HashMap, sync::LazyLock};
use std::{
collections::HashMap,
io::Cursor,
net::{SocketAddr, ToSocketAddrs},
sync::{Arc, LazyLock},
};
use faker::Faker;
use packet::{rw::read_packet, Packet};
use player::PlayerActor;
use tokio::{net::TcpListener, sync::RwLock};
use tracing::{error, info};
use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
use xtra::Address;
use server::web_main;
use tokio::{
net::{TcpListener, UdpSocket},
sync::RwLock,
};
use tracing::{error, info, Level};
use tracing_subscriber::{filter::FilterFn, layer::SubscriberExt, util::SubscriberInitExt, Layer};
use xtra::{prelude::MessageChannel, Mailbox};
pub fn clients() -> &'static RwLock<HashMap<u128, Address<PlayerActor>>> {
static CLIENTS: LazyLock<RwLock<HashMap<u128, Address<PlayerActor>>>> = LazyLock::new(|| RwLock::default());
type PacketChannel = MessageChannel<Packet, (), xtra::refcount::Strong>;
pub fn clients() -> &'static RwLock<HashMap<u128, (PacketChannel, SocketAddr)>> {
static CLIENTS: LazyLock<RwLock<HashMap<u128, (PacketChannel, SocketAddr)>>> = LazyLock::new(|| RwLock::default());
&CLIENTS
}
pub async fn broadcast_packet(packet: Packet) {
for (id, (address, _)) in clients().read().await.iter() {
if *id != packet.user_id {
let _ = address.send(packet.clone()).detach().await.unwrap();
}
}
}
#[tokio::main]
async fn main() {
//.with(EnvFilter::from_default_env())
tracing_subscriber::registry().with(tracing_subscriber::fmt::layer()).init();
tracing_subscriber::registry()
.with(tracing_subscriber::fmt::layer().with_filter(FilterFn::new(|meta| {
*meta.level() < Level::INFO || meta.module_path().unwrap().contains("smo_server")
})))
.init();
let tcp = TcpListener::bind("0.0.0.0:1027").await.unwrap();
let manager = web_main();
{
let (address, mailbox) = Mailbox::unbounded();
let address = xtra::spawn_tokio(
Faker {
address: address.clone(),
},
(address, mailbox),
);
clients().write().await.insert(
1,
(
MessageChannel::new(address),
"0.0.0.0:1027".to_socket_addrs().unwrap().next().unwrap(),
),
);
}
let socket = Arc::new(UdpSocket::bind("0.0.0.0:1027").await.unwrap());
tokio::spawn({
let socket = socket.clone();
async move {
loop {
let mut packet = [0u8; 256];
let (size, _addr) = socket.recv_from(&mut packet).await.unwrap();
// trace!("got packet from {_addr:?}");
match read_packet(&mut Cursor::new(&mut packet[..size]), true).await {
Ok(packet) => {
// info!("packet from udp {packet:?}");
if let Some(client) = clients().read().await.get(&packet.user_id) {
let _ = client.0.send(packet).detach().await;
}
}
Err(error) => {
error!("udp packet error: {error:?}");
}
};
}
}
});
info!("listening on port 1027");
loop {
match tcp.accept().await {
Ok((stream, addr)) => {
PlayerActor::new_connection(stream, addr);
PlayerActor::new_connection(stream, addr, socket.clone(), manager.clone());
}
Err(error) => {
error!("failed to handle connection: {error:?}");
}
}
}
// let socket = Arc::new(UdpSocket::bind("0.0.0.0:1027").await.unwrap());
// let mut packet = [0u8; 1024];
}

213
src/packet.rs
View file

@ -1,213 +0,0 @@
use core::str;
use std::{
ffi::CStr,
fmt::{Debug, Display},
};
use anyhow::{bail, Context};
use newtype_enum::newtype_enum;
use zerocopy::{
FromBytes, FromZeros, Immutable, IntoBytes, KnownLayout, Unaligned,
};
#[derive(Debug, Clone, Copy, FromZeros, IntoBytes, KnownLayout, Immutable)]
#[repr(C, packed)]
pub struct PacketHeader {
pub user_id: u128,
pub kind: PacketKind,
pub size: u16,
}
#[derive(Debug, Clone, Copy, FromZeros, IntoBytes, KnownLayout, Immutable)]
#[repr(u16)]
pub enum PacketKind {
Unknown = 0,
Init = 1,
Player = 2,
Cap = 3,
Game = 4,
Tag = 5,
Connect = 6,
Disconnect = 7,
Costume = 8,
Shine = 9,
Capture = 10,
ChangeStage = 11,
Command = 12,
UdpInit = 13,
HolePunch = 14
}
const COSTUME_NAME_SIZE: usize = 0x20;
const CAP_ANIM_SIZE: usize = 0x30;
const STAGE_GAME_NAME_SIZE: usize = 0x40;
const STAGE_CHANGE_NAME_SIZE: usize = 0x30;
const STAGE_ID_SIZE: usize = 0x10;
const CLIENT_NAME_SIZE: usize = COSTUME_NAME_SIZE;
#[derive(Debug)]
pub struct Packet {
pub user_id: u128,
pub data: PacketData,
}
#[newtype_enum]
#[derive(Debug)]
pub enum PacketData {
Unknown(Vec<u8>),
#[derive(FromZeros, IntoBytes, KnownLayout, Immutable)]
#[repr(C, packed)]
Init {
pub max_players: u16,
},
#[derive(FromZeros, IntoBytes, KnownLayout, Immutable)]
#[repr(C, packed)]
Player {
position: [f32; 3],
rotation: [f32; 4],
weights: [f32; 6],
action: u16,
subaction: u16,
},
#[derive(FromZeros, IntoBytes, KnownLayout, Immutable)]
#[repr(C, packed)]
Cap {
position: [f32; 3],
rotation: [f32; 4],
out: Bool,
anim: String<CAP_ANIM_SIZE>,
},
#[derive(FromZeros, IntoBytes, KnownLayout, Immutable)]
#[repr(C, packed)]
Game {
is_2d: u8,
scenario_num: u8,
stage: String<STAGE_GAME_NAME_SIZE>,
},
#[derive(FromZeros, IntoBytes, KnownLayout, Immutable)]
#[repr(C, packed)]
Tag {
update_type: u8,
is_it: Bool,
seconds: u8,
minutes: u16,
},
#[derive(FromZeros, IntoBytes, KnownLayout, Immutable)]
#[repr(C, packed)]
Connect {
kind: ConnectionKind,
max_player: u16,
client_name: String<CLIENT_NAME_SIZE>,
},
Disconnect,
#[derive(FromZeros, IntoBytes, KnownLayout, Immutable)]
#[repr(C, packed)]
Costume {
body_name: String<COSTUME_NAME_SIZE>,
cap_name: String<COSTUME_NAME_SIZE>,
},
#[derive(FromZeros, IntoBytes, KnownLayout, Immutable)]
#[repr(C, packed)]
Shine {
shine_id: i32,
is_grand: Bool,
},
#[derive(FromZeros, IntoBytes, KnownLayout, Immutable)]
#[repr(C, packed)]
Capture {
model: String<COSTUME_NAME_SIZE>,
},
#[derive(FromZeros, IntoBytes, KnownLayout, Immutable)]
#[repr(C, packed)]
ChangeStage {
stage: String<STAGE_CHANGE_NAME_SIZE>,
id: String<STAGE_ID_SIZE>,
scenario: i8,
sub_scenario: u8,
},
Command,
#[derive(FromZeros, IntoBytes, KnownLayout, Immutable)]
#[repr(C, packed)]
UdpInit {
port: u16
},
HolePunch
}
pub enum TagUpdateBit {
Time = 0,
State = 1,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq, FromZeros, IntoBytes, KnownLayout, Immutable)]
#[repr(u32)]
pub enum ConnectionKind {
New = 0,
Old = 1,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq, FromBytes, IntoBytes, KnownLayout, Immutable, Unaligned)]
#[repr(C)]
pub struct Bool(u8);
impl Bool {
pub fn new(value: bool) -> Bool {
Bool(if value { 1 } else { 0 })
}
pub fn get(&self) -> bool {
self.0 != 0
}
}
impl From<bool> for Bool {
fn from(value: bool) -> Self {
Bool::new(value)
}
}
impl From<Bool> for bool {
fn from(value: Bool) -> Self {
value.get()
}
}
#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord, FromBytes, IntoBytes, KnownLayout, Immutable, Unaligned)]
#[repr(C)]
pub struct String<const N: usize>([u8; N]);
impl<const N: usize> String<N> {
pub fn to_str(&self) -> anyhow::Result<&str> {
let cstr = CStr::from_bytes_until_nul(&self.0).context("interpreting bytes as c-string")?;
cstr.to_str().context("verifying string has utf-8")
// let str = str::from_utf8(&self.0).context("verifying string has utf-8")?;
// Ok(str.trim_end_matches('\0'))
}
pub fn assert_valid(&self) -> anyhow::Result<()> {
self.to_str().map(drop)
}
}
impl<const N: usize> TryFrom<&str> for String<N> {
type Error = anyhow::Error;
fn try_from(value: &str) -> Result<Self, Self::Error> {
let mut buf = [0; N];
if value.len() > N {
bail!("seggs")
}
value.write_to_prefix(&mut buf).unwrap();
Ok(Self(buf))
}
}
impl<const N: usize> Display for String<N> {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
Display::fmt(self.to_str().expect("failed to parse string"), f)
}
}
impl<const N: usize> Debug for String<N> {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
Debug::fmt(self.to_str().expect("failed to parse string"), f)
}
}

334
src/player.rs
View file

@ -1,183 +1,231 @@
use std::{net::SocketAddr, time::Duration};
use std::{io::Cursor, net::SocketAddr, sync::Arc};
use anyhow::{bail, Context};
use glam::Vec3;
use tokio::{
io::{AsyncReadExt, AsyncWriteExt, BufReader, BufWriter},
net::{
tcp::{OwnedReadHalf, OwnedWriteHalf},
TcpStream,
}, sync::mpsc, time::Instant,
io::{BufReader, BufWriter},
net::{TcpStream, UdpSocket},
sync::mpsc,
};
use tracing::{info, info_span, trace, Instrument};
use zerocopy::{IntoBytes, TryFromBytes};
use tracing::{error, info, info_span, trace, Instrument};
use xtra::{prelude::MessageChannel, scoped, Actor, Address, Handler, Mailbox};
use crate::packet::{
Packet, PacketData,
PacketData_variants::{Cap, Command, Connect, Disconnect, HolePunch, Init, Player},
PacketHeader, PacketKind, String,
use crate::{
broadcast_packet, clients,
packet::{
rw::{read_packet, write_packet},
ConnectionKind, Packet, PacketData,
PacketData_variants::{Connect, HolePunch, Init, UdpInit},
CLIENT_NAME_SIZE,
},
protocol::String,
server::{ChangedStage, Manager, PlayerConnected, PlayerMoved},
};
// #[derive(Actor)]
pub struct PlayerActor {
write_half: BufWriter<OwnedWriteHalf>,
id: u128,
connection_kind: ConnectionKind,
name: String<CLIENT_NAME_SIZE>,
write_sender: mpsc::UnboundedSender<WriteMessage>,
manager: Address<Manager>,
}
enum WriteMessage {
Data(Packet),
SetUdp(u16),
}
impl PlayerActor {
pub async fn read_packet(reader: &mut BufReader<OwnedReadHalf>) -> anyhow::Result<Packet> {
let mut header = [0; size_of::<PacketHeader>()];
reader.read_exact(&mut header).await.context("reading data")?;
let Ok(header) = PacketHeader::try_read_from_bytes(&header) else {
bail!("parsing packet buffer")
};
macro_rules! read_data {
($read: expr, $size: expr, $ty: ident $(, $field:ident => $assert: expr)*) => {{
async fn read_data(reader: &mut BufReader<OwnedReadHalf>, size: u16) -> anyhow::Result<PacketData> {
type T = crate::packet::PacketData_variants::$ty;
let size = size as usize;
if size < size_of::<T>() {
bail!("buffer too small for packet: expected {}, got {size}", size_of::<T>())
}
let mut data = [0; u16::MAX as usize];
reader.read_exact(&mut data[..size]).await.context("reading data")?;
let packet_data = match T::try_read_from_bytes(&data[..size_of::<T>()]) {
Ok(data) => data,
Err(error) => {
bail!(concat!("interpreting ", stringify!($ty), ": {:?}"), error)
}
};
$(
{
let $field = packet_data.$field;
$assert
}?;
)*
Ok(PacketData::$ty(packet_data))
}
type _FixIntellisense = crate::packet::PacketData_variants::$ty;
read_data($read, $size).await?
}};
}
let data: PacketData = match header.kind {
PacketKind::Unknown => {
let mut data = vec![0; header.size.into()];
reader.read_exact(&mut data).await.context("reading unknown data")?;
PacketData::Unknown(data)
}
PacketKind::Init => read_data!(reader, header.size, Init),
PacketKind::Player => read_data!(reader, header.size, Player),
PacketKind::Cap => read_data!(reader, header.size, Cap, anim => anim.assert_valid()),
PacketKind::Game => read_data!(reader, header.size, Game, stage => stage.assert_valid()),
PacketKind::Tag => read_data!(reader, header.size, Tag),
PacketKind::Connect => read_data!(reader, header.size, Connect),
PacketKind::Disconnect => PacketData::Disconnect(Disconnect),
PacketKind::Costume => read_data!(reader, header.size, Costume),
PacketKind::Shine => read_data!(reader, header.size, Shine),
PacketKind::Capture => read_data!(reader, header.size, Capture),
PacketKind::ChangeStage => read_data!(reader, header.size, ChangeStage),
PacketKind::Command => PacketData::Command(Command),
PacketKind::UdpInit => read_data!(reader, header.size, UdpInit),
PacketKind::HolePunch => PacketData::HolePunch(HolePunch),
};
Ok(Packet {
user_id: header.user_id,
data,
})
}
pub async fn write_packet(writer: &mut BufWriter<OwnedWriteHalf>, id: u128, data: PacketData) -> anyhow::Result<()> {
let (kind, slice) = match &data {
PacketData::Unknown(vec) => {
if vec.len() >= (u16::MAX as usize) {
bail!("unknown packet vec too large")
}
(PacketKind::Unknown, vec.as_slice())
}
PacketData::Init(init) => (PacketKind::Init, init.as_bytes()),
PacketData::Player(player) => (PacketKind::Player, player.as_bytes()),
PacketData::Cap(cap) => (PacketKind::Cap, cap.as_bytes()),
PacketData::Game(game) => (PacketKind::Game, game.as_bytes()),
PacketData::Tag(tag) => (PacketKind::Tag, tag.as_bytes()),
PacketData::Connect(connect) => (PacketKind::Connect, connect.as_bytes()),
PacketData::Disconnect(..) => (PacketKind::Disconnect, [].as_slice()),
PacketData::Costume(costume) => (PacketKind::Costume, costume.as_bytes()),
PacketData::Shine(shine) => (PacketKind::Shine, shine.as_bytes()),
PacketData::Capture(capture) => (PacketKind::Capture, capture.as_bytes()),
PacketData::ChangeStage(change_stage) => (PacketKind::ChangeStage, change_stage.as_bytes()),
PacketData::Command(..) => (PacketKind::Command, [].as_slice()),
PacketData::UdpInit(udp_init) => (PacketKind::UdpInit, udp_init.as_bytes()),
PacketData::HolePunch(..) => (PacketKind::HolePunch, [].as_slice()),
};
writer
.write_all(
PacketHeader {
kind,
size: slice.len() as u16,
user_id: id,
}
.as_bytes(),
)
.await
.context("writing header")?;
writer.write_all(slice).await.context("writing data")?;
writer.flush().await.context("flushing writer")
}
pub fn new_connection(stream: TcpStream, addr: SocketAddr) {
pub fn new_connection(stream: TcpStream, addr: SocketAddr, socket: Arc<UdpSocket>, manager: Address<Manager>) {
tokio::spawn(async move {
info!("accepted connection from {addr}");
let (reader, writer) = stream.into_split();
let mut reader = BufReader::new(reader);
let mut writer = BufWriter::new(writer);
let packet = Self::read_packet(&mut reader).await.unwrap();
let packet = read_packet(&mut reader, false).await.unwrap();
let Packet {
user_id,
data: PacketData::Connect(connect),
..
} = packet
else {
tracing::error!("not a valid player!");
return;
};
let bot_id = user_id + 1;
Self::write_packet(&mut writer, user_id, PacketData::Init(Init { max_players: 8 }))
.await
.expect("msfrarausfhsdagsdgkog");
Self::write_packet(&mut writer, bot_id, PacketData::Connect(Connect { kind: crate::packet::ConnectionKind::New, client_name: String::try_from("Bot").unwrap(), max_player: 8 }))
write_packet(&mut writer, user_id, PacketData::Init(Init { max_players: 8 }))
.await
.expect("msfrarausfhsdagsdgkog");
let span = info_span!("", player = connect.client_name.to_string());
let span = info_span!("Client", player = connect.client_name.to_string());
let (address, mailbox) = Mailbox::bounded(512);
let (sender, mut receiver) = mpsc::unbounded_channel();
tokio::spawn(async move {
while let Some((instant, packet)) = receiver.recv().await {
tokio::time::sleep_until(instant).await;
Self::write_packet(&mut writer, bot_id, packet).await.unwrap();
}
}.instrument(span.clone()));
tokio::spawn(scoped(&address.downgrade(), {
let address = address.downgrade();
async move {
let ip = addr.ip();
let mut addr = None;
while let Some(message) = receiver.recv().await {
match message {
WriteMessage::Data(Packet { user_id, udp, data }) => {
// trace!("writing packet {udp:?}, {data:?}");
let res = if udp {
async {
let Some(addr) = addr else {
trace!("no address set yet");
return Ok(());
};
// trace!("sending udp packet: {data:?}");
let mut buf = [0; 256];
let mut writer = Cursor::new(buf.as_mut_slice());
write_packet(&mut writer, user_id, data).await?;
socket.send_to(&buf, &addr).await.expect("kys");
Ok(())
}
.await
} else {
write_packet(&mut writer, user_id, data).await
};
if let Err(error) = res {
error!("error while writing packet: {error:?}");
let _ = address.send(StopError).detach().await;
return;
}
}
WriteMessage::SetUdp(port) => {
trace!("set udp port, connected on udp!");
addr = Some(SocketAddr::new(ip, port))
}
}
}
}
.instrument(span.clone())
}));
tokio::spawn(scoped(
&address.downgrade(),
{
let address = address.downgrade();
async move {
info!("connected {user_id}");
loop {
let packet = Self::read_packet(&mut reader).await.unwrap();
trace!("got packet {packet:?}");
let sender = sender.clone();
tokio::spawn(async move {
let _ = sender.send((Instant::now().checked_add(Duration::from_millis(200)).unwrap(), match packet.data {
PacketData::Player(player) => PacketData::Player(Player {position: [player.position[0], -20., player.position[2]], ..player}),
PacketData::Cap(cap) => PacketData::Cap(Cap {position: [cap.position[0], -20., cap.position[2]], ..cap}),
pass => pass,
}));
});
match read_packet(&mut reader, false).await {
Ok(packet) => {
let _ = address.send(packet).detach().await;
}
Err(error) => {
error!("error while reading packet: {error:?}");
let _ = address.send(StopError).detach().await;
break;
}
}
}
}
}
.instrument(span.clone()),
));
clients().write().await.insert(user_id, (MessageChannel::new(address), addr));
xtra::run(
mailbox,
PlayerActor {
id: user_id,
connection_kind: connect.kind,
name: connect.client_name,
write_sender: sender,
manager,
},
)
.instrument(span)
.await;
});
}
}
impl Actor for PlayerActor {
type Stop = ();
async fn started(&mut self, _: &Mailbox<Self>) -> Result<(), Self::Stop> {
broadcast_packet(Packet {
user_id: self.id,
udp: false,
data: PacketData::Connect(Connect {
kind: self.connection_kind,
max_player: 8,
client_name: self.name,
}),
})
.await;
self
.write_sender
.send(WriteMessage::Data(Packet {
user_id: 0,
udp: false,
data: PacketData::UdpInit(UdpInit { port: 1027 }),
}))
.map_err(drop)?;
self
.manager
.send(PlayerConnected {
id: self.id,
name: self.name,
})
.await
.unwrap();
Ok(())
}
async fn stopped(self) -> Self::Stop {}
}
struct StopError;
impl Handler<StopError> for PlayerActor {
type Return = ();
async fn handle(&mut self, _: StopError, ctx: &mut xtra::Context<Self>) -> Self::Return {
ctx.stop_all();
}
}
impl Handler<Packet> for PlayerActor {
type Return = ();
async fn handle(&mut self, packet: Packet, _: &mut xtra::Context<Self>) {
if matches!(packet.data, PacketData::HolePunch(..)) {
trace!("hole puncher");
return;
}
if packet.user_id == self.id {
match packet.data {
PacketData::UdpInit(UdpInit { port }) => {
trace!("hole puncher");
let _ = self.write_sender.send(WriteMessage::SetUdp(port));
let _ = self.write_sender.send(WriteMessage::Data(Packet {
user_id: 0,
udp: true,
data: PacketData::HolePunch(HolePunch),
}));
return;
}
PacketData::Connect(..) | PacketData::Init(..) => {
trace!("suspicious packet sent: {packet:?}");
return;
}
PacketData::Player(ref player) => {
let _ = self.manager.send(PlayerMoved { id: self.id, position: Vec3::from_array(player.position)}).detach().await.unwrap();
}
PacketData::Game(ref game) => {
let _ = self.manager.send(ChangedStage { id: self.id, stage: game.stage}).detach().await.unwrap();
}
_ => {}
}
broadcast_packet(packet).await;
} else {
let _ = self.write_sender.send(WriteMessage::Data(packet));
}
}
}