clusterflux-public/crates/clusterflux-coordinator/src/service/tcp.rs

use std::io::{BufRead, BufReader, Write};
use std::net::{SocketAddr, TcpListener, TcpStream};
use std::sync::{Arc, Mutex};

use super::{CoordinatorRequest, CoordinatorResponse, CoordinatorService, CoordinatorServiceError};

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum ClientAuthorityMode {
    Strict,
    LocalTrustedLoopback,
}

impl CoordinatorService {
    pub fn serve_tcp(self, listener: TcpListener) -> Result<(), CoordinatorServiceError> {
        if !listener.local_addr()?.ip().is_loopback() {
            return Err(CoordinatorServiceError::Protocol(
                "the native coordinator transport is plaintext and restricted to loopback; expose a remote coordinator only through a secure transport"
                    .to_owned(),
            ));
        }
        self.serve_tcp_with_authority(listener, ClientAuthorityMode::Strict)
    }

    pub fn serve_tcp_local_trusted(
        self,
        listener: TcpListener,
    ) -> Result<(), CoordinatorServiceError> {
        if !listener.local_addr()?.ip().is_loopback() {
            return Err(CoordinatorServiceError::Protocol(
                "local trusted request mode is restricted to a loopback listener".to_owned(),
            ));
        }
        self.serve_tcp_with_authority(listener, ClientAuthorityMode::LocalTrustedLoopback)
    }

    fn serve_tcp_with_authority(
        self,
        listener: TcpListener,
        authority_mode: ClientAuthorityMode,
    ) -> Result<(), CoordinatorServiceError> {
        let shared = Arc::new(Mutex::new(self));
        for stream in listener.incoming() {
            let stream = stream?;
            let service = Arc::clone(&shared);
            std::thread::spawn(move || {
                if let Err(err) = handle_shared_stream(service, stream, authority_mode) {
                    eprintln!("coordinator stream failed: {err}");
                }
            });
        }
        Ok(())
    }

    pub fn handle_stream(&mut self, stream: TcpStream) -> Result<(), CoordinatorServiceError> {
        self.handle_stream_with_authority(stream, ClientAuthorityMode::Strict)
    }

    #[cfg(test)]
    pub(super) fn handle_stream_local_trusted(
        &mut self,
        stream: TcpStream,
    ) -> Result<(), CoordinatorServiceError> {
        self.handle_stream_with_authority(stream, ClientAuthorityMode::LocalTrustedLoopback)
    }

    fn handle_stream_with_authority(
        &mut self,
        stream: TcpStream,
        authority_mode: ClientAuthorityMode,
    ) -> Result<(), CoordinatorServiceError> {
        let mut reader = BufReader::new(stream.try_clone()?);
        let mut writer = stream;
        loop {
            let mut line = String::new();
            if reader.read_line(&mut line)? == 0 {
                return Ok(());
            }
            if line.trim().is_empty() {
                continue;
            }
            let response = match decode_wire_request(&line) {
                Ok(request) => match authorize_client_request(&request, authority_mode)
                    .and_then(|()| self.handle_request(request))
                {
                    Ok(response) => response,
                    Err(err) => CoordinatorResponse::Error {
                        message: err.to_string(),
                    },
                },
                Err(err) => CoordinatorResponse::Error {
                    message: err.to_string(),
                },
            };
            serde_json::to_writer(&mut writer, &response)?;
            writer.write_all(b"\n")?;
            writer.flush()?;
        }
    }
}

fn handle_shared_stream(
    service: Arc<Mutex<CoordinatorService>>,
    stream: TcpStream,
    authority_mode: ClientAuthorityMode,
) -> Result<(), CoordinatorServiceError> {
    let mut reader = BufReader::new(stream.try_clone()?);
    let mut writer = stream;
    loop {
        let mut line = String::new();
        if reader.read_line(&mut line)? == 0 {
            return Ok(());
        }
        if line.trim().is_empty() {
            continue;
        }
        let response = match decode_wire_request(&line) {
            Ok(request) => match authorize_client_request(&request, authority_mode) {
                Ok(()) => match service.lock() {
                    Ok(mut service) => match service.handle_request(request) {
                        Ok(response) => response,
                        Err(err) => CoordinatorResponse::Error {
                            message: err.to_string(),
                        },
                    },
                    Err(_) => CoordinatorResponse::Error {
                        message: "coordinator service lock poisoned".to_owned(),
                    },
                },
                Err(err) => CoordinatorResponse::Error {
                    message: err.to_string(),
                },
            },
            Err(err) => CoordinatorResponse::Error {
                message: err.to_string(),
            },
        };
        serde_json::to_writer(&mut writer, &response)?;
        writer.write_all(b"\n")?;
        writer.flush()?;
    }
}

pub fn bind_listener(addr: &str) -> Result<(TcpListener, SocketAddr), CoordinatorServiceError> {
    let listener = TcpListener::bind(addr)?;
    let addr = listener.local_addr()?;
    Ok((listener, addr))
}

