flayer.profiles.artifacts

Module source

flayer.profiles.artifacts

Bounded artifact bundles with exclusive ownership and verified local cleanup.

MAX_ARTIFACT_BYTES module-attribute

MAX_ARTIFACT_BYTES = 1048576

MAX_BUNDLE_BYTES module-attribute

MAX_BUNDLE_BYTES = 2097152

MAX_MANIFEST_BYTES module-attribute

MAX_MANIFEST_BYTES = 16384

MAX_BUNDLE_FILES module-attribute

MAX_BUNDLE_FILES = 8

ArtifactError

Bases: ContractError

An artifact is invalid, foreign, changed, busy, or unavailable for safe storage.

Artifact dataclass

Artifact(name: str, content: bytes, sensitive: bool = True)

Memory-only generated bytes; sensitive payloads are never included in repr.

name instance-attribute

name: str

content class-attribute instance-attribute

content: bytes = field(repr=False)

sensitive class-attribute instance-attribute

sensitive: bool = True

ArtifactBundle dataclass

ArtifactBundle(identity: StackIdentity, kind: str, name: str, files: tuple[Artifact, ...])

A named set owned by one complete stack identity, independent of provider state.

identity instance-attribute

identity: StackIdentity

kind instance-attribute

kind: str

name instance-attribute

name: str

files class-attribute instance-attribute

files: tuple[Artifact, ...] = field(repr=False)

BuildDeviceBundle

BuildDeviceBundle(identity: StackIdentity, device_id: str, transport: str, files: tuple[Artifact, ...]) -> ArtifactBundle

Wrap externally issued transport files without generating or validating credentials.

Source code in installed/flayer/profiles/artifacts.py

def BuildDeviceBundle(
    identity: StackIdentity, device_id: str, transport: str, files: tuple[Artifact, ...],
) -> ArtifactBundle:
    """Wrap externally issued transport files without generating or validating credentials."""

    ValidateName(transport, "transport")
    ValidateName(device_id, "device_id")

    if not isinstance(files, tuple) or not 1 <= len(files) < MAX_BUNDLE_FILES or any(
        not isinstance(item, Artifact) for item in files
    ):
        raise ArtifactError("Device files must contain 1..7 immutable Artifact values")

    if any(not item.sensitive for item in files):
        raise ArtifactError("Externally issued device profiles must be marked sensitive")

    descriptor = Artifact("device-info.json", (json.dumps({
        "schema_version": 1, "transport": transport, "device_id": device_id,
        "status": "externally-issued-unverified",
    }, sort_keys=True) + "\n").encode("utf-8"), sensitive=False)

    return ArtifactBundle(identity, "device", device_id, files + (descriptor,))

ReadArtifactFile

ReadArtifactFile(path: str | Path) -> bytes

Read only a private artifact; never expose filesystem paths or bytes in failures.

Source code in installed/flayer/profiles/artifacts.py

def ReadArtifactFile(path: str | Path) -> bytes:
    """Read only a private artifact; never expose filesystem paths or bytes in failures."""

    artifact_path = Path(path)

    try:
        _ValidatePath(artifact_path)

        with _Directory(artifact_path.parent) as directory:
            return _ReadFile(directory, artifact_path.name, MAX_ARTIFACT_BYTES)

    except (OSError, ValueError, RecursionError):
        raise ArtifactError("Unable to read a valid private artifact") from None

WriteArtifactBundle

WriteArtifactBundle(root: str | Path, bundle: ArtifactBundle) -> Path

Publish a new manifest-last bundle exclusively and roll back only created files.

Source code in installed/flayer/profiles/artifacts.py

