flayer.profiles.vpn

Module source

flayer.profiles.vpn

Transport-neutral VPN intent, private artifact ownership, and replacement checks.

VPN_SCHEMA_VERSION module-attribute

VPN_SCHEMA_VERSION = 1

MAX_VPN_PROFILE_BYTES module-attribute

MAX_VPN_PROFILE_BYTES = 65536

MAX_VPN_DEVICES module-attribute

MAX_VPN_DEVICES = 16

MAX_VPN_TRANSPORTS module-attribute

MAX_VPN_TRANSPORTS = 8

VpnError

Bases: ContractError

VPN intent or artifact ownership is invalid before any external operation.

VpnDevice dataclass

VpnDevice(device_id: str, ipv4_address: str | None = None)

One named client; an optional private IPv4 address belongs to an IP tunnel.

device_id instance-attribute

device_id: str

ipv4_address class-attribute instance-attribute

ipv4_address: str | None = None

VpnRoutes dataclass

VpnRoutes(mode: str, ipv4_cidrs: tuple[str, ...], dns_servers: tuple[str, ...], ipv6_policy: str)

IPv4 destination policy; clients must disable IPv6 outside this initial contract.

mode instance-attribute

mode: str

ipv4_cidrs instance-attribute

ipv4_cidrs: tuple[str, ...]

dns_servers instance-attribute

dns_servers: tuple[str, ...]

ipv6_policy instance-attribute

ipv6_policy: str

VpnEndpoint dataclass

VpnEndpoint(host: str, port: int)

Public transport destination; a DNS name still needs adapter-side IPv4 resolution.

host instance-attribute

host: str

port instance-attribute

port: int

VpnListener dataclass

VpnListener(transport: str, protocol: str, port: int, source_cidrs: tuple[str, ...])

One explicit TCP or UDP ingress allowance for a named transport implementation.

transport instance-attribute

transport: str

protocol instance-attribute

protocol: str

port instance-attribute

port: int

source_cidrs instance-attribute

source_cidrs: tuple[str, ...]

VpnCapabilities dataclass

VpnCapabilities(transport: str, network_mode: str, route_modes: tuple[str, ...], ipv4: bool = True, ipv6: bool = False)

Transport-declared traffic scope; an application proxy is not an OS-wide tunnel.

transport instance-attribute

transport: str

network_mode instance-attribute

network_mode: str

route_modes instance-attribute

route_modes: tuple[str, ...]

ipv4 class-attribute instance-attribute

ipv4: bool = True

ipv6 class-attribute instance-attribute

ipv6: bool = False

VpnProfile dataclass

VpnProfile(identity: StackIdentity, devices: tuple[VpnDevice, ...], routes: VpnRoutes, listeners: tuple[VpnListener, ...], schema_version: int = VPN_SCHEMA_VERSION)

Public device and routing intent owned by one exact infrastructure stack.

identity instance-attribute

identity: StackIdentity

devices instance-attribute

devices: tuple[VpnDevice, ...]

routes instance-attribute

routes: VpnRoutes

listeners instance-attribute

listeners: tuple[VpnListener, ...]

schema_version class-attribute instance-attribute

schema_version: int = VPN_SCHEMA_VERSION

Fingerprint

Fingerprint() -> str

Hash only public intent for binding prepared artifacts to an exact profile.

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

def Fingerprint(self) -> str:
    """Hash only public intent for binding prepared artifacts to an exact profile."""

    payload = json.dumps(asdict(self), sort_keys=True, separators=(",", ":"))

    return hashlib.sha256(payload.encode("utf-8")).hexdigest()

PreparedVpnTransport dataclass

PreparedVpnTransport(profile: VpnProfile, capabilities: VpnCapabilities, server_bundle: ArtifactBundle, device_bundles: tuple[ArtifactBundle, ...])

Sensitive server and client artifacts; preparation is not deployment or health evidence.

profile instance-attribute

profile: VpnProfile

