pub mod actions;
use std::collections::HashMap;
use std::convert::Infallible;
use std::net::SocketAddr;
use std::sync::Arc;
use bytes::Bytes;
use http_body_util::{BodyExt, Full};
use hyper::body::Incoming;
use hyper::server::conn::http1;
use hyper::service::service_fn;
use hyper::{Request, Response};
use hyper_util::rt::TokioIo;
use tokio::sync::mpsc;
use tracing::{debug, error, info, trace};
use crate::llm::ollama_client::OllamaClient;
use crate::llm::ActionResult;
use crate::logging::emit::Log;
use crate::server::connection::ConnectionId;
use crate::server::IppProtocol;
use crate::state::app_state::AppState;
use crate::utils::WireFailure;
use crate::{console_error, console_info};
use actions::{build_ipp_response, ipp_status_code};
pub const MAX_IPP_BODY_BYTES: usize = 8 * 1024 * 1024;
const LINGER_DRAIN_BYTES: usize = 8 * 1024 * 1024;
const LINGER_DRAIN_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5);
enum BodyRead {
Complete(Bytes),
TooLarge { drained: usize, whole_body: bool },
}
async fn read_body_bounded(body: Incoming) -> BodyRead {
let mut body = std::pin::pin!(body);
let mut buffered: Vec<u8> = Vec::new();
let mut over = false;
let mut drained = 0usize;
let deadline = tokio::time::Instant::now() + LINGER_DRAIN_TIMEOUT;
loop {
let next = if over {
match tokio::time::timeout_at(deadline, body.frame()).await {
Ok(next) => next,
Err(_) => {
return BodyRead::TooLarge {
drained,
whole_body: false,
}
}
}
} else {
body.frame().await
};
let Some(frame) = next else {
return if over {
BodyRead::TooLarge {
drained,
whole_body: true,
}
} else {
BodyRead::Complete(Bytes::from(buffered))
};
};
let Ok(frame) = frame else {
return BodyRead::TooLarge {
drained,
whole_body: false,
};
};
let Ok(data) = frame.into_data() else {
continue; };
if over {
drained = drained.saturating_add(data.len());
if drained >= LINGER_DRAIN_BYTES {
return BodyRead::TooLarge {
drained,
whole_body: false,
};
}
continue;
}
if buffered.len() + data.len() > MAX_IPP_BODY_BYTES {
over = true;
buffered = Vec::new();
drained = data.len();
continue;
}
buffered.extend_from_slice(&data);
}
}
const MAX_CONNECTIONS: usize = crate::server::accept_bounded::DEFAULT_MAX_CONNECTIONS;
const CONNECTION_CAP_BODY: &str = "Too many connections, try again later.\n";
static CONNECTION_CAP_REFUSAL: std::sync::LazyLock<Vec<u8>> = std::sync::LazyLock::new(|| {
format!(
"HTTP/1.1 503 Service Unavailable\r\nRetry-After: 30\r\nContent-Type: \
text/plain\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
CONNECTION_CAP_BODY.len(),
CONNECTION_CAP_BODY
)
.into_bytes()
});
pub struct IppServer;
impl IppServer {
pub async fn spawn_with_llm_actions(
listen_addr: SocketAddr,
llm_client: OllamaClient,
app_state: Arc<AppState>,
status_tx: mpsc::UnboundedSender<String>,
_send_first: bool,
server_id: crate::state::ServerId,
) -> anyhow::Result<SocketAddr> {
let listener =
crate::server::socket_helpers::create_reusable_tcp_listener(listen_addr).await?;
let local_addr = listener.local_addr()?;
console_info!(status_tx, "IPP server listening on {}", local_addr);
let protocol = Arc::new(IppProtocol::new());
let task_registrar = app_state.clone();
let limiter = crate::server::accept_bounded::ConnectionLimiter::new(MAX_CONNECTIONS);
let accept_handle = tokio::spawn(async move {
loop {
match crate::server::accept_bounded::accept_bounded(
&listener,
&limiter,
&CONNECTION_CAP_REFUSAL,
"IPP",
Some(&status_tx),
)
.await
{
Ok((stream, remote_addr, permit)) => {
let connection_id =
ConnectionId::new(app_state.get_next_unified_id().await);
let local_addr_conn = stream.local_addr().unwrap_or(local_addr);
info!("IPP connection {} from {}", connection_id, remote_addr);
Log::new(Some(&status_tx))
.info(format!("IPP connection from {}", remote_addr));
use crate::state::server::{
ConnectionState as ServerConnectionState, ConnectionStatus,
ProtocolConnectionInfo,
};
let now = crate::utils::clock::Instant::now();
let conn_state = ServerConnectionState {
id: connection_id,
remote_addr,
local_addr: local_addr_conn,
bytes_sent: 0,
bytes_received: 0,
packets_sent: 0,
packets_received: 0,
last_activity: now,
status: ConnectionStatus::Active,
status_changed_at: now,
protocol_info: ProtocolConnectionInfo::empty(),
};
app_state
.add_connection_to_server(server_id, conn_state)
.await;
let _ = status_tx.send("__UPDATE_UI__".to_string());
let llm_client_clone = llm_client.clone();
let app_state_clone = app_state.clone();
let status_tx_clone = status_tx.clone();
let protocol_clone = protocol.clone();
let task_owner = app_state.clone();
task_owner
.spawn_server_task(server_id, async move {
let _permit = permit;
let io = TokioIo::new(stream);
let status_for_service = status_tx_clone.clone();
let app_state_for_service = app_state_clone.clone();
let service = service_fn(move |req: Request<Incoming>| {
let llm_clone = llm_client_clone.clone();
let state_clone = app_state_for_service.clone();
let status_clone = status_for_service.clone();
let protocol_clone = protocol_clone.clone();
handle_ipp_request_with_llm(
req,
connection_id,
llm_clone,
state_clone,
status_clone,
protocol_clone,
server_id,
)
});
if let Err(err) =
http1::Builder::new().serve_connection(io, service).await
{
error!("Error serving IPP connection: {:?}", err);
}
app_state_clone
.close_connection_on_server(server_id, connection_id)
.await;
Log::new(Some(&status_tx_clone))
.info(format!("IPP connection {} closed", connection_id));
let _ = status_tx_clone.send("__UPDATE_UI__".to_string());
})
.await;
}
Err(e) => {
console_error!(status_tx, "Failed to accept IPP connection: {}", e);
break;
}
}
}
});
task_registrar
.register_server_task(server_id, accept_handle)
.await;
Ok(local_addr)
}
}
async fn handle_ipp_request_with_llm(
req: Request<Incoming>,
connection_id: ConnectionId,
llm_client: OllamaClient,
app_state: Arc<AppState>,
status_tx: mpsc::UnboundedSender<String>,
protocol: Arc<IppProtocol>,
server_id: crate::state::ServerId,
) -> Result<Response<Full<Bytes>>, Infallible> {
let method = req.method().to_string();
let uri = req.uri().to_string();
let mut headers = HashMap::new();
for (name, value) in req.headers() {
if let Ok(value_str) = value.to_str() {
headers.insert(name.to_string(), value_str.to_string());
}
}
let body_bytes = match read_body_bounded(req.into_body()).await {
BodyRead::Complete(bytes) => bytes,
BodyRead::TooLarge {
drained,
whole_body,
} => {
console_error!(
status_tx,
"IPP {} refusing request body; limit is {} bytes",
connection_id,
MAX_IPP_BODY_BYTES
);
Log::new(Some(&status_tx)).warn(format!(
"IPP {} decision=fail_closed_body_too_large (limit {} bytes, drained {} more, \
peer finished writing: {})",
connection_id, MAX_IPP_BODY_BYTES, drained, whole_body
));
let body = build_ipp_response(
ipp_status_code("client-error-request-entity-too-large"),
Some("request body exceeds this server's limit"),
None,
);
return Ok(ipp_response_counted(&app_state, server_id, connection_id, 413, body).await);
}
};
app_state
.update_connection_stats(
server_id,
connection_id,
Some(body_bytes.len() as u64),
None,
Some(1),
None,
)
.await;
Log::new(Some(&status_tx)).debug(format!(
"IPP {} {} ({} bytes)",
method,
uri,
body_bytes.len()
));
let header = parse_ipp_header(&body_bytes);
let operation_name = header
.as_ref()
.map(|h| h.operation.clone())
.unwrap_or_else(|| {
if body_bytes.is_empty() {
"Empty".to_string()
} else {
"Malformed".to_string()
}
});
let request_id = header.as_ref().map(|h| h.request_id).unwrap_or(0);
let ipp_version = header
.as_ref()
.map(|h| format!("{}.{}", h.version_major, h.version_minor))
.unwrap_or_else(|| "unknown".to_string());
trace!(
"IPP operation: {} (request-id {}, v{})",
operation_name,
request_id,
ipp_version
);
let event = crate::protocol::Event::new(
&actions::IPP_REQUEST_EVENT,
serde_json::json!({
"method": method,
"uri": uri,
"operation": operation_name,
"request_id": request_id,
"ipp_version": ipp_version,
"answer_with": actions::answer_with_for_operation(&operation_name),
}),
);
let llm_result = crate::llm::action_helper::call_llm(
&llm_client,
&app_state,
server_id,
Some(connection_id),
&event,
protocol.as_ref(),
)
.await;
match llm_result {
Ok(execution_result) => {
for result in execution_result.protocol_results {
if let ActionResult::Custom { name, data } = result {
if name == "ipp_response" {
let status = data
.get("http_status")
.and_then(|v| v.as_u64())
.unwrap_or(200) as u16;
let body_hex = data.get("body_hex").and_then(|v| v.as_str()).unwrap_or("");
let Ok(mut body) = hex::decode(body_hex) else {
error!(
"IPP {} could not decode its own encoded response body \
({} chars); answering server-error-internal-error",
connection_id,
body_hex.len()
);
Log::new(Some(&status_tx)).warn(format!(
"IPP {} decision=fail_closed_encoding_bug",
connection_id
));
return Ok(ipp_response_counted(
&app_state,
server_id,
connection_id,
200,
internal_error_body(header.as_ref(), request_id),
)
.await);
};
stamp_response_header(&mut body, header.as_ref(), request_id);
debug!(
"IPP {} decision=model_answer http={} request-id={} ({} bytes)",
connection_id,
status,
request_id,
body.len()
);
Log::new(Some(&status_tx)).debug(format!("IPP → {} response", status));
return Ok(ipp_response_counted(
&app_state,
server_id,
connection_id,
status,
body,
)
.await);
}
}
}
let tag = if execution_result.failures.is_empty() {
"model_silent"
} else {
"fail_closed_action_error"
};
debug!(
"IPP {} decision={} - no ipp_* response action, \
answering server-error-internal-error",
connection_id, tag
);
Log::new(Some(&status_tx)).warn(format!(
"IPP {} decision={}: no ipp_* response action, sending \
server-error-internal-error",
connection_id, tag
));
Ok(ipp_response_counted(
&app_state,
server_id,
connection_id,
200,
internal_error_body(header.as_ref(), request_id),
)
.await)
}
Err(e) => {
let failure = WireFailure::classify(&e);
console_error!(
status_tx,
"IPP {} decision=fail_closed_llm_error: {}",
connection_id,
e
);
let (status_name, message) = match failure {
WireFailure::Overloaded => (
"server-error-busy",
"netget: backend at capacity, retry later",
),
WireFailure::Unavailable => (
"server-error-internal-error",
"netget: request could not be processed",
),
};
let mut body = build_ipp_response(ipp_status_code(status_name), Some(message), None);
stamp_response_header(&mut body, header.as_ref(), request_id);
Ok(ipp_response_counted(&app_state, server_id, connection_id, 200, body).await)
}
}
}
async fn ipp_response_counted(
app_state: &Arc<AppState>,
server_id: crate::state::ServerId,
connection_id: ConnectionId,
status: u16,
body: Vec<u8>,
) -> Response<Full<Bytes>> {
let sent = body.len() as u64;
app_state
.update_connection_stats(server_id, connection_id, None, Some(sent), None, Some(1))
.await;
ipp_http_response(status, body)
}
fn ipp_http_response(status: u16, body: Vec<u8>) -> Response<Full<Bytes>> {
Response::builder()
.status(status)
.header("Content-Type", "application/ipp")
.body(Full::new(Bytes::from(body)))
.unwrap_or_else(|_| {
Response::builder()
.status(500)
.body(Full::new(Bytes::new()))
.expect("500 with an empty body is always constructible")
})
}
fn internal_error_body(header: Option<&IppHeader>, request_id: u32) -> Vec<u8> {
let mut body = vec![
0x02, 0x00, 0x05, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01, 0x47, ];
body.extend_from_slice(&[0x00, 0x12]);
body.extend_from_slice(b"attributes-charset");
body.extend_from_slice(&[0x00, 0x05]);
body.extend_from_slice(b"utf-8");
body.push(0x48); body.extend_from_slice(&[0x00, 0x1b]);
body.extend_from_slice(b"attributes-natural-language");
body.extend_from_slice(&[0x00, 0x05]);
body.extend_from_slice(b"en-us");
body.push(0x03); stamp_response_header(&mut body, header, request_id);
body
}
fn stamp_response_header(body: &mut [u8], header: Option<&IppHeader>, request_id: u32) {
use actions::REQUEST_ID_OFFSET;
if let (Some(header), true) = (header, body.len() >= 2) {
body[0] = header.version_major;
body[1] = header.version_minor;
}
if body.len() >= REQUEST_ID_OFFSET + 4 {
body[REQUEST_ID_OFFSET..REQUEST_ID_OFFSET + 4].copy_from_slice(&request_id.to_be_bytes());
}
}
struct IppHeader {
version_major: u8,
version_minor: u8,
operation: String,
request_id: u32,
}
fn parse_ipp_header(body: &[u8]) -> Option<IppHeader> {
if body.len() < 8 {
return None;
}
let operation_id = u16::from_be_bytes([body[2], body[3]]);
let request_id = u32::from_be_bytes([body[4], body[5], body[6], body[7]]);
Some(IppHeader {
version_major: body[0],
version_minor: body[1],
operation: ipp_operation_name(operation_id),
request_id,
})
}
fn ipp_operation_name(operation_id: u16) -> String {
let name = match operation_id {
0x0002 => "Print-Job",
0x0003 => "Print-URI",
0x0004 => "Validate-Job",
0x0005 => "Create-Job",
0x0006 => "Send-Document",
0x0007 => "Send-URI",
0x0008 => "Cancel-Job",
0x0009 => "Get-Job-Attributes",
0x000A => "Get-Jobs",
0x000B => "Get-Printer-Attributes",
0x000C => "Hold-Job",
0x000D => "Release-Job",
0x000E => "Restart-Job",
0x000F => "Pause-Printer",
0x0010 => "Resume-Printer",
0x0011 => "Purge-Jobs",
0x0012 => "Set-Printer-Attributes",
0x0013 => "Set-Job-Attributes",
0x003B => "Close-Job",
0x003C => "Identify-Printer",
0x003D => "Validate-Document",
_ => return format!("Operation-0x{:04X}", operation_id),
};
name.to_string()
}