aion-server 0.30.0

//! Run-document handler: the deployed AWL document a run is executing, read
//! under the RUN'S permission.
//!
//! A run's `WorkflowStarted` records the content hash of the package it
//! started under (`WorkflowSummary::package_version`). The archived document
//! for that hash is what produced the run's history, and an operator who may
//! describe the run may read it — this is not a deploy-surface read. The
//! deploy grant gates the CATALOG (`/awl/deployed`, `/deploy/versions`: every
//! version, whether or not any run the caller can see started under it); this
//! read is scoped to one workflow and answers only a hash that workflow's own
//! history recorded, so a caller can never use a workflow they may see to pull
//! a package they may not.

use std::sync::Arc;

use aion_core::{Event, WorkflowId};
use aion_proto::WireError;

use super::error::workflow_not_found_error;
use crate::awl::deployed::DeployedDocument;
use crate::{CallerIdentity, NamespaceGuard, NamespaceOperation, ServerError, WorkflowTarget};

/// The error type of a well-formed hash no generation of the workflow started
/// under. Distinct from `WorkflowNotFound` (the workflow itself) and from the
/// archive refusals (`DeployedVersionNotFound`, `DeployedAwlSourceAbsent`).
pub const RUN_PACKAGE_NOT_RECORDED: &str = "RunPackageNotRecorded";

/// An authorized read of one workflow's document under one package hash:
/// the caller is scoped, the workflow is theirs, and the hash is one the
/// workflow's own history recorded. Holds the scoped engine so the archive
/// read that follows happens in the same scope the check ran in.
#[derive(Clone)]
pub struct RunDocumentAccess {
    /// The namespace the caller was scoped to.
    pub namespace: String,
    /// The workflow type the LATEST generation that recorded `content_hash`
    /// started as — the archive entry the document is registered under. The
    /// resource is "the document that produced this history segment", so the
    /// type comes from the generation that ran it, never from the summary
    /// (whose type belongs to the latest generation, which may never have
    /// run this package). When two generations recorded the same hash under
    /// different types (continue-as-new back onto an old package after a
    /// rename), the later one labels the answer; the type only labels.
    pub workflow_type: String,
    /// The canonical 64-hex content hash, as requested.
    pub content_hash: String,
    engine: Arc<aion::Engine>,
}

impl std::fmt::Debug for RunDocumentAccess {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("RunDocumentAccess")
            .field("namespace", &self.namespace)
            .field("workflow_type", &self.workflow_type)
            .field("content_hash", &self.content_hash)
            .finish_non_exhaustive()
    }
}

impl RunDocumentAccess {
    /// Reads the archived document this access was granted for.
    ///
    /// # Errors
    ///
    /// Returns the archive reader's refusals as stable [`WireError`]s: the
    /// version is not persisted here (`DeployedVersionNotFound`, 404), the
    /// archive carries no AWL source (`DeployedAwlSourceAbsent`, 404 — a
    /// Gleam-authored package has no document to show), or the archive cannot
    /// be read or staged (a backend error).
    pub async fn read(&self) -> Result<DeployedDocument, WireError> {
        crate::awl::deployed::read_document(&self.engine, &self.workflow_type, &self.content_hash)
            .await
            .map_err(|error| error.to_wire_error())
    }
}