def WriteArtifactBundle(root: str | Path, bundle: ArtifactBundle) -> Path:
    """Publish a new manifest-last bundle exclusively and roll back only created files."""

    if not isinstance(bundle, ArtifactBundle):
        raise ArtifactError("bundle must be an ArtifactBundle")

    root_path = Path(root)
    bundle_name = f"{bundle.kind}-{bundle.name}"
    created: list[tuple[str, _FileReceipt]] = []
    directory: int | None = None

    try:
        with _Directory(root_path, create=True) as parent:
            with _BundleLock(parent, bundle_name):
                try:
                    os.mkdir(bundle_name, 0o700, dir_fd=parent)
                    directory = os.open(bundle_name, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) | getattr(os, "O_NOFOLLOW", 0), dir_fd=parent)

                    for artifact in bundle.files:
                        _AssertDirectory(parent, bundle_name, directory)
                        receipt = _WriteFile(directory, artifact.name, artifact.content)
                        created.append((artifact.name, receipt))

                    for name, receipt in created:
                        _AssertFile(directory, name, receipt)

                    _AssertDirectory(parent, bundle_name, directory)
                    receipt = _WriteFile(directory, _MANIFEST_NAME, _Manifest(bundle))
                    created.append((_MANIFEST_NAME, receipt))
                    os.fsync(directory)
                    os.fsync(parent)

                except BaseException:
                    if directory is not None:
                        _AssertDirectory(parent, bundle_name, directory)

                        for name, receipt in reversed(created):
                            _UnlinkFile(directory, name, receipt)

                        _AssertDirectory(parent, bundle_name, directory)
                        os.rmdir(bundle_name, dir_fd=parent)

                    raise

                finally:
                    if directory is not None:
                        os.close(directory)

    except (OSError, ValueError, RecursionError):
        raise ArtifactError("Unable to publish artifact bundle; existing bundles are never replaced") from None

    return root_path / bundle_name

ReadOwnedArtifactFile

ReadOwnedArtifactFile(path: str | Path, identity: StackIdentity, *, kind: str, name: str) -> bytes

Authorize one exact file through the complete unchanged bundle ownership manifest.

Source code in installed/flayer/profiles/artifacts.py

def ReadOwnedArtifactFile(
    path: str | Path, identity: StackIdentity, *, kind: str, name: str,
) -> bytes:
    """Authorize one exact file through the complete unchanged bundle ownership manifest."""

    if not isinstance(identity, StackIdentity) or not isinstance(kind, str) or kind not in {"server", "device"}:
        raise ArtifactError("Artifact read requires explicit identity and bundle kind")

    ValidateName(name, "artifact name")
    artifact_path = Path(path)

    if artifact_path.parent.name != f"{kind}-{name}":
        raise ArtifactError("Artifact path does not match the expected bundle name")

    try:
        _ValidatePath(artifact_path)

        with _Directory(artifact_path.parent) as directory:
            snapshot = _ValidateManifest(directory, identity, kind, name)
            file = next((item for item in snapshot.files if item.name == artifact_path.name), None)

            if file is None:
                raise ArtifactError("Artifact filename is not owned by its bundle manifest")

            return file.content

    except (OSError, ValueError, RecursionError):
        raise ArtifactError("Unable to read an unchanged owned artifact bundle") from None

RemoveArtifactBundle

RemoveArtifactBundle(root: str | Path, identity: StackIdentity, *, kind: str, name: str) -> bool

Remove only exact, unchanged owned files; never recurse or delete provider resources.

Source code in installed/flayer/profiles/artifacts.py

def RemoveArtifactBundle(
    root: str | Path, identity: StackIdentity, *, kind: str, name: str,
) -> bool:
    """Remove only exact, unchanged owned files; never recurse or delete provider resources."""

    if not isinstance(identity, StackIdentity) or not isinstance(kind, str) or kind not in {"server", "device"}:
        raise ArtifactError("Artifact removal requires explicit identity and bundle kind")

    ValidateName(name, "artifact name")
    root_path = Path(root)
    bundle_name = f"{kind}-{name}"

    try:
        _ValidatePath(root_path)

        if not root_path.exists():
            return False

        with _Directory(root_path) as parent:
            with _BundleLock(parent, bundle_name):
                try:
                    descriptor = os.open(bundle_name, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) | getattr(os, "O_NOFOLLOW", 0), dir_fd=parent)

                except FileNotFoundError:
                    return False

                try:
                    _RequireOwned(os.fstat(descriptor), directory=True)
                    snapshot = _ValidateManifest(descriptor, identity, kind, name)
                    _AssertDirectory(parent, bundle_name, descriptor)

                    for file in snapshot.files:
                        _AssertFile(descriptor, file.name, file.receipt)

                    _AssertFile(descriptor, _MANIFEST_NAME, snapshot.manifest)

                    for file in snapshot.files:
                        _AssertDirectory(parent, bundle_name, descriptor)
                        _UnlinkFile(descriptor, file.name, file.receipt)

                    _UnlinkFile(descriptor, _MANIFEST_NAME, snapshot.manifest)
                    os.fsync(descriptor)
                    _AssertDirectory(parent, bundle_name, descriptor)
                    os.rmdir(bundle_name, dir_fd=parent)

                finally:
                    os.close(descriptor)

                os.fsync(parent)

    except (OSError, ValueError, RecursionError):
        raise ArtifactError("Unable to remove a valid owned artifact bundle; partial cleanup may require recovery") from None

    return True