aion_server/api/handlers/
document.rs

1//! Run-document handler: the deployed AWL document a run is executing, read
2//! under the RUN'S permission.
3//!
4//! A run's `WorkflowStarted` records the content hash of the package it
5//! started under (`WorkflowSummary::package_version`). The archived document
6//! for that hash is what produced the run's history, and an operator who may
7//! describe the run may read it — this is not a deploy-surface read. The
8//! deploy grant gates the CATALOG (`/awl/deployed`, `/deploy/versions`: every
9//! version, whether or not any run the caller can see started under it); this
10//! read is scoped to one workflow and answers only a hash that workflow's own
11//! history recorded, so a caller can never use a workflow they may see to pull
12//! a package they may not.
13
14use std::sync::Arc;
15
16use aion_core::{Event, WorkflowId};
17use aion_proto::WireError;
18
19use super::error::workflow_not_found_error;
20use crate::awl::deployed::DeployedDocument;
21use crate::{CallerIdentity, NamespaceGuard, NamespaceOperation, ServerError, WorkflowTarget};
22
23/// The error type of a well-formed hash no generation of the workflow started
24/// under. Distinct from `WorkflowNotFound` (the workflow itself) and from the
25/// archive refusals (`DeployedVersionNotFound`, `DeployedAwlSourceAbsent`).
26pub const RUN_PACKAGE_NOT_RECORDED: &str = "RunPackageNotRecorded";
27
28/// An authorized read of one workflow's document under one package hash:
29/// the caller is scoped, the workflow is theirs, and the hash is one the
30/// workflow's own history recorded. Holds the scoped engine so the archive
31/// read that follows happens in the same scope the check ran in.
32#[derive(Clone)]
33pub struct RunDocumentAccess {
34    /// The namespace the caller was scoped to.
35    pub namespace: String,
36    /// The workflow type the LATEST generation that recorded `content_hash`
37    /// started as — the archive entry the document is registered under. The
38    /// resource is "the document that produced this history segment", so the
39    /// type comes from the generation that ran it, never from the summary
40    /// (whose type belongs to the latest generation, which may never have
41    /// run this package). When two generations recorded the same hash under
42    /// different types (continue-as-new back onto an old package after a
43    /// rename), the later one labels the answer; the type only labels.
44    pub workflow_type: String,
45    /// The canonical 64-hex content hash, as requested.
46    pub content_hash: String,
47    engine: Arc<aion::Engine>,
48}
49
50impl std::fmt::Debug for RunDocumentAccess {
51    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
52        formatter
53            .debug_struct("RunDocumentAccess")
54            .field("namespace", &self.namespace)
55            .field("workflow_type", &self.workflow_type)
56            .field("content_hash", &self.content_hash)
57            .finish_non_exhaustive()
58    }
59}
60
61impl RunDocumentAccess {
62    /// Reads the archived document this access was granted for.
63    ///
64    /// # Errors
65    ///
66    /// Returns the archive reader's refusals as stable [`WireError`]s: the
67    /// version is not persisted here (`DeployedVersionNotFound`, 404), the
68    /// archive carries no AWL source (`DeployedAwlSourceAbsent`, 404 — a
69    /// Gleam-authored package has no document to show), or the archive cannot
70    /// be read or staged (a backend error).
71    pub async fn read(&self) -> Result<DeployedDocument, WireError> {
72        crate::awl::deployed::read_document(&self.engine, &self.workflow_type, &self.content_hash)
73            .await
74            .map_err(|error| error.to_wire_error())
75    }
76}
77
78/// Authorizes a run-document read: scopes the caller to `namespace` and
79/// verifies the workflow exactly as `describe` does, then requires
80/// `content_hash` to be recorded on one of the workflow's `WorkflowStarted`
81/// events (any generation of its continue-as-new chain).
82///
83/// # Errors
84///
85/// - `InvalidInput` when `content_hash` is not the canonical 64-character
86///   lowercase hexadecimal form — refused before any store read, so a
87///   malformed path never reaches the archive set. This check runs BEFORE
88///   the namespace scope on purpose: an ungranted caller probing with a
89///   malformed hash learns only that its own input is malformed (zero bits
90///   about any tenant's workflows or packages), and refusing garbage before
91///   engine resolution is the right economics. Do not "fix" the ordering.
92/// - the namespace guard's refusal (`namespace_denied`) when the caller may
93///   not read this workflow.
94/// - `WorkflowNotFound` when the workflow has no history in this scope.
95/// - [`RUN_PACKAGE_NOT_RECORDED`] (not found) when the workflow exists but no
96///   generation of it started under `content_hash`.
97pub async fn authorize_run_document(
98    guard: &NamespaceGuard,
99    caller: &CallerIdentity,
100    namespace: &str,
101    workflow_id: &WorkflowId,
102    content_hash: &str,
103) -> Result<RunDocumentAccess, WireError> {
104    require_canonical_hash(content_hash)?;
105    let target = WorkflowTarget::workflow(workflow_id);
106    let scoped = guard
107        .scope(
108            caller,
109            &NamespaceOperation::read_document(namespace, target),
110        )
111        .await
112        .map_err(|error| error.to_wire_error())?;
113    let engine = scoped.engine().map_err(|error| error.to_wire_error())?;
114    let history = engine
115        .store()
116        .read_history(workflow_id)
117        .await
118        .map_err(|error| ServerError::from(error).to_wire_error())?;
119    if history.is_empty() {
120        return Err(workflow_not_found_error(workflow_id));
121    }
122    let workflow_type = generation_started_under(&history, content_hash).ok_or_else(|| {
123        WireError::not_found_with_type(
124            RUN_PACKAGE_NOT_RECORDED,
125            format!(
126                "workflow {workflow_id} recorded no generation started under package \
127                 {content_hash}"
128            ),
129        )
130    })?;
131    Ok(RunDocumentAccess {
132        namespace: scoped.namespace().to_owned(),
133        workflow_type,
134        content_hash: content_hash.to_owned(),
135        engine: std::sync::Arc::clone(engine),
136    })
137}
138
139/// The workflow type of the LATEST generation whose `WorkflowStarted` recorded
140/// `content_hash`, or `None` when no generation did.
141fn generation_started_under(history: &[Event], content_hash: &str) -> Option<String> {
142    history.iter().rev().find_map(|event| match event {
143        Event::WorkflowStarted {
144            workflow_type,
145            package_version,
146            ..
147        } if package_version.as_str() == content_hash => Some(workflow_type.clone()),
148        _ => None,
149    })
150}
151
152/// The package hash's canonical textual form: 64 lowercase hexadecimal
153/// characters, exactly what `WorkflowStarted.package_version` records.
154fn require_canonical_hash(content_hash: &str) -> Result<(), WireError> {
155    let canonical = content_hash.len() == 64
156        && content_hash
157            .bytes()
158            .all(|byte| matches!(byte, b'0'..=b'9' | b'a'..=b'f'));
159    if canonical {
160        Ok(())
161    } else {
162        Err(WireError::invalid_input(
163            "content_hash must be the package's 64-character lowercase hexadecimal content \
164             hash, as `package_version` on the run's summary carries it",
165        ))
166    }
167}
168
169#[cfg(test)]
170mod tests {
171    use aion_core::{Event, PackageVersion, RunId};
172    use aion_package::AwlSource;
173    use aion_proto::WireErrorCode;
174    use aion_store::WriteToken;
175
176    use super::super::test_support::{
177        NAMESPACE, append_started, context, event_envelope, payload, workflow_id,
178    };
179    use super::{RUN_PACKAGE_NOT_RECORDED, authorize_run_document};
180    use crate::awl::deployed::fixtures::{DOCUMENT, manifest, record};
181
182    /// The fixture start event's hash (`test_support::started_event`).
183    const STARTED_HASH: &str = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa";
184
185    #[tokio::test]
186    async fn a_scoped_caller_reads_the_document_of_the_hash_the_run_recorded()
187    -> Result<(), Box<dyn std::error::Error>> {
188        let context = context().await?;
189        context.ownership.record(workflow_id(), NAMESPACE)?;
190        // The run's own generation: type `fixture`, started under the archive's hash.
191        let row = record(
192            manifest("fixture"),
193            Some(AwlSource::new(
194                "fixture.awl",
195                DOCUMENT,
196                std::iter::empty::<(String, Vec<u8>)>(),
197            )),
198            1_700_000_000,
199        )?;
200        let hash = row.content_hash.clone();
201        context.store.put_package(row).await?;
202        context
203            .store
204            .append(
205                WriteToken::recorder(),
206                &workflow_id(),
207                &[Event::WorkflowStarted {
208                    envelope: event_envelope(1),
209                    workflow_type: "fixture".to_owned(),
210                    input: payload()?,
211                    run_id: RunId::new(uuid::Uuid::from_u128(1)),
212                    parent_run_id: None,
213                    parent_workflow_id: None,
214                    package_version: PackageVersion::new(hash.clone()),
215                }],
216                0,
217            )
218            .await?;
219
220        let access = authorize_run_document(
221            &context.guard,
222            &context.caller,
223            NAMESPACE,
224            &workflow_id(),
225            &hash,
226        )
227        .await?;
228        assert_eq!(access.namespace, NAMESPACE);
229        assert_eq!(access.workflow_type, "fixture");
230        let document = access.read().await?;
231        assert_eq!(document.content_hash, hash);
232        assert_eq!(document.workflow_type, "fixture");
233        assert_eq!(document.source, DOCUMENT);
234        Ok(())
235    }
236
237    /// Two generations of one workflow, each under its own hash and type: each
238    /// hash resolves to the type of the generation that recorded it, and when
239    /// a later generation records an already-recorded hash under a new type,
240    /// the LATEST generation's type labels the answer (rev scan, not first
241    /// match).
242    #[tokio::test]
243    async fn each_hash_resolves_to_the_latest_generation_that_recorded_it()
244    -> Result<(), Box<dyn std::error::Error>> {
245        let context = context().await?;
246        context.ownership.record(workflow_id(), NAMESPACE)?;
247        let hash_a = "a".repeat(64);
248        let hash_b = "b".repeat(64);
249        let started = |seq: u64,
250                       workflow_type: &str,
251                       hash: &str|
252         -> Result<Event, Box<dyn std::error::Error>> {
253            Ok(Event::WorkflowStarted {
254                envelope: event_envelope(seq),
255                workflow_type: workflow_type.to_owned(),
256                input: payload()?,
257                run_id: RunId::new(uuid::Uuid::from_u128(u128::from(seq))),
258                parent_run_id: None,
259                parent_workflow_id: None,
260                package_version: PackageVersion::new(hash.to_owned()),
261            })
262        };
263        // gen 1: fixture @ A; gen 2: fixture_v2 @ B; gen 3: fixture_renamed @ A.
264        context
265            .store
266            .append(
267                WriteToken::recorder(),
268                &workflow_id(),
269                &[
270                    started(1, "fixture", &hash_a)?,
271                    started(2, "fixture_v2", &hash_b)?,
272                    started(3, "fixture_renamed", &hash_a)?,
273                ],
274                0,
275            )
276            .await?;
277
278        let under_b = authorize_run_document(
279            &context.guard,
280            &context.caller,
281            NAMESPACE,
282            &workflow_id(),
283            &hash_b,
284        )
285        .await?;
286        assert_eq!(under_b.workflow_type, "fixture_v2");
287        let under_a = authorize_run_document(
288            &context.guard,
289            &context.caller,
290            NAMESPACE,
291            &workflow_id(),
292            &hash_a,
293        )
294        .await?;
295        assert_eq!(
296            under_a.workflow_type, "fixture_renamed",
297            "the LATEST generation that recorded the hash labels the answer"
298        );
299        Ok(())
300    }
301
302    #[tokio::test]
303    async fn a_hash_the_workflow_never_started_under_is_not_found_by_its_own_type()
304    -> Result<(), Box<dyn std::error::Error>> {
305        let context = context().await?;
306        context.ownership.record(workflow_id(), NAMESPACE)?;
307        append_started(context.store.as_ref()).await?;
308        // A real, persisted archive the caller could otherwise read — the
309        // workflow simply never ran it. Reachability through the workflow
310        // must not leak it.
311        let foreign = record(
312            manifest("fixture"),
313            Some(AwlSource::new(
314                "fixture.awl",
315                DOCUMENT,
316                std::iter::empty::<(String, Vec<u8>)>(),
317            )),
318            1_700_000_000,
319        )?;
320        let foreign_hash = foreign.content_hash.clone();
321        context.store.put_package(foreign).await?;
322
323        let error = authorize_run_document(
324            &context.guard,
325            &context.caller,
326            NAMESPACE,
327            &workflow_id(),
328            &foreign_hash,
329        )
330        .await
331        .err()
332        .ok_or("a hash the run never recorded must be refused")?;
333        assert_eq!(error.code, WireErrorCode::NotFound);
334        assert_eq!(error.error_type.as_deref(), Some(RUN_PACKAGE_NOT_RECORDED));
335        Ok(())
336    }
337
338    #[tokio::test]
339    async fn a_recorded_hash_with_no_persisted_archive_is_the_archive_reader_s_not_found()
340    -> Result<(), Box<dyn std::error::Error>> {
341        let context = context().await?;
342        context.ownership.record(workflow_id(), NAMESPACE)?;
343        append_started(context.store.as_ref()).await?;
344
345        let access = authorize_run_document(
346            &context.guard,
347            &context.caller,
348            NAMESPACE,
349            &workflow_id(),
350            STARTED_HASH,
351        )
352        .await?;
353        let error = access
354            .read()
355            .await
356            .err()
357            .ok_or("an unpersisted archive must be refused")?;
358        assert_eq!(error.code, WireErrorCode::NotFound);
359        assert_eq!(error.error_type.as_deref(), Some("DeployedVersionNotFound"));
360        Ok(())
361    }
362
363    #[tokio::test]
364    async fn a_malformed_hash_is_refused_before_any_read() -> Result<(), Box<dyn std::error::Error>>
365    {
366        let context = context().await?;
367        for malformed in ["", "abc", &"A".repeat(64), &"g".repeat(64), &"a".repeat(63)] {
368            let error = authorize_run_document(
369                &context.guard,
370                &context.caller,
371                NAMESPACE,
372                &workflow_id(),
373                malformed,
374            )
375            .await
376            .err()
377            .ok_or_else(|| format!("{malformed:?} must be refused"))?;
378            assert_eq!(error.code, WireErrorCode::InvalidInput, "{malformed:?}");
379        }
380        Ok(())
381    }
382
383    #[tokio::test]
384    async fn an_unowned_workflow_is_refused_by_the_namespace_guard()
385    -> Result<(), Box<dyn std::error::Error>> {
386        let context = context().await?;
387        // Owned by ANOTHER namespace; the caller is scoped to NAMESPACE.
388        context.ownership.record(workflow_id(), "tenant-b")?;
389        append_started(context.store.as_ref()).await?;
390
391        let error = authorize_run_document(
392            &context.guard,
393            &context.caller,
394            NAMESPACE,
395            &workflow_id(),
396            STARTED_HASH,
397        )
398        .await
399        .err()
400        .ok_or("a foreign workflow must be refused")?;
401        // The guard's anti-leak answer: the workflow is "not found in
402        // namespace tenant-a" — the same words a truly absent workflow gets —
403        // and NOT the hash verdict, which would confirm the workflow exists.
404        // (The hash IS recorded, so a guard that let the caller through would
405        // have returned an access, not an error.)
406        assert_eq!(error.code, WireErrorCode::NotFound, "{error:?}");
407        assert_eq!(error.error_type, None, "{error:?}");
408        assert!(
409            error.message.contains("not found in namespace tenant-a"),
410            "{error:?}"
411        );
412        Ok(())
413    }
414
415    #[tokio::test]
416    async fn an_unknown_workflow_is_workflow_not_found() -> Result<(), Box<dyn std::error::Error>> {
417        let context = context().await?;
418        context.ownership.record(workflow_id(), NAMESPACE)?;
419        let error = authorize_run_document(
420            &context.guard,
421            &context.caller,
422            NAMESPACE,
423            &workflow_id(),
424            STARTED_HASH,
425        )
426        .await
427        .err()
428        .ok_or("a workflow with no history must be not found")?;
429        assert_eq!(error.code, WireErrorCode::NotFound);
430        assert_eq!(error.error_type.as_deref(), Some("WorkflowNotFound"));
431        Ok(())
432    }
433}
aion_server/api/handlers/document.rs

aion_server/api/handlers/
document.rs
