flayer.vpn_project

Module source

flayer.vpn_project

Private local VPN projects composed with owned Yandex lifecycle and trusted guest provisioning.

ADMIN_USERNAME module-attribute

ADMIN_USERNAME = 'flayer-admin'

MAX_SERIAL_BYTES module-attribute

MAX_SERIAL_BYTES = 1024 * 1024

PROJECT_TEMPLATE module-attribute

PROJECT_TEMPLATE = '# Edit the folder, image, management address, and Reality target before preparation.\n# Project files contain private keys and must remain outside Git and shared directories.\nschema_version = 1\n\n[identity]\nproject = "my-vpn"\nstack = "gateway"\nprovider = "yandex-cloud"\nscope_id = "replace-with-folder-id"\nowner_id = "my-team"\n\n[cloud]\nyc_profile = "default"\nzone_id = "ru-central1-a"\nimage_id = "replace-with-ubuntu-2404-amd64-image-id"\nsubnet_cidr = "10.42.0.0/24"\nmanagement_cidrs = ["198.51.100.42/32"] # Replace with your current public IPv4 address.\n\n# Optional explicit sizing: uncomment the entire block before prepare.\n# Confirm regional availability and image minimum disk size first.\n# [resources]\n# platform_id = "standard-v3"\n# cores = 2\n# core_fraction = 50\n# memory_gib = 2\n# disk_size_gib = 10\n# disk_type = "network-hdd"\n\n[routes]\nmode = "full"\nipv4_cidrs = ["0.0.0.0/0"]\ndns_servers = ["1.1.1.1"]\nipv6_policy = "disabled" # Disable client IPv6 to prevent traffic bypass.\n\n[[devices]]\ndevice_id = "laptop"\nipv4_address = "10.66.0.2"\n\n[amneziawg]\nport = 51820\ntunnel_cidr = "10.66.0.0/24"\n\n[vless]\nport = 443\n# Example only: choose an accessible TLS 1.3 / HTTP/2 site for your server location.\ntarget_host = "www.microsoft.com"\nserver_name = "www.microsoft.com"\n'

CloudPlacement dataclass

CloudPlacement(yc_profile: str, zone_id: str, image_id: str, subnet_cidr: str, management_cidrs: tuple[str, ...])

Public placement and narrowly scoped administrator ingress for one Ubuntu guest.

yc_profile instance-attribute

yc_profile: str

zone_id instance-attribute

zone_id: str

image_id instance-attribute

image_id: str

subnet_cidr instance-attribute

subnet_cidr: str

management_cidrs instance-attribute

management_cidrs: tuple[str, ...]

VpnResources dataclass

VpnResources(platform_id: str, cores: int, core_fraction: int, memory_gib: int, disk_size_gib: int, disk_type: str)

Explicit initial server sizing; absence preserves the legacy project contract.

platform_id instance-attribute

platform_id: str

cores instance-attribute

cores: int

core_fraction instance-attribute

core_fraction: int

memory_gib instance-attribute

memory_gib: int

disk_size_gib instance-attribute

disk_size_gib: int

disk_type instance-attribute

disk_type: str

VpnProject dataclass

VpnProject(root: Path, vpn: VpnProfile, cloud: CloudPlacement, amneziawg: AmneziaWgSettings, vless: VlessRealitySettings, ssh_public_key: str, resources: VpnResources | None = None)

Validated public project intent with explicit private local storage references.

root class-attribute instance-attribute

root: Path = field(repr=False)

vpn instance-attribute

vpn: VpnProfile

cloud instance-attribute

cloud: CloudPlacement

amneziawg instance-attribute

amneziawg: AmneziaWgSettings

vless instance-attribute

vless: VlessRealitySettings

ssh_public_key class-attribute instance-attribute

ssh_public_key: str = field(repr=False)

resources class-attribute instance-attribute

resources: VpnResources | None = None

Artifacts property

Artifacts: Path

Locate the existing private owned-artifact store within this project.

Fingerprint

Fingerprint() -> str

Bind prepared material to all public project settings without persisting secrets.

Source code in installed/flayer/vpn_project.py

