2 files changed, 138 insertions, 0 deletions
diff --git a/sip/src/transaction/auth.rs b/sip/src/transaction/auth.rs
new file mode 100644
index 0000000..4c46f9b
--- /dev/null
+++ b/sip/src/transaction/auth.rs
@@ -0,0 +1,51 @@
+use crate::encoding::{
+    headers::{Authorization, WWWAuthenticate},
+    method::Method,
+    request::Request,
+    response::Response,
+    uri::Uri,
+};
+use anyhow::Result;
+
+impl Authorization {
+    pub fn construct(
+        request: &Request,
+        failed_response: &Response,
+        username: &str,
+        password: &str,
+    ) -> Result<Authorization> {
+        let challenge = failed_response.headers.get_res::<WWWAuthenticate>()?;
+
+        Ok(Authorization {
+            response: response_digest(
+                username.to_string(),
+                challenge.realm.clone(),
+                password.to_string(),
+                request.method,
+                challenge.nonce.clone(),
+                request.uri.clone(),
+            ),
+            nonce: challenge.nonce,
+            realm: challenge.realm,
+            uri: request.uri.to_string(),
+            username: username.to_string(),
+        })
+    }
+}
+
+fn response_digest(
+    username: String,
+    realm: String,
+    password: String,
+    method: Method,
+    nonce: String,
+    uri: Uri,
+) -> String {
+    let h = |s: String| hex::encode(md5::compute(s.as_bytes()).0);
+    let kd = |secret, data| h(format!("{secret}:{data}"));
+
+    let a1 = format!("{username}:{realm}:{password}");
+    let a2 = format!("{method}:{uri}");
+    let response_digest = kd(h(a1), format!("{nonce}:{}", h(a2)));
+    return response_digest;
+}
diff --git a/sip/src/transaction/mod.rs b/sip/src/transaction/mod.rs
new file mode 100644
index 0000000..3368c47
--- /dev/null
+++ b/sip/src/transaction/mod.rs
@@ -0,0 +1,87 @@
+pub mod auth;
+
+use crate::{
+    encoding::{
+        headers::{CSeq, CallID, ContentLength},
+        request::Request,
+        response::Response,
+        Message,
+    },
+    transport::Transport,
+};
+use anyhow::{anyhow, Result};
+use std::{
+    collections::HashMap,
+    sync::atomic::{AtomicU32, Ordering},
+};
+use tokio::sync::{
+    mpsc::{self, channel},
+    RwLock,
+};
+
+pub struct TransactionUser<T> {
+    transport: T,
+    sequence: AtomicU32,
+    pending_requests: RwLock<HashMap<CSeq, mpsc::Sender<Response>>>,
+}
+
+impl<T: Transport> TransactionUser<T> {
+    pub fn new(transport: T) -> Self {
+        Self {
+            sequence: 0.into(),
+            pending_requests: Default::default(),
+            transport,
+        }
+    }
+
+    pub async fn process_incoming(&self) -> Result<Request> {
+        loop {
+            let mesg = self.transport.recv().await?;
+            match mesg {
+                Message::Request(req) => break Ok(req),
+                Message::Response(resp) => {
+                    let cseq = resp
+                        .headers
+                        .get()
+                        .ok_or(anyhow!("response cseq missing"))??;
+                    self.pending_requests
+                        .write()
+                        .await
+                        .get_mut(&cseq)
+                        .ok_or(anyhow!("message was not requested"))?
+                        .send(resp)
+                        .await?;
+                }
+            }
+        }
+    }
+    pub async fn respond(&self, req: &Request, mut resp: Response) -> Result<()> {
+        resp.headers.insert(
+            req.headers
+                .get::<CSeq>()
+                .ok_or(anyhow!("cseq is mandatory"))??,
+        );
+        resp.headers.insert(
+            req.headers
+                .get::<CallID>()
+                .ok_or(anyhow!("call-id is mandatory"))??,
+        );
+        resp.headers.insert(ContentLength(resp.body.len()));
+        self.transport.send(Message::Response(resp)).await?;
+        Ok(())
+    }
+
+    pub async fn transact(&self, mut request: Request) -> Result<mpsc::Receiver<Response>> {
+        let seq = self.sequence.fetch_add(1, Ordering::Relaxed);
+        let cseq = CSeq(seq, request.method);
+        request.headers.insert(cseq);
+        request.headers.insert(ContentLength(request.body.len()));
+
+        let (tx, rx) = channel(4);
+
+        self.transport.send(Message::Request(request)).await?;
+        self.pending_requests.write().await.insert(cseq, tx);
+
+        Ok(rx)
+    }
+}