fn decode_wire_request(line: &str) -> Result<CoordinatorRequest, CoordinatorServiceError> {
    serde_json::from_str::<super::CoordinatorWireRequest>(line)?
        .into_request()
        .map_err(CoordinatorServiceError::Protocol)
}

fn authorize_client_request(
    request: &CoordinatorRequest,
    authority_mode: ClientAuthorityMode,
) -> Result<(), CoordinatorServiceError> {
    if authority_mode == ClientAuthorityMode::LocalTrustedLoopback {
        return Ok(());
    }
    match request {
        CoordinatorRequest::Ping
        | CoordinatorRequest::Authenticated { .. }
        | CoordinatorRequest::ExchangeNodeEnrollmentGrant { .. }
        | CoordinatorRequest::SignedNode { .. }
        | CoordinatorRequest::NodeHeartbeat {
            node_signature: Some(_),
            ..
        }
        | CoordinatorRequest::StartProcess {
            actor_agent: Some(_),
            agent_signature: Some(_),
            ..
        }
        | CoordinatorRequest::LaunchTask {
            actor_agent: Some(_),
            agent_signature: Some(_),
            ..
        }
        | CoordinatorRequest::AdminStatus { .. }
        | CoordinatorRequest::SuspendTenant { .. } => Ok(()),
        _ => Err(CoordinatorServiceError::Protocol(
            "strict Core Client authority requires an authenticated CLI session, signed Agent, signed Node, enrollment grant exchange, or admin credential; request-body identity fields are not authority"
                .to_owned(),
        )),
    }
}

#[cfg(test)]
mod transport_boundary_tests {
    use std::io::{BufRead as _, BufReader};

    use clusterflux_core::{coordinator_wire_request, ProjectId, TenantId, UserId};
    use serde_json::json;

    use super::*;

    #[test]
    fn native_plaintext_service_refuses_non_loopback_listener() {
        let (listener, _) = bind_listener("0.0.0.0:0").unwrap();
        let error = CoordinatorService::new(1).serve_tcp(listener).unwrap_err();
        assert!(error.to_string().contains("restricted to loopback"));
    }

    #[test]
    fn malformed_identifiers_are_rejected_before_the_shared_service_lock_and_do_not_poison_it() {
        let (listener, addr) = bind_listener("127.0.0.1:0").unwrap();
        let mut coordinator = CoordinatorService::new(11);
        coordinator
            .issue_cli_session(
                TenantId::from("tenant"),
                ProjectId::from("project"),
                UserId::from("user"),
                "healthy-session",
                None,
            )
            .unwrap();
        let shared = Arc::new(Mutex::new(coordinator));
        let server_shared = Arc::clone(&shared);
        let server = std::thread::spawn(move || {
            let (stream, _) = listener.accept().unwrap();
            handle_shared_stream(server_shared, stream, ClientAuthorityMode::Strict).unwrap();
        });

        let mut stream = TcpStream::connect(addr).unwrap();
        let mut reader = BufReader::new(stream.try_clone().unwrap());
        for (index, malformed_process) in [
            String::new(),
            "   ".to_owned(),
            "bad\0process".to_owned(),
            "bad process!".to_owned(),
            "x".repeat(clusterflux_core::MAX_EXTERNAL_ID_BYTES + 1),
        ]
        .into_iter()
        .enumerate()
        {
            let malformed = coordinator_wire_request(
                format!("malformed-{index}"),
                json!({
                    "type": "authenticated",
                    "session_secret": "healthy-session",
                    "request": {
                        "type": "abort_process",
                        "process": malformed_process,
                        "launch_attempt": "valid-attempt"
                    }
                }),
            );
            serde_json::to_writer(&mut stream, &malformed).unwrap();
            stream.write_all(b"\n").unwrap();
            stream.flush().unwrap();

            let mut line = String::new();
            reader.read_line(&mut line).unwrap();
            let CoordinatorResponse::Error { message } =
                serde_json::from_str::<CoordinatorResponse>(&line).unwrap()
            else {
                panic!("malformed identifier request unexpectedly succeeded");
            };
            assert!(
                message.contains("malformed external identifier")
                    && message.contains("request.request.process"),
                "unexpected malformed identifier response: {message}"
            );

            let valid = coordinator_wire_request(
                format!("healthy-{index}"),
                json!({
                    "type": "authenticated",
                    "session_secret": "healthy-session",
                    "request": { "type": "auth_status" }
                }),
            );
            serde_json::to_writer(&mut stream, &valid).unwrap();
            stream.write_all(b"\n").unwrap();
            stream.flush().unwrap();

            line.clear();
            reader.read_line(&mut line).unwrap();
            assert!(
                matches!(
                    serde_json::from_str::<CoordinatorResponse>(&line).unwrap(),
                    CoordinatorResponse::AuthStatus {
                        authenticated: true,
                        ..
                    }
                ),
                "valid authenticated traffic failed after malformed request {index}"
            );
        }

        stream.shutdown(std::net::Shutdown::Both).unwrap();
        server.join().unwrap();
        assert!(!shared.is_poisoned());
    }
}