flayer.providers.yandex_lifecycle

Module source

flayer.providers.yandex_lifecycle

Bounded Yandex mutations with explicit ownership and recoverable uncertain outcomes.

MAX_USER_DATA_BYTES module-attribute

MAX_USER_DATA_BYTES = 262144

USERNAME_PATTERN module-attribute

USERNAME_PATTERN = re.compile('[a-z_][a-z0-9_-]{0,31}\\Z')

SHA256_PATTERN module-attribute

SHA256_PATTERN = re.compile('[a-f0-9]{64}\\Z')

OPERATION_ID_PATTERN module-attribute

OPERATION_ID_PATTERN = re.compile('[a-f0-9]{32}\\Z')

RESOURCE_NAME_PATTERN module-attribute

RESOURCE_NAME_PATTERN = re.compile('[a-z](?:[a-z0-9-]{0,61}[a-z0-9])?\\Z')

REQUIRED_PARAMETERS module-attribute

REQUIRED_PARAMETERS: dict[ResourceKind, frozenset[str]] = {ResourceKind.NETWORK: frozenset(), ResourceKind.SUBNET: frozenset({'zone_id', 'ipv4_cidr', 'network_dependency'}), ResourceKind.SECURITY_GROUP: frozenset({'network_dependency', 'rules'}), ResourceKind.ADDRESS: frozenset({'zone_id'}), ResourceKind.DISK: frozenset({'zone_id', 'image_id', 'size_gib'}), ResourceKind.INSTANCE: frozenset({'zone_id', 'cores', 'memory_gib', 'boot_disk_dependency', 'subnet_dependency', 'security_group_dependency', 'address_dependency', 'ssh_public_key'})}

OPTIONAL_PARAMETERS module-attribute

OPTIONAL_PARAMETERS: dict[ResourceKind, frozenset[str]] = {ResourceKind.NETWORK: frozenset(), ResourceKind.SUBNET: frozenset(), ResourceKind.SECURITY_GROUP: frozenset(), ResourceKind.ADDRESS: frozenset(), ResourceKind.DISK: frozenset({'type'}), ResourceKind.INSTANCE: frozenset({'ssh_username', 'user_data_file', 'user_data_sha256'})}

DEPENDENCY_KINDS module-attribute

DEPENDENCY_KINDS = {'network_dependency': ResourceKind.NETWORK, 'boot_disk_dependency': ResourceKind.DISK, 'subnet_dependency': ResourceKind.SUBNET, 'security_group_dependency': ResourceKind.SECURITY_GROUP, 'address_dependency': ResourceKind.ADDRESS}

KNOWN_STATUSES module-attribute

KNOWN_STATUSES = frozenset({'UNKNOWN', 'ACTIVE', 'CREATING', 'PROVISIONING', 'READY', 'RUNNING', 'STARTING', 'STOPPING', 'STOPPED', 'RESTARTING', 'UPDATING', 'DELETING', 'ERROR', 'FAILED'})

USER_DATA_FIELDS module-attribute

USER_DATA_FIELDS = frozenset({'ssh_pwauth', 'disable_root', 'users', 'package_update', 'package_upgrade', 'packages', 'write_files', 'runcmd'})

YandexLifecycleProvider

YandexLifecycleProvider(settings: YandexCloudSettings, runner: CommandRunner | None = None)

Bases: YandexCloudProvider

Create and delete an explicit six-kind subset using existing read-only observation.

Source code in installed/flayer/providers/yandex.py

def __init__(
    self, settings: YandexCloudSettings, runner: CommandRunner | None = None
) -> None:
    """Construct a provider without executing commands or reading credential files."""

    self._settings = settings
    self._runner = runner if runner is not None else SubprocessCommandRunner()
    self._identity = ProviderIdentity(PROVIDER_ID, settings.folder_id, "cli-profile")

Capabilities property

Capabilities: frozenset[ProviderCapability]

Advertise implemented mutations without promising credentials or account permission.

ValidateSpec

ValidateSpec(spec: ResourceSpec) -> None

Validate the complete supported option/dependency subset without cloud access.

Source code in installed/flayer/providers/yandex_lifecycle.py