def Fingerprint(self) -> str:
    """Bind prepared material to all public project settings without persisting secrets."""

    payload: dict[str, object] = {
        "vpn": asdict(self.vpn), "cloud": asdict(self.cloud),
        "amneziawg": asdict(self.amneziawg), "vless": asdict(self.vless),
        "ssh_public_key": self.ssh_public_key,
    }

    if self.resources is not None:
        payload["resources"] = asdict(self.resources)

    return hashlib.sha256(json.dumps(payload, sort_keys=True).encode("utf-8")).hexdigest()

VpnProjectReport dataclass

VpnProjectReport(action: str, status: str, cloud_status: str = 'not-requested', guest_status: str = 'not-requested', client_exports: tuple[str, ...] = (), recovery_required: bool = False, details: str = '', connectivity_status: str = 'not-verified')

Sanitized lifecycle and guest observations without transport material or provider output.

action instance-attribute

action: str

status instance-attribute

status: str

cloud_status class-attribute instance-attribute

cloud_status: str = 'not-requested'

guest_status class-attribute instance-attribute

guest_status: str = 'not-requested'

client_exports class-attribute instance-attribute

client_exports: tuple[str, ...] = ()

recovery_required class-attribute instance-attribute

recovery_required: bool = False

details class-attribute instance-attribute

details: str = ''

connectivity_status class-attribute instance-attribute

connectivity_status: str = 'not-verified'

ExitCode

ExitCode() -> int

Treat incomplete cloud or guest operations as unsuccessful without claiming connectivity.

Source code in installed/flayer/vpn_project.py

def ExitCode(self) -> int:
    """Treat incomplete cloud or guest operations as unsuccessful without claiming connectivity."""

    return 0 if self.status == "complete" else 1

VpnYandexProvider

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

Bases: YandexLifecycleProvider

Add one authenticated read-only public host-key channel to the existing lifecycle adapter.

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

ReadHostKey

ReadHostKey(instance: ProviderResource, fingerprint: str) -> str | None

Extract one exact bootstrap marker without logging unrelated serial-console contents.

Source code in installed/flayer/vpn_project.py

def ReadHostKey(self, instance: ProviderResource, fingerprint: str) -> str | None:
    """Extract one exact bootstrap marker without logging unrelated serial-console contents."""

    observed = self.GetResource(instance.reference)

    if observed.reference != instance.reference or observed.labels != instance.labels:
        raise VpnError("Instance ownership changed before authenticated host-key retrieval")

    payload = self._ReadJson(("compute", "instance", "get-serial-port-output", "--id", instance.reference.resource_id, "--port", "1"), "guest-host-key")

    if not isinstance(payload, dict) or not isinstance(payload.get("contents"), str):
        raise VpnError("Authenticated serial-console response is unavailable or malformed")

    contents = payload["contents"]

    try:
        if len(contents.encode("utf-8")) > MAX_SERIAL_BYTES:
            raise VpnError("Authenticated serial-console response exceeds its bounded size")

        marker = "FLAYER_HOSTKEY " + fingerprint + " "
        keys = {_PublicKey(line[len(marker):].encode("ascii")) for line in contents.splitlines() if line.startswith(marker)}

    except UnicodeError:
        raise VpnError("Authenticated serial-console host-key marker contains invalid encoding") from None

    if len(keys) > 1:
        raise VpnError("Authenticated serial console contains conflicting host-key pins")

    return next(iter(keys)) if keys else None

InitializeVpnProject

InitializeVpnProject(directory: str | Path) -> Path

Create one exclusive private project and an administration key without cloud access.

Source code in installed/flayer/vpn_project.py

def InitializeVpnProject(directory: str | Path) -> Path:
    """Create one exclusive private project and an administration key without cloud access."""

    path = Path(directory).absolute()
    ValidateSshPathSyntax(path)

    if os.name != "posix" or ".." in path.parts or any(item.is_symlink() for item in (path, *path.parents)):
        raise VpnError("Project initialization requires a POSIX path without symlinks or traversal")

    try:
        path.mkdir(mode=0o700)

    except OSError:
        raise VpnError("Project directory must be new and its parent must already exist") from None

    WritePrivate(path / "project.toml", PROJECT_TEMPLATE.encode("utf-8"))
    WritePrivate(path / ".gitignore", b"*\n")

    try:
        result = subprocess.run(
            ("ssh-keygen", "-q", "-t", "ed25519", "-N", "", "-C", "flayer-admin", "-f", str(path / "admin-key")),
            stdin=subprocess.DEVNULL, capture_output=True, check=False, timeout=30,
        )

        if result.returncode != 0:
            raise VpnError("Unable to generate the project administration key; inspect the private project directory")

        (path / "admin-key.pub").chmod(0o600)
        PrivatePath(path / "admin-key")
        _PublicKey(ReadPrivate(path / "admin-key.pub"))
        (path / "artifacts").mkdir(mode=0o700)

    except (OSError, subprocess.TimeoutExpired):
        raise VpnError("Project initialization requires the OpenSSH ssh-keygen command") from None

    return path

