[arbiter-authentication/3.1.1] Shared bounded carrier and streaming peer

.github/workflows/ci.yml

@@ -161,7 +161,7 @@ jobs:
         env:
           ARBITER_GO_CLIENT_SMOKE_OUTDIR: .ci/go-client-smoke
           GOCACHE: ${{ runner.temp }}/go-build
-        run: nox -s unit -- server/tests/integration/test_credential_peer.py server/tests/integration/test_provisional_peer.py server/tests/integration/test_pairing_peer.py server/tests/integration/test_enrollment_cli_peer.py server/tests/integration/test_control_tls_peer.py server/tests/integration/test_enrollment_receipt_peer.py server/tests/integration/test_activation_restart_peer.py server/tests/integration/test_installed_enrollment.py
+        run: nox -s unit -- server/tests/integration/test_credential_peer.py server/tests/integration/test_provisional_peer.py server/tests/integration/test_pairing_peer.py server/tests/integration/test_enrollment_cli_peer.py server/tests/integration/test_control_tls_peer.py server/tests/integration/test_authentication_peer.py server/tests/integration/test_enrollment_receipt_peer.py server/tests/integration/test_activation_restart_peer.py server/tests/integration/test_installed_enrollment.py
 
       - name: Verify native local-control rejection across OS users
         if: ${{ startsWith(matrix.platform_id, 'linux-') || startsWith(matrix.platform_id, 'macos-') }}

client/go-cli/internal/identity/authentication_peer_test.go