capabilities instance-attribute

capabilities: VpnCapabilities

server_bundle class-attribute instance-attribute

server_bundle: ArtifactBundle = field(repr=False)

device_bundles class-attribute instance-attribute

device_bundles: tuple[ArtifactBundle, ...] = field(repr=False)

Fingerprint

Fingerprint() -> str

Bind replacement approval to public intent and exact private payload digests.

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

def Fingerprint(self) -> str:
    """Bind replacement approval to public intent and exact private payload digests."""

    bundles = (self.server_bundle, *self.device_bundles)
    document = {
        "profile": self.profile.Fingerprint(), "capabilities": asdict(self.capabilities),
        "artifacts": sorted((
            bundle.kind, bundle.name, artifact.name,
            hashlib.sha256(artifact.content).hexdigest(),
        ) for bundle in bundles for artifact in bundle.files),
    }
    payload = json.dumps(document, sort_keys=True, separators=(",", ":"))

    return hashlib.sha256(payload.encode("utf-8")).hexdigest()

ValidateVpnTransport

ValidateVpnTransport(profile: VpnProfile, capabilities: VpnCapabilities) -> VpnListener

Require the selected implementation to support this exact declared routing intent.

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

def ValidateVpnTransport(profile: VpnProfile, capabilities: VpnCapabilities) -> VpnListener:
    """Require the selected implementation to support this exact declared routing intent."""

    if not isinstance(profile, VpnProfile) or not isinstance(capabilities, VpnCapabilities):
        raise VpnError("VPN preparation requires a VpnProfile and VpnCapabilities")

    listener = next((item for item in profile.listeners if item.transport == capabilities.transport), None)

    if listener is None:
        raise VpnError("VPN transport must have an explicitly declared listener")

    if profile.routes.mode not in capabilities.route_modes:
        raise VpnError("VPN implementation does not support the requested routing mode")

    if capabilities.network_mode == "ip-tunnel" and any(item.ipv4_address is None for item in profile.devices):
        raise VpnError("IP tunnel devices require explicit private IPv4 addresses")

    return listener

BuildVpnServerBundle

BuildVpnServerBundle(profile: VpnProfile, transport: str, files: tuple[Artifact, ...]) -> ArtifactBundle

Prepare private server bytes for authenticated guest provisioning, never cloud metadata.

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

def BuildVpnServerBundle(
    profile: VpnProfile, transport: str, files: tuple[Artifact, ...],
) -> ArtifactBundle:
    """Prepare private server bytes for authenticated guest provisioning, never cloud metadata."""

    _Transport(profile, transport)
    _SensitiveFiles(files)

    return ArtifactBundle(profile.identity, "server", f"{profile.identity.stack}-{transport}", files)

BuildVpnDeviceBundle

BuildVpnDeviceBundle(profile: VpnProfile, transport: str, device_id: str, files: tuple[Artifact, ...]) -> ArtifactBundle

Prepare owned client bytes without overwriting or revoking existing credentials.

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

def BuildVpnDeviceBundle(
    profile: VpnProfile, transport: str, device_id: str, files: tuple[Artifact, ...],
) -> ArtifactBundle:
    """Prepare owned client bytes without overwriting or revoking existing credentials."""

    _Transport(profile, transport)
    _Name(device_id, "device_id")
    _SensitiveFiles(files)

    if not any(item.device_id == device_id for item in profile.devices):
        raise VpnError("VPN client artifacts require an explicitly declared device")

    return ArtifactBundle(profile.identity, "device", f"{device_id}-{transport}", files)

ValidateVpnReplacement

ValidateVpnReplacement(previous: PreparedVpnTransport, replacement: PreparedVpnTransport, *, expected_fingerprint: str) -> None

Require exact old artifact consent; guest revocation and local cleanup remain separate.

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

