#!/usr/bin/env python3
"""Gate web connector: Python 3.10+ and OpenSSH, with no third-party packages."""
import argparse
import base64
from collections import deque
from contextlib import contextmanager
import fcntl
import hashlib
import hmac
import http.client
import ipaddress
import json
import os
from pathlib import Path
import re
import secrets
import selectors
import shlex
import shutil
import socket
import ssl
import stat
import struct
import subprocess
import sys
import tempfile
import threading
import time
import urllib.parse
import uuid

MAX_JSON = 16384
MAX_FRAME = 16384
WINDOW = 131072
MAX_SAFE_INTEGER = 9007199254740991
USER_AGENT = "Instinct-Gate/1.8"


class GateError(Exception):
    pass


def require(ok, message="Invalid Gate connection details."):
    if not ok:
        raise GateError(message)


def object_pairs(items):
    result = {}
    for key, value in items:
        require(key not in result, "Gate returned duplicate JSON fields.")
        result[key] = value
    return result


def decode(data):
    require(isinstance(data, bytes) and len(data) <= MAX_JSON)
    def invalid_constant(_value):
        raise GateError("Invalid Gate response.")
    try:
        return json.loads(data.decode("utf-8"), object_pairs_hook=object_pairs,
                          parse_constant=invalid_constant)
    except (ValueError, UnicodeError, RecursionError) as error:
        raise GateError("Invalid Gate response.") from error


def fields(value, names):
    require(isinstance(value, dict) and set(value) == set(names.split()))


def encoded(value):
    data = json.dumps(value, sort_keys=True, separators=(",", ":"), allow_nan=False).encode()
    require(len(data) + 1 <= MAX_JSON)
    return data + b"\n"


def canonical_uuid(value):
    require(isinstance(value, str))
    try:
        require(str(uuid.UUID(value)) == value)
    except ValueError as error:
        raise GateError("Invalid Gate device identity.") from error
    return value


def generation(value):
    require(type(value) is int and 1 <= value <= MAX_SAFE_INTEGER)
    return value


def token(value):
    require(isinstance(value, str) and re.fullmatch(r"[A-Za-z0-9_-]{43}", value))
    raw = base64.urlsafe_b64decode(value + "=")
    require(base64.urlsafe_b64encode(raw).decode().rstrip("=") == value and len(raw) == 32)
    return raw


def token_digest(value):
    return hashlib.sha256(token(value)).hexdigest()


def public_key(value):
    require(isinstance(value, str) and value.startswith("ssh-ed25519 "))
    try:
        raw = base64.b64decode(value[12:], validate=True)
    except ValueError as error:
        raise GateError("Invalid SSH identity.") from error
    require(raw[:19] == struct.pack(">I", 11) + b"ssh-ed25519" + struct.pack(">I", 32)
            and len(raw) == 51 and value == "ssh-ed25519 " + base64.b64encode(raw).decode())
    return value


class LocalTest:
    """Explicit localhost injection confined to private temporary test state."""
    def __init__(self, service_url, state_dir):
        url = urllib.parse.urlsplit(service_url)
        require(url.scheme == "http" and url.hostname == "127.0.0.1" and url.port
                and url.netloc == "127.0.0.1:" + str(url.port)
                and not url.path and not url.query and not url.fragment)
        self.state_dir = Path(state_dir).absolute()
        roots = (Path(tempfile.gettempdir()).resolve(), Path("/tmp").resolve())
        require(any(root in self.state_dir.resolve().parents for root in roots),
                "Local tests require isolated temporary state.")
        private_dir(self.state_dir)
        self.service_url = service_url


def origin(value, testing=None):
    require(isinstance(value, str) and len(value) <= 300)
    if testing is not None and value == testing.service_url:
        return value
    try:
        url = urllib.parse.urlsplit(value)
        require(url.scheme == "https" and url.hostname and url.port in (None, 443)
                and not url.username and not url.password and not url.path
                and not url.query and not url.fragment)
        hostname = url.hostname
        require(re.fullmatch(r"(?=.{1,253}$)(?:[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?\.)+[a-z0-9]{1,63}", hostname))
        require(url.netloc in (hostname, hostname + ":443")
                and value == "https://" + url.netloc)
        try:
            address = ipaddress.ip_address(hostname)
        except ValueError:
            address = None
        require(address is None or address.is_global)
    except ValueError as error:
        raise GateError("Gate requires its publisher's HTTPS address.") from error
    return value


def profile(value, testing=None):
    fields(value, "version transport serviceURL")
    require(type(value["version"]) is int and value["version"] == 1
            and value["transport"] == "wss")
    origin(value["serviceURL"], testing)
    return value


