liminal-server 0.25.0

use super::*;
use crate::auth_lys::live::drafts::tests::{Issuer, TargetStore, limits};

type TestResult = Result<(), Box<dyn std::error::Error>>;

fn adapter() -> Result<std::sync::Arc<LysPassAdapter>, lys_pass::Error> {
    Ok(std::sync::Arc::new(LysPassAdapter::from_keys_json(
        "{\"keys\":[]}",
        "https://issuer.example".to_owned(),
    )?))
}

#[tokio::test]
async fn unauthenticated_admin_request_cannot_pull_the_connector() -> TestResult {
    let dir = tempfile::tempdir()?;
    let issuer = Issuer::new(vec![])?;
    let puller = DraftPuller::new(
        &issuer.api,
        TargetStore::new(dir.path().join("target")),
        limits()?,
    )?;
    let mut route = AdminDrafts::new(
        puller,
        adapter()?,
        "network-a".to_owned(),
        AdminLimits {
            credential_bytes: limits()?.metadata_bytes,
            body_bytes: limits()?.metadata_bytes,
        },
    )?;
    let request = AdminRequest {
        method: "POST",
        path: PULL_PATH,
        authorization: None,
        body: b"",
        at_ms: 100_000,
    };
    let response = route.handle(request, "connector-fixture").await?;
    assert_eq!(response.status, 401);
    assert_eq!(response.body["refusal"], "draft_pull_pass_missing");
    assert_eq!(issuer.calls()?, 0);
    issuer.finish()
}

#[tokio::test]
async fn caller_cannot_replace_network_or_request_automatic_authority() -> TestResult {
    let dir = tempfile::tempdir()?;
    let issuer = Issuer::new(vec![])?;
    let puller = DraftPuller::new(
        &issuer.api,
        TargetStore::new(dir.path().join("target")),
        limits()?,
    )?;
    let mut route = AdminDrafts::new(
        puller,
        adapter()?,
        "network-a".to_owned(),
        AdminLimits {
            credential_bytes: limits()?.metadata_bytes,
            body_bytes: limits()?.metadata_bytes,
        },
    )?;
    for (method, path, body, status) in [
        ("GET", PULL_PATH, b"".as_slice(), 405),
        (
            "POST",
            "/admin/drafts/pull?network=other",
            b"".as_slice(),
            404,
        ),
        (
            "POST",
            PULL_PATH,
            b"{\"network\":\"other\"}".as_slice(),
            400,
        ),
    ] {
        let request = AdminRequest {
            method,
            path,
            authorization: None,
            body,
            at_ms: 100_000,
        };
        assert_eq!(
            route.handle(request, "connector-fixture").await?.status,
            status
        );
    }
    assert_eq!(issuer.calls()?, 0);
    issuer.finish()
}

#[tokio::test]
async fn forged_bearer_is_refused_before_any_lys_request() -> TestResult {
    let dir = tempfile::tempdir()?;
    let issuer = Issuer::new(vec![])?;
    let puller = DraftPuller::new(
        &issuer.api,
        TargetStore::new(dir.path().join("target")),
        limits()?,
    )?;
    let mut route = AdminDrafts::new(
        puller,
        adapter()?,
        "network-a".to_owned(),
        AdminLimits {
            credential_bytes: limits()?.metadata_bytes,
            body_bytes: limits()?.metadata_bytes,
        },
    )?;
    let request = AdminRequest {
        method: "POST",
        path: PULL_PATH,
        authorization: Some("Bearer forged-fixture"),
        body: b"{}",
        at_ms: 100_000,
    };
    let response = route.handle(request, "connector-fixture").await?;
    assert_eq!(response.status, 401);
    assert!(!response.body.to_string().contains("forged-fixture"));
    assert_eq!(issuer.calls()?, 0);
    issuer.finish()
}

