compliance-scanner-agent/compliance-agent/tests/grounded_surface_live.rs

//! 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"
    );
}