diff --git a/app/services/plugin_artifacts.py b/app/services/plugin_artifacts.py new file mode 100644 index 0000000..5b76ca9 --- /dev/null +++ b/app/services/plugin_artifacts.py @@ -0,0 +1,243 @@ +"""Plugin artifact integrity verification (#384). + +A marketplace artifact is only trustworthy if the bytes downloaded match the +sha256 digest recorded in the catalog. This module provides the single +verification path used before an artifact is registered: + +1. normalize the catalog digest (accepting ``sha256:`` / ``sha256-`` prefixes + and bare hex); +2. hash the artifact (streaming for files, never loading a whole package into + memory); +3. compare the two in constant time with :func:`hmac.compare_digest`; +4. on mismatch, record a bounded audit event (when a workspace context is + supplied) and raise :class:`ArtifactIntegrityError`. + +:func:`verify_and_register_artifact` makes the ordering explicit: the +registration callback is invoked **only after** verification succeeds, so a +tampered artifact can never reach the registry. + +This module does not download anything and does not execute plugin code. +""" + +from __future__ import annotations + +import hashlib +import hmac +import logging +import os +import re +from collections.abc import Callable +from pathlib import Path +from typing import TYPE_CHECKING, Any, BinaryIO + +from app.services.plugins import PluginError + +if TYPE_CHECKING: # pragma: no cover - typing only + from app.models.user import User + +logger = logging.getLogger(__name__) + +#: Read size used when hashing a file or stream. +CHUNK_SIZE = 1024 * 1024 + +#: Accepted sha256 spellings: ``sha256:``, ``sha256-``, and bare hex. +_SHA256_RE = re.compile(r"^(?:sha256[:=-])?([0-9a-fA-F]{64})$", re.IGNORECASE) + +#: A filesystem path, in-memory bytes, or a binary file-like object. +ArtifactSource = str | os.PathLike[str] | bytes | bytearray | memoryview | BinaryIO + + +class ArtifactError(PluginError): + """A plugin artifact could not be read or validated.""" + + +class ArtifactIntegrityError(ArtifactError): + """An artifact's sha256 digest does not match the catalog value.""" + + def __init__(self, expected: str, actual: str, *, plugin_id: str | None = None) -> None: + label = f" for plugin {plugin_id}" if plugin_id else "" + super().__init__(f"Artifact sha256 mismatch{label}: expected {expected}, got {actual}") + self.expected = expected + self.actual = actual + self.plugin_id = plugin_id + + +def normalize_sha256(value: str) -> str: + """Normalize a sha256 digest to lowercase 64-character hex. + + Raises: + ArtifactError: If ``value`` is not a valid sha256 hex digest. + """ + if not isinstance(value, str): + raise ArtifactError(f"Invalid sha256 digest: {value!r}") + + text = value.strip() + match = _SHA256_RE.match(text) + if match is None: + algorithm = text.split(":", 1)[0].lower() if ":" in text else "" + if algorithm and algorithm != "sha256": + raise ArtifactError(f"Unsupported artifact integrity algorithm: {algorithm}") + raise ArtifactError(f"Invalid sha256 digest: {value!r}") + return match.group(1).lower() + + +def _update_from_stream(digest: Any, stream: BinaryIO, chunk_size: int) -> None: + """Feed a binary stream into ``digest`` without reading it all at once.""" + while True: + chunk = stream.read(chunk_size) + if not chunk: + break + if not isinstance(chunk, (bytes, bytearray, memoryview)): + raise ArtifactError("Artifact stream must be opened in binary mode") + digest.update(chunk) + + +def compute_sha256(source: ArtifactSource, *, chunk_size: int = CHUNK_SIZE) -> str: + """Return the lowercase hex sha256 digest of ``source``. + + ``source`` may be a path (``str``/``os.PathLike``), in-memory bytes, or a + readable binary stream. Files are streamed in ``chunk_size`` blocks. + + Raises: + ArtifactError: If the artifact cannot be read or ``chunk_size`` is not + positive. + """ + if not isinstance(chunk_size, int) or chunk_size <= 0: + raise ArtifactError("Artifact chunk size must be a positive integer") + + digest = hashlib.sha256() + + if isinstance(source, (bytes, bytearray, memoryview)): + digest.update(source) + return digest.hexdigest() + + if hasattr(source, "read"): + try: + _update_from_stream(digest, source, chunk_size) # type: ignore[arg-type] + except OSError as exc: + raise ArtifactError(f"Cannot read artifact stream: {exc}") from exc + return digest.hexdigest() + + if isinstance(source, (str, os.PathLike)): + path = Path(source) + try: + with path.open("rb") as handle: + _update_from_stream(digest, handle, chunk_size) + except OSError as exc: + raise ArtifactError(f"Cannot read artifact file {path}: {exc}") from exc + return digest.hexdigest() + + raise ArtifactError(f"Unsupported artifact source: {type(source).__name__}") + + +def _record_mismatch( + *, + plugin_id: str | None, + workspace_id: int | None, + actor: User | None, +) -> None: + """Best-effort bounded audit event for a refused artifact. + + The integrity error is the security-relevant outcome; a failure to write the + audit row must not mask it, so recording failures are logged and swallowed. + Outside an application context (for example a unit test or CLI helper with + no workspace) there is nowhere to persist the event and recording is a + no-op. + """ + if workspace_id is None: + return + + try: + from flask import has_app_context + + if not has_app_context(): + return + + from app.services.plugin_audit import record_plugin_audit + + record_plugin_audit( + workspace_id, + "denied", + actor=actor, + plugin_id=plugin_id, + outcome="denied", + reason="artifact integrity mismatch", + ) + except Exception: # pragma: no cover - audit is best effort + logger.warning("Could not record artifact integrity audit event", exc_info=True) + + +def verify_artifact_integrity( + source: ArtifactSource, + expected_sha256: str, + *, + plugin_id: str | None = None, + workspace_id: int | None = None, + actor: User | None = None, +) -> str: + """Verify ``source`` against ``expected_sha256`` in constant time. + + Returns the artifact's computed (normalized) digest on success. On mismatch + the catalog value is recorded as a plugin audit denial for ``workspace_id`` + (when provided) and :class:`ArtifactIntegrityError` is raised, so the caller + must not register the artifact. + + Args: + source: Artifact path, bytes, or binary stream. + expected_sha256: Catalog digest (``sha256:``, ``sha256-``, or + bare hex). + plugin_id: Plugin being installed, for the error/audit facts. + workspace_id: Workspace performing the install, for the audit event. + actor: User performing the install, for the audit event. + + Raises: + ArtifactIntegrityError: If the hashes do not match. + ArtifactError: If the artifact or expected digest cannot be read/parsed. + """ + expected = normalize_sha256(expected_sha256) + actual = compute_sha256(source) + + if not hmac.compare_digest(actual, expected): + _record_mismatch(plugin_id=plugin_id, workspace_id=workspace_id, actor=actor) + raise ArtifactIntegrityError(expected, actual, plugin_id=plugin_id) + + return actual + + +def verify_and_register_artifact[T]( + source: ArtifactSource, + expected_sha256: str, + register: Callable[[str], T], + *, + plugin_id: str | None = None, + workspace_id: int | None = None, + actor: User | None = None, +) -> T: + """Verify an artifact and only then invoke ``register``. + + ``register`` receives the verified digest and is never called for a hash + mismatch, which keeps a tampered artifact out of the registry. + + Returns: + The value returned by ``register``. + """ + digest = verify_artifact_integrity( + source, + expected_sha256, + plugin_id=plugin_id, + workspace_id=workspace_id, + actor=actor, + ) + return register(digest) + + +__all__ = [ + "CHUNK_SIZE", + "ArtifactError", + "ArtifactIntegrityError", + "ArtifactSource", + "compute_sha256", + "normalize_sha256", + "verify_and_register_artifact", + "verify_artifact_integrity", +] diff --git a/app/services/plugin_compat.py b/app/services/plugin_compat.py index 1a454e7..9719bc9 100644 --- a/app/services/plugin_compat.py +++ b/app/services/plugin_compat.py @@ -28,6 +28,8 @@ from packaging.specifiers import InvalidSpecifier, SpecifierSet +from app.services.plugin_versions import InvalidPluginVersionError, parse_version + logger = logging.getLogger(__name__) #: Value used when the application version cannot be determined. Kept as a @@ -101,11 +103,11 @@ def plugin_compatible(compatibility: str | None, version: str | None = None) -> """ try: specifier = parse_compatibility(compatibility) - target = (version or "").strip() or app_version() + target = parse_version((version or "").strip() or app_version()) return specifier.contains(target, prereleases=True) - except InvalidCompatibilityError: + except (InvalidCompatibilityError, InvalidPluginVersionError): return False - except ValueError: # unparseable application version -> not compatible + except ValueError: # additional defensive fallback -> not compatible return False diff --git a/app/services/plugin_deps.py b/app/services/plugin_deps.py index 5dd3349..0ea060a 100644 --- a/app/services/plugin_deps.py +++ b/app/services/plugin_deps.py @@ -34,7 +34,8 @@ from importlib import metadata from packaging.requirements import InvalidRequirement, Requirement -from packaging.version import InvalidVersion, Version + +from app.services.plugin_versions import InvalidPluginVersionError, parse_version logger = logging.getLogger(__name__) @@ -203,8 +204,8 @@ def _check_package_dependency( ) from None try: - version = Version(installed) - except InvalidVersion: + version = parse_version(installed) + except InvalidPluginVersionError: raise PluginDependencyVersionError( root_id, f"dependency '{requirement}' cannot be checked against " diff --git a/app/services/plugin_lockfile.py b/app/services/plugin_lockfile.py new file mode 100644 index 0000000..618e906 --- /dev/null +++ b/app/services/plugin_lockfile.py @@ -0,0 +1,266 @@ +"""Deterministic plugin lockfile serialization (#398). + +A plugin lockfile records the exact versions, sources, and artifact hashes a +workspace resolved so an install can be reproduced. That only helps if the file +is stable: resolving the same set of plugins twice must produce byte-identical +output, otherwise every run creates a spurious diff. + +This module therefore defines one canonical serialization: + +* entries are sorted by plugin id (the order they were resolved in is ignored); +* versions are normalized to their canonical PEP 440 spelling; +* sha256 integrity hashes are normalized to ``sha256:``; +* dependency requirement strings are sorted; +* JSON object keys are sorted and the document ends with exactly one ``\\n``; +* volatile fields (such as a per-resolution ``resolved_at`` timestamp) are not + written, because they would defeat reproducibility. + +Callers can pass :class:`LockfileEntry` objects or mappings/objects with the +matching fields, so the serializer works with the persisted lock model and with +plain dictionaries from a resolver:: + + dumps_lockfile([ + {"plugin_id": "b", "version": "1.0", "source": "catalog", "integrity": "..."}, + ]) + +Use :func:`write_lockfile` to write the canonical bytes atomically. +""" + +from __future__ import annotations + +import json +import os +import re +import tempfile +from collections.abc import Iterable, Mapping +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +from packaging.version import InvalidVersion, Version + +#: Format version written at the top of every lockfile. +LOCKFILE_VERSION = 1 + +#: Accepted sha256 spellings: ``sha256:``, ``sha256-``, and bare hex. +_SHA256_RE = re.compile(r"^(?:sha256[:=-])?([0-9a-fA-F]{64})$", re.IGNORECASE) + +#: Mapping field aliases accepted for the integrity hash. +_INTEGRITY_ALIASES = ("integrity", "integrity_hash", "artifact_hash", "hash") + + +class LockfileError(ValueError): + """Raised when lockfile entries cannot be serialized deterministically.""" + + +@dataclass(frozen=True) +class LockfileEntry: + """One resolved plugin entry in a lockfile. + + ``resolved_at`` is intentionally absent: a timestamp changes between runs + and would make the lockfile non-deterministic. Callers that persist it in a + database row should keep it out of the serialized output. + """ + + plugin_id: str + version: str + source: str + integrity: str + dependencies: tuple[str, ...] = field(default_factory=tuple) + + def to_dict(self) -> dict[str, Any]: + """Return the canonical mapping written to the lockfile.""" + return { + "plugin_id": self.plugin_id, + "version": self.version, + "source": self.source, + "integrity": self.integrity, + "dependencies": list(self.dependencies), + } + + +def normalize_integrity(value: str) -> str: + """Normalize a sha256 integrity hash to ``sha256:``. + + Accepts ``sha256:`` / ``sha256-`` prefixes and bare uppercase hex so hashes + copied from a catalog, a checksum file, or a previous lockfile all converge + on the same output. + + Raises: + LockfileError: If the value is not a 64-character sha256 hex digest. + """ + if not isinstance(value, str): + raise LockfileError(f"Invalid integrity hash: {value!r}") + + text = value.strip() + match = _SHA256_RE.match(text) + if match is None: + algorithm = text.split(":", 1)[0].lower() if ":" in text else "" + if algorithm and algorithm != "sha256": + raise LockfileError(f"Unsupported integrity algorithm: {algorithm}") + raise LockfileError(f"Invalid sha256 integrity hash: {value!r}") + return f"sha256:{match.group(1).lower()}" + + +def _normalize_version(value: str) -> str: + """Return the canonical PEP 440 spelling of a plugin version.""" + if not isinstance(value, str) or not value.strip(): + raise LockfileError(f"Invalid plugin version: {value!r}") + try: + return str(Version(value.strip())) + except InvalidVersion as exc: + raise LockfileError(f"Invalid plugin version: {value!r}") from exc + + +def _required_text(entry: Mapping[str, Any] | object, field_name: str) -> str: + """Return a required, non-empty string field from a mapping or object.""" + if isinstance(entry, Mapping): + value = entry.get(field_name) + else: + value = getattr(entry, field_name, None) + if not isinstance(value, str) or not value.strip(): + raise LockfileError(f"Invalid lockfile entry: {field_name} is required") + return value.strip() + + +def _entry_integrity(entry: Mapping[str, Any] | object) -> str: + """Return the first integrity alias present on a mapping or object.""" + for alias in _INTEGRITY_ALIASES: + value = entry.get(alias) if isinstance(entry, Mapping) else getattr(entry, alias, None) + if value is not None: + return normalize_integrity(value) + raise LockfileError("Invalid lockfile entry: integrity is required") + + +def _entry_dependencies(entry: Mapping[str, Any] | object) -> tuple[str, ...]: + """Return the sorted dependency requirement strings for an entry.""" + if isinstance(entry, Mapping): + raw = entry.get("dependencies", ()) + else: + raw = getattr(entry, "dependencies", ()) + if raw is None: + return () + if isinstance(raw, str): + raise LockfileError("Invalid lockfile entry: dependencies must be a list") + try: + items = list(raw) + except TypeError as exc: + raise LockfileError("Invalid lockfile entry: dependencies must be a list") from exc + if any(not isinstance(item, str) or not item.strip() for item in items): + raise LockfileError("Invalid lockfile entry: dependencies must be non-empty strings") + return tuple(sorted(item.strip() for item in items)) + + +def _coerce_entry(entry: LockfileEntry | Mapping[str, Any] | object) -> LockfileEntry: + """Normalize an entry object/mapping into a :class:`LockfileEntry`.""" + if isinstance(entry, LockfileEntry): + plugin_id = _required_text(entry, "plugin_id") + version = _normalize_version(entry.version) + source = _required_text(entry, "source") + integrity = normalize_integrity(entry.integrity) + dependencies = _entry_dependencies(entry) + else: + plugin_id = _required_text(entry, "plugin_id") + version = _normalize_version(_required_text(entry, "version")) + source = _required_text(entry, "source") + integrity = _entry_integrity(entry) + dependencies = _entry_dependencies(entry) + + return LockfileEntry( + plugin_id=plugin_id, + version=version, + source=source, + integrity=integrity, + dependencies=dependencies, + ) + + +def build_lockfile( + entries: Iterable[LockfileEntry | Mapping[str, Any] | object], +) -> dict[str, Any]: + """Build the canonical lockfile document from ``entries``. + + Entries are normalized, checked for duplicate plugin ids, and sorted by + plugin id. The returned mapping is plain JSON-serializable data (no model + objects or timestamps). + + Raises: + LockfileError: For duplicate ids or invalid entry fields. + """ + normalized = [_coerce_entry(entry) for entry in entries] + normalized.sort(key=lambda item: item.plugin_id) + + seen: set[str] = set() + for entry in normalized: + if entry.plugin_id in seen: + raise LockfileError(f"Duplicate lockfile entry for plugin: {entry.plugin_id}") + seen.add(entry.plugin_id) + + return { + "lockfile_version": LOCKFILE_VERSION, + "plugins": [entry.to_dict() for entry in normalized], + } + + +def dumps_lockfile(entries: Iterable[LockfileEntry | Mapping[str, Any] | object]) -> str: + """Serialize ``entries`` to the canonical, byte-stable lockfile string. + + The output uses sorted keys, two-space indentation, UTF-8 text, and a single + trailing newline. Sorting and normalization happen before serialization, so + the same entries always produce the same bytes regardless of input order. + """ + document = build_lockfile(entries) + return ( + json.dumps( + document, + indent=2, + sort_keys=True, + ensure_ascii=False, + ) + + "\n" + ) + + +def write_lockfile( + path: str | Path, + entries: Iterable[LockfileEntry | Mapping[str, Any] | object], +) -> Path: + """Write the canonical lockfile to ``path`` atomically. + + The file is written to a temporary file in the destination directory and + moved into place with :func:`os.replace`, so readers never observe a + partially written lockfile. Newlines are forced to ``\\n`` on every + platform to avoid line-ending churn in version control. + + Returns: + The destination path. + """ + destination = Path(path) + destination.parent.mkdir(parents=True, exist_ok=True) + contents = dumps_lockfile(entries) + + fd, temporary = tempfile.mkstemp( + prefix=f".{destination.name}.", + suffix=".tmp", + dir=destination.parent, + ) + try: + with os.fdopen(fd, "w", encoding="utf-8", newline="\n") as handle: + handle.write(contents) + os.replace(temporary, destination) + except Exception: + if os.path.exists(temporary): + os.unlink(temporary) + raise + return destination + + +__all__ = [ + "LOCKFILE_VERSION", + "LockfileEntry", + "LockfileError", + "build_lockfile", + "dumps_lockfile", + "normalize_integrity", + "write_lockfile", +] diff --git a/app/services/plugin_versions.py b/app/services/plugin_versions.py new file mode 100644 index 0000000..3925aed --- /dev/null +++ b/app/services/plugin_versions.py @@ -0,0 +1,158 @@ +"""PEP 440 plugin version comparison helpers (#399). + +Plugin versions are PEP 440 versions (the manifest already validates them as +semantic versions, which are a subset of PEP 440). This module wraps +:mod:`packaging.version` behind a small, uniform API so update checks, +dependency checks, and lockfile ordering compare versions the same way and +report invalid values consistently instead of raising bare ``packaging`` +errors. + +The helpers are intentionally side-effect free and never coerce a bad value +into a "comparable" one: an invalid version raises +:class:`InvalidPluginVersionError` (or is reported as invalid by +:func:`is_valid_version`). Callers on a security-sensitive path should fail +closed rather than treating an unparseable version as newest. + +Examples: + >>> from app.services.plugin_versions import compare_versions, sort_versions + >>> compare_versions("1.2.0", "1.10.0") + -1 + >>> sort_versions(["2.0.0", "1.0.0", "1.5.0"]) + ['1.0.0', '1.5.0', '2.0.0'] +""" + +from __future__ import annotations + +from collections.abc import Iterable + +from packaging.version import InvalidVersion, Version + +#: Value returned by :func:`compare_versions` when the versions are equal under +#: PEP 440 ordering (e.g. ``1.0`` and ``1.0.0``). +EQUAL = 0 +#: ``left`` sorts before (is older than) ``right``. +OLDER = -1 +#: ``left`` sorts after (is newer than) ``right``. +NEWER = 1 + + +class InvalidPluginVersionError(ValueError): + """Raised when a value is not a valid PEP 440 version.""" + + +def parse_version(value: str | Version) -> Version: + """Parse ``value`` into a :class:`packaging.version.Version`. + + Strings are stripped before parsing. Passing an already-parsed + :class:`~packaging.version.Version` is a no-op, which makes the helper safe + to reuse in code that already works with ``Version`` objects. + + Raises: + InvalidPluginVersionError: If ``value`` is not a valid PEP 440 version + (including empty strings and non-string values). + """ + if isinstance(value, Version): + return value + if not isinstance(value, str): + raise InvalidPluginVersionError(f"Invalid plugin version: {value!r}") + + text = value.strip() + if not text: + raise InvalidPluginVersionError("Invalid plugin version: version is empty") + + try: + return Version(text) + except InvalidVersion as exc: + raise InvalidPluginVersionError(f"Invalid plugin version: {value!r}") from exc + + +def is_valid_version(value: object) -> bool: + """Return ``True`` when ``value`` is a valid PEP 440 version string.""" + if isinstance(value, Version): + return True + try: + parse_version(value) # type: ignore[arg-type] + except InvalidPluginVersionError: + return False + return True + + +def normalize_version(value: str | Version) -> str: + """Return the canonical string form of a PEP 440 version. + + ``"1.0.0"`` stays ``"1.0.0"``; whitespace is removed and equivalent + spellings are normalized by ``packaging``. This keeps hashing, diffing, and + display stable without changing ordering. + """ + return str(parse_version(value)) + + +def compare_versions(left: str | Version, right: str | Version) -> int: + """Compare two PEP 440 versions. + + Returns: + :data:`OLDER` (``-1``) when ``left < right``, :data:`EQUAL` (``0``) + when they compare equal, and :data:`NEWER` (``1``) when + ``left > right``. + + Raises: + InvalidPluginVersionError: If either value is not a valid version. + """ + left_version = parse_version(left) + right_version = parse_version(right) + return (left_version > right_version) - (left_version < right_version) + + +def is_newer(candidate: str | Version, current: str | Version) -> bool: + """Return ``True`` when ``candidate`` is strictly newer than ``current``.""" + return compare_versions(candidate, current) == NEWER + + +def sort_versions(values: Iterable[str | Version], *, reverse: bool = False) -> list[str]: + """Sort versions in ascending PEP 440 order (oldest first). + + The original values are returned (as strings) so callers can preserve the + representation they were given. Sorting is stable: versions that compare + equal (``1.0`` and ``1.0.0``) keep their input order. + + Raises: + InvalidPluginVersionError: If any value is not a valid version. + """ + parsed = [(parse_version(value), _original(value)) for value in values] + parsed.sort(key=lambda item: item[0], reverse=reverse) + return [original for _, original in parsed] + + +def _original(value: str | Version) -> str: + """Return the caller's original string representation for ``value``.""" + return value if isinstance(value, str) else str(value) + + +def latest_version(values: Iterable[str | Version]) -> str | None: + """Return the newest version in ``values``, or ``None`` for an empty input. + + Ties are resolved by input order (the first of the equal versions wins). + """ + parsed = [(parse_version(value), _original(value)) for value in values] + if not parsed: + return None + newest = parsed[0] + for candidate in parsed[1:]: + if candidate[0] > newest[0]: + newest = candidate + return newest[1] + + +__all__ = [ + "EQUAL", + "NEWER", + "OLDER", + "InvalidPluginVersionError", + "compare_versions", + "is_newer", + "is_valid_version", + "latest_version", + "normalize_version", + "parse_version", + "sort_versions", +] diff --git a/app/static/css/style.css b/app/static/css/style.css index 727f2a7..579512c 100644 --- a/app/static/css/style.css +++ b/app/static/css/style.css @@ -1906,6 +1906,52 @@ a:hover { align-items: flex-start; } +/* Trust, signature and compatibility badges (Phase 9 #419). Every badge also + renders explicit text and an aria-label, so state is never color-only. */ +.plugin-badges { + display: inline-flex; + flex-wrap: wrap; + gap: 4px; + margin-left: 4px; + vertical-align: middle; +} + +.plugin-badge { + display: inline-flex; + align-items: center; + gap: 4px; + padding: 1px 8px; + background: var(--bg-elevated); + border: 1px solid var(--border); + border-radius: 999px; + font-size: 11px; + font-weight: 600; + line-height: 1.6; + white-space: nowrap; +} + +.plugin-badge-icon { + font-weight: 700; +} + +.plugin-badge-trusted, +.plugin-badge-compatible { + border-color: var(--success); + color: var(--success); +} + +.plugin-badge-invalid, +.plugin-badge-incompatible { + border-color: var(--error); + color: var(--error); +} + +.plugin-badge-unverified, +.plugin-badge-unknown { + border-color: var(--warning); + color: var(--warning); +} + .plugin-actions { display: flex; flex-direction: column; diff --git a/app/static/js/plugin_badges.js b/app/static/js/plugin_badges.js new file mode 100644 index 0000000..0398727 --- /dev/null +++ b/app/static/js/plugin_badges.js @@ -0,0 +1,138 @@ +// AI Code Assistant - trust, signature and compatibility badges. +// +// Marketplace and plugin-management surfaces must make a plugin's trust and +// compatibility state understandable without relying on color alone. Every +// badge therefore renders an icon (hidden from assistive technology), explicit +// visible text, and an `aria-label` that spells out the full state. +// +// The module is dependency-free and DOM-free so it can be unit tested under +// Node and reused by any marketplace view: +// +// var html = window.AICAPluginBadges.render(plugin); +// +// Issue #419: trust and compatibility badges UI. + +(function (root, factory) { + "use strict"; + + var api = factory(); + if (typeof module === "object" && module.exports) { + module.exports = api; + } else { + root.AICAPluginBadges = api; + } +})(typeof globalThis !== "undefined" ? globalThis : this, function () { + "use strict"; + + var TRUST_LABELS = { + trusted: "Trusted publisher", + unverified: "Unverified publisher", + invalid: "Invalid signature", + }; + + var TRUST_ICONS = { + trusted: "\u2713", + unverified: "?", + invalid: "\u2715", + }; + + function escapeHtml(text) { + return String(text == null ? "" : text).replace(/[&<>"']/g, function (c) { + return { "&": "&", "<": "<", ">": ">", '"': """, "'": "'" }[c]; + }); + } + + function trustState(plugin) { + var value = plugin && plugin.trust_state != null ? String(plugin.trust_state).trim().toLowerCase() : ""; + if (value === "trusted" || value === "verified") return "trusted"; + if (value === "invalid") return "invalid"; + return "unverified"; + } + + function compatibilityState(plugin) { + if (plugin && plugin.compatible_with_app === true) return "compatible"; + if (plugin && plugin.compatible_with_app === false) return "incompatible"; + return "unknown"; + } + + function badge(kind, tone, icon, label, detail) { + var text = label + (detail ? " " + detail : ""); + var aria = kind + ": " + text; + return ( + '' + + '" + + '' + escapeHtml(text) + "" + + "" + ); + } + + function publisherDetail(plugin) { + var publisher = plugin && plugin.trust_publisher != null ? String(plugin.trust_publisher).trim() : ""; + return publisher ? "(" + publisher + ")" : ""; + } + + function signatureLabel(state) { + if (state === "trusted") return "Signature verified"; + if (state === "invalid") return "Signature invalid"; + return "No publisher signature"; + } + + function trustBadge(plugin) { + var state = trustState(plugin); + return badge("Trust", state, TRUST_ICONS[state], TRUST_LABELS[state], publisherDetail(plugin)); + } + + function signatureBadge(plugin) { + var state = trustState(plugin); + var tone = state === "trusted" ? "trusted" : state === "invalid" ? "invalid" : "unverified"; + return badge("Signature", tone, TRUST_ICONS[state], signatureLabel(state), ""); + } + + function compatibilityBadge(plugin) { + var state = compatibilityState(plugin); + var appVersion = plugin && plugin.app_version != null ? String(plugin.app_version).trim() : ""; + var required = plugin && plugin.compatibility != null ? String(plugin.compatibility).trim() : ""; + + if (state === "compatible") { + return badge( + "Compatibility", + "compatible", + "\u2713", + "Compatible with app" + (appVersion ? " " + appVersion : ""), + "" + ); + } + if (state === "incompatible") { + var reason = required && required !== "any" ? "(requires " + required + ")" : ""; + return badge( + "Compatibility", + "incompatible", + "\u2715", + "Incompatible with app" + (appVersion ? " " + appVersion : ""), + reason + ); + } + return badge("Compatibility", "unknown", "?", "Compatibility unknown", ""); + } + + function render(plugin) { + return ( + '' + + trustBadge(plugin) + + signatureBadge(plugin) + + compatibilityBadge(plugin) + + "" + ); + } + + return { + escapeHtml: escapeHtml, + trustState: trustState, + compatibilityState: compatibilityState, + trustBadge: trustBadge, + signatureBadge: signatureBadge, + compatibilityBadge: compatibilityBadge, + render: render, + }; +}); diff --git a/app/static/js/plugins.js b/app/static/js/plugins.js index e8a3a8c..07128d0 100644 --- a/app/static/js/plugins.js +++ b/app/static/js/plugins.js @@ -60,6 +60,17 @@ }).join(""); } + function pluginBadges(plugin) { + if ( + typeof window !== "undefined" && + window.AICAPluginBadges && + typeof window.AICAPluginBadges.render === "function" + ) { + return " " + window.AICAPluginBadges.render(plugin); + } + return ""; + } + function pluginRow(plugin) { var stateBadge; if (!plugin.installed) { @@ -93,7 +104,8 @@ '
' + "
" + '
' + escapeHtml(plugin.name) + - ' v' + escapeHtml(plugin.version) + " " + stateBadge + "
" + + ' v' + escapeHtml(plugin.version) + " " + stateBadge + + pluginBadges(plugin) + "
" + "

