use std::net::SocketAddr;
use axum::extract::{ConnectInfo, FromRequestParts};
use axum::http::request::Parts;
use mail4agent_api::{
Address, Declared, MailError, ParticipantId, SessionAttested, SessionCard, SessionCorroborated, SessionId,
};
use mail4agent_attest::{AttestError, PeerProcess};
use sha2::{Digest, Sha256};
use crate::service::MailboxService;
pub struct PeerAddr(pub Option<SocketAddr>);
impl<S> FromRequestParts<S> for PeerAddr
where
S: Send + Sync,
{
type Rejection = std::convert::Infallible;
async fn from_request_parts(parts: &mut Parts, state: &S) -> Result<Self, Self::Rejection> {
match ConnectInfo::<SocketAddr>::from_request_parts(parts, state).await {
Ok(ConnectInfo(addr)) => Ok(Self(Some(addr))),
Err(_) => Ok(Self(None)),
}
}
}
const DERIVED_SESSION_ID_HEX_LEN: usize = 32;
#[derive(Debug)]
pub enum SessionError {
MissingConnectInfo,
Attest(AttestError),
Mailbox(MailError),
}
impl std::fmt::Display for SessionError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::MissingConnectInfo => write!(
f,
"ConnectInfo<SocketAddr> is absent from this request -- the server is not wired to supply it"
),
Self::Attest(err) => write!(f, "could not attest the calling process: {err}"),
Self::Mailbox(err) => write!(f, "session registration refused: {err}"),
}
}
}
impl std::error::Error for SessionError {}
impl From<AttestError> for SessionError {
fn from(err: AttestError) -> Self {
Self::Attest(err)
}
}
pub fn derive_session_id(pid: u32, started_at_unix_ms: u64) -> SessionId {
let mut hasher = Sha256::new();
hasher.update(pid.to_le_bytes());
hasher.update(started_at_unix_ms.to_le_bytes());
let digest = hasher.finalize();
let hex_digest = hex::encode(digest);
let body = &hex_digest[..DERIVED_SESSION_ID_HEX_LEN];
SessionId::new(format!("{}{body}", SessionId::PREFIX))
.expect("a fixed-length lower-hex body within SessionId's own bounds always validates")
}
#[derive(Clone, Debug, Default, Eq, PartialEq)]
pub struct CorroboratedCommandLine {
pub provider_session_id: Option<String>,
pub model: Option<String>,
}
pub fn parse_claude_command_line(command_line: &str) -> CorroboratedCommandLine {
let tokens: Vec<&str> = command_line.split_whitespace().collect();
let mut result = CorroboratedCommandLine::default();
let mut index = 0;
while index < tokens.len() {
let token = tokens[index];
if let Some(value) = token.strip_prefix("--resume=") {
if !value.is_empty() {
result.provider_session_id = Some(value.to_string());
}
} else if token == "--model" {
if let Some(value) = tokens.get(index + 1) {
result.model = Some((*value).to_string());
index += 1;
}
}
index += 1;
}
result
}
pub fn session_card_from_peer_process(peer: &PeerProcess) -> SessionCard {
let parsed = peer
.command_line
.as_ref()
.map(|declared| parse_claude_command_line(declared.inner_ref()))
.unwrap_or_default();
SessionCard {
attested: SessionAttested { pid: peer.pid, started_at_unix_ms: peer.started_at_unix_ms, exe: peer.exe.clone() },
corroborated: SessionCorroborated {
provider_session_id: parsed.provider_session_id.map(Declared::new),
model: parsed.model.map(Declared::new),
cwd: peer.cwd.as_ref().map(|declared| Declared::new(declared.inner_ref().clone())),
},
declared: Default::default(),
}
}
pub async fn ensure_session_from_peer_process(
service: &MailboxService,
account: ParticipantId,
peer_process: &PeerProcess,
now_unix_ms: u64,
) -> Result<Address, SessionError> {
let session_id = derive_session_id(peer_process.pid, peer_process.started_at_unix_ms);
let card = session_card_from_peer_process(peer_process);
service
.ensure_session(account.clone(), session_id.clone(), card, now_unix_ms)
.await
.map_err(SessionError::Mailbox)?;
Ok(Address::Session { participant: account, session: session_id })
}
pub async fn resolve_session(
service: &MailboxService,
account: ParticipantId,
peer: SocketAddr,
local: SocketAddr,
now_unix_ms: u64,
) -> Result<Address, SessionError> {
let peer_process = mail4agent_attest::attest(peer, local)?;
ensure_session_from_peer_process(service, account, &peer_process, now_unix_ms).await
}
#[cfg(test)]
mod tests {
use super::*;
use mail4agent_core::ParticipantPermissions;
use mail4agent_store_sqlite::SqliteMailStore;
fn peer_process(pid: u32, started_at_unix_ms: u64) -> PeerProcess {
PeerProcess { pid, started_at_unix_ms, exe: Some("C:\\claude.exe".to_string()), command_line: None, cwd: None }
}
async fn test_service() -> MailboxService {
let engine_store = SqliteMailStore::open_in_memory().expect("in-memory store opens and migrates");
let reader_store = SqliteMailStore::new(engine_store.db());
MailboxService::new(engine_store, reader_store)
}
async fn register(service: &MailboxService, id: &ParticipantId) {
let permissions = ParticipantPermissions { may_send: true, may_read: true, operator: false };
service.register_participant(id.clone(), None, permissions).await.expect("register test participant");
}
#[test]
fn derive_session_id_is_stable_for_the_same_pid_and_start_time() {
assert_eq!(derive_session_id(4242, 1_000), derive_session_id(4242, 1_000));
}
#[test]
fn derive_session_id_differs_when_start_time_differs_for_the_same_pid() {
assert_ne!(derive_session_id(4242, 1_000), derive_session_id(4242, 2_000));
}
#[test]
fn derive_session_id_differs_across_pids() {
assert_ne!(derive_session_id(1, 1_000), derive_session_id(2, 1_000));
}
#[test]
fn parse_claude_command_line_recognises_resume_and_model() {
let parsed = parse_claude_command_line("claude --resume=1234-uuid --model sonnet");
assert_eq!(parsed.provider_session_id.as_deref(), Some("1234-uuid"));
assert_eq!(parsed.model.as_deref(), Some("sonnet"));
}
#[test]
fn parse_claude_command_line_tolerates_either_order() {
let parsed = parse_claude_command_line("claude --model opus --resume=abc");
assert_eq!(parsed.provider_session_id.as_deref(), Some("abc"));
assert_eq!(parsed.model.as_deref(), Some("opus"));
}
#[test]
fn parse_claude_command_line_yields_nothing_for_an_unrecognised_line() {
let parsed = parse_claude_command_line("some-other-cli --flag value");
assert_eq!(parsed, CorroboratedCommandLine::default());
}
#[test]
fn parse_claude_command_line_ignores_a_dangling_model_flag_with_no_value() {
let parsed = parse_claude_command_line("claude --model");
assert_eq!(parsed.model, None);
}
#[test]
fn parse_claude_command_line_ignores_an_empty_resume_value() {
let parsed = parse_claude_command_line("claude --resume=");
assert_eq!(parsed.provider_session_id, None);
}
#[tokio::test]
async fn ensure_session_from_peer_process_returns_a_session_address_for_the_account() {
let service = test_service().await;
let account = ParticipantId::new("alice").expect("valid participant id");
register(&service, &account).await;
let address = ensure_session_from_peer_process(&service, account.clone(), &peer_process(111, 222), 1_000)
.await
.expect("resolves a session address");
match address {
Address::Session { participant, session } => {
assert_eq!(participant, account);
assert_eq!(session, derive_session_id(111, 222));
}
other => panic!("expected a session address, got {other:?}"),
}
}
#[tokio::test]
async fn ensure_session_from_peer_process_is_idempotent_for_the_same_process() {
let service = test_service().await;
let account = ParticipantId::new("alice").expect("valid participant id");
register(&service, &account).await;
let first = ensure_session_from_peer_process(&service, account.clone(), &peer_process(111, 222), 1_000)
.await
.expect("first call resolves");
let second = ensure_session_from_peer_process(&service, account.clone(), &peer_process(111, 222), 2_000)
.await
.expect("second call resolves the same session");
assert_eq!(first, second);
}
#[tokio::test]
async fn ensure_session_from_peer_process_gives_different_processes_different_sessions() {
let service = test_service().await;
let account = ParticipantId::new("alice").expect("valid participant id");
register(&service, &account).await;
let first = ensure_session_from_peer_process(&service, account.clone(), &peer_process(111, 222), 1_000)
.await
.expect("first process resolves");
let second = ensure_session_from_peer_process(&service, account.clone(), &peer_process(333, 444), 1_000)
.await
.expect("second, distinct process resolves");
assert_ne!(first, second);
}
#[tokio::test]
async fn ensure_session_from_peer_process_refuses_a_session_already_owned_by_another_account() {
let service = test_service().await;
let alice = ParticipantId::new("alice").expect("valid participant id");
let bob = ParticipantId::new("bob").expect("valid participant id");
register(&service, &alice).await;
register(&service, &bob).await;
ensure_session_from_peer_process(&service, alice, &peer_process(111, 222), 1_000)
.await
.expect("alice's session registers");
let err = ensure_session_from_peer_process(&service, bob, &peer_process(111, 222), 2_000)
.await
.expect_err("the same (pid, started_at) presented under a different account must be refused");
assert!(matches!(err, SessionError::Mailbox(MailError::SessionAccountMismatch { .. })));
}
}