flayer.profiles.static_site

Module source

flayer.profiles.static_site

Prepare a bounded static HTTP page for an explicitly selected fresh Ubuntu guest.

PROFILE_NAME module-attribute

PROFILE_NAME = 'static-site'

GUEST_CONTRACT module-attribute

GUEST_CONTRACT = 'ubuntu-24.04-cloud-init'

MAX_PROFILE_BYTES module-attribute

MAX_PROFILE_BYTES = 32768

CLOUD_INIT_FILENAME module-attribute

CLOUD_INIT_FILENAME = 'static-site-cloud-init.json'

PLAN_FILENAME module-attribute

PLAN_FILENAME = 'deployment-plan.toml'

StaticSiteError

Bases: ContractError

The static site violates its bounded profile or prepared artifact contract.

StaticSiteProfile dataclass

StaticSiteProfile(identity: StackIdentity, zone_id: str, image_id: str, subnet_cidr: str, admin_username: str, admin_public_key: str, admin_source_cidrs: tuple[str, ...], http_source_cidrs: tuple[str, ...], title: str, body: str, cores: int = 2, memory_gib: int = 2, boot_disk_gib: int = 20, schema_version: int = 1, guest_contract: str = GUEST_CONTRACT)

Explicit Yandex placement and one escaped static page with separate SSH authority.

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

admin_username instance-attribute

admin_username: str

admin_public_key class-attribute instance-attribute

admin_public_key: str = field(repr=False)

admin_source_cidrs instance-attribute

admin_source_cidrs: tuple[str, ...]

http_source_cidrs instance-attribute

http_source_cidrs: tuple[str, ...]

title class-attribute instance-attribute

title: str = field(repr=False)

body class-attribute instance-attribute

body: str = field(repr=False)

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 = 1

guest_contract class-attribute instance-attribute

guest_contract: str = GUEST_CONTRACT

StaticSitePlan dataclass

StaticSitePlan(profile: StaticSiteProfile)

Immutable profile intent whose serialized plan binds an exact owned initialization.

profile instance-attribute

profile: StaticSiteProfile

StaticSitePreparation dataclass

StaticSitePreparation(server_directory: Path, plan_directory: Path)

Local bundle references for a prepared plan; deployment and readiness remain unverified.

server_directory instance-attribute

server_directory: Path

plan_directory instance-attribute

plan_directory: Path

PlanFile property

PlanFile: Path

Locate the prepared TOML inside its separately owned private plan bundle.

ParseStaticSiteProfile

ParseStaticSiteProfile(data: Mapping[str, object]) -> StaticSiteProfile

Decode a strict concrete profile without accepting commands or private credentials.

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

def ParseStaticSiteProfile(data: Mapping[str, object]) -> StaticSiteProfile:
    """Decode a strict concrete profile without accepting commands or private credentials."""

    try:
        ValidateFields(data, frozenset({
            "schema_version", "profile", "guest_contract", "identity", "placement", "admin", "site",
        }), frozenset(), "static-site profile")
        ValidateSchemaVersion(data["schema_version"])

        if data["profile"] != PROFILE_NAME or data["guest_contract"] != GUEST_CONTRACT:
            raise StaticSiteError("profile and guest_contract must select the supported static site")

        placement = RequireTable(data["placement"], "placement")
        admin = RequireTable(data["admin"], "admin")
        site = RequireTable(data["site"], "site")
        ValidateFields(placement, _PLACEMENT_FIELDS, _SIZING_FIELDS, "placement")
        ValidateFields(admin, frozenset({"username", "public_key", "source_cidrs"}),
                       frozenset(), "admin")
        ValidateFields(site, frozenset({"title", "body", "http_source_cidrs"}),
                       frozenset(), "site")

        if not isinstance(admin["source_cidrs"], list) or not isinstance(site["http_source_cidrs"], list):
            raise StaticSiteError("Admin and HTTP source CIDRs must be explicit arrays")

        return StaticSiteProfile(
            identity=ParseIdentity(data["identity"]),
            zone_id=ValidateIdentifier(placement["zone_id"], "zone_id"),
            image_id=ValidateIdentifier(placement["image_id"], "image_id"),
            subnet_cidr=_Cidr(placement["subnet_cidr"], 16, "subnet_cidr"),
            admin_username=_Username(admin["username"]),
            admin_public_key=_PublicKey(admin["public_key"]),
            admin_source_cidrs=tuple(admin["source_cidrs"]),
            http_source_cidrs=tuple(site["http_source_cidrs"]),
            title=_Text(site["title"], 160, "site title"),
            body=_Text(site["body"], 8192, "site body", multiline=True),
            cores=_Integer(placement.get("cores", 2), 2, 32, "cores"),
            memory_gib=_Integer(placement.get("memory_gib", 2), 1, 64, "memory_gib"),
            boot_disk_gib=_Integer(placement.get("boot_disk_gib", 20), 10, 256, "boot_disk_gib"),
        )

    except ContractError as error:
        raise StaticSiteError(str(error)) from None