def enrollment(value):
    fields(value, "version inviteID deviceID pairingNonce tunnelPublicKey desktopHostKey desktopUser desktopName")
    require(type(value["version"]) is int and value["version"] == 2 and value["desktopName"] == "My Mac")
    canonical_uuid(value["inviteID"])
    canonical_uuid(value["deviceID"])
    token(value["pairingNonce"])
    require(isinstance(value["desktopUser"], str)
            and re.fullmatch(r"[a-zA-Z_][a-zA-Z0-9_.-]{0,63}", value["desktopUser"]))
    public_key(value["tunnelPublicKey"])
    public_key(value["desktopHostKey"])
    require(value["tunnelPublicKey"] != value["desktopHostKey"])
    return value


def parse_invitation(text, testing=None):
    require(isinstance(text, str) and len(text.encode()) <= 4096)
    lines = text.splitlines()
    require(len(lines) == 2 and lines[0] == "Instinct Gate would like to pair",
            "Paste both lines of Gate's invitation through standard input.")
    url = urllib.parse.urlsplit(lines[1])
    service = origin(urllib.parse.urlunsplit((url.scheme, url.netloc, "", "", "")), testing)
    require(url.path == "/pair" and not url.query and lines[1] == service + "/pair#" + url.fragment)
    parts = url.fragment.split(".")
    require(len(parts) == 3)
    device = canonical_uuid(parts[0])
    require(re.fullmatch(r"[1-9][0-9]{0,15}", parts[1]))
    number = generation(int(parts[1]))
    token(parts[2])
    return service, device, number, parts[2]


def private_dir(path):
    path.mkdir(parents=True, exist_ok=True, mode=0o700)
    info = path.lstat()
    require(stat.S_ISDIR(info.st_mode) and info.st_uid == os.getuid()
            and info.st_mode & 0o077 == 0,
            "Gate storage must be owned by you and private (mode 700).")


def private_read(path):
    descriptor = os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK)
    with os.fdopen(descriptor, "rb") as stream:
        info = os.fstat(stream.fileno())
        require(stat.S_ISREG(info.st_mode) and info.st_uid == os.getuid()
                and info.st_mode & 0o077 == 0 and info.st_nlink == 1
                and info.st_size <= MAX_JSON, "Unsafe saved Gate file.")
        return stream.read(MAX_JSON + 1)


def exists(path):
    return os.path.lexists(path)


def sync_directory(directory):
    descriptor = os.open(directory, os.O_RDONLY)
    try:
        os.fsync(descriptor)
    finally:
        os.close(descriptor)


def save(path, data):
    require(len(data) <= MAX_JSON)
    if exists(path):
        private_read(path)
    descriptor, temporary = tempfile.mkstemp(prefix=".gate-", dir=path.parent)
    try:
        with os.fdopen(descriptor, "wb") as stream:
            stream.write(data)
            stream.flush()
            os.fsync(stream.fileno())
        os.replace(temporary, path)
        sync_directory(path.parent)
    finally:
        Path(temporary).unlink(missing_ok=True)


@contextmanager
def state_lock(state, testing=None):
    state = Path(state).absolute()
    if testing is not None:
        require(state == testing.state_dir, "Local tests require their isolated state directory.")
    private_dir(state)
    descriptor = os.open(state / ".lock", os.O_RDWR | os.O_CREAT | os.O_NOFOLLOW, 0o600)
    try:
        info = os.fstat(descriptor)
        require(stat.S_ISREG(info.st_mode) and info.st_uid == os.getuid()
                and info.st_mode & 0o077 == 0 and info.st_nlink == 1)
        fcntl.flock(descriptor, fcntl.LOCK_EX | fcntl.LOCK_NB)
        yield state
    except BlockingIOError as error:
        raise GateError("Another Gate operation is using this state. Retry shortly.") from error
    finally:
        os.close(descriptor)


def bounded_command(args):
    with subprocess.Popen(args, stdin=subprocess.DEVNULL, stdout=subprocess.PIPE,
                          stderr=subprocess.PIPE) as process:
        result, errors = bytearray(), bytearray()
        try:
            with selectors.DefaultSelector() as selector:
                selector.register(process.stdout, selectors.EVENT_READ, result)
                selector.register(process.stderr, selectors.EVENT_READ, errors)
                deadline = time.monotonic() + 20
                while selector.get_map():
                    require(time.monotonic() < deadline, "OpenSSH did not complete in time.")
                    for key, _events in selector.select(0.5):
                        chunk = os.read(key.fileobj.fileno(), MAX_JSON + 1)
                        if chunk:
                            key.data.extend(chunk)
                            require(len(key.data) <= MAX_JSON, "Unexpected OpenSSH response.")
                        else:
                            selector.unregister(key.fileobj)
            require(process.wait(timeout=2) == 0, "OpenSSH could not load Gate's private identity.")
            return bytes(result)
        except BaseException:
            process.kill()
            process.wait()
            raise