LoadVpnProject

LoadVpnProject(directory: str | Path) -> VpnProject

Read explicit editable public TOML and administration public key from private storage.

Source code in installed/flayer/vpn_project.py

def LoadVpnProject(directory: str | Path) -> VpnProject:
    """Read explicit editable public TOML and administration public key from private storage."""

    ValidateSshPathSyntax(Path(directory).absolute())
    root = PrivatePath(Path(directory), directory=True)
    PrivatePath(root / "artifacts", directory=True)

    try:
        data = tomllib.loads(ReadPrivate(root / "project.toml").decode("utf-8"))
        ValidateFields(data, frozenset({
            "schema_version", "identity", "cloud", "routes", "devices", "amneziawg", "vless",
        }), frozenset({"resources"}), "VPN project")
        ValidateSchemaVersion(data["schema_version"])
        cloud = _Table(data, "cloud", frozenset({"yc_profile", "zone_id", "image_id", "subnet_cidr", "management_cidrs"}))
        awg = _Table(data, "amneziawg", frozenset({"port", "tunnel_cidr"}))
        vless = _Table(data, "vless", frozenset({"port", "target_host", "server_name"}))
        identity = ParseIdentity(data["identity"])

        if identity.provider != "yandex-cloud" or identity.scope_id.startswith("replace-with-"):
            raise VpnError("Project requires an explicit Yandex Cloud folder ID")

        if not isinstance(cloud["management_cidrs"], list) or any(
            not isinstance(cloud[key], str) for key in ("yc_profile", "zone_id", "image_id", "subnet_cidr")
        ):
            raise VpnError("Cloud placement requires string settings and a management CIDR array")

        if not isinstance(awg["tunnel_cidr"], str) or not isinstance(vless["target_host"], str) or not isinstance(vless["server_name"], str):
            raise VpnError("Transport network and TLS target settings must be strings")

        vpn = ParseVpnProfile({
            "schema_version": 1, "identity": data["identity"], "routes": data["routes"],
            "devices": data["devices"], "listeners": [
                {"transport": "amneziawg", "protocol": "udp", "port": awg["port"], "source_cidrs": ["0.0.0.0/0"]},
                {"transport": "vless-reality", "protocol": "tcp", "port": vless["port"], "source_cidrs": ["0.0.0.0/0"]},
            ],
        })

        if vless["port"] == 22:
            raise VpnError("VLESS listener must not collide with the administration SSH port")

        for suffix in ("credentials", "bootstrap", "trust"):
            ValidateName(identity.stack + "-" + suffix, "project artifact name")

        placement = CloudPlacement(
            cast(str, cloud["yc_profile"]), cast(str, cloud["zone_id"]), cast(str, cloud["image_id"]),
            cast(str, cloud["subnet_cidr"]), tuple(cloud["management_cidrs"]),
        )
        resources = None

        if "resources" in data:
            sizing = _Table(data, "resources", frozenset({
                "platform_id", "cores", "core_fraction", "memory_gib", "disk_size_gib", "disk_type",
            }))
            resources = VpnResources(
                cast(str, sizing["platform_id"]), cast(int, sizing["cores"]),
                cast(int, sizing["core_fraction"]), cast(int, sizing["memory_gib"]),
                cast(int, sizing["disk_size_gib"]), cast(str, sizing["disk_type"]),
            )

            if placement.zone_id.startswith("kz1-") and resources.platform_id != "standard-v3":
                raise VpnError("Kazakhstan VPN deployments require platform_id standard-v3")

        settings = AmneziaWgSettings(awg["tunnel_cidr"])

        if ipaddress.IPv4Network(settings.tunnel_cidr).overlaps(ipaddress.IPv4Network(placement.subnet_cidr)):
            raise VpnError("VPN tunnel and cloud subnet must not overlap")

        return VpnProject(
            root, vpn, placement, settings, VlessRealitySettings(vless["target_host"], vless["server_name"]),
            _PublicKey(ReadPrivate(root / "admin-key.pub")), resources,
        )

    except (ValueError, RecursionError) as error:
        if isinstance(error, VpnError):
            raise

        raise VpnError("Project configuration is invalid; check its documented fields and types") from None

