daemon.rs 93.0%

213 of 229 lines run, 33 of 92 functions · run #35 on main · run missed

//! The daemon's state (spec §7.1): the ledger it alone writes, the
//! principals kept current from the bus, the authenticator, and presence
//! with its timers. HTTP lives elsewhere (`spectra-mcp`, `radar`); this is
//! what they share.

use crate::authn::{Authenticator, Caller, Refusal};
use crate::event::{Auth, Event, Method, NewEvent};
use crate::keys::KeyPair;
use crate::ledger::{KeyResolver, Ledger};
use crate::presence::Presence;
use crate::principals::{Principal, Principals};
use crate::proof::Proof;
use anyhow::{bail, Context, Result};
use serde::{Deserialize, Serialize};
use std::path::{Path, PathBuf};
use std::sync::Arc;
use tokio::sync::{Mutex, RwLock};

/// Kinds the owner's CLI may hand the daemon, already signed over their
/// payload (spec §8.5, "One writer").
pub const OWNER_KINDS: &[&str] = &[
    "identity.rotated",
    "agent.added",
    "agent.disabled",
    "agent.scoped",
    "project.added",
    "project.linked",
    "project.unlinked",
    "brief.set",
];

pub struct Daemon {
    pub data: PathBuf,
    pub ledger: Ledger,
    pub principals: RwLock<Principals>,
    pub authn: Authenticator,
    pub presence: Mutex<Presence>,
    /// Per agent: empty `wait`s in a row, and until when `wait` is refused.
    pub idle_waits: Mutex<std::collections::HashMap<String, (u32, u64)>>,
}

impl Daemon {
    /// Open the data directory's ledger for writing and start the timers.
    pub async fn start(data: &Path) -> Result<Arc<Daemon>> {
        let db = data.join("spectra.db");
        if !db.exists() {
            bail!("no Spectra in {} (run `spectra init`)", data.display());
        }
        let instance = KeyPair::instance(data, "spectra")?;
        let ledger = Ledger::open(&db, Some(instance)).await?;
        let principals = Principals::load(&ledger).await?;
        let d = Arc::new(Daemon {
            data: data.to_path_buf(),
            ledger,
            principals: RwLock::new(principals),
            authn: Authenticator::default(),
            presence: Mutex::new(Presence::default()),
            idle_waits: Mutex::new(Default::default()),
        });
        d.clone().watch_principals();
        d.clone().tick_presence();
        Ok(d)
    }

    /// Reload who is who whenever an identity or agent event is appended.
    fn watch_principals(self: Arc<Self>) {
        let mut rx = self.ledger.subscribe();
        tokio::spawn(async move {
            loop {
                match rx.recv().await {
                    Ok(e)
                        if e.kind.starts_with("identity.")
                            || e.kind.starts_with("agent.") && e.kind != "agent.presence" && e.kind != "agent.status" =>
                    {
                        if let Ok(p) = Principals::load(&self.ledger).await {
                            *self.principals.write().await = p;
                        }
                    }
                    Ok(_) => {}
                    Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => {
                        if let Ok(p) = Principals::load(&self.ledger).await {
                            *self.principals.write().await = p;
                        }
                    }
                    Err(_) => return,
                }
            }
        });
    }

    /// Record presence changes and quiet sessions every 15 s, and write a
    /// checkpoint when one is due by time.
    fn tick_presence(self: Arc<Self>) {
        tokio::spawn(async move {
            let mut every = tokio::time::interval(std::time::Duration::from_secs(15));
            loop {
                every.tick().await;
                let events = self.presence.lock().await.tick(crate::ids::now_ms());
                if !events.is_empty() {
                    let _ = self.ledger.append(events).await;
                }
                let _ = self.ledger.append(vec![]).await;
            }
        });
    }

