#!/usr/bin/env python3 """Create wg-xeddsa-v1 HTTP Message Signature headers without exposing the key.""" import argparse import base64 import getpass import hashlib import json import os import pathlib import re import stat import time import urllib.parse import warnings import xeddsa METHODS = ("GET", "POST", "PUT", "DELETE") HOSTNAME = re.compile(r"^[a-z0-9]([a-z0-9-]*[a-z0-9])?(\.[a-z0-9]([a-z0-9-]*[a-z0-9])?)*$") def read_private_key(path: str | None) -> bytes: if path: # One descriptor for the checks and the read, so both see the same file. descriptor = os.open(path, os.O_RDONLY) try: info = os.fstat(descriptor) if not stat.S_ISREG(info.st_mode): raise SystemExit("private key file must be a regular file") if stat.S_IMODE(info.st_mode) & 0o077: raise SystemExit("private key file must not be readable by group or others") raw = os.read(descriptor, 1024) finally: os.close(descriptor) try: encoded = raw.decode("ascii").strip() except UnicodeDecodeError as exc: raise SystemExit("invalid base64 WireGuard private key") from exc else: # getpass falls back to echoing the input when it can't hide it; refuse instead. try: with warnings.catch_warnings(): warnings.simplefilter("error", getpass.GetPassWarning) encoded = getpass.getpass("WireGuard private key (base64): ").strip() except getpass.GetPassWarning as exc: raise SystemExit("cannot hide the key while it is typed; use --private-key-file") from exc try: key = base64.b64decode(encoded, validate=True) except ValueError as exc: raise SystemExit("invalid base64 WireGuard private key") from exc if len(key) != 32 or base64.b64encode(key).decode() != encoded: raise SystemExit("WireGuard private key must be canonical base64 for 32 bytes") return key def read_body(path: str | None) -> bytes: if path is None: return b"" return pathlib.Path(path).read_bytes() def parse_url(url: str) -> urllib.parse.SplitResult: # Validate the text itself: urlsplit strips some characters and accepts others. if any(not 0x21 <= ord(c) <= 0x7E for c in url): raise SystemExit("--url must be printable ASCII without spaces") if "#" in url: raise SystemExit("--url must not have a fragment") parsed = urllib.parse.urlsplit(url) if parsed.scheme != "https" or not parsed.hostname: raise SystemExit("--url must be an absolute HTTPS URL") if parsed.username is not None or parsed.password is not None: raise SystemExit("credentials are not allowed in --url") try: port = parsed.port except ValueError as exc: raise SystemExit("--url has an invalid port") from exc if port not in (None, 443): raise SystemExit("--url must use the default HTTPS port") return parsed def header_value(option: str, value: str) -> str: # The server trims header values; sign what it will see, on one line of printable ASCII. value = value.strip(" \t") if not value or any(not 0x20 <= ord(c) <= 0x7E for c in value): raise SystemExit(f"{option} must be printable ASCII on one line") return value def main() -> None: parser = argparse.ArgumentParser() parser.add_argument("--method", required=True, choices=METHODS, help="sent and signed exactly as given") parser.add_argument("--url", required=True, help="send the request to exactly this URL") parser.add_argument("--authority", help="host to sign for when the URL's host differs (a proxy); default: the URL's host") parser.add_argument("--body-file") parser.add_argument("--content-type") parser.add_argument("--payment-signature") parser.add_argument("--private-key-file") args = parser.parse_args() parsed = parse_url(args.url) authority = (args.authority or parsed.hostname).lower() if not HOSTNAME.match(authority): raise SystemExit("--authority must be a hostname, without a port") content_type = header_value("--content-type", args.content_type) if args.content_type else None payment_signature = header_value("--payment-signature", args.payment_signature) if args.payment_signature else None private_key = read_private_key(args.private_key_file) public_key = xeddsa.priv_to_curve25519_pub(private_key) public_key_b64 = base64.b64encode(public_key).decode() body = read_body(args.body_file) content_digest = "sha-256=:" + base64.b64encode(hashlib.sha256(body).digest()).decode() + ":" nonce = base64.urlsafe_b64encode(os.urandom(32)).decode().rstrip("=") created = int(time.time()) expires = created + 300 headers = { "content-digest": content_digest, "x-windscribe-wireguard-key": public_key_b64, } components = ["@method", "@scheme", "@authority", "@path", "@query"] if content_type: headers["content-type"] = content_type components.append("content-type") components.extend(["content-digest", "x-windscribe-wireguard-key"]) if payment_signature: headers["payment-signature"] = payment_signature components.append("payment-signature") quoted_components = " ".join(json.dumps(component) for component in components) signature_params = ( f"({quoted_components});created={created};expires={expires};nonce={json.dumps(nonce)};" f"keyid={json.dumps('wg:' + public_key_b64)};alg=\"xeddsa-x25519-sha512\";" 'tag="windscribe-vpn-api-v1"' ) path = parsed.path or "/" query = "?" + parsed.query values = { "@method": args.method, "@scheme": "https", "@authority": authority, "@path": path, "@query": query, **headers, } lines = [f'"{component}": {values[component]}' for component in components] lines.append(f'"@signature-params": {signature_params}') signature_base = "\n".join(lines).encode("ascii") signing_scalar = xeddsa.priv_force_sign(private_key, False) signature = xeddsa.ed25519_priv_sign(signing_scalar, signature_base, os.urandom(64)) headers["signature-input"] = "vpn=" + signature_params headers["signature"] = "vpn=:" + base64.b64encode(signature).decode() + ":" output = { "profile": "wg-xeddsa-v1", "public_key": public_key_b64, "headers": {name.title(): value for name, value in headers.items()}, } print(json.dumps(output, indent=2)) if __name__ == "__main__": main()