flayer.profiles.amneziawg

Module source

flayer.profiles.amneziawg

AmneziaWG 3.1 preparation with private keys and a pinned Ubuntu guest installer.

AMNEZIAWG_CAPABILITIES module-attribute

AMNEZIAWG_CAPABILITIES = VpnCapabilities('amneziawg', 'ip-tunnel', ('full', 'split'))

AMNEZIAWG_GO_VERSION module-attribute

AMNEZIAWG_GO_VERSION = 'v3.1.20260828'

AMNEZIAWG_GO_REVISION module-attribute

AMNEZIAWG_GO_REVISION = 'b5928efb6ca19f0153958460c3d141f04abc5c2e'

AMNEZIAWG_TOOLS_REVISION module-attribute

AMNEZIAWG_TOOLS_REVISION = 'ee0f0a9aa34ff0a0da4b3433b9512781cfe02843'

GO_VERSION module-attribute

GO_VERSION = '1.27.2'

GO_LINUX_AMD64_SHA256 module-attribute

GO_LINUX_AMD64_SHA256 = 'ecbadb99091a3f46e31f5f934b068b1864eafa7995211b39eaddf76996045fe5'

AmneziaWgDeviceKey dataclass

AmneziaWgDeviceKey(device_id: str, private_key: str)

One device credential, excluded from diagnostic object representations.

device_id instance-attribute

device_id: str

private_key class-attribute instance-attribute

private_key: str = field(repr=False)

AmneziaWgSecrets dataclass

AmneziaWgSecrets(server_private_key: str, header_protection_key: str, device_private_keys: tuple[AmneziaWgDeviceKey, ...])

Private preparation input kept out of generic configuration, state, and repr.

server_private_key class-attribute instance-attribute

server_private_key: str = field(repr=False)

header_protection_key class-attribute instance-attribute

header_protection_key: str = field(repr=False)

device_private_keys class-attribute instance-attribute

device_private_keys: tuple[AmneziaWgDeviceKey, ...] = field(repr=False)

AmneziaWgSettings dataclass

AmneziaWgSettings(tunnel_cidr: str = '10.66.0.0/24', mtu: int = 1280)

Bounded IPv4 tunnel subnet and conservative client packet size.

tunnel_cidr class-attribute instance-attribute

tunnel_cidr: str = '10.66.0.0/24'

mtu class-attribute instance-attribute

mtu: int = 1280

GenerateAmneziaWgSecrets

GenerateAmneziaWgSecrets(profile: VpnProfile, previous: AmneziaWgSecrets | None = None, *, rotate_device_ids: tuple[str, ...] = ()) -> AmneziaWgSecrets

Keep unchanged credentials while explicitly adding, rotating, or removing devices.

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

def GenerateAmneziaWgSecrets(
    profile: VpnProfile, previous: AmneziaWgSecrets | None = None,
    *, rotate_device_ids: tuple[str, ...] = (),
) -> AmneziaWgSecrets:
    """Keep unchanged credentials while explicitly adding, rotating, or removing devices."""

    ValidateVpnTransport(profile, AMNEZIAWG_CAPABILITIES)
    device_ids = {item.device_id for item in profile.devices}

    if (
        not isinstance(rotate_device_ids, tuple)
        or any(not isinstance(item, str) or item not in device_ids for item in rotate_device_ids)
        or len(set(rotate_device_ids)) != len(rotate_device_ids)
        or (previous is not None and not isinstance(previous, AmneziaWgSecrets))
    ):
        raise VpnError("AmneziaWG rotation requires unique currently declared device IDs")

    old = {} if previous is None else {
        item.device_id: item.private_key for item in previous.device_private_keys
    }
    device_keys = tuple(AmneziaWgDeviceKey(
        item.device_id,
        old[item.device_id] if item.device_id in old and item.device_id not in rotate_device_ids
        else _PrivateKey(),
    ) for item in profile.devices)

    return AmneziaWgSecrets(
        _PrivateKey() if previous is None else previous.server_private_key,
        base64.b64encode(secrets.token_bytes(32)).decode("ascii")
        if previous is None else previous.header_protection_key,
        device_keys,
    )

EncodeAmneziaWgSecrets

EncodeAmneziaWgSecrets(secret_values: AmneziaWgSecrets) -> bytes

Serialize controller-only credentials for an owned private material bundle.

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

def EncodeAmneziaWgSecrets(secret_values: AmneziaWgSecrets) -> bytes:
    """Serialize controller-only credentials for an owned private material bundle."""

    if not isinstance(secret_values, AmneziaWgSecrets):
        raise VpnError("AmneziaWG material must be validated private credentials")

    data = {"schema_version": 1, **asdict(secret_values)}

    return (json.dumps(data, sort_keys=True, separators=(",", ":")) + "\n").encode("utf-8")

DecodeAmneziaWgSecrets

DecodeAmneziaWgSecrets(content: bytes) -> AmneziaWgSecrets

Validate bounded private material without returning rejected bytes in errors.

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