def identity(directory, create=False):
    path = directory / "agent-key"
    keygen = shutil.which("ssh-keygen")
    require(keygen, "OpenSSH ssh-keygen is required.")
    if not exists(path):
        require(create, "Gate's saved private key is missing. Restore its existing state.")
        with tempfile.TemporaryDirectory(prefix=".identity-", dir=directory) as scratch:
            temporary = Path(scratch) / "key"
            bounded_command([keygen, "-q", "-t", "ed25519", "-N", "", "-C", "", "-f", str(temporary)])
            # ssh-keygen closing its output is not a durability guarantee. Flush
            # the private key's bytes before publishing its final directory entry.
            descriptor = os.open(temporary, os.O_RDONLY | os.O_NOFOLLOW)
            try:
                os.fsync(descriptor)
            finally:
                os.close(descriptor)
            os.link(temporary, path)
        sync_directory(directory)
    private_read(path)
    pub = public_key(bounded_command([keygen, "-y", "-P", "", "-f", str(path)]).decode().strip())
    return path, pub


def json_request(service, method, path, authorization=None, body=None, testing=None):
    origin(service, testing)
    url = urllib.parse.urlsplit(service)
    headers = {"User-Agent": USER_AGENT, "Accept": "application/json", "Connection": "close"}
    if authorization is not None:
        token(authorization)
        headers["Authorization"] = "Bearer " + authorization
    data = None if body is None else encoded(body)
    if data is not None:
        headers["Content-Type"] = "application/json"
    if testing is not None and service == testing.service_url:
        connection = http.client.HTTPConnection(url.hostname, url.port, timeout=15)
    else:
        connection = http.client.HTTPSConnection(url.hostname, url.port or 443,
                                                context=ssl.create_default_context(), timeout=15)
    try:
        connection.request(method, path, body=data, headers=headers)
        response = connection.getresponse()
        require(not 300 <= response.status < 400, "Gate redirected. Keep the original publisher address.")
        require(response.status == 200,
                "Gate could not approve this invitation. Retry the same invitation or check it on the Mac.")
        require(response.getheader("Content-Type", "").split(";")[0].strip() == "application/json")
        return decode(response.read(MAX_JSON + 1))
    finally:
        connection.close()


def location(state, service, device):
    directory = state / hashlib.sha256(service.encode()).hexdigest() / canonical_uuid(device)
    private_dir(directory.parent)
    private_dir(directory)
    return directory


def credentials(value, testing=None):
    fields(value, "version service deviceID generation claimToken agentToken agentPublicKey")
    require(type(value["version"]) is int and value["version"] == 1)
    profile(value["service"], testing)
    canonical_uuid(value["deviceID"])
    generation(value["generation"])
    token(value["claimToken"])
    token(value["agentToken"])
    require(value["agentToken"] != value["claimToken"])
    public_key(value["agentPublicKey"])
    return value


def check_claim(value, saved):
    fields(value, "version state request generation agentPublicKey agentAuthDigest")
    require(type(value["version"]) is int and value["version"] == 1
            and value["state"] in ("claimed", "paired"))
    enrollment(value["request"])
    generation(value["generation"])
    require(value["request"]["deviceID"] == saved["deviceID"]
            and value["generation"] == saved["generation"]
            and value["agentPublicKey"] == saved["agentPublicKey"]
            and value["agentAuthDigest"] == token_digest(saved["agentToken"]),
            "Gate returned a different pairing identity. Existing access was preserved.")
    require(len({saved["agentPublicKey"], value["request"]["tunnelPublicKey"],
                 value["request"]["desktopHostKey"]}) == 3)
    return value


def load_binding(directory, testing=None):
    binding = decode(private_read(directory / "binding.json"))
    fields(binding, "version credentials claim")
    require(type(binding["version"]) is int and binding["version"] == 1)
    saved = credentials(binding["credentials"], testing)
    claim = check_claim(binding["claim"], saved)
    return saved, claim