def ValidateVpnReplacement(
    previous: PreparedVpnTransport, replacement: PreparedVpnTransport, *, expected_fingerprint: str,
) -> None:
    """Require exact old artifact consent; guest revocation and local cleanup remain separate."""

    if not isinstance(previous, PreparedVpnTransport) or not isinstance(replacement, PreparedVpnTransport):
        raise VpnError("VPN replacement requires prepared previous and replacement artifacts")

    if previous.profile.identity != replacement.profile.identity or previous.capabilities.transport != replacement.capabilities.transport:
        raise VpnError("VPN replacement must preserve exact stack identity and transport")

    if not isinstance(expected_fingerprint, str) or previous.Fingerprint() != expected_fingerprint:
        raise VpnError("VPN replacement requires the exact expected previous fingerprint")

ParseVpnProfile

ParseVpnProfile(data: Mapping[str, object]) -> VpnProfile

Decode only public VPN intent and reject unknown fields, including inline credentials.

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

def ParseVpnProfile(data: Mapping[str, object]) -> VpnProfile:
    """Decode only public VPN intent and reject unknown fields, including inline credentials."""

    try:
        ValidateFields(data, frozenset({
            "schema_version", "identity", "devices", "routes", "listeners",
        }), frozenset(), "VPN profile")
        ValidateSchemaVersion(data["schema_version"])
        routes = RequireTable(data["routes"], "VPN routes")
        ValidateFields(routes, frozenset({
            "mode", "ipv4_cidrs", "dns_servers", "ipv6_policy",
        }), frozenset(), "VPN routes")

        if (
            not isinstance(routes["mode"], str) or not isinstance(routes["ipv6_policy"], str)
            or not isinstance(routes["ipv4_cidrs"], list) or not isinstance(routes["dns_servers"], list)
            or not isinstance(data["devices"], list) or not isinstance(data["listeners"], list)
        ):
            raise VpnError("VPN profile requires string policies and explicit array values")

        devices: list[VpnDevice] = []
        listeners: list[VpnListener] = []

        for raw_device in data["devices"]:
            device = RequireTable(raw_device, "VPN device")
            ValidateFields(device, frozenset({"device_id"}), frozenset({"ipv4_address"}), "VPN device")
            address = device.get("ipv4_address")

            if address is not None:
                address = _Address(address, "device IPv4 address")

            devices.append(VpnDevice(_Name(device["device_id"], "device_id"), address))

        for raw_listener in data["listeners"]:
            listener = RequireTable(raw_listener, "VPN listener")
            ValidateFields(listener, frozenset({
                "transport", "protocol", "port", "source_cidrs",
            }), frozenset(), "VPN listener")

            if not isinstance(listener["protocol"], str) or not isinstance(listener["source_cidrs"], list):
                raise VpnError("VPN listener requires a protocol string and a CIDR array")

            listeners.append(VpnListener(
                _Name(listener["transport"], "transport"), listener["protocol"],
                _Port(listener["port"]), tuple(listener["source_cidrs"]),
            ))

        return VpnProfile(
            ParseIdentity(data["identity"]), tuple(devices),
            VpnRoutes(routes["mode"], tuple(routes["ipv4_cidrs"]), tuple(routes["dns_servers"]), routes["ipv6_policy"]),
            tuple(listeners),
        )

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

LoadVpnProfile

LoadVpnProfile(path: str | Path) -> VpnProfile

Read a bounded public TOML profile without reading credentials or contacting a provider.

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

def LoadVpnProfile(path: str | Path) -> VpnProfile:
    """Read a bounded public TOML profile without reading credentials or contacting a provider."""

    try:
        with Path(path).open("rb") as stream:
            content = stream.read(MAX_VPN_PROFILE_BYTES + 1)

        if len(content) > MAX_VPN_PROFILE_BYTES:
            raise VpnError("VPN profile exceeds its bounded input size")

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

    except (OSError, ValueError, RecursionError):
        raise VpnError("Unable to read a valid bounded VPN profile") from None

    return ParseVpnProfile(data)