use std::sync::{Arc, Mutex};
use super::{ServedState, ServerError};
use crate::auth_lys::drafts::admin::{AdminRequest, AdminResponse};
pub trait DraftEndpoint: Send + Sync + std::fmt::Debug {
fn limits(&self) -> crate::auth_lys::drafts::admin::AdminLimits;
fn handle(&self, request: AdminRequest<'_>) -> Result<AdminResponse, ServerError>;
}
pub(super) type DraftRoute = Arc<Mutex<Option<Arc<dyn DraftEndpoint>>>>;
impl super::HealthServerHandle {
pub(crate) fn install_drafts(
&self,
endpoint: Arc<dyn DraftEndpoint>,
) -> Result<(), ServerError> {
let mut held = self
.drafts
.lock()
.map_err(|_| refused("draft route lock poisoned"))?;
if held.is_some() {
return Err(refused("draft route already installed"));
}
*held = Some(endpoint);
drop(held);
Ok(())
}
pub(crate) fn release_drafts(&self) -> Result<(), ServerError> {
drop(
self.drafts
.lock()
.map_err(|_| refused("draft route lock poisoned"))?
.take(),
);
Ok(())
}
}
fn refused(message: &str) -> ServerError {
ServerError::HealthEndpoint {
message: message.into(),
}
}
pub(super) fn route(bytes: &[u8], served: &ServedState) -> Result<Option<Vec<u8>>, ServerError> {
let Some(end) = bytes.windows(4).position(|part| part == b"\r\n\r\n") else {
return Ok(None);
};
let Ok(headers) = std::str::from_utf8(&bytes[..end]) else {
return Ok(None);
};
let mut lines = headers.split("\r\n");
let Some((method, path)) = lines.next().and_then(super::parse_request_line) else {
return Ok(None);
};
if path != crate::auth_lys::drafts::admin::PULL_PATH
&& path != crate::auth_lys::drafts::admin::CREATE_PATH
{
return Ok(None);
}
let endpoint = served
.drafts
.lock()
.map_err(|_| refused("draft route lock poisoned"))?
.clone();
let Some(endpoint) = endpoint else {
return Ok(Some(super::empty_response(
super::StatusCode::ServiceUnavailable,
)));
};
let mut authorization = None;
for line in lines {
let Some((name, value)) = line.split_once(':') else {
return Ok(Some(super::empty_response(super::StatusCode::BadRequest)));
};
if name.eq_ignore_ascii_case("authorization") {
if authorization.is_some() {
return Ok(Some(super::empty_response(super::StatusCode::BadRequest)));
}
authorization = Some(value.trim());
}
}
let at_ms = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map_err(|_| refused("draft pull clock is before epoch"))?;
let at_ms =
u64::try_from(at_ms.as_millis()).map_err(|_| refused("draft pull clock overflow"))?;
let answer = endpoint.handle(AdminRequest {
method,
path,
authorization,
body: &bytes[end + 4..],
at_ms,
})?;
let body =
serde_json::to_vec(&answer.body).map_err(|_| refused("draft reply encoding failed"))?;
let mut response = format!("HTTP/1.1 {} {}\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n", answer.status,
match answer.status { 200 => "OK", 400 => "Bad Request", 401 => "Unauthorized", 403 => "Forbidden", 404 => "Not Found", 405 => "Method Not Allowed", _ => "Service Unavailable" }, body.len()).into_bytes();
response.extend(body);
Ok(Some(response))
}
pub(super) fn read(
stream: &mut std::net::TcpStream,
served: &ServedState,
) -> Result<Vec<u8>, ServerError> {
use std::io::Read;
let endpoint = served
.drafts
.lock()
.map_err(|_| refused("draft route lock poisoned"))?
.clone();
let limit = endpoint
.as_ref()
.map_or(Some(super::READ_BUFFER_BYTES), |endpoint| {
let limits = endpoint.limits();
limits
.credential_bytes
.get()
.checked_add(limits.body_bytes.get())?
.checked_add(super::READ_BUFFER_BYTES)
})
.ok_or_else(|| refused("draft HTTP bound overflow"))?;
let mut bytes = Vec::new();
let mut expected = None;
loop {
let mut part = [0; super::READ_BUFFER_BYTES];
let remaining = limit.saturating_sub(bytes.len());
if remaining == 0 {
return Err(refused("health HTTP request bound exceeded"));
}
let count = stream
.read(&mut part[..remaining.min(super::READ_BUFFER_BYTES)])
.map_err(|_| refused("health HTTP request read failed"))?;
if count == 0 {
return if bytes.is_empty() {
Ok(bytes)
} else {
Err(refused("health HTTP request incomplete"))
};
}
bytes.extend_from_slice(&part[..count]);
if expected.is_none() {
if let Some(end) = bytes.windows(4).position(|part| part == b"\r\n\r\n") {
let head = std::str::from_utf8(&bytes[..end])
.map_err(|_| refused("health HTTP headers invalid"))?;
let mut length = None;
for line in head.split("\r\n").skip(1) {
let (name, value) = line
.split_once(':')
.ok_or_else(|| refused("health HTTP header invalid"))?;
if name.eq_ignore_ascii_case("transfer-encoding") {
return Err(refused("health HTTP transfer encoding refused"));
}
if name.eq_ignore_ascii_case("content-length") {
if length.is_some() {
return Err(refused("health HTTP content length repeated"));
}
length = Some(
value
.trim()
.parse::<usize>()
.map_err(|_| refused("health HTTP content length invalid"))?,
);
}
}
expected = Some(
(end + 4)
.checked_add(length.unwrap_or(0))
.filter(|size| *size <= limit)
.ok_or_else(|| refused("health HTTP body bound exceeded"))?,
);
}
}
if let Some(expected) = expected {
if bytes.len() > expected {
return Err(refused("health HTTP trailing bytes refused"));
}
if bytes.len() == expected {
return Ok(bytes);
}
}
}
}