| 1 |
|
| 2 |
|
| 3 |
|
| 4 |
|
| 5 |
|
| 6 |
|
| 7 |
|
| 8 |
|
| 9 |
|
| 10 |
use askama::Template; |
| 11 |
use axum::body::Bytes; |
| 12 |
use axum::extract::connect_info::ConnectInfo; |
| 13 |
use axum::extract::{Form, Path, Query, State}; |
| 14 |
use axum::http::{header, HeaderMap, HeaderValue, StatusCode}; |
| 15 |
use axum::response::sse::{Event as SseEvent, KeepAlive, Sse}; |
| 16 |
use axum::response::{Html, IntoResponse, Redirect, Response}; |
| 17 |
use axum::routing::{get, post}; |
| 18 |
use axum::Router; |
| 19 |
use futures::StreamExt; |
| 20 |
use serde::Deserialize; |
| 21 |
use spectra_core::conversation::{self as conv, Author, Post, Signer}; |
| 22 |
use spectra_core::event::{Auth, Method}; |
| 23 |
use spectra_core::ids::{format_time, now_ms}; |
| 24 |
use spectra_core::principals::{Principal, Principals}; |
| 25 |
use spectra_mcp::http::{AppState, PeerInfo}; |
| 26 |
use spectra_mcp::session::{self, Session}; |
| 27 |
use std::time::Duration; |
| 28 |
|
| 29 |
const MARK: &str = include_str!("../assets/spectra.svg"); |
| 30 |
const STYLE: &str = include_str!("../assets/style.css"); |
| 31 |
const SCRIPT: &str = include_str!("../assets/radar.js"); |
| 32 |
const COOKIE: &str = "spectra_session"; |
| 33 |
|
| 34 |
#[derive(Clone)] |
| 35 |
pub struct RadarState { |
| 36 |
pub app: AppState, |
| 37 |
} |
| 38 |
|
| 39 |
impl RadarState { |
| 40 |
fn sessions(&self) -> &session::Sessions { |
| 41 |
&self.app.sessions |
| 42 |
} |
| 43 |
} |
| 44 |
|
| 45 |
|
| 46 |
pub fn router(app: AppState) -> Router { |
| 47 |
let state = RadarState { app }; |
| 48 |
Router::new() |
| 49 |
.route("/", get(live)) |
| 50 |
.route("/channels", get(channels)) |
| 51 |
.route("/c/{name}", get(channel).post(post_channel)) |
| 52 |
.route("/c/{name}/t/{id}", get(thread).post(post_thread)) |
| 53 |
.route("/log", get(log)) |
| 54 |
.route("/log/verify", post(verify)) |
| 55 |
.route("/radar/link", post(link)) |
| 56 |
.route("/signin", get(signin)) |
| 57 |
.route("/signout", post(signout)) |
| 58 |
.route("/radar/events", get(events)) |
| 59 |
.route("/radar/style.css", get(|| async { asset("text/css; charset=utf-8", STYLE) })) |
| 60 |
.route("/radar/radar.js", get(|| async { asset("text/javascript; charset=utf-8", SCRIPT) })) |
| 61 |
.route("/radar/spectra.svg", get(|| async { asset("image/svg+xml", MARK) })) |
| 62 |
.layer(axum::middleware::map_response(security_headers)) |
| 63 |
.with_state(state) |
| 64 |
} |
| 65 |
|
| 66 |
fn asset(kind: &'static str, body: &'static str) -> Response { |
| 67 |
([(header::CONTENT_TYPE, kind), (header::CACHE_CONTROL, "max-age=300")], body).into_response() |
| 68 |
} |
| 69 |
|
| 70 |
|
| 71 |
async fn security_headers(mut r: Response) -> Response { |
| 72 |
let h = r.headers_mut(); |
| 73 |
h.insert("content-security-policy", HeaderValue::from_static("default-src 'self'; script-src 'self'; style-src 'self'; img-src 'self' data:; frame-ancestors 'none'; form-action 'self'; base-uri 'none'")); |
| 74 |
h.insert("x-content-type-options", HeaderValue::from_static("nosniff")); |
| 75 |
|
| 76 |
h.insert("referrer-policy", HeaderValue::from_static("same-origin")); |
| 77 |
h.insert("x-frame-options", HeaderValue::from_static("DENY")); |
| 78 |
r |
| 79 |
} |
| 80 |
|
| 81 |
|
| 82 |
pub struct Shell { |
| 83 |
pub page: &'static str, |
| 84 |
pub owner: String, |
| 85 |
pub head: u64, |
| 86 |
pub version: &'static str, |
| 87 |
pub mark: &'static str, |
| 88 |
} |
| 89 |
|
| 90 |
|
| 91 |
struct Owner { |
| 92 |
session: Session, |
| 93 |
principal: Principal, |
| 94 |
shell: Shell, |
| 95 |
} |
| 96 |
|
| 97 |
impl Owner { |
| 98 |
fn author(&self) -> Author<'static> { |
| 99 |
Author { |
| 100 |
principal: self.principal.clone(), |
| 101 |
signer: Signer::Request(Auth { |
| 102 |
key: Some(self.session.key), |
| 103 |
method: Method::Session, |
| 104 |
assurance: 2, |
| 105 |
proof: Some(self.session.proof.clone()), |
| 106 |
}), |
| 107 |
} |
| 108 |
} |
| 109 |
} |
| 110 |
|
| 111 |
fn cookie(headers: &HeaderMap) -> Option<String> { |
| 112 |
let c = headers.get(header::COOKIE)?.to_str().ok()?; |
| 113 |
c.split(';').map(str::trim).find_map(|kv| kv.strip_prefix(&format!("{COOKIE}=")).map(String::from)) |
| 114 |
} |
| 115 |
|
| 116 |
async fn owner(st: &RadarState, headers: &HeaderMap, page: &'static str) -> Result<Owner, Response> { |
| 117 |
let now = now_ms(); |
| 118 |
let session = cookie(headers) |
| 119 |
.and_then(|id| st.sessions().get(&id, now)) |
| 120 |
.ok_or_else(|| signin_page("Radar shows Spectra to its owner only.", StatusCode::UNAUTHORIZED))?; |
| 121 |
let d = &st.app.daemon; |
| 122 |
let principal = d |
| 123 |
.principals |
| 124 |
.read() |
| 125 |
.await |
| 126 |
.principal(&key_id(session.key), now) |
| 127 |
.map_err(|why| signin_page(&format!("This session no longer belongs to the owner: {why}."), StatusCode::UNAUTHORIZED))?; |
| 128 |
if !matches!(principal, Principal::Owner { .. }) { |
| 129 |
return Err(signin_page("Radar shows Spectra to its owner only.", StatusCode::FORBIDDEN)); |
| 130 |
} |
| 131 |
let shell = Shell { page, owner: principal.display(), head: d.ledger.head().await.0, version: env!("CARGO_PKG_VERSION"), mark: MARK }; |
| 132 |
Ok(Owner { session, principal, shell }) |
| 133 |
} |
| 134 |
|
| 135 |
fn key_id(k: [u8; 32]) -> spectra_core::identity::KeyId { |
| 136 |
spectra_core::identity::KeyId(k) |
| 137 |
} |
| 138 |
|
| 139 |
#[derive(Template)] |
| 140 |
#[template(path = "signin.html")] |
| 141 |
struct SigninPage<'a> { |
| 142 |
mark: &'static str, |
| 143 |
message: &'a str, |
| 144 |
} |
| 145 |
|
| 146 |
fn signin_page(message: &str, status: StatusCode) -> Response { |
| 147 |
(status, render(&SigninPage { mark: MARK, message })).into_response() |
| 148 |
} |
| 149 |
|
| 150 |
fn render<T: Template>(t: &T) -> Response { |
| 151 |
match t.render() { |
| 152 |
Ok(html) => Html(html).into_response(), |
| 153 |
Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, format!("could not render the page: {e}")).into_response(), |
| 154 |
} |
| 155 |
} |
| 156 |
|
| 157 |
|
| 158 |
|
| 159 |
fn check_form(st: &RadarState, o: &Owner, headers: &HeaderMap, csrf: &str) -> Option<Response> { |
| 160 |
if !spectra_mcp::http::origin_ok(headers, &st.app.origins) { |
| 161 |
return Some((StatusCode::FORBIDDEN, "this origin may not change Spectra").into_response()); |
| 162 |
} |
| 163 |
if csrf != o.session.csrf { |
| 164 |
return Some((StatusCode::FORBIDDEN, "the form is stale: reload the page").into_response()); |
| 165 |
} |
| 166 |
None |
| 167 |
} |
| 168 |
|
| 169 |
|
| 170 |
fn markdown(body: &str) -> String { |
| 171 |
let mut html = String::new(); |
| 172 |
pulldown_cmark::html::push_html( |
| 173 |
&mut html, |
| 174 |
pulldown_cmark::Parser::new_ext(body, pulldown_cmark::Options::ENABLE_STRIKETHROUGH | pulldown_cmark::Options::ENABLE_TABLES), |
| 175 |
); |
| 176 |
ammonia::clean(&html) |
| 177 |
} |
| 178 |
|
| 179 |
|
| 180 |
|
| 181 |
struct AgentCard { |
| 182 |
name: String, |
| 183 |
owner: String, |
| 184 |
kind: String, |
| 185 |
project: String, |
| 186 |
presence: String, |
| 187 |
status: Option<String>, |
| 188 |
last_tool: Option<String>, |
| 189 |
last_at: String, |
| 190 |
} |
| 191 |
|
| 192 |
struct EventRow { |
| 193 |
seq: i64, |
| 194 |
at: String, |
| 195 |
kind: String, |
| 196 |
actor: String, |
| 197 |
subject: Option<String>, |
| 198 |
} |
| 199 |
|
| 200 |
#[derive(Template)] |
| 201 |
#[template(path = "live.html")] |
| 202 |
struct LivePage { |
| 203 |
shell: Shell, |
| 204 |
agents: Vec<AgentCard>, |
| 205 |
unread: i64, |
| 206 |
events: Vec<EventRow>, |
| 207 |
} |
| 208 |
|
| 209 |
async fn live(State(st): State<RadarState>, headers: HeaderMap) -> Response { |
| 210 |
let o = match owner(&st, &headers, "live").await { |
| 211 |
Ok(o) => o, |
| 212 |
Err(r) => return r, |
| 213 |
}; |
| 214 |
let d = &st.app.daemon; |
| 215 |
let pool = d.ledger.pool(); |
| 216 |
let rows: Vec<(String, String, String, String, Option<String>)> = |
| 217 |
sqlx::query_as("SELECT a.name, a.owner_name, a.kind, a.project, p.status_text FROM agents a LEFT JOIN presence p ON p.agent = a.name WHERE a.disabled_ms IS NULL ORDER BY a.name") |
| 218 |
.fetch_all(pool) |
| 219 |
.await |
| 220 |
.unwrap_or_default(); |
| 221 |
let mut agents = Vec::new(); |
| 222 |
{ |
| 223 |
let now = now_ms(); |
| 224 |
for (name, owner, kind, project, status) in rows { |
| 225 |
let last: Option<(String, i64)> = |
| 226 |
sqlx::query_as("SELECT tool, at_ms FROM tool_calls WHERE agent = ? ORDER BY seq DESC LIMIT 1") |
| 227 |
.bind(&name) |
| 228 |
.fetch_optional(pool) |
| 229 |
.await |
| 230 |
.ok() |
| 231 |
.flatten(); |
| 232 |
let presence = d.presence.lock().await.state(&name, now).to_string(); |
| 233 |
agents.push(AgentCard { |
| 234 |
name, |
| 235 |
owner, |
| 236 |
kind, |
| 237 |
project, |
| 238 |
presence, |
| 239 |
status, |
| 240 |
last_tool: last.as_ref().map(|(t, _)| t.clone()), |
| 241 |
last_at: last.map(|(_, at)| format_time(at as u64)).unwrap_or_default(), |
| 242 |
}); |
| 243 |
} |
| 244 |
} |
| 245 |
let unread = conv::unread(&d.ledger, "human").await.unwrap_or(0); |
| 246 |
let events = recent(&st, None, None, 50) |
| 247 |
.await |
| 248 |
.into_iter() |
| 249 |
.map(|e| EventRow { seq: e.seq, at: e.at, kind: e.kind, actor: e.actor, subject: e.subject }) |
| 250 |
.collect(); |
| 251 |
render(&LivePage { shell: o.shell, agents, unread, events }) |
| 252 |
} |
| 253 |
|
| 254 |
|
| 255 |
|
| 256 |
struct ChannelRow { |
| 257 |
name: String, |
| 258 |
display: String, |
| 259 |
kind: String, |
| 260 |
purpose: String, |
| 261 |
count: i64, |
| 262 |
last: String, |
| 263 |
} |
| 264 |
|
| 265 |
|
| 266 |
type ChannelCount = (String, String, String, Option<String>, i64, Option<i64>); |
| 267 |
|
| 268 |
#[derive(Template)] |
| 269 |
#[template(path = "channels.html")] |
| 270 |
struct ChannelsPage { |
| 271 |
shell: Shell, |
| 272 |
channels: Vec<ChannelRow>, |
| 273 |
} |
| 274 |
|
| 275 |
async fn channels(State(st): State<RadarState>, headers: HeaderMap) -> Response { |
| 276 |
let o = match owner(&st, &headers, "channels").await { |
| 277 |
Ok(o) => o, |
| 278 |
Err(r) => return r, |
| 279 |
}; |
| 280 |
let rows: Vec<ChannelCount> = sqlx::query_as( |
| 281 |
"SELECT c.id, c.name, c.kind, c.purpose, (SELECT COUNT(*) FROM messages m WHERE m.channel = c.id), (SELECT MAX(at_ms) FROM messages m WHERE m.channel = c.id) \ |
| 282 |
FROM channels c ORDER BY c.kind = 'direct', c.kind = 'system', c.name", |
| 283 |
) |
| 284 |
.fetch_all(st.app.daemon.ledger.pool()) |
| 285 |
.await |
| 286 |
.unwrap_or_default(); |
| 287 |
let channels = rows |
| 288 |
.into_iter() |
| 289 |
.map(|(_, name, kind, purpose, count, last)| ChannelRow { |
| 290 |
display: match name.strip_prefix("dm:") { |
| 291 |
Some(p) => format!("@{p}"), |
| 292 |
None => format!("#{name}"), |
| 293 |
}, |
| 294 |
name, |
| 295 |
kind, |
| 296 |
purpose: purpose.unwrap_or_default(), |
| 297 |
count, |
| 298 |
last: last.map(|t| format_time(t as u64)).unwrap_or_default(), |
| 299 |
}) |
| 300 |
.collect(); |
| 301 |
render(&ChannelsPage { shell: o.shell, channels }) |
| 302 |
} |
| 303 |
|
| 304 |
|
| 305 |
|
| 306 |
struct MessageRow { |
| 307 |
id: String, |
| 308 |
author: String, |
| 309 |
at: String, |
| 310 |
kind: String, |
| 311 |
replies: i64, |
| 312 |
redacted: bool, |
| 313 |
body_html: String, |
| 314 |
} |
| 315 |
|
| 316 |
#[derive(Template)] |
| 317 |
#[template(path = "channel.html")] |
| 318 |
struct ChannelPage { |
| 319 |
shell: Shell, |
| 320 |
name: String, |
| 321 |
display: String, |
| 322 |
purpose: Option<String>, |
| 323 |
thread: Option<String>, |
| 324 |
messages: Vec<MessageRow>, |
| 325 |
can_post: bool, |
| 326 |
action: String, |
| 327 |
csrf: String, |
| 328 |
} |
| 329 |
|
| 330 |
async fn show(st: &RadarState, o: Owner, name: &str, thread: Option<&str>) -> Response { |
| 331 |
let d = &st.app.daemon; |
| 332 |
let (ch, msgs) = match conv::read(&d.ledger, &o.principal, if thread.is_some() { None } else { Some(name) }, thread, None, 200).await { |
| 333 |
Ok(r) => r, |
| 334 |
Err(e) => return (StatusCode::NOT_FOUND, format!("{e:#}")).into_response(), |
| 335 |
}; |
| 336 |
let mut messages = Vec::new(); |
| 337 |
for m in msgs { |
| 338 |
messages.push(MessageRow { |
| 339 |
author: spectra_core::conversation::display_name(&d.ledger, &m.author).await, |
| 340 |
at: format_time(m.at_ms as u64), |
| 341 |
redacted: m.redacted.is_some(), |
| 342 |
body_html: if m.redacted.is_some() { ammonia::clean_text(&m.body) } else { markdown(&m.body) }, |
| 343 |
id: m.id, |
| 344 |
kind: m.kind, |
| 345 |
replies: m.replies, |
| 346 |
}); |
| 347 |
} |
| 348 |
let can_post = conv::can_post(&o.principal, &ch).is_ok(); |
| 349 |
let action = match thread { |
| 350 |
Some(t) => format!("/c/{}/t/{t}", ch.name), |
| 351 |
None => format!("/c/{}", ch.name), |
| 352 |
}; |
| 353 |
let csrf = o.session.csrf.clone(); |
| 354 |
render(&ChannelPage { |
| 355 |
shell: o.shell, |
| 356 |
name: ch.name.clone(), |
| 357 |
display: ch.display(), |
| 358 |
purpose: ch.purpose.clone(), |
| 359 |
thread: thread.map(String::from), |
| 360 |
messages, |
| 361 |
can_post, |
| 362 |
action, |
| 363 |
csrf, |
| 364 |
}) |
| 365 |
} |
| 366 |
|
| 367 |
async fn channel(State(st): State<RadarState>, headers: HeaderMap, Path(name): Path<String>) -> Response { |
| 368 |
match owner(&st, &headers, "channels").await { |
| 369 |
Ok(o) => show(&st, o, &name, None).await, |
| 370 |
Err(r) => r, |
| 371 |
} |
| 372 |
} |
| 373 |
|
| 374 |
async fn thread(State(st): State<RadarState>, headers: HeaderMap, Path((name, id)): Path<(String, String)>) -> Response { |
| 375 |
match owner(&st, &headers, "channels").await { |
| 376 |
Ok(o) => show(&st, o, &name, Some(&id)).await, |
| 377 |
Err(r) => r, |
| 378 |
} |
| 379 |
} |
| 380 |
|
| 381 |
#[derive(Deserialize)] |
| 382 |
struct PostForm { |
| 383 |
_csrf: String, |
| 384 |
body: String, |
| 385 |
reply_to: Option<String>, |
| 386 |
} |
| 387 |
|
| 388 |
async fn post_to(st: &RadarState, headers: &HeaderMap, name: &str, form: PostForm, back: String) -> Response { |
| 389 |
let o = match owner(st, headers, "channels").await { |
| 390 |
Ok(o) => o, |
| 391 |
Err(r) => return r, |
| 392 |
}; |
| 393 |
if let Some(r) = check_form(st, &o, headers, &form._csrf) { |
| 394 |
return r; |
| 395 |
} |
| 396 |
let post = Post { |
| 397 |
channel: if form.reply_to.is_some() { None } else { Some(name.to_string()) }, |
| 398 |
body: form.body, |
| 399 |
reply_to: form.reply_to, |
| 400 |
..Default::default() |
| 401 |
}; |
| 402 |
match conv::post(&st.app.daemon.ledger, &o.author(), post, now_ms()).await { |
| 403 |
Ok(_) => Redirect::to(&back).into_response(), |
| 404 |
Err(e) => (StatusCode::BAD_REQUEST, format!("{e:#}")).into_response(), |
| 405 |
} |
| 406 |
} |
| 407 |
|
| 408 |
async fn post_channel(State(st): State<RadarState>, headers: HeaderMap, Path(name): Path<String>, Form(form): Form<PostForm>) -> Response { |
| 409 |
let back = format!("/c/{name}"); |
| 410 |
post_to(&st, &headers, &name, form, back).await |
| 411 |
} |
| 412 |
|
| 413 |
async fn post_thread( |
| 414 |
State(st): State<RadarState>, |
| 415 |
headers: HeaderMap, |
| 416 |
Path((name, id)): Path<(String, String)>, |
| 417 |
Form(mut form): Form<PostForm>, |
| 418 |
) -> Response { |
| 419 |
form.reply_to = Some(id.clone()); |
| 420 |
let back = format!("/c/{name}/t/{id}"); |
| 421 |
post_to(&st, &headers, &name, form, back).await |
| 422 |
} |
| 423 |
|
| 424 |
|
| 425 |
|
| 426 |
struct LogRow { |
| 427 |
seq: i64, |
| 428 |
id: String, |
| 429 |
at: String, |
| 430 |
kind: String, |
| 431 |
actor: String, |
| 432 |
subject: Option<String>, |
| 433 |
method: String, |
| 434 |
assurance: i64, |
| 435 |
hash: String, |
| 436 |
prev: String, |
| 437 |
key: String, |
| 438 |
proof: String, |
| 439 |
payload: String, |
| 440 |
} |
| 441 |
|
| 442 |
async fn recent(st: &RadarState, kind: Option<&str>, before: Option<i64>, limit: i64) -> Vec<LogRow> { |
| 443 |
#[allow(clippy::type_complexity)] |
| 444 |
let rows: Vec<( |
| 445 |
i64, |
| 446 |
String, |
| 447 |
i64, |
| 448 |
String, |
| 449 |
String, |
| 450 |
Option<String>, |
| 451 |
i64, |
| 452 |
i64, |
| 453 |
Vec<u8>, |
| 454 |
Vec<u8>, |
| 455 |
Option<Vec<u8>>, |
| 456 |
Option<Vec<u8>>, |
| 457 |
Option<Vec<u8>>, |
| 458 |
)> = sqlx::query_as( |
| 459 |
"SELECT seq, id, at_ms, kind, actor, subject, auth_method, auth_assurance, hash, prev, auth_key, auth_proof, payload FROM events \ |
| 460 |
WHERE (? IS NULL OR kind = ?) AND (? IS NULL OR seq < ?) ORDER BY seq DESC LIMIT ?", |
| 461 |
) |
| 462 |
.bind(kind) |
| 463 |
.bind(kind) |
| 464 |
.bind(before) |
| 465 |
.bind(before) |
| 466 |
.bind(limit) |
| 467 |
.fetch_all(st.app.daemon.ledger.pool()) |
| 468 |
.await |
| 469 |
.unwrap_or_default(); |
| 470 |
let mut out = Vec::new(); |
| 471 |
for (seq, id, at, kind, actor, subject, method, assurance, hash, prev, key, proof, payload) in rows { |
| 472 |
let payload = payload |
| 473 |
.and_then(|p| serde_json::from_slice::<serde_json::Value>(&p).ok()) |
| 474 |
.map(|v| serde_json::to_string_pretty(&v).unwrap_or_default()) |
| 475 |
.unwrap_or_else(|| "(pruned to its stub)".into()); |
| 476 |
out.push(LogRow { |
| 477 |
seq, |
| 478 |
id, |
| 479 |
at: format_time(at as u64), |
| 480 |
kind, |
| 481 |
actor: spectra_core::conversation::display_name(&st.app.daemon.ledger, &actor).await, |
| 482 |
subject, |
| 483 |
method: Method::from_u8(method as u8).map(|m| m.name()).unwrap_or("?").to_string(), |
| 484 |
assurance, |
| 485 |
hash: hex::encode(hash), |
| 486 |
prev: hex::encode(prev), |
| 487 |
key: key.map(hex::encode).unwrap_or_else(|| "none (Spectra itself; covered by the next checkpoint)".into()), |
| 488 |
proof: proof.map(|p| format!("{} bytes", p.len())).unwrap_or_else(|| "none".into()), |
| 489 |
payload, |
| 490 |
}); |
| 491 |
} |
| 492 |
out |
| 493 |
} |
| 494 |
|
| 495 |
#[derive(Template)] |
| 496 |
#[template(path = "log.html")] |
| 497 |
struct LogPage { |
| 498 |
shell: Shell, |
| 499 |
head_hash: String, |
| 500 |
verified: String, |
| 501 |
csrf: String, |
| 502 |
kinds: Vec<String>, |
| 503 |
kind: String, |
| 504 |
events: Vec<LogRow>, |
| 505 |
older: Option<i64>, |
| 506 |
} |
| 507 |
|
| 508 |
#[derive(Deserialize)] |
| 509 |
struct LogQuery { |
| 510 |
kind: Option<String>, |
| 511 |
before: Option<i64>, |
| 512 |
around: Option<i64>, |
| 513 |
} |
| 514 |
|
| 515 |
async fn log(State(st): State<RadarState>, headers: HeaderMap, Query(q): Query<LogQuery>) -> Response { |
| 516 |
let o = match owner(&st, &headers, "log").await { |
| 517 |
Ok(o) => o, |
| 518 |
Err(r) => return r, |
| 519 |
}; |
| 520 |
let d = &st.app.daemon; |
| 521 |
let (_, head) = d.ledger.head().await; |
| 522 |
let last: Option<(Vec<u8>, i64)> = |
| 523 |
sqlx::query_as("SELECT payload, at_ms FROM events WHERE kind = 'ledger.verified' ORDER BY seq DESC LIMIT 1") |
| 524 |
.fetch_optional(d.ledger.pool()) |
| 525 |
.await |
| 526 |
.ok() |
| 527 |
.flatten(); |
| 528 |
let verified = match last { |
| 529 |
Some((p, at)) => { |
| 530 |
let v: serde_json::Value = serde_json::from_slice(&p).unwrap_or_default(); |
| 531 |
let n = v["findings"].as_u64().unwrap_or(0); |
| 532 |
format!( |
| 533 |
"{} at {} over {} events", |
| 534 |
if n == 0 { "clean".to_string() } else { format!("{n} finding(s)") }, |
| 535 |
format_time(at as u64), |
| 536 |
v["events"].as_u64().unwrap_or(0) |
| 537 |
) |
| 538 |
} |
| 539 |
None => "not yet".into(), |
| 540 |
}; |
| 541 |
let kinds: Vec<String> = |
| 542 |
sqlx::query_scalar("SELECT kind FROM event_counts ORDER BY n DESC LIMIT 12").fetch_all(d.ledger.pool()).await.unwrap_or_default(); |
| 543 |
let kind = q.kind.unwrap_or_default(); |
| 544 |
let before = q.before.or(q.around.map(|a| a + 25)); |
| 545 |
let events = recent(&st, (!kind.is_empty()).then_some(kind.as_str()), before, 100).await; |
| 546 |
let older = (events.len() == 100).then(|| events.last().map(|e| e.seq)).flatten(); |
| 547 |
let csrf = o.session.csrf.clone(); |
| 548 |
render(&LogPage { shell: o.shell, head_hash: hex::encode(head), verified, csrf, kinds, kind, events, older }) |
| 549 |
} |
| 550 |
|
| 551 |
#[derive(Deserialize)] |
| 552 |
struct CsrfForm { |
| 553 |
_csrf: String, |
| 554 |
} |
| 555 |
|
| 556 |
|
| 557 |
async fn verify(State(st): State<RadarState>, headers: HeaderMap, Form(f): Form<CsrfForm>) -> Response { |
| 558 |
let o = match owner(&st, &headers, "log").await { |
| 559 |
Ok(o) => o, |
| 560 |
Err(r) => return r, |
| 561 |
}; |
| 562 |
if let Some(r) = check_form(&st, &o, &headers, &f._csrf) { |
| 563 |
return r; |
| 564 |
} |
| 565 |
let ledger = &st.app.daemon.ledger; |
| 566 |
match ledger.verify(&mut Principals::default()).await { |
| 567 |
Ok(v) => { |
| 568 |
let e = spectra_core::event::NewEvent::system( |
| 569 |
"ledger.verified", |
| 570 |
serde_json::json!({ "events": v.events, "checkpoints": v.checkpoints, "findings": v.findings.len(), "head": hex::encode(v.head), "first": v.findings.first().map(|f| format!("#{}: {}", f.seq, f.detail)) }), |
| 571 |
); |
| 572 |
let _ = ledger.append_one(e).await; |
| 573 |
Redirect::to("/log").into_response() |
| 574 |
} |
| 575 |
Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, format!("{e:#}")).into_response(), |
| 576 |
} |
| 577 |
} |
| 578 |
|
| 579 |
|
| 580 |
|
| 581 |
|
| 582 |
async fn link(State(st): State<RadarState>, ConnectInfo(peer): ConnectInfo<PeerInfo>, headers: HeaderMap, body: Bytes) -> Response { |
| 583 |
let caller = match spectra_mcp::http::authenticate(&st.app, peer, &headers, "POST /radar/link", &body).await { |
| 584 |
Ok(c) => c, |
| 585 |
Err(r) => return r, |
| 586 |
}; |
| 587 |
if !matches!(caller.principal, Principal::Owner { .. }) { |
| 588 |
return (StatusCode::FORBIDDEN, axum::Json(serde_json::json!({ "title": "only the owner signs in to Radar" }))).into_response(); |
| 589 |
} |
| 590 |
let (Some(key), Some(proof)) = (caller.auth.key, caller.auth.proof) else { |
| 591 |
return StatusCode::UNAUTHORIZED.into_response(); |
| 592 |
}; |
| 593 |
let code = st.sessions().link(key, proof, now_ms()); |
| 594 |
let base = st.app.origins.first().cloned().unwrap_or_default(); |
| 595 |
axum::Json(serde_json::json!({ "url": format!("{base}/signin?code={code}"), "valid_ms": session::LINK_MS })).into_response() |
| 596 |
} |
| 597 |
|
| 598 |
#[derive(Deserialize)] |
| 599 |
struct SigninQuery { |
| 600 |
code: Option<String>, |
| 601 |
} |
| 602 |
|
| 603 |
async fn signin(State(st): State<RadarState>, Query(q): Query<SigninQuery>) -> Response { |
| 604 |
let Some(id) = q.code.and_then(|c| st.sessions().open(&c, now_ms())) else { |
| 605 |
return signin_page("That sign-in link has been used or has expired.", StatusCode::UNAUTHORIZED); |
| 606 |
}; |
| 607 |
let cookie = format!("{COOKIE}={id}; HttpOnly; SameSite=Strict; Path=/; Max-Age={}", session::SESSION_MS / 1000); |
| 608 |
let mut r = Redirect::to("/").into_response(); |
| 609 |
r.headers_mut().insert(header::SET_COOKIE, HeaderValue::from_str(&cookie).unwrap()); |
| 610 |
r |
| 611 |
} |
| 612 |
|
| 613 |
async fn signout(State(st): State<RadarState>, headers: HeaderMap, Form(f): Form<CsrfForm>) -> Response { |
| 614 |
if let Ok(o) = owner(&st, &headers, "live").await { |
| 615 |
if check_form(&st, &o, &headers, &f._csrf).is_none() { |
| 616 |
if let Some(id) = cookie(&headers) { |
| 617 |
st.sessions().close(&id); |
| 618 |
} |
| 619 |
} |
| 620 |
} |
| 621 |
signin_page("Signed out.", StatusCode::OK) |
| 622 |
} |
| 623 |
|
| 624 |
|
| 625 |
|
| 626 |
|
| 627 |
|
| 628 |
async fn events(State(st): State<RadarState>, headers: HeaderMap) -> Response { |
| 629 |
if let Err(r) = owner(&st, &headers, "live").await { |
| 630 |
return r; |
| 631 |
} |
| 632 |
let rx = st.app.daemon.ledger.subscribe(); |
| 633 |
let stream = tokio_stream::wrappers::BroadcastStream::new(rx).filter_map(|e| async move { |
| 634 |
let e = e.ok()?; |
| 635 |
let data = serde_json::json!({ "seq": e.seq, "kind": e.kind, "actor": e.actor, "subject": e.subject }); |
| 636 |
Some(Ok::<_, std::convert::Infallible>(SseEvent::default().data(data.to_string()))) |
| 637 |
}); |
| 638 |
Sse::new(stream).keep_alive(KeepAlive::new().interval(Duration::from_secs(15))).into_response() |
| 639 |
} |