    /// Authenticate a request (spec §8.5); an agent used from the wrong Unix
    /// user is recorded as `guard.impersonation`, and a first connection
    /// that binds an agent's uid as `agent.uid_bound`.
    pub async fn authenticate(&self, proof: Option<&Proof>, method: &str, body: &[u8], peer: Option<Peer>) -> Result<Caller, Refusal> {
        let now = crate::ids::now_ms();
        let r = {
            let p = self.principals.read().await;
            if p.owner().is_none() {
                return Err(Refusal::NotAPrincipal("this Spectra has no owner yet: its owner claims it with `spectra connect`".into()));
            }
            self.authn.authenticate(&p, proof, method, body, now, peer.map(|x| x.uid))
        };
        match &r {
            Err(Refusal::WrongUid { agent, bound, from }) => {
                let e = NewEvent::system(
                    "guard.impersonation",
                    serde_json::json!({ "agent": agent, "bound_uid": bound, "from_uid": from, "from_pid": peer.map(|x| x.pid), "method": method }),
                )
                .subject(agent.clone());
                let _ = self.ledger.append_one(e).await;
                tracing::warn!(agent, bound, from, "a key used from the wrong Unix user");
            }
            Err(Refusal::BoundOverNetwork { agent, bound }) => {
                let e = NewEvent::system(
                    "guard.impersonation",
                    serde_json::json!({ "agent": agent, "bound_uid": bound, "from_uid": null, "over": "tcp", "method": method }),
                )
                .subject(agent.clone());
                let _ = self.ledger.append_one(e).await;
                tracing::warn!(agent, bound, "a uid-bound key used over the network");
            }
            Ok(Caller { principal: Principal::Agent { name, .. }, learn_uid: Some(uid), .. }) => {
                let e = NewEvent::system("agent.uid_bound", serde_json::json!({ "name": name, "uid": uid })).subject(name.clone());
                let _ = self.ledger.append_one(e).await;
            }
            _ => {}
        }
        r
    }

    /// Note a request from an agent: its session, and the event that
    /// starts one when this is the first in ten minutes.
    pub async fn touch(&self, who: &Principal, new_session: bool) -> Option<String> {
        let Principal::Agent { name, .. } = who else { return None };
        let (id, started) = self.presence.lock().await.request(name, crate::ids::now_ms(), new_session);
        if let Some(e) = started {
            let _ = self.ledger.append_one(e).await;
        }
        Some(id)
    }

    /// Append events the owner's CLI signed over their payload, after
    /// checking each as `spectra admin verify` would.
    pub async fn accept_owner_events(&self, events: Vec<WireEvent>) -> Result<Vec<u64>> {
        let mut principals = self.principals.read().await.clone();
        let projects: Vec<String> = events
            .iter()
            .filter(|e| e.kind == "project.added")
            .filter_map(|e| e.payload.get("name").and_then(|n| n.as_str()).map(String::from))
            .collect();
        let mut out = Vec::new();
        for w in events {
            let n = w.into_new_event()?;
            if n.kind == "channel.created" {
                // A project's channel comes with its project, unsigned, as `project add` makes it.
                let p = &n.payload;
                let ok = p.get("kind").and_then(|k| k.as_str()) == Some("project")
                    && p.get("name").and_then(|x| x.as_str()).is_some_and(|x| projects.iter().any(|q| q == x));
                if !ok || n.auth.method != Method::System {
                    bail!("a channel is only made with its project");
                }
                out.push(n);
                continue;
            }
            if !OWNER_KINDS.contains(&n.kind.as_str()) {
                bail!("{} cannot be handed to the daemon", n.kind);
            }
            if n.actor != "human" {
                bail!("only the owner hands events to the daemon");
            }
            if n.kind == "brief.set" {
                crate::rules::check(n.payload.get("text").and_then(|t| t.as_str()).unwrap_or(""))?;
            }
            // Check it as verify will, against who is who including the
            // batch so far.
            let probe = probe_event(&n);
            principals.check(&probe).map_err(|why| anyhow::anyhow!("refused {}: {why}", n.kind))?;
            if n.kind != "identity.rotated" {
                let key = crate::identity::KeyId(n.auth.key.context("unsigned")?);
                if !matches!(principals.principal(&key, crate::ids::now_ms()), Ok(Principal::Owner { .. })) {
                    bail!("refused {}: not signed by the owner", n.kind);
                }
            }
            out.push(n);
        }
        let appended = self.ledger.append(out).await?;
        let _ = self.ledger.checkpoint().await;
        // Who is who has changed: know it before the next request arrives.
        *self.principals.write().await = Principals::load(&self.ledger).await?;
        Ok(appended.iter().map(|e| e.seq).collect())
    }

