93 lines
3.3 KiB
Rust
93 lines
3.3 KiB
Rust
//! Live validation of the grounded surface path (Stage 5d) for absence-based CRA
|
|
//! controls. Ignored (hits api-dev CRA catalog + LiteLLM). Run:
|
|
//! export ... (LITELLM_* + BREAKPILOT_BASE_URL)
|
|
//! cargo test -p compliance-agent --test grounded_surface_live -- --ignored --nocapture
|
|
//!
|
|
//! Builds a fixture whose code surfaces trigger several absence-based controls
|
|
//! (no rate limiting, no security logging, unverified update) and checks that the
|
|
//! grounded checker produces control-tagged findings.
|
|
|
|
mod common;
|
|
|
|
use std::sync::Arc;
|
|
|
|
use compliance_agent::llm::LlmClient;
|
|
use compliance_core::config::BreakpilotConfig;
|
|
use secrecy::SecretString;
|
|
|
|
fn env(k: &str) -> String {
|
|
std::env::var(k).unwrap_or_else(|_| panic!("env {k} must be set"))
|
|
}
|
|
|
|
fn write(repo: &std::path::Path, rel: &str, body: &str) {
|
|
let p = repo.join(rel);
|
|
if let Some(parent) = p.parent() {
|
|
std::fs::create_dir_all(parent).unwrap();
|
|
}
|
|
std::fs::write(p, body).unwrap();
|
|
}
|
|
|
|
#[tokio::test]
|
|
#[ignore = "live: api-dev CRA catalog + LiteLLM"]
|
|
async fn grounded_surface_flags_absence_controls() {
|
|
let llm = Arc::new(LlmClient::new(
|
|
env("LITELLM_URL"),
|
|
SecretString::from(env("LITELLM_API_KEY")),
|
|
env("LITELLM_MODEL"),
|
|
env("LITELLM_EMBED_MODEL"),
|
|
));
|
|
let mut config = common::dev_config("mongodb://unused".into(), "grounded".into());
|
|
config.breakpilot = BreakpilotConfig {
|
|
base_url: Some(env("BREAKPILOT_BASE_URL")),
|
|
token: None,
|
|
snapshot_dir: std::env::temp_dir()
|
|
.join("grounded-snap")
|
|
.to_string_lossy()
|
|
.into_owned(),
|
|
semantic_mapping: false,
|
|
grounded_control_checks: true,
|
|
};
|
|
|
|
let repo = std::env::temp_dir().join("grounded-fixture-repo");
|
|
let _ = std::fs::remove_dir_all(&repo);
|
|
// cra-ai-11: login endpoint with no rate limiting / lockout
|
|
write(
|
|
&repo,
|
|
"app/auth.py",
|
|
"@app.route('/login', methods=['POST'])\ndef login():\n u = request.form['username']\n p = request.form['password']\n if authenticate(u, p):\n return redirect('/')\n return 'bad credentials', 401\n",
|
|
);
|
|
// cra-ai-24: privileged admin action with no security/audit logging
|
|
write(
|
|
&repo,
|
|
"app/admin.py",
|
|
"@app.route('/admin/delete_user', methods=['POST'])\ndef admin_delete_user():\n uid = request.form['uid']\n db.users.delete_one({'_id': uid})\n return 'ok', 200\n",
|
|
);
|
|
// cra-ai-28/29/30: firmware update applied without signature / checksum verification
|
|
write(
|
|
&repo,
|
|
"app/updater.py",
|
|
"def apply_firmware_update(url):\n blob = download(url)\n install_firmware(blob)\n reboot_device()\n",
|
|
);
|
|
|
|
let findings =
|
|
compliance_agent::controls::grounded_surface_findings(&config, llm, &repo, "repo-grounded")
|
|
.await;
|
|
|
|
println!("\n=== Grounded surface findings ({}) ===", findings.len());
|
|
for f in &findings {
|
|
println!(
|
|
" {:24} {}:{:?} {}",
|
|
f.control_refs.join(","),
|
|
f.file_path.as_deref().unwrap_or(""),
|
|
f.line_number,
|
|
f.title
|
|
);
|
|
}
|
|
let _ = std::fs::remove_dir_all(&repo);
|
|
|
|
assert!(
|
|
!findings.is_empty(),
|
|
"expected the grounded pass to flag at least one absence-based control"
|
|
);
|
|
}
|