def DecodeAmneziaWgSecrets(content: bytes) -> AmneziaWgSecrets:
    """Validate bounded private material without returning rejected bytes in errors."""

    try:
        if not isinstance(content, bytes) or not 1 <= len(content) <= 16384:
            raise ValueError

        data = json.loads(content.decode("utf-8"), object_pairs_hook=_UniqueObject)

        if (
            not isinstance(data, dict)
            or set(data) != {"schema_version", "server_private_key", "header_protection_key", "device_private_keys"}
            or type(data["schema_version"]) is not int or data["schema_version"] != 1
            or not isinstance(data["device_private_keys"], list)
        ):
            raise ValueError

        keys = []

        for item in data["device_private_keys"]:
            if not isinstance(item, dict) or set(item) != {"device_id", "private_key"}:
                raise ValueError

            keys.append(AmneziaWgDeviceKey(item["device_id"], item["private_key"]))

        values = AmneziaWgSecrets(data["server_private_key"], data["header_protection_key"], tuple(keys))

    except (ValueError, TypeError, KeyError):
        raise VpnError("AmneziaWG private material is invalid or unsupported") from None

    return values

AmneziaWgServiceName

AmneziaWgServiceName(profile: VpnProfile) -> str

Derive a stable guest service name from complete stack ownership, not credentials.

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

def AmneziaWgServiceName(profile: VpnProfile) -> str:
    """Derive a stable guest service name from complete stack ownership, not credentials."""

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

    return "flayer-awg-" + hashlib.sha256(payload.encode("utf-8")).hexdigest()[:12]

PrepareAmneziaWg

PrepareAmneziaWg(profile: VpnProfile, endpoint: VpnEndpoint, settings: AmneziaWgSettings, secret_values: AmneziaWgSecrets) -> PreparedVpnTransport

Produce deterministic private server/client artifacts without any external operation.

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

def PrepareAmneziaWg(
    profile: VpnProfile, endpoint: VpnEndpoint, settings: AmneziaWgSettings,
    secret_values: AmneziaWgSecrets,
) -> PreparedVpnTransport:
    """Produce deterministic private server/client artifacts without any external operation."""

    listener = ValidateVpnTransport(profile, AMNEZIAWG_CAPABILITIES)

    if (
        not isinstance(endpoint, VpnEndpoint) or endpoint.port != listener.port
        or listener.protocol != "udp" or not isinstance(settings, AmneziaWgSettings)
        or not isinstance(secret_values, AmneziaWgSecrets)
    ):
        raise VpnError("AmneziaWG requires matching UDP endpoint, settings, and private keys")

    device_keys = {item.device_id: item.private_key for item in secret_values.device_private_keys}

    if set(device_keys) != {item.device_id for item in profile.devices}:
        raise VpnError("AmneziaWG secrets must match exactly the declared devices")

    network = ipaddress.IPv4Network(settings.tunnel_cidr)

    for device in profile.devices:
        address = ipaddress.IPv4Address(str(device.ipv4_address))

        if address not in network or address in {
            network.network_address, network.network_address + 1, network.broadcast_address,
        }:
            raise VpnError("AmneziaWG device addresses must be usable nonserver tunnel hosts")

    public_keys = {key: _PublicKey(value) for key, value in device_keys.items()}
    server_public_key = _PublicKey(secret_values.server_private_key)

    if len(set(public_keys.values()) | {server_public_key}) != len(public_keys) + 1:
        raise VpnError("AmneziaWG peers must have distinct Curve25519 identities")

    parameters = _Parameters(secret_values)
    server_lines = ["[Interface]", f"PrivateKey = {secret_values.server_private_key}",
                    f"ListenPort = {listener.port}", *parameters]

    for device in profile.devices:
        server_lines.extend(["", "[Peer]", f"PublicKey = {public_keys[device.device_id]}",
                             f"AllowedIPs = {device.ipv4_address}/32"])

    service = AmneziaWgServiceName(profile)
    interface = "fa" + service.removeprefix("flayer-awg-")
    base = "/etc/flayer/" + service
    server_bundle = BuildVpnServerBundle(profile, "amneziawg", (
        Artifact("server.conf", ("\n".join(server_lines) + "\n").encode()),
        Artifact("install.sh", _Installer(service, base, interface).encode()),
        Artifact("firewall.nft", _Firewall(profile, interface, interface).encode()),
        Artifact("network.sh", _NetworkScript(base, interface, interface, settings).encode()),
        Artifact("service.service", _Service(service, base, interface).encode()),
    ))
    device_bundles = []

    for device in profile.devices:
        client_lines = [
            "[Interface]", f"PrivateKey = {device_keys[device.device_id]}",
            f"Address = {device.ipv4_address}/32", f"MTU = {settings.mtu}",
            "DNS = " + ", ".join(profile.routes.dns_servers), *parameters, "", "[Peer]",
            f"PublicKey = {server_public_key}", f"Endpoint = {endpoint.host}:{endpoint.port}",
            "AllowedIPs = " + ", ".join(profile.routes.ipv4_cidrs), "PersistentKeepalive = 25", "",
        ]
        device_bundles.append(BuildVpnDeviceBundle(profile, "amneziawg", device.device_id, (
            Artifact("amneziawg.conf", "\n".join(client_lines).encode()),
        )))

    return PreparedVpnTransport(profile, AMNEZIAWG_CAPABILITIES, server_bundle, tuple(device_bundles))