    /// The owner `spectra init --owner` named, while none is recorded.
    pub fn named_owner(&self) -> Option<String> {
        let text = std::fs::read_to_string(self.data.join("spectra.toml")).ok()?;
        let t: toml::Table = text.parse().ok()?;
        t.get("owner")?.get("key")?.as_str().map(|k| k.to_ascii_lowercase())
    }

    /// The owner claims a Spectra set up with `spectra init --owner`
    /// (spec §8.6): the named identity, signed by itself over its payload,
    /// then any successors up to the owner's current identity, each a
    /// rotation as `identity refresh` makes. Accepted only while no owner
    /// is recorded, and only from the key init named.
    pub async fn claim(&self, events: Vec<WireEvent>) -> Result<Vec<u64>> {
        let mut principals = self.principals.read().await.clone();
        if principals.owner().is_some() {
            bail!("this Spectra already has an owner");
        }
        let named = self.named_owner().context("this Spectra names no owner (spectra init --owner)")?;
        let mut out = Vec::new();
        for (i, w) in events.into_iter().enumerate() {
            let n = w.into_new_event()?;
            let want = if i == 0 { "identity.owner" } else { "identity.rotated" };
            if n.kind != want || n.actor != "human" {
                bail!("a claim is the owner's identity.owner, then their rotations");
            }
            if i == 0 {
                let key = n.payload.get("key").and_then(|k| k.as_str()).unwrap_or("").to_ascii_lowercase();
                if key != named || n.auth.key.map(hex::encode) != Some(named.clone()) {
                    bail!("this Spectra was set up for another owner");
                }
            }
            principals.check(&probe_event(&n)).map_err(|why| anyhow::anyhow!("refused the claim: {why}"))?;
            out.push(n);
        }
        if out.is_empty() {
            bail!("an empty claim");
        }
        let appended = self.ledger.append(out).await?;
        let _ = self.ledger.checkpoint().await;
        *self.principals.write().await = Principals::load(&self.ledger).await?;
        Ok(appended.iter().map(|e| e.seq).collect())
    }

    /// Write a checkpoint covering everything (at shutdown).
    pub async fn shutdown(&self) {
        let _ = self.ledger.checkpoint().await;
    }
}

/// The Unix peer of a socket connection.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct Peer {
    pub uid: u32,
    pub pid: Option<i32>,
}

/// An event as the CLI hands it to the daemon: signed already.
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct WireEvent {
    pub kind: String,
    pub actor: String,
    pub subject: Option<String>,
    pub project: Option<String>,
    pub payload: serde_json::Value,
    /// Hex of the auth key and the encoded proof; absent for a project's channel.
    pub key: Option<String>,
    pub proof: Option<String>,
}

impl WireEvent {
    pub fn from_new(n: &NewEvent) -> WireEvent {
        WireEvent {
            kind: n.kind.clone(),
            actor: n.actor.clone(),
            subject: n.subject.clone(),
            project: n.project.clone(),
            payload: n.payload.clone(),
            key: n.auth.key.map(hex::encode),
            proof: n.auth.proof.as_ref().map(hex::encode),
        }
    }

    fn into_new_event(self) -> Result<NewEvent> {
        let auth = match (&self.key, &self.proof) {
            (Some(k), Some(p)) => Auth {
                key: Some(hex::decode(k).ok().and_then(|b| b.try_into().ok()).context("a key is 32 bytes of hex")?),
                method: Method::Signature,
                assurance: 2,
                proof: Some(hex::decode(p).context("a proof is hex")?),
            },
            _ => Auth::system(),
        };
        let mut n = NewEvent::system(&self.kind, self.payload).by(self.actor, auth);
        n.subject = self.subject;
        n.project = self.project;
        Ok(n)
    }
}