" + escapeHtml(plugin.description || "No description.") + "

" + "

author: " + escapeHtml(plugin.author || "unknown") + " · id: " + escapeHtml(plugin.id) + "

" + diff --git a/app/templates/plugins/index.html b/app/templates/plugins/index.html index ccc0300..730f0c5 100644 --- a/app/templates/plugins/index.html +++ b/app/templates/plugins/index.html @@ -48,5 +48,6 @@

Registered plugins

{% endblock %} {% block scripts %} + {% endblock %} diff --git a/tests/js/plugin_badges.test.js b/tests/js/plugin_badges.test.js new file mode 100644 index 0000000..69778b3 --- /dev/null +++ b/tests/js/plugin_badges.test.js @@ -0,0 +1,199 @@ +/** + * Unit tests for the plugin trust, signature and compatibility badges (#419). + * + * Every badge must state its status in text (never color alone), expose an + * accessible label, and safely escape plugin-controlled values. + */ +import { readFileSync } from "node:fs"; +import { resolve } from "node:path"; + +import { beforeAll, describe, expect, it } from "vitest"; + +const MODULE_PATH = resolve(process.cwd(), "app/static/js/plugin_badges.js"); + +let badges; + +beforeAll(() => { + window.eval(readFileSync(MODULE_PATH, "utf8")); + badges = window.AICAPluginBadges; + if (!badges || typeof badges.render !== "function") { + throw new Error("plugin_badges.js did not expose window.AICAPluginBadges.render"); + } +}); + +function mount(html) { + const container = document.createElement("div"); + container.innerHTML = html; + document.body.appendChild(container); + return container; +} + +function basePlugin(overrides) { + return Object.assign( + { + id: "example-plugin", + trust_state: "Trusted", + trust_publisher: "Acme", + compatibility: "any", + compatible_with_app: true, + app_version: "0.1.0", + }, + overrides || {} + ); +} + +describe("testing surface", () => { + it("publishes the badge helpers", () => { + for (const name of [ + "render", + "trustBadge", + "signatureBadge", + "compatibilityBadge", + "trustState", + "compatibilityState", + ]) { + expect(typeof badges[name]).toBe("function"); + } + }); +}); + +describe("trust badges", () => { + it("renders a verified publisher with text and an accessible label", () => { + const container = mount(badges.render(basePlugin())); + const trust = container.querySelector(".plugin-badge-trusted"); + + expect(trust).not.toBeNull(); + expect(trust.textContent).toContain("Trusted publisher"); + expect(trust.textContent).toContain("Acme"); + expect(trust.getAttribute("aria-label")).toBe("Trust: Trusted publisher (Acme)"); + expect(trust.getAttribute("role")).toBe("note"); + }); + + it("renders an unverified publisher", () => { + const container = mount(badges.render(basePlugin({ trust_state: "Unverified", trust_publisher: null }))); + const trust = container.querySelector(".plugin-badge-unverified"); + + expect(trust).not.toBeNull(); + expect(trust.textContent).toContain("Unverified publisher"); + expect(trust.getAttribute("aria-label")).toContain("Unverified publisher"); + }); + + it("renders an invalid signature as distinct from unverified", () => { + const container = mount(badges.render(basePlugin({ trust_state: "Invalid", trust_publisher: "Acme" }))); + const trust = container.querySelector(".plugin-badge-invalid"); + + expect(trust).not.toBeNull(); + expect(trust.textContent).toContain("Invalid signature"); + expect(container.textContent).not.toContain("Unverified publisher"); + }); + + it("treats an unknown or missing trust state as unverified", () => { + expect(badges.trustState({})).toBe("unverified"); + expect(badges.trustState({ trust_state: "trusted" })).toBe("trusted"); + expect(badges.trustState({ trust_state: "VERIFIED" })).toBe("trusted"); + }); +}); + +describe("signature badges", () => { + it("says the signature is verified for a trusted publisher", () => { + const container = mount(badges.render(basePlugin())); + expect(container.textContent).toContain("Signature verified"); + }); + + it("says there is no publisher signature for an unverified plugin", () => { + const container = mount(badges.render(basePlugin({ trust_state: "Unverified" }))); + expect(container.textContent).toContain("No publisher signature"); + }); + + it("says the signature is invalid for an invalid plugin", () => { + const container = mount(badges.render(basePlugin({ trust_state: "Invalid" }))); + expect(container.textContent).toContain("Signature invalid"); + }); +}); + +describe("compatibility badges", () => { + it("renders a compatible plugin with the app version", () => { + const container = mount(badges.render(basePlugin())); + const compat = container.querySelector(".plugin-badge-compatible"); + + expect(compat).not.toBeNull(); + expect(compat.textContent).toContain("Compatible with app 0.1.0"); + expect(compat.getAttribute("aria-label")).toContain("Compatibility:"); + }); + + it("renders an incompatible plugin with the required range", () => { + const container = mount( + badges.render( + basePlugin({ + trust_state: "Unverified", + compatible_with_app: false, + compatibility: ">=99.0.0", + }) + ) + ); + const compat = container.querySelector(".plugin-badge-incompatible"); + + expect(compat).not.toBeNull(); + expect(compat.textContent).toContain("Incompatible with app 0.1.0"); + expect(compat.textContent).toContain("requires >=99.0.0"); + }); + + it("renders an explicit unknown state when the backend omits compatibility", () => { + const container = mount(badges.render({ trust_state: "Unverified" })); + expect(container.textContent).toContain("Compatibility unknown"); + expect(container.querySelector(".plugin-badge-unknown")).not.toBeNull(); + }); + + it("identifies compatibility state from the backend flag", () => { + expect(badges.compatibilityState({ compatible_with_app: true })).toBe("compatible"); + expect(badges.compatibilityState({ compatible_with_app: false })).toBe("incompatible"); + expect(badges.compatibilityState({})).toBe("unknown"); + }); +}); + +describe("accessibility and safety", () => { + it("never relies on color alone", () => { + const container = mount( + badges.render( + basePlugin({ + trust_state: "Invalid", + compatible_with_app: false, + compatibility: ">=99.0.0", + }) + ) + ); + + const rendered = container.querySelectorAll(".plugin-badge"); + expect(rendered.length).toBe(3); + rendered.forEach((badge) => { + const text = badge.querySelector(".plugin-badge-text"); + const icon = badge.querySelector(".plugin-badge-icon"); + expect(text.textContent.trim().length).toBeGreaterThan(0); + expect(badge.getAttribute("aria-label").trim().length).toBeGreaterThan(0); + expect(icon.getAttribute("aria-hidden")).toBe("true"); + }); + }); + + it("escapes plugin-controlled values", () => { + const container = mount( + badges.render( + basePlugin({ + trust_publisher: '', + compatibility: '">', + compatible_with_app: false, + }) + ) + ); + + expect(container.querySelector("img")).toBeNull(); + expect(container.querySelector("script")).toBeNull(); + expect(container.textContent).toContain(''); + }); + + it("renders a safe fallback for a missing plugin", () => { + const container = mount(badges.render(null)); + expect(container.textContent).toContain("Unverified publisher"); + expect(container.textContent).toContain("No publisher signature"); + expect(container.textContent).toContain("Compatibility unknown"); + }); +}); diff --git a/tests/test_plugin_artifacts.py b/tests/test_plugin_artifacts.py new file mode 100644 index 0000000..272ae7e --- /dev/null +++ b/tests/test_plugin_artifacts.py @@ -0,0 +1,197 @@ +"""Tests for plugin artifact sha256 verification (#384). + +Covers hashing bytes/files/streams, accepting the catalog digest spellings, +constant-time match/mismatch handling, refusing registration on a mismatch, and +recording the refused install in the plugin audit trail. +""" + +import hashlib +import io + +import pytest + +from app.extensions import db +from app.models import ActivityEvent, Workspace +from app.models.activity_event import EVENT_PLUGIN_DENIED +from app.services.plugin_artifacts import ( + ArtifactError, + ArtifactIntegrityError, + compute_sha256, + normalize_sha256, + verify_and_register_artifact, + verify_artifact_integrity, +) + +PAYLOAD = b"the plugin artifact bytes" +DIGEST = hashlib.sha256(PAYLOAD).hexdigest() +OTHER_DIGEST = hashlib.sha256(b"tampered").hexdigest() + + +class TestNormalizeSha256: + @pytest.mark.parametrize( + "value", + [DIGEST, DIGEST.upper(), f"sha256:{DIGEST}", f"sha256-{DIGEST}", f"SHA256:{DIGEST}"], + ) + def test_accepts_catalog_spellings(self, value): + assert normalize_sha256(value) == DIGEST + + def test_strips_whitespace(self): + assert normalize_sha256(f" sha256:{DIGEST} ") == DIGEST + + @pytest.mark.parametrize( + "value", + ["", "a" * 63, "a" * 65, "g" * 64, "sha512:" + "a" * 128, None, 123], + ) + def test_rejects_invalid_digests(self, value): + with pytest.raises(ArtifactError): + normalize_sha256(value) + + def test_names_an_unsupported_algorithm(self): + with pytest.raises(ArtifactError, match="sha512"): + normalize_sha256("sha512:" + "a" * 128) + + +class TestComputeSha256: + def test_hashes_bytes(self): + assert compute_sha256(PAYLOAD) == DIGEST + + def test_hashes_empty_bytes(self): + assert compute_sha256(b"") == hashlib.sha256(b"").hexdigest() + + def test_hashes_a_file_path(self, tmp_path): + artifact = tmp_path / "plugin.zip" + artifact.write_bytes(PAYLOAD) + assert compute_sha256(artifact) == DIGEST + assert compute_sha256(str(artifact)) == DIGEST + + def test_hashes_a_binary_stream(self): + assert compute_sha256(io.BytesIO(PAYLOAD)) == DIGEST + + def test_streaming_chunk_size_does_not_change_the_digest(self): + assert compute_sha256(PAYLOAD, chunk_size=1) == DIGEST + assert compute_sha256(io.BytesIO(PAYLOAD), chunk_size=2) == DIGEST + + def test_missing_file_is_an_artifact_error(self, tmp_path): + with pytest.raises(ArtifactError, match="Cannot read artifact file"): + compute_sha256(tmp_path / "missing.zip") + + def test_rejects_invalid_chunk_size(self): + with pytest.raises(ArtifactError, match="chunk size"): + compute_sha256(PAYLOAD, chunk_size=0) + + def test_rejects_text_streams(self): + with pytest.raises(ArtifactError, match="binary"): + compute_sha256(io.StringIO("not bytes")) + + def test_rejects_unsupported_sources(self): + with pytest.raises(ArtifactError, match="Unsupported artifact source"): + compute_sha256(12345) + + +class TestVerifyArtifactIntegrity: + def test_match_returns_the_digest(self): + assert verify_artifact_integrity(PAYLOAD, DIGEST) == DIGEST + assert verify_artifact_integrity(PAYLOAD, f"sha256:{DIGEST.upper()}") == DIGEST + + def test_mismatch_raises_with_both_digests(self): + with pytest.raises(ArtifactIntegrityError) as excinfo: + verify_artifact_integrity(PAYLOAD, OTHER_DIGEST, plugin_id="evil-plugin") + + error = excinfo.value + assert error.expected == OTHER_DIGEST + assert error.actual == DIGEST + assert error.plugin_id == "evil-plugin" + assert OTHER_DIGEST in str(error) + assert DIGEST in str(error) + assert "evil-plugin" in str(error) + + def test_mismatch_without_workspace_does_not_need_an_app_context(self): + with pytest.raises(ArtifactIntegrityError): + verify_artifact_integrity(PAYLOAD, OTHER_DIGEST, workspace_id=None) + + def test_match_does_not_record_an_audit_event(self, app, make_user): + user = make_user() + workspace = Workspace(user_id=user.id, name="Artifacts") + db.session.add(workspace) + db.session.commit() + + verify_artifact_integrity( + PAYLOAD, + DIGEST, + plugin_id="safe-plugin", + workspace_id=workspace.id, + actor=user, + ) + db.session.flush() + + assert ( + ActivityEvent.query.filter_by( + workspace_id=workspace.id, event_type=EVENT_PLUGIN_DENIED + ).count() + == 0 + ) + + +class TestVerifyAndRegisterArtifact: + def test_registers_only_after_a_match(self): + calls = [] + + def register(digest): + calls.append(digest) + return "registered" + + assert ( + verify_and_register_artifact(PAYLOAD, DIGEST, register, plugin_id="safe-plugin") + == "registered" + ) + assert calls == [DIGEST] + + def test_does_not_register_a_tampered_artifact(self): + calls = [] + + def register(digest): # pragma: no cover - must never run + calls.append(digest) + return "registered" + + with pytest.raises(ArtifactIntegrityError): + verify_and_register_artifact(PAYLOAD, OTHER_DIGEST, register, plugin_id="evil-plugin") + + assert calls == [] + + def test_propagates_registration_errors_after_verification(self): + def register(digest): + raise RuntimeError("registration failed") + + with pytest.raises(RuntimeError, match="registration failed"): + verify_and_register_artifact(PAYLOAD, DIGEST, register) + + +class TestMismatchAudit: + def test_mismatch_records_a_bounded_audit_event(self, app, make_user): + user = make_user() + workspace = Workspace(user_id=user.id, name="Artifacts") + db.session.add(workspace) + db.session.commit() + + with pytest.raises(ArtifactIntegrityError): + verify_artifact_integrity( + PAYLOAD, + OTHER_DIGEST, + plugin_id="evil-plugin", + workspace_id=workspace.id, + actor=user, + ) + + db.session.flush() + event = ActivityEvent.query.filter_by( + workspace_id=workspace.id, event_type=EVENT_PLUGIN_DENIED + ).one() + assert event.actor_id == user.id + assert event.event_metadata["plugin_id"] == "evil-plugin" + assert event.event_metadata["action"] == "denied" + assert event.event_metadata["outcome"] == "denied" + assert event.event_metadata["reason"] == "artifact integrity mismatch" + # No artifact contents or digests are written to the audit trail. + serialized = str(event.event_metadata) + assert DIGEST not in serialized + assert OTHER_DIGEST not in serialized diff --git a/tests/test_plugin_lockfile.py b/tests/test_plugin_lockfile.py new file mode 100644 index 0000000..d2a36da --- /dev/null +++ b/tests/test_plugin_lockfile.py @@ -0,0 +1,193 @@ +"""Tests for deterministic plugin lockfile serialization (#398). + +The lockfile is meant to be diffed and committed, so the important properties +are byte stability across runs/input order, sorted entries, normalized hashes, +and the absence of volatile fields such as per-resolution timestamps. +""" + +import json +from dataclasses import dataclass + +import pytest + +from app.services.plugin_lockfile import ( + LOCKFILE_VERSION, + LockfileEntry, + LockfileError, + build_lockfile, + dumps_lockfile, + normalize_integrity, + write_lockfile, +) + +HASH_A = "a" * 64 +HASH_B = "B" * 64 + + +def _entry(plugin_id, version="1.0.0", source="catalog", integrity=HASH_A, dependencies=()): + return LockfileEntry( + plugin_id=plugin_id, + version=version, + source=source, + integrity=integrity, + dependencies=tuple(dependencies or ()), + ) + + +class TestNormalizeIntegrity: + @pytest.mark.parametrize("prefix", ["", "sha256:", "sha256-", "SHA256:"]) + def test_accepts_common_spellings(self, prefix): + assert normalize_integrity(prefix + HASH_A.upper()) == f"sha256:{HASH_A}" + + def test_strips_whitespace(self): + assert normalize_integrity(f" sha256:{HASH_A} ") == f"sha256:{HASH_A}" + + @pytest.mark.parametrize( + "value", + ["", "a" * 63, "a" * 65, "g" * 64, "sha512:" + "a" * 128, None, 123], + ) + def test_rejects_invalid_hashes(self, value): + with pytest.raises(LockfileError): + normalize_integrity(value) + + def test_names_an_unsupported_algorithm(self): + with pytest.raises(LockfileError, match="sha512"): + normalize_integrity("sha512:" + "a" * 128) + + +class TestBuildLockfile: + def test_sorts_entries_by_plugin_id(self): + document = build_lockfile([_entry("zebra"), _entry("alpha"), _entry("middle")]) + assert [entry["plugin_id"] for entry in document["plugins"]] == [ + "alpha", + "middle", + "zebra", + ] + assert document["lockfile_version"] == LOCKFILE_VERSION + + def test_normalizes_versions_and_hashes(self): + document = build_lockfile( + [_entry("plugin", version=" 1.0.0RC1 ", integrity=HASH_B.lower())] + ) + entry = document["plugins"][0] + assert entry["version"] == "1.0.0rc1" + assert entry["integrity"] == f"sha256:{HASH_B.lower()}" + + def test_sorts_and_trims_dependencies(self): + document = build_lockfile([_entry("plugin", dependencies=(" z>=1 ", "a>=2"))]) + assert document["plugins"][0]["dependencies"] == ["a>=2", "z>=1"] + + def test_tolerates_none_dependencies(self): + document = build_lockfile([_entry("plugin", dependencies=None)]) + assert document["plugins"][0]["dependencies"] == [] + + def test_accepts_mappings_and_aliases(self): + document = build_lockfile( + [ + { + "plugin_id": "plugin", + "version": "2.0", + "source": "https://example.test/catalog", + "integrity_hash": HASH_B, + "dependencies": ("dep>=1",), + "resolved_at": "2026-10-17T00:00:00Z", + } + ] + ) + entry = document["plugins"][0] + assert entry["integrity"] == f"sha256:{HASH_B.lower()}" + assert "resolved_at" not in entry # volatile data must not be serialized + + def test_accepts_objects_with_attributes(self): + @dataclass + class Row: + plugin_id: str + version: str + source: str + integrity: str + dependencies: tuple[str, ...] + + document = build_lockfile([Row("plugin", "1.0", "catalog", HASH_A, ("dep>=1",))]) + assert document["plugins"][0]["plugin_id"] == "plugin" + + def test_ignores_volatile_resolved_at_on_entries(self): + entry = LockfileEntry( + plugin_id="plugin", + version="1.0.0", + source="catalog", + integrity=HASH_A, + dependencies=(), + ) + assert "resolved_at" not in entry.to_dict() + + def test_duplicate_plugin_ids_are_refused(self): + with pytest.raises(LockfileError, match="Duplicate"): + build_lockfile([_entry("plugin"), _entry("plugin", version="2.0.0")]) + + @pytest.mark.parametrize( + "entry", + [ + {"version": "1.0", "source": "catalog", "integrity": HASH_A}, + {"plugin_id": "p", "source": "catalog", "integrity": HASH_A}, + {"plugin_id": "p", "version": "1.0", "integrity": HASH_A}, + {"plugin_id": "p", "version": "1.0", "source": "catalog"}, + ], + ) + def test_missing_required_fields_are_refused(self, entry): + with pytest.raises(LockfileError): + build_lockfile([entry]) + + def test_invalid_version_is_refused(self): + with pytest.raises(LockfileError, match="version"): + build_lockfile([_entry("plugin", version="not-a-version")]) + + def test_empty_input(self): + assert build_lockfile([]) == {"lockfile_version": LOCKFILE_VERSION, "plugins": []} + + +class TestDumpsLockfile: + def test_output_is_stable_across_input_order(self): + entries = [_entry("b"), _entry("a", version="2.0.0")] + first = dumps_lockfile(entries) + second = dumps_lockfile(list(reversed(entries))) + assert first == second + + def test_output_is_stable_across_repeated_runs(self): + entries = [_entry("a"), _entry("b")] + assert len({dumps_lockfile(entries) for _ in range(5)}) == 1 + + def test_ends_with_exactly_one_newline(self): + output = dumps_lockfile([_entry("a")]) + assert output.endswith("\n") + assert not output.endswith("\n\n") + + def test_object_keys_are_sorted(self): + output = dumps_lockfile([_entry("a")]) + assert output.index('"lockfile_version"') < output.index('"plugins"') + entry = json.loads(output)["plugins"][0] + assert list(entry) == ["dependencies", "integrity", "plugin_id", "source", "version"] + + def test_keeps_unicode(self): + output = dumps_lockfile([_entry("plugin", source="catalogue-é")]) + assert "catalogue-é" in output + assert "\\u00e9" not in output + + +class TestWriteLockfile: + def test_writes_canonical_bytes_and_newlines(self, tmp_path): + destination = tmp_path / "plugins.lock.json" + write_lockfile(destination, [_entry("b"), _entry("a")]) + + contents = destination.read_bytes() + assert contents == dumps_lockfile([_entry("a"), _entry("b")]).encode("utf-8") + assert b"\r\n" not in contents + + def test_creates_parent_directories(self, tmp_path): + destination = tmp_path / "nested" / "dir" / "plugins.lock.json" + write_lockfile(destination, [_entry("a")]) + assert destination.is_file() + + def test_leaves_no_temporary_files(self, tmp_path): + destination = tmp_path / "plugins.lock.json" + write_lockfile(destination, [_entry("a")]) + assert list(tmp_path.glob(".plugins.lock.json.*")) == [] diff --git a/tests/test_plugin_versions.py b/tests/test_plugin_versions.py new file mode 100644 index 0000000..4c380e1 --- /dev/null +++ b/tests/test_plugin_versions.py @@ -0,0 +1,171 @@ +"""Tests for the PEP 440 version comparison helpers (#399). + +The helpers back update and compatibility checks, so the tests cover the +ordering rules that are easy to get wrong (release padding, prereleases, +post/dev releases, epochs, and local versions), invalid input handling, and +stable sorting. +""" + +import pytest +from packaging.version import Version + +from app.services.plugin_versions import ( + EQUAL, + NEWER, + OLDER, + InvalidPluginVersionError, + compare_versions, + is_newer, + is_valid_version, + latest_version, + normalize_version, + parse_version, + sort_versions, +) + + +class TestParseVersion: + @pytest.mark.parametrize( + "value", + ["1", "1.0", "v1.0", "1.2.3", " 2.1.0 ", "1.0.0rc1", "1.0.post1", "1!2.0", "1.0+local"], + ) + def test_accepts_valid_versions(self, value): + assert isinstance(parse_version(value), Version) + + def test_accepts_an_existing_version(self): + version = Version("1.2.3") + assert parse_version(version) is version + + @pytest.mark.parametrize( + "value", + ["", " ", "not-a-version", "1.0.0-", None, 123, object()], + ) + def test_rejects_invalid_versions(self, value): + with pytest.raises(InvalidPluginVersionError): + parse_version(value) + + def test_invalid_error_has_a_clear_message(self): + with pytest.raises(InvalidPluginVersionError, match="not-a-version"): + parse_version("not-a-version") + + +class TestValidation: + @pytest.mark.parametrize("value", ["1.0", "1.0.0rc1", Version("2.0")]) + def test_valid(self, value): + assert is_valid_version(value) is True + + @pytest.mark.parametrize("value", ["", "nope", None, 1.0, []]) + def test_invalid(self, value): + assert is_valid_version(value) is False + + +class TestNormalizeVersion: + @pytest.mark.parametrize( + ("value", "expected"), + [ + ("1.0", "1.0"), + ("v1.0", "1.0"), + (" 1.0.0 ", "1.0.0"), + ("01.002.0003", "1.2.3"), + ("1.0.0RC1", "1.0.0rc1"), + ("1.0.0+ABC", "1.0.0+abc"), + ], + ) + def test_normalizes(self, value, expected): + assert normalize_version(value) == expected + + +class TestCompareVersions: + @pytest.mark.parametrize( + ("left", "right", "expected"), + [ + ("1.0.0", "1.0.0", EQUAL), + ("1.0", "1.0.0", EQUAL), # missing release segments are zero + ("1.0.1", "1.0.0", NEWER), + ("0.9.9", "1.0.0", OLDER), + ("1.0.0rc1", "1.0.0", OLDER), # prerelease sorts before release + ("1.0.0rc2", "1.0.0rc1", NEWER), + ("1.0.0.post1", "1.0.0", NEWER), # postrelease sorts after release + ("1.0.0.dev1", "1.0.0a1", OLDER), # dev sorts before pre-release + ("1!1.0", "2.0", NEWER), # epoch dominates + ("1.0+abc", "1.0", NEWER), # local version sorts after plain + ("1.0+abc", "1.0+abd", OLDER), + ], + ) + def test_ordering(self, left, right, expected): + assert compare_versions(left, right) == expected + + def test_compare_is_antisymmetric(self): + left, right = "1.2.0", "1.10.0" + assert compare_versions(left, right) == -compare_versions(right, left) + + def test_invalid_operand_raises(self): + with pytest.raises(InvalidPluginVersionError): + compare_versions("1.0", "bogus") + + +class TestIsNewer: + def test_newer(self): + assert is_newer("1.0.1", "1.0.0") is True + + def test_equal_is_not_newer(self): + assert is_newer("1.0", "1.0.0") is False + + def test_older_is_not_newer(self): + assert is_newer("0.9.0", "1.0.0") is False + + +class TestSortVersions: + def test_sorts_ascending(self): + assert sort_versions(["2.0.0", "1.0.0", "1.5.0"]) == ["1.0.0", "1.5.0", "2.0.0"] + + def test_sorts_with_prereleases(self): + assert sort_versions(["1.0.0", "1.0.0rc1", "0.9.0"]) == [ + "0.9.0", + "1.0.0rc1", + "1.0.0", + ] + + def test_reverse(self): + assert sort_versions(["1.0.0", "2.0.0"], reverse=True) == ["2.0.0", "1.0.0"] + + def test_stable_for_equal_versions(self): + assert sort_versions(["1.0.0", "1.0"]) == ["1.0.0", "1.0"] + + def test_preserves_input_representation(self): + assert sort_versions([" 2.0 ", "1.0"]) == ["1.0", " 2.0 "] + + def test_empty(self): + assert sort_versions([]) == [] + + def test_invalid_value_raises(self): + with pytest.raises(InvalidPluginVersionError): + sort_versions(["1.0", "nope"]) + + +class TestLatestVersion: + def test_latest(self): + assert latest_version(["0.9.0", "1.2.0", "1.10.0"]) == "1.10.0" + + def test_empty_returns_none(self): + assert latest_version([]) is None + + def test_tie_keeps_first(self): + assert latest_version(["1.0.0", "1.0"]) == "1.0.0" + + def test_invalid_value_raises(self): + with pytest.raises(InvalidPluginVersionError): + latest_version(["1.0", "bad"]) + + +class TestCompatibilityUsesTheHelpers: + def test_invalid_app_version_fails_closed(self): + from app.services.plugin_compat import plugin_compatible + + assert plugin_compatible(">=1.0.0", "not-a-version") is False + + def test_version_comparison_is_pep440(self): + from app.services.plugin_compat import plugin_compatible + + assert plugin_compatible(">=1.10.0", "1.9.9") is False + assert plugin_compatible(">=1.10.0", "1.11.0") is True