windmill/backend/tests/jobs_read_auth.rs

//! Regression test for the single-job read authorization bypass.
//!
//! The single-job read endpoints (`/jobs_u/get`, `/completed/get`,
//! `/completed/get_result`, `/get_args`, `/get_logs`, `/getupdate`, ...) fetch a
//! job through the root DB handle, filtered only by job id + workspace. That is
//! required for the unauthenticated approval / public-trigger / anonymous-job
//! flows, but for a *logged-in* user it meant any workspace member — including a
//! plain viewer with no ACL on the runnable — could read another user's job
//! args/result/logs simply by obtaining the job UUID, even though the same job is
//! hidden from them in `jobs/list` (RLS-filtered) and the underlying script
//! returns 404.
//!
//! The fix (`require_job_read_access`) gates the authenticated case: a caller may
//! read a job they created (covers app components / webhooks / their own runs)
//! or one visible to them under the same RLS as `jobs/list` (admins bypass);
//! otherwise 404. Unauthenticated access is unchanged (anonymous jobs only).
//!
//! This test pins down, against the `jobs_read_auth` fixture:
//!   - a viewer is denied the victim job's full record / result / result_maybe /
//!     args / logs / live update by UUID, and the secret never appears in the
//!     body (the core fix; pre-fix these returned 200 with the secret),
//!   - the job's owner and an admin can still read it (no over-blocking),
//!   - the "app component" affordance survives: a viewer who *launched* a job
//!     (created_by) running as someone else's identity can still read its result,
//!   - unauthenticated behavior is unchanged: anonymous jobs readable, the
//!     non-anonymous victim job rejected,
//!   - `queue/cancel` and `queue/force_cancel` are gated by that same access, so
//!     a viewer cannot kill a run hidden from them while its owner still can, and
//!     force cancel gates on the ancestor it actually kills rather than the id in
//!     the URL.

use sqlx::{Pool, Postgres};
use windmill_test_utils::*;

const VICTIM: &str = "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa";
const APP_JOB: &str = "bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb";
const ANON_JOB: &str = "cccccccc-cccc-cccc-cccc-cccccccccccc";
const FLOW_JOB: &str = "dddddddd-dddd-dddd-dddd-dddddddddddd";
const STEP_JOB: &str = "eeeeeeee-eeee-eeee-eeee-eeeeeeeeeeee";
// Deep nesting: top (not visible) -> mid (visible via folder) -> deep leaf.
const TOP_SECRET_FLOW: &str = "ffffffff-ffff-ffff-ffff-ffffffffffff";
const DEEP_LEAF_JOB: &str = "88888888-8888-8888-8888-888888888888";
// A queued/running job (no completed row) owned by test-user-2.
const RUNNING_JOB: &str = "77777777-7777-7777-7777-777777777777";
// An app-component job launched BY the admin embed viewer (created_by test-user).
const EMBED_OWN_JOB: &str = "12121212-1212-1212-1212-121212121212";
// A QUEUED job launched by the embed viewer (created_by test-user) — cancelable by it.
const EMBED_OWN_QUEUED: &str = "13131313-1313-1313-1313-131313131313";
// `singlestepflow` wrappers (as native retry / scheduled runs produce), one around a
// SCRIPT and one around a FLOW.
const WRAPPED_JOB: &str = "14141414-1414-1414-1414-141414141414";
const WRAPPED_FLOW_JOB: &str = "15151515-1515-1515-1515-151515151515";
// An inline-script component run of app `u/test-user-2/dash` (`trigger_kind = 'app'`).
const APP_INLINE_JOB: &str = "16161616-1616-1616-1616-161616161616";
// Queued sub-flow test-user-3 can see (folder `shared`), whose parent top flow they
// cannot. Force cancel walks up to that parent.
const QUEUED_VISIBLE_MID: &str = "55555555-5555-5555-5555-555555555555";

// Secrets that must never leak to an unauthorized viewer.
const RESULT_SECRET: &str = "RESULT_SECRET";
const ARGS_SECRET: &str = "LEAK_TEST_ARGS";
const LOGS_SECRET: &str = "LEAK_TEST_LOGS";

fn client() -> reqwest::Client {
    reqwest::Client::new()
}

async fn get(base: &str, path: &str, token: Option<&str>) -> (reqwest::StatusCode, String) {
    let mut req = client().get(format!("{base}/{path}"));
    if let Some(token) = token {
        req = req.header("Authorization", format!("Bearer {token}"));
    }
    let resp = req.send().await.expect("request");
    let status = resp.status();
    let body = resp.text().await.expect("body");
    (status, body)
}

async fn post(base: &str, path: &str, token: Option<&str>) -> (reqwest::StatusCode, String) {
    let mut req = client()
        .post(format!("{base}/{path}"))
        .json(&serde_json::json!({}));
    if let Some(token) = token {
        req = req.header("Authorization", format!("Bearer {token}"));
    }
    let resp = req.send().await.expect("request");
    let status = resp.status();
    let body = resp.text().await.expect("body");
    (status, body)
}

async fn post_json(
    url: &str,
    token: Option<&str>,
    body: serde_json::Value,
) -> (reqwest::StatusCode, String) {
    let mut req = client().post(url).json(&body);
    if let Some(token) = token {
        req = req.header("Authorization", format!("Bearer {token}"));
    }
    let resp = req.send().await.expect("request");
    let status = resp.status();
    let body = resp.text().await.expect("body");
    (status, body)
}

async fn delete(url: &str, token: Option<&str>) -> (reqwest::StatusCode, String) {
    let mut req = client().delete(url);
    if let Some(token) = token {
        req = req.header("Authorization", format!("Bearer {token}"));
    }
    let resp = req.send().await.expect("request");
    let status = resp.status();
    let body = resp.text().await.expect("body");
    (status, body)
}