def ValidateSpec(self, spec: ResourceSpec) -> None:
    """Validate the complete supported option/dependency subset without cloud access."""

    failure = False

    try:
        if not isinstance(spec, ResourceSpec):
            raise ValueError("Lifecycle requests require ResourceSpec")

        if RESOURCE_NAME_PATTERN.fullmatch(spec.name) is None:
            raise ValueError("Resource name violates provider naming requirements")

        options = dict(spec.parameters)
        required = REQUIRED_PARAMETERS[spec.kind]

        if not required <= options.keys() or options.keys() - required - OPTIONAL_PARAMETERS[
            spec.kind
        ]:
            raise ValueError("Unsupported lifecycle options")

        expected: set[str] = set()

        for key in options:
            if key in DEPENDENCY_KINDS:
                dependency = _String(options[key])
                ValidateName(dependency, "dependency")
                expected.add(dependency)

        dependency_options = sum(key in DEPENDENCY_KINDS for key in options)

        if expected != set(spec.dependencies) or len(expected) != dependency_options:
            raise ValueError("Dependencies must match explicit typed dependency options")

        if "zone_id" in options:
            ValidateIdentifier(_String(options["zone_id"]))

        if spec.kind == ResourceKind.SUBNET:
            ipaddress.IPv4Network(_String(options["ipv4_cidr"]), strict=True)

        elif spec.kind == ResourceKind.SECURITY_GROUP:
            _Rules(options["rules"])

        elif spec.kind == ResourceKind.DISK:
            ValidateIdentifier(_String(options["image_id"]))
            _Integer(options["size_gib"], 10, 1024)

            if options.get("type", "network-ssd") not in {"network-ssd", "network-hdd"}:
                raise ValueError("Unsupported boot disk type")

        elif spec.kind == ResourceKind.INSTANCE:
            _Integer(options["cores"], 2, 32)
            _Integer(options["memory_gib"], 1, 128)
            _PublicKey(options["ssh_public_key"])

            if USERNAME_PATTERN.fullmatch(_String(options.get("ssh_username", "yc-user"))) is None:
                raise ValueError("Unsupported SSH username")

            if ("user_data_file" in options) != ("user_data_sha256" in options):
                raise ValueError("Initialization file requires its immutable content digest")

            if "user_data_file" in options:
                path = Path(_String(options["user_data_file"]))

                if not str(path) or ".." in path.parts or "\x00" in str(path):
                    raise ValueError("Initialization file path is unsafe")

                if SHA256_PATTERN.fullmatch(_String(options["user_data_sha256"])) is None:
                    raise ValueError("Initialization digest must be SHA256")

    except (ValueError, TypeError, KeyError):
        failure = True

    if failure:
        raise MutationError(ProviderErrorCode.UNSUPPORTED, "validate-spec", False)

GetResource

GetResource(reference: ResourceReference) -> ProviderResource

Use the scoped base adapter while restricting observed lifecycle fields.

Source code in installed/flayer/providers/yandex_lifecycle.py

def GetResource(self, reference: ResourceReference) -> ProviderResource:
    """Use the scoped base adapter while restricting observed lifecycle fields."""

    return self._ValidateObservation(super().GetResource(reference))

FindResource

FindResource(spec: ResourceSpec, identity: StackIdentity) -> ProviderResource | None

Reconcile exact logical ownership; duplicates and changed desired content conflict.

Source code in installed/flayer/providers/yandex_lifecycle.py

def FindResource(
    self, spec: ResourceSpec, identity: StackIdentity
) -> ProviderResource | None:
    """Reconcile exact logical ownership; duplicates and changed desired content conflict."""

    self.ValidateSpec(spec)
    self._ValidateIdentity(identity)
    labels = identity.OwnershipLabels()
    labels[LOGICAL_ID_LABEL] = spec.logical_id
    ownership = tuple(sorted(labels.items()))
    matches = tuple(
        resource for resource in self.ListResources(spec.kind) if resource.HasLabels(ownership)
    )

    if len(matches) > 1:
        raise ProviderError(ProviderErrorCode.CONFLICT, "find")

    if not matches:
        return None

    resource = self._ValidateObservation(matches[0])

    if not resource.HasLabels(spec.OwnershipLabels(identity)) or resource.name != spec.name:
        raise ProviderError(ProviderErrorCode.CONFLICT, "find")

    options = dict(spec.parameters)

    if "zone_id" in options and self._ResourceZone(resource) != options["zone_id"]:
        raise ProviderError(ProviderErrorCode.SCOPE_MISMATCH, "find")

    return resource

CreateResource

CreateResource(spec: ResourceSpec, identity: StackIdentity, dependencies: Mapping[str, ProviderResource], operation_id: str) -> ProviderResource

Create once with stable ownership labels and an explicit, separately owned boot disk.

Source code in installed/flayer/providers/yandex_lifecycle.py