def current_binding(state, testing=None):
    current = decode(private_read(state / "current.json"))
    fields(current, "version serviceURL deviceID")
    require(type(current["version"]) is int and current["version"] == 1)
    service = origin(current["serviceURL"], testing)
    directory = location(state, service, current["deviceID"])
    saved, claim = load_binding(directory, testing)
    require(saved["service"]["serviceURL"] == service and saved["deviceID"] == current["deviceID"])
    key, pub = identity(directory)
    require(pub == saved["agentPublicKey"], "Gate's saved private identity changed.")
    return directory, saved, claim, key


def pair(state, invitation, testing=None):
    service, device, number, claim_token = parse_invitation(invitation, testing)
    expected_profile = {"version": 1, "transport": "wss", "serviceURL": service}
    with state_lock(state, testing) as state:
        if exists(state / "current.json"):
            _old_dir, old_saved, _old_claim, _old_key = current_binding(state, testing)
            require(old_saved["service"] == expected_profile, "Gate's publisher changed. The saved connection was not replaced.")
            if old_saved["deviceID"] != device:
                # Check the saved origin with the saved credential, never a URL
                # or proof supplied by the new invitation. Keep old state until
                # the fresh claim commits; an interrupted retry is safe.
                previous = json_request(service, "POST", f"/v1/device/{old_saved['deviceID']}/agent-status",
                                        old_saved["agentToken"], {"version": 1, "generation": old_saved["generation"]}, testing)
                fields(previous, "version state generation")
                require(type(previous["version"]) is int and previous["version"] == 1
                        and type(previous["generation"]) is int and previous["generation"] == old_saved["generation"]
                        and previous["state"] == "revoked",
                        "Revoke the previous pairing on the Mac before pairing again.")
        directory = location(state, service, device)
        path = directory / "credentials.json"
        prior_claim = None
        prior_saved = None
        if exists(directory / "binding.json"):
            prior_saved, prior_claim = load_binding(directory, testing)
            require(prior_saved["service"] == expected_profile and prior_saved["deviceID"] == device)
        if exists(path):
            saved = credentials(decode(private_read(path)), testing)
            require(saved["service"] == expected_profile and saved["deviceID"] == device)
            _key, pub = identity(directory)
            require(pub == saved["agentPublicKey"], "Gate's saved private identity changed.")
            if prior_saved is not None:
                require(saved["agentPublicKey"] == prior_saved["agentPublicKey"]
                        and hmac.compare_digest(saved["agentToken"], prior_saved["agentToken"])
                        and saved["generation"] >= prior_saved["generation"],
                        "Gate's saved private identity changed.")
                require(prior_claim["state"] != "paired" or number == prior_saved["generation"],
                        "This Mac is already verified. Reconnect using its saved connection.")
            require(number >= saved["generation"], "This invitation is older than the saved request.")
            if number == saved["generation"]:
                require(hmac.compare_digest(claim_token, saved["claimToken"]),
                        "This invitation does not match the saved request.")
            else:
                saved = dict(saved, generation=number, claimToken=claim_token)
        else:
            require(not exists(directory / "binding.json"), "Gate's saved credentials are missing.")
            _key, pub = identity(directory, create=True)
            agent_token = secrets.token_urlsafe(32)
            while agent_token == claim_token:
                agent_token = secrets.token_urlsafe(32)
            saved = {"version": 1, "service": expected_profile, "deviceID": device,
                     "generation": number, "claimToken": claim_token,
                     "agentToken": agent_token, "agentPublicKey": pub}
        # Both private identity and retry credentials exist durably before ANY HTTP.
        save(path, encoded(saved))
        advertised = profile(json_request(service, "GET", "/profile.json", testing=testing), testing)
        require(advertised == expected_profile, "Gate returned a different publisher profile.")
        body = {"version": 1, "generation": number, "agentPublicKey": saved["agentPublicKey"],
                "agentAuthDigest": token_digest(saved["agentToken"])}
        reply = check_claim(json_request(service, "POST", f"/v1/device/{device}/claim",
                                         claim_token, body, testing), saved)
        if prior_claim is not None:
            require(prior_claim["request"] == reply["request"]
                    and prior_claim["agentPublicKey"] == reply["agentPublicKey"]
                    and prior_claim["agentAuthDigest"] == reply["agentAuthDigest"],
                    "Gate's saved Mac identity changed. Existing access was preserved.")
            require(prior_claim["state"] != "paired" or reply["state"] == "paired",
                    "Gate's verified binding cannot return to an unverified state.")
        # One commit retains a usable old binding if a newer request fails or the
        # process stops between saving pending credentials and receiving approval.
        save(directory / "binding.json", encoded({"version": 1, "credentials": saved, "claim": reply}))
        save(state / "current.json", encoded({"version": 1, "serviceURL": service, "deviceID": device}))
        return directory / "binding.json"