#[sqlx::test(fixtures("base", "jobs_read_auth"))]
async fn test_single_job_read_authorization(db: Pool<Postgres>) -> anyhow::Result<()> {
    initialize_tracing().await;

    let server = ApiServer::start(db.clone()).await?;
    let port = server.addr.port();
    let base = format!("http://localhost:{port}/api/w/test-workspace/jobs_u");
    // result_by_id / get_otel_traces live on the authed `/jobs` service, not `/jobs_u`.
    let authed_base = format!("http://localhost:{port}/api/w/test-workspace/jobs");
    let rules_url =
        format!("http://localhost:{port}/api/w/test-workspace/workspaces/protection_rules");

    // The endpoints that return the victim job's sensitive data by UUID.
    let endpoints = [
        ("get", format!("get/{VICTIM}")),
        ("completed/get", format!("completed/get/{VICTIM}")),
        (
            "completed/get_result",
            format!("completed/get_result/{VICTIM}"),
        ),
        (
            "completed/get_result_maybe",
            format!("completed/get_result_maybe/{VICTIM}"),
        ),
        ("get_args", format!("get_args/{VICTIM}")),
        ("get_logs", format!("get_logs/{VICTIM}")),
        (
            "get_completed_logs_tail",
            format!("get_completed_logs_tail/{VICTIM}"),
        ),
        ("get_flow_all_logs", format!("get_flow_all_logs/{VICTIM}")),
        (
            "get_flow_all_logs_structured",
            format!("get_flow_all_logs_structured/{VICTIM}"),
        ),
        (
            "completed/get_timing",
            format!("completed/get_timing/{VICTIM}"),
        ),
        ("getupdate", format!("getupdate/{VICTIM}?only_result=true")),
    ];

    // ---- CORE REGRESSION: the viewer (test-user-3) is denied on every endpoint
    //      and no secret ever appears in the body. Pre-fix these returned 200
    //      and leaked the secret.
    for (name, path) in &endpoints {
        let (status, body) = get(&base, path, Some("SECRET_TOKEN_3")).await;
        assert_eq!(
            status,
            reqwest::StatusCode::FORBIDDEN,
            "viewer must get 403 on {name} (got {status}): {body}"
        );
        for secret in [RESULT_SECRET, ARGS_SECRET, LOGS_SECRET] {
            assert!(
                !body.contains(secret),
                "viewer response for {name} leaked `{secret}`: {body}"
            );
        }
    }

    // The 403 for an existing-but-forbidden job carries actionable guidance
    // (request a share link), distinguishing it from a plain not-found.
    let (status, body) = get(
        &base,
        &format!("completed/get_result/{VICTIM}"),
        Some("SECRET_TOKEN_3"),
    )
    .await;
    assert_eq!(status, reqwest::StatusCode::FORBIDDEN);
    assert!(
        body.to_lowercase().contains("share"),
        "403 body should guide the user to request a share link: {body}"
    );

    // A genuinely non-existent job is a 404, not a 403 — existence is only disclosed
    // for jobs that actually exist in the workspace.
    let missing = "00000000-0000-4000-8000-000000000000";
    let (status, _) = get(
        &base,
        &format!("completed/get_result/{missing}"),
        Some("SECRET_TOKEN_3"),
    )
    .await;
    assert_eq!(
        status,
        reqwest::StatusCode::NOT_FOUND,
        "a non-existent job must be 404, not 403 (got {status})"
    );

    // ---- NO OVER-BLOCKING: the job's owner (test-user-2) can read its result.
    let (status, body) = get(
        &base,
        &format!("completed/get_result/{VICTIM}"),
        Some("SECRET_TOKEN_2"),
    )
    .await;
    assert!(
        status.is_success(),
        "owner must still read their own job result (got {status}): {body}"
    );
    assert!(
        body.contains(RESULT_SECRET),
        "owner result must contain the value: {body}"
    );

    // ---- ADMIN BYPASS: an admin (test-user) can read any job in the workspace.
    let (status, body) = get(
        &base,
        &format!("completed/get_result/{VICTIM}"),
        Some("SECRET_TOKEN"),
    )
    .await;
    assert!(
        status.is_success(),
        "admin must read any job (got {status}): {body}"
    );
    assert!(body.contains(RESULT_SECRET), "admin result body: {body}");

    // ---- APP AFFORDANCE: a viewer who LAUNCHED a job (created_by = viewer) that
    //      runs as another identity (permissioned_as = test-user-2,
    //      visible_to_owner = false) can still read its result. This is the app
    //      component-polling path; the fix must not break it.
    let (status, body) = get(
        &base,
        &format!("completed/get_result/{APP_JOB}"),
        Some("SECRET_TOKEN_3"),
    )
    .await;
    assert!(
        status.is_success(),
        "launcher must read a job they created even without ACL on the runnable (got {status}): {body}"
    );
    assert!(
        body.contains("visible_to_launcher"),
        "launcher should get the result they polled: {body}"
    );

    // ---- AUTHED `/jobs` endpoints in the same class: result_by_id (flow node
    //      result) and get_otel_traces (job telemetry). The viewer must be denied
    //      the victim by UUID. The auth gate runs before result/trace resolution,
    //      so 404 here is the gate, not incidental resolution failure.
    let (status, body) = get(
        &authed_base,
        &format!("result_by_id/{VICTIM}/somenode"),
        Some("SECRET_TOKEN_3"),
    )
    .await;
    assert_eq!(
        status,
        reqwest::StatusCode::FORBIDDEN,
        "viewer must get 403 on result_by_id (got {status}): {body}"
    );
    assert!(!body.contains(RESULT_SECRET), "result_by_id leaked: {body}");

    let (status, body) = get(
        &authed_base,
        &format!("get_otel_traces/{VICTIM}"),
        Some("SECRET_TOKEN_3"),
    )
    .await;
    assert_eq!(
        status,
        reqwest::StatusCode::FORBIDDEN,
        "viewer must get 403 on get_otel_traces (got {status}): {body}"
    );

    // ---- FLOW VISIBILITY INHERITANCE: test-user-3 has folder ACL on the flow
    //      `f/shared/flow1` (run by test-user-2) but did NOT launch it, and has no
    //      ACL on the step's inner runnable `u/test-user-2/inner_secret`. They must
    //      still be able to (a) read the flow they can see, and (b) inspect its
    //      step result — visibility is inherited from the flow root. A naive
    //      "same as list" gate would 404 the step and break the flow-run UI.
    let (status, body) = get(
        &base,
        &format!("completed/get_result/{FLOW_JOB}"),
        Some("SECRET_TOKEN_3"),
    )
    .await;
    assert!(
        status.is_success(),
        "viewer with folder ACL must read the flow they can see (got {status}): {body}"
    );
    let (status, body) = get(
        &base,
        &format!("completed/get_result/{STEP_JOB}"),
        Some("SECRET_TOKEN_3"),
    )
    .await;
    assert!(
        status.is_success(),
        "viewer must inspect a step of a flow they can see, even without ACL on the step's runnable (got {status}): {body}"
    );
    assert!(
        body.contains("STEP_RESULT_INHERITED"),
        "step result should be returned via flow-root inheritance: {body}"
    );

    // ---- DEEP NESTING / MIDDLE-LAYER VISIBILITY: the deep leaf's root_job is the
    //      top flow (NOT visible to test-user-3), but an intermediate sub-flow
    //      (f/shared/mid) IS visible. Reading the leaf must succeed via that middle
    //      ancestor — i.e. the full parent chain is walked, not just [self, root].
    let (status, body) = get(
        &base,
        &format!("completed/get_result/{DEEP_LEAF_JOB}"),
        Some("SECRET_TOKEN_3"),
    )
    .await;
    assert!(
        status.is_success(),
        "deep leaf must be readable via a visible intermediate sub-flow (got {status}): {body}"
    );
    assert!(
        body.contains("DEEP_STEP_INHERITED"),
        "deep leaf result should be returned via mid-ancestor visibility: {body}"
    );
    // ...but the top flow itself, in a folder the viewer cannot read, stays denied.
    let (status, body) = get(
        &base,
        &format!("completed/get_result/{TOP_SECRET_FLOW}"),
        Some("SECRET_TOKEN_3"),
    )
    .await;
    assert_eq!(
        status,
        reqwest::StatusCode::FORBIDDEN,
        "top flow in an unreadable folder must stay denied (got {status}): {body}"
    );

    // ---- APP EMBED TOKEN: confined to jobs the viewer LAUNCHED, not everything
    //      the (admin) viewer can otherwise read. The token carries the `app_embed`
    //      sentinel; an admin's normal token reads VICTIM (asserted above), but the
    //      embed token must stop at the `created_by == viewer` grant so user-authored
    //      app JS can't reuse it to read unrelated jobs by UUID.
    // Its own launched component job (created_by == viewer) still reads.
    let (status, body) = get(
        &base,
        &format!("completed/get_result/{EMBED_OWN_JOB}"),
        Some("EMBED_APP_TOKEN"),
    )
    .await;
    assert!(
        status.is_success(),
        "embed token must read a job it launched (got {status}): {body}"
    );
    assert!(
        body.contains("EMBED_OWN_RESULT"),
        "embed token should get its own launched job result: {body}"
    );
    // The VICTIM job — created by another user but readable by this admin viewer's
    // normal token (asserted above) — is denied to the embed token across result /
    // logs / live update. NotFound (not 403) so the untrusted app can't even probe
    // existence, and no secret leaks.
    for path in [
        format!("completed/get_result/{VICTIM}"),
        format!("get_logs/{VICTIM}"),
        format!("getupdate/{VICTIM}?only_result=true"),
    ] {
        let (status, body) = get(&base, &path, Some("EMBED_APP_TOKEN")).await;
        assert_eq!(
            status,
            reqwest::StatusCode::NOT_FOUND,
            "embed token must not read a job it did not launch ({path}, got {status}): {body}"
        );
        for secret in [RESULT_SECRET, ARGS_SECRET, LOGS_SECRET] {
            assert!(
                !body.contains(secret),
                "embed token response for {path} leaked `{secret}`: {body}"
            );
        }
    }

    // ---- PATH-SCOPED RUN TOKEN: confined to jobs of the runnable it may start.
    //      RUN_SCOPED_TOKEN is test-user-2's `jobs:run:flows:f/shared/flow1` webhook
    //      token, and test-user-2 created every job asserted on below — so `created_by`
    //      alone would hand it all of them.
    // Its own flow run reads, and so do the steps beneath it: a step's `runnable_path`
    // is the inner script's, so the scope has to be satisfied through the ancestor.
    for (path, expected) in [
        (
            format!("completed/get_result/{FLOW_JOB}"),
            r#""flow": "done""#,
        ),
        (
            format!("completed/get_result/{STEP_JOB}"),
            "STEP_RESULT_INHERITED",
        ),
    ] {
        let (status, body) = get(&base, &path, Some("RUN_SCOPED_TOKEN")).await;
        assert!(
            status.is_success(),
            "run-scoped token must read its own flow run ({path}, got {status}): {body}"
        );
        assert!(
            body.contains(expected),
            "run-scoped token should get {expected} for {path}: {body}"
        );
    }
    // A job of any other runnable is out of scope, even though the same user created it.
    for path in [
        format!("completed/get_result/{VICTIM}"),
        format!("get_args/{VICTIM}"),
        format!("get_logs/{VICTIM}"),
        format!("getupdate/{VICTIM}?only_result=true"),
    ] {
        let (status, body) = get(&base, &path, Some("RUN_SCOPED_TOKEN")).await;
        assert_eq!(
            status,
            reqwest::StatusCode::NOT_FOUND,
            "run-scoped token must not read a job outside its scope ({path}, got {status}): {body}"
        );
        for secret in [RESULT_SECRET, ARGS_SECRET, LOGS_SECRET] {
            assert!(
                !body.contains(secret),
                "run-scoped token response for {path} leaked `{secret}`: {body}"
            );
        }
    }
    // A `singlestepflow` wrapper (native retry / scheduled run) belongs to the runnable
    // it wraps, not to the flow domain its `kind` suggests. Each wrapper is readable by
    // the token scoped to the wrapped kind, and only by that one.
    for (job, reader, denied) in [
        (WRAPPED_JOB, "RUN_SCOPED_SCRIPT_TOKEN", "RUN_SCOPED_TOKEN"),
        (
            WRAPPED_FLOW_JOB,
            "RUN_SCOPED_TOKEN",
            "RUN_SCOPED_SCRIPT_TOKEN",
        ),
    ] {
        let (status, body) = get(&base, &format!("completed/get_result/{job}"), Some(reader)).await;
        assert!(
            status.is_success() && body.contains("WRAPPED"),
            "{reader} must read the singlestepflow wrapping its runnable (got {status}): {body}"
        );
        let (status, body) = get(&base, &format!("completed/get_result/{job}"), Some(denied)).await;
        assert_eq!(
            status,
            reqwest::StatusCode::NOT_FOUND,
            "{denied} must not read a wrapper around the other kind (got {status}): {body}"
        );
    }

    // An agent run and a chat turn of it belong to the agent, readable by the token scoped to
    // it and by no flow-scoped one.
    for (job, expected) in [
        ("17171717-1717-1717-1717-171717171717", "AGENT_RESULT"),
        ("18181818-1818-1818-1818-181818181818", "AGENT_CHAT_RESULT"),
    ] {
        let path = format!("completed/get_result/{job}");
        let (status, body) = get(&base, &path, Some("RUN_SCOPED_AGENT_TOKEN")).await;
        assert!(
            status.is_success() && body.contains(expected),
            "agent-scoped token must read its agent's run {job} (got {status}): {body}"
        );
        let (status, body) = get(&base, &path, Some("RUN_SCOPED_TOKEN")).await;
        assert_eq!(
            status,
            reqwest::StatusCode::NOT_FOUND,
            "a flow-scoped token must not read an agent run {job} (got {status}): {body}"
        );
    }

    // A lock job belongs to `jobs:run:dependencies`: its path is whatever the caller sent,
    // so a token scoped to run the flow at that path must not read it.
    let path = "completed/get_result/19191919-1919-1919-1919-191919191919";
    let (status, body) = get(&base, path, Some("RUN_SCOPED_DEPS_TOKEN")).await;
    assert!(
        status.is_success() && body.contains("DEPS_RESULT"),
        "dependencies-scoped token must read its lock job (got {status}): {body}"
    );
    let (status, body) = get(&base, path, Some("RUN_SCOPED_TOKEN")).await;
    assert_eq!(
        status,
        reqwest::StatusCode::NOT_FOUND,
        "a flow-scoped token must not read a lock job (got {status}): {body}"
    );
    let (status, body) = get(
        &base,
        &format!("completed/get_result/{FLOW_JOB}"),
        Some("RUN_SCOPED_DEPS_TOKEN"),
    )
    .await;
    assert_eq!(
        status,
        reqwest::StatusCode::NOT_FOUND,
        "dependencies-scoped token must not read a flow run at its path (got {status}): {body}"
    );
    // A lock job filed at `*` matches no path-scoped grant: the stored path is not a pattern.
    let (status, body) = get(
        &base,
        "completed/get_result/20202020-2020-2020-2020-202020202020",
        Some("RUN_SCOPED_DEPS_TOKEN"),
    )
    .await;
    assert_eq!(
        status,
        reqwest::StatusCode::NOT_FOUND,
        "a path-scoped token must not read a lock job filed at `*` (got {status}): {body}"
    );

    // An `apps:run:<app>` scope is a start grant too: the inline-script component run it
    // launched — a kind no `jobs:run` scope can name — stays readable to a token scoped
    // to that app, and stays out of reach for one that is only scoped to run jobs.
    let (status, body) = get(
        &base,
        &format!("completed/get_result/{APP_INLINE_JOB}"),
        Some("APP_RUNNER_TOKEN"),
    )
    .await;
    assert!(
        status.is_success() && body.contains("APP_INLINE_RESULT"),
        "app-scoped token must read the component run its app launched (got {status}): {body}"
    );
    let (status, body) = get(
        &base,
        &format!("completed/get_result/{APP_INLINE_JOB}"),
        Some("RUN_SCOPED_SCRIPT_TOKEN"),
    )
    .await;
    assert_eq!(
        status,
        reqwest::StatusCode::NOT_FOUND,
        "a token with no scope on the app must not read its component run (got {status}): {body}"
    );

    // An approval link is a bypass of the read gate, so the confinement is re-applied on
    // top of it: it must not become a way for a scoped token to read an out-of-scope job.
    // The link itself is untouched — a logged-out approver still reads the same job.
    let approval_token =
        windmill_common::variables::generate_approval_token("test-workspace", VICTIM.parse()?, &db)
            .await?;
    let (status, body) = get(
        &base,
        &format!("get/{VICTIM}?approval_token={approval_token}"),
        None,
    )
    .await;
    assert!(
        status.is_success(),
        "an approval link must still authorize a logged-out read (got {status}): {body}"
    );
    let (status, body) = get(
        &base,
        &format!("get/{VICTIM}?approval_token={approval_token}"),
        Some("RUN_SCOPED_TOKEN"),
    )
    .await;
    assert_eq!(
        status,
        reqwest::StatusCode::NOT_FOUND,
        "an approval link must not lift the run-scope confinement (got {status}): {body}"
    );

    // Same for the resume-secret bypass on the result route, which the approval page uses.
    let (status, secret) = get(
        &authed_base,
        &format!("job_signature/{STEP_JOB}/0"),
        Some("SECRET_TOKEN"),
    )
    .await;
    assert!(
        status.is_success(),
        "an admin must mint a resume secret: {secret}"
    );
    let secret = secret.trim().trim_matches('"').to_string();
    let approval_result = format!(
        "completed/get_result/{STEP_JOB}?suspended_job={STEP_JOB}&resume_id=0&secret={secret}"
    );
    let (status, body) = get(&base, &approval_result, None).await;
    assert!(
        status.is_success(),
        "a resume secret must still authorize a logged-out result read (got {status}): {body}"
    );
    let (status, body) = get(&base, &approval_result, Some("RUN_SCOPED_SCRIPT_TOKEN")).await;
    assert_eq!(
        status,
        reqwest::StatusCode::NOT_FOUND,
        "a resume secret must not lift the run-scope confinement (got {status}): {body}"
    );

    // And a run grant is not an enumeration grant: the whole listing surface is denied.
    let (status, body) = get(&authed_base, "list", Some("RUN_SCOPED_TOKEN")).await;
    assert_eq!(
        status,
        reqwest::StatusCode::FORBIDDEN,
        "run-scoped token must not enumerate jobs (got {status}): {body}"
    );

    // ---- PATH-SCOPED READ TOKEN: READ_SCOPED_TOKEN is test-user-2's
    //      `jobs:read:f/shared/flow1`. It reads that flow's runs and the steps beneath
    //      them, and nothing else its owner created.
    for (path, expected) in [
        (
            format!("completed/get_result/{FLOW_JOB}"),
            r#""flow": "done""#,
        ),
        (
            format!("completed/get_result/{STEP_JOB}"),
            "STEP_RESULT_INHERITED",
        ),
    ] {
        let (status, body) = get(&base, &path, Some("READ_SCOPED_TOKEN")).await;
        assert!(
            status.is_success() && body.contains(expected),
            "read-scoped token must read a run of its flow ({path}, got {status}): {body}"
        );
    }
    for path in [
        format!("completed/get_result/{VICTIM}"),
        format!("get/{VICTIM}"),
        format!("flow/approval_info/{VICTIM}"),
    ] {
        let (status, body) = get(&base, &path, Some("READ_SCOPED_TOKEN")).await;
        assert_eq!(
            status,
            reqwest::StatusCode::NOT_FOUND,
            "read-scoped token must not read a job outside its paths ({path}, got {status}): {body}"
        );
    }
    // The lists answer with the top-level runs at its paths only, where the same user's
    // unscoped token sees the other runs too.
    for list in ["list", "completed/list"] {
        let (status, body) = get(&authed_base, list, Some("READ_SCOPED_TOKEN")).await;
        assert!(status.is_success(), "{list} (got {status}): {body}");
        let jobs: Vec<serde_json::Value> = serde_json::from_str(&body)?;
        let mut ids: Vec<&str> = jobs.iter().filter_map(|j| j["id"].as_str()).collect();
        ids.sort();
        assert_eq!(ids, [WRAPPED_FLOW_JOB, FLOW_JOB], "{list}: {body}");
    }
    let (_, body) = get(&authed_base, "list", Some("SECRET_TOKEN_2")).await;
    assert!(body.contains(VICTIM), "unscoped list: {body}");
    let (status, body) = get(&authed_base, "completed/count", Some("READ_SCOPED_TOKEN")).await;
    assert_eq!(
        status,
        reqwest::StatusCode::FORBIDDEN,
        "read-scoped token must not count the workspace's jobs (got {status}): {body}"
    );

    // ---- APP EMBED TOKEN: cancellation confined to the app's own jobs. The token
    //      may cancel a job it launched (created_by == viewer), but `cancel_job_api`
    //      denies (NotFound) a job created by someone else, even one the (admin)
    //      viewer could otherwise cancel.
    let (status, body) = post(
        &base,
        &format!("queue/cancel/{EMBED_OWN_QUEUED}"),
        Some("EMBED_APP_TOKEN"),
    )
    .await;
    assert!(
        status.is_success(),
        "embed token must cancel a job it launched (got {status}): {body}"
    );
    let (status, body) = post(
        &base,
        &format!("queue/cancel/{RUNNING_JOB}"),
        Some("EMBED_APP_TOKEN"),
    )
    .await;
    assert_eq!(
        status,
        reqwest::StatusCode::NOT_FOUND,
        "embed token must not cancel another user's job (got {status}): {body}"
    );

    // ---- UNAUTHENTICATED, unchanged: an anonymous-created job is readable
    //      without a token (public trigger / public app result polling).
    let (status, body) = get(&base, &format!("completed/get_result/{ANON_JOB}"), None).await;
    assert!(
        status.is_success(),
        "anonymous job must remain readable unauthenticated (got {status}): {body}"
    );

    // ---- UNAUTHENTICATED, unchanged: the non-anonymous victim job is rejected
    //      for an unauthenticated caller (400, the pre-existing guard).
    let (status, body) = get(&base, &format!("completed/get_result/{VICTIM}"), None).await;
    assert_eq!(
        status,
        reqwest::StatusCode::BAD_REQUEST,
        "unauthenticated access to a non-anonymous job must stay rejected (got {status}): {body}"
    );
    assert!(
        !body.contains(RESULT_SECRET),
        "unauth body must not leak: {body}"
    );

    // ---- SHARE READ LINK (view_token) ----
    // The owner (test-user-2) mints a share token for the victim job.
    let (status, mint_body) = get(
        &authed_base,
        &format!("job_view_token/{VICTIM}"),
        Some("SECRET_TOKEN_2"),
    )
    .await;
    assert!(
        status.is_success(),
        "owner must be able to mint a share token (got {status}): {mint_body}"
    );
    let token = mint_body.trim().trim_matches('"').to_string();
    assert!(
        token.starts_with(VICTIM),
        "token must encode the job id: {token}"
    );

    // The viewer (no ACL) can now read the victim job via the share link.
    let (status, body) = get(
        &base,
        &format!("completed/get_result/{VICTIM}?view_token={token}"),
        Some("SECRET_TOKEN_3"),
    )
    .await;
    assert!(
        status.is_success(),
        "view_token must grant the viewer read of the shared job (got {status}): {body}"
    );
    assert!(
        body.contains(RESULT_SECRET),
        "shared job result must be returned with a valid view_token: {body}"
    );
    // ...and its args/logs too (whole detail page).
    let (status, _) = get(
        &base,
        &format!("get_args/{VICTIM}?view_token={token}"),
        Some("SECRET_TOKEN_3"),
    )
    .await;
    assert!(
        status.is_success(),
        "view_token must also grant args (got {status})"
    );

    // The token is scoped: it does NOT authorize an unrelated job.
    let (status, _) = get(
        &base,
        &format!("completed/get_result/{ANON_JOB}?view_token={token}"),
        Some("SECRET_TOKEN_3"),
    )
    .await;
    assert_eq!(
        status,
        reqwest::StatusCode::FORBIDDEN,
        "a victim-scoped token must not authorize a different job (got {status})"
    );

    // A garbage token is rejected (falls through to the normal 404).
    let (status, _) = get(
        &base,
        &format!("completed/get_result/{VICTIM}?view_token={VICTIM}.deadbeef"),
        Some("SECRET_TOKEN_3"),
    )
    .await;
    assert_eq!(
        status,
        reqwest::StatusCode::FORBIDDEN,
        "an invalid view_token must not grant access (got {status})"
    );

    // A share token authorizes the shared job's whole flow subtree: the owner mints
    // for the top secret flow, and the viewer can then read its deep leaf.
    let (status, mint_body) = get(
        &authed_base,
        &format!("job_view_token/{TOP_SECRET_FLOW}"),
        Some("SECRET_TOKEN_2"),
    )
    .await;
    assert!(
        status.is_success(),
        "owner mints token for top flow (got {status}): {mint_body}"
    );
    let top_token = mint_body.trim().trim_matches('"').to_string();
    let (status, body) = get(
        &base,
        &format!("completed/get_result/{DEEP_LEAF_JOB}?view_token={top_token}"),
        Some("SECRET_TOKEN_3"),
    )
    .await;
    assert!(
        status.is_success(),
        "a flow's share token must authorize its deep descendants (got {status}): {body}"
    );

    // A viewer who cannot read a job cannot mint a share token for it.
    let (status, _) = get(
        &authed_base,
        &format!("job_view_token/{TOP_SECRET_FLOW}"),
        Some("SECRET_TOKEN_3"),
    )
    .await;
    assert_eq!(
        status,
        reqwest::StatusCode::FORBIDDEN,
        "a non-reader must not be able to mint a share token (got {status})"
    );

    // ---- PUBLIC SHARE READ LINK (public view_token) ----
    // A member-audience token must never turn into a public one: links already handed
    // out stay confined to logged-in members.
    let (status, body) = get(
        &base,
        &format!("completed/get_result/{VICTIM}?view_token={token}"),
        None,
    )
    .await;
    assert_eq!(
        status,
        reqwest::StatusCode::BAD_REQUEST,
        "a member view_token must not grant unauthenticated read (got {status}): {body}"
    );
    assert!(
        !body.contains(RESULT_SECRET),
        "unauth body must not leak with a member token: {body}"
    );

    // The owner mints the public flavor for the top flow.
    let (status, mint_body) = get(
        &authed_base,
        &format!("job_public_view_token/{TOP_SECRET_FLOW}"),
        Some("SECRET_TOKEN_2"),
    )
    .await;
    assert!(
        status.is_success(),
        "owner must be able to mint a public share token (got {status}): {mint_body}"
    );
    let public_token = mint_body.trim().trim_matches('"').to_string();

    // It grants a logged-out visitor the shared job and its whole flow subtree.
    for path in [
        format!("get/{TOP_SECRET_FLOW}?view_token={public_token}"),
        format!("get_args/{TOP_SECRET_FLOW}?view_token={public_token}"),
        format!("getupdate/{TOP_SECRET_FLOW}?view_token={public_token}"),
        format!("completed/get_result/{DEEP_LEAF_JOB}?view_token={public_token}"),
        format!("get_logs/{DEEP_LEAF_JOB}?view_token={public_token}"),
    ] {
        let (status, body) = get(&base, &path, None).await;
        assert!(
            status.is_success(),
            "a public view_token must grant unauthenticated read of {path} (got {status}): {body}"
        );
    }

    // Same scoping as the member flavor: another job, and a tampered signature, are refused.
    for path in [
        format!("get/{VICTIM}?view_token={public_token}"),
        format!("get/{TOP_SECRET_FLOW}?view_token={TOP_SECRET_FLOW}.deadbeef"),
    ] {
        let (status, body) = get(&base, &path, None).await;
        assert_eq!(
            status,
            reqwest::StatusCode::BAD_REQUEST,
            "an out-of-scope or forged public token must not grant read of {path} (got {status}): {body}"
        );
    }

    // The public audience is a superset of the member one: it must also satisfy the
    // authenticated ACL check, so a viewer handed a public link is not worse off.
    let (status, body) = get(
        &base,
        &format!("completed/get_result/{DEEP_LEAF_JOB}?view_token={public_token}"),
        Some("SECRET_TOKEN_3"),
    )
    .await;
    assert!(
        status.is_success(),
        "a public view_token must also grant an authenticated member read (got {status}): {body}"
    );

    // Tag-scoped caller, job inside its scope: the member flavor is allowed (asserted
    // further down), the public one must not be — i.e. strictly stricter.
    let (status, body) = get(
        &authed_base,
        &format!("job_public_view_token/{VICTIM}"),
        Some("SCOPED_DENO_TOKEN"),
    )
    .await;
    assert_eq!(
        status,
        reqwest::StatusCode::FORBIDDEN,
        "a tag-scoped token must NOT mint a public link, even for an in-scope job (got {status}): {body}"
    );

    // Minting the public flavor takes the same read access as the member one.
    let (status, _) = get(
        &authed_base,
        &format!("job_public_view_token/{TOP_SECRET_FLOW}"),
        Some("SECRET_TOKEN_3"),
    )
    .await;
    assert_eq!(
        status,
        reqwest::StatusCode::FORBIDDEN,
        "a non-reader must not be able to mint a public share token (got {status})"
    );

    // Publishing a run is gated by `RestrictPublicRunSharing`, so a wrong bitflag or
    // rule-kind arm would silently let a restricted member expose one. Admins bypass, and
    // the member flavor is untouched by the rule.
    let (status, body) = post_json(
        &rules_url,
        Some("SECRET_TOKEN"),
        serde_json::json!({
            "name": "no-public-runs",
            "rules": ["RestrictPublicRunSharing"],
            "bypass_users": [],
            "bypass_groups": []
        }),
    )
    .await;
    assert!(
        status.is_success(),
        "admin should create the protection rule (got {status}): {body}"
    );
    let (status, body) = get(
        &authed_base,
        &format!("job_public_view_token/{TOP_SECRET_FLOW}"),
        Some("SECRET_TOKEN_2"),
    )
    .await;
    assert_eq!(
        status,
        reqwest::StatusCode::FORBIDDEN,
        "the rule must block a non-admin owner from minting a public link (got {status}): {body}"
    );
    let (status, body) = get(
        &authed_base,
        &format!("job_view_token/{TOP_SECRET_FLOW}"),
        Some("SECRET_TOKEN_2"),
    )
    .await;
    assert!(
        status.is_success(),
        "the rule must NOT affect the member flavor (got {status}): {body}"
    );
    let (status, body) = get(
        &authed_base,
        &format!("job_public_view_token/{TOP_SECRET_FLOW}"),
        Some("SECRET_TOKEN"),
    )
    .await;
    assert!(
        status.is_success(),
        "an admin must bypass the rule (got {status}): {body}"
    );

    // Later assertions in this test mint public tokens; drop the rule so they see the
    // same workspace state as before.
    let (status, _) = delete(&format!("{rules_url}/no-public-runs"), Some("SECRET_TOKEN")).await;
    assert!(status.is_success(), "admin should delete the rule");

    // ---- TAG-SCOPED token must not mint a token outside its allowed tags ----
    // SCOPED_DENO_TOKEN (test-user-2, scope `if_jobs:filter_tags:deno`) can read both
    // VICTIM (tag deno) and FLOW_JOB (tag flow) by RLS, but minting must honor the
    // tag scope: allowed for the deno job, denied for the flow job.
    let (status, body) = get(
        &authed_base,
        &format!("job_view_token/{VICTIM}"),
        Some("SCOPED_DENO_TOKEN"),
    )
    .await;
    assert!(
        status.is_success(),
        "tag-scoped token may mint for an in-scope (deno) job (got {status}): {body}"
    );
    let (status, _) = get(
        &authed_base,
        &format!("job_view_token/{FLOW_JOB}"),
        Some("SCOPED_DENO_TOKEN"),
    )
    .await;
    assert_eq!(
        status,
        reqwest::StatusCode::NOT_FOUND,
        "tag-scoped token must NOT mint for an out-of-scope (flow) job (got {status})"
    );

    // ---- USE side: a tag-scoped token must not use someone else's valid view_token
    //      to read an out-of-scope job, even via handlers that don't tag-filter their
    //      data query (result_by_id, get_otel_traces, get_flow_debug_info). ----
    // An unscoped owner mints a valid token for the flow (tag 'flow').
    let (status, mint_body) = get(
        &authed_base,
        &format!("job_view_token/{FLOW_JOB}"),
        Some("SECRET_TOKEN_2"),
    )
    .await;
    assert!(
        status.is_success(),
        "owner mints flow token (got {status}): {mint_body}"
    );
    let flow_token = mint_body.trim().trim_matches('"').to_string();

    // The deno-scoped token presents that valid flow token to the non-tag-filtered
    // endpoints — must still be denied (flow tag is out of its scope).
    for path in [
        format!("get_otel_traces/{FLOW_JOB}?view_token={flow_token}"),
        format!("result_by_id/{FLOW_JOB}/somenode?view_token={flow_token}"),
    ] {
        let (status, _) = get(&authed_base, &path, Some("SCOPED_DENO_TOKEN")).await;
        assert_eq!(
            status,
            reqwest::StatusCode::NOT_FOUND,
            "tag-scoped token must not use a view_token to read an out-of-scope job ({path}, got {status})"
        );
    }

    // ...but the deno-scoped token CAN use an in-scope (deno) view_token.
    let (status, body) = get(
        &base,
        &format!("completed/get_result/{VICTIM}?view_token={token}"),
        Some("SCOPED_DENO_TOKEN"),
    )
    .await;
    assert!(
        status.is_success(),
        "tag-scoped token may use a view_token for an in-scope (deno) job (got {status}): {body}"
    );

    // ---- get_result_maybe?get_started=true must authorize before disclosing the
    //      running-state of a queued (not-yet-completed) private job. ----
    // Viewer (no ACL) must be denied rather than told the job is started.
    let (status, body) = get(
        &base,
        &format!("completed/get_result_maybe/{RUNNING_JOB}?get_started=true"),
        Some("SECRET_TOKEN_3"),
    )
    .await;
    assert_eq!(
        status,
        reqwest::StatusCode::FORBIDDEN,
        "viewer must be denied the running-state of a private queued job (got {status}): {body}"
    );
    assert!(
        !body.contains("\"started\""),
        "denied response must not disclose started-state: {body}"
    );
    // The owner still gets the in-progress response.
    let (status, body) = get(
        &base,
        &format!("completed/get_result_maybe/{RUNNING_JOB}?get_started=true"),
        Some("SECRET_TOKEN_2"),
    )
    .await;
    assert!(
        status.is_success() && body.contains("\"started\":true"),
        "owner must see the running job as started (got {status}): {body}"
    );

    // ---- CANCEL / FORCE_CANCEL are gated by the same per-job access as reading:
    //      knowing the UUID of a run hidden from you must not let you kill it. ----
    for path in [
        format!("queue/cancel/{RUNNING_JOB}"),
        format!("queue/force_cancel/{RUNNING_JOB}"),
    ] {
        let (status, body) = post(&base, &path, Some("SECRET_TOKEN_3")).await;
        assert_eq!(
            status,
            reqwest::StatusCode::FORBIDDEN,
            "viewer must not cancel another user's job ({path}, got {status}): {body}"
        );
    }
    // A public share link grants read, never the right to kill the run: cancelling stays
    // confined to anonymously-created jobs for a logged-out caller.
    let (status, mint_body) = get(
        &authed_base,
        &format!("job_public_view_token/{RUNNING_JOB}"),
        Some("SECRET_TOKEN_2"),
    )
    .await;
    assert!(status.is_success(), "owner mints for the running job");
    let running_public = mint_body.trim().trim_matches('"').to_string();
    let (status, body) = get(
        &base,
        &format!("get/{RUNNING_JOB}?view_token={running_public}"),
        None,
    )
    .await;
    assert!(
        status.is_success(),
        "the public token must grant the anonymous read it is for (got {status}): {body}"
    );
    let (status, body) = post(
        &base,
        &format!("queue/cancel/{RUNNING_JOB}?view_token={running_public}"),
        None,
    )
    .await;
    assert!(
        !status.is_success(),
        "a public token must not let an anonymous caller cancel the run (got {status}): {body}"
    );
    // ---- RUN_NOW is gated like cancel, and only moves a job still waiting for a
    //      future start. ----
    let run_now = format!("queue/run_now/{RUNNING_JOB}");
    let (status, body) = post(&authed_base, &run_now, Some("SECRET_TOKEN_3")).await;
    assert_eq!(
        status,
        reqwest::StatusCode::FORBIDDEN,
        "viewer must not start another user's job early (got {status}): {body}"
    );
    let (status, body) = post(&authed_base, &run_now, Some("SECRET_TOKEN_2")).await;
    assert_eq!(
        status,
        reqwest::StatusCode::BAD_REQUEST,
        "a running job has no future start to move (got {status}): {body}"
    );
    sqlx::query(
        "UPDATE v2_job_queue SET running = false, scheduled_for = now() + interval '1 hour'
         WHERE id = $1::uuid",
    )
    .bind(RUNNING_JOB)
    .execute(&db)
    .await?;
    let (status, body) = post(&authed_base, &run_now, Some("SECRET_TOKEN_2")).await;
    assert!(
        status.is_success(),
        "owner must start their deferred job now (got {status}): {body}"
    );
    let due: bool =
        sqlx::query_scalar("SELECT scheduled_for <= now() FROM v2_job_queue WHERE id = $1::uuid")
            .bind(RUNNING_JOB)
            .fetch_one(&db)
            .await?;
    assert!(due, "run_now must make the job due immediately");

    // A schedule tick that is not due yet is refused: the schedule would queue that same
    // tick again on completion and run twice.
    sqlx::query(
        "INSERT INTO schedule (workspace_id, path, edited_by, schedule, script_path, permissioned_as)
         VALUES ('test-workspace', 'u/test-user-2/daily', 'test-user-2', '0 0 3 * * *',
                 'u/test-user-2/running_secret', 'u/test-user-2')",
    )
    .execute(&db)
    .await?;
    sqlx::query(
        "UPDATE v2_job SET trigger_kind = 'schedule', trigger = 'u/test-user-2/daily'
         WHERE id = $1::uuid",
    )
    .bind(RUNNING_JOB)
    .execute(&db)
    .await?;
    sqlx::query(
        "UPDATE v2_job_queue SET scheduled_for =
            (date_trunc('day', now() AT TIME ZONE 'UTC') + interval '1 day 3 hours') AT TIME ZONE 'UTC'
         WHERE id = $1::uuid",
    )
    .bind(RUNNING_JOB)
    .execute(&db)
    .await?;
    let (status, body) = post(&authed_base, &run_now, Some("SECRET_TOKEN_2")).await;
    assert_eq!(
        status,
        reqwest::StatusCode::BAD_REQUEST,
        "an upcoming schedule tick must not be started early (got {status}): {body}"
    );
    // A tick a concurrency limit pushed past its time is off the cron occurrences and
    // may still be started early.
    sqlx::query(
        "UPDATE v2_job_queue SET scheduled_for = now() + interval '1 hour 37 seconds 123 milliseconds'
         WHERE id = $1::uuid",
    )
    .bind(RUNNING_JOB)
    .execute(&db)
    .await?;
    let (status, body) = post(&authed_base, &run_now, Some("SECRET_TOKEN_2")).await;
    assert!(
        status.is_success(),
        "a deferred schedule tick must still start now (got {status}): {body}"
    );
    sqlx::query("UPDATE v2_job SET trigger_kind = NULL, trigger = NULL WHERE id = $1::uuid")
        .bind(RUNNING_JOB)
        .execute(&db)
        .await?;

    // The owner still cancels their own job (no over-blocking). Keep this last: it
    // takes RUNNING_JOB out of the queue.
    let (status, body) = post(
        &base,
        &format!("queue/cancel/{RUNNING_JOB}"),
        Some("SECRET_TOKEN_2"),
    )
    .await;
    assert!(
        status.is_success(),
        "owner must still cancel their own job (got {status}): {body}"
    );

    // Force cancel kills the highest queued ancestor, not the job named in the URL, so it
    // must authorize that ancestor: the viewer can see the sub-flow (asserted first, or the
    // denial below would prove nothing) but not the top flow force-cancelling it would kill.
    let (status, body) = get(
        &base,
        &format!("get/{QUEUED_VISIBLE_MID}"),
        Some("SECRET_TOKEN_3"),
    )
    .await;
    assert!(
        status.is_success(),
        "viewer must be able to read the sub-flow (got {status}): {body}"
    );
    let (status, body) = post(
        &base,
        &format!("queue/force_cancel/{QUEUED_VISIBLE_MID}"),
        Some("SECRET_TOKEN_3"),
    )
    .await;
    assert_eq!(
        status,
        reqwest::StatusCode::FORBIDDEN,
        "viewer must not force-cancel up into a flow they cannot see (got {status}): {body}"
    );

    Ok(())
}