@@ -0,0 +1,107 @@
+package identity
+
+import (
+	"bytes"
+	"crypto/tls"
+	"encoding/json"
+	"fmt"
+	"io"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/gorilla/websocket"
+)
+
+// A bounded synthetic peer adapter can batch two TLS records or split records
+// across WS messages. Production still uses its existing controlSocket.
+type streamPeerSocket struct {
+	*controlSocket
+	batch, fragment bool
+	pending         []byte
+}
+
+func (c *streamPeerSocket) Write(p []byte) (int, error) {
+	if c.batch {
+		if len(c.pending)+len(p) > 65536 {
+			return 0, io.ErrShortBuffer
+		}
+		c.pending = append(c.pending, p...)
+		return len(p), nil
+	}
+	if !c.fragment {
+		return c.controlSocket.Write(p)
+	}
+	n := len(p)
+	for len(p) > 0 {
+		size := min(2049, len(p))
+		if err := c.WriteMessage(websocket.BinaryMessage, p[:size]); err != nil {
+			return n - len(p), err
+		}
+		p = p[size:]
+	}
+	return n, nil
+}
+
+func TestPythonAuthenticationStreamPeer(t *testing.T) {
+	path := os.Getenv("ARBITER_TEST_AUTH_STREAM")
+	if path == "" {
+		t.Skip("private Python stream peer only")
+	}
+	var fixture struct {
+		URL, ServerID string
+		Authority     []byte
+		Fragment      bool
+	}
+	raw, err := os.ReadFile(path)
+	if err != nil || json.Unmarshal(raw, &fixture) != nil {
+		t.Fatal("invalid synthetic peer fixture")
+	}
+	dialer := websocket.Dialer{HandshakeTimeout: 5 * time.Second,
+		TLSClientConfig: &tls.Config{InsecureSkipVerify: true}}
+	ws, _, err := dialer.Dial(fixture.URL, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer ws.Close()
+	socket := &streamPeerSocket{controlSocket: &controlSocket{Conn: ws}, fragment: fixture.Fragment}
+	if err := socket.SetDeadline(time.Now().Add(8 * time.Second)); err != nil {
+		t.Fatal(err)
+	}
+	conn := tls.Client(socket, &tls.Config{MinVersion: tls.VersionTLS13, MaxVersion: tls.VersionTLS13,
+		NextProtos: []string{controlALPN}, SessionTicketsDisabled: true, InsecureSkipVerify: true,
+		VerifyConnection: func(state tls.ConnectionState) error {
+			if state.Version != tls.VersionTLS13 || state.NegotiatedProtocol != controlALPN ||
+				state.DidResume || len(state.PeerCertificates) != 1 {
+				return errControl
+			}
+			_, err := VerifyCredential(state.PeerCertificates[0].Raw,
+				Identity{Role: ServerTLS, SubjectID: fixture.ServerID},
+				&Issuer{DER: fixture.Authority, ID: fixture.ServerID}, time.Now().UTC())
+			return err
+		}})
+	if err := conn.Handshake(); err != nil {
+		t.Fatal(err)
+	}
+	payload := bytes.Repeat([]byte("bounded-stream!!"), 2048) // 32 KiB / two chunks.
+	for round := range 16 {
+		payload[0] = byte(round)
+		socket.batch = !fixture.Fragment
+		for offset := 0; offset < len(payload); offset += 16384 {
+			if _, err := conn.Write(payload[offset : offset+16384]); err != nil {
+				t.Fatal(err)
+			}
+		}
+		if socket.batch {
+			if err := ws.WriteMessage(websocket.BinaryMessage, socket.pending); err != nil {
+				t.Fatal(err)
+			}
+			socket.pending = nil
+		}
+		got := make([]byte, len(payload))
+		if _, err := io.ReadFull(conn, got); err != nil || !bytes.Equal(got, payload) {
+			t.Fatalf("stream round %d: %v", round, err)
+		}
+	}
+	fmt.Println("ARBITER_STREAM_PEER=pass")
+}

docs/authentication-cli.md

@@ -0,0 +1,7 @@
+# Native authentication delivery status
+
+The shared bounded TLS stream is an internal transport primitive. Its private
+Python/Go interoperability peer does not enable ordinary authenticated API
+requests or add a user command. Existing enrollment and fixed control behavior
+remain available. The approved ordinary native API and enrollment selection
+are delivered by the following cuts in the authentication implementation plan.

server/news/+auth-3-1-1.feature.md

@@ -0,0 +1,4 @@
+Extract the shared bounded native TLS stream pump and carrier credit accounting,
+with private Python/Go streaming interoperability tests. Existing enrollment and
+fixed control behavior remain unchanged; ordinary authenticated API access is
+not enabled by this internal transport step.

server/src/arbiter_server/control_tls.py

@@ -2,15 +2,14 @@
 
 from __future__ import annotations
 
-from collections.abc import Callable
 from pathlib import Path
-from typing import Any
 
 from cryptography import x509
 from OpenSSL import SSL
 
 from .pairing_carrier import PairingProtocol
 from .pairing_identity import load_pairing_identity
+from .native_stream import TLSStream
 
 ALPN = b"arbiter-control-v1"
 MAX_CONTROL = 65536
@@ -42,17 +41,14 @@ def select(connection: SSL.Connection, protocols: list[bytes]) -> bytes:
     return connection
 
 
-class ControlTLS:
+class ControlTLS(TLSStream):
     """One bounded exchange over the existing carrier and crypto ownership.
 
     Every BIO/crypto call is serialized through the carrier's shared worker pool.
     A cancelled worker retains its lease until actual completion. Outgoing BIO
     data is drained before accepting more input; no background read-ahead exists.
     """
 
-    def __init__(self, carrier: PairingProtocol, connection: SSL.Connection) -> None:
-        self.carrier, self.connection = carrier, connection
-
     @classmethod
     async def accept(cls, carrier: PairingProtocol, root: Path) -> ControlTLS:
         assert carrier.lease is not None
@@ -68,53 +64,6 @@ async def accept(cls, carrier: PairingProtocol, root: Path) -> ControlTLS:
         carrier.begin_control()
         return stream
 
-    async def _call(self, operation: Callable[[], Any]) -> Any:
-        carrier, connection = self.carrier, self.connection
-        assert carrier.lease is not None
-        first = True
-        result: Any = None
-        needs_input = False
-        while True:
-
-            def step() -> tuple[Any, bool, bytes | None]:
-                operation_result: Any = None
-                operation_needs_input = False
-                if first:
-                    try:
-                        operation_result = operation()
-                    except SSL.WantReadError:
-                        operation_needs_input = True
-                try:
-                    # Leave headroom for the encrypted WebSocket frame and
-                    # the transport's own write-buffer copy.
-                    chunk = connection.bio_read(max(1, carrier.frame_limit // 2))
-                except SSL.WantReadError:
-                    chunk = None
-                carrier.update_control_tls_pending(connection.pending())
-                return operation_result, operation_needs_input, chunk
-
-            step_result, step_needs_input, chunk = await carrier.responder._run(
-                carrier.lease, step
-            )
-            if first:
-                result, needs_input = step_result, step_needs_input
-                first = False
-            if chunk is not None:
-                await carrier.send(chunk)
-            if not needs_input:
-                if chunk is None:
-                    return result
-                continue
-            if chunk is not None:
-                continue
-            data = await carrier.receive()
-            carrier.reserve_control_tls_input(len(data))
-            await carrier.responder._run(
-                carrier.lease, lambda: connection.bio_write(data)
-            )
-            del data  # BIO owns the input before the next carrier read.
-            first = True
-
     async def receive(self) -> bytes:
         async def exact(size: int) -> bytes:
             result = bytearray()
@@ -138,12 +87,9 @@ async def send(self, message: bytes) -> None:
             raise ValueError("Control exceeded limit")
         await self._call(lambda: self.connection.send(len(message).to_bytes(4, "big")))
         view = memoryview(message)
-        try:
-            while view:
-                written = await self._call(lambda: self.connection.send(view))
-                view = view[written:]
-        finally:
-            view.release()
+        while view:
+            written = await self._call(lambda: self.connection.send(view))
+            view = view[written:]
 
     def exporter(self) -> bytes:
         return self.connection.export_keying_material(