def CreateResource(
    self, spec: ResourceSpec, identity: StackIdentity,
    dependencies: Mapping[str, ProviderResource], operation_id: str,
) -> ProviderResource:
    """Create once with stable ownership labels and an explicit, separately owned boot disk."""

    self.ValidateSpec(spec)
    self._ValidateIdentity(identity)

    if not isinstance(operation_id, str) or OPERATION_ID_PATTERN.fullmatch(operation_id) is None:
        raise MutationError(ProviderErrorCode.UNSUPPORTED, "operation-id", False)

    options = dict(spec.parameters)
    content: bytes | None = None
    existing: ProviderResource | None = None
    arguments: tuple[str, ...] = ()
    preflight_failure: ProviderErrorCode | None = None

    try:
        content = self._ReadUserData(
            Path(_String(options["user_data_file"])), _String(options["user_data_sha256"])
        ) if "user_data_file" in options else None
        existing = self.FindResource(spec, identity)

        if existing is None:
            refreshed = self._Dependencies(spec, identity, dependencies)
            arguments = self._Arguments(spec, refreshed)

    except MutationError:
        raise

    except ProviderError as error:
        preflight_failure = error.code

    except Exception:
        preflight_failure = ProviderErrorCode.COMMAND_FAILED

    if preflight_failure is not None:
        raise MutationError(preflight_failure, "create", False)

    if existing is not None:
        return existing

    creation_labels = (*spec.OwnershipLabels(identity), (OPERATION_ID_LABEL, operation_id))
    labels = ",".join(f"{key}={value}" for key, value in creation_labels)
    command = (*self._CommandForKind(spec.kind), "create", "--name", spec.name, "--labels", labels)
    temporary_path: str | None = None
    mutation_started = False
    failure: ProviderErrorCode | None = None
    created: ProviderResource | None = None

    try:
        if content is not None:
            with tempfile.NamedTemporaryFile(mode="wb", delete=False) as destination:
                temporary_path = destination.name
                destination.write(content)
                destination.flush()
                os.fsync(destination.fileno())

            arguments = (*arguments, "--metadata-from-file", f"user-data={temporary_path}")

        mutation_started = True
        result = self._MutationRun((*command, *arguments), "create")

        try:
            payload: object = json.loads(result.stdout)

        except (ValueError, RecursionError):
            failure = ProviderErrorCode.INVALID_RESPONSE

        else:
            resource = self._ValidateObservation(self._NormalizeResource(payload, spec.kind, "create"))

            if not resource.HasLabels(creation_labels) or resource.name != spec.name:
                failure = ProviderErrorCode.SCOPE_MISMATCH

            elif "zone_id" in options and resource.zone_id != options["zone_id"]:
                failure = ProviderErrorCode.SCOPE_MISMATCH

            else:
                created = resource

    except MutationError:
        raise

    except ProviderError as error:
        failure = error.code

    except (OSError, ValueError):
        failure = ProviderErrorCode.COMMAND_FAILED

    finally:
        if temporary_path is not None:
            try:
                os.unlink(temporary_path)

            except FileNotFoundError:
                pass

            except OSError:
                failure = ProviderErrorCode.COMMAND_FAILED

    if failure is None and created is not None:
        return created

    raise MutationError(failure or ProviderErrorCode.INVALID_RESPONSE, "create", mutation_started)

DeleteResource

DeleteResource(reference: ResourceReference, identity: StackIdentity, logical_id: str, operation_id: str | None = None) -> None

Observe and verify ownership before exact-ID deletion, then confirm absence once.

Source code in installed/flayer/providers/yandex_lifecycle.py

def DeleteResource(
    self, reference: ResourceReference, identity: StackIdentity, logical_id: str,
    operation_id: str | None = None,
) -> None:
    """Observe and verify ownership before exact-ID deletion, then confirm absence once."""

    self._ValidateIdentity(identity)
    ValidateName(logical_id, "delete.logical_id")
    labels = identity.OwnershipLabels()
    labels[LOGICAL_ID_LABEL] = logical_id

    if operation_id is not None:
        if not isinstance(operation_id, str) or OPERATION_ID_PATTERN.fullmatch(operation_id) is None:
            raise MutationError(ProviderErrorCode.UNSUPPORTED, "operation-id", False)

        labels[OPERATION_ID_LABEL] = operation_id

    try:
        resource = self.GetResource(reference)

    except ProviderError as error:
        if error.code == ProviderErrorCode.NOT_FOUND:
            return

        raise

    if not resource.HasLabels(tuple(sorted(labels.items()))):
        raise MutationError(ProviderErrorCode.SCOPE_MISMATCH, "delete", False)

    command = self._CommandForKind(reference.kind)
    self._MutationRun((*command, "delete", "--id", reference.resource_id), "delete")
    failure = ProviderErrorCode.CONFLICT

    try:
        self.GetResource(reference)

    except ProviderError as error:
        if error.code == ProviderErrorCode.NOT_FOUND:
            return

        failure = error.code

    raise MutationError(failure, "delete", True)