/// Authorizes a run-document read: scopes the caller to `namespace` and
/// verifies the workflow exactly as `describe` does, then requires
/// `content_hash` to be recorded on one of the workflow's `WorkflowStarted`
/// events (any generation of its continue-as-new chain).
///
/// # Errors
///
/// - `InvalidInput` when `content_hash` is not the canonical 64-character
///   lowercase hexadecimal form — refused before any store read, so a
///   malformed path never reaches the archive set. This check runs BEFORE
///   the namespace scope on purpose: an ungranted caller probing with a
///   malformed hash learns only that its own input is malformed (zero bits
///   about any tenant's workflows or packages), and refusing garbage before
///   engine resolution is the right economics. Do not "fix" the ordering.
/// - the namespace guard's refusal (`namespace_denied`) when the caller may
///   not read this workflow.
/// - `WorkflowNotFound` when the workflow has no history in this scope.
/// - [`RUN_PACKAGE_NOT_RECORDED`] (not found) when the workflow exists but no
///   generation of it started under `content_hash`.
pub async fn authorize_run_document(
    guard: &NamespaceGuard,
    caller: &CallerIdentity,
    namespace: &str,
    workflow_id: &WorkflowId,
    content_hash: &str,
) -> Result<RunDocumentAccess, WireError> {
    require_canonical_hash(content_hash)?;
    let target = WorkflowTarget::workflow(workflow_id);
    let scoped = guard
        .scope(
            caller,
            &NamespaceOperation::read_document(namespace, target),
        )
        .await
        .map_err(|error| error.to_wire_error())?;
    let engine = scoped.engine().map_err(|error| error.to_wire_error())?;
    let history = engine
        .store()
        .read_history(workflow_id)
        .await
        .map_err(|error| ServerError::from(error).to_wire_error())?;
    if history.is_empty() {
        return Err(workflow_not_found_error(workflow_id));
    }
    let workflow_type = generation_started_under(&history, content_hash).ok_or_else(|| {
        WireError::not_found_with_type(
            RUN_PACKAGE_NOT_RECORDED,
            format!(
                "workflow {workflow_id} recorded no generation started under package \
                 {content_hash}"
            ),
        )
    })?;
    Ok(RunDocumentAccess {
        namespace: scoped.namespace().to_owned(),
        workflow_type,
        content_hash: content_hash.to_owned(),
        engine: std::sync::Arc::clone(engine),
    })
}

/// The workflow type of the LATEST generation whose `WorkflowStarted` recorded
/// `content_hash`, or `None` when no generation did.
fn generation_started_under(history: &[Event], content_hash: &str) -> Option<String> {
    history.iter().rev().find_map(|event| match event {
        Event::WorkflowStarted {
            workflow_type,
            package_version,
            ..
        } if package_version.as_str() == content_hash => Some(workflow_type.clone()),
        _ => None,
    })
}

/// The package hash's canonical textual form: 64 lowercase hexadecimal
/// characters, exactly what `WorkflowStarted.package_version` records.
fn require_canonical_hash(content_hash: &str) -> Result<(), WireError> {
    let canonical = content_hash.len() == 64
        && content_hash
            .bytes()
            .all(|byte| matches!(byte, b'0'..=b'9' | b'a'..=b'f'));
    if canonical {
        Ok(())
    } else {
        Err(WireError::invalid_input(
            "content_hash must be the package's 64-character lowercase hexadecimal content \
             hash, as `package_version` on the run's summary carries it",
        ))
    }
}

#[cfg(test)]
mod tests {
    use aion_core::{Event, PackageVersion, RunId};
    use aion_package::AwlSource;
    use aion_proto::WireErrorCode;
    use aion_store::WriteToken;

    use super::super::test_support::{
        NAMESPACE, append_started, context, event_envelope, payload, workflow_id,
    };
    use super::{RUN_PACKAGE_NOT_RECORDED, authorize_run_document};
    use crate::awl::deployed::fixtures::{DOCUMENT, manifest, record};

    /// The fixture start event's hash (`test_support::started_event`).
    const STARTED_HASH: &str = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa";

