implemented user and started security

harsh-server/src/gateway.rs

@@ -1,7 +1,10 @@
-use harsh_common::{client, server, ClientRequest, ServerRequest};
+use harsh_common::{client, server, ClientRequest, ServerEvent};
 use telecomande::{Processor, Remote};
 
-use crate::{Addr, Id, SessionCmd, SessionProc, StorageCmd, StorageProc};
+use crate::{
+    sessions::SessionExt, storage::Perm, Addr, Id, SecurityCmd, SecurityProc, SessionCmd,
+    SessionProc, StorageCmd, StorageProc,
+};
 
 #[derive(Debug)]
 pub enum GatewayCmd {
@@ -12,193 +15,246 @@ pub enum GatewayCmd {
 pub struct GatewayProc {
     sessions: Remote<SessionProc>,
     storage: Remote<StorageProc>,
+    security: Remote<SecurityProc>,
 }
 
+use client::*;
+
 impl GatewayProc {
-    async fn handle_request(&mut self, address: Addr, request: ClientRequest) {
-        use client as c;
-        use ClientRequest::*;
+    async fn handle_request(&mut self, addr: Addr, request: ClientRequest) -> Result<(), String> {
+        use client::*;
+        use ClientRequest as CR;
+
+        // auth-free API
+        let request = match request {
+            CR::Ping(ping) => return self.on_ping(ping, addr),
+            CR::Authenticate(authenticate) => {
+                return self.on_authenticate(authenticate, addr).await
+            }
+            _ => request,
+        };
+
+        // let user = self.sessions.get_user(addr.clone()).await.ok_or("owno")?;
+
+        // auth API
         match request {
-            Ping(c::Ping { content }) => self.on_ping(content, address),
+            CR::Ping(_) | CR::Authenticate(_) => unreachable!(),
 
-            ChannelCreate(c::ChannelCreate { name }) => self.on_channel_create(name).await,
-            ChannelDelete(c::ChannelDelete { id }) => self.on_channel_delete(id),
-            ChannelList(c::ChannelList {}) => self.on_channel_list(address).await,
-            ChannelGetName(c::ChannelGetName { id }) => self.on_channel_get_name(id, address).await,
-            ChannelSetName(c::ChannelSetName { id, name }) => self.on_channel_set_name(id, name),
+            CR::ChannelCreate(req) => self.on_channel_create(req).await,
+            CR::ChannelDelete(req) => todo!(), //self.on_channel_delete(req, user).await,
+            CR::ChannelList(req) => self.on_channel_list(req, addr).await,
+            CR::ChannelGetName(req) => self.on_channel_get_name(req, addr).await,
+            CR::ChannelSetName(req) => self.on_channel_set_name(req),
 
-            MessageList(c::MessageList { channel_id }) => {
-                self.on_message_list(channel_id, address).await
-            }
-            MessageCreate(c::MessageCreate {
-                channel_id,
-                content,
-            }) => self.on_message_create(channel_id, content).await,
-            MessageDelete(c::MessageDelete { channel_id, id }) => {
-                self.on_message_delete(channel_id, id)
-            }
-            MessageGetContent(c::MessageGetContent { channel_id, id }) => {
-                self.on_message_get_content(channel_id, id, address).await
-            }
+            CR::MessageList(req) => self.on_message_list(req, addr).await,
+            CR::MessageCreate(req) => self.on_message_create(req).await,
+            CR::MessageDelete(req) => self.on_message_delete(req),
+            CR::MessageGetContent(req) => self.on_message_get_content(req, addr).await,
+            CR::MessageSetContent(req) => self.on_message_set_content(req),
 
-            MessageSetContent(c::MessageSetContent {
-                channel_id,
-                id,
-                content,
-            }) => {
-                self.on_message_set_content(channel_id, id, content);
-            }
+            CR::UserList(req) => self.on_user_list(req, addr).await,
+            CR::UserCreate(req) => self.on_user_create(req).await,
+            CR::UserDelete(req) => self.on_user_delete(req),
+            CR::UserGetName(req) => self.on_user_get_name(req, addr).await,
+            CR::UserSetName(req) => self.on_user_set_name(req),
+            CR::UserSetPass(req) => self.on_user_set_pass(req, addr),
+        };
+        Ok(())
+    }
 
-            // TODO: user
-            UserList(c::UserList {}) => {
-                let (cmd, rec) = StorageCmd::new_user_list();
-                self.storage.send(cmd).unwrap();
-                let result = rec.await.unwrap().iter().map(Id::to_u64).collect();
-                let request = ServerRequest::new_user_list(result);
-                let command = SessionCmd::new_send(address, request);
-                self.sessions.send(command).unwrap();
-            }
+    async fn on_authenticate(
+        &mut self,
+        Authenticate { id, pass }: Authenticate,
+        address: Addr,
+    ) -> Result<(), String> {
+        let (cmd, rec) = SecurityCmd::new_authenticate(id.into(), pass);
+        self.security.send(cmd).unwrap();
+        if rec.await.unwrap() {
+            let command = SessionCmd::new_set_user(address, Some(id.into()));
+            self.sessions.send(command).unwrap();
+        } else {
+            Err("Invalid password")?;
+        };
+        Ok(())
+    }
 
-            UserCreate(c::UserCreate { name, pass }) => {
-                let (cmd, rec) = StorageCmd::new_user_create(name.clone(), pass);
-                self.storage.send(cmd).unwrap();
-                let id = rec.await.unwrap();
-                let request = ServerRequest::new_user_create(id.into(), name);
-                let command = SessionCmd::new_broadcast(request);
-                self.sessions.send(command).unwrap();
-            }
-
-            UserDelete(c::UserDelete { id }) => {
-                let command = StorageCmd::new_user_delete(id.into());
-                self.storage.send(command).unwrap();
-                let request = ServerRequest::new_user_delete(id.into());
-                let command = SessionCmd::new_broadcast(request);
-                self.sessions.send(command).unwrap();
-            }
-
-            UserGetName(c::UserGetName { id }) => {
-                let (cmd, rec) = StorageCmd::new_user_get_name(id.into());
-                self.storage.send(cmd).unwrap();
-                let name = rec.await.unwrap();
-                let request = ServerRequest::new_user_get_name(id.into(), name);
-                let command = SessionCmd::new_send(address, request);
-                self.sessions.send(command).unwrap();
-            }
-
-            UserSetName(c::UserSetName { id, name }) => {
-                let command = StorageCmd::new_user_set_name(id.into(), name.clone());
-                self.storage.send(command).unwrap();
-                let request = ServerRequest::new_user_set_name(id.into(), name);
-                let command = SessionCmd::new_broadcast(request);
-                self.sessions.send(command).unwrap();
-            }
-
-            UserSetPass(c::UserSetPass { id, pass }) => {
-                let command = StorageCmd::new_user_set_pass(id.into(), pass);
-                self.storage.send(command).unwrap();
-                let request = ServerRequest::new_user_set_pass(id.into());
-                let command = SessionCmd::new_send(address, request);
-                self.sessions.send(command).unwrap();
-            }
+    pub fn new(
+        sessions: Remote<SessionProc>,
+        storage: Remote<StorageProc>,
+        security: Remote<SecurityProc>,
+    ) -> Self {
+        Self {
+            sessions,
+            storage,
+            security,
         }
     }
 
-    pub fn new(sessions: Remote<SessionProc>, storage: Remote<StorageProc>) -> Self {
-        Self { sessions, storage }
-    }
-
-    fn on_ping(&mut self, content: String, address: Addr) {
+    fn on_ping(&mut self, Ping { content }: Ping, address: Addr) -> Result<(), String> {
         println!("[gateway/PING] '{content:?}'");
-        let request = ServerRequest::Pong(server::Pong { content });
+        let request = ServerEvent::Pong(server::Pong { content });
         let command = SessionCmd::new_send(address, request);
         self.sessions.send(command).unwrap();
+        Ok(())
     }
 
-    async fn on_channel_create(&mut self, name: String) {
+    async fn on_channel_create(&mut self, ChannelCreate { name }: ChannelCreate) {
         let (cmd, rec) = StorageCmd::new_channel_create(name.clone());
         self.storage.send(cmd).unwrap();
         let id = rec.await.unwrap().to_u64();
-        let request = ServerRequest::new_channel_create(id, name);
+        let request = ServerEvent::new_channel_create(id, name);
         let command = SessionCmd::new_broadcast(request);
         self.sessions.send(command).unwrap();
     }
 
-    fn on_channel_delete(&mut self, id: u64) {
+    async fn on_channel_delete(&mut self, ChannelDelete { id }: ChannelDelete, user: Id) {
+        // TODO: verify is OP
+        let (cmd, req) = SecurityCmd::new_verify(user, Perm::OpChannel(id.into()));
+        self.security.send(cmd).unwrap();
+        req.await.unwrap();
         let command = StorageCmd::new_channel_delete(id.into());
         self.storage.send(command).unwrap();
-        let request = ServerRequest::new_channel_delete(id);
+        let request = ServerEvent::new_channel_delete(id);
         let command = SessionCmd::new_broadcast(request);
         self.sessions.send(command).unwrap();
     }
 
-    async fn on_channel_list(&mut self, address: Addr) {
+    async fn on_channel_list(&mut self, _: ChannelList, address: Addr) {
         let (cmd, rec) = StorageCmd::new_channel_list();
         self.storage.send(cmd).unwrap();
         let channels = rec.await.unwrap().iter().map(|id| id.to_u64()).collect();
-        let request = ServerRequest::new_channel_list(channels);
+        let request = ServerEvent::new_channel_list(channels);
         let command = SessionCmd::new_send(address, request);
         self.sessions.send(command).unwrap();
     }
 
-    async fn on_channel_get_name(&mut self, id: u64, address: Addr) {
+    async fn on_channel_get_name(&mut self, ChannelGetName { id }: ChannelGetName, address: Addr) {
         let (cmd, rec) = StorageCmd::new_channel_get_name(id.into());
         self.storage.send(cmd).unwrap();
         let name = rec.await.unwrap();
-        let request = ServerRequest::new_channel_get_name(id, name);
+        let request = ServerEvent::new_channel_get_name(id, name);
         let command = SessionCmd::new_send(address, request);
         self.sessions.send(command).unwrap();
     }
 
-    fn on_channel_set_name(&mut self, id: u64, name: String) {
+    fn on_channel_set_name(&mut self, ChannelSetName { id, name }: ChannelSetName) {
         let command = StorageCmd::new_channel_set_name(id.into(), name.clone());
         self.storage.send(command).unwrap();
-        let request = ServerRequest::new_channel_set_name(id, name);
+        let request = ServerEvent::new_channel_set_name(id, name);
         let command = SessionCmd::new_broadcast(request);
         self.sessions.send(command).unwrap();
     }
 
-    async fn on_message_list(&mut self, channel_id: u64, address: Addr) {
+    async fn on_message_list(&mut self, MessageList { channel_id }: MessageList, address: Addr) {
         let (cmd, rec) = StorageCmd::new_message_list(channel_id.into());
         self.storage.send(cmd).unwrap();
         let messages = rec.await.unwrap().iter().map(Id::to_u64).collect();
-        let request = ServerRequest::new_message_list(channel_id, messages);
+        let request = ServerEvent::new_message_list(channel_id, messages);
         let command = SessionCmd::new_send(address, request);
         self.sessions.send(command).unwrap();
     }
 
-    async fn on_message_create(&mut self, channel_id: u64, content: String) {
+    async fn on_message_create(
+        &mut self,
+        MessageCreate {
+            channel_id,
+            content,
+        }: MessageCreate,
+    ) {
         let (cmd, rec) = StorageCmd::new_message_create(channel_id.into(), content.clone());
         self.storage.send(cmd).unwrap();
         let id = rec.await.unwrap();
-        let request = ServerRequest::new_message_create(channel_id, id.to_u64(), content);
+        let request = ServerEvent::new_message_create(channel_id, id.to_u64(), content);
         let command = SessionCmd::new_broadcast(request);
         self.sessions.send(command).unwrap();
     }
 
-    fn on_message_delete(&mut self, channel_id: u64, id: u64) {
+    fn on_message_delete(&mut self, MessageDelete { channel_id, id }: MessageDelete) {
         let command = StorageCmd::new_message_delete(channel_id.into(), id.into());
         self.storage.send(command).unwrap();
-        let request = ServerRequest::new_message_delete(channel_id, id);
+        let request = ServerEvent::new_message_delete(channel_id, id);
         let command = SessionCmd::new_broadcast(request);
         self.sessions.send(command).unwrap();
     }
 
-    async fn on_message_get_content(&mut self, channel_id: u64, id: u64, address: Addr) {
+    async fn on_message_get_content(
+        &mut self,
+        MessageGetContent { channel_id, id }: MessageGetContent,
+        address: Addr,
+    ) {
         let (cmd, rec) = StorageCmd::new_message_get_content(channel_id.into(), id.into());
         self.storage.send(cmd).unwrap();
-        let request = ServerRequest::new_message_get_content(channel_id, id, rec.await.unwrap());
+        let request = ServerEvent::new_message_get_content(channel_id, id, rec.await.unwrap());
         let command = SessionCmd::new_send(address, request);
         self.sessions.send(command).unwrap();
     }
 
-    fn on_message_set_content(&mut self, channel_id: u64, id: u64, content: String) {
+    fn on_message_set_content(
+        &mut self,
+        MessageSetContent {
+            channel_id,
+            id,
+            content,
+        }: MessageSetContent,
+    ) {
         let command =
             StorageCmd::new_message_set_content(channel_id.into(), id.into(), content.clone());
         self.storage.send(command).unwrap();
-        let request = ServerRequest::new_message_set_content(channel_id, id, content);
+        let request = ServerEvent::new_message_set_content(channel_id, id, content);
         let command = SessionCmd::new_broadcast(request);
         self.sessions.send(command).unwrap();
     }
+
+    fn on_user_set_pass(&mut self, UserSetPass { id, pass }: UserSetPass, address: Addr) {
+        let command = StorageCmd::new_user_set_pass(id.into(), pass);
+        self.storage.send(command).unwrap();
+        let request = ServerEvent::new_user_set_pass(id);
+        let command = SessionCmd::new_send(address, request);
+        self.sessions.send(command).unwrap();
+    }
+
+    fn on_user_set_name(&mut self, UserSetName { id, name }: UserSetName) {
+        let command = StorageCmd::new_user_set_name(id.into(), name.clone());
+        self.storage.send(command).unwrap();
+        let request = ServerEvent::new_user_set_name(id, name);
+        let command = SessionCmd::new_broadcast(request);
+        self.sessions.send(command).unwrap();
+    }
+
+    async fn on_user_get_name(&mut self, UserGetName { id }: UserGetName, address: Addr) {
+        let (cmd, rec) = StorageCmd::new_user_get_name(id.into());
+        self.storage.send(cmd).unwrap();
+        let name = rec.await.unwrap();
+        let request = ServerEvent::new_user_get_name(id, name);
+        let command = SessionCmd::new_send(address, request);
+        self.sessions.send(command).unwrap();
+    }
+
+    fn on_user_delete(&mut self, UserDelete { id }: UserDelete) {
+        let command = StorageCmd::new_user_delete(id.into());
+        self.storage.send(command).unwrap();
+        let request = ServerEvent::new_user_delete(id);
+        let command = SessionCmd::new_broadcast(request);
+        self.sessions.send(command).unwrap();
+    }
+
+    async fn on_user_create(&mut self, UserCreate { name, pass }: UserCreate) {
+        let (cmd, rec) = StorageCmd::new_user_create(name.clone(), pass);
+        self.storage.send(cmd).unwrap();
+        let id = rec.await.unwrap();
+        let request = ServerEvent::new_user_create(id.into(), name);
+        let command = SessionCmd::new_broadcast(request);
+        self.sessions.send(command).unwrap();
+    }
+
+    async fn on_user_list(&mut self, _: UserList, address: Addr) {
+        let (cmd, rec) = StorageCmd::new_user_list();
+        self.storage.send(cmd).unwrap();
+        let result = rec.await.unwrap().iter().map(Id::to_u64).collect();
+        let request = ServerEvent::new_user_list(result);
+        let command = SessionCmd::new_send(address, request);
+        self.sessions.send(command).unwrap();
+    }
 }
 
 #[telecomande::async_trait]
@@ -210,9 +266,11 @@ impl Processor for GatewayProc {
             GatewayCmd::Request(address, request) => {
                 if let Some(request) = ClientRequest::try_parse(&request) {
                     println!("[session/info] received command '{request:?}'");
-                    self.handle_request(address, request).await;
+                    if let Err(reason) = self.handle_request(address, request).await {
+                        eprintln!("[gateway/warn] exception '{reason}'");
+                    }
                 } else {
-                    println!("[session/warn] failed to parse command");
+                    println!("[session/info] failed to parse command");
                 }
             }
             GatewayCmd::ClosedConnection(address) => self

harsh-server/src/main.rs

@@ -1,23 +1,31 @@
 use telecomande::{Executor, SimpleExecutor};
 use tokio::net::TcpListener;
 
-const ADDRESS: &'static str = "localhost:42069";
-const DB_PATH: &'static str = "./db.test";
+const ADDRESS: &str = "localhost:42000";
+const DB_PATH: &str = "./db.test";
 
 #[tokio::main]
 async fn main() {
     println!("[main/info] starting server ...");
+
     let sessions = SimpleExecutor::new(SessionProc::default()).spawn();
     println!("[main/info] spawned sessions");
+
     let storage = SimpleExecutor::new(StorageProc::new(DB_PATH)).spawn();
     println!("[main/info] spawned storage");
-    let gateway =
-        SimpleExecutor::new(GatewayProc::new(sessions.remote(), storage.remote())).spawn();
+
+    let security = SimpleExecutor::new(SecurityProc::new(storage.remote())).spawn();
+
+    let gateway = SimpleExecutor::new(GatewayProc::new(
+        sessions.remote(),
+        storage.remote(),
+        security.remote(),
+    ))
+    .spawn();
     println!("[main/info] spawned gateway");
 
     let listener = TcpListener::bind(ADDRESS).await.unwrap();
     println!("[main/info] listening on '{ADDRESS}' ...");
-
     let client_handler = sessions.remote();
     loop {
         let (stream, address) = listener.accept().await.unwrap();
@@ -42,3 +50,6 @@ pub use sessions::{SessionCmd, SessionProc};
 
 mod storage;
 pub use storage::{StorageCmd, StorageProc};
+
+mod security;
+pub use security::{SecurityCmd, SecurityProc};

harsh-server/src/security.rs

@@ -0,0 +1,99 @@
+use telecomande::{Processor, Remote};
+use tokio::sync::oneshot::{self, Receiver, Sender};
+
+use crate::{storage::Perm, Id, StorageCmd, StorageProc};
+
+#[derive(Debug)]
+pub enum SecurityCmd {
+    Verify(Id, Perm, Sender<bool>),
+    Authenticate(Id, String, Sender<bool>),
+    StorePass(Id, String),
+}
+
+impl SecurityCmd {
+    pub fn new_verify(user_id: Id, permission: Perm) -> (Self, Receiver<bool>) {
+        let (sender, receiver) = oneshot::channel();
+        let command = Self::Verify(user_id, permission, sender);
+        (command, receiver)
+    }
+
+    pub fn new_authenticate(user_id: Id, pass: String) -> (Self, Receiver<bool>) {
+        let (sender, receiver) = oneshot::channel();
+        let command = Self::Authenticate(user_id, pass, sender);
+        (command, receiver)
+    }
+
+    pub fn new_store_pass(user_id: Id, pass: String) -> Self {
+        Self::StorePass(user_id, pass)
+    }
+}
+
+pub struct SecurityProc {
+    storage: Remote<StorageProc>,
+}
+
+impl SecurityProc {
+    pub fn new(storage: Remote<StorageProc>) -> Self {
+        Self { storage }
+    }
+
+    async fn handle_command(&mut self, command: SecurityCmd) {
+        match command {
+            SecurityCmd::Verify(user, perm, sender) => {
+                let (cmd, req) = StorageCmd::new_perm_server_get_op();
+                self.storage.send(cmd).unwrap();
+                let serv_ops = req.await.unwrap();
+                let is_serv_op = serv_ops.into_iter().any(|i| i == user);
+                let result = match (is_serv_op, perm) {
+                    (true, _) => true,
+                    (false, Perm::OpChannel(chan_id)) => {
+                        let (cmd, req) = StorageCmd::new_perm_channel_get_op(chan_id);
+                        self.storage.send(cmd).unwrap();
+                        let channel_ops = req.await.unwrap();
+                        channel_ops.into_iter().any(|i| i == user)
+                    }
+                    _ => false,
+                };
+                sender.send(result).unwrap();
+            }
+            SecurityCmd::Authenticate(user, pass, sender) => {
+                let (cmd, rec) = StorageCmd::new_user_get_pass(user);
+                self.storage.send(cmd).unwrap();
+                let stored = rec.await.unwrap();
+                let result = stored.map(|stored| stored == hash(pass)).unwrap_or(false);
+                sender.send(result).unwrap();
+            }
+            SecurityCmd::StorePass(user, pass) => {
+                let pass = hash(pass);
+                let command = StorageCmd::new_user_set_pass(user, pass);
+                self.storage.send(command).unwrap();
+            }
+        }
+    }
+}
+
+const SALT: &str = ":)";
+
+fn hash(input: String) -> String {
+    let hash = blake3::hash((input + SALT).as_bytes());
+    format!("{hash}")
+}
+
+#[test]
+fn test_hash() {
+    assert_eq!(
+        &hash("arbre".into()),
+        "c2d3a87dcb76c21a8a935b8e988745f31663c3650a0d3732430eaa323f12ee0f"
+    );
+}
+
+#[telecomande::async_trait]
+impl Processor for SecurityProc {
+    type Command = SecurityCmd;
+    type Error = ();
+
+    async fn handle(&mut self, command: Self::Command) -> Result<(), Self::Error> {
+        self.handle_command(command).await;
+        Ok(())
+    }
+}

harsh-server/src/sessions.rs

@@ -1,6 +1,6 @@
 use std::{collections::HashMap, net::SocketAddr};
 
-use harsh_common::ServerRequest;
+use harsh_common::ServerEvent;
 use telecomande::{Processor, Remote};
 use tokio::{
     io::{AsyncBufReadExt, AsyncWriteExt, BufReader},
@@ -8,33 +8,58 @@ use tokio::{
         tcp::{OwnedReadHalf, OwnedWriteHalf},
         TcpStream,
     },
+    sync::oneshot::{self, Receiver, Sender},
     task::JoinHandle,
 };
 
-use crate::{gateway, Addr};
+use crate::{gateway, Addr, Id};
 #[derive(Debug)]
 pub enum SessionCmd {
     AddSession(TcpStream, SocketAddr, Remote<gateway::GatewayProc>),
     RemoveSession(Addr),
     Send(Addr, String),
     Broadcast(String),
+    GetUser(Addr, Sender<Option<Id>>),
+    SetUser(Addr, Option<Id>),
 }
 
 impl SessionCmd {
-    pub fn new_send(address: Addr, request: ServerRequest) -> Self {
+    pub fn new_add_session(
+        stream: TcpStream,
+        address: SocketAddr,
+        gateway: Remote<gateway::GatewayProc>,
+    ) -> Self {
+        Self::AddSession(stream, address, gateway)
+    }
+
+    pub fn new_remove_session(address: Addr) -> Self {
+        Self::RemoveSession(address)
+    }
+
+    pub fn new_send(address: Addr, request: ServerEvent) -> Self {
         let content = request.serialize();
         Self::Send(address, content)
     }
 
-    pub fn new_broadcast(request: ServerRequest) -> Self {
+    pub fn new_broadcast(request: ServerEvent) -> Self {
         let content = request.serialize();
         Self::Broadcast(content)
     }
+
+    pub fn new_get_user(address: Addr) -> (Self, Receiver<Option<Id>>) {
+        let (sender, receiver) = oneshot::channel();
+        let command = Self::GetUser(address, sender);
+        (command, receiver)
+    }
+
+    pub fn new_set_user(address: Addr, user: Option<Id>) -> Self {
+        Self::SetUser(address, user)
+    }
 }
 
 #[derive(Debug, Default)]
 pub struct SessionProc {
-    clients: HashMap<Addr, (OwnedWriteHalf, JoinHandle<()>)>,
+    clients: HashMap<Addr, Client>,
 }
 
 impl SessionProc {
@@ -46,7 +71,7 @@ impl SessionProc {
     ) {
         let (reader, writer) = stream.into_split();
         let handle = tokio::spawn(session(address.clone(), reader, remote));
-        self.clients.insert(address, (writer, handle));
+        self.clients.insert(address, Client::new(writer, handle));
     }
 }
 
@@ -65,24 +90,29 @@ impl Processor for SessionProc {
             }
             SessionCmd::RemoveSession(address) => {
                 println!("[sessions/info] closed connection from '{address:?}'");
-                if let Some((_writer, handle)) = self.clients.remove(&address) {
-                    handle.await.unwrap();
+                if let Some(client) = self.clients.remove(&address) {
+                    client.unwrap().await.unwrap();
                 }
             }
             SessionCmd::Send(address, content) => {
-                if let Some((client, _)) = self.clients.get_mut(&address) {
+                if let Some(client) = self.clients.get_mut(&address) {
                     println!("[session/info] sending '{content}' to '{address:?}'");
-                    client.write_all(content.as_bytes()).await.unwrap();
-                    client.write_all(b"\n").await.unwrap();
-                } else {
-                    eprintln!("failed to find session with address '{address:?}'")
+                    client.send(&content).await;
                 }
             }
             SessionCmd::Broadcast(content) => {
-                for (client, _) in self.clients.values_mut() {
-                    println!("[session/info] broadcasting '{content}'");
-                    client.write_all(content.as_bytes()).await.unwrap();
-                    client.write_all(b"\n").await.unwrap();
+                println!("[session/info] broadcasting '{content}'");
+                for client in self.clients.values_mut() {
+                    client.send(&content).await;
+                }
+            }
+            SessionCmd::GetUser(address, sender) => {
+                let user = self.clients.get_mut(&address).and_then(|c| c.get_user());
+                sender.send(user).unwrap();
+            }
+            SessionCmd::SetUser(address, user) => {
+                if let Some(client) = self.clients.get_mut(&address) {
+                    client.set_user(user);
                 }
             }
         };
@@ -90,6 +120,45 @@ impl Processor for SessionProc {
     }
 }
 
+#[derive(Debug)]
+pub struct Client {
+    writer: OwnedWriteHalf,
+    handle: JoinHandle<()>,
+    user: Option<Id>,
+}
+
+impl Client {
+    pub fn new(writer: OwnedWriteHalf, handle: JoinHandle<()>) -> Self {
+        let user = None;
+        Self {
+            handle,
+            user,
+            writer,
+        }
+    }
+
+    pub fn unwrap(self) -> JoinHandle<()> {
+        let Self {
+            writer,
+            handle,
+            user,
+        } = self;
+        drop((writer, user));
+        handle
+    }
+
+    pub async fn send(&mut self, message: &str) {
+        self.writer.write_all(message.as_bytes()).await.unwrap();
+        self.writer.write_all(b"\n").await.unwrap();
+    }
+    pub fn set_user(&mut self, id: Option<Id>) {
+        self.user = id;
+    }
+    pub fn get_user(&self) -> Option<Id> {
+        self.user
+    }
+}
+
 async fn session(address: Addr, reader: OwnedReadHalf, remote: Remote<gateway::GatewayProc>) {
     let mut reader = BufReader::new(reader);
     loop {
@@ -107,3 +176,24 @@ async fn session(address: Addr, reader: OwnedReadHalf, remote: Remote<gateway::G
         .send(gateway::GatewayCmd::ClosedConnection(address))
         .unwrap();
 }
+
+#[telecomande::async_trait]
+pub trait SessionExt {
+    fn send(&self, cmd: SessionCmd);
+
+    async fn is_logged(&self, address: Addr) -> bool {
+        self.get_user(address).await.is_some()
+    }
+
+    async fn get_user(&self, address: Addr) -> Option<Id> {
+        let (cmd, rec) = SessionCmd::new_get_user(address);
+        self.send(cmd);
+        rec.await.unwrap()
+    }
+}
+
+impl SessionExt for Remote<SessionProc> {
+    fn send(&self, cmd: SessionCmd) {
+        self.send(cmd).unwrap();
+    }
+}

harsh-server/src/storage.rs

@@ -23,6 +23,12 @@ pub enum StorageCmd {
     UserSetName(Id, String),
     UserGetPass(Id, Sender<Option<String>>),
     UserSetPass(Id, String),
+    PermServerAddOp(Id),
+    PermServerRemoveOp(Id),
+    PermServerGetOp(Sender<Vec<Id>>),
+    PermChannelAddOp(Id, Id),
+    PermChannelRemoveOp(Id, Id),
+    PermChannelGetOp(Id, Sender<Vec<Id>>),
 }
 
 impl StorageCmd {
@@ -110,6 +116,34 @@ impl StorageCmd {
     pub fn new_user_set_pass(id: Id, pass: String) -> Self {
         Self::UserSetPass(id, pass)
     }
+
+    pub fn new_perm_server_add_op(user_id: Id) -> Self {
+        Self::PermServerAddOp(user_id)
+    }
+
+    pub fn new_perm_server_remove_op(user_id: Id) -> Self {
+        Self::PermServerRemoveOp(user_id)
+    }
+
+    pub fn new_perm_server_get_op() -> (Self, Receiver<Vec<Id>>) {
+        let (sender, receiver) = oneshot::channel();
+        let cmd = Self::PermServerGetOp(sender);
+        (cmd, receiver)
+    }
+
+    pub fn new_perm_channel_add_op(channel_id: Id, user_id: Id) -> Self {
+        Self::PermChannelAddOp(channel_id, user_id)
+    }
+
+    pub fn new_perm_channel_remove_op(channel_id: Id, user_id: Id) -> Self {
+        Self::PermChannelRemoveOp(channel_id, user_id)
+    }
+
+    pub fn new_perm_channel_get_op(channel_id: Id) -> (Self, Receiver<Vec<Id>>) {
+        let (sender, receiver) = oneshot::channel();
+        let command = Self::PermChannelGetOp(channel_id, sender);
+        (command, receiver)
+    }
 }
 
 pub struct StorageProc {
@@ -194,48 +228,32 @@ impl StorageProc {
             //
             // User
             //
-            UserList(sender) => {
-                let users = self.list("/users/");
-                sender.send(users).unwrap();
-            }
+            UserList(sender) => self.on_user_list(sender),
+            UserCreate(name, pass, sender) => self.on_user_create(name, pass, sender),
+            UserDelete(id) => self.on_user_delete(id),
+            UserGetName(id, sender) => self.on_user_get_name(id, sender),
+            UserSetName(id, name) => self.on_user_set_name(id, name),
+            UserGetPass(id, sender) => self.on_user_get_pass(id, sender),
+            UserSetPass(id, pass) => self.on_user_set_pass(id, pass),
 
-            UserCreate(name, pass, sender) => {
-                let user = User::new(name, pass);
-                let id = user.get_id();
-                self.set(format!("/users/{id}"), user);
-                sender.send(id).unwrap();
+            //
+            // Perms
+            //
+            PermServerGetOp(sender) => {
+                let result = self.list("/op/serv/".to_string());
+                sender.send(result).unwrap();
             }
-
-            UserDelete(id) => {
-                self.remove(format!("/users/{id}"));
+            PermServerAddOp(user_id) => self.set(format!("/op/serv/{user_id}"), true),
+            PermServerRemoveOp(user_id) => self.remove(format!("/op/serv/{user_id}")),
+            PermChannelAddOp(channel_id, user_id) => {
+                self.set(format!("/op/channels/{channel_id}/{user_id}"), true)
             }
-
-            UserGetName(id, sender) => {
-                let user = self.get::<_, User>(format!("/users/{id}"));
-                let name = user.map(|u| u.get_name().to_string());
-                sender.send(name).unwrap();
+            PermChannelRemoveOp(channel_id, user_id) => {
+                self.remove(format!("/op/channels/{channel_id}/{user_id}"))
             }
-
-            UserSetName(id, name) => {
-                let path = format!("/users/{id}");
-                if let Some(mut user) = self.get::<_, User>(&path) {
-                    user.set_name(name);
-                    self.set(path, user);
-                }
-            }
-
-            UserGetPass(id, sender) => {
-                let user = self.get::<_, User>(format!("/users/{id}"));
-                let name = user.map(|u| u.get_pass().to_string());
-                sender.send(name).unwrap();
-            }
-
-            UserSetPass(id, pass) => {
-                let path = format!("/users/{id}");
-                if let Some(mut user) = self.get::<_, User>(&path) {
-                    user.set_pass(pass);
-                    self.set(path, user);
-                }
+            PermChannelGetOp(channel_id, sender) => {
+                let result = self.list(format!("/op/channels/{channel_id}/"));
+                sender.send(result).unwrap();
             }
         };
     }
@@ -308,6 +326,54 @@ impl StorageProc {
             self.set(path, message);
         }
     }
+
+    //
+    // User
+    //
+
+    fn on_user_list(&mut self, sender: Sender<Vec<Id>>) {
+        let users = self.list("/users/");
+        sender.send(users).unwrap();
+    }
+
+    fn on_user_create(&mut self, name: String, pass: String, sender: Sender<Id>) {
+        let user = User::new(name, pass);
+        let id = user.get_id();
+        self.set(format!("/users/{id}"), user);
+        sender.send(id).unwrap();
+    }
+
+    fn on_user_delete(&mut self, id: Id) {
+        self.remove(format!("/users/{id}"));
+    }
+
+    fn on_user_get_name(&mut self, id: Id, sender: Sender<Option<String>>) {
+        let user = self.get::<_, User>(format!("/users/{id}"));
+        let name = user.map(|u| u.get_name().to_string());
+        sender.send(name).unwrap();
+    }
+
+    fn on_user_set_name(&mut self, id: Id, name: String) {
+        let path = format!("/users/{id}");
+        if let Some(mut user) = self.get::<_, User>(&path) {
+            user.set_name(name);
+            self.set(path, user);
+        }
+    }
+
+    fn on_user_get_pass(&mut self, id: Id, sender: Sender<Option<String>>) {
+        let user = self.get::<_, User>(format!("/users/{id}"));
+        let name = user.map(|u| u.get_pass().to_string());
+        sender.send(name).unwrap();
+    }
+
+    fn on_user_set_pass(&mut self, id: Id, pass: String) {
+        let path = format!("/users/{id}");
+        if let Some(mut user) = self.get::<_, User>(&path) {
+            user.set_pass(pass);
+            self.set(path, user);
+        }
+    }
 }
 
 #[telecomande::async_trait]
@@ -323,7 +389,7 @@ impl Processor for StorageProc {
 }
 
 mod models;
-pub use models::{Channel, Message, SerDeser, User};
+pub use models::{Channel, Message, Perm, SerDeser, User};
 
 fn list(db: &Db, path: String) -> Vec<Id> {
     let len = path.len();

harsh-server/src/storage/models.rs

@@ -114,3 +114,9 @@ where
         db.insert(path, bytes).unwrap();
     }
 }
+
+#[derive(Debug, Serialize, Deserialize, PartialEq, Eq, Clone)]
+pub enum Perm {
+    OpServer,
+    OpChannel(Id),
+}