player_plugin_loader/
catalog.rs

1//! Metadata-only plugin catalog importing and indexing.
2//!
3//! This module is deliberately separate from [`crate::PluginRegistry`].  It
4//! validates manifest-owned values and artifact bytes, but it never opens a
5//! dynamic library, creates a WASM instance, or retains a runtime owner.
6
7use std::collections::BTreeMap;
8use std::fs::File;
9use std::io::{self, Read};
10use std::path::{Path, PathBuf};
11
12use player_plugin::{PluginCatalog, PluginCatalogError, PluginCatalogRecord};
13use sha2::{Digest, Sha256};
14use thiserror::Error;
15
16/// Upper bound for one streamed digest check.  Catalog import must remain
17/// bounded even when a package points at an unexpectedly large file.
18pub const MAX_PLUGIN_CATALOG_IMPORT_ARTIFACT_BYTES: u64 = 512 * 1024 * 1024;
19const DIGEST_BUFFER_BYTES: usize = 64 * 1024;
20
21/// A structured failure produced before a catalog candidate is committed.
22#[derive(Debug, Error)]
23pub enum PluginCatalogImportError {
24    #[error(transparent)]
25    Catalog(PluginCatalogError),
26    #[error(
27        "catalog import rejected duplicate identity `{identity}` from `{first_path}` and `{duplicate_path}`"
28    )]
29    DuplicateIdentity {
30        identity: String,
31        first_path: String,
32        duplicate_path: String,
33    },
34    #[error("failed to read plugin artifact `{path}`: {source}")]
35    ReadArtifact {
36        path: String,
37        #[source]
38        source: io::Error,
39    },
40    #[error("plugin artifact `{path}` is not a regular file")]
41    ArtifactNotFile { path: String },
42    #[error(
43        "plugin artifact `{path}` is {actual_bytes} bytes; maximum allowed for catalog import is {maximum_bytes}"
44    )]
45    ArtifactTooLarge {
46        path: String,
47        actual_bytes: u64,
48        maximum_bytes: u64,
49    },
50    #[error("stale plugin artifact digest for `{path}`: declared {expected}, actual {actual}")]
51    StaleDigest {
52        path: String,
53        expected: String,
54        actual: String,
55    },
56    #[error("catalog import path `{path}` is not valid: {message}")]
57    InvalidPath { path: String, message: String },
58}
59
60impl From<PluginCatalogError> for PluginCatalogImportError {
61    fn from(error: PluginCatalogError) -> Self {
62        match error {
63            PluginCatalogError::DuplicateIdentity {
64                identity,
65                first_path,
66                duplicate_path,
67            } => Self::DuplicateIdentity {
68                identity,
69                first_path,
70                duplicate_path,
71            },
72            other => Self::Catalog(other),
73        }
74    }
75}
76
77/// A read-only, deterministic index over a validated catalog.
78///
79/// The index stores only offsets into the value-owned [`PluginCatalog`].  No
80/// file descriptor, library handle, worker, queue, or media buffer can enter
81/// this type.
82#[derive(Debug, Clone, PartialEq, Eq)]
83pub struct PluginCatalogIndex {
84    catalog: PluginCatalog,
85    identity_index: BTreeMap<String, usize>,
86    plugin_index: BTreeMap<String, Vec<usize>>,
87}
88
89impl Default for PluginCatalogIndex {
90    fn default() -> Self {
91        Self {
92            catalog: PluginCatalog::empty(),
93            identity_index: BTreeMap::new(),
94            plugin_index: BTreeMap::new(),
95        }
96    }
97}
98
99impl PluginCatalogIndex {
100    /// Builds an index from a validated catalog without touching artifact paths.
101    pub fn from_catalog(catalog: PluginCatalog) -> Result<Self, PluginCatalogImportError> {
102        let mut identity_index = BTreeMap::new();
103        let mut plugin_index: BTreeMap<String, Vec<usize>> = BTreeMap::new();
104        for (index, record) in catalog.records().iter().enumerate() {
105            let canonical_identity = record.canonical_identity_key();
106            if let Some(first_index) = identity_index.insert(canonical_identity, index) {
107                let first_path = catalog.records()[first_index].artifact_path().to_owned();
108                return Err(PluginCatalogImportError::DuplicateIdentity {
109                    identity: record.identity_key(),
110                    first_path,
111                    duplicate_path: record.artifact_path().to_owned(),
112                });
113            }
114            plugin_index
115                .entry(record.descriptor().plugin_id.clone())
116                .or_default()
117                .push(index);
118        }
119        Ok(Self {
120            catalog,
121            identity_index,
122            plugin_index,
123        })
124    }
125
126    pub fn catalog(&self) -> &PluginCatalog {
127        &self.catalog
128    }
129
130    pub fn records(&self) -> &[PluginCatalogRecord] {
131        self.catalog.records()
132    }
133
134    pub fn len(&self) -> usize {
135        self.catalog.len()
136    }
137
138    pub fn is_empty(&self) -> bool {
139        self.catalog.is_empty()
140    }
141
142    pub fn fingerprint(&self) -> &str {
143        self.catalog.fingerprint()
144    }
145
146    pub fn get(&self, identity: &str) -> Option<&PluginCatalogRecord> {
147        self.identity_index
148            .get(identity)
149            .and_then(|index| self.catalog.records().get(*index))
150    }
151
152    pub fn find(&self, plugin_id: &str) -> impl Iterator<Item = &PluginCatalogRecord> {
153        self.plugin_index
154            .get(plugin_id)
155            .into_iter()
156            .flat_map(|indices| indices.iter())
157            .filter_map(|index| self.catalog.records().get(*index))
158    }
159}
160
161/// Transactional importer for catalog records.
162///
163/// Each mutation builds a complete candidate catalog first.  If validation or
164/// digest verification fails, the existing index remains byte-for-byte
165/// unchanged and no runtime owner has been created.
166#[derive(Debug, Clone, Default)]
167pub struct PluginCatalogImporter {
168    index: PluginCatalogIndex,
169}
170
171impl PluginCatalogImporter {
172    pub fn new() -> Self {
173        Self::default()
174    }
175
176    pub fn from_catalog(catalog: PluginCatalog) -> Result<Self, PluginCatalogImportError> {
177        Ok(Self {
178            index: PluginCatalogIndex::from_catalog(catalog)?,
179        })
180    }
181
182    /// Creates an immutable index in one step from catalog records.
183    pub fn import(
184        records: impl IntoIterator<Item = PluginCatalogRecord>,
185    ) -> Result<PluginCatalogIndex, PluginCatalogImportError> {
186        let catalog =
187            PluginCatalog::from_records(records).map_err(PluginCatalogImportError::from)?;
188        PluginCatalogIndex::from_catalog(catalog)
189    }
190
191    /// Creates an immutable index from a canonical catalog JSON snapshot.
192    pub fn import_json(bytes: &[u8]) -> Result<PluginCatalogIndex, PluginCatalogImportError> {
193        let catalog = PluginCatalog::from_json(bytes).map_err(PluginCatalogImportError::from)?;
194        PluginCatalogIndex::from_catalog(catalog)
195    }
196
197    pub fn index(&self) -> &PluginCatalogIndex {
198        &self.index
199    }
200
201    pub fn into_index(self) -> PluginCatalogIndex {
202        self.index
203    }
204
205    /// Imports metadata without reading the artifact path.
206    pub fn import_record(
207        &mut self,
208        record: PluginCatalogRecord,
209    ) -> Result<(), PluginCatalogImportError> {
210        self.commit_records(std::iter::once(record))
211    }
212
213    /// Imports a batch atomically.  A single invalid or duplicate record
214    /// leaves the previous index untouched.
215    pub fn import_records(
216        &mut self,
217        records: impl IntoIterator<Item = PluginCatalogRecord>,
218    ) -> Result<(), PluginCatalogImportError> {
219        self.commit_records(records)
220    }
221
222    pub fn import_json_into(&mut self, bytes: &[u8]) -> Result<(), PluginCatalogImportError> {
223        let catalog = PluginCatalog::from_json(bytes).map_err(PluginCatalogImportError::from)?;
224        self.commit_records(catalog.records().iter().cloned())
225    }
226
227    /// Verifies one artifact's bytes and commits its metadata only after the
228    /// digest matches.  The file is streamed and never loaded as a runtime.
229    pub fn import_record_at(
230        &mut self,
231        record: PluginCatalogRecord,
232        path: impl AsRef<Path>,
233    ) -> Result<(), PluginCatalogImportError> {
234        let path = path.as_ref();
235        record.validate().map_err(PluginCatalogImportError::from)?;
236        verify_artifact_digest(&record, path)?;
237        self.import_record(record)
238    }
239
240    /// Resolves a record path against a package/install root, verifies its
241    /// digest, and commits it atomically. Absolute installed paths are kept.
242    pub fn import_record_from_root(
243        &mut self,
244        record: PluginCatalogRecord,
245        root: impl AsRef<Path>,
246    ) -> Result<(), PluginCatalogImportError> {
247        let path = resolve_record_path(&record, root.as_ref())?;
248        self.import_record_at(record, path)
249    }
250
251    pub fn import_records_from_root(
252        &mut self,
253        records: impl IntoIterator<Item = PluginCatalogRecord>,
254        root: impl AsRef<Path>,
255    ) -> Result<(), PluginCatalogImportError> {
256        let root = root.as_ref();
257        let records = records.into_iter().collect::<Vec<_>>();
258        for record in &records {
259            record.validate().map_err(PluginCatalogImportError::from)?;
260            let path = resolve_record_path(record, root)?;
261            verify_artifact_digest(record, &path)?;
262        }
263        self.commit_records(records)
264    }
265
266    fn commit_records(
267        &mut self,
268        records: impl IntoIterator<Item = PluginCatalogRecord>,
269    ) -> Result<(), PluginCatalogImportError> {
270        let mut candidate = self.index.records().to_vec();
271        candidate.extend(records);
272        let catalog =
273            PluginCatalog::from_records(candidate).map_err(PluginCatalogImportError::from)?;
274        let index = PluginCatalogIndex::from_catalog(catalog)?;
275        self.index = index;
276        Ok(())
277    }
278}
279
280fn resolve_record_path(
281    record: &PluginCatalogRecord,
282    root: &Path,
283) -> Result<PathBuf, PluginCatalogImportError> {
284    if root.as_os_str().is_empty() {
285        return Err(PluginCatalogImportError::InvalidPath {
286            path: record.artifact_path().to_owned(),
287            message: "root path must not be empty".to_owned(),
288        });
289    }
290    let path = Path::new(record.artifact_path());
291    if path.is_absolute() {
292        Ok(path.to_path_buf())
293    } else {
294        Ok(root.join(path))
295    }
296}
297
298fn verify_artifact_digest(
299    record: &PluginCatalogRecord,
300    path: &Path,
301) -> Result<(), PluginCatalogImportError> {
302    let path_string = path.display().to_string();
303    let metadata =
304        std::fs::metadata(path).map_err(|source| PluginCatalogImportError::ReadArtifact {
305            path: path_string.clone(),
306            source,
307        })?;
308    if !metadata.is_file() {
309        return Err(PluginCatalogImportError::ArtifactNotFile { path: path_string });
310    }
311    if metadata.len() > MAX_PLUGIN_CATALOG_IMPORT_ARTIFACT_BYTES {
312        return Err(PluginCatalogImportError::ArtifactTooLarge {
313            path: path_string,
314            actual_bytes: metadata.len(),
315            maximum_bytes: MAX_PLUGIN_CATALOG_IMPORT_ARTIFACT_BYTES,
316        });
317    }
318
319    let mut file = File::open(path).map_err(|source| PluginCatalogImportError::ReadArtifact {
320        path: path.display().to_string(),
321        source,
322    })?;
323    let mut hasher = Sha256::new();
324    let mut buffer = [0_u8; DIGEST_BUFFER_BYTES];
325    let mut total_read = 0_u64;
326    loop {
327        let read =
328            file.read(&mut buffer)
329                .map_err(|source| PluginCatalogImportError::ReadArtifact {
330                    path: path.display().to_string(),
331                    source,
332                })?;
333        if read == 0 {
334            break;
335        }
336        total_read = total_read.saturating_add(read as u64);
337        if total_read > MAX_PLUGIN_CATALOG_IMPORT_ARTIFACT_BYTES {
338            return Err(PluginCatalogImportError::ArtifactTooLarge {
339                path: path.display().to_string(),
340                actual_bytes: total_read,
341                maximum_bytes: MAX_PLUGIN_CATALOG_IMPORT_ARTIFACT_BYTES,
342            });
343        }
344        hasher.update(&buffer[..read]);
345    }
346    let actual = hex::encode(hasher.finalize());
347    if actual != record.artifact_sha256() {
348        return Err(PluginCatalogImportError::StaleDigest {
349            path: path.display().to_string(),
350            expected: record.artifact_sha256().to_owned(),
351            actual,
352        });
353    }
354    Ok(())
355}
356
357#[cfg(test)]
358mod tests {
359    use super::*;
360    use player_plugin::{
361        PLUGIN_CATALOG_MIGRATION_VERSION, PLUGIN_CATALOG_SCHEMA_VERSION, PluginArtifactCapability,
362        PluginArtifactDescriptor, PluginArtifactFormat, PluginArtifactTransport,
363        PluginCatalogSource, PluginResourcePolicy,
364    };
365
366    fn descriptor(plugin_id: &str, instance_id: &str) -> PluginArtifactDescriptor {
367        PluginArtifactDescriptor {
368            schema_version: PLUGIN_CATALOG_SCHEMA_VERSION,
369            plugin_id: plugin_id.to_owned(),
370            version: "1.0.0".to_owned(),
371            publisher: "dev.vesper.publisher".to_owned(),
372            transport: PluginArtifactTransport::Native,
373            target: "aarch64-apple-darwin".to_owned(),
374            format: PluginArtifactFormat::Dylib,
375            architecture: "arm64".to_owned(),
376            abi_major: 1,
377            abi_minor_min: 0,
378            abi_minor_max: 0,
379            capabilities: vec![PluginArtifactCapability {
380                interface_id: "e9479dbc-42d2-575e-b39e-a24bc512fbc7".to_owned(),
381                instance_id: instance_id.to_owned(),
382            }],
383            requires: Vec::new(),
384            provides: Vec::new(),
385            runtime_dependencies: Vec::new(),
386            resource_policy: PluginResourcePolicy::default(),
387            migration_version: PLUGIN_CATALOG_MIGRATION_VERSION.to_owned(),
388        }
389    }
390
391    fn record(plugin_id: &str, path: &str) -> PluginCatalogRecord {
392        record_with_digest(
393            plugin_id,
394            path,
395            "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
396        )
397    }
398
399    fn record_with_digest(plugin_id: &str, path: &str, digest: &str) -> PluginCatalogRecord {
400        PluginCatalogRecord::new(
401            descriptor(plugin_id, &format!("{plugin_id}.primary")),
402            path,
403            digest,
404            PluginCatalogSource::Development,
405        )
406        .expect("fixture record")
407    }
408
409    #[test]
410    fn failed_duplicate_import_keeps_the_previous_index() {
411        let mut importer = PluginCatalogImporter::new();
412        importer
413            .import_record(record("dev.vesper.one", "/tmp/one"))
414            .expect("first record");
415        let before = importer.index().fingerprint().to_owned();
416        let error = importer
417            .import_record(record("dev.vesper.one", "/tmp/duplicate"))
418            .expect_err("duplicate identity");
419        assert!(matches!(
420            error,
421            PluginCatalogImportError::DuplicateIdentity { .. }
422        ));
423        assert_eq!(before, importer.index().fingerprint());
424        assert_eq!(importer.index().len(), 1);
425    }
426
427    #[test]
428    fn json_import_rebuilds_an_immutable_index() {
429        let record = record("dev.vesper.one", "/tmp/one");
430        let catalog = PluginCatalog::from_records([record]).expect("catalog");
431        let bytes = catalog.to_json().expect("json");
432        let index = PluginCatalogImporter::import_json(&bytes).expect("index");
433        assert_eq!(index.fingerprint(), catalog.fingerprint());
434        assert_eq!(index.find("dev.vesper.one").count(), 1);
435    }
436
437    #[test]
438    fn verified_artifact_import_commits_only_after_digest_match() {
439        let directory = tempfile::tempdir().expect("temporary directory");
440        let path = directory.path().join("fixture.bin");
441        let bytes = b"catalog importer fixture";
442        std::fs::write(&path, bytes).expect("artifact bytes");
443        let digest = hex::encode(Sha256::digest(bytes));
444        let mut importer = PluginCatalogImporter::new();
445
446        importer
447            .import_record_at(
448                record_with_digest("dev.vesper.one", &path.to_string_lossy(), &digest),
449                &path,
450            )
451            .expect("matching digest");
452        let before = importer.index().fingerprint().to_owned();
453        let error = importer
454            .import_record_at(
455                PluginCatalogRecord::new(
456                    descriptor("dev.vesper.two", "dev.vesper.two.primary"),
457                    path.to_string_lossy(),
458                    "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb",
459                    PluginCatalogSource::Development,
460                )
461                .expect("stale record"),
462                &path,
463            )
464            .expect_err("stale digest");
465        assert!(matches!(
466            error,
467            PluginCatalogImportError::StaleDigest { .. }
468        ));
469        assert_eq!(before, importer.index().fingerprint());
470        assert_eq!(importer.index().len(), 1);
471    }
472}
player_plugin_loader/catalog.rs

player_plugin_loader/
catalog.rs