PrepareVpnProject

PrepareVpnProject(project: VpnProject) -> VpnProjectReport

Prepare private materials and verify both renderers locally without exporting false endpoints.

Source code in installed/flayer/vpn_project.py

def PrepareVpnProject(project: VpnProject) -> VpnProjectReport:
    """Prepare private materials and verify both renderers locally without exporting false endpoints."""

    ValidateSshPathSyntax(project.root)
    _VerifyAdministrationKey(project)
    material = _MaterialBundle(project)
    transports = _Transports(project, material, "192.0.2.1")
    bootstrap = _Bootstrap(project)
    _Publish(project, material)

    for transport in transports:
        _Publish(project, transport.server_bundle)

    _Publish(project, bootstrap)

    return VpnProjectReport("prepare", "complete", details="Local server artifacts prepared; client exports follow actual address allocation")

CompileVpnProject

CompileVpnProject(project: VpnProject) -> DeploymentPlan

Compile six owned resources while binding only the verified nonsecret bootstrap file.

Source code in installed/flayer/vpn_project.py

def CompileVpnProject(project: VpnProject) -> DeploymentPlan:
    """Compile six owned resources while binding only the verified nonsecret bootstrap file."""

    bundle = _Bootstrap(project)
    directory = project.Artifacts / f"server-{bundle.name}"
    path = directory / "cloud-init.json"
    actual = ReadOwnedArtifactFile(path, project.vpn.identity, kind="server", name=bundle.name)

    if actual != bundle.files[0].content:
        raise VpnError("Prepared bootstrap does not match the current public project settings")

    resources: list[ResourceSpec] = []

    def Add(logical_id: str, kind: ResourceKind, dependencies: tuple[str, ...], options: dict[str, object]) -> None:
        """Append one exact typed resource to the owned dependency graph."""

        resources.append(ResourceSpec(logical_id, kind, ValidateName(project.vpn.identity.stack + "-" + logical_id, "resource name"),
                                      tuple((key, _Freeze(value)) for key, value in options.items()), dependencies))

    Add("network", ResourceKind.NETWORK, (), {})
    Add("subnet", ResourceKind.SUBNET, ("network",), {
        "zone_id": project.cloud.zone_id, "ipv4_cidr": project.cloud.subnet_cidr, "network_dependency": "network",
    })
    rules: list[object] = [{"direction": "ingress", "protocol": "tcp", "from_port": 22, "to_port": 22, "cidr": cidr}
                           for cidr in project.cloud.management_cidrs]
    rules.extend({"direction": "ingress", "protocol": listener.protocol, "from_port": listener.port,
                  "to_port": listener.port, "cidr": cidr} for listener in project.vpn.listeners for cidr in listener.source_cidrs)
    rules.append({"direction": "egress", "protocol": "any", "cidr": "0.0.0.0/0"})
    Add("firewall", ResourceKind.SECURITY_GROUP, ("network",), {"network_dependency": "network", "rules": rules})
    Add("address", ResourceKind.ADDRESS, (), {"zone_id": project.cloud.zone_id})
    disk: dict[str, object] = {"zone_id": project.cloud.zone_id, "image_id": project.cloud.image_id, "size_gib": 20}
    instance: dict[str, object] = {
        "zone_id": project.cloud.zone_id, "cores": 2, "memory_gib": 2, "boot_disk_dependency": "boot-disk",
        "subnet_dependency": "subnet", "security_group_dependency": "firewall", "address_dependency": "address",
        "ssh_public_key": project.ssh_public_key, "ssh_username": ADMIN_USERNAME, "user_data_file": str(path),
        "user_data_sha256": hashlib.sha256(actual).hexdigest(),
    }

    if project.resources is not None:
        sizing = project.resources
        disk.update(size_gib=sizing.disk_size_gib, type=sizing.disk_type)
        instance.update(cores=sizing.cores, memory_gib=sizing.memory_gib,
                        platform_id=sizing.platform_id, core_fraction=sizing.core_fraction)

    Add("boot-disk", ResourceKind.DISK, (), disk)
    Add("instance", ResourceKind.INSTANCE, ("subnet", "firewall", "address", "boot-disk"), instance)

    return DeploymentPlan(project.vpn.identity, tuple(resources))