def option_path(path):
    require(not any(character in str(path) for character in "\r\n\x00%"), "Unsupported Gate storage path.")
    return '"' + str(path).replace("\\", "\\\\").replace('"', '\\"') + '"'


def connect_args(state, command, testing=None):
    with state_lock(state, testing) as state:
        directory, saved, claim, key = current_binding(state, testing)
        alias = "instinct-gate-" + saved["deviceID"]
        pins = directory / "mac-known-hosts"
        save(pins, (alias + " " + claim["request"]["desktopHostKey"] + "\n").encode())
        script = Path(__file__).resolve()
        option_path(script)
        option_path(state)
        proxy = [sys.executable, str(script), "--state-dir", str(state)]
        if testing is not None:
            proxy.append("--allow-local-test")
        proxy.append("pipe")
        ssh = shutil.which("ssh")
        require(ssh, "OpenSSH is required.")
        return [ssh, "-F", "/dev/null", "-T", "-o", "BatchMode=yes", "-o", "IdentitiesOnly=yes",
                "-o", "IdentityAgent=none", "-o", "StrictHostKeyChecking=yes",
                "-o", "UserKnownHostsFile=" + option_path(pins), "-o", "GlobalKnownHostsFile=/dev/null",
                "-o", "PasswordAuthentication=no", "-o", "KbdInteractiveAuthentication=no",
                "-o", "HostKeyAlgorithms=ssh-ed25519", "-o", "CheckHostIP=no",
                "-o", "HostKeyAlias=" + alias, "-o", "ClearAllForwardings=yes",
                "-o", "ServerAliveInterval=15", "-o", "ServerAliveCountMax=3",
                "-o", "ConnectTimeout=25", "-o", "LogLevel=ERROR", "-i", str(key),
                "-o", "ProxyCommand=" + shlex.join(proxy), "-p", "22",
                claim["request"]["desktopUser"] + "@" + alias, shlex.join(command)]


def client_frame(opcode, payload):
    require(opcode in (1, 2, 8, 9, 10) and isinstance(payload, bytes))
    require(len(payload) <= (125 if opcode >= 8 else MAX_FRAME))
    mask = os.urandom(4)
    length = len(payload)
    header = bytes([0x80 | opcode, 0x80 | length]) if length < 126 else bytes([0x80 | opcode, 0xFE]) + struct.pack(">H", length)
    return header + mask + bytes(value ^ mask[index % 4] for index, value in enumerate(payload))


class FrameDecoder:
    """Strict RFC6455 server frames, with one bounded fragmented message."""
    def __init__(self):
        self.buffer = bytearray()
        self.fragment_opcode = None
        self.fragment = bytearray()

    def feed(self, data):
        require(len(data) <= MAX_FRAME + 14)
        self.buffer.extend(data)
        events = []
        while len(self.buffer) >= 2:
            first, second = self.buffer[0], self.buffer[1]
            opcode, final = first & 15, bool(first & 0x80)
            require(first & 0x70 == 0 and second & 0x80 == 0,
                    "Gate sent an invalid WebSocket frame.")
            require(opcode in (0, 1, 2, 8, 9, 10))
            length, offset = second & 127, 2
            if length == 126:
                if len(self.buffer) < 4:
                    break
                length, offset = struct.unpack(">H", self.buffer[2:4])[0], 4
                require(length >= 126)
            elif length == 127:
                # All 64-bit lengths exceed this protocol's 16KiB maximum.
                raise GateError("Gate sent an oversized WebSocket frame.")
            require(length <= MAX_FRAME)
            require(opcode < 8 or (final and length <= 125))
            if len(self.buffer) < offset + length:
                break
            payload = bytes(self.buffer[offset:offset + length])
            del self.buffer[:offset + length]
            if opcode >= 8:
                if opcode == 8:
                    require(len(payload) != 1)
                    if payload:
                        code = struct.unpack(">H", payload[:2])[0]
                        require(code in (1000, 1001, 1002, 1003, 1007, 1008, 1009, 1010, 1011, 1012, 1013, 1014)
                                or 3000 <= code <= 4999)
                        payload[2:].decode("utf-8")
                events.append((opcode, payload))
                continue
            if opcode == 0:
                require(self.fragment_opcode is not None)
            else:
                require(self.fragment_opcode is None)
                self.fragment_opcode = opcode
            self.fragment.extend(payload)
            require(len(self.fragment) <= MAX_FRAME)
            if final:
                complete = bytes(self.fragment)
                if self.fragment_opcode == 1:
                    complete.decode("utf-8")
                else:
                    require(complete, "Gate sent an empty data frame.")
                events.append((self.fragment_opcode, complete))
                self.fragment.clear()
                self.fragment_opcode = None
        require(len(self.buffer) <= MAX_FRAME + 4)
        return events


