use std::collections::{BTreeMap, HashSet};
use std::sync::{Arc, Mutex};
use clawft_types::placement::{Capability, NodeFacts, ProbeNote, Provenance};
use ed25519_dalek::{Signature, Signer, SigningKey, Verifier, VerifyingKey};
use serde::{Deserialize, Serialize};
use serde_json::{Value, json};
use sha2::{Digest, Sha256};
use super::seed::{API_TIMEOUT, SeedApiRuntime};
use super::seed_http::Method;
use super::types::{RuntimeError, WorkloadRuntime};
use crate::chain::{ChainManager, EVENT_KIND_WORKLOAD_NODE_BIND, EVENT_KIND_WORKLOAD_REFUSE};
use crate::node_facts_advert::{NodeFactsAdvertError, SignedNodeFacts, sign_node_facts};
use crate::node_registry::node_id_from_pubkey;
use crate::workload_pkg::codec::hex_encode;
use crate::workload_pkg::manifest::valid_token;
pub const BIND_CHAIN_SOURCE: &str = "workload.bind";
pub const BIND_DOMAIN: &[u8] = b"weftos.workload.node.bind.v1\0";
pub const DEFAULT_MAX_BIND_AGE_SECS: u64 = 600;
pub const BIND_STATE_FILE: &str = "workload-seed-binds.json";
const MAX_STATE_BYTES: u64 = 256 * 1024;
const MAX_BOUND_DEVICES: usize = 1000;
pub const BIND_CLOCK_SKEW_SECS: u64 = 60;
pub const SEED_FACTS_TTL_SECS: u64 = 300;
const MAX_RECORD_BYTES: usize = 4096;
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct BindRecord {
pub device_id: String,
pub device_pubkey: String,
pub node_id: String,
pub bound_at: u64,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct SignedBind {
pub record: String,
pub operator_key: Vec<u8>,
pub signature: Vec<u8>,
}
pub fn seed_node_id(adapter: &SigningKey) -> String {
node_id_from_pubkey(&adapter.verifying_key().to_bytes())
}
fn signed_bytes(record: &str) -> Vec<u8> {
let mut v = BIND_DOMAIN.to_vec();
v.extend_from_slice(record.as_bytes());
v
}
pub fn sign_bind(record: &BindRecord, operator: &SigningKey) -> SignedBind {
let record = serde_json::to_string(record).unwrap_or_default();
let signature = operator.sign(&signed_bytes(&record)).to_bytes().to_vec();
SignedBind {
record,
operator_key: operator.verifying_key().to_bytes().to_vec(),
signature,
}
}
#[non_exhaustive]
#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
pub enum BindError {
#[error("malformed bind record: {0}")]
Malformed(String),
#[error("bind record is not signed by a pinned operator key")]
UntrustedOperator,
#[error("bind record signature does not verify")]
BadSignature,
#[error("bind record is for node {record}, this adapter is node {adapter}")]
WrongNode {
record: String,
adapter: String,
},
#[error("the Seed's identity does not match the bind record ({0})")]
IdentityMismatch(&'static str),
#[error("bind record is outside its validity window")]
Expired,
#[error("bind record replayed or older than the current binding")]
Replayed,
#[error("node {0} is already bound to another device")]
NodeTaken(String),
#[error("{0}")]
UnpinnedTransport(String),
#[cfg(all(feature = "mesh", feature = "ecc", feature = "exochain", feature = "native"))]
#[error("{0}")]
Licence(crate::licence::LicenceError),
#[error("device {device_id} is already bound to another mesh (seed_bound_elsewhere); unbind it first")]
SeedBoundElsewhere {
device_id: String,
},
#[error("bind state not saved: {0}")]
State(String),
#[error("could not read the Seed identity: {0}")]
Seed(String),
}
impl BindError {
pub fn code(&self) -> &'static str {
match self {
Self::Malformed(_) => "malformed",
Self::UntrustedOperator => "untrusted_operator",
Self::BadSignature => "bad_signature",
Self::WrongNode { .. } => "wrong_node",
Self::IdentityMismatch(_) => "identity_mismatch",
Self::Expired => "expired",
Self::Replayed => "replayed",
Self::NodeTaken(_) => "node_taken",
Self::UnpinnedTransport(_) => "unpinned_transport",
#[cfg(all(feature = "mesh", feature = "ecc", feature = "exochain", feature = "native"))]
Self::Licence(e) => v2::licence_code(e),
Self::SeedBoundElsewhere { .. } => "seed_bound_elsewhere",
Self::State(_) => "state_unwritable",
Self::Seed(_) => "seed_unreachable",
}
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Binding {
record: BindRecord,
}
impl Binding {
pub fn record(&self) -> &BindRecord {
&self.record
}
}
pub struct SeedBinder {
operators: Vec<[u8; 32]>,
chain: Arc<ChainManager>,
max_age_secs: u64,
seen: Mutex<HashSet<[u8; 32]>>,
bound: Mutex<BTreeMap<String, (u64, String)>>,
state_file: Option<std::path::PathBuf>,
}
#[derive(Debug, Default, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct BindState {
version: u32,
bound: BTreeMap<String, BoundEntry>,
}
#[derive(Debug, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct BoundEntry {
bound_at: u64,
node_id: String,
}
fn write_state(
path: &std::path::Path,
bound: &BTreeMap<String, (u64, String)>,
) -> Result<(), String> {
use std::io::Write;
use std::os::unix::fs::OpenOptionsExt;
let st = BindState {
version: 1,
bound: bound
.iter()
.map(|(d, (at, n))| {
(
d.clone(),
BoundEntry {
bound_at: *at,
node_id: n.clone(),
},
)
})
.collect(),
};
let bytes = serde_json::to_vec_pretty(&st).map_err(|e| e.to_string())?;
let tmp = path.with_extension("json.tmp");
let mut f = std::fs::OpenOptions::new()
.create(true)
.write(true)
.truncate(true)
.mode(0o600)
.open(&tmp)
.map_err(|e| e.to_string())?;
f.write_all(&bytes).map_err(|e| e.to_string())?;
f.sync_all().map_err(|e| e.to_string())?;
std::fs::rename(&tmp, path).map_err(|e| e.to_string())
}
impl SeedBinder {
pub fn new(operators: Vec<[u8; 32]>, chain: Arc<ChainManager>) -> Self {
Self {
operators,
chain,
max_age_secs: DEFAULT_MAX_BIND_AGE_SECS,
seen: Mutex::new(HashSet::new()),
bound: Mutex::new(BTreeMap::new()),
state_file: None,
}
}
pub fn with_state_file(
mut self,
path: impl Into<std::path::PathBuf>,
) -> Result<Self, BindError> {
let path = path.into();
let bad = |m: String| BindError::State(format!("{}: {m}", path.display()));
if let Ok(meta) = std::fs::symlink_metadata(&path) {
if !meta.file_type().is_file() || meta.len() > MAX_STATE_BYTES {
return Err(bad("not a regular file of at most 256 KiB".into()));
}
let text = std::fs::read_to_string(&path).map_err(|e| bad(e.to_string()))?;
let st: BindState = serde_json::from_str(&text).map_err(|e| bad(e.to_string()))?;
if st.version != 1 || st.bound.len() > MAX_BOUND_DEVICES {
return Err(bad("unsupported version or too many devices".into()));
}
if let Ok(mut b) = self.bound.lock() {
for (d, e) in st.bound {
b.insert(d, (e.bound_at, e.node_id));
}
}
}
self.state_file = Some(path);
Ok(self)
}
pub fn with_max_age_secs(mut self, secs: u64) -> Self {
self.max_age_secs = secs;
self
}
pub fn device_of(&self, node_id: &str) -> Option<String> {
self.bound
.lock()
.ok()?
.iter()
.find(|(_, (_, n))| n == node_id)
.map(|(d, _)| d.clone())
}
fn refuse(&self, signed: &SignedBind, e: &BindError) {
self.refuse_record(&signed.record, e);
}
fn refuse_record(&self, record: &str, e: &BindError) {
let record_hash = hex_encode(&Sha256::digest(record.as_bytes()));
let claimed = serde_json::from_str::<Value>(record).unwrap_or(Value::Null);
let clip = |k: &str| {
claimed
.get(k)
.and_then(Value::as_str)
.map(|s| s.chars().take(64).collect::<String>())
};
self.chain.append(
BIND_CHAIN_SOURCE,
EVENT_KIND_WORKLOAD_REFUSE,
Some(json!({
"phase": "node.bind", "code": e.code(), "reason": e.to_string(),
"record_hash": record_hash,
"claimed_node_id": clip("node_id"), "claimed_device_id": clip("device_id"),
})),
);
}
pub async fn bind(
&self,
signed: &SignedBind,
rt: &SeedApiRuntime,
now: u64,
) -> Result<Binding, BindError> {
match self.verify(signed, rt, now).await {
Ok(b) => Ok(b),
Err(e) => {
self.refuse(signed, &e);
Err(e)
}
}
}
async fn verify(
&self,
signed: &SignedBind,
rt: &SeedApiRuntime,
now: u64,
) -> Result<Binding, BindError> {
let bad = |m: &str| BindError::Malformed(m.into());
rt.link_security()
.require_pinned(rt.node_id())
.map_err(BindError::UnpinnedTransport)?;
if signed.record.len() > MAX_RECORD_BYTES {
return Err(bad("record too large"));
}
let key: [u8; 32] = signed
.operator_key
.as_slice()
.try_into()
.map_err(|_| bad("operator key is not 32 bytes"))?;
let sig: [u8; 64] = signed
.signature
.as_slice()
.try_into()
.map_err(|_| bad("signature is not 64 bytes"))?;
if !self.operators.contains(&key) {
return Err(BindError::UntrustedOperator);
}
let vk = VerifyingKey::from_bytes(&key).map_err(|_| bad("bad operator key"))?;
vk.verify(&signed_bytes(&signed.record), &Signature::from_bytes(&sig))
.map_err(|_| BindError::BadSignature)?;
let record: BindRecord = serde_json::from_str(&signed.record)
.map_err(|e| BindError::Malformed(e.to_string()))?;
if !valid_token(&record.device_id, 128)
|| !valid_token(&record.node_id, 64)
|| record.device_pubkey.is_empty()
|| record.device_pubkey.len() > 256
{
return Err(bad("device id, device key or node id is not well formed"));
}
let adapter = rt.node_id().to_string();
if record.node_id != adapter {
return Err(BindError::WrongNode {
record: record.node_id,
adapter,
});
}
if record.bound_at > now.saturating_add(BIND_CLOCK_SKEW_SECS)
|| now.saturating_sub(record.bound_at) > self.max_age_secs
{
return Err(BindError::Expired);
}
let (device_id, device_key) = rt
.identity()
.await
.map_err(|e| BindError::Seed(e.to_string()))?;
if device_id != record.device_id {
return Err(BindError::IdentityMismatch("device id"));
}
if device_key != record.device_pubkey {
return Err(BindError::IdentityMismatch("device key"));
}
let digest: [u8; 32] =
Sha256::digest([signed.record.as_bytes(), signed.signature.as_slice()].concat()).into();
let mut seen = self.seen.lock().map_err(|_| bad("binder poisoned"))?;
let mut bound = self.bound.lock().map_err(|_| bad("binder poisoned"))?;
if seen.contains(&digest) {
return Err(BindError::Replayed);
}
if bound
.get(&record.device_id)
.is_some_and(|(at, _)| record.bound_at <= *at)
{
return Err(BindError::Replayed);
}
if bound
.iter()
.any(|(d, (_, n))| *n == record.node_id && *d != record.device_id)
{
return Err(BindError::NodeTaken(record.node_id));
}
if !bound.contains_key(&record.device_id) && bound.len() >= MAX_BOUND_DEVICES {
return Err(BindError::State("too many bound devices".into()));
}
let mut next = bound.clone();
next.insert(
record.device_id.clone(),
(record.bound_at, record.node_id.clone()),
);
if let Some(path) = &self.state_file {
write_state(path, &next).map_err(BindError::State)?;
}
*bound = next;
seen.insert(digest);
self.chain.append(
BIND_CHAIN_SOURCE,
EVENT_KIND_WORKLOAD_NODE_BIND,
Some(json!({
"device_id": record.device_id, "device_pubkey": record.device_pubkey,
"node_id": record.node_id, "bound_at": record.bound_at,
"operator_key": hex_encode(&key),
"record_hash": hex_encode(&Sha256::digest(signed.record.as_bytes())),
})),
);
Ok(Binding { record })
}
}
#[cfg(all(feature = "mesh", feature = "ecc", feature = "exochain", feature = "native"))]
mod v2;
#[cfg(all(feature = "mesh", feature = "ecc", feature = "exochain", feature = "native"))]
pub use v2::{StewardBind, UnbindOutcome, grant_fingerprint};
impl SeedApiRuntime {
pub fn node_id(&self) -> &str {
&self.cfg.node_id
}
pub async fn identity(&self) -> Result<(String, String), RuntimeError> {
let v = self
.api(Method::Get, "/api/v1/identity", None, API_TIMEOUT)
.await?;
let field = |k: &str| {
v.get(k)
.and_then(Value::as_str)
.filter(|s| !s.is_empty() && s.len() <= 256)
.map(str::to_string)
.ok_or_else(|| RuntimeError::Backend(format!("seed /api/v1/identity: no {k}")))
};
Ok((field("device_id")?, field("public_key")?))
}
}
pub fn attest_seed_facts(
binding: &Binding,
rt: &SeedApiRuntime,
adapter: &SigningKey,
now: u64,
seq: u64,
) -> Result<SignedNodeFacts, NodeFactsAdvertError> {
let node_id = seed_node_id(adapter);
if node_id != binding.record.node_id {
return Err(NodeFactsAdvertError::NodeMismatch {
claimed: binding.record.node_id.clone(),
derived: node_id,
});
}
let mut facts = NodeFacts::new(node_id, now, SEED_FACTS_TTL_SECS, seq);
facts.capabilities = rt
.provides()
.into_iter()
.map(|c: Capability| Capability {
provenance: Provenance::Claimed,
..c
})
.collect();
facts.notes.push(ProbeNote::new(
"remote.api",
format!(
"claimed by the adapter for Seed device {} (bound by an operator record), not probed",
binding.record.device_id
),
));
sign_node_facts(&facts, adapter)
}