You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@bifromq.apache.org by gu...@apache.org on 2026/10/10 19:14:55 UTC
(bifromq-tenon) branch main updated: Support runtime policy updates and custom plugin package formats (#17)
This is an automated email from the ASF dual-hosted git repository.
Gujiawei-Edinburgh pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/bifromq-tenon.git
The following commit(s) were added to refs/heads/main by this push:
new 5ba53a1 Support runtime policy updates and custom plugin package formats (#17)
5ba53a1 is described below
commit 5ba53a15adcc1abb965130716ce768e2cf9131df
Author: Yonny(Yu) Hao <po...@gmail.com>
AuthorDate: Sun Oct 11 03:14:50 2026 +0800
Support runtime policy updates and custom plugin package formats (#17)
* Add Runner policy updates and plugin package extensions
* Remove plugin input size limits
---
SECURITY.md | 10 +-
guide/http-api.md | 6 +-
guide/plugins.md | 40 +-
guide/quickstart.md | 2 +-
guide/runner-extensions.md | 109 +++++-
guide/runner.md | 2 +-
plugin/README.md | 2 +-
src/lib.rs | 9 +-
src/runner/document_store/mod.rs | 8 +-
src/runner/document_store/tests.rs | 28 +-
src/runner/execution.rs | 160 +++++++-
src/runner/{extensions.rs => extensions/mod.rs} | 109 +++++-
src/runner/extensions/tar_gz.rs | 300 ++++++++++++++
src/runner/http/mod.rs | 2 +-
src/runner/http/plugins/mod.rs | 12 +-
src/runner/http/plugins/tests.rs | 62 ++-
src/runner/main_loop/failure.rs | 13 +
src/runner/main_loop/mod.rs | 10 +-
src/runner/main_loop/shutdown.rs | 2 +-
src/runner/main_loop/startup.rs | 67 +++-
src/runner/main_loop/tests.rs | 5 +-
src/runner/management/jobs.rs | 7 +-
src/runner/management/state/mod.rs | 34 +-
src/runner/management/supervisor/mod.rs | 45 ++-
.../management/supervisor/tests/policy_changes.rs | 160 ++++++++
src/runner/management/supervisor/tests/programs.rs | 21 +-
src/runner/management/supervisor/tests/uploads.rs | 8 +-
src/runner/mod.rs | 10 -
src/runner/pipeline/programs/tests.rs | 19 +-
src/runner/pipeline/runtime_resolution/tests.rs | 32 +-
src/runner/pipeline/test_support.rs | 15 +-
src/runner/plugin/package/mod.rs | 42 +-
src/runner/plugin/package/safety/mod.rs | 433 +++------------------
src/runner/plugin/package/safety/output.rs | 190 +++++++++
src/runner/plugin/package/safety/output/tests.rs | 119 ++++++
src/runner/plugin/package/safety/tests.rs | 75 ++--
src/runner/plugin/package/tests.rs | 57 ++-
src/runner/plugin/store/filesystem.rs | 96 +----
src/runner/plugin/store/mod.rs | 160 +++++---
src/runner/plugin/store/package_input.rs | 302 ++++++++++++++
src/runner/plugin/store/protected.rs | 163 ++++++++
src/runner/plugin/store/tests.rs | 224 ++++++-----
src/runner/plugin/store/tests/protected.rs | 181 +++++++++
src/runner/plugin/store/upload.rs | 171 --------
src/runner/recovery/mod.rs | 32 +-
src/runner/recovery/tests.rs | 194 +++++----
src/runner/test_support.rs | 18 +-
tests/fixtures/runner_extensions.rs | 96 ++++-
tests/fixtures/runner_extensions/live_policy.rs | 139 +++++++
tests/runner_cli.rs | 50 ++-
tests/runner_cli/plugin_fixture.rs | 72 +++-
tests/runner_extensions.rs | 37 +-
tests/runner_extensions/artifacts.rs | 228 +++++++++++
tests/runner_extensions/http_authorization.rs | 4 +-
tests/runner_extensions/policy_changes.rs | 133 +++++++
tests/runner_http.rs | 115 ++----
tests/runner_http/flow_contract.rs | 5 +-
tests/runner_java_archetype.rs | 15 +-
tests/runner_java_archetype/topology.rs | 5 +-
tests/runner_metrics.rs | 8 +-
tests/runner_metrics/flows.rs | 5 +-
tests/runner_metrics/performance.rs | 5 +-
tests/runner_metrics/states.rs | 14 +-
tests/runner_resource_limits.rs | 3 +-
tests/runner_resource_limits/linux.rs | 31 +-
tools/example/local.sh | 2 +-
tools/verify-container-resource-limits.py | 2 +-
tools/verify-debug-plugins.py | 2 +-
tools/verify-rust-scaffold.py | 4 +-
69 files changed, 3426 insertions(+), 1315 deletions(-)
diff --git a/SECURITY.md b/SECURITY.md
index b0a7ab1..6394659 100644
--- a/SECURITY.md
+++ b/SECURITY.md
@@ -95,7 +95,9 @@ Lua scripts can transform or discard data, send to configured Sinks and reveal d
Linux Document CPU and memory limits require delegated cgroup v2 support. On macOS, these limits are ignored and execution continues. Per-VM and per-Flow limits do not impose a total node budget. See [resource enforcement](guide/runner.md#cpu-and-memory). The supplied [Docker configuration](deploy/container/README.md) grants `CAP_SYS_ADMIN` for cgroup management. That broad capability is inherited by child programs; use trusted plugins and retain container security policies.
-The HTTP API has no total Document or compressed-upload byte limit and no built-in per-principal rate or execution quota policy. Archive extraction has separate fixed bounds, which do not bound aggregate concurrent uploads, retained installations or all management costs. Deployers must set request size, rate, connection, execution, disk and resource budgets appropriate to their environment and protect the listener against network exhaustion. A TLS handshake deadline is not general denial [...]
+The HTTP API has no total byte limit for a Document or a received plugin package. The core applies fixed directory limits to every package format. The default decoder keeps limits on the decompressed tar stream and tar entries. These limits do not control total upload concurrency, installed data size, or all management costs. The core has no built-in rate limit or execution quota per caller.
+
+Deployers must set request size, rate, connection, execution, disk, and resource limits for their environment. Protect the listener against network exhaustion. A TLS handshake deadline does not prevent all denial-of-service attacks.
## Storage, secrets and observations
@@ -105,11 +107,13 @@ Use a separate state directory for each Runner and prevent access by untrusted w
## Extension responsibilities
-A custom Runner distribution can provide HTTP authorization, execution admission and Document storage protection. These features require custom code; they are not enabled by installing a plugin. See the [extension guide](guide/runner-extensions.md).
+A custom Runner distribution can provide HTTP authorization, execution admission and artifact protection. These features require custom code; they are not enabled by installing a plugin. See the [extension guide](guide/runner-extensions.md).
- HTTP authorization can restrict methods and paths using request headers. The hook does not receive the body, query or TLS client identity. Policies that need those inputs require a gateway or other distribution support.
- Execution admission can restrict the saved Document set or impose an entitlement deadline. It does not authenticate API callers or inspect plugin uploads.
-- Document protection can encrypt stored Documents. GET responses, plugin configuration and runtime memory still contain plaintext; protect access to those separately.
+- Artifact protection can encrypt stored Documents and decode package formats through the core directory writer. The core validates the directory before publication. A failed hook cannot publish partial output.
+
+ The core keeps the received package as the durable source for both standard archives and envelopes. GET responses, plugin configuration, and runtime files still contain plaintext. A host administrator can copy those files. The extension cannot prevent use of copied standard plugin materials.
Custom distributions must document their policies, key management and failure behavior. The default Runner allows all requests and execution and stores Documents as plaintext.
diff --git a/guide/http-api.md b/guide/http-api.md
index f4694af..cdfb926 100644
--- a/guide/http-api.md
+++ b/guide/http-api.md
@@ -45,7 +45,7 @@ GET returns original `application/jsonc` bytes and a strong ETag. Preserve the c
DELETE of an existing Document requires the exact `If-Match` and returns `204` once the saved Document has been removed. An absent Document is an idempotent `204`. A successful save means the original bytes are durable and admitted; application proceeds separately. After a lost response, read the Document and compare bytes/ETags before deciding whether to retry. Do not blindly overwrite concurrent edits.
-The API imposes no total Document or compressed upload byte limit. Deployment admission and resource controls must fit the environment. JSONC applies to Document/configuration input, while ordinary API responses are JSON. Schemas use `application/schema+json`; payload descriptors use `application/x-protobuf`.
+The API imposes no total Document byte limit. The API and Store impose no total byte limit on received plugin packages. All formats use the same [directory limits](plugins.md#package-structure). Deployment admission and resource controls must fit the environment. JSONC applies to Document/configuration input, while ordinary API responses are JSON. Schemas use `application/schema+json`; payload descriptors use `application/x-protobuf`.
## Runtime status
@@ -61,7 +61,9 @@ An unready status includes `runtimeIssues`, such as missing Program, platform mi
## Packages and errors
-POST `/plugins` accepts raw gzip tar bytes with `Content-Type: application/vnd.apache.tenon.plugin+tar+gzip`. It does not accept multipart, Base64 or a download URL. Identity and interface come from `manifest.json`. A new installation returns `201`, an identical normalized file set returns `204`, and conflicting content for an installed identity returns `409`. See [package rules](plugins.md).
+POST `/plugins` accepts package bytes only with `Content-Type: application/octet-stream`. The package hook selects the format from the bytes. The default hook accepts `tar.gz`; a [custom hook](runner-extensions.md#plugin-packages) can accept other formats. Multipart, Base64, and download URLs are not accepted.
+
+Identity and interface come from `manifest.json`. A new installation returns `201`. An identical normalized file set returns `204`; conflicting content for an installed identity returns `409`. The Store keeps the received bytes and reconstructs runtime files at each Runner startup. See the [package rules](plugins.md).
Uninstallation is blocked while persistent Documents or active execution still reference the Program. Removing a Document does not guarantee that its old processes have already exited. Query again after shutdown rather than forcing removal.
diff --git a/guide/plugins.md b/guide/plugins.md
index a5af5c4..aad0b54 100644
--- a/guide/plugins.md
+++ b/guide/plugins.md
@@ -25,7 +25,7 @@ A Program is identified by `programName` and an exact semantic version. It imple
## Package structure
-Each distributable is a gzip tar containing these fixed root files:
+The core requires a plugin directory with these fixed root files:
```text
manifest.json
@@ -35,9 +35,15 @@ payload.descriptor.pb
Other ordinary files contain the launcher, executable, libraries, optional runtime and resources. The [manifest Schema](../contracts/plugin/manifest.schema.json) defines identity, display metadata, interface, a nonempty `platforms` list and an argv `command`. The descriptor must include the standard top-level `SourceRecordPayload` and/or `SinkRecordPayload` roots exactly for the interfaces implemented. Include source information in the descriptor; the standard packaging tools generate it.
-Tar paths must be relative and canonical. Absolute paths, `.`/`..`, duplicate normalized paths, symlinks, hard links and special files are rejected. Directory entries and archived permission bits are not package identity. Installed directories use `0700` and ordinary files `0500`.
+Package paths must be relative and canonical. Absolute paths, `.`/`..`, duplicate normalized paths, symlinks, hard links and special files are rejected. Directory entries and package permission bits are not package identity. Extracted runtime directories use `0700` and ordinary files `0500`.
-`command` is executed without a shell, with the installation directory as cwd. Bare executable names use the inherited PATH. Tenon appends its reserved `--sdk-config` argument; do not include it yourself. Use stdout and stderr for diagnostic text.
+The standard generators produce `tar.gz` packages. The default package hook decodes this format. A custom hook can decode another format into the same directory contract. Upload every format with `Content-Type: application/octet-stream`.
+
+The core does not impose a total byte limit on received packages. Runtime contents have a 4 GiB total file limit and a 100,000-entry limit, including implicit directories. `manifest.json` and `config.schema.json` each have a 16 MiB limit. `payload.descriptor.pb` has a 64 MiB limit. Paths have a 4,096-byte limit; each component has a 255-byte limit.
+
+The default decoder limits the decompressed tar stream to 8 GiB. It does not impose a separate byte limit on compressed input. It permits at most 100,000 tar entries. Each archive metadata record has a 64 KiB limit. Custom decoders must bound their format-specific work. The core applies directory limits to all formats.
+
+The Runner executes `command` without a shell, with the extracted runtime directory as cwd. Bare executable names use the inherited PATH. Tenon appends its reserved `--sdk-config` argument; do not include it yourself. Use stdout and stderr for diagnostic text.
## Display metadata
@@ -55,7 +61,29 @@ Package authors must declare genuine compatibility and document external depende
On the same Runner, the normalized ordinary-file paths and every file's original bytes determine duplicate installation. Equal content is idempotent; any difference for the same identity conflicts. Across platform variants of one logical version, interface, Config Schema, payload descriptors and business semantics must remain identical, while launcher/runtime/platform material may differ. Change the exact version for a semantic change.
-Install through POST `/plugins`, then refer to the exact identity from a Document. To upgrade, install the new version, update the Document, observe application and shutdown of the old instance, then uninstall the old version. Do not modify the installed tree. Damaged installed packages can stop the Runner or leave dependent Documents unready; restore them by reinstalling the original bundle.
+Install through POST `/plugins`, then refer to the exact identity from a Document. To upgrade, install the new version, update the Document, observe application and shutdown of the old instance, then uninstall the old version. Do not modify saved packages or runtime files while the Runner runs. A missing package leaves dependent Documents `unready`. A saved package that fails validation stops startup. Stop the Runner before you restore a damaged package from a known valid copy.
+
+## Package storage and recovery
+
+The Program Store keeps the received package bytes without changes. Standard packages remain `tar.gz` archives. Distribution envelopes remain envelopes. Extracted runtime files are temporary. They are not the durable source.
+
+All packages use this path under the state directory:
+
+```text
+plugins/programs/<programName>/.tenon-artifact-<versionHash>
+```
+
+`versionHash` is the lowercase hexadecimal SHA-256 of the exact version string in UTF-8. It keeps the filename length fixed. It does not authenticate the package. Package files must be ordinary files with mode `0600` and one hard link. Namespace directories must have mode `0700`.
+
+Installation first calls the package hook to write a private runtime directory. The core then validates that directory. It then synchronizes the received package file, renames it to the final path, and synchronizes the parent directory. Only then does the Store publish the validated Entry. An equal normalized file set keeps the existing package bytes and Entry, even if the new archive bytes differ.
+
+Each Runner startup reconstructs runtime files from the saved packages before it serves requests or starts Pipelines. Recovery validates package contents and requires the package identity to match its path. A package access, format, permission, or identity failure stops startup and keeps that package. Recovery keeps valid packages for other platforms; their Documents remain `unready` on this platform.
+
+Recovery removes incomplete uploads, deletion tombstones, and entries outside the supported layout. An invalid or unreadable namespace is removed as a whole. Cleanup does not follow symbolic links. A failure to read the Store root or complete required cleanup stops startup.
+
+A stopped Runner can receive a package through a file copy to the path above, with the required permissions. Runtime files are never a recovery source. Plugin process restarts within one Runner reuse the validated runtime files. Normal shutdown removes these files after process cleanup. If process ownership is uncertain, the runtime directory remains for the existing recovery process.
+
+Uninstall requires the absence of Document and runtime references. The Store renames the saved package to a deletion tombstone and synchronizes its parent. It then removes the Entry, runtime files, and tombstone, and synchronizes the parent again. A failure after rename can leave a committed deletion. After Runner recovery, query the identity before a retry.
## SDKs and generators
@@ -70,3 +98,7 @@ The [quickstart](quickstart.md) builds an entirely local generated plugin and ex
## Build artifacts for distribution
Use `cargo tenon bundle --release` for a Rust distribution; the quickstart's default debug bundle is for local testing. Rust binaries can contain build-machine paths in diagnostics and debug information. For a distributable, set compiler `--remap-path-prefix` mappings for the checkout, Cargo cache and any local dependencies through `RUSTFLAGS` or `CARGO_ENCODED_RUSTFLAGS`, then inspect the finished archive. A release profile alone does not remove every embedded source path. Keep build lo [...]
+
+## Custom package formats
+
+A custom Runner can decode another package format through `ArtifactProtection`. It can also decode an envelope around a standard package. The core validates the resulting directory and owns all files and processes. Standard generators and the default Runner continue to use `tar.gz`. See the [extension contract](runner-extensions.md#plugin-packages) for storage, recovery, access failures, and runtime cleanup.
diff --git a/guide/quickstart.md b/guide/quickstart.md
index 0ca93f3..bfbeea9 100644
--- a/guide/quickstart.md
+++ b/guide/quickstart.md
@@ -59,7 +59,7 @@ Once the Runner is running, the essential API calls are:
```sh
curl --fail -i -X POST http://127.0.0.1:18080/plugins \
- -H 'Content-Type: application/vnd.apache.tenon.plugin+tar+gzip' \
+ -H 'Content-Type: application/octet-stream' \
--data-binary @/absolute/path/to/hello-tenon.tar.gz
curl --fail -i -X PUT http://127.0.0.1:18080/documents/hello \
-H 'Content-Type: application/jsonc' -H 'If-None-Match: *' \
diff --git a/guide/runner-extensions.md b/guide/runner-extensions.md
index 2162e0f..832a9d8 100644
--- a/guide/runner-extensions.md
+++ b/guide/runner-extensions.md
@@ -21,7 +21,7 @@ under the License.
Tenon is part of Apache BifroMQ (Incubating). See the [incubation disclaimer](../DISCLAIMER).
-A Rust executable can reuse Tenon's Runner and Pipeline with three hooks: HTTP request authorization, execution admission, and complete-Document storage protection. Tenon does not supply licensing, billing, encryption algorithms or credential management. The public interfaces are exported by the `tenon` library.
+A Rust executable can use Tenon's Runner and Pipeline with three hooks: HTTP request authorization, execution admission, and artifact protection. Tenon does not supply licensing, billing, encryption algorithms or credential management. The `tenon` library exports the public interfaces.
The [security policy and threat model](../SECURITY.md#extension-responsibilities) describes the limits of each hook and the responsibilities of a custom distribution. Hook availability alone does not enable access control or turn plugin processes into security sandboxes.
@@ -33,25 +33,114 @@ fn main() -> std::process::ExitCode {
`run_main_with` receives a one-time initializer over validated `RunnerConfig` and returns `ExitCode`. It runs once at Runner startup, before saved Documents are loaded or HTTP requests are accepted. Distribution settings belong to the optional `extra` object. Sanitize errors returned during initialization.
-`RunnerHooks::new(policy, protection, authorization)` installs concrete implementations. Defaults are `AllowAll`, `Plaintext` and `NoHttpAuth`.
+`RunnerHooks::new(policy, protection, authorization)` installs concrete implementations. Defaults are `AllowAll`, `ByPass` and `NoHttpAuth`.
-## DocumentProtection
+## ArtifactProtection
-`DocumentProtection: Send + Sync` exposes synchronous `protect(source: &[u8], output: &mut dyn Write)` and `unprotect(stored: &[u8], output: &mut dyn Write)`, both returning `io::Result<()>`. Implement the byte transformation; Tenon handles file storage.
+`ArtifactProtection: Send + Sync` controls Document storage bytes and package decoding. All three methods have default implementations. `protect` and `unprotect` write unchanged Document bytes. `open_plugin_package` decodes a standard `tar.gz` package. `ByPass` uses these defaults without encryption or decryption.
-A failed `protect` leaves the previous saved Document intact. A failed `unprotect` prevents startup and preserves the file. Methods may be called concurrently for different Documents. Bound their memory and computation costs; cancellation does not roll back a file operation already underway.
+A custom implementation can override any method. It can call the default decoder after it opens its own envelope:
-ETags, GET responses and plugin configuration always use original plaintext bytes. Randomized protected bytes do not change the ETag. Storage protection is not API authorization, field-level secret expansion or encryption of runtime memory.
+```rust
+use std::io::{self, Read};
+use tenon::{ArtifactProtection, ByPass, PluginPackageOutput};
+
+struct DistributionProtection;
+
+impl ArtifactProtection for DistributionProtection {
+ fn open_plugin_package(
+ &self,
+ source: &mut dyn Read,
+ output: &mut dyn PluginPackageOutput,
+ ) -> io::Result<()> {
+ ByPass.open_plugin_package(source, output)
+ }
+}
+```
+
+This example delegates unchanged input. An envelope implementation supplies its decoded reader instead. The Document methods keep their defaults.
+
+The core owns files, paths, permissions, commits, and cleanup. The package hook receives a reader and a directory writer. It does not receive a destination path or a process handle. Methods run in blocking Store work and can run concurrently. The implementation must bound its memory use and work. Cancellation does not interrupt a file operation that has started.
+
+### Documents
+
+`protect` writes the stored representation. `unprotect` writes the original Document bytes. If `protect` fails before file replacement, the saved Document stays unchanged. A failure after replacement does not imply rollback. If `unprotect` fails, startup stops and keeps the saved file. The core discards partial output.
+
+ETags, GET responses, and plugin configuration use the original plaintext. Random protected bytes do not change the ETag. Document protection does not provide API authorization or protect runtime memory.
+
+### Plugin packages
+
+The core receives the full upload in a temporary file inside the Program Store. Installation and recovery read saved package files through the same decoding and validation process. After installation checks pass, the core renames the temporary original to its final path. A rejected package does not replace an installed package.
+
+`PluginPackageOutput` is a public trait. The Runner supplies its private implementation as `&mut dyn PluginPackageOutput`. The trait exposes only `write_file` and `create_dir`. The Runner owns the directory, failure records, and final validation. The hook decodes its format and sends each file through this trait. The core does not require a tar archive between the hook and directory validation.
+
+`output.write_file(relative_path, reader)` writes one ordinary file to EOF and creates missing parent directories. `output.create_dir(relative_path)` creates an explicit directory. The writer checks paths, duplicates, file conflicts, entry counts, and byte limits. It sets fixed permissions. It has no operation for links or special files. See the [directory contract and limits](plugins.md#package-structure).
+
+The hook must finish format, integrity, and access checks before it returns success. A recognized envelope that fails a check must return an error. It must not fall back to another format. A streaming encryption format must detect truncation and missing or reordered content. No output can run before the hook succeeds and the core validates the complete directory.
+
+| Result | Core action |
+| --- | --- |
+| `Ok(())` with no recorded stream or output failure | Validate the directory. Save the received input without changes. Publish the temporary runtime files. |
+| Hook error with `InvalidData` or `UnexpectedEof` | Reject a new upload with HTTP 422 and `plugin_package_invalid`. Discard the temporary directory. |
+| Hook error with `FileTooLarge` | Reject a new upload with HTTP 413 and `plugin_package_too_large`. Discard the temporary directory. |
+| Other hook error | Reject a new upload with HTTP 422 and `plugin_package_open_failed`. Discard the temporary directory. |
+| Invalid output | Return the package validation error. A size limit returns HTTP 413. Discard the temporary directory. |
+| Local input or output filesystem failure | Stop the Runner. Discard the temporary directory if cleanup succeeds. |
+
+The method returns `io::Result<()>`. A hook can put a custom error in `io::Error`. The core uses the error kind for the categories above. It does not inspect the custom error type. Use `PermissionDenied` to reject package access. Do not include secrets in errors.
+
+The core records failures at its own reader and writer. These failures take priority over the hook result. The core does not identify local filesystem failures from `ErrorKind` alone. Returning success cannot hide a recorded failure. Any failure during recovery stops startup and keeps the saved package.
+
+The default decoder checks gzip, tar headers, archive metadata, and end markers. It uses only the public directory writer and `io::Error`, as a custom decoder does. The writer checks file size limits against the bytes it reads. The decoder also checks that each file has the length stated in its tar header. The core checks the manifest, platform, Schema, descriptor, and normalized file contents. HTTP accepts only `application/octet-stream` for every format.
+
+All plugins use the same storage process. A standard package stays as the received `tar.gz`. An envelope stays as the received envelope. The hook does not choose a storage form. The Program Store owns package files, validation, publication, and cleanup.
+
+Each Runner startup calls the hook for every saved package and reconstructs the runtime files. A stopped Runner can receive either package type through a file copy. The [package storage contract](plugins.md#package-storage-and-recovery) defines paths, permissions, identity checks, and recovery failures. The default Runner accepts standard archives at these paths. Other formats require a custom hook.
+
+Identity stays `programName + exactVersion`. An identical normalized file set returns HTTP 204 and keeps the existing storage bytes and Entry. A new envelope does not replace an existing one. Different content returns HTTP 409. Envelope replacement requires uninstall and installation, or a new version.
+
+The hook runs at installation and each Runner recovery. Process restarts use the verified runtime files again. Plans keep the existing shared Entry until their process owners finish. Uninstall requires no external Entry references. It removes the stored input and runtime files.
+
+Normal shutdown removes runtime files after process cleanup. If process ownership is uncertain, the existing recovery rules keep the runtime directory.
+
+A host administrator can copy plaintext runtime files. This interface does not prevent that copy or prevent execution of a repackaged standard plugin elsewhere.
## ExecutionPolicy
-`authorize(ExecutionScope)` synchronously receives the complete proposed set of admitted, statically verified Documents and returns `Result<ExecutionPermit, ExecutionDenied>`. Scope references are borrowed for that call and have no ordering guarantee. Quota decisions use the proposed Document set. Missing-plugin Documents can still consume admission quota.
+```rust
+pub type PolicyChanges = tokio::sync::watch::Receiver<()>;
+
+pub trait ExecutionPolicy {
+ fn authorize(&self, scope: ExecutionScope<'_>)
+ -> Result<ExecutionPermit, ExecutionDenied>;
+
+ fn changes(&self) -> Option<PolicyChanges> {
+ None
+ }
+}
+```
+
+`authorize` receives the full proposed set of accepted, verified Documents. The Runner uses these references only during this call. Set order is unspecified. Accepted Documents with missing plugins stay in the set.
+
+The Runner thread calls the policy synchronously. The policy needs neither `Send` nor `Sync`. It must return promptly without blocking I/O. Callback counts are not a quota ledger or a commit receipt.
+
+The Runner calls `changes()` once, before its first decision. It first evaluates an empty set before Store recovery, even if no Documents exist. Denial stops startup. Recovery then tries Documents one by one in id byte order. A denied candidate stays readable and deletable, with no accepted Pipeline. PUT of identical bytes can retry a denied Document.
+
+PUT evaluates the full set with the candidate after static and conditional checks, before persistence. Candidate denial returns the policy's public code and message as HTTP 403. It keeps the previous accepted set. DELETE removes a Document without a policy call. Plugin installation and process restarts do not cause another decision for unchanged Documents.
+
+Both outcomes carry `entitlement_until: Option<Instant>`. Each outcome immediately replaces the single current deadline. `None` removes it. File success or failure does not change this deadline. Expiry starts bounded shutdown, including when there is no API traffic. It does not mean all processes disappear at once.
+
+A policy updater must validate a full replacement and publish its cached state before it sends a notification. Invalid replacement data must not extend the old deadline. The notification is a latest-state hint; several notifications can merge. The Runner keeps the same receiver across waits. It evaluates the full accepted set when notified, including an empty set. Denial of this current set starts shutdown and keeps Documents.
+
+If a notification arrives during a Document commit, the Runner also evaluates the full set after a successful commit, before Pipeline publication. Denial starts shutdown and keeps the saved Document. Failed commits need no second decision. Without a notification, a commit does not cause a second decision.
+
+The Runner processes pending notifications from recovery before initial Pipeline execution. A closed channel emits `runner.policy_changes_closed`, keeps the current deadline, and disables the notification wait. External shutdown and an already expired deadline take priority in the main loop. An update must be accepted before shutdown starts. Later updates cannot reopen the Runner.
-Calls happen on the Runner thread; the policy need not be Send/Sync and must return promptly without blocking I/O. Do not maintain an independent quota ledger by counting callbacks. An allowed Document can still fail to save.
+The notification has no generation or acknowledgement. It does not atomically change package keys, cancel package work, or add a policy call before each process spawn. Work already started can finish its disk commit during shutdown. It cannot publish new Pipeline targets then.
-Startup tries Documents one by one in id byte order. A denied Document remains readable and deletable but has no admitted Pipeline; retrying identical bytes through PUT rechecks admission. PUT evaluates a replacement in the full desired set after static and conditional checks and before persistence. Denial returns the hook's sanitized public code/message as HTTP 403. DELETE frees desired quota without another policy call. Plugin installation, automatic restart and reconfiguration do not [...]
+Distributions must keep access identity stable during ordinary renewal. A change of customer or decryption identity requires a Runner restart. Old saved envelopes still need compatible keys.
-Both accepted and denied results carry `entitlement_until: Option<Instant>`. Every result immediately replaces the current deadline, including `None` to remove it, independently of whether a later file operation succeeds. Expiry initiates bounded Runner shutdown even without API traffic. It does not promise instantaneous process disappearance. Once shutdown begins, later results cannot reopen admission. Update policy inputs in your distribution; `authorize` runs only during startup admis [...]
+Existing policy implementations can omit `changes()`. The initial empty-set call is new behavior. Implementations must support it even when they do not supply notifications.
## HttpApiAuthorization
diff --git a/guide/runner.md b/guide/runner.md
index 5339cf8..f76348f 100644
--- a/guide/runner.md
+++ b/guide/runner.md
@@ -51,7 +51,7 @@ The deadlines limit Pipeline startup, configuration updates and shutdown. If an
Give each Runner its own writable `stateDirectory`, protected from other writers. The Runner creates its state directories with mode `0700`; existing directories must have that mode and must not be symlinks. Modify Documents and installed packages through the HTTP API.
-Saved Documents and installed plugins survive restarts. A missing or damaged plugin leaves dependent Documents `unready`; reinstall the package to restore them. Invalid or unreadable saved Documents prevent startup. Back up configuration and installed packages with a consistent filesystem snapshot or while the Runner is stopped.
+Saved Documents and plugin packages survive restarts. Each startup reconstructs plugin runtime files from the saved packages. A missing package leaves dependent Documents `unready`. A saved package that fails access or validation checks stops startup and remains on disk. Invalid or unreadable saved Documents also prevent startup. Back up configuration and packages with a consistent filesystem snapshot or while the Runner is stopped. See [package storage and recovery](plugins.md#package-s [...]
In-flight messages, Lua state and timers do not survive restarts. Use upstream replay and downstream deduplication as required by your delivery policy.
diff --git a/plugin/README.md b/plugin/README.md
index 395803f..80b6148 100644
--- a/plugin/README.md
+++ b/plugin/README.md
@@ -53,7 +53,7 @@ Start a local Runner with the [quickstart script](../guide/quickstart.md#run-the
for report in /tmp/tenon-dummy-bundle.json /tmp/tenon-stdout-bundle.json; do
bundle="$(python3 -c 'import json,sys; print(json.load(open(sys.argv[1]))["bundle"])' "$report")"
curl --fail -i -X POST http://127.0.0.1:18080/plugins \
- -H 'Content-Type: application/vnd.apache.tenon.plugin+tar+gzip' \
+ -H 'Content-Type: application/octet-stream' \
--data-binary "@$bundle"
done
```
diff --git a/src/lib.rs b/src/lib.rs
index 85eb6fe..67949a0 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -20,7 +20,7 @@
//! Tenon's executable library boundary.
//!
//! The default binary enters through [`run_main`]. Trusted distributions use
-//! [`run_main_with`] to install HTTP authorization, execution, and Document protection hooks. Runtime
+//! [`run_main_with`] to install HTTP authorization, execution, and artifact protection hooks. Runtime
//! owners stay private; hooks receive immutable configuration and verified facts.
use std::env;
@@ -32,8 +32,9 @@ pub use identifiers::{
ExactVersion, FlowId, PluginInstanceId, PluginProgramIdentity, ProgramName, TenonDocumentId,
};
pub use runner::extensions::{
- AllowAll, DocumentProtection, ExecutionDenied, ExecutionPermit, ExecutionPolicy,
- ExecutionScope, HttpApiAuthorization, HttpAuthRejection, NoHttpAuth, Plaintext, RunnerHooks,
+ AllowAll, ArtifactProtection, ByPass, ExecutionDenied, ExecutionPermit, ExecutionPolicy,
+ ExecutionScope, HttpApiAuthorization, HttpAuthRejection, NoHttpAuth, PluginPackageOutput,
+ PolicyChanges, RunnerHooks,
};
pub use tenon_document::verified::{Flow, PluginInstance};
pub use tenon_document::{ResourceLimits, VerifiedTenonDocument};
@@ -77,7 +78,7 @@ pub fn run_main() -> ExitCode {
/// Runs this distribution with exactly one Runner-only initialization callback.
///
/// The callback receives the fully validated, frozen configuration before Store
-/// recovery and returns both hooks. It is never called in a Pipeline subprocess.
+/// recovery and returns the three hooks. It is never called in a Pipeline subprocess.
/// Initialization failures produce a non-zero exit code without opening the API.
#[must_use]
pub fn run_main_with(
diff --git a/src/runner/document_store/mod.rs b/src/runner/document_store/mod.rs
index 4ecba5f..2699988 100644
--- a/src/runner/document_store/mod.rs
+++ b/src/runner/document_store/mod.rs
@@ -35,7 +35,7 @@
//! external filesystem changes.
use crate::identifiers::TenonDocumentId;
-use crate::runner::extensions::DocumentProtection;
+use crate::runner::extensions::ArtifactProtection;
use sha2::{Digest as _, Sha256};
use std::error::Error;
use std::ffi::OsStr;
@@ -165,7 +165,7 @@ impl TenonDocumentStore {
/// failure.
pub(crate) fn sources(
&self,
- protection: Arc<dyn DocumentProtection>,
+ protection: Arc<dyn ArtifactProtection>,
) -> Result<
impl Iterator<Item = Result<StoredTenonDocumentSource, TenonDocumentStoreError>> + use<>,
TenonDocumentStoreError,
@@ -204,7 +204,7 @@ impl TenonDocumentStore {
&self,
document_id: &TenonDocumentId,
source: &[u8],
- protection: &dyn DocumentProtection,
+ protection: &dyn ArtifactProtection,
) -> Result<TenonDocumentEtag, TenonDocumentStoreError> {
let id_sha256: [u8; SHA256_BYTES] = Sha256::digest(document_id.as_str().as_bytes()).into();
let committed = self.directory.join(format!(
@@ -476,7 +476,7 @@ fn remove_temporary_entry(entry: &DirEntry) -> Result<(), TenonDocumentStoreErro
fn read_source_entry(
entry: DirEntry,
- protection: &dyn DocumentProtection,
+ protection: &dyn ArtifactProtection,
) -> Result<StoredTenonDocumentSource, TenonDocumentStoreError> {
let path = entry.path();
let file_name = entry.file_name();
diff --git a/src/runner/document_store/tests.rs b/src/runner/document_store/tests.rs
index 7a79211..4b5349c 100644
--- a/src/runner/document_store/tests.rs
+++ b/src/runner/document_store/tests.rs
@@ -21,7 +21,7 @@ use super::{
SHA256_BYTES, TenonDocumentEtag, TenonDocumentStore, TenonDocumentStoreError, read_fixed_length,
};
use crate::identifiers::TenonDocumentId;
-use crate::runner::extensions::{DocumentProtection, Plaintext};
+use crate::runner::extensions::{ArtifactProtection, ByPass};
use crate::runner::state_directory::TENON_DOCUMENT_STORE_DIRECTORY_NAME;
use sha2::{Digest as _, Sha256};
use std::error::Error as _;
@@ -33,7 +33,7 @@ use std::{env, fs, process};
struct PartialProtectionFailure;
-impl DocumentProtection for PartialProtectionFailure {
+impl ArtifactProtection for PartialProtectionFailure {
fn protect(&self, _source: &[u8], output: &mut dyn io::Write) -> io::Result<()> {
output.write_all(b"partial protected output")?;
Err(io::Error::other("Test protection failed"))
@@ -53,7 +53,7 @@ fn partial_protection_failure_preserves_the_committed_file() -> io::Result<()> {
let store = directory.store();
let original = b"original source";
store
- .commit(&id, original, &Plaintext)
+ .commit(&id, original, &ByPass)
.map_err(io::Error::other)?;
let error = store
.commit(&id, b"replacement", &PartialProtectionFailure)
@@ -156,7 +156,7 @@ fn missing_store_is_a_read_failure_and_is_not_created() -> io::Result<()> {
let error = state_directory
.store()
- .sources(Arc::new(Plaintext))
+ .sources(Arc::new(ByPass))
.err()
.ok_or_else(|| io::Error::other("missing Store was treated as empty"))?;
@@ -172,9 +172,7 @@ fn source_bytes_are_read_when_iteration_reaches_the_file() -> io::Result<()> {
let path = store_directory.join(committed_file_name("document"));
fs::write(&path, b"old document")?;
let store = state_directory.store();
- let mut sources = store
- .sources(Arc::new(Plaintext))
- .map_err(io::Error::other)?;
+ let mut sources = store.sources(Arc::new(ByPass)).map_err(io::Error::other)?;
fs::write(path, b"current source")?;
let source = sources
@@ -203,9 +201,7 @@ fn first_physical_error_stops_source_iteration() -> io::Result<()> {
let store_directory = state_directory.create_store()?;
fs::write(store_directory.join("operator-note.txt"), b"note")?;
let store = state_directory.store();
- let mut sources = store
- .sources(Arc::new(Plaintext))
- .map_err(io::Error::other)?;
+ let mut sources = store.sources(Arc::new(ByPass)).map_err(io::Error::other)?;
let result = sources
.next()
@@ -225,9 +221,7 @@ fn malformed_committed_file_name_stops_source_iteration() -> io::Result<()> {
let store_directory = state_directory.create_store()?;
fs::write(store_directory.join("not-a-sha256.jsonc"), b"invalid")?;
let store = state_directory.store();
- let mut sources = store
- .sources(Arc::new(Plaintext))
- .map_err(io::Error::other)?;
+ let mut sources = store.sources(Arc::new(ByPass)).map_err(io::Error::other)?;
let result = sources
.next()
@@ -293,7 +287,7 @@ fn commit_does_not_recreate_a_missing_store() -> io::Result<()> {
let error = state_directory
.store()
- .commit(&document_id, b"document source", &Plaintext)
+ .commit(&document_id, b"document source", &ByPass)
.err()
.ok_or_else(|| io::Error::other("commit recreated a missing Store"))?;
@@ -419,7 +413,7 @@ fn committed_directory_is_rejected() -> io::Result<()> {
fs::create_dir(store_directory.join(committed_file_name("directory")))?;
let store = state_directory.store();
let result = store
- .sources(Arc::new(Plaintext))
+ .sources(Arc::new(ByPass))
.map_err(io::Error::other)?
.next()
.ok_or_else(|| io::Error::other("committed directory was ignored"))?;
@@ -449,7 +443,7 @@ fn committed_symbolic_link_is_rejected() -> io::Result<()> {
)?;
let store = state_directory.store();
let result = store
- .sources(Arc::new(Plaintext))
+ .sources(Arc::new(ByPass))
.map_err(io::Error::other)?
.next()
.ok_or_else(|| io::Error::other("committed symbolic link was ignored"))?;
@@ -474,7 +468,7 @@ fn existing_store_path_must_be_a_directory() -> io::Result<()> {
b"not a directory",
)?;
- let Err(error) = state_directory.store().sources(Arc::new(Plaintext)) else {
+ let Err(error) = state_directory.store().sources(Arc::new(ByPass)) else {
return Err(io::Error::other("store file was accepted as a directory"));
};
assert_eq!(error.code(), "tenon_document_store.directory_read_failed");
diff --git a/src/runner/execution.rs b/src/runner/execution.rs
index b499cb4..c031dd5 100644
--- a/src/runner/execution.rs
+++ b/src/runner/execution.rs
@@ -17,35 +17,86 @@
* under the License.
*/
-//! Separates policy decisions from the Runner's current entitlement deadline.
+//! Contains the policy, change notifications, and the Runner's entitlement deadline.
-use crate::runner::extensions::{ExecutionDenied, ExecutionPolicy, ExecutionScope};
-use std::future;
+use crate::runner::extensions::{ExecutionDenied, ExecutionPolicy, ExecutionScope, PolicyChanges};
+use std::future::{self, Future};
use std::time::Instant;
use tokio::sync::watch;
use tokio::time;
pub(super) struct RunnerExecution {
policy: Box<dyn ExecutionPolicy>,
- pub(super) expiry: ExecutionExpiry,
+ changes: Option<PolicyChanges>,
+ expiry: ExecutionExpiry,
}
impl RunnerExecution {
pub(super) fn new(policy: Box<dyn ExecutionPolicy>) -> Self {
Self {
+ changes: policy.changes(),
policy,
expiry: ExecutionExpiry::default(),
}
}
pub(super) fn check(&self, scope: ExecutionScope<'_>) -> Result<(), ExecutionDenied> {
- let decision = self.policy.authorize(scope);
- self.expiry.update(match &decision {
+ Self::check_policy(self.policy.as_ref(), &self.expiry, scope)
+ }
+
+ /// Gives Pipeline supervision a handle to the same deadline.
+ pub(super) fn expiry(&self) -> ExecutionExpiry {
+ self.expiry.clone()
+ }
+
+ pub(super) async fn wait_for_expiry(&self) {
+ self.expiry.wait_for_expiry().await;
+ }
+
+ /// The event loop can wait for changes and do policy checks at the same time.
+ pub(super) fn split(
+ &mut self,
+ ) -> (
+ impl Fn(ExecutionScope<'_>) -> Result<(), ExecutionDenied> + '_,
+ impl Future<Output = bool> + '_,
+ &ExecutionExpiry,
+ ) {
+ let policy = self.policy.as_ref();
+ let expiry = &self.expiry;
+ (
+ move |scope| Self::check_policy(policy, expiry, scope),
+ Self::policy_changed(&mut self.changes),
+ expiry,
+ )
+ }
+
+ fn check_policy(
+ policy: &dyn ExecutionPolicy,
+ expiry: &ExecutionExpiry,
+ scope: ExecutionScope<'_>,
+ ) -> Result<(), ExecutionDenied> {
+ let decision = policy.authorize(scope);
+ expiry.update(match &decision {
Ok(permit) => permit.entitlement_until,
Err(denied) => denied.entitlement_until,
});
decision.map(|_| ())
}
+
+ async fn policy_changed(changes: &mut Option<PolicyChanges>) -> bool {
+ let Some(receiver) = changes else {
+ return future::pending().await;
+ };
+ if receiver.changed().await.is_ok() {
+ true
+ } else {
+ *changes = None;
+ eprintln!(
+ "runner.policy_changes_closed: Policy notifications stopped; the current deadline does not change"
+ );
+ false
+ }
+ }
}
#[derive(Clone)]
@@ -105,6 +156,103 @@ mod tests {
use std::time::{Duration, Instant};
use tokio::task;
+ struct NotifiedPolicy {
+ state: Rc<RefCell<Result<ExecutionPermit, ExecutionDenied>>>,
+ receiver: PolicyChanges,
+ subscriptions: Rc<Cell<usize>>,
+ }
+
+ impl ExecutionPolicy for NotifiedPolicy {
+ fn authorize(
+ &self,
+ _scope: ExecutionScope<'_>,
+ ) -> Result<ExecutionPermit, ExecutionDenied> {
+ self.state.borrow().clone()
+ }
+
+ fn changes(&self) -> Option<PolicyChanges> {
+ self.subscriptions.set(self.subscriptions.get() + 1);
+ Some(self.receiver.clone())
+ }
+ }
+
+ #[tokio::test(start_paused = true)]
+ async fn notifications_survive_cancellation_and_use_the_latest_state()
+ -> Result<(), ExecutionDenied> {
+ let (updates, receiver) = watch::channel(());
+ let state = Rc::new(RefCell::new(Ok(ExecutionPermit {
+ entitlement_until: Some((time::Instant::now() + Duration::from_secs(1)).into_std()),
+ })));
+ let subscriptions = Rc::new(Cell::new(0));
+ let mut execution = RunnerExecution::new(Box::new(NotifiedPolicy {
+ state: Rc::clone(&state),
+ receiver,
+ subscriptions: Rc::clone(&subscriptions),
+ }));
+ execution.check(ExecutionScope { documents: &[] })?;
+ {
+ let (check, policy_changed, _) = execution.split();
+ tokio::select! {
+ biased;
+ _ = policy_changed => panic!("No update was sent"),
+ result = async {
+ task::yield_now().await;
+ check(ExecutionScope { documents: &[] })
+ } => result?,
+ }
+ }
+ *state.borrow_mut() = Err(ExecutionDenied::new(
+ "test.denied",
+ "Execution is not permitted",
+ None,
+ ));
+ updates.send_replace(());
+ *state.borrow_mut() = Ok(ExecutionPermit::default());
+ updates.send_replace(());
+ {
+ let (check, policy_changed, _) = execution.split();
+ assert!(policy_changed.await);
+ check(ExecutionScope { documents: &[] })?;
+ }
+ time::advance(Duration::from_secs(2)).await;
+ assert!(!execution.expiry.is_expired());
+ assert_eq!(subscriptions.get(), 1);
+ let (_, policy_changed, _) = execution.split();
+ tokio::select! {
+ biased;
+ _ = policy_changed => panic!("A merged update was read twice"),
+ () = task::yield_now() => {},
+ }
+ Ok(())
+ }
+
+ #[tokio::test(start_paused = true)]
+ async fn closed_notifications_preserve_expiry_and_disable_the_wait()
+ -> Result<(), ExecutionDenied> {
+ let (updates, receiver) = watch::channel(());
+ let state = Rc::new(RefCell::new(Ok(ExecutionPermit {
+ entitlement_until: Some((time::Instant::now() + Duration::from_secs(1)).into_std()),
+ })));
+ let mut execution = RunnerExecution::new(Box::new(NotifiedPolicy {
+ state,
+ receiver,
+ subscriptions: Rc::new(Cell::new(0)),
+ }));
+ execution.check(ExecutionScope { documents: &[] })?;
+ drop(updates);
+ let (_, policy_changed, _) = execution.split();
+ assert!(!policy_changed.await);
+ let (_, policy_changed, _) = execution.split();
+ tokio::select! {
+ biased;
+ _ = policy_changed => panic!("Closed channel caused a ready loop"),
+ () = task::yield_now() => {},
+ }
+ time::advance(Duration::from_secs(1)).await;
+ assert!(execution.expiry.is_expired());
+ Ok(())
+ }
+
struct Policy(Rc<RefCell<Result<ExecutionPermit, ExecutionDenied>>>);
impl ExecutionPolicy for Policy {
diff --git a/src/runner/extensions.rs b/src/runner/extensions/mod.rs
similarity index 69%
rename from src/runner/extensions.rs
rename to src/runner/extensions/mod.rs
index 9a1cdd6..642ca56 100644
--- a/src/runner/extensions.rs
+++ b/src/runner/extensions/mod.rs
@@ -17,27 +17,31 @@
* under the License.
*/
-//! Trusted distribution hooks for HTTP admission, execution, and Document storage.
+//! Trusted distribution hooks for HTTP admission, execution, and artifact protection.
//!
//! The Runner initializes these implementations once after configuration parsing.
//! Only its serialized control loop calls the execution policy. Blocking Store
//! jobs share the protection implementation, while the core retains all file,
//! commit, process, cancellation, and shutdown ownership.
+mod tar_gz;
+
use crate::tenon_document::VerifiedTenonDocument;
use axum::http::{HeaderMap, HeaderValue, Method};
use std::error::Error;
use std::fmt;
use std::future::Future;
-use std::io::{self, Write};
+use std::io::{self, Read, Write};
use std::pin::Pin;
use std::sync::Arc;
use std::time::Instant;
+use tar_gz::open_tar_gz;
+use tokio::sync::watch;
/// The three implementations installed by one Runner initialization callback.
pub struct RunnerHooks {
pub(crate) execution_policy: Box<dyn ExecutionPolicy>,
- pub(crate) document_protection: Arc<dyn DocumentProtection>,
+ pub(crate) artifact_protection: Arc<dyn ArtifactProtection>,
pub(crate) http_authorization: Arc<dyn HttpApiAuthorization>,
}
@@ -46,12 +50,12 @@ impl RunnerHooks {
#[must_use]
pub fn new(
execution_policy: impl ExecutionPolicy + 'static,
- document_protection: impl DocumentProtection + 'static,
+ artifact_protection: impl ArtifactProtection + 'static,
http_authorization: impl HttpApiAuthorization + 'static,
) -> Self {
Self {
execution_policy: Box::new(execution_policy),
- document_protection: Arc::new(document_protection),
+ artifact_protection: Arc::new(artifact_protection),
http_authorization: Arc::new(http_authorization),
}
}
@@ -59,7 +63,7 @@ impl RunnerHooks {
impl Default for RunnerHooks {
fn default() -> Self {
- Self::new(AllowAll, Plaintext, NoHttpAuth)
+ Self::new(AllowAll, ByPass, NoHttpAuth)
}
}
@@ -85,9 +89,24 @@ pub trait ExecutionPolicy {
///
/// A denied candidate leaves the accepted desired state unchanged. Startup
/// reports and skips it; an HTTP change returns the rejection to its caller.
+ /// Initial empty-set denial prevents recovery. Current-set denial after a
+ /// notification starts shutdown and keeps saved Documents.
fn authorize(&self, scope: ExecutionScope<'_>) -> Result<ExecutionPermit, ExecutionDenied>;
+
+ /// Supplies optional notifications that the policy state has changed.
+ ///
+ /// The Runner calls this once, before the first authorization. Publish the
+ /// full policy state before you send a notification. The Runner can combine
+ /// notifications and evaluates the current state, including an empty Document set.
+ fn changes(&self) -> Option<PolicyChanges> {
+ None
+ }
}
+/// A notification channel for changes to the current policy state.
+/// Values carry no decision, generation, or acknowledgement.
+pub type PolicyChanges = watch::Receiver<()>;
+
/// Immutable facts borrowed only for one synchronous policy decision.
#[derive(Debug)]
pub struct ExecutionScope<'a> {
@@ -150,25 +169,81 @@ impl fmt::Display for ExecutionDenied {
impl Error for ExecutionDenied {}
-/// Converts complete Document bytes without owning filesystem operations.
+/// Converts Document bytes and package envelopes without owning filesystem operations.
///
/// Methods run in blocking Store work. The same implementation is shared by
/// writes and startup recovery. Formats and key material remain implementation
/// details; the core never guesses formats or falls back to plaintext.
-pub trait DocumentProtection: Send + Sync {
+/// Default Document methods write unchanged bytes. The package method decodes tar.gz.
+pub trait ArtifactProtection: Send + Sync {
/// Writes the protected representation into the core-owned temporary file.
///
/// # Errors
///
/// An error leaves the formal Document unchanged, even after partial output.
- fn protect(&self, source: &[u8], output: &mut dyn Write) -> io::Result<()>;
+ fn protect(&self, source: &[u8], output: &mut dyn Write) -> io::Result<()> {
+ output.write_all(source)
+ }
/// Writes the recovered original bytes into the core-owned output buffer.
///
/// # Errors
///
/// An error aborts startup; partial output is discarded without parsing.
- fn unprotect(&self, stored: &[u8], output: &mut dyn Write) -> io::Result<()>;
+ fn unprotect(&self, stored: &[u8], output: &mut dyn Write) -> io::Result<()> {
+ output.write_all(stored)
+ }
+
+ /// Decodes a package into files through the core-owned directory writer.
+ ///
+ /// This method runs during installation and recovery. Complete all format,
+ /// integrity, and access checks before returning success. The core validates
+ /// the directory contract and keeps the received input without changes.
+ /// The default decodes a standard tar.gz package with fixed format limits.
+ /// Custom hooks can call `ByPass.open_plugin_package` to reuse this decoder.
+ ///
+ /// # Errors
+ ///
+ /// Return an error if the package cannot be opened. Do not put secrets in
+ /// errors. The core discards partial output and does not try another format.
+ /// Use `InvalidData` or `UnexpectedEof` for invalid package data, and
+ /// `FileTooLarge` for a size limit. Other error kinds reject package access.
+ /// Failures from its input reader or directory writer remain core failures.
+ fn open_plugin_package(
+ &self,
+ source: &mut dyn Read,
+ output: &mut dyn PluginPackageOutput,
+ ) -> io::Result<()> {
+ open_tar_gz(source, output)
+ }
+}
+
+/// Receives decoded package files and directories.
+///
+/// The Runner supplies its implementation to [`ArtifactProtection::open_plugin_package`].
+/// Paths are relative to the package root. Files and directories use fixed
+/// owner-only permissions. The writer rejects duplicate paths, path conflicts,
+/// and output that exceeds the fixed limits. A failed call invalidates the
+/// whole output, even if the hook ignores its error.
+pub trait PluginPackageOutput {
+ /// Writes one ordinary file from its content stream to EOF.
+ ///
+ /// The writer creates missing parent directories. The file is not available
+ /// for execution until the hook and core validation have both succeeded.
+ ///
+ /// # Errors
+ ///
+ /// Returns an error for an invalid path, a duplicate, a size limit,
+ /// a failed content read, or a local file operation failure.
+ fn write_file(&mut self, path: &str, source: &mut dyn Read) -> io::Result<()>;
+
+ /// Creates one explicit directory and any missing parent directories.
+ ///
+ /// # Errors
+ ///
+ /// Returns an error for an invalid path, a duplicate explicit path,
+ /// a file conflict, an entry limit, or a local file operation failure.
+ fn create_dir(&mut self, path: &str) -> io::Result<()>;
}
/// Decides whether one HTTP request may enter its handler without reading its body.
@@ -236,16 +311,8 @@ impl ExecutionPolicy for AllowAll {
}
}
-/// The open-source protection implementation, preserving exact original bytes.
+/// Uses all default artifact methods without encryption or decryption.
#[derive(Debug, Clone, Copy, Default)]
-pub struct Plaintext;
-
-impl DocumentProtection for Plaintext {
- fn protect(&self, source: &[u8], output: &mut dyn Write) -> io::Result<()> {
- output.write_all(source)
- }
+pub struct ByPass;
- fn unprotect(&self, stored: &[u8], output: &mut dyn Write) -> io::Result<()> {
- output.write_all(stored)
- }
-}
+impl ArtifactProtection for ByPass {}
diff --git a/src/runner/extensions/tar_gz.rs b/src/runner/extensions/tar_gz.rs
new file mode 100644
index 0000000..5941340
--- /dev/null
+++ b/src/runner/extensions/tar_gz.rs
@@ -0,0 +1,300 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+//! Decodes tar.gz packages for the default ArtifactProtection method.
+
+use super::PluginPackageOutput;
+use flate2::bufread::GzDecoder;
+use std::io::{self, BufReader, Read};
+use tar::{Archive, PaxExtensions};
+
+const MAX_ARCHIVE_ENTRIES: usize = 100_000;
+const MAX_ARCHIVE_BYTES: u64 = 8 * 1024 * 1024 * 1024;
+const MAX_ARCHIVE_METADATA_BYTES: u64 = 64 * 1024;
+const TAR_END_MARKER_BYTES: u64 = 512;
+const TAR_READ_BUFFER_BYTES: usize = 64 * 1024;
+
+pub(super) fn open_tar_gz(
+ package: &mut dyn Read,
+ output: &mut dyn PluginPackageOutput,
+) -> io::Result<()> {
+ let input = BufReader::new(package);
+ let gzip = GzDecoder::new(input);
+ let bounded = BoundedReader::new(gzip, MAX_ARCHIVE_BYTES);
+ let mut archive = Archive::new(bounded);
+ let mut pending_metadata = PendingArchiveMetadata::default();
+ let mut entry_count = 0_usize;
+ {
+ let entries = archive.entries().map_err(archive_error)?.raw(true);
+ for entry in entries {
+ entry_count += 1;
+ if entry_count > MAX_ARCHIVE_ENTRIES {
+ return Err(io::Error::from(io::ErrorKind::FileTooLarge));
+ }
+ let mut entry = entry.map_err(archive_error)?;
+ let entry_type = entry.header().entry_type();
+ if entry_type.is_gnu_longname() {
+ if pending_metadata.gnu_path.is_some() {
+ return Err(io::Error::from(io::ErrorKind::InvalidData));
+ }
+ let mut path = read_archive_metadata(&mut entry)?;
+ if path.last() == Some(&0) {
+ path.pop();
+ }
+ if path.is_empty() {
+ return Err(io::Error::from(io::ErrorKind::InvalidData));
+ }
+ pending_metadata.gnu_path = Some(path);
+ continue;
+ }
+ if entry_type.is_pax_local_extensions() {
+ if pending_metadata.has_pax_local() {
+ return Err(io::Error::from(io::ErrorKind::InvalidData));
+ }
+ let metadata = read_archive_metadata(&mut entry)?;
+ pending_metadata.set_pax_local(&metadata)?;
+ continue;
+ }
+ if entry_type.is_pax_global_extensions() {
+ if pending_metadata.has_local() {
+ return Err(io::Error::from(io::ErrorKind::InvalidData));
+ }
+ let metadata = read_archive_metadata(&mut entry)?;
+ validate_pax_global(&metadata)?;
+ continue;
+ }
+ if !entry_type.is_file() && !entry_type.is_dir() {
+ return Err(io::Error::from(io::ErrorKind::InvalidData));
+ }
+ let size = entry.header().size().map_err(archive_error)?;
+ let path_bytes = pending_metadata.take_path(entry.header().path_bytes().as_ref())?;
+ if pending_metadata
+ .take_size()
+ .is_some_and(|value| value != size)
+ {
+ return Err(io::Error::from(io::ErrorKind::InvalidData));
+ }
+ let path = std::str::from_utf8(&path_bytes)
+ .map_err(|_| io::Error::from(io::ErrorKind::InvalidData))?;
+ if entry_type.is_dir() {
+ if size != 0 {
+ return Err(io::Error::from(io::ErrorKind::InvalidData));
+ }
+ output.create_dir(path)?;
+ } else {
+ let mut content = entry.take(size);
+ output.write_file(path, &mut content)?;
+ if content.limit() != 0 {
+ return Err(io::ErrorKind::UnexpectedEof.into());
+ }
+ }
+ }
+ }
+ if pending_metadata.has_local() {
+ return Err(io::Error::from(io::ErrorKind::InvalidData));
+ }
+ let mut bounded = archive.into_inner();
+ validate_tar_end(&mut bounded)?;
+ let gzip = bounded.into_inner();
+ let mut compressed_input = gzip.into_inner();
+ let mut trailing = [0_u8; 1];
+ if compressed_input
+ .read(&mut trailing)
+ .map_err(archive_error)?
+ != 0
+ {
+ return Err(io::Error::from(io::ErrorKind::InvalidData));
+ }
+ Ok(())
+}
+
+#[derive(Debug, Default)]
+struct PendingArchiveMetadata {
+ gnu_path: Option<Vec<u8>>,
+ pax_path: Option<Vec<u8>>,
+ pax_size: Option<u64>,
+ pax_local_seen: bool,
+}
+
+impl PendingArchiveMetadata {
+ fn has_pax_local(&self) -> bool {
+ self.pax_local_seen
+ }
+
+ fn has_local(&self) -> bool {
+ self.gnu_path.is_some() || self.has_pax_local()
+ }
+
+ fn set_pax_local(&mut self, metadata: &[u8]) -> io::Result<()> {
+ self.pax_local_seen = true;
+ for extension in PaxExtensions::new(metadata) {
+ let extension = extension.map_err(archive_error)?;
+ match extension.key_bytes() {
+ b"path" => {
+ if self.pax_path.is_some() || extension.value_bytes().is_empty() {
+ return Err(io::Error::from(io::ErrorKind::InvalidData));
+ }
+ self.pax_path = Some(extension.value_bytes().to_vec());
+ }
+ b"size" => {
+ if self.pax_size.is_some() {
+ return Err(io::Error::from(io::ErrorKind::InvalidData));
+ }
+ let value = std::str::from_utf8(extension.value_bytes())
+ .ok()
+ .and_then(|value| value.parse::<u64>().ok())
+ .ok_or(io::Error::from(io::ErrorKind::InvalidData))?;
+ self.pax_size = Some(value);
+ }
+ b"linkpath" => return Err(io::Error::from(io::ErrorKind::InvalidData)),
+ key if key.starts_with(b"GNU.sparse.") => {
+ return Err(io::Error::from(io::ErrorKind::InvalidData));
+ }
+ _ => {}
+ }
+ }
+ Ok(())
+ }
+
+ fn take_path(&mut self, header_path: &[u8]) -> io::Result<Vec<u8>> {
+ match (self.gnu_path.take(), self.pax_path.take()) {
+ (Some(_), Some(_)) => Err(io::Error::from(io::ErrorKind::InvalidData)),
+ (Some(path), None) | (None, Some(path)) => Ok(path),
+ (None, None) => Ok(header_path.to_vec()),
+ }
+ }
+
+ fn take_size(&mut self) -> Option<u64> {
+ self.pax_local_seen = false;
+ self.pax_size.take()
+ }
+}
+
+fn validate_pax_global(metadata: &[u8]) -> io::Result<()> {
+ for extension in PaxExtensions::new(metadata) {
+ let extension = extension.map_err(archive_error)?;
+ let key = extension.key_bytes();
+ if matches!(key, b"path" | b"size" | b"linkpath") || key.starts_with(b"GNU.sparse.") {
+ return Err(io::Error::from(io::ErrorKind::InvalidData));
+ }
+ }
+ Ok(())
+}
+
+fn read_archive_metadata(input: &mut impl Read) -> io::Result<Vec<u8>> {
+ let mut bytes = Vec::new();
+ input
+ .take(MAX_ARCHIVE_METADATA_BYTES + 1)
+ .read_to_end(&mut bytes)
+ .map_err(archive_error)?;
+ if bytes.len() as u64 > MAX_ARCHIVE_METADATA_BYTES {
+ return Err(io::Error::from(io::ErrorKind::FileTooLarge));
+ }
+ Ok(bytes)
+}
+
+fn validate_tar_end(input: &mut impl Read) -> io::Result<()> {
+ let mut trailing_bytes = 0_u64;
+ let mut buffer = [0_u8; TAR_READ_BUFFER_BYTES];
+ loop {
+ let count = input.read(&mut buffer).map_err(archive_error)?;
+ if count == 0 {
+ break;
+ }
+ if buffer[..count].iter().any(|byte| *byte != 0) {
+ return Err(io::Error::from(io::ErrorKind::InvalidData));
+ }
+ trailing_bytes = trailing_bytes
+ .checked_add(count as u64)
+ .ok_or(io::Error::from(io::ErrorKind::FileTooLarge))?;
+ }
+ if trailing_bytes < TAR_END_MARKER_BYTES {
+ return Err(io::Error::from(io::ErrorKind::InvalidData));
+ }
+ Ok(())
+}
+
+#[derive(Debug)]
+struct BoundedReader<R> {
+ inner: R,
+ remaining: u64,
+}
+
+impl<R> BoundedReader<R> {
+ const fn new(inner: R, limit: u64) -> Self {
+ Self {
+ inner,
+ remaining: limit,
+ }
+ }
+
+ fn into_inner(self) -> R {
+ self.inner
+ }
+}
+
+impl<R: Read> Read for BoundedReader<R> {
+ fn read(&mut self, buffer: &mut [u8]) -> io::Result<usize> {
+ if buffer.is_empty() {
+ return Ok(0);
+ }
+ if self.remaining == 0 {
+ let mut probe = [0_u8; 1];
+ return match self.inner.read(&mut probe)? {
+ 0 => Ok(0),
+ _ => Err(io::ErrorKind::FileTooLarge.into()),
+ };
+ }
+ let available = usize::try_from(self.remaining)
+ .unwrap_or(usize::MAX)
+ .min(buffer.len());
+ let count = self.inner.read(&mut buffer[..available])?;
+ self.remaining -= count as u64;
+ Ok(count)
+ }
+}
+
+fn archive_error(source: io::Error) -> io::Error {
+ if source.kind() == io::ErrorKind::FileTooLarge {
+ source
+ } else {
+ io::Error::new(io::ErrorKind::InvalidData, source)
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn stream_limit_accepts_the_boundary_and_rejects_one_more_byte() -> io::Result<()> {
+ for length in [3, 4] {
+ let mut input = BoundedReader::new(io::repeat(0).take(length), 3);
+ let result = io::copy(&mut input, &mut io::sink());
+ if length == 3 {
+ assert_eq!(result?, 3);
+ } else {
+ assert!(
+ matches!(result, Err(error) if error.kind() == io::ErrorKind::FileTooLarge)
+ );
+ }
+ }
+ Ok(())
+ }
+}
diff --git a/src/runner/http/mod.rs b/src/runner/http/mod.rs
index 29104f4..7d1cbeb 100644
--- a/src/runner/http/mod.rs
+++ b/src/runner/http/mod.rs
@@ -290,7 +290,7 @@ mod server_tests {
connection
.write_all(
format!(
- "POST /plugins HTTP/1.1\r\nHost: {address}\r\nConnection: close\r\nContent-Type: application/vnd.apache.tenon.plugin+tar+gzip\r\nContent-Length: 1\r\nExpect: 100-continue\r\n\r\n"
+ "POST /plugins HTTP/1.1\r\nHost: {address}\r\nConnection: close\r\nContent-Type: application/octet-stream\r\nContent-Length: 1\r\nExpect: 100-continue\r\n\r\n"
)
.as_bytes(),
)
diff --git a/src/runner/http/plugins/mod.rs b/src/runner/http/plugins/mod.rs
index 6ed4084..9b25cbf 100644
--- a/src/runner/http/plugins/mod.rs
+++ b/src/runner/http/plugins/mod.rs
@@ -41,8 +41,6 @@ use serde::{Deserialize, Serialize};
use tokio_stream::StreamExt as _;
use utoipa::ToSchema;
-const PLUGIN_PACKAGE: &str = "application/vnd.apache.tenon.plugin+tar+gzip";
-
/// The Program collection response: one summary per available Program.
#[derive(Serialize, ToSchema)]
struct ProgramListBody<'a> {
@@ -79,15 +77,17 @@ struct PluginInUseBody {
post,
path = "/plugins",
tag = "plugins",
- request_body(content = Vec<u8>, description = "A Plugin package (gzip-compressed tar)", content_type = "application/vnd.apache.tenon.plugin+tar+gzip"),
+ request_body(content(
+ (Vec<u8> = "application/octet-stream"),
+ ), description = "Plugin package bytes for the configured decoder"),
responses(
(status = 201, description = "Installed; the Location header holds the Program path"),
(status = 204, description = "Unchanged; the Location header holds the Program path"),
(status = 400, description = "The request body could not be read", body = ErrorEnvelope),
(status = 409, description = "A different package already holds this version", body = ErrorEnvelope),
(status = 413, description = "The package exceeds the size limit", body = ErrorEnvelope),
- (status = 415, description = "The Content-Type is not the Plugin package media type", body = ErrorEnvelope),
- (status = 422, description = "The package or its contracts are invalid, or the platform does not match", body = ErrorEnvelope),
+ (status = 415, description = "The Content-Type is not application/octet-stream", body = ErrorEnvelope),
+ (status = 422, description = "Package access was rejected, package validation failed, or the platform does not match", body = ErrorEnvelope),
(status = 503, description = "Runner is shutting down", body = ErrorEnvelope),
)
)]
@@ -97,7 +97,7 @@ pub(super) async fn install_program(
headers: HeaderMap,
body: Body,
) -> Response {
- if !has_content_type(&headers, PLUGIN_PACKAGE) {
+ if !has_content_type(&headers, "application/octet-stream") {
return unsupported_media_type();
}
let Ok(upload) = state.management.begin_program_install().await else {
diff --git a/src/runner/http/plugins/tests.rs b/src/runner/http/plugins/tests.rs
index 5db951c..89222ea 100644
--- a/src/runner/http/plugins/tests.rs
+++ b/src/runner/http/plugins/tests.rs
@@ -37,7 +37,7 @@ use axum::routing::get;
use bytes::Bytes;
use serde_json::{Value, json};
use std::fs;
-use std::io::{self, Write as _};
+use std::io::{self, Read as _, Write as _};
use tonic::codegen::Service as _;
#[tokio::test(flavor = "current_thread")]
@@ -52,7 +52,7 @@ async fn upload_publishes_all_interfaces_with_exact_locations_and_idempotent_ret
let package = valid_program_package(interface)?;
for status in [StatusCode::CREATED, StatusCode::NO_CONTENT] {
let request = Request::post("/plugins")
- .header("Content-Type", PLUGIN_PACKAGE)
+ .header("Content-Type", "application/octet-stream")
.body(Body::from(package.clone()))
.map_err(io::Error::other)?;
let response = exchange(&mut supervisor, &mut app, request).await?;
@@ -88,8 +88,12 @@ async fn upload_publishes_all_interfaces_with_exact_locations_and_idempotent_ret
}
supervisor.shutdown().await.map_err(io::Error::other)?;
drop(supervisor);
- let recovered = PluginProgramStore::recover(directory.path().join("plugins/programs"))
- .map_err(io::Error::other)?;
+ let recovered = PluginProgramStore::recover(
+ directory.path().join("plugins/programs"),
+ directory.path().join("plugins/programs"),
+ std::sync::Arc::new(crate::runner::extensions::ByPass),
+ )
+ .map_err(io::Error::other)?;
assert_eq!(recovered.programs().count(), 3);
for (_, _, entry) in recovered.programs() {
assert_eq!(entry.display_name(), "Example Plugin");
@@ -128,8 +132,8 @@ async fn normalized_retry_and_conflict_keep_the_committed_package() -> io::Resul
}
let root = directory.path().join("plugins/programs");
assert_eq!(
- fs::read(root.join("com.example.source/1.0.0/resources/data.bin"))?,
- b"data"
+ fs::read(root.join("com.example.source/.tenon-artifact-92521fc3cbd964bdc9f584a991b89fddaa5754ed1cc96d6d42445338669c1305"))?,
+ valid_program_package(PluginInterface::Source)?
);
assert_eq!(fs::read_dir(root)?.count(), 1);
supervisor.shutdown().await.map_err(io::Error::other)?;
@@ -165,6 +169,11 @@ async fn invalid_uploads_and_expansion_limits_do_not_poison_the_store() -> io::R
StatusCode::PAYLOAD_TOO_LARGE,
"plugin_package_too_large",
),
+ (
+ truncated_manifest_archive()?,
+ StatusCode::UNPROCESSABLE_ENTITY,
+ "plugin_package_invalid",
+ ),
] {
let response = exchange(
&mut supervisor,
@@ -191,14 +200,19 @@ async fn invalid_uploads_and_expansion_limits_do_not_poison_the_store() -> io::R
async fn invalid_content_type_and_failed_body_publish_nothing() -> io::Result<()> {
let package = valid_program_package(PluginInterface::Source)?;
let (directory, mut supervisor, mut app) = program_http()?;
- let request = Request::post("/plugins")
- .header("Content-Type", "application/json")
- .body(Body::empty())
- .map_err(io::Error::other)?;
- assert_eq!(
- exchange(&mut supervisor, &mut app, request).await?.status(),
- StatusCode::UNSUPPORTED_MEDIA_TYPE
- );
+ for media_type in [
+ "application/json",
+ "application/vnd.apache.tenon.plugin+tar+gzip",
+ ] {
+ let request = Request::post("/plugins")
+ .header("Content-Type", media_type)
+ .body(Body::empty())
+ .map_err(io::Error::other)?;
+ assert_eq!(
+ exchange(&mut supervisor, &mut app, request).await?.status(),
+ StatusCode::UNSUPPORTED_MEDIA_TYPE
+ );
+ }
let complete_archive = Bytes::from(package.clone());
let body = Body::from_stream(tokio_stream::iter([
Ok(complete_archive),
@@ -231,6 +245,19 @@ async fn invalid_content_type_and_failed_body_publish_nothing() -> io::Result<()
}
fn oversized_manifest_archive() -> io::Result<Vec<u8>> {
+ let size = 16 * 1024 * 1024 + 1;
+ let mut header = tar::Header::new_gnu();
+ header.set_path("manifest.json")?;
+ header.set_mode(0o500);
+ header.set_size(size);
+ header.set_cksum();
+ let gzip = flate2::write::GzEncoder::new(Vec::new(), flate2::Compression::fast());
+ let mut archive = tar::Builder::new(gzip);
+ archive.append(&header, io::repeat(0).take(size))?;
+ archive.into_inner()?.finish()
+}
+
+fn truncated_manifest_archive() -> io::Result<Vec<u8>> {
let mut header = tar::Header::new_gnu();
header.set_path("manifest.json")?;
header.set_mode(0o500);
@@ -243,7 +270,7 @@ fn oversized_manifest_archive() -> io::Result<Vec<u8>> {
fn upload_request(body: Body) -> io::Result<Request<Body>> {
Request::post("/plugins")
- .header("Content-Type", PLUGIN_PACKAGE)
+ .header("Content-Type", "application/octet-stream")
.body(body)
.map_err(io::Error::other)
}
@@ -281,14 +308,15 @@ fn program_http() -> io::Result<(tempfile::TempDir, RunnerManagementSupervisor,
let recovered = recover(
&config,
&layout,
- &RunnerHooks::default().document_protection,
+ &RunnerHooks::default().artifact_protection,
+ &layout.pipeline_runtime_directory(),
)
.map_err(io::Error::other)?;
let recovered = RunnerManagementSupervisor::recover(
directory.path(),
config.script_vm_limits(),
recovered,
- RunnerHooks::default().document_protection,
+ RunnerHooks::default().artifact_protection,
None,
|_| Ok(()),
)?;
diff --git a/src/runner/main_loop/failure.rs b/src/runner/main_loop/failure.rs
index 57057e3..861f764 100644
--- a/src/runner/main_loop/failure.rs
+++ b/src/runner/main_loop/failure.rs
@@ -22,9 +22,11 @@
use crate::error::ErrorChain;
use crate::identifiers::TenonDocumentId;
use crate::runner::control_server::RunnerControlServerError;
+use crate::runner::extensions::ExecutionDenied;
use crate::runner::http::RunnerHttpServerError;
use crate::runner::management::RunnerManagementSupervisorError;
use crate::runner::pipeline_supervisor::PipelineSupervisorError;
+use crate::runner::recovery::RunnerRecoveryError;
use crate::runner::runtime_resources::RunnerRuntimeResourcesError;
use crate::tenon_document::TenonDocumentVerifierInitializationError;
use std::error::Error;
@@ -149,6 +151,8 @@ pub(super) enum RunnerMainLoopError {
CpuObservation(io::Error),
ProcessResources(io::Error),
ExecutionExpired,
+ Recovery(RunnerRecoveryError),
+ ExecutionDenied(ExecutionDenied),
RuntimeResources(RunnerRuntimeResourcesError),
ShutdownSignal(io::Error),
ControlServer(RunnerControlServerError),
@@ -168,11 +172,16 @@ impl RunnerMainLoopError {
match self {
Self::CpuObservation(_) => "runner.cpu_observation_failed",
Self::ExecutionExpired => "runner.execution_permit_expired",
+ Self::Recovery(source) => source.code(),
+ Self::ExecutionDenied(source) => source.code(),
Self::ProcessResources(_) => "runner.process_resource_recovery_failed",
Self::RuntimeResources(source) => source.code(),
Self::ShutdownSignal(_) => "runner.shutdown_signal_failed",
Self::ControlServer(_) => "runner.pipeline_control_failed",
Self::HttpServer(_) => "runner.http_server_failed",
+ Self::Management(RunnerManagementSupervisorError::ExecutionDenied(source)) => {
+ source.code()
+ }
Self::ManagementInitialization(_) | Self::Management(_) => "runner.management_failed",
Self::Pipeline(_) => "runner.pipeline_lifecycle_failed",
Self::AdditionalFailures { primary, .. } => primary.code(),
@@ -192,6 +201,8 @@ impl fmt::Display for RunnerMainLoopError {
Self::ExecutionExpired => {
formatter.write_str("Runner execution entitlement has expired")
}
+ Self::Recovery(_) => formatter.write_str("Runner startup recovery failed"),
+ Self::ExecutionDenied(_) => formatter.write_str("Runner execution is not permitted"),
Self::RuntimeResources(source) => write!(formatter, "{source}"),
Self::ShutdownSignal(_) => formatter.write_str("Runner shutdown signal stream failed"),
Self::ControlServer(_) => {
@@ -223,6 +234,8 @@ impl Error for RunnerMainLoopError {
Self::CpuObservation(source) => Some(source),
Self::ProcessResources(source) => Some(source),
Self::ExecutionExpired => None,
+ Self::Recovery(source) => Some(source),
+ Self::ExecutionDenied(source) => Some(source),
Self::RuntimeResources(source) => Some(source),
Self::ShutdownSignal(source) => Some(source),
Self::ControlServer(source) => Some(source),
diff --git a/src/runner/main_loop/mod.rs b/src/runner/main_loop/mod.rs
index 31591db..3a5d305 100644
--- a/src/runner/main_loop/mod.rs
+++ b/src/runner/main_loop/mod.rs
@@ -70,21 +70,27 @@ impl RunnerMainLoop {
) -> Result<RunnerShutdownReport, RunnerMainLoopFailure> {
tokio::pin!(shutdown);
let primary_failure = loop {
+ let (check, policy_changed, expiry) = self.execution.split();
tokio::select! {
biased;
result = &mut shutdown => {
break result.err().map(RunnerMainLoopError::ShutdownSignal);
}
- () = self.execution.expiry.wait_for_expiry() => {
+ () = expiry.wait_for_expiry() => {
break Some(RunnerMainLoopError::ExecutionExpired);
}
+ changed = policy_changed => {
+ if changed && let Err(source) = self.management.check_execution(&check) {
+ break Some(RunnerMainLoopError::ExecutionDenied(source));
+ }
+ }
source = self.control_server.wait() => {
break Some(RunnerMainLoopError::ControlServer(source));
}
source = self.http_server.wait() => {
break Some(RunnerMainLoopError::HttpServer(source));
}
- event = self.management.next_event(|scope| self.execution.check(scope)) => {
+ event = self.management.next_event(&check) => {
match event {
RunnerManagementEvent::CommitReady(commit) => {
if let Err(source) =
diff --git a/src/runner/main_loop/shutdown.rs b/src/runner/main_loop/shutdown.rs
index 23dc6a3..866da42 100644
--- a/src/runner/main_loop/shutdown.rs
+++ b/src/runner/main_loop/shutdown.rs
@@ -54,7 +54,7 @@ impl RunnerMainLoop {
loop {
tokio::select! {
biased;
- () = execution.expiry.wait_for_expiry() => {
+ () = execution.wait_for_expiry() => {
failures.push(Some(RunnerMainLoopError::ExecutionExpired));
failures.push(http_server.abort().await.map(RunnerMainLoopError::HttpServer));
break;
diff --git a/src/runner/main_loop/startup.rs b/src/runner/main_loop/startup.rs
index f4c7f44..e009f92 100644
--- a/src/runner/main_loop/startup.rs
+++ b/src/runner/main_loop/startup.rs
@@ -30,13 +30,13 @@ use crate::metrics::MetricsRuntime;
use crate::runner::diagnostics::RunnerDiagnostics;
use crate::runner::executable::CapturedRunnerExecutable;
use crate::runner::execution::RunnerExecution;
-use crate::runner::extensions::RunnerHooks;
+use crate::runner::extensions::{ExecutionScope, RunnerHooks};
use crate::runner::http::tls::TlsServerConfig;
use crate::runner::http::{RunnerHttpServer, RunnerHttpServerError};
use crate::runner::management::RunnerManagementSupervisor;
use crate::runner::metrics::RunnerMetrics;
use crate::runner::process_resources;
-use crate::runner::recovery::RecoveredRunnerState;
+use crate::runner::recovery::recover;
use crate::runner::runtime_resources::{
RunnerRuntimeCleanup, remove_stale_runtime_directory, stale_runner_directories,
};
@@ -55,7 +55,6 @@ impl RunnerMainLoop {
config: RunnerConfig,
resources: Arc<RunnerResources>,
state_layout: RunnerStateLayout,
- recovered: RecoveredRunnerState,
executable_image: CapturedRunnerExecutable,
tls: Option<TlsServerConfig>,
hooks: RunnerHooks,
@@ -63,21 +62,15 @@ impl RunnerMainLoop {
) -> Result<Self, RunnerMainLoopFailure> {
let meter = metrics.meter();
let metrics = RunnerMetrics::new(metrics, config.metrics().timeout());
- let execution = RunnerExecution::new(hooks.execution_policy);
+ let mut execution = RunnerExecution::new(hooks.execution_policy);
+ execution
+ .check(ExecutionScope { documents: &[] })
+ .map_err(RunnerMainLoopError::ExecutionDenied)?;
let document_verifier = config.tenon_document_verifier().map_err(|source| {
RunnerMainLoopFailure::from(RunnerMainLoopError::ManagementInitialization(source))
})?;
let pipeline_runtime_directory = state_layout.pipeline_runtime_directory();
- let recovered_management = RunnerManagementSupervisor::recover(
- state_layout.state_directory(),
- config.script_vm_limits(),
- recovered,
- hooks.document_protection,
- Some(&meter),
- |scope| execution.check(scope),
- )
- .map_err(RunnerMainLoopError::CpuObservation)?;
resources.recover_stale_groups().await.map_err(|source| {
RunnerMainLoopFailure::from(RunnerMainLoopError::ProcessResources(source))
})?;
@@ -87,6 +80,52 @@ impl RunnerMainLoop {
.map_err(|source| {
RunnerMainLoopFailure::from(RunnerMainLoopError::RuntimeResources(source))
})?;
+ let prepared: Result<_, RunnerMainLoopError> = async {
+ let recovered = recover(
+ &config,
+ &state_layout,
+ &hooks.artifact_protection,
+ runtime_resources.directory(),
+ )
+ .map_err(RunnerMainLoopError::Recovery)?;
+ let recovered_management = RunnerManagementSupervisor::recover(
+ state_layout.state_directory(),
+ config.script_vm_limits(),
+ recovered,
+ hooks.artifact_protection,
+ Some(&meter),
+ |scope| execution.check(scope),
+ )
+ .map_err(RunnerMainLoopError::CpuObservation)?;
+ // Process notifications received during recovery before any Pipeline starts.
+ let (check, policy_changed, _) = execution.split();
+ let policy_result = tokio::select! {
+ biased;
+ changed = policy_changed => {
+ if changed {
+ recovered_management.check_execution(check)
+ } else {
+ Ok(())
+ }
+ }
+ () = std::future::ready(()) => Ok(()),
+ };
+ policy_result.map_err(RunnerMainLoopError::ExecutionDenied)?;
+ Ok(recovered_management)
+ }
+ .await;
+ let recovered_management = match prepared {
+ Ok(management) => management,
+ Err(source) => {
+ let cleanup = runtime_resources
+ .cleanup(RunnerRuntimeCleanup::Remove)
+ .err()
+ .map(RunnerMainLoopError::RuntimeResources)
+ .into_iter()
+ .collect();
+ return Err(combine_failures(source, cleanup).into());
+ }
+ };
let diagnostics = RunnerDiagnostics::new();
let control_server = match RunnerControlServer::start(
runtime_resources.directory(),
@@ -124,7 +163,7 @@ impl RunnerMainLoop {
.pipeline_launcher()
.with_metrics(Some(&meter)),
initial_pipeline_directives,
- execution.expiry.clone(),
+ execution.expiry(),
);
let http_server = match RunnerHttpServer::start(
config.http_listen_address(),
diff --git a/src/runner/main_loop/tests.rs b/src/runner/main_loop/tests.rs
index 2bec23f..1e3aea9 100644
--- a/src/runner/main_loop/tests.rs
+++ b/src/runner/main_loop/tests.rs
@@ -142,14 +142,15 @@ async fn admitted_http_handler_drain_keeps_advancing_pipeline_state() -> io::Res
let recovered = recover(
&config,
&state_layout,
- &RunnerHooks::default().document_protection,
+ &RunnerHooks::default().artifact_protection,
+ &state_layout.pipeline_runtime_directory(),
)
.map_err(io::Error::other)?;
let (state, initial_directives) = RunnerManagementState::recover(
config.state_directory(),
config.script_vm_limits(),
recovered,
- RunnerHooks::default().document_protection,
+ RunnerHooks::default().artifact_protection,
None,
|_| Ok(()),
)?;
diff --git a/src/runner/management/jobs.rs b/src/runner/management/jobs.rs
index 6751edd..ea319ee 100644
--- a/src/runner/management/jobs.rs
+++ b/src/runner/management/jobs.rs
@@ -31,7 +31,7 @@ use super::{
};
use crate::identifiers::{ExactVersion, ProgramName, TenonDocumentId};
use crate::runner::document_store::{TenonDocumentStore, TenonDocumentStoreError};
-use crate::runner::extensions::DocumentProtection;
+use crate::runner::extensions::ArtifactProtection;
use crate::runner::plugin::package::PluginPackageError;
use crate::runner::plugin::store::{
PluginProgramInstallResult, PluginProgramStore, PluginStoreError,
@@ -73,7 +73,7 @@ pub(super) enum ManagementMutationJob {
outcome: PutDocumentOutcome,
response: ManagementReply<Result<PutDocumentOutcome, PutDocumentFailure>>,
document_store_directory: PathBuf,
- protection: Arc<dyn DocumentProtection>,
+ protection: Arc<dyn ArtifactProtection>,
},
DeleteDocument {
id: TenonDocumentId,
@@ -220,6 +220,9 @@ fn program_install_failure(
error: PluginStoreError,
) -> Result<PluginOperationFailure, PluginStoreError> {
match error {
+ PluginStoreError::PackageAccessRejected => {
+ Ok(PluginOperationFailure::Invalid { code: error.code() })
+ }
PluginStoreError::PlatformMismatch { platforms } => {
Ok(PluginOperationFailure::PlatformMismatch { platforms })
}
diff --git a/src/runner/management/state/mod.rs b/src/runner/management/state/mod.rs
index dfc6ad0..8af9954 100644
--- a/src/runner/management/state/mod.rs
+++ b/src/runner/management/state/mod.rs
@@ -43,7 +43,7 @@ use crate::config::ScriptVmLimits;
use crate::contracts::core::{PipelineStatusSnapshot, PluginInstanceState};
use crate::identifiers::TenonDocumentId;
use crate::runner::document_store::TenonDocumentEtag;
-use crate::runner::extensions::{DocumentProtection, ExecutionDenied, ExecutionScope};
+use crate::runner::extensions::{ArtifactProtection, ExecutionDenied, ExecutionScope};
use crate::runner::pipeline::{
PipelineLifecycleTarget, RuntimeResolution, RuntimeResolutionIssue, RuntimeResolver,
};
@@ -76,7 +76,7 @@ pub(crate) enum PipelineDirective {
/// The only owner of Runner management facts and business decisions.
pub(crate) struct RunnerManagementState {
state_directory: PathBuf,
- document_protection: Arc<dyn DocumentProtection>,
+ artifact_protection: Arc<dyn ArtifactProtection>,
script_vm_limits: ScriptVmLimits,
available_cpu_count: NonZeroUsize,
documents: HashMap<TenonDocumentId, RunnerDocumentState>,
@@ -89,19 +89,34 @@ pub(crate) struct RunnerManagementState {
impl RunnerManagementState {
// ===== Recovery and lifecycle intake =====
+ pub(crate) fn check_execution(
+ &self,
+ authorize: impl FnOnce(ExecutionScope<'_>) -> Result<(), ExecutionDenied>,
+ ) -> Result<(), ExecutionDenied> {
+ let documents = self
+ .documents
+ .values()
+ .filter(|state| state.is_accepted())
+ .map(|state| state.document.as_ref())
+ .collect::<Vec<_>>();
+ authorize(ExecutionScope {
+ documents: &documents,
+ })
+ }
+
/// Restores the complete management state and derives initial targets.
pub(crate) fn recover(
state_directory: &Path,
script_vm_limits: ScriptVmLimits,
recovered: RecoveredRunnerState,
- document_protection: Arc<dyn DocumentProtection>,
+ artifact_protection: Arc<dyn ArtifactProtection>,
metrics: Option<&Meter>,
mut authorize: impl FnMut(ExecutionScope<'_>) -> Result<(), ExecutionDenied>,
) -> io::Result<(Self, Box<[PipelineDirective]>)> {
let (documents, programs) = recovered.into_parts();
let mut owner = Self {
state_directory: state_directory.to_path_buf(),
- document_protection,
+ artifact_protection,
script_vm_limits,
available_cpu_count: process_resources::available_cpu_count()?,
documents: HashMap::new(),
@@ -324,7 +339,7 @@ impl RunnerManagementState {
outcome,
response,
document_store_directory: self.document_store_directory(),
- protection: Arc::clone(&self.document_protection),
+ protection: Arc::clone(&self.artifact_protection),
})
}
@@ -831,7 +846,7 @@ fn plugin_instance_views(
#[cfg(test)]
pub(crate) mod test_support {
use super::*;
- use crate::runner::extensions::Plaintext;
+ use crate::runner::extensions::ByPass;
use crate::runner::plugin::store::test_support::empty_store;
#[allow(
@@ -844,7 +859,7 @@ pub(crate) mod test_support {
) -> RunnerManagementState {
RunnerManagementState {
state_directory: state_directory.to_path_buf(),
- document_protection: Arc::new(Plaintext),
+ artifact_protection: Arc::new(ByPass),
script_vm_limits,
available_cpu_count: process_resources::available_cpu_count()
.expect("Runner must determine available CPU count"),
@@ -916,14 +931,15 @@ mod tests {
let recovered = recover(
&config,
&layout,
- &RunnerHooks::default().document_protection,
+ &RunnerHooks::default().artifact_protection,
+ &layout.pipeline_runtime_directory(),
)
.map_err(io::Error::other)?;
let (mut state, initial) = RunnerManagementState::recover(
directory.path(),
config.script_vm_limits(),
recovered,
- RunnerHooks::default().document_protection,
+ RunnerHooks::default().artifact_protection,
None,
|_| Ok(()),
)?;
diff --git a/src/runner/management/supervisor/mod.rs b/src/runner/management/supervisor/mod.rs
index 3ea1a1d..2708e81 100644
--- a/src/runner/management/supervisor/mod.rs
+++ b/src/runner/management/supervisor/mod.rs
@@ -32,7 +32,7 @@ use super::{
RunnerManagementQuery,
};
use crate::config::ScriptVmLimits;
-use crate::runner::extensions::{DocumentProtection, ExecutionDenied, ExecutionScope};
+use crate::runner::extensions::{ArtifactProtection, ExecutionDenied, ExecutionScope};
use crate::runner::plugin::store::PluginStoreError;
use crate::runner::recovery::RecoveredRunnerState;
use crate::tenon_document::verified::TenonDocumentVerifier;
@@ -61,6 +61,15 @@ pub(crate) struct RecoveredRunnerManagement {
initial_pipeline_directives: Box<[PipelineDirective]>,
}
+impl RecoveredRunnerManagement {
+ pub(crate) fn check_execution(
+ &self,
+ authorize: impl FnOnce(ExecutionScope<'_>) -> Result<(), ExecutionDenied>,
+ ) -> Result<(), ExecutionDenied> {
+ self.state.check_execution(authorize)
+ }
+}
+
/// Owns all scheduling around the Runner's mutable management facts.
pub(crate) struct RunnerManagementSupervisor {
state: RunnerManagementState,
@@ -69,15 +78,24 @@ pub(crate) struct RunnerManagementSupervisor {
mutation_task: Option<JoinHandle<Result<ManagementMutationResult, PluginStoreError>>>,
document_preparations: VecDeque<JoinHandle<ManagementDocumentPreparationResult>>,
pipeline_publication: PipelinePublication,
+ // A policy notification arrived during this mutation. Clear after completion.
+ recheck_after_mutation: bool,
}
impl RunnerManagementSupervisor {
+ pub(crate) fn check_execution(
+ &mut self,
+ authorize: impl FnOnce(ExecutionScope<'_>) -> Result<(), ExecutionDenied>,
+ ) -> Result<(), ExecutionDenied> {
+ self.recheck_after_mutation |= self.mutation_task.is_some();
+ self.state.check_execution(authorize)
+ }
/// Restores management facts without opening their command interface.
pub(crate) fn recover(
state_directory: &Path,
script_vm_limits: ScriptVmLimits,
recovered: RecoveredRunnerState,
- document_protection: Arc<dyn DocumentProtection>,
+ artifact_protection: Arc<dyn ArtifactProtection>,
metrics: Option<&Meter>,
authorize: impl FnMut(ExecutionScope<'_>) -> Result<(), ExecutionDenied>,
) -> io::Result<RecoveredRunnerManagement> {
@@ -85,7 +103,7 @@ impl RunnerManagementSupervisor {
state_directory,
script_vm_limits,
recovered,
- document_protection,
+ artifact_protection,
metrics,
authorize,
)?;
@@ -109,6 +127,7 @@ impl RunnerManagementSupervisor {
mutation_task: None,
document_preparations: VecDeque::new(),
pipeline_publication: PipelinePublication::Open,
+ recheck_after_mutation: false,
},
management,
recovered.initial_pipeline_directives,
@@ -128,8 +147,16 @@ impl RunnerManagementSupervisor {
tokio::select! {
result = wait_for_mutation(&mut self.mutation_task), if self.mutation_task.is_some() => {
self.mutation_task = None;
+ let recheck = std::mem::take(&mut self.recheck_after_mutation)
+ && self.pipeline_publication == PipelinePublication::Open
+ && matches!(&result, Ok(Ok(ManagementMutationResult::PutDocument { result: Ok(_), .. })));
match self.complete_mutation(result) {
- Ok(Some(commit)) => return RunnerManagementEvent::CommitReady(commit),
+ Ok(Some(commit)) => {
+ if recheck && let Err(source) = self.state.check_execution(&mut authorize) {
+ return RunnerManagementEvent::Failure(RunnerManagementSupervisorError::ExecutionDenied(source));
+ }
+ return RunnerManagementEvent::CommitReady(commit);
+ }
Ok(None) => {}
Err(source) => return RunnerManagementEvent::Failure(source),
}
@@ -224,6 +251,7 @@ impl RunnerManagementSupervisor {
self.mutation_task.as_ref()?;
let result = wait_for_mutation(&mut self.mutation_task).await;
self.mutation_task = None;
+ self.recheck_after_mutation = false;
self.complete_mutation(result).err()
}
@@ -328,12 +356,14 @@ pub(crate) enum RunnerManagementSupervisorError {
Task(JoinError),
/// Preserves the Store failure that requires the whole Runner to stop.
ProgramStore(PluginStoreError),
+ ExecutionDenied(ExecutionDenied),
InterfaceClosed,
}
impl fmt::Display for RunnerManagementSupervisorError {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
+ Self::ExecutionDenied(_) => formatter.write_str("Runner execution is not permitted"),
Self::Task(_) => formatter.write_str("Runner management task failed"),
Self::ProgramStore(_) => formatter.write_str("Runner Plugin Program Store failed"),
Self::InterfaceClosed => formatter.write_str("Runner management interface closed"),
@@ -344,6 +374,7 @@ impl fmt::Display for RunnerManagementSupervisorError {
impl Error for RunnerManagementSupervisorError {
fn source(&self) -> Option<&(dyn Error + 'static)> {
match self {
+ Self::ExecutionDenied(source) => Some(source),
Self::Task(source) => Some(source),
Self::ProgramStore(source) => Some(source),
Self::InterfaceClosed => None,
@@ -372,6 +403,7 @@ pub(crate) mod test_support {
mutation_task: None,
document_preparations: VecDeque::new(),
pipeline_publication: PipelinePublication::Open,
+ recheck_after_mutation: false,
},
management,
)
@@ -380,6 +412,7 @@ pub(crate) mod test_support {
#[cfg(test)]
mod tests {
+ mod policy_changes;
mod programs;
mod uploads;
@@ -431,6 +464,7 @@ mod tests {
mutation_task: Some(mutation_task),
document_preparations: VecDeque::new(),
pipeline_publication: PipelinePublication::Open,
+ recheck_after_mutation: false,
};
let _ = finish.send(());
let RunnerManagementEvent::CommitReady(commit) = supervisor.next_event(|_| Ok(())).await
@@ -492,6 +526,7 @@ mod tests {
mutation_task: Some(mutation_task),
document_preparations: VecDeque::new(),
pipeline_publication: PipelinePublication::Open,
+ recheck_after_mutation: false,
};
let shutdown = supervisor.shutdown();
@@ -536,6 +571,7 @@ mod tests {
mutation_task: Some(mutation_task),
document_preparations: VecDeque::new(),
pipeline_publication: PipelinePublication::Open,
+ recheck_after_mutation: false,
};
let fill_queue = async move {
@@ -635,6 +671,7 @@ mod tests {
mutation_task: None,
document_preparations: VecDeque::from([first_task, second_task]),
pipeline_publication: PipelinePublication::Open,
+ recheck_after_mutation: false,
};
let release_preparations = async move {
diff --git a/src/runner/management/supervisor/tests/policy_changes.rs b/src/runner/management/supervisor/tests/policy_changes.rs
new file mode 100644
index 0000000..24d2860
--- /dev/null
+++ b/src/runner/management/supervisor/tests/policy_changes.rs
@@ -0,0 +1,160 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use super::programs::program_supervisor;
+use super::*;
+use crate::runner::recovery::recover;
+use crate::runner::test_support::write_document;
+use std::fs;
+use std::io::Write;
+use std::sync::Mutex;
+
+struct PausedDocumentWrite {
+ started: Mutex<Option<oneshot::Sender<()>>>,
+ release: Mutex<std_mpsc::Receiver<()>>,
+}
+
+impl ArtifactProtection for PausedDocumentWrite {
+ fn protect(&self, source: &[u8], output: &mut dyn Write) -> io::Result<()> {
+ if let Some(started) = self
+ .started
+ .lock()
+ .map_err(|_| io::Error::other("Lock failed"))?
+ .take()
+ {
+ let _ = started.send(());
+ let _ = self
+ .release
+ .lock()
+ .map_err(|_| io::Error::other("Lock failed"))?
+ .recv();
+ }
+ output.write_all(source)
+ }
+
+ fn unprotect(&self, stored: &[u8], output: &mut dyn Write) -> io::Result<()> {
+ output.write_all(stored)
+ }
+}
+
+#[tokio::test(flavor = "current_thread")]
+async fn policy_notification_cannot_miss_a_document_commit() -> io::Result<()> {
+ for during_write in [true, false] {
+ let directory = tempfile::tempdir()?;
+ drop(program_supervisor(directory.path())?);
+ write_document(directory.path(), "first", document("first"))?;
+ let config = load_config(directory.path())?;
+ let layout = prepare_runner_state_directory(directory.path()).map_err(io::Error::other)?;
+ let runtime = tempfile::tempdir()?;
+ let (started, was_started) = oneshot::channel();
+ let (release, released) = std_mpsc::channel();
+ let protection: Arc<dyn ArtifactProtection> = Arc::new(PausedDocumentWrite {
+ started: Mutex::new(Some(started)),
+ release: Mutex::new(released),
+ });
+ let recovered =
+ recover(&config, &layout, &protection, runtime.path()).map_err(io::Error::other)?;
+ let recovered = RunnerManagementSupervisor::recover(
+ directory.path(),
+ config.script_vm_limits(),
+ recovered,
+ protection,
+ None,
+ |_| Ok(()),
+ )?;
+ let (mut supervisor, management, initial) = RunnerManagementSupervisor::start(
+ recovered,
+ config.tenon_document_verifier().map_err(io::Error::other)?,
+ );
+ assert_eq!(initial.len(), 1);
+ let source = document("second");
+ let put = management.put_document(
+ TenonDocumentId::try_from("second").map_err(io::Error::other)?,
+ PutDocumentPrecondition::Create,
+ source.clone().into_bytes().into_boxed_slice(),
+ );
+ tokio::pin!(put);
+ let mut old_decisions = Vec::new();
+ tokio::select! {
+ _ = supervisor.next_event(|scope| { old_decisions.push(scope.documents.len()); Ok(()) }) => {
+ return Err(io::Error::other("Document commit completed before release"));
+ }
+ _ = &mut put => return Err(io::Error::other("PUT completed before release")),
+ result = was_started => result.map_err(io::Error::other)?,
+ }
+ assert_eq!(old_decisions, [2]);
+
+ let mut new_decisions = Vec::new();
+ let mut limit_to_one = |scope: ExecutionScope<'_>| {
+ new_decisions.push(scope.documents.len());
+ if scope.documents.len() > 1 {
+ Err(ExecutionDenied::new(
+ "test.document_limit",
+ "One Document is permitted",
+ None,
+ ))
+ } else {
+ Ok(())
+ }
+ };
+ if during_write {
+ supervisor
+ .check_execution(&mut limit_to_one)
+ .map_err(io::Error::other)?;
+ }
+ let _ = release.send(());
+ if during_write {
+ assert!(matches!(
+ supervisor.next_event(&mut limit_to_one).await,
+ RunnerManagementEvent::Failure(RunnerManagementSupervisorError::ExecutionDenied(error))
+ if error.code() == "test.document_limit"
+ ));
+ assert!(matches!(put.await, Err(RunnerUnavailable)));
+ assert_eq!(new_decisions, [1, 2]);
+ } else {
+ let RunnerManagementEvent::CommitReady(commit) = supervisor
+ .next_event(|_| {
+ Err(ExecutionDenied::new(
+ "test.unexpected_call",
+ "No policy call is necessary",
+ None,
+ ))
+ })
+ .await
+ else {
+ return Err(io::Error::other("Document commit did not complete"));
+ };
+ commit.publish_after(|directives| {
+ assert_eq!(directives.len(), 1);
+ Ok::<_, io::Error>(())
+ })?;
+ assert!(matches!(put.await, Ok(Ok(_))));
+ assert!(supervisor.check_execution(&mut limit_to_one).is_err());
+ assert_eq!(new_decisions, [2]);
+ }
+ assert!(!supervisor.recheck_after_mutation);
+ supervisor.shutdown().await.map_err(io::Error::other)?;
+ let saved = fs::read_dir(directory.path().join("tenon-documents"))?
+ .map(|entry| fs::read(entry?.path()))
+ .collect::<io::Result<Vec<_>>>()?;
+ assert_eq!(saved.len(), 2);
+ assert!(saved.iter().any(|bytes| bytes == source.as_bytes()));
+ }
+ Ok(())
+}
diff --git a/src/runner/management/supervisor/tests/programs.rs b/src/runner/management/supervisor/tests/programs.rs
index ae0d5e7..1dda6a6 100644
--- a/src/runner/management/supervisor/tests/programs.rs
+++ b/src/runner/management/supervisor/tests/programs.rs
@@ -166,7 +166,7 @@ async fn program_store_failure_rejects_queued_work_in_running_and_draining_modes
let release = hold_mutation_before_filesystem_work(&mut supervisor, job);
let target = directory
.path()
- .join("plugins/programs/com.example.source/1.0.0");
+ .join("plugins/programs/com.example.source/.tenon-artifact-92521fc3cbd964bdc9f584a991b89fddaa5754ed1cc96d6d42445338669c1305");
fs::rename(&target, directory.path().join("externally-moved-program"))?;
let (response, queued_mutation) = oneshot::channel();
@@ -249,8 +249,8 @@ async fn program_store_failure_rejects_queued_work_in_running_and_draining_modes
assert!(
directory
.path()
- .join("plugins/programs/com.example.sink/1.0.0")
- .is_dir()
+ .join("plugins/programs/com.example.sink/.tenon-artifact-92521fc3cbd964bdc9f584a991b89fddaa5754ed1cc96d6d42445338669c1305")
+ .is_file()
);
}
Ok(())
@@ -273,7 +273,7 @@ async fn shutdown_propagates_started_program_failure_and_joins_other_work() -> i
fs::rename(
directory
.path()
- .join("plugins/programs/com.example.source/1.0.0"),
+ .join("plugins/programs/com.example.source/.tenon-artifact-92521fc3cbd964bdc9f584a991b89fddaa5754ed1cc96d6d42445338669c1305"),
directory.path().join("externally-moved-program"),
)?;
let (response, query) = oneshot::channel();
@@ -336,8 +336,12 @@ pub(super) fn program_supervisor(
) -> io::Result<(RunnerManagementSupervisor, RunnerManagementClient)> {
let layout = prepare_runner_state_directory(directory).map_err(io::Error::other)?;
{
- let mut programs = PluginProgramStore::recover(layout.plugin_program_store_directory())
- .map_err(io::Error::other)?;
+ let mut programs = PluginProgramStore::recover(
+ layout.plugin_program_store_directory(),
+ layout.plugin_program_store_directory(),
+ std::sync::Arc::new(crate::runner::extensions::ByPass),
+ )
+ .map_err(io::Error::other)?;
for interface in [PluginInterface::Source, PluginInterface::Sink] {
programs
.install(Cursor::new(valid_program_package(interface)?))
@@ -348,14 +352,15 @@ pub(super) fn program_supervisor(
let recovered = recover(
&config,
&layout,
- &RunnerHooks::default().document_protection,
+ &RunnerHooks::default().artifact_protection,
+ &layout.pipeline_runtime_directory(),
)
.map_err(io::Error::other)?;
let (state, directives) = RunnerManagementState::recover(
directory,
config.script_vm_limits(),
recovered,
- RunnerHooks::default().document_protection,
+ RunnerHooks::default().artifact_protection,
None,
|_| Ok(()),
)?;
diff --git a/src/runner/management/supervisor/tests/uploads.rs b/src/runner/management/supervisor/tests/uploads.rs
index 4e11d28..c9ac84c 100644
--- a/src/runner/management/supervisor/tests/uploads.rs
+++ b/src/runner/management/supervisor/tests/uploads.rs
@@ -122,8 +122,12 @@ async fn cancellation_after_end_does_not_abandon_installation_during_shutdown()
shutdown.await.map_err(io::Error::other)?;
}
drop(supervisor);
- let recovered = PluginProgramStore::recover(directory.path().join("plugins/programs"))
- .map_err(io::Error::other)?;
+ let recovered = PluginProgramStore::recover(
+ directory.path().join("plugins/programs"),
+ directory.path().join("plugins/programs"),
+ std::sync::Arc::new(crate::runner::extensions::ByPass),
+ )
+ .map_err(io::Error::other)?;
assert_eq!(recovered.programs().count(), 3);
Ok(())
}
diff --git a/src/runner/mod.rs b/src/runner/mod.rs
index 19de155..1be92f0 100644
--- a/src/runner/mod.rs
+++ b/src/runner/mod.rs
@@ -48,7 +48,6 @@ use crate::metrics::{CoreProcess, MetricsRuntime};
use executable::CapturedRunnerExecutable;
use extensions::RunnerHooks;
use main_loop::{RunnerMainLoop, RunnerMainLoopFailure, RunnerShutdownReport};
-use recovery::{RunnerRecoveryError, recover};
use state_directory::{RunnerStateDirectoryError, prepare_runner_state_directory};
use std::error::Error;
use std::ffi::OsString;
@@ -108,8 +107,6 @@ fn run(
CapturedRunnerExecutable::capture_current().map_err(RunnerStartupError::ExecutableImage)?;
let state_layout = prepare_runner_state_directory(config.state_directory())
.map_err(RunnerStartupError::StateDirectory)?;
- let recovered = recover(&config, &state_layout, &hooks.document_protection)
- .map_err(RunnerStartupError::Recovery)?;
let runtime = runtime::Builder::new_current_thread()
.enable_all()
.build()
@@ -125,7 +122,6 @@ fn run(
config,
resources,
state_layout,
- recovered,
executable_image,
tls,
hooks,
@@ -170,7 +166,6 @@ enum RunnerStartupError {
ProcessResources(io::Error),
MetricsIdentity(getrandom::Error),
StateDirectory(RunnerStateDirectoryError),
- Recovery(RunnerRecoveryError),
ExecutableImage(io::Error),
RuntimeInitialization(io::Error),
ShutdownSignalRegistration(io::Error),
@@ -187,7 +182,6 @@ impl RunnerStartupError {
Self::ProcessResources(_) => "runner.process_resources_initialization_failed",
Self::MetricsIdentity(_) => "runner.metrics_identity_unavailable",
Self::StateDirectory(error) => error.code(),
- Self::Recovery(error) => error.code(),
Self::ExecutableImage(_) => "runner.executable_image_unavailable",
Self::RuntimeInitialization(_) => "runner.runtime_initialization_failed",
Self::ShutdownSignalRegistration(_) => "runner.shutdown_signal_failed",
@@ -204,7 +198,6 @@ impl RunnerStartupError {
| Self::ProcessResources(_)
| Self::Tls(_)
| Self::StateDirectory(_)
- | Self::Recovery(_)
| Self::ExecutableImage(_)
| Self::RuntimeInitialization(_)
| Self::ShutdownSignalRegistration(_)
@@ -222,7 +215,6 @@ impl RunnerStartupError {
| Self::ProcessResources(_)
| Self::Tls(_)
| Self::StateDirectory(_)
- | Self::Recovery(_)
| Self::ExecutableImage(_)
| Self::RuntimeInitialization(_)
| Self::ShutdownSignalRegistration(_) => None,
@@ -246,7 +238,6 @@ impl fmt::Display for RunnerStartupError {
}
Self::Tls(_) => formatter.write_str("Runner HTTPS identity could not be initialized"),
Self::StateDirectory(_) => formatter.write_str("Runner state directory setup failed"),
- Self::Recovery(_) => formatter.write_str("Runner startup recovery failed"),
Self::ExecutableImage(_) => {
formatter.write_str("Runner executable image could not be captured")
}
@@ -271,7 +262,6 @@ impl Error for RunnerStartupError {
Self::Tls(error) => Some(error),
Self::MetricsIdentity(error) => Some(error),
Self::StateDirectory(error) => Some(error),
- Self::Recovery(error) => Some(error),
Self::ExecutableImage(error)
| Self::RuntimeInitialization(error)
| Self::ShutdownSignalRegistration(error) => Some(error),
diff --git a/src/runner/pipeline/programs/tests.rs b/src/runner/pipeline/programs/tests.rs
index 81535f3..7112da1 100644
--- a/src/runner/pipeline/programs/tests.rs
+++ b/src/runner/pipeline/programs/tests.rs
@@ -94,7 +94,7 @@ fn all_interfaces_share_one_sorted_runtime_per_exact_program() -> io::Result<()>
#[test]
fn only_the_last_snapshot_release_allows_selected_program_uninstall() -> io::Result<()> {
- let (directory, mut store) = empty_store()?;
+ let (_directory, mut store) = empty_store()?;
for interface in [PluginInterface::Source, PluginInterface::Sink] {
store
.install(Cursor::new(valid_program_package(interface)?))
@@ -104,7 +104,12 @@ fn only_the_last_snapshot_release_allows_selected_program_uninstall() -> io::Res
let sink = identity("com.example.sink")?;
let first = capture_programs(&store, [source.clone(), source.clone()])?;
let second = capture_programs(&store, [source.clone()])?;
- let source_directory = directory.path().join("com.example.source/1.0.0");
+ let source_directory = std::path::PathBuf::from(&first.runtimes()[0].program_directory);
+ let sink_directory = store
+ .lookup(sink.program_name(), sink.exact_version())
+ .ok_or_else(|| io::Error::other("Sink Program is missing"))?
+ .directory()
+ .to_path_buf();
assert!(matches!(
store.uninstall(source.program_name(), source.exact_version()),
@@ -127,7 +132,7 @@ fn only_the_last_snapshot_release_allows_selected_program_uninstall() -> io::Res
.lookup(sink.program_name(), sink.exact_version())
.is_none()
);
- assert!(!directory.path().join("com.example.sink/1.0.0").exists());
+ assert!(!sink_directory.exists());
drop(first);
assert!(matches!(
@@ -194,8 +199,12 @@ fn equivalent_install_and_unrelated_changes_preserve_snapshot_material_and_reten
fn empty_store() -> io::Result<(TempDir, PluginProgramStore)> {
let directory = tempfile::tempdir()?;
- let store =
- PluginProgramStore::recover(directory.path().to_owned()).map_err(io::Error::other)?;
+ let store = PluginProgramStore::recover(
+ directory.path().to_owned(),
+ directory.path().to_owned(),
+ std::sync::Arc::new(crate::runner::extensions::ByPass),
+ )
+ .map_err(io::Error::other)?;
Ok((directory, store))
}
diff --git a/src/runner/pipeline/runtime_resolution/tests.rs b/src/runner/pipeline/runtime_resolution/tests.rs
index bd1813d..6a24f5c 100644
--- a/src/runner/pipeline/runtime_resolution/tests.rs
+++ b/src/runner/pipeline/runtime_resolution/tests.rs
@@ -324,7 +324,7 @@ fn bidirectional_plans_retain_original_entries_until_the_last_plan_is_dropped()
store.uninstall(&name, &version).map_err(io::Error::other)?,
PluginUninstallOutcome::Uninstalled
);
- assert!(!directory.path().join("com.example.gateway/1.0.0").exists());
+ assert!(!directory.path().join("com.example.gateway/.tenon-artifact-92521fc3cbd964bdc9f584a991b89fddaa5754ed1cc96d6d42445338669c1305").exists());
Ok(())
}
@@ -448,8 +448,12 @@ struct RuntimeCase {
fn empty_store() -> io::Result<(TempDir, PluginProgramStore)> {
let directory = tempfile::tempdir()?;
- let store =
- PluginProgramStore::recover(directory.path().to_owned()).map_err(io::Error::other)?;
+ let store = PluginProgramStore::recover(
+ directory.path().to_owned(),
+ directory.path().to_owned(),
+ std::sync::Arc::new(crate::runner::extensions::ByPass),
+ )
+ .map_err(io::Error::other)?;
Ok((directory, store))
}
@@ -559,7 +563,6 @@ fn issues(resolution: &RuntimeResolution) -> io::Result<Value> {
#[test]
fn recovered_foreign_material_stays_queryable_while_old_and_unrelated_plans_survive()
-> io::Result<()> {
- use crate::runner::plugin::package::stage_plugin_program_package;
use crate::runner::plugin::package::tests::package_with_platforms;
use std::os::unix::fs::PermissionsExt as _;
let (directory, mut store) = empty_store()?;
@@ -572,16 +575,20 @@ fn recovered_foreign_material_stays_queryable_while_old_and_unrelated_plans_surv
let foreign_program = program("com.example.foreign", "source-and-sink");
let platforms = json!([crate::runner::plugin::package::tests::foreign_platform()]);
let bytes = package_with_platforms(&package(&foreign_program)?, &platforms)?;
- let staged = stage_plugin_program_package(Cursor::new(bytes), directory.path())
- .map_err(io::Error::other)?;
- let manifest_before = fs::read(staged.path().join("manifest.json"))?;
let namespace = directory.path().join("com.example.foreign");
fs::create_dir(&namespace)?;
fs::set_permissions(&namespace, fs::Permissions::from_mode(0o700))?;
- fs::rename(staged.path(), namespace.join("1.0.0"))?;
+ let saved = namespace
+ .join(".tenon-artifact-92521fc3cbd964bdc9f584a991b89fddaa5754ed1cc96d6d42445338669c1305");
+ fs::write(&saved, &bytes)?;
+ fs::set_permissions(&saved, fs::Permissions::from_mode(0o600))?;
drop(store);
- let mut recovered =
- PluginProgramStore::recover(directory.path().to_owned()).map_err(io::Error::other)?;
+ let mut recovered = PluginProgramStore::recover(
+ directory.path().to_owned(),
+ directory.path().to_owned(),
+ std::sync::Arc::new(crate::runner::extensions::ByPass),
+ )
+ .map_err(io::Error::other)?;
let RuntimeResolution::Ready(old_plan) = resolver.resolve(&original, &recovered) else {
return Err(io::Error::other("Original revision did not resolve"));
};
@@ -591,10 +598,7 @@ fn recovered_foreign_material_stays_queryable_while_old_and_unrelated_plans_surv
.lookup(&name, &version)
.ok_or_else(|| io::Error::other("Foreign material was deleted"))?;
assert_eq!(serde_json::to_value(entry.platforms())?, platforms);
- assert_eq!(
- fs::read(entry.directory().join("manifest.json"))?,
- manifest_before
- );
+ assert_eq!(fs::read(&saved)?, bytes);
let mut changed = dual_document();
changed["pluginInstances"]["left"]["programName"] = json!("com.example.foreign");
changed["pluginInstances"]["right"]["programName"] = json!("com.example.foreign");
diff --git a/src/runner/pipeline/test_support.rs b/src/runner/pipeline/test_support.rs
index 1119b68..97ed0e5 100644
--- a/src/runner/pipeline/test_support.rs
+++ b/src/runner/pipeline/test_support.rs
@@ -52,7 +52,12 @@ pub(in crate::runner) struct TargetFixture {
impl TargetFixture {
pub(in crate::runner) fn new(directory: &Path, config: &RunnerConfig) -> io::Result<Self> {
- let store = PluginProgramStore::recover(directory.to_owned()).map_err(io::Error::other)?;
+ let store = PluginProgramStore::recover(
+ directory.to_owned(),
+ directory.to_owned(),
+ std::sync::Arc::new(crate::runner::extensions::ByPass),
+ )
+ .map_err(io::Error::other)?;
let limits = config.script_vm_limits();
let mut fixture = Self {
store,
@@ -135,8 +140,12 @@ pub(super) struct RevisionFixture {
impl RevisionFixture {
pub(super) fn new(programs: &[&str]) -> io::Result<Self> {
let directory = tempfile::tempdir()?;
- let mut store =
- PluginProgramStore::recover(directory.path().to_owned()).map_err(io::Error::other)?;
+ let mut store = PluginProgramStore::recover(
+ directory.path().to_owned(),
+ directory.path().to_owned(),
+ std::sync::Arc::new(crate::runner::extensions::ByPass),
+ )
+ .map_err(io::Error::other)?;
let descriptor = valid_program_descriptor(PluginInterface::SourceAndSink)?;
for name in programs {
let manifest = json!({"programName": name, "exactVersion": "1.0.0", "interface": "source-and-sink", "platforms": [crate::runner::plugin::platform::Platform::CURRENT], "displayName": "Example Plugin", "description": "Read and write example records.", "command": ["./bin/start"]});
diff --git a/src/runner/plugin/package/mod.rs b/src/runner/plugin/package/mod.rs
index b4cbc31..041c14a 100644
--- a/src/runner/plugin/package/mod.rs
+++ b/src/runner/plugin/package/mod.rs
@@ -19,7 +19,7 @@
//! Safe staging and validation for Plugin Program packages.
//!
-//! Archive and Config Schema safety stay behind this package boundary.
+//! This module checks directory paths and Config Schemas.
//! It does not publish Store state, expose HTTP resources, or start processes.
mod safety;
@@ -28,12 +28,14 @@ use super::platform::Platform;
use crate::contracts::plugin::manifest_schema_bytes;
use crate::identifiers::{ExactVersion, ProgramName};
use crate::payload_contract::{PluginInterface, PluginProgramPayloadContract};
+use crate::runner::extensions;
use crate::strict_jsonc::parse_json;
+use extensions::ArtifactProtection;
pub(super) use safety::PluginConfigSchema;
pub(crate) use safety::PluginPackageError;
use safety::{
- InspectedPackageDirectory, StagedPackageArchive, inspect_package_directory,
- package_files_are_identical, stage_package_archive,
+ InspectedPackageDirectory, StagedPackageDirectory, inspect_package_directory,
+ package_files_are_identical, stage_package_directory,
};
use serde::Deserialize;
use std::collections::BTreeMap;
@@ -44,7 +46,7 @@ use std::path::Path;
#[must_use = "dropping a staged Program package removes its temporary directory"]
#[derive(Debug)]
pub(crate) struct StagedPluginProgramPackage {
- archive: StagedPackageArchive,
+ directory: StagedPackageDirectory,
validated: ValidatedPluginProgramPackage,
}
@@ -55,13 +57,13 @@ impl StagedPluginProgramPackage {
///
/// Returns the filesystem error when the staging tree cannot be removed.
pub(super) fn close(self) -> io::Result<()> {
- self.archive.close()
+ self.directory.close()
}
/// Returns the root of the private validated temporary tree.
#[must_use]
pub(crate) fn path(&self) -> &Path {
- self.archive.path()
+ self.directory.path()
}
/// Returns the validated Program name used for the Store identity path.
@@ -94,7 +96,7 @@ impl StagedPluginProgramPackage {
package_files_are_identical(
self.path(),
&self.validated.manifest_bytes,
- self.archive.file_digests(),
+ self.directory.file_digests(),
existing_root,
&existing.validated.manifest_bytes,
&existing.file_digests,
@@ -107,9 +109,15 @@ impl StagedPluginProgramPackage {
pub(super) fn into_snapshot(self) -> ValidatedPluginProgramSnapshot {
ValidatedPluginProgramSnapshot {
validated: self.validated,
- file_digests: self.archive.into_file_digests(),
+ file_digests: self.directory.into_file_digests(),
}
}
+
+ /// Transfers file cleanup to the Runner runtime owner.
+ pub(super) fn into_runtime(mut self) -> (std::path::PathBuf, ValidatedPluginProgramSnapshot) {
+ self.directory.keep();
+ (self.path().to_path_buf(), self.into_snapshot())
+ }
}
/// Parsed immutable Program materials shared by staging and startup recovery.
@@ -201,21 +209,25 @@ impl ValidatedPluginProgramSnapshot {
/// archive, fixed files, manifest, Config Schema, or shared
/// payload descriptor violates its current contract.
pub(crate) fn stage_plugin_program_package(
+ protection: &dyn ArtifactProtection,
package: impl Read,
staging_parent: &Path,
) -> Result<StagedPluginProgramPackage, PluginPackageError> {
- let archive = stage_package_archive(package, staging_parent)?;
+ let directory = stage_package_directory(protection, package, staging_parent)?;
let validated = (|| {
validate_program_materials(
- archive.read_manifest()?,
- archive.read_config_schema()?,
- archive.read_payload_descriptor()?,
+ directory.read_manifest()?,
+ directory.read_config_schema()?,
+ directory.read_payload_descriptor()?,
)
})();
match validated {
- Ok(validated) => Ok(StagedPluginProgramPackage { archive, validated }),
+ Ok(validated) => Ok(StagedPluginProgramPackage {
+ directory,
+ validated,
+ }),
Err(error) => {
- archive
+ directory
.close()
.map_err(|source| PluginPackageError::FilesystemOperationFailed { source })?;
Err(error)
@@ -223,7 +235,7 @@ pub(crate) fn stage_plugin_program_package(
}
}
-/// Revalidates one committed current Program directory through the shared tree boundary.
+/// Revalidates one Program runtime directory through the shared tree boundary.
///
/// # Errors
///
diff --git a/src/runner/plugin/package/safety/mod.rs b/src/runner/plugin/package/safety/mod.rs
index 1725892..56be7f7 100644
--- a/src/runner/plugin/package/safety/mod.rs
+++ b/src/runner/plugin/package/safety/mod.rs
@@ -17,30 +17,31 @@
* under the License.
*/
-//! Safe archive, installed-tree, and Config Schema boundaries for Plugin packages.
+//! Safe package directories and Config Schemas.
//!
-//! Untrusted archives are extracted into an owned private directory with fixed
-//! resource limits. Installed scans reject unsafe filesystem objects. The
-//! Program package boundary owns manifest and payload semantics; this module
-//! never publishes Store state or starts a process.
+//! The selected hook writes files through a bounded directory writer.
+//! Directory scans reject unsafe filesystem objects. Package validation owns
+//! manifest and payload semantics. This module does not publish Store state.
+
+mod output;
use crate::identifiers::IdentifierParseError;
use crate::payload_contract::PayloadContractError;
+use crate::runner::extensions::ArtifactProtection;
use crate::runner::private_filesystem::{
has_owner_only_directory_permission, has_owner_only_file_permission,
- set_owner_only_directory_permission, set_owner_only_file_permission,
+ set_owner_only_directory_permission,
};
use crate::strict_jsonc::{StrictJsonError, parse_json};
-use flate2::bufread::GzDecoder;
+use output::PackageOutput;
use serde_json::Value;
use sha2::{Digest, Sha256};
use std::collections::{BTreeMap, HashSet};
use std::error::Error;
use std::fmt;
-use std::fs::{self, OpenOptions};
-use std::io::{self, BufReader, Read, Write};
+use std::fs;
+use std::io::{self, Read};
use std::path::{Path, PathBuf};
-use tar::{Archive, PaxExtensions};
use tempfile::TempDir;
const MANIFEST_PATH: &str = "manifest.json";
@@ -48,14 +49,11 @@ const CONFIG_SCHEMA_PATH: &str = "config.schema.json";
const PAYLOAD_DESCRIPTOR_PATH: &str = "payload.descriptor.pb";
pub(crate) const STAGING_DIRECTORY_PREFIX: &str = ".tenon-plugin-";
const COPY_BUFFER_BYTES: usize = 64 * 1024;
-const TAR_END_MARKER_BYTES: u64 = 512;
-const MAX_ARCHIVE_ENTRIES: usize = 100_000;
+const MAX_PACKAGE_ENTRIES: usize = 100_000;
const MAX_EXTRACTED_FILE_BYTES: u64 = 4 * 1024 * 1024 * 1024;
-const MAX_TAR_STREAM_BYTES: u64 = 8 * 1024 * 1024 * 1024;
const MAX_MANIFEST_BYTES: u64 = 16 * 1024 * 1024;
const MAX_CONFIG_SCHEMA_BYTES: u64 = 16 * 1024 * 1024;
const MAX_PAYLOAD_DESCRIPTOR_BYTES: u64 = 64 * 1024 * 1024;
-const MAX_ARCHIVE_METADATA_BYTES: u64 = 64 * 1024;
const MAX_PACKAGE_PATH_BYTES: usize = 4096;
const MAX_PACKAGE_PATH_COMPONENT_BYTES: usize = 255;
const DRAFT_2020_12_URI: &str = "https://json-schema.org/draft/2020-12/schema";
@@ -137,37 +135,36 @@ impl fmt::Debug for PluginConfigSchema {
}
}
-/// Safely extracts one package into an owner-only temporary tree.
-///
-/// # Errors
-///
-/// Returns [`PluginPackageError`] when the gzip/tar stream, normalized path
-/// set, entry types, permissions, or fixed extraction limits are invalid, or
-/// when the private staging filesystem cannot complete the operation.
-pub(super) fn stage_package_archive(
+/// Writes package contents into a private directory through the selected hook.
+pub(super) fn stage_package_directory(
+ protection: &dyn ArtifactProtection,
package: impl Read,
staging_parent: &Path,
-) -> Result<StagedPackageArchive, PluginPackageError> {
- let extracted = extract_package_archive(package, staging_parent, PackageLimits::production())?;
- Ok(StagedPackageArchive {
- directory: extracted.directory,
- file_digests: extracted.file_digests,
- })
+) -> Result<StagedPackageDirectory, PluginPackageError> {
+ extract_package_directory(
+ protection,
+ package,
+ staging_parent,
+ PackageLimits::production(),
+ )
}
-fn extract_package_archive(
- package: impl Read,
+fn extract_package_directory(
+ protection: &dyn ArtifactProtection,
+ mut package: impl Read,
staging_parent: &Path,
limits: PackageLimits,
-) -> Result<ExtractedPackageArchive, PluginPackageError> {
+) -> Result<StagedPackageDirectory, PluginPackageError> {
let directory = tempfile::Builder::new()
.prefix(STAGING_DIRECTORY_PREFIX)
.tempdir_in(staging_parent)
.map_err(|source| PluginPackageError::FilesystemOperationFailed { source })?;
set_owner_only_directory_permission(directory.path())
.map_err(|source| PluginPackageError::FilesystemOperationFailed { source })?;
- let extracted = match extract_archive(package, directory.path(), limits) {
- Ok(extracted) => extracted,
+ let mut output = PackageOutput::new(directory.path(), limits);
+ let result = protection.open_plugin_package(&mut package, &mut output);
+ let file_digests = match output.finish(result) {
+ Ok(file_digests) => file_digests,
Err(error) => {
directory
.close()
@@ -175,10 +172,9 @@ fn extract_package_archive(
return Err(error);
}
};
-
- Ok(ExtractedPackageArchive {
+ Ok(StagedPackageDirectory {
directory,
- file_digests: extracted.file_digests,
+ file_digests,
})
}
@@ -225,7 +221,7 @@ pub(super) fn inspect_package_directory(
entry_count = entry_count
.checked_add(1)
.ok_or(PluginPackageError::PackageTooLarge)?;
- if entry_count > limits.archive_entries {
+ if entry_count > limits.entries {
return Err(PluginPackageError::PackageTooLarge);
}
@@ -436,9 +432,8 @@ fn has_multiple_hard_links(_metadata: &fs::Metadata) -> bool {
#[derive(Clone, Copy, Debug)]
struct PackageLimits {
- archive_entries: usize,
+ entries: usize,
extracted_file_bytes: u64,
- tar_stream_bytes: u64,
manifest_bytes: u64,
config_schema_bytes: u64,
payload_descriptor_bytes: u64,
@@ -447,9 +442,8 @@ struct PackageLimits {
impl PackageLimits {
const fn production() -> Self {
Self {
- archive_entries: MAX_ARCHIVE_ENTRIES,
+ entries: MAX_PACKAGE_ENTRIES,
extracted_file_bytes: MAX_EXTRACTED_FILE_BYTES,
- tar_stream_bytes: MAX_TAR_STREAM_BYTES,
manifest_bytes: MAX_MANIFEST_BYTES,
config_schema_bytes: MAX_CONFIG_SCHEMA_BYTES,
payload_descriptor_bytes: MAX_PAYLOAD_DESCRIPTOR_BYTES,
@@ -462,12 +456,12 @@ impl PackageLimits {
/// Extraction has validated archive structure and filesystem safety only;
/// callers still own all fixed-file parsing and contract validation.
#[derive(Debug)]
-pub(super) struct StagedPackageArchive {
+pub(super) struct StagedPackageDirectory {
directory: TempDir,
file_digests: BTreeMap<Box<str>, [u8; 32]>,
}
-impl StagedPackageArchive {
+impl StagedPackageDirectory {
/// Removes a rejected or unused temporary tree and reports cleanup failure.
///
/// # Errors
@@ -496,6 +490,10 @@ impl StagedPackageArchive {
self.file_digests
}
+ pub(super) fn keep(&mut self) {
+ self.directory.disable_cleanup(true);
+ }
+
/// Reads the bounded fixed `manifest.json` file.
///
/// # Errors
@@ -531,242 +529,6 @@ impl StagedPackageArchive {
}
}
-#[derive(Debug)]
-struct ExtractedPackageArchive {
- directory: TempDir,
- file_digests: BTreeMap<Box<str>, [u8; 32]>,
-}
-
-#[derive(Debug)]
-struct ExtractedArchive {
- file_digests: BTreeMap<Box<str>, [u8; 32]>,
-}
-
-fn extract_archive(
- package: impl Read,
- output_root: &Path,
- limits: PackageLimits,
-) -> Result<ExtractedArchive, PluginPackageError> {
- let input = BufReader::new(package);
- let gzip = GzDecoder::new(input);
- let bounded = BoundedReader::new(gzip, limits.tar_stream_bytes);
- let mut archive = Archive::new(bounded);
- let mut file_digests = BTreeMap::new();
- let mut explicit_paths = HashSet::new();
- let mut known_directories = HashSet::new();
- let mut known_files = HashSet::new();
- let mut pending_metadata = PendingArchiveMetadata::default();
- let mut entry_count = 0_usize;
- let mut extracted_file_bytes = 0_u64;
-
- {
- let entries = archive.entries().map_err(archive_error)?.raw(true);
- for entry in entries {
- entry_count = entry_count
- .checked_add(1)
- .ok_or(PluginPackageError::PackageTooLarge)?;
- if entry_count > limits.archive_entries {
- return Err(PluginPackageError::PackageTooLarge);
- }
-
- let mut entry = entry.map_err(archive_error)?;
- let entry_type = entry.header().entry_type();
- if entry_type.is_gnu_longname() {
- if pending_metadata.gnu_path.is_some() {
- return Err(PluginPackageError::PackageInvalid);
- }
- let mut path = read_archive_metadata(&mut entry)?;
- if path.last() == Some(&0) {
- path.pop();
- }
- if path.is_empty() || path.len() > MAX_PACKAGE_PATH_BYTES {
- return Err(PluginPackageError::PackageInvalid);
- }
- pending_metadata.gnu_path = Some(path);
- continue;
- }
- if entry_type.is_pax_local_extensions() {
- if pending_metadata.has_pax_local() {
- return Err(PluginPackageError::PackageInvalid);
- }
- let metadata = read_archive_metadata(&mut entry)?;
- pending_metadata.set_pax_local(&metadata)?;
- continue;
- }
- if entry_type.is_pax_global_extensions() {
- if pending_metadata.has_local() {
- return Err(PluginPackageError::PackageInvalid);
- }
- let metadata = read_archive_metadata(&mut entry)?;
- validate_pax_global(&metadata)?;
- continue;
- }
- if !entry_type.is_file() && !entry_type.is_dir() {
- return Err(PluginPackageError::PackageInvalid);
- }
- let entry_kind = if entry_type.is_dir() {
- PackageEntryKind::Directory
- } else {
- PackageEntryKind::File
- };
- let header_size = entry.header().size().map_err(archive_error)?;
- let path_bytes = pending_metadata.take_path(entry.header().path_bytes().as_ref())?;
- if pending_metadata
- .take_size()
- .is_some_and(|pax_size| pax_size != header_size)
- {
- return Err(PluginPackageError::PackageInvalid);
- }
- let normalized = NormalizedPackagePath::parse(&path_bytes, entry_kind)?;
- if !explicit_paths.insert(normalized.text.clone()) {
- return Err(PluginPackageError::PackageInvalid);
- }
- register_path_shape(&normalized, &mut known_directories, &mut known_files)?;
- if known_directories.len() + known_files.len() > limits.archive_entries {
- return Err(PluginPackageError::PackageTooLarge);
- }
-
- let target = normalized.to_path(output_root);
- if normalized.kind == PackageEntryKind::Directory {
- if header_size != 0 {
- return Err(PluginPackageError::PackageInvalid);
- }
- create_owner_only_directories(output_root, &normalized.components)?;
- continue;
- }
-
- let size = header_size;
- extracted_file_bytes = extracted_file_bytes
- .checked_add(size)
- .ok_or(PluginPackageError::PackageTooLarge)?;
- if extracted_file_bytes > limits.extracted_file_bytes
- || control_file_limit(normalized.as_str(), limits).is_some_and(|limit| size > limit)
- {
- return Err(PluginPackageError::PackageTooLarge);
- }
- let parent_count = normalized.components.len().saturating_sub(1);
- create_owner_only_directories(output_root, &normalized.components[..parent_count])?;
- let mut output = OpenOptions::new()
- .write(true)
- .create_new(true)
- .open(&target)
- .map_err(|source| PluginPackageError::FilesystemOperationFailed { source })?;
- let digest = copy_and_digest(&mut entry, &mut output, size)?;
- set_owner_only_file_permission(&target)
- .map_err(|source| PluginPackageError::FilesystemOperationFailed { source })?;
- if normalized.as_str() != MANIFEST_PATH {
- file_digests.insert(normalized.text, digest);
- }
- }
- }
- if pending_metadata.has_local() {
- return Err(PluginPackageError::PackageInvalid);
- }
-
- let mut bounded = archive.into_inner();
- validate_tar_end(&mut bounded)?;
- let gzip = bounded.into_inner();
- let mut compressed_input = gzip.into_inner();
- let mut trailing = [0_u8; 1];
- if compressed_input
- .read(&mut trailing)
- .map_err(archive_error)?
- != 0
- {
- return Err(PluginPackageError::PackageInvalid);
- }
-
- Ok(ExtractedArchive { file_digests })
-}
-
-#[derive(Debug, Default)]
-struct PendingArchiveMetadata {
- gnu_path: Option<Vec<u8>>,
- pax_path: Option<Vec<u8>>,
- pax_size: Option<u64>,
- pax_local_seen: bool,
-}
-
-impl PendingArchiveMetadata {
- fn has_pax_local(&self) -> bool {
- self.pax_local_seen
- }
-
- fn has_local(&self) -> bool {
- self.gnu_path.is_some() || self.has_pax_local()
- }
-
- fn set_pax_local(&mut self, metadata: &[u8]) -> Result<(), PluginPackageError> {
- self.pax_local_seen = true;
- for extension in PaxExtensions::new(metadata) {
- let extension = extension.map_err(archive_error)?;
- match extension.key_bytes() {
- b"path" => {
- if self.pax_path.is_some()
- || extension.value_bytes().is_empty()
- || extension.value_bytes().len() > MAX_PACKAGE_PATH_BYTES
- {
- return Err(PluginPackageError::PackageInvalid);
- }
- self.pax_path = Some(extension.value_bytes().to_vec());
- }
- b"size" => {
- if self.pax_size.is_some() {
- return Err(PluginPackageError::PackageInvalid);
- }
- let value = std::str::from_utf8(extension.value_bytes())
- .ok()
- .and_then(|value| value.parse::<u64>().ok())
- .ok_or(PluginPackageError::PackageInvalid)?;
- self.pax_size = Some(value);
- }
- b"linkpath" => return Err(PluginPackageError::PackageInvalid),
- key if key.starts_with(b"GNU.sparse.") => {
- return Err(PluginPackageError::PackageInvalid);
- }
- _ => {}
- }
- }
- Ok(())
- }
-
- fn take_path(&mut self, header_path: &[u8]) -> Result<Vec<u8>, PluginPackageError> {
- match (self.gnu_path.take(), self.pax_path.take()) {
- (Some(_), Some(_)) => Err(PluginPackageError::PackageInvalid),
- (Some(path), None) | (None, Some(path)) => Ok(path),
- (None, None) => Ok(header_path.to_vec()),
- }
- }
-
- fn take_size(&mut self) -> Option<u64> {
- self.pax_local_seen = false;
- self.pax_size.take()
- }
-}
-
-fn validate_pax_global(metadata: &[u8]) -> Result<(), PluginPackageError> {
- for extension in PaxExtensions::new(metadata) {
- let extension = extension.map_err(archive_error)?;
- let key = extension.key_bytes();
- if matches!(key, b"path" | b"size" | b"linkpath") || key.starts_with(b"GNU.sparse.") {
- return Err(PluginPackageError::PackageInvalid);
- }
- }
- Ok(())
-}
-
-fn read_archive_metadata(input: &mut impl Read) -> Result<Vec<u8>, PluginPackageError> {
- let mut bytes = Vec::new();
- input
- .take(MAX_ARCHIVE_METADATA_BYTES + 1)
- .read_to_end(&mut bytes)
- .map_err(archive_error)?;
- if bytes.len() as u64 > MAX_ARCHIVE_METADATA_BYTES {
- return Err(PluginPackageError::PackageTooLarge);
- }
- Ok(bytes)
-}
-
fn control_file_limit(path: &str, limits: PackageLimits) -> Option<u64> {
match path {
MANIFEST_PATH => Some(limits.manifest_bytes),
@@ -804,57 +566,6 @@ fn register_path_shape(
Ok(())
}
-fn copy_and_digest(
- input: &mut impl Read,
- output: &mut impl Write,
- expected_size: u64,
-) -> Result<[u8; 32], PluginPackageError> {
- let mut digest = Sha256::new();
- let mut copied = 0_u64;
- let mut buffer = [0_u8; COPY_BUFFER_BYTES];
- loop {
- let count = input.read(&mut buffer).map_err(archive_error)?;
- if count == 0 {
- break;
- }
- output
- .write_all(&buffer[..count])
- .map_err(|source| PluginPackageError::FilesystemOperationFailed { source })?;
- digest.update(&buffer[..count]);
- copied = copied
- .checked_add(count as u64)
- .ok_or(PluginPackageError::PackageTooLarge)?;
- }
- if copied != expected_size {
- return Err(PluginPackageError::PackageInvalid);
- }
- output
- .flush()
- .map_err(|source| PluginPackageError::FilesystemOperationFailed { source })?;
- Ok(digest.finalize().into())
-}
-
-fn validate_tar_end(input: &mut impl Read) -> Result<(), PluginPackageError> {
- let mut trailing_bytes = 0_u64;
- let mut buffer = [0_u8; COPY_BUFFER_BYTES];
- loop {
- let count = input.read(&mut buffer).map_err(archive_error)?;
- if count == 0 {
- break;
- }
- if buffer[..count].iter().any(|byte| *byte != 0) {
- return Err(PluginPackageError::PackageInvalid);
- }
- trailing_bytes = trailing_bytes
- .checked_add(count as u64)
- .ok_or(PluginPackageError::PackageTooLarge)?;
- }
- if trailing_bytes < TAR_END_MARKER_BYTES {
- return Err(PluginPackageError::PackageInvalid);
- }
- Ok(())
-}
-
fn read_control_file(
root: &Path,
relative_path: &str,
@@ -1110,67 +821,19 @@ impl NormalizedPackagePath {
}
}
-#[derive(Debug)]
-struct BoundedReader<R> {
- inner: R,
- remaining: u64,
-}
-
-impl<R> BoundedReader<R> {
- const fn new(inner: R, limit: u64) -> Self {
- Self {
- inner,
- remaining: limit,
- }
- }
-
- fn into_inner(self) -> R {
- self.inner
- }
-}
-
-impl<R: Read> Read for BoundedReader<R> {
- fn read(&mut self, buffer: &mut [u8]) -> io::Result<usize> {
- if buffer.is_empty() {
- return Ok(0);
- }
- if self.remaining == 0 {
- let mut probe = [0_u8; 1];
- return match self.inner.read(&mut probe)? {
- 0 => Ok(0),
- _ => Err(io::Error::from(io::ErrorKind::FileTooLarge)),
- };
- }
- let available = usize::try_from(self.remaining)
- .unwrap_or(usize::MAX)
- .min(buffer.len());
- let count = self.inner.read(&mut buffer[..available])?;
- self.remaining -= count as u64;
- Ok(count)
- }
-}
-
-fn archive_error(source: io::Error) -> PluginPackageError {
- match source.downcast::<PluginPackageError>() {
- Ok(error) => error,
- Err(source) if source.kind() == io::ErrorKind::FileTooLarge => {
- PluginPackageError::PackageTooLarge
- }
- Err(source) => PluginPackageError::ArchiveInvalid { source },
- }
-}
-
/// A stable failure category for Plugin package staging and validation.
#[derive(Debug)]
#[non_exhaustive]
pub(crate) enum PluginPackageError {
+ /// The selected package hook could not open the package.
+ AccessRejected,
/// The expanded archive exceeds a fixed Runner implementation limit.
PackageTooLarge,
/// The archive structure, path set, or exact ordinary file set is invalid.
PackageInvalid,
- /// gzip or tar decoding failed.
+ /// Package decoding failed.
ArchiveInvalid {
- /// The underlying streaming decoder failure.
+ /// The underlying decoder failure.
source: io::Error,
},
/// The embedded manifest Schema cannot be parsed.
@@ -1225,6 +888,7 @@ impl PluginPackageError {
#[must_use]
pub(crate) const fn code(&self) -> &'static str {
match self {
+ Self::AccessRejected => "plugin_package_open_failed",
Self::PackageTooLarge => "plugin_package_too_large",
Self::PackageInvalid | Self::ArchiveInvalid { .. } => "plugin_package_invalid",
Self::ManifestJsonInvalid { .. } | Self::ManifestSchemaInvalid { .. } => {
@@ -1244,9 +908,10 @@ impl PluginPackageError {
impl fmt::Display for PluginPackageError {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter.write_str(match self {
+ Self::AccessRejected => "Plugin package access was rejected",
Self::PackageTooLarge => "Plugin package exceeds an implementation limit",
Self::PackageInvalid => "Plugin package structure is invalid",
- Self::ArchiveInvalid { .. } => "Plugin package archive is invalid",
+ Self::ArchiveInvalid { .. } => "Plugin package format is invalid",
Self::InternalContractInvalid { .. }
| Self::InternalManifestSchemaCompilationFailed { .. } => {
"Embedded Plugin manifest contract is invalid"
@@ -1280,7 +945,7 @@ impl Error for PluginPackageError {
Self::ManifestIdentifierProjectionFailed { source } => Some(source),
Self::ConfigSchemaInvalid { source } => Some(source),
Self::PayloadContractInvalid { source } => Some(source),
- Self::PackageTooLarge | Self::PackageInvalid => None,
+ Self::AccessRejected | Self::PackageTooLarge | Self::PackageInvalid => None,
}
}
}
diff --git a/src/runner/plugin/package/safety/output.rs b/src/runner/plugin/package/safety/output.rs
new file mode 100644
index 0000000..63eac3e
--- /dev/null
+++ b/src/runner/plugin/package/safety/output.rs
@@ -0,0 +1,190 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+//! Writes decoded package files while the core owns their temporary directory.
+
+use super::{
+ COPY_BUFFER_BYTES, MANIFEST_PATH, NormalizedPackagePath, PackageEntryKind, PackageLimits,
+ PluginPackageError, control_file_limit, create_owner_only_directories, register_path_shape,
+};
+use crate::runner::extensions::PluginPackageOutput;
+use crate::runner::private_filesystem::set_owner_only_file_permission;
+use sha2::{Digest, Sha256};
+use std::collections::{BTreeMap, HashSet};
+use std::fs::OpenOptions;
+use std::io::{self, Read, Write};
+use std::path::Path;
+
+/// Writes package files and records failures for core validation.
+#[derive(Debug)]
+pub(super) struct PackageOutput<'a> {
+ root: &'a Path,
+ limits: PackageLimits,
+ explicit_paths: HashSet<Box<str>>,
+ directories: HashSet<Box<str>>,
+ files: HashSet<Box<str>>,
+ file_digests: BTreeMap<Box<str>, [u8; 32]>,
+ file_bytes: u64,
+ failure: Option<PluginPackageError>,
+}
+
+impl<'a> PackageOutput<'a> {
+ pub(super) fn new(root: &'a Path, limits: PackageLimits) -> Self {
+ Self {
+ root,
+ limits,
+ explicit_paths: HashSet::new(),
+ directories: HashSet::new(),
+ files: HashSet::new(),
+ file_digests: BTreeMap::new(),
+ file_bytes: 0,
+ failure: None,
+ }
+ }
+
+ pub(super) fn finish(
+ self,
+ result: io::Result<()>,
+ ) -> Result<BTreeMap<Box<str>, [u8; 32]>, PluginPackageError> {
+ if let Some(error) = self.failure {
+ return Err(error);
+ }
+ result.map_err(|error| match error.kind() {
+ io::ErrorKind::FileTooLarge => PluginPackageError::PackageTooLarge,
+ io::ErrorKind::InvalidData | io::ErrorKind::UnexpectedEof => {
+ PluginPackageError::ArchiveInvalid { source: error }
+ }
+ _ => PluginPackageError::AccessRejected,
+ })?;
+ Ok(self.file_digests)
+ }
+
+ fn require_usable(&self) -> io::Result<()> {
+ if self.failure.is_some() {
+ Err(io::Error::other("Plugin package output has already failed"))
+ } else {
+ Ok(())
+ }
+ }
+
+ fn record(&mut self, result: Result<(), PluginPackageError>) -> io::Result<()> {
+ result.map_err(|source| {
+ let kind = match &source {
+ PluginPackageError::FilesystemOperationFailed { source } => source.kind(),
+ PluginPackageError::PackageTooLarge => io::ErrorKind::FileTooLarge,
+ _ => io::ErrorKind::InvalidData,
+ };
+ self.failure = Some(source);
+ io::Error::new(kind, "Plugin package output failed")
+ })
+ }
+
+ fn register(
+ &mut self,
+ path: &str,
+ kind: PackageEntryKind,
+ ) -> Result<NormalizedPackagePath, PluginPackageError> {
+ let path = NormalizedPackagePath::parse(path.as_bytes(), kind)?;
+ if !self.explicit_paths.insert(path.text.clone()) {
+ return Err(PluginPackageError::PackageInvalid);
+ }
+ register_path_shape(&path, &mut self.directories, &mut self.files)?;
+ if self.directories.len() + self.files.len() > self.limits.entries {
+ return Err(PluginPackageError::PackageTooLarge);
+ }
+ Ok(path)
+ }
+
+ fn write_file_contents(
+ &mut self,
+ path: &str,
+ source: &mut dyn Read,
+ ) -> Result<(), PluginPackageError> {
+ let path = self.register(path, PackageEntryKind::File)?;
+ let limit = control_file_limit(path.as_str(), self.limits)
+ .unwrap_or(self.limits.extracted_file_bytes);
+ let target = path.to_path(self.root);
+ let parents = path.components.len().saturating_sub(1);
+ create_owner_only_directories(self.root, &path.components[..parents])?;
+ let mut file = OpenOptions::new()
+ .write(true)
+ .create_new(true)
+ .open(&target)
+ .map_err(|source| PluginPackageError::FilesystemOperationFailed { source })?;
+ let mut file_bytes = 0_u64;
+ let mut digest = Sha256::new();
+ let mut buffer = [0_u8; COPY_BUFFER_BYTES];
+ loop {
+ let count = match source.read(&mut buffer) {
+ Ok(count) => count,
+ Err(error) if error.kind() == io::ErrorKind::Interrupted => continue,
+ Err(error) => {
+ return Err(match error.kind() {
+ io::ErrorKind::FileTooLarge => PluginPackageError::PackageTooLarge,
+ _ => PluginPackageError::PackageInvalid,
+ });
+ }
+ };
+ if count == 0 {
+ break;
+ }
+ file_bytes = file_bytes
+ .checked_add(count as u64)
+ .ok_or(PluginPackageError::PackageTooLarge)?;
+ self.file_bytes = self
+ .file_bytes
+ .checked_add(count as u64)
+ .ok_or(PluginPackageError::PackageTooLarge)?;
+ if file_bytes > limit || self.file_bytes > self.limits.extracted_file_bytes {
+ return Err(PluginPackageError::PackageTooLarge);
+ }
+ file.write_all(&buffer[..count])
+ .map_err(|source| PluginPackageError::FilesystemOperationFailed { source })?;
+ digest.update(&buffer[..count]);
+ }
+ file.flush()
+ .map_err(|source| PluginPackageError::FilesystemOperationFailed { source })?;
+ set_owner_only_file_permission(&target)
+ .map_err(|source| PluginPackageError::FilesystemOperationFailed { source })?;
+ if path.as_str() != MANIFEST_PATH {
+ self.file_digests
+ .insert(path.text, digest.finalize().into());
+ }
+ Ok(())
+ }
+}
+
+impl PluginPackageOutput for PackageOutput<'_> {
+ fn write_file(&mut self, path: &str, source: &mut dyn Read) -> io::Result<()> {
+ self.require_usable()?;
+ let result = self.write_file_contents(path, source);
+ self.record(result)
+ }
+
+ fn create_dir(&mut self, path: &str) -> io::Result<()> {
+ self.require_usable()?;
+ let result = self
+ .register(path, PackageEntryKind::Directory)
+ .and_then(|path| create_owner_only_directories(self.root, &path.components));
+ self.record(result)
+ }
+}
+
+#[cfg(test)]
+mod tests;
diff --git a/src/runner/plugin/package/safety/output/tests.rs b/src/runner/plugin/package/safety/output/tests.rs
new file mode 100644
index 0000000..9f78753
--- /dev/null
+++ b/src/runner/plugin/package/safety/output/tests.rs
@@ -0,0 +1,119 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use super::*;
+use crate::runner::extensions::ArtifactProtection;
+use crate::runner::plugin::package::safety::stage_package_directory;
+use std::io::Cursor;
+
+#[test]
+fn size_limits_reject_excess_bytes_before_writing_them() -> io::Result<()> {
+ for (path, limits) in [
+ (
+ "program",
+ PackageLimits {
+ extracted_file_bytes: 3,
+ ..PackageLimits::production()
+ },
+ ),
+ (
+ "manifest.json",
+ PackageLimits {
+ manifest_bytes: 3,
+ ..PackageLimits::production()
+ },
+ ),
+ ] {
+ let directory = tempfile::tempdir()?;
+ let mut output = PackageOutput::new(directory.path(), limits);
+ let mut source = Cursor::new(b"abc").chain(Cursor::new(b"d"));
+ assert!(output.write_file(path, &mut source).is_err());
+ assert_eq!(std::fs::read(directory.path().join(path))?, b"abc");
+ assert!(matches!(
+ output.finish(Ok(())),
+ Err(PluginPackageError::PackageTooLarge)
+ ));
+ }
+ Ok(())
+}
+
+#[test]
+fn ignored_path_and_filesystem_errors_keep_the_output_invalid() -> io::Result<()> {
+ for path in ["../outside", "occupied"] {
+ let directory = tempfile::tempdir()?;
+ std::fs::write(directory.path().join("occupied"), b"existing")?;
+ let mut output = PackageOutput::new(directory.path(), PackageLimits::production());
+ assert!(output.write_file(path, &mut Cursor::new(b"new")).is_err());
+ assert!(output.create_dir("later").is_err());
+ let error = output
+ .finish(Ok(()))
+ .err()
+ .ok_or_else(|| io::Error::other("Failed output was accepted"))?;
+ if path == "occupied" {
+ assert!(matches!(
+ error,
+ PluginPackageError::FilesystemOperationFailed { .. }
+ ));
+ } else {
+ assert!(matches!(error, PluginPackageError::PackageInvalid));
+ }
+ assert_eq!(
+ std::fs::read(directory.path().join("occupied"))?,
+ b"existing"
+ );
+ }
+ Ok(())
+}
+
+#[test]
+fn a_hook_cannot_accept_partial_file_content_after_a_read_failure() -> io::Result<()> {
+ struct FailedContent(io::ErrorKind);
+ impl Read for FailedContent {
+ fn read(&mut self, _output: &mut [u8]) -> io::Result<usize> {
+ Err(self.0.into())
+ }
+ }
+ struct IgnoreFailure(io::ErrorKind);
+ impl ArtifactProtection for IgnoreFailure {
+ fn open_plugin_package(
+ &self,
+ _source: &mut dyn Read,
+ output: &mut dyn PluginPackageOutput,
+ ) -> io::Result<()> {
+ let _ = output.write_file(
+ "program",
+ &mut Cursor::new(b"partial").chain(FailedContent(self.0)),
+ );
+ Ok(())
+ }
+ }
+ for (kind, code) in [
+ (io::ErrorKind::PermissionDenied, "plugin_package_invalid"),
+ (io::ErrorKind::UnexpectedEof, "plugin_package_invalid"),
+ (io::ErrorKind::FileTooLarge, "plugin_package_too_large"),
+ ] {
+ let parent = tempfile::tempdir()?;
+ let error = stage_package_directory(&IgnoreFailure(kind), io::empty(), parent.path())
+ .err()
+ .ok_or_else(|| io::Error::other("Failed content was accepted"))?;
+ assert_eq!(error.code(), code);
+ assert!(parent.path().read_dir()?.next().is_none());
+ }
+ Ok(())
+}
diff --git a/src/runner/plugin/package/safety/tests.rs b/src/runner/plugin/package/safety/tests.rs
index 61f6534..d20c546 100644
--- a/src/runner/plugin/package/safety/tests.rs
+++ b/src/runner/plugin/package/safety/tests.rs
@@ -18,9 +18,9 @@
*/
use super::{
- MAX_ARCHIVE_METADATA_BYTES, MAX_PACKAGE_PATH_BYTES, MAX_PACKAGE_PATH_COMPONENT_BYTES,
- PackageEntryKind, PackageLimits, PluginConfigSchema, PluginPackageError,
- extract_package_archive, package_files_are_identical, stage_package_archive,
+ MAX_PACKAGE_PATH_BYTES, MAX_PACKAGE_PATH_COMPONENT_BYTES, PackageEntryKind, PackageLimits,
+ PluginConfigSchema, PluginPackageError, extract_package_directory, package_files_are_identical,
+ stage_package_directory,
};
use flate2::Compression;
use flate2::read::GzDecoder as ReadGzDecoder;
@@ -60,12 +60,14 @@ fn archive_modes_do_not_change_private_staging_permissions() -> io::Result<()> {
let restrictive_parent = tempfile::tempdir()?;
let permissive_parent = tempfile::tempdir()?;
- let restrictive = stage_package_archive(
+ let restrictive = stage_package_directory(
+ &crate::runner::extensions::ByPass,
Cursor::new(package_with_modes(0o000)?),
restrictive_parent.path(),
)
.map_err(io::Error::other)?;
- let permissive = stage_package_archive(
+ let permissive = stage_package_directory(
+ &crate::runner::extensions::ByPass,
Cursor::new(package_with_modes(0o777)?),
permissive_parent.path(),
)
@@ -96,8 +98,12 @@ fn archive_modes_do_not_change_private_staging_permissions() -> io::Result<()> {
#[test]
fn staged_directory_is_removed_when_owner_is_dropped() -> io::Result<()> {
let staging_parent = tempfile::tempdir()?;
- let staged = stage_package_archive(Cursor::new(archive_fixture()?), staging_parent.path())
- .map_err(io::Error::other)?;
+ let staged = stage_package_directory(
+ &crate::runner::extensions::ByPass,
+ Cursor::new(archive_fixture()?),
+ staging_parent.path(),
+ )
+ .map_err(io::Error::other)?;
let path = staged.path().to_path_buf();
assert!(path.is_dir());
@@ -179,15 +185,23 @@ fn bounded_gnu_and_pax_paths_remain_ordinary_program_files() -> io::Result<()> {
let gnu_path = format!("lib/{}.bin", "long-name-".repeat(20));
let gnu_package = package_with_path(&gnu_path, b"gnu path")?;
let staging_parent = tempfile::tempdir()?;
- let staged = stage_package_archive(Cursor::new(gnu_package), staging_parent.path())
- .map_err(io::Error::other)?;
+ let staged = stage_package_directory(
+ &crate::runner::extensions::ByPass,
+ Cursor::new(gnu_package),
+ staging_parent.path(),
+ )
+ .map_err(io::Error::other)?;
assert_eq!(std::fs::read(staged.path().join(gnu_path))?, b"gnu path");
let pax_path = format!("resources/{}.txt", "pax-name-".repeat(20));
let pax_package = package_with_pax_path(&pax_path)?;
let staging_parent = tempfile::tempdir()?;
- let staged = stage_package_archive(Cursor::new(pax_package), staging_parent.path())
- .map_err(io::Error::other)?;
+ let staged = stage_package_directory(
+ &crate::runner::extensions::ByPass,
+ Cursor::new(pax_package),
+ staging_parent.path(),
+ )
+ .map_err(io::Error::other)?;
assert_eq!(std::fs::read(staged.path().join(pax_path))?, b"pax path");
Ok(())
}
@@ -301,17 +315,13 @@ fn implementation_limits_fail_before_unbounded_extraction() -> io::Result<()> {
let package = archive_fixture()?;
for limits in [
PackageLimits {
- archive_entries: 2,
+ entries: 2,
..PackageLimits::production()
},
PackageLimits {
extracted_file_bytes: 1,
..PackageLimits::production()
},
- PackageLimits {
- tar_stream_bytes: 512,
- ..PackageLimits::production()
- },
] {
assert_package_limit(package.clone(), limits)?;
}
@@ -321,8 +331,12 @@ fn implementation_limits_fail_before_unbounded_extraction() -> io::Result<()> {
#[test]
fn fixed_control_files_are_bounded_before_reading() -> io::Result<()> {
let staging_parent = tempfile::tempdir()?;
- let staged = stage_package_archive(Cursor::new(archive_fixture()?), staging_parent.path())
- .map_err(io::Error::other)?;
+ let staged = stage_package_directory(
+ &crate::runner::extensions::ByPass,
+ Cursor::new(archive_fixture()?),
+ staging_parent.path(),
+ )
+ .map_err(io::Error::other)?;
for path in [
"manifest.json",
"config.schema.json",
@@ -342,7 +356,7 @@ fn materialized_entry_limit_includes_implicit_directories() -> io::Result<()> {
assert_package_limit(
package,
PackageLimits {
- archive_entries: 2,
+ entries: 2,
..PackageLimits::production()
},
)
@@ -352,7 +366,7 @@ fn materialized_entry_limit_includes_implicit_directories() -> io::Result<()> {
fn oversized_archive_metadata_is_rejected_before_allocation_can_grow_unbounded() -> io::Result<()> {
let encoder = GzEncoder::new(Vec::new(), Compression::default());
let mut archive = Builder::new(encoder);
- let oversized = vec![b'x'; MAX_ARCHIVE_METADATA_BYTES as usize];
+ let oversized = vec![b'x'; 64 * 1024];
archive.append_pax_extensions([("comment", oversized.as_slice())])?;
append_file(&mut archive, "entry", b"value", 0o644)?;
let encoder = archive.into_inner()?;
@@ -399,9 +413,14 @@ proptest! {
fn assert_package_limit(package: Vec<u8>, limits: PackageLimits) -> io::Result<()> {
let staging_parent = tempfile::tempdir()?;
- let error = extract_package_archive(Cursor::new(package), staging_parent.path(), limits)
- .err()
- .ok_or_else(|| io::Error::other("oversized Plugin package was accepted"))?;
+ let error = extract_package_directory(
+ &crate::runner::extensions::ByPass,
+ Cursor::new(package),
+ staging_parent.path(),
+ limits,
+ )
+ .err()
+ .ok_or_else(|| io::Error::other("oversized Plugin package was accepted"))?;
assert_eq!(error.code(), "plugin_package_too_large");
assert!(staging_parent.path().read_dir()?.next().is_none());
Ok(())
@@ -415,9 +434,13 @@ fn assert_error_code(package: Vec<u8>, expected: &str) -> io::Result<()> {
fn stage_error(package: Vec<u8>) -> io::Result<PluginPackageError> {
let staging_parent = tempfile::tempdir()?;
- stage_package_archive(Cursor::new(package), staging_parent.path())
- .err()
- .ok_or_else(|| io::Error::other("invalid Plugin package was accepted"))
+ stage_package_directory(
+ &crate::runner::extensions::ByPass,
+ Cursor::new(package),
+ staging_parent.path(),
+ )
+ .err()
+ .ok_or_else(|| io::Error::other("invalid Plugin package was accepted"))
}
fn archive_fixture() -> io::Result<Vec<u8>> {
diff --git a/src/runner/plugin/package/tests.rs b/src/runner/plugin/package/tests.rs
index 6072883..4758131 100644
--- a/src/runner/plugin/package/tests.rs
+++ b/src/runner/plugin/package/tests.rs
@@ -161,13 +161,17 @@ fn current_payload_vectors_stage_one_shared_program_contract() -> io::Result<()>
let descriptor = compile_descriptor(&vector.files, &vector.entry_files)?;
let package = program_package(vector.interface, descriptor.clone(), valid_config_schema())?;
let staging_parent = tempfile::tempdir()?;
- let staged = stage_plugin_program_package(Cursor::new(package), staging_parent.path())
- .map_err(|error| {
- io::Error::other(format!(
- "valid payload vector {} was rejected: {error}",
- vector.name
- ))
- })?;
+ let staged = stage_plugin_program_package(
+ &crate::runner::extensions::ByPass,
+ Cursor::new(package),
+ staging_parent.path(),
+ )
+ .map_err(|error| {
+ io::Error::other(format!(
+ "valid payload vector {} was rejected: {error}",
+ vector.name
+ ))
+ })?;
let (program_name, exact_version, interface, command) = manifest_fields(&staged);
assert_eq!(interface, vector.interface);
@@ -314,8 +318,12 @@ fn fixed_contract_files_are_required_but_other_files_come_from_the_archive() ->
let package = program_package(PluginInterface::Source, descriptor, valid_config_schema())?;
let staging_parent = tempfile::tempdir()?;
- let staged = stage_plugin_program_package(Cursor::new(package), staging_parent.path())
- .map_err(io::Error::other)?;
+ let staged = stage_plugin_program_package(
+ &crate::runner::extensions::ByPass,
+ Cursor::new(package),
+ staging_parent.path(),
+ )
+ .map_err(io::Error::other)?;
assert_eq!(fs::read(staged.path().join("resources/data.bin"))?, b"data");
Ok(())
}
@@ -360,10 +368,18 @@ fn failed_validation_and_owner_drop_remove_the_staging_tree() -> io::Result<()>
source_descriptor()?,
valid_config_schema(),
)?;
- assert!(stage_plugin_program_package(Cursor::new(invalid), staging_parent.path()).is_err());
+ assert!(
+ stage_plugin_program_package(
+ &crate::runner::extensions::ByPass,
+ Cursor::new(invalid),
+ staging_parent.path()
+ )
+ .is_err()
+ );
assert!(staging_parent.path().read_dir()?.next().is_none());
let staged = stage_plugin_program_package(
+ &crate::runner::extensions::ByPass,
Cursor::new(program_package(
PluginInterface::Source,
source_descriptor()?,
@@ -380,9 +396,13 @@ fn failed_validation_and_owner_drop_remove_the_staging_tree() -> io::Result<()>
fn stage_error(package: Vec<u8>) -> io::Result<PluginPackageError> {
let staging_parent = tempfile::tempdir()?;
- stage_plugin_program_package(Cursor::new(package), staging_parent.path())
- .err()
- .ok_or_else(|| io::Error::other("invalid Plugin Program package was accepted"))
+ stage_plugin_program_package(
+ &crate::runner::extensions::ByPass,
+ Cursor::new(package),
+ staging_parent.path(),
+ )
+ .err()
+ .ok_or_else(|| io::Error::other("invalid Plugin Program package was accepted"))
}
#[cfg(unix)]
@@ -407,9 +427,13 @@ mod cleanup {
input: Cursor::new(package),
permission: &permission,
};
- let error = stage_plugin_program_package(input, directory.path())
- .err()
- .ok_or_else(|| io::Error::other("Invalid input was accepted"))?;
+ let error = stage_plugin_program_package(
+ &crate::runner::extensions::ByPass,
+ input,
+ directory.path(),
+ )
+ .err()
+ .ok_or_else(|| io::Error::other("Invalid input was accepted"))?;
drop(permission);
assert!(
matches!(error, PluginPackageError::FilesystemOperationFailed { source }
@@ -426,6 +450,7 @@ mod cleanup {
}
let directory = tempfile::tempdir()?;
let staged = stage_plugin_program_package(
+ &crate::runner::extensions::ByPass,
Cursor::new(valid_source_program_package()?),
directory.path(),
)
diff --git a/src/runner/plugin/store/filesystem.rs b/src/runner/plugin/store/filesystem.rs
index f0b1045..3096069 100644
--- a/src/runner/plugin/store/filesystem.rs
+++ b/src/runner/plugin/store/filesystem.rs
@@ -22,7 +22,7 @@
//! The Program Store owns mutation ordering and the live reference index.
//! This boundary synchronizes, renames, and inspects its private filesystem.
-use crate::identifiers::{ExactVersion, ProgramName};
+use crate::identifiers::ProgramName;
use crate::runner::plugin::package::PluginPackageError;
use crate::runner::plugin::platform::Platform;
use crate::runner::private_filesystem::{
@@ -36,6 +36,14 @@ use std::path::{Path, PathBuf};
/// Filesystem operations whose ordering defines Store recovery and publication.
pub(super) trait PublicationFilesystem {
+ fn sync_file(&self, path: &Path) -> Result<(), PluginStoreError> {
+ fs::File::open(path)
+ .and_then(|file| file.sync_all())
+ .map_err(|source| PluginStoreError::FilesystemOperationFailed {
+ path: path.to_path_buf(),
+ source,
+ })
+ }
/// Reads all children before recovery mutates the directory.
///
/// # Errors
@@ -69,14 +77,7 @@ pub(super) trait PublicationFilesystem {
})
}
- /// Makes every ordinary file and directory below `root` durable before publication.
- ///
- /// # Errors
- ///
- /// Returns [`PluginStoreError`] when the tree cannot be traversed or synchronized.
- fn sync_package_tree(&self, root: &Path) -> Result<(), PluginStoreError>;
-
- /// Atomically makes the staged tree visible at its final identity path.
+ /// Atomically moves the package file to its final path.
///
/// # Errors
///
@@ -104,10 +105,6 @@ pub(super) trait PublicationFilesystem {
pub(super) struct DurablePublicationFilesystem;
impl PublicationFilesystem for DurablePublicationFilesystem {
- fn sync_package_tree(&self, root: &Path) -> Result<(), PluginStoreError> {
- sync_package_tree(root)
- }
-
fn rename(&self, source: &Path, target: &Path) -> Result<(), PluginStoreError> {
fs::rename(source, target).map_err(|source| PluginStoreError::FilesystemOperationFailed {
path: target.to_path_buf(),
@@ -123,6 +120,8 @@ impl PublicationFilesystem for DurablePublicationFilesystem {
/// A stable failure from package validation, Store state, or private filesystem I/O.
#[derive(Debug)]
pub(crate) enum PluginStoreError {
+ /// The distribution did not grant package access.
+ PackageAccessRejected,
/// The uploaded package failed the shared package boundary.
PackageInvalid {
/// The precise package failure.
@@ -136,7 +135,7 @@ pub(crate) enum PluginStoreError {
ProgramInUse,
/// A committed package no longer passes its complete package contract.
InstalledPackageInvalid {
- /// The committed program directory.
+ /// The package or runtime directory.
path: PathBuf,
/// The precise package failure.
source: PluginPackageError,
@@ -167,6 +166,7 @@ impl PluginStoreError {
#[must_use]
pub(crate) const fn code(&self) -> &'static str {
match self {
+ Self::PackageAccessRejected => "plugin_package_open_failed",
Self::PackageInvalid { source } => source.code(),
Self::PlatformMismatch { .. } => "plugin_platform_mismatch",
Self::VersionConflict => "plugin_version_conflict",
@@ -184,6 +184,9 @@ impl PluginStoreError {
impl fmt::Display for PluginStoreError {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
+ Self::PackageAccessRejected => {
+ formatter.write_str("Plugin package access was rejected")
+ }
Self::PackageInvalid { .. } => formatter.write_str("Plugin package is invalid"),
Self::PlatformMismatch { platforms } => write!(
formatter,
@@ -228,6 +231,7 @@ impl fmt::Display for PluginStoreError {
impl Error for PluginStoreError {
fn source(&self) -> Option<&(dyn Error + 'static)> {
match self {
+ Self::PackageAccessRejected => None,
Self::PackageInvalid { source } | Self::InstalledPackageInvalid { source, .. } => {
Some(source)
}
@@ -256,21 +260,6 @@ pub(super) fn parse_program_entry(entry: &DirEntry) -> Result<ProgramName, Plugi
.map_err(|_| PluginStoreError::StoreIntegrityInvalid { path: entry.path() })
}
-/// Validates one private directory entry and parses its name as an exact version.
-///
-/// # Errors
-///
-/// Returns [`PluginStoreError`] when the entry is unsafe or its name is not canonical.
-pub(super) fn parse_version_entry(entry: &DirEntry) -> Result<ExactVersion, PluginStoreError> {
- validate_private_directory_entry(entry)?;
- let version = entry
- .file_name()
- .into_string()
- .map_err(|_| PluginStoreError::StoreIntegrityInvalid { path: entry.path() })?;
- ExactVersion::try_from(version)
- .map_err(|_| PluginStoreError::StoreIntegrityInvalid { path: entry.path() })
-}
-
/// Accepts only an owner-private ordinary directory without following a link.
///
/// # Errors
@@ -296,55 +285,6 @@ pub(super) fn validate_private_directory_entry(entry: &DirEntry) -> Result<(), P
}
}
-/// Synchronizes every file, then every directory from leaves through `root`.
-///
-/// # Errors
-///
-/// Returns [`PluginStoreError`] when traversal, object validation, or synchronization fails.
-pub(super) fn sync_package_tree(root: &Path) -> Result<(), PluginStoreError> {
- let mut pending = vec![root.to_path_buf()];
- let mut directories = Vec::new();
- while let Some(directory) = pending.pop() {
- directories.push(directory.clone());
- let entries = fs::read_dir(&directory).map_err(|source| {
- PluginStoreError::FilesystemOperationFailed {
- path: directory.clone(),
- source,
- }
- })?;
- for entry in entries {
- let entry = entry.map_err(|source| PluginStoreError::FilesystemOperationFailed {
- path: directory.clone(),
- source,
- })?;
- let path = entry.path();
- let file_type = entry.file_type().map_err(|source| {
- PluginStoreError::FilesystemOperationFailed {
- path: path.clone(),
- source,
- }
- })?;
- if file_type.is_dir() {
- pending.push(path);
- } else if file_type.is_file() {
- fs::File::open(&path)
- .and_then(|file| file.sync_all())
- .map_err(|source| PluginStoreError::FilesystemOperationFailed {
- path,
- source,
- })?;
- } else {
- return Err(PluginStoreError::StoreIntegrityInvalid { path });
- }
- }
- }
- directories.sort_by_key(|path| std::cmp::Reverse(path.components().count()));
- for directory in directories {
- sync_directory(&directory)?;
- }
- Ok(())
-}
-
/// Synchronizes one directory entry set to stable storage.
///
/// # Errors
diff --git a/src/runner/plugin/store/mod.rs b/src/runner/plugin/store/mod.rs
index c6d1438..f84bb70 100644
--- a/src/runner/plugin/store/mod.rs
+++ b/src/runner/plugin/store/mod.rs
@@ -19,20 +19,22 @@
//! Durable and in-memory ownership of current unified Plugin Programs.
//!
-//! Recovery validates the prepared filesystem tree and removes unusable child
-//! entries before publishing the only runtime identity map. The map contains
-//! only immutable validated Entries; there is no saved failure or missing state.
+//! The Store keeps received packages and reconstructs temporary runtime files.
+//! Package access or validation failure stops recovery and keeps the saved input.
+//! The only runtime identity map contains immutable validated Entries.
+//! There is no saved failure or missing state.
//! Runtime mutations are exclusive. Persistence or installed-layout failures
//! must terminate the Runner; only the next startup may recover the disk.
mod filesystem;
-mod upload;
+mod package_input;
+mod protected;
pub(crate) use filesystem::PluginStoreError;
use super::package::{PluginConfigSchema, PluginPackageError};
use super::package::{
- ValidatedPluginProgramMaterial, ValidatedPluginProgramSnapshot, stage_plugin_program_package,
+ ValidatedPluginProgramMaterial, ValidatedPluginProgramSnapshot,
validate_plugin_program_directory,
};
use super::platform::Platform;
@@ -41,11 +43,13 @@ use crate::identifiers::{ExactVersion, PluginProgramIdentity, ProgramName};
use crate::payload_contract::{
PluginInterface, PluginProgramPayloadContract, PluginProgramPayloadContractProjection,
};
+use crate::runner::extensions::ArtifactProtection;
use crate::runner::private_filesystem::has_owner_only_directory_permission;
use filesystem::{
- DurablePublicationFilesystem, PublicationFilesystem, parse_program_entry, parse_version_entry,
+ DurablePublicationFilesystem, PublicationFilesystem, parse_program_entry,
set_private_directory_permission,
};
+use protected::{ORIGINAL_PREFIX, StoredOriginal, original_file_name, version_digest};
use std::collections::{BTreeMap, HashMap};
use std::fs;
use std::io::{self, Read};
@@ -64,25 +68,45 @@ pub(crate) enum PluginProgramInstallResult {
const DELETION_TOMBSTONE_PREFIX: &str = ".tenon-plugin-delete-";
/// The only owner of durable Program layout and the validated runtime index.
-#[derive(Debug)]
pub(crate) struct PluginProgramStore {
directory: PathBuf,
+ runtime_directory: PathBuf,
+ protection: Arc<dyn ArtifactProtection>,
programs: HashMap<PluginProgramIdentity, Arc<PluginProgramEntry>>,
}
+impl std::fmt::Debug for PluginProgramStore {
+ fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ formatter
+ .debug_struct("PluginProgramStore")
+ .field("directory", &self.directory)
+ .field("programs", &self.programs)
+ .finish_non_exhaustive()
+ }
+}
+
impl PluginProgramStore {
- /// Validates Programs and removes unusable children from the prepared root.
+ /// Reconstructs runtime files from the saved packages in the prepared root.
///
- /// An invalid exact version is deleted without affecting valid siblings.
- /// An invalid or unreadable namespace is removed as a whole. Cleanup never
- /// follows symbolic links or removes the fixed root.
+ /// Package access or validation failure stops recovery and keeps the saved input.
+ /// An invalid or unreadable namespace is removed as a whole. Cleanup removes
+ /// unsupported entries without following symbolic links or removing the fixed root.
///
/// # Errors
///
- /// Returns [`PluginStoreError`] when the root cannot be completely
- /// enumerated or a required cleanup or directory sync fails.
- pub(crate) fn recover(directory: PathBuf) -> Result<Self, PluginStoreError> {
- Self::recover_with_filesystem(directory, &DurablePublicationFilesystem)
+ /// Returns [`PluginStoreError`] for a package failure, incomplete root
+ /// enumeration, or a required cleanup or directory sync failure.
+ pub(crate) fn recover(
+ directory: PathBuf,
+ runtime_directory: PathBuf,
+ protection: Arc<dyn ArtifactProtection>,
+ ) -> Result<Self, PluginStoreError> {
+ Self::recover_using(
+ directory,
+ runtime_directory,
+ protection,
+ &DurablePublicationFilesystem,
+ )
}
/// Installs and publishes one immutable Program in the same owner mutation.
@@ -141,12 +165,16 @@ impl PluginProgramStore {
.map(|(identity, entry)| (identity.program_name(), identity.exact_version(), entry))
}
- fn recover_with_filesystem(
+ fn recover_using(
directory: PathBuf,
+ runtime_directory: PathBuf,
+ protection: Arc<dyn ArtifactProtection>,
filesystem: &impl PublicationFilesystem,
) -> Result<Self, PluginStoreError> {
let mut store = Self {
directory,
+ runtime_directory,
+ protection,
programs: HashMap::new(),
};
// Collect each directory before deleting children, so a failed
@@ -166,20 +194,26 @@ impl PluginProgramStore {
};
for version in versions {
let target = version.path();
- let recovered = parse_version_entry(&version).and_then(|exact_version| {
- let installed = load_installed(&target)?;
- validate_identity(&installed, &program_name, &exact_version, &target)?;
- Ok((exact_version, installed))
- });
- match recovered {
- Ok((exact_version, installed)) => {
- store.programs.insert(
- PluginProgramIdentity::from_parts(program_name.clone(), exact_version),
- Arc::new(program_entry(target, installed)),
- );
- }
- Err(issue) => discard_invalid_entry(&target, &path, &issue, filesystem)?,
+ if version
+ .file_name()
+ .to_str()
+ .is_some_and(|name| name.starts_with(ORIGINAL_PREFIX))
+ {
+ let (exact_version, entry) = store.load_original(&target, &program_name)?;
+ store.programs.insert(
+ PluginProgramIdentity::from_parts(program_name.clone(), exact_version),
+ Arc::new(entry),
+ );
+ continue;
}
+ discard_invalid_entry(
+ &target,
+ &path,
+ &PluginStoreError::StoreIntegrityInvalid {
+ path: target.clone(),
+ },
+ filesystem,
+ )?;
}
}
Ok(store)
@@ -190,19 +224,12 @@ impl PluginProgramStore {
package: impl Read,
publication: &impl PublicationFilesystem,
) -> Result<PluginProgramInstallResult, PluginStoreError> {
- let package = upload::receive(package, &self.directory)?;
- let staged =
- stage_plugin_program_package(package, &self.directory).map_err(
- |source| match source {
- PluginPackageError::FilesystemOperationFailed { source } => {
- PluginStoreError::FilesystemOperationFailed {
- path: self.directory.clone(),
- source,
- }
- }
- source => PluginStoreError::PackageInvalid { source },
- },
- )?;
+ let mut package = package_input::receive(package, &self.directory)?;
+ let staged = package_input::open(
+ package.as_file_mut(),
+ &self.runtime_directory,
+ self.protection.as_ref(),
+ )?;
if !staged.platforms().contains(&Platform::CURRENT) {
let platforms = staged.platforms().into();
staged
@@ -217,7 +244,7 @@ impl PluginProgramStore {
let identity =
PluginProgramIdentity::from_parts(program_name.clone(), staged.exact_version().clone());
let parent = self.directory.join(program_name.as_str());
- let target = parent.join(staged.exact_version().as_str());
+ let target = parent.join(original_file_name(staged.exact_version()));
let namespace_presence = inspect_program_namespace(&parent)?;
if matches!(namespace_presence, ProgramNamespacePresence::Absent)
@@ -230,18 +257,22 @@ impl PluginProgramStore {
}
if let Some(entry) = self.programs.get(&identity) {
- let existing = load_installed(&target)?;
+ entry.original.verify()?;
+ let target = &entry.directory;
+ let existing = load_installed(target)?;
validate_identity(
&existing,
staged.program_name(),
staged.exact_version(),
- &target,
+ target,
)?;
if !existing.matches_snapshot(&entry.manifest_bytes, &entry.file_digests) {
- return Err(PluginStoreError::StoreIntegrityInvalid { path: target });
+ return Err(PluginStoreError::StoreIntegrityInvalid {
+ path: target.clone(),
+ });
}
let identical = staged
- .has_same_files_as(&existing, &target)
+ .has_same_files_as(&existing, target)
.map_err(|source| PluginStoreError::InstalledPackageInvalid {
path: target.clone(),
source,
@@ -260,11 +291,13 @@ impl PluginProgramStore {
require_absent(&target, publication)?;
self.prepare_program_directory(&parent, namespace_presence, publication)?;
- publication.sync_package_tree(staged.path())?;
- publication.rename(staged.path(), &target)?;
+ let original = StoredOriginal::from_file(target, package.as_file_mut())?;
+ publication.sync_file(package.path())?;
+ publication.rename(package.path(), &original.path)?;
publication.sync_directory(&parent)?;
-
- let entry = Arc::new(program_entry(target, staged.into_snapshot()));
+ let (directory, snapshot) = staged.into_runtime();
+ let entry = program_entry(directory, original, snapshot);
+ let entry = Arc::new(entry);
self.programs.insert(identity.clone(), entry);
Ok(PluginProgramInstallResult::Installed(identity))
}
@@ -285,23 +318,14 @@ impl PluginProgramStore {
}
let parent = self.directory.join(program_name.as_str());
- let target = parent.join(exact_version.as_str());
+ let target = entry.original.path.clone();
if matches!(
inspect_program_namespace(&parent)?,
ProgramNamespacePresence::Absent
) {
return Err(PluginStoreError::StoreIntegrityInvalid { path: parent });
}
- let metadata = publication.symlink_metadata(&target).map_err(|source| {
- PluginStoreError::FilesystemOperationFailed {
- path: target.clone(),
- source,
- }
- })?;
- if !metadata.is_dir() || !has_owner_only_directory_permission(&metadata) {
- return Err(PluginStoreError::StoreIntegrityInvalid { path: target });
- }
-
+ entry.original.verify()?;
let tombstone = deletion_tombstone_path(&parent, exact_version);
require_absent(&tombstone, publication)?;
publication.rename(&target, &tombstone)?;
@@ -311,6 +335,9 @@ impl PluginProgramStore {
removed.is_some(),
"Program must remain owned until uninstall commits"
);
+ if let Some(entry) = &removed {
+ publication.remove_entry(&entry.directory)?;
+ }
publication.remove_entry(&tombstone)?;
publication.sync_directory(&parent)?;
Ok(PluginUninstallOutcome::Uninstalled)
@@ -349,6 +376,7 @@ pub(crate) enum PluginUninstallOutcome {
/// One immutable validated Program shared by the Store and ready Plans.
pub(crate) struct PluginProgramEntry {
directory: PathBuf,
+ original: StoredOriginal,
manifest_bytes: Box<[u8]>,
display_name: String,
description: String,
@@ -363,7 +391,7 @@ pub(crate) struct PluginProgramEntry {
}
impl PluginProgramEntry {
- /// Returns the exact durable Program directory.
+ /// Returns the validated runtime directory for this Program.
#[must_use]
pub(crate) fn directory(&self) -> &Path {
&self.directory
@@ -439,6 +467,7 @@ impl std::fmt::Debug for PluginProgramEntry {
fn program_entry(
directory: PathBuf,
+ original: StoredOriginal,
snapshot: ValidatedPluginProgramSnapshot,
) -> PluginProgramEntry {
let ValidatedPluginProgramSnapshot {
@@ -457,6 +486,7 @@ fn program_entry(
} = validated.into_material();
PluginProgramEntry {
directory,
+ original,
manifest_bytes,
display_name,
description,
@@ -472,7 +502,7 @@ fn program_entry(
fn deletion_tombstone_path(parent: &Path, exact_version: &ExactVersion) -> PathBuf {
parent.join(format!(
"{DELETION_TOMBSTONE_PREFIX}{}",
- exact_version.as_str()
+ version_digest(exact_version)
))
}
@@ -573,6 +603,8 @@ pub(crate) mod test_support {
pub(crate) fn empty_store(directory: PathBuf) -> PluginProgramStore {
PluginProgramStore {
+ runtime_directory: directory.clone(),
+ protection: std::sync::Arc::new(crate::runner::extensions::ByPass),
directory,
programs: HashMap::new(),
}
diff --git a/src/runner/plugin/store/package_input.rs b/src/runner/plugin/store/package_input.rs
new file mode 100644
index 0000000..e1c7a40
--- /dev/null
+++ b/src/runner/plugin/store/package_input.rs
@@ -0,0 +1,302 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+//! Receives new packages and opens local packages for installation and recovery.
+//!
+//! Incoming stream failure rejects only the installation. Local file failures
+//! require Runner exit. The core records these failures so a hook cannot hide them.
+
+use super::{PluginPackageError, PluginStoreError};
+use crate::runner::extensions::ArtifactProtection;
+use crate::runner::plugin::package::{StagedPluginProgramPackage, stage_plugin_program_package};
+use std::io::{self, Read, Seek as _, Write};
+use std::path::Path;
+use tempfile::NamedTempFile;
+
+/// Receives the full input into a temporary file under the Store root.
+///
+/// # Errors
+///
+/// Input read failure is an invalid upload; local file I/O requires Runner exit.
+pub(super) fn receive(
+ mut input: impl Read,
+ directory: &Path,
+) -> Result<NamedTempFile, PluginStoreError> {
+ let filesystem_error = |source| PluginStoreError::FilesystemOperationFailed {
+ path: directory.to_path_buf(),
+ source,
+ };
+ let mut spool = tempfile::Builder::new()
+ .prefix(".tenon-plugin-upload-")
+ .tempfile_in(directory)
+ .map_err(filesystem_error)?;
+ receive_into(&mut input, &mut spool, directory)?;
+ spool.rewind().map_err(filesystem_error)?;
+ Ok(spool)
+}
+
+pub(super) fn open(
+ input: &mut dyn Read,
+ directory: &Path,
+ protection: &dyn ArtifactProtection,
+) -> Result<StagedPluginProgramPackage, PluginStoreError> {
+ let filesystem_error = |source| PluginStoreError::FilesystemOperationFailed {
+ path: directory.to_path_buf(),
+ source,
+ };
+ let mut input = LocalPackageReader {
+ input,
+ failure: None,
+ };
+ let result = stage_plugin_program_package(protection, &mut input, directory);
+ if let Some(source) = input.failure {
+ if let Ok(staged) = result {
+ staged.close().map_err(filesystem_error)?;
+ }
+ return Err(filesystem_error(source));
+ }
+ result.map_err(|source| match source {
+ PluginPackageError::AccessRejected => PluginStoreError::PackageAccessRejected,
+ PluginPackageError::FilesystemOperationFailed { source } => filesystem_error(source),
+ source => PluginStoreError::PackageInvalid { source },
+ })
+}
+
+struct LocalPackageReader<'a> {
+ input: &'a mut dyn Read,
+ failure: Option<io::Error>,
+}
+
+impl Read for LocalPackageReader<'_> {
+ fn read(&mut self, output: &mut [u8]) -> io::Result<usize> {
+ if let Some(source) = &self.failure {
+ return Err(io::Error::new(
+ source.kind(),
+ "Plugin package input has already failed",
+ ));
+ }
+ loop {
+ match self.input.read(output) {
+ Ok(count) => return Ok(count),
+ Err(source) if source.kind() == io::ErrorKind::Interrupted => continue,
+ Err(source) => {
+ let error = io::Error::new(source.kind(), "Plugin package input read failed");
+ self.failure = Some(source);
+ return Err(error);
+ }
+ }
+ }
+ }
+}
+
+fn receive_into(
+ input: &mut impl Read,
+ spool: &mut impl Write,
+ directory: &Path,
+) -> Result<(), PluginStoreError> {
+ let mut buffer = [0_u8; 64 * 1024];
+ loop {
+ let count = input
+ .read(&mut buffer)
+ .map_err(|source| PluginStoreError::PackageInvalid {
+ source: PluginPackageError::ArchiveInvalid { source },
+ })?;
+ if count == 0 {
+ return Ok(());
+ }
+ spool.write_all(&buffer[..count]).map_err(|source| {
+ PluginStoreError::FilesystemOperationFailed {
+ path: directory.to_path_buf(),
+ source,
+ }
+ })?;
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::runner::extensions::ByPass;
+ use crate::runner::plugin::package::tests::valid_source_program_package;
+ use std::io::Cursor;
+
+ #[test]
+ fn input_failure_after_a_complete_archive_still_rejects_the_upload() -> io::Result<()> {
+ let directory = tempfile::tempdir()?;
+ let input = Cursor::new(valid_source_program_package()?)
+ .chain(FailingIo(io::ErrorKind::ConnectionReset));
+ let error = receive(input, directory.path())
+ .err()
+ .ok_or_else(|| io::Error::other("Interrupted body was accepted"))?;
+ assert!(matches!(error, PluginStoreError::PackageInvalid {
+ source: PluginPackageError::ArchiveInvalid { source }
+ } if source.kind() == io::ErrorKind::ConnectionReset));
+ assert!(directory.path().read_dir()?.next().is_none());
+ Ok(())
+ }
+
+ #[test]
+ fn spool_write_failure_is_not_an_invalid_upload() -> io::Result<()> {
+ let directory = tempfile::tempdir()?;
+ let error = receive_into(
+ &mut Cursor::new(b"body"),
+ &mut FailingIo(io::ErrorKind::StorageFull),
+ directory.path(),
+ )
+ .err()
+ .ok_or_else(|| io::Error::other("Disk write failure was ignored"))?;
+ assert!(
+ matches!(error, PluginStoreError::FilesystemOperationFailed { path, source }
+ if path == directory.path() && source.kind() == io::ErrorKind::StorageFull)
+ );
+ Ok(())
+ }
+
+ #[test]
+ fn local_read_failures_survive_gzip_tar_and_trailer_decoding() -> io::Result<()> {
+ let package = valid_source_program_package()?;
+ for (offset, kind) in [
+ (0, io::ErrorKind::Other),
+ (10, io::ErrorKind::Other),
+ (package.len() / 2, io::ErrorKind::Other),
+ (package.len() - 4, io::ErrorKind::Other),
+ (package.len(), io::ErrorKind::Other),
+ (0, io::ErrorKind::FileTooLarge),
+ ] {
+ let directory = tempfile::tempdir()?;
+ let mut input = Cursor::new(&package[..offset]).chain(FailingIo(kind));
+ let error = open(&mut input, directory.path(), &ByPass)
+ .err()
+ .ok_or_else(|| io::Error::other("Local read failure was ignored"))?;
+ assert!(
+ matches!(error, PluginStoreError::FilesystemOperationFailed { source, .. }
+ if source.kind() == kind),
+ "offset {offset}, kind {kind}"
+ );
+ assert!(directory.path().read_dir()?.next().is_none());
+ }
+ Ok(())
+ }
+
+ #[test]
+ fn input_above_eight_gib_is_read_to_the_end() -> io::Result<()> {
+ let directory = tempfile::tempdir()?;
+ let mut input = io::repeat(0).take(8 * 1024 * 1024 * 1024 + 1);
+ receive_into(&mut input, &mut io::sink(), directory.path()).map_err(io::Error::other)?;
+ assert_eq!(input.limit(), 0);
+ Ok(())
+ }
+
+ #[test]
+ fn local_reader_failure_cannot_be_hidden_by_a_successful_hook() -> io::Result<()> {
+ struct IgnoreReadFailure(Vec<u8>);
+ impl ArtifactProtection for IgnoreReadFailure {
+ fn open_plugin_package(
+ &self,
+ source: &mut dyn Read,
+ output: &mut dyn crate::PluginPackageOutput,
+ ) -> io::Result<()> {
+ let _ = source.read(&mut [0; 1]);
+ ByPass.open_plugin_package(&mut self.0.as_slice(), output)
+ }
+ }
+ let directory = tempfile::tempdir()?;
+ let protection = IgnoreReadFailure(valid_source_program_package()?);
+ let result = open(
+ &mut FailingIo(io::ErrorKind::PermissionDenied),
+ directory.path(),
+ &protection,
+ );
+ assert!(
+ matches!(result, Err(PluginStoreError::FilesystemOperationFailed { source, .. }) if source.kind() == io::ErrorKind::PermissionDenied)
+ );
+ assert!(directory.path().read_dir()?.next().is_none());
+ Ok(())
+ }
+
+ #[test]
+ fn hook_error_kinds_select_package_failures_and_remove_partial_output() -> io::Result<()> {
+ struct Reject(io::ErrorKind);
+ impl ArtifactProtection for Reject {
+ fn open_plugin_package(
+ &self,
+ _source: &mut dyn Read,
+ output: &mut dyn crate::PluginPackageOutput,
+ ) -> io::Result<()> {
+ output.write_file("program", &mut io::Cursor::new(b"partial"))?;
+ Err(io::Error::new(self.0, "Fixture package access failed"))
+ }
+ }
+ let directory = tempfile::tempdir()?;
+ for (kind, code) in [
+ (io::ErrorKind::InvalidData, Some("plugin_package_invalid")),
+ (io::ErrorKind::UnexpectedEof, Some("plugin_package_invalid")),
+ (
+ io::ErrorKind::FileTooLarge,
+ Some("plugin_package_too_large"),
+ ),
+ (io::ErrorKind::PermissionDenied, None),
+ (io::ErrorKind::Other, None),
+ ] {
+ let error = open(&mut io::empty(), directory.path(), &Reject(kind))
+ .err()
+ .ok_or_else(|| io::Error::other("Rejected package was accepted"))?;
+ match (error, code) {
+ (PluginStoreError::PackageInvalid { source }, Some(code)) => {
+ assert_eq!(source.code(), code);
+ if matches!(
+ kind,
+ io::ErrorKind::InvalidData | io::ErrorKind::UnexpectedEof
+ ) {
+ assert_eq!(
+ std::error::Error::source(&source).map(ToString::to_string),
+ Some(String::from("Fixture package access failed")),
+ );
+ }
+ }
+ (PluginStoreError::PackageAccessRejected, None) => {}
+ (error, code) => {
+ return Err(io::Error::other(format!(
+ "Unexpected package failure: {error:?}; expected {code:?}"
+ )));
+ }
+ }
+ assert!(directory.path().read_dir()?.next().is_none());
+ }
+ Ok(())
+ }
+
+ struct FailingIo(io::ErrorKind);
+
+ impl Read for FailingIo {
+ fn read(&mut self, _output: &mut [u8]) -> io::Result<usize> {
+ Err(io::Error::new(self.0, "Injected read failure"))
+ }
+ }
+
+ impl Write for FailingIo {
+ fn write(&mut self, _input: &[u8]) -> io::Result<usize> {
+ Err(io::Error::new(self.0, "Injected write failure"))
+ }
+
+ fn flush(&mut self) -> io::Result<()> {
+ Ok(())
+ }
+ }
+}
diff --git a/src/runner/plugin/store/protected.rs b/src/runner/plugin/store/protected.rs
new file mode 100644
index 0000000..287dbc5
--- /dev/null
+++ b/src/runner/plugin/store/protected.rs
@@ -0,0 +1,163 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+//! Opens saved original packages and creates validated runtime files.
+
+use super::{
+ PluginProgramEntry, PluginProgramStore, PluginStoreError, package_input, program_entry,
+};
+use crate::identifiers::{ExactVersion, ProgramName};
+use sha2::{Digest as _, Sha256};
+use std::fs::{self, File};
+use std::io::{self, Read as _, Seek as _};
+use std::os::unix::fs::{MetadataExt as _, PermissionsExt as _};
+use std::path::{Path, PathBuf};
+
+pub(super) const ORIGINAL_PREFIX: &str = ".tenon-artifact-";
+
+pub(super) fn original_file_name(version: &ExactVersion) -> String {
+ format!("{ORIGINAL_PREFIX}{}", version_digest(version))
+}
+
+pub(super) fn version_digest(version: &ExactVersion) -> String {
+ const HEX: &[u8; 16] = b"0123456789abcdef";
+ let mut encoded = String::with_capacity(64);
+ for byte in Sha256::digest(version.as_str().as_bytes()) {
+ encoded.push(char::from(HEX[usize::from(byte >> 4)]));
+ encoded.push(char::from(HEX[usize::from(byte & 0x0f)]));
+ }
+ encoded
+}
+
+pub(super) struct StoredOriginal {
+ pub(super) path: PathBuf,
+ digest: [u8; 32],
+}
+
+impl StoredOriginal {
+ pub(super) fn from_file(path: PathBuf, file: &mut File) -> Result<Self, PluginStoreError> {
+ let digest =
+ file_digest(file).map_err(|source| PluginStoreError::FilesystemOperationFailed {
+ path: path.clone(),
+ source,
+ })?;
+ Ok(Self { path, digest })
+ }
+
+ pub(super) fn verify(&self) -> Result<(), PluginStoreError> {
+ let mut file = open_original(&self.path)?;
+ let current = Self::from_file(self.path.clone(), &mut file)?;
+ if current.digest != self.digest {
+ return Err(PluginStoreError::StoreIntegrityInvalid {
+ path: self.path.clone(),
+ });
+ }
+ Ok(())
+ }
+}
+
+impl PluginProgramStore {
+ pub(super) fn load_original(
+ &self,
+ path: &Path,
+ name: &ProgramName,
+ ) -> Result<(ExactVersion, PluginProgramEntry), PluginStoreError> {
+ let mut input = open_original(path)?;
+ let original = StoredOriginal::from_file(path.to_path_buf(), &mut input)?;
+ let staged = package_input::open(
+ &mut input,
+ &self.runtime_directory,
+ self.protection.as_ref(),
+ )
+ .map_err(|error| match error {
+ PluginStoreError::PackageInvalid { source } => {
+ PluginStoreError::InstalledPackageInvalid {
+ path: path.to_path_buf(),
+ source,
+ }
+ }
+ error => error,
+ })?;
+ if staged.program_name() != name
+ || path.file_name()
+ != Some(std::ffi::OsStr::new(&original_file_name(
+ staged.exact_version(),
+ )))
+ {
+ return Err(PluginStoreError::StoreIntegrityInvalid {
+ path: path.to_path_buf(),
+ });
+ }
+ let version = staged.exact_version().clone();
+ let (directory, snapshot) = staged.into_runtime();
+ let entry = program_entry(directory, original, snapshot);
+ Ok((version, entry))
+ }
+}
+
+fn open_original(path: &Path) -> Result<File, PluginStoreError> {
+ let metadata = fs::symlink_metadata(path).map_err(|source| {
+ PluginStoreError::FilesystemOperationFailed {
+ path: path.to_path_buf(),
+ source,
+ }
+ })?;
+ if !metadata.is_file()
+ || metadata.permissions().mode() & 0o777 != 0o600
+ || metadata.nlink() != 1
+ {
+ return Err(PluginStoreError::StoreIntegrityInvalid {
+ path: path.to_path_buf(),
+ });
+ }
+ File::open(path).map_err(|source| PluginStoreError::FilesystemOperationFailed {
+ path: path.to_path_buf(),
+ source,
+ })
+}
+
+fn file_digest(file: &mut File) -> io::Result<[u8; 32]> {
+ file.rewind()?;
+ let mut digest = Sha256::new();
+ let mut buffer = [0_u8; 64 * 1024];
+ loop {
+ let count = file.read(&mut buffer)?;
+ if count == 0 {
+ break;
+ }
+ digest.update(&buffer[..count]);
+ }
+ file.rewind()?;
+ Ok(digest.finalize().into())
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn saved_original_above_eight_gib_can_be_opened() -> io::Result<()> {
+ let original = tempfile::NamedTempFile::new()?;
+ let length = 8 * 1024 * 1024 * 1024 + 1;
+ original.as_file().set_len(length)?;
+ let file = open_original(original.path()).map_err(io::Error::other)?;
+ assert_eq!(file.metadata()?.len(), length);
+ Ok(())
+ }
+}
diff --git a/src/runner/plugin/store/tests.rs b/src/runner/plugin/store/tests.rs
index 51ab2b0..0183269 100644
--- a/src/runner/plugin/store/tests.rs
+++ b/src/runner/plugin/store/tests.rs
@@ -18,7 +18,7 @@
*/
use super::PluginProgramInstallResult;
-use super::filesystem::{DurablePublicationFilesystem, sync_directory, sync_package_tree};
+use super::filesystem::{DurablePublicationFilesystem, sync_directory};
use super::test_support::program_count;
use super::{
PluginProgramEntry, PluginProgramStore, PluginStoreError, PluginUninstallOutcome,
@@ -44,8 +44,9 @@ fn first_install_publishes_disk_and_memory_in_one_store_mutation() -> io::Result
let root = prepare_store_root(parent.path())?;
let mut store = recover_store(root.clone())?;
+ let package = valid_source_program_package()?;
let outcome = store
- .install(Cursor::new(valid_source_program_package()?))
+ .install(Cursor::new(&package))
.map_err(io::Error::other)?;
let PluginProgramInstallResult::Installed(identity) = outcome else {
return Err(io::Error::other("first install was reported as unchanged"));
@@ -56,7 +57,9 @@ fn first_install_publishes_disk_and_memory_in_one_store_mutation() -> io::Result
);
let entry = available_entry(&store, "com.example.source")?;
- assert_eq!(entry.directory(), root.join("com.example.source/1.0.0"));
+ assert!(entry.directory().starts_with(&store.runtime_directory));
+ assert!(!entry.directory().starts_with(&root));
+ assert_eq!(fs::read(&entry.original.path)?, package);
assert_eq!(fs::read(entry.directory().join("bin/start"))?, b"program");
#[cfg(unix)]
{
@@ -138,8 +141,12 @@ fn same_identity_with_different_content_is_never_overwritten() -> io::Result<()>
assert_eq!(error.code(), "plugin_version_conflict");
assert!(
- fs::read_to_string(root.join("com.example.source/1.0.0/manifest.json"))?
- .contains("./bin/start")
+ fs::read_to_string(
+ available_entry(&store, "com.example.source")?
+ .directory()
+ .join("manifest.json")
+ )?
+ .contains("./bin/start")
);
assert_available(&store, "com.example.source")?;
Ok(())
@@ -160,7 +167,9 @@ fn recovery_owns_all_interfaces_in_one_identity_map() -> io::Result<()> {
.map_err(io::Error::other)?;
}
+ let old_runtime = store.runtime_directory.clone();
drop(store);
+ fs::remove_dir_all(old_runtime)?;
let recovered = recover_store(root)?;
assert_eq!(program_count(&recovered), 3);
@@ -189,7 +198,7 @@ fn recovery_owns_all_interfaces_in_one_identity_map() -> io::Result<()> {
}
#[test]
-fn recovery_deletes_invalid_versions_and_keeps_valid_siblings() -> io::Result<()> {
+fn recovery_preserves_invalid_original_packages_and_valid_siblings() -> io::Result<()> {
let parent = tempfile::tempdir()?;
let root = prepare_store_root(parent.path())?;
let mut store = recover_store(root.clone())?;
@@ -198,22 +207,20 @@ fn recovery_deletes_invalid_versions_and_keeps_valid_siblings() -> io::Result<()
.install(Cursor::new(valid_program_package(interface)?))
.map_err(io::Error::other)?;
}
+ let invalid = available_entry(&store, "com.example.sink")?
+ .original
+ .path
+ .clone();
+ let valid = available_entry(&store, "com.example.source")?
+ .original
+ .path
+ .clone();
+ let valid_bytes = fs::read(&valid)?;
drop(store);
- let invalid = root.join("com.example.sink/1.0.0/config.schema.json");
- make_writable(&invalid)?;
- fs::write(&invalid, b"{}")?;
- make_private_file(&invalid)?;
- let invalid_sibling = root.join("com.example.source/2.0.0");
- fs::create_dir(&invalid_sibling)?;
- make_private_directory(&invalid_sibling)?;
-
- let recovered = recover_store(root.clone())?;
-
- assert_available(&recovered, "com.example.source")?;
- assert!(lookup(&recovered, "com.example.sink")?.is_none());
- assert!(!root.join("com.example.sink/1.0.0").exists());
- assert!(!invalid_sibling.exists());
- assert_eq!(program_count(&recovered), 1);
+ fs::write(&invalid, b"invalid package")?;
+ assert!(recover_store(root).is_err());
+ assert_eq!(fs::read(invalid)?, b"invalid package");
+ assert_eq!(fs::read(valid)?, valid_bytes);
Ok(())
}
@@ -234,60 +241,59 @@ fn healthy_namespace_absence_is_missing_without_a_saved_state() -> io::Result<()
#[cfg(unix)]
#[test]
-fn recovery_rejects_identity_permission_and_link_drift_inside_a_package() -> io::Result<()> {
+fn recovery_preserves_original_packages_with_identity_permission_or_link_errors() -> io::Result<()>
+{
use std::os::unix::fs::{PermissionsExt as _, symlink};
-
enum Corruption {
- DirectoryIdentity,
- FilePermission,
+ Identity,
+ Permission,
SymbolicLink,
HardLink,
}
for corruption in [
- Corruption::DirectoryIdentity,
- Corruption::FilePermission,
+ Corruption::Identity,
+ Corruption::Permission,
Corruption::SymbolicLink,
Corruption::HardLink,
] {
let parent = tempfile::tempdir()?;
let root = prepare_store_root(parent.path())?;
let mut store = recover_store(root.clone())?;
- for interface in [PluginInterface::Source, PluginInterface::Sink] {
- store
- .install(Cursor::new(valid_program_package(interface)?))
- .map_err(io::Error::other)?;
- }
+ store
+ .install(Cursor::new(valid_source_program_package()?))
+ .map_err(io::Error::other)?;
+ let saved = available_entry(&store, "com.example.source")?
+ .original
+ .path
+ .clone();
drop(store);
- let version = root.join("com.example.source/1.0.0");
- let program = version.join("bin/start");
- let outside = parent.path().join("outside-program");
+ let outside = parent.path().join("outside-package");
fs::write(&outside, b"external-owner")?;
make_private_file(&outside)?;
- let invalid_version = match corruption {
- Corruption::DirectoryIdentity => {
- let destination = root.join("com.example.source/2.0.0");
- fs::rename(&version, &destination)?;
+ let invalid = match corruption {
+ Corruption::Identity => {
+ let destination =
+ saved.with_file_name(format!(".tenon-artifact-{}", "0".repeat(64)));
+ fs::rename(&saved, &destination)?;
destination
}
- Corruption::FilePermission => {
- fs::set_permissions(&program, fs::Permissions::from_mode(0o644))?;
- version
+ Corruption::Permission => {
+ fs::set_permissions(&saved, fs::Permissions::from_mode(0o644))?;
+ saved
}
Corruption::SymbolicLink => {
- fs::remove_file(&program)?;
- symlink(&outside, &program)?;
- version
+ fs::remove_file(&saved)?;
+ symlink(&outside, &saved)?;
+ saved
}
Corruption::HardLink => {
- fs::remove_file(&program)?;
- fs::hard_link(&outside, &program)?;
- version
+ fs::remove_file(&saved)?;
+ fs::hard_link(&outside, &saved)?;
+ saved
}
};
- let recovered = recover_store(root)?;
- assert_eq!(program_count(&recovered), 1);
- assert_available(&recovered, "com.example.sink")?;
- assert!(!invalid_version.exists());
+ assert!(recover_store(root).is_err());
+ assert!(fs::symlink_metadata(invalid).is_ok());
assert_eq!(fs::read(&outside)?, b"external-owner");
assert_eq!(fs::metadata(&outside)?.permissions().mode() & 0o777, 0o500);
}
@@ -385,7 +391,7 @@ fn install_failures_before_rename_recover_without_publishing_an_entry() -> io::R
for fail_at in [
PublicationPoint::SyncProgramDirectory,
PublicationPoint::SyncStoreRoot,
- PublicationPoint::SyncPackageTree,
+ PublicationPoint::SyncPackageFile,
PublicationPoint::Rename,
] {
let parent = tempfile::tempdir()?;
@@ -401,8 +407,10 @@ fn install_failures_before_rename_recover_without_publishing_an_entry() -> io::R
assert_eq!(error.code(), "plugin_store_internal_error");
assert!(std::error::Error::source(&error).is_some());
+ assert_eq!(program_count(&store), 0);
+ assert_eq!(fs::read_dir(&store.runtime_directory)?.count(), 0);
drop(store);
- assert!(!root.join("com.example.source/1.0.0").exists());
+ assert!(!source_package_path(&root)?.exists());
let mut recovered = recover_store(root)?;
assert!(lookup(&recovered, "com.example.source")?.is_none());
let retry = recovered
@@ -428,6 +436,9 @@ fn install_failure_after_rename_recovers_the_published_program() -> io::Result<(
assert_eq!(error.code(), "plugin_store_internal_error");
assert!(std::error::Error::source(&error).is_some());
+ assert_eq!(program_count(&store), 0);
+ assert_eq!(fs::read_dir(&store.runtime_directory)?.count(), 0);
+ assert_eq!(fs::read(source_package_path(&root)?)?, package);
drop(store);
let mut recovered = recover_store(root)?;
assert_available(&recovered, "com.example.source")?;
@@ -450,14 +461,14 @@ fn install_reports_external_deletion_and_retains_entry_ownership_until_shutdown(
.map_err(io::Error::other)?;
let entry = Arc::clone(available_entry(&store, "com.example.source")?);
let weak = Arc::downgrade(&entry);
- fs::remove_dir_all(root.join("com.example.source/1.0.0"))?;
+ fs::remove_file(source_package_path(&root)?)?;
let error = store
.install(Cursor::new(package))
.err()
.ok_or_else(|| io::Error::other("external deletion was silently repaired"))?;
- assert_eq!(error.code(), "plugin_store_integrity_invalid");
+ assert_eq!(error.code(), "plugin_store_internal_error");
drop(entry);
assert!(weak.upgrade().is_some());
drop(store);
@@ -473,7 +484,9 @@ fn install_reports_external_byte_drift_before_conflict_comparison() -> io::Resul
store
.install(Cursor::new(valid_source_program_package()?))
.map_err(io::Error::other)?;
- let resource = root.join("com.example.source/1.0.0/resources/data.bin");
+ let resource = available_entry(&store, "com.example.source")?
+ .directory()
+ .join("resources/data.bin");
make_writable(&resource)?;
fs::write(&resource, b"beta")?;
make_private_file(&resource)?;
@@ -508,7 +521,7 @@ fn uninstall_removes_disk_and_memory_in_one_store_mutation() -> io::Result<()> {
assert_eq!(outcome, PluginUninstallOutcome::Uninstalled);
assert!(store.lookup(&program_name, &exact_version).is_none());
- assert!(!root.join("com.example.source/1.0.0").exists());
+ assert!(!source_package_path(&root)?.exists());
assert!(!deletion_tombstone(&root).exists());
Ok(())
}
@@ -531,7 +544,7 @@ fn uninstall_rejects_an_entry_with_an_external_reference() -> io::Result<()> {
.ok_or_else(|| io::Error::other("referenced Program was uninstalled"))?;
assert_eq!(error.code(), "plugin_in_use");
- assert!(root.join("com.example.source/1.0.0").is_dir());
+ assert!(source_package_path(&root)?.is_file());
assert!(matches!(
store.lookup(&program_name, &exact_version),
Some(current) if Arc::ptr_eq(current, &entry)
@@ -557,7 +570,7 @@ fn uninstall_rejects_a_weak_reference_until_it_is_released() -> io::Result<()> {
.ok_or_else(|| io::Error::other("weakly referenced Program was uninstalled"))?;
assert_eq!(error.code(), "plugin_in_use");
- assert!(root.join("com.example.source/1.0.0").is_dir());
+ assert!(source_package_path(&root)?.is_file());
drop(weak);
assert_eq!(
store
@@ -576,7 +589,7 @@ fn uninstall_reports_an_exact_path_removed_outside_the_store() -> io::Result<()>
store
.install(Cursor::new(valid_source_program_package()?))
.map_err(io::Error::other)?;
- fs::remove_dir_all(root.join("com.example.source/1.0.0"))?;
+ fs::remove_file(source_package_path(&root)?)?;
let error = store
.uninstall(&source_program_name()?, &exact_version()?)
@@ -608,7 +621,7 @@ fn uninstall_of_a_missing_identity_is_idempotent() -> io::Result<()> {
#[test]
fn uninstall_failures_before_rename_recover_the_existing_program() -> io::Result<()> {
- for fail_at in [UninstallPoint::InspectTarget, UninstallPoint::Rename] {
+ for fail_at in [UninstallPoint::InspectTombstone, UninstallPoint::Rename] {
let parent = tempfile::tempdir()?;
let root = prepare_store_root(parent.path())?;
let mut store = recover_store(root.clone())?;
@@ -635,7 +648,7 @@ fn uninstall_failures_before_rename_recover_the_existing_program() -> io::Result
.uninstall(&source_program_name()?, &exact_version()?)
.map_err(io::Error::other)?;
assert_eq!(outcome, PluginUninstallOutcome::Uninstalled);
- assert!(!root.join("com.example.source/1.0.0").exists());
+ assert!(!source_package_path(&root)?.exists());
}
Ok(())
}
@@ -644,6 +657,7 @@ fn uninstall_failures_before_rename_recover_the_existing_program() -> io::Result
fn uninstall_failures_after_rename_recover_as_missing() -> io::Result<()> {
for fail_at in [
UninstallPoint::SyncParent,
+ UninstallPoint::RemoveRuntime,
UninstallPoint::RemoveTombstone,
UninstallPoint::SyncCleanup,
] {
@@ -666,7 +680,7 @@ fn uninstall_failures_after_rename_recover_as_missing() -> io::Result<()> {
assert_eq!(error.code(), "plugin_store_internal_error");
assert!(std::error::Error::source(&error).is_some());
- assert!(!root.join("com.example.source/1.0.0").exists());
+ assert!(!source_package_path(&root)?.exists());
assert_eq!(
deletion_tombstone(&root).exists(),
fail_at != UninstallPoint::SyncCleanup
@@ -760,8 +774,7 @@ fn incomplete_namespace_enumeration_deletes_the_whole_namespace() -> io::Result<
path: namespace.clone(),
};
- let recovered =
- PluginProgramStore::recover_with_filesystem(root, &filesystem).map_err(io::Error::other)?;
+ let recovered = recover_with_filesystem(root, &filesystem).map_err(io::Error::other)?;
assert!(!namespace.exists());
assert!(lookup(&recovered, "com.example.source")?.is_none());
@@ -781,7 +794,7 @@ fn incomplete_root_enumeration_aborts_without_deleting_the_root() -> io::Result<
path: root.clone(),
};
- let error = PluginProgramStore::recover_with_filesystem(root.clone(), &filesystem)
+ let error = recover_with_filesystem(root.clone(), &filesystem)
.err()
.ok_or_else(|| io::Error::other("incomplete root enumeration was accepted"))?;
@@ -817,7 +830,7 @@ fn cleanup_and_parent_sync_failures_abort_recovery_and_can_be_retried_at_startup
},
};
- let error = PluginProgramStore::recover_with_filesystem(root.clone(), &filesystem)
+ let error = recover_with_filesystem(root.clone(), &filesystem)
.err()
.ok_or_else(|| io::Error::other("cleanup failure was ignored"))?;
@@ -837,9 +850,13 @@ fn missing_root_is_fatal_and_is_not_created() -> io::Result<()> {
let parent = tempfile::tempdir()?;
let root = parent.path().join("plugins/programs");
- let error = PluginProgramStore::recover(root.clone())
- .err()
- .ok_or_else(|| io::Error::other("missing Program Store was treated as empty"))?;
+ let error = PluginProgramStore::recover(
+ root.clone(),
+ root.clone(),
+ Arc::new(crate::runner::extensions::ByPass),
+ )
+ .err()
+ .ok_or_else(|| io::Error::other("missing Program Store was treated as empty"))?;
assert_eq!(error.code(), "plugin_store_internal_error");
assert!(!root.exists());
@@ -960,8 +977,37 @@ fn java_target_bundles_install_idempotently_and_recover() -> io::Result<()> {
Ok(())
}
+fn recover_with_filesystem(
+ directory: PathBuf,
+ filesystem: &impl PublicationFilesystem,
+) -> Result<PluginProgramStore, PluginStoreError> {
+ let parent = directory
+ .parent()
+ .ok_or_else(|| PluginStoreError::StoreIntegrityInvalid {
+ path: directory.clone(),
+ })?;
+ let runtime = tempfile::tempdir_in(parent)
+ .map_err(|source| PluginStoreError::FilesystemOperationFailed {
+ path: directory.clone(),
+ source,
+ })?
+ .keep();
+ PluginProgramStore::recover_using(
+ directory,
+ runtime,
+ Arc::new(crate::runner::extensions::ByPass),
+ filesystem,
+ )
+}
+
fn recover_store(root: PathBuf) -> io::Result<PluginProgramStore> {
- PluginProgramStore::recover(root).map_err(io::Error::other)
+ recover_with_filesystem(root, &DurablePublicationFilesystem).map_err(io::Error::other)
+}
+
+fn source_package_path(root: &Path) -> io::Result<PathBuf> {
+ Ok(root
+ .join("com.example.source")
+ .join(super::original_file_name(&exact_version()?)))
}
fn lookup<'a>(
@@ -982,7 +1028,7 @@ fn exact_version() -> io::Result<ExactVersion> {
}
fn deletion_tombstone(root: &Path) -> PathBuf {
- root.join("com.example.source/.tenon-plugin-delete-1.0.0")
+ root.join("com.example.source/.tenon-plugin-delete-92521fc3cbd964bdc9f584a991b89fddaa5754ed1cc96d6d42445338669c1305")
}
fn available_entry<'a>(
@@ -1049,7 +1095,7 @@ fn make_private_file(_path: &Path) -> io::Result<()> {
enum PublicationPoint {
SyncProgramDirectory,
SyncStoreRoot,
- SyncPackageTree,
+ SyncPackageFile,
Rename,
SyncPublishedDirectory,
}
@@ -1090,11 +1136,10 @@ impl FaultingPublicationFilesystem {
}
impl PublicationFilesystem for FaultingPublicationFilesystem {
- fn sync_package_tree(&self, root: &Path) -> Result<(), PluginStoreError> {
- self.record(PublicationPoint::SyncPackageTree, root)?;
- sync_package_tree(root)
+ fn sync_file(&self, path: &Path) -> Result<(), PluginStoreError> {
+ self.record(PublicationPoint::SyncPackageFile, path)?;
+ DurablePublicationFilesystem.sync_file(path)
}
-
fn rename(&self, source: &Path, target: &Path) -> Result<(), PluginStoreError> {
self.record(PublicationPoint::Rename, target)?;
fs::rename(source, target).map_err(|source| PluginStoreError::FilesystemOperationFailed {
@@ -1111,9 +1156,10 @@ impl PublicationFilesystem for FaultingPublicationFilesystem {
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum UninstallPoint {
- InspectTarget,
+ InspectTombstone,
Rename,
SyncParent,
+ RemoveRuntime,
RemoveTombstone,
SyncCleanup,
}
@@ -1133,10 +1179,6 @@ impl FaultingUninstallFilesystem {
}
impl PublicationFilesystem for FaultingUninstallFilesystem {
- fn sync_package_tree(&self, root: &Path) -> Result<(), PluginStoreError> {
- sync_package_tree(root)
- }
-
fn rename(&self, source: &Path, target: &Path) -> Result<(), PluginStoreError> {
if self.fail_at == UninstallPoint::Rename {
return Err(PluginStoreError::FilesystemOperationFailed {
@@ -1151,7 +1193,7 @@ impl PublicationFilesystem for FaultingUninstallFilesystem {
}
fn symlink_metadata(&self, path: &Path) -> io::Result<fs::Metadata> {
- if self.fail_at == UninstallPoint::InspectTarget {
+ if self.fail_at == UninstallPoint::InspectTombstone {
return Err(io::Error::other("Injected uninstall metadata failure"));
}
fs::symlink_metadata(path)
@@ -1172,7 +1214,13 @@ impl PublicationFilesystem for FaultingUninstallFilesystem {
}
fn remove_entry(&self, path: &Path) -> Result<(), PluginStoreError> {
- if self.fail_at == UninstallPoint::RemoveTombstone {
+ let is_tombstone = path
+ .file_name()
+ .and_then(|name| name.to_str())
+ .is_some_and(|name| name.starts_with(super::DELETION_TOMBSTONE_PREFIX));
+ if (self.fail_at == UninstallPoint::RemoveTombstone && is_tombstone)
+ || (self.fail_at == UninstallPoint::RemoveRuntime && !is_tombstone)
+ {
return Err(PluginStoreError::FilesystemOperationFailed {
path: path.to_path_buf(),
source: io::Error::other("Injected tombstone cleanup failure"),
@@ -1232,10 +1280,6 @@ impl PublicationFilesystem for FaultingRecoveryFilesystem {
sync_directory(path)
}
- fn sync_package_tree(&self, root: &Path) -> Result<(), PluginStoreError> {
- sync_package_tree(root)
- }
-
fn rename(&self, source: &Path, target: &Path) -> Result<(), PluginStoreError> {
DurablePublicationFilesystem.rename(source, target)
}
@@ -1318,3 +1362,5 @@ fn installation_does_not_execute_or_probe_the_declared_command() -> io::Result<(
assert!(!parent.path().join("marker-command.executed").exists());
Ok(())
}
+
+mod protected;
diff --git a/src/runner/plugin/store/tests/protected.rs b/src/runner/plugin/store/tests/protected.rs
new file mode 100644
index 0000000..6c3042c
--- /dev/null
+++ b/src/runner/plugin/store/tests/protected.rs
@@ -0,0 +1,181 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use super::*;
+use crate::runner::extensions::{ArtifactProtection, ByPass, PluginPackageOutput};
+use std::io::Read;
+
+struct RejectAfterWrite;
+
+impl ArtifactProtection for RejectAfterWrite {
+ fn open_plugin_package(
+ &self,
+ source: &mut dyn Read,
+ output: &mut dyn PluginPackageOutput,
+ ) -> io::Result<()> {
+ ByPass.open_plugin_package(source, output)?;
+ Err(io::Error::from(io::ErrorKind::PermissionDenied))
+ }
+}
+
+#[test]
+fn access_failure_discards_output_and_saved_damage_stops_recovery() -> io::Result<()> {
+ let root = tempfile::tempdir()?;
+ let store_dir = prepare_store_root(root.path())?;
+ let runtime = tempfile::tempdir()?;
+ let mut store = PluginProgramStore::recover(
+ store_dir.clone(),
+ runtime.path().to_path_buf(),
+ Arc::new(RejectAfterWrite),
+ )
+ .map_err(io::Error::other)?;
+ let bytes = valid_source_program_package()?;
+ assert!(matches!(
+ store.install(Cursor::new(&bytes)),
+ Err(PluginStoreError::PackageAccessRejected)
+ ));
+ assert_eq!(store_dir.read_dir()?.count(), 0);
+ assert_eq!(runtime.path().read_dir()?.count(), 0);
+ store.protection = Arc::new(crate::runner::extensions::ByPass);
+ store
+ .install(Cursor::new(&bytes))
+ .map_err(io::Error::other)?;
+ let path = store_dir.join("com.example.source/.tenon-artifact-92521fc3cbd964bdc9f584a991b89fddaa5754ed1cc96d6d42445338669c1305");
+ fs::write(&path, b"damaged")?;
+ assert!(matches!(
+ store.install(Cursor::new(bytes)),
+ Err(PluginStoreError::StoreIntegrityInvalid { .. })
+ ));
+ let protection = Arc::clone(&store.protection);
+ drop(store);
+ let recovery = PluginProgramStore::recover(store_dir, runtime.path().to_path_buf(), protection);
+ assert!(matches!(
+ recovery,
+ Err(PluginStoreError::InstalledPackageInvalid { .. })
+ ));
+ assert_eq!(fs::read(path)?, b"damaged");
+ Ok(())
+}
+
+#[test]
+fn long_versions_install_recover_and_uninstall_original_packages() -> io::Result<()> {
+ let version =
+ ExactVersion::try_from(format!("1.0.0-{}", "a".repeat(240))).map_err(io::Error::other)?;
+ let bytes = source_package_version(&version)?;
+ let root = tempfile::tempdir()?;
+ let store_dir = prepare_store_root(root.path())?;
+ let runtime = tempfile::tempdir()?;
+ let protection: Arc<dyn ArtifactProtection> = Arc::new(crate::runner::extensions::ByPass);
+ let mut store = PluginProgramStore::recover(
+ store_dir.clone(),
+ runtime.path().to_path_buf(),
+ Arc::clone(&protection),
+ )
+ .map_err(io::Error::other)?;
+ assert!(matches!(
+ store
+ .install(Cursor::new(&bytes))
+ .map_err(io::Error::other)?,
+ PluginProgramInstallResult::Installed(_)
+ ));
+ assert!(matches!(
+ store
+ .install(Cursor::new(&bytes))
+ .map_err(io::Error::other)?,
+ PluginProgramInstallResult::Unchanged(_)
+ ));
+ drop(store);
+ drop(runtime);
+
+ let runtime = tempfile::tempdir()?;
+ let mut store =
+ PluginProgramStore::recover(store_dir.clone(), runtime.path().to_path_buf(), protection)
+ .map_err(io::Error::other)?;
+ let name = source_program_name()?;
+ let entry = store
+ .lookup(&name, &version)
+ .ok_or_else(|| io::Error::other("Program is missing after recovery"))?;
+ let material = entry.directory().to_path_buf();
+ assert_eq!(fs::read(material.join("bin/start"))?, b"program");
+ assert_eq!(
+ store.uninstall(&name, &version).map_err(io::Error::other)?,
+ PluginUninstallOutcome::Uninstalled
+ );
+ assert!(store.lookup(&name, &version).is_none());
+ assert!(!material.exists());
+ assert_eq!(store_dir.join(name.as_str()).read_dir()?.count(), 0);
+ Ok(())
+}
+
+#[test]
+fn original_filename_must_match_the_validated_version() -> io::Result<()> {
+ let root = tempfile::tempdir()?;
+ let store_dir = prepare_store_root(root.path())?;
+ let runtime = tempfile::tempdir()?;
+ let protection: Arc<dyn ArtifactProtection> = Arc::new(crate::runner::extensions::ByPass);
+ let mut store = PluginProgramStore::recover(
+ store_dir.clone(),
+ runtime.path().to_path_buf(),
+ Arc::clone(&protection),
+ )
+ .map_err(io::Error::other)?;
+ let bytes = valid_source_program_package()?;
+ store
+ .install(Cursor::new(&bytes))
+ .map_err(io::Error::other)?;
+ drop(store);
+ let namespace = store_dir.join("com.example.source");
+ let saved = namespace
+ .read_dir()?
+ .next()
+ .ok_or_else(|| io::Error::other("Saved package is missing"))??
+ .path();
+ let wrong_path = namespace.join(format!(".tenon-artifact-{}", "0".repeat(64)));
+ fs::rename(saved, &wrong_path)?;
+ assert!(matches!(
+ PluginProgramStore::recover(store_dir, runtime.path().to_path_buf(), protection),
+ Err(PluginStoreError::StoreIntegrityInvalid { path }) if path == wrong_path
+ ));
+ assert_eq!(fs::read(wrong_path)?, bytes);
+ Ok(())
+}
+
+fn source_package_version(version: &ExactVersion) -> io::Result<Vec<u8>> {
+ use crate::runner::plugin::package::tests::{
+ archive, valid_config_schema, valid_program_descriptor,
+ };
+ let manifest = serde_json::to_vec(&serde_json::json!({
+ "programName": "com.example.source",
+ "exactVersion": version.as_str(),
+ "interface": "source",
+ "platforms": [crate::runner::plugin::platform::Platform::CURRENT],
+ "displayName": "Example Plugin",
+ "description": "Read example records.",
+ "command": ["./bin/start"]
+ }))?;
+ archive(vec![
+ ("manifest.json", manifest),
+ ("config.schema.json", valid_config_schema()),
+ (
+ "payload.descriptor.pb",
+ valid_program_descriptor(PluginInterface::Source)?,
+ ),
+ ("bin/start", b"program".to_vec()),
+ ])
+}
diff --git a/src/runner/plugin/store/upload.rs b/src/runner/plugin/store/upload.rs
deleted file mode 100644
index 46771ed..0000000
--- a/src/runner/plugin/store/upload.rs
+++ /dev/null
@@ -1,171 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * https://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-//! Owns the complete compressed upload before the Store begins decoding it.
-//!
-//! Input failure rejects only the upload. Local spool I/O is tagged at its
-//! source so gzip/tar decoding cannot disguise disk failure as an invalid body.
-
-use super::{PluginPackageError, PluginStoreError};
-use std::fs::File;
-use std::io::{self, Read, Seek as _, Write};
-use std::path::Path;
-
-/// Receives the complete input into an unnamed file under the Store root.
-///
-/// # Errors
-///
-/// Input read failure is an invalid upload; local file I/O requires Runner exit.
-pub(super) fn receive(
- mut input: impl Read,
- directory: &Path,
-) -> Result<LocalArchiveReader<File>, PluginStoreError> {
- let filesystem_error = |source| PluginStoreError::FilesystemOperationFailed {
- path: directory.to_path_buf(),
- source,
- };
- let mut spool = tempfile::tempfile_in(directory).map_err(filesystem_error)?;
- receive_into(&mut input, &mut spool, directory)?;
- spool.rewind().map_err(filesystem_error)?;
- Ok(LocalArchiveReader(spool))
-}
-
-/// Preserves local read provenance through the shared streaming decoder.
-#[derive(Debug)]
-pub(super) struct LocalArchiveReader<R>(R);
-
-impl<R: Read> Read for LocalArchiveReader<R> {
- fn read(&mut self, output: &mut [u8]) -> io::Result<usize> {
- self.0.read(output).map_err(|source| {
- io::Error::new(
- source.kind(),
- PluginPackageError::FilesystemOperationFailed { source },
- )
- })
- }
-}
-
-fn receive_into(
- input: &mut impl Read,
- spool: &mut impl Write,
- directory: &Path,
-) -> Result<(), PluginStoreError> {
- let mut buffer = [0_u8; 64 * 1024];
- loop {
- let count = input
- .read(&mut buffer)
- .map_err(|source| PluginStoreError::PackageInvalid {
- source: PluginPackageError::ArchiveInvalid { source },
- })?;
- if count == 0 {
- return Ok(());
- }
- spool.write_all(&buffer[..count]).map_err(|source| {
- PluginStoreError::FilesystemOperationFailed {
- path: directory.to_path_buf(),
- source,
- }
- })?;
- }
-}
-
-#[cfg(test)]
-mod tests {
- use super::*;
- use crate::runner::plugin::package::{
- stage_plugin_program_package, tests::valid_source_program_package,
- };
- use std::io::Cursor;
-
- #[test]
- fn input_failure_after_a_complete_archive_still_rejects_the_upload() -> io::Result<()> {
- let directory = tempfile::tempdir()?;
- let input = Cursor::new(valid_source_program_package()?)
- .chain(FailingIo(io::ErrorKind::ConnectionReset));
- let error = receive(input, directory.path())
- .err()
- .ok_or_else(|| io::Error::other("Interrupted body was accepted"))?;
- assert!(matches!(error, PluginStoreError::PackageInvalid {
- source: PluginPackageError::ArchiveInvalid { source }
- } if source.kind() == io::ErrorKind::ConnectionReset));
- assert!(directory.path().read_dir()?.next().is_none());
- Ok(())
- }
-
- #[test]
- fn spool_write_failure_is_not_an_invalid_upload() -> io::Result<()> {
- let directory = tempfile::tempdir()?;
- let error = receive_into(
- &mut Cursor::new(b"body"),
- &mut FailingIo(io::ErrorKind::StorageFull),
- directory.path(),
- )
- .err()
- .ok_or_else(|| io::Error::other("Disk write failure was ignored"))?;
- assert!(
- matches!(error, PluginStoreError::FilesystemOperationFailed { path, source }
- if path == directory.path() && source.kind() == io::ErrorKind::StorageFull)
- );
- Ok(())
- }
-
- #[test]
- fn local_read_failures_survive_gzip_tar_and_trailer_decoding() -> io::Result<()> {
- let package = valid_source_program_package()?;
- for (offset, kind) in [
- (0, io::ErrorKind::Other),
- (10, io::ErrorKind::Other),
- (package.len() / 2, io::ErrorKind::Other),
- (package.len() - 4, io::ErrorKind::Other),
- (package.len(), io::ErrorKind::Other),
- (0, io::ErrorKind::FileTooLarge),
- ] {
- let directory = tempfile::tempdir()?;
- let input = Cursor::new(&package[..offset]).chain(FailingIo(kind));
- let error = stage_plugin_program_package(LocalArchiveReader(input), directory.path())
- .err()
- .ok_or_else(|| io::Error::other("Local read failure was ignored"))?;
- assert!(
- matches!(error, PluginPackageError::FilesystemOperationFailed { source }
- if source.kind() == kind),
- "offset {offset}, kind {kind}"
- );
- assert!(directory.path().read_dir()?.next().is_none());
- }
- Ok(())
- }
-
- struct FailingIo(io::ErrorKind);
-
- impl Read for FailingIo {
- fn read(&mut self, _output: &mut [u8]) -> io::Result<usize> {
- Err(io::Error::new(self.0, "Injected read failure"))
- }
- }
-
- impl Write for FailingIo {
- fn write(&mut self, _input: &[u8]) -> io::Result<usize> {
- Err(io::Error::new(self.0, "Injected write failure"))
- }
-
- fn flush(&mut self) -> io::Result<()> {
- Ok(())
- }
- }
-}
diff --git a/src/runner/recovery/mod.rs b/src/runner/recovery/mod.rs
index ef62e4b..a85fdd9 100644
--- a/src/runner/recovery/mod.rs
+++ b/src/runner/recovery/mod.rs
@@ -19,15 +19,16 @@
//! Restores Runner desired state from the private Stores.
//!
-//! Startup verifies committed Documents, then asks the sole Program Store to
-//! recover and clean its private installation tree. Invalid Documents or Store
-//! root/cleanup failures stop startup before any UDS or child is created.
-//! Invalid Programs are removed; their Documents remain as unready desired state.
+//! Startup validates saved Documents, then restores the Program Store.
+//! Invalid Documents or Store root and cleanup failures stop startup.
+//! Package access or validation failure stops startup and keeps the saved input.
+//! Runtime files are reconstructed from the saved packages.
+//! No new UDS or child process is created before recovery completes.
use crate::config::RunnerConfig;
use crate::identifiers::TenonDocumentId;
use crate::runner::document_store::{TenonDocumentStore, TenonDocumentStoreError};
-use crate::runner::extensions::DocumentProtection;
+use crate::runner::extensions::ArtifactProtection;
use crate::runner::plugin::store::{PluginProgramStore, PluginStoreError};
use crate::runner::state_directory::RunnerStateLayout;
use crate::tenon_document::verified::VerifiedTenonDocument;
@@ -38,6 +39,7 @@ use crate::tenon_document::{
use sha2::{Digest as _, Sha256};
use std::error::Error;
use std::fmt;
+use std::path::Path;
use std::sync::Arc;
use zeroize::Zeroizing;
@@ -78,11 +80,13 @@ impl RecoveredTenonDocument {
///
/// Returns [`RunnerRecoveryError`] when a root Store operation fails, a
/// committed Document cannot be parsed or verified, its filename identity is
-/// inconsistent, or Program cleanup cannot complete. Invalid Programs are removed.
+/// inconsistent, or Program recovery or cleanup cannot complete.
+/// Package access or validation failure keeps the saved input.
pub(crate) fn recover(
config: &RunnerConfig,
state_layout: &RunnerStateLayout,
- protection: &Arc<dyn DocumentProtection>,
+ protection: &Arc<dyn ArtifactProtection>,
+ runtime_directory: &Path,
) -> Result<RecoveredRunnerState, RunnerRecoveryError> {
let verifier = config
.tenon_document_verifier()
@@ -122,7 +126,12 @@ pub(crate) fn recover(
}
verified.sort_unstable_by(|left, right| left.1.id().as_str().cmp(right.1.id().as_str()));
- let programs = load_program_store(state_layout)?;
+ let programs = PluginProgramStore::recover(
+ state_layout.plugin_program_store_directory(),
+ runtime_directory.to_path_buf(),
+ Arc::clone(protection),
+ )
+ .map_err(RunnerRecoveryError::PluginStore)?;
let documents = verified
.into_iter()
.map(|(source, document)| RecoveredTenonDocument { source, document })
@@ -135,13 +144,6 @@ pub(crate) fn recover(
})
}
-fn load_program_store(
- state_layout: &RunnerStateLayout,
-) -> Result<PluginProgramStore, RunnerRecoveryError> {
- PluginProgramStore::recover(state_layout.plugin_program_store_directory())
- .map_err(RunnerRecoveryError::PluginStore)
-}
-
/// A fatal private-state failure discovered before the Runner starts serving.
#[derive(Debug)]
pub(crate) enum RunnerRecoveryError {
diff --git a/src/runner/recovery/tests.rs b/src/runner/recovery/tests.rs
index 97a04b8..2390d92 100644
--- a/src/runner/recovery/tests.rs
+++ b/src/runner/recovery/tests.rs
@@ -19,7 +19,6 @@
use super::recover;
use crate::config::RunnerConfig;
-use crate::identifiers::{ExactVersion, ProgramName};
use crate::payload_contract::PluginInterface;
use crate::pipeline::test_support::PipelineRevision;
use crate::runner::document_store::TenonDocumentEtag;
@@ -63,7 +62,8 @@ fn startup_resolution_reflects_the_current_plugin_registry() -> io::Result<()> {
let (documents, programs) = recover(
&config,
&state_layout,
- &RunnerHooks::default().document_protection,
+ &RunnerHooks::default().artifact_protection,
+ &state_layout.pipeline_runtime_directory(),
)
.map_err(io::Error::other)?
.into_parts();
@@ -111,7 +111,8 @@ fn startup_resolution_reflects_the_current_plugin_registry() -> io::Result<()> {
let (documents, programs) = recover(
&config,
&state_layout,
- &RunnerHooks::default().document_protection,
+ &RunnerHooks::default().artifact_protection,
+ &state_layout.pipeline_runtime_directory(),
)
.map_err(io::Error::other)?
.into_parts();
@@ -149,8 +150,12 @@ fn startup_recovers_all_program_interfaces_in_one_store() -> io::Result<()> {
install_plugin(state_directory.path(), PluginInterface::Sink)?;
let state_layout =
prepare_runner_state_directory(state_directory.path()).map_err(io::Error::other)?;
- let mut store = PluginProgramStore::recover(state_layout.plugin_program_store_directory())
- .map_err(io::Error::other)?;
+ let mut store = PluginProgramStore::recover(
+ state_layout.plugin_program_store_directory(),
+ state_layout.plugin_program_store_directory(),
+ std::sync::Arc::new(crate::runner::extensions::ByPass),
+ )
+ .map_err(io::Error::other)?;
store
.install(Cursor::new(valid_program_package(
PluginInterface::SourceAndSink,
@@ -161,7 +166,8 @@ fn startup_recovers_all_program_interfaces_in_one_store() -> io::Result<()> {
let (_documents, programs) = recover(
&config,
&state_layout,
- &RunnerHooks::default().document_protection,
+ &RunnerHooks::default().artifact_protection,
+ &state_layout.pipeline_runtime_directory(),
)
.map_err(io::Error::other)?
.into_parts();
@@ -171,12 +177,16 @@ fn startup_recovers_all_program_interfaces_in_one_store() -> io::Result<()> {
}
#[test]
-fn startup_deletes_a_corrupt_unified_program_without_deleting_documents() -> io::Result<()> {
+fn startup_preserves_a_corrupt_original_package_and_documents() -> io::Result<()> {
let state_directory = tempfile::tempdir()?;
let state_layout =
prepare_runner_state_directory(state_directory.path()).map_err(io::Error::other)?;
- let mut store = PluginProgramStore::recover(state_layout.plugin_program_store_directory())
- .map_err(io::Error::other)?;
+ let mut store = PluginProgramStore::recover(
+ state_layout.plugin_program_store_directory(),
+ state_layout.pipeline_runtime_directory(),
+ std::sync::Arc::new(crate::runner::extensions::ByPass),
+ )
+ .map_err(io::Error::other)?;
store
.install(Cursor::new(valid_source_program_package()?))
.map_err(io::Error::other)?;
@@ -186,30 +196,31 @@ fn startup_deletes_a_corrupt_unified_program_without_deleting_documents() -> io:
"keep-document",
document("keep-document"),
)?;
- let manifest = state_directory
+ let package = state_directory
.path()
- .join("plugins/programs/com.example.source/1.0.0/manifest.json");
- fs::set_permissions(&manifest, fs::Permissions::from_mode(0o700))?;
- fs::write(&manifest, b"{}")?;
- fs::set_permissions(&manifest, fs::Permissions::from_mode(0o500))?;
+ .join("plugins/programs/com.example.source/.tenon-artifact-92521fc3cbd964bdc9f584a991b89fddaa5754ed1cc96d6d42445338669c1305");
+ fs::write(&package, b"invalid archive")?;
let config = load_config(state_directory.path())?;
-
- let (documents, programs) = recover(
+ let error = recover(
&config,
&state_layout,
- &RunnerHooks::default().document_protection,
+ &RunnerHooks::default().artifact_protection,
+ &state_layout.pipeline_runtime_directory(),
)
- .map_err(io::Error::other)?
- .into_parts();
- let program_name =
- ProgramName::try_from(String::from("com.example.source")).map_err(io::Error::other)?;
- let exact_version = ExactVersion::try_from(String::from("1.0.0")).map_err(io::Error::other)?;
- assert!(programs.lookup(&program_name, &exact_version).is_none());
- assert!(!manifest.exists());
- assert_eq!(documents.len(), 1);
- let (source, document) = documents.into_vec().remove(0).into_parts();
- assert_eq!(document.id().as_str(), "keep-document");
- assert!(!source.is_empty());
+ .err()
+ .ok_or_else(|| io::Error::other("Corrupt package did not stop recovery"))?;
+ assert_eq!(error.code(), "plugin_store_integrity_invalid");
+ assert_eq!(fs::read(&package)?, b"invalid archive");
+ assert_eq!(
+ fs::read(
+ fs::read_dir(state_layout.tenon_document_store_directory())?
+ .next()
+ .transpose()?
+ .ok_or_else(|| io::Error::other("Saved Document is missing"))?
+ .path()
+ )?,
+ document("keep-document").as_bytes()
+ );
Ok(())
}
@@ -228,7 +239,8 @@ fn startup_rejects_a_committed_file_whose_name_does_not_match_its_document_id()
let error = recover(
&config,
&state_layout,
- &RunnerHooks::default().document_protection,
+ &RunnerHooks::default().artifact_protection,
+ &state_layout.pipeline_runtime_directory(),
)
.err()
.ok_or_else(|| io::Error::other("mismatched committed identity was accepted"))?;
@@ -243,18 +255,11 @@ async fn empty_runner_starts_and_stops_its_private_control_service() -> io::Resu
let config = load_config(state_directory.path())?;
let state_layout =
prepare_runner_state_directory(state_directory.path()).map_err(io::Error::other)?;
- let recovered = recover(
- &config,
- &state_layout,
- &RunnerHooks::default().document_protection,
- )
- .map_err(io::Error::other)?;
RunnerMainLoop::start(
config,
crate::runner::process_resources::test_support::unavailable(),
state_layout,
- recovered,
CapturedRunnerExecutable::capture_current()?,
None,
RunnerHooks::default(),
@@ -285,19 +290,12 @@ async fn failed_http_start_cleans_already_prepared_runtime_owners() -> io::Resul
let config = load_config(state_directory.path())?;
let state_layout =
prepare_runner_state_directory(state_directory.path()).map_err(io::Error::other)?;
- let recovered = recover(
- &config,
- &state_layout,
- &RunnerHooks::default().document_protection,
- )
- .map_err(io::Error::other)?;
let _occupied = TcpListener::bind(config.http_listen_address())?;
let error = RunnerMainLoop::start(
config,
crate::runner::process_resources::test_support::unavailable(),
state_layout,
- recovered,
CapturedRunnerExecutable::capture_current()?,
None,
RunnerHooks::default(),
@@ -327,12 +325,6 @@ async fn shutdown_ready_before_lifecycle_poll_prevents_pipeline_spawn() -> io::R
let config = load_config(state_directory.path())?;
let state_layout =
prepare_runner_state_directory(state_directory.path()).map_err(io::Error::other)?;
- let recovered = recover(
- &config,
- &state_layout,
- &RunnerHooks::default().document_protection,
- )
- .map_err(io::Error::other)?;
let marker = state_directory.path().join("pipeline-spawned");
let executable = state_directory.path().join("pipeline.sh");
fs::write(
@@ -345,7 +337,6 @@ async fn shutdown_ready_before_lifecycle_poll_prevents_pipeline_spawn() -> io::R
config,
crate::runner::process_resources::test_support::unavailable(),
state_layout,
- recovered,
CapturedRunnerExecutable::from_file(File::open(&executable)?)?,
None,
RunnerHooks::default(),
@@ -374,12 +365,6 @@ async fn shutdown_period_plugin_commit_does_not_start_a_new_pipeline() -> io::Re
let http_address = config.http_listen_address();
let state_layout =
prepare_runner_state_directory(state_directory.path()).map_err(io::Error::other)?;
- let recovered = recover(
- &config,
- &state_layout,
- &RunnerHooks::default().document_protection,
- )
- .map_err(io::Error::other)?;
let marker = state_directory
.path()
.join("pipeline-spawned-after-shutdown");
@@ -393,7 +378,6 @@ async fn shutdown_period_plugin_commit_does_not_start_a_new_pipeline() -> io::Re
config,
crate::runner::process_resources::test_support::unavailable(),
state_layout,
- recovered,
CapturedRunnerExecutable::from_file(File::open(&executable)?)?,
None,
RunnerHooks::default(),
@@ -409,7 +393,7 @@ async fn shutdown_period_plugin_commit_does_not_start_a_new_pipeline() -> io::Re
stream
.write_all(
format!(
- "POST /plugins HTTP/1.1\r\nHost: {http_address}\r\nConnection: close\r\nContent-Type: application/vnd.apache.tenon.plugin+tar+gzip\r\nContent-Length: {}\r\nExpect: 100-continue\r\n\r\n",
+ "POST /plugins HTTP/1.1\r\nHost: {http_address}\r\nConnection: close\r\nContent-Type: application/octet-stream\r\nContent-Length: {}\r\nExpect: 100-continue\r\n\r\n",
package.len(),
)
.as_bytes(),
@@ -502,12 +486,6 @@ async fn observe_unattached_pipeline(shutdown: UnattachedShutdown) -> io::Result
let config = RunnerConfig::load(&config_path).map_err(io::Error::other)?;
let state_layout =
prepare_runner_state_directory(state_directory.path()).map_err(io::Error::other)?;
- let recovered = recover(
- &config,
- &state_layout,
- &RunnerHooks::default().document_protection,
- )
- .map_err(io::Error::other)?;
let process_marker = state_directory.path().join("pipeline.pid");
let executable = state_directory.path().join("pipeline.sh");
fs::write(
@@ -524,7 +502,6 @@ async fn observe_unattached_pipeline(shutdown: UnattachedShutdown) -> io::Result
config,
crate::runner::process_resources::test_support::unavailable(),
state_layout,
- recovered,
CapturedRunnerExecutable::from_file(File::open(&executable)?)?,
None,
RunnerHooks::default(),
@@ -534,7 +511,9 @@ async fn observe_unattached_pipeline(shutdown: UnattachedShutdown) -> io::Result
.map_err(io::Error::other)?
.run_until_shutdown(async {
time::timeout(Duration::from_secs(5), async {
- while !shutdown_marker.is_file() {
+ while !fs::read_to_string(&shutdown_marker)
+ .is_ok_and(|value| value.trim().parse::<i32>().is_ok())
+ {
time::sleep(Duration::from_millis(5)).await;
}
let mut observed_starting = false;
@@ -611,3 +590,86 @@ async fn observe_unattached_pipeline(shutdown: UnattachedShutdown) -> io::Result
metrics.shutdown();
Ok(())
}
+
+#[tokio::test(flavor = "current_thread")]
+async fn policy_change_during_recovery_is_applied_before_pipeline_start() -> io::Result<()> {
+ use crate::runner::extensions::{
+ ArtifactProtection, ExecutionDenied, ExecutionPermit, ExecutionPolicy, ExecutionScope,
+ NoHttpAuth, PolicyChanges,
+ };
+ use std::sync::{
+ Arc,
+ atomic::{AtomicBool, Ordering},
+ };
+ struct Policy {
+ denied: Arc<AtomicBool>,
+ changes: PolicyChanges,
+ }
+ impl ExecutionPolicy for Policy {
+ fn authorize(
+ &self,
+ _scope: ExecutionScope<'_>,
+ ) -> Result<ExecutionPermit, ExecutionDenied> {
+ if self.denied.load(Ordering::Acquire) {
+ Err(ExecutionDenied::new(
+ "test.recovery_denied",
+ "Execution is not permitted",
+ None,
+ ))
+ } else {
+ Ok(ExecutionPermit::default())
+ }
+ }
+ fn changes(&self) -> Option<PolicyChanges> {
+ Some(self.changes.clone())
+ }
+ }
+ struct Protection {
+ denied: Arc<AtomicBool>,
+ changes: tokio::sync::watch::Sender<()>,
+ }
+ impl ArtifactProtection for Protection {
+ fn protect(&self, source: &[u8], output: &mut dyn io::Write) -> io::Result<()> {
+ output.write_all(source)
+ }
+ fn unprotect(&self, stored: &[u8], output: &mut dyn io::Write) -> io::Result<()> {
+ output.write_all(stored)?;
+ self.denied.store(true, Ordering::Release);
+ self.changes.send_replace(());
+ Ok(())
+ }
+ }
+ let root = tempfile::tempdir()?;
+ let config = load_config(root.path())?;
+ let layout = prepare_runner_state_directory(root.path()).map_err(io::Error::other)?;
+ write_document(root.path(), "notify-recovery", document("notify-recovery"))?;
+ let denied = Arc::new(AtomicBool::new(false));
+ let (changes, receiver) = tokio::sync::watch::channel(());
+ let hooks = RunnerHooks::new(
+ Policy {
+ denied: Arc::clone(&denied),
+ changes: receiver,
+ },
+ Protection { denied, changes },
+ NoHttpAuth,
+ );
+ let error = RunnerMainLoop::start(
+ config,
+ crate::runner::process_resources::test_support::unavailable(),
+ layout,
+ CapturedRunnerExecutable::capture_current()?,
+ None,
+ hooks,
+ crate::metrics::test_support::runner()?,
+ )
+ .await
+ .err()
+ .ok_or_else(|| io::Error::other("Recovery notification was lost"))?;
+ assert_eq!(error.code(), "test.recovery_denied");
+ assert_eq!(
+ fs::read_dir(root.path().join("tenon-documents"))?.count(),
+ 1
+ );
+ assert_eq!(fs::read_dir(root.path().join("pipelines"))?.count(), 0);
+ Ok(())
+}
diff --git a/src/runner/test_support.rs b/src/runner/test_support.rs
index 58043f5..294a1bc 100644
--- a/src/runner/test_support.rs
+++ b/src/runner/test_support.rs
@@ -50,11 +50,16 @@ pub(super) fn captured_runner_executable() -> io::Result<CapturedRunnerExecutabl
pub(super) fn install_plugin(state_directory: &Path, interface: PluginInterface) -> io::Result<()> {
let layout = prepare_runner_state_directory(state_directory).map_err(io::Error::other)?;
- PluginProgramStore::recover(layout.plugin_program_store_directory())
- .map_err(io::Error::other)?
- .install(Cursor::new(valid_program_package(interface)?))
- .map(drop)
- .map_err(io::Error::other)
+ let runtime = tempfile::tempdir_in(layout.pipeline_runtime_directory())?;
+ PluginProgramStore::recover(
+ layout.plugin_program_store_directory(),
+ runtime.path().to_path_buf(),
+ std::sync::Arc::new(crate::runner::extensions::ByPass),
+ )
+ .map_err(io::Error::other)?
+ .install(Cursor::new(valid_program_package(interface)?))
+ .map(drop)
+ .map_err(io::Error::other)
}
pub(super) fn write_document(
@@ -122,7 +127,8 @@ pub(super) fn target(
let (documents, programs) = recover(
config,
&state_layout,
- &RunnerHooks::default().document_protection,
+ &RunnerHooks::default().artifact_protection,
+ &state_layout.pipeline_runtime_directory(),
)
.map_err(io::Error::other)?
.into_parts();
diff --git a/tests/fixtures/runner_extensions.rs b/tests/fixtures/runner_extensions.rs
index 611f5a6..db05035 100644
--- a/tests/fixtures/runner_extensions.rs
+++ b/tests/fixtures/runner_extensions.rs
@@ -19,20 +19,25 @@
//! A real distribution binary exercising only Tenon's public extension API.
+#[path = "runner_extensions/live_policy.rs"]
+mod live_policy;
+
use axum::http::{HeaderMap, HeaderValue, Method};
use serde_json::Value;
use std::cell::Cell;
use std::fs::{File, OpenOptions};
use std::future::Future;
-use std::io::{self, Write};
+use std::io::{self, BufRead as _, Read, Write};
use std::path::PathBuf;
use std::pin::Pin;
use std::process::ExitCode;
use std::rc::Rc;
+use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
use tenon::{
- DocumentProtection, ExecutionDenied, ExecutionPermit, ExecutionPolicy, ExecutionScope,
- HttpApiAuthorization, HttpAuthRejection, NoHttpAuth, RunnerHooks,
+ ArtifactProtection, ByPass, ExecutionDenied, ExecutionPermit, ExecutionPolicy, ExecutionScope,
+ HttpApiAuthorization, HttpAuthRejection, NoHttpAuth, PluginPackageOutput, PolicyChanges,
+ RunnerHooks,
};
fn main() -> ExitCode {
@@ -44,8 +49,15 @@ fn main() -> ExitCode {
if let Some(marker) = extra.get("initializeMarker").and_then(Value::as_str) {
File::create_new(marker)?;
}
+ let live = extra
+ .get("policyFile")
+ .and_then(Value::as_str)
+ .map(|path| live_policy::LivePolicy::start(PathBuf::from(path)))
+ .transpose()?;
+ let package_state = live.as_ref().map(live_policy::LivePolicy::state);
Ok(RunnerHooks::new(
Policy {
+ live,
maximum_documents: extra.get("maximumDocuments").and_then(Value::as_u64),
maximum_authorizations: extra.get("maximumAuthorizations").and_then(Value::as_u64),
denied_document: extra
@@ -66,7 +78,14 @@ fn main() -> ExitCode {
.and_then(Value::as_u64)
.unwrap_or(1),
},
- Protection,
+ Protection {
+ package_state,
+ customer: extra
+ .get("customer")
+ .and_then(Value::as_str)
+ .unwrap_or("customer-a")
+ .to_owned(),
+ },
HttpAuthorization {
enabled: extra
.get("httpAuth")
@@ -78,6 +97,7 @@ fn main() -> ExitCode {
}
struct Policy {
+ live: Option<live_policy::LivePolicy>,
maximum_documents: Option<u64>,
maximum_authorizations: Option<u64>,
denied_document: Option<String>,
@@ -144,13 +164,62 @@ impl ExecutionPolicy for Policy {
entitlement_until,
));
}
- Ok(ExecutionPermit { entitlement_until })
+ match &self.live {
+ Some(live) => live.decision(),
+ None => Ok(ExecutionPermit { entitlement_until }),
+ }
+ }
+
+ fn changes(&self) -> Option<PolicyChanges> {
+ self.live.as_ref().map(live_policy::LivePolicy::changes)
}
}
-struct Protection;
+struct Protection {
+ package_state: Option<Arc<Mutex<live_policy::Snapshot>>>,
+ customer: String,
+}
+
+impl ArtifactProtection for Protection {
+ fn open_plugin_package(
+ &self,
+ source: &mut dyn Read,
+ output: &mut dyn PluginPackageOutput,
+ ) -> io::Result<()> {
+ let mut source = io::BufReader::new(source);
+ let prefix = if source.fill_buf()?.starts_with(b"fixture-package:") {
+ "fixture-package:"
+ } else if source.fill_buf()?.starts_with(b"fixture-files:") {
+ "fixture-files:"
+ } else {
+ return ByPass.open_plugin_package(&mut source, output);
+ };
+ let customer = match &self.package_state {
+ Some(state) => state
+ .lock()
+ .map_err(|_| io::Error::other("Fixture package state is unavailable"))?
+ .customer
+ .clone(),
+ None => self.customer.clone(),
+ };
+ let header = format!("{prefix}{customer}\n");
+ let mut actual = vec![0; header.len()];
+ source.read_exact(&mut actual)?;
+ if actual != header.as_bytes() {
+ return Err(io::Error::from(io::ErrorKind::PermissionDenied));
+ }
+ if prefix == "fixture-files:" {
+ let files: std::collections::BTreeMap<String, Vec<u8>> =
+ serde_json::from_reader(source)?;
+ for (path, bytes) in files {
+ output.write_file(&path, &mut bytes.as_slice())?;
+ }
+ Ok(())
+ } else {
+ ByPass.open_plugin_package(&mut DecodedPackage(source), output)
+ }
+ }
-impl DocumentProtection for Protection {
fn protect(&self, source: &[u8], output: &mut dyn Write) -> io::Result<()> {
output.write_all(b"fixture:")?;
if source
@@ -179,6 +248,19 @@ impl DocumentProtection for Protection {
}
}
+// This reversible test stream does not provide encryption.
+struct DecodedPackage<R>(R);
+
+impl<R: Read> Read for DecodedPackage<R> {
+ fn read(&mut self, output: &mut [u8]) -> io::Result<usize> {
+ let count = self.0.read(output)?;
+ for byte in &mut output[..count] {
+ *byte ^= 0x80;
+ }
+ Ok(count)
+ }
+}
+
struct HttpAuthorization {
enabled: bool,
}
diff --git a/tests/fixtures/runner_extensions/live_policy.rs b/tests/fixtures/runner_extensions/live_policy.rs
new file mode 100644
index 0000000..0728722
--- /dev/null
+++ b/tests/fixtures/runner_extensions/live_policy.rs
@@ -0,0 +1,139 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+//! A file updater owned by the test distribution, independent of the Runner internals.
+
+use serde::Deserialize;
+use std::fs;
+use std::io;
+use std::path::PathBuf;
+use std::sync::{Arc, Mutex, mpsc};
+use std::thread::{self, JoinHandle};
+use std::time::{Duration, Instant};
+use tenon::{ExecutionDenied, ExecutionPermit, PolicyChanges};
+use tokio::sync::watch;
+
+#[derive(Deserialize)]
+struct FileState {
+ customer: String,
+ allowed: bool,
+ valid_for_ms: u64,
+}
+
+pub(super) struct Snapshot {
+ pub(super) customer: String,
+ allowed: bool,
+ deadline: Instant,
+}
+
+pub(super) struct LivePolicy {
+ state: Arc<Mutex<Snapshot>>,
+ updates: PolicyChanges,
+ stop: mpsc::Sender<()>,
+ worker: Option<JoinHandle<()>>,
+}
+
+impl LivePolicy {
+ pub(super) fn start(path: PathBuf) -> io::Result<Self> {
+ let initial = fs::read(&path)?;
+ let snapshot =
+ parse(&initial).ok_or_else(|| io::Error::other("Initial policy file is invalid"))?;
+ let state = Arc::new(Mutex::new(snapshot));
+ let (sender, updates) = watch::channel(());
+ let (stop, stopped) = mpsc::channel();
+ let shared = Arc::clone(&state);
+ let worker = thread::spawn(move || {
+ let mut last = initial;
+ while matches!(
+ stopped.recv_timeout(Duration::from_millis(5)),
+ Err(mpsc::RecvTimeoutError::Timeout)
+ ) {
+ let Ok(bytes) = fs::read(&path) else { continue };
+ if bytes == last {
+ continue;
+ }
+ let Some(next) = parse(&bytes) else { continue };
+ let Ok(mut state) = shared.lock() else { break };
+ // This fixture permits renewal only for the original customer.
+ if state.customer != next.customer {
+ continue;
+ }
+ *state = next;
+ drop(state);
+ last = bytes;
+ sender.send_replace(());
+ }
+ });
+ Ok(Self {
+ state,
+ updates,
+ stop,
+ worker: Some(worker),
+ })
+ }
+
+ pub(super) fn decision(&self) -> Result<ExecutionPermit, ExecutionDenied> {
+ let state = self.state.lock().map_err(|_| {
+ ExecutionDenied::new(
+ "fixture.state_failed",
+ "Policy state is unavailable",
+ Some(Instant::now()),
+ )
+ })?;
+ let deadline = Some(state.deadline);
+ if state.allowed {
+ Ok(ExecutionPermit {
+ entitlement_until: deadline,
+ })
+ } else {
+ Err(ExecutionDenied::new(
+ "fixture.license_denied",
+ "Execution is not permitted",
+ deadline,
+ ))
+ }
+ }
+
+ pub(super) fn changes(&self) -> PolicyChanges {
+ self.updates.clone()
+ }
+
+ pub(super) fn state(&self) -> Arc<Mutex<Snapshot>> {
+ Arc::clone(&self.state)
+ }
+}
+
+impl Drop for LivePolicy {
+ fn drop(&mut self) {
+ let _ = self.stop.send(());
+ if let Some(worker) = self.worker.take() {
+ let _ = worker.join();
+ }
+ }
+}
+
+fn parse(bytes: &[u8]) -> Option<Snapshot> {
+ let state: FileState = serde_json::from_slice(bytes).ok()?;
+ let deadline = Instant::now().checked_add(Duration::from_millis(state.valid_for_ms))?;
+ Some(Snapshot {
+ customer: state.customer,
+ allowed: state.allowed,
+ deadline,
+ })
+}
diff --git a/tests/runner_cli.rs b/tests/runner_cli.rs
index e7a7606..d686092 100644
--- a/tests/runner_cli.rs
+++ b/tests/runner_cli.rs
@@ -30,8 +30,7 @@ use std::sync::{Mutex, MutexGuard};
use std::time::{Duration, Instant};
use std::{fs, iter, thread};
-#[path = "support/file_tree.rs"]
-mod file_tree;
+use plugin_fixture::file_tree;
#[path = "runner_cli/plugin_fixture.rs"]
mod plugin_fixture;
use file_tree::named_files;
@@ -515,14 +514,11 @@ fn unified_program_delete_failures_stop_all_pipelines_and_exit_nonzero() -> io::
#[derive(Clone, Copy, Debug)]
enum DeletionFault {
NamespaceDrift,
- TombstoneCleanup,
+ RuntimeCleanup,
}
- for fault in [
- DeletionFault::NamespaceDrift,
- DeletionFault::TombstoneCleanup,
- ] {
- if matches!(fault, DeletionFault::TombstoneCleanup) && geteuid().is_root() {
+ for fault in [DeletionFault::NamespaceDrift, DeletionFault::RuntimeCleanup] {
+ if matches!(fault, DeletionFault::RuntimeCleanup) && geteuid().is_root() {
eprintln!("Skipping permission-denial scenario: it requires a non-root test user");
continue;
}
@@ -536,15 +532,8 @@ fn unified_program_delete_failures_stop_all_pipelines_and_exit_nonzero() -> io::
let namespace = directory
.path()
.join("plugins/programs/com.example.gateway");
- let package = namespace.join("1.0.0");
- let resources = package.join("resources");
- fs::create_dir(&resources)?;
- fs::set_permissions(&resources, fs::Permissions::from_mode(0o700))?;
- fs::write(resources.join("data.bin"), b"resource")?;
- fs::set_permissions(
- resources.join("data.bin"),
- fs::Permissions::from_mode(0o500),
- )?;
+ let package =
+ plugin_fixture::stored_package_path(directory.path(), "com.example.gateway", "1.0.0");
for id in ["store-failure-a", "store-failure-b"] {
write_document(
directory.path(),
@@ -572,7 +561,15 @@ fn unified_program_delete_failures_stop_all_pipelines_and_exit_nonzero() -> io::
}
})?;
request.set_write_timeout(Some(PROCESS_DEADLINE))?;
- let _permission_guard = match fault {
+ let runtime_program = plugin_fixture::runtime_program_directory(
+ directory.path(),
+ "com.example.gateway",
+ "1.0.0",
+ )?;
+ let resources = runtime_program.join("resources");
+ fs::create_dir(&resources)?;
+ fs::write(resources.join("data.bin"), b"resource")?;
+ let permission_guard = match fault {
DeletionFault::NamespaceDrift => {
fs::rename(
&namespace,
@@ -581,7 +578,7 @@ fn unified_program_delete_failures_stop_all_pipelines_and_exit_nonzero() -> io::
fs::write(&namespace, b"not a namespace")?;
None
}
- DeletionFault::TombstoneCleanup => Some(NonWritableDirectory::new(&resources)?),
+ DeletionFault::RuntimeCleanup => Some(NonWritableDirectory::new(&resources)?),
};
write!(
@@ -590,6 +587,7 @@ fn unified_program_delete_failures_stop_all_pipelines_and_exit_nonzero() -> io::
)?;
request.flush()?;
wait_for_pipeline_shutdown_requests(&runtime_root, 2)?;
+ drop(permission_guard);
let output = runner.wait()?;
let diagnostic = String::from_utf8_lossy(&output.stderr);
@@ -607,26 +605,26 @@ fn unified_program_delete_failures_stop_all_pipelines_and_exit_nonzero() -> io::
diagnostic.contains("Runner Plugin Program Store failed"),
"{diagnostic}"
);
- assert!(
- diagnostic.contains(namespace.to_string_lossy().as_ref()),
- "{diagnostic}"
- );
match fault {
DeletionFault::NamespaceDrift => {
assert!(
diagnostic.contains("Plugin Store integrity is invalid"),
"{diagnostic}"
);
+ assert!(
+ diagnostic.contains(namespace.to_string_lossy().as_ref()),
+ "{diagnostic}"
+ );
assert!(namespace.is_file());
}
- DeletionFault::TombstoneCleanup => {
+ DeletionFault::RuntimeCleanup => {
assert!(
- diagnostic.contains(".tenon-plugin-delete-1.0.0"),
+ diagnostic.contains(runtime_program.to_string_lossy().as_ref()),
"{diagnostic}"
);
assert!(diagnostic.contains("Permission denied"), "{diagnostic}");
assert!(!package.exists());
- assert!(namespace.join(".tenon-plugin-delete-1.0.0").is_dir());
+ assert!(namespace.join(".tenon-plugin-delete-92521fc3cbd964bdc9f584a991b89fddaa5754ed1cc96d6d42445338669c1305").is_file());
}
}
wait_for_processes_to_exit(pipelines.into_iter().chain(plugins))?;
diff --git a/tests/runner_cli/plugin_fixture.rs b/tests/runner_cli/plugin_fixture.rs
index f07f6b6..b9286e7 100644
--- a/tests/runner_cli/plugin_fixture.rs
+++ b/tests/runner_cli/plugin_fixture.rs
@@ -29,11 +29,15 @@ use prost_types::{
DescriptorProto, FileDescriptorProto, FileDescriptorSet, SourceCodeInfo, source_code_info,
};
use serde_json::json;
+use sha2::{Digest as _, Sha256};
use std::collections::BTreeMap;
use std::fs;
use std::io;
use std::os::unix::fs::PermissionsExt as _;
-use std::path::Path;
+use std::path::{Path, PathBuf};
+
+#[path = "../support/file_tree.rs"]
+pub(crate) mod file_tree;
pub(crate) use tenon::runner_test_support::contracts::core::PluginInterface;
#[path = "../support/controlled_plugin.rs"]
@@ -53,20 +57,57 @@ pub(crate) fn install_program(
) -> io::Result<InstalledProgramFixture> {
let files = program_files(interface, None)?;
let (program_name, _) = program_identity(interface);
- let plugins = state_directory.join("plugins");
+ let fixture = InstalledProgramFixture {
+ config_schema: files["config.schema.json"].clone(),
+ payload_descriptor: files["payload.descriptor.pb"].clone(),
+ };
+ write_package(state_directory, program_name, "1.0.0", &archive(files)?)?;
+ Ok(fixture)
+}
+
+pub(crate) fn stored_package_path(state: &Path, name: &str, version: &str) -> PathBuf {
+ let hash: String = Sha256::digest(version.as_bytes())
+ .iter()
+ .map(|byte| format!("{byte:02x}"))
+ .collect();
+ state
+ .join("plugins/programs")
+ .join(name)
+ .join(format!(".tenon-artifact-{hash}"))
+}
+
+pub(crate) fn write_package(
+ state: &Path,
+ name: &str,
+ version: &str,
+ bytes: &[u8],
+) -> io::Result<()> {
+ let plugins = state.join("plugins");
let programs = plugins.join("programs");
- let program_root = programs.join(program_name);
- let package = program_root.join("1.0.0");
- for directory in [&plugins, &programs, &program_root, &package] {
+ let namespace = programs.join(name);
+ for directory in [&plugins, &programs, &namespace] {
create_private_directory(directory)?;
}
- for (path, bytes) in &files {
- write_private_file(&package.join(path), bytes)?;
+ let path = stored_package_path(state, name, version);
+ fs::write(&path, bytes)?;
+ fs::set_permissions(path, fs::Permissions::from_mode(0o600))
+}
+
+pub(crate) fn runtime_program_directory(
+ state: &Path,
+ name: &str,
+ version: &str,
+) -> io::Result<PathBuf> {
+ for manifest in file_tree::named_files(&state.join("pipelines"), "manifest.json")? {
+ let value: serde_json::Value = serde_json::from_slice(&fs::read(&manifest)?)?;
+ if value["programName"] == name && value["exactVersion"] == version {
+ return manifest
+ .parent()
+ .map(Path::to_path_buf)
+ .ok_or_else(|| io::Error::other("Runtime manifest has no parent"));
+ }
}
- Ok(InstalledProgramFixture {
- config_schema: files["config.schema.json"].clone(),
- payload_descriptor: files["payload.descriptor.pb"].clone(),
- })
+ Err(io::Error::other("Runtime Program directory is missing"))
}
pub(crate) fn plugin_package(interface: PluginInterface) -> io::Result<Vec<u8>> {
@@ -80,7 +121,7 @@ pub(crate) fn plugin_package_with_program(
archive(program_files(interface, Some(program))?)
}
-fn program_files(
+pub(crate) fn program_files(
interface: PluginInterface,
program: Option<&[u8]>,
) -> io::Result<BTreeMap<String, Vec<u8>>> {
@@ -131,7 +172,7 @@ fn program_identity(interface: PluginInterface) -> (&'static str, &'static str)
}
}
-fn archive(files: BTreeMap<String, Vec<u8>>) -> io::Result<Vec<u8>> {
+pub(crate) fn archive(files: BTreeMap<String, Vec<u8>>) -> io::Result<Vec<u8>> {
let encoder = GzEncoder::new(Vec::new(), Compression::default());
let mut archive = tar::Builder::new(encoder);
for (path, bytes) in files {
@@ -194,11 +235,6 @@ fn create_private_directory(path: &Path) -> io::Result<()> {
fs::set_permissions(path, fs::Permissions::from_mode(0o700))
}
-fn write_private_file(path: &Path, bytes: &[u8]) -> io::Result<()> {
- fs::write(path, bytes)?;
- fs::set_permissions(path, fs::Permissions::from_mode(0o500))
-}
-
pub(crate) fn platforms() -> serde_json::Value {
json!([{"os":"linux","architecture":"amd64"},{"os":"linux","architecture":"arm64"},{"os":"darwin","architecture":"amd64"},{"os":"darwin","architecture":"arm64"}])
}
diff --git a/tests/runner_extensions.rs b/tests/runner_extensions.rs
index 9d850a1..f077ace 100644
--- a/tests/runner_extensions.rs
+++ b/tests/runner_extensions.rs
@@ -19,12 +19,15 @@
//! Exercises a separately linked distribution through real HTTP and child processes.
-#[path = "support/file_tree.rs"]
-mod file_tree;
+#[path = "runner_extensions/artifacts.rs"]
+mod artifacts;
+use plugin_fixture::file_tree;
#[path = "runner_extensions/http_authorization.rs"]
mod http_authorization;
#[path = "runner_cli/plugin_fixture.rs"]
mod plugin_fixture;
+#[path = "runner_extensions/policy_changes.rs"]
+mod policy_changes;
#[path = "support/runner_http.rs"]
mod runner_http_support;
@@ -273,7 +276,12 @@ fn startup_accepts_cumulatively_and_continues_after_a_middle_rejection() -> io::
.collect::<Vec<_>>();
assert_eq!(
ids,
- [vec!["doc1"], vec!["doc1", "doc2"], vec!["doc1", "doc3"]]
+ [
+ vec![],
+ vec!["doc1"],
+ vec!["doc1", "doc2"],
+ vec!["doc1", "doc3"]
+ ]
);
for id in ["doc1", "doc2", "doc3"] {
assert_eq!(
@@ -387,7 +395,7 @@ fn plugin_installation_uses_the_already_authorized_desired_state() -> io::Result
let config = configure(
root.path(),
address,
- serde_json::json!({"maximumAuthorizations": 1}),
+ serde_json::json!({"maximumAuthorizations": 2}),
)?;
let mut runner = TestRunner::spawn_with_executable(&config, Path::new(EXECUTABLE))?;
wait_for_http(&mut runner, address)?;
@@ -396,17 +404,12 @@ fn plugin_installation_uses_the_already_authorized_desired_state() -> io::Result
address,
"POST",
"/plugins",
- &[(
- "Content-Type",
- "application/vnd.apache.tenon.plugin+tar+gzip",
- )],
+ &[("Content-Type", "application/octet-stream")],
&package,
)?;
assert_eq!(installed.status, 201, "{}", installed.body_text());
assert!(
- root.path()
- .join("plugins/programs/com.example.modbus/1.0.0/manifest.json")
- .is_file()
+ plugin_fixture::stored_package_path(root.path(), "com.example.modbus", "1.0.0").is_file()
);
assert_eq!(
request(address, "GET", "/documents/waiting", &[], &[])?.status,
@@ -505,7 +508,7 @@ fn failed_document_write_keeps_the_new_authorization_deadline() -> io::Result<()
let config = configure(
root.path(),
address,
- serde_json::json!({"expiresAfterMs": 1500, "expiresFromAuthorization": 2}),
+ serde_json::json!({"expiresAfterMs": 1500, "expiresFromAuthorization": 3}),
)?;
let mut runner = TestRunner::spawn_with_executable(&config, Path::new(EXECUTABLE))?;
wait_for_http(&mut runner, address)?;
@@ -541,7 +544,7 @@ fn rejected_document_updates_entitlement_and_expires_running_processes() -> io::
let config = configure(
root.path(),
address,
- serde_json::json!({"expiresAfterMs": 1500, "expiresFromAuthorization": 2, "maximumDocuments": 1}),
+ serde_json::json!({"expiresAfterMs": 1500, "expiresFromAuthorization": 3, "maximumDocuments": 1}),
)?;
let mut runner = TestRunner::spawn_with_executable(&config, Path::new(EXECUTABLE))?;
wait_for_http(&mut runner, address)?;
@@ -578,7 +581,7 @@ fn automatic_pipeline_restart_uses_the_already_authorized_desired_state() -> io:
let config = configure(
root.path(),
address,
- serde_json::json!({"maximumAuthorizations": 1, "initializeMarker": root.path().join("initialized")}),
+ serde_json::json!({"maximumAuthorizations": 2, "initializeMarker": root.path().join("initialized")}),
)?;
let mut runner = TestRunner::spawn_with_executable(&config, Path::new(EXECUTABLE))?;
let processes = running_processes(root.path())?;
@@ -619,7 +622,7 @@ fn unfinished_upload_cannot_delay_expiry_even_after_normal_shutdown_begins() ->
upload.set_read_timeout(Some(runner_http_support::DEADLINE))?;
write!(
upload,
- "POST /plugins HTTP/1.1\r\nHost: {address}\r\nContent-Type: application/vnd.apache.tenon.plugin+tar+gzip\r\nContent-Length: 1000000\r\nExpect: 100-continue\r\n\r\n"
+ "POST /plugins HTTP/1.1\r\nHost: {address}\r\nContent-Type: application/octet-stream\r\nContent-Length: 1000000\r\nExpect: 100-continue\r\n\r\n"
)?;
runner_http_support::read_continue_response(&mut upload)?;
if begin_normal_shutdown {
@@ -794,8 +797,8 @@ fn policy_observes_only_desired_documents_while_old_revisions_are_still_running(
.lines()
.map(serde_json::from_str)
.collect::<Result<_, _>>()?;
- assert_eq!(scopes.len(), 4);
- let mut final_scope = scopes[3]
+ assert_eq!(scopes.len(), 5);
+ let mut final_scope = scopes[4]
.as_array()
.ok_or_else(|| io::Error::other("Scope is not an array"))?
.clone();
diff --git a/tests/runner_extensions/artifacts.rs b/tests/runner_extensions/artifacts.rs
new file mode 100644
index 0000000..dd4e2da
--- /dev/null
+++ b/tests/runner_extensions/artifacts.rs
@@ -0,0 +1,228 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+//! Proves that a distribution can use package protection through public APIs.
+
+use super::*;
+use std::os::unix::fs::PermissionsExt as _;
+
+fn envelope(interface: PluginInterface, customer: &str) -> io::Result<Vec<u8>> {
+ let mut bytes = format!("fixture-package:{customer}\n").into_bytes();
+ bytes.extend(
+ plugin_fixture::plugin_package(interface)?
+ .into_iter()
+ .map(|byte| byte ^ 0x80),
+ );
+ Ok(bytes)
+}
+
+fn file_package(interface: PluginInterface, customer: &str) -> io::Result<Vec<u8>> {
+ let mut bytes = format!("fixture-files:{customer}\n").into_bytes();
+ serde_json::to_writer(&mut bytes, &plugin_fixture::program_files(interface, None)?)?;
+ Ok(bytes)
+}
+
+fn install(address: SocketAddr, body: &[u8]) -> io::Result<runner_http_support::HttpResponse> {
+ request(
+ address,
+ "POST",
+ "/plugins",
+ &[("Content-Type", "application/octet-stream")],
+ body,
+ )
+}
+
+#[test]
+fn envelopes_survive_restart_while_runtime_materials_follow_process_lifetime() -> io::Result<()> {
+ let root = tempfile::tempdir()?;
+ let address = available_address()?;
+ let policy = root.path().join("license.json");
+ let log = root.path().join("scopes.jsonl");
+ super::policy_changes::publish(&policy, true, 60_000)?;
+ let config = configure(
+ root.path(),
+ address,
+ serde_json::json!({"policyFile": policy, "scopeLog": log}),
+ )?;
+ let mut runner = TestRunner::spawn_with_executable(&config, Path::new(EXECUTABLE))?;
+ wait_for_http(&mut runner, address)?;
+ let source = envelope(PluginInterface::Source, "customer-a")?;
+ let sink = file_package(PluginInterface::Sink, "customer-a")?;
+ for body in [&source, &sink] {
+ assert_eq!(install(address, body)?.status, 201);
+ assert_eq!(install(address, body)?.status, 204);
+ }
+ let saved = root
+ .path()
+ .join("plugins/programs/com.example.modbus/.tenon-artifact-92521fc3cbd964bdc9f584a991b89fddaa5754ed1cc96d6d42445338669c1305");
+ assert_eq!(fs::read(&saved)?, source);
+ assert!(file_tree::named_files(&root.path().join("plugins"), "manifest.json")?.is_empty());
+ let created = request(
+ address,
+ "PUT",
+ "/documents/protected-plugin",
+ &[
+ ("Content-Type", "application/jsonc"),
+ ("If-None-Match", "*"),
+ ],
+ &document("protected-plugin", "test"),
+ )?;
+ assert_eq!(created.status, 201, "{}", created.body_text());
+ let processes = running_processes(root.path())?;
+ super::policy_changes::publish(&policy, true, 120_000)?;
+ runner_http_support::wait_until(|| {
+ Ok((fs::read_to_string(&log)?.lines().count() == 3).then_some(()))
+ })?;
+ assert_eq!(running_processes(root.path())?, processes);
+ let material = file_tree::named_files(&root.path().join("pipelines"), "manifest.json")?;
+ assert_eq!(material.len(), 2);
+ assert_eq!(
+ request(
+ address,
+ "DELETE",
+ "/plugins/com.example.modbus/1.0.0",
+ &[],
+ &[]
+ )?
+ .status,
+ 409
+ );
+ runner.terminate()?;
+ assert_reaped(&processes);
+ assert!(material.iter().all(|path| !path.exists()));
+ assert_eq!(fs::read(&saved)?, source);
+
+ let mut runner = TestRunner::spawn_with_executable(&config, Path::new(EXECUTABLE))?;
+ let processes = running_processes(root.path())?;
+ wait_for_http(&mut runner, address)?;
+ assert_eq!(
+ request(address, "GET", "/documents/protected-plugin", &[], &[])?.status,
+ 200
+ );
+ super::policy_changes::publish(&policy, false, 120_000)?;
+ runner.wait_for_failure("fixture.license_denied")?;
+ assert_reaped(&processes);
+
+ let wrong = configure(
+ root.path(),
+ address,
+ serde_json::json!({"customer": "customer-b"}),
+ )?;
+ let mut runner = TestRunner::spawn_with_executable(&wrong, Path::new(EXECUTABLE))?;
+ runner.wait_for_failure("plugin_package_open_failed")?;
+ assert_eq!(fs::read(&saved)?, source);
+ assert_eq!(fs::read_dir(root.path().join("pipelines"))?.count(), 0);
+ Ok(())
+}
+
+#[test]
+fn custom_file_format_import_uses_access_checks_and_stock_runner_preserves_it() -> io::Result<()> {
+ let root = tempfile::tempdir()?;
+ let address = available_address()?;
+ let config = configure(root.path(), address, serde_json::json!({}))?;
+ let namespace = root.path().join("plugins/programs/com.example.modbus");
+ fs::create_dir_all(&namespace)?;
+ for directory in [
+ root.path().join("plugins"),
+ root.path().join("plugins/programs"),
+ ] {
+ fs::set_permissions(directory, fs::Permissions::from_mode(0o700))?;
+ }
+ fs::set_permissions(&namespace, fs::Permissions::from_mode(0o700))?;
+ let path = namespace
+ .join(".tenon-artifact-92521fc3cbd964bdc9f584a991b89fddaa5754ed1cc96d6d42445338669c1305");
+ let bytes = file_package(PluginInterface::Source, "customer-a")?;
+ fs::write(&path, &bytes)?;
+ fs::set_permissions(&path, fs::Permissions::from_mode(0o600))?;
+ let mut stock = TestRunner::spawn(&config)?;
+ stock.wait_for_failure("plugin_store_integrity_invalid")?;
+ assert_eq!(fs::read(&path)?, bytes);
+ let mut runner = TestRunner::spawn_with_executable(&config, Path::new(EXECUTABLE))?;
+ wait_for_http(&mut runner, address)?;
+ assert_eq!(
+ request(
+ address,
+ "GET",
+ "/plugins/com.example.modbus/1.0.0",
+ &[],
+ &[]
+ )?
+ .status,
+ 200
+ );
+ let wrong = envelope(PluginInterface::Sink, "customer-b")?;
+ assert_eq!(install(address, &wrong)?.status, 422);
+ let invalid = b"fixture-package:customer-a\ninvalid";
+ assert_eq!(install(address, invalid)?.status, 422);
+ assert_eq!(
+ request(
+ address,
+ "DELETE",
+ "/plugins/com.example.modbus/1.0.0",
+ &[],
+ &[]
+ )?
+ .status,
+ 204
+ );
+ assert!(!path.exists());
+ assert!(file_tree::named_files(&root.path().join("pipelines"), "manifest.json")?.is_empty());
+ runner.terminate()
+}
+
+#[test]
+fn standard_packages_survive_restart_and_rebuild_runtime_materials() -> io::Result<()> {
+ let root = tempfile::tempdir()?;
+ let address = available_address()?;
+ let config = configure(root.path(), address, serde_json::json!({}))?;
+ let package = plugin_fixture::plugin_package(PluginInterface::Source)?;
+ let saved = plugin_fixture::stored_package_path(root.path(), "com.example.modbus", "1.0.0");
+ let mut stock = TestRunner::spawn(&config)?;
+ wait_for_http(&mut stock, address)?;
+ assert_eq!(install(address, &package)?.status, 201);
+ let runtime =
+ plugin_fixture::runtime_program_directory(root.path(), "com.example.modbus", "1.0.0")?;
+ assert!(runtime.join("manifest.json").is_file());
+ assert_eq!(fs::read(&saved)?, package);
+ assert!(file_tree::named_files(&root.path().join("plugins"), "manifest.json")?.is_empty());
+ stock.terminate()?;
+ assert!(!runtime.exists());
+
+ let mut runner = TestRunner::spawn_with_executable(&config, Path::new(EXECUTABLE))?;
+ wait_for_http(&mut runner, address)?;
+ let rebuilt =
+ plugin_fixture::runtime_program_directory(root.path(), "com.example.modbus", "1.0.0")?;
+ assert_ne!(runtime, rebuilt);
+ assert!(rebuilt.join("manifest.json").is_file());
+ assert_eq!(fs::read(&saved)?, package);
+ assert_eq!(
+ request(
+ address,
+ "DELETE",
+ "/plugins/com.example.modbus/1.0.0",
+ &[],
+ &[]
+ )?
+ .status,
+ 204
+ );
+ assert!(!saved.exists());
+ assert!(!rebuilt.exists());
+ runner.terminate()
+}
diff --git a/tests/runner_extensions/http_authorization.rs b/tests/runner_extensions/http_authorization.rs
index 303d96f..dc098c8 100644
--- a/tests/runner_extensions/http_authorization.rs
+++ b/tests/runner_extensions/http_authorization.rs
@@ -97,7 +97,7 @@ fn authorization_controls_real_http_and_tls_requests_before_upload() -> io::Resu
let mut upload = transport.connect(address)?;
write!(
upload,
- "POST /plugins HTTP/1.1\r\nHost: {address}\r\nConnection: close\r\nContent-Type: application/vnd.apache.tenon.plugin+tar+gzip\r\nContent-Length: 1000000\r\nExpect: 100-continue\r\n\r\n"
+ "POST /plugins HTTP/1.1\r\nHost: {address}\r\nConnection: close\r\nContent-Type: application/octet-stream\r\nContent-Length: 1000000\r\nExpect: 100-continue\r\n\r\n"
)?;
upload.flush()?;
let mut bytes = Vec::new();
@@ -190,7 +190,7 @@ fn http_refusal_preserves_running_processes_and_does_not_refresh_execution_deadl
)?;
assert_eq!(rejected.status, 403);
assert_eq!(fs::read(formal_file(root.path(), "runtime"))?, source);
- assert_eq!(fs::read_to_string(&log)?.lines().count(), 1);
+ assert_eq!(fs::read_to_string(&log)?.lines().count(), 2);
for pid in &processes {
assert!(process::test_kill_process(*pid).is_ok());
}
diff --git a/tests/runner_extensions/policy_changes.rs b/tests/runner_extensions/policy_changes.rs
new file mode 100644
index 0000000..58be102
--- /dev/null
+++ b/tests/runner_extensions/policy_changes.rs
@@ -0,0 +1,133 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+//! Exercises file renewal and revocation without a Document write.
+
+use super::*;
+use std::os::unix::fs::PermissionsExt as _;
+
+pub(super) fn publish(path: &Path, allowed: bool, valid_for_ms: u64) -> io::Result<()> {
+ let temporary = path.with_extension("new");
+ fs::write(
+ &temporary,
+ serde_json::to_vec(&serde_json::json!({
+ "customer": "customer-a", "allowed": allowed, "valid_for_ms": valid_for_ms,
+ }))?,
+ )?;
+ fs::rename(temporary, path)
+}
+
+#[test]
+fn initial_denial_precedes_document_recovery_even_when_no_document_can_parse() -> io::Result<()> {
+ let root = tempfile::tempdir()?;
+ let address = available_address()?;
+ let policy = root.path().join("license.json");
+ publish(&policy, false, 60_000)?;
+ let config = configure(
+ root.path(),
+ address,
+ serde_json::json!({"policyFile": policy}),
+ )?;
+ let path = formal_file(root.path(), "invalid");
+ fs::create_dir_all(
+ path.parent()
+ .ok_or_else(|| io::Error::other("Document parent is absent"))?,
+ )?;
+ fs::set_permissions(
+ root.path().join("tenon-documents"),
+ fs::Permissions::from_mode(0o700),
+ )?;
+ fs::write(&path, b"invalid")?;
+ let mut runner = TestRunner::spawn_with_executable(&config, Path::new(EXECUTABLE))?;
+ runner.wait_for_failure("fixture.license_denied")?;
+ assert_eq!(fs::read(path)?, b"invalid");
+ Ok(())
+}
+
+#[test]
+fn idle_runner_renews_and_rejects_an_invalid_replacement() -> io::Result<()> {
+ let root = tempfile::tempdir()?;
+ let address = available_address()?;
+ let policy = root.path().join("license.json");
+ let log = root.path().join("scopes.jsonl");
+ publish(&policy, true, 60_000)?;
+ let config = configure(
+ root.path(),
+ address,
+ serde_json::json!({"policyFile": policy, "scopeLog": log}),
+ )?;
+ let mut runner = TestRunner::spawn_with_executable(&config, Path::new(EXECUTABLE))?;
+ wait_for_http(&mut runner, address)?;
+ publish(&policy, true, 1_500)?;
+ runner_http_support::wait_until(|| {
+ Ok((fs::read_to_string(&log)?.lines().count() == 2).then_some(()))
+ })?;
+ fs::write(&policy, b"invalid renewal")?;
+ // Only the accepted update can set the deadline. No HTTP call follows it.
+ runner.wait_for_failure("runner.execution_permit_expired")?;
+ let scopes = fs::read_to_string(&log)?;
+ assert!(scopes.lines().all(|line| line == "[]"));
+ assert_eq!(scopes.lines().count(), 2);
+ Ok(())
+}
+
+#[test]
+fn revocation_rechecks_unready_documents_and_preserves_their_files() -> io::Result<()> {
+ let root = tempfile::tempdir()?;
+ let address = available_address()?;
+ let policy = root.path().join("license.json");
+ let log = root.path().join("scopes.jsonl");
+ publish(&policy, true, 60_000)?;
+ let config = configure(
+ root.path(),
+ address,
+ serde_json::json!({"policyFile": policy, "scopeLog": log}),
+ )?;
+ let mut runner = TestRunner::spawn_with_executable(&config, Path::new(EXECUTABLE))?;
+ wait_for_http(&mut runner, address)?;
+ assert_eq!(
+ request(
+ address,
+ "PUT",
+ "/documents/waiting",
+ &[
+ ("Content-Type", "application/jsonc"),
+ ("If-None-Match", "*"),
+ ],
+ &document("waiting", "unready")
+ )?
+ .status,
+ 201
+ );
+ let path = formal_file(root.path(), "waiting");
+ let stored = fs::read(&path)?;
+ publish(&policy, false, 60_000)?;
+ runner.wait_for_failure("fixture.license_denied")?;
+ assert_eq!(fs::read(&path)?, stored);
+ let scopes = fs::read_to_string(&log)?;
+ let last: Value = serde_json::from_str(
+ scopes
+ .lines()
+ .last()
+ .ok_or_else(|| io::Error::other("Policy scope is absent"))?,
+ )?;
+ assert_eq!(last[0]["id"], "waiting");
+ assert_eq!(last.as_array().map(Vec::len), Some(1));
+ Ok(())
+}
diff --git a/tests/runner_http.rs b/tests/runner_http.rs
index d0e6dd8..6357455 100644
--- a/tests/runner_http.rs
+++ b/tests/runner_http.rs
@@ -51,28 +51,20 @@ fn unified_document_reaches_real_pipeline_and_keeps_applied_when_latest_is_unrea
] {
let directory = tempfile::tempdir()?;
install_program(directory.path(), PluginInterface::SourceAndSink)?;
- let foreign_manifest = if expected_issue == "plugin_platform_mismatch" {
- let namespace = directory
- .path()
- .join("plugins/programs/com.example.gateway");
- let foreign = namespace.join("2.0.0");
- fs::create_dir(&foreign)?;
- fs::set_permissions(&foreign, fs::Permissions::from_mode(0o700))?;
- for entry in fs::read_dir(namespace.join("1.0.0"))? {
- let entry = entry?;
- let mut bytes = fs::read(entry.path())?;
- if entry.file_name() == "manifest.json" {
- let mut manifest: serde_json::Value = serde_json::from_slice(&bytes)?;
- manifest["exactVersion"] = serde_json::json!("2.0.0");
- manifest["platforms"] =
- serde_json::json!([plugin_platform::foreign_platform()]);
- bytes = serde_json::to_vec(&manifest)?;
- }
- let target = foreign.join(entry.file_name());
- fs::write(&target, bytes)?;
- fs::set_permissions(target, fs::Permissions::from_mode(0o500))?;
- }
- Some(fs::read(foreign.join("manifest.json"))?)
+ let foreign_package = if expected_issue == "plugin_platform_mismatch" {
+ let mut files = plugin_fixture::program_files(PluginInterface::SourceAndSink, None)?;
+ let mut manifest: serde_json::Value = serde_json::from_slice(&files["manifest.json"])?;
+ manifest["exactVersion"] = serde_json::json!("2.0.0");
+ manifest["platforms"] = serde_json::json!([plugin_platform::foreign_platform()]);
+ files.insert("manifest.json".into(), serde_json::to_vec(&manifest)?);
+ let bytes = plugin_fixture::archive(files)?;
+ plugin_fixture::write_package(
+ directory.path(),
+ "com.example.gateway",
+ "2.0.0",
+ &bytes,
+ )?;
+ Some(bytes)
} else {
None
};
@@ -154,7 +146,7 @@ fn unified_document_reaches_real_pipeline_and_keeps_applied_when_latest_is_unrea
assert_eq!(latest["pluginInstances"][0]["exactVersion"], "1.0.0");
assert_eq!(latest["runtimeIssues"][0]["code"], expected_issue);
assert_summary(&latest)?;
- if let Some(manifest_bytes) = foreign_manifest {
+ if let Some(package_bytes) = foreign_package {
let summary = request(
address,
"GET",
@@ -168,12 +160,12 @@ fn unified_document_reaches_real_pipeline_and_keeps_applied_when_latest_is_unrea
serde_json::json!([plugin_platform::foreign_platform()])
);
assert_eq!(
- fs::read(
- directory
- .path()
- .join("plugins/programs/com.example.gateway/2.0.0/manifest.json")
- )?,
- manifest_bytes
+ fs::read(plugin_fixture::stored_package_path(
+ directory.path(),
+ "com.example.gateway",
+ "2.0.0"
+ ))?,
+ package_bytes
);
let in_use = request(
address,
@@ -487,9 +479,7 @@ fn unified_program_delete_rejects_invalid_paths_without_touching_the_store() ->
)?;
assert_eq!(entry.status, 200, "{}", entry.body_text());
assert!(
- directory
- .path()
- .join("plugins/programs/com.example.modbus/1.0.0/manifest.json")
+ plugin_fixture::stored_package_path(directory.path(), "com.example.modbus", "1.0.0")
.is_file()
);
runner.terminate()?;
@@ -500,13 +490,8 @@ fn unified_program_delete_rejects_invalid_paths_without_touching_the_store() ->
fn unified_program_queries_only_return_usable_metadata_summaries() -> io::Result<()> {
let directory = tempfile::tempdir()?;
install_program(directory.path(), PluginInterface::Source)?;
- install_program(directory.path(), PluginInterface::Sink)?;
install_program(directory.path(), PluginInterface::SourceAndSink)?;
let program_root = directory.path().join("plugins/programs");
- let invalid = program_root.join("com.example.kafka/1.0.0/manifest.json");
- fs::set_permissions(&invalid, fs::Permissions::from_mode(0o700))?;
- fs::write(&invalid, b"{}")?;
- fs::set_permissions(&invalid, fs::Permissions::from_mode(0o500))?;
let broken_namespace = program_root.join("com.example.broken");
fs::write(&broken_namespace, b"not a namespace")?;
let address = available_address()?;
@@ -558,7 +543,6 @@ fn unified_program_queries_only_return_usable_metadata_summaries() -> io::Result
assert_eq!(missing.json()["error"]["code"], "plugin_not_found");
}
}
- assert!(!program_root.join("com.example.kafka/1.0.0").exists());
assert!(!broken_namespace.exists());
runner.terminate()?;
@@ -856,10 +840,7 @@ fn document_plugin_and_reconcile_surface_is_one_consistent_boundary() -> io::Res
address,
"POST",
"/plugins",
- &[(
- "Content-Type",
- "application/vnd.apache.tenon.plugin+tar+gzip",
- )],
+ &[("Content-Type", "application/octet-stream")],
&package,
)?;
assert_eq!(installed.status, 201, "{}", installed.body_text());
@@ -867,10 +848,7 @@ fn document_plugin_and_reconcile_surface_is_one_consistent_boundary() -> io::Res
address,
"POST",
"/plugins",
- &[(
- "Content-Type",
- "application/vnd.apache.tenon.plugin+tar+gzip",
- )],
+ &[("Content-Type", "application/octet-stream")],
&package,
)?;
assert_eq!(unchanged.status, 204, "{}", unchanged.body_text());
@@ -916,10 +894,7 @@ fn document_plugin_and_reconcile_surface_is_one_consistent_boundary() -> io::Res
address,
"POST",
"/plugins",
- &[(
- "Content-Type",
- "application/vnd.apache.tenon.plugin+tar+gzip",
- )],
+ &[("Content-Type", "application/octet-stream")],
&conflicting_package,
)?;
assert_eq!(conflict.status, 409, "{}", conflict.body_text());
@@ -937,9 +912,9 @@ fn document_plugin_and_reconcile_surface_is_one_consistent_boundary() -> io::Res
let schema_path = "/plugins/com.example.modbus/1.0.0/config-schema";
let original_schema = request(address, "GET", schema_path, &[], &[])?;
assert_eq!(original_schema.status, 200);
- let private_schema = directory
- .path()
- .join("plugins/programs/com.example.modbus/1.0.0/config.schema.json");
+ let private_schema =
+ plugin_fixture::runtime_program_directory(directory.path(), "com.example.modbus", "1.0.0")?
+ .join("config.schema.json");
fs::set_permissions(&private_schema, fs::Permissions::from_mode(0o700))?;
fs::write(&private_schema, br#"{"type":"null"}"#)?;
let schema_after_private_edit = request(address, "GET", schema_path, &[], &[])?;
@@ -1119,6 +1094,13 @@ fn openapi_document_is_served_and_covers_every_public_operation() -> io::Result<
"{transport:?}: missing path {path}"
);
}
+ let upload_types = document["paths"]["/plugins"]["post"]["requestBody"]["content"]
+ .as_object()
+ .ok_or_else(|| io::Error::other("Plugin upload content types are missing"))?;
+ assert_eq!(
+ upload_types.keys().map(String::as_str).collect::<Vec<_>>(),
+ ["application/octet-stream"]
+ );
let document_item = &document["paths"]["/documents/{id}"];
assert!(
document_item["get"].is_object(),
@@ -1241,10 +1223,7 @@ fn validation_and_restart_recovery_are_publicly_observable() -> io::Result<()> {
address,
"POST",
"/plugins",
- &[(
- "Content-Type",
- "application/vnd.apache.tenon.plugin+tar+gzip",
- )],
+ &[("Content-Type", "application/octet-stream")],
&package,
)?;
assert_eq!(installed.status, 201, "{}", installed.body_text());
@@ -1319,7 +1298,7 @@ fn graceful_shutdown_drains_a_started_plugin_upload() -> io::Result<()> {
let mut stream = transport.connect(address)?;
write!(
stream,
- "POST /plugins HTTP/1.1\r\nHost: {address}\r\nConnection: close\r\nContent-Type: application/vnd.apache.tenon.plugin+tar+gzip\r\nContent-Length: {}\r\nExpect: 100-continue\r\n\r\n",
+ "POST /plugins HTTP/1.1\r\nHost: {address}\r\nConnection: close\r\nContent-Type: application/octet-stream\r\nContent-Length: {}\r\nExpect: 100-continue\r\n\r\n",
package.len()
)?;
stream.flush()?;
@@ -1360,7 +1339,7 @@ fn incomplete_plugin_bodies_never_publish_a_plugin() -> io::Result<()> {
let mut large = transport.connect(address)?;
write!(
large,
- "POST /plugins HTTP/1.1\r\nHost: {address}\r\nConnection: close\r\nContent-Type: application/vnd.apache.tenon.plugin+tar+gzip\r\nContent-Length: {}\r\nExpect: 100-continue\r\n\r\n",
+ "POST /plugins HTTP/1.1\r\nHost: {address}\r\nConnection: close\r\nContent-Type: application/octet-stream\r\nContent-Length: {}\r\nExpect: 100-continue\r\n\r\n",
64_u64 * 1024 * 1024 + 1,
)?;
large.flush()?;
@@ -1372,7 +1351,7 @@ fn incomplete_plugin_bodies_never_publish_a_plugin() -> io::Result<()> {
let mut truncated = transport.connect(address)?;
write!(
truncated,
- "POST /plugins HTTP/1.1\r\nHost: {address}\r\nConnection: close\r\nContent-Type: application/vnd.apache.tenon.plugin+tar+gzip\r\nContent-Length: {}\r\n\r\n",
+ "POST /plugins HTTP/1.1\r\nHost: {address}\r\nConnection: close\r\nContent-Type: application/octet-stream\r\nContent-Length: {}\r\n\r\n",
package.len() + 16,
)?;
truncated.write_all(&package)?;
@@ -1386,9 +1365,7 @@ fn incomplete_plugin_bodies_never_publish_a_plugin() -> io::Result<()> {
runner.terminate()?;
assert!(
- !directory
- .path()
- .join("plugins/programs/com.example.modbus/1.0.0")
+ !plugin_fixture::stored_package_path(directory.path(), "com.example.modbus", "1.0.0")
.exists()
);
}
@@ -1414,10 +1391,7 @@ fn plugin_staging_failure_stops_the_runner_without_publishing_a_program() -> io:
address,
"POST",
"/plugins",
- &[(
- "Content-Type",
- "application/vnd.apache.tenon.plugin+tar+gzip",
- )],
+ &[("Content-Type", "application/octet-stream")],
b"package",
);
@@ -1527,17 +1501,14 @@ fn large_plugin_package_installs_with_content_length_and_chunked_transfer() -> i
address,
"POST",
"/plugins",
- &[(
- "Content-Type",
- "application/vnd.apache.tenon.plugin+tar+gzip",
- )],
+ &[("Content-Type", "application/octet-stream")],
&package,
)?;
assert_eq!(response.status, 201, "{}", response.body_text());
let mut stream = transport.connect(address)?;
write!(
stream,
- "POST /plugins HTTP/1.1\r\nHost: {address}\r\nConnection: close\r\nContent-Type: application/vnd.apache.tenon.plugin+tar+gzip\r\nTransfer-Encoding: chunked\r\n\r\n",
+ "POST /plugins HTTP/1.1\r\nHost: {address}\r\nConnection: close\r\nContent-Type: application/octet-stream\r\nTransfer-Encoding: chunked\r\n\r\n",
)?;
for chunk in package.chunks(64 * 1024) {
write!(stream, "{:x}\r\n", chunk.len())?;
diff --git a/tests/runner_http/flow_contract.rs b/tests/runner_http/flow_contract.rs
index 44e6380..f9a4bd4 100644
--- a/tests/runner_http/flow_contract.rs
+++ b/tests/runner_http/flow_contract.rs
@@ -478,10 +478,7 @@ fn install_interfaces(address: std::net::SocketAddr) -> io::Result<()> {
address,
"POST",
"/plugins",
- &[(
- "Content-Type",
- "application/vnd.apache.tenon.plugin+tar+gzip",
- )],
+ &[("Content-Type", "application/octet-stream")],
&plugin_package(interface)?,
)?;
assert_eq!(response.status, 201, "{}", response.body_text());
diff --git a/tests/runner_java_archetype.rs b/tests/runner_java_archetype.rs
index 6c86ddd..d5e5479 100644
--- a/tests/runner_java_archetype.rs
+++ b/tests/runner_java_archetype.rs
@@ -233,10 +233,7 @@ fn run_generated_java_pipeline_with_path(
address,
"POST",
"/plugins",
- &[(
- "Content-Type",
- "application/vnd.apache.tenon.plugin+tar+gzip",
- )],
+ &[("Content-Type", "application/octet-stream")],
source_package,
)?;
assert_eq!(
@@ -250,10 +247,7 @@ fn run_generated_java_pipeline_with_path(
address,
"POST",
"/plugins",
- &[(
- "Content-Type",
- "application/vnd.apache.tenon.plugin+tar+gzip",
- )],
+ &[("Content-Type", "application/octet-stream")],
sink_package,
)?;
assert_eq!(installed_sink.status, 201, "{}", installed_sink.body_text());
@@ -1937,10 +1931,7 @@ fn verify_missing_external_java(source: &[u8], sink: &[u8]) -> io::Result<()> {
address,
"POST",
"/plugins",
- &[(
- "Content-Type",
- "application/vnd.apache.tenon.plugin+tar+gzip",
- )],
+ &[("Content-Type", "application/octet-stream")],
bundle,
)?;
assert_eq!(response.status, 201, "{}", response.body_text());
diff --git a/tests/runner_java_archetype/topology.rs b/tests/runner_java_archetype/topology.rs
index 576d35b..26a40f2 100644
--- a/tests/runner_java_archetype/topology.rs
+++ b/tests/runner_java_archetype/topology.rs
@@ -200,10 +200,7 @@ fn install_bundle(address: SocketAddr, bundle: &[u8]) -> io::Result<()> {
address,
"POST",
"/plugins",
- &[(
- "Content-Type",
- "application/vnd.apache.tenon.plugin+tar+gzip",
- )],
+ &[("Content-Type", "application/octet-stream")],
bundle,
)?;
assert_eq!(installed.status, 201, "{}", installed.body_text());
diff --git a/tests/runner_metrics.rs b/tests/runner_metrics.rs
index 74a0d3e..007ad99 100644
--- a/tests/runner_metrics.rs
+++ b/tests/runner_metrics.rs
@@ -19,8 +19,7 @@
//! Verifies HTTP metrics against real Runner and Plugin process behavior.
-#[path = "support/file_tree.rs"]
-mod file_tree;
+use plugin_fixture::file_tree;
#[path = "runner_cli/plugin_fixture.rs"]
mod plugin_fixture;
#[path = "support/runner_http.rs"]
@@ -93,10 +92,7 @@ fn real_core_snapshots_follow_configuration_and_actual_process_restarts() -> io:
address,
"POST",
"/plugins",
- &[(
- "Content-Type",
- "application/vnd.apache.tenon.plugin+tar+gzip",
- )],
+ &[("Content-Type", "application/octet-stream")],
&plugin_package(PluginInterface::SourceAndSink)?,
)?;
assert_eq!(installed.status, 201, "{}", installed.body_text());
diff --git a/tests/runner_metrics/flows.rs b/tests/runner_metrics/flows.rs
index e1a7b45..5414fd2 100644
--- a/tests/runner_metrics/flows.rs
+++ b/tests/runner_metrics/flows.rs
@@ -31,10 +31,7 @@ fn real_pipeline_flow_metrics_survive_replacement_and_retire_with_the_flow() ->
address,
"POST",
"/plugins",
- &[(
- "Content-Type",
- "application/vnd.apache.tenon.plugin+tar+gzip"
- )],
+ &[("Content-Type", "application/octet-stream")],
&plugin_package(PluginInterface::SourceAndSink)?
)?
.status,
diff --git a/tests/runner_metrics/performance.rs b/tests/runner_metrics/performance.rs
index 38bb863..d3d2000 100644
--- a/tests/runner_metrics/performance.rs
+++ b/tests/runner_metrics/performance.rs
@@ -63,10 +63,7 @@ fn measure(mode: Mode, round: usize) -> io::Result<serde_json::Value> {
address,
"POST",
"/plugins",
- &[(
- "Content-Type",
- "application/vnd.apache.tenon.plugin+tar+gzip"
- )],
+ &[("Content-Type", "application/octet-stream")],
&plugin_package(PluginInterface::SourceAndSink)?
)?
.status,
diff --git a/tests/runner_metrics/states.rs b/tests/runner_metrics/states.rs
index c138ca7..3b875d8 100644
--- a/tests/runner_metrics/states.rs
+++ b/tests/runner_metrics/states.rs
@@ -50,10 +50,7 @@ fn multiple_objects_report_transitions_and_only_successful_automatic_spawns() ->
address,
"POST",
"/plugins",
- &[(
- "Content-Type",
- "application/vnd.apache.tenon.plugin+tar+gzip"
- )],
+ &[("Content-Type", "application/octet-stream")],
&package
)?
.status,
@@ -194,9 +191,12 @@ fn multiple_objects_report_transitions_and_only_successful_automatic_spawns() ->
),
BTreeMap::from([("gateway".into(), 1.0)])
);
- let executable = directory
- .path()
- .join("plugins/programs/com.example.gateway/1.0.0/plugin.sh");
+ let executable = plugin_fixture::runtime_program_directory(
+ directory.path(),
+ "com.example.gateway",
+ "1.0.0",
+ )?
+ .join("plugin.sh");
fs::set_permissions(&executable, fs::Permissions::from_mode(0o600))?;
kill(read_pid(&alpha_instance)?)?;
wait_plugin_values(
diff --git a/tests/runner_resource_limits.rs b/tests/runner_resource_limits.rs
index 4892e28..4507225 100644
--- a/tests/runner_resource_limits.rs
+++ b/tests/runner_resource_limits.rs
@@ -19,8 +19,7 @@
//! Resource-limit behavior through real Runner HTTP and child processes.
-#[path = "support/file_tree.rs"]
-mod file_tree;
+use plugin_fixture::file_tree;
#[path = "runner_cli/plugin_fixture.rs"]
mod plugin_fixture;
#[path = "support/runner_http.rs"]
diff --git a/tests/runner_resource_limits/linux.rs b/tests/runner_resource_limits/linux.rs
index e0c06ac..765c0d9 100644
--- a/tests/runner_resource_limits/linux.rs
+++ b/tests/runner_resource_limits/linux.rs
@@ -156,10 +156,8 @@ fn quote(value: &str) -> String {
}
fn install_load(state: &Path, mode: &str, markers: &Path) -> io::Result<()> {
- use std::os::unix::fs::PermissionsExt as _;
- install_program(state, PluginInterface::SourceAndSink)?;
- let directory = state.join("plugins/programs/com.example.gateway/1.0.0");
- let mut manifest: Value = serde_json::from_slice(&fs::read(directory.join("manifest.json"))?)?;
+ let mut files = plugin_fixture::program_files(PluginInterface::SourceAndSink, None)?;
+ let mut manifest: Value = serde_json::from_slice(&files["manifest.json"])?;
let control = plugin_fixture::controlled_program_command(PluginInterface::SourceAndSink)?
.iter()
.map(|arg| quote(arg))
@@ -179,24 +177,15 @@ fn install_load(state: &Path, mode: &str, markers: &Path) -> io::Result<()> {
quote(&markers.to_string_lossy()),
quote(&child_command)
);
- let script_path = directory.join("resource-plugin.sh");
- fs::write(&script_path, script)?;
- fs::set_permissions(&script_path, fs::Permissions::from_mode(0o500))?;
+ files.insert("resource-plugin.sh".into(), script.into_bytes());
manifest["command"] = json!(["/bin/sh", "./resource-plugin.sh"]);
- // Installation is a fixture setup before the Runner establishes its Store.
- fs::set_permissions(
- directory.join("manifest.json"),
- fs::Permissions::from_mode(0o600),
- )?;
- fs::write(
- directory.join("manifest.json"),
- serde_json::to_vec(&manifest)?,
- )?;
- fs::set_permissions(
- directory.join("manifest.json"),
- fs::Permissions::from_mode(0o500),
- )?;
- Ok(())
+ files.insert("manifest.json".into(), serde_json::to_vec(&manifest)?);
+ plugin_fixture::write_package(
+ state,
+ "com.example.gateway",
+ "1.0.0",
+ &plugin_fixture::archive(files)?,
+ )
}
#[test]
diff --git a/tools/example/local.sh b/tools/example/local.sh
index 62b305f..1a39aa0 100755
--- a/tools/example/local.sh
+++ b/tools/example/local.sh
@@ -211,7 +211,7 @@ fi
printf 'Installing the generated and built-in Programs...\n'
for package in "$bundle" "${builtin_bundles[@]}"; do
curl --silent --show-error --fail -X POST "$api/plugins" \
- -H 'Content-Type: application/vnd.apache.tenon.plugin+tar+gzip' \
+ -H 'Content-Type: application/octet-stream' \
--data-binary "@$package" -o /dev/null
done
printf 'Saving the two Pipeline Documents...\n'
diff --git a/tools/verify-container-resource-limits.py b/tools/verify-container-resource-limits.py
index e188ae8..403e69a 100644
--- a/tools/verify-container-resource-limits.py
+++ b/tools/verify-container-resource-limits.py
@@ -86,7 +86,7 @@ def verify(image, fixtures, *, init, grant):
wait_for(lambda: request("GET", "/documents")[0] == 200)
status, _, body = request("POST", "/plugins", (fixtures / "plugin.tar.gz").read_bytes(),
- {"Content-Type": "application/vnd.apache.tenon.plugin+tar+gzip"})
+ {"Content-Type": "application/octet-stream"})
assert status == 201, (status, body)
document = json.loads((fixtures / "document.json").read_text())
diff --git a/tools/verify-debug-plugins.py b/tools/verify-debug-plugins.py
index 11b13f0..ca88b2a 100644
--- a/tools/verify-debug-plugins.py
+++ b/tools/verify-debug-plugins.py
@@ -188,7 +188,7 @@ end
for bundle in bundles:
for expected in [201, 204]:
status, _, body = request("POST", "/plugins", bundle.read_bytes(),
- {"Content-Type": "application/vnd.apache.tenon.plugin+tar+gzip"})
+ {"Content-Type": "application/octet-stream"})
assert status == expected, (status, body)
for identity, document in documents.items():
status, _, body = request("PUT", f"/documents/{identity}", json.dumps(document).encode(),
diff --git a/tools/verify-rust-scaffold.py b/tools/verify-rust-scaffold.py
index 8bbfd63..101d495 100644
--- a/tools/verify-rust-scaffold.py
+++ b/tools/verify-rust-scaffold.py
@@ -337,7 +337,7 @@ def verify_runner(runner, bundles, scenario, *, chain_programs=("source", "sink"
for bundle in bundles.values():
for expected in [201, 204]:
status, _, body = request("POST", "/plugins", bundle.read_bytes(),
- {"Content-Type": "application/vnd.apache.tenon.plugin+tar+gzip"})
+ {"Content-Type": "application/octet-stream"})
assert status == expected, (status, body)
for identity, source, sink in [("rust-chain", *chain_programs),
("rust-loop", "source-and-sink", "source-and-sink")]:
@@ -575,7 +575,7 @@ def verify_rejected_uploads(request, bundle):
for content, code in [(b"not gzip", "plugin_package_invalid"),
(foreign_bundle.getvalue(), "plugin_platform_mismatch")]:
status, _, body = request("POST", "/plugins", content,
- {"Content-Type": "application/vnd.apache.tenon.plugin+tar+gzip"})
+ {"Content-Type": "application/octet-stream"})
assert status == 422 and json.loads(body)["error"]["code"] == code, (status, body)
status, _, body = request("GET", "/plugins")
assert status == 200 and json.loads(body)["plugins"] == [], (status, body)