/// The event `n` would be, for checking before it is appended.
fn probe_event(n: &NewEvent) -> Event {
    let now = crate::ids::now_ms();
    Event {
        seq: 0,
        id: String::new(),
        at_ms: now,
        hlc: crate::hlc::Hlc { wall_ms: now, logical: 0 },
        prev: [0; 32],
        hash: [0; 32],
        auth: n.auth.clone(),
        kind: n.kind.clone(),
        actor: n.actor.clone(),
        session: n.session.clone(),
        project: n.project.clone(),
        subject: n.subject.clone(),
        payload: Some(crate::event::canonical_json(&n.payload)),
        trace: n.trace.clone(),
    }
}

`/`	Search
`g` `c`	Code
`g` `h`	History
`g` `b`	Bookmarks
`g` `t`	Trust
`y`	Permalink to this state (files and trees)
`?`	This sheet

1		//! The daemon's state (spec §7.1): the ledger it alone writes, the
2		//! principals kept current from the bus, the authenticator, and presence
3		//! with its timers. HTTP lives elsewhere (`spectra-mcp`, `radar`); this is
4		//! what they share.
5
6		use crate::authn::{Authenticator, Caller, Refusal};
7		use crate::event::{Auth, Event, Method, NewEvent};
8		use crate::keys::KeyPair;
9		use crate::ledger::{KeyResolver, Ledger};
10		use crate::presence::Presence;
11		use crate::principals::{Principal, Principals};
12		use crate::proof::Proof;
13		use anyhow::{bail, Context, Result};
14		use serde::{Deserialize, Serialize};
15		use std::path::{Path, PathBuf};
16		use std::sync::Arc;
17		use tokio::sync::{Mutex, RwLock};
18
19		/// Kinds the owner's CLI may hand the daemon, already signed over their
20		/// payload (spec §8.5, "One writer").
21		pub const OWNER_KINDS: &[&str] = &[
22		"identity.rotated",
23		"agent.added",
24		"agent.disabled",
25		"agent.scoped",
26		"project.added",
27		"project.linked",
28		"project.unlinked",
29		"brief.set",
30		];
31
32		pub struct Daemon {
33		pub data: PathBuf,
34		pub ledger: Ledger,
35		pub principals: RwLock<Principals>,
36		pub authn: Authenticator,
37		pub presence: Mutex<Presence>,
38		/// Per agent: empty `wait`s in a row, and until when `wait` is refused.
39		pub idle_waits: Mutex<std::collections::HashMap<String, (u32, u64)>>,
40		}
41
42		impl Daemon {
43		/// Open the data directory's ledger for writing and start the timers.
44	22×	pub async fn start(data: &Path) -> Result<Arc<Daemon>> {
45	22×	let db = data.join("spectra.db");
46	22×	if !db.exists() {
47	0×	bail!("no Spectra in {} (run `spectra init`)", data.display());
48	22×	}
49	22×	let instance = KeyPair::instance(data, "spectra")?;
50	22×	let ledger = Ledger::open(&db, Some(instance)).await?;
51	22×	let principals = Principals::load(&ledger).await?;
52	22×	let d = Arc::new(Daemon {
53	22×	data: data.to_path_buf(),
54	22×	ledger,
55	22×	principals: RwLock::new(principals),
56	22×	authn: Authenticator::default(),
57	22×	presence: Mutex::new(Presence::default()),
58	22×	idle_waits: Mutex::new(Default::default()),
59	22×	});
60	22×	d.clone().watch_principals();
61	22×	d.clone().tick_presence();
62	22×	Ok(d)
63	22×	}
64
65		/// Reload who is who whenever an identity or agent event is appended.
66	22×	fn watch_principals(self: Arc<Self>) {
67	22×	let mut rx = self.ledger.subscribe();
68	22×	tokio::spawn(async move {
69		loop {
70	177×	match rx.recv().await {
71	37×	Ok(e)
72	157×	if e.kind.starts_with("identity.")
73	148×	\|\| e.kind.starts_with("agent.") && e.kind != "agent.presence" && e.kind != "agent.status" =>
74		{
75	37×	if let Ok(p) = Principals::load(&self.ledger).await {
76	35×	*self.principals.write().await = p;
77	0×	}
78		}
79	120×	Ok(_) => {}
80		Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => {
81	0×	if let Ok(p) = Principals::load(&self.ledger).await {
82	0×	*self.principals.write().await = p;
83	0×	}
84		}
85	0×	Err(_) => return,
86		}
87		}
88	0×	});
89	22×	}
90
91		/// Record presence changes and quiet sessions every 15 s, and write a
92		/// checkpoint when one is due by time.
93	22×	fn tick_presence(self: Arc<Self>) {
94	22×	tokio::spawn(async move {
95	22×	let mut every = tokio::time::interval(std::time::Duration::from_secs(15));
96		loop {
97	44×	every.tick().await;
98	22×	let events = self.presence.lock().await.tick(crate::ids::now_ms());
99	22×	if !events.is_empty() {
100	0×	let _ = self.ledger.append(events).await;
101	22×	}
102	22×	let _ = self.ledger.append(vec![]).await;
103		}
104		});
105	22×	}
106
107		/// Authenticate a request (spec §8.5); an agent used from the wrong Unix
108		/// user is recorded as `guard.impersonation`, and a first connection
109		/// that binds an agent's uid as `agent.uid_bound`.
110	148×	pub async fn authenticate(&self, proof: Option<&Proof>, method: &str, body: &[u8], peer: Option<Peer>) -> Result<Caller, Refusal> {
111	148×	let now = crate::ids::now_ms();
112	147×	let r = {
113	148×	let p = self.principals.read().await;
114	148×	if p.owner().is_none() {
115	1×	return Err(Refusal::NotAPrincipal("this Spectra has no owner yet: its owner claims it with `spectra connect`".into()));
116	147×	}
117	147×	self.authn.authenticate(&p, proof, method, body, now, peer.map(\|x\| x.uid))
118		};
119	84×	match &r {
120	1×	Err(Refusal::WrongUid { agent, bound, from }) => {
121	1×	let e = NewEvent::system(
122	1×	"guard.impersonation",
123	1×	serde_json::json!({ "agent": agent, "bound_uid": bound, "from_uid": from, "from_pid": peer.map(\|x\| x.pid), "method": method }),
124		)
125	1×	.subject(agent.clone());
126	1×	let _ = self.ledger.append_one(e).await;
127	1×	tracing::warn!(agent, bound, from, "a key used from the wrong Unix user");
128		}
129	1×	Err(Refusal::BoundOverNetwork { agent, bound }) => {
130	1×	let e = NewEvent::system(
131	1×	"guard.impersonation",
132	1×	serde_json::json!({ "agent": agent, "bound_uid": bound, "from_uid": null, "over": "tcp", "method": method }),
133		)
134	1×	.subject(agent.clone());
135	1×	let _ = self.ledger.append_one(e).await;
136	1×	tracing::warn!(agent, bound, "a uid-bound key used over the network");
137		}
138	1×	Ok(Caller { principal: Principal::Agent { name, .. }, learn_uid: Some(uid), .. }) => {
139	1×	let e = NewEvent::system("agent.uid_bound", serde_json::json!({ "name": name, "uid": uid })).subject(name.clone());
140	1×	let _ = self.ledger.append_one(e).await;
141		}
142	144×	_ => {}
143		}
144	147×	r
145	148×	}
146
147		/// Note a request from an agent: its session, and the event that
148		/// starts one when this is the first in ten minutes.
149	62×	pub async fn touch(&self, who: &Principal, new_session: bool) -> Option<String> {
150	62×	let Principal::Agent { name, .. } = who else { return None };
151	62×	let (id, started) = self.presence.lock().await.request(name, crate::ids::now_ms(), new_session);
152	62×	if let Some(e) = started {
153	21×	let _ = self.ledger.append_one(e).await;
154	41×	}
155	62×	Some(id)
156	62×	}
157
158		/// Append events the owner's CLI signed over their payload, after
159		/// checking each as `spectra admin verify` would.
160	12×	pub async fn accept_owner_events(&self, events: Vec<WireEvent>) -> Result<Vec<u64>> {
161	12×	let mut principals = self.principals.read().await.clone();
162	12×	let projects: Vec<String> = events
163	12×	.iter()
164	18×	.filter(\|e\| e.kind == "project.added")
165	12×	.filter_map(\|e\| e.payload.get("name").and_then(\|n\| n.as_str()).map(String::from))
166	12×	.collect();
167	12×	let mut out = Vec::new();
168	18×	for w in events {
169	18×	let n = w.into_new_event()?;
170	18×	if n.kind == "channel.created" {
171		// A project's channel comes with its project, unsigned, as `project add` makes it.
172	2×	let p = &n.payload;
173	2×	let ok = p.get("kind").and_then(\|k\| k.as_str()) == Some("project")
174	2×	&& p.get("name").and_then(\|x\| x.as_str()).is_some_and(\|x\| projects.iter().any(\|q\| q == x));
175	2×	if !ok \|\| n.auth.method != Method::System {
176	0×	bail!("a channel is only made with its project");
177	2×	}
178	2×	out.push(n);
179	2×	continue;
180	16×	}
181	16×	if !OWNER_KINDS.contains(&n.kind.as_str()) {
182	0×	bail!("{} cannot be handed to the daemon", n.kind);
183	16×	}
184	16×	if n.actor != "human" {
185	0×	bail!("only the owner hands events to the daemon");
186	16×	}
187	16×	if n.kind == "brief.set" {
188	4×	crate::rules::check(n.payload.get("text").and_then(\|t\| t.as_str()).unwrap_or(""))?;
189	12×	}
190		// Check it as verify will, against who is who including the
191		// batch so far.
192	16×	let probe = probe_event(&n);
193	16×	principals.check(&probe).map_err(\|why\| anyhow::anyhow!("refused {}: {why}", n.kind))?;
194	16×	if n.kind != "identity.rotated" {
195	12×	let key = crate::identity::KeyId(n.auth.key.context("unsigned")?);
196	12×	if !matches!(principals.principal(&key, crate::ids::now_ms()), Ok(Principal::Owner { .. })) {
197	0×	bail!("refused {}: not signed by the owner", n.kind);
198	12×	}
199	4×	}
200	16×	out.push(n);
201		}
202	12×	let appended = self.ledger.append(out).await?;
203	12×	let _ = self.ledger.checkpoint().await;
204		// Who is who has changed: know it before the next request arrives.
205	12×	*self.principals.write().await = Principals::load(&self.ledger).await?;
206	18×	Ok(appended.iter().map(\|e\| e.seq).collect())
207	12×	}
208
209		/// The owner `spectra init --owner` named, while none is recorded.
210	12×	pub fn named_owner(&self) -> Option<String> {
211	12×	let text = std::fs::read_to_string(self.data.join("spectra.toml")).ok()?;
212	12×	let t: toml::Table = text.parse().ok()?;
213	12×	t.get("owner")?.get("key")?.as_str().map(\|k\| k.to_ascii_lowercase())
214	12×	}
215
216		/// The owner claims a Spectra set up with `spectra init --owner`
217		/// (spec §8.6): the named identity, signed by itself over its payload,
218		/// then any successors up to the owner's current identity, each a
219		/// rotation as `identity refresh` makes. Accepted only while no owner
220		/// is recorded, and only from the key init named.
221	6×	pub async fn claim(&self, events: Vec<WireEvent>) -> Result<Vec<u64>> {
222	6×	let mut principals = self.principals.read().await.clone();
223	6×	if principals.owner().is_some() {
224	1×	bail!("this Spectra already has an owner");
225	5×	}
226	5×	let named = self.named_owner().context("this Spectra names no owner (spectra init --owner)")?;
227	5×	let mut out = Vec::new();
228	7×	for (i, w) in events.into_iter().enumerate() {
229	7×	let n = w.into_new_event()?;
230	7×	let want = if i == 0 { "identity.owner" } else { "identity.rotated" };
231	7×	if n.kind != want \|\| n.actor != "human" {
232	0×	bail!("a claim is the owner's identity.owner, then their rotations");
233	7×	}
234	7×	if i == 0 {
235	5×	let key = n.payload.get("key").and_then(\|k\| k.as_str()).unwrap_or("").to_ascii_lowercase();
236	5×	if key != named \|\| n.auth.key.map(hex::encode) != Some(named.clone()) {
237	2×	bail!("this Spectra was set up for another owner");
238	3×	}
239	2×	}
240	5×	principals.check(&probe_event(&n)).map_err(\|why\| anyhow::anyhow!("refused the claim: {why}"))?;
241	5×	out.push(n);
242		}
243	3×	if out.is_empty() {
244	0×	bail!("an empty claim");
245	3×	}
246	3×	let appended = self.ledger.append(out).await?;
247	3×	let _ = self.ledger.checkpoint().await;
248	3×	*self.principals.write().await = Principals::load(&self.ledger).await?;
249	5×	Ok(appended.iter().map(\|e\| e.seq).collect())
250	6×	}
251
252		/// Write a checkpoint covering everything (at shutdown).
253	22×	pub async fn shutdown(&self) {
254	22×	let _ = self.ledger.checkpoint().await;
255	22×	}
256		}
257
258		/// The Unix peer of a socket connection.
259		#[derive(Clone, Copy, Debug, PartialEq, Eq)]
260		pub struct Peer {
261		pub uid: u32,
262		pub pid: Option<i32>,
263		}
264
265		/// An event as the CLI hands it to the daemon: signed already.
266		#[derive(Clone, Debug, Serialize, Deserialize)]
267		pub struct WireEvent {
268		pub kind: String,
269		pub actor: String,
270		pub subject: Option<String>,
271		pub project: Option<String>,
272		pub payload: serde_json::Value,
273		/// Hex of the auth key and the encoded proof; absent for a project's channel.
274		pub key: Option<String>,
275		pub proof: Option<String>,
276		}
277
278		impl WireEvent {
279	27×	pub fn from_new(n: &NewEvent) -> WireEvent {
280	27×	WireEvent {
281	27×	kind: n.kind.clone(),
282	27×	actor: n.actor.clone(),
283	27×	subject: n.subject.clone(),
284	27×	project: n.project.clone(),
285	27×	payload: n.payload.clone(),
286	27×	key: n.auth.key.map(hex::encode),
287	27×	proof: n.auth.proof.as_ref().map(hex::encode),
288	27×	}
289	27×	}
290
291	25×	fn into_new_event(self) -> Result<NewEvent> {
292	25×	let auth = match (&self.key, &self.proof) {
293	23×	(Some(k), Some(p)) => Auth {
294	23×	key: Some(hex::decode(k).ok().and_then(\|b\| b.try_into().ok()).context("a key is 32 bytes of hex")?),
295	23×	method: Method::Signature,
296		assurance: 2,
297	23×	proof: Some(hex::decode(p).context("a proof is hex")?),
298		},
299	2×	_ => Auth::system(),
300		};
301	25×	let mut n = NewEvent::system(&self.kind, self.payload).by(self.actor, auth);
302	25×	n.subject = self.subject;
303	25×	n.project = self.project;
304	25×	Ok(n)
305	25×	}
306		}
307
308		/// The event `n` would be, for checking before it is appended.
309	21×	fn probe_event(n: &NewEvent) -> Event {
310	21×	let now = crate::ids::now_ms();
311	21×	Event {
312	21×	seq: 0,
313	21×	id: String::new(),
314	21×	at_ms: now,
315	21×	hlc: crate::hlc::Hlc { wall_ms: now, logical: 0 },
316	21×	prev: [0; 32],
317	21×	hash: [0; 32],
318	21×	auth: n.auth.clone(),
319	21×	kind: n.kind.clone(),
320	21×	actor: n.actor.clone(),
321	21×	session: n.session.clone(),
322	21×	project: n.project.clone(),
323	21×	subject: n.subject.clone(),
324	21×	payload: Some(crate::event::canonical_json(&n.payload)),
325	21×	trace: n.trace.clone(),
326	21×	}
327	21×	}

Coverage / crates/spectra-core/src/daemon.rs 93.0%

Keyboard