use crate::{
Error,
model::{ArchivedRecord, ChainPoint, ObjectStatus, PinReceipt, PruneReceipt, ServiceStatus},
names::{
MAX_NAMES_PAGE, MAX_NAMES_PAGE_BYTES, MAX_NAMES_STATE_BYTES, NamesCursor, NamesEntry,
NamesPage, NamesResolution, NamesStatus, NamesStep,
},
projection::{MAX_WIRE_PAGE, WireCursor, WirePage},
};
use bitcoin::Txid;
use std::{
io::{self, Read, Write},
net::{SocketAddr, TcpStream},
time::{Duration, Instant},
};
use urma_names::name::Name;
mod overlay;
const CONNECT_TIMEOUT: Duration = Duration::from_secs(2);
const OPERATION_TIMEOUT: Duration = Duration::from_secs(5);
const MAX_HEADER: usize = 8192;
const MAX_BODY: usize = 32 * 1024 * 1024;
pub struct Client {
address: SocketAddr,
token: String,
}
impl Client {
pub fn new(address: SocketAddr, token: String) -> Result<Self, Error> {
if !address.ip().is_loopback()
|| token.len() != 64
|| !token.bytes().all(|byte| byte.is_ascii_hexdigit())
{
return Err(invalid(
"client requires loopback and a 64-hex bearer token",
));
}
Ok(Self { address, token })
}
pub fn status(&self) -> Result<ServiceStatus, Error> {
Ok(serde_json::from_value(self.status_json()?)?)
}
pub fn status_json(&self) -> Result<serde_json::Value, Error> {
let deadline = Instant::now() + OPERATION_TIMEOUT;
let (code, body) = self.request("GET", "/v1/status", deadline)?;
if code != 200 {
return Err(invalid(format!("local status HTTP {code}")));
}
let value = serde_json::from_slice(&body)?;
remaining(deadline)?;
Ok(value)
}
pub fn record(&self, txid: Txid) -> Result<Option<ArchivedRecord>, Error> {
let deadline = Instant::now() + OPERATION_TIMEOUT;
let (code, body) = self.request("GET", &format!("/v1/records/{txid}"), deadline)?;
match code {
200 => {
let record: ArchivedRecord = serde_json::from_slice(&body)?;
if record.receipt.txid != txid {
return Err(invalid("local record receipt TXID differs from request"));
}
remaining(deadline)?;
Ok(Some(record))
}
404 => {
let _: serde_json::Value = serde_json::from_slice(&body)?;
remaining(deadline)?;
Ok(None)
}
_ => Err(invalid(format!("local record HTTP {code}"))),
}
}
pub fn object_status(&self, root: Txid) -> Result<Option<ObjectStatus>, Error> {
let deadline = Instant::now() + OPERATION_TIMEOUT;
let (code, body) = self.request("GET", &format!("/v1/objects/{root}"), deadline)?;
match code {
200 => {
let status: ObjectStatus = serde_json::from_slice(&body)?;
remaining(deadline)?;
if status.root != root {
return Err(invalid("local object status root differs from request"));
}
if status.members_total == 0
|| status.members_local > status.members_total
|| status.members_selected > status.members_total
{
return Err(invalid("local object status has impossible member counts"));
}
Ok(Some(status))
}
404 => {
let _: serde_json::Value = serde_json::from_slice(&body)?;
remaining(deadline)?;
Ok(None)
}
_ => Err(invalid(format!("local object status HTTP {code}"))),
}
}
pub fn set_pin(&self, root: Txid, pin: bool) -> Result<PinReceipt, Error> {
let deadline = Instant::now() + OPERATION_TIMEOUT;
let action = if pin { "pin" } else { "unpin" };
let (code, body) =
self.request("POST", &format!("/v1/objects/{root}/{action}"), deadline)?;
if code != 200 {
return Err(invalid(format!("local {action} HTTP {code}")));
}
let receipt: PinReceipt = serde_json::from_slice(&body)?;
remaining(deadline)?;
if receipt.root != root || receipt.pinned != pin {
return Err(invalid("local pin receipt differs from request"));
}
Ok(receipt)
}
pub fn prune_cache(&self) -> Result<PruneReceipt, Error> {
let deadline = Instant::now() + OPERATION_TIMEOUT;
let (code, body) = self.request("POST", "/v1/cache/prune", deadline)?;
if code != 200 {
return Err(invalid(format!("local cache prune HTTP {code}")));
}
let receipt = serde_json::from_slice(&body)?;
remaining(deadline)?;
Ok(receipt)
}
pub fn wire_page(&self, cursor: Option<&WireCursor>, limit: u32) -> Result<WirePage, Error> {
let deadline = Instant::now() + OPERATION_TIMEOUT;
if !(1..=MAX_WIRE_PAGE).contains(&limit) {
return Err(invalid("local Wire limit must be 1..128"));
}
let mut path = format!("/v1/wire?limit={limit}");
if let Some(cursor) = cursor {
validate_instance(&cursor.snapshot.instance)?;
if cursor.height > i64::MAX as u64 {
return Err(invalid("local Wire cursor height is out of range"));
}
let encoded = serde_json::to_vec(cursor)?;
if encoded.len() > 512 {
return Err(invalid("local Wire cursor exceeds 1024 hex characters"));
}
path.push_str("&cursor=");
path.push_str(&hex::encode(encoded));
}
let (code, body) = self.request("GET", &path, deadline)?;
if code == 409 {
return Err(Error::Busy(
"local Wire snapshot changed; discard traversal and restart".into(),
));
}
if code != 200 {
return Err(invalid(format!("local Wire HTTP {code}")));
}
let page: WirePage = serde_json::from_slice(&body)?;
remaining(deadline)?;
validate_wire_page(&page, cursor, limit, deadline)?;
remaining(deadline)?;
Ok(page)
}
pub fn register_names(&self, genesis: Txid, block: ChainPoint) -> Result<NamesStatus, Error> {
let deadline = Instant::now() + OPERATION_TIMEOUT;
if block.height > i64::MAX as u64 {
return Err(invalid("local Names registration height is out of range"));
}
let path = format!(
"/v1/names/{genesis}/register?height={}&hash={}",
block.height, block.hash
);
let (code, body) = self.names_request("POST", &path, deadline)?;
if code != 200 {
return Err(invalid(format!("local Names registration HTTP {code}")));
}
let status: NamesStatus = serde_json::from_slice(&body)?;
remaining(deadline)?;
validate_names_status(&status, genesis)?;
if status.genesis_block != block {
return Err(invalid(
"local Names genesis block differs from registration",
));
}
remaining(deadline)?;
Ok(status)
}
pub fn names_step(&self, genesis: Txid) -> Result<NamesStep, Error> {
let deadline = Instant::now() + OPERATION_TIMEOUT;
let (code, body) =
self.names_request("POST", &format!("/v1/names/{genesis}/step"), deadline)?;
if code != 200 {
return Err(invalid(format!("local Names step HTTP {code}")));
}
let step: NamesStep = serde_json::from_slice(&body)?;
remaining(deadline)?;
validate_names_status(&step.status, genesis)?;
if step
.applied
.is_some_and(|point| point != step.status.projected)
{
return Err(invalid("local Names applied block differs from projection"));
}
remaining(deadline)?;
Ok(step)
}
pub fn names_status(&self, genesis: Txid) -> Result<Option<NamesStatus>, Error> {
let deadline = Instant::now() + OPERATION_TIMEOUT;
let (code, body) = self.names_request("GET", &format!("/v1/names/{genesis}"), deadline)?;
match code {
200 => {
let status: NamesStatus = serde_json::from_slice(&body)?;
remaining(deadline)?;
validate_names_status(&status, genesis)?;
remaining(deadline)?;
Ok(Some(status))
}
404 => {
let _: serde_json::Value = serde_json::from_slice(&body)?;
remaining(deadline)?;
Ok(None)
}
_ => Err(invalid(format!("local Names status HTTP {code}"))),
}
}
pub fn names_resolve(&self, genesis: Txid, name: &Name) -> Result<NamesResolution, Error> {
let deadline = Instant::now() + OPERATION_TIMEOUT;
let path = format!("/v1/names/{genesis}/resolve/{}", name.as_str());
let (code, body) = self.names_request("GET", &path, deadline)?;
if code != 200 {
return Err(invalid(format!("local Names resolution HTTP {code}")));
}
let resolution: NamesResolution = serde_json::from_slice(&body)?;
remaining(deadline)?;
validate_names_status(&resolution.status, genesis)?;
remaining(deadline)?;
Ok(resolution)
}
pub fn names_view(
&self,
genesis: Txid,
cursor: Option<&NamesCursor>,
limit: u32,
) -> Result<NamesPage, Error> {
let deadline = Instant::now() + OPERATION_TIMEOUT;
if !(1..=MAX_NAMES_PAGE).contains(&limit) {
return Err(invalid("local Names view limit must be 1..128"));
}
let mut path = format!("/v1/names/{genesis}/view?limit={limit}");
if let Some(cursor) = cursor {
if cursor.genesis != genesis {
return Err(invalid("local Names cursor genesis differs from request"));
}
validate_names_instance(&cursor.snapshot.instance)?;
let encoded = serde_json::to_vec(cursor)?;
if encoded.len() > 512 {
return Err(invalid("local Names cursor exceeds 1024 hex characters"));
}
path.push_str("&cursor=");
path.push_str(&hex::encode(encoded));
}
let (code, body) = self.names_request("GET", &path, deadline)?;
if code != 200 {
return Err(invalid(format!("local Names view HTTP {code}")));
}
if body.len() > MAX_NAMES_PAGE_BYTES {
return Err(invalid("local Names page exceeds 16 MiB"));
}
let page: NamesPage = serde_json::from_slice(&body)?;
remaining(deadline)?;
validate_names_page(&page, genesis, cursor, limit, deadline)?;
remaining(deadline)?;
Ok(page)
}
fn names_request(
&self,
method: &str,
path: &str,
deadline: Instant,
) -> Result<(u16, Vec<u8>), Error> {
let response = self.request(method, path, deadline)?;
if response.0 == 409 {
return Err(Error::Busy(
"local Names operation busy or selected state changed".into(),
));
}
Ok(response)
}
fn request(
&self,
method: &str,
path: &str,
deadline: Instant,
) -> Result<(u16, Vec<u8>), Error> {
let mut stream =
TcpStream::connect_timeout(&self.address, CONNECT_TIMEOUT.min(remaining(deadline)?))?;
let body_length = if method == "POST" {
"Content-Length: 0\r\n"
} else {
""
};
let request = format!(
"{method} {path} HTTP/1.1\r\nHost: {}\r\nAuthorization: Bearer {}\r\nAccept: application/json\r\nAccept-Encoding: identity\r\n{body_length}Connection: close\r\n\r\n",
self.address, self.token
);
let mut pending = request.as_bytes();
while !pending.is_empty() {
stream.set_write_timeout(Some(remaining(deadline)?))?;
let count = match stream.write(pending) {
Err(error) if error.kind() == io::ErrorKind::Interrupted => continue,
result => result?,
};
remaining(deadline)?;
if count == 0 {
return Err(io::Error::from(io::ErrorKind::WriteZero).into());
}
pending = &pending[count..];
}
let mut header = Vec::new();
while !header.ends_with(b"\r\n\r\n") {
if header.len() == MAX_HEADER {
return Err(invalid("local HTTP header exceeds 8192 bytes"));
}
let mut byte = [0];
read(&mut stream, &mut byte, deadline)?;
header.push(byte[0]);
}
let (code, length) = parse_header(&header)?;
let mut body = vec![0; length];
let mut received = 0;
while received < length {
received += read(&mut stream, &mut body[received..], deadline)?;
}
remaining(deadline)?;
Ok((code, body))
}
}
fn invalid(message: impl Into<String>) -> Error {
Error::Invalid(message.into())
}
fn validate_names_status(status: &NamesStatus, genesis: Txid) -> Result<(), Error> {
if status.genesis != genesis {
return Err(invalid("local Names genesis differs from request"));
}
let tip = status
.selected_tip
.ok_or_else(|| invalid("local Names projection has no selected tip"))?;
if status.projected.height < status.genesis_block.height || tip.height < status.projected.height
{
return Err(invalid("local Names projection heights are inconsistent"));
}
for (left, right) in [
(status.genesis_block, status.projected),
(status.projected, tip),
(status.genesis_block, tip),
] {
if left.height == right.height && left.hash != right.hash {
return Err(invalid("local Names equal-height block hashes differ"));
}
}
if status.complete_to_tip != (status.projected == tip) {
return Err(invalid(
"local Names completion differs from projection and tip",
));
}
Ok(())
}
fn validate_names_instance(instance: &str) -> Result<(), Error> {
if instance.len() != 32 || !instance.bytes().all(|byte| byte.is_ascii_hexdigit()) {
return Err(invalid(
"local Names snapshot instance must be 32 hex characters",
));
}
Ok(())
}
fn names_json_bound(value: &impl serde::Serialize, max: usize, message: &str) -> Result<(), Error> {
struct Counter {
count: usize,
max: usize,
exceeded: bool,
}
impl Write for Counter {
fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
if bytes.len() > self.max - self.count {
self.exceeded = true;
return Err(io::Error::other("serialized Names byte bound"));
}
self.count += bytes.len();
Ok(bytes.len())
}
fn flush(&mut self) -> io::Result<()> {
Ok(())
}
}
let mut counter = Counter {
count: 0,
max,
exceeded: false,
};
serde_json::to_writer(&mut counter, value).map_err(|error| {
if counter.exceeded {
invalid(message)
} else {
Error::Json(error)
}
})
}
fn validate_names_page(
page: &NamesPage,
genesis: Txid,
cursor: Option<&NamesCursor>,
limit: u32,
deadline: Instant,
) -> Result<(), Error> {
validate_names_instance(&page.snapshot.instance)?;
validate_names_status(&page.status, genesis)?;
if page.entries.len() > limit as usize {
return Err(invalid(
"local Names page entry count exceeds requested limit",
));
}
if cursor.is_some_and(|cursor| cursor.snapshot != page.snapshot) {
return Err(invalid("local Names page snapshot differs from cursor"));
}
names_json_bound(
&page.rules,
MAX_NAMES_STATE_BYTES,
"local Names rules exceed 64 KiB",
)?;
remaining(deadline)?;
page.rules
.validate()
.map_err(|_| invalid("local Names rules are invalid"))?;
let mut previous = cursor.map(|cursor| cursor.after.order_key());
for entry in &page.entries {
remaining(deadline)?;
let key = entry.key().order_key();
if previous.as_ref().is_some_and(|previous| &key <= previous) {
return Err(invalid(
"local Names keys are not strictly after cursor/in order",
));
}
match entry {
NamesEntry::Name { state, .. } => names_json_bound(
state,
MAX_NAMES_STATE_BYTES,
"local Names state exceeds 64 KiB",
)?,
NamesEntry::Pending { record, .. } => names_json_bound(
record,
MAX_NAMES_STATE_BYTES,
"local Names pending state exceeds 64 KiB",
)?,
}
previous = Some(key);
}
if let Some(next) = &page.next {
let last = page
.entries
.last()
.ok_or_else(|| invalid("empty local Names page has a continuation"))?;
if next.genesis != genesis || next.snapshot != page.snapshot || next.after != last.key() {
return Err(invalid(
"local Names continuation differs from last key/genesis/snapshot",
));
}
}
remaining(deadline)?;
names_json_bound(
page,
MAX_NAMES_PAGE_BYTES,
"local Names page exceeds 16 MiB",
)?;
remaining(deadline)?;
Ok(())
}
fn validate_instance(instance: &str) -> Result<(), Error> {
if instance.len() != 32 || !instance.bytes().all(|byte| byte.is_ascii_hexdigit()) {
return Err(invalid(
"local Wire snapshot instance must be 32 hex characters",
));
}
Ok(())
}
fn validate_wire_page(
page: &WirePage,
cursor: Option<&WireCursor>,
limit: u32,
deadline: Instant,
) -> Result<(), Error> {
validate_instance(&page.snapshot.instance)?;
if page.archive.api_version != crate::model::API_VERSION || page.records.len() > limit as usize
{
return Err(invalid(
"local Wire page version or record count is invalid",
));
}
if cursor.is_some_and(|cursor| cursor.snapshot != page.snapshot) {
return Err(invalid("local Wire page snapshot differs from cursor"));
}
let mut previous_coordinate = cursor.map(|cursor| (cursor.height, cursor.position));
let mut previous_block = None;
for record in &page.records {
remaining(deadline)?;
let entry = &record.entry;
let coordinate = (entry.height, entry.position);
if previous_coordinate.is_some_and(|previous| coordinate <= previous) {
return Err(invalid(
"local Wire entries are not strictly after the cursor/in order",
));
}
let start = page.archive.application_start;
let tip = page
.archive
.selected_tip
.ok_or_else(|| invalid("local Wire records require a selected tip"))?;
if entry.height < start.height || entry.height > tip.height {
return Err(invalid(
"local Wire entry lies outside the application/selected range",
));
}
if (entry.height == start.height && record.block_hash != start.hash)
|| (entry.height == tip.height && record.block_hash != tip.hash)
|| previous_block
.is_some_and(|(height, hash)| height == entry.height && hash != record.block_hash)
{
return Err(invalid(
"local Wire block hashes disagree at the same height",
));
}
let txid = entry
.txid
.parse::<Txid>()
.map_err(|_| invalid("local Wire entry has an invalid TXID"))?;
let author = entry
.author
.parse::<bitcoin::XOnlyPublicKey>()
.map_err(|_| invalid("local Wire entry has an invalid author"))?;
if entry.record.len() > urma_core::format::Urma::MAX_PUBLIC_BYTES * 2 {
return Err(invalid("local Wire record exceeds the public byte bound"));
}
let raw = hex::decode(&entry.record)
.map_err(|_| invalid("local Wire entry has invalid record hex"))?;
let parsed = urma_wire::index::entry_from_record(
&raw,
txid,
author,
None,
entry.height,
entry.position,
)
.map_err(|error| invalid(format!("invalid local Wire record: {error}")))?;
if parsed.is_none() {
return Err(invalid("local Wire entry is not a Wire record kind"));
}
previous_coordinate = Some(coordinate);
previous_block = Some((entry.height, record.block_hash));
}
if let Some(next) = &page.next {
let last = page
.records
.last()
.ok_or_else(|| invalid("empty local Wire page has a continuation"))?;
if next.snapshot != page.snapshot
|| (next.height, next.position) != (last.entry.height, last.entry.position)
{
return Err(invalid(
"local Wire continuation differs from the last entry/snapshot",
));
}
}
Ok(())
}
fn remaining(deadline: Instant) -> Result<Duration, Error> {
deadline
.checked_duration_since(Instant::now())
.filter(|duration| !duration.is_zero())
.ok_or_else(|| {
io::Error::new(io::ErrorKind::TimedOut, "local HTTP operation deadline").into()
})
}
fn read(stream: &mut TcpStream, bytes: &mut [u8], deadline: Instant) -> Result<usize, Error> {
loop {
stream.set_read_timeout(Some(remaining(deadline)?))?;
let count = match stream.read(bytes) {
Err(error) if error.kind() == io::ErrorKind::Interrupted => continue,
result => result?,
};
remaining(deadline)?;
if count == 0 {
return Err(io::Error::new(
io::ErrorKind::UnexpectedEof,
"truncated local HTTP response",
)
.into());
}
return Ok(count);
}
}
fn is_token(byte: u8) -> bool {
byte.is_ascii_alphanumeric() || b"!#$%&'*+-.^_`|~".contains(&byte)
}
fn json_content_type(value: &str) -> bool {
let mut parts = value.split(';');
if !parts
.next()
.unwrap()
.trim()
.eq_ignore_ascii_case("application/json")
{
return false;
}
if let Some(parameter) = parts.next() {
let Some((name, charset)) = parameter.trim().split_once('=') else {
return false;
};
let charset = charset.trim();
let charset = charset
.strip_prefix('"')
.and_then(|s| s.strip_suffix('"'))
.unwrap_or(charset);
if !name.trim().eq_ignore_ascii_case("charset")
|| charset.is_empty()
|| !charset.bytes().all(is_token)
{
return false;
}
}
parts.next().is_none()
}
fn parse_header(bytes: &[u8]) -> Result<(u16, usize), Error> {
let text =
std::str::from_utf8(bytes).map_err(|_| invalid("invalid local HTTP header encoding"))?;
let mut lines = text[..text.len() - 4].split("\r\n");
let mut status = lines.next().unwrap().splitn(3, ' ');
if status.next() != Some("HTTP/1.1") {
return Err(invalid("local response requires HTTP/1.1"));
}
let code = status.next().unwrap_or("");
if code.len() != 3 || !code.bytes().all(|byte| byte.is_ascii_digit()) {
return Err(invalid("malformed local HTTP status"));
}
let reason = status
.next()
.ok_or_else(|| invalid("malformed local HTTP status"))?;
if !reason
.bytes()
.all(|byte| byte == b'\t' || (32..=126).contains(&byte))
{
return Err(invalid("malformed local HTTP status"));
}
let code: u16 = code
.parse()
.map_err(|_| invalid("malformed local HTTP status"))?;
if !(100..=599).contains(&code) {
return Err(invalid("malformed local HTTP status"));
}
let mut length = None;
let mut content_type = None;
for line in lines {
let (name, value) = line
.split_once(':')
.ok_or_else(|| invalid("malformed local HTTP header"))?;
if name.is_empty()
|| !name.bytes().all(is_token)
|| !value
.bytes()
.all(|byte| byte == b'\t' || (32..=126).contains(&byte))
{
return Err(invalid("malformed local HTTP header"));
}
let value = value.trim_matches([' ', '\t']);
if name.eq_ignore_ascii_case("transfer-encoding") {
return Err(invalid("local HTTP transfer encoding is unsupported"));
}
if name.eq_ignore_ascii_case("content-encoding") && !value.eq_ignore_ascii_case("identity")
{
return Err(invalid("local HTTP content encoding is unsupported"));
}
if name.eq_ignore_ascii_case("content-length") {
if length.is_some()
|| value.is_empty()
|| !value.bytes().all(|byte| byte.is_ascii_digit())
{
return Err(invalid(
"local HTTP requires a single decimal Content-Length",
));
}
let parsed: usize = value
.parse()
.map_err(|_| invalid("invalid local HTTP Content-Length"))?;
if parsed > MAX_BODY {
return Err(invalid("local HTTP body exceeds 32 MiB"));
}
length = Some(parsed);
}
if name.eq_ignore_ascii_case("content-type") {
if content_type.is_some() || !json_content_type(value) {
return Err(invalid(
"local HTTP requires application/json with optional charset",
));
}
content_type = Some(());
}
}
if content_type.is_none() {
return Err(invalid("local HTTP Content-Type is missing"));
}
Ok((
code,
length.ok_or_else(|| invalid("local HTTP Content-Length is missing"))?,
))
}