flayer.profiles¶

flayer.profiles ¶

Explicit deployment profiles and owned generated artifacts.

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)

ArtifactError ¶

Bases: ContractError

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

GatewayDevice `dataclass` ¶

GatewayDevice(device_id: str, public_key: str, source_cidrs: tuple[str, ...])

One device public credential and explicitly bounded connection source networks.

device_id `instance-attribute` ¶

device_id: str

public_key `class-attribute` `instance-attribute` ¶

public_key: str = field(repr=False)

source_cidrs `instance-attribute` ¶

source_cidrs: tuple[str, ...]

GatewayError ¶

Bases: ContractError

Profile intent or a prepared initialization artifact violates its contract.

GatewayPlan `dataclass` ¶

GatewayPlan(profile: GatewayProfile)

Immutable composition input; serialized lifecycle plans bind an exported artifact.

profile `instance-attribute` ¶

profile: GatewayProfile

GatewayPort `dataclass` ¶

GatewayPort(protocol: str, port: int, cidrs: tuple[str, ...])

One explicit service allowance; an allowance does not install a listener.

protocol `instance-attribute` ¶

protocol: str

port `instance-attribute` ¶

port: int

cidrs `instance-attribute` ¶

cidrs: tuple[str, ...]

GatewayProfile `dataclass` ¶

GatewayProfile(identity: StackIdentity, zone_id: str, image_id: str, subnet_cidr: str, ssh_public_key: str, management_cidrs: tuple[str, ...], transport: GatewayTransport, service_ports: tuple[GatewayPort, ...] = (), ssh_username: str = 'gateway-admin', ssh_port: int = 22, cores: int = 2, memory_gib: int = 2, boot_disk_gib: int = 20, schema_version: int = PROFILE_SCHEMA_VERSION)

Explicit cloud placement and Ubuntu 24.04 cloud-init security intent.

identity `instance-attribute` ¶

identity: StackIdentity

zone_id `instance-attribute` ¶

zone_id: str

image_id `instance-attribute` ¶

image_id: str

subnet_cidr `instance-attribute` ¶

subnet_cidr: str

ssh_public_key `class-attribute` `instance-attribute` ¶

ssh_public_key: str = field(repr=False)

management_cidrs `instance-attribute` ¶

management_cidrs: tuple[str, ...]

transport `instance-attribute` ¶

transport: GatewayTransport

service_ports `class-attribute` `instance-attribute` ¶

service_ports: tuple[GatewayPort, ...] = ()

ssh_username `class-attribute` `instance-attribute` ¶

ssh_username: str = 'gateway-admin'

ssh_port `class-attribute` `instance-attribute` ¶

ssh_port: int = 22

cores `class-attribute` `instance-attribute` ¶

cores: int = 2

memory_gib `class-attribute` `instance-attribute` ¶

memory_gib: int = 2

boot_disk_gib `class-attribute` `instance-attribute` ¶

boot_disk_gib: int = 20

schema_version `class-attribute` `instance-attribute` ¶

schema_version: int = PROFILE_SCHEMA_VERSION

GatewayTarget `dataclass` ¶

GatewayTarget(name: str, host: str, port: int)

One exact TCP destination reachable through the SSH gateway.

name `instance-attribute` ¶

name: str

host `instance-attribute` ¶

host: str

port `instance-attribute` ¶

port: int

GatewayTransport `dataclass` ¶

GatewayTransport(username: str, targets: tuple[GatewayTarget, ...], devices: tuple[GatewayDevice, ...])

A restricted SSH local-forward transport with separately declared device keys.

username `instance-attribute` ¶

username: str

targets `instance-attribute` ¶

targets: tuple[GatewayTarget, ...]

devices `instance-attribute` ¶

devices: tuple[GatewayDevice, ...]

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,))

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

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