LoadStaticSiteProfile

LoadStaticSiteProfile(path: str | Path) -> StaticSiteProfile

Read bounded regular UTF-8 TOML without following symlinks or blocking on special files.

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

def LoadStaticSiteProfile(path: str | Path) -> StaticSiteProfile:
    """Read bounded regular UTF-8 TOML without following symlinks or blocking on special files."""

    descriptor: int | None = None

    try:
        profile_path = Path(path).absolute()

        if ".." in profile_path.parts or any(item.is_symlink() for item in (profile_path, *profile_path.parents)):
            raise StaticSiteError("Profile paths must not contain traversal or symlinks")

        descriptor = os.open(profile_path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)
                             | getattr(os, "O_NONBLOCK", 0))
        info = os.fstat(descriptor)

        if not stat.S_ISREG(info.st_mode) or not 1 <= info.st_size <= MAX_PROFILE_BYTES:
            raise StaticSiteError("Profile input must be a bounded regular file")

        with os.fdopen(descriptor, "rb") as source:
            descriptor = None
            raw = source.read(MAX_PROFILE_BYTES + 1)

        if len(raw) > MAX_PROFILE_BYTES:
            raise StaticSiteError("Static-site profile exceeds its bounded input size")

        data = tomllib.loads(raw.decode("utf-8"))

    except (OSError, ValueError, RecursionError):
        raise StaticSiteError("Unable to read a valid bounded static-site profile") from None

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

    return ParseStaticSiteProfile(data)

CompileStaticSite

CompileStaticSite(profile: StaticSiteProfile) -> StaticSitePlan

Validate composed names without invoking storage, lifecycle, or provider mutation.

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

def CompileStaticSite(profile: StaticSiteProfile) -> StaticSitePlan:
    """Validate composed names without invoking storage, lifecycle, or provider mutation."""

    plan = StaticSitePlan(profile)

    for logical_id in ("network", "subnet", "firewall", "address", "boot-disk", "instance", "plan"):
        ValidateName(profile.identity.stack + "-" + logical_id, "resource or bundle name")

    return plan

BuildStaticSiteBundle

BuildStaticSiteBundle(profile: StaticSiteProfile) -> ArtifactBundle

Create public initialization bytes without credentials or guest readiness claims.

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

def BuildStaticSiteBundle(profile: StaticSiteProfile) -> ArtifactBundle:
    """Create public initialization bytes without credentials or guest readiness claims."""

    CompileStaticSite(profile)

    return ArtifactBundle(profile.identity, "server", profile.identity.stack, (
        Artifact(CLOUD_INIT_FILENAME, _CloudInit(profile), sensitive=False),
    ))

RenderStaticSitePlan

RenderStaticSitePlan(plan: StaticSitePlan, *, user_data_file: str | Path) -> str

Bind the existing six-resource lifecycle contract to verified owned initialization.

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