def websocket(saved, testing=None):
    service = origin(saved["service"]["serviceURL"], testing)
    url = urllib.parse.urlsplit(service)
    raw = socket.create_connection((url.hostname, url.port or 443), timeout=15)
    stream = raw
    try:
        if not (testing is not None and service == testing.service_url):
            stream = ssl.create_default_context().wrap_socket(raw, server_hostname=url.hostname)
        nonce = base64.b64encode(os.urandom(16)).decode()
        path = f"/v1/device/{saved['deviceID']}/connect"
        headers = [f"GET {path} HTTP/1.1", "Host: " + url.netloc, "Upgrade: websocket",
                   "Connection: Upgrade", "Sec-WebSocket-Version: 13", "Sec-WebSocket-Key: " + nonce,
                   "User-Agent: " + USER_AGENT, "Authorization: Bearer " + saved["agentToken"],
                   "X-Gate-Generation: " + str(saved["generation"]), "", ""]
        stream.sendall("\r\n".join(headers).encode("ascii"))
        received = bytearray()
        while b"\r\n\r\n" not in received:
            chunk = stream.recv(4096)
            require(chunk, "Gate closed the connection before pairing was ready.")
            received.extend(chunk)
            require(len(received) <= MAX_JSON + 4096, "Gate returned oversized upgrade headers.")
        header, leftover = bytes(received).split(b"\r\n\r\n", 1)
        require(len(header) <= MAX_JSON)
        lines = header.decode("ascii").split("\r\n")
        status = lines[0].split(" ", 2)
        require(len(status) >= 2 and status[0] == "HTTP/1.1" and status[1] == "101",
                "Gate is unavailable. Check that pairing is complete and Mac access is On.")
        parsed = {}
        for line in lines[1:]:
            require(":" in line and not line.startswith((" ", "\t")))
            name, value = line.split(":", 1)
            require(re.fullmatch(r"[!#$%&'*+.^_`|~0-9A-Za-z-]+", name))
            name, value = name.lower(), value.strip()
            require(name not in parsed and not any(ord(c) < 32 and c != "\t" for c in value))
            parsed[name] = value
        expected = base64.b64encode(hashlib.sha1((nonce + "258EAFA5-E914-47DA-95CA-C5AB0DC85B11").encode()).digest()).decode()
        require(parsed.get("upgrade", "").lower() == "websocket"
                and "upgrade" in [part.strip().lower() for part in parsed.get("connection", "").split(",")]
                and hmac.compare_digest(parsed.get("sec-websocket-accept", ""), expected)
                and "sec-websocket-extensions" not in parsed and "sec-websocket-protocol" not in parsed,
                "Gate returned an invalid WebSocket upgrade.")
        return stream, leftover
    except BaseException:
        stream.close()
        if stream is not raw:
            raw.close()
        raise


class Credit:
    """Credits track queued, transmitted and consumed bytes separately."""
    def __init__(self):
        self.available = WINDOW
        self.queued = 0
        self.outstanding = 0

    def reserve(self, count):
        require(type(count) is int and 1 <= count <= MAX_FRAME and count <= self.available)
        self.available -= count
        self.queued += count

    def transmitted(self, count):
        require(1 <= count <= self.queued)
        self.queued -= count
        self.outstanding += count

    def restore(self, count):
        require(type(count) is int and 1 <= count <= self.outstanding)
        self.outstanding -= count
        self.available += count
        require(self.available + self.queued + self.outstanding == WINDOW)


