OXYGENHUB
# 8cfb9269bb14 spectra / crates / spectra-core / src / daemon.rs
10.3 KiB · Rust · last changed The 0.1 tools: brief, rules, agents, status, say, read, inbox, wait with the idle guard, ack, search, redact; every call on the record (#8) 2 days ago source blame history raw
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] = &["identity.rotated", "agent.added", "agent.disabled", "project.added"];
22
23 
pub struct Daemon {
24 
    pub data: PathBuf,
25 
    pub ledger: Ledger,
26 
    pub principals: RwLock<Principals>,
27 
    pub authn: Authenticator,
28 
    pub presence: Mutex<Presence>,
29 
    /// Per agent: empty `wait`s in a row, and until when `wait` is refused.
30 
    pub idle_waits: Mutex<std::collections::HashMap<String, (u32, u64)>>,
31 
}
32
33 
impl Daemon {
34 
    /// Open the data directory's ledger for writing and start the timers.
35 
    pub async fn start(data: &Path) -> Result<Arc<Daemon>> {
36 
        let db = data.join("spectra.db");
37 
        if !db.exists() {
38 
            bail!("no Spectra in {} (run `spectra init`)", data.display());
39 
        }
40 
        let instance = KeyPair::instance(data, "spectra")?;
41 
        let ledger = Ledger::open(&db, Some(instance)).await?;
42 
        let principals = Principals::load(&ledger).await?;
43 
        let d = Arc::new(Daemon {
44 
            data: data.to_path_buf(),
45 
            ledger,
46 
            principals: RwLock::new(principals),
47 
            authn: Authenticator::default(),
48 
            presence: Mutex::new(Presence::default()),
49 
            idle_waits: Mutex::new(Default::default()),
50 
        });
51 
        d.clone().watch_principals();
52 
        d.clone().tick_presence();
53 
        Ok(d)
54 
    }
55
56 
    /// Reload who is who whenever an identity or agent event is appended.
57 
    fn watch_principals(self: Arc<Self>) {
58 
        let mut rx = self.ledger.subscribe();
59 
        tokio::spawn(async move {
60 
            loop {
61 
                match rx.recv().await {
62 
                    Ok(e)
63 
                        if e.kind.starts_with("identity.")
64 
                            || e.kind.starts_with("agent.") && e.kind != "agent.presence" && e.kind != "agent.status" =>
65 
                    {
66 
                        if let Ok(p) = Principals::load(&self.ledger).await {
67 
                            *self.principals.write().await = p;
68 
                        }
69 
                    }
70 
                    Ok(_) => {}
71 
                    Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => {
72 
                        if let Ok(p) = Principals::load(&self.ledger).await {
73 
                            *self.principals.write().await = p;
74 
                        }
75 
                    }
76 
                    Err(_) => return,
77 
                }
78 
            }
79 
        });
80 
    }
81
82 
    /// Record presence changes and quiet sessions every 15 s, and write a
83 
    /// checkpoint when one is due by time.
84 
    fn tick_presence(self: Arc<Self>) {
85 
        tokio::spawn(async move {
86 
            let mut every = tokio::time::interval(std::time::Duration::from_secs(15));
87 
            loop {
88 
                every.tick().await;
89 
                let events = self.presence.lock().await.tick(crate::ids::now_ms());
90 
                if !events.is_empty() {
91 
                    let _ = self.ledger.append(events).await;
92 
                }
93 
                let _ = self.ledger.append(vec![]).await;
94 
            }
95 
        });
96 
    }
97
98 
    /// Authenticate a request (spec §8.5); an agent used from the wrong Unix
99 
    /// user is recorded as `guard.impersonation`, and a first connection
100 
    /// that binds an agent's uid as `agent.uid_bound`.
101 
    pub async fn authenticate(&self, proof: Option<&Proof>, method: &str, body: &[u8], peer: Option<Peer>) -> Result<Caller, Refusal> {
102 
        let now = crate::ids::now_ms();
103 
        let r = {
104 
            let p = self.principals.read().await;
105 
            self.authn.authenticate(&p, proof, method, body, now, peer.map(|x| x.uid))
106 
        };
107 
        match &r {
108 
            Err(Refusal::WrongUid { agent, bound, from }) => {
109 
                let e = NewEvent::system(
110 
                    "guard.impersonation",
111 
                    serde_json::json!({ "agent": agent, "bound_uid": bound, "from_uid": from, "from_pid": peer.map(|x| x.pid), "method": method }),
112 
                )
113 
                .subject(agent.clone());
114 
                let _ = self.ledger.append_one(e).await;
115 
                tracing::warn!(agent, bound, from, "a key used from the wrong Unix user");
116 
            }
117 
            Ok(Caller { principal: Principal::Agent { name, .. }, learn_uid: Some(uid), .. }) => {
118 
                let e = NewEvent::system("agent.uid_bound", serde_json::json!({ "name": name, "uid": uid })).subject(name.clone());
119 
                let _ = self.ledger.append_one(e).await;
120 
            }
121 
            _ => {}
122 
        }
123 
        r
124 
    }
125
126 
    /// Note a request from an agent: its session, and the event that
127 
    /// starts one when this is the first in ten minutes.
128 
    pub async fn touch(&self, who: &Principal, new_session: bool) -> Option<String> {
129 
        let Principal::Agent { name, .. } = who else { return None };
130 
        let (id, started) = self.presence.lock().await.request(name, crate::ids::now_ms(), new_session);
131 
        if let Some(e) = started {
132 
            let _ = self.ledger.append_one(e).await;
133 
        }
134 
        Some(id)
135 
    }
136
137 
    /// Append events the owner's CLI signed over their payload, after
138 
    /// checking each as `spectra admin verify` would.
139 
    pub async fn accept_owner_events(&self, events: Vec<WireEvent>) -> Result<Vec<u64>> {
140 
        let mut principals = self.principals.read().await.clone();
141 
        let projects: Vec<String> = events
142 
            .iter()
143 
            .filter(|e| e.kind == "project.added")
144 
            .filter_map(|e| e.payload.get("name").and_then(|n| n.as_str()).map(String::from))
145 
            .collect();
146 
        let mut out = Vec::new();
147 
        for w in events {
148 
            let n = w.into_new_event()?;
149 
            if n.kind == "channel.created" {
150 
                // A project's channel comes with its project, unsigned, as `project add` makes it.
151 
                let p = &n.payload;
152 
                let ok = p.get("kind").and_then(|k| k.as_str()) == Some("project")
153 
                    && p.get("name").and_then(|x| x.as_str()).is_some_and(|x| projects.iter().any(|q| q == x));
154 
                if !ok || n.auth.method != Method::System {
155 
                    bail!("a channel is only made with its project");
156 
                }
157 
                out.push(n);
158 
                continue;
159 
            }
160 
            if !OWNER_KINDS.contains(&n.kind.as_str()) {
161 
                bail!("{} cannot be handed to the daemon", n.kind);
162 
            }
163 
            if n.actor != "human" {
164 
                bail!("only the owner hands events to the daemon");
165 
            }
166 
            // Check it as verify will, against who is who including the
167 
            // batch so far.
168 
            let probe = probe_event(&n);
169 
            principals.check(&probe).map_err(|why| anyhow::anyhow!("refused {}: {why}", n.kind))?;
170 
            if n.kind != "identity.rotated" {
171 
                let key = oxy_core::hash::ObjectRef(n.auth.key.context("unsigned")?);
172 
                if !matches!(principals.principal(&key, crate::ids::now_ms()), Ok(Principal::Owner { .. })) {
173 
                    bail!("refused {}: not signed by the owner", n.kind);
174 
                }
175 
            }
176 
            out.push(n);
177 
        }
178 
        let appended = self.ledger.append(out).await?;
179 
        let _ = self.ledger.checkpoint().await;
180 
        // Who is who has changed: know it before the next request arrives.
181 
        *self.principals.write().await = Principals::load(&self.ledger).await?;
182 
        Ok(appended.iter().map(|e| e.seq).collect())
183 
    }
184
185 
    /// Write a checkpoint covering everything (at shutdown).
186 
    pub async fn shutdown(&self) {
187 
        let _ = self.ledger.checkpoint().await;
188 
    }
189 
}
190
191 
/// The Unix peer of a socket connection.
192 
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
193 
pub struct Peer {
194 
    pub uid: u32,
195 
    pub pid: Option<i32>,
196 
}
197
198 
/// An event as the CLI hands it to the daemon: signed already.
199 
#[derive(Clone, Debug, Serialize, Deserialize)]
200 
pub struct WireEvent {
201 
    pub kind: String,
202 
    pub actor: String,
203 
    pub subject: Option<String>,
204 
    pub project: Option<String>,
205 
    pub payload: serde_json::Value,
206 
    /// Hex of the auth key and the encoded proof; absent for a project's channel.
207 
    pub key: Option<String>,
208 
    pub proof: Option<String>,
209 
}
210
211 
impl WireEvent {
212 
    pub fn from_new(n: &NewEvent) -> WireEvent {
213 
        WireEvent {
214 
            kind: n.kind.clone(),
215 
            actor: n.actor.clone(),
216 
            subject: n.subject.clone(),
217 
            project: n.project.clone(),
218 
            payload: n.payload.clone(),
219 
            key: n.auth.key.map(hex::encode),
220 
            proof: n.auth.proof.as_ref().map(hex::encode),
221 
        }
222 
    }
223
224 
    fn into_new_event(self) -> Result<NewEvent> {
225 
        let auth = match (&self.key, &self.proof) {
226 
            (Some(k), Some(p)) => Auth {
227 
                key: Some(hex::decode(k).ok().and_then(|b| b.try_into().ok()).context("a key is 32 bytes of hex")?),
228 
                method: Method::Signature,
229 
                assurance: 2,
230 
                proof: Some(hex::decode(p).context("a proof is hex")?),
231 
            },
232 
            _ => Auth::system(),
233 
        };
234 
        let mut n = NewEvent::system(&self.kind, self.payload).by(self.actor, auth);
235 
        n.subject = self.subject;
236 
        n.project = self.project;
237 
        Ok(n)
238 
    }
239 
}
240
241 
/// The event `n` would be, for checking before it is appended.
242 
fn probe_event(n: &NewEvent) -> Event {
243 
    let now = crate::ids::now_ms();
244 
    Event {
245 
        seq: 0,
246 
        id: String::new(),
247 
        at_ms: now,
248 
        hlc: crate::hlc::Hlc { wall_ms: now, logical: 0 },
249 
        prev: [0; 32],
250 
        hash: [0; 32],
251 
        auth: n.auth.clone(),
252 
        kind: n.kind.clone(),
253 
        actor: n.actor.clone(),
254 
        session: n.session.clone(),
255 
        project: n.project.clone(),
256 
        subject: n.subject.clone(),
257 
        payload: Some(crate::event::canonical_json(&n.payload)),
258 
        trace: n.trace.clone(),
259 
    }
260 
}