server/lib/tests/ai_models_through_the_booted_doors.rs
//! ★ **AI models and blocked pickles through a booted server's doors** (lane OX-WIRED4).
//!
//! `ai_models_on_the_intake_chain` drives the chain [`Holger::intake_chain`] builds, in
//! process. Here the same config is BOOTED (`Holger::start`) and driven only through its
//! HTTP doors, by static-token principals with duties (`ida` Importer, `bob` and `carol`
//! Security Approvers):
//!
//! * a GGUF, a Safetensors and an ONNX file go up `PUT /incoming/…`, and
//! `POST /-/intake/advance` moves each to SCANNED; their model metadata is in the
//! server's property store, so `GET /-/search?property=model.architecture:<arch>` finds
//! each by it;
//! * a pickle, a `.pt` zip holding `data.pkl` and a pickle renamed `.safetensors` go up,
//! and the advance is refused with the model scanner's frozen code; each stays
//! INGESTED (`GET /-/intake/state`), and production serves none of them;
//! * `models.allow_pickle` naming an undeclared repository refuses the BOOT
//! (`holger-model-8`), not only the chain's construction.
// A server's test, never a gateway build's (lane HOL-CLEANUP): a gateway build is a
// downstream, and it scans, and a downstream runs no scanners (holger-downstream-1).
#![cfg(not(feature = "gateway"))]
use std::io::Write;
use std::time::Duration;
use holger_server_lib::scanner::model::fixtures;
use holger_server_lib::{read_ron_config, wire_holger, Holger};
use traits::test_support::hold_port;
fn token(who: &str) -> String {
format!("{who}-token-of-sufficient-length-for-holger")
}
struct Running {
_data: tempfile::TempDir,
_cfg: tempfile::NamedTempFile,
_holger: Holger,
base: String,
}
fn ron(d: &str, grpc: u16, http: u16, models: &str) -> String {
let repo = |name: &str| {
format!(
r#"(ron_name: "{name}", ron_repo_type: "raw", ron_upstreams: [], ron_in: None,
ron_out: Some((ron_storage_endpoint: "storage", ron_exposed_endpoint: "door")),
ron_data_dir: Some("{d}/{name}")),"#
)
};
let methods: Vec<String> = ["ida", "bob", "carol"]
.iter()
.map(|p| {
format!(
r#"StaticToken(subject: "{p}", token: Some("{}"))"#,
token(p)
)
})
.collect();
format!(
r#"(
repositories: [ {} {} {} {} ],
exposed_endpoints: [ (ron_name: "door", ron_url: "127.0.0.1:{grpc}", ron_http_url: Some("127.0.0.1:{http}")) ],
storage_endpoints: [ (ron_name: "storage", ron_storage_type: "znippy", ron_path: "{d}") ],
audit: (dir: "{d}/audit", enabled: true),
auth: Some((
methods: [ {methods} ],
duties: {{ "ida": [Importer], "bob": [SecurityApprover], "carol": [SecurityApprover] }},
)),
promotion: Some((
ladder_name: "intake",
stages: [
(name: "incoming", repo: "incoming"), (name: "quarantine", repo: "quarantine"),
(name: "approved", repo: "approved"), (name: "production", repo: "production"),
],
intake: Some((rules: (sandbox: off),
incoming: "incoming", quarantine: "quarantine", approved: "approved", production: "production",
approvals: 2,
scan: Some((required: [model])),
)),
)),
{models}
)"#,
repo("incoming"),
repo("quarantine"),
repo("approved"),
repo("production"),
methods = methods.join(", "),
)
}
/// Boot the config with `models`; the error of the first step that refuses.
fn boot(models: &str) -> Result<Running, String> {
let mut last = String::new();
for _ in 0..8 {
let (grpc, http) = (hold_port(), hold_port());
let data = tempfile::tempdir().unwrap();
let mut cfg = tempfile::NamedTempFile::new().unwrap();
cfg.write_all(ron(&data.path().display().to_string(), grpc, http, models).as_bytes())
.unwrap();
let go = || -> Result<Holger, String> {
let mut h = read_ron_config(cfg.path()).map_err(|e| format!("config: {e:#}"))?;
h.instantiate_backends()
.map_err(|e| format!("backends: {e:#}"))?;
wire_holger(&mut h).map_err(|e| format!("wire: {e:#}"))?;
h.start().map_err(|e| format!("start: {e:#}"))?;
Ok(h)
};
match go() {
Ok(h) => {
return Ok(Running {
_data: data,
_cfg: cfg,
_holger: h,
base: format!("http://127.0.0.1:{http}"),
})
}
// A refusal by name is the answer, not a port race.
Err(e) if e.contains("[holger-") || e.starts_with("config") => return Err(e),
Err(e) => last = e,
}
}
panic!("server never started; last error: {last}");
}
fn client() -> reqwest::Client {
let _ = rustls::crypto::ring::default_provider().install_default();
reqwest::Client::builder()
.tls_certs_only(
webpki_root_certs::TLS_SERVER_ROOT_CERTS
.iter()
.filter_map(|c| reqwest::Certificate::from_der(c).ok()),
)
.build()
.expect("reqwest client")
}
async fn wait_healthy(c: &reqwest::Client, base: &str) {
for _ in 0..250 {
if let Ok(r) = c.get(format!("{base}/healthz")).send().await {
if r.status() == 200 {
return;
}
}
tokio::time::sleep(Duration::from_millis(20)).await;
}
panic!("holger never became healthy at {base}");
}
async fn put(c: &reqwest::Client, r: &Running, name: &str, bytes: Vec<u8>) {
let resp = c
.put(format!("{}/incoming/{name}", r.base))
.bearer_auth(token("ida"))
.body(bytes)
.send()
.await
.expect("the door answers");
let st = resp.status().as_u16();
assert_eq!(
st,
201,
"{name} goes into incoming: {}",
resp.text().await.unwrap_or_default()
);
}
/// `POST /-/intake/advance` as the importer: (status, body).
async fn advance(c: &reqwest::Client, r: &Running, name: &str) -> (u16, String) {
let resp = c
.post(format!(
"{}/-/intake/advance?name={name}&verb=advance",
r.base
))
.bearer_auth(token("ida"))
.send()
.await
.expect("the door answers");
(
resp.status().as_u16(),
resp.text().await.unwrap_or_default(),
)
}
/// `GET /-/intake/state` as an approver: the state's name.
async fn state(c: &reqwest::Client, r: &Running, name: &str) -> String {
let resp = c
.get(format!("{}/-/intake/state?name={name}", r.base))
.bearer_auth(token("bob"))
.send()
.await
.expect("the door answers");
assert_eq!(resp.status().as_u16(), 200);
let v: serde_json::Value = resp.json().await.unwrap();
v["state"].as_str().unwrap_or_default().to_string()
}
#[tokio::test(flavor = "multi_thread")]
async fn models_are_scanned_and_found_by_their_metadata_through_the_doors() {
let r = boot("").expect("the server boots");
let c = client();
wait_healthy(&c, &r.base).await;
for (name, bytes) in [
("llm/tiny.gguf", fixtures::gguf()),
("bert/model.safetensors", fixtures::safetensors()),
("mlp/tiny.onnx", fixtures::onnx()),
] {
put(&c, &r, name, bytes).await;
let (st, body) = advance(&c, &r, name).await;
assert_eq!(st, 200, "{name}: {body}");
let v: serde_json::Value = serde_json::from_str(&body).unwrap();
assert_eq!(v["state"], "SCANNED", "{name}: {v}");
assert_eq!(state(&c, &r, name).await, "SCANNED", "{name}");
}
for (key, value, want) in [
("model.format", "onnx", "mlp/tiny.onnx"),
("model.architecture", "llama", "llm/tiny.gguf"),
("model.architecture", "bert", "bert/model.safetensors"),
] {
let resp = c
.get(format!(
"{}/-/search?property={key}:{value}&repo=incoming",
r.base
))
.bearer_auth(token("bob"))
.send()
.await
.unwrap();
assert_eq!(resp.status().as_u16(), 200, "{key}={value}");
let v: serde_json::Value = resp.json().await.unwrap();
let names: Vec<&str> = v["artifacts"]
.as_array()
.unwrap_or_else(|| panic!("{v}"))
.iter()
.filter_map(|h| h["id"]["name"].as_str())
.collect();
assert_eq!(names, vec![want], "{key}={value}: {v}");
}
}
#[tokio::test(flavor = "multi_thread")]
async fn the_pickle_family_is_refused_at_the_door_with_its_frozen_code() {
let r = boot("").expect("the server boots");
let c = client();
wait_healthy(&c, &r.base).await;
let pt = zip_entry::fixture::zip(&[
("archive/data.pkl", &fixtures::pickle()[..]),
("archive/version", &b"3\n"[..]),
]);
for (name, bytes, code) in [
("tiny.pkl", fixtures::pickle(), "holger-model-1"),
("fake.pt", pt, "holger-model-2"),
("renamed.safetensors", fixtures::pickle(), "holger-model-1"),
(
"lying.safetensors",
fixtures::safetensors_lying_offsets(),
"holger-model-4",
),
] {
put(&c, &r, name, bytes).await;
let (st, body) = advance(&c, &r, name).await;
assert!(st >= 400, "{name} is refused at SCANNED: {st} {body}");
assert!(body.contains(code), "{name} names {code}: {body}");
assert_eq!(state(&c, &r, name).await, "INGESTED", "{name} did not move");
let served = c
.get(format!("{}/production/{name}", r.base))
.bearer_auth(token("bob"))
.send()
.await
.unwrap();
assert!(
served.status().as_u16() >= 400,
"production serves nothing of {name}: {}",
served.status()
);
}
}
#[tokio::test(flavor = "multi_thread")]
async fn allow_pickle_naming_an_undeclared_repository_refuses_the_boot() {
let err = boot(r#"models: (allow_pickle: ["no-such-repo"]),"#)
.err()
.expect("the boot is refused");
assert!(err.contains("[holger-model-8]"), "{err}");
let r = boot(r#"models: (allow_pickle: ["incoming"]),"#).expect("a declared repository boots");
let c = client();
wait_healthy(&c, &r.base).await;
put(&c, &r, "approved-research.pkl", fixtures::pickle()).await;
let (st, body) = advance(&c, &r, "approved-research.pkl").await;
assert_eq!(
st, 200,
"a pickle in an allowed repository is a warning, not a block: {body}"
);
}