def bridge(stream, leftover=b"", source_fd=0, sink_fd=1, trace=None):
    """One TLS IO loop; two bounded local pipe workers never touch TLS."""
    condition = threading.Condition()
    wake_read, wake_write = socket.socketpair()
    wake_read.setblocking(False)
    wake_write.setblocking(False)
    outgoing, incoming = deque(), deque()
    credit = Credit()
    shared = {"ready": False, "stop": False, "error": None, "eof": False,
              "received": 0, "writing": False, "sentTotal": 0, "receivedTotal": 0,
              "writtenTotal": 0, "peerClose": None}

    def diagnostic(event):
        if trace is not None:
            with condition:
                # Fixed field names and integers only; never dump bytes, tokens,
                # credentials, exception text or a peer-provided close reason.
                trace(event, {"sent": shared["sentTotal"], "received": shared["receivedTotal"],
                              "written": shared["writtenTotal"], "available": credit.available,
                              "queued": credit.queued, "outstanding": credit.outstanding,
                              "uncredited": shared["received"], "peerClose": shared["peerClose"]})

    def wake():
        try:
            wake_write.send(b"x")
        except (BlockingIOError, OSError):
            pass

    def fail():
        with condition:
            shared["error"] = GateError("Gate's desktop connection closed unexpectedly.")
            shared["stop"] = True
            condition.notify_all()
        wake()

    def enqueue(opcode, payload, kind=None, count=0):
        with condition:
            require(len(outgoing) < 2 * WINDOW + 16, "Gate exceeded its bounded transfer queue.")
            outgoing.append((client_frame(opcode, payload), kind, count))
        wake()

    def read_source():
        try:
            while True:
                with condition:
                    condition.wait_for(lambda: shared["stop"] or (shared["ready"] and credit.available))
                    if shared["stop"]:
                        return
                    maximum = min(MAX_FRAME, credit.available)
                data = os.read(source_fd, maximum)
                with condition:
                    if shared["stop"]:
                        return
                    if not data:
                        shared["eof"] = True
                        diagnostic("source_eof")
                        enqueue(8, struct.pack(">H", 1000), "close")
                        return
                    credit.reserve(len(data))
                    enqueue(2, data, "data", len(data))
        except (OSError, GateError):
            fail()

    def write_sink():
        try:
            while True:
                with condition:
                    condition.wait_for(lambda: shared["stop"] or incoming)
                    if not incoming:
                        return
                    data = incoming.popleft()
                    shared["writing"] = True
                view = memoryview(data)
                while view:
                    count = os.write(sink_fd, view)
                    require(count > 0)
                    view = view[count:]
                with condition:
                    shared["writing"] = False
                    shared["writtenTotal"] += len(data)
                    if not shared["stop"]:
                        enqueue(1, encoded({"type": "credit", "bytes": len(data)}), "credit", len(data))
                    condition.notify_all()
                wake()
        except (OSError, GateError):
            fail()

    reader = threading.Thread(target=read_source, daemon=True)
    writer = threading.Thread(target=write_sink, daemon=True)
    reader.start()
    writer.start()
    decoder = FrameDecoder()
    pending = None
    offset = 0
    receive_wants_write = False
    send_wants_read = False
    closing_at = None
    started = time.monotonic()
    stream.setblocking(False)

    def receive(data):
        nonlocal closing_at
        for opcode, payload in decoder.feed(data):
            if opcode == 9:
                enqueue(10, payload)
            elif opcode == 10:
                continue
            elif opcode == 8:
                with condition:
                    shared["peerClose"] = struct.unpack(">H", payload[:2])[0] if payload else 1005
                    shared["stop"] = True
                    condition.notify_all()
                diagnostic("peer_close")
                enqueue(8, payload, "close")
                closing_at = closing_at or time.monotonic()
            elif opcode == 1:
                message = decode(payload)
                with condition:
                    if not shared["ready"]:
                        fields(message, "type window maxFrame")
                        require(message["type"] == "ready" and type(message["window"]) is int
                                and message["window"] == WINDOW and type(message["maxFrame"]) is int
                                and message["maxFrame"] == MAX_FRAME)
                        shared["ready"] = True
                    else:
                        fields(message, "type bytes")
                        require(message["type"] == "credit")
                        credit.restore(message["bytes"])
                    condition.notify_all()
            else:
                with condition:
                    require(shared["ready"] and not shared["stop"])
                    require(shared["received"] + len(payload) <= WINDOW,
                            "Gate exceeded its receive credit.")
                    shared["received"] += len(payload)
                    shared["receivedTotal"] += len(payload)
                    incoming.append(payload)
                    condition.notify_all()

    try:
        if leftover:
            receive(leftover)
        with selectors.DefaultSelector() as selector:
            selector.register(wake_read, selectors.EVENT_READ)
            selector.register(stream, selectors.EVENT_READ)
            while True:
                with condition:
                    if shared["error"]:
                        raise shared["error"]
                    if pending is None and outgoing:
                        pending, offset = outgoing.popleft(), 0
                    drained = not incoming and not shared["writing"]
                    ready = shared["ready"]
                if closing_at is not None and ((pending is None and drained) or time.monotonic() - closing_at > 2):
                    break
                require(ready or time.monotonic() - started < 25, "Gate timed out waiting for the Mac.")
                events = selectors.EVENT_READ
                if (pending is not None and not send_wants_read) or receive_wants_write:
                    events |= selectors.EVENT_WRITE
                selector.modify(stream, events)
                buffered = isinstance(stream, ssl.SSLSocket) and stream.pending() > 0
                selected = selector.select(0 if buffered else 0.5)
                readable = buffered or any(key.fileobj is stream and event & selectors.EVENT_READ for key, event in selected)
                writable = any(key.fileobj is stream and event & selectors.EVENT_WRITE for key, event in selected)
                for key, _event in selected:
                    if key.fileobj is wake_read:
                        try:
                            wake_read.recv(4096)
                        except BlockingIOError:
                            pass
                if pending is not None and (writable or (readable and send_wants_read)):
                    try:
                        sent = stream.send(pending[0][offset:])
                        require(sent > 0)
                        send_wants_read = False
                        offset += sent
                        if offset == len(pending[0]):
                            with condition:
                                if pending[1] == "data":
                                    credit.transmitted(pending[2])
                                    shared["sentTotal"] += pending[2]
                                elif pending[1] == "credit":
                                    shared["received"] -= pending[2]
                                elif pending[1] == "close":
                                    closing_at = closing_at or time.monotonic()
                            pending = None
                    except ssl.SSLWantReadError:
                        send_wants_read = True
                    except (ssl.SSLWantWriteError, BlockingIOError):
                        send_wants_read = False
                if readable or (writable and receive_wants_write):
                    try:
                        data = stream.recv(4096)
                        receive_wants_write = False
                        if not data:
                            diagnostic("transport_eof")
                            with condition:
                                shared["stop"] = True
                                condition.notify_all()
                            closing_at = closing_at or time.monotonic()
                            if drained:
                                break
                        else:
                            receive(data)
                    except ssl.SSLWantWriteError:
                        receive_wants_write = True
                    except (ssl.SSLWantReadError, BlockingIOError):
                        receive_wants_write = False
    except BaseException:
        diagnostic("bridge_error")
        raise
    finally:
        diagnostic("bridge_done")
        with condition:
            shared["stop"] = True
            condition.notify_all()
        stream.close()
        wake_read.close()
        wake_write.close()


