[MG_Remote, MG_IntegrationTest, CI, scripts] (P7): add Ph fuzz arm 1 for malformed control frames

.github/workflows/test.yml

@@ -1482,6 +1482,8 @@ jobs:
       # the supervisor's Busy / token / listen / version / build-policy / reap controls zero times.
       # --check is passed so this step cannot quietly regenerate a different committed header;
       # keeping that honest is flatc-check's job and this step must not overlap with it.
+      # P7 F2: `TcpLane.ControlFrameFuzz` (Ph fuzz arm 1, scripts/ci/ph_fuzz_control_frames.py)
+      # needs the same flatc and is required PASSED by name on the same junit_tally line.
       - name: Check out FlatBuffers (TCP lane)
         run: git submodule update --init 3rdparty/flatbuffers
 
@@ -1516,7 +1518,7 @@ jobs:
           ctest --show-only=json-v1 -L '^integration-tcp$' > "${RUNNER_TEMP}/tcp-discovery.json"
           ctest --output-on-failure -L '^integration-tcp$' --no-tests=error -j 4 --output-junit "${RUNNER_TEMP}/tcp-baseline.xml"
           python3 ../scripts/ci/junit_tally.py "${RUNNER_TEMP}/tcp-baseline.xml" --require-tcp-ran --require-run-ahead --discovery "${RUNNER_TEMP}/tcp-discovery.json"
-          python3 ../scripts/ci/junit_tally.py "${RUNNER_TEMP}/tcp-baseline.xml" --require-entry-passed TcpLane.SupervisorProtocolControls --require-entry-passed TcpLane.EventForfeitPeer
+          python3 ../scripts/ci/junit_tally.py "${RUNNER_TEMP}/tcp-baseline.xml" --require-entry-passed TcpLane.SupervisorProtocolControls --require-entry-passed TcpLane.EventForfeitPeer --require-entry-passed TcpLane.ControlFrameFuzz
 
       # Both full transport runs are hard gates. The per-case comparison below
       # remains diagnostic because explicit mechanism controls retain their own mode.

MobileGL/MG_IntegrationTest/CMakeLists.txt

@@ -2129,6 +2129,20 @@ if (MOBILEGL_BUILD_DISAGGREGATED)
             LABELS integration-tcp TIMEOUT 120
             ENVIRONMENT "${MGL_ITEST_EVENT_FORFEIT_ENVIRONMENT}")
     endif()
+    # P7 F2: Ph fuzz arm 1 (PLAN-PH-P34B-P7.md §1.1, "gate: fuzz arm 1 malformed control
+    # frames"). Five malformed control frames - bad identifier, truncated length prefix, oversize
+    # length, garbage flatbuffer, wrong union type - sent to a live supervisor as a FIRST frame
+    # (answered by name with no child forked, PH-7 (5)) and inside an established session (the
+    # session ends with a named line and is not counted as a fault), with a well-formed client
+    # welcomed after each phase. Same shape as TcpLane.SupervisorProtocolControls: its own supervisor on its own
+    # port, the lane's resource lock, flatc or SKIP (77) - and CI requires it PASSED by name.
+    add_test(NAME TcpLane.ControlFrameFuzz COMMAND "${Python3_EXECUTABLE}"
+        "${CMAKE_SOURCE_DIR}/scripts/ci/ph_fuzz_control_frames.py"
+        --server "${MGL_ITEST_SPLIT_SERVER_PATH}"
+        --out "${CMAKE_CURRENT_BINARY_DIR}/tcp-control-frame-fuzz")
+    set_tests_properties(TcpLane.ControlFrameFuzz PROPERTIES
+        LABELS integration-tcp TIMEOUT 180 SKIP_RETURN_CODE 77
+        RESOURCE_LOCK mobilegl-tcp)
     # D2: the texture-view opt-in on both TCP lists too, so the three shared GLES split arms
     # carry one configuration. See the note over MGL_ITEST_GLES_SPLIT_ENVIRONMENT.
     #

MobileGL/MG_Remote/Server/ServerMain.cpp