    #[tokio::test]
    async fn a_scoped_caller_reads_the_document_of_the_hash_the_run_recorded()
    -> Result<(), Box<dyn std::error::Error>> {
        let context = context().await?;
        context.ownership.record(workflow_id(), NAMESPACE)?;
        // The run's own generation: type `fixture`, started under the archive's hash.
        let row = record(
            manifest("fixture"),
            Some(AwlSource::new(
                "fixture.awl",
                DOCUMENT,
                std::iter::empty::<(String, Vec<u8>)>(),
            )),
            1_700_000_000,
        )?;
        let hash = row.content_hash.clone();
        context.store.put_package(row).await?;
        context
            .store
            .append(
                WriteToken::recorder(),
                &workflow_id(),
                &[Event::WorkflowStarted {
                    envelope: event_envelope(1),
                    workflow_type: "fixture".to_owned(),
                    input: payload()?,
                    run_id: RunId::new(uuid::Uuid::from_u128(1)),
                    parent_run_id: None,
                    parent_workflow_id: None,
                    package_version: PackageVersion::new(hash.clone()),
                }],
                0,
            )
            .await?;

        let access = authorize_run_document(
            &context.guard,
            &context.caller,
            NAMESPACE,
            &workflow_id(),
            &hash,
        )
        .await?;
        assert_eq!(access.namespace, NAMESPACE);
        assert_eq!(access.workflow_type, "fixture");
        let document = access.read().await?;
        assert_eq!(document.content_hash, hash);
        assert_eq!(document.workflow_type, "fixture");
        assert_eq!(document.source, DOCUMENT);
        Ok(())
    }

    /// Two generations of one workflow, each under its own hash and type: each
    /// hash resolves to the type of the generation that recorded it, and when
    /// a later generation records an already-recorded hash under a new type,
    /// the LATEST generation's type labels the answer (rev scan, not first
    /// match).
    #[tokio::test]
    async fn each_hash_resolves_to_the_latest_generation_that_recorded_it()
    -> Result<(), Box<dyn std::error::Error>> {
        let context = context().await?;
        context.ownership.record(workflow_id(), NAMESPACE)?;
        let hash_a = "a".repeat(64);
        let hash_b = "b".repeat(64);
        let started = |seq: u64,
                       workflow_type: &str,
                       hash: &str|
         -> Result<Event, Box<dyn std::error::Error>> {
            Ok(Event::WorkflowStarted {
                envelope: event_envelope(seq),
                workflow_type: workflow_type.to_owned(),
                input: payload()?,
                run_id: RunId::new(uuid::Uuid::from_u128(u128::from(seq))),
                parent_run_id: None,
                parent_workflow_id: None,
                package_version: PackageVersion::new(hash.to_owned()),
            })
        };
        // gen 1: fixture @ A; gen 2: fixture_v2 @ B; gen 3: fixture_renamed @ A.
        context
            .store
            .append(
                WriteToken::recorder(),
                &workflow_id(),
                &[
                    started(1, "fixture", &hash_a)?,
                    started(2, "fixture_v2", &hash_b)?,
                    started(3, "fixture_renamed", &hash_a)?,
                ],
                0,
            )
            .await?;

        let under_b = authorize_run_document(
            &context.guard,
            &context.caller,
            NAMESPACE,
            &workflow_id(),
            &hash_b,
        )
        .await?;
        assert_eq!(under_b.workflow_type, "fixture_v2");
        let under_a = authorize_run_document(
            &context.guard,
            &context.caller,
            NAMESPACE,
            &workflow_id(),
            &hash_a,
        )
        .await?;
        assert_eq!(
            under_a.workflow_type, "fixture_renamed",
            "the LATEST generation that recorded the hash labels the answer"
        );
        Ok(())
    }

    #[tokio::test]
    async fn a_hash_the_workflow_never_started_under_is_not_found_by_its_own_type()
    -> Result<(), Box<dyn std::error::Error>> {
        let context = context().await?;
        context.ownership.record(workflow_id(), NAMESPACE)?;
        append_started(context.store.as_ref()).await?;
        // A real, persisted archive the caller could otherwise read — the
        // workflow simply never ran it. Reachability through the workflow
        // must not leak it.
        let foreign = record(
            manifest("fixture"),
            Some(AwlSource::new(
                "fixture.awl",
                DOCUMENT,
                std::iter::empty::<(String, Vec<u8>)>(),
            )),
            1_700_000_000,
        )?;
        let foreign_hash = foreign.content_hash.clone();
        context.store.put_package(foreign).await?;

        let error = authorize_run_document(
            &context.guard,
            &context.caller,
            NAMESPACE,
            &workflow_id(),
            &foreign_hash,
        )
        .await
        .err()
        .ok_or("a hash the run never recorded must be refused")?;
        assert_eq!(error.code, WireErrorCode::NotFound);
        assert_eq!(error.error_type.as_deref(), Some(RUN_PACKAGE_NOT_RECORDED));
        Ok(())
    }