def pipe(state, source_fd=0, sink_fd=1, testing=None):
    with state_lock(state, testing) as state:
        _directory, saved, _claim, _key = current_binding(state, testing)
    stream, leftover = websocket(saved, testing)
    def trace(event, counters):
        print("Gate local test: " + event + " " + json.dumps(counters, sort_keys=True), file=sys.stderr)
    bridge(stream, leftover, source_fd, sink_fd, trace if testing is not None else None)


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--state-dir", type=Path)
    parser.add_argument("--allow-local-test", action="store_true", help=argparse.SUPPRESS)
    commands = parser.add_subparsers(dest="action", required=True)
    commands.add_parser("pair", help="Read the owner's two-line invitation from standard input.")
    commands.add_parser("pipe", help=argparse.SUPPRESS)
    run = commands.add_parser("run", help="Run a command on the paired Mac while access is On.")
    run.add_argument("command", nargs=argparse.REMAINDER)
    args = parser.parse_args()
    try:
        invitation = sys.stdin.read(4097) if args.action == "pair" else None
        testing = None
        if args.allow_local_test:
            require(args.state_dir is not None, "Local tests require an explicit state directory.")
            if invitation is not None:
                lines = invitation.splitlines()
                require(len(lines) == 2)
                url = urllib.parse.urlsplit(lines[1])
                service = urllib.parse.urlunsplit((url.scheme, url.netloc, "", "", ""))
            else:
                service = decode(private_read(args.state_dir / "current.json"))["serviceURL"]
            testing = LocalTest(service, args.state_dir)
        state = args.state_dir or Path.home() / ".local/share/instinct-gate-web"
        if args.action == "pair":
            pair(state, invitation, testing)
            print("Invitation approved. Gate will finish pairing on the Mac. Access stays Off until its owner turns it On.")
            print("After access is On, test: python3 client.py run -- pwd")
        elif args.action == "pipe":
            pipe(state, testing=testing)
        else:
            command = args.command[1:] if args.command[:1] == ["--"] else args.command
            require(command, "Provide a command, for example: run -- pwd")
            result = subprocess.run(connect_args(state, command, testing), check=False)
            if result.returncode == 255:
                print("Gate is unavailable. Check that Mac access is On and the Mac is awake and online.", file=sys.stderr)
            return result.returncode
    except (GateError, OSError, ValueError, KeyError, TypeError, UnicodeError,
            http.client.HTTPException, subprocess.SubprocessError):
        # Never print a network exception: it can contain a private request/header.
        print("Gate could not complete this operation. Keep its saved state and retry the same invitation or connection.", file=sys.stderr)
        return 1
    return 0


if __name__ == "__main__":
    sys.exit(main())