fn signed_operator(
    issuer: &str,
    network: &str,
    mode: &str,
    expires: u64,
) -> Result<(std::sync::Arc<LysPassAdapter>, String), Box<dyn std::error::Error>> {
    let identity = lys_core::keys::Ed25519Identity::ephemeral();
    let keys = serde_json::json!({"keys":[{"kty":"OKP","crv":"Ed25519",
        "kid":"draft-test","x":base64(&identity.public_key_bytes()),"alg":"EdDSA","use":"sig"}]});
    let adapter = LysPassAdapter::from_keys_json(&keys.to_string(), issuer.to_owned())?;
    let claims = serde_json::json!({"iss":issuer,"sub":"person-operator","aud":"liminal",
        "iat":0,"exp":expires,"holder":{"kind":"person","id":"person-operator","responsible":null},
        "rights":[{"resource":{"kind":"liminal.network","id":network},
            "actions":[PULL_ACTION],"mode":mode,"grant":"grant-operator"}]});
    let header = base64(br#"{"alg":"EdDSA","typ":"JWT","kid":"draft-test"}"#);
    let signed = format!("{header}.{}", base64(&serde_json::to_vec(&claims)?));
    let token = format!("{signed}.{}", base64(&identity.sign(signed.as_bytes())));
    Ok((std::sync::Arc::new(adapter), token))
}

pub fn base64(bytes: &[u8]) -> String {
    const ALPHABET: &[u8; 64] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-_";
    let mut encoded = String::new();
    for chunk in bytes.chunks(3) {
        let first = chunk[0];
        let second = chunk.get(1).copied().unwrap_or(0);
        let third = chunk.get(2).copied().unwrap_or(0);
        encoded.push(char::from(ALPHABET[usize::from(first >> 2)]));
        encoded.push(char::from(
            ALPHABET[usize::from((first & 3) << 4 | second >> 4)],
        ));
        if chunk.len() > 1 {
            encoded.push(char::from(
                ALPHABET[usize::from((second & 15) << 2 | third >> 6)],
            ));
        }
        if chunk.len() > 2 {
            encoded.push(char::from(ALPHABET[usize::from(third & 63)]));
        }
    }
    encoded
}

#[tokio::test]
async fn signed_operator_is_checked_live_and_only_machine_pass_pulls() -> TestResult {
    let dir = tempfile::tempdir()?;
    let issuer = Issuer::new(vec![
        (
            200,
            serde_json::json!({"revision":1,"degraded":null,
        "results":[{"allowed":true,"grant":"grant-operator","mode":"outright",
            "path":[],"via":null,"refusal":null,"reason":null,"degraded":null}]}),
        ),
        (200, serde_json::json!({"drafts":[],"next":null,"total":0})),
    ])?;
    let (adapter, token) = signed_operator(&issuer.api, "network-a", "outright", 200)?;
    // The live check is asked with the app's own credential; the person is the subject.
    issuer.permit_check("connector-fixture")?;
    let puller = DraftPuller::new(
        &issuer.api,
        TargetStore::new(dir.path().join("target")),
        limits()?,
    )?;
    let mut route = AdminDrafts::new(
        puller,
        adapter,
        "network-a".to_owned(),
        AdminLimits {
            credential_bytes: limits()?.metadata_bytes,
            body_bytes: limits()?.metadata_bytes,
        },
    )?;
    let header = format!("Bearer {token}");
    let response = route
        .handle(
            AdminRequest {
                method: "POST",
                path: PULL_PATH,
                authorization: Some(&header),
                body: b"{}",
                at_ms: 100_000,
            },
            "connector-fixture",
        )
        .await?;
    assert_eq!(response.status, 200);
    assert_eq!(response.body["pages"], 1);
    assert_eq!(issuer.calls()?, 2);
    issuer.finish()
}

#[tokio::test]
async fn wrong_network_held_and_expired_rights_cannot_trigger_the_connector() -> TestResult {
    for (network, mode, expires, status) in [
        ("network-b", "outright", 200, 403),
        ("network-a", "by_draft", 200, 403),
        ("network-a", "outright", 100, 401),
    ] {
        let dir = tempfile::tempdir()?;
        let issuer = Issuer::new(vec![])?;
        let (adapter, token) = signed_operator(&issuer.api, network, mode, expires)?;
        let puller = DraftPuller::new(
            &issuer.api,
            TargetStore::new(dir.path().join("target")),
            limits()?,
        )?;
        let mut route = AdminDrafts::new(
            puller,
            adapter,
            "network-a".to_owned(),
            AdminLimits {
                credential_bytes: limits()?.metadata_bytes,
                body_bytes: limits()?.metadata_bytes,
            },
        )?;
        let header = format!("Bearer {token}");
        assert_eq!(
            route
                .handle(
                    AdminRequest {
                        method: "POST",
                        path: PULL_PATH,
                        authorization: Some(&header),
                        body: b"",
                        at_ms: 100_000
                    },
                    "connector-fixture"
                )
                .await?
                .status,
            status
        );
        assert_eq!(issuer.calls()?, 0);
        issuer.finish()?;
    }
    Ok(())
}

#[tokio::test]
async fn live_revocation_refuses_an_unexpired_operator_pass() -> TestResult {
    let dir = tempfile::tempdir()?;
    let issuer = Issuer::new(vec![(
        200,
        serde_json::json!({"revision":2,"degraded":null,
        "results":[{"allowed":false,"grant":null,"mode":null,"path":[],"via":null,
            "refusal":"grant_revoked","reason":"The grant was withdrawn.","degraded":null}]}),
    )])?;
    let (adapter, token) = signed_operator(&issuer.api, "network-a", "outright", 200)?;
    // The live check is asked with the app's own credential; the person is the subject.
    issuer.permit_check("connector-fixture")?;
    let puller = DraftPuller::new(
        &issuer.api,
        TargetStore::new(dir.path().join("target")),
        limits()?,
    )?;
    let mut route = AdminDrafts::new(
        puller,
        adapter,
        "network-a".to_owned(),
        AdminLimits {
            credential_bytes: limits()?.metadata_bytes,
            body_bytes: limits()?.metadata_bytes,
        },
    )?;
    let header = format!("Bearer {token}");
    let response = route
        .handle(
            AdminRequest {
                method: "POST",
                path: PULL_PATH,
                authorization: Some(&header),
                body: b"",
                at_ms: 100_000,
            },
            "connector-fixture",
        )
        .await?;
    assert_eq!(response.status, 403);
    assert_eq!(response.body["refusal"], "grant_revoked");
    assert_eq!(response.body["reason"], "The grant was withdrawn.");
    assert_eq!(issuer.calls()?, 1);
    issuer.finish()
}

/// Verifies the real fixture signature; it never constructs trusted claims directly.
pub fn verified_holder(
    holder: &str,
    kind: &str,
    target: &lys_pass::Target,
) -> Result<std::sync::Arc<lys_pass::VerifiedPass>, Box<dyn std::error::Error>> {
    let identity = lys_core::keys::Ed25519Identity::from_seed(&zeroize::Zeroizing::new([41; 32]));
    let keys = serde_json::json!({"keys":[{"kty":"OKP","crv":"Ed25519","kid":"prepared-fixture","alg":"EdDSA","use":"sig","x":base64(&identity.public_key_bytes())}]});
    let claims = serde_json::json!({"iss":"https://prepared.fixture","sub":holder,"aud":"liminal","iat":0,"exp":1_000_000,"holder":{"kind":kind,"id":holder,"responsible":if kind == "agent" { Some("fixture-owner") } else { None }},"rights":[{"resource":{"kind":target.kind,"id":target.id},"actions":[target.action],"mode":"by_draft","grant":"grant-prepared"}]});
    let signed = format!(
        "{}.{}",
        base64(br#"{"alg":"EdDSA","typ":"JWT","kid":"prepared-fixture"}"#),
        base64(&serde_json::to_vec(&claims)?)
    );
    let token = format!("{signed}.{}", base64(&identity.sign(signed.as_bytes())));
    Ok(std::sync::Arc::new(lys_pass::VerifiedPass::verify(
        &token,
        &lys_pass::KeySet::from_json(&keys.to_string())?,
        "https://prepared.fixture",
        "liminal",
        1,
    )?))
}

// Two workers let sockets progress while a registry call waits for HTTP.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn native_site_pull_records_one_transition_and_a_retired_target_is_terminal() -> TestResult {
    use crate::auth_lys::live::drafts::tests::{closed, page};
    use crate::auth_lys::live::drafts::waiting::tests::{DiskFixture, held, keeper, placer};
    use crate::auth_lys::live::drafts::waiting::{AgentDelivery, WaitingLimits};
    use crate::services::lifecycle::site::executor::SiteDraftStore;
    let positive = |value| NonZeroUsize::new(value).ok_or("zero fixture bound");
    let (draft, gone) = fixture_site_drafts()?;
    let issuer = Issuer::new(vec![
        (200, page(&draft, None)),
        // One placement each for the site, its subject and its plan subject.
        (200, serde_json::json!({})),
        (200, serde_json::json!({})),
        (200, serde_json::json!({})),
        (200, closed(&draft, "executed")),
        (200, page(&draft, None)),
        (200, page(&gone, None)),
        (200, closed(&gone, "refused_on_execution")),
        (200, page(&gone, None)),
    ])?;
    let disk = DiskFixture::new()?;
    let placement = placer(&issuer)?;
    let owners = disk.owners(placement.clone())?;
    let lifecycle = disk.lifecycle(placement)?;
    crate::auth_lys::live::drafts::waiting::tests::configure_sites(
        &disk, &issuer, &owners, &lifecycle,
    )?;
    held(
        &keeper(
            owners.clone(),
            lifecycle.clone(),
            "fixture-owner",
            "person",
            &draft,
        )?,
        &draft,
    )?;
    let waiting = WaitingLimits {
        drafts: positive(8)?,
        words_bytes: positive(4096)?,
        identifier_bytes: positive(256)?,
    };
    let delivery = AgentDelivery::new(
        &owners,
        std::collections::BTreeMap::new(),
        "fixture-exchange".into(),
        waiting,
    )?;
    let mut puller = DraftPuller::new(
        &issuer.api,
        SiteDraftStore::new(lifecycle.clone(), owners.journal().clone(), delivery, 2000),
        limits()?,
    )?;
    assert_eq!(
        puller
            .pull("connector-fixture", PullSignal::Startup, 2000)
            .await?
            .executed,
        1
    );
    assert_eq!(
        puller
            .pull("connector-fixture", PullSignal::Console, 3000)
            .await?
            .replayed,
        1
    );
    let record = lifecycle.record("construction")?;
    assert_eq!((record.transitions.len(), record.approvals.len()), (1, 1));
    assert_eq!(record.approvals[0].draft, draft.id);
    assert_eq!(record.approvals[0].digest, draft.request_digest);
    assert_eq!(record.approvals[0].approved_by, ["fixture-owner"]);
    assert_eq!(record.approvals[0].executed_at_ms, 2000);
    held(
        &keeper(
            owners.clone(),
            lifecycle.clone(),
            "fixture-owner",
            "person",
            &gone,
        )?,
        &gone,
    )?;
    lifecycle.retire("construction", "fixture-owner", 4000)?;
    assert_eq!(
        puller
            .pull("connector-fixture", PullSignal::Console, 5000)
            .await?
            .refused,
        1
    );
    assert_eq!(
        puller
            .pull("connector-fixture", PullSignal::Console, 6000)
            .await?
            .replayed,
        1
    );
    assert_eq!(lifecycle.record("construction")?.approvals.len(), 1);
    assert_eq!(issuer.calls()?, 9);
    issuer.finish()
}

pub fn fixture_child_request(
    owners: std::sync::Arc<crate::auth_lys::live::drafts::waiting::durable::ApprovalAgents>,
    lifecycle: std::sync::Arc<crate::services::lifecycle::Lifecycle>,
) -> Result<lys_pass::drafts::ApprovedDraft, Box<dyn std::error::Error>> {
    use crate::agents::records::{AgentAct, AgentEvent, AgentHarness, AgentSpec};
    use crate::auth_lys::live::PreparedDraftKeeper;
    use crate::auth_lys::live::drafts::waiting::tests::keeper;
    use lys_pass::drafts::{ApprovedDraft, DraftRequest, request_digest};
    let target = lys_pass::Target::new("liminal.agent", "child", "create_agent")?;
    let identity = verified_holder("builder", "agent", &target)?;
    let change = AgentEvent::Create(AgentSpec {
        harness: AgentHarness::Cdx,
        purpose: "Build a second approved service.".into(),
        service: None,
    });
    let words =
        serde_json::to_string(&AgentAct::prepare("child".into(), change, &identity, 1000)?)?;
    let draft = ApprovedDraft {
        id: "child-create".into(),
        app: "liminal".into(),
        grant: "grant-prepared".into(),
        target,
        request_digest: request_digest(&words),
        approved_by: vec!["fixture-owner".into()],
        mode: lys_pass::Mode::ByDraft,
        words,
    };
    let owner = keeper(owners, lifecycle, "builder", "agent", &draft)?;
    let request = DraftRequest {
        operation: "child-operation".into(),
        grant: draft.grant.clone(),
        target: draft.target.clone(),
        request_digest: draft.request_digest.clone(),
        words: draft.words.clone(),
    };
    owner.prepare("builder", &request, 1000)?;
    owner.held("builder", &draft.id, &request, 1000)?;
    Ok(draft)
}