def RenderStaticSitePlan(plan: StaticSitePlan, *, user_data_file: str | Path) -> str:
    """Bind the existing six-resource lifecycle contract to verified owned initialization."""

    if not isinstance(plan, StaticSitePlan):
        raise StaticSiteError("plan must be a StaticSitePlan")

    profile = plan.profile
    artifact_path = Path(user_data_file)
    expected = _CloudInit(profile)

    if artifact_path.name != CLOUD_INIT_FILENAME:
        raise StaticSiteError("Initialization must reference the exact static-site artifact filename")

    try:
        actual = ReadOwnedArtifactFile(artifact_path, profile.identity, kind="server",
                                       name=profile.identity.stack)

    except ContractError:
        raise StaticSiteError("Prepared static-site artifact ownership or integrity is invalid") from None

    if actual != expected:
        raise StaticSiteError("Initialization artifact does not match the compiled static-site profile")

    def Quote(value: str) -> str:
        """Serialize Unicode scalar strings as valid TOML without path expansion."""

        return json.dumps(value, ensure_ascii=False)

    lines = ["schema_version = 1", f"profile = {Quote(PROFILE_NAME)}", "", "[identity]"]

    for key in sorted(IDENTITY_FIELDS):
        lines.append(f"{key} = {Quote(getattr(profile.identity, key))}")

    def AddResource(
        logical_id: str, kind: str, dependencies: tuple[str, ...], parameters: Mapping[str, object],
    ) -> None:
        """Render deterministic scalar resource intent without observed cloud identifiers."""

        lines.extend([
            "", "[[resources]]", f"logical_id = {Quote(logical_id)}", f"kind = {Quote(kind)}",
            f"name = {Quote(ValidateName(profile.identity.stack + '-' + logical_id, 'resource name'))}",
            "dependencies = [" + ", ".join(Quote(item) for item in dependencies) + "]",
            "[resources.parameters]",
        ])

        for key, value in parameters.items():
            if isinstance(value, str):
                rendered = Quote(value)

            elif type(value) is int:
                rendered = str(value)

            else:
                raise StaticSiteError("Resource parameters must be explicit scalar values")

            lines.append(f"{key} = {rendered}")

    AddResource("network", "network", (), {})
    AddResource("subnet", "subnet", ("network",), {
        "zone_id": profile.zone_id, "ipv4_cidr": profile.subnet_cidr,
        "network_dependency": "network",
    })
    AddResource("firewall", "security-group", ("network",), {"network_dependency": "network"})

    for port, cidrs in ((22, profile.admin_source_cidrs), (80, profile.http_source_cidrs)):
        for cidr in cidrs:
            lines.extend([
                "[[resources.parameters.rules]]", 'direction = "ingress"', 'protocol = "tcp"',
                f"from_port = {port}", f"to_port = {port}", f"cidr = {Quote(cidr)}",
            ])

    lines.extend([
        "[[resources.parameters.rules]]", 'direction = "egress"', 'protocol = "any"',
        'cidr = "0.0.0.0/0"',
    ])
    AddResource("address", "address", (), {"zone_id": profile.zone_id})
    AddResource("boot-disk", "disk", (), {
        "zone_id": profile.zone_id, "image_id": profile.image_id, "size_gib": profile.boot_disk_gib,
    })
    AddResource("instance", "instance", ("subnet", "firewall", "address", "boot-disk"), {
        "zone_id": profile.zone_id, "cores": profile.cores, "memory_gib": profile.memory_gib,
        "boot_disk_dependency": "boot-disk", "subnet_dependency": "subnet",
        "security_group_dependency": "firewall", "address_dependency": "address",
        "ssh_public_key": profile.admin_public_key, "ssh_username": profile.admin_username,
        "user_data_file": str(artifact_path.absolute()),
        "user_data_sha256": hashlib.sha256(expected).hexdigest(),
    })

    return "\n".join(lines) + "\n"

PrepareStaticSite

PrepareStaticSite(profile: StaticSiteProfile, *, artifact_root: str | Path) -> StaticSitePreparation

Publish exclusive owned bundles; preserve a verified plan after publication failure.

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

def PrepareStaticSite(profile: StaticSiteProfile, *, artifact_root: str | Path) -> StaticSitePreparation:
    """Publish exclusive owned bundles; preserve a verified plan after publication failure."""

    plan = CompileStaticSite(profile)
    server_bundle = BuildStaticSiteBundle(profile)
    server_directory = WriteArtifactBundle(artifact_root, server_bundle)
    plan_text: str | None = None

    try:
        plan_text = RenderStaticSitePlan(plan, user_data_file=server_directory / CLOUD_INIT_FILENAME)
        plan_directory = WriteArtifactBundle(artifact_root, ArtifactBundle(
            profile.identity, "server", profile.identity.stack + "-plan", (
                Artifact(PLAN_FILENAME, plan_text.encode("utf-8"), sensitive=False),
            ),
        ))

    except BaseException:
        published_plan = False

        if plan_text is not None:
            try:
                actual = ReadOwnedArtifactFile(
                    Path(artifact_root) / ("server-" + profile.identity.stack + "-plan") / PLAN_FILENAME,
                    profile.identity, kind="server", name=profile.identity.stack + "-plan",
                )
                published_plan = actual == plan_text.encode("utf-8")

            except ContractError:
                pass

        if not published_plan:
            RemoveArtifactBundle(
                artifact_root, profile.identity, kind="server", name=profile.identity.stack,
                expected_bundle=server_bundle,
            )

        raise

    return StaticSitePreparation(server_directory, plan_directory)

Main

Main(argv: Sequence[str] | None = None) -> int

Prepare bounded local artifacts only; never instantiate or invoke a cloud provider.

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

def Main(argv: Sequence[str] | None = None) -> int:
    """Prepare bounded local artifacts only; never instantiate or invoke a cloud provider."""

    parser = argparse.ArgumentParser(description="Prepare an owned static-site lifecycle plan locally")
    parser.add_argument("profile_file", help="Explicit version-one static-site TOML profile")
    parser.add_argument("--artifact-root", required=True, help="Private POSIX artifact root")
    arguments = parser.parse_args(argv)

    try:
        prepared = PrepareStaticSite(LoadStaticSiteProfile(arguments.profile_file),
                                     artifact_root=arguments.artifact_root)

    except ContractError:
        print(json.dumps({"status": "failed", "reason": "Profile or owned artifact preparation failed"}))

        return 1

    print(json.dumps({"status": "prepared-unverified", "plan_file": str(prepared.PlanFile.absolute())},
                     sort_keys=True))

    return 0