    #[tokio::test]
    async fn a_recorded_hash_with_no_persisted_archive_is_the_archive_reader_s_not_found()
    -> Result<(), Box<dyn std::error::Error>> {
        let context = context().await?;
        context.ownership.record(workflow_id(), NAMESPACE)?;
        append_started(context.store.as_ref()).await?;

        let access = authorize_run_document(
            &context.guard,
            &context.caller,
            NAMESPACE,
            &workflow_id(),
            STARTED_HASH,
        )
        .await?;
        let error = access
            .read()
            .await
            .err()
            .ok_or("an unpersisted archive must be refused")?;
        assert_eq!(error.code, WireErrorCode::NotFound);
        assert_eq!(error.error_type.as_deref(), Some("DeployedVersionNotFound"));
        Ok(())
    }

    #[tokio::test]
    async fn a_malformed_hash_is_refused_before_any_read() -> Result<(), Box<dyn std::error::Error>>
    {
        let context = context().await?;
        for malformed in ["", "abc", &"A".repeat(64), &"g".repeat(64), &"a".repeat(63)] {
            let error = authorize_run_document(
                &context.guard,
                &context.caller,
                NAMESPACE,
                &workflow_id(),
                malformed,
            )
            .await
            .err()
            .ok_or_else(|| format!("{malformed:?} must be refused"))?;
            assert_eq!(error.code, WireErrorCode::InvalidInput, "{malformed:?}");
        }
        Ok(())
    }

    #[tokio::test]
    async fn an_unowned_workflow_is_refused_by_the_namespace_guard()
    -> Result<(), Box<dyn std::error::Error>> {
        let context = context().await?;
        // Owned by ANOTHER namespace; the caller is scoped to NAMESPACE.
        context.ownership.record(workflow_id(), "tenant-b")?;
        append_started(context.store.as_ref()).await?;

        let error = authorize_run_document(
            &context.guard,
            &context.caller,
            NAMESPACE,
            &workflow_id(),
            STARTED_HASH,
        )
        .await
        .err()
        .ok_or("a foreign workflow must be refused")?;
        // The guard's anti-leak answer: the workflow is "not found in
        // namespace tenant-a" — the same words a truly absent workflow gets —
        // and NOT the hash verdict, which would confirm the workflow exists.
        // (The hash IS recorded, so a guard that let the caller through would
        // have returned an access, not an error.)
        assert_eq!(error.code, WireErrorCode::NotFound, "{error:?}");
        assert_eq!(error.error_type, None, "{error:?}");
        assert!(
            error.message.contains("not found in namespace tenant-a"),
            "{error:?}"
        );
        Ok(())
    }

    #[tokio::test]
    async fn an_unknown_workflow_is_workflow_not_found() -> Result<(), Box<dyn std::error::Error>> {
        let context = context().await?;
        context.ownership.record(workflow_id(), NAMESPACE)?;
        let error = authorize_run_document(
            &context.guard,
            &context.caller,
            NAMESPACE,
            &workflow_id(),
            STARTED_HASH,
        )
        .await
        .err()
        .ok_or("a workflow with no history must be not found")?;
        assert_eq!(error.code, WireErrorCode::NotFound);
        assert_eq!(error.error_type.as_deref(), Some("WorkflowNotFound"));
        Ok(())
    }
}