use crate::{
Error, MAX_OBJECT_BYTES, MAX_TRANSACTION_OBJECT_BYTES, MergePair, NodeData, NodeId, ObjectId,
Owner, Provenance, Result, TransactionId, WriterId, model::MAX_TRANSACTION_BYTES,
};
use chrono::{DateTime, Utc};
use ed25519_dalek::{Signature, Signer, SigningKey, VerifyingKey};
use sha2::{Digest, Sha256};
use std::str;
const MAGIC: &[u8; 8] = b"KWTX\0\x02\0\0";
const SIGNATURE_BYTES: usize = 64;
#[derive(Clone, Debug, Eq, PartialEq)]
pub(crate) struct ObjectDeclaration {
pub id: ObjectId,
pub length: u64,
pub sha256: [u8; 32],
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub(crate) struct NodeOperation {
pub id: NodeId,
pub data: NodeData,
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub(crate) struct UnsignedTransaction {
pub writer: WriterId,
pub committed_at: DateTime<Utc>,
pub heads: Vec<TransactionId>,
pub provenance: Provenance,
pub merge_pairs: Vec<MergePair>,
pub objects: Vec<ObjectDeclaration>,
pub creates: Vec<NodeOperation>,
pub updates: Vec<NodeOperation>,
}
#[derive(Clone, Debug)]
pub(crate) struct ParsedTransaction {
pub id: TransactionId,
pub unsigned: UnsignedTransaction,
}
pub(crate) fn object_hash(bytes: &[u8]) -> [u8; 32] {
Sha256::digest(bytes).into()
}
pub(crate) fn build_signed(
transaction: &UnsignedTransaction,
signing_key: &SigningKey,
) -> Result<Vec<u8>> {
validate_unsigned(transaction)?;
if transaction.writer != WriterId(signing_key.verifying_key().to_bytes()) {
return Err(Error::invalid_input(
"transaction writer does not match signing key",
));
}
let unsigned_length = unsigned_encoded_length(transaction)?;
let signed_length = unsigned_length
.checked_add(SIGNATURE_BYTES)
.ok_or_else(|| Error::invalid_input("canonical transaction length overflow"))?;
if signed_length > MAX_TRANSACTION_BYTES {
return Err(Error::invalid_input("canonical transaction exceeds 64 MiB"));
}
let mut bytes = Vec::new();
bytes
.try_reserve_exact(signed_length)
.map_err(|_| Error::invalid_input("cannot allocate canonical transaction"))?;
encode_unsigned(&mut bytes, transaction)?;
if bytes.len() != unsigned_length {
return Err(Error::corrupt(
"canonical transaction length calculation mismatch",
));
}
let signature = signing_key.sign(&bytes);
bytes.extend_from_slice(&signature.to_bytes());
Ok(bytes)
}
pub(crate) fn parse_signed(bytes: &[u8]) -> Result<ParsedTransaction> {
parse_signed_inner(bytes, None)
}
pub(crate) fn parse_signed_authorized(
bytes: &[u8],
writers: &[WriterId],
) -> Result<ParsedTransaction> {
parse_signed_inner(bytes, Some(writers))
}
fn parse_signed_inner(bytes: &[u8], writers: Option<&[WriterId]>) -> Result<ParsedTransaction> {
if bytes.len() > MAX_TRANSACTION_BYTES || bytes.len() < MAGIC.len() + 32 + SIGNATURE_BYTES {
return Err(Error::invalid_transaction(
"signed transaction has an invalid length",
));
}
if &bytes[..MAGIC.len()] != MAGIC {
return Err(Error::invalid_transaction(
"unsupported canonical transaction format; v2 is required",
));
}
let unsigned_length = bytes.len() - SIGNATURE_BYTES;
let writer = WriterId(
bytes[MAGIC.len()..MAGIC.len() + 32]
.try_into()
.map_err(|_| Error::invalid_transaction("writer field is truncated"))?,
);
if writers.is_some_and(|writers| !writers.contains(&writer)) {
return Err(Error::invalid_transaction(format!(
"writer {writer} is not configured"
)));
}
let verifying_key = VerifyingKey::from_bytes(&writer.0)
.map_err(|_| Error::invalid_transaction("writer is not a valid Ed25519 key"))?;
let signature = Signature::from_bytes(
bytes[unsigned_length..]
.try_into()
.map_err(|_| Error::invalid_transaction("invalid signature length"))?,
);
verifying_key
.verify_strict(&bytes[..unsigned_length], &signature)
.map_err(|_| Error::invalid_transaction("Ed25519 signature verification failed"))?;
let mut reader = Reader::new(&bytes[..unsigned_length]);
let unsigned = reader.unsigned()?;
if !reader.remaining().is_empty() {
return Err(Error::invalid_transaction(
"trailing canonical transaction bytes",
));
}
validate_unsigned(&unsigned).map_err(|error| Error::invalid_transaction(error.to_string()))?;
if unsigned.writer != writer {
return Err(Error::invalid_transaction(
"decoded writer does not match the authenticated envelope",
));
}
Ok(ParsedTransaction {
id: TransactionId::for_signed_bytes(bytes),
unsigned,
})
}
fn validate_unsigned(transaction: &UnsignedTransaction) -> Result<()> {
transaction.provenance.validate()?;
ensure_sorted_unique(&transaction.heads, "heads")?;
ensure_sorted_unique(&transaction.merge_pairs, "merge pairs")?;
ensure_sorted_by(
&transaction.objects,
|value| value.id,
"object declarations",
)?;
ensure_sorted_by(&transaction.creates, |value| value.id, "create operations")?;
ensure_sorted_by(&transaction.updates, |value| value.id, "update operations")?;
if sorted_operations_overlap(&transaction.creates, &transaction.updates) {
return Err(Error::invalid_input(
"one transaction cannot create and update the same node",
));
}
for pair in &transaction.merge_pairs {
if pair.first >= pair.second {
return Err(Error::invalid_input("merge pairs are not canonical"));
}
}
let mut total = 0_u64;
for object in &transaction.objects {
if !object.id.valid_domain() || object.length > MAX_OBJECT_BYTES {
return Err(Error::invalid_input(
"object declaration exceeds 32 GiB or has wrong domain",
));
}
total = total
.checked_add(object.length)
.ok_or_else(|| Error::invalid_input("object declaration length overflow"))?;
if total > MAX_TRANSACTION_OBJECT_BYTES {
return Err(Error::invalid_input(
"transaction object payload total exceeds 32 GiB",
));
}
}
for operation in transaction.creates.iter().chain(&transaction.updates) {
if !operation.id.valid_domain() {
return Err(Error::invalid_input(
"node operation has wrong locator domain",
));
}
operation.data.validate()?;
}
Ok(())
}
fn sorted_operations_overlap(left: &[NodeOperation], right: &[NodeOperation]) -> bool {
let (mut left_index, mut right_index) = (0, 0);
while left_index < left.len() && right_index < right.len() {
match left[left_index].id.cmp(&right[right_index].id) {
std::cmp::Ordering::Less => left_index += 1,
std::cmp::Ordering::Greater => right_index += 1,
std::cmp::Ordering::Equal => return true,
}
}
false
}
fn ensure_sorted_unique<T: Ord>(values: &[T], name: &str) -> Result<()> {
if values.windows(2).any(|pair| pair[0] >= pair[1]) {
return Err(Error::invalid_input(format!(
"{name} must be sorted and unique"
)));
}
Ok(())
}
fn ensure_sorted_by<T, K: Ord + Copy>(
values: &[T],
key: impl Fn(&T) -> K,
name: &str,
) -> Result<()> {
if values.windows(2).any(|pair| key(&pair[0]) >= key(&pair[1])) {
return Err(Error::invalid_input(format!(
"{name} must be sorted and unique"
)));
}
Ok(())
}
fn unsigned_encoded_length(transaction: &UnsignedTransaction) -> Result<usize> {
let mut length = MAGIC.len() + 32 + 8 + 4;
add_vector_length(&mut length, transaction.heads.len(), 32)?;
add_string_length(&mut length, &transaction.provenance.author)?;
add_string_length(&mut length, &transaction.provenance.source)?;
add_length(&mut length, 8 + 4)?;
add_string_length(&mut length, &transaction.provenance.data)?;
add_vector_length(&mut length, transaction.merge_pairs.len(), 64)?;
add_vector_length(&mut length, transaction.objects.len(), 6 + 8 + 32)?;
add_length(&mut length, 8)?;
for operation in &transaction.creates {
add_length(&mut length, node_operation_encoded_length(operation)?)?;
}
add_length(&mut length, 8)?;
for operation in &transaction.updates {
add_length(&mut length, node_operation_encoded_length(operation)?)?;
}
Ok(length)
}
fn node_operation_encoded_length(operation: &NodeOperation) -> Result<usize> {
let mut length = 6;
add_string_length(&mut length, &operation.data.short_name)?;
add_string_length(&mut length, &operation.data.short_description)?;
add_string_length(&mut length, &operation.data.long_description)?;
add_length(
&mut length,
match operation.data.owner {
Owner::Node(_) => 1 + 6,
Owner::Unowned | Owner::SelfNode => 1,
},
)?;
add_vector_length(&mut length, operation.data.fixed_connections.len(), 6)?;
add_vector_length(&mut length, operation.data.recent_connections.len(), 6)?;
add_vector_length(&mut length, operation.data.objects.len(), 6)?;
Ok(length)
}
fn add_vector_length(length: &mut usize, count: usize, item_bytes: usize) -> Result<()> {
let payload = count
.checked_mul(item_bytes)
.ok_or_else(|| Error::invalid_input("canonical vector length overflow"))?;
add_length(length, 8)?;
add_length(length, payload)
}
fn add_string_length(length: &mut usize, value: &str) -> Result<()> {
u32::try_from(value.len())
.map_err(|_| Error::invalid_input("canonical string length does not fit u32"))?;
add_length(length, 4)?;
add_length(length, value.len())
}
fn add_length(length: &mut usize, addition: usize) -> Result<()> {
*length = length
.checked_add(addition)
.ok_or_else(|| Error::invalid_input("canonical transaction length overflow"))?;
Ok(())
}
fn encode_unsigned(output: &mut Vec<u8>, transaction: &UnsignedTransaction) -> Result<()> {
output.extend_from_slice(MAGIC);
output.extend_from_slice(&transaction.writer.0);
output.extend_from_slice(&transaction.committed_at.timestamp().to_be_bytes());
output.extend_from_slice(
&transaction
.committed_at
.timestamp_subsec_nanos()
.to_be_bytes(),
);
vector(output, &transaction.heads, |output, value| {
output.extend_from_slice(&value.0)
});
string(output, &transaction.provenance.author)?;
string(output, &transaction.provenance.source)?;
output.extend_from_slice(
&transaction
.provenance
.source_created_at
.timestamp()
.to_be_bytes(),
);
output.extend_from_slice(
&transaction
.provenance
.source_created_at
.timestamp_subsec_nanos()
.to_be_bytes(),
);
string(output, &transaction.provenance.data)?;
vector(output, &transaction.merge_pairs, |output, pair| {
output.extend_from_slice(&pair.first.0);
output.extend_from_slice(&pair.second.0);
});
vector(output, &transaction.objects, |output, object| {
output.extend_from_slice(&object.id.0);
output.extend_from_slice(&object.length.to_be_bytes());
output.extend_from_slice(&object.sha256);
});
output.extend_from_slice(&(transaction.creates.len() as u64).to_be_bytes());
for operation in &transaction.creates {
node_operation(output, operation)?;
}
output.extend_from_slice(&(transaction.updates.len() as u64).to_be_bytes());
for operation in &transaction.updates {
node_operation(output, operation)?;
}
Ok(())
}
fn vector<T>(output: &mut Vec<u8>, values: &[T], mut write: impl FnMut(&mut Vec<u8>, &T)) {
output.extend_from_slice(&(values.len() as u64).to_be_bytes());
for value in values {
write(output, value);
}
}
fn string(output: &mut Vec<u8>, value: &str) -> Result<()> {
let length = u32::try_from(value.len())
.map_err(|_| Error::invalid_input("canonical string length does not fit u32"))?;
output.extend_from_slice(&length.to_be_bytes());
output.extend_from_slice(value.as_bytes());
Ok(())
}
fn node_operation(output: &mut Vec<u8>, operation: &NodeOperation) -> Result<()> {
output.extend_from_slice(&operation.id.0);
node_data(output, &operation.data)
}
fn node_data(output: &mut Vec<u8>, data: &NodeData) -> Result<()> {
string(output, &data.short_name)?;
string(output, &data.short_description)?;
string(output, &data.long_description)?;
match data.owner {
Owner::Unowned => output.push(0),
Owner::SelfNode => output.push(1),
Owner::Node(id) => {
output.push(2);
output.extend_from_slice(&id.0);
}
}
vector(output, &data.fixed_connections, |output, id| {
output.extend_from_slice(&id.0)
});
vector(output, &data.recent_connections, |output, id| {
output.extend_from_slice(&id.0)
});
vector(output, &data.objects, |output, id| {
output.extend_from_slice(&id.0)
});
Ok(())
}
struct Reader<'a> {
input: &'a [u8],
offset: usize,
}
impl<'a> Reader<'a> {
fn new(input: &'a [u8]) -> Self {
Self { input, offset: 0 }
}
fn remaining(&self) -> &'a [u8] {
&self.input[self.offset..]
}
fn take(&mut self, length: usize) -> Result<&'a [u8]> {
let end = self
.offset
.checked_add(length)
.ok_or_else(|| Error::invalid_transaction("canonical length overflow"))?;
if end > self.input.len() {
return Err(Error::invalid_transaction(
"truncated canonical transaction",
));
}
let bytes = &self.input[self.offset..end];
self.offset = end;
Ok(bytes)
}
fn array<const N: usize>(&mut self) -> Result<[u8; N]> {
self.take(N)?
.try_into()
.map_err(|_| Error::invalid_transaction("invalid fixed-width field"))
}
fn u8(&mut self) -> Result<u8> {
Ok(self.array::<1>()?[0])
}
fn u32(&mut self) -> Result<u32> {
Ok(u32::from_be_bytes(self.array()?))
}
fn u64(&mut self) -> Result<u64> {
Ok(u64::from_be_bytes(self.array()?))
}
fn i64(&mut self) -> Result<i64> {
Ok(i64::from_be_bytes(self.array()?))
}
fn datetime(&mut self) -> Result<DateTime<Utc>> {
DateTime::from_timestamp(self.i64()?, self.u32()?)
.ok_or_else(|| Error::invalid_transaction("timestamp is outside supported range"))
}
fn string(&mut self) -> Result<String> {
let length = self.u32()? as usize;
if length > self.remaining().len() {
return Err(Error::invalid_transaction(
"canonical string has invalid length",
));
}
str::from_utf8(self.take(length)?)
.map(str::to_owned)
.map_err(|_| Error::invalid_transaction("canonical string is not UTF-8"))
}
fn count(&mut self, minimum_item_bytes: usize) -> Result<usize> {
let count = usize::try_from(self.u64()?).map_err(|_| {
Error::invalid_transaction("canonical vector length does not fit usize")
})?;
let minimum = count
.checked_mul(minimum_item_bytes)
.ok_or_else(|| Error::invalid_transaction("canonical vector length overflow"))?;
if minimum > self.remaining().len() {
return Err(Error::invalid_transaction(
"canonical vector cannot fit inside its enclosing record",
));
}
Ok(count)
}
fn unsigned(&mut self) -> Result<UnsignedTransaction> {
if self.take(MAGIC.len())? != MAGIC {
return Err(Error::invalid_transaction(
"unsupported transaction version",
));
}
let writer = WriterId(self.array()?);
let committed_at = self.datetime()?;
let head_count = self.count(32)?;
let mut heads = Vec::with_capacity(head_count);
for _ in 0..head_count {
heads.push(TransactionId(self.array()?));
}
let provenance = Provenance {
author: self.string()?,
source: self.string()?,
source_created_at: self.datetime()?,
data: self.string()?,
};
let merge_count = self.count(64)?;
let mut merge_pairs = Vec::with_capacity(merge_count);
for _ in 0..merge_count {
merge_pairs.push(MergePair {
first: TransactionId(self.array()?),
second: TransactionId(self.array()?),
});
}
let object_count = self.count(46)?;
let mut objects = Vec::with_capacity(object_count);
for _ in 0..object_count {
objects.push(ObjectDeclaration {
id: ObjectId(self.array()?),
length: self.u64()?,
sha256: self.array()?,
});
}
let create_count = self.count(43)?;
let mut creates = Vec::new();
for _ in 0..create_count {
creates.push(self.node_operation()?);
}
let update_count = self.count(43)?;
let mut updates = Vec::new();
for _ in 0..update_count {
updates.push(self.node_operation()?);
}
Ok(UnsignedTransaction {
writer,
committed_at,
heads,
provenance,
merge_pairs,
objects,
creates,
updates,
})
}
fn node_operation(&mut self) -> Result<NodeOperation> {
Ok(NodeOperation {
id: NodeId(self.array()?),
data: self.node_data()?,
})
}
fn node_ids(&mut self) -> Result<Vec<NodeId>> {
let count = self.count(6)?;
let mut values = Vec::with_capacity(count);
for _ in 0..count {
values.push(NodeId(self.array()?));
}
Ok(values)
}
fn object_ids(&mut self) -> Result<Vec<ObjectId>> {
let count = self.count(6)?;
let mut values = Vec::with_capacity(count);
for _ in 0..count {
values.push(ObjectId(self.array()?));
}
Ok(values)
}
fn node_data(&mut self) -> Result<NodeData> {
let short_name = self.string()?;
let short_description = self.string()?;
let long_description = self.string()?;
let owner = match self.u8()? {
0 => Owner::Unowned,
1 => Owner::SelfNode,
2 => Owner::Node(NodeId(self.array()?)),
_ => return Err(Error::invalid_transaction("unknown owner tag")),
};
Ok(NodeData {
short_name,
short_description,
long_description,
owner,
fixed_connections: self.node_ids()?,
recent_connections: self.node_ids()?,
objects: self.object_ids()?,
})
}
}
#[cfg(test)]
mod tests {
use super::*;
use chrono::TimeZone;
fn fixture() -> (UnsignedTransaction, SigningKey) {
let signing_key = SigningKey::from_bytes(&[7; 32]);
let timestamp = Utc.timestamp_opt(1_700_000_000, 123).unwrap();
(
UnsignedTransaction {
writer: WriterId(signing_key.verifying_key().to_bytes()),
committed_at: timestamp,
heads: Vec::new(),
provenance: Provenance {
author: "Kennedy".into(),
source: "wire-golden".into(),
source_created_at: timestamp,
data: "fixed".into(),
},
merge_pairs: Vec::new(),
objects: vec![ObjectDeclaration {
id: ObjectId([128, 2, 3, 4, 5, 6]),
length: 3,
sha256: object_hash(b"abc"),
}],
creates: vec![NodeOperation {
id: NodeId([6, 5, 4, 3, 2, 1]),
data: NodeData {
short_name: "Node".into(),
short_description: "short".into(),
long_description: "long description".into(),
owner: Owner::SelfNode,
fixed_connections: Vec::new(),
recent_connections: Vec::new(),
objects: vec![ObjectId([128, 2, 3, 4, 5, 6])],
},
}],
updates: Vec::new(),
},
signing_key,
)
}
#[test]
fn v2_round_trip_is_deterministic() {
let (transaction, signing_key) = fixture();
let first = build_signed(&transaction, &signing_key).unwrap();
assert_eq!(first, build_signed(&transaction, &signing_key).unwrap());
assert_eq!(&first[..8], MAGIC);
assert_eq!(parse_signed(&first).unwrap().unsigned, transaction);
}
#[test]
fn node_text_round_trips_without_database_policy_limits() {
let (mut transaction, signing_key) = fixture();
let data = &mut transaction.creates[0].data;
data.short_name.clear();
data.short_description = "x".repeat(201);
data.long_description = "x".repeat(1024 * 1024 + 1);
let signed = build_signed(&transaction, &signing_key).unwrap();
assert_eq!(parse_signed(&signed).unwrap().unsigned, transaction);
}
#[test]
fn v1_and_declared_vector_overflow_fail_closed() {
let (transaction, signing_key) = fixture();
let mut signed = build_signed(&transaction, &signing_key).unwrap();
signed[5] = 1;
assert!(parse_signed(&signed).is_err());
let mut signed = build_signed(&transaction, &signing_key).unwrap();
let heads_offset = 8 + 32 + 8 + 4;
signed[heads_offset..heads_offset + 8].copy_from_slice(&u64::MAX.to_be_bytes());
assert!(parse_signed(&signed).is_err());
}
}