RunVpnProject

RunVpnProject(project: VpnProject, action: str, *, allow_mutation: bool = False, scope_confirm: str = '', rollback: bool = False, provider: VpnYandexProvider | None = None, guest_factory: Callable[..., SshGuestProvisioner] = SshGuestProvisioner) -> VpnProjectReport

Compose explicit cloud consent, durable resources, authenticated guest install, and client export.

Source code in installed/flayer/vpn_project.py

def RunVpnProject(
    project: VpnProject, action: str, *, allow_mutation: bool = False, scope_confirm: str = "",
    rollback: bool = False, provider: VpnYandexProvider | None = None,
    guest_factory: Callable[..., SshGuestProvisioner] = SshGuestProvisioner,
) -> VpnProjectReport:
    """Compose explicit cloud consent, durable resources, authenticated guest install, and client export."""

    ValidateSshPathSyntax(project.root)

    if action not in {"prepare", "deploy", "status", "destroy", "recover"}:
        raise VpnError("VPN project action is unsupported")

    if action == "prepare":
        return PrepareVpnProject(project)

    if action in {"deploy", "destroy", "recover"} and (not allow_mutation or scope_confirm != project.vpn.identity.scope_id):
        raise VpnError("Cloud mutation requires --allow-mutation and exact --scope-confirm folder ID")

    if action == "deploy":
        state = LoadState(project.root / "state.json", project.vpn.identity)
        old_trust = project.Artifacts / f"server-{project.vpn.identity.stack}-trust"
        old_exports = tuple(project.Artifacts / f"device-{device.device_id}-{listener.transport}"
                            for device in project.vpn.devices for listener in project.vpn.listeners)

        if (state is None or not state.resources) and any(path.exists() or path.is_symlink() for path in (old_trust, *old_exports)):
            raise VpnError("Previous server trust or client exports remain after cloud cleanup. Initialize a new private project before another deployment")

        PrepareVpnProject(project)

    plan = CompileVpnProject(project)
    provider = provider or VpnYandexProvider(YandexCloudSettings(project.vpn.identity.scope_id, project.cloud.yc_profile))
    engine = LifecycleEngine(plan, provider, project.root / "state.json")
    _CheckCloudAccess(provider)

    if action == "destroy":
        report = engine.Destroy()

        if report.status != "complete":
            return _Incomplete(action, report)

        return VpnProjectReport(action, "complete", "complete", "destroyed", details="Cloud resources removed; private local artifacts retained")

    if action == "recover":
        report = engine.Recover(rollback=rollback)

        if report.status != "complete":
            return _Incomplete(action, report)

        return VpnProjectReport(action, "complete", "complete", details="Cloud recovery completed; run deploy to provision and check both VPN services")

    report = engine.Create() if action == "deploy" else engine.Status()

    if report.status != "complete":
        return _Incomplete(action, report)

    instance, endpoint = _OwnedEndpoints(project, provider)
    trust = _HostTrust(project, provider, instance, endpoint, wait=action == "deploy")
    material = _MaterialBundle(project)
    transports = _Transports(project, material, endpoint)
    guest = guest_factory(VpnEndpoint(endpoint, 22), ADMIN_USERNAME, project.root / "admin-key", trust)

    if action == "deploy":
        for transport in transports:
            _Publish(project, transport.server_bundle)

        guest.Install(tuple((item.capabilities.transport, item.server_bundle) for item in transports))

    services = (AmneziaWgServiceName(project.vpn) + ".service", VlessServiceName(project.vpn))
    observations = guest.ServiceStatus(services)

    if not all(active for _, active in observations):
        return VpnProjectReport(action, "incomplete", "complete", "inactive", details="Cloud resources exist but a VPN service is not active")

    exports = []

    if action == "deploy":
        for transport in transports:
            for bundle in transport.device_bundles:
                directory = _Publish(project, bundle)
                exports.append(str(directory.relative_to(project.root)))

    return VpnProjectReport(action, "complete", "complete", "services-active", tuple(exports),
                            details="Guest services are active; authenticated client traffic, DNS and Internet access are not verified")