@@ -420,6 +420,14 @@ void SendLogAck(void* pointer) {
             // below, so an apply thread that the stop wakes inside ReserveEventOrBlock names the
             // hangup (PeerGone) rather than the stop (PH-6 fix round).
             session.NoteControlStreamEnded();
+            // Fuzz arm 1, steady state. A bad magic or an over-cap length has already been named
+            // by the framing latch (Framing.h) as it latched; a clean close is the client's normal
+            // end and says nothing. The one silent case was a close INSIDE a frame - a truncated
+            // length prefix or payload - which ended the session exactly like a clean close.
+            if (result == MOBILEGL_ERR_TRANSPORT_CLOSED && control->BufferedBytes() != 0)
+                WireLogError("MG_Remote server: control connection closed inside a frame (%llu bytes of it "
+                             "received); ending the session",
+                             static_cast<unsigned long long>(control->BufferedBytes()));
             break;
         }
         // THE FILE IDENTIFIER, ASKED FIRST AND SAID OUT LOUD (P7 wave 0).
@@ -478,7 +486,15 @@ void SendLogAck(void* pointer) {
             continue;
         }
         const auto* operation = envelope->msg_as_SurfaceOp();
-        if (!operation) break;
+        if (!operation) {
+            // Fuzz arm 1, steady state: a well-formed envelope of a type this loop does not take
+            // (or a SurfaceOp tag with no table behind it) ended the session without a word.
+            WireLogError("MG_Remote server: control frame of type %u (%s) is not a LogFlush or SurfaceOp "
+                         "(size=%llu); ending the session",
+                         static_cast<unsigned>(envelope->msg_type()), Protocol::EnumNameCtrlMsg(envelope->msg_type()),
+                         static_cast<unsigned long long>(size));
+            break;
+        }
         Server::SurfaceControlFrame reply{};
         (void)ServerApplyWireSurfaceOp(*operation, &reply);
         // PH-1 (3): an op that latched (its own bytes, or a record the apply thread was on) is

MobileGL/MG_Remote/Transport/SocketTransport.cpp

@@ -777,6 +777,11 @@ namespace MobileGL::MG_Remote::Transport {
         return m_reader.PendingMessageSize();
     }
 
