johnandersen777.bsky.social/knotless
git that lives in your Atmosphere account
Forked from chadtmiller.com/knotless
0
0
//! Usage, counted and read: the pages' hits, the requests, what the index
//! says the network did, and `knotless admin stats` showing it to an
//! operator, an account on the PDS. The index is filled by hand.
#[path = "../../knotless/tests/common/mod.rs"]
mod common;
use common::Sandbox;
use knotless::records::nsid;
use knotless::usage::{Report, Stats};
use knotless_mcp::http::Event;
use knotless_serve::gateway::Gateway;
use knotless_serve::index::Index;
use knotless_serve::logs::{self, Logs};
use knotless_serve::usage::{Usage, counted, mcp_events};
use knotless_serve::{App, router};
use serde_json::json;
use std::sync::Arc;
use std::time::{Duration, SystemTime};
/// `knotless <args>` as `who`, at the appview at `base`.
fn cli(who: &mut Sandbox, base: &str, args: &[&str]) -> (i32, String) {
who.set("KNOTLESS_APPVIEW", base);
who.cli(args, b"")
}
/// An appview, unserved yet, at a port of its own, as a gateway too: its
/// `did:web` is what its operator's service tokens are for. The account
/// `operator` signs in to runs it.
struct Appview {
base: String,
listener: tokio::net::TcpListener,
operator: Sandbox,
}
impl Appview {
fn new(rt: &tokio::runtime::Runtime) -> Appview {
let listener = rt
.block_on(tokio::net::TcpListener::bind("127.0.0.1:0"))
.unwrap();
let port = listener.local_addr().unwrap().port();
let mut operator = Sandbox::new();
operator.set("KNOTLESS_PASSWORD", "");
let password = operator.account.password.clone();
let handle = operator.account.handle.clone();
let login = operator.cli(&["login", &handle, "--password-stdin"], password.as_bytes());
assert_eq!(login.0, 0, "{}", login.1);
Appview {
base: format!("http://localhost:{port}"),
listener,
operator,
}
}
fn app(&self, index: Option<Index>) -> Arc<App> {
let mut server = Sandbox::with(self.operator.account.clone());
server.set("KNOTLESS_DID_WEB_SCHEME", "http");
server.set("KNOTLESS_ADMIN_DIDS", &self.operator.account.did);
let ctx = common::ctx(&server);
let gateway = Gateway::new(&self.base).unwrap();
App::build(ctx, Duration::from_secs(60), Some(gateway), index)
}
/// Serves `app`, counted if it counts, and says where, and as whom.
fn serve(self, rt: &tokio::runtime::Runtime, app: Arc<App>) -> (String, Sandbox) {
let routes = match app.usage.clone() {
Some(usage) => counted(router(app), usage),
None => router(app),
};
let listener = self.listener;
rt.spawn(async move { axum::serve(listener, routes).await.unwrap() });
(self.base, self.operator)
}
}
const BROWSER: &str = "Mozilla/5.0 (Macintosh) Safari/605";
#[test]
fn hits_requests_and_the_network() {
let rt = tokio::runtime::Runtime::new().unwrap();
let dir = tempfile::tempdir().unwrap();
let now = humantime::format_rfc3339_seconds(SystemTime::now()).to_string();
let index = Index::open(&dir.path().join("index.db")).unwrap();
let repo = json!({"uri": "at://did:a/dev.knotless.repo/r", "key": "k1"});
for (did, collection, rkey, value) in [
(
"did:a",
nsid::REPO,
"r",
json!({"key": "k1", "createdAt": now}),
),
(
"did:a",
nsid::CLAIM,
"c",
json!({"repo": "k1", "name": "refs/heads/main", "to": "abc", "createdAt": now}),
),
(
"did:b",
nsid::PULL,
"p",
json!({"repo": repo, "createdAt": now}),
),
(
"did:b",
nsid::ISSUE,
"i",
json!({"repo": repo, "createdAt": now}),
),
(
"did:b",
nsid::COMMENT,
"c",
json!({"repo": repo, "createdAt": now}),
),
(
"did:b",
nsid::STATE,
"s",
json!({"repo": repo, "createdAt": now}),
),
// Dated past the days asked about: counted in the network alone.
(
"did:c",
nsid::COMMENT,
"later",
json!({"repo": repo, "createdAt": "2999-01-01T00:00:00Z"}),
),
(
"did:run",
nsid::CHECK,
"x",
json!({"repo": repo, "ref": "refs/heads/main", "commit": "abc", "startedAt": now}),
),
] {
index.put(did, collection, rkey, &value).unwrap();
}
let appview = Appview::new(&rt);
let mut app = appview.app(Some(index));
let usage = Usage::open(&dir.path().join("usage.db")).unwrap();
Arc::get_mut(&mut app).unwrap().usage = Some(Arc::new(usage));
// What the handlers tell of: two accounts sign in to MCP, one goes on to
// call a tool, twice; one delegates to the gateway and pushes.
let mcp = mcp_events(app.usage.clone().unwrap());
let ms = Duration::from_millis;
for (name, subject, error, took, did) in [
("mcp.signin", "", None, ms(9), "did:x"),
("mcp.signin", "", None, ms(9), "did:y"),
("mcp.tool", "read_file", None, ms(100), "did:x"),
(
"mcp.tool",
"read_file",
Some("no such file"),
ms(300),
"did:x",
),
] {
let did = Some(did);
mcp(Event {
name,
subject,
ok: error.is_none(),
took,
did,
error,
});
}
app.event("gateway.delegated", "", true, Duration::ZERO, Some("did:x"));
app.event("gateway.push", "", true, ms(40), Some("did:x"));
app.event("gateway.served", "", true, ms(40), Some("did:x"));
let (base, mut operator) = appview.serve(&rt, app);
let http = reqwest::Client::new();
let hit_from = |from: Option<&str>, agent: &str, gpc: bool, body: &str| {
let mut req = http
.post(format!("{base}/_hit"))
.header("user-agent", agent)
.body(body.to_string());
if let Some(from) = from {
req = req.header("x-forwarded-for", from);
}
if gpc {
req = req.header("sec-gpc", "1");
}
rt.block_on(async { req.send().await.unwrap().status().as_u16() })
};
let hit = |agent: &str, gpc: bool, body: &str| hit_from(Some("1.2.3.4"), agent, gpc, body);
let page = r#"{"route": "/[authority]/[repo]", "referrer": "https://bsky.app/profile/x"}"#;
assert_eq!(hit(BROWSER, false, page), 204);
assert_eq!(hit(BROWSER, false, r#"{"route": "/"}"#), 204);
// Not counted: one asking not to be, and a crawler.
assert_eq!(hit(BROWSER, true, page), 204);
assert_eq!(hit("Googlebot/2.1", false, page), 204);
assert_eq!(hit("git/2.45.0", false, page), 204);
assert_eq!(hit("curl/8.7.1", false, page), 204);
// Visitors apart: one over IPv6, and one with no address at all.
assert_eq!(hit_from(Some("2001:db8::1"), BROWSER, false, page), 204);
assert_eq!(hit_from(None, BROWSER, false, page), 204);
assert_eq!(hit(BROWSER, false, "nonsense"), 400);
assert_eq!(hit(BROWSER, false, r#"{"route": "no slash"}"#), 400);
let listed = rt.block_on(async {
let url = format!("{base}/xrpc/dev.knotless.listRepos");
let res = http
.get(url)
.header("user-agent", "knotless/0.1.0")
.send()
.await;
res.unwrap().status().as_u16()
});
assert_eq!(listed, 200);
// A crawler, counted by its agent.
let crawled = rt.block_on(async {
let url = format!("{base}/xrpc/dev.knotless.listRepos");
let res = http
.get(url)
.header("user-agent", "curl/8.7.1")
.send()
.await;
res.unwrap().status().as_u16()
});
assert_eq!(crawled, 200);
let unauthorized = rt.block_on(async {
http.get(format!("{base}/xrpc/dev.knotless.getUsage"))
.send()
.await
.unwrap()
.status()
.as_u16()
});
assert_eq!(unauthorized, 401);
let (code, json) = cli(
&mut operator,
&base,
&["admin", "stats", "--days", "2", "--json"],
);
assert_eq!(code, 0, "{json}");
let report: Report = serde_json::from_str(&json).unwrap();
assert_eq!(report.days.len(), 2);
let today = report.days.last().unwrap();
assert_eq!(today.day, now[..10]);
assert_eq!(
(today.visitors, today.views, today.requests, today.errors),
(3, 4, 3, 0)
);
assert_eq!(
(
today.repos,
today.pushes,
today.pulls,
today.issues,
today.comments,
today.checks
),
(1, 1, 1, 1, 1, 1)
);
assert_eq!(today.authors, 2);
let network = report.network.unwrap();
assert_eq!(
(
network.accounts,
network.repos,
network.authors_week,
network.authors_month
),
(3, 1, 3, 3)
);
// The listing, and the appview's DID document the CLI read to know
// whom its token is for.
let routes: Vec<(&str, &str, u64)> = report
.routes
.iter()
.map(|r| (r.surface.as_str(), r.route.as_str(), r.requests))
.collect();
assert!(
routes.contains(&("xrpc", "/xrpc/dev.knotless.listRepos", 2)),
"{routes:?}"
);
assert!(
routes.contains(&("other", "/.well-known/did.json", 1)),
"{routes:?}"
);
assert_eq!(report.referrers[0].name, "bsky.app");
assert_eq!(report.clients[0].name, "cli");
let event = |name: &str, subject: &str| {
let e = report
.events
.iter()
.find(|e| e.name == name && e.subject == subject);
e.unwrap_or_else(|| panic!("{name} {subject}: {:?}", report.events))
.clone()
};
let tool = event("mcp.tool", "read_file");
assert_eq!(
(tool.calls, tool.failed, tool.mean_ms, tool.accounts),
(2, 1, 200, 1)
);
assert_eq!(event("mcp.signin", "").accounts, 2);
let failure = &report.failures[0];
assert_eq!(
(failure.subject.as_str(), failure.error.as_str(), failure.n),
("read_file", "no such file", 1)
);
assert_eq!(report.agents[0].name, "curl/8.7.1");
assert!(event("cli.version", env!("CARGO_PKG_VERSION")).calls >= 1);
let funnels: Vec<(&str, &str, u64, u64)> = report
.funnels
.iter()
.map(|f| (f.from.as_str(), f.to.as_str(), f.started, f.reached))
.collect();
assert_eq!(
funnels,
[
("gateway.delegated", "gateway.served", 1, 1),
("mcp.signin", "mcp.tool", 2, 1)
]
);
// Anyone's: the days and the network, without what only an operator reads.
let stats: Stats = rt.block_on(async {
let res = http
.get(format!("{base}/xrpc/dev.knotless.getStats"))
.send()
.await
.unwrap();
assert_eq!(
res.headers()["cache-control"].to_str().unwrap(),
"public, max-age=300"
);
res.json().await.unwrap()
});
assert_eq!(stats.days.len(), 30);
let today = stats.days.last().unwrap();
assert_eq!(today.day, now[..10]);
assert_eq!(
(today.visitors, today.views, today.pushes, today.authors),
(3, 4, 1, 2)
);
assert_eq!(stats.network.unwrap().accounts, 3);
let (code, text) = cli(&mut operator, &base, &["admin", "stats"]);
assert_eq!(code, 0, "{text}");
assert!(
text.starts_with("network\t3 accounts, 1 repositories"),
"{text}"
);
assert!(
text.contains("route\txrpc\tGET /xrpc/dev.knotless.listRepos\t2 requests"),
"{text}"
);
assert_ne!(
cli(&mut operator, &base, &["admin", "stats", "--days", "x"]).0,
0
);
assert_ne!(
cli(&mut operator, &base, &["admin", "stats", "--weeks"]).0,
0
);
}
#[test]
fn a_server_counting_nothing() {
let rt = tokio::runtime::Runtime::new().unwrap();
let appview = Appview::new(&rt);
let app = appview.app(None);
let (base, mut operator) = appview.serve(&rt, app);
let http = reqwest::Client::new();
let hit = rt.block_on(async {
let res = http
.post(format!("{base}/_hit"))
.body(r#"{"route": "/"}"#)
.send()
.await;
res.unwrap().status().as_u16()
});
assert_eq!(hit, 404);
// Still a report, of days with nothing in them.
let (code, json) = cli(
&mut operator,
&base,
&["admin", "stats", "--days", "1", "--json"],
);
assert_eq!(code, 0, "{json}");
let report: Report = serde_json::from_str(&json).unwrap();
assert_eq!((report.days.len(), report.network), (1, None));
let stats: Stats = rt.block_on(async {
let url = format!("{base}/xrpc/dev.knotless.getStats");
http.get(url).send().await.unwrap().json().await.unwrap()
});
assert_eq!((stats.days.len(), stats.network), (30, None));
// Nor are its logs kept: reading them fails.
let (code, _) = cli(&mut operator, &base, &["admin", "logs", "--level", "warn"]);
assert_ne!(code, 0);
// And an answer that is not one says what it was.
let (code, _) = cli(
&mut operator,
&format!("{base}/nowhere"),
&["admin", "stats"],
);
assert_ne!(code, 0);
}
/// The lines serve logged, read back by an operator: a request's by its
/// trace, or the last of a level; asking for nothing is refused.
#[test]
fn logs_read_back() {
let rt = tokio::runtime::Runtime::new().unwrap();
let appview = Appview::new(&rt);
let dir = tempfile::tempdir().unwrap();
let logs = Arc::new(Logs::open(&dir.path().join("logs.db"), logs::KEEP).unwrap());
for (level, message, trace) in [
("INFO", "request", "t1"),
("WARN", "outgoing", "t1"),
("INFO", "request", "t2"),
] {
let line = json!({"timestamp": "2999-01-01T00:00:00Z", "level": level,
"message": message, "trace_id": trace});
logs.keep(format!("{line}\n").as_bytes());
}
let mut app = appview.app(None);
Arc::get_mut(&mut app).unwrap().logs = Some(logs);
let (base, mut operator) = appview.serve(&rt, app);
let (code, text) = cli(&mut operator, &base, &["admin", "logs", "--trace", "t1"]);
assert_eq!(code, 0, "{text}");
let lines: Vec<serde_json::Value> = text
.lines()
.map(|l| serde_json::from_str(l).unwrap())
.collect();
let messages: Vec<&str> = lines
.iter()
.map(|l| l["message"].as_str().unwrap())
.collect();
assert_eq!(messages, ["request", "outgoing"]);
let (code, text) = cli(
&mut operator,
&base,
&["admin", "logs", "--level", "warn", "--limit", "5"],
);
assert_eq!(code, 0, "{text}");
assert!(text.contains("\"outgoing\""), "{text}");
assert!(!text.contains("\"t2\""), "{text}");
// Asking for nothing is refused.
let (code, _) = cli(&mut operator, &base, &["admin", "logs"]);
assert_ne!(code, 0);
assert_ne!(
cli(&mut operator, &base, &["admin", "logs", "--since"]).0,
0
);
assert_ne!(
cli(&mut operator, &base, &["admin", "logs", "--trace"]).0,
0
);
}