+    std::uint64_t SocketTransport::BufferedBytes() {
+        std::lock_guard<std::mutex> lock(m_recvMutex);
+        return m_reader.BufferedBytes();
+    }
+
     MobileGLResult SocketTransport::ShareFd(int fd, MobileGLByteSpan sideband) {
         if (m_tcp || m_auxFd < 0) {
             // A link that declared no descriptor passing. Rule G: it does not
@@ -843,6 +848,7 @@ namespace MobileGL::MG_Remote::Transport {
         return MOBILEGL_ERR_UNSUPPORTED;
     }
     std::uint64_t SocketTransport::PeekFrameSize() { return 0; }
+    std::uint64_t SocketTransport::BufferedBytes() { return 0; }
     MobileGLResult SocketTransport::ShareFd(int, MobileGLByteSpan) {
         return MOBILEGL_ERR_UNSUPPORTED;
     }

MobileGL/MG_Remote/Transport/SocketTransport.h

@@ -167,6 +167,10 @@ namespace MobileGL::MG_Remote::Transport {
         void CloseLocalCopy();
         // End requests while retaining the receive side for final diagnostics/EOF.
         MobileGLResult ShutdownSend();
+        // Bytes received but not yet returned as a whole frame. Non-zero after a TRANSPORT_CLOSED
+        // means the peer closed INSIDE a frame, which the steady-state control loop names (fuzz
+        // arm 1) rather than treating it as the client's normal end.
+        std::uint64_t BufferedBytes();
 
     private:
         // Pulls whatever the socket has into the reassembler. Returns

MobileGL/MG_Test/Wire/PreAuthGateTest.cpp

@@ -11,8 +11,8 @@
 // These are the cheap halves. The end-to-end halves - a live supervisor that must answer every
 // malformed first frame by name, fork for nothing it has not authenticated, refuse a silent peer
 // at its deadline, cap its pending connections and back an address off - are
-// TcpLane.SupervisorProtocolControls (scripts/ci/tcp_supervisor_smoke.py). What is pinned here
-// is what
+// TcpLane.SupervisorProtocolControls (scripts/ci/tcp_supervisor_smoke.py) and
+// TcpLane.ControlFrameFuzz (scripts/ci/ph_fuzz_control_frames.py). What is pinned here is what
 // those scripts cannot see from outside: that each malformation lands on the shape it is named
 // for (and not merely on SOME refusal), that the assembler never reads a byte past the first
 // frame, and the backoff's arithmetic.

scripts/ci/ph_fuzz_control_frames.py

@@ -0,0 +1,278 @@
+#!/usr/bin/env python3
+"""Ph fuzz arm 1: malformed control frames against a live TCP supervisor, each refused by name.
+
+PLAN-PH-P34B-P7.md §1.1, row "gate: fuzz arm 1 malformed control frames": the mechanism was in the
+tree (Framing.h's latch, ControlInbox) and the end-to-end half was not. This is that half. It is
+registered as `TcpLane.ControlFrameFuzz` (label integration-tcp) and required PASSED by CI the way
+`TcpLane.SupervisorProtocolControls` is - a SKIP there is red.
+
+Five malformations, delivered at the two places a control frame is read:
+
+  bad_identifier           a well-formed Hello envelope finished with the file identifier 'XXXX'
+  truncated_length_prefix  five bytes of the eight-byte frame header, then the peer's write side
+                           closes
+  oversize_length          'MGLF' and a length of 0xFFFFFFF0
+  garbage_flatbuffer       a frame whose payload carries our identifier and a root offset that
+                           points nowhere, so only the verifier can refuse it
+  wrong_union_type         a VALID envelope of a type that is not expected there (a LogFlush as a
+                           first frame; a DataBind in an established session)
+
+(1) FIRST FRAME, read by the supervisor before it has forked anything (PH-7 (5)). Each must be
+    answered with a Refuse frame naming the shape - never a bare close - with NO session child
+    forked for it (/proc, and no reap line), and the supervisor must go on serving: after all of
+    them, a well-formed client is welcomed and its session comes up. Before PH-7 (5) every one of
+    these forked a child that `_exit(67)`ed without a word (or, for the oversize length, after
+    the frame reader refused to allocate), counted as a faulted session. Three extra shapes ride
+    along because they are cheap and each has its own words: a wrong magic, the 24-byte envelope
+    whose tag says Hello and whose table is NULL, and a peer that says nothing at all.
+
+(2) STEADY STATE, read by an authenticated session child after Welcome. Each must end the session
+    with a named line in the server log - `WireLogError`, since a Refuse is a handshake answer and
+    the session is past its handshake - and the child must exit 0 (the peer's malformation is not
+    the server's fault); the next client is then welcomed. Before this package two of the five
+    (the truncated frame and the wrong union type) ended the session without a word.
+
+(3) SINGLE-SESSION SHAPE (no --serve), where the session process reads its own first frame
+    through the same classification: a wrong identifier and a silent peer are each answered by
+    name and the process exits 0, where it used to `_exit(67)` without a word.
+
+The supervisor runs with the authentication backoff OFF (MOBILEGL_IPC_AUTH_BACKOFF_AFTER=0): every
+malformation above is an authentication failure from 127.0.0.1, and with the default threshold the
+closing well-formed client would be backed off - which TcpLane.SupervisorProtocolControls' own
+`auth_backoff` control asserts on purpose, and which this arm is not about.
+
+Exit 77 (SKIP) without flatc or without the server image, like the supervisor smoke.
+"""
+import argparse
+import json
+import os
+from pathlib import Path
+import socket
+import struct
+import sys
+import tempfile
+import time
+
+sys.path.insert(0, str(Path(__file__).resolve().parent))
+import tcp_supervisor_smoke as smoke  # noqa: E402  (the shared helpers and the mirrored details)
+
+
+def framed(payload):
+    return struct.pack('<4sI', b'MGLF', len(payload)) + payload
+
+
+def envelope(schema, flatbuffers, kind, build_body=None, identifier=b'MGLC'):
+    """A framed CtrlEnvelope of union type `kind`; `build_body(builder)` returns its table, or
+    None for the NULL-table shape."""
+    builder = flatbuffers.Builder(128)
+    body = build_body(builder) if build_body else None
+    root = schema['CtrlEnvelope']
+    root.Start(builder)
+    root.AddMsgType(builder, kind)
+    if body is not None:
+        root.AddMsg(builder, body)
+    builder.Finish(root.End(builder), identifier)
+    return framed(bytes(builder.Output()))
+
+
+def log_flush(schema, flatbuffers):
+    def body(builder):
+        message = schema['LogFlush']
+        message.Start(builder)
+        message.AddSeq(builder, 1)
+        return message.End(builder)
+    return envelope(schema, flatbuffers, schema['CtrlMsg'].CtrlMsg.LogFlush, body)
+
+
+def data_bind_envelope(schema, flatbuffers):
+    def body(builder):
+        vector = builder.CreateByteVector(os.urandom(16))
+        message = schema['DataBind']
+        message.Start(builder)
+        message.AddNonce(builder, vector)
+        return message.End(builder)
+    return envelope(schema, flatbuffers, schema['CtrlMsg'].CtrlMsg.DataBind, body)
+
+
+def garbage_flatbuffer():
+    # Root offset 0xFFFFFF00 (far past the end), then OUR identifier: the identifier check passes
+    # and the verifier is the one that has to say no. Deterministic, so a green run is not luck.
+    return framed(struct.pack('<I', 0xFFFFFF00) + b'MGLC' + bytes(range(56)))
+
+
+def malformations(schema, flatbuffers, fingerprint):
+    """name -> (bytes to send, close the write side afterwards)."""
+    return {
+        'bad_identifier': (smoke.hello(schema, flatbuffers, fingerprint=fingerprint, identifier=b'XXXX'), False),
+        'truncated_length_prefix': (b'MGLF\x10', True),
+        'oversize_length': (b'MGLF' + struct.pack('<I', 0xFFFFFFF0) + bytes(56), False),
+        'garbage_flatbuffer': (garbage_flatbuffer(), False),
+        'wrong_union_type': (log_flush(schema, flatbuffers), False),
+    }
+
+
+def send_first_frame(port, schema, payload, half_close):
+    """One connection, one malformed first frame, and whatever the supervisor answers."""
+    peer = smoke.connect_control(port)
+    try:
+        peer.settimeout(10)
+        if payload:
+            peer.sendall(payload)
+        if half_close:
+            peer.shutdown(socket.SHUT_WR)
+        began = time.monotonic()
+        try:
+            answer = smoke.receive(peer, schema)
+        except (RuntimeError, OSError) as error:
+            answer = {'error': repr(error)}
+        answer['seconds'] = round(time.monotonic() - began, 3)
+        return answer
+    finally:
+        peer.close()
+
+
+def reap_lines(path):
+    return path.read_text(errors='replace').count(' reaped ') if path.is_file() else 0
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('--server', required=True)
+    parser.add_argument('--flatc', default='')
+    parser.add_argument('--out', type=Path, required=True)
+    args = parser.parse_args()
+    args.out.mkdir(parents=True, exist_ok=True)
+    flatc = smoke.resolve_flatc(args.flatc)
+    if flatc is None:
+        print('ph_fuzz_control_frames: no flatc found (--flatc, $MOBILEGL_FLATC_EXECUTABLE, '
+              '<repo>/../flatc-build/flatc); skipping', flush=True)
+        return smoke.kSkipExit
+    if not os.path.isfile(args.server):
+        print(f'ph_fuzz_control_frames: no server image at {args.server}; skipping', flush=True)
+        return smoke.kSkipExit
+    evidence = {'first_frame': {}, 'steady_state': {}}
+    with tempfile.TemporaryDirectory(prefix='mgl-fuzz-schema-') as directory:
+        flatbuffers, schema = smoke.load_schema(flatc, directory)
+        knobs = {'MOBILEGL_IPC_AUTH_BACKOFF_AFTER': '0', 'MOBILEGL_IPC_PREAUTH_MS': '1500'}
+        with smoke.supervisor(args.server, args.out / 'fuzz-supervisor.log', extra_env=knobs) as (
+                port, process, serverLog):
+            fingerprint = smoke.learn_fingerprint(port, schema, flatbuffers)['expected']
+            # The child that answered it is reaped before the phase starts counting reaps (and its
+            # reap line, written just after the waitpid, has landed).
+            smoke.wait_for_no_session(process.pid)
+            time.sleep(.2)
+
+            # ---- (1) FIRST FRAME: the supervisor, before any fork ---------------------------------
+            expected = {
+                'bad_identifier': (8, smoke.kDetailNoIdentifier),
+                'truncated_length_prefix': (8, smoke.kDetailTruncated),
+                'oversize_length': (8, smoke.kDetailNotAFrame),
+                'garbage_flatbuffer': (8, smoke.kDetailUnverifiable),
+                'wrong_union_type': (8, smoke.kDetailNotAHello),
+                'bad_magic': (8, smoke.kDetailNotAFrame),
+                'null_hello_table': (8, smoke.kDetailNotAHello),
+                'silent': (4, smoke.kDetailDeadline),
+            }
+            cases = dict(malformations(schema, flatbuffers, fingerprint))
+            cases['bad_magic'] = (b'XXXX' + bytes(60), False)
+            cases['null_hello_table'] = (envelope(schema, flatbuffers, schema['CtrlMsg'].CtrlMsg.Hello), False)
+            cases['silent'] = (b'', False)
+            reaped_before = reap_lines(serverLog)
+            for name, (payload, half_close) in cases.items():
+                answer = send_first_frame(port, schema, payload, half_close)
+                forked = smoke.child_pids(process.pid)
+                code, detail = expected[name]
+                assert answer.get('code') == code and answer.get('detail') == detail, (
+                    f'first-frame {name}: expected Refuse code {code} "{detail}", got {answer}; see {serverLog}')
+                assert not forked, f'first-frame {name}: the supervisor forked {forked} for it'
+                assert process.poll() is None, f'first-frame {name} killed the supervisor'
+                evidence['first_frame'][name] = answer
+            text = serverLog.read_text(errors='replace')
+            for name, (code, detail) in expected.items():
+                word = 'MalformedHello' if code == 8 else 'Authentication'
+                assert f'Refuse{{{word}}} {detail}' in text, (
+                    f'first-frame {name}: the supervisor log does not name "Refuse{{{word}}} {detail}"')
+            assert reap_lines(serverLog) == reaped_before, (
+                'a session child was reaped during the first-frame phase, so one was forked for a '
+                f'malformed peer; see {serverLog}')
+
+            # A well-formed client after all of them: welcomed, and its session comes up.
+            with smoke.held_session(port, schema, flatbuffers,
+                                    smoke.hello(schema, flatbuffers, fingerprint=fingerprint)) as (control, welcome):
+                assert welcome.get('welcome', 0), welcome
+                ready = smoke.wait_for_log(serverLog, f'pid={welcome["welcome"]} transport=spawn role=server ready')
+                assert ready is not None, f'the well-formed client after the first-frame phase never came up'
+            evidence['first_frame']['then_well_formed'] = {'welcome': welcome['welcome']}
+
+            # ---- (2) STEADY STATE: an authenticated session child, after Welcome ---------------
+            steady = malformations(schema, flatbuffers, fingerprint)
+            # A LogFlush is a type the steady-state loop DOES take, so the wrong union type there
+            # is a DataBind - valid, verifiable, and not a message a session reads on control.
+            steady['wrong_union_type'] = (data_bind_envelope(schema, flatbuffers), False)
+            named = {
+                'bad_identifier': 'control frame carries no CtrlEnvelope identifier',
+                'truncated_length_prefix': 'control connection closed inside a frame',
+                'oversize_length': 'frame length 4294967280 exceeds',
+                'garbage_flatbuffer': 'control frame did not verify as a CtrlEnvelope',
+                'wrong_union_type': 'is not a LogFlush or SurfaceOp',
+            }
+            for name, (payload, half_close) in steady.items():
+                with smoke.held_session(port, schema, flatbuffers,
+                                        smoke.hello(schema, flatbuffers, fingerprint=fingerprint)) as (control, welcome):
+                    assert welcome.get('welcome', 0), (name, welcome)
+                    child = welcome['welcome']
+                    ready = smoke.wait_for_log(serverLog, f'pid={child} transport=spawn role=server ready')
+                    assert ready is not None, f'steady-state {name}: session pid={child} never came up'
+                    before = len(serverLog.read_text(errors='replace'))
+                    control.sendall(payload)
+                    if half_close:
+                        control.shutdown(socket.SHUT_WR)
+                    reaped = smoke.wait_for_log(serverLog, f'pid={child} reaped exit=')
+                assert reaped is not None, f'steady-state {name}: session pid={child} did not end'
+                said = reaped[before:]
+                assert named[name] in said, (
+                    f'steady-state {name}: session pid={child} ended without naming why '
+                    f'(wanted "{named[name]}"); see {serverLog}')
+                assert f'pid={child} reaped exit=0 ' in reaped, (
+                    f'steady-state {name}: session pid={child} was counted as a fault')
+                evidence['steady_state'][name] = next(
+                    l.strip() for l in said.splitlines() if named[name] in l)
+                assert process.poll() is None, f'steady-state {name} killed the supervisor'
+
+            # And the supervisor still serves.
+            with smoke.held_session(port, schema, flatbuffers,
+                                    smoke.hello(schema, flatbuffers, fingerprint=fingerprint)) as (control, welcome):
+                assert welcome.get('welcome', 0), welcome
+                ready = smoke.wait_for_log(serverLog, f'pid={welcome["welcome"]} transport=spawn role=server ready')
+                assert ready is not None, 'the well-formed client after the steady-state phase never came up'
+            evidence['steady_state']['then_well_formed'] = {'welcome': welcome['welcome']}
+
+        # ---- (3) THE SINGLE-SESSION SHAPE: the process that IS the session reads its own first frame
+        #
+        # Without --serve there is no supervisor to read for it: RunSession reads the first frame
+        # itself (the unix endpoint's children do the same), through the same classification.
+        # That read used to `_exit(67)` on every malformation without a word; now the peer is
+        # answered by name and the process exits 0 - the malformation is the peer's, not a fault.
+        evidence['single_session'] = {}
+        for name, payload, code, detail in (
+                ('bad_identifier', smoke.hello(schema, flatbuffers, fingerprint=fingerprint, identifier=b'XXXX'),
+                 8, smoke.kDetailNoIdentifier),
+                ('silent', b'', 4, smoke.kDetailDeadline)):
+            with smoke.supervisor(args.server, args.out / f'single-session-{name}.log', serve=False,
+                                  extra_env=knobs) as (port, process, serverLog):
+                answer = send_first_frame(port, schema, payload, False)
+                try:
+                    exited = process.wait(timeout=10)
+                except Exception:  # noqa: BLE001 - still running is the red shape here
+                    exited = None
+            assert answer.get('code') == code and answer.get('detail') == detail, (
+                f'single-session {name}: expected Refuse code {code} "{detail}", got {answer}; see {serverLog}')
+            assert exited == 0, f'single-session {name}: the session process exited {exited}, not 0'
+            evidence['single_session'][name] = {'refusal': answer, 'exit': exited}
+        (args.out / 'fuzz-control-frames.json').write_text(json.dumps(evidence, indent=2))
+        print(json.dumps(evidence, indent=2))
+    return 0
+
+
+if __name__ == '__main__':
+    sys.exit(main())

scripts/ci/tcp_supervisor_smoke.py

@@ -17,7 +17,8 @@ P7 F2, PH-7 (5): the supervisor now reads and authenticates every first frame be
 the controls here assert that an unauthenticated peer learns nothing of ours
 (`unauthenticated_wrong_layout`), that a silent one costs no fork and is answered at the pre-auth
 deadline (`silent_peer`), that Busy is the answer to an AUTHENTICATED second client (`busy`), the
-pending cap and the backoff (`pending_cap`, `auth_backoff`).
+pending cap and the backoff (`pending_cap`, `auth_backoff`). The five malformed-frame shapes
+are fuzz arm 1's own entry, scripts/ci/ph_fuzz_control_frames.py.
 
 It needs flatc, which is deliberately absent from the default build graph (gen_protocol.py's
 header block says why). Resolution order is gen_protocol.py's, minus the build-it-for-you arm: